432 lines
17 KiB
Go
432 lines
17 KiB
Go
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
|
||
}
|