fix: harden DingTalk student synchronization

2026-07-18 14:51:53 +08:00
parent d25e451b61
commit c11f6bb614
15 changed files with 758 additions and 356 deletions
--- a/apps/server/src/sync/sync.controller.spec.ts
+++ b/apps/server/src/sync/sync.controller.spec.ts
@@ -1,3 +1,4 @@
+import { BadRequestException } from '@nestjs/common';
 import { SyncController } from './sync.controller';
 import { ScheduleSyncQueryDto } from './dto/schedule-sync.dto';

@@ -20,4 +21,26 @@ describe('SyncController — schedule sync options', () => {
      true,
    );
  });
+
+  it.each(['1abc', '0', '-1', '9007199254740992'])(
+    'rejects invalid root department id %s',
+    async (rootDeptId) => {
+      const syncService = { triggerSync: jest.fn() };
+      const controller = new SyncController(syncService as never);
+
+      await expect(controller.triggerSync('dingtalk_students', rootDeptId)).rejects.toBeInstanceOf(
+        BadRequestException,
+      );
+      expect(syncService.triggerSync).not.toHaveBeenCalled();
+    },
+  );
+
+  it('accepts a positive integer root department id', async () => {
+    const syncService = { triggerSync: jest.fn().mockResolvedValue([]) };
+    const controller = new SyncController(syncService as never);
+
+    await controller.triggerSync('dingtalk_students', '12');
+
+    expect(syncService.triggerSync).toHaveBeenCalledWith('dingtalk_students', 12);
+  });
 });
--- a/apps/server/src/sync/sync.controller.ts
+++ b/apps/server/src/sync/sync.controller.ts
@@ -25,10 +25,12 @@ export class SyncController {
  @Get('status')
  @RequirePermission('sync:read')
  async getStatus() {
-    const lastDingTalk = await this.syncService.getLastSync('dingtalk');
+    const lastDingTalkStudents = await this.syncService.getLastSync('dingtalk_students');
+    const lastDingTalkAttendance = await this.syncService.getLastSync('dingtalk_attendance');
    const lastWeCom = await this.syncService.getLastSync('wecom');
    return {
-      dingTalk: lastDingTalk ? { lastSyncAt: lastDingTalk.finishedAt, status: lastDingTalk.status } : null,
+      dingTalkStudents: lastDingTalkStudents ? { lastSyncAt: lastDingTalkStudents.finishedAt, status: lastDingTalkStudents.status } : null,
+      dingTalkAttendance: lastDingTalkAttendance ? { lastSyncAt: lastDingTalkAttendance.finishedAt, status: lastDingTalkAttendance.status } : null,
      weCom: lastWeCom ? { lastSyncAt: lastWeCom.finishedAt, status: lastWeCom.status } : null,
    };
  }
@@ -101,9 +103,12 @@ export class SyncController {


  private parseRootDeptId(rootDeptId: string): number {
-    const parsed = parseInt(rootDeptId, 10);
-    if (isNaN(parsed)) {
-      throw new BadRequestException('rootDeptId must be a valid integer');
+    if (!/^\d+$/.test(rootDeptId)) {
+      throw new BadRequestException('rootDeptId must be a positive integer');
+    }
+    const parsed = Number(rootDeptId);
+    if (!Number.isSafeInteger(parsed) || parsed < 1) {
+      throw new BadRequestException('rootDeptId must be a positive integer');
    }
    return parsed;
  }
--- a/apps/server/src/sync/sync.service.spec.ts
+++ b/apps/server/src/sync/sync.service.spec.ts
@@ -1,8 +1,97 @@
+import { ConflictException, ServiceUnavailableException } from '@nestjs/common';
+import { SyncLog } from '../entities';
+import { SyncService } from './sync.service';

+function queryBuilder(affected = 1) {
+  const builder = {
+    insert: jest.fn(),
+    update: jest.fn(),
+    values: jest.fn(),
+    orIgnore: jest.fn(),
+    set: jest.fn(),
+    where: jest.fn(),
+    andWhere: jest.fn(),
+    execute: jest.fn().mockResolvedValue({ affected }),
+  };
+  for (const method of ['insert', 'update', 'values', 'orIgnore', 'set', 'where', 'andWhere'] as const) {
+    builder[method].mockReturnValue(builder);
+  }
+  return builder;
+}

-describe('SyncService', () => {
-  it('should be defined', () => {
-    // SyncService module compiles — full tests removed with importDingTalkUsers
-    expect(true).toBe(true);
+function createService(options?: {
+  affected?: number;
+  attendanceResult?: { success: boolean; imported: number; errors: string[] };
+}) {
+  const builders = [queryBuilder(), queryBuilder(options?.affected), queryBuilder()];
+  const syncStateRepo = {
+    createQueryBuilder: jest.fn().mockImplementation(() => builders.shift()),
+    findOne: jest.fn().mockResolvedValue({ lastSyncAt: null }),
+    update: jest.fn().mockResolvedValue({ affected: 1 }),
+  };
+  const syncLogRepo = {
+    create: jest.fn().mockImplementation((value: Partial<SyncLog>) => value),
+    save: jest.fn().mockImplementation(async (value: SyncLog) => value),
+    findOne: jest.fn(),
+    find: jest.fn(),
+  };
+  const dingTalkService = {
+    syncAll: jest.fn().mockResolvedValue({ created: 1, updated: 2, conflicts: [] }),
+  };
+  const attendanceImportService = {
+    importFromDingTalk: jest.fn().mockResolvedValue(options?.attendanceResult ?? {
+      success: true,
+      imported: 3,
+      errors: [],
+    }),
+  };
+  const service = new SyncService(
+    syncLogRepo as never,
+    syncStateRepo as never,
+    { find: jest.fn().mockResolvedValue([{ dingUserId: 'u1' }]) } as never,
+    dingTalkService as never,
+    { syncAll: jest.fn().mockResolvedValue({ userCount: 0 }) } as never,
+    attendanceImportService as never,
+    {} as never,
+  );
+  return { service, syncStateRepo, syncLogRepo, dingTalkService, attendanceImportService };
+}
+
+describe('SyncService — safe DingTalk orchestration', () => {
+  it('uses a dedicated student cursor and records a successful student sync', async () => {
+    const { service, syncStateRepo, syncLogRepo } = createService();
+
+    const log = await service.syncDingTalkStudents(9);
+
+    expect(log).toMatchObject({
+      platform: 'dingtalk_students',
+      status: 'success',
+      recordsCount: 3,
+    });
+    expect(syncStateRepo.update).toHaveBeenCalledWith(
+      { platform: 'dingtalk_students' },
+      expect.objectContaining({ lastSyncAt: expect.any(Date) }),
+    );
+    expect(syncLogRepo.save).toHaveBeenCalled();
+  });
+
+  it('rejects a second run when the database lease is held', async () => {
+    const { service } = createService({ affected: 0 });
+
+    await expect(service.syncDingTalkStudents()).rejects.toBeInstanceOf(ConflictException);
+  });
+
+  it('does not advance attendance cursor when import reports failure', async () => {
+    const { service, syncStateRepo, syncLogRepo } = createService({
+      attendanceResult: { success: false, imported: 0, errors: ['upstream failed'] },
+    });
+
+    await expect(service.syncDingTalkAttendance()).rejects.toBeInstanceOf(
+      ServiceUnavailableException,
+    );
+    expect(syncStateRepo.update).not.toHaveBeenCalled();
+    expect(syncLogRepo.save).toHaveBeenLastCalledWith(
+      expect.objectContaining({ status: 'failed', errorMessage: expect.stringContaining('upstream failed') }),
+    );
  });
 });
--- a/apps/server/src/sync/sync.service.ts
+++ b/apps/server/src/sync/sync.service.ts
@@ -1,16 +1,19 @@
-import { Injectable, Logger } from '@nestjs/common';
+import { ConflictException, Injectable, Logger, ServiceUnavailableException } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
+import { randomUUID } from 'node:crypto';
 import { Repository } from 'typeorm';
 import { SyncLog, SyncState, StudentDingMapping } from '../entities';
-import type { SyncPlatform, SyncType, SyncStatus } from '../entities/sync-log.entity';
+import type { SyncPlatform, SyncStatus, SyncType } from '../entities/sync-log.entity';
+import { AttendanceImportService } from '../attendance/attendance-import.service';
 import { DingTalkService } from '../integration/dingtalk.service';
 import { WeComService } from '../integration/wecom.service';
-import { AttendanceImportService } from '../attendance/attendance-import.service';
 import { ScheduleSyncService } from './schedule-sync.service';

@Injectable()
 export class SyncService {
  private readonly logger = new Logger(SyncService.name);
+  private static readonly LEASE_MS = 30 * 60 * 1000;
+
  constructor(
    @InjectRepository(SyncLog)
    private readonly syncLogRepo: Repository<SyncLog>,
@@ -18,82 +21,70 @@ export class SyncService {
    private readonly syncStateRepo: Repository<SyncState>,
    @InjectRepository(StudentDingMapping)
    private readonly studentDingMappingRepo: Repository<StudentDingMapping>,
-
    private readonly dingTalkService: DingTalkService,
    private readonly weComService: WeComService,
    private readonly attendanceImportService: AttendanceImportService,
    private readonly scheduleSyncService: ScheduleSyncService,
  ) {}

-  // ── Scheduled sync disabled — use manual trigger via UI ──
-
-  // ── Sync DingTalk ──
-  async syncDingTalk(rootDeptId = 1): Promise<SyncLog> {
-    const platform: SyncPlatform = 'dingtalk';
-    const syncType = await this.determineSyncType(platform);
-
-    const log = await this.createSyncLog(platform, syncType, 'running');
-
-    try {
-      const lastSyncAt = await this.getLastSyncAt(platform);
-      this.logger.log(`Syncing ${platform} (${syncType}), lastSyncAt: ${lastSyncAt}`);
-
-      // ── Call existing integration APIs ──
-      // Integration hooks — extend here to call DingTalk APIs with lastSyncAt
-      const recordsCount = await this.performDingTalkSync(lastSyncAt, rootDeptId);
-
-      await this.updateLastSyncAt(platform);
-      await this.finishSyncLog(log, 'success', recordsCount);
-      this.logger.log(`DingTalk sync complete: ${recordsCount} records`);
-    } catch (error: unknown) {
-      const message = error instanceof Error ? error.message : String(error);
-      await this.finishSyncLog(log, 'failed', 0, message);
-      this.logger.error(`DingTalk sync failed: ${message}`, error instanceof Error ? error.stack : undefined);
-    }
-
-    return log;
+  async syncDingTalkStudents(rootDeptId = 1): Promise<SyncLog> {
+    return this.runSync('dingtalk_students', async () => {
+      const result = await this.dingTalkService.syncAll(rootDeptId);
+      return {
+        recordsCount: result.created + result.updated,
+        status: result.conflicts.length ? 'partial' : 'success',
+        message: result.conflicts.length ? JSON.stringify(result.conflicts.slice(0, 20)) : undefined,
+      };
+    });
+  }
+
+  async syncDingTalkAttendance(): Promise<SyncLog> {
+    return this.runSync('dingtalk_attendance', async (lastSyncAt) => {
+      const endDate = new Date();
+      const startDate = lastSyncAt ? new Date(lastSyncAt) : new Date(endDate);
+      if (!lastSyncAt) startDate.setDate(startDate.getDate() - 7);
+
+      const mappings = await this.studentDingMappingRepo.find();
+      const userIds = [...new Set(mappings.map((mapping) => mapping.dingUserId).filter(Boolean))];
+      if (userIds.length === 0) {
+        throw new ServiceUnavailableException('没有可用于考勤导入的钉钉学生映射');
+      }
+
+      const result = await this.attendanceImportService.importFromDingTalk({
+        startDate: startDate.toISOString().slice(0, 10),
+        endDate: endDate.toISOString().slice(0, 10),
+        userIds,
+        autoMatch: true,
+      });
+      if (!result.success) {
+        throw new ServiceUnavailableException(result.errors.join('；') || '钉钉考勤导入失败');
+      }
+      return { recordsCount: result.imported, status: 'success' };
+    });
  }

-  // ── Sync WeCom ──
  async syncWeCom(): Promise<SyncLog> {
-    const platform: SyncPlatform = 'wecom';
-    const syncType = await this.determineSyncType(platform);
-
-    const log = await this.createSyncLog(platform, syncType, 'running');
-
-    try {
-      const lastSyncAt = await this.getLastSyncAt(platform);
-      this.logger.log(`Syncing ${platform} (${syncType}), lastSyncAt: ${lastSyncAt}`);
-
-      // ── Call existing integration APIs ──
-      // Integration hooks — extend here to call WeCom APIs with lastSyncAt
-      const recordsCount = await this.performWeComSync(lastSyncAt);
-
-      await this.updateLastSyncAt(platform);
-      await this.finishSyncLog(log, 'success', recordsCount);
-      this.logger.log(`WeCom sync complete: ${recordsCount} records`);
-    } catch (error: unknown) {
-      const message = error instanceof Error ? error.message : String(error);
-      await this.finishSyncLog(log, 'failed', 0, message);
-      this.logger.error(`WeCom sync failed: ${message}`, error instanceof Error ? error.stack : undefined);
-    }
-
-    return log;
+    return this.runSync('wecom', async () => {
+      const result = await this.weComService.syncAll();
+      return { recordsCount: result.userCount, status: 'success' };
+    });
  }

-  // ── Manual trigger ──
  async triggerSync(platform?: SyncPlatform, rootDeptId = 1): Promise<SyncLog[]> {
-    if (platform === 'dingtalk') return [await this.syncDingTalk(rootDeptId)];
+    if (platform === 'dingtalk_students') return [await this.syncDingTalkStudents(rootDeptId)];
+    if (platform === 'dingtalk_attendance') return [await this.syncDingTalkAttendance()];
    if (platform === 'wecom') return [await this.syncWeCom()];
-    return [await this.syncDingTalk(rootDeptId), await this.syncWeCom()];
+    return [
+      await this.syncDingTalkStudents(rootDeptId),
+      await this.syncDingTalkAttendance(),
+      await this.syncWeCom(),
+    ];
  }

-  /** 获取钉钉组织部门树，供前端选择同步起点 */
  async getDingTalkOrgTree(rootDeptId = 1) {
    return this.dingTalkService.fetchOrgTree(rootDeptId);
  }

-  /** 获取钉钉组织部门树（含用户），供前端同步用户选择器使用 */
  async getDingTalkOrgTreeWithUsers(rootDeptId = 1) {
    return this.dingTalkService.fetchOrgTreeWithUsers(rootDeptId);
  }
@@ -121,9 +112,6 @@ export class SyncService {
    return { total: groups.length, deleted, failed };
  }

-  // ── 排班同步 ──
-
-  /** 将本地排课同步到钉钉考勤排班 */
  async syncScheduleToDingTalk(
    dateFrom?: string,
    days = 30,
@@ -132,31 +120,82 @@ export class SyncService {
    return this.scheduleSyncService.syncAll(dateFrom, days, 'manager', attendanceMachineOnly);
  }

-  /** 获取排班同步状态（当前仅返回活跃排课统计） */
  async getScheduleSyncStatus(date?: string) {
-    const targetDate = date || new Date().toISOString().slice(0, 10);
-    return this.scheduleSyncService.getStatus(targetDate);
+    return this.scheduleSyncService.getStatus(date || new Date().toISOString().slice(0, 10));
  }

-  // ── Sync log queries ──
-  async getLogs(platform?: SyncPlatform, limit: number = 50): Promise<SyncLog[]> {
+  async getLogs(platform?: SyncPlatform, limit = 50): Promise<SyncLog[]> {
    const where: Record<string, SyncPlatform> = {};
    if (platform) where.platform = platform;
    return this.syncLogRepo.find({ where, order: { createdAt: 'DESC' }, take: limit });
  }

-  async getLastSync(platform: SyncPlatform): Promise<SyncLog | null> {
-    return this.syncLogRepo.findOne({
-      where: { platform },
-      order: { createdAt: 'DESC' },
-    });
+  async getLastSync(platform: SyncPlatform | 'dingtalk'): Promise<SyncLog | null> {
+    if (platform === 'dingtalk') {
+      return this.syncLogRepo.findOne({
+        where: [{ platform: 'dingtalk_students' }, { platform: 'dingtalk_attendance' }],
+        order: { createdAt: 'DESC' },
+      });
+    }
+    return this.syncLogRepo.findOne({ where: { platform }, order: { createdAt: 'DESC' } });
  }

-  // ── Private helpers ──
+  private async runSync(
+    platform: SyncPlatform,
+    operation: (lastSyncAt: Date | null) => Promise<{
+      recordsCount: number;
+      status: Extract<SyncStatus, 'success' | 'partial'>;
+      message?: string;
+    }>,
+  ): Promise<SyncLog> {
+    const runId = await this.acquireLease(platform);
+    let log: SyncLog | undefined;
+    try {
+      const lastSyncAt = await this.getLastSyncAt(platform);
+      log = await this.createSyncLog(platform, lastSyncAt ? 'incremental' : 'full', 'running');
+      const result = await operation(lastSyncAt);
+      await this.syncStateRepo.update({ platform }, { lastSyncAt: new Date() });
+      await this.finishSyncLog(log, result.status, result.recordsCount, result.message);
+      return log;
+    } catch (error: unknown) {
+      const message = error instanceof Error ? error.message : String(error);
+      if (log) await this.finishSyncLog(log, 'failed', 0, message);
+      this.logger.error(`${platform} sync failed: ${message}`, error instanceof Error ? error.stack : undefined);
+      throw error;
+    } finally {
+      await this.releaseLease(platform, runId);
+    }
+  }

-  private async determineSyncType(platform: SyncPlatform): Promise<SyncType> {
-    const state = await this.syncStateRepo.findOne({ where: { platform } });
-    return state?.lastSyncAt ? 'incremental' : 'full';
+  private async acquireLease(platform: SyncPlatform): Promise<string> {
+    await this.syncStateRepo
+      .createQueryBuilder()
+      .insert()
+      .values({ platform, lastSyncAt: null, runId: null, runningSince: null })
+      .orIgnore()
+      .execute();
+
+    const runId = randomUUID();
+    const result = await this.syncStateRepo
+      .createQueryBuilder()
+      .update()
+      .set({ runId, runningSince: new Date() })
+      .where('platform = :platform', { platform })
+      .andWhere('(running_since IS NULL OR running_since < :staleBefore)', {
+        staleBefore: new Date(Date.now() - SyncService.LEASE_MS),
+      })
+      .execute();
+    if (result.affected !== 1) throw new ConflictException(`${platform} 同步正在进行中`);
+    return runId;
+  }
+
+  private async releaseLease(platform: SyncPlatform, runId: string): Promise<void> {
+    await this.syncStateRepo
+      .createQueryBuilder()
+      .update()
+      .set({ runId: null, runningSince: null })
+      .where('platform = :platform AND run_id = :runId', { platform, runId })
+      .execute();
  }

  private async getLastSyncAt(platform: SyncPlatform): Promise<Date | null> {
@@ -164,26 +203,20 @@ export class SyncService {
    return state?.lastSyncAt ?? null;
  }

-  private async updateLastSyncAt(platform: SyncPlatform): Promise<void> {
-    await this.syncStateRepo.upsert(
-      { platform, lastSyncAt: new Date() },
-      ['platform'],
-    );
-  }
-
  private async createSyncLog(
    platform: SyncPlatform,
    syncType: SyncType,
    status: SyncStatus,
  ): Promise<SyncLog> {
-    const log = this.syncLogRepo.create({
-      platform,
-      syncType,
-      status,
-      recordsCount: 0,
-      startedAt: new Date(),
-    });
-    return this.syncLogRepo.save(log);
+    return this.syncLogRepo.save(
+      this.syncLogRepo.create({
+        platform,
+        syncType,
+        status,
+        recordsCount: 0,
+        startedAt: new Date(),
+      }),
+    );
  }

  private async finishSyncLog(
@@ -195,52 +228,7 @@ export class SyncService {
    log.status = status;
    log.recordsCount = recordsCount;
    log.finishedAt = new Date();
-    if (errorMessage) log.errorMessage = errorMessage;
+    log.errorMessage = errorMessage ?? null;
    await this.syncLogRepo.save(log);
  }
-
-  private async performDingTalkSync(lastSyncAt: Date | null, rootDeptId = 1): Promise<number> {
-    // Stage 1: Sync departments and users
-    const result = await this.dingTalkService.syncAll(rootDeptId);
-    let total = result.deptCount + result.userCount;
-
-    // Stage 2: Import attendance data (last 7 days or since last sync)
-    try {
-      const endDate = new Date();
-      const startDate = new Date();
-      // If never synced, import last 7 days; otherwise import since last sync
-      if (lastSyncAt) {
-        startDate.setTime(lastSyncAt.getTime());
-      } else {
-        startDate.setDate(startDate.getDate() - 7);
-      }
-
-      const start = startDate.toISOString().slice(0, 10);
-      const end = endDate.toISOString().slice(0, 10);
-
-      this.logger.log(`Importing DingTalk attendance: ${start} ~ ${end}`);
-      const mappings = await this.studentDingMappingRepo.find();
-      const userIds = mappings.map((m) => m.dingUserId);
-      const importResult = await this.attendanceImportService.importFromDingTalk({
-        startDate: start,
-        endDate: end,
-        userIds: userIds.length > 0 ? userIds : undefined,
-        autoMatch: true,
-      });
-
-      total += importResult.imported;
-      this.logger.log(`DingTalk attendance import: ${importResult.imported} imported, ${importResult.skipped} skipped`);
-    } catch (err: unknown) {
-      // Attendance import failure should not block the sync
-      const msg = err instanceof Error ? err.message : String(err);
-      this.logger.warn(`DingTalk attendance import failed (non-fatal): ${msg}`);
-    }
-
-    return total;
-  }
-
-  private async performWeComSync(_lastSyncAt: Date | null): Promise<number> {
-    const result = await this.weComService.syncAll();
-    return result.userCount;
-  }
 }