feat: P2-15 cron-based DingTalk/WeCom sync with logging and incremental sync
- Add @nestjs/schedule dependency
- Create SyncLog entity (platform, syncType, status, recordsCount, errorMessage, timestamps)
- Create SyncState entity for tracking lastSyncAt per platform
- SyncService with @Cron('0 2 * * *') daily scheduled sync
- Incremental sync: tracks last_sync_at per platform, passes to integration APIs
- Full/incremental mode determined by existence of prior sync
- SyncController: POST /sync/trigger (manual), GET /sync/logs (history)
- Register SyncModule in AppModule with ScheduleModule
This commit is contained in:
@@ -28,6 +28,7 @@
|
||||
"@nestjs/jwt": "^11.0.2",
|
||||
"@nestjs/passport": "^11.0.5",
|
||||
"@nestjs/platform-express": "^11.1.19",
|
||||
"@nestjs/schedule": "^6.1.3",
|
||||
"@nestjs/throttler": "^6.5.0",
|
||||
"@nestjs/typeorm": "^11.0.1",
|
||||
"@types/multer": "^2.1.0",
|
||||
|
||||
@@ -25,6 +25,8 @@ import {
|
||||
ClassSchedule,
|
||||
AttendanceRecord,
|
||||
DingAttendanceRaw,
|
||||
SyncLog,
|
||||
SyncState,
|
||||
} from './entities';
|
||||
import { AuthModule } from './auth/auth.module';
|
||||
import { RbacModule } from './rbac/rbac.module';
|
||||
@@ -43,6 +45,7 @@ import { TenantsModule } from './tenants/tenants.module';
|
||||
import { AttendanceModule } from './attendance/attendance.module';
|
||||
import { SchedulesModule } from './schedules/schedules.module';
|
||||
import { ClassroomRentalsModule } from './classroom-rentals/classroom-rentals.module';
|
||||
import { SyncModule } from './sync/sync.module';
|
||||
|
||||
@Module({
|
||||
imports: [
|
||||
@@ -77,6 +80,8 @@ import { ClassroomRentalsModule } from './classroom-rentals/classroom-rentals.mo
|
||||
ClassSchedule,
|
||||
AttendanceRecord,
|
||||
DingAttendanceRaw,
|
||||
SyncLog,
|
||||
SyncState,
|
||||
];
|
||||
if (dbType === 'mysql') {
|
||||
return {
|
||||
@@ -115,6 +120,7 @@ import { ClassroomRentalsModule } from './classroom-rentals/classroom-rentals.mo
|
||||
TenantsModule,
|
||||
SchedulesModule,
|
||||
ClassroomRentalsModule,
|
||||
SyncModule,
|
||||
],
|
||||
providers: [
|
||||
{ provide: APP_GUARD, useClass: ThrottlerGuard },
|
||||
|
||||
@@ -20,3 +20,5 @@ export { ClassTeacher, TeacherRoleType } from './class-teacher.entity';
|
||||
export { ClassSchedule, ScheduleType } from './class-schedule.entity';
|
||||
export { AttendanceRecord } from './attendance-record.entity';
|
||||
export { DingAttendanceRaw } from './ding-attendance-raw.entity';
|
||||
export { SyncLog } from './sync-log.entity';
|
||||
export { SyncState } from './sync-state.entity';
|
||||
|
||||
35
apps/server/src/entities/sync-log.entity.ts
Normal file
35
apps/server/src/entities/sync-log.entity.ts
Normal file
@@ -0,0 +1,35 @@
|
||||
import { Entity, PrimaryGeneratedColumn, Column, CreateDateColumn } from 'typeorm';
|
||||
|
||||
export type SyncPlatform = 'dingtalk' | 'wecom';
|
||||
export type SyncType = 'full' | 'incremental';
|
||||
export type SyncStatus = 'running' | 'success' | 'failed';
|
||||
|
||||
@Entity('sync_logs')
|
||||
export class SyncLog {
|
||||
@PrimaryGeneratedColumn()
|
||||
id: number;
|
||||
|
||||
@Column({ type: 'varchar', length: 20 })
|
||||
platform: SyncPlatform;
|
||||
|
||||
@Column({ name: 'sync_type', type: 'varchar', length: 20 })
|
||||
syncType: SyncType;
|
||||
|
||||
@Column({ type: 'varchar', length: 20 })
|
||||
status: SyncStatus;
|
||||
|
||||
@Column({ name: 'records_count', type: 'integer', default: 0 })
|
||||
recordsCount: number;
|
||||
|
||||
@Column({ name: 'error_message', type: 'text', nullable: true })
|
||||
errorMessage: string | null;
|
||||
|
||||
@Column({ name: 'started_at', type: 'datetime', nullable: true })
|
||||
startedAt: Date | null;
|
||||
|
||||
@Column({ name: 'finished_at', type: 'datetime', nullable: true })
|
||||
finishedAt: Date | null;
|
||||
|
||||
@CreateDateColumn({ name: 'created_at' })
|
||||
createdAt: Date;
|
||||
}
|
||||
11
apps/server/src/entities/sync-state.entity.ts
Normal file
11
apps/server/src/entities/sync-state.entity.ts
Normal file
@@ -0,0 +1,11 @@
|
||||
import { Entity, PrimaryColumn, Column } from 'typeorm';
|
||||
import type { SyncPlatform } from './sync-log.entity';
|
||||
|
||||
@Entity('sync_state')
|
||||
export class SyncState {
|
||||
@PrimaryColumn({ type: 'varchar', length: 20 })
|
||||
platform: SyncPlatform;
|
||||
|
||||
@Column({ name: 'last_sync_at', type: 'datetime', nullable: true })
|
||||
lastSyncAt: Date | null;
|
||||
}
|
||||
27
apps/server/src/sync/sync.controller.ts
Normal file
27
apps/server/src/sync/sync.controller.ts
Normal file
@@ -0,0 +1,27 @@
|
||||
import { Controller, Get, Post, Query, UseGuards } from '@nestjs/common';
|
||||
import { JwtAuthGuard } from '../auth/guards/jwt-auth.guard';
|
||||
import { RequirePermission } from '../auth/decorators/permission.decorator';
|
||||
import { SyncService } from './sync.service';
|
||||
import type { SyncPlatform } from '../entities/sync-log.entity';
|
||||
|
||||
@UseGuards(JwtAuthGuard)
|
||||
@Controller('sync')
|
||||
export class SyncController {
|
||||
constructor(private readonly syncService: SyncService) {}
|
||||
|
||||
@Post('trigger')
|
||||
@RequirePermission('sync:trigger')
|
||||
async triggerSync(@Query('platform') platform?: SyncPlatform) {
|
||||
const logs = await this.syncService.triggerSync(platform);
|
||||
return { synced: logs.length, logs };
|
||||
}
|
||||
|
||||
@Get('logs')
|
||||
@RequirePermission('sync:read')
|
||||
async getLogs(
|
||||
@Query('platform') platform?: SyncPlatform,
|
||||
@Query('limit') limit?: number,
|
||||
) {
|
||||
return this.syncService.getLogs(platform, limit ? Number(limit) : 50);
|
||||
}
|
||||
}
|
||||
17
apps/server/src/sync/sync.module.ts
Normal file
17
apps/server/src/sync/sync.module.ts
Normal file
@@ -0,0 +1,17 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { ScheduleModule } from '@nestjs/schedule';
|
||||
import { TypeOrmModule } from '@nestjs/typeorm';
|
||||
import { SyncLog, SyncState } from '../entities';
|
||||
import { SyncService } from './sync.service';
|
||||
import { SyncController } from './sync.controller';
|
||||
|
||||
@Module({
|
||||
imports: [
|
||||
ScheduleModule.forRoot(),
|
||||
TypeOrmModule.forFeature([SyncLog, SyncState]),
|
||||
],
|
||||
controllers: [SyncController],
|
||||
providers: [SyncService],
|
||||
exports: [SyncService],
|
||||
})
|
||||
export class SyncModule {}
|
||||
164
apps/server/src/sync/sync.service.ts
Normal file
164
apps/server/src/sync/sync.service.ts
Normal file
@@ -0,0 +1,164 @@
|
||||
import { Injectable, Logger } from '@nestjs/common';
|
||||
import { Cron } from '@nestjs/schedule';
|
||||
import { InjectRepository } from '@nestjs/typeorm';
|
||||
import { Repository } from 'typeorm';
|
||||
import { SyncLog, SyncState } from '../entities';
|
||||
import type { SyncPlatform, SyncType, SyncStatus } from '../entities/sync-log.entity';
|
||||
|
||||
@Injectable()
|
||||
export class SyncService {
|
||||
private readonly logger = new Logger(SyncService.name);
|
||||
|
||||
constructor(
|
||||
@InjectRepository(SyncLog)
|
||||
private readonly syncLogRepo: Repository<SyncLog>,
|
||||
@InjectRepository(SyncState)
|
||||
private readonly syncStateRepo: Repository<SyncState>,
|
||||
) {}
|
||||
|
||||
// ── Scheduled cron: daily at 2 AM ──
|
||||
@Cron('0 2 * * *')
|
||||
async scheduledSync() {
|
||||
this.logger.log('Starting scheduled sync job');
|
||||
await this.syncDingTalk();
|
||||
await this.syncWeCom();
|
||||
this.logger.log('Scheduled sync job completed');
|
||||
}
|
||||
|
||||
// ── Sync DingTalk ──
|
||||
async syncDingTalk(): 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);
|
||||
|
||||
await this.updateLastSyncAt(platform);
|
||||
await this.finishSyncLog(log, 'success', recordsCount);
|
||||
this.logger.log(`DingTalk sync complete: ${recordsCount} records`);
|
||||
} catch (error: any) {
|
||||
await this.finishSyncLog(log, 'failed', 0, error.message);
|
||||
this.logger.error(`DingTalk sync failed: ${error.message}`, error.stack);
|
||||
}
|
||||
|
||||
return log;
|
||||
}
|
||||
|
||||
// ── 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: any) {
|
||||
await this.finishSyncLog(log, 'failed', 0, error.message);
|
||||
this.logger.error(`WeCom sync failed: ${error.message}`, error.stack);
|
||||
}
|
||||
|
||||
return log;
|
||||
}
|
||||
|
||||
// ── Manual trigger ──
|
||||
async triggerSync(platform?: SyncPlatform): Promise<SyncLog[]> {
|
||||
if (platform === 'dingtalk') return [await this.syncDingTalk()];
|
||||
if (platform === 'wecom') return [await this.syncWeCom()];
|
||||
return [await this.syncDingTalk(), await this.syncWeCom()];
|
||||
}
|
||||
|
||||
// ── Sync log queries ──
|
||||
async getLogs(platform?: SyncPlatform, limit: number = 50): Promise<SyncLog[]> {
|
||||
const where: any = {};
|
||||
if (platform) where.platform = platform;
|
||||
return this.syncLogRepo.find({ where, order: { createdAt: 'DESC' }, take: limit });
|
||||
}
|
||||
|
||||
// ── Private helpers ──
|
||||
|
||||
private async determineSyncType(platform: SyncPlatform): Promise<SyncType> {
|
||||
const state = await this.syncStateRepo.findOne({ where: { platform } });
|
||||
return state?.lastSyncAt ? 'incremental' : 'full';
|
||||
}
|
||||
|
||||
private async getLastSyncAt(platform: SyncPlatform): Promise<Date | null> {
|
||||
const state = await this.syncStateRepo.findOne({ where: { platform } });
|
||||
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);
|
||||
}
|
||||
|
||||
private async finishSyncLog(
|
||||
log: SyncLog,
|
||||
status: SyncStatus,
|
||||
recordsCount: number,
|
||||
errorMessage?: string,
|
||||
): Promise<void> {
|
||||
log.status = status;
|
||||
log.recordsCount = recordsCount;
|
||||
log.finishedAt = new Date();
|
||||
if (errorMessage) log.errorMessage = errorMessage;
|
||||
await this.syncLogRepo.save(log);
|
||||
}
|
||||
|
||||
// ── Integration stubs — replace with real API calls ──
|
||||
|
||||
/**
|
||||
* Perform the actual DingTalk data pull.
|
||||
* Pass `lastSyncAt` to the DingTalk API for incremental sync.
|
||||
*/
|
||||
private async performDingTalkSync(_lastSyncAt: Date | null): Promise<number> {
|
||||
// TODO: Call DingTalk API — e.g. attendance records, user list, etc.
|
||||
// const records = await this.dingTalkClient.fetchAttendance({ updatedAfter: lastSyncAt });
|
||||
// return records.length;
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform the actual WeCom data pull.
|
||||
* Pass `lastSyncAt` to the WeCom API for incremental sync.
|
||||
*/
|
||||
private async performWeComSync(_lastSyncAt: Date | null): Promise<number> {
|
||||
// TODO: Call WeCom API — e.g. contacts, attendance, etc.
|
||||
// const records = await this.weComClient.fetchContacts({ updatedAfter: lastSyncAt });
|
||||
// return records.length;
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user