diff --git a/lib/__tests__/ding-workflow-sync.test.ts b/lib/__tests__/ding-workflow-sync.test.ts index 5a10fb9..78b4ce1 100644 --- a/lib/__tests__/ding-workflow-sync.test.ts +++ b/lib/__tests__/ding-workflow-sync.test.ts @@ -1,5 +1,8 @@ import { + DING_WORKFLOW_ALLOWED_PROCESS_CODES, + isAllowedDingWorkflowProcessCode, isDingWorkflowOngoing, + normalizeDingWorkflowProcessCode, sourceWorkflowTimestampToIso, } from '@/lib/ding-workflow-sync' @@ -29,4 +32,21 @@ describe('ding workflow snapshot sync semantics', () => { expect(sourceWorkflowTimestampToIso('2026-08-20T10:30:00Z')).toBe('2026-08-20T10:30:00.000Z') expect(sourceWorkflowTimestampToIso('invalid')).toBeNull() }) + + it('only accepts the two approved process types', () => { + expect(DING_WORKFLOW_ALLOWED_PROCESS_CODES).toEqual([ + 'PROC-3571C915-0506-4490-A91A-8AA7EA33F838', + 'PROC-AE9FF3E1-A442-4E6A-85CF-0F2B86409673', + ]) + expect(isAllowedDingWorkflowProcessCode(' proc-3571c915-0506-4490-a91a-8aa7ea33f838 ')).toBe(true) + expect(isAllowedDingWorkflowProcessCode('PROC-AE9FF3E1-A442-4E6A-85CF-0F2B86409673')).toBe(true) + expect(isAllowedDingWorkflowProcessCode('PROC-OTHER')).toBe(false) + expect(isAllowedDingWorkflowProcessCode(null)).toBe(false) + }) + + it('normalizes process codes before persistence and comparison', () => { + expect(normalizeDingWorkflowProcessCode(' proc-ae9ff3e1-a442-4e6a-85cf-0f2b86409673 ')) + .toBe('PROC-AE9FF3E1-A442-4E6A-85CF-0F2B86409673') + expect(normalizeDingWorkflowProcessCode('')).toBeNull() + }) }) diff --git a/lib/ding-workflow-sync.ts b/lib/ding-workflow-sync.ts index 1ea191d..404b75f 100644 --- a/lib/ding-workflow-sync.ts +++ b/lib/ding-workflow-sync.ts @@ -12,6 +12,13 @@ const DEFAULT_SYNC_INTERVAL_MS = 60 * 60 * 1000 const INITIAL_SYNC_DELAY_MS = 5_000 const UPSERT_BATCH_SIZE = 750 +export const DING_WORKFLOW_ALLOWED_PROCESS_CODES = [ + 'PROC-3571C915-0506-4490-A91A-8AA7EA33F838', + 'PROC-AE9FF3E1-A442-4E6A-85CF-0F2B86409673', +] as const +const ALLOWED_PROCESS_CODE_VALUES: string[] = [...DING_WORKFLOW_ALLOWED_PROCESS_CODES] +const ALLOWED_PROCESS_CODE_SET = new Set(DING_WORKFLOW_ALLOWED_PROCESS_CODES) + const WORKFLOW_TABLE = { schema: process.env.REPORT_DB_WORKFLOW_SCHEMA || 'rpt', table: process.env.REPORT_DB_WORKFLOW_TABLE || 'ding_Workflow', @@ -129,6 +136,15 @@ function clean(value: unknown): string | null { return normalized || null } +export function normalizeDingWorkflowProcessCode(value: unknown): string | null { + return clean(value)?.toUpperCase() || null +} + +export function isAllowedDingWorkflowProcessCode(value: unknown): boolean { + const processCode = normalizeDingWorkflowProcessCode(value) + return Boolean(processCode && ALLOWED_PROCESS_CODE_SET.has(processCode)) +} + function nonNegativeInteger(value: unknown): number { const number = Number(value) return Number.isFinite(number) && number > 0 ? Math.floor(number) : 0 @@ -216,6 +232,9 @@ async function assertDashboardAccess(session: Session | null) { async function fetchSourceWorkflows(pool: ConnectionPool, cursor: string | null) { const request = pool.request() request.input('cursor', mssql.NVarChar(64), cursor) + DING_WORKFLOW_ALLOWED_PROCESS_CODES.forEach((processCode, index) => { + request.input(`allowedProcessCode${index}`, mssql.NVarChar(255), processCode) + }) const result = await request.query(` SELECT CONVERT(nvarchar(255), processInstanceId) AS processInstanceId, @@ -235,10 +254,16 @@ async function fetchSourceWorkflows(pool: ConnectionPool, cursor: string | null) CONVERT(nvarchar(64), finishTime) AS sourceFinishedAtRaw FROM ${tableRef(WORKFLOW_TABLE)} WHERE - @cursor IS NULL - OR UPPER(LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(100), status), '')))) NOT IN ('COMPLETED', 'TERMINATED', 'CANCELED') - OR TRY_CONVERT(datetime2, NULLIF(CONVERT(nvarchar(64), updateAt), '')) >= - DATEADD(minute, -5, TRY_CONVERT(datetime2, @cursor)) + UPPER(LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(255), processCode), '')))) IN ( + @allowedProcessCode0, + @allowedProcessCode1 + ) + AND ( + @cursor IS NULL + OR UPPER(LTRIM(RTRIM(COALESCE(CONVERT(nvarchar(100), status), '')))) NOT IN ('COMPLETED', 'TERMINATED', 'CANCELED') + OR TRY_CONVERT(datetime2, NULLIF(CONVERT(nvarchar(64), updateAt), '')) >= + DATEADD(minute, -5, TRY_CONVERT(datetime2, @cursor)) + ) `) return result.recordset } @@ -324,7 +349,7 @@ function buildSnapshotPayload( originatorDeptName: clean(row.originatorDeptName), status: status.toUpperCase(), bizAction: clean(row.bizAction), - processCode: clean(row.processCode), + processCode: normalizeDingWorkflowProcessCode(row.processCode), flowType: clean(row.flowType), joinedAtRaw: clean(row.joinedAtRaw), sourceCreatedAtRaw: clean(row.sourceCreatedAtRaw), @@ -427,7 +452,7 @@ export async function syncDingWorkflowSnapshots() { const startedAt = new Date() const leaseOwner = `${hostname()}:${process.pid}:${randomUUID()}` const state = await acquireSyncLease(leaseOwner, startedAt) - if (!state) return { skipped: true, rowsUpserted: 0, ongoingCount: null } + if (!state) return { skipped: true, rowsUpserted: 0, rowsDeleted: 0, ongoingCount: null } const pool = new mssql.ConnectionPool(buildSourceConfig()) try { @@ -445,7 +470,20 @@ export async function syncDingWorkflowSnapshots() { await upsertSnapshotBatch(rows.slice(index, index + UPSERT_BATCH_SIZE), syncedAt) } - const ongoingCount = await prisma.dingWorkflowSnapshot.count({ where: { isOngoing: true } }) + const deleted = await prisma.dingWorkflowSnapshot.deleteMany({ + where: { + OR: [ + { processCode: null }, + { processCode: { notIn: ALLOWED_PROCESS_CODE_VALUES } }, + ], + }, + }) + const ongoingCount = await prisma.dingWorkflowSnapshot.count({ + where: { + isOngoing: true, + processCode: { in: ALLOWED_PROCESS_CODE_VALUES }, + }, + }) const sourceCursorRaw = maxSourceCursor(rows, state.sourceCursorRaw) await prisma.dingWorkflowSyncState.updateMany({ where: { id: SYNC_STATE_ID, leaseOwner }, @@ -460,7 +498,7 @@ export async function syncDingWorkflowSnapshots() { currentOngoingCount: ongoingCount, }, }) - return { skipped: false, rowsUpserted: rows.length, ongoingCount } + return { skipped: false, rowsUpserted: rows.length, rowsDeleted: deleted.count, ongoingCount } } catch (error) { const message = error instanceof Error ? error.message : String(error) await prisma.dingWorkflowSyncState.updateMany({ @@ -487,7 +525,12 @@ function normalizeSyncStatus(state: { status: string; lastCompletedAt: Date | nu async function readSummaryWithoutAccessCheck(): Promise { const [total, state] = await prisma.$transaction([ - prisma.dingWorkflowSnapshot.count({ where: { isOngoing: true } }), + prisma.dingWorkflowSnapshot.count({ + where: { + isOngoing: true, + processCode: { in: ALLOWED_PROCESS_CODE_VALUES }, + }, + }), prisma.dingWorkflowSyncState.findUnique({ where: { id: SYNC_STATE_ID }, select: { status: true, lastCompletedAt: true }, @@ -515,6 +558,7 @@ export async function getOngoingDingWorkflowProjects( const query = clean(params?.query) const where = { isOngoing: true, + processCode: { in: ALLOWED_PROCESS_CODE_VALUES }, ...(query ? { OR: [ { title: { contains: query, mode: 'insensitive' as const } }, @@ -604,7 +648,7 @@ export function ensureHourlyDingWorkflowSyncStarted() { console.info('[ding-workflow-sync] 另一实例正在同步,本轮跳过') return } - console.info(`[ding-workflow-sync] 同步完成:更新 ${result.rowsUpserted} 条,进行中 ${result.ongoingCount} 条`) + console.info(`[ding-workflow-sync] 同步完成:更新 ${result.rowsUpserted} 条,清理 ${result.rowsDeleted} 条,进行中 ${result.ongoingCount} 条`) }) .catch((error) => console.warn('[ding-workflow-sync] 同步失败:', error)) }