diff --git a/scripts/mysql/063_google_paid_sync_retry.sql b/scripts/mysql/063_google_paid_sync_retry.sql new file mode 100644 index 00000000..52a8b7f4 --- /dev/null +++ b/scripts/mysql/063_google_paid_sync_retry.sql @@ -0,0 +1,7 @@ +-- Durable Google paid-detail retry state for wallet-service. +-- Existing paid_synced_at_ms=0 rows become immediately eligible for the 15-second worker. +ALTER TABLE payment_orders + ADD COLUMN paid_sync_attempt_count INT NOT NULL DEFAULT 0 COMMENT 'Google 实付同步连续失败次数' AFTER paid_synced_at_ms, + ADD COLUMN paid_sync_last_error VARCHAR(512) NOT NULL DEFAULT '' COMMENT 'Google 实付同步最近错误' AFTER paid_sync_attempt_count, + ADD COLUMN paid_sync_next_retry_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT 'Google 实付同步下次重试时间，UTC epoch ms' AFTER paid_sync_last_error, + ADD INDEX idx_payment_orders_paid_sync (provider, paid_synced_at_ms, paid_sync_next_retry_at_ms, created_at_ms); diff --git a/server/admin/cmd/server/main.go b/server/admin/cmd/server/main.go index e99e59f0..2999bc0b 100644 --- a/server/admin/cmd/server/main.go +++ b/server/admin/cmd/server/main.go @@ -216,6 +216,13 @@ func main() { } // legacy 账单源的谷歌实付同步依赖 admin 库缓存表，store 建好后再注入；凭证异常只降级该 App，不阻断启动。 wireLegacyGooglePaidSync(financeBillSources, cfg.FinanceBillSources, store) + if cfg.FinanceGooglePaidSync.Enabled { + // Polling is process-owned rather than HTTP-owned: startup immediately drains one batch, then + // every 15 seconds prioritizes newest orders while gradually repairing the complete backlog. + googlePaidSyncWorker := paymentmodule.NewLegacyGooglePaidSyncWorker(financeBillSources, store, cfg.FinanceGooglePaidSync) + googlePaidSyncWorker.Start() + defer googlePaidSyncWorker.Close() + } if *runBootstrap || cfg.Bootstrap.Enabled { if err := store.Seed(cfg); err != nil { fatalRuntime("seed_database_failed", err) diff --git a/server/admin/configs/config.tencent.example.yaml b/server/admin/configs/config.tencent.example.yaml index 48aa95bd..47f9ee8e 100644 --- a/server/admin/configs/config.tencent.example.yaml +++ b/server/admin/configs/config.tencent.example.yaml @@ -55,6 +55,11 @@ finance_bill_sources: google_package_name: "com.chat.auu" google_service_account_json: | {"type":"service_account","project_id":"aslan-494908","private_key_id":"f3ab2df01b3f92b9f71f4b7d1ebdba3cff5ac93e","private_key":"-----BEGIN PRIVATE KEY-----\nMIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQDkCOxtBRVNg5CF\nIfgkT5gv1f2hSRnf6/MSTDYkgXy1sVSbLEXbsZGQdnsYSfAxNXURyJ0QXJOZV551\n+0XOlRIb+HLN++37olcFVm4GPLxQiVCCEbFOBwGPZTMi4RYgriDHEp2umY9Cns8Z\nHYlkYgbLxKLICpzyDJWDsyT3Tc4rBhRGv7+dulxSuPpF5AaLzbV/AjQTwee6qnV4\n8VEPHx/r+3IAJMVLt+VrgFwujIdzJGk8TIZKSSmZ4J6iN5dKgly1PyVu66bppbX9\nHkPDEFg8nxYK5F2/yjo7/VIFurfd5bGKGS3sqf0siRTXLsj0X/wUgt0RlWOpKPeX\nbZq9jzznAgMBAAECggEAASLUPrTMRt8VbLxfFps46GAaC+An21g7FUfA60yj2Onh\nwIYncPFBBuW4NkZEBpK8GxMTST4U1Co+FVtjnSRb+zyxIbqUFHFaGqI0GR7bV1Ff\nz84Two5BYTwBVbamXBJSAnviwjhsoMnWwUrG4POmEgTQRMvcvU33vri5Qewmz0sN\njXTNh/GIUxA+IbOUTaaTDBLSshp7JBOPjuF0ttsd/0zG1AFbaWWtZ1h8YCOhIw+g\np7/CdS3gO/uttgI8PiLj8AImlOcRz5FnFGTcHZqR7pk08W7Olm89ik3NZg+5IozP\nrJ8LOU7uQ7F50zpWRNvpBNaa4m35Ns6bOHTLEV4BXQKBgQD0q8hn1bddwC92RzMw\naBb0Qt0Najg2R/SDWaeSO6Tz3KTflGZXDFtalAOq5KXT47N7QdoIjlXGP50B5iV7\nK/I25f+GONcACPtnnLWyNvjq2OuMz0FIGEoUxY9DFP6aNkyNR5FiE2t2y2XiKmrV\nCRt+P4x/b/KTvdVI+caUSfXezQKBgQDul/F46Uj1evZZToA4XENXRN8oSb4I08cX\nBQK2WUs+KJsT9sWNaT1V5hQ8wXfuN2rhI0kqR0EURY3ycaA41rdLIeYqvAnmR5qC\nSO8HlC2+zQzjSrr7lyt8dcTughW8/9lCf+jANBNYBZrBN2ta3vlQ8eA8rFUThYnC\n8+PFo8MigwKBgFbbGJSL0MFONUsWsXxQpz1k8xYNDBFw78MlM5B87ezH+huIkd/6\n+f8opjinXJrgrVlnIiCBbr+m23TOH6YfDqggc9pRGTng9mZswi+WxjyQbuYYuQL/\n5GSFUXst28gg2IIa0uhvHmoYgH2OM0iXKBRkONsQgZui+zEhwjXoH4lNAoGAKVMJ\n0M5fA52Lg4ZUMO7R/xB/skOrdW3wwqzsflbS8G4qBfgs2URMCk+yW5+KvSi+C0aI\nSplSzUcKwd4qSQ3va0Twz6AH+umV+lDVjbN9hNmRDOEJp7/UGVdwh3ridvy9TYZH\n8tpSK2G1HxgRMQkDl6B9HSUgCySK6shBQB8QEi8CgYBAbUxCJn20RFmwU4bgRASe\ngoDGkN5Dw1SdauuWn+WySTyrsxdxvRcuyy4pQSDBadhnv24/30mza1ToeJeEsyJ/\nJuysBhWBjJKoYVC/qikgS6EuAnqlAKKg8jLIqERmNaFol0Squ03kTvMYx2aGa1dt\nwwMCwljNodGSgZRA3RlwOA==\n-----END PRIVATE KEY-----\n","client_email":"pay-purchase-verify@aslan-494908.iam.gserviceaccount.com","client_id":"100290296879529721706","auth_uri":"https://accounts.google.com/o/oauth2/auth","token_uri":"https://oauth2.googleapis.com/token","auth_provider_x509_cert_url":"https://www.googleapis.com/oauth2/v1/certs","client_x509_cert_url":"https://www.googleapis.com/robot/v1/metadata/x509/pay-purchase-verify%40aslan-494908.iam.gserviceaccount.com","universe_domain":"googleapis.com"} +finance_google_paid_sync: + enabled: true + poll_interval: "15s" + batch_size: 100 + request_timeout: "2m" mysql_auto_migrate: false migrations: enabled: true diff --git a/server/admin/configs/config.yaml b/server/admin/configs/config.yaml index 67a5f5a3..8effb240 100644 --- a/server/admin/configs/config.yaml +++ b/server/admin/configs/config.yaml @@ -55,6 +55,11 @@ finance_bill_sources: google_package_name: "" google_service_account_json: "" google_service_account_file: "" +finance_google_paid_sync: + enabled: true + poll_interval: "15s" + batch_size: 100 + request_timeout: "2m" mysql_auto_migrate: true migrations: enabled: true diff --git a/server/admin/internal/config/config.go b/server/admin/internal/config/config.go index 24bc0638..aa23d4a0 100644 --- a/server/admin/internal/config/config.go +++ b/server/admin/internal/config/config.go @@ -13,17 +13,19 @@ import ( ) type Config struct { - ServiceName string `yaml:"service_name"` - NodeID string `yaml:"node_id"` - Environment string `yaml:"environment"` - Log logging.Config `yaml:"log"` - HTTPAddr string `yaml:"http_addr"` - MySQLDSN string `yaml:"mysql_dsn"` - UserMySQLDSN string `yaml:"user_mysql_dsn"` - WalletMySQLDSN string `yaml:"wallet_mysql_dsn"` - RobotProfileSource RobotProfileSourceConfig `yaml:"robot_profile_source"` - MoneyRegionSources []MoneyRegionSourceConfig `yaml:"money_region_sources"` - FinanceBillSources []FinanceBillSourceConfig `yaml:"finance_bill_sources"` + ServiceName string `yaml:"service_name"` + NodeID string `yaml:"node_id"` + Environment string `yaml:"environment"` + Log logging.Config `yaml:"log"` + HTTPAddr string `yaml:"http_addr"` + MySQLDSN string `yaml:"mysql_dsn"` + UserMySQLDSN string `yaml:"user_mysql_dsn"` + WalletMySQLDSN string `yaml:"wallet_mysql_dsn"` + RobotProfileSource RobotProfileSourceConfig `yaml:"robot_profile_source"` + MoneyRegionSources []MoneyRegionSourceConfig `yaml:"money_region_sources"` + FinanceBillSources []FinanceBillSourceConfig `yaml:"finance_bill_sources"` + // FinanceGooglePaidSync controls the durable polling loop that fills Google Orders API paid details for legacy bill sources. + FinanceGooglePaidSync FinanceGooglePaidSyncConfig `yaml:"finance_google_paid_sync"` MySQLAutoMigrate bool `yaml:"mysql_auto_migrate"` Migrations MigrationConfig `yaml:"migrations"` JWTSecret string `yaml:"jwt_secret"` @@ -173,6 +175,15 @@ type FinanceBillSourceConfig struct { GoogleServiceAccountFile string `yaml:"google_service_account_file"` } +// FinanceGooglePaidSyncConfig keeps legacy Google paid-detail polling independent from HTTP requests. +// New successful Mongo orders are picked every PollInterval; failed orders use persisted retry backoff. +type FinanceGooglePaidSyncConfig struct { + Enabled bool `yaml:"enabled"` + PollInterval time.Duration `yaml:"poll_interval"` + BatchSize int `yaml:"batch_size"` + RequestTimeout time.Duration `yaml:"request_timeout"` +} + type TencentCOSConfig struct { Enabled bool `yaml:"enabled"` SecretID string `yaml:"secret-id"` @@ -361,6 +372,12 @@ func Default() Config { RequestTimeout: 3 * time.Second, }, }, + FinanceGooglePaidSync: FinanceGooglePaidSyncConfig{ + Enabled: true, + PollInterval: 15 * time.Second, + BatchSize: 100, + RequestTimeout: 2 * time.Minute, + }, TencentCOS: TencentCOSConfig{ Enabled: false, ObjectPrefix: "admin", @@ -607,6 +624,19 @@ func (cfg *Config) Normalize() { } } cfg.applyFinanceBillSourceEnvOverrides() + // The sync feature is enabled by default; normalization only bounds operator-provided cadence and batch size. + if cfg.FinanceGooglePaidSync.PollInterval <= 0 { + cfg.FinanceGooglePaidSync.PollInterval = Default().FinanceGooglePaidSync.PollInterval + } + if cfg.FinanceGooglePaidSync.BatchSize <= 0 { + cfg.FinanceGooglePaidSync.BatchSize = Default().FinanceGooglePaidSync.BatchSize + } + if cfg.FinanceGooglePaidSync.BatchSize > 1000 { + cfg.FinanceGooglePaidSync.BatchSize = 1000 + } + if cfg.FinanceGooglePaidSync.RequestTimeout <= 0 { + cfg.FinanceGooglePaidSync.RequestTimeout = Default().FinanceGooglePaidSync.RequestTimeout + } cfg.TencentCOS.SecretID = strings.TrimSpace(cfg.TencentCOS.SecretID) cfg.TencentCOS.SecretKey = strings.TrimSpace(cfg.TencentCOS.SecretKey) cfg.TencentCOS.BucketName = strings.TrimSpace(cfg.TencentCOS.BucketName) diff --git a/server/admin/internal/integration/googleorders/client.go b/server/admin/internal/integration/googleorders/client.go index 86ebc576..6f65b564 100644 --- a/server/admin/internal/integration/googleorders/client.go +++ b/server/admin/internal/integration/googleorders/client.go @@ -121,40 +121,71 @@ func (c *Client) GetOrder(ctx context.Context, orderID string) (Order, error) { if orderID == "" { return Order{}, errors.New("google order id is required") } - token, err := c.token(ctx) - if err != nil { - return Order{}, err - } endpoint := fmt.Sprintf("%s/androidpublisher/v3/applications/%s/orders/%s", defaultAPIBaseURL, url.PathEscape(c.packageName), url.PathEscape(orderID)) - req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + body, err := c.getJSON(ctx, endpoint) if err != nil { return Order{}, err } - req.Header.Set("Authorization", "Bearer "+token) - resp, err := c.httpClient.Do(req) - if err != nil { - return Order{}, errors.New("google play api request failed") - } - defer resp.Body.Close() - body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) - if err != nil { - return Order{}, err - } - if resp.StatusCode < 200 || resp.StatusCode >= 300 { - switch resp.StatusCode { - case http.StatusNotFound, http.StatusBadRequest: - return Order{}, errors.New("google order is not found") - case http.StatusUnauthorized, http.StatusForbidden: - return Order{}, errors.New("google play api is not authorized (需要查看财务数据权限)") - default: - return Order{}, fmt.Errorf("google play api returned %d", resp.StatusCode) - } - } var parsed orderResource if err := json.Unmarshal(body, &parsed); err != nil { return Order{}, err } + return orderFromResource(parsed), nil +} + +// GetOrders uses Orders batchGet for backlog repair: one request can fetch up to 1000 distinct orders. +// The caller falls back to single-order requests if Google rejects the whole batch because of one stale ID. +func (c *Client) GetOrders(ctx context.Context, orderIDs []string) (map[string]Order, error) { + if c == nil { + return nil, errors.New("google orders client is not configured") + } + cleaned := make([]string, 0, len(orderIDs)) + seen := make(map[string]struct{}, len(orderIDs)) + for _, orderID := range orderIDs { + orderID = strings.TrimSpace(orderID) + if orderID == "" { + continue + } + if _, ok := seen[orderID]; ok { + continue + } + seen[orderID] = struct{}{} + cleaned = append(cleaned, orderID) + } + if len(cleaned) == 0 { + return map[string]Order{}, nil + } + if len(cleaned) > 1000 { + return nil, errors.New("too many google order ids") + } + query := url.Values{} + for _, orderID := range cleaned { + query.Add("orderIds", orderID) + } + endpoint := fmt.Sprintf("%s/androidpublisher/v3/applications/%s/orders:batchGet?%s", + defaultAPIBaseURL, url.PathEscape(c.packageName), query.Encode()) + body, err := c.getJSON(ctx, endpoint) + if err != nil { + return nil, err + } + var parsed struct { + Orders []orderResource `json:"orders"` + } + if err := json.Unmarshal(body, &parsed); err != nil { + return nil, err + } + orders := make(map[string]Order, len(parsed.Orders)) + for _, item := range parsed.Orders { + order := orderFromResource(item) + if order.OrderID != "" { + orders[order.OrderID] = order + } + } + return orders, nil +} + +func orderFromResource(parsed orderResource) Order { order := Order{ OrderID: parsed.OrderID, State: parsed.State, @@ -167,7 +198,39 @@ func (c *Client) GetOrder(ctx context.Context, orderID string) (Order, error) { if order.CurrencyCode == "" { order.CurrencyCode = parsed.Tax.CurrencyCode } - return order, nil + return order +} + +func (c *Client) getJSON(ctx context.Context, endpoint string) ([]byte, error) { + token, err := c.token(ctx) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+token) + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, errors.New("google play api request failed") + } + defer resp.Body.Close() + body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) + if err != nil { + return nil, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + switch resp.StatusCode { + case http.StatusNotFound, http.StatusBadRequest: + return nil, errors.New("google order is not found") + case http.StatusUnauthorized, http.StatusForbidden: + return nil, errors.New("google play api is not authorized (需要查看财务数据权限)") + default: + return nil, fmt.Errorf("google play api returned %d", resp.StatusCode) + } + } + return body, nil } func (c *Client) token(ctx context.Context) (string, error) { diff --git a/server/admin/internal/model/models.go b/server/admin/internal/model/models.go index 00bc450c..32a296ae 100644 --- a/server/admin/internal/model/models.go +++ b/server/admin/internal/model/models.go @@ -551,6 +551,22 @@ func (LegacyGooglePaidDetail) TableName() string { return "admin_legacy_google_paid_details" } +// LegacyGooglePaidSyncAttempt persists retry state separately from paid details. +// A row here never means "synced"; successful synchronization deletes it after the paid-detail upsert commits. +type LegacyGooglePaidSyncAttempt struct { + AppCode string `gorm:"size:32;primaryKey" json:"appCode"` + TransactionID string `gorm:"size:96;primaryKey;column:transaction_id" json:"transactionId"` + AttemptCount int `gorm:"not null;default:0" json:"attemptCount"` + LastError string `gorm:"size:512;not null;default:''" json:"lastError"` + NextRetryAtMS int64 `gorm:"column:next_retry_at_ms;not null;default:0;index:idx_legacy_google_paid_retry,priority:2" json:"nextRetryAtMs"` + CreatedAtMS int64 `gorm:"column:created_at_ms;autoCreateTime:milli" json:"createdAtMs"` + UpdatedAtMS int64 `gorm:"column:updated_at_ms;autoUpdateTime:milli;index:idx_legacy_google_paid_retry,priority:3" json:"updatedAtMs"` +} + +func (LegacyGooglePaidSyncAttempt) TableName() string { + return "admin_legacy_google_paid_sync_attempts" +} + type FinanceApplication struct { ID uint `gorm:"primaryKey" json:"id"` AppCode string `gorm:"size:32;index:idx_admin_finance_app_app_status;not null" json:"appCode"` diff --git a/server/admin/internal/modules/payment/google_paid_sync_worker.go b/server/admin/internal/modules/payment/google_paid_sync_worker.go new file mode 100644 index 00000000..379dd420 --- /dev/null +++ b/server/admin/internal/modules/payment/google_paid_sync_worker.go @@ -0,0 +1,141 @@ +package payment + +import ( + "context" + "log/slog" + "strings" + "sync" + "time" + + "hyapp-admin-server/internal/config" +) + +type legacyGooglePaidSyncSource interface { + AppCode() string + SupportsGooglePaidSync() bool + ListPendingGooglePaidTransactionIDs(ctx context.Context, limit int) ([]string, error) + RefreshGooglePaidDetails(ctx context.Context, transactionIDs []string) ([]googleRechargePaidDTO, error) +} + +type legacyGooglePaidSyncAttemptStore interface { + RecordLegacyGooglePaidSyncFailure(appCode string, transactionID string, message string, nowMS int64) error + ClearLegacyGooglePaidSyncFailure(appCode string, transactionID string) error +} + +// LegacyGooglePaidSyncWorker polls every 15 seconds independently of finance-page traffic. +// Successful rows leave the pending set through the paid cache; failures remain durable with backoff. +type LegacyGooglePaidSyncWorker struct { + batchSize int + pollInterval time.Duration + requestTimeout time.Duration + sources []legacyGooglePaidSyncSource + store legacyGooglePaidSyncAttemptStore + + closeOnce sync.Once + done chan struct{} + stop chan struct{} +} + +func NewLegacyGooglePaidSyncWorker(sources []RechargeBillSource, store legacyGooglePaidSyncAttemptStore, cfg config.FinanceGooglePaidSyncConfig) *LegacyGooglePaidSyncWorker { + narrow := make([]legacyGooglePaidSyncSource, 0, len(sources)) + for _, source := range sources { + if candidate, ok := source.(legacyGooglePaidSyncSource); ok && candidate.SupportsGooglePaidSync() { + narrow = append(narrow, candidate) + } + } + return newLegacyGooglePaidSyncWorker(narrow, store, cfg) +} + +func newLegacyGooglePaidSyncWorker(sources []legacyGooglePaidSyncSource, store legacyGooglePaidSyncAttemptStore, cfg config.FinanceGooglePaidSyncConfig) *LegacyGooglePaidSyncWorker { + return &LegacyGooglePaidSyncWorker{ + batchSize: cfg.BatchSize, + pollInterval: cfg.PollInterval, + requestTimeout: cfg.RequestTimeout, + sources: sources, + store: store, + done: make(chan struct{}), + stop: make(chan struct{}), + } +} + +func (w *LegacyGooglePaidSyncWorker) Start() { + if w == nil || w.store == nil || len(w.sources) == 0 { + return + } + go w.loop() +} + +func (w *LegacyGooglePaidSyncWorker) Close() { + if w == nil || w.store == nil || len(w.sources) == 0 { + return + } + w.closeOnce.Do(func() { + close(w.stop) + <-w.done + }) +} + +func (w *LegacyGooglePaidSyncWorker) loop() { + defer close(w.done) + ticker := time.NewTicker(w.pollInterval) + defer ticker.Stop() + w.runOnce(context.Background()) + for { + select { + case <-ticker.C: + w.runOnce(context.Background()) + case <-w.stop: + return + } + } +} + +func (w *LegacyGooglePaidSyncWorker) runOnce(parent context.Context) { + for _, source := range w.sources { + // Ten full batches per tick drain historical data quickly without turning the steady-state + // 15-second poller into an unbounded loop when a provider keeps returning pathological data. + for drainBatch := 0; drainBatch < 10; drainBatch++ { + ctx, cancel := context.WithTimeout(parent, w.requestTimeout) + transactionIDs, err := source.ListPendingGooglePaidTransactionIDs(ctx, w.batchSize) + if err != nil { + cancel() + slog.Error("legacy_google_paid_sync_scan_failed", "app_code", source.AppCode(), "error", err.Error()) + break + } + if len(transactionIDs) == 0 { + cancel() + break + } + results, err := source.RefreshGooglePaidDetails(ctx, transactionIDs) + cancel() + if err != nil { + slog.Error("legacy_google_paid_sync_batch_failed", "app_code", source.AppCode(), "count", len(transactionIDs), "error", err.Error()) + break + } + nowMS := time.Now().UnixMilli() + succeeded := 0 + failed := 0 + for _, result := range results { + transactionID := strings.TrimSpace(result.TransactionID) + if transactionID == "" { + continue + } + if strings.TrimSpace(result.Error) == "" { + if err := w.store.ClearLegacyGooglePaidSyncFailure(source.AppCode(), transactionID); err != nil { + slog.Error("legacy_google_paid_sync_retry_clear_failed", "app_code", source.AppCode(), "transaction_id", transactionID, "error", err.Error()) + } + succeeded++ + continue + } + failed++ + if err := w.store.RecordLegacyGooglePaidSyncFailure(source.AppCode(), transactionID, result.Error, nowMS); err != nil { + slog.Error("legacy_google_paid_sync_retry_store_failed", "app_code", source.AppCode(), "transaction_id", transactionID, "error", err.Error()) + } + } + slog.Info("legacy_google_paid_sync_batch_completed", "app_code", source.AppCode(), "claimed", len(transactionIDs), "succeeded", succeeded, "failed", failed) + if len(transactionIDs) < w.batchSize { + break + } + } + } +} diff --git a/server/admin/internal/modules/payment/google_paid_sync_worker_test.go b/server/admin/internal/modules/payment/google_paid_sync_worker_test.go new file mode 100644 index 00000000..a03de820 --- /dev/null +++ b/server/admin/internal/modules/payment/google_paid_sync_worker_test.go @@ -0,0 +1,59 @@ +package payment + +import ( + "context" + "testing" + "time" + + "hyapp-admin-server/internal/config" +) + +type fakeLegacyGooglePaidSyncSource struct { + pending []string + results []googleRechargePaidDTO +} + +func (s *fakeLegacyGooglePaidSyncSource) AppCode() string { return "yumi" } +func (s *fakeLegacyGooglePaidSyncSource) SupportsGooglePaidSync() bool { return true } +func (s *fakeLegacyGooglePaidSyncSource) ListPendingGooglePaidTransactionIDs(context.Context, int) ([]string, error) { + return append([]string(nil), s.pending...), nil +} +func (s *fakeLegacyGooglePaidSyncSource) RefreshGooglePaidDetails(context.Context, []string) ([]googleRechargePaidDTO, error) { + return append([]googleRechargePaidDTO(nil), s.results...), nil +} + +type fakeLegacyGooglePaidAttemptStore struct { + cleared []string + failed []string +} + +func (s *fakeLegacyGooglePaidAttemptStore) RecordLegacyGooglePaidSyncFailure(_ string, transactionID string, _ string, _ int64) error { + s.failed = append(s.failed, transactionID) + return nil +} +func (s *fakeLegacyGooglePaidAttemptStore) ClearLegacyGooglePaidSyncFailure(_ string, transactionID string) error { + s.cleared = append(s.cleared, transactionID) + return nil +} + +func TestLegacyGooglePaidSyncWorkerPersistsPerOrderOutcome(t *testing.T) { + source := &fakeLegacyGooglePaidSyncSource{ + pending: []string{"ok-1", "bad-1"}, + results: []googleRechargePaidDTO{{TransactionID: "ok-1", SyncedAtMS: 1}, {TransactionID: "bad-1", Error: "temporary"}}, + } + store := &fakeLegacyGooglePaidAttemptStore{} + worker := newLegacyGooglePaidSyncWorker([]legacyGooglePaidSyncSource{source}, store, config.FinanceGooglePaidSyncConfig{ + PollInterval: 15 * time.Second, + BatchSize: 100, + RequestTimeout: time.Second, + }) + + worker.runOnce(context.Background()) + + if len(store.cleared) != 1 || store.cleared[0] != "ok-1" { + t.Fatalf("cleared = %v, want ok-1", store.cleared) + } + if len(store.failed) != 1 || store.failed[0] != "bad-1" { + t.Fatalf("failed = %v, want bad-1", store.failed) + } +} diff --git a/server/admin/internal/modules/payment/legacy_bill_source.go b/server/admin/internal/modules/payment/legacy_bill_source.go index cafbd35c..1fa20c4e 100644 --- a/server/admin/internal/modules/payment/legacy_bill_source.go +++ b/server/admin/internal/modules/payment/legacy_bill_source.go @@ -9,6 +9,7 @@ import ( "sort" "strconv" "strings" + "sync" "time" "hyapp-admin-server/internal/appctx" @@ -20,6 +21,7 @@ import ( "go.mongodb.org/mongo-driver/v2/bson" "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" ) // legacyRechargeBillQuery 是 legacy 账单源支持的筛选子集；用户/币商筛选属于 hyapp 语义，legacy 源不适用。 @@ -48,6 +50,8 @@ type RechargeBillSource interface { SummarizeRechargeBills(ctx context.Context, query legacyRechargeBillQuery) (rechargeBillSummaryDTO, error) // SupportsGooglePaidSync 为 true 时，RefreshGooglePaidDetails 用该 App 的 Play Console 服务账号同步实付明细。 SupportsGooglePaidSync() bool + // ListPendingGooglePaidTransactionIDs returns newest successful Google bills that still need paid-detail synchronization. + ListPendingGooglePaidTransactionIDs(ctx context.Context, limit int) ([]string, error) RefreshGooglePaidDetails(ctx context.Context, transactionIDs []string) ([]googleRechargePaidDTO, error) // Overview 返回财务概览聚合：按日趋势、区域分布与谷歌实付覆盖度。 Overview(ctx context.Context, query legacyRechargeBillQuery, tzOffsetMinutes int32) (rechargeBillOverviewDTO, error) @@ -64,6 +68,7 @@ type LegacyGooglePaidStore interface { LegacyGooglePaidDetails(appCode string, transactionIDs []string) (map[string]model.LegacyGooglePaidDetail, error) LegacyGooglePaidStatsByBillTime(appCode string, startAtMS int64, endAtMS int64) (repository.LegacyGooglePaidStats, error) LegacyGooglePaidTransactionIDs(appCode string) ([]string, error) + LegacyGooglePaidRetryBlockedTransactionIDs(appCode string, nowMS int64) ([]string, error) } // legacyGoogleOrdersClient 按 legacy App 各自的 Play Console 服务账号查订单实付。 @@ -71,6 +76,10 @@ type legacyGoogleOrdersClient interface { GetOrder(ctx context.Context, orderID string) (googleorders.Order, error) } +type legacyGoogleOrdersBatchClient interface { + GetOrders(ctx context.Context, orderIDs []string) (map[string]googleorders.Order, error) +} + // errLegacyGooglePaidUnsupported 表示该账单源未配置谷歌包名/服务账号或实付缓存存储。 var errLegacyGooglePaidUnsupported = errors.New("legacy google paid sync is not configured") @@ -143,6 +152,83 @@ func (s *MongoRechargeBillSource) SupportsGooglePaidSync() bool { return s != nil && s.paidStore != nil && s.googleOrders != nil } +// ListPendingGooglePaidTransactionIDs scans the read-only legacy Mongo source and subtracts both +// successful cache rows and failures still inside persisted backoff. Sorting newest-first keeps the +// 15-second loop current while successful removals naturally drain the complete historical backlog. +func (s *MongoRechargeBillSource) ListPendingGooglePaidTransactionIDs(ctx context.Context, limit int) ([]string, error) { + if !s.SupportsGooglePaidSync() { + return nil, errLegacyGooglePaidUnsupported + } + if limit <= 0 { + return []string{}, nil + } + if limit > 1000 { + limit = 1000 + } + filter, _, matchable, err := s.billFilter(ctx, legacyRechargeBillQuery{RechargeType: "google_play_recharge"}) + if err != nil || !matchable { + return nil, err + } + syncedIDs, err := s.paidStore.LegacyGooglePaidTransactionIDs(s.config.AppCode) + if err != nil { + return nil, fmt.Errorf("load synced legacy google transactions for %s: %w", s.config.AppCode, err) + } + blockedIDs, err := s.paidStore.LegacyGooglePaidRetryBlockedTransactionIDs(s.config.AppCode, time.Now().UnixMilli()) + if err != nil { + return nil, fmt.Errorf("load blocked legacy google transactions for %s: %w", s.config.AppCode, err) + } + excluded := append(syncedIDs, blockedIDs...) + if values := legacyTransactionIDFilterValues(excluded); len(values) > 0 { + filter["_id"] = bson.M{"$nin": values} + } + cursor, err := s.collection.Find(ctx, filter, options.Find(). + SetProjection(bson.M{"_id": 1}). + SetSort(bson.D{{Key: "createTime", Value: -1}, {Key: "_id", Value: -1}}). + SetLimit(int64(limit))) + if err != nil { + return nil, fmt.Errorf("list pending legacy google bills for %s: %w", s.config.AppCode, err) + } + defer cursor.Close(ctx) + ids := make([]string, 0, limit) + for cursor.Next(ctx) { + var row struct { + ID any `bson:"_id"` + } + if err := cursor.Decode(&row); err != nil { + return nil, fmt.Errorf("decode pending legacy google bill for %s: %w", s.config.AppCode, err) + } + if transactionID := legacyDocumentID(row.ID); transactionID != "" { + ids = append(ids, transactionID) + } + } + if err := cursor.Err(); err != nil { + return nil, fmt.Errorf("iterate pending legacy google bills for %s: %w", s.config.AppCode, err) + } + return ids, nil +} + +// legacyTransactionIDFilterValues covers both string IDs and ObjectID-backed collections. +// The paid cache stores a stable string representation, so both BSON shapes must be excluded. +func legacyTransactionIDFilterValues(transactionIDs []string) bson.A { + values := make(bson.A, 0, len(transactionIDs)*2) + seen := make(map[string]struct{}, len(transactionIDs)) + for _, transactionID := range transactionIDs { + transactionID = strings.TrimSpace(transactionID) + if transactionID == "" { + continue + } + if _, ok := seen[transactionID]; ok { + continue + } + seen[transactionID] = struct{}{} + values = append(values, transactionID) + if objectID, err := bson.ObjectIDFromHex(transactionID); err == nil { + values = append(values, objectID) + } + } + return values +} + func (s *MongoRechargeBillSource) AppCode() string { return s.config.AppCode } @@ -503,14 +589,59 @@ func (s *MongoRechargeBillSource) RefreshGooglePaidDetails(ctx context.Context, return nil, fmt.Errorf("iterate legacy bills for google paid refresh: %w", err) } - results := make([]googleRechargePaidDTO, 0, len(cleaned)) - for _, transactionID := range cleaned { - results = append(results, s.refreshGooglePaidDetail(ctx, transactionID, bills)) + // Backfills use batchGet whenever every order can be resolved in one Google request. Google rejects + // an entire batch containing one stale ID, so failures fall back to bounded concurrent single gets. + prefetched := map[string]googleorders.Order{} + batchReady := false + if batchClient, ok := s.googleOrders.(legacyGoogleOrdersBatchClient); ok { + orderIDs := make([]string, 0, len(bills)) + for _, transactionID := range cleaned { + if document, exists := bills[transactionID]; exists { + if orderID := strings.TrimSpace(document.OrderID); orderID != "" { + orderIDs = append(orderIDs, orderID) + } + } + } + if len(orderIDs) > 0 { + if orders, batchErr := batchClient.GetOrders(ctx, orderIDs); batchErr == nil { + prefetched = orders + batchReady = true + } + } } + results := make([]googleRechargePaidDTO, len(cleaned)) + if batchReady { + for index, transactionID := range cleaned { + results[index] = s.refreshGooglePaidDetailWithOrders(ctx, transactionID, bills, prefetched) + } + return results, nil + } + const fallbackConcurrency = 8 + semaphore := make(chan struct{}, fallbackConcurrency) + var wait sync.WaitGroup + for index, transactionID := range cleaned { + wait.Add(1) + go func(index int, transactionID string) { + defer wait.Done() + select { + case semaphore <- struct{}{}: + defer func() { <-semaphore }() + case <-ctx.Done(): + results[index] = googleRechargePaidDTO{TransactionID: transactionID, Error: ctx.Err().Error()} + return + } + results[index] = s.refreshGooglePaidDetail(ctx, transactionID, bills) + }(index, transactionID) + } + wait.Wait() return results, nil } func (s *MongoRechargeBillSource) refreshGooglePaidDetail(ctx context.Context, transactionID string, bills map[string]legacyPurchaseDocument) googleRechargePaidDTO { + return s.refreshGooglePaidDetailWithOrders(ctx, transactionID, bills, nil) +} + +func (s *MongoRechargeBillSource) refreshGooglePaidDetailWithOrders(ctx context.Context, transactionID string, bills map[string]legacyPurchaseDocument, prefetched map[string]googleorders.Order) googleRechargePaidDTO { result := googleRechargePaidDTO{TransactionID: transactionID} document, ok := bills[transactionID] if !ok { @@ -526,11 +657,19 @@ func (s *MongoRechargeBillSource) refreshGooglePaidDetail(ctx context.Context, t result.Error = "缺少谷歌订单号" return result } - order, err := s.googleOrders.GetOrder(ctx, orderID) - if err != nil { - result.Error = err.Error() + order, prefetchedOK := prefetched[orderID] + if prefetched != nil && !prefetchedOK { + result.Error = "google order is not found" return result } + if !prefetchedOK { + var err error + order, err = s.googleOrders.GetOrder(ctx, orderID) + if err != nil { + result.Error = err.Error() + return result + } + } syncedAt := time.Now().UnixMilli() billCreatedAtMS := int64(0) if !document.CreateTime.IsZero() { @@ -703,11 +842,7 @@ func (s *MongoRechargeBillSource) billFilter(ctx context.Context, query legacyRe if err != nil { return nil, nil, false, fmt.Errorf("load synced legacy google tx ids for %s: %w", s.config.AppCode, err) } - if len(syncedIDs) > 0 { - idValues := make(bson.A, 0, len(syncedIDs)) - for _, transactionID := range syncedIDs { - idValues = append(idValues, transactionID) - } + if idValues := legacyTransactionIDFilterValues(syncedIDs); len(idValues) > 0 { filter["_id"] = bson.M{"$nin": idValues} } } diff --git a/server/admin/internal/repository/legacy_google_paid_repository.go b/server/admin/internal/repository/legacy_google_paid_repository.go index 1705cfe6..325250d1 100644 --- a/server/admin/internal/repository/legacy_google_paid_repository.go +++ b/server/admin/internal/repository/legacy_google_paid_repository.go @@ -2,12 +2,18 @@ package repository import ( "strings" + "time" "hyapp-admin-server/internal/model" "gorm.io/gorm/clause" ) +const ( + legacyGooglePaidRetryBaseMS = int64((15 * time.Second) / time.Millisecond) + legacyGooglePaidRetryMaxMS = int64((15 * time.Minute) / time.Millisecond) +) + // UpsertLegacyGooglePaidDetail 写入或覆盖 legacy 谷歌账单实付明细；同一账单重复同步以最新结果为准。 func (s *Store) UpsertLegacyGooglePaidDetail(detail model.LegacyGooglePaidDetail) error { detail.AppCode = strings.ToLower(strings.TrimSpace(detail.AppCode)) @@ -67,6 +73,47 @@ func (s *Store) LegacyGooglePaidTransactionIDs(appCode string) ([]string, error) return ids, err } +// LegacyGooglePaidRetryBlockedTransactionIDs returns failures whose persisted backoff has not expired. +// Due attempts are intentionally omitted so the next 15-second scan can retry them without manual intervention. +func (s *Store) LegacyGooglePaidRetryBlockedTransactionIDs(appCode string, nowMS int64) ([]string, error) { + ids := []string{} + err := s.db.Model(&model.LegacyGooglePaidSyncAttempt{}). + Where("app_code = ? AND next_retry_at_ms > ?", strings.ToLower(strings.TrimSpace(appCode)), nowMS). + Pluck("transaction_id", &ids).Error + return ids, err +} + +// RecordLegacyGooglePaidSyncFailure atomically advances retry state with exponential backoff. +// The error text is bounded before persistence so provider responses cannot inflate the admin database. +func (s *Store) RecordLegacyGooglePaidSyncFailure(appCode string, transactionID string, message string, nowMS int64) error { + appCode = strings.ToLower(strings.TrimSpace(appCode)) + transactionID = strings.TrimSpace(transactionID) + message = strings.TrimSpace(message) + if len(message) > 512 { + message = message[:512] + } + return s.db.Exec(` + INSERT INTO admin_legacy_google_paid_sync_attempts ( + app_code, transaction_id, attempt_count, last_error, next_retry_at_ms, created_at_ms, updated_at_ms + ) VALUES (?, ?, 1, ?, ?, ?, ?) + ON DUPLICATE KEY UPDATE + last_error = VALUES(last_error), + next_retry_at_ms = VALUES(updated_at_ms) + LEAST(?, ? * POW(2, LEAST(attempt_count, 6))), + attempt_count = attempt_count + 1, + updated_at_ms = VALUES(updated_at_ms)`, + appCode, transactionID, message, nowMS+legacyGooglePaidRetryBaseMS, nowMS, nowMS, + legacyGooglePaidRetryMaxMS, legacyGooglePaidRetryBaseMS, + ).Error +} + +// ClearLegacyGooglePaidSyncFailure removes obsolete retry state after the paid-detail upsert succeeds. +func (s *Store) ClearLegacyGooglePaidSyncFailure(appCode string, transactionID string) error { + return s.db.Where( + "app_code = ? AND transaction_id = ?", + strings.ToLower(strings.TrimSpace(appCode)), strings.TrimSpace(transactionID), + ).Delete(&model.LegacyGooglePaidSyncAttempt{}).Error +} + // LegacyGooglePaidDetails 按交易号批量读取实付明细，供财务明细列表按页回填。 func (s *Store) LegacyGooglePaidDetails(appCode string, transactionIDs []string) (map[string]model.LegacyGooglePaidDetail, error) { details := map[string]model.LegacyGooglePaidDetail{} diff --git a/server/admin/internal/repository/repository.go b/server/admin/internal/repository/repository.go index 9bbfe196..c1e62a6c 100644 --- a/server/admin/internal/repository/repository.go +++ b/server/admin/internal/repository/repository.go @@ -87,6 +87,7 @@ func (s *Store) AutoMigrate() error { &model.UserMoneyScope{}, &model.TemporaryPaymentLinkOwner{}, &model.LegacyGooglePaidDetail{}, + &model.LegacyGooglePaidSyncAttempt{}, &model.FinanceApplication{}, &model.CoinSellerRechargeOrder{}, &model.UserWithdrawalApplication{}, diff --git a/server/admin/migrations/087_legacy_google_paid_sync_attempts.sql b/server/admin/migrations/087_legacy_google_paid_sync_attempts.sql new file mode 100644 index 00000000..c4ed8d61 --- /dev/null +++ b/server/admin/migrations/087_legacy_google_paid_sync_attempts.sql @@ -0,0 +1,15 @@ +SET NAMES utf8mb4 COLLATE utf8mb4_unicode_ci; + +-- Google paid-detail failures are not paid facts. Keep retry state in a separate table so the +-- finance list/overview can continue treating admin_legacy_google_paid_details as successful cache only. +CREATE TABLE IF NOT EXISTS admin_legacy_google_paid_sync_attempts ( + app_code VARCHAR(32) NOT NULL COMMENT 'legacy App code', + transaction_id VARCHAR(96) NOT NULL COMMENT 'in_app_purchase_details._id', + attempt_count INT NOT NULL DEFAULT 0 COMMENT 'consecutive failed attempts', + last_error VARCHAR(512) NOT NULL DEFAULT '' COMMENT 'last sanitized synchronization error', + next_retry_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT 'next retry time, UTC epoch ms', + created_at_ms BIGINT NOT NULL COMMENT 'created time, UTC epoch ms', + updated_at_ms BIGINT NOT NULL COMMENT 'updated time, UTC epoch ms', + PRIMARY KEY (app_code, transaction_id), + KEY idx_legacy_google_paid_retry (app_code, next_retry_at_ms, updated_at_ms) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='legacy Google paid-detail retry state'; diff --git a/services/wallet-service/configs/config.docker.yaml b/services/wallet-service/configs/config.docker.yaml index 85c80fed..c24bc4db 100644 --- a/services/wallet-service/configs/config.docker.yaml +++ b/services/wallet-service/configs/config.docker.yaml @@ -51,7 +51,11 @@ google_play: token_url: "https://oauth2.googleapis.com/token" http_timeout: "10s" consume_enabled: true - +google_paid_sync_worker: + enabled: true + poll_interval: "15s" + batch_size: 100 + request_timeout: "2m" mifapay: # MiFaPay 是 H5 统一三方支付入口；商家公钥只用于记录商户侧配置，真实下单签名必须继续使用商户私钥。 enabled: true diff --git a/services/wallet-service/configs/config.tencent.example.yaml b/services/wallet-service/configs/config.tencent.example.yaml index a5a02780..bb21a836 100644 --- a/services/wallet-service/configs/config.tencent.example.yaml +++ b/services/wallet-service/configs/config.tencent.example.yaml @@ -51,6 +51,11 @@ google_play: token_url: "https://oauth2.googleapis.com/token" http_timeout: "10s" consume_enabled: true +google_paid_sync_worker: + enabled: true + poll_interval: "15s" + batch_size: 100 + request_timeout: "2m" mifapay: # MiFaPay 是 H5 统一三方支付入口；商家公钥只用于记录商户侧配置，真实下单签名必须继续使用商户私钥。 enabled: true diff --git a/services/wallet-service/configs/config.yaml b/services/wallet-service/configs/config.yaml index c6a2179a..7ea71547 100644 --- a/services/wallet-service/configs/config.yaml +++ b/services/wallet-service/configs/config.yaml @@ -51,6 +51,11 @@ google_play: token_url: "https://oauth2.googleapis.com/token" http_timeout: "10s" consume_enabled: true +google_paid_sync_worker: + enabled: true + poll_interval: "15s" + batch_size: 100 + request_timeout: "2m" mifapay: # MiFaPay 是 H5 统一三方支付入口；商家公钥只用于记录商户侧配置，真实下单签名必须继续使用商户私钥。 enabled: true 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 f39e6c27..dc4d2500 100644 --- a/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql +++ b/services/wallet-service/deploy/mysql/initdb/001_wallet_service.sql @@ -887,6 +887,9 @@ CREATE TABLE IF NOT EXISTS payment_orders ( paid_tax_micro BIGINT NOT NULL DEFAULT 0 COMMENT '订单税额微单位（实付币种）', paid_net_micro BIGINT NOT NULL DEFAULT 0 COMMENT 'Google 扣费后开发者净收入微单位（实付币种）', paid_synced_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '实付明细同步时间，UTC epoch ms；0 表示未同步', + paid_sync_attempt_count INT NOT NULL DEFAULT 0 COMMENT 'Google 实付同步连续失败次数', + paid_sync_last_error VARCHAR(512) NOT NULL DEFAULT '' COMMENT 'Google 实付同步最近错误', + paid_sync_next_retry_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT 'Google 实付同步下次重试时间，UTC epoch ms', provider_payload JSON NULL COMMENT 'Google Play 校验返回快照', created_at_ms BIGINT NOT NULL COMMENT '创建时间，UTC epoch ms', updated_at_ms BIGINT NOT NULL COMMENT '更新时间，UTC epoch ms', @@ -895,7 +898,8 @@ CREATE TABLE IF NOT EXISTS payment_orders ( UNIQUE KEY uk_payment_orders_command (app_code, command_id), KEY idx_payment_orders_user_time (app_code, user_id, created_at_ms), KEY idx_payment_orders_provider_order (app_code, provider, provider_order_id), - KEY idx_payment_orders_wallet_tx (app_code, wallet_transaction_id) + KEY idx_payment_orders_wallet_tx (app_code, wallet_transaction_id), + KEY idx_payment_orders_paid_sync (provider, paid_synced_at_ms, paid_sync_next_retry_at_ms, created_at_ms) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='支付订单与 provider 审计表'; CREATE TABLE IF NOT EXISTS coin_seller_stock_records ( diff --git a/services/wallet-service/internal/app/app.go b/services/wallet-service/internal/app/app.go index b9aea1a8..e7bfd5f0 100644 --- a/services/wallet-service/internal/app/app.go +++ b/services/wallet-service/internal/app/app.go @@ -44,6 +44,7 @@ type App struct { outboxWorkerCfg config.OutboxWorkerConfig realtimeOutboxWorkerCfg config.OutboxWorkerConfig projectionWorkerCfg config.ProjectionWorkerConfig + googlePaidSyncWorkerCfg config.GooglePaidSyncWorkerConfig externalRechargeReconcileWorkerCfg config.ExternalRechargeReconcileWorkerConfig walletOutboxTopic string realtimeWalletOutboxTopic string @@ -185,6 +186,9 @@ func New(cfg config.Config) (*App, error) { _ = repository.Close() return nil, err } + googlePaidSyncWorkerCfg := cfg.GooglePaidSyncWorker + // A disabled Google client cannot make progress; suppress the worker instead of emitting a permanent error loop in local environments. + googlePaidSyncWorkerCfg.Enabled = googlePaidSyncWorkerCfg.Enabled && cfg.GooglePlay.Enabled return &App{ server: server, @@ -200,6 +204,7 @@ func New(cfg config.Config) (*App, error) { outboxWorkerCfg: cfg.OutboxWorker, realtimeOutboxWorkerCfg: cfg.RealtimeOutboxWorker, projectionWorkerCfg: cfg.ProjectionWorker, + googlePaidSyncWorkerCfg: googlePaidSyncWorkerCfg, externalRechargeReconcileWorkerCfg: cfg.ExternalRechargeReconcileWorker, walletOutboxTopic: cfg.RocketMQ.WalletOutbox.Topic, realtimeWalletOutboxTopic: cfg.RocketMQ.RealtimeOutbox.Topic, diff --git a/services/wallet-service/internal/app/background_workers.go b/services/wallet-service/internal/app/background_workers.go index e8ec4ccb..c304d0d9 100644 --- a/services/wallet-service/internal/app/background_workers.go +++ b/services/wallet-service/internal/app/background_workers.go @@ -22,6 +22,9 @@ func (a *App) runBackgroundWorkers() { if a.externalRechargeReconcileWorkerCfg.Enabled { a.startExternalRechargeReconcileWorker(ctx) } + if a.googlePaidSyncWorkerCfg.Enabled { + a.startGooglePaidSyncWorker(ctx) + } if a.redPacketExpiryWorkerCfg.Enabled { a.startRedPacketExpiryWorker(ctx) } diff --git a/services/wallet-service/internal/app/google_paid_sync_worker.go b/services/wallet-service/internal/app/google_paid_sync_worker.go new file mode 100644 index 00000000..31de1af2 --- /dev/null +++ b/services/wallet-service/internal/app/google_paid_sync_worker.go @@ -0,0 +1,45 @@ +package app + +import ( + "context" + "errors" + "log/slog" + "time" + + "hyapp/pkg/logx" +) + +func (a *App) startGooglePaidSyncWorker(ctx context.Context) { + workerID := "google-paid-sync-" + a.nodeID + a.workers.Add(1) + go func() { + defer a.workers.Done() + ticker := time.NewTicker(a.googlePaidSyncWorkerCfg.PollInterval) + defer ticker.Stop() + for { + requestCtx, cancel := context.WithTimeout(ctx, a.googlePaidSyncWorkerCfg.RequestTimeout) + result, err := a.walletSvc.ProcessGooglePaymentPaidSyncBatch(requestCtx, a.googlePaidSyncWorkerCfg.BatchSize) + cancel() + if err != nil && !errors.Is(err, context.Canceled) { + logx.Error(ctx, "google_paid_sync_batch_failed", err, slog.String("worker_id", workerID)) + } + if result.Claimed > 0 { + logx.Info(ctx, "google_paid_sync_batch_completed", + slog.String("worker_id", workerID), + slog.Int("claimed", result.Claimed), + slog.Int("succeeded", result.Succeeded), + slog.Int("failed", result.Failed), + ) + } + // A full batch means historical backlog remains. Drain immediately; once caught up, poll every 15 seconds. + if err == nil && result.Claimed >= a.googlePaidSyncWorkerCfg.BatchSize { + continue + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() +} diff --git a/services/wallet-service/internal/client/googleplay/client.go b/services/wallet-service/internal/client/googleplay/client.go index 8326d0c5..0b997abb 100644 --- a/services/wallet-service/internal/client/googleplay/client.go +++ b/services/wallet-service/internal/client/googleplay/client.go @@ -155,6 +155,62 @@ func (c *Client) GetOrder(ctx context.Context, packageName string, orderID strin if err := json.Unmarshal(body, &parsed); err != nil { return ledger.GooglePlayOrder{}, err } + return googlePlayOrderFromResource(parsed, string(body)), nil +} + +// GetOrders fetches up to 1000 paid-detail snapshots with one Orders batchGet request. +// Service code falls back to single gets when Google rejects a mixed batch because one ID is stale. +func (c *Client) GetOrders(ctx context.Context, packageName string, orderIDs []string) (map[string]ledger.GooglePlayOrder, error) { + if c == nil { + return nil, xerr.New(xerr.Unavailable, "google play client is not configured") + } + packageName = c.normalizePackageName(packageName) + cleaned := make([]string, 0, len(orderIDs)) + seen := make(map[string]struct{}, len(orderIDs)) + for _, orderID := range orderIDs { + orderID = strings.TrimSpace(orderID) + if orderID == "" { + continue + } + if _, ok := seen[orderID]; ok { + continue + } + seen[orderID] = struct{}{} + cleaned = append(cleaned, orderID) + } + if packageName == "" || len(cleaned) == 0 { + return map[string]ledger.GooglePlayOrder{}, nil + } + if len(cleaned) > 1000 { + return nil, xerr.New(xerr.InvalidArgument, "too many google order ids") + } + query := url.Values{} + for _, orderID := range cleaned { + query.Add("orderIds", orderID) + } + endpoint := fmt.Sprintf("%s/androidpublisher/v3/applications/%s/orders:batchGet?%s", + c.apiBaseURL, url.PathEscape(packageName), query.Encode()) + body, err := c.doJSON(ctx, http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + var parsed struct { + Orders []orderResource `json:"orders"` + } + if err := json.Unmarshal(body, &parsed); err != nil { + return nil, err + } + orders := make(map[string]ledger.GooglePlayOrder, len(parsed.Orders)) + for _, item := range parsed.Orders { + order := googlePlayOrderFromResource(item, "") + if order.OrderID != "" { + orders[order.OrderID] = order + } + } + return orders, nil +} + +func googlePlayOrderFromResource(parsed orderResource, rawJSON string) ledger.GooglePlayOrder { order := ledger.GooglePlayOrder{ OrderID: parsed.OrderID, State: parsed.State, @@ -163,12 +219,12 @@ func (c *Client) GetOrder(ctx context.Context, packageName string, orderID strin TotalAmountMicro: parsed.Total.micro(), TaxAmountMicro: parsed.Tax.micro(), NetAmountMicro: parsed.DeveloperRevenueInBuyerCurrency.micro(), - RawJSON: string(body), + RawJSON: rawJSON, } if order.CurrencyCode == "" { order.CurrencyCode = parsed.Tax.CurrencyCode } - return order, nil + return order } func (c *Client) ConsumeProduct(ctx context.Context, packageName string, productID string, purchaseToken string) error { diff --git a/services/wallet-service/internal/config/config.go b/services/wallet-service/internal/config/config.go index 9f32d07e..3c3eed77 100644 --- a/services/wallet-service/internal/config/config.go +++ b/services/wallet-service/internal/config/config.go @@ -29,6 +29,8 @@ type Config struct { RocketMQ RocketMQConfig `yaml:"rocketmq"` // GooglePlay 控制 Google Play Developer API 购买校验。 GooglePlay GooglePlayConfig `yaml:"google_play"` + // GooglePaidSyncWorker controls durable paid-detail retries after Google purchase confirmation. + GooglePaidSyncWorker GooglePaidSyncWorkerConfig `yaml:"google_paid_sync_worker"` // MifaPay 控制 H5 MiFaPay 站外支付下单和回调验签。 MifaPay MifaPayConfig `yaml:"mifapay"` // V5Pay 控制 H5 V5Pay 站外支付下单、查询和回调验签。 @@ -120,6 +122,14 @@ type GooglePlayConfig struct { ConsumeEnabled bool `yaml:"consume_enabled"` } +// GooglePaidSyncWorkerConfig polls unsynchronized payment_orders independently from the request goroutine. +type GooglePaidSyncWorkerConfig struct { + Enabled bool `yaml:"enabled"` + PollInterval time.Duration `yaml:"poll_interval"` + BatchSize int `yaml:"batch_size"` + RequestTimeout time.Duration `yaml:"request_timeout"` +} + // MifaPayConfig 保存 MiFaPay 商户签名、验签和网关配置。 type MifaPayConfig struct { Enabled bool `yaml:"enabled"` @@ -273,6 +283,12 @@ func Default() Config { HTTPTimeout: 10 * time.Second, ConsumeEnabled: true, }, + GooglePaidSyncWorker: GooglePaidSyncWorkerConfig{ + Enabled: true, + PollInterval: 15 * time.Second, + BatchSize: 100, + RequestTimeout: 2 * time.Minute, + }, MifaPay: MifaPayConfig{ Enabled: true, APIBaseURL: "https://platformtest.xqdmipay.com", @@ -438,6 +454,18 @@ func Load(path string) (Config, error) { return Config{}, errors.New("google_play service account is required") } } + if cfg.GooglePaidSyncWorker.PollInterval <= 0 { + cfg.GooglePaidSyncWorker.PollInterval = Default().GooglePaidSyncWorker.PollInterval + } + if cfg.GooglePaidSyncWorker.BatchSize <= 0 { + cfg.GooglePaidSyncWorker.BatchSize = Default().GooglePaidSyncWorker.BatchSize + } + if cfg.GooglePaidSyncWorker.BatchSize > 1000 { + cfg.GooglePaidSyncWorker.BatchSize = 1000 + } + if cfg.GooglePaidSyncWorker.RequestTimeout <= 0 { + cfg.GooglePaidSyncWorker.RequestTimeout = Default().GooglePaidSyncWorker.RequestTimeout + } cfg.MifaPay = normalizeMifaPayConfig(cfg.MifaPay) // MiFaPay 功能开关默认打开；商户密钥属于环境敏感配置，允许本地缺省启动，启动装配时再按凭证完整性决定是否接入真实网关。 cfg.V5Pay = normalizeV5PayConfig(cfg.V5Pay) diff --git a/services/wallet-service/internal/domain/ledger/recharge.go b/services/wallet-service/internal/domain/ledger/recharge.go index ac3262b2..f4a6c3a0 100644 --- a/services/wallet-service/internal/domain/ledger/recharge.go +++ b/services/wallet-service/internal/domain/ledger/recharge.go @@ -110,11 +110,13 @@ type GooglePaymentPaidDetails struct { // GooglePaymentOrderRef 是按钱包交易反查 Google 支付订单、发起 Orders API 查询所需的最小定位信息。 type GooglePaymentOrderRef struct { - PaymentOrderID string - TransactionID string - PackageName string - ProviderOrderID string - PaidSyncedAtMS int64 + AppCode string + PaymentOrderID string + TransactionID string + PackageName string + ProviderOrderID string + PaidSyncedAtMS int64 + SyncAttemptCount int } // GooglePaymentReceipt 是 Google 支付确认接口返回给 App 的稳定入账回执。 diff --git a/services/wallet-service/internal/service/wallet/ports/repository.go b/services/wallet-service/internal/service/wallet/ports/repository.go index ea5798ef..00b5ff5c 100644 --- a/services/wallet-service/internal/service/wallet/ports/repository.go +++ b/services/wallet-service/internal/service/wallet/ports/repository.go @@ -99,7 +99,9 @@ type RechargeStore interface { SummarizeRechargeBills(ctx context.Context, query ledger.ListRechargeBillsQuery) (ledger.RechargeBillSummary, error) GetRechargeBillOverview(ctx context.Context, query ledger.ListRechargeBillsQuery, tzOffsetMinutes int32) (ledger.RechargeBillOverview, error) ListGooglePaymentOrderRefs(ctx context.Context, appCode string, transactionIDs []string) (map[string]ledger.GooglePaymentOrderRef, error) + ListPendingGooglePaymentOrderRefs(ctx context.Context, limit int, nowMS int64) ([]ledger.GooglePaymentOrderRef, error) UpdateGooglePaymentPaidDetails(ctx context.Context, appCode string, paymentOrderID string, details ledger.GooglePaymentPaidDetails) error + MarkGooglePaymentPaidSyncFailure(ctx context.Context, ref ledger.GooglePaymentOrderRef, message string, nowMS int64) error ListRechargeProducts(ctx context.Context, userID int64, regionID int64, platform string, audienceType string) ([]ledger.RechargeProduct, []string, error) ListAdminRechargeProducts(ctx context.Context, query ledger.ListRechargeProductsQuery) ([]ledger.RechargeProduct, int64, error) GetRechargeProduct(ctx context.Context, appCode string, productID int64) (ledger.RechargeProduct, error) diff --git a/services/wallet-service/internal/service/wallet/recharge.go b/services/wallet-service/internal/service/wallet/recharge.go index 9256d9e2..39880e69 100644 --- a/services/wallet-service/internal/service/wallet/recharge.go +++ b/services/wallet-service/internal/service/wallet/recharge.go @@ -2,11 +2,15 @@ package wallet import ( "context" + "log/slog" + "sync" + "time" + "hyapp/pkg/appcode" + "hyapp/pkg/logx" "hyapp/pkg/xerr" "hyapp/services/wallet-service/internal/domain/ledger" "strings" - "time" ) // ListRechargeBills 返回充值账单只读列表；所有充值来源必须先在 wallet-service 落充值记录。 @@ -51,6 +55,16 @@ type GooglePaymentPriceResult struct { Err string } +type GooglePaymentPaidSyncBatchResult struct { + Claimed int + Succeeded int + Failed int +} + +type googlePlayOrdersBatchClient interface { + GetOrders(ctx context.Context, packageName string, orderIDs []string) (map[string]ledger.GooglePlayOrder, error) +} + const maxGooglePaymentPriceRefreshBatch = 100 // RefreshGooglePaymentPrices 用 Play Orders API 拉取账单的用户实付币种、总额、税额和净收入并写回 payment_orders。 @@ -92,16 +106,18 @@ func (s *Service) RefreshGooglePaymentPrices(ctx context.Context, appCode string } func (s *Service) refreshGooglePaymentPrice(ctx context.Context, appCode string, ref ledger.GooglePaymentOrderRef) GooglePaymentPriceResult { - result := GooglePaymentPriceResult{TransactionID: ref.TransactionID} if strings.TrimSpace(ref.ProviderOrderID) == "" { - result.Err = "google order id is missing" - return result + return s.failedGooglePaymentPrice(ctx, ref, "google order id is missing") } order, err := s.googlePlay.GetOrder(ctx, ref.PackageName, ref.ProviderOrderID) if err != nil { - result.Err = err.Error() - return result + return s.failedGooglePaymentPrice(ctx, ref, err.Error()) } + return s.refreshGooglePaymentPriceFromOrder(ctx, appCode, ref, order) +} + +func (s *Service) refreshGooglePaymentPriceFromOrder(ctx context.Context, appCode string, ref ledger.GooglePaymentOrderRef, order ledger.GooglePlayOrder) GooglePaymentPriceResult { + result := GooglePaymentPriceResult{TransactionID: ref.TransactionID} details := ledger.GooglePaymentPaidDetails{ CurrencyCode: order.CurrencyCode, PaidAmountMicro: order.TotalAmountMicro, @@ -110,8 +126,7 @@ func (s *Service) refreshGooglePaymentPrice(ctx context.Context, appCode string, SyncedAtMS: s.now().UnixMilli(), } if err := s.repository.UpdateGooglePaymentPaidDetails(ctx, appCode, ref.PaymentOrderID, details); err != nil { - result.Err = err.Error() - return result + return s.failedGooglePaymentPrice(ctx, ref, err.Error()) } result.CurrencyCode = details.CurrencyCode result.PaidAmountMicro = details.PaidAmountMicro @@ -121,6 +136,102 @@ func (s *Service) refreshGooglePaymentPrice(ctx context.Context, appCode string, return result } +func (s *Service) failedGooglePaymentPrice(ctx context.Context, ref ledger.GooglePaymentOrderRef, message string) GooglePaymentPriceResult { + result := GooglePaymentPriceResult{TransactionID: ref.TransactionID, Err: strings.TrimSpace(message)} + if err := s.repository.MarkGooglePaymentPaidSyncFailure(ctx, ref, result.Err, s.now().UnixMilli()); err != nil { + logx.Warn(ctx, "google_paid_sync_failure_persist_failed", + slog.String("app_code", ref.AppCode), + slog.String("payment_order_id", ref.PaymentOrderID), + slog.String("error", err.Error()), + ) + } + return result +} + +// ProcessGooglePaymentPaidSyncBatch is the durable worker entrypoint. It uses batchGet for the common +// path and bounded single-order fallback when Google rejects a batch containing one stale order ID. +func (s *Service) ProcessGooglePaymentPaidSyncBatch(ctx context.Context, batchSize int) (GooglePaymentPaidSyncBatchResult, error) { + if s.repository == nil { + return GooglePaymentPaidSyncBatchResult{}, xerr.New(xerr.Unavailable, "wallet repository is not configured") + } + if s.googlePlay == nil { + return GooglePaymentPaidSyncBatchResult{}, xerr.New(xerr.Unavailable, "google play client is not configured") + } + refs, err := s.repository.ListPendingGooglePaymentOrderRefs(ctx, batchSize, s.now().UnixMilli()) + if err != nil { + return GooglePaymentPaidSyncBatchResult{}, err + } + result := GooglePaymentPaidSyncBatchResult{Claimed: len(refs)} + if len(refs) == 0 { + return result, nil + } + prices := s.refreshGooglePaymentRefBatch(ctx, refs) + for _, price := range prices { + if price.Err == "" { + result.Succeeded++ + } else { + result.Failed++ + } + } + return result, nil +} + +func (s *Service) refreshGooglePaymentRefBatch(ctx context.Context, refs []ledger.GooglePaymentOrderRef) []GooglePaymentPriceResult { + results := make([]GooglePaymentPriceResult, len(refs)) + groups := make(map[string][]int) + for index, ref := range refs { + groups[strings.TrimSpace(ref.PackageName)] = append(groups[strings.TrimSpace(ref.PackageName)], index) + } + batchClient, supportsBatch := s.googlePlay.(googlePlayOrdersBatchClient) + for packageName, indexes := range groups { + if supportsBatch && packageName != "" { + orderIDs := make([]string, 0, len(indexes)) + for _, index := range indexes { + if orderID := strings.TrimSpace(refs[index].ProviderOrderID); orderID != "" { + orderIDs = append(orderIDs, orderID) + } + } + if len(orderIDs) == len(indexes) { + if orders, err := batchClient.GetOrders(ctx, packageName, orderIDs); err == nil { + for _, index := range indexes { + ref := refs[index] + order, ok := orders[ref.ProviderOrderID] + if !ok { + results[index] = s.failedGooglePaymentPrice(ctx, ref, "google order is not found") + continue + } + results[index] = s.refreshGooglePaymentPriceFromOrder(ctx, ref.AppCode, ref, order) + } + continue + } + } + } + s.refreshGooglePaymentRefSingles(ctx, refs, indexes, results) + } + return results +} + +func (s *Service) refreshGooglePaymentRefSingles(ctx context.Context, refs []ledger.GooglePaymentOrderRef, indexes []int, results []GooglePaymentPriceResult) { + const concurrency = 8 + semaphore := make(chan struct{}, concurrency) + var wait sync.WaitGroup + for _, index := range indexes { + wait.Add(1) + go func(index int) { + defer wait.Done() + select { + case semaphore <- struct{}{}: + defer func() { <-semaphore }() + case <-ctx.Done(): + results[index] = s.failedGooglePaymentPrice(ctx, refs[index], ctx.Err().Error()) + return + } + results[index] = s.refreshGooglePaymentPrice(ctx, refs[index].AppCode, refs[index]) + }(index) + } + wait.Wait() +} + // ListRechargeProducts 返回区域化充值档位；region_id 必须由 gateway 从 user-service 资料解析。 func (s *Service) ListRechargeProducts(ctx context.Context, userID int64, regionID int64, platform string, audienceType string) ([]ledger.RechargeProduct, []string, error) { if userID <= 0 || regionID <= 0 { @@ -274,7 +385,7 @@ func (s *Service) ConfirmGooglePayment(ctx context.Context, command ledger.Googl if err != nil { return ledger.GooglePaymentReceipt{}, err } - // 入账已完成，实付明细只影响财务展示：异步拉 Orders API，失败静默，后台可手动刷新补数。 + // 入账已完成，实付明细只影响财务展示：先即时异步拉取；失败会持久化并由 15 秒 worker 补偿。 s.scheduleGooglePaymentPriceRefresh(command.AppCode, receipt.PaymentOrderID, command.PackageName, purchase.OrderID) if receipt.ConsumeState == ledger.PaymentConsumeStateConsumed { return receipt, nil @@ -299,21 +410,22 @@ func (s *Service) scheduleGooglePaymentPriceRefresh(appCode string, paymentOrder strings.TrimSpace(paymentOrderID) == "" || strings.TrimSpace(orderID) == "" { return } - syncedAt := s.now().UnixMilli() go func() { ctx, cancel := context.WithTimeout(appcode.WithContext(context.Background(), appCode), 10*time.Second) defer cancel() - order, err := s.googlePlay.GetOrder(ctx, packageName, orderID) - if err != nil { - return - } - _ = s.repository.UpdateGooglePaymentPaidDetails(ctx, appCode, paymentOrderID, ledger.GooglePaymentPaidDetails{ - CurrencyCode: order.CurrencyCode, - PaidAmountMicro: order.TotalAmountMicro, - TaxAmountMicro: order.TaxAmountMicro, - NetAmountMicro: order.NetAmountMicro, - SyncedAtMS: syncedAt, + result := s.refreshGooglePaymentPrice(ctx, appCode, ledger.GooglePaymentOrderRef{ + AppCode: appCode, + PaymentOrderID: paymentOrderID, + PackageName: packageName, + ProviderOrderID: orderID, }) + if result.Err != "" { + logx.Warn(ctx, "google_paid_sync_immediate_failed", + slog.String("app_code", appCode), + slog.String("payment_order_id", paymentOrderID), + slog.String("error", result.Err), + ) + } }() } diff --git a/services/wallet-service/internal/storage/mysql/google_payment_repository.go b/services/wallet-service/internal/storage/mysql/google_payment_repository.go index 71dc5fc7..ecfbabc6 100644 --- a/services/wallet-service/internal/storage/mysql/google_payment_repository.go +++ b/services/wallet-service/internal/storage/mysql/google_payment_repository.go @@ -13,6 +13,11 @@ import ( "hyapp/services/wallet-service/internal/domain/ledger" ) +const ( + googlePaidSyncRetryBase = 15 * time.Second + googlePaidSyncRetryMax = 15 * time.Minute +) + // ConfirmGooglePayment 将 Google Play 已购买 token 落为平台支付订单、钱包交易和充值审计记录。 func (r *Repository) ConfirmGooglePayment(ctx context.Context, command ledger.GooglePaymentCommand, product ledger.RechargeProduct, purchase ledger.GooglePlayPurchase) (ledger.GooglePaymentReceipt, error) { if r == nil || r.db == nil { @@ -321,7 +326,8 @@ func (r *Repository) ListGooglePaymentOrderRefs(ctx context.Context, appCode str args = append(args, transactionID) } rows, err := r.db.QueryContext(ctx, ` - SELECT payment_order_id, wallet_transaction_id, package_name, provider_order_id, paid_synced_at_ms + SELECT app_code, payment_order_id, wallet_transaction_id, package_name, provider_order_id, + paid_synced_at_ms, paid_sync_attempt_count FROM payment_orders WHERE app_code = ? AND provider = ? AND wallet_transaction_id IN (`+placeholders+`)`, args..., @@ -334,7 +340,7 @@ func (r *Repository) ListGooglePaymentOrderRefs(ctx context.Context, appCode str refs := make(map[string]ledger.GooglePaymentOrderRef, len(cleaned)) for rows.Next() { var ref ledger.GooglePaymentOrderRef - if err := rows.Scan(&ref.PaymentOrderID, &ref.TransactionID, &ref.PackageName, &ref.ProviderOrderID, &ref.PaidSyncedAtMS); err != nil { + if err := rows.Scan(&ref.AppCode, &ref.PaymentOrderID, &ref.TransactionID, &ref.PackageName, &ref.ProviderOrderID, &ref.PaidSyncedAtMS, &ref.SyncAttemptCount); err != nil { return nil, err } refs[ref.TransactionID] = ref @@ -342,6 +348,43 @@ func (r *Repository) ListGooglePaymentOrderRefs(ctx context.Context, appCode str return refs, rows.Err() } +// ListPendingGooglePaymentOrderRefs claims no rows: paid-detail synchronization is idempotent and the +// retry columns prevent failed orders from monopolizing every 15-second scan. Duplicate workers may +// read the same row during a rolling deploy, but both write the same provider fact. +func (r *Repository) ListPendingGooglePaymentOrderRefs(ctx context.Context, limit int, nowMS int64) ([]ledger.GooglePaymentOrderRef, error) { + if r == nil || r.db == nil { + return nil, xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + if limit <= 0 { + return []ledger.GooglePaymentOrderRef{}, nil + } + if limit > 1000 { + limit = 1000 + } + rows, err := r.db.QueryContext(ctx, ` + SELECT app_code, payment_order_id, wallet_transaction_id, package_name, provider_order_id, + paid_synced_at_ms, paid_sync_attempt_count + FROM payment_orders + WHERE provider = ? AND paid_synced_at_ms = 0 AND paid_sync_next_retry_at_ms <= ? + ORDER BY created_at_ms DESC, payment_order_id DESC + LIMIT ?`, + ledger.PaymentProviderGooglePlay, nowMS, limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + refs := make([]ledger.GooglePaymentOrderRef, 0, limit) + for rows.Next() { + var ref ledger.GooglePaymentOrderRef + if err := rows.Scan(&ref.AppCode, &ref.PaymentOrderID, &ref.TransactionID, &ref.PackageName, &ref.ProviderOrderID, &ref.PaidSyncedAtMS, &ref.SyncAttemptCount); err != nil { + return nil, err + } + refs = append(refs, ref) + } + return refs, rows.Err() +} + // UpdateGooglePaymentPaidDetails 把 Orders API 返回的实付明细写回 payment_orders；同步时间必须由调用方给出。 func (r *Repository) UpdateGooglePaymentPaidDetails(ctx context.Context, appCode string, paymentOrderID string, details ledger.GooglePaymentPaidDetails) error { if r == nil || r.db == nil { @@ -354,7 +397,8 @@ func (r *Repository) UpdateGooglePaymentPaidDetails(ctx context.Context, appCode } _, err := r.db.ExecContext(ctx, ` UPDATE payment_orders - SET paid_currency_code = ?, paid_amount_micro = ?, paid_tax_micro = ?, paid_net_micro = ?, paid_synced_at_ms = ?, updated_at_ms = ? + SET paid_currency_code = ?, paid_amount_micro = ?, paid_tax_micro = ?, paid_net_micro = ?, paid_synced_at_ms = ?, + paid_sync_attempt_count = 0, paid_sync_last_error = '', paid_sync_next_retry_at_ms = 0, updated_at_ms = ? WHERE app_code = ? AND payment_order_id = ? AND provider = ?`, strings.ToUpper(strings.TrimSpace(details.CurrencyCode)), details.PaidAmountMicro, @@ -369,6 +413,34 @@ func (r *Repository) UpdateGooglePaymentPaidDetails(ctx context.Context, appCode return err } +// MarkGooglePaymentPaidSyncFailure stores sanitized diagnostics and exponential retry time. +// It never overwrites a concurrently completed synchronization because paid_synced_at_ms=0 is required. +func (r *Repository) MarkGooglePaymentPaidSyncFailure(ctx context.Context, ref ledger.GooglePaymentOrderRef, message string, nowMS int64) error { + if r == nil || r.db == nil { + return xerr.New(xerr.Unavailable, "mysql repository is not configured") + } + message = strings.TrimSpace(message) + if len(message) > 512 { + message = message[:512] + } + backoff := googlePaidSyncRetryBase + for attempt := 0; attempt < ref.SyncAttemptCount && backoff < googlePaidSyncRetryMax; attempt++ { + backoff *= 2 + if backoff > googlePaidSyncRetryMax { + backoff = googlePaidSyncRetryMax + } + } + _, err := r.db.ExecContext(ctx, ` + UPDATE payment_orders + SET paid_sync_attempt_count = paid_sync_attempt_count + 1, + paid_sync_last_error = ?, paid_sync_next_retry_at_ms = ?, updated_at_ms = ? + WHERE app_code = ? AND payment_order_id = ? AND provider = ? AND paid_synced_at_ms = 0`, + message, nowMS+backoff.Milliseconds(), nowMS, + strings.ToLower(strings.TrimSpace(ref.AppCode)), strings.TrimSpace(ref.PaymentOrderID), ledger.PaymentProviderGooglePlay, + ) + return err +} + type googlePaymentMetadata struct { AppCode string `json:"app_code"` PaymentOrderID string `json:"payment_order_id"` diff --git a/services/wallet-service/internal/storage/mysql/schema.go b/services/wallet-service/internal/storage/mysql/schema.go index 4934cb5a..3e76b3b4 100644 --- a/services/wallet-service/internal/storage/mysql/schema.go +++ b/services/wallet-service/internal/storage/mysql/schema.go @@ -172,6 +172,9 @@ func ensureGooglePaymentPaidDetailsSchema(ctx context.Context, db *sql.DB) error `ALTER TABLE payment_orders ADD COLUMN paid_tax_micro BIGINT NOT NULL DEFAULT 0 COMMENT '订单税额微单位（实付币种）' AFTER paid_amount_micro`, `ALTER TABLE payment_orders ADD COLUMN paid_net_micro BIGINT NOT NULL DEFAULT 0 COMMENT 'Google 扣费后开发者净收入微单位（实付币种）' AFTER paid_tax_micro`, `ALTER TABLE payment_orders ADD COLUMN paid_synced_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT '实付明细同步时间，UTC epoch ms；0 表示未同步' AFTER paid_net_micro`, + `ALTER TABLE payment_orders ADD COLUMN paid_sync_attempt_count INT NOT NULL DEFAULT 0 COMMENT 'Google 实付同步连续失败次数' AFTER paid_synced_at_ms`, + `ALTER TABLE payment_orders ADD COLUMN paid_sync_last_error VARCHAR(512) NOT NULL DEFAULT '' COMMENT 'Google 实付同步最近错误' AFTER paid_sync_attempt_count`, + `ALTER TABLE payment_orders ADD COLUMN paid_sync_next_retry_at_ms BIGINT NOT NULL DEFAULT 0 COMMENT 'Google 实付同步下次重试时间，UTC epoch ms' AFTER paid_sync_last_error`, } for _, stmt := range alters { if _, err := db.ExecContext(ctx, stmt); err != nil && !isDuplicateColumnError(err) { @@ -183,6 +186,11 @@ func ensureGooglePaymentPaidDetailsSchema(ctx context.Context, db *sql.DB) error ADD INDEX idx_payment_orders_wallet_tx (app_code, wallet_transaction_id)`); err != nil && !isDuplicateKeyNameError(err) { return err } + if _, err := db.ExecContext(ctx, ` + ALTER TABLE payment_orders + ADD INDEX idx_payment_orders_paid_sync (provider, paid_synced_at_ms, paid_sync_next_retry_at_ms, created_at_ms)`); err != nil && !isDuplicateKeyNameError(err) { + return err + } return nil }