From d0cd7ab9797d4c11033bdf4236d21e71e807a0ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B1=9F=E6=9E=97=E9=93=96?= Date: Thu, 17 Sep 2026 16:14:14 +0800 Subject: [PATCH] feat: reconcile workflow product ids daily --- PRODUCT.md | 2 +- .../PerformanceConversionDashboard.test.tsx | 33 +++- .../PerformanceConversionDashboard.tsx | 41 ++++- .../PerformanceConversionDetail.test.tsx | 1 + .../PerformanceConversionDetail.tsx | 2 +- .../performance-conversion.ts | 9 +- instrumentation.ts | 4 +- lib/__tests__/ding-workflow-sync.test.ts | 38 +++- ...ormance-conversion-workflow-filter.test.ts | 10 ++ lib/ding-workflow-sync.ts | 168 ++++++++++++++---- lib/performance-conversion.ts | 15 +- prisma/schema.prisma | 2 +- 12 files changed, 263 insertions(+), 62 deletions(-) diff --git a/PRODUCT.md b/PRODUCT.md index e1293bc..0f266e6 100644 --- a/PRODUCT.md +++ b/PRODUCT.md @@ -30,7 +30,7 @@ web - “进行中项目”项目明细按业务导出表展示审批标题、审批编号、需求人、需求部门、平台、品牌、类目、SPU、需求类型、标准时效、当前节点与负责人；需求人取流程发起人，需求部门按 HR 人员架构匹配。 - 美术设计流程的明细开始时间取“品牌策划主管分配任务”节点创建时间，结束时间取“品牌策划主管审核”节点完成时间；内容视频流程的明细开始时间取已确认的“内容主管分配任务”节点创建时间，结束时间按规则分支取“编导审核”或“内容主管审核”节点完成时间。内容流程中多个主管分配节点、编导制作节点及普通领导者任务均不等同于最终审核；未知 activityId 不得仅凭负责人身份、AGREE 结果或完成状态推断为审核节点。 - 美术设计申请流程的 `ding_WorkflowTask.activityId` 按已确认流程定义映射为精确中文节点名并写入本地快照；未知节点继续使用同步字段推断或明确回退，不改写上游只读数据。 -- 钉钉流程快照每小时从只读报表库的 `ding_Workflow`、`ding_WorkflowTask`、`ding_WorkflowScheme` 联合增量同步，只接收已批准的美术设计申请和内容视频流程；三表复合游标写入现有同步状态并继续复用同一批生产快照数据。 +- 钉钉流程快照每小时从只读报表库的 `ding_Workflow`、`ding_WorkflowTask`、`ding_WorkflowScheme` 联合增量同步，只接收已批准的美术设计申请和内容视频流程；三表复合游标写入现有同步状态，并每 24 小时全量校准流程、品牌和商品 ID，修正上游未推进游标的历史表单改动。 - “本月上新”只读取表单字段“需求类型”为“新品”的六品牌美术设计流程，不以流程标题关键词判断，也不计“优化”等其他需求类型；分子为本月整条流程已完结数量，分母为本月已完结数量与当前进行中数量之和；其中进行中流程直接复用“进行中项目”的美术设计流程状态、时间与当前品牌中心处理人条件。 - “按时交付”只统计对应主管/规则审核节点在本月完成、审批状态为已完结且审批结论未拒绝的美术设计与内容视频流程；待计算流程不进入分母。详情页可在全部流程、美术设计流程和内容视频流程之间筛选，公式、数量与明细必须共用当前筛选口径。 - “首轮通过”读取作品过稿数据，分子为 V1 版本审核通过的作品数，分母为当前已通过的作品数；详情页列出部门范围内所有含 V1 的作品，搜索只过滤明细，不改变顶部完整口径。 diff --git a/components/department-dashboard/PerformanceConversionDashboard.test.tsx b/components/department-dashboard/PerformanceConversionDashboard.test.tsx index 3f0690e..993e854 100644 --- a/components/department-dashboard/PerformanceConversionDashboard.test.tsx +++ b/components/department-dashboard/PerformanceConversionDashboard.test.tsx @@ -16,7 +16,11 @@ const emptyList = { page: 1, pageSize: 20, hasMore: false, - lastWorkflowSyncedAt: '2026-09-17T00:00:00.000Z', + workflowSync: { + status: 'ready' as const, + lastCompletedAt: '2099-09-17T01:00:00.000Z', + lastFullReconciledAt: '2099-09-17T00:00:00.000Z', + }, } const workflowItem = { @@ -65,6 +69,33 @@ describe('PerformanceConversionDashboard filters', () => { expect(requestUrl).toContain('completedTo=2026-07-31') expect(window.location.search).toContain('completedFrom=2026-07-01') expect(window.location.search).toContain('completedTo=2026-07-31') + expect(screen.getByText('流程与商品 ID 每日增量同步')).toBeInTheDocument() + expect(screen.getByText('最近成功 2099/09/17 09:00')).toBeInTheDocument() + }) + + it.each([ + ['syncing', '正在同步流程与商品 ID'], + ['error', '流程与商品 ID 同步异常'], + ['pending', '等待首次流程与商品 ID 同步'], + ] as const)('shows the %s workflow synchronization state', async (status, label) => { + global.fetch = jest.fn().mockResolvedValue({ + ok: true, + json: async () => ({ + success: true, + data: { + ...emptyList, + workflowSync: { + status, + lastCompletedAt: status === 'pending' ? null : '2099-09-17T01:00:00.000Z', + lastFullReconciledAt: status === 'pending' ? null : '2099-09-17T00:00:00.000Z', + }, + }, + }), + } as Response) + + render() + + expect(await screen.findByText(label)).toBeInTheDocument() }) it('opens each product ID on a separate detail page and preserves the list context', async () => { diff --git a/components/department-dashboard/PerformanceConversionDashboard.tsx b/components/department-dashboard/PerformanceConversionDashboard.tsx index 4e8d9a6..3b8c2f6 100644 --- a/components/department-dashboard/PerformanceConversionDashboard.tsx +++ b/components/department-dashboard/PerformanceConversionDashboard.tsx @@ -4,15 +4,19 @@ import Link from 'next/link' import { FormEvent, useCallback, useEffect, useRef, useState } from 'react' import { Activity, + AlertTriangle, CalendarDays, + CheckCircle2, ChevronLeft, ChevronRight, + Clock3, RefreshCw, Search, } from 'lucide-react' import DepartmentShell from './DepartmentShell' import { BUSINESS_TREND_BRANDS, + type PerformanceConversionWorkflowSync, type PerformanceConversionWorkflowItem, type PerformanceConversionWorkflowList, isValidDateKey, @@ -43,6 +47,41 @@ function formatDate(value: string | null, includeTime = false) { }).format(date) } +function WorkflowSyncStatus({ sync }: { sync: PerformanceConversionWorkflowSync }) { + const latestSuccess = sync.lastCompletedAt || sync.lastFullReconciledAt + const latestSuccessLabel = latestSuccess ? formatDate(latestSuccess, true) : null + const latestReconciliation = sync.lastFullReconciledAt || sync.lastCompletedAt + const isDelayed = sync.status === 'ready' + && latestReconciliation !== null + && Date.now() - new Date(latestReconciliation).getTime() > 30 * 60 * 60 * 1000 + + const view = sync.status === 'syncing' + ? { label: '正在同步流程与商品 ID', detail: latestSuccessLabel ? `上次成功 ${latestSuccessLabel}` : '首次同步进行中', tone: 'text-[#0050cb]', icon: RefreshCw, spin: true } + : sync.status === 'error' + ? { label: '流程与商品 ID 同步异常', detail: latestSuccessLabel ? `暂显示 ${latestSuccessLabel} 的数据` : '暂无可用同步记录', tone: 'text-amber-700', icon: AlertTriangle, spin: false } + : sync.status === 'pending' + ? { label: '等待首次流程与商品 ID 同步', detail: '后台任务尚未完成', tone: 'text-slate-500', icon: Clock3, spin: false } + : isDelayed + ? { label: '流程与商品 ID 每日校准延迟', detail: `最近增量成功 ${latestSuccessLabel}`, tone: 'text-amber-700', icon: AlertTriangle, spin: false } + : { label: '流程与商品 ID 每日增量同步', detail: latestSuccessLabel ? `最近成功 ${latestSuccessLabel}` : '等待首次成功记录', tone: 'text-emerald-700', icon: CheckCircle2, spin: false } + const Icon = view.icon + + return ( +
+

+

+

{view.detail}

+
+ ) +} + function WorkflowTable({ items, detailHref }: { items: PerformanceConversionWorkflowItem[] detailHref: (item: PerformanceConversionWorkflowItem) => string @@ -178,7 +217,7 @@ export default function PerformanceConversionDashboard() {
-

当前品牌

{selectedBrand}

{appliedFilters.completedFrom || appliedFilters.completedTo ? '区间内 ' : ''}{selectedBrandCount} 个有效流程商品 ID{listState.data?.invalidProductIdCount ? ` · ${listState.data.invalidProductIdCount} 条流程未填写有效 ID` : ''}

流程同步 {formatDate(listState.data?.lastWorkflowSyncedAt || null, true)}
+

当前品牌

{selectedBrand}

{appliedFilters.completedFrom || appliedFilters.completedTo ? '区间内 ' : ''}{selectedBrandCount} 个有效流程商品 ID{listState.data?.invalidProductIdCount ? ` · ${listState.data.invalidProductIdCount} 条流程未填写有效 ID` : ''}

{listState.data && }
完结日期 diff --git a/components/department-dashboard/PerformanceConversionDetail.test.tsx b/components/department-dashboard/PerformanceConversionDetail.test.tsx index 0383512..1d9a417 100644 --- a/components/department-dashboard/PerformanceConversionDetail.test.tsx +++ b/components/department-dashboard/PerformanceConversionDetail.test.tsx @@ -104,6 +104,7 @@ describe('PerformanceConversionDetail', () => { expect.objectContaining({ cache: 'no-store' }), ) expect(screen.getByRole('link', { name: '返回业绩转化' })).toHaveAttribute('href', returnHref) + expect(screen.getByText(/经营数据更新至/)).toBeInTheDocument() expect(screen.getAllByText('12.50%').length).toBeGreaterThan(0) expect(screen.getAllByText('¥1,299.50').length).toBeGreaterThan(0) expect(screen.getByRole('img', { name: /支付金额从/ })).toBeInTheDocument() diff --git a/components/department-dashboard/PerformanceConversionDetail.tsx b/components/department-dashboard/PerformanceConversionDetail.tsx index 8faba0d..36bddc9 100644 --- a/components/department-dashboard/PerformanceConversionDetail.tsx +++ b/components/department-dashboard/PerformanceConversionDetail.tsx @@ -161,7 +161,7 @@ function DetailContent({ data }: { data: PerformanceConversionTimeseries }) {
审批编号
审批编号 {data.workflow.approvalNumber || '—'}
流程完结
流程完结 {formatDate(data.workflow.completedAt, true)}
-
数据更新
数据更新至 {formatDate(data.coverage.sourceLatestDate)}
+
经营数据更新
经营数据更新至 {formatDate(data.coverage.sourceLatestDate)}
{data.product.brandMatchesWorkflow === false && ( diff --git a/components/department-dashboard/performance-conversion.ts b/components/department-dashboard/performance-conversion.ts index 38cd64d..d2adcb2 100644 --- a/components/department-dashboard/performance-conversion.ts +++ b/components/department-dashboard/performance-conversion.ts @@ -1,8 +1,13 @@ import { BUSINESS_TREND_BRANDS, type BusinessTrendBrandName } from './business-trends' - export { BUSINESS_TREND_BRANDS } export type PerformanceConversionBrand = BusinessTrendBrandName +export type PerformanceConversionWorkflowSync = { + status: 'pending' | 'syncing' | 'ready' | 'error' + lastCompletedAt: string | null + lastFullReconciledAt: string | null +} + export const PERFORMANCE_CONVERSION_METRICS = [ { key: 'impressions', label: '展现量', group: '流量', format: 'integer' }, { key: 'visitors', label: '商品访客数', group: '流量', format: 'integer' }, @@ -38,7 +43,7 @@ export type PerformanceConversionWorkflowList = { page: number pageSize: number hasMore: boolean - lastWorkflowSyncedAt: string | null + workflowSync: PerformanceConversionWorkflowSync } export type PerformanceConversionDailyMetric = { diff --git a/instrumentation.ts b/instrumentation.ts index bf4a817..708baa5 100644 --- a/instrumentation.ts +++ b/instrumentation.ts @@ -120,12 +120,12 @@ export async function register() { console.error('[Instrumentation] 调度组织架构每日同步任务失败:', error) } - // 钉钉审批流程快照：启动后建立基线，之后每小时增量同步。 + // 钉钉审批流程快照：每小时增量同步，并每 24 小时全量校准流程与商品 ID。 // 看板只读本地 PostgreSQL 快照，避免用户请求直接扫描上游无索引大表。 try { const { ensureHourlyDingWorkflowSyncStarted } = await import('./lib/ding-workflow-sync') ensureHourlyDingWorkflowSyncStarted() - console.log('[Instrumentation] 钉钉流程每小时增量同步任务已调度') + console.log('[Instrumentation] 钉钉流程增量同步与每日全量校准任务已调度') } catch (error) { console.error('[Instrumentation] 调度钉钉流程每小时同步任务失败:', error) } diff --git a/lib/__tests__/ding-workflow-sync.test.ts b/lib/__tests__/ding-workflow-sync.test.ts index 92cf7b7..da75c5f 100644 --- a/lib/__tests__/ding-workflow-sync.test.ts +++ b/lib/__tests__/ding-workflow-sync.test.ts @@ -28,7 +28,9 @@ import { parseDingWorkflowSourceCursor, resolveDingWorkflowInitiator, resolveDingWorkflowDetailMilestones, + selectLatestDingWorkflowSchemeValue, serializeDingWorkflowSourceCursor, + shouldRunDingWorkflowDailyReconciliation, sourceWorkflowTimestampToIso, storedWorkflowTimestampToIso, } from '@/lib/ding-workflow-sync' @@ -182,29 +184,32 @@ describe('ding workflow snapshot sync semantics', () => { it('upgrades the legacy workflow-only cursor to a three-table cursor', () => { expect(parseDingWorkflowSourceCursor('2026-08-26 19:42:36')).toEqual({ - version: 3, + version: 4, workflowUpdatedAt: '2026-08-26 19:42:36', taskId: null, schemeId: null, + lastFullReconciledAt: null, }) }) - it('round-trips the Workflow, Task and Scheme incremental cursors', () => { + it('round-trips the Workflow, Task and Scheme cursors with the daily reconciliation watermark', () => { const cursor = { - version: 3 as const, + version: 4 as const, workflowUpdatedAt: '2026-08-26 19:42:36', taskId: 2248339, schemeId: 10634556, + lastFullReconciledAt: '2026-09-16T18:00:00.000Z', } expect(parseDingWorkflowSourceCursor(serializeDingWorkflowSourceCursor(cursor))).toEqual(cursor) }) it('preserves empty Task and Scheme cursors so the next run performs alignment', () => { const cursor = { - version: 3 as const, + version: 4 as const, workflowUpdatedAt: '2026-08-26 19:42:36', taskId: null, schemeId: null, + lastFullReconciledAt: null, } expect(parseDingWorkflowSourceCursor(serializeDingWorkflowSourceCursor(cursor))).toEqual(cursor) }) @@ -216,27 +221,46 @@ describe('ding workflow snapshot sync semantics', () => { taskId: 2248339, schemeId: 10634556, }))).toEqual({ - version: 3, + version: 4, workflowUpdatedAt: '2026-08-26 19:42:36', taskId: null, schemeId: null, + lastFullReconciledAt: null, }) }) - it('rejects malformed version 3 cursor ids and safely requests re-alignment', () => { + it('upgrades version 3 cursors and requests the first daily reconciliation', () => { expect(parseDingWorkflowSourceCursor(JSON.stringify({ version: 3, workflowUpdatedAt: '2026-08-26 19:42:36', taskId: -1, schemeId: 'invalid', }))).toEqual({ - version: 3, + version: 4, workflowUpdatedAt: '2026-08-26 19:42:36', taskId: null, schemeId: null, + lastFullReconciledAt: null, }) }) + it('runs a full reconciliation at least once every 24 hours', () => { + const now = new Date('2026-09-17T02:00:00.000Z') + expect(shouldRunDingWorkflowDailyReconciliation(null, now)).toBe(true) + expect(shouldRunDingWorkflowDailyReconciliation('2026-09-16T02:00:01.000Z', now)).toBe(false) + expect(shouldRunDingWorkflowDailyReconciliation('2026-09-16T02:00:00.000Z', now)).toBe(true) + expect(shouldRunDingWorkflowDailyReconciliation('invalid', now)).toBe(true) + }) + + it('selects the latest non-empty Scheme row across product ID aliases', () => { + expect(selectLatestDingWorkflowSchemeValue([ + { sourceRowId: 101, name: '商品ID', value: '100000001' }, + { sourceRowId: 109, name: '商品Id', value: '100000009' }, + { sourceRowId: 110, name: '商品id', value: ' ' }, + { sourceRowId: 108, name: '品牌', value: '亿家达' }, + ], ['商品ID', '商品Id', '商品id'])).toBe('100000009') + }) + it('only queries ongoing workflows whose current assignee belongs to Brand Center', () => { expect(buildBrandCenterOngoingWorkflowWhere()).toEqual({ isOngoing: true, diff --git a/lib/__tests__/performance-conversion-workflow-filter.test.ts b/lib/__tests__/performance-conversion-workflow-filter.test.ts index 87b66fa..a6a37c5 100644 --- a/lib/__tests__/performance-conversion-workflow-filter.test.ts +++ b/lib/__tests__/performance-conversion-workflow-filter.test.ts @@ -19,6 +19,11 @@ jest.mock('@/lib/prisma', () => ({ jest.mock('@/lib/ding-workflow-sync', () => ({ DING_WORKFLOW_ALLOWED_PROCESS_CODES: ['art-process'], DING_WORKFLOW_STARTED_AT_CUTOFF_ISO: '2026-03-31T16:00:00.000Z', + getDingWorkflowSyncMetadata: jest.fn().mockResolvedValue({ + status: 'ready', + lastCompletedAt: '2026-09-17T01:30:00.000Z', + lastFullReconciledAt: '2026-09-17T00:00:00.000Z', + }), isDingWorkflowApprovedCompletion: jest.fn(() => true), })) @@ -70,6 +75,11 @@ describe('performance conversion completion date filters', () => { expect(result.invalidProductIdCount).toBe(1) expect(result.brands.find((item) => item.brand === '亿家达')?.count).toBe(2) expect(result.brands.find((item) => item.brand === '卓禾')?.count).toBe(1) + expect(result.workflowSync).toEqual({ + status: 'ready', + lastCompletedAt: '2026-09-17T01:30:00.000Z', + lastFullReconciledAt: '2026-09-17T00:00:00.000Z', + }) }) it('rejects invalid or reversed date ranges before reading snapshots', async () => { diff --git a/lib/ding-workflow-sync.ts b/lib/ding-workflow-sync.ts index cc8338e..ff54df4 100644 --- a/lib/ding-workflow-sync.ts +++ b/lib/ding-workflow-sync.ts @@ -31,9 +31,9 @@ export type { TeamCapacityWorkflowRow, } from '@/lib/ding-workflow-team-capacity' -// v4 forces one full refresh so historical snapshots also receive the newly -// collected 商品ID/现商品ID fields used by the performance conversion view. -const SYNC_STATE_ID = 'ding-workflow-hourly-v4-product-id' +// Keep this persisted key stable: changing it would discard the production +// cursor and force another baseline scan of the upstream tables. +export const DING_WORKFLOW_SYNC_STATE_ID = 'ding-workflow-hourly-v4-product-id' const TERMINAL_WORKFLOW_STATUSES = new Set(['COMPLETED', 'TERMINATED', 'CANCELED']) const REJECTED_WORKFLOW_DECISIONS = new Set([ 'REFUSE', @@ -50,6 +50,7 @@ const REJECTED_WORKFLOW_DECISIONS = new Set([ const LEASE_DURATION_MS = 20 * 60 * 1000 const DEFAULT_SYNC_INTERVAL_MS = 60 * 60 * 1000 const INITIAL_SYNC_DELAY_MS = 5_000 +const DAILY_RECONCILIATION_INTERVAL_MS = 24 * 60 * 60 * 1000 const UPSERT_BATCH_SIZE = 750 export const DING_WORKFLOW_STARTED_AT_CUTOFF_LOCAL = '2026-04-01 00:00:00' @@ -235,19 +236,24 @@ type SourceSchemeAggregate = { schemeFieldCount: unknown } -type SourceSchemeField = { - processInstanceId: unknown - componentType: unknown +export type DingWorkflowSchemeValueRow = { name: unknown value: unknown + sourceRowId?: unknown +} + +type SourceSchemeField = DingWorkflowSchemeValueRow & { + processInstanceId: unknown + componentType: unknown extValue: unknown } type DingWorkflowSourceCursor = { - version: 3 + version: 4 workflowUpdatedAt: string | null taskId: number | null schemeId: number | null + lastFullReconciledAt: string | null } type SourceDeltaState = { @@ -642,13 +648,31 @@ function extractContactAssignments(fields: SourceSchemeField[]): ContactAssignme ])).values()) } +export function selectLatestDingWorkflowSchemeValue( + fields: DingWorkflowSchemeValueRow[], + names: string[], +): string | null { + const acceptedNames = new Set(names) + const candidates = fields.flatMap((field, index) => { + const name = clean(field.name) + const value = clean(field.value) + if (!name || !acceptedNames.has(name) || !value) return [] + const sourceRowId = normalizeSourceCursorId(field.sourceRowId) + return [{ value, sourceRowId, index }] + }) + candidates.sort((left, right) => { + if (left.sourceRowId !== null && right.sourceRowId !== null && left.sourceRowId !== right.sourceRowId) { + return right.sourceRowId - left.sourceRowId + } + if (left.sourceRowId !== null) return -1 + if (right.sourceRowId !== null) return 1 + return left.index - right.index + }) + return candidates[0]?.value || null +} + function firstSchemeValue(fields: SourceSchemeField[], names: string[]): string | null { - for (const name of names) { - const value = fields.find((field) => clean(field.name) === name && clean(field.value))?.value - const normalized = clean(value) - if (normalized) return normalized - } - return null + return selectLatestDingWorkflowSchemeValue(fields, names) } function collectSchemeDisplayValues(value: unknown): string[] { @@ -1309,43 +1333,75 @@ function normalizeSourceCursorId(value: unknown): number | null { /** * 兼容旧版只保存 ding_Workflow.updateAt 的游标。 * - * v3 新增交付时效所需表单字段；读取旧版 v2 时保留主表游标，但清空 - * Task / Scheme 游标触发一次全量对齐，之后继续按三表复合游标增量同步。 - * 后续每小时只读取主表变化和 Task / Scheme 新增版本所关联的流程。 + * v4 在三表复合游标之外记录最近一次全量校准时间。小时增量继续负责 + * 快速更新，24 小时校准负责修正上游原地修改但未推进任何游标的历史表单。 */ export function parseDingWorkflowSourceCursor(raw: string | null): DingWorkflowSourceCursor { const normalized = clean(raw) if (!normalized) { - return { version: 3, workflowUpdatedAt: null, taskId: null, schemeId: null } + return { + version: 4, + workflowUpdatedAt: null, + taskId: null, + schemeId: null, + lastFullReconciledAt: null, + } } try { const parsed = JSON.parse(normalized) as Record - if (parsed.version === 3) { + if (parsed.version === 4) { return { - version: 3, + version: 4, workflowUpdatedAt: clean(parsed.workflowUpdatedAt), taskId: normalizeSourceCursorId(parsed.taskId), schemeId: normalizeSourceCursorId(parsed.schemeId), + lastFullReconciledAt: clean(parsed.lastFullReconciledAt), + } + } + if (parsed.version === 3) { + return { + version: 4, + workflowUpdatedAt: clean(parsed.workflowUpdatedAt), + taskId: normalizeSourceCursorId(parsed.taskId), + schemeId: normalizeSourceCursorId(parsed.schemeId), + lastFullReconciledAt: null, } } if (parsed.version === 2) { return { - version: 3, + version: 4, workflowUpdatedAt: clean(parsed.workflowUpdatedAt), taskId: null, schemeId: null, + lastFullReconciledAt: null, } } } catch { // Legacy cursor is a plain ding_Workflow.updateAt string. } - return { version: 3, workflowUpdatedAt: normalized, taskId: null, schemeId: null } + return { + version: 4, + workflowUpdatedAt: normalized, + taskId: null, + schemeId: null, + lastFullReconciledAt: null, + } } export function serializeDingWorkflowSourceCursor(cursor: DingWorkflowSourceCursor): string { return JSON.stringify(cursor) } +export function shouldRunDingWorkflowDailyReconciliation( + lastFullReconciledAt: string | null, + now = new Date(), +) { + if (!lastFullReconciledAt) return true + const previous = new Date(lastFullReconciledAt) + if (Number.isNaN(previous.getTime())) return true + return now.getTime() - previous.getTime() >= DAILY_RECONCILIATION_INTERVAL_MS +} + export function isReportWorkflowDbConfigured() { return Boolean( process.env.REPORT_DB_SERVER @@ -1605,6 +1661,7 @@ async function fetchWorkflowAggregates(pool: ConnectionPool, processInstanceIds: ) SELECT CONVERT(nvarchar(255), scheme.processInstanceId) AS processInstanceId, + scheme.id AS sourceRowId, CONVERT(nvarchar(500), scheme.componentType) AS componentType, CONVERT(nvarchar(2000), scheme.name) AS name, CONVERT(nvarchar(max), scheme.value) AS value, @@ -1622,7 +1679,8 @@ async function fetchWorkflowAggregates(pool: ConnectionPool, processInstanceIds: '数量', '需求数量', '拍摄数量', '摄像数量', '摄影数量', '策划任务统计', '平面任务统计', '出图任务统计' ) - OR CONVERT(nvarchar(500), scheme.componentType) IN ('InnerContactField', 'TableField'); + OR CONVERT(nvarchar(500), scheme.componentType) IN ('InnerContactField', 'TableField') + ORDER BY scheme.processInstanceId, scheme.id DESC; `) const taskRows = (result.recordsets?.[0] || []) as SourceTaskAggregate[] @@ -1920,13 +1978,13 @@ async function upsertSnapshotBatch(rows: WorkflowSnapshotPayload[], syncedAt: Da async function acquireSyncLease(leaseOwner: string, startedAt: Date) { await prisma.dingWorkflowSyncState.upsert({ - where: { id: SYNC_STATE_ID }, - create: { id: SYNC_STATE_ID }, + where: { id: DING_WORKFLOW_SYNC_STATE_ID }, + create: { id: DING_WORKFLOW_SYNC_STATE_ID }, update: {}, }) const claimed = await prisma.dingWorkflowSyncState.updateMany({ where: { - id: SYNC_STATE_ID, + id: DING_WORKFLOW_SYNC_STATE_ID, OR: [ { leaseExpiresAt: null }, { leaseExpiresAt: { lt: startedAt } }, @@ -1942,7 +2000,16 @@ async function acquireSyncLease(leaseOwner: string, startedAt: Date) { }, }) if (claimed.count !== 1) return null - return prisma.dingWorkflowSyncState.findUniqueOrThrow({ where: { id: SYNC_STATE_ID } }) + return prisma.dingWorkflowSyncState.findUniqueOrThrow({ where: { id: DING_WORKFLOW_SYNC_STATE_ID } }) +} + +async function renewSyncLease(leaseOwner: string) { + const now = new Date() + const renewed = await prisma.dingWorkflowSyncState.updateMany({ + where: { id: DING_WORKFLOW_SYNC_STATE_ID, leaseOwner }, + data: { leaseExpiresAt: new Date(now.getTime() + LEASE_DURATION_MS) }, + }) + if (renewed.count !== 1) throw new Error('钉钉流程同步租约已失效，本轮停止写入') } export async function syncDingWorkflowSnapshots() { @@ -1960,6 +2027,8 @@ export async function syncDingWorkflowSnapshots() { await pool.connect() const sourceCursor = parseDingWorkflowSourceCursor(state.sourceCursorRaw) const sourceDelta = await fetchSourceDeltaState(pool, sourceCursor) + const performFullReconciliation = sourceDelta.requiresFullSync + || shouldRunDingWorkflowDailyReconciliation(sourceCursor.lastFullReconciledAt, startedAt) const knownOngoingRows = await prisma.dingWorkflowSnapshot.findMany({ where: { isOngoing: true, @@ -1970,10 +2039,11 @@ export async function syncDingWorkflowSnapshots() { const knownOngoingWorkflowIds = knownOngoingRows.map((row) => row.processInstanceId) const sourceRows = await fetchSourceWorkflows( pool, - sourceDelta.requiresFullSync ? null : sourceCursor.workflowUpdatedAt, + performFullReconciliation ? null : sourceCursor.workflowUpdatedAt, sourceDelta.changedWorkflowIds, knownOngoingWorkflowIds, ) + await renewSyncLease(leaseOwner) const workflowIds = sourceRows.flatMap((row) => { const id = clean(row.processInstanceId) return id ? [id] : [] @@ -1984,6 +2054,7 @@ export async function syncDingWorkflowSnapshots() { schemeByWorkflow, schemeRowsByWorkflow, } = await fetchWorkflowAggregates(pool, workflowIds) + await renewSyncLease(leaseOwner) const employeeDirectory = await readEmployeeDirectory(sourceRows, taskRowsByWorkflow) const activityNodeNames = inferActivityNodeNames(sourceRows, taskRowsByWorkflow, schemeRowsByWorkflow) const rows = buildSnapshotPayload( @@ -1999,6 +2070,7 @@ export async function syncDingWorkflowSnapshots() { for (let index = 0; index < rows.length; index += UPSERT_BATCH_SIZE) { await upsertSnapshotBatch(rows.slice(index, index + UPSERT_BATCH_SIZE), syncedAt) + await renewSyncLease(leaseOwner) } const deleted = await prisma.dingWorkflowSnapshot.deleteMany({ @@ -2015,13 +2087,16 @@ export async function syncDingWorkflowSnapshots() { where: buildBrandCenterOngoingWorkflowWhere(), }) const sourceCursorRaw = serializeDingWorkflowSourceCursor({ - version: 3, + version: 4, workflowUpdatedAt: maxSourceCursor(rows, sourceCursor.workflowUpdatedAt), taskId: sourceDelta.maxTaskId, schemeId: sourceDelta.maxSchemeId, + lastFullReconciledAt: performFullReconciliation + ? syncedAt.toISOString() + : sourceCursor.lastFullReconciledAt, }) await prisma.dingWorkflowSyncState.updateMany({ - where: { id: SYNC_STATE_ID, leaseOwner }, + where: { id: DING_WORKFLOW_SYNC_STATE_ID, leaseOwner }, data: { status: 'ready', sourceCursorRaw, @@ -2037,7 +2112,7 @@ export async function syncDingWorkflowSnapshots() { } catch (error) { const message = error instanceof Error ? error.message : String(error) await prisma.dingWorkflowSyncState.updateMany({ - where: { id: SYNC_STATE_ID, leaseOwner }, + where: { id: DING_WORKFLOW_SYNC_STATE_ID, leaseOwner }, data: { status: 'error', lastError: message.slice(0, 2_000), @@ -2058,13 +2133,32 @@ function normalizeSyncStatus(state: { status: string; lastCompletedAt: Date | nu return 'ready' } +export type DingWorkflowSyncMetadata = { + status: DingWorkflowSyncStatus + lastCompletedAt: string | null + lastFullReconciledAt: string | null +} + +export async function getDingWorkflowSyncMetadata(): Promise { + const state = await prisma.dingWorkflowSyncState.findUnique({ + where: { id: DING_WORKFLOW_SYNC_STATE_ID }, + select: { status: true, lastCompletedAt: true, sourceCursorRaw: true }, + }) + const sourceCursor = parseDingWorkflowSourceCursor(state?.sourceCursorRaw || null) + return { + status: normalizeSyncStatus(state), + lastCompletedAt: state?.lastCompletedAt?.toISOString() || null, + lastFullReconciledAt: sourceCursor.lastFullReconciledAt, + } +} + async function readSummaryWithoutAccessCheck(): Promise { const [total, state] = await prisma.$transaction([ prisma.dingWorkflowSnapshot.count({ where: buildBrandCenterOngoingWorkflowWhere(), }), prisma.dingWorkflowSyncState.findUnique({ - where: { id: SYNC_STATE_ID }, + where: { id: DING_WORKFLOW_SYNC_STATE_ID }, select: { status: true, lastCompletedAt: true }, }), ]) @@ -2333,7 +2427,7 @@ async function readBusinessTrendsInsightsWithoutAccessCheck( ], }), prisma.dingWorkflowSyncState.findUnique({ - where: { id: SYNC_STATE_ID }, + where: { id: DING_WORKFLOW_SYNC_STATE_ID }, select: { status: true, lastCompletedAt: true }, }), ]) @@ -2392,7 +2486,7 @@ async function readTeamCapacityInsightsWithoutAccessCheck( ], }), prisma.dingWorkflowSyncState.findUnique({ - where: { id: SYNC_STATE_ID }, + where: { id: DING_WORKFLOW_SYNC_STATE_ID }, select: { status: true, lastCompletedAt: true }, }), ]) @@ -2542,7 +2636,7 @@ async function readMonthlyLaunchInsightsWithoutAccessCheck( select: MONTHLY_LAUNCH_WORKFLOW_SELECT, }), prisma.dingWorkflowSyncState.findUnique({ - where: { id: SYNC_STATE_ID }, + where: { id: DING_WORKFLOW_SYNC_STATE_ID }, select: { status: true, lastCompletedAt: true }, }), ]) @@ -2632,7 +2726,7 @@ async function readOnTimeDeliveryInsightsWithoutAccessCheck( ], }), prisma.dingWorkflowSyncState.findUnique({ - where: { id: SYNC_STATE_ID }, + where: { id: DING_WORKFLOW_SYNC_STATE_ID }, select: { status: true, lastCompletedAt: true }, }), ]) @@ -2732,7 +2826,7 @@ export function ensureHourlyDingWorkflowSyncStarted() { const configuredHours = Number(process.env.DING_WORKFLOW_SYNC_INTERVAL_HOURS || 1) const intervalMs = Number.isFinite(configuredHours) && configuredHours > 0 - ? Math.max(5 * 60 * 1000, configuredHours * 60 * 60 * 1000) + ? Math.min(DAILY_RECONCILIATION_INTERVAL_MS, Math.max(5 * 60 * 1000, configuredHours * 60 * 60 * 1000)) : DEFAULT_SYNC_INTERVAL_MS const runOnce = () => { void syncDingWorkflowSnapshots() @@ -2750,5 +2844,5 @@ export function ensureHourlyDingWorkflowSyncStarted() { initialTimer.unref?.() const interval = setInterval(runOnce, intervalMs) interval.unref?.() - console.info(`[ding-workflow-sync] 已调度每小时增量同步，周期 ${Math.round(intervalMs / 60_000)} 分钟`) + console.info(`[ding-workflow-sync] 已调度流程增量同步，周期 ${Math.round(intervalMs / 60_000)} 分钟，并保证每 24 小时全量校准流程与商品 ID`) } diff --git a/lib/performance-conversion.ts b/lib/performance-conversion.ts index 1ac4622..fe18657 100644 --- a/lib/performance-conversion.ts +++ b/lib/performance-conversion.ts @@ -5,6 +5,7 @@ import { prisma } from '@/lib/prisma' import { DING_WORKFLOW_ALLOWED_PROCESS_CODES, DING_WORKFLOW_STARTED_AT_CUTOFF_ISO, + getDingWorkflowSyncMetadata, isDingWorkflowApprovedCompletion, } from '@/lib/ding-workflow-sync' import { @@ -49,7 +50,6 @@ type SnapshotRow = { brand: string | null sourceFinishedAtRaw: string | null formFields: unknown - syncedAt: Date } type SourceMetricRow = { @@ -138,7 +138,6 @@ async function readEligibleSnapshots(): Promise { brand: true, sourceFinishedAtRaw: true, formFields: true, - syncedAt: true, }, orderBy: [ { sourceFinishedAtRaw: 'desc' }, @@ -172,7 +171,10 @@ export async function getPerformanceConversionWorkflows( if (completedFrom && completedTo && compareDateKeys(completedFrom, completedTo) > 0) { throw new PerformanceConversionError('完结开始日期不能晚于结束日期') } - const snapshots = await readEligibleSnapshots() + const [snapshots, workflowSync] = await Promise.all([ + readEligibleSnapshots(), + getDingWorkflowSyncMetadata(), + ]) const rangeSnapshots = snapshots.filter((row) => { const completedDate = row.sourceFinishedAtRaw ? toShanghaiDateKey(row.sourceFinishedAtRaw) : null if (!completedDate) return false @@ -195,11 +197,6 @@ export async function getPerformanceConversionWorkflows( item.title, ].some((value) => String(value || '').toLocaleLowerCase('zh-CN').includes(query)))) const offset = (page - 1) * pageSize - const latestSyncedAt = snapshots.reduce( - (latest, row) => !latest || row.syncedAt > latest ? row.syncedAt : latest, - null, - ) - return { brands: BUSINESS_TREND_BRANDS.map((item) => ({ brand: item, count: brandCounts.get(item) || 0 })), items: filtered.slice(offset, offset + pageSize), @@ -208,7 +205,7 @@ export async function getPerformanceConversionWorkflows( page, pageSize, hasMore: offset + pageSize < filtered.length, - lastWorkflowSyncedAt: latestSyncedAt?.toISOString() || null, + workflowSync, } } diff --git a/prisma/schema.prisma b/prisma/schema.prisma index af9a612..beab4d3 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -1469,7 +1469,7 @@ model TaskDetailChangeNotification { /// /// 上游 rpt.ding_Workflow / ding_WorkflowTask / ding_WorkflowScheme 均为无索引大表， /// 部门看板不得在用户请求链路中直连扫描。后台任务首次建立全量基线，之后每小时 -/// 增量同步主流程，并为全部未完结流程刷新任务与表单字段汇总。 +/// 增量同步主流程，并每 24 小时全量校准流程与商品 ID 表单字段。 model DingWorkflowSnapshot { processInstanceId String @id @map("process_instance_id") businessId String? @map("business_id")