修复：保护平台订单原始事实不被清理和覆盖

apps/backend/src/services/scheduler/data-retention-job.test.ts

@@ -6,21 +6,25 @@ async function readJobSource() {
   return readFile(new URL('./data-retention-job.ts', import.meta.url), 'utf8')
 }
 
-test('保留清理必须覆盖 work_order_events，否则这张只增不减的表会一直拖慢历史单检索', async () => {
+test('数据维护不得覆盖原始报文或删除永久业务事实', async () => {
   const src = await readJobSource()
-  assert.ok(
-    src.includes("deleteOldRowsInBatches('work_order_events'"),
-    'work_order_events 未纳入保留清理（生产上 71 万行 / 269MB 且只增不减）',
-  )
-  assert.ok(
-    src.includes("'task_events' | 'webhook_events' | 'admin_audit_logs' | 'work_order_events'"),
-    '删除助手需要显式允许 work_order_events（表名是白名单联合类型，避免拼接注入）',
-  )
-  assert.ok(src.includes('retention.workOrderEventDays'), '保留天数必须走配置项，便于按环境调整')
-  assert.ok(
-    src.includes('workOrderEventRows: number') && src.includes('workOrderEventRows,'),
-    '清理结果要带上工单事件的删除行数，方便定时任务面板核对',
-  )
+  assert.equal(src.includes('raw_payload_json ='), false)
+  assert.equal(src.includes('DELETE FROM'), false)
+  for (const table of [
+    'orders',
+    'kuaishou_industry_vouchers',
+    'task_events',
+    'webhook_events',
+    'admin_audit_logs',
+    'work_order_events',
+  ]) {
+    assert.equal(
+      new RegExp(`(?:UPDATE|DELETE)\\s+${table}\\b`, 'i').test(src),
+      false,
+      `${table} 属于永久业务事实，不能在维护任务中清理或覆盖`,
+    )
+  }
+  assert.ok(src.includes('永久业务事实未清理'))
 })
 
 test('成本异常只归档已处理的行，open 队列不能被藏起来', async () => {
@@ -64,29 +68,28 @@ test('保留任务要回填漏写的付款派生列，并在有命中时打日
   )
 })
 
-test('保留清理默认窗口：敏感报文/事件 180 天，审计 365 天，工单事件 180 天', async () => {
+test('数据维护只保留可派生数据的成本异常归档窗口', async () => {
   const defaults = await readFile(new URL('../../config/defaults.ts', import.meta.url), 'utf8')
+  assert.ok(defaults.includes('costExceptionDays: 60'))
   for (const line of [
-    'rawPayloadDays: 180',
-    'taskEventDays: 180',
-    'webhookEventDays: 180',
-    'auditLogDays: 365',
-    'workOrderEventDays: 180',
-    'costExceptionDays: 60',
+    'rawPayloadDays:',
+    'taskEventDays:',
+    'webhookEventDays:',
+    'auditLogDays:',
+    'workOrderEventDays:',
   ]) {
-    assert.ok(defaults.includes(line), `默认保留策略缺少 ${line}`)
+    assert.equal(defaults.includes(line), false, `永久事实不能存在清理窗口 ${line}`)
   }
 
   const overrides = await readFile(
     new URL('../../config/env-overrides.ts', import.meta.url),
     'utf8',
   )
-  assert.ok(
-    overrides.includes(
-      "integerEnv('WORK_ORDER_EVENT_RETENTION_DAYS', ['retention', 'workOrderEventDays'])",
-    ),
-    'WORK_ORDER_EVENT_RETENTION_DAYS 需要环境变量出口，否则生产无法调整',
-  )
+  assert.equal(overrides.includes('RAW_PAYLOAD_RETENTION_DAYS'), false)
+  assert.equal(overrides.includes('TASK_EVENT_RETENTION_DAYS'), false)
+  assert.equal(overrides.includes('WEBHOOK_EVENT_RETENTION_DAYS'), false)
+  assert.equal(overrides.includes('AUDIT_LOG_RETENTION_DAYS'), false)
+  assert.equal(overrides.includes('WORK_ORDER_EVENT_RETENTION_DAYS'), false)
   assert.ok(
     overrides.includes(
       "integerEnv('COST_EXCEPTION_RETENTION_DAYS', ['retention', 'costExceptionDays'])",

apps/backend/src/services/scheduler/data-retention-job.ts

@@ -8,49 +8,29 @@ import { pruneWorkProductMatchLogs } from '../../repositories/worker-platform/pr
 export type DataRetentionResult = {
   status: 'ok'
   message: string
-  rawPayloadRows: number
-  taskEventRows: number
-  webhookEventRows: number
-  auditLogRows: number
-  workOrderEventRows: number
-  matchLogRows: number
   matchLogDeletedRows: number
   sourceOrderAmountBackfillRows: number
   costExceptionArchivedRows: number
 }
 
-/** 清理敏感原始报文并删除已过保留期的事件/审计记录。 */
+/**
+ * 维护可重建的派生数据。
+ *
+ * 原始订单、平台报文、工单/任务事件、审计日志和资金记录都是永久事实，
+ * 不在这里按时间清理或覆盖。
+ */
 export async function runDataRetentionJob(): Promise<DataRetentionResult> {
   const retention = runtimeConfig.retention || {}
-  const rawPayloadDays = positiveDays(retention.rawPayloadDays, 180)
-  const taskEventDays = positiveDays(retention.taskEventDays, 180)
-  const webhookEventDays = positiveDays(retention.webhookEventDays, 180)
-  const auditLogDays = positiveDays(retention.auditLogDays, 365)
-  // 工单事件表以前完全没纳入保留：717k 行 / 269MB 且只增不减，会一直拖慢历史单检索。
-  const workOrderEventDays = positiveDays(retention.workOrderEventDays, 180)
   // 成本异常只归档「已处理」的行；open 行是待办队列，不能因为超期就藏起来。
   const costExceptionDays = positiveDays(retention.costExceptionDays, 60)
 
-  // 1. 按配置清理超出上限或过期的打手商品匹配日志
+  // 商品匹配日志是诊断数据，不是订单或履约事实。
   const matchConfig = getWorkerProductMatchConfig()
   const matchLogPruneResult = await pruneWorkProductMatchLogs({
     maxRows: matchConfig.maxMatchLogs,
     retentionDays: matchConfig.matchLogRetentionDays,
   })
 
-  // 2. 清理其他报文与日志
-  const rawPayloadRows =
-    (await scrubJsonColumnInBatches('orders', 'updated_at', rawPayloadDays)) +
-    (await scrubJsonColumnInBatches('kuaishou_industry_vouchers', 'updated_at', rawPayloadDays))
-  const matchLogRows = await scrubJsonColumnInBatches(
-    'work_product_match_logs',
-    'created_at',
-    rawPayloadDays,
-  )
-  const taskEventRows = await deleteOldRowsInBatches('task_events', taskEventDays)
-  const webhookEventRows = await deleteOldRowsInBatches('webhook_events', webhookEventDays)
-  const auditLogRows = await deleteOldRowsInBatches('admin_audit_logs', auditLogDays)
-  const workOrderEventRows = await deleteOldRowsInBatches('work_order_events', workOrderEventDays)
   // 统计热路径已不再读 material_json。若将来有第三个写入点忘了维护派生列，
   // 金额会静默回退 orders.total_amount。这里把漏写补上并打日志，让口径变化可见。
   const costExceptionArchive = await archiveResolvedCostExceptions({
@@ -66,14 +46,8 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
 
   return {
     status: 'ok',
-    message: '数据保留清理完成',
-    rawPayloadRows,
-    matchLogRows,
+    message: '派生数据维护完成；永久业务事实未清理',
     matchLogDeletedRows: matchLogPruneResult.totalDeleted,
-    taskEventRows,
-    webhookEventRows,
-    auditLogRows,
-    workOrderEventRows,
     sourceOrderAmountBackfillRows,
     costExceptionArchivedRows,
   }
@@ -92,59 +66,6 @@ async function backfillMissingSourceOrderAmount() {
   return result.rowCount || 0
 }
 
-async function scrubJsonColumnInBatches(
-  table: 'orders' | 'kuaishou_industry_vouchers' | 'work_product_match_logs',
-  dateColumn: 'created_at' | 'updated_at',
-  retentionDays: number,
-) {
-  let total = 0
-  while (true) {
-    const result = await query(
-      `
-        WITH candidates AS (
-          SELECT ctid
-          FROM ${table}
-          WHERE ${dateColumn} < NOW() - make_interval(days => $1::int)
-            AND raw_payload_json <> '{}'::jsonb
-          LIMIT 1000
-        )
-        UPDATE ${table} AS target
-        SET raw_payload_json = '{}'::jsonb
-        FROM candidates
-        WHERE target.ctid = candidates.ctid
-      `,
-      [retentionDays],
-    )
-    total += result.rowCount || 0
-    if ((result.rowCount || 0) < 1000) return total
-  }
-}
-
-async function deleteOldRowsInBatches(
-  table: 'task_events' | 'webhook_events' | 'admin_audit_logs' | 'work_order_events',
-  retentionDays: number,
-) {
-  let total = 0
-  while (true) {
-    const result = await query(
-      `
-        WITH candidates AS (
-          SELECT ctid
-          FROM ${table}
-          WHERE created_at < NOW() - make_interval(days => $1::int)
-          LIMIT 1000
-        )
-        DELETE FROM ${table} AS target
-        USING candidates
-        WHERE target.ctid = candidates.ctid
-      `,
-      [retentionDays],
-    )
-    total += result.rowCount || 0
-    if ((result.rowCount || 0) < 1000) return total
-  }
-}
-
 function positiveDays(value: unknown, fallback: number) {
   const days = Number(value)
   return Number.isInteger(days) && days > 0 ? days : fallback

apps/backend/src/services/scheduler/scheduler-service.ts

@@ -169,12 +169,9 @@ async function runJob(job: JsonObject, { manual = false }: { manual?: boolean }
       status,
       ...(job.type === 'data_retention'
         ? {
-            rawPayloadRows: Number(summary.rawPayloadRows || 0),
-            matchLogRows: Number(summary.matchLogRows || 0),
-            taskEventRows: Number(summary.taskEventRows || 0),
-            webhookEventRows: Number(summary.webhookEventRows || 0),
-            auditLogRows: Number(summary.auditLogRows || 0),
+            matchLogDeletedRows: Number(summary.matchLogDeletedRows || 0),
             sourceOrderAmountBackfillRows: Number(summary.sourceOrderAmountBackfillRows || 0),
+            costExceptionArchivedRows: Number(summary.costExceptionArchivedRows || 0),
           }
         : {}),
     })