import { ConflictException, Injectable, Logger } from '@nestjs/common'; import { InjectRepository } from '@nestjs/typeorm'; import { randomUUID } from 'node:crypto'; import { Repository } from 'typeorm'; import { SyncLog, SyncState } from '../entities'; import type { SyncPlatform, SyncStatus, SyncType } from '../entities/sync-log.entity'; const LEASE_MS = 30 * 60 * 1000; @Injectable() export class SyncRunner { private readonly logger = new Logger('SyncRunner'); constructor( @InjectRepository(SyncState) private readonly syncStateRepo: Repository, @InjectRepository(SyncLog) private readonly syncLogRepo: Repository, ) {} async run( platform: SyncPlatform, operation: (lastSyncAt: Date | null) => Promise<{ recordsCount: number; status: Extract; message?: string; }>, ): Promise { 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 { // releaseLease 自身兜底:释放租约失败不能掩盖原始的同步结果/错误 try { await this.releaseLease(platform, runId); } catch (error) { this.logger.error( `${platform} sync lease release failed`, error instanceof Error ? error.stack : undefined, ); } } } private async acquireLease(platform: SyncPlatform): Promise { 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() - LEASE_MS), }) .execute(); if (result.affected !== 1) throw new ConflictException(`${platform} 同步正在进行中`); return runId; } private async releaseLease(platform: SyncPlatform, runId: string): Promise { 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 { const state = await this.syncStateRepo.findOne({ where: { platform } }); return state?.lastSyncAt ?? null; } private async createSyncLog( platform: SyncPlatform, syncType: SyncType, status: SyncStatus, ): Promise { return this.syncLogRepo.save( this.syncLogRepo.create({ platform, syncType, status, recordsCount: 0, startedAt: new Date(), }), ); } private async finishSyncLog( log: SyncLog, status: SyncStatus, recordsCount: number, errorMessage?: string, ): Promise { log.status = status; log.recordsCount = recordsCount; log.finishedAt = new Date(); log.errorMessage = errorMessage ?? null; await this.syncLogRepo.save(log); } }