From 023802de618d53908e7c4f4bd44ce15fcee1697d Mon Sep 17 00:00:00 2001 From: zhx Date: Tue, 21 Jul 2026 13:53:28 +0800 Subject: [PATCH] feat(wallet): archive delivered outbox to COS --- .../wallet-service/configs/config.docker.yaml | 25 + .../configs/config.tencent.example.yaml | 27 ++ services/wallet-service/configs/config.yaml | 25 + .../mysql/initdb/001_wallet_service.sql | 34 ++ .../012_wallet_outbox_archive_receipts.sql | 32 ++ services/wallet-service/internal/app/app.go | 39 +- .../internal/app/background_workers.go | 4 + .../internal/app/outbox_archive.go | 431 ++++++++++++++++++ .../internal/app/outbox_archive_cos.go | 192 ++++++++ .../app/outbox_archive_credentials.go | 102 +++++ .../wallet-service/internal/config/config.go | 164 +++++++ .../internal/storage/mysql/outbox_archive.go | 215 +++++++++ 12 files changed, 1278 insertions(+), 12 deletions(-) create mode 100644 services/wallet-service/deploy/mysql/migrations/012_wallet_outbox_archive_receipts.sql create mode 100644 services/wallet-service/internal/app/outbox_archive.go create mode 100644 services/wallet-service/internal/app/outbox_archive_cos.go create mode 100644 services/wallet-service/internal/app/outbox_archive_credentials.go create mode 100644 services/wallet-service/internal/storage/mysql/outbox_archive.go diff --git a/services/wallet-service/configs/config.docker.yaml b/services/wallet-service/configs/config.docker.yaml index ce902c21..e5998d60 100644 --- a/services/wallet-service/configs/config.docker.yaml +++ b/services/wallet-service/configs/config.docker.yaml @@ -157,6 +157,31 @@ outbox_worker: max_retry_count: 10 initial_backoff: "5s" max_backoff: "5m" +outbox_archive: + # 未配置私有 COS 的容器默认不扫描热表；显式启用后先用 dry_run 核对范围。 + enabled: false + mode: "dry_run" + purge_enabled: false + max_age: "720h" + poll_interval: "5m" + query_batch_size: 500 + segment_max_rows: 10000 + segment_max_uncompressed_bytes: 67108864 + round_max_rows: 50000 + batch_pause: "50ms" + query_timeout: "10s" + operation_timeout: "2m" + object_prefix: "wallet-outbox/v1" + temp_dir: "" + cos: + credential_source: "static" + cam_role_name: "" + secret_id: "" + secret_key: "" + session_token: "" + bucket: "" + region: "" + http_timeout: "30s" realtime_outbox_worker: enabled: true app_codes: [] diff --git a/services/wallet-service/configs/config.tencent.example.yaml b/services/wallet-service/configs/config.tencent.example.yaml index fc665372..3be7bf1c 100644 --- a/services/wallet-service/configs/config.tencent.example.yaml +++ b/services/wallet-service/configs/config.tencent.example.yaml @@ -158,6 +158,33 @@ outbox_worker: max_retry_count: 10 initial_backoff: "5s" max_backoff: "5m" +outbox_archive: + # 模板默认关闭；生产私有配置补齐专用桶/角色后显式切 archive_only。 + enabled: false + mode: "dry_run" + # 目录版本、notice repair、statistics replay 和 legacy projection 仍读热表，当前任何环境都禁止 purge。 + purge_enabled: false + max_age: "720h" + poll_interval: "5m" + query_batch_size: 500 + segment_max_rows: 10000 + segment_max_uncompressed_bytes: 67108864 + round_max_rows: 50000 + batch_pause: "50ms" + query_timeout: "10s" + operation_timeout: "2m" + object_prefix: "wallet-outbox/v1" + temp_dir: "" + cos: + # 生产使用 CVM 实例角色的 STS 临时凭证，CAM 策略至少包含 PutObject/HeadObject/GetObject 的归档前缀最小权限。 + credential_source: "cvm_role" + cam_role_name: "HyAppWalletOutboxArchiveRole" + secret_id: "" + secret_key: "" + session_token: "" + bucket: "${TENCENT_COS_ARCHIVE_BUCKET}" + region: "${TENCENT_COS_ARCHIVE_REGION}" + http_timeout: "30s" realtime_outbox_worker: enabled: true # 实时 lane 必须与普通 lane 使用同一阶段 allowlist，避免同表出现跨 App 抢占差异。 diff --git a/services/wallet-service/configs/config.yaml b/services/wallet-service/configs/config.yaml index 2a6820f9..e9b216ed 100644 --- a/services/wallet-service/configs/config.yaml +++ b/services/wallet-service/configs/config.yaml @@ -159,6 +159,31 @@ outbox_worker: max_retry_count: 10 initial_backoff: "5s" max_backoff: "5m" +outbox_archive: + # 未配置私有 COS 的环境默认不扫描热表；显式启用后先用 dry_run 核对范围。 + enabled: false + mode: "dry_run" + purge_enabled: false + max_age: "720h" + poll_interval: "5m" + query_batch_size: 500 + segment_max_rows: 10000 + segment_max_uncompressed_bytes: 67108864 + round_max_rows: 50000 + batch_pause: "50ms" + query_timeout: "10s" + operation_timeout: "2m" + object_prefix: "wallet-outbox/v1" + temp_dir: "" + cos: + credential_source: "static" + cam_role_name: "" + secret_id: "" + secret_key: "" + session_token: "" + bucket: "" + region: "" + http_timeout: "30s" realtime_outbox_worker: enabled: true app_codes: [] 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 56f265b5..705b53cc 100644 --- a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql +++ b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql @@ -169,6 +169,40 @@ CREATE TABLE IF NOT EXISTS wallet_outbox ( KEY idx_wallet_outbox_tx (app_code, transaction_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='钱包事件 outbox 表'; +-- 只记录 data/manifest/_SUCCESS 均在私有 COS 读回校验成功的批次。 +-- 当前表不授权删除 wallet_outbox；notice/statistics/gift-wall 等读者迁移前仅用于推进 archive-only 稳定游标。 +CREATE TABLE IF NOT EXISTS wallet_outbox_archive_receipts ( + app_code VARCHAR(32) NOT NULL COMMENT '应用编码', + batch_id CHAR(64) NOT NULL COMMENT '归档批次内容指纹', + state VARCHAR(32) NOT NULL COMMENT '只有 VERIFIED 可作为后续归档游标，不代表已恢复演练', + data_object_key VARCHAR(512) NOT NULL, + manifest_object_key VARCHAR(512) NOT NULL, + success_object_key VARCHAR(512) NOT NULL, + schema_version INT 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, + min_created_at_ms BIGINT NOT NULL, + max_created_at_ms BIGINT 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_crc64 VARCHAR(32) NOT NULL, + success_crc64 VARCHAR(32) NOT NULL, + manifest_sha256 CHAR(64) NOT NULL, + uncompressed_bytes BIGINT NOT NULL, + compressed_bytes BIGINT NOT NULL, + verified_at_ms BIGINT NOT NULL, + created_at_ms BIGINT NOT NULL, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, batch_id), + 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 外部归档读回验证回执'; + -- 钱包运行侧收益政策实例：由 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/012_wallet_outbox_archive_receipts.sql b/services/wallet-service/deploy/mysql/migrations/012_wallet_outbox_archive_receipts.sql new file mode 100644 index 00000000..576371f7 --- /dev/null +++ b/services/wallet-service/deploy/mysql/migrations/012_wallet_outbox_archive_receipts.sql @@ -0,0 +1,32 @@ +-- wallet_outbox archive-only 回执表。本迁移不修改、不删除 wallet_outbox，对热表无重写和锁表风险。 +CREATE TABLE IF NOT EXISTS wallet_outbox_archive_receipts ( + app_code VARCHAR(32) NOT NULL COMMENT '应用编码', + batch_id CHAR(64) NOT NULL COMMENT '归档批次内容指纹', + state VARCHAR(32) NOT NULL COMMENT '只有 VERIFIED 可作为后续归档游标，不代表已恢复演练', + data_object_key VARCHAR(512) NOT NULL, + manifest_object_key VARCHAR(512) NOT NULL, + success_object_key VARCHAR(512) NOT NULL, + schema_version INT 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, + min_created_at_ms BIGINT NOT NULL, + max_created_at_ms BIGINT 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_crc64 VARCHAR(32) NOT NULL, + success_crc64 VARCHAR(32) NOT NULL, + manifest_sha256 CHAR(64) NOT NULL, + uncompressed_bytes BIGINT NOT NULL, + compressed_bytes BIGINT NOT NULL, + verified_at_ms BIGINT NOT NULL, + created_at_ms BIGINT NOT NULL, + updated_at_ms BIGINT NOT NULL, + PRIMARY KEY (app_code, batch_id), + 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 外部归档读回验证回执'; diff --git a/services/wallet-service/internal/app/app.go b/services/wallet-service/internal/app/app.go index 353f4169..84362616 100644 --- a/services/wallet-service/internal/app/app.go +++ b/services/wallet-service/internal/app/app.go @@ -32,18 +32,22 @@ import ( // App 装配 wallet-service gRPC 入口。 type App struct { - server *grpc.Server - listener net.Listener - health *grpchealth.ServingChecker - healthHTTP *healthhttp.Server - mysqlRepo *mysqlstorage.Repository - walletSvc *walletservice.Service - activityConn *grpc.ClientConn - outboxProducer *rocketmqx.Producer - realtimeOutboxProducer *rocketmqx.Producer - projectionConsumer *rocketmqx.Consumer - outboxWorkerCfg config.OutboxWorkerConfig - realtimeOutboxWorkerCfg config.OutboxWorkerConfig + server *grpc.Server + listener net.Listener + health *grpchealth.ServingChecker + healthHTTP *healthhttp.Server + mysqlRepo *mysqlstorage.Repository + walletSvc *walletservice.Service + activityConn *grpc.ClientConn + outboxProducer *rocketmqx.Producer + realtimeOutboxProducer *rocketmqx.Producer + projectionConsumer *rocketmqx.Consumer + outboxWorkerCfg config.OutboxWorkerConfig + outboxArchiveCfg config.OutboxArchiveConfig + outboxArchiveStore *walletOutboxArchiveCOSStore + // dry-run 游标只存内存，不会被误认为已上传回执；单归档 goroutine 顺序访问无需额外锁。 + outboxArchiveDryRunCursors map[string]mysqlstorage.WalletOutboxArchiveCursor + realtimeOutboxWorkerCfg config.OutboxWorkerConfig // 普通与实时 lane 共享同一张表但可独立配置事件类型；各自维护 App 轮转游标，避免一个 lane 的吞吐改变另一个 lane 的公平性。 outboxPartitions *outboxpartition.Partitions realtimeOutboxPartitions *outboxpartition.Partitions @@ -69,6 +73,14 @@ func New(cfg config.Config) (*App, error) { if err != nil { return nil, err } + var outboxArchiveStore *walletOutboxArchiveCOSStore + if cfg.OutboxArchive.Enabled && cfg.OutboxArchive.Mode == config.OutboxArchiveModeArchiveOnly { + outboxArchiveStore, err = newWalletOutboxArchiveCOSStore(cfg.OutboxArchive.COS, nil) + if err != nil { + _ = repository.Close() + return nil, err + } + } listener, err := net.Listen("tcp", cfg.GRPCAddr) if err != nil { @@ -207,6 +219,9 @@ func New(cfg config.Config) (*App, error) { realtimeOutboxProducer: realtimeOutboxProducer, projectionConsumer: projectionConsumer, outboxWorkerCfg: cfg.OutboxWorker, + outboxArchiveCfg: cfg.OutboxArchive, + outboxArchiveStore: outboxArchiveStore, + outboxArchiveDryRunCursors: make(map[string]mysqlstorage.WalletOutboxArchiveCursor), realtimeOutboxWorkerCfg: cfg.RealtimeOutboxWorker, outboxPartitions: outboxpartition.New(cfg.OutboxWorker.AppCodes, outboxpartition.DefaultDiscoveryRefreshInterval), realtimeOutboxPartitions: outboxpartition.New(cfg.RealtimeOutboxWorker.AppCodes, outboxpartition.DefaultDiscoveryRefreshInterval), diff --git a/services/wallet-service/internal/app/background_workers.go b/services/wallet-service/internal/app/background_workers.go index 5ac97f22..2a90658c 100644 --- a/services/wallet-service/internal/app/background_workers.go +++ b/services/wallet-service/internal/app/background_workers.go @@ -19,6 +19,10 @@ func (a *App) runBackgroundWorkers() { if a.realtimeOutboxWorkerCfg.Enabled && a.realtimeOutboxProducer != nil { 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 也持续用有界索引查询暴露容量，但两种模式都不删除热表行。 + a.startWalletOutboxArchiveWorker(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 new file mode 100644 index 00000000..fd30ba61 --- /dev/null +++ b/services/wallet-service/internal/app/outbox_archive.go @@ -0,0 +1,431 @@ +package app + +import ( + "compress/gzip" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "math/big" + "os" + "path" + "strings" + "time" + + "hyapp/pkg/appcode" + "hyapp/pkg/logx" + "hyapp/services/wallet-service/internal/config" + mysqlstorage "hyapp/services/wallet-service/internal/storage/mysql" +) + +const ( + walletOutboxArchiveSchema = "hyapp.wallet_outbox.archive" + walletOutboxArchiveSchemaVersion = 1 + walletOutboxArchiveManifestState = "DATA_VERIFIED" + walletOutboxArchiveVerifiedState = "VERIFIED" +) + +type walletOutboxArchiveManifest 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 mysqlstorage.WalletOutboxArchiveCursor `json:"first_cursor"` + LastCursor mysqlstorage.WalletOutboxArchiveCursor `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 walletOutboxArchiveSuccess 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"` +} + +type walletOutboxArchiveArtifact struct { + manifest walletOutboxArchiveManifest + manifestBody []byte + manifestSHA256 string + successBody []byte + successSHA256 string + dataFilePath string + dataObjectKey string + manifestObjectKey string + successObjectKey string +} + +func (a *App) startWalletOutboxArchiveWorker(ctx context.Context) { + a.workers.Add(1) + go func() { + defer a.workers.Done() + ticker := time.NewTicker(a.outboxArchiveCfg.PollInterval) + defer ticker.Stop() + for { + if err := a.runWalletOutboxArchiveRound(ctx); err != nil && !errors.Is(err, context.Canceled) { + logx.Error(ctx, "wallet_outbox_archive_round_failed", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() +} + +func (a *App) runWalletOutboxArchiveRound(ctx context.Context) error { + if a.mysqlRepo == nil { + return nil + } + queryCtx, cancel := context.WithTimeout(ctx, a.outboxArchiveCfg.QueryTimeout) + appCodes, err := a.mysqlRepo.ListWalletOutboxAppCodes(queryCtx) + cancel() + if err != nil { + return err + } + cutoffMS := time.Now().UTC().Add(-a.outboxArchiveCfg.MaxAge).UnixMilli() + for _, appCode := range appCodes { + var appErr error + if a.outboxArchiveCfg.Mode == config.OutboxArchiveModeDryRun { + appErr = a.inspectWalletOutboxArchivePage(ctx, appCode, cutoffMS) + } else { + appErr = a.archiveWalletOutboxAppRound(ctx, appCode, cutoffMS) + } + if appErr != nil { + // 单租户归档失败不得阻塞其他 App；失败 App 不写 receipt，下一轮仍从同一游标重试。 + logx.Error(ctx, "wallet_outbox_archive_app_failed", appErr, slog.String("app_code", appCode)) + } + } + return nil +} + +func (a *App) inspectWalletOutboxArchivePage(ctx context.Context, targetAppCode string, cutoffMS int64) error { + tenantCtx := appcode.WithContext(ctx, targetAppCode) + cursor := a.outboxArchiveDryRunCursors[targetAppCode] + queryCtx, cancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout) + records, err := a.mysqlRepo.ListDeliveredWalletOutboxArchiveBatch(queryCtx, cursor, cutoffMS, a.outboxArchiveCfg.QueryBatchSize) + cancel() + if err != nil || len(records) == 0 { + return err + } + first := records[0] + last := records[len(records)-1] + a.outboxArchiveDryRunCursors[targetAppCode] = mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID} + logx.Info(ctx, "wallet_outbox_archive_planned", + slog.String("app_code", targetAppCode), + slog.Int("row_count", len(records)), + slog.Int64("cutoff_ms", cutoffMS), + slog.Int64("first_updated_at_ms", first.UpdatedAtMS), + slog.Int64("last_updated_at_ms", last.UpdatedAtMS), + slog.String("state", "PLANNED"), + slog.Bool("purge_enabled", false), + ) + return nil +} + +func (a *App) archiveWalletOutboxAppRound(ctx context.Context, targetAppCode string, cutoffMS int64) error { + tenantCtx := appcode.WithContext(ctx, targetAppCode) + lockCtx, lockCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout) + release, acquired, err := a.mysqlRepo.AcquireWalletOutboxArchiveAppLock(lockCtx) + lockCancel() + if err != nil || !acquired { + return err + } + defer release() + queryCtx, cancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout) + cursor, err := a.mysqlRepo.LatestVerifiedWalletOutboxArchiveCursor(queryCtx) + cancel() + if err != nil { + return err + } + archivedRows := 0 + for archivedRows < a.outboxArchiveCfg.RoundMaxRows { + segmentLimit := min(a.outboxArchiveCfg.SegmentMaxRows, a.outboxArchiveCfg.RoundMaxRows-archivedRows) + records, err := a.collectWalletOutboxArchiveSegment(tenantCtx, cursor, cutoffMS, segmentLimit) + if err != nil || len(records) == 0 { + return err + } + if err := a.archiveWalletOutboxRecords(tenantCtx, targetAppCode, records); err != nil { + return err + } + last := records[len(records)-1] + cursor = mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID} + archivedRows += len(records) + if len(records) < segmentLimit { + break + } + } + return nil +} + +func (a *App) collectWalletOutboxArchiveSegment(ctx context.Context, cursor mysqlstorage.WalletOutboxArchiveCursor, cutoffMS int64, segmentLimit int) ([]mysqlstorage.WalletOutboxArchiveRecord, error) { + records := make([]mysqlstorage.WalletOutboxArchiveRecord, 0, segmentLimit) + partitionDay := "" + var uncompressedBytes int64 + for len(records) < segmentLimit { + pageLimit := min(a.outboxArchiveCfg.QueryBatchSize, segmentLimit-len(records)) + queryCtx, cancel := context.WithTimeout(ctx, a.outboxArchiveCfg.QueryTimeout) + page, err := a.mysqlRepo.ListDeliveredWalletOutboxArchiveBatch(queryCtx, cursor, cutoffMS, pageLimit) + cancel() + if err != nil { + return nil, err + } + if len(page) == 0 { + break + } + for _, record := range page { + recordDay := time.UnixMilli(record.UpdatedAtMS).UTC().Format("2006-01-02") + if partitionDay == "" { + partitionDay = recordDay + } + // 一个对象严格属于一个 UTC delivered 日，保证按日盘点/恢复不会遗漏跨日批次里的前半段。 + if recordDay != partitionDay { + return records, nil + } + line, err := json.Marshal(record) + if err != nil { + return nil, err + } + recordBytes := int64(len(line) + 1) + // 至少允许单行进入批次；超大单行仍由 operation timeout 和临时盘容量形成外层保护。 + if len(records) > 0 && uncompressedBytes+recordBytes > a.outboxArchiveCfg.SegmentMaxUncompressedBytes { + return records, nil + } + records = append(records, record) + uncompressedBytes += recordBytes + if len(records) == segmentLimit { + return records, nil + } + } + last := records[len(records)-1] + cursor = mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID} + if len(page) < pageLimit { + break + } + if a.outboxArchiveCfg.BatchPause > 0 { + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(a.outboxArchiveCfg.BatchPause): + } + } + } + return records, nil +} + +func (a *App) archiveWalletOutboxRecords(tenantCtx context.Context, targetAppCode string, records []mysqlstorage.WalletOutboxArchiveRecord) error { + if a.outboxArchiveStore == nil { + return fmt.Errorf("wallet outbox archive store is not configured") + } + artifact, err := buildWalletOutboxArchiveArtifact(a.outboxArchiveCfg, records) + if err != nil { + return err + } + defer artifact.remove() + operationCtx, operationCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.OperationTimeout) + defer operationCancel() + // 只有 data 读回通过后才上传 manifest，manifest 读回通过后才写 _SUCCESS； + // 任意中断都不会产生 MySQL verified receipt，后续可用相同内容键幂等重试。 + dataCRC64, err := a.outboxArchiveStore.putGzipAndVerify(operationCtx, artifact.dataObjectKey, artifact.dataFilePath, artifact.manifest.CompressedBytes, artifact.manifest.GzipSHA256, artifact.manifest.NDJSONSHA256) + if err != nil { + return err + } + if err := artifact.finalizeManifest(dataCRC64); err != nil { + return err + } + manifestCRC64, err := a.outboxArchiveStore.putBytesAndVerify(operationCtx, artifact.manifestObjectKey, artifact.manifestBody, "application/json", artifact.manifestSHA256) + if err != nil { + return err + } + if err := artifact.finalizeSuccess(manifestCRC64); err != nil { + return err + } + successCRC64, err := a.outboxArchiveStore.putBytesAndVerify(operationCtx, artifact.successObjectKey, artifact.successBody, "application/json", artifact.successSHA256) + if err != nil { + return err + } + nowMS := time.Now().UTC().UnixMilli() + receiptCtx, receiptCancel := context.WithTimeout(tenantCtx, a.outboxArchiveCfg.QueryTimeout) + err = a.mysqlRepo.RecordVerifiedWalletOutboxArchive(receiptCtx, mysqlstorage.WalletOutboxArchiveReceipt{ + AppCode: targetAppCode, BatchID: artifact.manifest.BatchID, State: walletOutboxArchiveVerifiedState, + DataObjectKey: artifact.dataObjectKey, ManifestObjectKey: artifact.manifestObjectKey, SuccessObjectKey: artifact.successObjectKey, + SchemaVersion: artifact.manifest.SchemaVersion, RowCount: artifact.manifest.RowCount, + FirstCursor: artifact.manifest.FirstCursor, LastCursor: artifact.manifest.LastCursor, + MinCreatedAtMS: artifact.manifest.MinCreatedAtMS, MaxCreatedAtMS: artifact.manifest.MaxCreatedAtMS, + AvailableDeltaTotal: artifact.manifest.AvailableDeltaTotal, FrozenDeltaTotal: artifact.manifest.FrozenDeltaTotal, + 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, + }) + receiptCancel() + if err != nil { + return err + } + logx.Info(tenantCtx, "wallet_outbox_archive_verified", + slog.String("app_code", targetAppCode), slog.String("batch_id", artifact.manifest.BatchID), + slog.Int("row_count", artifact.manifest.RowCount), slog.String("state", walletOutboxArchiveVerifiedState), + slog.Bool("purge_enabled", false), + ) + return nil +} + +func buildWalletOutboxArchiveArtifact(cfg config.OutboxArchiveConfig, records []mysqlstorage.WalletOutboxArchiveRecord) (walletOutboxArchiveArtifact, error) { + if len(records) == 0 { + return walletOutboxArchiveArtifact{}, fmt.Errorf("wallet outbox archive batch is empty") + } + tempFile, err := os.CreateTemp(cfg.TempDir, "wallet-outbox-*.ndjson.gz") + if err != nil { + return walletOutboxArchiveArtifact{}, err + } + tempPath := tempFile.Name() + removeOnError := true + defer func() { + if removeOnError { + _ = tempFile.Close() + _ = os.Remove(tempPath) + } + }() + // CreateTemp 默认 0600，仍显式收紧权限，避免自定义 umask 导致账务快照被同机其他用户读取。 + if err := tempFile.Chmod(0o600); err != nil { + return walletOutboxArchiveArtifact{}, err + } + gzipHash := sha256.New() + gzipWriter := gzip.NewWriter(io.MultiWriter(tempFile, gzipHash)) + gzipWriter.Header.ModTime = time.Unix(0, 0).UTC() + gzipWriter.Header.OS = 255 + ndjsonHash := sha256.New() + countedNDJSON := &countingWriter{writer: io.MultiWriter(gzipWriter, ndjsonHash)} + availableTotal := new(big.Int) + frozenTotal := new(big.Int) + minCreatedAtMS, maxCreatedAtMS := records[0].CreatedAtMS, records[0].CreatedAtMS + partitionDay := time.UnixMilli(records[0].UpdatedAtMS).UTC().Format("2006-01-02") + for _, record := range records { + if record.AppCode != records[0].AppCode || record.Status != "delivered" { + return walletOutboxArchiveArtifact{}, fmt.Errorf("wallet outbox archive batch crosses app boundary or contains non-delivered row") + } + if time.UnixMilli(record.UpdatedAtMS).UTC().Format("2006-01-02") != partitionDay { + return walletOutboxArchiveArtifact{}, fmt.Errorf("wallet outbox archive batch crosses UTC delivered day") + } + line, err := json.Marshal(record) + if err != nil { + return walletOutboxArchiveArtifact{}, err + } + line = append(line, '\n') + if _, err := countedNDJSON.Write(line); err != nil { + return walletOutboxArchiveArtifact{}, err + } + availableTotal.Add(availableTotal, big.NewInt(record.AvailableDelta)) + frozenTotal.Add(frozenTotal, big.NewInt(record.FrozenDelta)) + if record.CreatedAtMS < minCreatedAtMS { + minCreatedAtMS = record.CreatedAtMS + } + if record.CreatedAtMS > maxCreatedAtMS { + maxCreatedAtMS = record.CreatedAtMS + } + } + if err := gzipWriter.Close(); err != nil { + return walletOutboxArchiveArtifact{}, err + } + if err := tempFile.Sync(); err != nil { + return walletOutboxArchiveArtifact{}, err + } + info, err := tempFile.Stat() + if err != nil { + return walletOutboxArchiveArtifact{}, err + } + if err := tempFile.Close(); err != nil { + return walletOutboxArchiveArtifact{}, err + } + first, last := records[0], records[len(records)-1] + firstCursor := mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: first.UpdatedAtMS, EventID: first.EventID} + lastCursor := mysqlstorage.WalletOutboxArchiveCursor{UpdatedAtMS: last.UpdatedAtMS, EventID: last.EventID} + ndjsonSHA256 := hex.EncodeToString(ndjsonHash.Sum(nil)) + gzipSHA256 := hex.EncodeToString(gzipHash.Sum(nil)) + batchHash := sha256.Sum256([]byte(strings.Join([]string{first.AppCode, fmt.Sprint(len(records)), fmt.Sprint(firstCursor.UpdatedAtMS), firstCursor.EventID, fmt.Sprint(lastCursor.UpdatedAtMS), lastCursor.EventID, ndjsonSHA256}, "\n"))) + batchID := hex.EncodeToString(batchHash[:]) + objectDirectory := path.Join(cfg.ObjectPrefix, "app_code="+first.AppCode, "delivered_day="+partitionDay, "batch_id="+batchID) + dataObjectKey := path.Join(objectDirectory, "data.ndjson.gz") + manifest := walletOutboxArchiveManifest{ + Schema: walletOutboxArchiveSchema, SchemaVersion: walletOutboxArchiveSchemaVersion, State: walletOutboxArchiveManifestState, + AppCode: first.AppCode, BatchID: batchID, RowCount: len(records), FirstCursor: firstCursor, LastCursor: lastCursor, + MinCreatedAtMS: minCreatedAtMS, MaxCreatedAtMS: maxCreatedAtMS, MinUpdatedAtMS: first.UpdatedAtMS, MaxUpdatedAtMS: last.UpdatedAtMS, + AvailableDeltaTotal: availableTotal.String(), FrozenDeltaTotal: frozenTotal.String(), NDJSONSHA256: ndjsonSHA256, GzipSHA256: gzipSHA256, + UncompressedBytes: countedNDJSON.count, CompressedBytes: info.Size(), DataObjectKey: dataObjectKey, + } + removeOnError = false + return walletOutboxArchiveArtifact{ + manifest: manifest, dataFilePath: tempPath, + dataObjectKey: dataObjectKey, manifestObjectKey: path.Join(objectDirectory, "manifest.json"), successObjectKey: path.Join(objectDirectory, "_SUCCESS"), + }, nil +} + +func (a *walletOutboxArchiveArtifact) finalizeManifest(dataCRC64 string) error { + dataCRC64 = strings.TrimSpace(dataCRC64) + if dataCRC64 == "" { + return fmt.Errorf("wallet outbox archive data CRC64 is empty") + } + a.manifest.DataCRC64 = dataCRC64 + manifestBody, err := json.Marshal(a.manifest) + if err != nil { + return err + } + a.manifestBody = append(manifestBody, '\n') + manifestHash := sha256.Sum256(a.manifestBody) + a.manifestSHA256 = hex.EncodeToString(manifestHash[:]) + return nil +} + +func (a *walletOutboxArchiveArtifact) finalizeSuccess(manifestCRC64 string) error { + manifestCRC64 = strings.TrimSpace(manifestCRC64) + if manifestCRC64 == "" { + return fmt.Errorf("wallet outbox archive manifest CRC64 is empty") + } + successBody, err := json.Marshal(walletOutboxArchiveSuccess{ + Schema: walletOutboxArchiveSchema, SchemaVersion: walletOutboxArchiveSchemaVersion, + State: walletOutboxArchiveVerifiedState, BatchID: a.manifest.BatchID, + ManifestSHA256: a.manifestSHA256, ManifestCRC64: manifestCRC64, + }) + if err != nil { + return err + } + a.successBody = append(successBody, '\n') + successHash := sha256.Sum256(a.successBody) + a.successSHA256 = hex.EncodeToString(successHash[:]) + return nil +} + +func (a walletOutboxArchiveArtifact) remove() { + if strings.TrimSpace(a.dataFilePath) != "" { + _ = os.Remove(a.dataFilePath) + } +} + +type countingWriter struct { + writer io.Writer + count int64 +} + +func (w *countingWriter) Write(body []byte) (int, error) { + written, err := w.writer.Write(body) + w.count += int64(written) + return written, err +} diff --git a/services/wallet-service/internal/app/outbox_archive_cos.go b/services/wallet-service/internal/app/outbox_archive_cos.go new file mode 100644 index 00000000..2ad663ba --- /dev/null +++ b/services/wallet-service/internal/app/outbox_archive_cos.go @@ -0,0 +1,192 @@ +package app + +import ( + "compress/gzip" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "strings" + + cos "github.com/tencentyun/cos-go-sdk-v5" + "hyapp/services/wallet-service/internal/config" +) + +var errWalletOutboxArchiveObjectExists = errors.New("wallet outbox archive object already exists") + +// walletOutboxArchiveCOSStore 只使用私有 bucket API，不生成公网 URL，也不在对象上设置 public-read ACL。 +type walletOutboxArchiveCOSStore struct { + client *cos.Client +} + +func newWalletOutboxArchiveCOSStore(cfg config.OutboxArchiveCOSConfig, transport http.RoundTripper) (*walletOutboxArchiveCOSStore, error) { + bucketURL, err := url.Parse(fmt.Sprintf("https://%s.cos.%s.myqcloud.com", strings.TrimSpace(cfg.Bucket), strings.TrimSpace(cfg.Region))) + if err != nil { + return nil, fmt.Errorf("build wallet outbox archive COS url: %w", err) + } + if transport == nil { + transport = http.DefaultTransport + } + var authorizationTransport http.RoundTripper + if cfg.CredentialSource == "cvm_role" { + authorizationTransport = &cvmRoleAuthorizationTransport{provider: newCVMCAMCredentialProvider(cfg.CAMRoleName), transport: transport} + } else { + authorizationTransport = &cos.AuthorizationTransport{ + SecretID: cfg.SecretID, + SecretKey: cfg.SecretKey, + SessionToken: cfg.SessionToken, + Transport: transport, + } + } + return &walletOutboxArchiveCOSStore{client: cos.NewClient(&cos.BaseURL{BucketURL: bucketURL}, &http.Client{ + Timeout: cfg.HTTPTimeout, + Transport: authorizationTransport, + })}, nil +} + +func (s *walletOutboxArchiveCOSStore) putGzipAndVerify(ctx context.Context, key string, filePath string, sizeBytes int64, gzipSHA256 string, ndjsonSHA256 string) (string, error) { + file, err := os.Open(filePath) + if err != nil { + return "", err + } + defer file.Close() + // 对象本体已经是 gzip 文件，不设置 Content-Encoding。Go HTTP Transport 会自动解压带 + // Content-Encoding:gzip 的响应，导致后续无法同时核对“压缩对象 hash”和“解压 NDJSON hash”。 + putCRC64, err := s.put(ctx, key, file, sizeBytes, "application/gzip", "", gzipSHA256) + if err != nil && !errors.Is(err, errWalletOutboxArchiveObjectExists) { + return "", err + } + crc64, err := s.verifyHead(ctx, key, sizeBytes, gzipSHA256, putCRC64) + if err != nil { + return "", err + } + if err := s.verifyGzipReadback(ctx, key, gzipSHA256, ndjsonSHA256); err != nil { + return "", err + } + return crc64, nil +} + +func (s *walletOutboxArchiveCOSStore) putBytesAndVerify(ctx context.Context, key string, body []byte, contentType string, expectedSHA256 string) (string, error) { + putCRC64, err := s.put(ctx, key, strings.NewReader(string(body)), int64(len(body)), contentType, "", expectedSHA256) + if err != nil && !errors.Is(err, errWalletOutboxArchiveObjectExists) { + return "", err + } + crc64, err := s.verifyHead(ctx, key, int64(len(body)), expectedSHA256, putCRC64) + if err != nil { + return "", err + } + response, err := s.client.Object.Get(ctx, key, nil) + if err != nil { + return "", err + } + defer response.Body.Close() + hash := sha256.New() + if _, err := io.Copy(hash, response.Body); err != nil { + return "", err + } + if actual := hex.EncodeToString(hash.Sum(nil)); actual != expectedSHA256 { + return "", fmt.Errorf("wallet outbox archive COS readback hash mismatch for %s", key) + } + return crc64, nil +} + +func (s *walletOutboxArchiveCOSStore) put(ctx context.Context, key string, reader io.Reader, sizeBytes int64, contentType string, contentEncoding string, contentSHA256 string) (string, error) { + if s == nil || s.client == nil || strings.TrimSpace(key) == "" || reader == nil || sizeBytes <= 0 { + return "", fmt.Errorf("wallet outbox archive COS put input is invalid") + } + metadata := make(http.Header) + metadata.Set("x-cos-meta-sha256", contentSHA256) + optionHeaders := make(http.Header) + // 归档键由内容指纹决定；禁止覆盖可避免 bucket versioning 在并发/重试时生成重复版本。 + optionHeaders.Set("x-cos-forbid-overwrite", "true") + options := &cos.ObjectPutOptions{ObjectPutHeaderOptions: &cos.ObjectPutHeaderOptions{ + ContentLength: sizeBytes, + ContentType: contentType, + ContentEncoding: contentEncoding, + XCosMetaXXX: &metadata, + XOptionHeader: &optionHeaders, + XCosStorageClass: "STANDARD", + XCosServerSideEncryption: "AES256", + }} + response, err := s.client.Object.Put(ctx, strings.TrimLeft(key, "/"), reader, options) + crc64 := "" + if response != nil { + crc64 = strings.TrimSpace(response.Header.Get("x-cos-hash-crc64ecma")) + } + if response != nil && response.Body != nil { + _ = response.Body.Close() + } + if err != nil { + var cosErr *cos.ErrorResponse + if errors.As(err, &cosErr) && cosErr.Response != nil && (cosErr.Response.StatusCode == http.StatusConflict || cosErr.Response.StatusCode == http.StatusPreconditionFailed) { + return "", errWalletOutboxArchiveObjectExists + } + return "", err + } + if crc64 == "" { + return "", fmt.Errorf("wallet outbox archive COS PUT response has no CRC64 for %s", key) + } + return crc64, nil +} + +func (s *walletOutboxArchiveCOSStore) verifyHead(ctx context.Context, key string, expectedSize int64, expectedSHA256 string, expectedCRC64 string) (string, error) { + response, err := s.client.Object.Head(ctx, key, nil) + if err != nil { + return "", err + } + if response.Body != nil { + defer response.Body.Close() + } + if response.ContentLength != expectedSize || response.Header.Get("x-cos-meta-sha256") != expectedSHA256 { + return "", fmt.Errorf("wallet outbox archive COS HEAD mismatch for %s", key) + } + crc64 := strings.TrimSpace(response.Header.Get("x-cos-hash-crc64ecma")) + if crc64 == "" || (expectedCRC64 != "" && crc64 != expectedCRC64) { + return "", fmt.Errorf("wallet outbox archive COS HEAD CRC64 mismatch for %s", key) + } + // SSE-COS 是归档对象的强制约束；响应缺失加密标记时不能写 verified receipt。 + if !strings.EqualFold(response.Header.Get("x-cos-server-side-encryption"), "AES256") { + return "", fmt.Errorf("wallet outbox archive COS object is not SSE-COS encrypted: %s", key) + } + return crc64, nil +} + +func (s *walletOutboxArchiveCOSStore) verifyGzipReadback(ctx context.Context, key string, expectedGzipSHA256 string, expectedNDJSONSHA256 string) error { + response, err := s.client.Object.Get(ctx, key, nil) + if err != nil { + return err + } + defer response.Body.Close() + gzipHash := sha256.New() + compressedReader := io.TeeReader(response.Body, gzipHash) + gzipReader, err := gzip.NewReader(compressedReader) + if err != nil { + return err + } + ndjsonHash := sha256.New() + _, copyErr := io.Copy(ndjsonHash, gzipReader) + closeErr := gzipReader.Close() + // gzip.Reader 可能在内部缓冲区之后留下响应尾部；继续读完才能验证整个压缩对象的 hash。 + _, drainErr := io.Copy(io.Discard, compressedReader) + if copyErr != nil { + return copyErr + } + if closeErr != nil { + return closeErr + } + if drainErr != nil { + return drainErr + } + if actual := hex.EncodeToString(gzipHash.Sum(nil)); actual != expectedGzipSHA256 { + return fmt.Errorf("wallet outbox archive COS gzip hash mismatch for %s", key) + } + if actual := hex.EncodeToString(ndjsonHash.Sum(nil)); actual != expectedNDJSONSHA256 { + return fmt.Errorf("wallet outbox archive COS ndjson hash mismatch for %s", key) + } + return nil +} diff --git a/services/wallet-service/internal/app/outbox_archive_credentials.go b/services/wallet-service/internal/app/outbox_archive_credentials.go new file mode 100644 index 00000000..c167a8d7 --- /dev/null +++ b/services/wallet-service/internal/app/outbox_archive_credentials.go @@ -0,0 +1,102 @@ +package app + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strings" + "sync" + "time" + + cos "github.com/tencentyun/cos-go-sdk-v5" +) + +const cvmCAMCredentialMetadataBaseURL = "http://metadata.tencentyun.com/latest/meta-data/cam/security-credentials" + +type cvmCAMCredential struct { + TmpSecretID string `json:"TmpSecretId"` + TmpSecretKey string `json:"TmpSecretKey"` + Token string `json:"Token"` + ExpiredTime int64 `json:"ExpiredTime"` + Code string `json:"Code"` +} + +// cvmCAMCredentialProvider 在凭证过期前 5 分钟刷新，整个结构不暴露凭证方法，避免误打日志。 +type cvmCAMCredentialProvider struct { + roleName string + client *http.Client + mu sync.Mutex + cached cvmCAMCredential +} + +func newCVMCAMCredentialProvider(roleName string) *cvmCAMCredentialProvider { + return &cvmCAMCredentialProvider{ + roleName: strings.TrimSpace(roleName), + // metadata 是启动/请求链路的外部依赖，必须在 2 秒内失败，不能把 COS 归档 goroutine 长时间卡死。 + client: &http.Client{Timeout: 2 * time.Second}, + } +} + +func (p *cvmCAMCredentialProvider) get(ctx context.Context) (string, string, string, error) { + p.mu.Lock() + defer p.mu.Unlock() + now := time.Now().UTC().Unix() + if p.cached.ExpiredTime > now+int64((5*time.Minute).Seconds()) { + return p.cached.TmpSecretID, p.cached.TmpSecretKey, p.cached.Token, nil + } + previous := p.cached + if p.roleName == "" { + return "", "", "", fmt.Errorf("wallet outbox archive CVM CAM role name is empty") + } + metadataURL := cvmCAMCredentialMetadataBaseURL + "/" + url.PathEscape(p.roleName) + request, err := http.NewRequestWithContext(ctx, http.MethodGet, metadataURL, nil) + if err != nil { + return "", "", "", err + } + response, err := p.client.Do(request) + if err != nil { + if previous.ExpiredTime > now { + return previous.TmpSecretID, previous.TmpSecretKey, previous.Token, nil + } + return "", "", "", fmt.Errorf("load wallet outbox archive CVM role credential: %w", err) + } + defer response.Body.Close() + if response.StatusCode < 200 || response.StatusCode > 299 { + // 不读取/回传 metadata 错误体，避免平台异常响应携带临时凭证。 + if previous.ExpiredTime > now { + return previous.TmpSecretID, previous.TmpSecretKey, previous.Token, nil + } + return "", "", "", fmt.Errorf("load wallet outbox archive CVM role credential: metadata status %d", response.StatusCode) + } + var credential cvmCAMCredential + if err := json.NewDecoder(response.Body).Decode(&credential); err != nil { + return "", "", "", fmt.Errorf("decode wallet outbox archive CVM role credential: %w", err) + } + if credential.Code != "Success" || strings.TrimSpace(credential.TmpSecretID) == "" || strings.TrimSpace(credential.TmpSecretKey) == "" || strings.TrimSpace(credential.Token) == "" || credential.ExpiredTime <= now+30 { + return "", "", "", fmt.Errorf("wallet outbox archive CVM role returned incomplete or expired credential") + } + p.cached = credential + return credential.TmpSecretID, credential.TmpSecretKey, credential.Token, nil +} + +// cvmRoleAuthorizationTransport 每次签名前从缓存 provider 取 STS；只有刷新窗口才访问 metadata。 +type cvmRoleAuthorizationTransport struct { + provider *cvmCAMCredentialProvider + transport http.RoundTripper +} + +func (t *cvmRoleAuthorizationTransport) GetCredential() (string, string, string, error) { + return t.provider.get(context.Background()) +} + +func (t *cvmRoleAuthorizationTransport) RoundTrip(request *http.Request) (*http.Response, error) { + secretID, secretKey, token, err := t.provider.get(request.Context()) + if err != nil { + return nil, err + } + return (&cos.AuthorizationTransport{ + SecretID: secretID, SecretKey: secretKey, SessionToken: token, Transport: t.transport, + }).RoundTrip(request) +} diff --git a/services/wallet-service/internal/config/config.go b/services/wallet-service/internal/config/config.go index 991afd81..16e209e8 100644 --- a/services/wallet-service/internal/config/config.go +++ b/services/wallet-service/internal/config/config.go @@ -48,6 +48,8 @@ type Config struct { ExternalRechargeReconcileWorker ExternalRechargeReconcileWorkerConfig `yaml:"external_recharge_reconcile_worker"` // OutboxWorker 控制 wallet_outbox 到 MQ 的补偿投递。 OutboxWorker OutboxWorkerConfig `yaml:"outbox_worker"` + // OutboxArchive 只把已投递事实备份到外部存储；当前不提供 purge，避免破坏线上回放与修复读者。 + OutboxArchive OutboxArchiveConfig `yaml:"outbox_archive"` // RealtimeOutboxWorker 控制红包等实时 UI 事件到独立 MQ topic 的补偿投递。 RealtimeOutboxWorker OutboxWorkerConfig `yaml:"realtime_outbox_worker"` // ProjectionWorker 控制 wallet_outbox MQ 到本服务读模型的投影消费。 @@ -115,6 +117,46 @@ type OutboxWorkerConfig struct { EventTypes []string `yaml:"event_types"` } +// OutboxArchiveMode 区分无写入的容量评估与可验证的 COS 归档。 +type OutboxArchiveMode string + +const ( + OutboxArchiveModeDryRun OutboxArchiveMode = "dry_run" + OutboxArchiveModeArchiveOnly OutboxArchiveMode = "archive_only" +) + +// OutboxArchiveConfig 控制 delivered wallet_outbox 的 archive-only 备份链路。 +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"` + // 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"` + BatchPause time.Duration `yaml:"batch_pause"` + QueryTimeout time.Duration `yaml:"query_timeout"` + OperationTimeout time.Duration `yaml:"operation_timeout"` + ObjectPrefix string `yaml:"object_prefix"` + TempDir string `yaml:"temp_dir"` + COS OutboxArchiveCOSConfig `yaml:"cos"` +} + +// OutboxArchiveCOSConfig 只允许服务端环境注入私有桶凭证；归档对象不生成公网 URL。 +type OutboxArchiveCOSConfig struct { + CredentialSource string `yaml:"credential_source"` + CAMRoleName string `yaml:"cam_role_name"` + SecretID string `yaml:"secret_id"` + SecretKey string `yaml:"secret_key"` + SessionToken string `yaml:"session_token"` + Bucket string `yaml:"bucket"` + Region string `yaml:"region"` + HTTPTimeout time.Duration `yaml:"http_timeout"` +} + // ProjectionWorkerConfig 控制礼物墙和 badge 展示读模型从 wallet_outbox MQ 投影的节奏。 type ProjectionWorkerConfig struct { Enabled bool `yaml:"enabled"` @@ -349,6 +391,25 @@ func Default() Config { InitialBackoff: 5 * time.Second, MaxBackoff: 5 * time.Minute, }, + OutboxArchive: OutboxArchiveConfig{ + Enabled: false, + Mode: OutboxArchiveModeDryRun, + PurgeEnabled: false, + MaxAge: 30 * 24 * time.Hour, + PollInterval: 5 * time.Minute, + QueryBatchSize: 500, + SegmentMaxRows: 10000, + SegmentMaxUncompressedBytes: 64 << 20, + RoundMaxRows: 50000, + BatchPause: 50 * time.Millisecond, + QueryTimeout: 10 * time.Second, + OperationTimeout: 2 * time.Minute, + ObjectPrefix: "wallet-outbox/v1", + COS: OutboxArchiveCOSConfig{ + CredentialSource: "static", + HTTPTimeout: 30 * time.Second, + }, + }, RealtimeOutboxWorker: OutboxWorkerConfig{ Enabled: false, PollInterval: time.Second, @@ -517,6 +578,11 @@ func Load(path string) (Config, error) { if cfg.OutboxWorker.Enabled && !cfg.RocketMQ.WalletOutbox.Enabled { return Config{}, errors.New("outbox_worker requires rocketmq.wallet_outbox.enabled") } + outboxArchive, err := normalizeOutboxArchiveConfig(cfg.OutboxArchive, Default().OutboxArchive) + if err != nil { + return Config{}, err + } + cfg.OutboxArchive = outboxArchive cfg.RealtimeOutboxWorker = normalizeOutboxWorkerConfig(cfg.RealtimeOutboxWorker, Default().RealtimeOutboxWorker) if len(cfg.RealtimeOutboxWorker.EventTypes) == 0 { cfg.RealtimeOutboxWorker.EventTypes = defaultRealtimeOutboxEventTypes() @@ -847,6 +913,104 @@ func normalizeOutboxWorkerConfig(cfg OutboxWorkerConfig, defaults OutboxWorkerCo return cfg } +func normalizeOutboxArchiveConfig(cfg OutboxArchiveConfig, defaults OutboxArchiveConfig) (OutboxArchiveConfig, error) { + cfg.Mode = OutboxArchiveMode(strings.ToLower(strings.TrimSpace(string(cfg.Mode)))) + 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") + } + if cfg.MaxAge <= 0 { + cfg.MaxAge = defaults.MaxAge + } + if cfg.Enabled && cfg.MaxAge < 24*time.Hour { + return OutboxArchiveConfig{}, errors.New("outbox_archive.max_age must be at least 24h") + } + if cfg.PollInterval <= 0 { + cfg.PollInterval = defaults.PollInterval + } + if cfg.QueryBatchSize <= 0 { + cfg.QueryBatchSize = defaults.QueryBatchSize + } + if cfg.QueryBatchSize > 500 { + cfg.QueryBatchSize = 500 + } + if cfg.SegmentMaxRows <= 0 { + cfg.SegmentMaxRows = defaults.SegmentMaxRows + } + if cfg.SegmentMaxRows > 50000 { + cfg.SegmentMaxRows = 50000 + } + if cfg.SegmentMaxUncompressedBytes <= 0 { + cfg.SegmentMaxUncompressedBytes = defaults.SegmentMaxUncompressedBytes + } + if cfg.SegmentMaxUncompressedBytes > 256<<20 { + cfg.SegmentMaxUncompressedBytes = 256 << 20 + } + if cfg.RoundMaxRows <= 0 { + cfg.RoundMaxRows = defaults.RoundMaxRows + } + if cfg.RoundMaxRows < cfg.SegmentMaxRows { + cfg.RoundMaxRows = cfg.SegmentMaxRows + } + if cfg.RoundMaxRows > 250000 { + cfg.RoundMaxRows = 250000 + } + if cfg.BatchPause < 0 { + cfg.BatchPause = defaults.BatchPause + } + if cfg.QueryTimeout <= 0 { + cfg.QueryTimeout = defaults.QueryTimeout + } + if cfg.OperationTimeout <= 0 { + cfg.OperationTimeout = defaults.OperationTimeout + } + cfg.ObjectPrefix = strings.Trim(strings.TrimSpace(cfg.ObjectPrefix), "/") + if cfg.ObjectPrefix == "" { + cfg.ObjectPrefix = defaults.ObjectPrefix + } + cfg.TempDir = strings.TrimSpace(cfg.TempDir) + cfg.COS.SecretID = strings.TrimSpace(cfg.COS.SecretID) + cfg.COS.SecretKey = strings.TrimSpace(cfg.COS.SecretKey) + cfg.COS.SessionToken = strings.TrimSpace(cfg.COS.SessionToken) + cfg.COS.Bucket = strings.TrimSpace(cfg.COS.Bucket) + cfg.COS.Region = strings.TrimSpace(cfg.COS.Region) + cfg.COS.CredentialSource = strings.ToLower(strings.TrimSpace(cfg.COS.CredentialSource)) + if cfg.COS.CredentialSource == "" { + cfg.COS.CredentialSource = defaults.COS.CredentialSource + } + cfg.COS.CAMRoleName = strings.TrimSpace(cfg.COS.CAMRoleName) + if cfg.COS.HTTPTimeout <= 0 { + cfg.COS.HTTPTimeout = defaults.COS.HTTPTimeout + } + if cfg.Enabled && cfg.Mode == OutboxArchiveModeArchiveOnly { + if cfg.COS.Bucket == "" || cfg.COS.Region == "" { + return OutboxArchiveConfig{}, errors.New("outbox_archive archive_only mode requires private COS bucket and region") + } + switch cfg.COS.CredentialSource { + case "static": + if cfg.COS.SecretID == "" || cfg.COS.SecretKey == "" { + return OutboxArchiveConfig{}, errors.New("outbox_archive static COS credential source requires secret_id and secret_key") + } + case "cvm_role": + if cfg.COS.CAMRoleName == "" { + return OutboxArchiveConfig{}, errors.New("outbox_archive cvm_role COS credential source requires cam_role_name") + } + default: + return OutboxArchiveConfig{}, errors.New("outbox_archive.cos.credential_source must be static or cvm_role") + } + } + return cfg, nil +} + 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/storage/mysql/outbox_archive.go b/services/wallet-service/internal/storage/mysql/outbox_archive.go new file mode 100644 index 00000000..f7042a61 --- /dev/null +++ b/services/wallet-service/internal/storage/mysql/outbox_archive.go @@ -0,0 +1,215 @@ +package mysql + +import ( + "context" + "database/sql" + "strings" + "time" + + "hyapp/pkg/appcode" + "hyapp/pkg/xerr" +) + +const walletOutboxArchiveStateVerified = "VERIFIED" + +// WalletOutboxArchiveCursor 是 delivered 时间 + 事件 ID 的稳定复合游标。 +// delivered 后这两列不再变化，因此翻页不会因新写入而重复或跳行。 +type WalletOutboxArchiveCursor struct { + UpdatedAtMS int64 `json:"updated_at_ms"` + EventID string `json:"event_id"` +} + +// WalletOutboxArchiveRecord 保留 wallet_outbox 的完整行快照，便于恢复工具按原始事实校验。 +type WalletOutboxArchiveRecord 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"` +} + +// WalletOutboxArchiveReceipt 只在 COS data/manifest/_SUCCESS 都读回校验后写入。 +// VERIFIED 只代表对象完整性，不代表已完成真实 MySQL 恢复演练。 +type WalletOutboxArchiveReceipt struct { + AppCode string + BatchID string + State string + DataObjectKey string + ManifestObjectKey string + SuccessObjectKey string + SchemaVersion int + RowCount int + FirstCursor WalletOutboxArchiveCursor + LastCursor WalletOutboxArchiveCursor + MinCreatedAtMS int64 + MaxCreatedAtMS int64 + AvailableDeltaTotal string + FrozenDeltaTotal string + NDJSONSHA256 string + GzipSHA256 string + DataCRC64 string + ManifestCRC64 string + SuccessCRC64 string + ManifestSHA256 string + UncompressedBytes int64 + CompressedBytes int64 + VerifiedAtMS int64 +} + +// AcquireWalletOutboxArchiveAppLock 用 MySQL connection-scoped advisory lock 保证多副本下每个 App 只有一个归档者。 +// release 必须 defer 调用；专用 Conn 在释放前不会回到连接池，避免锁跟随错误会话。 +func (r *Repository) AcquireWalletOutboxArchiveAppLock(ctx context.Context) (release func(), acquired bool, err error) { + if r == nil || r.db == nil { + return nil, false, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + lockName := "wallet_outbox_archive:" + appcode.FromContext(ctx) + conn, err := r.db.Conn(ctx) + if err != nil { + return nil, false, err + } + var result sql.NullInt64 + if err := conn.QueryRowContext(ctx, `SELECT GET_LOCK(?, 0)`, lockName).Scan(&result); err != nil { + _ = conn.Close() + return nil, false, err + } + if !result.Valid || result.Int64 != 1 { + _ = conn.Close() + return nil, false, nil + } + release = func() { + releaseCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + var released sql.NullInt64 + // Close 同样会释放 connection-scoped lock；即使显式 release 超时也不能把会话留在连接池。 + _ = conn.QueryRowContext(releaseCtx, `SELECT RELEASE_LOCK(?)`, lockName).Scan(&released) + _ = conn.Close() + } + return release, true, nil +} + +// LatestVerifiedWalletOutboxArchiveCursor 只信任持久化的 VERIFIED 回执，上传中断的对象不会推进游标。 +func (r *Repository) LatestVerifiedWalletOutboxArchiveCursor(ctx context.Context) (WalletOutboxArchiveCursor, error) { + if r == nil || r.db == nil { + return WalletOutboxArchiveCursor{}, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + var cursor WalletOutboxArchiveCursor + err := r.db.QueryRowContext(ctx, ` + SELECT last_updated_at_ms, last_event_id + FROM wallet_outbox_archive_receipts FORCE INDEX (idx_wallet_outbox_archive_cursor) + WHERE app_code = ? AND state = ? + ORDER BY last_updated_at_ms DESC, last_event_id DESC + LIMIT 1`, appcode.FromContext(ctx), walletOutboxArchiveStateVerified).Scan(&cursor.UpdatedAtMS, &cursor.EventID) + if err == sql.ErrNoRows { + return WalletOutboxArchiveCursor{}, nil + } + return cursor, err +} + +// ListDeliveredWalletOutboxArchiveBatch 只读取一个 App 的旧 delivered 行。 +// WHERE 完整命中 (app_code,status,updated_at_ms,event_id) 索引前缀，游标下界和 cutoff 上界把每次 IO 限制在小范围内。 +func (r *Repository) ListDeliveredWalletOutboxArchiveBatch(ctx context.Context, cursor WalletOutboxArchiveCursor, cutoffMS int64, limit int) ([]WalletOutboxArchiveRecord, error) { + if r == nil || r.db == nil { + return nil, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + if cutoffMS <= 0 { + return nil, xerr.New(xerr.InvalidArgument, "archive cutoff_ms must be positive") + } + if limit <= 0 { + limit = 100 + } + if limit > 500 { + limit = 500 + } + rows, err := r.db.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 FORCE INDEX (idx_wallet_outbox_retention) + WHERE app_code = ? AND status = ? + AND updated_at_ms >= ? AND updated_at_ms < ? + AND (updated_at_ms > ? OR (updated_at_ms = ? AND event_id > ?)) + ORDER BY updated_at_ms ASC, event_id ASC + LIMIT ?`, + appcode.FromContext(ctx), outboxStatusDelivered, + cursor.UpdatedAtMS, cutoffMS, + cursor.UpdatedAtMS, cursor.UpdatedAtMS, strings.TrimSpace(cursor.EventID), limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + records := make([]WalletOutboxArchiveRecord, 0, limit) + for rows.Next() { + var record WalletOutboxArchiveRecord + var lockUntilMS, nextRetryAtMS sql.NullInt64 + var lastError sql.NullString + if err := rows.Scan( + &record.AppCode, &record.EventID, &record.EventType, &record.TransactionID, &record.CommandID, + &record.UserID, &record.AssetType, &record.AvailableDelta, &record.FrozenDelta, &record.PayloadJSON, + &record.Status, &record.WorkerID, &lockUntilMS, &record.RetryCount, &nextRetryAtMS, &lastError, + &record.CreatedAtMS, &record.UpdatedAtMS, + ); err != nil { + return nil, err + } + if lockUntilMS.Valid { + value := lockUntilMS.Int64 + record.LockUntilMS = &value + } + if nextRetryAtMS.Valid { + value := nextRetryAtMS.Int64 + record.NextRetryAtMS = &value + } + if lastError.Valid { + value := lastError.String + record.LastError = &value + } + records = append(records, record) + } + return records, rows.Err() +} + +// RecordVerifiedWalletOutboxArchive 是归档链路的最后一步；调用方必须先完成 COS HEAD + GET 读回校验。 +func (r *Repository) RecordVerifiedWalletOutboxArchive(ctx context.Context, receipt WalletOutboxArchiveReceipt) error { + if r == nil || r.db == nil { + return xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + 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 == "" { + return xerr.New(xerr.InvalidArgument, "verified archive receipt is incomplete") + } + _, err := r.db.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, + min_created_at_ms, max_created_at_ms, available_delta_total, frozen_delta_total, + ndjson_sha256, gzip_sha256, data_crc64, manifest_crc64, success_crc64, manifest_sha256, uncompressed_bytes, compressed_bytes, + verified_at_ms, created_at_ms, updated_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + state = VALUES(state), data_object_key = VALUES(data_object_key), manifest_object_key = VALUES(manifest_object_key), + success_object_key = VALUES(success_object_key), verified_at_ms = VALUES(verified_at_ms), updated_at_ms = VALUES(updated_at_ms)`, + receipt.AppCode, receipt.BatchID, receipt.State, receipt.DataObjectKey, receipt.ManifestObjectKey, receipt.SuccessObjectKey, + receipt.SchemaVersion, receipt.RowCount, receipt.FirstCursor.UpdatedAtMS, receipt.FirstCursor.EventID, + receipt.LastCursor.UpdatedAtMS, receipt.LastCursor.EventID, receipt.MinCreatedAtMS, receipt.MaxCreatedAtMS, + receipt.AvailableDeltaTotal, receipt.FrozenDeltaTotal, receipt.NDJSONSHA256, receipt.GzipSHA256, + receipt.DataCRC64, receipt.ManifestCRC64, receipt.SuccessCRC64, + receipt.ManifestSHA256, receipt.UncompressedBytes, receipt.CompressedBytes, + receipt.VerifiedAtMS, receipt.VerifiedAtMS, receipt.VerifiedAtMS, + ) + return err +}