From 4159a435d564b5be2af866781a29be30a9ab2dd5 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Mon, 9 Nov 2026 13:20:36 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E4=BF=9D=E6=8A=A4?= =?UTF-8?q?=E5=B9=B3=E5=8F=B0=E8=AE=A2=E5=8D=95=E5=8E=9F=E5=A7=8B=E4=BA=8B?= =?UTF-8?q?=E5=AE=9E=E4=B8=8D=E8=A2=AB=E6=B8=85=E7=90=86=E5=92=8C=E8=A6=86?= =?UTF-8?q?=E7=9B=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.server.example | 6 - README.md | 2 +- apps/backend/src/config/defaults.ts | 6 - apps/backend/src/config/env-overrides.test.ts | 4 - apps/backend/src/config/env-overrides.ts | 5 - .../src/config/runtime-validation.test.ts | 4 - apps/backend/src/config/runtime-validation.ts | 6 - .../152_order_source_facts_immutable.sql | 57 ++++++++ .../src/repositories/order-item-repo.test.ts | 33 +++-- .../src/repositories/order-item-repo.ts | 72 +-------- .../src/repositories/order-repo.test.ts | 10 ++ apps/backend/src/repositories/order-repo.ts | 8 +- .../order-source-facts.integration.test.ts | 137 ++++++++++++++++++ .../order-source-snapshot-repo.test.ts | 40 +++++ .../order-source-snapshot-repo.ts | 67 +++++++++ .../src/services/order/order-service.ts | 29 ++++ .../platforms/ninetyone/order-service.ts | 22 +-- .../scheduler/data-retention-job.test.ts | 59 ++++---- .../services/scheduler/data-retention-job.ts | 95 +----------- .../services/scheduler/scheduler-service.ts | 7 +- apps/backend/src/types/runtime-config.ts | 6 - docs/统一订单履约与财务模型-权威版.md | 21 ++- 22 files changed, 429 insertions(+), 267 deletions(-) create mode 100644 apps/backend/src/db/migrations/152_order_source_facts_immutable.sql create mode 100644 apps/backend/src/repositories/order-source-facts.integration.test.ts create mode 100644 apps/backend/src/repositories/order-source-snapshot-repo.test.ts create mode 100644 apps/backend/src/repositories/order-source-snapshot-repo.ts diff --git a/.env.server.example b/.env.server.example index 5470e05a..7f0c4de0 100644 --- a/.env.server.example +++ b/.env.server.example @@ -72,12 +72,6 @@ LOG_RETENTION_DAYS=30 LOG_MAX_FILE_SIZE_MB=20 LOG_MAX_FILES=5 -# 数据库敏感报文/事件保留策略（天）。清理任务每天执行一次。 -RAW_PAYLOAD_RETENTION_DAYS=180 -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 diff --git a/README.md b/README.md index fa38aca7..442afa75 100644 --- a/README.md +++ b/README.md @@ -109,7 +109,7 @@ SKIP_CHECKS=1 git push | 文件 | 含义 | | --- | --- | | `001_init.sql` | 空库基线，建立完整业务结构 | -| `002_xxx.sql` ~ `121_xxx.sql` | 后续结构与数据增量；当前生产发布基线为 119，120–121 为本次履约路由升级 | +| `002_xxx.sql` ~ `152_xxx.sql` | 后续结构与数据增量；当前生产基线为 151，152 为本地待发布的事实保护与路由升级 | 约定： diff --git a/apps/backend/src/config/defaults.ts b/apps/backend/src/config/defaults.ts index 437296df..a41d38fb 100644 --- a/apps/backend/src/config/defaults.ts +++ b/apps/backend/src/config/defaults.ts @@ -24,12 +24,6 @@ export function createDefaultRuntimeConfig(projectRoot: string): RuntimeConfig { }, retention: { - // 原始报文只保留追溯所需窗口，清理后保留订单/事件摘要。 - rawPayloadDays: 180, - taskEventDays: 180, - webhookEventDays: 180, - auditLogDays: 365, - workOrderEventDays: 180, // 已处理的成本异常保留 60 天，之后搬进 cost_exceptions_archive；open 行不归档。 costExceptionDays: 60, }, diff --git a/apps/backend/src/config/env-overrides.test.ts b/apps/backend/src/config/env-overrides.test.ts index a53db9de..4ac95bd7 100644 --- a/apps/backend/src/config/env-overrides.test.ts +++ b/apps/backend/src/config/env-overrides.test.ts @@ -96,10 +96,6 @@ function createRuntimeConfig(): RuntimeConfig { retentionDays: 30, }, retention: { - rawPayloadDays: 180, - taskEventDays: 180, - webhookEventDays: 180, - auditLogDays: 365, costExceptionDays: 60, }, database: { diff --git a/apps/backend/src/config/env-overrides.ts b/apps/backend/src/config/env-overrides.ts index d3502261..6410c2df 100644 --- a/apps/backend/src/config/env-overrides.ts +++ b/apps/backend/src/config/env-overrides.ts @@ -30,11 +30,6 @@ export const ENV_OVERRIDES: readonly EnvOverride[] = [ integerEnv('LOG_RETENTION_DAYS', ['logging', 'retentionDays']), integerEnv('LOG_MAX_FILE_SIZE_MB', ['logging', 'maxFileSizeMb']), integerEnv('LOG_MAX_FILES', ['logging', 'maxFiles']), - integerEnv('RAW_PAYLOAD_RETENTION_DAYS', ['retention', 'rawPayloadDays']), - integerEnv('TASK_EVENT_RETENTION_DAYS', ['retention', 'taskEventDays']), - 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']), diff --git a/apps/backend/src/config/runtime-validation.test.ts b/apps/backend/src/config/runtime-validation.test.ts index 59fa9223..6b06989b 100644 --- a/apps/backend/src/config/runtime-validation.test.ts +++ b/apps/backend/src/config/runtime-validation.test.ts @@ -160,10 +160,6 @@ function createRuntimeConfig(overrides: Record = {}): RuntimeCo integrationLevel: 'warn', }, retention: { - rawPayloadDays: 180, - taskEventDays: 180, - webhookEventDays: 180, - auditLogDays: 365, costExceptionDays: 60, }, database: { diff --git a/apps/backend/src/config/runtime-validation.ts b/apps/backend/src/config/runtime-validation.ts index a7d16176..2688fb82 100644 --- a/apps/backend/src/config/runtime-validation.ts +++ b/apps/backend/src/config/runtime-validation.ts @@ -82,12 +82,6 @@ export function validateRuntimeConfig( max: 100, }) if (config.retention) { - requireInteger(issues, 'retention.rawPayloadDays', config.retention.rawPayloadDays, { min: 1 }) - requireInteger(issues, 'retention.taskEventDays', config.retention.taskEventDays, { min: 1 }) - requireInteger(issues, 'retention.webhookEventDays', config.retention.webhookEventDays, { - min: 1, - }) - requireInteger(issues, 'retention.auditLogDays', config.retention.auditLogDays, { min: 1 }) requireInteger(issues, 'retention.costExceptionDays', config.retention.costExceptionDays, { min: 1, }) diff --git a/apps/backend/src/db/migrations/152_order_source_facts_immutable.sql b/apps/backend/src/db/migrations/152_order_source_facts_immutable.sql new file mode 100644 index 00000000..aa101765 --- /dev/null +++ b/apps/backend/src/db/migrations/152_order_source_facts_immutable.sql @@ -0,0 +1,57 @@ +-- 152_order_source_facts_immutable.sql +-- +-- 永久保留平台订单事实： +-- 1. 每次来源同步追加完整订单快照，供审计和后续重放； +-- 2. 快照只允许插入，不允许应用更新或删除； +-- 3. orders.raw_payload_json 由应用写入口保证首次写入后不再覆盖。 + +CREATE TABLE IF NOT EXISTS order_source_snapshots ( + id BIGSERIAL PRIMARY KEY, + order_id BIGINT NOT NULL REFERENCES orders(id) ON DELETE RESTRICT, + provider TEXT NOT NULL, + platform TEXT NOT NULL, + shop_id TEXT NOT NULL DEFAULT '', + platform_order_id TEXT NOT NULL, + source_label TEXT NOT NULL DEFAULT '', + payload_json JSONB NOT NULL, + payload_sha256 TEXT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT chk_order_source_snapshots_payload_sha256 + CHECK (payload_sha256 ~ '^[0-9a-f]{64}$'), + CONSTRAINT uq_order_source_snapshots_payload + UNIQUE (order_id, payload_sha256) +); + +CREATE INDEX IF NOT EXISTS idx_order_source_snapshots_order + ON order_source_snapshots(order_id, recorded_at DESC, id DESC); + +CREATE INDEX IF NOT EXISTS idx_order_source_snapshots_platform_order + ON order_source_snapshots(provider, platform, platform_order_id, recorded_at DESC); + +COMMENT ON TABLE order_source_snapshots IS + '平台订单同步的不可变原始快照；重复同步只追加，不覆盖订单主表原始报文'; +COMMENT ON COLUMN order_source_snapshots.payload_json IS + '平台来源事件、原始报文与订单项完整快照'; +COMMENT ON COLUMN order_source_snapshots.payload_sha256 IS + '完整快照内容 SHA-256，用于同一订单重复消息幂等'; + +CREATE OR REPLACE FUNCTION reject_order_source_snapshot_mutation() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +BEGIN + IF current_setting('app.allow_order_fact_repair', TRUE) = 'on' THEN + RETURN CASE WHEN TG_OP = 'DELETE' THEN OLD ELSE NEW END; + END IF; + + RAISE EXCEPTION 'order_source_snapshots is append-only' + USING ERRCODE = '23514'; +END; +$$; + +DROP TRIGGER IF EXISTS trg_order_source_snapshots_append_only + ON order_source_snapshots; +CREATE TRIGGER trg_order_source_snapshots_append_only +BEFORE UPDATE OR DELETE ON order_source_snapshots +FOR EACH ROW +EXECUTE FUNCTION reject_order_source_snapshot_mutation(); diff --git a/apps/backend/src/repositories/order-item-repo.test.ts b/apps/backend/src/repositories/order-item-repo.test.ts index 50a7d0ca..f9e75072 100644 --- a/apps/backend/src/repositories/order-item-repo.test.ts +++ b/apps/backend/src/repositories/order-item-repo.test.ts @@ -38,7 +38,7 @@ test('resolveOrderItemSyncPlan preserves matching order item ids across repeated }, ]) assert.deepEqual(plan.creates, []) - assert.deepEqual(plan.deletes, []) + assert.deepEqual(plan.surplusExistingIds, []) }) test('resolveOrderItemSyncPlan matches by identity before falling back to position', () => { @@ -74,10 +74,10 @@ test('resolveOrderItemSyncPlan matches by identity before falling back to positi [22, 21], ) assert.deepEqual(plan.creates, []) - assert.deepEqual(plan.deletes, []) + assert.deepEqual(plan.surplusExistingIds, []) }) -test('resolveOrderItemSyncPlan keeps surplus existing ids for conditional cleanup instead of recreating everything', () => { +test('resolveOrderItemSyncPlan reports surplus existing items without deleting them', () => { const existingItems = [ { id: 31, sku_code: 'sku-a', sku_name: '礼包 A', quantity: 1 }, { id: 32, sku_code: 'sku-b', sku_name: '礼包 B', quantity: 1 }, @@ -101,7 +101,7 @@ test('resolveOrderItemSyncPlan keeps surplus existing ids for conditional cleanu [31], ) assert.deepEqual(plan.creates, []) - assert.deepEqual(plan.deletes, [32]) + assert.deepEqual(plan.surplusExistingIds, [32]) }) test('已分配订单重推时冻结原渠道决策与中文内部 SKU', () => { @@ -155,7 +155,7 @@ test('已分配订单重推时冻结原渠道决策与中文内部 SKU', () => { }) }) -test('已分配订单重推写库时不会用平台新 SKU 覆盖冻结值', async () => { +test('平台订单重推只追加新订单项，不更新或删除已有原始订单项', async () => { const existing = { id: '41', order_id: 7, @@ -184,24 +184,37 @@ test('已分配订单重推写库时不会用平台新 SKU 覆盖冻结值', asy } const updateParams: unknown[][] = [] let selectCount = 0 + let writeSql = '' const executor = async (sql: string, params: unknown[] = []) => { if (sql.includes('SELECT *') && sql.includes('FROM v_order_item_current')) { selectCount += 1 - return { rows: selectCount === 1 ? [existing] : [] } + return { rows: selectCount === 1 ? [existing] : [{ ...existing, id: 42 }] } } - if (sql.includes('UPDATE order_items')) { + if (sql.includes('INSERT INTO order_items')) { updateParams.push(params) + writeSql += sql return { rows: [], rowCount: 1 } } + if (/UPDATE\s+order_items|DELETE\s+FROM\s+order_items/i.test(sql)) { + throw new Error(`permanent source item must not be mutated: ${sql}`) + } throw new Error(`unexpected SQL: ${sql}`) } - await replaceOrderItemsWithExecutor(executor as never, 7, [incoming]) + await replaceOrderItemsWithExecutor(executor as never, 7, [ + incoming, + { + ...incoming, + skuCode: 'platform-extra', + skuName: '平台新增商品', + }, + ]) assert.equal(updateParams.length, 1) - assert.equal(updateParams[0]?.[0], 'lucky_coin_x2') - assert.equal(updateParams[0]?.[1], '幸运币2个') + assert.ok(writeSql.includes('INSERT INTO order_items')) + assert.equal(updateParams[0]?.[1], 'platform-extra') + assert.match(String(updateParams[0]?.[2]), /平台新增商品/) }) test('所有替换订单项入口都在同一事务内分摊订单收入', async () => { diff --git a/apps/backend/src/repositories/order-item-repo.ts b/apps/backend/src/repositories/order-item-repo.ts index 0fd3f8c2..ed139a97 100644 --- a/apps/backend/src/repositories/order-item-repo.ts +++ b/apps/backend/src/repositories/order-item-repo.ts @@ -13,7 +13,7 @@ type OrderItemSyncPlan = { item: OrderItemReplaceInput }> creates: OrderItemReplaceInput[] - deletes: number[] + surplusExistingIds: number[] } type OrderItemIdentityInput = Partial & Partial @@ -100,36 +100,6 @@ export async function replaceOrderItemsWithExecutor( ): Promise { const existingItems = await listOrderItemsByOrderIdWithExecutor(executor, orderId) const plan = resolveOrderItemSyncPlan(existingItems, items) - const existingById = new Map(existingItems.map((item) => [Number(item.id), item])) - - for (const update of plan.updates) { - const effectiveItem = preserveAssignedOrderItemState( - existingById.get(Number(update.orderItemId)), - update.item, - ) - await executor( - ` - UPDATE order_items - SET - sku_code = $1, - sku_name = $2, - quantity = $3, - spec_json = $4::jsonb, - item_snapshot_json = $5::jsonb, - updated_at = $6 - WHERE id = $7 - `, - [ - effectiveItem.skuCode, - effectiveItem.skuName, - update.item.quantity, - safeJsonStringify(effectiveItem.specJson || '{}'), - safeJsonStringify(effectiveItem.itemSnapshotJson || effectiveItem.specJson || '{}'), - effectiveItem.updatedAt, - update.orderItemId, - ], - ) - } for (const create of plan.creates) { await executor( @@ -158,14 +128,6 @@ export async function replaceOrderItemsWithExecutor( ) } - if (plan.deletes.length > 0) { - const deletableIds = await listDeletableOrderItemIdsWithExecutor(executor, plan.deletes) - - if (deletableIds.length > 0) { - await executor('DELETE FROM order_items WHERE id = ANY($1::bigint[])', [deletableIds]) - } - } - return listOrderItemsByOrderIdWithExecutor(executor, orderId) } @@ -280,7 +242,9 @@ export function resolveOrderItemSyncPlan( return { updates, creates, - deletes: unmatchedExisting.map((item) => Number(item.id)).filter((value) => value > 0), + surplusExistingIds: unmatchedExisting + .map((item) => Number(item.id)) + .filter((value) => value > 0), } } @@ -301,34 +265,6 @@ function buildOrderItemIdentity(item: OrderItemIdentityInput): string { ].join('::') } -async function listDeletableOrderItemIdsWithExecutor( - executor: QueryExecutor, - orderItemIds: number[], -): Promise { - const normalizedIds = Array.isArray(orderItemIds) - ? orderItemIds.map((value) => Number(value)).filter((value) => value > 0) - : [] - - if (normalizedIds.length === 0) { - return [] - } - - const result = await executor( - ` - SELECT oi.id - FROM v_order_item_current oi - LEFT JOIN fulfillment_tasks ft ON ft.order_item_id = oi.id - WHERE oi.id = ANY($1::bigint[]) - GROUP BY oi.id - HAVING COUNT(ft.id) = 0 - ORDER BY oi.id ASC - `, - [normalizedIds], - ) - - return result.rows.map((row) => Number(row.id)).filter((value) => value > 0) -} - async function listOrderItemsByOrderIdWithExecutor( executor: QueryExecutor, orderId: number | string, diff --git a/apps/backend/src/repositories/order-repo.test.ts b/apps/backend/src/repositories/order-repo.test.ts index dd183799..33195cae 100644 --- a/apps/backend/src/repositories/order-repo.test.ts +++ b/apps/backend/src/repositories/order-repo.test.ts @@ -1,8 +1,18 @@ import assert from 'node:assert/strict' +import { readFile } from 'node:fs/promises' import test from 'node:test' import { listOrdersWithExecutor, type OrderQueryExecutor } from './order-repo.js' +test('订单更新永不覆盖首次写入的平台原始报文', async () => { + const source = await readFile(new URL('./order-repo.ts', import.meta.url), 'utf8') + const updateSource = source.slice( + source.indexOf('export async function updateOrderWithExecutor'), + source.indexOf('export async function getOrderById'), + ) + assert.doesNotMatch(updateSource, /raw_payload_json\s*=/) +}) + test('订单列表综合筛选下推到总数和分页查询', async () => { const calls: Array<{ sql: string; params: unknown[] }> = [] const executor = createExecutor(calls, 3) diff --git a/apps/backend/src/repositories/order-repo.ts b/apps/backend/src/repositories/order-repo.ts index 2874a936..b0765c1e 100644 --- a/apps/backend/src/repositories/order-repo.ts +++ b/apps/backend/src/repositories/order-repo.ts @@ -191,10 +191,9 @@ export async function updateOrderWithExecutor( receiver_contact = $9, total_amount = $10, currency = $11, - raw_payload_json = $12::jsonb, - paid_at = $13, - updated_at = $14 - WHERE id = $15 + paid_at = $12, + updated_at = $13 + WHERE id = $14 RETURNING * `, [ @@ -209,7 +208,6 @@ export async function updateOrderWithExecutor( sanitizeWellFormedUnicode(input.receiverContact), input.totalAmount, input.currency, - safeJsonStringify(input.rawPayloadJson || '{}'), input.paidAt || null, input.updatedAt, Number(orderId), diff --git a/apps/backend/src/repositories/order-source-facts.integration.test.ts b/apps/backend/src/repositories/order-source-facts.integration.test.ts new file mode 100644 index 00000000..138339f2 --- /dev/null +++ b/apps/backend/src/repositories/order-source-facts.integration.test.ts @@ -0,0 +1,137 @@ +/** + * 平台订单原始事实不可变集成测试（需要真实 PostgreSQL）。 + * + * ORDER_SOURCE_FACTS_TEST_DATABASE_URL=postgres://... \ + * npx tsx --test src/repositories/order-source-facts.integration.test.ts + */ +import assert from 'node:assert/strict' +import test from 'node:test' + +const TEST_DATABASE_URL = String(process.env.ORDER_SOURCE_FACTS_TEST_DATABASE_URL || '').trim() + +test( + '平台重推只追加原始事实，不覆盖订单报文、不删除已有订单项', + { skip: !TEST_DATABASE_URL ? '未设置 ORDER_SOURCE_FACTS_TEST_DATABASE_URL' : false }, + async () => { + process.env.DATABASE_URL = TEST_DATABASE_URL + const { closeDb, query, withTransaction } = await import('../db/client.js') + const { appendOrderSourceSnapshotWithExecutor } = + await import('./order-source-snapshot-repo.js') + const { replaceOrderItemsWithExecutor } = await import('./order-item-repo.js') + const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}` + let orderId = 0 + + try { + const order = await query<{ id: number }>( + ` + INSERT INTO orders ( + provider, platform, shop_id, shop_name, platform_order_id, + order_status, pay_status, total_amount, raw_payload_json, created_at, updated_at + ) VALUES ( + '91kaquan', 'kuaishou-91', $1, '原始事实测试店', $2, + 'paid', 'paid', 1000, $3::jsonb, NOW(), NOW() + ) RETURNING id + `, + [ + `source-facts-shop-${suffix}`, + `SOURCE-FACTS-${suffix}`, + JSON.stringify({ orderNo: `SOURCE-FACTS-${suffix}`, version: 1 }), + ], + ) + orderId = Number(order.rows[0]?.id || 0) + assert.ok(orderId > 0) + + const item = await query<{ id: number }>( + ` + INSERT INTO order_items ( + order_id, sku_code, sku_name, quantity, spec_json, item_snapshot_json, + revenue_amount_fen, revenue_basis, created_at, updated_at + ) VALUES ( + $1, 'sku-old', '原始商品', 1, '{}'::jsonb, '{}'::jsonb, + 1000, 'order_total', NOW(), NOW() + ) RETURNING id + `, + [orderId], + ) + const originalItemId = Number(item.rows[0]?.id || 0) + assert.ok(originalItemId > 0) + + const snapshotInput = { + orderId, + provider: '91kaquan', + platform: 'kuaishou-91', + shopId: `source-facts-shop-${suffix}`, + platformOrderId: `SOURCE-FACTS-${suffix}`, + sourceLabel: 'integration-test', + payload: { orderNo: `SOURCE-FACTS-${suffix}`, version: 1, items: ['sku-old'] }, + recordedAt: new Date().toISOString(), + } + assert.equal(await appendOrderSourceSnapshotWithExecutor(query, snapshotInput), true) + assert.equal(await appendOrderSourceSnapshotWithExecutor(query, snapshotInput), false) + + const snapshots = await query<{ count: number }>( + 'SELECT COUNT(*)::int AS count FROM order_source_snapshots WHERE order_id = $1', + [orderId], + ) + assert.equal(Number(snapshots.rows[0]?.count), 1) + + await replaceOrderItemsWithExecutor(query, orderId, [ + { + skuCode: 'sku-new', + skuName: '平台新商品', + quantity: 1, + specJson: '{}', + itemSnapshotJson: '{}', + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }, + { + skuCode: 'sku-extra', + skuName: '平台新增商品', + quantity: 1, + specJson: '{}', + itemSnapshotJson: '{}', + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }, + ]) + + const items = await query<{ original_exists: boolean; total: number }>( + ` + SELECT + EXISTS (SELECT 1 FROM order_items WHERE id = $1) AS original_exists, + (SELECT COUNT(*)::int FROM order_items WHERE order_id = $2) AS total + `, + [originalItemId, orderId], + ) + assert.equal(items.rows[0]?.original_exists, true) + assert.equal(Number(items.rows[0]?.total), 2) + + await replaceOrderItemsWithExecutor(query, orderId, [ + { + skuCode: 'sku-new', + skuName: '平台新商品', + quantity: 1, + specJson: '{}', + itemSnapshotJson: '{}', + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }, + ]) + const afterShrink = await query<{ total: number }>( + 'SELECT COUNT(*)::int AS total FROM order_items WHERE order_id = $1', + [orderId], + ) + assert.equal(Number(afterShrink.rows[0]?.total), 2) + } finally { + if (orderId > 0) { + await withTransaction(async (client) => { + await client.query(`SET LOCAL app.allow_order_fact_repair = 'on'`) + await client.query('DELETE FROM order_source_snapshots WHERE order_id = $1', [orderId]) + await client.query('DELETE FROM orders WHERE id = $1', [orderId]) + }) + } + await closeDb() + } + }, +) diff --git a/apps/backend/src/repositories/order-source-snapshot-repo.test.ts b/apps/backend/src/repositories/order-source-snapshot-repo.test.ts new file mode 100644 index 00000000..2fbcc6a7 --- /dev/null +++ b/apps/backend/src/repositories/order-source-snapshot-repo.test.ts @@ -0,0 +1,40 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { + appendOrderSourceSnapshotWithExecutor, + orderSourceSnapshotHash, +} from './order-source-snapshot-repo.js' + +test('订单来源快照哈希稳定且能区分不同报文', () => { + const first = orderSourceSnapshotHash({ orderNo: 'A-1', status: 'paid' }) + assert.equal(first, orderSourceSnapshotHash({ orderNo: 'A-1', status: 'paid' })) + assert.notEqual(first, orderSourceSnapshotHash({ orderNo: 'A-1', status: 'refunded' })) + assert.match(first, /^[0-9a-f]{64}$/) +}) + +test('同订单相同来源快照只追加一次', async () => { + const calls: unknown[][] = [] + const executor = async (sql: string, params: unknown[] = []) => { + calls.push(params) + assert.ok(sql.includes('INSERT INTO order_source_snapshots')) + assert.ok(sql.includes('ON CONFLICT (order_id, payload_sha256) DO NOTHING')) + return { rows: [{ id: 1 }] } + } + + const inserted = await appendOrderSourceSnapshotWithExecutor(executor as never, { + orderId: 7, + provider: 'kuaishou-91', + platform: 'kuaishou-91', + shopId: 'shop-1', + platformOrderId: 'ORDER-1', + sourceLabel: 'source-order-upsert', + payload: { orderNo: 'ORDER-1', status: 'paid' }, + recordedAt: '2026-11-09T00:00:00.000Z', + }) + + assert.equal(inserted, true) + assert.equal(calls.length, 1) + assert.equal(calls[0]?.[0], 7) + assert.equal(calls[0]?.[7], orderSourceSnapshotHash({ orderNo: 'ORDER-1', status: 'paid' })) +}) diff --git a/apps/backend/src/repositories/order-source-snapshot-repo.ts b/apps/backend/src/repositories/order-source-snapshot-repo.ts new file mode 100644 index 00000000..8b02422c --- /dev/null +++ b/apps/backend/src/repositories/order-source-snapshot-repo.ts @@ -0,0 +1,67 @@ +import { createHash } from 'node:crypto' +import type { QueryResult, QueryResultRow } from 'pg' + +import { safeJsonStringify } from '../utils/json.js' + +type QueryExecutor = ( + text: string, + params?: unknown[], +) => Promise> + +export type OrderSourceSnapshotInput = { + orderId: number + provider: string + platform: string + shopId: string + platformOrderId: string + sourceLabel: string + payload: unknown + recordedAt: string +} + +export function orderSourceSnapshotHash(payload: unknown): string { + return createHash('sha256') + .update(safeJsonStringify(payload || {})) + .digest('hex') +} + +/** + * 追加平台订单原始快照。 + * 同一订单的相同内容只写一次，不同内容永远追加，不覆盖订单主表原始报文。 + */ +export async function appendOrderSourceSnapshotWithExecutor( + executor: QueryExecutor, + input: OrderSourceSnapshotInput, +): Promise { + const payloadJson = safeJsonStringify(input.payload || {}) + const payloadHash = orderSourceSnapshotHash(input.payload) + const result = await executor<{ id: number }>( + ` + INSERT INTO order_source_snapshots ( + order_id, + provider, + platform, + shop_id, + platform_order_id, + source_label, + payload_json, + payload_sha256, + recorded_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9) + ON CONFLICT (order_id, payload_sha256) DO NOTHING + RETURNING id + `, + [ + input.orderId, + input.provider, + input.platform, + input.shopId, + input.platformOrderId, + input.sourceLabel, + payloadJson, + payloadHash, + input.recordedAt, + ], + ) + return result.rows.length > 0 +} diff --git a/apps/backend/src/services/order/order-service.ts b/apps/backend/src/services/order/order-service.ts index c717e4de..9255bff8 100644 --- a/apps/backend/src/services/order/order-service.ts +++ b/apps/backend/src/services/order/order-service.ts @@ -7,6 +7,7 @@ import { applyOrderItemRevenueAllocationWithExecutor, replaceOrderItemsWithExecutor, } from '../../repositories/order-item-repo.js' +import { appendOrderSourceSnapshotWithExecutor } from '../../repositories/order-source-snapshot-repo.js' import { withTransaction } from '../../db/client.js' import { syncDeliveryTasksForOrder } from './delivery-task-service.js' import { resolveOrderItemForFulfillment } from '../fulfillment/product-resolution-service.js' @@ -189,6 +190,34 @@ export async function upsertOrderFromSource( }) } + await appendOrderSourceSnapshotWithExecutor(executor, { + orderId: Number(order.id), + provider: event.provider, + platform: event.platform, + shopId: String(order.shop_id || event.shopId || '').trim(), + platformOrderId: event.platformOrderId, + sourceLabel, + payload: { + sourceLabel, + provider: event.provider, + platform: event.platform, + shopId: event.shopId, + shopName: event.shopName, + platformOrderId: event.platformOrderId, + orderStatus: event.orderStatus, + payStatus: event.payStatus, + buyerId: event.buyerId, + buyerName: event.buyerName, + receiverContact: event.receiverContact, + totalAmount: event.totalAmount, + currency: event.currency, + paidAt: event.paidAt || null, + rawPayload: event.rawPayload, + items: event.items, + }, + recordedAt: now, + }) + const orderItems = await replaceOrderItemsWithExecutor( executor, order.id, diff --git a/apps/backend/src/services/platforms/ninetyone/order-service.ts b/apps/backend/src/services/platforms/ninetyone/order-service.ts index 78153313..fe8e2eec 100644 --- a/apps/backend/src/services/platforms/ninetyone/order-service.ts +++ b/apps/backend/src/services/platforms/ninetyone/order-service.ts @@ -25,9 +25,8 @@ import type { OrderItemRow, OrderRow } from '../../../types/repository/rows.js' export const OPEN_91_PENDING_CONFIG_STATUS = 'pending_config' export const OPEN_91_MANUAL_FAILED_STATUS = 'manual_failed' -export async function markOpen91OrderPendingConfig(order: OrderRow, reason = '') { +export async function markOpen91OrderPendingConfig(order: OrderRow, _reason = '') { const now = nowIso() - const rawPayload = parseJsonObject(order.raw_payload_json) return updateOrder(order.id, { provider: order.provider, platform: order.platform, @@ -41,11 +40,6 @@ export async function markOpen91OrderPendingConfig(order: OrderRow, reason = '') receiverContact: order.receiver_contact, totalAmount: Number(order.total_amount) || 0, currency: order.currency, - rawPayloadJson: JSON.stringify({ - ...rawPayload, - pendingReason: String(reason || '').trim() || 'fulfillment_not_ready', - pendingAt: now, - }), paidAt: order.paid_at, updatedAt: now, }) @@ -154,11 +148,6 @@ export async function upsertOpen91PendingOrder(payload: JsonObject = {}, config: platformOrderId: event.platformOrderId, }) - const rawPayloadJson = JSON.stringify({ - ...event.rawPayload, - pendingReason: 'unconfigured_items', - pendingAt: now, - }) const orderPayload = { provider: event.provider, platform: event.platform, @@ -172,7 +161,6 @@ export async function upsertOpen91PendingOrder(payload: JsonObject = {}, config: receiverContact: '', totalAmount: event.totalAmount, currency: 'CNY', - rawPayloadJson, paidAt: existing?.paid_at || now, } @@ -245,10 +233,9 @@ export async function retryOpen91Order(orderId: string | number) { } } -export async function failOpen91Order(orderId: string | number, reason = '') { +export async function failOpen91Order(orderId: string | number, _reason = '') { const order = await getRequiredOpen91Order(orderId) const now = nowIso() - const rawPayload = parseJsonObject(order.raw_payload_json) const updated = await updateOrder(order.id, { provider: order.provider, platform: order.platform, @@ -262,11 +249,6 @@ export async function failOpen91Order(orderId: string | number, reason = '') { receiverContact: order.receiver_contact, totalAmount: order.total_amount, currency: order.currency, - rawPayloadJson: JSON.stringify({ - ...rawPayload, - manualFailedAt: now, - manualFailedReason: String(reason || '').trim() || '商家手动标记无法履约', - }), paidAt: order.paid_at, updatedAt: now, }) diff --git a/apps/backend/src/services/scheduler/data-retention-job.test.ts b/apps/backend/src/services/scheduler/data-retention-job.test.ts index ac70c39c..88bdffb0 100644 --- a/apps/backend/src/services/scheduler/data-retention-job.test.ts +++ b/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'])", diff --git a/apps/backend/src/services/scheduler/data-retention-job.ts b/apps/backend/src/services/scheduler/data-retention-job.ts index f5a9d7ea..0c4a2d66 100644 --- a/apps/backend/src/services/scheduler/data-retention-job.ts +++ b/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 { 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 { 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 diff --git a/apps/backend/src/services/scheduler/scheduler-service.ts b/apps/backend/src/services/scheduler/scheduler-service.ts index 6f36ca39..2190a16c 100644 --- a/apps/backend/src/services/scheduler/scheduler-service.ts +++ b/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), } : {}), }) diff --git a/apps/backend/src/types/runtime-config.ts b/apps/backend/src/types/runtime-config.ts index 0239e9c4..faa95205 100644 --- a/apps/backend/src/types/runtime-config.ts +++ b/apps/backend/src/types/runtime-config.ts @@ -28,12 +28,6 @@ export type RuntimeConfig = { maxFiles?: number } retention: { - rawPayloadDays: number - taskEventDays: number - webhookEventDays: number - auditLogDays: number - /** 工单事件（work_order_events）保留天数：这张表只增不减，必须纳入清理。 */ - workOrderEventDays: number /** 已处理成本异常（cost_exceptions）保留天数：超期归档，open 行不归档。 */ costExceptionDays: number } diff --git a/docs/统一订单履约与财务模型-权威版.md b/docs/统一订单履约与财务模型-权威版.md index 3519527f..7d1d797c 100644 --- a/docs/统一订单履约与财务模型-权威版.md +++ b/docs/统一订单履约与财务模型-权威版.md @@ -3,16 +3,16 @@ type: 权威架构与实施计划 description: 订单、内部 SKU、履约路由、履约执行、商品、成本与财务的唯一权威口径及重构计划。 tags: [订单, 内部SKU, 履约, 商品, 成本, 财务, 迁移] status: authoritative -version: 2.4 -updated_at: 2026-11-08 +version: 2.5 +updated_at: 2026-11-09 --- # 统一订单、履约与财务模型 > 状态：**唯一权威文档**。 -> 最后校准：2026-11-08。 -> 生产基线：迁移 `146`。 -> 开发基线：迁移 `150`（仅本地，尚未部署生产）。 +> 最后校准：2026-11-09。 +> 生产基线：迁移 `151`。 +> 开发基线：迁移 `152`（仅本地，尚未部署生产）。 > > 从本文件生效起，订单、内部 SKU、履约渠道、履约载体、标准商品、成本和财务相关实现， > 只能以本文件为业务口径。其他旧分析只作为历史资料，不再具备设计或验收效力。 @@ -313,7 +313,7 @@ finance_journal_* -- 总账凭证，可重建 | 信息 | 唯一写入方 | 唯一真源 | 明确禁止 | | --- | --- | --- | --- | | 平台订单与订单项 | 订单接入服务 | `orders` / `order_items` | 财务或履约服务改写平台事实 | -| 平台原始报文 | 各平台适配器 | 原始表 / `raw_payload_json` | 覆盖历史报文 | +| 平台原始报文 | 各平台适配器 | `order_source_snapshots` / 首次 `raw_payload_json` | 覆盖历史报文 | | 中文内部 SKU | 商品身份服务 | `internal_skus` | 各业务表各自发明同义 SKU | | 平台字段到内部 SKU | 映射服务 | `internal_sku_mappings` | 按名称猜测渠道 | | 主渠道与兜底 | 路由服务 | `sku_channel_routes` | UI 根据商品前缀筛选 | @@ -366,6 +366,7 @@ finance_journal_* -- 总账凭证，可重建 - `orders` - `order_items` 的原始平台字段 - 平台原始订单表与原始报文字段 +- `order_source_snapshots` 平台订单不可变事件快照 - 平台售后、退款、支付事实 - `work_orders`、`work_order_shares`、`work_order_events` - `fulfillment_tasks` 中真实执行与结果记录 @@ -377,6 +378,14 @@ finance_journal_* -- 总账凭证，可重建 A 级数据只能追加新事件或做字段兼容迁移，不能为了“模型整齐”重写历史事实。 +订单重复同步必须遵守： + +- 首次写入的 `orders.raw_payload_json` 不再覆盖； +- 每次来源同步追加 `order_source_snapshots`，相同内容幂等； +- 已有 `order_items` 的平台 SKU、名称、数量、规格和快照不因重推更新； +- 重推中缺少的订单项只记录为来源差异，不得物理删除订单项或级联删除履约记录； +- 数据保留任务不得清空或删除 A 级事实，只允许维护可重建的派生数据。 + ### B 级：业务输入，必须迁移而不是重建 - `internal_sku_mappings`