feat(cost-center): 成本异常 60 天归档

cost_exceptions 只增不减：本地快照 260,253 行（resolved 151,939 / open 108,314），已处理的行只是审计痕迹，留在热表会拖慢「状态 + 时间」分页，并让唯一索引持续膨胀。 - 迁移 108：新增 cost_exceptions_archive（保留原 id 与时间字段，加 archived_at），热表补部分索引 idx_cost_exceptions_archivable (COALESCE(resolved_at, created_at)) WHERE status <> 'open'，让每日任务不扫全表； - cost-exception-archive-service：只归档 status <> 'open' 且超期的行，每批 1000 行， DELETE ... RETURNING + INSERT 在同一语句内完成，单批原子，不存在「删了没归档」； - data-retention 每日任务接入；保留期走 COST_EXCEPTION_RETENTION_DAYS（默认 60）； - 新增 npm run cost-exceptions:archive（--dry-run / --days / --batch-size）。为什么 open 行不归档：open 是待办队列，超期不等于已解决，归档等于把问题藏起来。验证： - 3 条 resolved 回拨到 90 天前 → 归档 3 条，热表 resolved 151,939 → 151,936； - 1 条 open 回拨到 90 天前 → 仍在热表，未被归档； - 归档表保留原 id 与时间，archived_at 记录归档时刻； - 恢复测试数据后计数回到 151,939 / 108,314。
2026-10-25 12:09:20 +08:00
parent fcd9b607d4
commit d90c066888
15 changed files with 300 additions and 0 deletions
@@ -78,6 +78,8 @@ TASK_EVENT_RETENTION_DAYS=180
 WEBHOOK_EVENT_RETENTION_DAYS=180
 AUDIT_LOG_RETENTION_DAYS=365
 WORK_ORDER_EVENT_RETENTION_DAYS=180
+# 已处理成本异常保留天数，超期搬入 cost_exceptions_archive（open 行不归档）
+COST_EXCEPTION_RETENTION_DAYS=60

 # 文件存储：生产使用阿里云 OSS（生产编排已移除 MinIO，仅本地开发使用）。
 # STORAGE_MODE 取值 minio | dual | oss；dual 仅在 MinIO→OSS 迁移/回滚期间使用。
@@ -12,6 +12,7 @@
    "finance:sync": "tsx scripts/sync-cost-finance.ts",
    "fulfillment:backfill": "tsx scripts/backfill-fulfillment-mode.ts",
    "revenue:backfill": "tsx scripts/backfill-revenue-basis.ts",
+    "cost-exceptions:archive": "tsx scripts/archive-cost-exceptions.ts",
    "storage:migrate": "tsx scripts/migrate-storage.ts",
    "storage:migrate:built": "node dist/storage-migrate.js",
    "check:sql": "tsx scripts/check-sql-guard.ts",
@@ -0,0 +1,92 @@
+/**
+ * 手动归档已处理的成本异常（日常由 data-retention 定时任务自动执行）。
+ *
+ * 默认窗口与线上配置一致（60 天），可通过 --days 覆盖；open 行永远不归档。
+ *
+ * 用法：
+ *   npm run cost-exceptions:archive -- --dry-run
+ *   npm run cost-exceptions:archive -- --days=90
+ *   npm run cost-exceptions:archive -- --days=60 --batch-size=2000
+ */
+import process from 'node:process'
+
+import { closeDb } from '../src/db/client.js'
+import { runtimeConfig } from '../src/config/runtime.js'
+import {
+  archiveResolvedCostExceptions,
+  countArchivableCostExceptions,
+} from '../src/services/cost-center/cost-exception-archive-service.js'
+
+const options = parseOptions(process.argv.slice(2))
+const defaultDays = Number(runtimeConfig.retention?.costExceptionDays) || 60
+const retentionDays = normalizeInteger(options.days, defaultDays, 1, 3650)
+const batchSize = normalizeInteger(options.batchSize, 1000, 1, 10_000)
+const dryRun = options.dryRun === 'true'
+
+async function main() {
+  const startedAt = Date.now()
+  const archivable = await countArchivableCostExceptions(retentionDays)
+
+  if (dryRun) {
+    console.log(
+      JSON.stringify(
+        {
+          dryRun: true,
+          retentionDays,
+          archivableRows: archivable,
+          durationMs: Date.now() - startedAt,
+        },
+        null,
+        2,
+      ),
+    )
+    return
+  }
+
+  const result = await archiveResolvedCostExceptions({ retentionDays, batchSize })
+  console.log(
+    JSON.stringify(
+      { dryRun: false, ...result, batchSize, durationMs: Date.now() - startedAt },
+      null,
+      2,
+    ),
+  )
+
+  if (result.remainingArchivableRows > 0) {
+    console.error(`仍有 ${result.remainingArchivableRows} 行待归档；请重跑（幂等）。`)
+    process.exitCode = 2
+  }
+}
+
+main()
+  .catch((error) => {
+    console.error(error instanceof Error ? error.message : error)
+    process.exitCode = 1
+  })
+  .finally(async () => {
+    await closeDb()
+  })
+
+function parseOptions(args: string[]) {
+  const result: Record<string, string> = {}
+  for (const arg of args) {
+    if (!arg.startsWith('--')) continue
+    const body = arg.slice(2)
+    const index = body.indexOf('=')
+    const key = index === -1 ? body : body.slice(0, index)
+    const value = index === -1 ? 'true' : body.slice(index + 1)
+    result[key] = value
+    result[toCamelCase(key)] = value
+  }
+  return result
+}
+
+function toCamelCase(value: string) {
+  return value.replace(/-([a-z])/g, (_, char: string) => char.toUpperCase())
+}
+
+function normalizeInteger(value: unknown, fallback: number, min: number, max: number) {
+  const parsed = Number(value)
+  if (!Number.isInteger(parsed)) return fallback
+  return Math.min(max, Math.max(min, parsed))
+}
@@ -30,6 +30,8 @@ export function createDefaultRuntimeConfig(projectRoot: string): RuntimeConfig {
      webhookEventDays: 180,
      auditLogDays: 365,
      workOrderEventDays: 180,
+      // 已处理的成本异常保留 60 天，之后搬进 cost_exceptions_archive；open 行不归档。
+      costExceptionDays: 60,
    },

    database: {
@@ -100,6 +100,7 @@ function createRuntimeConfig(): RuntimeConfig {
      taskEventDays: 180,
      webhookEventDays: 180,
      auditLogDays: 365,
+      costExceptionDays: 60,
    },
    database: {
      url: 'postgres://postgres:postgres@127.0.0.1:5432/order_site',
@@ -35,6 +35,7 @@ export const ENV_OVERRIDES: readonly EnvOverride[] = [
  integerEnv('WEBHOOK_EVENT_RETENTION_DAYS', ['retention', 'webhookEventDays']),
  integerEnv('AUDIT_LOG_RETENTION_DAYS', ['retention', 'auditLogDays']),
  integerEnv('WORK_ORDER_EVENT_RETENTION_DAYS', ['retention', 'workOrderEventDays']),
+  integerEnv('COST_EXCEPTION_RETENTION_DAYS', ['retention', 'costExceptionDays']),
  stringEnv('DATABASE_URL', ['database', 'url']),
  booleanEnv('DATABASE_SSL', ['database', 'ssl']),
  integerEnv('DATABASE_MAX_CONNECTIONS', ['database', 'maxConnections']),
@@ -164,6 +164,7 @@ function createRuntimeConfig(overrides: Record<string, unknown> = {}): RuntimeCo
        taskEventDays: 180,
        webhookEventDays: 180,
        auditLogDays: 365,
+        costExceptionDays: 60,
      },
      database: {
        url: 'postgres://postgres:postgres@postgres:5432/order_site',
@@ -88,6 +88,9 @@ export function validateRuntimeConfig(
      min: 1,
    })
    requireInteger(issues, 'retention.auditLogDays', config.retention.auditLogDays, { min: 1 })
+    requireInteger(issues, 'retention.costExceptionDays', config.retention.costExceptionDays, {
+      min: 1,
+    })
  }
  const storageMode = normalizeStorageMode(config.storage?.mode)
  if (!storageMode) {
@@ -0,0 +1,49 @@
+-- 108_cost_exception_archive.sql —— 成本异常归档表。
+--
+-- 背景（2026-10-25）：
+--   cost_exceptions 只增不减。本地快照上 260,253 行（resolved 151,939 / open 108,314），
+--   其中 open 的 product_mapping_missing 与 cost_missing 仍是可操作队列，
+--   但已处理（resolved / ignored）的历史行只是审计痕迹，继续留在热表里拖慢：
+--     - 按状态 + 时间分页要扫过大量已完成行；
+--     - 唯一索引 uq_cost_exceptions_subject_reason 随行数膨胀。
+--
+-- 策略：
+--   1. 只归档**已处理**（status <> 'open'）且超过保留期（默认 60 天）的行；
+--   2. open 行永远留在热表，避免把未解决的问题藏起来；
+--   3. 归档按批（每次 1000 行）在 data-retention 定时任务里执行，不锁全表；
+--   4. 归档表不建外键：历史异常可能指向已删除的订单项/工单，保留原值即可。
+--
+-- 保留期由 COST_EXCEPTION_RETENTION_DAYS 控制（默认 60）。
+
+CREATE TABLE IF NOT EXISTS cost_exceptions_archive (
+  id                       BIGINT PRIMARY KEY,
+  subject_type             TEXT NOT NULL,
+  subject_id               BIGINT NOT NULL,
+  order_item_id            BIGINT,
+  work_order_id            BIGINT,
+  reason                   TEXT NOT NULL,
+  status                   TEXT NOT NULL,
+  detail_json              JSONB NOT NULL DEFAULT '{}'::jsonb,
+  created_at               TIMESTAMPTZ NOT NULL,
+  resolved_at              TIMESTAMPTZ,
+  resolved_by_admin_user_id BIGINT,
+  resolution_note          TEXT NOT NULL DEFAULT '',
+  archived_at              TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_cost_exceptions_archive_archived_at
+  ON cost_exceptions_archive (archived_at DESC);
+CREATE INDEX IF NOT EXISTS idx_cost_exceptions_archive_subject
+  ON cost_exceptions_archive (subject_type, subject_id);
+CREATE INDEX IF NOT EXISTS idx_cost_exceptions_archive_reason
+  ON cost_exceptions_archive (reason, archived_at DESC);
+
+COMMENT ON TABLE cost_exceptions_archive IS
+  '已处理成本异常的归档：data-retention 任务把超过保留期（默认 60 天）的 resolved/ignored 行搬到这里，open 行不归档';
+COMMENT ON COLUMN cost_exceptions_archive.archived_at IS
+  '归档时间；原表 created_at / resolved_at 原样保留，便于按原时间追溯';
+
+-- 热表上补一个「待归档」用的部分索引，避免每日任务扫全表。
+CREATE INDEX IF NOT EXISTS idx_cost_exceptions_archivable
+  ON cost_exceptions (COALESCE(resolved_at, created_at))
+  WHERE status <> 'open';
@@ -0,0 +1,89 @@
+/**
+ * 成本异常归档：把超过保留期的「已处理」异常搬进 cost_exceptions_archive。
+ *
+ * 口径：
+ *   - 只归档 status <> 'open' 的行；open 是待办队列，超期也必须留着；
+ *   - 时钟取 COALESCE(resolved_at, created_at)：ignored 行没有 resolved_at 时按创建时间；
+ *   - 每批 1000 行，DELETE ... RETURNING + INSERT 在同一语句内完成，单批原子，
+ *     不存在「删了但没归档」的窗口；
+ *   - 归档保留 id 与原时间字段，可按原口径追溯。
+ *
+ * 调用方：data-retention 定时任务（每日）、scripts/archive-cost-exceptions.ts（手动）。
+ */
+import { query } from '../../db/client.js'
+
+export type ArchiveCostExceptionsResult = {
+  archivedRows: number
+  remainingArchivableRows: number
+  retentionDays: number
+}
+
+export async function countArchivableCostExceptions(retentionDays: number) {
+  const { rows } = await query<{ count: number }>(
+    `
+      SELECT COUNT(*)::int AS count
+      FROM cost_exceptions
+      WHERE status <> 'open'
+        AND COALESCE(resolved_at, created_at) < NOW() - make_interval(days => $1::int)
+    `,
+    [retentionDays],
+  )
+  return rows[0]?.count ?? 0
+}
+
+export async function archiveResolvedCostExceptions({
+  retentionDays,
+  batchSize = 1000,
+  maxBatches = 10_000,
+}: {
+  retentionDays: number
+  batchSize?: number
+  maxBatches?: number
+}): Promise<ArchiveCostExceptionsResult> {
+  const archiveExists = await query<{ exists: boolean }>(
+    `SELECT to_regclass('public.cost_exceptions_archive') IS NOT NULL AS exists`,
+  )
+  if (!archiveExists.rows[0]?.exists) {
+    return { archivedRows: 0, remainingArchivableRows: 0, retentionDays }
+  }
+
+  let archivedRows = 0
+  let batches = 0
+  while (batches < maxBatches) {
+    const result = await query(
+      `
+        WITH moved AS (
+          DELETE FROM cost_exceptions
+          WHERE id IN (
+            SELECT id
+            FROM cost_exceptions
+            WHERE status <> 'open'
+              AND COALESCE(resolved_at, created_at) < NOW() - make_interval(days => $1::int)
+            ORDER BY id
+            LIMIT $2
+          )
+          RETURNING id, subject_type, subject_id, order_item_id, work_order_id,
+                    reason, status, detail_json, created_at, resolved_at,
+                    resolved_by_admin_user_id, resolution_note
+        )
+        INSERT INTO cost_exceptions_archive (
+          id, subject_type, subject_id, order_item_id, work_order_id,
+          reason, status, detail_json, created_at, resolved_at,
+          resolved_by_admin_user_id, resolution_note, archived_at
+        )
+        SELECT id, subject_type, subject_id, order_item_id, work_order_id,
+               reason, status, detail_json, created_at, resolved_at,
+               resolved_by_admin_user_id, resolution_note, NOW()
+        FROM moved
+      `,
+      [retentionDays, batchSize],
+    )
+    const movedRows = result.rowCount || 0
+    archivedRows += movedRows
+    batches += 1
+    if (movedRows < batchSize) break
+  }
+
+  const remainingArchivableRows = await countArchivableCostExceptions(retentionDays)
+  return { archivedRows, remainingArchivableRows, retentionDays }
+}
@@ -23,6 +23,30 @@ test('保留清理必须覆盖 work_order_events，否则这张只增不减的
  )
 })

+test('成本异常只归档已处理的行，open 队列不能被藏起来', async () => {
+  const job = await readJobSource()
+  assert.ok(
+    job.includes('archiveResolvedCostExceptions'),
+    '成本异常必须走归档，不能直接 DELETE 掉审计痕迹',
+  )
+  assert.ok(job.includes('retention.costExceptionDays'), '保留天数必须走配置项')
+
+  // 归档 SQL 在独立服务里，检查服务源码而不是任务源码。
+  const service = await readFile(
+    new URL('../cost-center/cost-exception-archive-service.ts', import.meta.url),
+    'utf8',
+  )
+  assert.ok(service.includes("status <> 'open'"), 'open 行是待办队列，不能因超期被归档')
+  assert.ok(
+    service.includes('COALESCE(resolved_at, created_at)'),
+    '归档时钟优先用 resolved_at，ignored 行没有 resolved_at 时回退 created_at',
+  )
+  assert.ok(
+    service.includes('INSERT INTO cost_exceptions_archive'),
+    '归档必须落到 cost_exceptions_archive，保持可追溯',
+  )
+})
+
 test('保留任务要回填漏写的付款派生列，并在有命中时打日志', async () => {
  const src = await readJobSource()
  assert.ok(
@@ -48,6 +72,7 @@ test('保留清理默认窗口：敏感报文/事件 180 天，审计 365 天，
    'webhookEventDays: 180',
    'auditLogDays: 365',
    'workOrderEventDays: 180',
+    'costExceptionDays: 60',
  ]) {
    assert.ok(defaults.includes(line), `默认保留策略缺少 ${line}`)
  }
@@ -62,4 +87,10 @@ test('保留清理默认窗口：敏感报文/事件 180 天，审计 365 天，
    ),
    'WORK_ORDER_EVENT_RETENTION_DAYS 需要环境变量出口，否则生产无法调整',
  )
+  assert.ok(
+    overrides.includes(
+      "integerEnv('COST_EXCEPTION_RETENTION_DAYS', ['retention', 'costExceptionDays'])",
+    ),
+    'COST_EXCEPTION_RETENTION_DAYS 需要环境变量出口，否则生产无法调整归档窗口',
+  )
 })
@@ -2,6 +2,7 @@ import { query } from '../../db/client.js'
 import { runtimeConfig } from '../../config/runtime.js'
 import { logWarn } from '../../utils/logger.js'
 import { getWorkerProductMatchConfig } from '../worker-platform/worker-product-match-config-service.js'
+import { archiveResolvedCostExceptions } from '../cost-center/cost-exception-archive-service.js'
 import { pruneWorkProductMatchLogs } from '../../repositories/worker-platform/product-match-repo.js'

 export type DataRetentionResult = {
@@ -15,6 +16,7 @@ export type DataRetentionResult = {
  matchLogRows: number
  matchLogDeletedRows: number
  sourceOrderAmountBackfillRows: number
+  costExceptionArchivedRows: number
 }

 /** 清理敏感原始报文并删除已过保留期的事件/审计记录。 */
@@ -26,6 +28,8 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
  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()
@@ -49,6 +53,10 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
  const workOrderEventRows = await deleteOldRowsInBatches('work_order_events', workOrderEventDays)
  // 统计热路径已不再读 material_json。若将来有第三个写入点忘了维护派生列，
  // 金额会静默回退 orders.total_amount。这里把漏写补上并打日志，让口径变化可见。
+  const costExceptionArchive = await archiveResolvedCostExceptions({
+    retentionDays: costExceptionDays,
+  })
+  const costExceptionArchivedRows = costExceptionArchive.archivedRows
  const sourceOrderAmountBackfillRows = await backfillMissingSourceOrderAmount()
  if (sourceOrderAmountBackfillRows > 0) {
    logWarn('[data-retention]', 'source_order_amount_fen 出现漏写，已按 JSON 回填', {
@@ -67,6 +75,7 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
    auditLogRows,
    workOrderEventRows,
    sourceOrderAmountBackfillRows,
+    costExceptionArchivedRows,
  }
 }

@@ -34,6 +34,8 @@ export type RuntimeConfig = {
    auditLogDays: number
    /** 工单事件（work_order_events）保留天数：这张表只增不减，必须纳入清理。 */
    workOrderEventDays: number
+    /** 已处理成本异常（cost_exceptions）保留天数：超期归档，open 行不归档。 */
+    costExceptionDays: number
  }
  database: {
    url: string
@@ -85,6 +85,7 @@
 | 履约方式统一 | **已实施 P0** | 迁移 `105`；375,447 项全部回填：接单平台 153,148、人工 115,194、历史平台 53,901、自动发货 53,204 |
 | 人工商品归一 | **已实施 P1（本次快照恢复执行）** | 人工 115,194 项已归一为 1,195 个 `man:<内部SKU>` 商品；历史人工支出仍不补录 |
 | 影子总账 | 未实施 | 现有财务页仍是 `UNION ALL` 业务事件拼装，不是复式总账 |
+| 成本异常归档 | **已实施** | 迁移 `108`；已处理异常超 60 天搬入 `cost_exceptions_archive`，open 行不归档；`COST_EXCEPTION_RETENTION_DAYS` 可调 |
 | 退款冲减 / 平台对账 / 期间关账 | 未实施 | 仅方案 |

 > 分支状态（2026-10-25）：`dev @ 465a9f65` 之上完成迁移收敛（10 → 5，`103`–`107`），并已与 `main` 的 worker-hall 改动合并。本地库按「清 `schema_migrations` 103–112 后重跑」对齐，升级结果与全新构建的 schema 完全一致；生产 101 → 107 的路径已验证。**未推送、未部署生产**；生产口径仍以 `main` 为准。
@@ -468,6 +468,22 @@ npm run revenue:backfill -- --batch-size=2000 --max-batches=100
 | `idx_work_orders_rule_cover` | 成本版本页按接单规则聚合 |
 | `idx_order_items_revenue_cover` | 收入聚合 |
 | `idx_work_categories_name` | 商品目录按分类名 lateral join |
+| `idx_cost_exceptions_archivable` | 归档任务扫描「已处理且超期」的异常（部分索引，`status <> 'open'`） |
+
+### 9.1 成本异常归档（60 天）
+
+`cost_exceptions` 只增不减：本地快照 260,253 行（resolved 151,939 / open 108,314）。已处理的行只是审计痕迹，继续留在热表会拖慢按状态+时间的分页，并让唯一索引持续膨胀。
+
+| 项 | 规则 |
+| --- | --- |
+| 归档对象 | `status <> 'open'` 且 `COALESCE(resolved_at, created_at)` 超过保留期 |
+| open 行 | **永不归档**——它是待办队列，超期也要留在页面上 |
+| 保留期 | `COST_EXCEPTION_RETENTION_DAYS`，默认 **60** 天 |
+| 落点 | `cost_exceptions_archive`（保留原 id 与时间字段，加 `archived_at`） |
+| 执行 | `data-retention` 每日任务；每批 1000 行，`DELETE ... RETURNING` + `INSERT` 同语句，单批原子 |
+| 手动 | `npm run cost-exceptions:archive -- --dry-run` / `-- --days=90` |
+
+归档不删除审计痕迹，但热表查询与页面上不会再出现这些历史行；需要追溯时查归档表。

 ---