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, @InjectRepository(SyncState) private readonly syncStateRepo: Repository, ) {} // ── 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 { 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 { 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 { 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 { const where: any = {}; if (platform) where.platform = platform; return this.syncLogRepo.find({ where, order: { createdAt: 'DESC' }, take: limit }); } async getLastSync(platform: SyncPlatform): Promise { return this.syncLogRepo.findOne({ where: { platform }, order: { createdAt: 'DESC' }, }); } // ── Private helpers ── private async determineSyncType(platform: SyncPlatform): Promise { const state = await this.syncStateRepo.findOne({ where: { platform } }); return state?.lastSyncAt ? 'incremental' : 'full'; } private async getLastSyncAt(platform: SyncPlatform): Promise { const state = await this.syncStateRepo.findOne({ where: { platform } }); return state?.lastSyncAt ?? null; } private async updateLastSyncAt(platform: SyncPlatform): Promise { await this.syncStateRepo.upsert( { platform, lastSyncAt: new Date() }, ['platform'], ); } private async createSyncLog( platform: SyncPlatform, syncType: SyncType, status: SyncStatus, ): Promise { 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 { 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 { const appKey = process.env.DINGTALK_APP_KEY; if (!appKey) { this.logger.warn('DingTalk not configured (DINGTALK_APP_KEY missing), skipping sync'); return 0; } // 接入指引: // 1. 创建 apps/server/src/integration/dingtalk.service.ts,实现 fetchDepartments()/fetchUsers() // 2. 在 SyncModule 中注入 DingTalkService // 3. 取消以下注释并调用 this.dingTalkService.fetchDepartments() this.logger.warn('DingTalk sync stub: create DingTalkService in src/integration/ to enable real sync'); return 0; } private async performWeComSync(_lastSyncAt: Date | null): Promise { const corpId = process.env.WECOM_CORP_ID; if (!corpId) { this.logger.warn('WeCom not configured (WECOM_CORP_ID missing), skipping sync'); return 0; } this.logger.warn( 'WeCom integration service not yet implemented — add WeComService to SyncModule to enable real sync', ); return 0; } }