From 48507a386189ef855596ab87729bf77f18b81016 Mon Sep 17 00:00:00 2001 From: zhx Date: Tue, 21 Jul 2026 15:49:11 +0800 Subject: [PATCH] feat(wallet): safely purge verified outbox archives --- .../wallet-outbox-archive-restore/README.md | 53 +++ .../cmd/wallet-outbox-archive-restore/main.go | 95 ++++ .../wallet-service/configs/config.docker.yaml | 9 +- .../configs/config.tencent.example.yaml | 9 +- services/wallet-service/configs/config.yaml | 9 +- .../mysql/initdb/001_wallet_service.sql | 54 +++ ...3_wallet_outbox_archive_purge_receipts.sql | 53 +++ services/wallet-service/internal/app/app.go | 4 + .../internal/app/background_workers.go | 5 +- .../internal/app/outbox_archive.go | 8 + .../internal/app/outbox_purge.go | 107 +++++ .../wallet-service/internal/config/config.go | 129 ++++- .../internal/outboxarchiverestore/mysql.go | 440 ++++++++++++++++++ .../internal/outboxarchiverestore/verify.go | 431 +++++++++++++++++ .../internal/storage/mysql/outbox_archive.go | 132 +++++- .../internal/storage/mysql/outbox_purge.go | 378 +++++++++++++++ 16 files changed, 1894 insertions(+), 22 deletions(-) create mode 100644 services/wallet-service/cmd/wallet-outbox-archive-restore/README.md create mode 100644 services/wallet-service/cmd/wallet-outbox-archive-restore/main.go create mode 100644 services/wallet-service/deploy/mysql/migrations/013_wallet_outbox_archive_purge_receipts.sql create mode 100644 services/wallet-service/internal/app/outbox_purge.go create mode 100644 services/wallet-service/internal/outboxarchiverestore/mysql.go create mode 100644 services/wallet-service/internal/outboxarchiverestore/verify.go create mode 100644 services/wallet-service/internal/storage/mysql/outbox_purge.go diff --git a/services/wallet-service/cmd/wallet-outbox-archive-restore/README.md b/services/wallet-service/cmd/wallet-outbox-archive-restore/README.md new file mode 100644 index 00000000..126185a5 --- /dev/null +++ b/services/wallet-service/cmd/wallet-outbox-archive-restore/README.md @@ -0,0 +1,53 @@ +# Wallet outbox archive restore rehearsal + +This command accepts the exact `data.ndjson.gz`, `manifest.json`, and `_SUCCESS` +files from one private COS batch. It verifies `_SUCCESS` before opening MySQL, +then checks manifest/data SHA-256, COS CRC64-ECMA, gzip integrity, row count, +cursor bounds, UTC time bounds, unique source PKs, and both delta control totals. + +On a pay node, place the three files in a mode `0700` temporary directory using +the attached `HyAppWalletOutboxArchiveRole`; do not copy the operator CSV key or +long-lived credentials onto the node. Keep the files mode `0600` and remove the +temporary directory after the report is captured. + +Verify files without database writes: + +```bash +go run ./services/wallet-service/cmd/wallet-outbox-archive-restore \ + --data /secure/batch/data.ndjson.gz \ + --manifest /secure/batch/manifest.json \ + --success /secure/batch/_SUCCESS +``` + +Restore to the standard isolation contract. The DSN is read from an environment +variable so it does not appear in shell history or process arguments. The exact +database check prevents a mistyped DSN from silently selecting another schema. + +```bash +read -rsp 'Restore MySQL DSN: ' WALLET_OUTBOX_ARCHIVE_RESTORE_DSN && export WALLET_OUTBOX_ARCHIVE_RESTORE_DSN +go run ./services/wallet-service/cmd/wallet-outbox-archive-restore \ + --data /secure/batch/data.ndjson.gz \ + --manifest /secure/batch/manifest.json \ + --success /secure/batch/_SUCCESS \ + --restore \ + --expected-database wallet_archive_restore \ + --rehearsal-id rehearsal-20260721-001 \ + --confirm RESTORE:rehearsal-20260721-001: +``` + +The command creates only these versioned isolation tables: + +- `wallet_outbox_restore_rehearsals_v1`: one atomic state/evidence row per rehearsal. +- `wallet_outbox_restore_rows_v1`: wallet_outbox-compatible fields, keyed by + `(rehearsal_id, app_code, event_id)` and tagged with `source_batch_id`. + +`RESTORE_VERIFIED` means the three files were cryptographically verified, every +row was inserted into the isolation table, and SQL readback reproduced the +manifest controls. It is not purge authorization. Any later purge workflow must +also match the computed data/manifest/success CRC values to the owner receipt, +prove all consumer watermarks are beyond the batch cursor, and delete only small +exact `(app_code,event_id)` PK sets under its own concurrency gate. + +The batch states are intentionally monotonic: manifest `DATA_VERIFIED` -> marker +`VERIFIED` -> CLI `FILES_VERIFIED` -> transactional `RESTORING` -> persisted +`RESTORE_VERIFIED`. No state in this command maps to `PURGE_ELIGIBLE`. diff --git a/services/wallet-service/cmd/wallet-outbox-archive-restore/main.go b/services/wallet-service/cmd/wallet-outbox-archive-restore/main.go new file mode 100644 index 00000000..8de527df --- /dev/null +++ b/services/wallet-service/cmd/wallet-outbox-archive-restore/main.go @@ -0,0 +1,95 @@ +// Command wallet-outbox-archive-restore validates one downloaded COS batch and, +// only after _SUCCESS and every checksum pass, writes it to isolation tables. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "os" + "strings" + "time" + + "hyapp/services/wallet-service/internal/outboxarchiverestore" +) + +const defaultDSNEnv = "WALLET_OUTBOX_ARCHIVE_RESTORE_DSN" + +func main() { + dataPath := flag.String("data", "", "local data.ndjson.gz downloaded from the private archive bucket") + manifestPath := flag.String("manifest", "", "local manifest.json from the same batch directory") + successPath := flag.String("success", "", "local _SUCCESS from the same batch directory") + restore := flag.Bool("restore", false, "persist the verified batch into fixed isolation tables") + dsnEnv := flag.String("dsn-env", defaultDSNEnv, "environment variable containing the restore-target MySQL DSN") + expectedDatabase := flag.String("expected-database", "", "exact restore-target database returned by SELECT DATABASE()") + rehearsalID := flag.String("rehearsal-id", "", "stable 1..64 character restore rehearsal key") + confirmation := flag.String("confirm", "", "restore confirmation RESTORE::") + timeout := flag.Duration("timeout", 10*time.Minute, "whole verification/restore timeout, at most 30m") + flag.Parse() + + if strings.TrimSpace(*dataPath) == "" || strings.TrimSpace(*manifestPath) == "" || strings.TrimSpace(*successPath) == "" { + fatal(errors.New("--data, --manifest, and --success are required")) + } + if *timeout <= 0 || *timeout > 30*time.Minute { + fatal(errors.New("--timeout must be within (0,30m]")) + } + ctx, cancel := context.WithTimeout(context.Background(), *timeout) + defer cancel() + + verification, err := outboxarchiverestore.VerifyLocalFiles(*dataPath, *manifestPath, *successPath, outboxarchiverestore.DefaultLimits()) + if err != nil { + fatal(err) + } + if !*restore { + printJSON(map[string]any{ + "state": "FILES_VERIFIED", "batch_id": verification.Manifest.BatchID, + "app_code": verification.Manifest.AppCode, "row_count": verification.Manifest.RowCount, + "data_object_key": verification.Manifest.DataObjectKey, + "first_cursor": verification.Manifest.FirstCursor, "last_cursor": verification.Manifest.LastCursor, + "ndjson_sha256": verification.Manifest.NDJSONSHA256, "gzip_sha256": verification.Manifest.GzipSHA256, + "data_crc64": verification.Manifest.DataCRC64, + "manifest_sha256": verification.ManifestSHA256, "manifest_crc64": verification.ManifestCRC64, + "success_sha256": verification.SuccessSHA256, "success_crc64": verification.SuccessCRC64, + "available_delta_total": verification.Manifest.AvailableDeltaTotal, + "frozen_delta_total": verification.Manifest.FrozenDeltaTotal, "no_database_writes": true, + }) + return + } + + rehearsal := strings.TrimSpace(*rehearsalID) + expectedConfirmation := "RESTORE:" + rehearsal + ":" + verification.Manifest.BatchID + if rehearsal == "" || strings.TrimSpace(*confirmation) != expectedConfirmation { + fatal(fmt.Errorf("--restore requires --rehearsal-id and --confirm=%s", expectedConfirmation)) + } + envName := strings.TrimSpace(*dsnEnv) + if envName == "" { + fatal(errors.New("--dsn-env cannot be empty")) + } + dsn := strings.TrimSpace(os.Getenv(envName)) + if dsn == "" { + fatal(fmt.Errorf("restore DSN environment variable %s is empty", envName)) + } + result, err := outboxarchiverestore.Restore(ctx, dsn, *expectedDatabase, rehearsal, verification) + if err != nil { + fatal(err) + } + printJSON(struct { + outboxarchiverestore.RestoreResult + DSNEnv string `json:"dsn_env"` + }{RestoreResult: result, DSNEnv: envName}) +} + +func printJSON(value any) { + body, err := json.Marshal(value) + if err != nil { + fatal(err) + } + fmt.Println(string(body)) +} + +func fatal(err error) { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) +} diff --git a/services/wallet-service/configs/config.docker.yaml b/services/wallet-service/configs/config.docker.yaml index e5998d60..96c7fc92 100644 --- a/services/wallet-service/configs/config.docker.yaml +++ b/services/wallet-service/configs/config.docker.yaml @@ -158,10 +158,17 @@ outbox_worker: initial_backoff: "5s" max_backoff: "5m" outbox_archive: - # 未配置私有 COS 的容器默认不扫描热表；显式启用后先用 dry_run 核对范围。 + # 未配置私有 COS 的容器默认不扫描热表；purge 还需要显式租户、恢复凭证和生产授权记录。 enabled: false mode: "dry_run" purge_enabled: false + purge_app_codes: [] + purge_max_age: "720h" + purge_poll_interval: "5s" + purge_batch_size: 100 + purge_receipt_grace: "1h" + purge_policy_version: "financial-v1" + purge_restore_proofs: [] max_age: "720h" poll_interval: "5m" query_batch_size: 500 diff --git a/services/wallet-service/configs/config.tencent.example.yaml b/services/wallet-service/configs/config.tencent.example.yaml index 3be7bf1c..4430e068 100644 --- a/services/wallet-service/configs/config.tencent.example.yaml +++ b/services/wallet-service/configs/config.tencent.example.yaml @@ -162,8 +162,15 @@ outbox_archive: # 模板默认关闭；生产私有配置补齐专用桶/角色后显式切 archive_only。 enabled: false mode: "dry_run" - # 目录版本、notice repair、statistics replay 和 legacy projection 仍读热表，当前任何环境都禁止 purge。 + # 自动清理必须再配置租户 allowlist、真实恢复凭证和生产库授权；模板永远不默认打开。 purge_enabled: false + purge_app_codes: [] + purge_max_age: "720h" + purge_poll_interval: "5s" + purge_batch_size: 100 + purge_receipt_grace: "1h" + purge_policy_version: "financial-v1" + purge_restore_proofs: [] max_age: "720h" poll_interval: "5m" query_batch_size: 500 diff --git a/services/wallet-service/configs/config.yaml b/services/wallet-service/configs/config.yaml index e9b216ed..9d15aba7 100644 --- a/services/wallet-service/configs/config.yaml +++ b/services/wallet-service/configs/config.yaml @@ -160,10 +160,17 @@ outbox_worker: initial_backoff: "5s" max_backoff: "5m" outbox_archive: - # 未配置私有 COS 的环境默认不扫描热表；显式启用后先用 dry_run 核对范围。 + # 未配置私有 COS 的环境默认不扫描热表；purge 还需要显式租户、恢复凭证和生产授权记录。 enabled: false mode: "dry_run" purge_enabled: false + purge_app_codes: [] + purge_max_age: "720h" + purge_poll_interval: "5s" + purge_batch_size: 100 + purge_receipt_grace: "1h" + purge_policy_version: "financial-v1" + purge_restore_proofs: [] max_age: "720h" poll_interval: "5m" query_batch_size: 500 diff --git a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql index 705b53cc..212da4cc 100644 --- a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql +++ b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql @@ -203,6 +203,60 @@ CREATE TABLE IF NOT EXISTS wallet_outbox_archive_receipts ( KEY idx_wallet_outbox_archive_cursor (app_code, state, last_updated_at_ms, last_event_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 外部归档读回验证回执'; +-- wallet_outbox 自动清理只新增独立窄表，不 ALTER 热 outbox，也不会触发表重写；member 随未清理归档行增减。 +-- 授权行必须由部署流程在隔离库完成 RESTORE_VERIFIED 并核对归档回执后手工写入；初始化脚本绝不生成清理授权。 +CREATE TABLE IF NOT EXISTS wallet_outbox_archive_members ( + app_code VARCHAR(32) NOT NULL COMMENT '应用编码', + event_id VARCHAR(128) NOT NULL COMMENT '已归档 wallet_outbox 事件 ID', + updated_at_ms BIGINT NOT NULL COMMENT '归档时 delivered 行的精确更新时间，UTC epoch ms', + batch_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '包含该精确事件的归档批次 ID', + manifest_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '包含该事件的 manifest SHA-256', + receipt_verified_at_ms BIGINT NOT NULL COMMENT '归档 receipt 完成 COS 读回验证的时间，UTC epoch ms', + PRIMARY KEY (app_code, event_id), + KEY idx_wallet_outbox_archive_member_purge (app_code, updated_at_ms, event_id), + KEY idx_wallet_outbox_archive_member_batch (app_code, batch_id, event_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 已验证归档精确成员'; + +CREATE TABLE IF NOT EXISTS wallet_outbox_archive_purge_authorizations ( + app_code VARCHAR(32) NOT NULL COMMENT '应用编码', + archive_batch_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '完成恢复演练的归档批次内容指纹', + manifest_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '恢复演练核对通过的 manifest SHA-256', + schema_version INT NOT NULL COMMENT '恢复演练核对通过的归档 schema 版本', + row_count INT NOT NULL COMMENT '恢复演练核对通过的精确行数', + data_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 data COS CRC64', + manifest_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 manifest COS CRC64', + success_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 _SUCCESS COS CRC64', + rehearsal_receipt_id VARCHAR(64) NOT NULL COMMENT '隔离库 RESTORE_VERIFIED 演练回执 ID', + state VARCHAR(32) NOT NULL COMMENT '固定为 RESTORE_VERIFIED，其他状态不能授权清理', + verified_at_ms BIGINT NOT NULL COMMENT '恢复演练验证完成时间，UTC epoch ms', + created_at_ms BIGINT NOT NULL COMMENT '授权证据录入时间，UTC epoch ms', + updated_at_ms BIGINT NOT NULL COMMENT '授权证据更新时间，UTC epoch ms', + PRIMARY KEY (app_code, archive_batch_id, manifest_sha256), + KEY idx_wallet_outbox_purge_auth_verified (app_code, state, verified_at_ms, archive_batch_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 恢复演练清理授权证据'; + +-- 一行只对应一个已提交的精确 PK 删除批次；删除与回执写入同事务，失败或回滚时不会留下虚假清理证据。 +CREATE TABLE IF NOT EXISTS wallet_outbox_archive_purge_receipts ( + app_code VARCHAR(32) NOT NULL COMMENT '应用编码', + purge_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '清理批次稳定指纹', + state VARCHAR(32) NOT NULL COMMENT '固定为 PURGED', + policy_version VARCHAR(32) NOT NULL COMMENT '正向安全事件白名单版本', + cutoff_ms BIGINT NOT NULL COMMENT '本批固定的 delivered 保留截止时间，UTC epoch ms，删除严格小于该值', + receipt_verified_before_ms BIGINT NOT NULL COMMENT '独立归档回执成熟截止时间，只有更早 VERIFIED receipt 可清理', + candidate_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '按游标排序的精确 (app_code,event_id) 候选集合 SHA-256', + candidate_count INT NOT NULL COMMENT '事务内锁定的候选行数', + deleted_count INT NOT NULL COMMENT '精确 PK 删除影响行数，必须与 candidate_count 一致', + first_updated_at_ms BIGINT NOT NULL COMMENT '首条候选 delivered 游标时间', + first_event_id VARCHAR(128) NOT NULL COMMENT '首条候选事件 ID', + last_updated_at_ms BIGINT NOT NULL COMMENT '末条候选 delivered 游标时间', + last_event_id VARCHAR(128) NOT NULL COMMENT '末条候选事件 ID', + purged_at_ms BIGINT NOT NULL COMMENT '清理提交时间，UTC epoch ms', + created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', + updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', + PRIMARY KEY (app_code, purge_id), + KEY idx_wallet_outbox_purge_receipt_time (app_code, purged_at_ms, purge_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 精确 PK 清理审计回执'; + -- 钱包运行侧收益政策实例：由 admin-server 发布写入，wallet 送礼、提现和结算只读本库运行快照。 CREATE TABLE IF NOT EXISTS wallet_policy_instances ( app_code VARCHAR(32) NOT NULL COMMENT '应用编码', diff --git a/services/wallet-service/deploy/mysql/migrations/013_wallet_outbox_archive_purge_receipts.sql b/services/wallet-service/deploy/mysql/migrations/013_wallet_outbox_archive_purge_receipts.sql new file mode 100644 index 00000000..aec8d039 --- /dev/null +++ b/services/wallet-service/deploy/mysql/migrations/013_wallet_outbox_archive_purge_receipts.sql @@ -0,0 +1,53 @@ +-- wallet_outbox 自动清理只新增独立窄表，不 ALTER 热 outbox，也不会触发表重写；member 随未清理归档行增减。 +-- 授权行必须由部署流程在隔离库完成 RESTORE_VERIFIED 并核对归档回执后手工写入；迁移本身绝不生成清理授权。 +CREATE TABLE IF NOT EXISTS wallet_outbox_archive_members ( + app_code VARCHAR(32) NOT NULL COMMENT '应用编码', + event_id VARCHAR(128) NOT NULL COMMENT '已归档 wallet_outbox 事件 ID', + updated_at_ms BIGINT NOT NULL COMMENT '归档时 delivered 行的精确更新时间，UTC epoch ms', + batch_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '包含该精确事件的归档批次 ID', + manifest_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '包含该事件的 manifest SHA-256', + receipt_verified_at_ms BIGINT NOT NULL COMMENT '归档 receipt 完成 COS 读回验证的时间，UTC epoch ms', + PRIMARY KEY (app_code, event_id), + KEY idx_wallet_outbox_archive_member_purge (app_code, updated_at_ms, event_id), + KEY idx_wallet_outbox_archive_member_batch (app_code, batch_id, event_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 已验证归档精确成员'; + +CREATE TABLE IF NOT EXISTS wallet_outbox_archive_purge_authorizations ( + app_code VARCHAR(32) NOT NULL COMMENT '应用编码', + archive_batch_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '完成恢复演练的归档批次内容指纹', + manifest_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '恢复演练核对通过的 manifest SHA-256', + schema_version INT NOT NULL COMMENT '恢复演练核对通过的归档 schema 版本', + row_count INT NOT NULL COMMENT '恢复演练核对通过的精确行数', + data_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 data COS CRC64', + manifest_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 manifest COS CRC64', + success_crc64 VARCHAR(32) NOT NULL COMMENT '恢复演练核对通过的 _SUCCESS COS CRC64', + rehearsal_receipt_id VARCHAR(64) NOT NULL COMMENT '隔离库 RESTORE_VERIFIED 演练回执 ID', + state VARCHAR(32) NOT NULL COMMENT '固定为 RESTORE_VERIFIED，其他状态不能授权清理', + verified_at_ms BIGINT NOT NULL COMMENT '恢复演练验证完成时间，UTC epoch ms', + created_at_ms BIGINT NOT NULL COMMENT '授权证据录入时间，UTC epoch ms', + updated_at_ms BIGINT NOT NULL COMMENT '授权证据更新时间，UTC epoch ms', + PRIMARY KEY (app_code, archive_batch_id, manifest_sha256), + KEY idx_wallet_outbox_purge_auth_verified (app_code, state, verified_at_ms, archive_batch_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 恢复演练清理授权证据'; + +-- 一行只对应一个已提交的精确 PK 删除批次；删除与回执写入同事务，失败或回滚时不会留下虚假清理证据。 +CREATE TABLE IF NOT EXISTS wallet_outbox_archive_purge_receipts ( + app_code VARCHAR(32) NOT NULL COMMENT '应用编码', + purge_id CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '清理批次稳定指纹', + state VARCHAR(32) NOT NULL COMMENT '固定为 PURGED', + policy_version VARCHAR(32) NOT NULL COMMENT '正向安全事件白名单版本', + cutoff_ms BIGINT NOT NULL COMMENT '本批固定的 delivered 保留截止时间，UTC epoch ms，删除严格小于该值', + receipt_verified_before_ms BIGINT NOT NULL COMMENT '独立归档回执成熟截止时间，只有更早 VERIFIED receipt 可清理', + candidate_sha256 CHAR(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '按游标排序的精确 (app_code,event_id) 候选集合 SHA-256', + candidate_count INT NOT NULL COMMENT '事务内锁定的候选行数', + deleted_count INT NOT NULL COMMENT '精确 PK 删除影响行数，必须与 candidate_count 一致', + first_updated_at_ms BIGINT NOT NULL COMMENT '首条候选 delivered 游标时间', + first_event_id VARCHAR(128) NOT NULL COMMENT '首条候选事件 ID', + last_updated_at_ms BIGINT NOT NULL COMMENT '末条候选 delivered 游标时间', + last_event_id VARCHAR(128) NOT NULL COMMENT '末条候选事件 ID', + purged_at_ms BIGINT NOT NULL COMMENT '清理提交时间，UTC epoch ms', + created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', + updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', + PRIMARY KEY (app_code, purge_id), + KEY idx_wallet_outbox_purge_receipt_time (app_code, purged_at_ms, purge_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox 精确 PK 清理审计回执'; diff --git a/services/wallet-service/internal/app/app.go b/services/wallet-service/internal/app/app.go index 84362616..d38c1cfa 100644 --- a/services/wallet-service/internal/app/app.go +++ b/services/wallet-service/internal/app/app.go @@ -73,6 +73,10 @@ func New(cfg config.Config) (*App, error) { if err != nil { return nil, err } + if err := requireWalletOutboxPurgeAuthorizations(startupCtx, repository, cfg.OutboxArchive); err != nil { + _ = repository.Close() + return nil, err + } var outboxArchiveStore *walletOutboxArchiveCOSStore if cfg.OutboxArchive.Enabled && cfg.OutboxArchive.Mode == config.OutboxArchiveModeArchiveOnly { outboxArchiveStore, err = newWalletOutboxArchiveCOSStore(cfg.OutboxArchive.COS, nil) diff --git a/services/wallet-service/internal/app/background_workers.go b/services/wallet-service/internal/app/background_workers.go index 2a90658c..82bdfd1d 100644 --- a/services/wallet-service/internal/app/background_workers.go +++ b/services/wallet-service/internal/app/background_workers.go @@ -20,9 +20,12 @@ func (a *App) runBackgroundWorkers() { a.startWalletOutboxWorkers(ctx, "wallet-realtime-outbox", a.realtimeOutboxWorkerCfg, a.realtimeOutboxPartitions, a.realtimeOutboxProducer, a.realtimeWalletOutboxTopic, a.realtimeOutboxWorkerCfg.EventTypes, nil) } if a.outboxArchiveCfg.Enabled { - // archive-only 与 MQ fanout 独立；dry_run 也持续用有界索引查询暴露容量，但两种模式都不删除热表行。 + // 归档与 MQ fanout 独立；purge 另有恢复凭证、exact membership 和 allowlist 三重门槛。 a.startWalletOutboxArchiveWorker(ctx) } + if a.outboxArchiveCfg.PurgeEnabled { + a.startWalletOutboxPurgeWorker(ctx) + } if a.externalRechargeReconcileWorkerCfg.Enabled { a.startExternalRechargeReconcileWorker(ctx) } diff --git a/services/wallet-service/internal/app/outbox_archive.go b/services/wallet-service/internal/app/outbox_archive.go index fd30ba61..d880faae 100644 --- a/services/wallet-service/internal/app/outbox_archive.go +++ b/services/wallet-service/internal/app/outbox_archive.go @@ -264,6 +264,13 @@ func (a *App) archiveWalletOutboxRecords(tenantCtx context.Context, targetAppCod return err } nowMS := time.Now().UTC().UnixMilli() + members := make([]mysqlstorage.WalletOutboxArchiveMember, 0, len(records)) + for _, record := range records { + // 精确成员证明与 VERIFIED receipt 同事务落库；purge 不再用游标范围猜测某一行是否真的存在于 COS 对象。 + members = append(members, mysqlstorage.WalletOutboxArchiveMember{ + AppCode: record.AppCode, EventID: record.EventID, UpdatedAtMS: record.UpdatedAtMS, + }) + } receiptCtx, receiptCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout) err = a.mysqlRepo.RecordVerifiedWalletOutboxArchive(receiptCtx, mysqlstorage.WalletOutboxArchiveReceipt{ AppCode: targetAppCode, BatchID: artifact.manifest.BatchID, State: walletOutboxArchiveVerifiedState, @@ -275,6 +282,7 @@ func (a *App) archiveWalletOutboxRecords(tenantCtx context.Context, targetAppCod NDJSONSHA256: artifact.manifest.NDJSONSHA256, GzipSHA256: artifact.manifest.GzipSHA256, ManifestSHA256: artifact.manifestSHA256, DataCRC64: dataCRC64, ManifestCRC64: manifestCRC64, SuccessCRC64: successCRC64, UncompressedBytes: artifact.manifest.UncompressedBytes, CompressedBytes: artifact.manifest.CompressedBytes, VerifiedAtMS: nowMS, + Members: members, }) receiptCancel() if err != nil { diff --git a/services/wallet-service/internal/app/outbox_purge.go b/services/wallet-service/internal/app/outbox_purge.go new file mode 100644 index 00000000..7ade144e --- /dev/null +++ b/services/wallet-service/internal/app/outbox_purge.go @@ -0,0 +1,107 @@ +package app + +import ( + "context" + "errors" + "fmt" + "log/slog" + "time" + + "hyapp/pkg/appcode" + "hyapp/pkg/logx" + "hyapp/services/wallet-service/internal/config" + mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql" +) + +func requireWalletOutboxPurgeAuthorizations(ctx context.Context, repository *mysqlstorage.Repository, cfg config.OutboxArchiveConfig) error { + if !cfg.PurgeEnabled { + return nil + } + if repository == nil { + return errors.New("wallet outbox purge repository is not configured") + } + for _, proof := range cfg.PurgeRestoreProofs { + proofCtx := appcode.WithContext(ctx, proof.AppCode) + if err := repository.RequireWalletOutboxPurgeAuthorization(proofCtx, proof.AppCode, proof.BatchID, proof.ManifestSHA256); err != nil { + // App 启动失败比“配置显示 purge=true、实际没有恢复凭证”更安全；滚动发布会停在第一台并保留另一台服务流量。 + return fmt.Errorf("wallet outbox purge authorization rejected for app %s: %w", proof.AppCode, err) + } + } + return nil +} + +func (a *App) startWalletOutboxPurgeWorker(ctx context.Context) { + a.workers.Add(1) + go func() { + defer a.workers.Done() + ticker := time.NewTicker(a.outboxArchiveCfg.PurgePollInterval) + defer ticker.Stop() + for { + if err := a.runWalletOutboxPurgeRound(ctx); err != nil && !errors.Is(err, context.Canceled) { + logx.Error(ctx, "wallet_outbox_purge_round_failed", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() +} + +func (a *App) runWalletOutboxPurgeRound(ctx context.Context) error { + if a.mysqlRepo == nil { + return nil + } + now := time.Now().UTC() + cutoffMS := now.Add(-a.outboxArchiveCfg.PurgeMaxAge).UnixMilli() + receiptVerifiedBeforeMS := now.Add(-a.outboxArchiveCfg.PurgeReceiptGrace).UnixMilli() + for _, targetAppCode := range a.outboxArchiveCfg.PurgeAppCodes { + tenantCtx := appcode.WithContext(ctx, targetAppCode) + lockCtx, lockCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout) + release, acquired, err := a.mysqlRepo.AcquireWalletOutboxArchiveAppLock(lockCtx) + lockCancel() + if err != nil { + logx.Error(ctx, "wallet_outbox_purge_app_lock_failed", err, slog.String("app_code", targetAppCode)) + continue + } + if !acquired { + continue + } + + purgeCtx, purgeCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout) + receipt, purgeErr := a.mysqlRepo.PurgeDeliveredWalletOutboxBatch( + purgeCtx, + targetAppCode, + a.outboxArchiveCfg.PurgePolicyVersion, + cutoffMS, + receiptVerifiedBeforeMS, + a.outboxArchiveCfg.PurgeBatchSize, + now.UnixMilli(), + ) + purgeCancel() + release() + if purgeErr != nil { + // 单 App 失败不能扩大到其他租户；事务会回滚，下一轮仍从同一 exact member 重试。 + logx.Error(ctx, "wallet_outbox_purge_app_failed", purgeErr, slog.String("app_code", targetAppCode)) + continue + } + if receipt.DeletedCount == 0 { + continue + } + logx.Info(ctx, "wallet_outbox_purged", + slog.String("app_code", receipt.AppCode), + slog.String("purge_id", receipt.PurgeID), + slog.String("policy_version", receipt.PolicyVersion), + slog.Int("deleted_count", receipt.DeletedCount), + slog.Int64("cutoff_ms", receipt.CutoffMS), + slog.Int64("receipt_verified_before_ms", receipt.ReceiptVerifiedBeforeMS), + slog.Int64("first_updated_at_ms", receipt.FirstCursor.UpdatedAtMS), + slog.String("first_event_id", receipt.FirstCursor.EventID), + slog.Int64("last_updated_at_ms", receipt.LastCursor.UpdatedAtMS), + slog.String("last_event_id", receipt.LastCursor.EventID), + slog.String("candidate_sha256", receipt.CandidateSHA256), + ) + } + return nil +} diff --git a/services/wallet-service/internal/config/config.go b/services/wallet-service/internal/config/config.go index 16e209e8..963569dd 100644 --- a/services/wallet-service/internal/config/config.go +++ b/services/wallet-service/internal/config/config.go @@ -1,6 +1,7 @@ package config import ( + "encoding/hex" "errors" "os" "path/filepath" @@ -48,7 +49,7 @@ type Config struct { ExternalRechargeReconcileWorker ExternalRechargeReconcileWorkerConfig `yaml:"external_recharge_reconcile_worker"` // OutboxWorker 控制 wallet_outbox 到 MQ 的补偿投递。 OutboxWorker OutboxWorkerConfig `yaml:"outbox_worker"` - // OutboxArchive 只把已投递事实备份到外部存储；当前不提供 purge，避免破坏线上回放与修复读者。 + // OutboxArchive 把已投递事实归档到外部存储；purge 仅对恢复验证且命中正向安全策略的精确成员开放。 OutboxArchive OutboxArchiveConfig `yaml:"outbox_archive"` // RealtimeOutboxWorker 控制红包等实时 UI 事件到独立 MQ topic 的补偿投递。 RealtimeOutboxWorker OutboxWorkerConfig `yaml:"realtime_outbox_worker"` @@ -125,15 +126,28 @@ const ( OutboxArchiveModeArchiveOnly OutboxArchiveMode = "archive_only" ) -// OutboxArchiveConfig 控制 delivered wallet_outbox 的 archive-only 备份链路。 +// OutboxArchiveConfig 控制 delivered wallet_outbox 的归档与受控清理链路。 type OutboxArchiveConfig struct { - Enabled bool `yaml:"enabled"` - Mode OutboxArchiveMode `yaml:"mode"` - PurgeEnabled bool `yaml:"purge_enabled"` - MaxAge time.Duration `yaml:"max_age"` - PollInterval time.Duration `yaml:"poll_interval"` - QueryBatchSize int `yaml:"query_batch_size"` - SegmentMaxRows int `yaml:"segment_max_rows"` + Enabled bool `yaml:"enabled"` + Mode OutboxArchiveMode `yaml:"mode"` + // PurgeEnabled 只有在归档、真实恢复演练凭证和租户 allowlist 同时满足时才允许启动。 + PurgeEnabled bool `yaml:"purge_enabled"` + // PurgeAppCodes 永远是显式 allowlist；清理不能像归档一样自动发现新租户。 + PurgeAppCodes []string `yaml:"purge_app_codes"` + // PurgeMaxAge 与归档 cutoff 分离，防止调快归档任务时意外同步缩短热数据保留期。 + PurgeMaxAge time.Duration `yaml:"purge_max_age"` + PurgePollInterval time.Duration `yaml:"purge_poll_interval"` + PurgeBatchSize int `yaml:"purge_batch_size"` + // PurgeReceiptGrace 强制归档回执先稳定一段时间，避免刚完成上传就立即删除热表事实。 + PurgeReceiptGrace time.Duration `yaml:"purge_receipt_grace"` + // PurgePolicyVersion 对应代码内正向事件类型清单；未知的新事件默认永不删除。 + PurgePolicyVersion string `yaml:"purge_policy_version"` + // PurgeRestoreProofs 把一次真实 RESTORE_VERIFIED 结果绑定到 App 和归档 manifest；生产库还必须有相同授权记录。 + PurgeRestoreProofs []OutboxArchivePurgeRestoreProofConfig `yaml:"purge_restore_proofs"` + MaxAge time.Duration `yaml:"max_age"` + PollInterval time.Duration `yaml:"poll_interval"` + QueryBatchSize int `yaml:"query_batch_size"` + SegmentMaxRows int `yaml:"segment_max_rows"` // SegmentMaxUncompressedBytes caps one in-memory/archive artifact independently of row count because payload sizes vary. SegmentMaxUncompressedBytes int64 `yaml:"segment_max_uncompressed_bytes"` RoundMaxRows int `yaml:"round_max_rows"` @@ -145,6 +159,13 @@ type OutboxArchiveConfig struct { COS OutboxArchiveCOSConfig `yaml:"cos"` } +// OutboxArchivePurgeRestoreProofConfig 是自动删除的双重确认材料，不包含任何 COS 或数据库密钥。 +type OutboxArchivePurgeRestoreProofConfig struct { + AppCode string `yaml:"app_code"` + BatchID string `yaml:"batch_id"` + ManifestSHA256 string `yaml:"manifest_sha256"` +} + // OutboxArchiveCOSConfig 只允许服务端环境注入私有桶凭证；归档对象不生成公网 URL。 type OutboxArchiveCOSConfig struct { CredentialSource string `yaml:"credential_source"` @@ -395,6 +416,11 @@ func Default() Config { Enabled: false, Mode: OutboxArchiveModeDryRun, PurgeEnabled: false, + PurgeMaxAge: 30 * 24 * time.Hour, + PurgePollInterval: 5 * time.Second, + PurgeBatchSize: 100, + PurgeReceiptGrace: time.Hour, + PurgePolicyVersion: "financial-v1", MaxAge: 30 * 24 * time.Hour, PollInterval: 5 * time.Minute, QueryBatchSize: 500, @@ -918,15 +944,10 @@ func normalizeOutboxArchiveConfig(cfg OutboxArchiveConfig, defaults OutboxArchiv if cfg.Mode == "" { cfg.Mode = defaults.Mode } - // wallet_outbox 仍是 GiftCatalogVersion、gift-wall/badge 回填、notice repair 和 statistics replay 的在线事实源。 - // 在这些读者切到热表 + 归档恢复层之前，任何 purge 配置都必须启动失败，不允许静默降级为裸删。 - if cfg.PurgeEnabled { - return OutboxArchiveConfig{}, errors.New("outbox_archive.purge_enabled is unavailable until wallet_outbox online readers use the archive restore path") - } switch cfg.Mode { case OutboxArchiveModeDryRun, OutboxArchiveModeArchiveOnly: default: - return OutboxArchiveConfig{}, errors.New("outbox_archive.mode must be dry_run or archive_only; purge modes are intentionally unsupported") + return OutboxArchiveConfig{}, errors.New("outbox_archive.mode must be dry_run or archive_only") } if cfg.MaxAge <= 0 { cfg.MaxAge = defaults.MaxAge @@ -934,6 +955,76 @@ func normalizeOutboxArchiveConfig(cfg OutboxArchiveConfig, defaults OutboxArchiv if cfg.Enabled && cfg.MaxAge < 24*time.Hour { return OutboxArchiveConfig{}, errors.New("outbox_archive.max_age must be at least 24h") } + cfg.PurgeAppCodes = outboxpartition.Normalize(cfg.PurgeAppCodes, false) + if cfg.PurgeMaxAge <= 0 { + cfg.PurgeMaxAge = defaults.PurgeMaxAge + } + if cfg.PurgePollInterval <= 0 { + cfg.PurgePollInterval = defaults.PurgePollInterval + } + if cfg.PurgeBatchSize <= 0 { + cfg.PurgeBatchSize = defaults.PurgeBatchSize + } + if cfg.PurgeReceiptGrace <= 0 { + cfg.PurgeReceiptGrace = defaults.PurgeReceiptGrace + } + cfg.PurgePolicyVersion = strings.ToLower(strings.TrimSpace(cfg.PurgePolicyVersion)) + if cfg.PurgePolicyVersion == "" { + cfg.PurgePolicyVersion = defaults.PurgePolicyVersion + } + proofsByApp := make(map[string]OutboxArchivePurgeRestoreProofConfig, len(cfg.PurgeRestoreProofs)) + for _, proof := range cfg.PurgeRestoreProofs { + proof.AppCode = strings.ToLower(strings.TrimSpace(proof.AppCode)) + proof.BatchID = strings.ToLower(strings.TrimSpace(proof.BatchID)) + proof.ManifestSHA256 = strings.ToLower(strings.TrimSpace(proof.ManifestSHA256)) + if proof.AppCode == "" || !isSHA256Hex(proof.BatchID) || !isSHA256Hex(proof.ManifestSHA256) { + return OutboxArchiveConfig{}, errors.New("outbox_archive purge restore proof requires app_code and 64-character hex batch/manifest hashes") + } + if _, duplicated := proofsByApp[proof.AppCode]; duplicated { + return OutboxArchiveConfig{}, errors.New("outbox_archive purge restore proofs must contain each app_code once") + } + proofsByApp[proof.AppCode] = proof + } + cfg.PurgeRestoreProofs = cfg.PurgeRestoreProofs[:0] + for _, appCode := range cfg.PurgeAppCodes { + proof, ok := proofsByApp[appCode] + if !ok && cfg.PurgeEnabled { + return OutboxArchiveConfig{}, errors.New("outbox_archive every purge app_code requires an exact restore proof") + } + if ok { + cfg.PurgeRestoreProofs = append(cfg.PurgeRestoreProofs, proof) + delete(proofsByApp, appCode) + } + } + if cfg.PurgeEnabled { + // 删除开关必须 fail closed：只能在持续 COS 归档模式下，对显式租户逐小批运行。 + if !cfg.Enabled || cfg.Mode != OutboxArchiveModeArchiveOnly { + return OutboxArchiveConfig{}, errors.New("outbox_archive purge requires enabled archive_only mode") + } + if len(cfg.PurgeAppCodes) == 0 || len(proofsByApp) != 0 { + return OutboxArchiveConfig{}, errors.New("outbox_archive purge requires a non-empty app allowlist with no unmatched restore proofs") + } + for _, appCode := range cfg.PurgeAppCodes { + if appCode != "lalu" && appCode != "fami" { + return OutboxArchiveConfig{}, errors.New("outbox_archive financial-v1 purge only supports explicitly approved lalu and fami apps") + } + } + if cfg.PurgeMaxAge < 30*24*time.Hour || cfg.PurgeMaxAge < cfg.MaxAge { + return OutboxArchiveConfig{}, errors.New("outbox_archive purge_max_age must be at least 30d and not shorter than archive max_age") + } + if cfg.PurgePollInterval < 5*time.Second { + return OutboxArchiveConfig{}, errors.New("outbox_archive purge_poll_interval must be at least 5s") + } + if cfg.PurgeReceiptGrace < time.Hour { + return OutboxArchiveConfig{}, errors.New("outbox_archive purge_receipt_grace must be at least 1h") + } + if cfg.PurgePolicyVersion != "financial-v1" { + return OutboxArchiveConfig{}, errors.New("outbox_archive purge_policy_version must be financial-v1") + } + if cfg.PurgeBatchSize > 500 { + return OutboxArchiveConfig{}, errors.New("outbox_archive purge_batch_size must not exceed 500") + } + } if cfg.PollInterval <= 0 { cfg.PollInterval = defaults.PollInterval } @@ -1011,6 +1102,14 @@ func normalizeOutboxArchiveConfig(cfg OutboxArchiveConfig, defaults OutboxArchiv return cfg, nil } +func isSHA256Hex(value string) bool { + if len(value) != 64 { + return false + } + decoded, err := hex.DecodeString(value) + return err == nil && len(decoded) == 32 +} + func normalizeStringSlice(values []string) []string { normalized := make([]string, 0, len(values)) seen := make(map[string]struct{}, len(values)) diff --git a/services/wallet-service/internal/outboxarchiverestore/mysql.go b/services/wallet-service/internal/outboxarchiverestore/mysql.go new file mode 100644 index 00000000..46fde6df --- /dev/null +++ b/services/wallet-service/internal/outboxarchiverestore/mysql.go @@ -0,0 +1,440 @@ +package outboxarchiverestore + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "errors" + "fmt" + "reflect" + "strings" + "time" + + _ "github.com/go-sql-driver/mysql" +) + +const ( + RehearsalTable = "wallet_outbox_restore_rehearsals_v1" + RowsTable = "wallet_outbox_restore_rows_v1" + restoreState = "RESTORING" +) + +// RestoreResult is safe to print: it identifies the selected schema and fixed +// table contract but never includes the DSN or credentials used to reach it. +type RestoreResult struct { + State string `json:"state"` + RehearsalID string `json:"rehearsal_id"` + BatchID string `json:"batch_id"` + AppCode string `json:"app_code"` + RowCount int `json:"row_count"` + TargetDatabase string `json:"target_database"` + RehearsalTable string `json:"rehearsal_table"` + RowsTable string `json:"rows_table"` + DataObjectKey string `json:"data_object_key"` + FirstCursor Cursor `json:"first_cursor"` + LastCursor Cursor `json:"last_cursor"` + NDJSONSHA256 string `json:"ndjson_sha256"` + GzipSHA256 string `json:"gzip_sha256"` + DataCRC64 string `json:"data_crc64"` + ManifestSHA256 string `json:"manifest_sha256"` + ManifestCRC64 string `json:"manifest_crc64"` + SuccessSHA256 string `json:"success_sha256"` + SuccessCRC64 string `json:"success_crc64"` + AvailableTotal string `json:"available_delta_total"` + FrozenTotal string `json:"frozen_delta_total"` + VerifiedAtMS int64 `json:"verified_at_ms"` + IdempotentReadback bool `json:"idempotent_readback"` +} + +// Restore persists a fully verified batch into versioned isolation tables. All +// DDL targets new/fixed table names; all DML is scoped by rehearsal PK. It never +// reads, locks, updates, or deletes wallet_outbox. +func Restore(ctx context.Context, dsn, expectedDatabase, rehearsalID string, verification Verification) (RestoreResult, error) { + dsn = strings.TrimSpace(dsn) + expectedDatabase = strings.TrimSpace(expectedDatabase) + rehearsalID = strings.TrimSpace(rehearsalID) + if dsn == "" || expectedDatabase == "" { + return RestoreResult{}, errors.New("restore DSN and exact expected database are required") + } + if !validRehearsalID(rehearsalID) { + return RestoreResult{}, errors.New("rehearsal_id must be 1..64 ASCII letters, digits, dot, underscore, or dash") + } + if verification.State != filesVerifiedState || len(verification.Records) != verification.Manifest.RowCount { + return RestoreResult{}, errors.New("local archive verification is incomplete") + } + + db, err := sql.Open("mysql", dsn) + if err != nil { + return RestoreResult{}, err + } + defer db.Close() + // A rehearsal is intentionally single-connection and serial; this prevents a + // restore tool from consuming the production-sized connection pool by mistake. + db.SetMaxOpenConns(1) + db.SetMaxIdleConns(1) + db.SetConnMaxLifetime(10 * time.Minute) + conn, err := db.Conn(ctx) + if err != nil { + return RestoreResult{}, err + } + defer conn.Close() + + var databaseName string + if err := conn.QueryRowContext(ctx, `SELECT DATABASE()`).Scan(&databaseName); err != nil { + return RestoreResult{}, err + } + if databaseName != expectedDatabase { + return RestoreResult{}, fmt.Errorf("restore database mismatch: connected=%q expected=%q", databaseName, expectedDatabase) + } + if err := ensureIsolationSchema(ctx, conn); err != nil { + return RestoreResult{}, err + } + release, acquired, err := acquireRestoreLock(ctx, conn, rehearsalID) + if err != nil { + return RestoreResult{}, err + } + if !acquired { + return RestoreResult{}, errors.New("another process holds the same restore rehearsal lock") + } + defer release() + + if existing, found, err := readExistingRestore(ctx, conn, rehearsalID, verification); err != nil { + return RestoreResult{}, err + } else if found { + return existing, nil + } + + tx, err := conn.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return RestoreResult{}, err + } + defer tx.Rollback() + nowMS := time.Now().UTC().UnixMilli() + if err := insertRehearsal(ctx, tx, rehearsalID, databaseName, verification, nowMS); err != nil { + return RestoreResult{}, err + } + for offset := 0; offset < len(verification.Records); offset += 100 { + end := min(offset+100, len(verification.Records)) + if err := insertRestoreRows(ctx, tx, rehearsalID, verification.Manifest.BatchID, nowMS, verification.Records[offset:end]); err != nil { + return RestoreResult{}, err + } + } + // These aggregates only scan one bounded rehearsal partition in the isolation + // table. They never aggregate the live outbox and are the database-level proof + // that JSON conversion and INSERT semantics preserved the archive controls. + if err := verifyRestoredControls(ctx, tx, rehearsalID, verification); err != nil { + return RestoreResult{}, err + } + if err := verifyRestoredRowsExact(ctx, tx, rehearsalID, verification.Records); err != nil { + return RestoreResult{}, err + } + result := newRestoreResult(rehearsalID, databaseName, verification, nowMS, false) + update, err := tx.ExecContext(ctx, ` + UPDATE wallet_outbox_restore_rehearsals_v1 + SET state = ?, verified_at_ms = ?, updated_at_ms = ? + WHERE rehearsal_id = ? AND state = ?`, restoreVerifiedState, nowMS, nowMS, rehearsalID, restoreState) + if err != nil { + return RestoreResult{}, err + } + if affected, err := update.RowsAffected(); err != nil || affected != 1 { + return RestoreResult{}, errors.New("restore rehearsal state transition did not affect exactly one row") + } + if err := tx.Commit(); err != nil { + return RestoreResult{}, err + } + return result, nil +} + +func ensureIsolationSchema(ctx context.Context, conn *sql.Conn) error { + // CREATE IF NOT EXISTS takes metadata locks only on the dedicated table names; + // unlike ALTER wallet_outbox it cannot rebuild or block the hot owner table. + statements := []string{` + CREATE TABLE IF NOT EXISTS wallet_outbox_restore_rehearsals_v1 ( + rehearsal_id VARCHAR(64) NOT NULL, + source_app_code VARCHAR(32) NOT NULL, + source_batch_id CHAR(64) NOT NULL, + state VARCHAR(32) NOT NULL, + schema_version INT NOT NULL, + data_object_key VARCHAR(512) NOT NULL, + row_count INT NOT NULL, + first_updated_at_ms BIGINT NOT NULL, + first_event_id VARCHAR(128) NOT NULL, + last_updated_at_ms BIGINT NOT NULL, + last_event_id VARCHAR(128) NOT NULL, + available_delta_total VARCHAR(80) NOT NULL, + frozen_delta_total VARCHAR(80) NOT NULL, + ndjson_sha256 CHAR(64) NOT NULL, + gzip_sha256 CHAR(64) NOT NULL, + data_crc64 VARCHAR(32) NOT NULL, + manifest_sha256 CHAR(64) NOT NULL, + manifest_crc64 VARCHAR(32) NOT NULL, + success_sha256 CHAR(64) NOT NULL, + success_crc64 VARCHAR(32) NOT NULL, + target_database VARCHAR(128) NOT NULL, + verified_at_ms BIGINT NOT NULL, + created_at_ms BIGINT NOT NULL, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (rehearsal_id), + KEY idx_wallet_outbox_restore_batch (source_app_code, source_batch_id, state) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='wallet outbox archive restore rehearsal evidence v1'`, ` + CREATE TABLE IF NOT EXISTS wallet_outbox_restore_rows_v1 ( + rehearsal_id VARCHAR(64) NOT NULL, + source_batch_id CHAR(64) NOT NULL, + app_code VARCHAR(32) NOT NULL, + event_id VARCHAR(128) NOT NULL, + event_type VARCHAR(64) NOT NULL, + transaction_id VARCHAR(96) NOT NULL, + command_id VARCHAR(128) NOT NULL, + user_id BIGINT NOT NULL, + asset_type VARCHAR(32) NOT NULL, + available_delta BIGINT NOT NULL, + frozen_delta BIGINT NOT NULL, + payload JSON NOT NULL, + status VARCHAR(32) NOT NULL, + worker_id VARCHAR(128) NOT NULL, + lock_until_ms BIGINT NULL, + retry_count INT NOT NULL, + next_retry_at_ms BIGINT NULL, + last_error TEXT NULL, + created_at_ms BIGINT NOT NULL, + updated_at_ms BIGINT NOT NULL, + restored_at_ms BIGINT NOT NULL, + PRIMARY KEY (rehearsal_id, app_code, event_id), + KEY idx_wallet_outbox_restore_cursor (rehearsal_id, app_code, updated_at_ms, event_id), + KEY idx_wallet_outbox_restore_batch (source_batch_id, app_code, event_id) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='isolated wallet_outbox-compatible restored rows v1'`} + for _, statement := range statements { + if _, err := conn.ExecContext(ctx, statement); err != nil { + return err + } + } + return nil +} + +func insertRehearsal(ctx context.Context, tx *sql.Tx, rehearsalID, databaseName string, verification Verification, nowMS int64) error { + manifest := verification.Manifest + _, err := tx.ExecContext(ctx, ` + INSERT INTO wallet_outbox_restore_rehearsals_v1 ( + rehearsal_id, source_app_code, source_batch_id, state, schema_version, data_object_key, row_count, + first_updated_at_ms, first_event_id, last_updated_at_ms, last_event_id, + available_delta_total, frozen_delta_total, ndjson_sha256, gzip_sha256, data_crc64, + manifest_sha256, manifest_crc64, success_sha256, success_crc64, target_database, + verified_at_ms, created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?)`, + rehearsalID, manifest.AppCode, manifest.BatchID, restoreState, manifest.SchemaVersion, manifest.DataObjectKey, manifest.RowCount, + manifest.FirstCursor.UpdatedAtMS, manifest.FirstCursor.EventID, manifest.LastCursor.UpdatedAtMS, manifest.LastCursor.EventID, + manifest.AvailableDeltaTotal, manifest.FrozenDeltaTotal, manifest.NDJSONSHA256, manifest.GzipSHA256, manifest.DataCRC64, + verification.ManifestSHA256, verification.ManifestCRC64, verification.SuccessSHA256, verification.SuccessCRC64, databaseName, + nowMS, nowMS) + return err +} + +func insertRestoreRows(ctx context.Context, tx *sql.Tx, rehearsalID, batchID string, restoredAtMS int64, records []Record) error { + if len(records) == 0 || len(records) > 100 { + return errors.New("restore insert batch must contain 1..100 rows") + } + const columns = `(rehearsal_id, source_batch_id, app_code, event_id, event_type, transaction_id, command_id, + user_id, asset_type, available_delta, frozen_delta, payload, status, worker_id, lock_until_ms, + retry_count, next_retry_at_ms, last_error, created_at_ms, updated_at_ms, restored_at_ms)` + placeholders := make([]string, 0, len(records)) + args := make([]any, 0, len(records)*21) + for _, record := range records { + placeholders = append(placeholders, `(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`) + args = append(args, + rehearsalID, batchID, record.AppCode, record.EventID, record.EventType, record.TransactionID, record.CommandID, + record.UserID, record.AssetType, record.AvailableDelta, record.FrozenDelta, record.PayloadJSON, record.Status, record.WorkerID, + nullInt64(record.LockUntilMS), record.RetryCount, nullInt64(record.NextRetryAtMS), nullString(record.LastError), + record.CreatedAtMS, record.UpdatedAtMS, restoredAtMS, + ) + } + _, err := tx.ExecContext(ctx, `INSERT INTO wallet_outbox_restore_rows_v1 `+columns+` VALUES `+strings.Join(placeholders, ","), args...) + return err +} + +func verifyRestoredControls(ctx context.Context, query interface { + QueryRowContext(context.Context, string, ...any) *sql.Row +}, rehearsalID string, verification Verification) error { + var count int + var availableTotal, frozenTotal string + var minCreated, maxCreated, minUpdated, maxUpdated int64 + if err := query.QueryRowContext(ctx, ` + SELECT COUNT(*), + CAST(COALESCE(SUM(available_delta), 0) AS CHAR), + CAST(COALESCE(SUM(frozen_delta), 0) AS CHAR), + COALESCE(MIN(created_at_ms), 0), COALESCE(MAX(created_at_ms), 0), + COALESCE(MIN(updated_at_ms), 0), COALESCE(MAX(updated_at_ms), 0) + FROM wallet_outbox_restore_rows_v1 FORCE INDEX (PRIMARY) + WHERE rehearsal_id = ?`, rehearsalID).Scan(&count, &availableTotal, &frozenTotal, &minCreated, &maxCreated, &minUpdated, &maxUpdated); err != nil { + return err + } + manifest := verification.Manifest + if count != manifest.RowCount || availableTotal != manifest.AvailableDeltaTotal || frozenTotal != manifest.FrozenDeltaTotal || minCreated != manifest.MinCreatedAtMS || maxCreated != manifest.MaxCreatedAtMS || minUpdated != manifest.MinUpdatedAtMS || maxUpdated != manifest.MaxUpdatedAtMS { + return errors.New("isolation-table row count, time bounds, or control totals do not match manifest") + } + var firstUpdated, lastUpdated int64 + var firstEvent, lastEvent string + if err := query.QueryRowContext(ctx, ` + SELECT updated_at_ms, event_id FROM wallet_outbox_restore_rows_v1 FORCE INDEX (idx_wallet_outbox_restore_cursor) + WHERE rehearsal_id = ? AND app_code = ? ORDER BY updated_at_ms ASC, event_id ASC LIMIT 1`, rehearsalID, manifest.AppCode).Scan(&firstUpdated, &firstEvent); err != nil { + return err + } + if err := query.QueryRowContext(ctx, ` + SELECT updated_at_ms, event_id FROM wallet_outbox_restore_rows_v1 FORCE INDEX (idx_wallet_outbox_restore_cursor) + WHERE rehearsal_id = ? AND app_code = ? ORDER BY updated_at_ms DESC, event_id DESC LIMIT 1`, rehearsalID, manifest.AppCode).Scan(&lastUpdated, &lastEvent); err != nil { + return err + } + if (Cursor{UpdatedAtMS: firstUpdated, EventID: firstEvent}) != manifest.FirstCursor || (Cursor{UpdatedAtMS: lastUpdated, EventID: lastEvent}) != manifest.LastCursor { + return errors.New("isolation-table cursor bounds do not match manifest") + } + return nil +} + +func readExistingRestore(ctx context.Context, conn *sql.Conn, rehearsalID string, verification Verification) (RestoreResult, bool, error) { + var state, appCode, batchID, manifestSHA, manifestCRC, successCRC, availableTotal, frozenTotal string + var rowCount int + var verifiedAtMS int64 + err := conn.QueryRowContext(ctx, ` + SELECT state, source_app_code, source_batch_id, row_count, manifest_sha256, manifest_crc64, success_crc64, + available_delta_total, frozen_delta_total, verified_at_ms + FROM wallet_outbox_restore_rehearsals_v1 WHERE rehearsal_id = ?`, rehearsalID). + Scan(&state, &appCode, &batchID, &rowCount, &manifestSHA, &manifestCRC, &successCRC, &availableTotal, &frozenTotal, &verifiedAtMS) + if err == sql.ErrNoRows { + return RestoreResult{}, false, nil + } + if err != nil { + return RestoreResult{}, false, err + } + manifest := verification.Manifest + if state != restoreVerifiedState || appCode != manifest.AppCode || batchID != manifest.BatchID || rowCount != manifest.RowCount || manifestSHA != verification.ManifestSHA256 || manifestCRC != verification.ManifestCRC64 || successCRC != verification.SuccessCRC64 || availableTotal != manifest.AvailableDeltaTotal || frozenTotal != manifest.FrozenDeltaTotal { + return RestoreResult{}, false, errors.New("rehearsal_id already exists with non-matching or incomplete evidence") + } + if err := verifyRestoredControls(ctx, conn, rehearsalID, verification); err != nil { + return RestoreResult{}, false, fmt.Errorf("existing restore readback failed: %w", err) + } + if err := verifyRestoredRowsExact(ctx, conn, rehearsalID, verification.Records); err != nil { + return RestoreResult{}, false, fmt.Errorf("existing restore exact-row readback failed: %w", err) + } + result := newRestoreResult(rehearsalID, "", verification, verifiedAtMS, true) + if err := conn.QueryRowContext(ctx, `SELECT DATABASE()`).Scan(&result.TargetDatabase); err != nil { + return RestoreResult{}, false, err + } + return result, true, nil +} + +func newRestoreResult(rehearsalID, databaseName string, verification Verification, verifiedAtMS int64, idempotent bool) RestoreResult { + manifest := verification.Manifest + return RestoreResult{ + State: restoreVerifiedState, RehearsalID: rehearsalID, BatchID: manifest.BatchID, AppCode: manifest.AppCode, + RowCount: manifest.RowCount, TargetDatabase: databaseName, RehearsalTable: RehearsalTable, RowsTable: RowsTable, + DataObjectKey: manifest.DataObjectKey, FirstCursor: manifest.FirstCursor, LastCursor: manifest.LastCursor, + NDJSONSHA256: manifest.NDJSONSHA256, GzipSHA256: manifest.GzipSHA256, DataCRC64: manifest.DataCRC64, + ManifestSHA256: verification.ManifestSHA256, ManifestCRC64: verification.ManifestCRC64, + SuccessSHA256: verification.SuccessSHA256, SuccessCRC64: verification.SuccessCRC64, + AvailableTotal: manifest.AvailableDeltaTotal, FrozenTotal: manifest.FrozenDeltaTotal, + VerifiedAtMS: verifiedAtMS, IdempotentReadback: idempotent, + } +} + +func verifyRestoredRowsExact(ctx context.Context, query interface { + QueryContext(context.Context, string, ...any) (*sql.Rows, error) +}, rehearsalID string, expected []Record) error { + rows, err := query.QueryContext(ctx, ` + SELECT app_code, event_id, event_type, transaction_id, command_id, user_id, asset_type, + available_delta, frozen_delta, CAST(payload AS CHAR), status, worker_id, + lock_until_ms, retry_count, next_retry_at_ms, last_error, created_at_ms, updated_at_ms + FROM wallet_outbox_restore_rows_v1 FORCE INDEX (idx_wallet_outbox_restore_cursor) + WHERE rehearsal_id = ? + ORDER BY app_code ASC, updated_at_ms ASC, event_id ASC`, rehearsalID) + if err != nil { + return err + } + defer rows.Close() + index := 0 + for rows.Next() { + if index >= len(expected) { + return errors.New("isolation table contains more rows than verified archive") + } + var actual Record + var lockUntilMS, nextRetryAtMS sql.NullInt64 + var lastError sql.NullString + if err := rows.Scan( + &actual.AppCode, &actual.EventID, &actual.EventType, &actual.TransactionID, &actual.CommandID, + &actual.UserID, &actual.AssetType, &actual.AvailableDelta, &actual.FrozenDelta, &actual.PayloadJSON, + &actual.Status, &actual.WorkerID, &lockUntilMS, &actual.RetryCount, &nextRetryAtMS, &lastError, + &actual.CreatedAtMS, &actual.UpdatedAtMS, + ); err != nil { + return err + } + if lockUntilMS.Valid { + value := lockUntilMS.Int64 + actual.LockUntilMS = &value + } + if nextRetryAtMS.Valid { + value := nextRetryAtMS.Int64 + actual.NextRetryAtMS = &value + } + if lastError.Valid { + value := lastError.String + actual.LastError = &value + } + // Source payload was archived from CAST(payload AS CHAR); casting the restored + // JSON back through MySQL proves JSON parsing did not alter the owner snapshot. + if !reflect.DeepEqual(actual, expected[index]) { + return fmt.Errorf("isolation table row %d does not exactly match archive", index+1) + } + index++ + } + if err := rows.Err(); err != nil { + return err + } + if index != len(expected) { + return errors.New("isolation table contains fewer rows than verified archive") + } + return nil +} + +func acquireRestoreLock(ctx context.Context, conn *sql.Conn, rehearsalID string) (func(), bool, error) { + hash := sha256.Sum256([]byte(rehearsalID)) + lockName := "wallet_outbox_restore:" + hex.EncodeToString(hash[:16]) + var acquired sql.NullInt64 + if err := conn.QueryRowContext(ctx, `SELECT GET_LOCK(?, 0)`, lockName).Scan(&acquired); err != nil { + return nil, false, err + } + if !acquired.Valid || acquired.Int64 != 1 { + return func() {}, false, nil + } + return func() { + releaseCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + var released sql.NullInt64 + _ = conn.QueryRowContext(releaseCtx, `SELECT RELEASE_LOCK(?)`, lockName).Scan(&released) + }, true, nil +} + +func validRehearsalID(value string) bool { + if len(value) == 0 || len(value) > 64 { + return false + } + for _, char := range value { + if (char >= 'a' && char <= 'z') || (char >= 'A' && char <= 'Z') || (char >= '0' && char <= '9') || char == '.' || char == '_' || char == '-' { + continue + } + return false + } + return true +} + +func nullInt64(value *int64) any { + if value == nil { + return nil + } + return *value +} + +func nullString(value *string) any { + if value == nil { + return nil + } + return *value +} diff --git a/services/wallet-service/internal/outboxarchiverestore/verify.go b/services/wallet-service/internal/outboxarchiverestore/verify.go new file mode 100644 index 00000000..5f73446e --- /dev/null +++ b/services/wallet-service/internal/outboxarchiverestore/verify.go @@ -0,0 +1,431 @@ +// Package outboxarchiverestore verifies one immutable wallet_outbox archive batch +// and restores it into versioned isolation tables. It deliberately contains no +// purge API: a successful rehearsal is evidence for a later retention gate, not +// authorization to mutate the owner outbox. +package outboxarchiverestore + +import ( + "bufio" + "bytes" + "compress/gzip" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "hash/crc64" + "io" + "math/big" + "os" + "path" + "strconv" + "strings" + "time" +) + +const ( + archiveSchema = "hyapp.wallet_outbox.archive" + archiveSchemaVersion = 1 + manifestState = "DATA_VERIFIED" + successState = "VERIFIED" + filesVerifiedState = "FILES_VERIFIED" + restoreVerifiedState = "RESTORE_VERIFIED" + + maxManifestBytes = 1 << 20 + maxSuccessBytes = 64 << 10 + maxNDJSONLine = 64 << 20 +) + +// Limits bounds local CPU, memory and disk exposure before the operator-selected +// restore database is opened. Values are intentionally independent of worker +// configuration so an accidentally supplied object cannot bypass the CLI guard. +type Limits struct { + MaxRows int + MaxCompressedBytes int64 + MaxUncompressedBytes int64 +} + +func DefaultLimits() Limits { + return Limits{MaxRows: 100000, MaxCompressedBytes: 512 << 20, MaxUncompressedBytes: 1 << 30} +} + +type Cursor struct { + UpdatedAtMS int64 `json:"updated_at_ms"` + EventID string `json:"event_id"` +} + +type Manifest struct { + Schema string `json:"schema"` + SchemaVersion int `json:"schema_version"` + State string `json:"state"` + AppCode string `json:"app_code"` + BatchID string `json:"batch_id"` + RowCount int `json:"row_count"` + FirstCursor Cursor `json:"first_cursor"` + LastCursor Cursor `json:"last_cursor"` + MinCreatedAtMS int64 `json:"min_created_at_ms"` + MaxCreatedAtMS int64 `json:"max_created_at_ms"` + MinUpdatedAtMS int64 `json:"min_updated_at_ms"` + MaxUpdatedAtMS int64 `json:"max_updated_at_ms"` + AvailableDeltaTotal string `json:"available_delta_total"` + FrozenDeltaTotal string `json:"frozen_delta_total"` + NDJSONSHA256 string `json:"ndjson_sha256"` + GzipSHA256 string `json:"gzip_sha256"` + DataCRC64 string `json:"data_crc64"` + UncompressedBytes int64 `json:"uncompressed_bytes"` + CompressedBytes int64 `json:"compressed_bytes"` + DataObjectKey string `json:"data_object_key"` +} + +type Success struct { + Schema string `json:"schema"` + SchemaVersion int `json:"schema_version"` + State string `json:"state"` + BatchID string `json:"batch_id"` + ManifestSHA256 string `json:"manifest_sha256"` + ManifestCRC64 string `json:"manifest_crc64"` +} + +// Record mirrors every wallet_outbox column. Restore tables add rehearsal and +// source-batch keys around this payload rather than changing owner-table fields, +// so later historical tools can compare one archived row without translation. +type Record struct { + AppCode string `json:"app_code"` + EventID string `json:"event_id"` + EventType string `json:"event_type"` + TransactionID string `json:"transaction_id"` + CommandID string `json:"command_id"` + UserID int64 `json:"user_id"` + AssetType string `json:"asset_type"` + AvailableDelta int64 `json:"available_delta"` + FrozenDelta int64 `json:"frozen_delta"` + PayloadJSON string `json:"payload_json"` + Status string `json:"status"` + WorkerID string `json:"worker_id"` + LockUntilMS *int64 `json:"lock_until_ms"` + RetryCount int `json:"retry_count"` + NextRetryAtMS *int64 `json:"next_retry_at_ms"` + LastError *string `json:"last_error"` + CreatedAtMS int64 `json:"created_at_ms"` + UpdatedAtMS int64 `json:"updated_at_ms"` +} + +// Verification is the complete local evidence set. SuccessCRC64 is computed +// locally because _SUCCESS cannot contain its own checksum; a future purge gate +// must compare all three CRC values with wallet_outbox_archive_receipts. +type Verification struct { + State string `json:"state"` + Manifest Manifest `json:"manifest"` + Records []Record `json:"-"` + ManifestSHA256 string `json:"manifest_sha256"` + ManifestCRC64 string `json:"manifest_crc64"` + SuccessSHA256 string `json:"success_sha256"` + SuccessCRC64 string `json:"success_crc64"` +} + +// VerifyLocalFiles establishes the _SUCCESS gate before any restore DSN is +// opened. The order is deliberate: an incomplete batch may have data/manifest +// objects, but without a valid marker it can never reach a database write path. +func VerifyLocalFiles(dataPath, manifestPath, successPath string, limits Limits) (Verification, error) { + limits = normalizeLimits(limits) + successBody, err := readBoundedFile(successPath, maxSuccessBytes) + if err != nil { + return Verification{}, fmt.Errorf("read _SUCCESS: %w", err) + } + var success Success + if err := decodeStrictJSON(successBody, &success); err != nil { + return Verification{}, fmt.Errorf("decode _SUCCESS: %w", err) + } + if success.Schema != archiveSchema || success.SchemaVersion != archiveSchemaVersion || success.State != successState || !validSHA256(success.ManifestSHA256) || !validCRC64(success.ManifestCRC64) || !validSHA256(success.BatchID) { + return Verification{}, errors.New("_SUCCESS schema, state, batch_id, or manifest checksum is invalid") + } + + manifestBody, err := readBoundedFile(manifestPath, maxManifestBytes) + if err != nil { + return Verification{}, fmt.Errorf("read manifest: %w", err) + } + manifestSHA256, manifestCRC64 := checksumBytes(manifestBody) + if manifestSHA256 != success.ManifestSHA256 || manifestCRC64 != success.ManifestCRC64 { + return Verification{}, errors.New("manifest does not match _SUCCESS SHA256/CRC64") + } + var manifest Manifest + if err := decodeStrictJSON(manifestBody, &manifest); err != nil { + return Verification{}, fmt.Errorf("decode manifest: %w", err) + } + if err := validateManifest(manifest, success, limits); err != nil { + return Verification{}, err + } + + records, err := verifyDataFile(dataPath, manifest, limits) + if err != nil { + return Verification{}, err + } + successSHA256, successCRC64 := checksumBytes(successBody) + return Verification{ + State: filesVerifiedState, Manifest: manifest, Records: records, + ManifestSHA256: manifestSHA256, ManifestCRC64: manifestCRC64, + SuccessSHA256: successSHA256, SuccessCRC64: successCRC64, + }, nil +} + +func normalizeLimits(limits Limits) Limits { + defaults := DefaultLimits() + if limits.MaxRows <= 0 { + limits.MaxRows = defaults.MaxRows + } + if limits.MaxCompressedBytes <= 0 { + limits.MaxCompressedBytes = defaults.MaxCompressedBytes + } + if limits.MaxUncompressedBytes <= 0 { + limits.MaxUncompressedBytes = defaults.MaxUncompressedBytes + } + return limits +} + +func validateManifest(manifest Manifest, success Success, limits Limits) error { + if manifest.Schema != archiveSchema || manifest.SchemaVersion != archiveSchemaVersion || manifest.State != manifestState { + return errors.New("manifest schema, version, or state is invalid") + } + if manifest.BatchID != success.BatchID || !validSHA256(manifest.BatchID) || strings.TrimSpace(manifest.AppCode) == "" || strings.TrimSpace(manifest.DataObjectKey) == "" { + return errors.New("manifest identity does not match _SUCCESS") + } + if manifest.RowCount <= 0 || manifest.RowCount > limits.MaxRows { + return fmt.Errorf("manifest row_count must be within [1,%d]", limits.MaxRows) + } + if manifest.CompressedBytes <= 0 || manifest.CompressedBytes > limits.MaxCompressedBytes || manifest.UncompressedBytes <= 0 || manifest.UncompressedBytes > limits.MaxUncompressedBytes { + return errors.New("manifest byte counts exceed restore safety limits") + } + if !validSHA256(manifest.NDJSONSHA256) || !validSHA256(manifest.GzipSHA256) || !validCRC64(manifest.DataCRC64) { + return errors.New("manifest data checksums are invalid") + } + if _, ok := new(big.Int).SetString(manifest.AvailableDeltaTotal, 10); !ok { + return errors.New("manifest available_delta_total is invalid") + } + if _, ok := new(big.Int).SetString(manifest.FrozenDeltaTotal, 10); !ok { + return errors.New("manifest frozen_delta_total is invalid") + } + if manifest.FirstCursor.UpdatedAtMS <= 0 || manifest.LastCursor.UpdatedAtMS <= 0 || strings.TrimSpace(manifest.FirstCursor.EventID) == "" || strings.TrimSpace(manifest.LastCursor.EventID) == "" { + return errors.New("manifest cursor is incomplete") + } + deliveredDay := time.UnixMilli(manifest.FirstCursor.UpdatedAtMS).UTC().Format("2006-01-02") + if time.UnixMilli(manifest.LastCursor.UpdatedAtMS).UTC().Format("2006-01-02") != deliveredDay { + return errors.New("manifest crosses UTC delivered day") + } + expectedSuffix := path.Join("app_code="+manifest.AppCode, "delivered_day="+deliveredDay, "batch_id="+manifest.BatchID, "data.ndjson.gz") + cleanObjectKey := path.Clean(manifest.DataObjectKey) + if cleanObjectKey != expectedSuffix && !strings.HasSuffix(cleanObjectKey, "/"+expectedSuffix) { + return errors.New("manifest data_object_key does not match app/day/batch identity") + } + return nil +} + +func verifyDataFile(path string, manifest Manifest, limits Limits) ([]Record, error) { + file, err := os.Open(strings.TrimSpace(path)) + if err != nil { + return nil, fmt.Errorf("open data object: %w", err) + } + defer file.Close() + info, err := file.Stat() + if err != nil { + return nil, err + } + if info.Size() != manifest.CompressedBytes || info.Size() > limits.MaxCompressedBytes { + return nil, errors.New("compressed data size does not match manifest") + } + + gzipHash := sha256.New() + gzipCRC := crc64.New(crc64.MakeTable(crc64.ECMA)) + compressedReader := io.TeeReader(file, io.MultiWriter(gzipHash, gzipCRC)) + gzipReader, err := gzip.NewReader(compressedReader) + if err != nil { + return nil, fmt.Errorf("open archived gzip: %w", err) + } + records, ndjsonSHA256, uncompressedBytes, totals, err := readAndValidateNDJSON(gzipReader, manifest, limits) + closeErr := gzipReader.Close() + _, drainErr := io.Copy(io.Discard, compressedReader) + if err != nil { + return nil, err + } + if closeErr != nil { + return nil, fmt.Errorf("close archived gzip: %w", closeErr) + } + if drainErr != nil { + return nil, fmt.Errorf("read archived gzip tail: %w", drainErr) + } + if hex.EncodeToString(gzipHash.Sum(nil)) != manifest.GzipSHA256 || strconv.FormatUint(gzipCRC.Sum64(), 10) != manifest.DataCRC64 { + return nil, errors.New("data object SHA256/CRC64 does not match manifest") + } + if ndjsonSHA256 != manifest.NDJSONSHA256 || uncompressedBytes != manifest.UncompressedBytes { + return nil, errors.New("NDJSON SHA256/byte count does not match manifest") + } + if totals.available.String() != manifest.AvailableDeltaTotal || totals.frozen.String() != manifest.FrozenDeltaTotal { + return nil, errors.New("NDJSON control totals do not match manifest") + } + return records, nil +} + +type controlTotals struct { + available *big.Int + frozen *big.Int +} + +func readAndValidateNDJSON(reader io.Reader, manifest Manifest, limits Limits) ([]Record, string, int64, controlTotals, error) { + buffered := bufio.NewReaderSize(reader, 64<<10) + hash := sha256.New() + records := make([]Record, 0, manifest.RowCount) + seen := make(map[string]struct{}, manifest.RowCount) + totals := controlTotals{available: new(big.Int), frozen: new(big.Int)} + var byteCount int64 + var minCreated, maxCreated int64 + for { + line, err := readRawLine(buffered, maxNDJSONLine) + if err == io.EOF { + break + } + if err != nil { + return nil, "", 0, totals, err + } + byteCount += int64(len(line)) + if byteCount > limits.MaxUncompressedBytes { + return nil, "", 0, totals, errors.New("NDJSON exceeds uncompressed byte limit") + } + _, _ = hash.Write(line) + var record Record + if err := decodeStrictJSON(bytes.TrimSuffix(line, []byte{'\n'}), &record); err != nil { + return nil, "", 0, totals, fmt.Errorf("decode NDJSON row %d: %w", len(records)+1, err) + } + if err := validateRecord(record, manifest, records, seen); err != nil { + return nil, "", 0, totals, fmt.Errorf("validate NDJSON row %d: %w", len(records)+1, err) + } + if len(records) == 0 { + minCreated, maxCreated = record.CreatedAtMS, record.CreatedAtMS + } else { + if record.CreatedAtMS < minCreated { + minCreated = record.CreatedAtMS + } + if record.CreatedAtMS > maxCreated { + maxCreated = record.CreatedAtMS + } + } + totals.available.Add(totals.available, big.NewInt(record.AvailableDelta)) + totals.frozen.Add(totals.frozen, big.NewInt(record.FrozenDelta)) + records = append(records, record) + if len(records) > limits.MaxRows || len(records) > manifest.RowCount { + return nil, "", 0, totals, errors.New("NDJSON row count exceeds manifest or safety limit") + } + } + if len(records) != manifest.RowCount { + return nil, "", 0, totals, errors.New("NDJSON row count does not match manifest") + } + first, last := records[0], records[len(records)-1] + if manifest.FirstCursor != (Cursor{UpdatedAtMS: first.UpdatedAtMS, EventID: first.EventID}) || manifest.LastCursor != (Cursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID}) || manifest.MinUpdatedAtMS != first.UpdatedAtMS || manifest.MaxUpdatedAtMS != last.UpdatedAtMS { + return nil, "", 0, totals, errors.New("NDJSON cursor bounds do not match manifest") + } + if manifest.MinCreatedAtMS != minCreated || manifest.MaxCreatedAtMS != maxCreated { + return nil, "", 0, totals, errors.New("NDJSON created_at bounds do not match manifest") + } + ndjsonSHA256 := hex.EncodeToString(hash.Sum(nil)) + batchHash := sha256.Sum256([]byte(strings.Join([]string{ + manifest.AppCode, strconv.Itoa(len(records)), strconv.FormatInt(first.UpdatedAtMS, 10), first.EventID, + strconv.FormatInt(last.UpdatedAtMS, 10), last.EventID, ndjsonSHA256, + }, "\n"))) + if hex.EncodeToString(batchHash[:]) != manifest.BatchID { + return nil, "", 0, totals, errors.New("recomputed batch_id does not match manifest") + } + return records, ndjsonSHA256, byteCount, totals, nil +} + +func validateRecord(record Record, manifest Manifest, prior []Record, seen map[string]struct{}) error { + if record.AppCode != manifest.AppCode || record.Status != "delivered" || strings.TrimSpace(record.EventID) == "" || strings.TrimSpace(record.EventType) == "" || record.UpdatedAtMS <= 0 || record.CreatedAtMS <= 0 { + return errors.New("row identity, state, or timestamp is invalid") + } + if !json.Valid([]byte(record.PayloadJSON)) { + return errors.New("payload_json is not valid JSON") + } + if _, exists := seen[record.EventID]; exists { + return errors.New("duplicate (app_code,event_id) archive PK") + } + seen[record.EventID] = struct{}{} + if len(prior) > 0 { + previous := prior[len(prior)-1] + if record.UpdatedAtMS < previous.UpdatedAtMS || (record.UpdatedAtMS == previous.UpdatedAtMS && record.EventID <= previous.EventID) { + return errors.New("rows are not strictly ordered by (updated_at_ms,event_id)") + } + } + return nil +} + +func readRawLine(reader *bufio.Reader, maxBytes int) ([]byte, error) { + line := make([]byte, 0, 4096) + for { + fragment, err := reader.ReadSlice('\n') + if len(line)+len(fragment) > maxBytes { + return nil, fmt.Errorf("NDJSON line exceeds %d bytes", maxBytes) + } + line = append(line, fragment...) + switch err { + case nil: + return line, nil + case bufio.ErrBufferFull: + continue + case io.EOF: + if len(line) == 0 { + return nil, io.EOF + } + return nil, errors.New("NDJSON final row is missing newline terminator") + default: + return nil, err + } + } +} + +func readBoundedFile(path string, maxBytes int64) ([]byte, error) { + file, err := os.Open(strings.TrimSpace(path)) + if err != nil { + return nil, err + } + defer file.Close() + body, err := io.ReadAll(io.LimitReader(file, maxBytes+1)) + if err != nil { + return nil, err + } + if int64(len(body)) > maxBytes || len(body) == 0 { + return nil, errors.New("file is empty or exceeds safety limit") + } + return body, nil +} + +func decodeStrictJSON(body []byte, destination any) error { + decoder := json.NewDecoder(bytes.NewReader(body)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(destination); err != nil { + return err + } + var trailing any + if err := decoder.Decode(&trailing); err != io.EOF { + if err == nil { + return errors.New("JSON contains more than one value") + } + return err + } + return nil +} + +func checksumBytes(body []byte) (string, string) { + sha := sha256.Sum256(body) + crc := crc64.Update(0, crc64.MakeTable(crc64.ECMA), body) + return hex.EncodeToString(sha[:]), strconv.FormatUint(crc, 10) +} + +func validSHA256(value string) bool { + if len(value) != sha256.Size*2 { + return false + } + _, err := hex.DecodeString(value) + return err == nil +} + +func validCRC64(value string) bool { + _, err := strconv.ParseUint(strings.TrimSpace(value), 10, 64) + return err == nil +} diff --git a/services/wallet-service/internal/storage/mysql/outbox_archive.go b/services/wallet-service/internal/storage/mysql/outbox_archive.go index f7042a61..bef6e515 100644 --- a/services/wallet-service/internal/storage/mysql/outbox_archive.go +++ b/services/wallet-service/internal/storage/mysql/outbox_archive.go @@ -3,6 +3,7 @@ package mysql import ( "context" "database/sql" + "fmt" "strings" "time" @@ -41,6 +42,14 @@ type WalletOutboxArchiveRecord struct { UpdatedAtMS int64 `json:"updated_at_ms"` } +// WalletOutboxArchiveMember 是一个已经进入同批 COS data/manifest/_SUCCESS 并完成读回验证的精确事件键。 +// 它只保存后续清理所需的窄字段，不复制 payload；旧 receipt 没有 member 时天然不具备清理资格。 +type WalletOutboxArchiveMember struct { + AppCode string + EventID string + UpdatedAtMS int64 +} + // WalletOutboxArchiveReceipt 只在 COS data/manifest/_SUCCESS 都读回校验后写入。 // VERIFIED 只代表对象完整性，不代表已完成真实 MySQL 恢复演练。 type WalletOutboxArchiveReceipt struct { @@ -67,6 +76,7 @@ type WalletOutboxArchiveReceipt struct { UncompressedBytes int64 CompressedBytes int64 VerifiedAtMS int64 + Members []WalletOutboxArchiveMember } // AcquireWalletOutboxArchiveAppLock 用 MySQL connection-scoped advisory lock 保证多副本下每个 App 只有一个归档者。 @@ -186,13 +196,23 @@ func (r *Repository) RecordVerifiedWalletOutboxArchive(ctx context.Context, rece if r == nil || r.db == nil { return xerr.New(xerr.Unavailable, "mysql repository is not configured") } + receipt.AppCode = appcode.Normalize(receipt.AppCode) if receipt.State != walletOutboxArchiveStateVerified || receipt.BatchID == "" || receipt.RowCount <= 0 || receipt.DataObjectKey == "" || receipt.ManifestObjectKey == "" || receipt.SuccessObjectKey == "" || receipt.NDJSONSHA256 == "" || receipt.GzipSHA256 == "" || receipt.ManifestSHA256 == "" || - receipt.DataCRC64 == "" || receipt.ManifestCRC64 == "" || receipt.SuccessCRC64 == "" { + receipt.DataCRC64 == "" || receipt.ManifestCRC64 == "" || receipt.SuccessCRC64 == "" || + receipt.VerifiedAtMS <= 0 || len(receipt.Members) != receipt.RowCount { return xerr.New(xerr.InvalidArgument, "verified archive receipt is incomplete") } - _, err := r.db.ExecContext(ctx, ` + if err := validateWalletOutboxArchiveMembers(receipt); err != nil { + return err + } + tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + _, err = tx.ExecContext(ctx, ` INSERT INTO wallet_outbox_archive_receipts ( app_code, batch_id, state, data_object_key, manifest_object_key, success_object_key, schema_version, row_count, first_updated_at_ms, first_event_id, last_updated_at_ms, last_event_id, @@ -211,5 +231,111 @@ func (r *Repository) RecordVerifiedWalletOutboxArchive(ctx context.Context, rece receipt.ManifestSHA256, receipt.UncompressedBytes, receipt.CompressedBytes, receipt.VerifiedAtMS, receipt.VerifiedAtMS, receipt.VerifiedAtMS, ) - return err + if err != nil { + return err + } + if err := verifyWalletOutboxArchiveReceiptControls(ctx, tx, receipt); err != nil { + return err + } + // receipt 与 exact members 必须原子提交：只看到 receipt、看不到成员的中间态会让 + // watermark 推断重新混入清理链路；只看到成员、receipt 未 VERIFIED 则无法证明 COS 对象完整。 + for offset := 0; offset < len(receipt.Members); offset += 500 { + end := min(offset+500, len(receipt.Members)) + if err := upsertAndVerifyWalletOutboxArchiveMembers(ctx, tx, receipt, receipt.Members[offset:end]); err != nil { + return err + } + } + return tx.Commit() +} + +func verifyWalletOutboxArchiveReceiptControls(ctx context.Context, tx *sql.Tx, receipt WalletOutboxArchiveReceipt) error { + var state, manifestSHA, dataCRC64, manifestCRC64, successCRC64 string + var schemaVersion, rowCount int + var verifiedAtMS int64 + if err := tx.QueryRowContext(ctx, ` + SELECT state, schema_version, row_count, manifest_sha256, + data_crc64, manifest_crc64, success_crc64, verified_at_ms + FROM wallet_outbox_archive_receipts FORCE INDEX (PRIMARY) + WHERE app_code = ? AND batch_id = ?`, receipt.AppCode, receipt.BatchID). + Scan(&state, &schemaVersion, &rowCount, &manifestSHA, &dataCRC64, &manifestCRC64, &successCRC64, &verifiedAtMS); err != nil { + return err + } + // batch_id 是内容指纹，但存储层仍逐字段核对恢复授权所依赖的控制值；同 PK 下任何 + // 非幂等 receipt 都必须在写 members 前失败，不能留下永远无法与 receipt JOIN 的孤立证明。 + if state != receipt.State || schemaVersion != receipt.SchemaVersion || rowCount != receipt.RowCount || + manifestSHA != receipt.ManifestSHA256 || dataCRC64 != receipt.DataCRC64 || + manifestCRC64 != receipt.ManifestCRC64 || successCRC64 != receipt.SuccessCRC64 || verifiedAtMS != receipt.VerifiedAtMS { + return xerr.New(xerr.Conflict, "verified archive receipt controls do not match existing batch") + } + return nil +} + +func validateWalletOutboxArchiveMembers(receipt WalletOutboxArchiveReceipt) error { + seen := make(map[string]struct{}, len(receipt.Members)) + for _, member := range receipt.Members { + if appcode.Normalize(member.AppCode) != receipt.AppCode || strings.TrimSpace(member.EventID) == "" || member.UpdatedAtMS <= 0 { + return xerr.New(xerr.InvalidArgument, "verified archive member is incomplete") + } + if _, exists := seen[member.EventID]; exists { + return xerr.New(xerr.InvalidArgument, "verified archive members contain duplicate event_id") + } + seen[member.EventID] = struct{}{} + } + return nil +} + +func upsertAndVerifyWalletOutboxArchiveMembers(ctx context.Context, tx *sql.Tx, receipt WalletOutboxArchiveReceipt, members []WalletOutboxArchiveMember) error { + values := make([]string, 0, len(members)) + args := make([]any, 0, len(members)*6) + for _, member := range members { + values = append(values, "(?, ?, ?, ?, ?, ?)") + args = append(args, receipt.AppCode, member.EventID, member.UpdatedAtMS, receipt.BatchID, receipt.ManifestSHA256, receipt.VerifiedAtMS) + } + // archive worker 的网络结果可能在 DB commit 后丢失；no-op duplicate 允许同一批重试， + // 随后的逐键读回会拒绝任何已经被另一 batch 占用或 metadata 不一致的成员。 + if _, err := tx.ExecContext(ctx, ` + INSERT INTO wallet_outbox_archive_members ( + app_code, event_id, updated_at_ms, batch_id, manifest_sha256, receipt_verified_at_ms + ) VALUES `+strings.Join(values, ",")+` + ON DUPLICATE KEY UPDATE event_id = VALUES(event_id)`, args...); err != nil { + return err + } + + placeholders := make([]string, 0, len(members)) + lookupArgs := make([]any, 0, len(members)+1) + lookupArgs = append(lookupArgs, receipt.AppCode) + expected := make(map[string]int64, len(members)) + for _, member := range members { + placeholders = append(placeholders, "?") + lookupArgs = append(lookupArgs, member.EventID) + expected[member.EventID] = member.UpdatedAtMS + } + rows, err := tx.QueryContext(ctx, ` + SELECT event_id, updated_at_ms, batch_id, manifest_sha256, receipt_verified_at_ms + FROM wallet_outbox_archive_members FORCE INDEX (PRIMARY) + WHERE app_code = ? AND event_id IN (`+strings.Join(placeholders, ",")+`)`, lookupArgs...) + if err != nil { + return err + } + defer rows.Close() + matched := 0 + for rows.Next() { + var eventID, batchID, manifestSHA string + var updatedAtMS, receiptVerifiedAtMS int64 + if err := rows.Scan(&eventID, &updatedAtMS, &batchID, &manifestSHA, &receiptVerifiedAtMS); err != nil { + return err + } + expectedUpdatedAtMS, exists := expected[eventID] + if !exists || updatedAtMS != expectedUpdatedAtMS || batchID != receipt.BatchID || manifestSHA != receipt.ManifestSHA256 || receiptVerifiedAtMS != receipt.VerifiedAtMS { + return fmt.Errorf("wallet outbox archive member conflicts with verified receipt: event_id=%s", eventID) + } + matched++ + } + if err := rows.Err(); err != nil { + return err + } + if matched != len(members) { + return fmt.Errorf("wallet outbox archive member readback mismatch: expected=%d matched=%d", len(members), matched) + } + return nil } diff --git a/services/wallet-service/internal/storage/mysql/outbox_purge.go b/services/wallet-service/internal/storage/mysql/outbox_purge.go new file mode 100644 index 00000000..f56395dd --- /dev/null +++ b/services/wallet-service/internal/storage/mysql/outbox_purge.go @@ -0,0 +1,378 @@ +package mysql + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/binary" + "encoding/hex" + "errors" + "fmt" + "hash" + "strings" + + "hyapp/pkg/appcode" + "hyapp/pkg/xerr" +) + +const ( + walletOutboxPurgeAuthorizationState = "RESTORE_VERIFIED" + walletOutboxPurgeReceiptState = "PURGED" + walletOutboxPurgePolicyFinancialV1 = "financial-v1" +) + +// financial-v1 是显式正向安全集合：只覆盖已经确认不被正常在线查询回读的账务事实。 +// 未知新事件、礼物/资源目录事件和低频资源/VIP 控制事件默认永久留热，新增清理类型必须发布新 policy version。 +var walletOutboxPurgeFinancialV1EventTypes = []string{ + "AgencyPointCredited", + "HostPeriodDiamondCredited", + "HostPointCredited", + "WalletAgencyOpeningRewardCredited", + "WalletBalanceChanged", + "WalletCoinSellerCoinCompensated", + "WalletCoinSellerStockPurchased", + "WalletCoinSellerSubTransferred", + "WalletCoinSellerTransferred", + "WalletGiftDebited", + "WalletGiftIncomeCredited", + "WalletGooglePaymentCredited", + "WalletInviteActivityRewardCredited", + "WalletLuckyGiftRewardCredited", + "WalletRechargeRecorded", + "WalletRedPacketClaimed", + "WalletRedPacketCreated", + "WalletRedPacketRefunded", + "WalletRoomTurnoverRewardCredited", + "WalletTaskRewardCredited", + "WalletWheelRewardCredited", +} + +// WalletOutboxPurgeReceipt 是一个已经提交的精确 PK 清理批次。 +// DeletedCount 为零时表示本轮没有成熟候选，此时 PurgeID 为空且数据库不会写空审计行。 +type WalletOutboxPurgeReceipt struct { + AppCode string + PurgeID string + State string + PolicyVersion string + CutoffMS int64 + ReceiptVerifiedBeforeMS int64 + CandidateSHA256 string + CandidateCount int + DeletedCount int + FirstCursor WalletOutboxArchiveCursor + LastCursor WalletOutboxArchiveCursor + PurgedAtMS int64 +} + +type walletOutboxPurgeCandidate struct { + EventID string + UpdatedAtMS int64 +} + +// RequireWalletOutboxPurgeAuthorization 要求生产 wallet DB 已手工固化一条与隔离库 +// RESTORE_VERIFIED 回执精确匹配的授权证据。迁移不会自动插入该表，因此仅部署人员 +// 明确核验过的 (App, archive batch, manifest hash) 能打开自动清理总闸。 +func (r *Repository) RequireWalletOutboxPurgeAuthorization(ctx context.Context, appCode, batchID, manifestSHA string) error { + if r == nil || r.db == nil { + return xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + ctx, normalizedAppCode, err := walletOutboxPurgeContext(ctx, appCode) + if err != nil { + return err + } + batchID, err = normalizeWalletOutboxPurgeSHA256("archive_batch_id", batchID) + if err != nil { + return err + } + manifestSHA, err = normalizeWalletOutboxPurgeSHA256("manifest_sha256", manifestSHA) + if err != nil { + return err + } + + var marker int + err = r.db.QueryRowContext(ctx, ` + SELECT 1 + FROM wallet_outbox_archive_purge_authorizations auth FORCE INDEX (PRIMARY) + JOIN wallet_outbox_archive_receipts receipt FORCE INDEX (PRIMARY) + ON receipt.app_code = auth.app_code AND receipt.batch_id = auth.archive_batch_id + AND receipt.manifest_sha256 = auth.manifest_sha256 + AND receipt.schema_version = auth.schema_version AND receipt.row_count = auth.row_count + AND receipt.data_crc64 = auth.data_crc64 AND receipt.manifest_crc64 = auth.manifest_crc64 + AND receipt.success_crc64 = auth.success_crc64 + WHERE auth.app_code = ? AND auth.archive_batch_id = ? AND auth.manifest_sha256 = ? + AND auth.state = ? AND receipt.state = ? + LIMIT 1`, + normalizedAppCode, batchID, manifestSHA, walletOutboxPurgeAuthorizationState, walletOutboxArchiveStateVerified, + ).Scan(&marker) + if errors.Is(err, sql.ErrNoRows) { + return xerr.New(xerr.Conflict, "wallet outbox purge restore rehearsal is not authorized") + } + return err +} + +// PurgeDeliveredWalletOutboxBatch 在单个 READ COMMITTED 事务内锁定、删除并审计至多 limit 条旧事实。 +// 调用方必须已经持有 AcquireWalletOutboxArchiveAppLock 返回的 App 级 GET_LOCK。候选只能从 +// exact member 表产生；每一行都重新连接当前 outbox 和 VERIFIED receipt，杜绝用批次首尾游标 +// 推断中间行、迟到写入或旧 receipt 自动获得删除资格。 +func (r *Repository) PurgeDeliveredWalletOutboxBatch(ctx context.Context, appCode, policyVersion string, cutoffMS, receiptVerifiedBeforeMS int64, limit int, nowMS int64) (WalletOutboxPurgeReceipt, error) { + if r == nil || r.db == nil { + return WalletOutboxPurgeReceipt{}, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + ctx, normalizedAppCode, err := walletOutboxPurgeContext(ctx, appCode) + if err != nil { + return WalletOutboxPurgeReceipt{}, err + } + eventTypes, err := walletOutboxPurgeEventTypes(policyVersion) + if err != nil { + return WalletOutboxPurgeReceipt{}, err + } + if cutoffMS <= 0 || receiptVerifiedBeforeMS <= 0 || nowMS <= 0 || cutoffMS >= nowMS || receiptVerifiedBeforeMS >= nowMS { + return WalletOutboxPurgeReceipt{}, xerr.New(xerr.InvalidArgument, "wallet outbox purge cutoff, receipt grace, and now are invalid") + } + if limit <= 0 || limit > 500 { + return WalletOutboxPurgeReceipt{}, xerr.New(xerr.InvalidArgument, "wallet outbox purge limit must be within [1,500]") + } + + tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted}) + if err != nil { + return WalletOutboxPurgeReceipt{}, err + } + defer func() { _ = tx.Rollback() }() + receipt := WalletOutboxPurgeReceipt{ + AppCode: normalizedAppCode, + State: walletOutboxPurgeReceiptState, + PolicyVersion: policyVersion, + CutoffMS: cutoffMS, + ReceiptVerifiedBeforeMS: receiptVerifiedBeforeMS, + PurgedAtMS: nowMS, + } + candidates, err := selectWalletOutboxPurgeCandidates(ctx, tx, normalizedAppCode, cutoffMS, receiptVerifiedBeforeMS, eventTypes, limit) + if err != nil { + return WalletOutboxPurgeReceipt{}, err + } + if len(candidates) == 0 { + return receipt, nil + } + + receipt.CandidateSHA256 = hashWalletOutboxPurgeCandidates(normalizedAppCode, candidates) + receipt.CandidateCount = len(candidates) + receipt.FirstCursor = WalletOutboxArchiveCursor{UpdatedAtMS: candidates[0].UpdatedAtMS, EventID: candidates[0].EventID} + last := candidates[len(candidates)-1] + receipt.LastCursor = WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID} + receipt.PurgeID = hashWalletOutboxPurgeReceipt(receipt) + + deleted, err := deleteWalletOutboxPurgeCandidates(ctx, tx, normalizedAppCode, cutoffMS, eventTypes, candidates) + if err != nil { + return WalletOutboxPurgeReceipt{}, err + } + if deleted != int64(len(candidates)) { + return WalletOutboxPurgeReceipt{}, fmt.Errorf("wallet outbox purge candidate/delete mismatch: candidates=%d deleted=%d", len(candidates), deleted) + } + deletedMembers, err := deleteWalletOutboxArchiveMembers(ctx, tx, normalizedAppCode, candidates) + if err != nil { + return WalletOutboxPurgeReceipt{}, err + } + if deletedMembers != int64(len(candidates)) { + // outbox 与 member 删除必须共同提交；member 数不一致时回滚 outbox 删除，避免留下无法审计的半完成批次。 + return WalletOutboxPurgeReceipt{}, fmt.Errorf("wallet outbox purge member/delete mismatch: candidates=%d deleted_members=%d", len(candidates), deletedMembers) + } + receipt.DeletedCount = int(deleted) + if err := insertWalletOutboxPurgeReceipt(ctx, tx, receipt); err != nil { + return WalletOutboxPurgeReceipt{}, err + } + if err := tx.Commit(); err != nil { + return WalletOutboxPurgeReceipt{}, err + } + return receipt, nil +} + +func selectWalletOutboxPurgeCandidates(ctx context.Context, tx *sql.Tx, appCode string, cutoffMS, receiptVerifiedBeforeMS int64, eventTypes []string, limit int) ([]walletOutboxPurgeCandidate, error) { + placeholders := walletOutboxPurgePlaceholders(len(eventTypes)) + args := make([]any, 0, len(eventTypes)+6) + args = append(args, appCode, cutoffMS, walletOutboxArchiveStateVerified, receiptVerifiedBeforeMS, outboxStatusDelivered) + for _, eventType := range eventTypes { + args = append(args, eventType) + } + args = append(args, limit) + rows, err := tx.QueryContext(ctx, ` + SELECT STRAIGHT_JOIN m.event_id, m.updated_at_ms + FROM wallet_outbox_archive_members m FORCE INDEX (idx_wallet_outbox_archive_member_purge) + JOIN wallet_outbox_archive_receipts ar FORCE INDEX (PRIMARY) + ON ar.app_code = m.app_code AND ar.batch_id = m.batch_id + AND ar.manifest_sha256 = m.manifest_sha256 AND ar.verified_at_ms = m.receipt_verified_at_ms + JOIN wallet_outbox wo FORCE INDEX (PRIMARY) + ON wo.app_code = m.app_code AND wo.event_id = m.event_id AND wo.updated_at_ms = m.updated_at_ms + WHERE m.app_code = ? AND m.updated_at_ms < ? + AND ar.state = ? AND ar.verified_at_ms < ? + AND wo.status = ? AND wo.event_type IN (`+placeholders+`) + ORDER BY m.updated_at_ms ASC, m.event_id ASC + LIMIT ? + FOR UPDATE SKIP LOCKED`, args...) + if err != nil { + return nil, err + } + defer rows.Close() + candidates := make([]walletOutboxPurgeCandidate, 0, limit) + for rows.Next() { + var candidate walletOutboxPurgeCandidate + if err := rows.Scan(&candidate.EventID, &candidate.UpdatedAtMS); err != nil { + return nil, err + } + candidates = append(candidates, candidate) + } + return candidates, rows.Err() +} + +func deleteWalletOutboxPurgeCandidates(ctx context.Context, tx *sql.Tx, appCode string, cutoffMS int64, eventTypes []string, candidates []walletOutboxPurgeCandidate) (int64, error) { + primaryKeys, args, err := walletOutboxPurgeExactKeys(appCode, candidates) + if err != nil { + return 0, err + } + args = append(args, appCode, outboxStatusDelivered, cutoffMS) + for _, eventType := range eventTypes { + args = append(args, eventType) + } + // 候选 SELECT 已锁住这些行；删除仍逐键复核 app/event/updated、delivered、cutoff 和正向 policy， + // 即使未来有人修改候选 JOIN，也不能把未授权类型或时间发生变化的行带入删除。 + result, err := tx.ExecContext(ctx, ` + DELETE wo + FROM wallet_outbox wo FORCE INDEX (PRIMARY) + WHERE (wo.app_code, wo.event_id, wo.updated_at_ms) IN (`+primaryKeys+`) + AND wo.app_code = ? AND wo.status = ? AND wo.updated_at_ms < ? + AND wo.event_type IN (`+walletOutboxPurgePlaceholders(len(eventTypes))+`)`, args...) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +func deleteWalletOutboxArchiveMembers(ctx context.Context, tx *sql.Tx, appCode string, candidates []walletOutboxPurgeCandidate) (int64, error) { + primaryKeys, args, err := walletOutboxPurgeExactKeys(appCode, candidates) + if err != nil { + return 0, err + } + result, err := tx.ExecContext(ctx, ` + DELETE m + FROM wallet_outbox_archive_members m FORCE INDEX (PRIMARY) + WHERE (m.app_code, m.event_id, m.updated_at_ms) IN (`+primaryKeys+`)`, args...) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + +func walletOutboxPurgeExactKeys(appCode string, candidates []walletOutboxPurgeCandidate) (string, []any, error) { + if len(candidates) == 0 || len(candidates) > 500 { + return "", nil, xerr.New(xerr.InvalidArgument, "wallet outbox purge candidate count must be within [1,500]") + } + keys := make([]string, 0, len(candidates)) + args := make([]any, 0, len(candidates)*3) + for _, candidate := range candidates { + keys = append(keys, "(?, ?, ?)") + args = append(args, appCode, candidate.EventID, candidate.UpdatedAtMS) + } + return strings.Join(keys, ","), args, nil +} + +func insertWalletOutboxPurgeReceipt(ctx context.Context, tx *sql.Tx, receipt WalletOutboxPurgeReceipt) error { + result, err := tx.ExecContext(ctx, ` + INSERT INTO wallet_outbox_archive_purge_receipts ( + app_code, purge_id, state, policy_version, cutoff_ms, receipt_verified_before_ms, + candidate_sha256, candidate_count, deleted_count, + first_updated_at_ms, first_event_id, last_updated_at_ms, last_event_id, + purged_at_ms, created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + receipt.AppCode, receipt.PurgeID, receipt.State, receipt.PolicyVersion, receipt.CutoffMS, receipt.ReceiptVerifiedBeforeMS, + receipt.CandidateSHA256, receipt.CandidateCount, receipt.DeletedCount, + receipt.FirstCursor.UpdatedAtMS, receipt.FirstCursor.EventID, receipt.LastCursor.UpdatedAtMS, receipt.LastCursor.EventID, + receipt.PurgedAtMS, receipt.PurgedAtMS, receipt.PurgedAtMS, + ) + if err != nil { + return err + } + affected, err := result.RowsAffected() + if err != nil { + return err + } + if affected != 1 { + return fmt.Errorf("wallet outbox purge receipt insert affected %d rows", affected) + } + return nil +} + +func walletOutboxPurgeContext(ctx context.Context, appCode string) (context.Context, string, error) { + if strings.TrimSpace(appCode) == "" { + return ctx, "", xerr.New(xerr.InvalidArgument, "wallet outbox purge app_code is required") + } + ctx = contextWithCommandApp(ctx, appCode) + normalized := appcode.FromContext(ctx) + // 本轮容量治理边界只有 Lalu/Fami；其他租户即使误配也必须在进入事务前失败。 + if normalized != "lalu" && normalized != "fami" { + return ctx, "", xerr.New(xerr.InvalidArgument, "wallet outbox purge app_code is not allowed") + } + return ctx, normalized, nil +} + +func walletOutboxPurgeEventTypes(policyVersion string) ([]string, error) { + policyVersion = strings.TrimSpace(policyVersion) + if policyVersion != walletOutboxPurgePolicyFinancialV1 { + return nil, xerr.New(xerr.InvalidArgument, "wallet outbox purge policy_version is unsupported") + } + return walletOutboxPurgeFinancialV1EventTypes, nil +} + +func normalizeWalletOutboxPurgeSHA256(field string, value string) (string, error) { + value = strings.ToLower(strings.TrimSpace(value)) + if len(value) != sha256.Size*2 { + return "", xerr.New(xerr.InvalidArgument, field+" must be a SHA-256 hex digest") + } + if _, err := hex.DecodeString(value); err != nil { + return "", xerr.New(xerr.InvalidArgument, field+" must be a SHA-256 hex digest") + } + return value, nil +} + +func walletOutboxPurgePlaceholders(count int) string { + values := make([]string, count) + for index := range values { + values[index] = "?" + } + return strings.Join(values, ",") +} + +func hashWalletOutboxPurgeCandidates(appCode string, candidates []walletOutboxPurgeCandidate) string { + digest := sha256.New() + writeWalletOutboxPurgeHashString(digest, appCode) + writeWalletOutboxPurgeHashUint64(digest, uint64(len(candidates))) + for _, candidate := range candidates { + // 候选已按 (updated_at_ms,event_id) 排序；哈希内容采用精确 PK 和归档时 updated_at 快照。 + writeWalletOutboxPurgeHashString(digest, candidate.EventID) + writeWalletOutboxPurgeHashUint64(digest, uint64(candidate.UpdatedAtMS)) + } + return hex.EncodeToString(digest.Sum(nil)) +} + +func hashWalletOutboxPurgeReceipt(receipt WalletOutboxPurgeReceipt) string { + digest := sha256.New() + writeWalletOutboxPurgeHashString(digest, receipt.AppCode) + writeWalletOutboxPurgeHashString(digest, receipt.PolicyVersion) + writeWalletOutboxPurgeHashString(digest, receipt.CandidateSHA256) + writeWalletOutboxPurgeHashUint64(digest, uint64(receipt.CutoffMS)) + writeWalletOutboxPurgeHashUint64(digest, uint64(receipt.ReceiptVerifiedBeforeMS)) + writeWalletOutboxPurgeHashUint64(digest, uint64(receipt.PurgedAtMS)) + return hex.EncodeToString(digest.Sum(nil)) +} + +func writeWalletOutboxPurgeHashString(digest hash.Hash, value string) { + // 长度前缀消除字段拼接歧义，使相同候选在不同进程上得到完全一致的审计指纹。 + var size [4]byte + binary.BigEndian.PutUint32(size[:], uint32(len(value))) + _, _ = digest.Write(size[:]) + _, _ = digest.Write([]byte(value)) +} + +func writeWalletOutboxPurgeHashUint64(digest hash.Hash, value uint64) { + var encoded [8]byte + binary.BigEndian.PutUint64(encoded[:], value) + _, _ = digest.Write(encoded[:]) +}