diff --git a/backend/internal/jobs/cleanup.go b/backend/internal/jobs/cleanup.go index 1f6ee6b..a16e9b4 100644 --- a/backend/internal/jobs/cleanup.go +++ b/backend/internal/jobs/cleanup.go @@ -1,14 +1,21 @@ package jobs import ( + "context" + "errors" + "fmt" + "log/slog" "time" "crussell/auth" + "crussell/db" authHandlers "crussell/handlers/auth" "crussell/handlers/payments" "crussell/handlers/scheduling" "crussell/handlers/user" "crussell/mw" + + "github.com/jackc/pgx/v5" ) // RegisterAll registers every background maintenance job on the scheduler. @@ -214,4 +221,41 @@ func RegisterAll(s *Scheduler) { Concurrency: 1, Handler: scheduling.CleanupExpiredRefreshTokens, }) + + // Staggered from cleanup-verification-codes / cleanup-refresh-tokens + // (both at 0 2 * * *) to avoid DB contention. + s.Register(Job{ + Name: "sweep-square-webhook-events", + Schedule: "30 2 * * *", // Daily at 2:30am + Timeout: 30 * time.Second, + Concurrency: 1, + Handler: SweepSquareWebhookEvents, + }) +} + +// SweepSquareWebhookEvents deletes square_webhook_events rows older than 90 +// days. Every accepted webhook event_id is stored permanently for restart-safe +// dedup (payment.updated fires on every payment update), so without retention +// the table would grow without bound. 90 days comfortably exceeds Square's +// webhook replay window while keeping the table bounded. +func SweepSquareWebhookEvents(ctx context.Context) (int, error) { + tx, err := db.Conn.Begin(ctx) + if err != nil { + return 0, fmt.Errorf("failed to begin transaction: %w", err) + } + defer func() { + if err := tx.Rollback(ctx); err != nil && !errors.Is(err, pgx.ErrTxClosed) { + slog.Error("failed to rollback transaction", "err", err) + } + }() + + tag, err := tx.Exec(ctx, ` + DELETE FROM square_webhook_events + WHERE received_at < NOW() - INTERVAL '90 days' + `) + if err != nil { + return 0, fmt.Errorf("failed to sweep square webhook events: %w", err) + } + + return int(tag.RowsAffected()), tx.Commit(ctx) } diff --git a/backend/internal/jobs/cleanup_test.go b/backend/internal/jobs/cleanup_test.go new file mode 100644 index 0000000..1fd3df6 --- /dev/null +++ b/backend/internal/jobs/cleanup_test.go @@ -0,0 +1,89 @@ +//go:build test + +package jobs + +import ( + "context" + "os" + "testing" + + "crussell/db" + "crussell/testutils/testdb" +) + +func TestMain(m *testing.M) { + pool := testdb.CreateTestDatabase("crussell_test_jobs") + db.Conn = db.NewPoolProxy(pool) + code := m.Run() + testdb.DestroyTestDatabase(pool, "crussell_test_jobs") + os.Exit(code) +} + +// ============================================================ +// SweepSquareWebhookEvents Tests +// ============================================================ + +func TestSweepSquareWebhookEvents_DeletesOldRows(t *testing.T) { + ctx := context.Background() + + // Recent event (NOW() — must be preserved) + if _, err := db.Conn.Exec(ctx, + "INSERT INTO square_webhook_events (event_id, received_at) VALUES ($1, NOW())", + "evt_recent"); err != nil { + t.Fatalf("failed to insert recent webhook event: %v", err) + } + + // Old event (100 days ago — must be swept) + if _, err := db.Conn.Exec(ctx, + "INSERT INTO square_webhook_events (event_id, received_at) VALUES ($1, NOW() - INTERVAL '100 days')", + "evt_old"); err != nil { + t.Fatalf("failed to insert old webhook event: %v", err) + } + t.Cleanup(func() { + _, _ = db.Conn.Exec(ctx, "DELETE FROM square_webhook_events WHERE event_id IN ('evt_recent', 'evt_old')") + }) + + n, err := SweepSquareWebhookEvents(ctx) + if err != nil { + t.Fatalf("SweepSquareWebhookEvents failed: %v", err) + } + if n != 1 { + t.Errorf("expected 1 row deleted, got %d", n) + } + + // Old row must be gone + var oldCount int + if err := db.Conn.QueryRow(ctx, + "SELECT COUNT(*) FROM square_webhook_events WHERE event_id = 'evt_old'").Scan(&oldCount); err != nil { + t.Fatalf("failed to count old event: %v", err) + } + if oldCount != 0 { + t.Errorf("expected old event to be deleted, got %d rows", oldCount) + } + + // Recent row must remain + var recentCount int + if err := db.Conn.QueryRow(ctx, + "SELECT COUNT(*) FROM square_webhook_events WHERE event_id = 'evt_recent'").Scan(&recentCount); err != nil { + t.Fatalf("failed to count recent event: %v", err) + } + if recentCount != 1 { + t.Errorf("expected recent event to be preserved, got %d rows", recentCount) + } +} + +func TestSweepSquareWebhookEvents_EmptyTable(t *testing.T) { + ctx := context.Background() + + if _, err := db.Conn.Exec(ctx, "DELETE FROM square_webhook_events"); err != nil { + t.Fatalf("failed to clear square_webhook_events: %v", err) + } + + n, err := SweepSquareWebhookEvents(ctx) + if err != nil { + t.Fatalf("SweepSquareWebhookEvents failed on empty table: %v", err) + } + if n != 0 { + t.Errorf("expected 0 rows deleted on empty table, got %d", n) + } +} diff --git a/backend/internal/jobs/scheduler_test.go b/backend/internal/jobs/scheduler_test.go index ceeec01..b9356f9 100644 --- a/backend/internal/jobs/scheduler_test.go +++ b/backend/internal/jobs/scheduler_test.go @@ -413,8 +413,8 @@ func TestRegisterAll_RegistersExpectedJobs(t *testing.T) { s := New() RegisterAll(s) - if got := len(s.registry); got != 23 { - t.Fatalf("RegisterAll() registered %d jobs, want 23", got) + if got := len(s.registry); got != 24 { + t.Fatalf("RegisterAll() registered %d jobs, want 24", got) } registered := make(map[string]Job, len(s.registry)) @@ -490,6 +490,7 @@ func expectedJobNames() map[string]bool { "transition-discount-campaigns": true, "cleanup-verification-codes": true, "cleanup-refresh-tokens": true, + "sweep-square-webhook-events": true, "apply-default-hours": true, } } diff --git a/backend/testutils/testdb/testdb.go b/backend/testutils/testdb/testdb.go index f8fda50..ac2ca45 100644 --- a/backend/testutils/testdb/testdb.go +++ b/backend/testutils/testdb/testdb.go @@ -53,6 +53,13 @@ func CreateTestDatabase(dbName string) *pgxpool.Pool { } defer adminPool.Close() + // Pre-flight check: pgxpool.New is lazy, so surface the admin connection + // failure here with actionable guidance instead of a confusing + // "database \"crussell_test_...\" does not exist" error further down. + if err := adminPool.Ping(ctx); err != nil { + log.Fatalf("testdb: run `docker compose up -d postgres` (see README Getting Started) — test database admin connection failed: %v", err) + } + safeName := pgx.Identifier{dbName}.Sanitize() // Drop existing database with retries (terminate connections first) diff --git a/init-scripts/init-script.sql b/init-scripts/init-script.sql index f3276b9..fd94068 100644 --- a/init-scripts/init-script.sql +++ b/init-scripts/init-script.sql @@ -2284,6 +2284,18 @@ CREATE TABLE square_deposits ( CREATE INDEX idx_square_deposits_deposited ON square_deposits(deposited_at); +-- ======================================= +-- SQUARE WEBHOOK EVENTS TABLE (Webhook Dedup) +-- ======================================= +-- Records every Square webhook event_id the handler has accepted. The webhook +-- handler inserts with ON CONFLICT (event_id) DO NOTHING and treats a 0-row +-- insert as a duplicate, giving restart-safe, at-most-once delivery with no +-- eviction limit (unlike the in-memory fast-path cache). +CREATE TABLE IF NOT EXISTS square_webhook_events ( + event_id TEXT PRIMARY KEY, + received_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + -- ======================================= -- CARDDAV / CALDAV TABLES (used by SabreDAV sync in Go backend) -- =======================================