Compare commits

...

2 Commits

Author SHA1 Message Date
a33cbce667 refactor(api): extract everyTick helper for reconciler goroutines 2026-06-15 01:30:57 +02:00
5dd196834f feat/sca pay redirects sweep (#50)
* feat(sca): sweep expired pay_redirects rows on a ticker

* fix(fix): revert the 10-second ticker to 10m after forgetting to revert it after testing
2026-06-15 00:39:31 +02:00
3 changed files with 127 additions and 40 deletions

View File

@@ -225,6 +225,23 @@ func ReconcileExpiredCards(ctx context.Context) {
}
}
// ReconcilePayRedirects deletes pay_redirects rows whose ExpiresAt had passed.
// Lazy deletion on read (PayRedirect) is the primary path, this sweep only
// reclaims tokens nobody ever followed so the table stays bounded.
func ReconcilePayRedirects(ctx context.Context) {
db := database.GetDB().WithContext(ctx)
res := db.Where("expires_at < ?", time.Now()).Delete(&models.PayRedirect{})
if res.Error != nil {
reconcileLog().Error("pay_redirects sweep failed", "error", res.Error)
return
}
if res.RowsAffected == 0 {
return
}
reconcileLog().Info("swept expired pay_redirects", "count", res.RowsAffected)
}
// ReconcileUnshipped finds paid orders older than 24h with no tracking
// number and posts a single Slack reminder. Pure read + notify.
func ReconcileUnshipped(ctx context.Context) {

View File

@@ -518,3 +518,81 @@ func seedRequiresActionOrder(t *testing.T, userID uint, piID string, updatedAt t
}
return order
}
// TestReconcilePayRedirects_DeleteExpiredKeepsLive verifies the predicate
// boundary: a row past ExpiresAt is hard-deleted while a row whose ExpiresAt
// is still in the future survives. Seeding both sides through GORM keeps the
// stored timestamp format identical to the one the sweep binds for time.Now(),
// so the comparison is exercised exactly as it runs in production.
func TestReconcilePayRedirects_DeleteExpiredKeepsLive(t *testing.T) {
testDB := setupTestDB(t)
defer func() { _ = os.Remove(testDB) }()
defer database.ResetForTesting()
db := database.GetDB()
now := time.Now()
expired := models.PayRedirect{
Token: "tok_expired",
URL: "https://example.test/expired",
Purpose: models.RedirectPurposeCheckout3DS,
CreatedAt: now.Add(-20 * time.Minute),
ExpiresAt: now.Add(-10 * time.Minute),
}
live := models.PayRedirect{
Token: "tok_live",
URL: "https://example.test/live",
Purpose: models.RedirectPurposeAddCard,
CreatedAt: now,
ExpiresAt: now.Add(10 * time.Minute),
}
if err := db.Create(&expired).Error; err != nil {
t.Fatalf("seed expired: %v", err)
}
if err := db.Create(&live).Error; err != nil {
t.Fatalf("seed live: %v", err)
}
ReconcilePayRedirects(context.Background())
var remaining []models.PayRedirect
if err := db.Find(&remaining).Error; err != nil {
t.Fatalf("list remaining: %v", err)
}
if len(remaining) != 1 {
t.Fatalf("want 1 row remaining, got %d", len(remaining))
}
if remaining[0].Token != "tok_live" {
t.Fatalf("wrong row survived: want tok_live, got %q", remaining[0].Token)
}
}
func TestReconcilePayRedirects_NoExpiredIsNoOp(t *testing.T) {
testDB := setupTestDB(t)
defer func() { _ = os.Remove(testDB) }()
defer database.ResetForTesting()
db := database.GetDB()
live := models.PayRedirect{
Token: "tok_live_only",
URL: "https://example.test/live",
Purpose: models.RedirectPurposeCheckout3DS,
CreatedAt: time.Now(),
ExpiresAt: time.Now().Add(10 * time.Minute),
}
if err := db.Create(&live).Error; err != nil {
t.Fatalf("seed live: %v", err)
}
ReconcilePayRedirects(context.Background())
var count int64
if err := db.Model(&models.PayRedirect{}).Count(&count).Error; err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("want 1 row preserved, got %d", count)
}
}

View File

@@ -25,6 +25,25 @@ import (
const version = "v0.1.0"
// everyTick runs fn on its own goroutine every d until ctx is cancelled.
// Shared by the reconciler sweeps so the ticker + shutdown plumbing lives in
// one place. ReconcileUnshipped intentionally does not use it, it needs a
// staggered warm-up run before its first tick.
func everyTick(ctx context.Context, d time.Duration, fn func(context.Context)) {
go func() {
ticker := time.NewTicker(d)
defer ticker.Stop()
for {
select {
case <-ticker.C:
fn(ctx)
case <-ctx.Done():
return
}
}
}()
}
func main() {
observability.InitLogger()
middleware.SetBuildInfo(version)
@@ -106,20 +125,8 @@ func main() {
}
}()
// Start reconciliation job. runs every 5 minutes to catch any orders
// that were charged but not recorded due to a crash.
go func() {
ticker := time.NewTicker(5 * time.Minute)
defer ticker.Stop()
for {
select {
case <-ticker.C:
handlers.ReconcileOrders(ctx)
case <-ctx.Done():
return
}
}
}()
// Catch orders charged but not recorded due to a crash. Every 5 minutes
everyTick(ctx, 5*time.Minute, handlers.ReconcileOrders)
go func() {
select {
@@ -142,33 +149,18 @@ func main() {
// Sweep saved cards whose retention deadline elapsed. Read paths filter
// expired rows from query results, so this sweeper only drives the Stripe
// detach + audit + physical row delete asynchronously.
go func() {
ticker := time.NewTicker(1 * time.Hour)
defer ticker.Stop()
for {
select {
case <-ticker.C:
handlers.ReconcileExpiredCards(ctx)
case <-ctx.Done():
return
}
}
}()
// detach + audit + physical row delete asynchronously. Hourly.
everyTick(ctx, 1*time.Hour, handlers.ReconcileExpiredCards)
go func() {
ticker := time.NewTicker(15 * time.Minute)
defer ticker.Stop()
threshold := time.Duration(cfg.Abandoned3DSThresholdMinutes) * time.Minute
for {
select {
case <-ticker.C:
handlers.ReconcileStale3DSOrders(ctx, threshold)
case <-ctx.Done():
return
}
}
}()
// Flip abandoned 3DS orders to failed. Every 15 minutes.
threshold := time.Duration(cfg.Abandoned3DSThresholdMinutes) * time.Minute
everyTick(ctx, 15*time.Minute, func(ctx context.Context) {
handlers.ReconcileStale3DSOrders(ctx, threshold)
})
// Sweep expired pay_redirects rows so the table stays bounded. Lazy delete
// on read is the primary path; this only reclaims tokens nobody follows.
everyTick(ctx, 10*time.Minute, handlers.ReconcilePayRedirects)
log.Println("API server started. Press Ctrl+C to shutdown.")