108 lines
3.6 KiB
Go
108 lines
3.6 KiB
Go
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
|
||
}
|