From 2e8c9715b64170f6d3b87b0670b1bdfe689bb24b Mon Sep 17 00:00:00 2001 From: yml2213 Date: Thu, 17 Sep 2026 13:34:39 +0800 Subject: [PATCH] =?UTF-8?q?fix(migration):=20079=20=E5=BB=BA=E5=86=BB?= =?UTF-8?q?=E7=BB=93=E5=B9=82=E7=AD=89=E7=B4=A2=E5=BC=95=E5=89=8D=E5=85=88?= =?UTF-8?q?=E5=8E=BB=E9=87=8D=E5=8E=86=E5=8F=B2=E9=87=8D=E5=A4=8D=E5=86=BB?= =?UTF-8?q?=E7=BB=93=E5=B9=B6=E9=80=80=E8=BF=98=E5=A4=9A=E5=86=BB=E6=8A=BC?= =?UTF-8?q?=E9=87=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 生产首跑 079 在 uniq_worker_wallet_ledgers_consignment_publish_freeze 索引上 23505：历史存在同提报单两条 consignment_publish 冻结流水 （如提报单 5），正是该索引要防的二次冻结缺陷的脏数据产物。 与 079 处理重复绑定工单同套路——先清数据再加约束： - 每组（worker+提报单）保留最早一条冻结流水，多余行仅改 payload 标记 consignment_publish_superseded（移出押金口径与索引范围，原行留审计） - 多冻的钱退还钱包：退额 = LEAST(多冻总额, 冻结池实有)，绝不无中生有； 写 deposit_release 修正流水（kind=consignment_publish_dedup_refund） - 同一打手多条多余流水先按钱包聚合再退，避免 UPDATE..FROM 漏退 - 再建幂等索引 已在一次性 Postgres 完整复现生产路径验证：078+脏数据 → 079/080 迁移 成功、钱包退回多冻 ¥20、索引生效拦截第三次冻结 --- .../079_drop_consignment_collect_links.sql | 80 ++++++++++++++++++- .../consignment-payout.test.ts | 21 +++++ 2 files changed, 100 insertions(+), 1 deletion(-) diff --git a/apps/backend/src/db/migrations/079_drop_consignment_collect_links.sql b/apps/backend/src/db/migrations/079_drop_consignment_collect_links.sql index 566a8a09..b40c8caf 100644 --- a/apps/backend/src/db/migrations/079_drop_consignment_collect_links.sql +++ b/apps/backend/src/db/migrations/079_drop_consignment_collect_links.sql @@ -96,7 +96,85 @@ CREATE UNIQUE INDEX IF NOT EXISTS uniq_consignment_inventories_bound_work_order -- 5) 寄卖发布押金的冻结核销行幂等键：同一提报单只能有一笔发布冻结， -- 防止未来任何绕过"先读台账再冻结"约定的调用方重复扣款。 +-- 生产库存在历史重复冻结（发布链路二次冻结缺陷的产物，如提报单 5 有两条冻结流水）， +-- 直接建索引会 23505 失败：先与上面"重复绑定工单"同套路清理——保留每组最早一条冻结流水， +-- 多余流水退还多冻的钱（退额 = LEAST(多冻总额, 冻结池实有)，绝不无中生有）并写修正流水， +-- 仅改 payload 标记 superseded（移出押金口径与索引范围，原行保留可审计），再建索引。 +WITH ranked_freezes AS ( + SELECT + id, + ROW_NUMBER() OVER ( + PARTITION BY worker_id, payload_json ->> 'consignmentSubmissionId' + ORDER BY id ASC + ) AS rn + FROM worker_wallet_ledgers + WHERE ledger_type = 'deposit_freeze' + AND COALESCE(payload_json ->> 'kind', '') = 'consignment_publish' +), +superseded_freezes AS ( + UPDATE worker_wallet_ledgers + SET + payload_json = payload_json || jsonb_build_object( + 'kind', 'consignment_publish_superseded', + 'supersededBy', '079_consignment_publish_freeze_dedup', + 'supersededReason', '同一提报单重复冻结，多余一笔已退还并移出押金口径' + ) + WHERE id IN (SELECT id FROM ranked_freezes WHERE rn > 1) + RETURNING id, worker_id, amount +), +-- 同一打手可能有多条多余流水：先按钱包聚合，避免 UPDATE ... FROM 只吃到其中一条 +superseded_totals AS ( + SELECT + worker_id, + SUM(ABS(amount)) AS dup_frozen_fen, + jsonb_agg(id) AS source_ledger_ids + FROM superseded_freezes + GROUP BY worker_id +), +refund_plan AS ( + SELECT + st.worker_id, + LEAST(st.dup_frozen_fen, COALESCE(w.frozen_deposit_amount, 0)) AS refund_fen, + COALESCE(w.available_amount, 0) + + LEAST(st.dup_frozen_fen, COALESCE(w.frozen_deposit_amount, 0)) AS available_after, + COALESCE(w.frozen_deposit_amount, 0) + - LEAST(st.dup_frozen_fen, COALESCE(w.frozen_deposit_amount, 0)) AS frozen_after, + st.source_ledger_ids + FROM superseded_totals st + LEFT JOIN worker_wallets w ON w.worker_id = st.worker_id +), +applied_refunds AS ( + UPDATE worker_wallets w + SET + available_amount = rp.available_after, + frozen_deposit_amount = rp.frozen_after, + updated_at = NOW() + FROM refund_plan rp + WHERE w.worker_id = rp.worker_id + RETURNING w.worker_id +) +INSERT INTO worker_wallet_ledgers ( + worker_id, ledger_type, amount, balance_after, frozen_after, + related_work_order_id, note, payload_json, created_at +) +SELECT + rp.worker_id, + 'deposit_release', + rp.refund_fen, + rp.available_after, + rp.frozen_after, + NULL, + '重复冻结押金自动退还（079 迁移去重）', + jsonb_build_object( + 'kind', 'consignment_publish_dedup_refund', + 'sourceLedgerIds', rp.source_ledger_ids + ), + NOW() +FROM refund_plan rp +JOIN applied_refunds ar ON ar.worker_id = rp.worker_id +WHERE rp.refund_fen > 0; + CREATE UNIQUE INDEX IF NOT EXISTS uniq_worker_wallet_ledgers_consignment_publish_freeze ON worker_wallet_ledgers ((payload_json ->> 'consignmentSubmissionId')) WHERE ledger_type = 'deposit_freeze' - AND payload_json ->> 'kind' = 'consignment_publish'; + AND COALESCE(payload_json ->> 'kind', '') = 'consignment_publish'; diff --git a/apps/backend/src/services/worker-platform/consignment-payout.test.ts b/apps/backend/src/services/worker-platform/consignment-payout.test.ts index 8c443a80..43b1797a 100644 --- a/apps/backend/src/services/worker-platform/consignment-payout.test.ts +++ b/apps/backend/src/services/worker-platform/consignment-payout.test.ts @@ -128,3 +128,24 @@ test('拼单份额全部验收（母单转已验收）也必须补寄卖钩子 '份额验收收官时必须结算供货货款（拼单母单无接单人，传 0 走供货渠道）', ) }) + +test('迁移 079：建冻结幂等索引前必须先去重历史重复冻结并退还多冻押金', async () => { + const migration = await readFile( + new URL('../../db/migrations/079_drop_consignment_collect_links.sql', import.meta.url), + 'utf8', + ) + // 生产曾因同提报单两条 consignment_publish 冻结流水建不起索引（23505）， + // 修复是"先清脏数据再建约束"：去重退还必须在 CREATE UNIQUE INDEX 之前，顺序不能反 + const dedupPos = migration.indexOf("'consignment_publish_superseded'") + const refundPos = migration.indexOf("'consignment_publish_dedup_refund'") + const indexPos = migration.indexOf( + 'CREATE UNIQUE INDEX IF NOT EXISTS uniq_worker_wallet_ledgers_consignment_publish_freeze', + ) + assert.ok(dedupPos > 0, '079 必须包含重复冻结去重（标记 superseded）') + assert.ok(refundPos > dedupPos, '去重后必须写退还修正流水') + assert.ok(indexPos > refundPos, '幂等索引必须建在去重退还之后，否则脏数据会 23505') + assert.ok( + migration.includes('LEAST(st.dup_frozen_fen, COALESCE(w.frozen_deposit_amount, 0))'), + '退还额必须封顶冻结池实有，绝不无中生有', + ) +})