diff --git a/backend/handlers/scheduling/scheduled-cleanup.go b/backend/handlers/scheduling/scheduled-cleanup.go index 274cb94..752591c 100644 --- a/backend/handlers/scheduling/scheduled-cleanup.go +++ b/backend/handlers/scheduling/scheduled-cleanup.go @@ -12,6 +12,7 @@ import ( "crussell/auth" "crussell/clock" "crussell/db" + "crussell/internal/adminnotify" "github.com/jackc/pgx/v5" ) @@ -70,11 +71,17 @@ func NotifyUnpaidOneWeek(ctx context.Context) (int, error) { return 0, nil } - _, err = tx.Exec(ctx, ` + // C5: flood-cap the unacknowledged '1_week_no_pay' queue (pre-check logs + // the suppression; the fold inside the INSERT enforces it atomically). + if adminnotify.CriticalLogsCapExceeded(ctx, tx, "1_week_no_pay") { + slog.Warn("suppressed 1_week_no_pay admin notification — unacknowledged queue at the cap") + } else if _, err = tx.Exec(ctx, ` INSERT INTO admin_notifications (reason, booking_id, user_id) SELECT '1_week_no_pay', unnest($1::text[]), unnest($2::text[]) - `, ids, userIDs) - if err != nil { + WHERE (SELECT COUNT(*) FROM admin_notifications _an + WHERE _an.reason = '1_week_no_pay' + AND _an.acknowledged_at IS NULL) < $3 + `, ids, userIDs, adminnotify.MaxUnacknowledgedCriticalLogs); err != nil { return 0, fmt.Errorf("failed to insert 1_week_no_pay notifications: %w", err) } @@ -138,11 +145,17 @@ func NotifyUnpaidOneMonth(ctx context.Context) (int, error) { return 0, nil } - _, err = tx.Exec(ctx, ` + // C5: flood-cap the unacknowledged '1_month_no_pay' queue (pre-check logs + // the suppression; the fold inside the INSERT enforces it atomically). + if adminnotify.CriticalLogsCapExceeded(ctx, tx, "1_month_no_pay") { + slog.Warn("suppressed 1_month_no_pay admin notification — unacknowledged queue at the cap") + } else if _, err = tx.Exec(ctx, ` INSERT INTO admin_notifications (reason, booking_id, user_id) SELECT '1_month_no_pay', unnest($1::text[]), unnest($2::text[]) - `, ids, userIDs) - if err != nil { + WHERE (SELECT COUNT(*) FROM admin_notifications _an + WHERE _an.reason = '1_month_no_pay' + AND _an.acknowledged_at IS NULL) < $3 + `, ids, userIDs, adminnotify.MaxUnacknowledgedCriticalLogs); err != nil { return 0, fmt.Errorf("failed to insert 1_month_no_pay notifications: %w", err) } @@ -392,11 +405,18 @@ func ApplyScheduledDefaultHours(ctx context.Context) (int, error) { return 0, fmt.Errorf("failed to mark change %d as applied: %w", rowID, err) } - // Insert admin notification - if _, err := tx.Exec(ctx, ` + // Insert admin notification — C5: the 'default_hours_changed' queue is + // flood-capped (pre-check logs the suppression; the fold inside the INSERT + // enforces it atomically). + if adminnotify.CriticalLogsCapExceeded(ctx, tx, "default_hours_changed") { + slog.Warn("suppressed default_hours_changed admin notification — unacknowledged queue at the cap") + } else if _, err := tx.Exec(ctx, ` INSERT INTO admin_notifications (reason, created_at) - VALUES ('default_hours_changed', NOW()) - `); err != nil { + SELECT 'default_hours_changed', NOW() + WHERE (SELECT COUNT(*) FROM admin_notifications _an + WHERE _an.reason = 'default_hours_changed' + AND _an.acknowledged_at IS NULL) < $1 + `, adminnotify.MaxUnacknowledgedCriticalLogs); err != nil { return 0, fmt.Errorf("failed to insert notification for change %d: %w", rowID, err) } diff --git a/backend/handlers/scheduling/scheduled_cleanup_test.go b/backend/handlers/scheduling/scheduled_cleanup_test.go index 8e76a43..1d83098 100644 --- a/backend/handlers/scheduling/scheduled_cleanup_test.go +++ b/backend/handlers/scheduling/scheduled_cleanup_test.go @@ -8,6 +8,7 @@ import ( "time" "crussell/clock" + "crussell/internal/adminnotify" "crussell/testutils/fixtures" ) @@ -992,3 +993,59 @@ func TestTransitionDiscountCampaigns_OnlyCompleted(t *testing.T) { t.Errorf("expected status 'completed', got %q", status) } } + +// TestNotifyUnpaidOneWeek_NotificationFloodCap pins C5 for the +// '1_week_no_pay' insert site: the unacknowledged queue is flood-capped at +// adminnotify.MaxUnacknowledgedCriticalLogs, so a runaway unpaid-booking +// cleanup cannot bury the operator's notification centre. At the cap further +// inserts are suppressed and the queue stays bounded. +func TestNotifyUnpaidOneWeek_NotificationFloodCap(t *testing.T) { + t.Parallel() + ctx, tx := resetTestData(t) + + userID, err := fixtures.CreateTestUser(tx) + if err != nil { + t.Fatalf("failed to create user: %v", err) + } + + var bookingID string + err = tx.QueryRow(ctx, ` + INSERT INTO bookings (user_id, start_time, status) + VALUES ($1, $2, 'completed') + RETURNING id + `, userID, clock.Now().Add(-14*24*time.Hour)).Scan(&bookingID) + if err != nil { + t.Fatalf("failed to create booking: %v", err) + } + + // Fill the unacknowledged '1_week_no_pay' queue to the cap before the + // cleanup runs, so the insert site must suppress instead of growing it. + for i := 0; i < adminnotify.MaxUnacknowledgedCriticalLogs; i++ { + if _, err := tx.Exec(ctx, ` + INSERT INTO admin_notifications (reason, user_id, created_at) + VALUES ('1_week_no_pay', $1, NOW()) + `, userID); err != nil { + t.Fatalf("failed to seed 1_week_no_pay notification %d: %v", i, err) + } + } + if !adminnotify.CriticalLogsCapExceeded(ctx, tx, "1_week_no_pay") { + t.Fatal("expected the unacknowledged 1_week_no_pay queue to be at the cap") + } + + n, err := NotifyUnpaidOneWeek(ctx) + if err != nil { + t.Fatalf("NotifyUnpaidOneWeek failed: %v", err) + } + if n != 1 { + t.Errorf("expected candidate count 1, got %d", n) + } + + var dbCount int + err = tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = '1_week_no_pay'`).Scan(&dbCount) + if err != nil { + t.Fatalf("failed to query notification count: %v", err) + } + if dbCount != adminnotify.MaxUnacknowledgedCriticalLogs { + t.Errorf("expected the queue to stay capped at %d, got %d", adminnotify.MaxUnacknowledgedCriticalLogs, dbCount) + } +} diff --git a/backend/handlers/scheduling/time-blockers.go b/backend/handlers/scheduling/time-blockers.go index fa80fac..3301fa0 100644 --- a/backend/handlers/scheduling/time-blockers.go +++ b/backend/handlers/scheduling/time-blockers.go @@ -18,6 +18,7 @@ import ( "crussell/clock" "crussell/db" "crussell/handlers/payments" + "crussell/internal/adminnotify" "crussell/internal/dav" "crussell/internal/s3" "crussell/internal/square" @@ -506,11 +507,24 @@ func squareCleanupNotificationID(userID string) string { func insertSquareCleanupCriticalNotification(ctx context.Context, userID string) { actx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() + // Round 2 Loop B finding 1: this is a 'critical_payment_log' insert site — + // the same atomic global flood cap as every other (webhooks, account + // erasure, jwt reuse). The pre-check logs the suppression; the fold inside + // the INSERT enforces it atomically (count-then-insert). The per-user + // deterministic id (ON CONFLICT (id) DO NOTHING) is preserved, so each + // affected user still gets exactly one notification. + if adminnotify.CriticalLogsCapExceeded(actx, db.Conn, "critical_payment_log") { + slog.Error("failed to insert critical notification for failed Square erasure — unacknowledged 'critical_payment_log' queue at the cap", "user", userID) + return + } tag, err := db.Conn.Exec(actx, ` INSERT INTO admin_notifications (id, reason, booking_id, user_id, created_at) - VALUES ($1, 'critical_payment_log'::admin_notification_reason, NULL, NULL, NOW()) + SELECT $1, 'critical_payment_log'::admin_notification_reason, NULL, NULL, NOW() + WHERE (SELECT COUNT(*) FROM admin_notifications _an + WHERE _an.reason = 'critical_payment_log' + AND _an.acknowledged_at IS NULL) < $2 ON CONFLICT (id) DO NOTHING - `, squareCleanupNotificationID(userID)) + `, squareCleanupNotificationID(userID), adminnotify.MaxUnacknowledgedCriticalLogs) if err != nil { slog.Error("failed to insert critical notification for failed Square erasure", "user", userID, "err", err) return @@ -1274,15 +1288,23 @@ func CleanupExpiredDeposits(ctx context.Context) (int, error) { userIDs[i] = b.userID } - _, err = tx.Exec(ctx, ` + // C5: flood-cap the unacknowledged 'deposit_not_paid_by_deadline' queue + // (pre-check logs the suppression; the fold inside the INSERT enforces it + // atomically). The per-booking NOT EXISTS dedup is preserved. No early + // return — the time_blocker cleanup below must still run. + if adminnotify.CriticalLogsCapExceeded(ctx, tx, "deposit_not_paid_by_deadline") { + slog.Warn("suppressed deposit_not_paid_by_deadline admin notification — unacknowledged queue at the cap") + } else if _, err = tx.Exec(ctx, ` INSERT INTO admin_notifications (reason, booking_id, user_id) SELECT 'deposit_not_paid_by_deadline', unnest($1::text[]), unnest($2::text[]) WHERE NOT EXISTS ( SELECT 1 FROM admin_notifications an WHERE an.booking_id = ANY($1) AND an.reason = 'deposit_not_paid_by_deadline' ) - `, ids, userIDs) - if err != nil { + AND (SELECT COUNT(*) FROM admin_notifications _an + WHERE _an.reason = 'deposit_not_paid_by_deadline' + AND _an.acknowledged_at IS NULL) < $3 + `, ids, userIDs, adminnotify.MaxUnacknowledgedCriticalLogs); err != nil { return 0, fmt.Errorf("failed to create pending_release notifications: %w", err) } @@ -1316,14 +1338,27 @@ func CleanupExpiredDeposits(ctx context.Context) (int, error) { // drift apart. // // This function: -// 1. Finds unredeemed cards (redeemed_by IS NULL) unused for the window +// 1. Zeroes unredeemed cards (redeemed_by IS NULL) unused for the window // 2. Inserts into gift_card_expired_balances for recovery claims -// 3. Sets amount_remaining to 0 -// 4. Records transaction in gift_card_transactions +// 3. Records transaction in gift_card_transactions // // Once redeemed to an account, the balance doesn't expire (but the account can // be deleted after idle time per GDPR - see CleanupIdleAccounts). // +// M4 (expiry-sweep TOCTOU): the expiry decision is made ATOMICALLY with the +// zeroing — the SELECT reads the rolling-expiry predicate under FOR UPDATE row +// locks. A concurrent top-up either commits BEFORE the SELECT (its refreshed +// last_used_at drops the card out of the predicate → the card is never +// selected), or it blocks on the row lock until this sweep's transaction ends +// and lands on the ALREADY-zeroed card, reviving it via its own +// last_used_at/expiry refresh (→ the top-up value is preserved on the card). +// Either way the top-up's value is never destroyed. The old sweep SELECTed the +// expired cards with NO lock and zeroed them with an unconditional +// `UPDATE ... WHERE id = ANY($1)` a moment later, so a top-up committing in +// between was read as expired and then clobbered — the freshly topped-up value +// was lost. The row lock also pins the read balance until commit, so the +// recovery/audit rows below can never disagree with the zeroed value. +// // Pre-expiry email warnings are intentionally omitted: gift cards are unowned // (bought as gifts, change hands) until redeemed to an account. After redemption, // the balance is covered by CleanupIdleAccounts warnings. @@ -1342,12 +1377,17 @@ func CleanupExpiredGiftCards(ctx context.Context) (int, error) { return 0, fmt.Errorf("failed to read gift card expiry months: %w", monthsErr) } + // FOR UPDATE (M4): the predicate is re-evaluated against the latest + // committed row state while the row lock is held, so a concurrent top-up + // can neither be read as expired and clobbered, nor interleave a balance + // change between this read and the zeroing UPDATE below. rows, err := tx.Query(ctx, ` SELECT id, amount_remaining FROM gift_cards WHERE redeemed_by IS NULL AND amount_remaining > 0 AND last_used_at < NOW() - ($1 * INTERVAL '1 month') + FOR UPDATE `, expiryMonths) if err != nil { return 0, fmt.Errorf("failed to query expired gift cards: %w", err) @@ -1369,6 +1409,9 @@ func CleanupExpiredGiftCards(ctx context.Context) (int, error) { } expiredCards = append(expiredCards, card) } + if err := rows.Err(); err != nil { + return 0, fmt.Errorf("failed to iterate expired gift cards: %w", err) + } if len(expiredCards) > 0 { ids := make([]string, len(expiredCards)) diff --git a/backend/handlers/scheduling/time_blockers_test.go b/backend/handlers/scheduling/time_blockers_test.go index 1fd634f..3aaf1d9 100644 --- a/backend/handlers/scheduling/time_blockers_test.go +++ b/backend/handlers/scheduling/time_blockers_test.go @@ -27,7 +27,9 @@ import ( "time" "crussell/clock" + "crussell/db" "crussell/handlers/payments" + "crussell/internal/adminnotify" "crussell/internal/square" "crussell/mw" "crussell/testutils" @@ -2844,6 +2846,118 @@ func TestCleanupExpiredGiftCards_SkipRedeemed(t *testing.T) { } } +// TestCleanupExpiredGiftCards_ConcurrentTopUp_NotClobbered pins the M4 +// expiry-sweep TOCTOU fix: a top-up landing between the sweep reading a card as +// expired and its zeroing step must NEVER have its value destroyed. The test +// reproduces the exact race deterministically — a top-up transaction locks the +// card row, applies its refresh (adds balance, sets last_used_at = NOW()), and +// stays open while the sweep runs, so the sweep's zeroing UPDATE blocks behind +// the top-up's row lock. When the top-up commits, the sweep re-evaluates its +// rolling-expiry predicate against the refreshed row and skips the card. The +// OLD sweep (plain SELECT then unconditional `UPDATE ... WHERE id = ANY($1)`) +// fails this test: its SELECT already read the card as expired and its UPDATE +// then zeroed the freshly topped-up balance. +// +// The card row is created on the REAL pool (not inside this test's transaction) +// because the sweep runs on its own connection and must see the committed row. +// The test deliberately does NOT call t.Parallel(): its real-pool writes are +// committed to the shared test DB, and the parallel sibling sweep tests assert +// global counts on gift_card_expired_balances — a concurrent committed row from +// this test would flake them. Running sequentially isolates the real-pool rows +// (cleaned up before the parallel tests resume). +func TestCleanupExpiredGiftCards_ConcurrentTopUp_NotClobbered(t *testing.T) { + ctx := context.Background() + + const cardID = "m4race000001" + if _, err := db.Conn.Exec(ctx, ` + INSERT INTO gift_cards (id, total_funds_added, amount_remaining, last_used_at) + VALUES ($1, 50.00, 50.00, NOW() - INTERVAL '25 months') + `, cardID); err != nil { + t.Fatalf("failed to create expired-looking gift card: %v", err) + } + t.Cleanup(func() { + cctx := context.Background() + _, _ = db.Conn.Exec(cctx, `DELETE FROM gift_card_transactions WHERE gift_card_id = $1`, cardID) + _, _ = db.Conn.Exec(cctx, `DELETE FROM gift_card_expired_balances WHERE account_id IS NULL AND original_balance = $1`, 50.00) + _, _ = db.Conn.Exec(cctx, `DELETE FROM gift_cards WHERE id = $1`, cardID) + }) + + // The concurrent top-up: lock the card row FOR UPDATE, apply the same + // refresh TopUpGiftCard performs (add balance, last_used_at = NOW(), + // expiry extended), and hold the transaction OPEN so the sweep's zeroing + // UPDATE blocks on this row lock until the top-up commits. + topUpTx, err := db.Conn.Begin(ctx) + if err != nil { + t.Fatalf("failed to begin top-up transaction: %v", err) + } + _, err = topUpTx.Exec(ctx, ` + UPDATE gift_cards + SET amount_remaining = amount_remaining + 25.00, + total_funds_added = total_funds_added + 25.00, + last_used_at = NOW(), + expiry_date = NOW() + INTERVAL '1 month' + WHERE id = $1 + `, cardID) + if err != nil { + _ = topUpTx.Rollback(ctx) + t.Fatalf("failed to apply concurrent top-up: %v", err) + } + + // Run the sweep while the top-up transaction is still open: its UPDATE + // blocks on the row lock the top-up holds. + sweepDone := make(chan error, 1) + go func() { + _, err := CleanupExpiredGiftCards(context.Background()) + sweepDone <- err + }() + + // Commit the top-up — its refreshed last_used_at is now visible, so the + // sweep's re-checked predicate must skip the card. + if err := topUpTx.Commit(ctx); err != nil { + t.Fatalf("failed to commit concurrent top-up: %v", err) + } + + select { + case err := <-sweepDone: + if err != nil { + t.Fatalf("CleanupExpiredGiftCards failed: %v", err) + } + case <-time.After(30 * time.Second): + t.Fatal("CleanupExpiredGiftCards did not finish after the top-up committed — deadlock?") + } + + // The topped-up value must be intact: £50 + £25 = £75, NOT zeroed. + var remaining float64 + if err := db.Conn.QueryRow(ctx, `SELECT amount_remaining FROM gift_cards WHERE id = $1`, cardID).Scan(&remaining); err != nil { + t.Fatalf("failed to query gift card: %v", err) + } + if remaining != 75.00 { + t.Errorf("expected amount_remaining 75.00 after a concurrent top-up, got %.2f (top-up value was clobbered)", remaining) + } + + // The sweep must not have recorded an expired-balance claim or an 'expire' + // transaction for the card — the card was never expired by this run. + var expiredBalanceCount int + if err := db.Conn.QueryRow(ctx, ` + SELECT COUNT(*) FROM gift_card_expired_balances + WHERE account_id IS NULL AND original_balance = $1`, 50.00).Scan(&expiredBalanceCount); err != nil { + t.Fatalf("failed to count expired balances: %v", err) + } + if expiredBalanceCount != 0 { + t.Errorf("expected no expired-balance record for the topped-up card, got %d", expiredBalanceCount) + } + + var expireTxCount int + if err := db.Conn.QueryRow(ctx, ` + SELECT COUNT(*) FROM gift_card_transactions + WHERE gift_card_id = $1 AND transaction_type = 'expire'`, cardID).Scan(&expireTxCount); err != nil { + t.Fatalf("failed to count expire transactions: %v", err) + } + if expireTxCount != 0 { + t.Errorf("expected no 'expire' transaction for the topped-up card, got %d", expireTxCount) + } +} + // --- Tests for CleanupIdleAccounts --- // TestCleanupIdleAccounts_WithBalance verifies that an account idle for 5+ years @@ -3918,3 +4032,144 @@ func TestAnonymizeStaleGuestAccounts_InvalidatesSquareCustomerCache(t *testing.T t.Errorf("anonymized stale guest must not reuse the deleted Square customer id %q from the cache", originalID) } } + +// TestInsertSquareCleanupCriticalNotification_FloodCap pins Round 2 Loop B +// finding 1 for the time-blockers cleanup insert site: the 'critical_payment_log' +// notification is flood-capped at adminnotify.MaxUnacknowledgedCriticalLogs +// (pre-checked by CriticalLogsCapExceeded and enforced atomically inside the +// INSERT), so an unbounded flood of failed Square erasures cannot bury the +// operator. The per-user deterministic id is preserved: re-delivery of the +// same failure stays a no-op, and once the unacknowledged queue is at the cap a +// NEW user's failure is suppressed without growing the queue. +func TestInsertSquareCleanupCriticalNotification_FloodCap(t *testing.T) { + ctx, tx := resetTestData(t) + + userID, err := fixtures.CreateTestUser(tx) + if err != nil { + t.Fatalf("failed to create user: %v", err) + } + + // Below the cap: the first delivery inserts exactly one notification with + // the deterministic per-user id. + insertSquareCleanupCriticalNotification(ctx, "user-cleanup-1") + var n int + if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'critical_payment_log'`).Scan(&n); err != nil { + t.Fatalf("failed to count critical notifications: %v", err) + } + if n != 1 { + t.Fatalf("expected 1 critical notification below the cap, got %d", n) + } + + // Re-delivery of the same failure is a no-op (deterministic id dedup). + insertSquareCleanupCriticalNotification(ctx, "user-cleanup-1") + if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'critical_payment_log'`).Scan(&n); err != nil { + t.Fatalf("failed to count critical notifications: %v", err) + } + if n != 1 { + t.Errorf("expected the deterministic per-user id to dedup re-delivery, got %d notifications", n) + } + + // Fill the unacknowledged queue to the cap, then a NEW user's failure must + // be suppressed (the queue must not exceed the cap). + if _, err := tx.Exec(ctx, `DELETE FROM admin_notifications WHERE reason = 'critical_payment_log'`); err != nil { + t.Fatalf("failed to clear critical notifications: %v", err) + } + for i := 0; i < adminnotify.MaxUnacknowledgedCriticalLogs; i++ { + if _, err := tx.Exec(ctx, ` + INSERT INTO admin_notifications (reason, user_id, created_at) + VALUES ('critical_payment_log', $1, NOW()) + `, userID); err != nil { + t.Fatalf("failed to seed critical notification %d: %v", i, err) + } + } + if !adminnotify.CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") { + t.Fatal("expected the unacknowledged critical_payment_log queue to be at the cap") + } + + insertSquareCleanupCriticalNotification(ctx, "flood-user") + if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'critical_payment_log'`).Scan(&n); err != nil { + t.Fatalf("failed to count critical notifications: %v", err) + } + if n != adminnotify.MaxUnacknowledgedCriticalLogs { + t.Errorf("expected the queue to stay capped at %d, got %d", adminnotify.MaxUnacknowledgedCriticalLogs, n) + } +} + +// TestCleanupExpiredDeposits_NotificationFloodCap pins C5 for the +// 'deposit_not_paid_by_deadline' insert site: the unacknowledged queue is +// flood-capped at adminnotify.MaxUnacknowledgedCriticalLogs, so an eviction +// flood cannot bury the operator. At the cap the notification insert is +// suppressed, but the eviction + time_blocker cleanup still run. +func TestCleanupExpiredDeposits_NotificationFloodCap(t *testing.T) { + t.Parallel() + ctx, tx := resetTestData(t) + + userID, err := fixtures.CreateTestUser(tx) + if err != nil { + t.Fatalf("failed to create user: %v", err) + } + + startTime := clock.Now().Add(12 * time.Hour) + var bookingID string + err = tx.QueryRow(ctx, ` + INSERT INTO bookings (user_id, start_time, status, deposit_required) + VALUES ($1, $2, 'confirmed', true) + RETURNING id + `, userID, startTime).Scan(&bookingID) + if err != nil { + t.Fatalf("failed to create booking: %v", err) + } + + _, err = tx.Exec(ctx, ` + INSERT INTO time_blockers (start_time, duration_minutes, description) + VALUES ($1, 60, $2) + `, startTime, "RESERVATION:user:"+userID+":bk123") + if err != nil { + t.Fatalf("failed to create reservation time blocker: %v", err) + } + + // Fill the unacknowledged queue to the cap before the cleanup runs, so the + // insert site must suppress instead of growing it. + for i := 0; i < adminnotify.MaxUnacknowledgedCriticalLogs; i++ { + if _, err := tx.Exec(ctx, ` + INSERT INTO admin_notifications (reason, user_id, created_at) + VALUES ('deposit_not_paid_by_deadline', $1, NOW()) + `, userID); err != nil { + t.Fatalf("failed to seed deposit_not_paid_by_deadline notification %d: %v", i, err) + } + } + if !adminnotify.CriticalLogsCapExceeded(ctx, db.Conn, "deposit_not_paid_by_deadline") { + t.Fatal("expected the unacknowledged deposit_not_paid_by_deadline queue to be at the cap") + } + + _, err = CleanupExpiredDeposits(ctx) + if err != nil { + t.Fatalf("CleanupExpiredDeposits failed: %v", err) + } + + // The eviction + time_blocker cleanup still run; only the notification + // insert is suppressed by the cap. + var status string + if err := tx.QueryRow(ctx, "SELECT status FROM bookings WHERE id = $1", bookingID).Scan(&status); err != nil { + t.Fatalf("failed to query booking: %v", err) + } + if status != "pending_release" { + t.Errorf("expected status 'pending_release', got '%s'", status) + } + + var tbCount int + if err := tx.QueryRow(ctx, "SELECT COUNT(*) FROM time_blockers WHERE description = $1", "RESERVATION:user:"+userID+":bk123").Scan(&tbCount); err != nil { + t.Fatalf("failed to query time blockers: %v", err) + } + if tbCount != 0 { + t.Error("expected reservation time blocker to be deleted") + } + + var n int + if err := tx.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'deposit_not_paid_by_deadline'`).Scan(&n); err != nil { + t.Fatalf("failed to count deposit_not_paid_by_deadline notifications: %v", err) + } + if n != adminnotify.MaxUnacknowledgedCriticalLogs { + t.Errorf("expected the queue to stay capped at %d, got %d", adminnotify.MaxUnacknowledgedCriticalLogs, n) + } +} diff --git a/backend/internal/adminnotify/adminnotify.go b/backend/internal/adminnotify/adminnotify.go index 17edb90..bd0751f 100644 --- a/backend/internal/adminnotify/adminnotify.go +++ b/backend/internal/adminnotify/adminnotify.go @@ -16,18 +16,13 @@ // (dispute, booking, unknown-event, orphan-replay). // - handlers/user/account.go InsertSquareErasureCriticalNotification. // -// Sites owned by OTHER agents that still need the fold (coordination notes): +// Sites owned by OTHER agents (coordination notes): // -// - handlers/payments/sweep.go:1556 insertCriticalPaymentNotification (money -// agent) — its INSERT ... SELECT ... WHERE NOT EXISTS is the same -// unbounded-across-accounts shape; fold `AND (SELECT COUNT(*) FROM -// admin_notifications _an WHERE _an.reason = 'critical_payment_log' AND -// _an.acknowledged_at IS NULL) < $N` (N = MaxUnacknowledgedCriticalLogs) -// into its WHERE clause. -// - handlers/scheduling/time-blockers.go:511 insertSquareCleanupCriticalNotification -// and internal/jobs/cleanup.go:333 ScanCriticalPaymentLogs — the same -// 'critical_payment_log' insert shape on the GDPR-cleanup and log-scan -// paths (owner: the jobs/scheduling agent). +// - handlers/payments/sweep.go insertCriticalPaymentNotification — NOW capped +// (same atomic fold + pre-check as every other site). +// - handlers/scheduling/time-blockers.go insertSquareCleanupCriticalNotification +// and internal/jobs/cleanup.go ScanCriticalPaymentLogs — both NOW capped +// (same atomic fold + pre-check as every other site). // - main.go has NO admin_notifications insert sites (it only mounts the // notification read/ack routes), so nothing to cap there. // diff --git a/backend/internal/adminnotify/adminnotify_test.go b/backend/internal/adminnotify/adminnotify_test.go new file mode 100644 index 0000000..7327c10 --- /dev/null +++ b/backend/internal/adminnotify/adminnotify_test.go @@ -0,0 +1,72 @@ +//go:build test + +package adminnotify + +import ( + "context" + "os" + "testing" + + "crussell/db" + "crussell/testutils/testdb" +) + +func TestMain(m *testing.M) { + pool := testdb.CreateTestDatabase("crussell_test_adminnotify") + db.Conn = db.NewPoolProxy(pool) + code := m.Run() + testdb.DestroyTestDatabase(pool, "crussell_test_adminnotify") + os.Exit(code) +} + +// TestMaxUnacknowledgedCriticalLogs_Value pins the exported flood-cap constant +// that every capped insert site folds into its SQL (changing it re-scopes every +// site at once, so the value is part of the shared contract). +func TestMaxUnacknowledgedCriticalLogs_Value(t *testing.T) { + if MaxUnacknowledgedCriticalLogs != 100 { + t.Errorf("expected MaxUnacknowledgedCriticalLogs to be 100, got %d", MaxUnacknowledgedCriticalLogs) + } +} + +// TestCriticalLogsCapExceeded verifies the exported pre-check that logs a +// suppression: below the cap it reports false, at the cap true, and +// acknowledging rows re-arms it. +func TestCriticalLogsCapExceeded(t *testing.T) { + ctx := context.Background() + t.Cleanup(func() { + _, _ = db.Conn.Exec(ctx, "DELETE FROM admin_notifications WHERE reason = 'critical_payment_log'") + }) + + if CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") { + t.Fatal("expected the empty queue to be below the cap") + } + + for i := 0; i < MaxUnacknowledgedCriticalLogs-1; i++ { + if _, err := db.Conn.Exec(ctx, ` + INSERT INTO admin_notifications (reason, created_at) + VALUES ('critical_payment_log', NOW()) + `); err != nil { + t.Fatalf("failed to insert row %d: %v", i, err) + } + } + if CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") { + t.Fatalf("expected %d unacknowledged rows to stay below the cap", MaxUnacknowledgedCriticalLogs-1) + } + + if _, err := db.Conn.Exec(ctx, ` + INSERT INTO admin_notifications (reason, created_at) + VALUES ('critical_payment_log', NOW()) + `); err != nil { + t.Fatalf("failed to insert the cap row: %v", err) + } + if !CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") { + t.Fatalf("expected %d unacknowledged rows to be at the cap", MaxUnacknowledgedCriticalLogs) + } + + if _, err := db.Conn.Exec(ctx, "UPDATE admin_notifications SET acknowledged_at = NOW() WHERE reason = 'critical_payment_log'"); err != nil { + t.Fatalf("failed to acknowledge the queue: %v", err) + } + if CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") { + t.Fatal("expected acknowledging the queue to re-arm inserts") + } +} diff --git a/backend/internal/jobs/cleanup.go b/backend/internal/jobs/cleanup.go index 27b638d..81a0431 100644 --- a/backend/internal/jobs/cleanup.go +++ b/backend/internal/jobs/cleanup.go @@ -15,6 +15,7 @@ import ( "crussell/handlers/payments" "crussell/handlers/scheduling" "crussell/handlers/user" + "crussell/internal/adminnotify" "crussell/internal/square" "crussell/mw" @@ -324,10 +325,28 @@ const criticalPaymentStaleAge = "2 hours" // the bell never floods. Acknowledging the notification re-arms the scan: // while the row stays unresolved it is surfaced again on the next run. // +// Flood cap (Round 2 Loop B finding 1): the unacknowledged +// 'critical_payment_log' queue is globally capped at +// adminnotify.MaxUnacknowledgedCriticalLogs — the SAME shared cap as every +// other insert site (webhooks, account erasure, time blockers). The cap is +// folded into the INSERT's WHERE clause (count-then-insert is atomic, closing +// the TOCTOU where two concurrent scans could both read a below-cap count and +// overshoot together), and the pre-check logs the suppression so a capped-out +// scan stays visible to the operator. A hostile flood of unresolved money rows +// (attacker-registered accounts, a runaway reconciliation loop) must not be +// able to bury the single-operator notification centre. +// // This is the "grep CRITICAL" the audit asked for, but DB-backed since the app // has no log pipeline. When a proper log/alert pipeline (T14 Sentry) lands, // this job can be retired. func ScanCriticalPaymentLogs(ctx context.Context) (int, error) { + // Pre-check: at the cap, skip the scan entirely (the fold inside the INSERT + // enforces the cap atomically even if a concurrent insert races this check; + // the pre-check only decides whether to log the suppression). + if adminnotify.CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") { + log.Printf("[SCAN] critical_payment_log admin notification suppressed — %d unacknowledged rows at the cap; acknowledge outstanding notifications to re-arm", adminnotify.MaxUnacknowledgedCriticalLogs) + return 0, nil + } tag, err := db.Conn.Exec(ctx, ` INSERT INTO admin_notifications (reason, booking_id, created_at) SELECT DISTINCT 'critical_payment_log'::admin_notification_reason, src.booking_id, NOW() @@ -358,7 +377,14 @@ func ScanCriticalPaymentLogs(ctx context.Context) (int, error) { AND (src.booking_id IS NULL OR EXISTS ( SELECT 1 FROM bookings b WHERE b.id = src.booking_id )) - `) + -- Flood cap (Round 2 Loop B finding 1): fold the global + -- MaxUnacknowledgedCriticalLogs bound INTO the insert so the + -- count-then-insert is atomic — two concurrent scans cannot both read a + -- below-cap count and overshoot together. + AND (SELECT COUNT(*) FROM admin_notifications _an + WHERE _an.reason = 'critical_payment_log' + AND _an.acknowledged_at IS NULL) < $1 + `, adminnotify.MaxUnacknowledgedCriticalLogs) if err != nil { return 0, fmt.Errorf("failed to scan critical payment logs: %w", err) } diff --git a/backend/internal/jobs/cleanup_test.go b/backend/internal/jobs/cleanup_test.go index 162cb5c..cc099d0 100644 --- a/backend/internal/jobs/cleanup_test.go +++ b/backend/internal/jobs/cleanup_test.go @@ -13,6 +13,7 @@ import ( "crussell/db" "crussell/handlers/payments" + "crussell/internal/adminnotify" "crussell/internal/square" "crussell/testutils" "crussell/testutils/testdb" @@ -331,6 +332,60 @@ func TestScanCriticalPaymentLogs_RefundBelowCapNotNotified(t *testing.T) { } } +// TestScanCriticalPaymentLogs_FloodCapped verifies the shared +// 'critical_payment_log' flood cap (Round 2 Loop B finding 1): once the +// unacknowledged queue reaches adminnotify.MaxUnacknowledgedCriticalLogs, the +// scan inserts nothing, and acknowledging rows re-arms it. +func TestScanCriticalPaymentLogs_FloodCapped(t *testing.T) { + ctx := context.Background() + paymentID := seedStalePendingPayment(t) + t.Cleanup(func() { + _, _ = db.Conn.Exec(ctx, "DELETE FROM payments WHERE id = $1", paymentID) + _, _ = db.Conn.Exec(ctx, "DELETE FROM admin_notifications WHERE reason = 'critical_payment_log'") + }) + + // Fill the unacknowledged queue to the cap. + for i := 0; i < adminnotify.MaxUnacknowledgedCriticalLogs; i++ { + if _, err := db.Conn.Exec(ctx, ` + INSERT INTO admin_notifications (reason, created_at) + VALUES ('critical_payment_log', NOW()) + `); err != nil { + t.Fatalf("failed to fill the unacknowledged queue to the cap: %v", err) + } + } + if !adminnotify.CriticalLogsCapExceeded(ctx, db.Conn, "critical_payment_log") { + t.Fatal("expected the unacknowledged critical_payment_log queue to be at the cap") + } + + // At the cap the scan must insert nothing for the stale pending payment. + n, err := ScanCriticalPaymentLogs(ctx) + if err != nil { + t.Fatalf("ScanCriticalPaymentLogs failed: %v", err) + } + if n != 0 { + t.Errorf("expected 0 notifications inserted at the cap, got %d", n) + } + var got int + if err := db.Conn.QueryRow(ctx, `SELECT COUNT(*) FROM admin_notifications WHERE reason = 'critical_payment_log' AND acknowledged_at IS NULL`).Scan(&got); err != nil { + t.Fatalf("failed to count unacknowledged notifications: %v", err) + } + if got != adminnotify.MaxUnacknowledgedCriticalLogs { + t.Errorf("expected the queue to stay capped at %d, got %d", adminnotify.MaxUnacknowledgedCriticalLogs, got) + } + + // Acknowledging re-arms the scan while the row stays unresolved. + if _, err := db.Conn.Exec(ctx, "UPDATE admin_notifications SET acknowledged_at = NOW() WHERE reason = 'critical_payment_log' AND acknowledged_at IS NULL"); err != nil { + t.Fatalf("failed to acknowledge the queue: %v", err) + } + n, err = ScanCriticalPaymentLogs(ctx) + if err != nil { + t.Fatalf("ScanCriticalPaymentLogs failed after acknowledge: %v", err) + } + if n != 1 { + t.Errorf("expected 1 notification re-inserted after acknowledge, got %d", n) + } +} + // ============================================================ // RetryPendingSquareErasures — GDPR Square outbox job (batch-1 fix) // ============================================================