package jobs import ( "context" "database/sql" "errors" "fmt" "log" "log/slog" "time" "crussell/auth" "crussell/db" authHandlers "crussell/handlers/auth" "crussell/handlers/payments" "crussell/handlers/scheduling" "crussell/handlers/user" "crussell/internal/square" "crussell/mw" "github.com/jackc/pgx/v5" ) // RegisterAll registers every background maintenance job on the scheduler. // Call once during server startup, before s.Start(). func RegisterAll(s *Scheduler) { // === HIGH FREQUENCY — every 5 minutes === s.Register(Job{ Name: "cleanup-reservations", Schedule: "*/5 * * * *", Timeout: 30 * time.Second, Concurrency: 1, Handler: scheduling.CleanupOldReservations, }) s.Register(Job{ Name: "cleanup-expired-deposits", Schedule: "*/5 * * * *", Timeout: 30 * time.Second, Concurrency: 1, Handler: scheduling.CleanupExpiredDeposits, }) s.Register(Job{ Name: "cleanup-rate-limiters", Schedule: "*/5 * * * *", Timeout: 30 * time.Second, Concurrency: 1, Handler: mw.CleanupAllRateLimiters, }) s.Register(Job{ Name: "cleanup-gdpr-export-cache", Schedule: "*/5 * * * *", Timeout: 10 * time.Second, Concurrency: 1, Handler: user.CleanupGDPRExportCache, }) s.Register(Job{ Name: "sweep-pending-square-refunds", Schedule: "*/5 * * * *", Timeout: 60 * time.Second, Concurrency: 1, Handler: payments.SweepPendingSquareRefunds, }) // Offset from the refund sweep (which also writes payments rows) by one // minute to avoid the two sweeps contending on the same table. s.Register(Job{ Name: "sweep-stale-pending-payments", Schedule: "1,6,11,16,21,26,31,36,41,46,51,56 * * * *", Timeout: 60 * time.Second, Concurrency: 1, Handler: payments.SweepStalePendingPayments, }) // Cancels terminal (card-machine) checkouts still pending at Square after // an hour — a never-polled checkout would otherwise sit live indefinitely // and complete into an invisible, untracked charge. s.Register(Job{ Name: "sweep-stale-terminal-checkouts", Schedule: "*/15 * * * *", Timeout: 60 * time.Second, Concurrency: 1, Handler: payments.SweepStaleTerminalCheckouts, }) // === MID FREQUENCY — every minute (progressive rate limiter was on 30s) === s.Register(Job{ Name: "cleanup-progressive-rate-limiter", Schedule: "* * * * *", Timeout: 10 * time.Second, Concurrency: 1, Handler: mw.CleanupProgressiveRateLimiter, }) // === MID FREQUENCY — hourly === s.Register(Job{ Name: "cleanup-expired-loyalty-redemptions", Schedule: "0 * * * *", Timeout: 30 * time.Second, Concurrency: 1, Handler: scheduling.CleanupExpiredLoyaltyRedemptions, }) s.Register(Job{ Name: "cleanup-old-idempotency-keys", Schedule: "0 * * * *", Timeout: 30 * time.Second, Concurrency: 1, Handler: scheduling.CleanupOldIdempotencyKeys, }) s.Register(Job{ Name: "cleanup-revoked-jtis", Schedule: "0 * * * *", Timeout: 30 * time.Second, Concurrency: 1, Handler: auth.CleanupRevokedJTIs, }) s.Register(Job{ Name: "cleanup-stale-login-entries", Schedule: "0 * * * *", Timeout: 30 * time.Second, Concurrency: 1, Handler: authHandlers.CleanupStaleLoginEntries, }) // === LOW FREQUENCY — daily, off-peak (staggered to avoid DB contention) === s.Register(Job{ Name: "anonymize-stale-guest-accounts", Schedule: "0 3 * * *", Timeout: 5 * time.Minute, Concurrency: 1, Handler: scheduling.AnonymizeStaleGuestAccounts, }) s.Register(Job{ Name: "cleanup-expired-financial-records", Schedule: "0 4 * * *", Timeout: 10 * time.Minute, Concurrency: 1, Handler: scheduling.CleanupExpiredFinancialRecords, }) s.Register(Job{ Name: "cleanup-idle-accounts", Schedule: "30 3 * * *", Timeout: 5 * time.Minute, Concurrency: 1, Handler: scheduling.CleanupIdleAccounts, }) s.Register(Job{ Name: "cleanup-expired-gift-cards", Schedule: "0 5 * * *", Timeout: 5 * time.Minute, Concurrency: 1, Handler: scheduling.CleanupExpiredGiftCards, }) s.Register(Job{ Name: "cleanup-old-name-history", Schedule: "30 4 * * *", Timeout: 30 * time.Second, Concurrency: 1, Handler: scheduling.CleanupOldNameHistory, }) // === STAGED HOURS CHANGE === s.Register(Job{ Name: "apply-default-hours", Schedule: "5 0 * * *", // Daily at 00:05 — after midnight to avoid race Timeout: 30 * time.Second, Concurrency: 1, Handler: scheduling.ApplyScheduledDefaultHours, }) // === BUSINESS LOGIC JOBS === s.Register(Job{ Name: "notify-unpaid-1-week", Schedule: "0 7 * * *", // Daily at 7am — end of business day + 7 days Timeout: 2 * time.Minute, Concurrency: 1, Handler: scheduling.NotifyUnpaidOneWeek, }) s.Register(Job{ Name: "notify-unpaid-1-month", Schedule: "30 7 * * *", // Daily at 7:30am (staggered from notify-unpaid-1-week) Timeout: 2 * time.Minute, Concurrency: 1, Handler: scheduling.NotifyUnpaidOneMonth, }) s.Register(Job{ Name: "transition-discount-campaigns", Schedule: "0 * * * *", // Hourly Timeout: 30 * time.Second, Concurrency: 1, Handler: scheduling.TransitionDiscountCampaigns, }) s.Register(Job{ Name: "cleanup-verification-codes", Schedule: "0 2 * * *", // Daily at 2am Timeout: 30 * time.Second, Concurrency: 1, Handler: scheduling.CleanupExpiredVerificationCodes, }) s.Register(Job{ Name: "cleanup-refresh-tokens", Schedule: "0 2 * * *", // Daily at 2am Timeout: 30 * time.Second, 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, }) // Surfaces unresolved money events (stale pending payments/till sales, // refunds at the retry cap) as admin notifications so the owner sees the // "money may have moved at Square but the DB couldn't record it" situations // that would otherwise live only in un-watched CRITICAL log lines (Gap // Backlog T14). Retire this job when a proper log/alert pipeline (T14 // Sentry) lands — the notifications page is the stopgap while the app has // none. Staggered at 2:45am between the 2am and 3am daily batches. s.Register(Job{ Name: "scan-critical-payment-logs", Schedule: "45 2 * * *", // Daily at 2:45am Timeout: 30 * time.Second, Concurrency: 1, Handler: ScanCriticalPaymentLogs, }) // Durable safety net for the GDPR account-deletion Square outbox (Fault // A1): retries the Square card/customer deletions that the // DeleteAccountHandler async cleanup could not finish (process crash or // exhausted retries), using the outbox rows the handler persisted inside // its anonymization tx before it committed. Hourly — the async goroutine // handles the common case within seconds, so this only catches stragglers. // The schedule is deliberately unshared so a long run cannot contend with // the payment sweeps. s.Register(Job{ Name: "retry-square-erasures", Schedule: "17 * * * *", // Hourly at :17 Timeout: 30 * time.Second, Concurrency: 1, Handler: RetryPendingSquareErasures, }) } // SweepSquareWebhookEvents deletes square_webhook_events rows older than the // shared auth.RefreshTokenLifetime window (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. The 90-day window deliberately reuses // auth.RefreshTokenLifetime so the two coinciding retention windows cannot // drift apart. 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() - make_interval(days => $1) `, int64(auth.RefreshTokenLifetime/(24*time.Hour))) if err != nil { return 0, fmt.Errorf("failed to sweep square webhook events: %w", err) } return int(tag.RowsAffected()), tx.Commit(ctx) } // criticalPaymentStaleAge is how long a pending payment / till sale may sit // unresolved before the scan surfaces it to the admin notification centre. A // pending row with no update in this window is exactly the "money may have // moved at Square but the DB couldn't record it" situation the CRITICAL payment // logs describe (handlers.go, giftcards.go, refunds.go, till.go). Deliberately // shorter than the 24h stale-pending sweep (handlers/payments/sweep.go) so the // owner hears about it while the row could still be rescued. const criticalPaymentStaleAge = "2 hours" // ScanCriticalPaymentLogs surfaces unresolved critical payment states as admin // notifications (reason='critical_payment_log'), giving the owner an in-app // view of money events that would otherwise be visible only in un-watched // CRITICAL log lines (the app has no log/alert pipeline — Gap Backlog T14). // // Candidates: // - payments / till_sales rows still 'pending' with no update for >2h: a // lost-response charge the DB never recorded. // - refunds rows still 'pending' at the 3-attempt retry cap: a refund the // sweep could not resolve (money may have moved at Square). // // Each candidate inserts one admin_notifications row keyed on (reason, // booking_id) — but ONLY if no unacknowledged notification for that key // already exists (NULL-safe via IS NOT DISTINCT FROM, matching the // insertRefundFailedNotifications dedup in handlers/payments/refunds.go), so // the bell never floods. Acknowledging the notification re-arms the scan: // while the row stays unresolved it is surfaced again on the next run. // // 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) { 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() FROM ( -- Lost-response payments: still pending with no update past the threshold. SELECT id, booking_id FROM payments WHERE status = 'pending' AND updated_at < NOW() - INTERVAL '`+criticalPaymentStaleAge+`' UNION ALL -- Lost-response till sales (no booking linkage — booking_id NULL). SELECT id, NULL::CHAR(12) FROM till_sales WHERE status = 'pending' AND updated_at < NOW() - INTERVAL '`+criticalPaymentStaleAge+`' UNION ALL -- Refunds stuck at the 3-attempt retry cap. SELECT id, booking_id FROM refunds WHERE status = 'pending' AND refund_attempts >= 3 ) src WHERE NOT EXISTS ( SELECT 1 FROM admin_notifications an WHERE an.reason = 'critical_payment_log' AND an.booking_id IS NOT DISTINCT FROM src.booking_id AND an.acknowledged_at IS NULL ) -- admin_notifications.booking_id is a hard FK to bookings(id); a -- candidate whose booking was hard-deleted (7-year retention) would -- otherwise fail the whole INSERT and silently kill ALL critical -- alerts. Skip orphans: keep NULL booking_ids (till sales) and only -- payments/refunds whose booking still exists. AND (src.booking_id IS NULL OR EXISTS ( SELECT 1 FROM bookings b WHERE b.id = src.booking_id )) `) if err != nil { return 0, fmt.Errorf("failed to scan critical payment logs: %w", err) } n := int(tag.RowsAffected()) if n > 0 { log.Printf("[SCAN] Inserted %d admin_notification(s) for unresolved critical payment states (pending payments/till sales > %s, refunds at retry cap)", n, criticalPaymentStaleAge) } return n, nil } // erasureNotificationKey derives the notification-dedup key for a pending // outbox row: the user id when the row still carries one (registered users // whose deletion tx has not yet run), otherwise a stable row-scoped key (guest // rows are unlinked by delete_guest_user and the account-deletion outbox // NULLs user_id on the rows it writes). func erasureNotificationKey(userID sql.NullString, rowID string) string { if userID.Valid && userID.String != "" { return userID.String } return "row:" + rowID } // raiseErasureNotification raises the deduped critical notification once per // key. Repeated failures across job runs collapse to a single alert (the // deterministic admin_notifications id in // user.InsertSquareErasureCriticalNotification does the ON CONFLICT dedup). func raiseErasureNotification(ctx context.Context, key string, notified map[string]bool) { if notified[key] { return } notified[key] = true user.InsertSquareErasureCriticalNotification(ctx, key) } // RetryPendingSquareErasures is the durable safety net for the account-deletion // Square outbox (Fault A1). DeleteAccountHandler persists the Square // card/customer erasure targets on the scrubbed, soft-deleted user_saved_cards // rows (last_4 = 'XXXX') inside its anonymization transaction, before it // commits. If the process crashes between that commit and the async cleanup // goroutine finishing, those rows are the only remaining record of the // Square-side PII — this job finds them and retries the Square deletion so the // card/customer is never permanently orphaned at Square. On success (or a // Square NOT_FOUND — the data is already gone) it drains the outbox columns; on // final failure it raises a critical payment notification, deduped per affected // user/row. Returns the number of outbox rows drained. func RetryPendingSquareErasures(ctx context.Context) (int, error) { if payments.SquareClient == nil { // Square not configured: no external erasure is possible, and the // handler only writes outbox entries when a client was configured. return 0, nil } client := payments.SquareClient rows, err := db.Conn.Query(ctx, ` SELECT id, user_id, square_card_id, square_customer_id FROM user_saved_cards WHERE deleted_at IS NOT NULL AND last_4 = 'XXXX' AND (square_card_id IS NOT NULL OR square_customer_id IS NOT NULL) ORDER BY id `) if err != nil { return 0, fmt.Errorf("failed to query pending Square erasures: %w", err) } defer rows.Close() type pendingErasure struct { rowID string userID sql.NullString cardID sql.NullString customerID sql.NullString } var pending []pendingErasure for rows.Next() { var p pendingErasure if err := rows.Scan(&p.rowID, &p.userID, &p.cardID, &p.customerID); err != nil { return 0, fmt.Errorf("failed to scan pending Square erasure: %w", err) } pending = append(pending, p) } if err := rows.Err(); err != nil { return 0, fmt.Errorf("failed to iterate pending Square erasures: %w", err) } if len(pending) == 0 { return 0, nil } // Group the outbox rows: each card id maps to exactly one row (per-user // UNIQUE), each customer id may map to several rows (shared across the // user's saved cards) and may also appear on other deleted accounts' rows. cardRows := map[string][]pendingErasure{} customerRows := map[string][]pendingErasure{} for _, p := range pending { if p.cardID.Valid && p.cardID.String != "" { cardRows[p.cardID.String] = append(cardRows[p.cardID.String], p) } if p.customerID.Valid && p.customerID.String != "" { customerRows[p.customerID.String] = append(customerRows[p.customerID.String], p) } } notified := map[string]bool{} drained := map[string]bool{} // Cards: each ccof: token is erased once. NOT_FOUND means Square no longer // has the card — the erasure is complete, so the outbox is drained rather // than alerted on. for cardID, cardPend := range cardRows { rowID := cardPend[0].rowID err := user.RetrySquareDeletion(ctx, func(actx context.Context) error { return client.DeleteCardOnFile(actx, cardID) }) if err != nil && !square.IsNotFound(err) { raiseErasureNotification(ctx, erasureNotificationKey(cardPend[0].userID, rowID), notified) log.Printf("Error: retry-square-erasures failed to delete Square card %s (outbox row %s): %v", square.TokenPrefix(cardID), rowID, err) slog.Error("square card erasure retry failed after attempts", "row", rowID, "card", square.TokenPrefix(cardID), "error", err) continue } if _, err := db.Conn.Exec(ctx, ` UPDATE user_saved_cards SET square_card_id = NULL WHERE id = $1 AND deleted_at IS NOT NULL `, rowID); err != nil { return 0, fmt.Errorf("failed to clear card erasure outbox row %s: %w", rowID, err) } drained[rowID] = true } // Customers: one DeleteCustomer per distinct id, guarded by the // still-referenced-by-another-account check (a shared Square customer must // survive while any active card of another account references it). for customerID, custPend := range customerRows { stillReferenced := false for _, p := range custPend { var ref bool if err := db.Conn.QueryRow(ctx, ` SELECT EXISTS( SELECT 1 FROM user_saved_cards WHERE square_customer_id = $1 AND deleted_at IS NULL AND user_id IS DISTINCT FROM $2 ) `, customerID, p.userID).Scan(&ref); err != nil { return 0, fmt.Errorf("failed to check Square customer %s references before deletion: %w", square.TokenPrefix(customerID), err) } if ref { stillReferenced = true break } } if stillReferenced { // Deliberately kept (shared customer) — not a pending erasure. // Drain the outbox rows so the job stops retrying a deletion that // must not happen; the customer is erased when the last referencing // account is itself erased. for _, p := range custPend { if _, err := db.Conn.Exec(ctx, ` UPDATE user_saved_cards SET square_customer_id = NULL WHERE id = $1 AND deleted_at IS NOT NULL `, p.rowID); err != nil { return 0, fmt.Errorf("failed to clear skipped customer erasure outbox row %s: %w", p.rowID, err) } drained[p.rowID] = true } continue } err := user.RetrySquareDeletion(ctx, func(actx context.Context) error { return client.DeleteCustomer(actx, customerID) }) if err != nil && !square.IsNotFound(err) { for _, p := range custPend { raiseErasureNotification(ctx, erasureNotificationKey(p.userID, p.rowID), notified) } log.Printf("Error: retry-square-erasures failed to delete Square customer %s (%d outbox rows): %v", square.TokenPrefix(customerID), len(custPend), err) slog.Error("square customer erasure retry failed after attempts", "customer", square.TokenPrefix(customerID), "rows", len(custPend), "error", err) continue } for _, p := range custPend { if _, err := db.Conn.Exec(ctx, ` UPDATE user_saved_cards SET square_customer_id = NULL WHERE id = $1 AND deleted_at IS NOT NULL `, p.rowID); err != nil { return 0, fmt.Errorf("failed to clear customer erasure outbox row %s: %w", p.rowID, err) } drained[p.rowID] = true } } if n := len(drained); n > 0 { log.Printf("[ERASURE] retry-square-erasures drained %d pending Square erasure outbox row(s)", n) } return len(drained), nil }