diff --git a/backend/internal/repository/proxy_repeated_expiry_integration_test.go b/backend/internal/repository/proxy_repeated_expiry_integration_test.go new file mode 100644 index 0000000000..6c84309234 --- /dev/null +++ b/backend/internal/repository/proxy_repeated_expiry_integration_test.go @@ -0,0 +1,69 @@ +//go:build integration + +package repository + +import ( + "encoding/json" + "time" + + "github.com/Wei-Shaw/sub2api/internal/service" +) + +func (s *ProxyExpirySuite) TestSweep_RepeatedExpiryPreservesOriginalProxy() { + for _, direct := range []bool{false, true} { + s.Run(map[bool]string{false: "next backup", true: "direct"}[direct], func() { + now := time.Now() + past := now.Add(-time.Hour) + soon := now.Add(time.Hour) + later := now.Add(48 * time.Hour) + originalID := s.mkProxy("repeated-original", service.FallbackModeProxy, &past, nil) + middleID := s.mkProxy("repeated-middle", service.FallbackModeDirect, &soon, nil) + lastID := s.mkProxy("repeated-last", service.FallbackModeNone, &later, nil) + // Populate directed foreign keys directly so this regression does not depend + // on the separate ORM backup-edge fix. + _, err := s.tx.ExecContext(s.ctx, `UPDATE proxies SET backup_proxy_id=$1 WHERE id=$2`, middleID, originalID) + s.Require().NoError(err) + if !direct { + _, err = s.tx.ExecContext(s.ctx, `UPDATE proxies SET fallback_mode='proxy',backup_proxy_id=$1 WHERE id=$2`, lastID, middleID) + s.Require().NoError(err) + } + accountID := s.mkAccountWithProxy(originalID) + changed, err := s.repo.SweepExpiredProxies(s.ctx, now) + s.Require().NoError(err) + s.Require().EqualValues(1, changed) + s.Require().Equal(&middleID, s.accountProxyID(accountID)) + + changed, err = s.repo.SweepExpiredProxies(s.ctx, now.Add(2*time.Hour)) + s.Require().NoError(err) + s.Require().EqualValues(1, changed, "accounts already in fallback must still be rerouted") + if direct { + s.Require().Nil(s.accountProxyID(accountID)) + } else { + s.Require().Equal(&lastID, s.accountProxyID(accountID)) + } + var origin *int64 + err = scanSingleRow(s.ctx, s.tx, `SELECT proxy_fallback_origin_id FROM accounts WHERE id=$1`, []any{accountID}, &origin) + s.Require().NoError(err) + s.Require().Equal(&originalID, origin, "manual revert must retain the first original proxy") + + var payloadRaw []byte + err = scanSingleRow(s.ctx, s.tx, `SELECT payload FROM scheduler_outbox WHERE event_type=$1 ORDER BY id DESC LIMIT 1`, []any{service.SchedulerOutboxEventAccountBulkChanged}, &payloadRaw) + s.Require().NoError(err) + var payload struct { + AccountIDs []int64 `json:"account_ids"` + } + s.Require().NoError(json.Unmarshal(payloadRaw, &payload)) + s.Require().Equal([]int64{accountID}, payload.AccountIDs) + + changed, err = s.repo.SweepExpiredProxies(s.ctx, now.Add(3*time.Hour)) + s.Require().NoError(err) + s.Require().Zero(changed, "repeating the scan must be idempotent") + accounts := newAccountRepositoryWithSQL(s.tx.Client(), s.tx, nil) + s.Require().NoError(accounts.RevertProxyFallback(s.ctx, accountID)) + s.Require().Equal(&originalID, s.accountProxyID(accountID)) + err = scanSingleRow(s.ctx, s.tx, `SELECT proxy_fallback_origin_id FROM accounts WHERE id=$1`, []any{accountID}, &origin) + s.Require().NoError(err) + s.Require().Nil(origin) + }) + } +} diff --git a/backend/internal/repository/proxy_repo.go b/backend/internal/repository/proxy_repo.go index 679b22b26f..07c92046bd 100644 --- a/backend/internal/repository/proxy_repo.go +++ b/backend/internal/repository/proxy_repo.go @@ -742,27 +742,29 @@ func (r *proxyRepository) sweepOneExpiredProxyOnExec(ctx context.Context, exec s rows *sql.Rows err error ) + // Match the current proxy even after an earlier fallback. Keep the first + // origin so manual revert still restores the originally assigned proxy. if target == nil { rows, err = exec.QueryContext(ctx, ` - UPDATE accounts SET proxy_id=NULL, proxy_fallback_origin_id=$1, + UPDATE accounts SET proxy_id=NULL, proxy_fallback_origin_id=COALESCE(proxy_fallback_origin_id,$1), extra=CASE WHEN type='apikey' AND extra ? 'upstream_billing_probe' THEN extra - 'upstream_billing_probe' ELSE extra END, updated_at=NOW() - WHERE proxy_id=$1 AND proxy_fallback_origin_id IS NULL AND deleted_at IS NULL + WHERE proxy_id=$1 AND deleted_at IS NULL RETURNING id`, proxyID) } else { rows, err = exec.QueryContext(ctx, ` - UPDATE accounts SET proxy_id=$2, proxy_fallback_origin_id=$1, + UPDATE accounts SET proxy_id=$2, proxy_fallback_origin_id=COALESCE(proxy_fallback_origin_id,$1), extra=CASE WHEN type='apikey' AND extra ? 'upstream_billing_probe' THEN extra - 'upstream_billing_probe' ELSE extra END, updated_at=NOW() - WHERE proxy_id=$1 AND proxy_fallback_origin_id IS NULL AND deleted_at IS NULL + WHERE proxy_id=$1 AND deleted_at IS NULL RETURNING id`, proxyID, *target) } if err != nil {