popertots
69a854d857
- adminnotify: MaxUnacknowledgedCriticalLogs global cap exposed as CriticalLogsCapExceeded — a pre-check helper every insert site pairs with the atomic fold inside its INSERT (count-then-insert is atomic, closing the TOCTOU where concurrent inserts could both read a below-cap count). - jobs/cleanup.go ScanCriticalPaymentLogs: capped at the shared cap, pre-check skips the scan and logs the suppression. - scheduling: 1_week_no_pay, 1_month_no_pay, default_hours_changed, deposit_not_paid_by_deadline and the Square-erasure critical notification all flood-capped with pre-check + atomic fold (per-booking/per-user dedup kept). - time-blockers.go CleanupExpiredGiftCards (M4): the expiry SELECT now runs under FOR UPDATE row locks so the read-expired-then-zero window is atomic — a concurrent top-up either commits before the SELECT (refreshed last_used_at drops the card out of the predicate) or blocks until the sweep's tx ends and revives the zeroed card via its own expiry refresh; the top-up value can never be destroyed by the sweep. - flood-cap tests added for 1_week_no_pay; adminnotify unit coverage added.
430 lines
14 KiB
Go
430 lines
14 KiB
Go
package scheduling
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"crussell/auth"
|
|
"crussell/clock"
|
|
"crussell/db"
|
|
"crussell/internal/adminnotify"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
// NotifyUnpaidOneWeek inserts admin_notifications for bookings that ended
|
|
// 7+ days ago with no completed payment. Runs daily.
|
|
// TODO: Notify affected user via email/SMS when SMTP is wired (E5).
|
|
func NotifyUnpaidOneWeek(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)
|
|
}
|
|
}()
|
|
|
|
rows, err := tx.Query(ctx, `
|
|
SELECT b.id, b.user_id
|
|
FROM bookings b
|
|
WHERE b.status NOT IN ('client_cancelled', 'we_cancelled')
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM payments p
|
|
WHERE p.booking_id = b.id AND p.status = 'completed'
|
|
)
|
|
AND b.end_time >= NOW() - INTERVAL '30 days'
|
|
AND b.end_time < NOW() - INTERVAL '7 days'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM admin_notifications an
|
|
WHERE an.booking_id = b.id AND an.reason = '1_week_no_pay'
|
|
)
|
|
`)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to query unpaid 1-week bookings: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var ids, userIDs []string
|
|
for rows.Next() {
|
|
var id, userID string
|
|
if err := rows.Scan(&id, &userID); err != nil {
|
|
return 0, fmt.Errorf("failed to scan row: %w", err)
|
|
}
|
|
ids = append(ids, id)
|
|
userIDs = append(userIDs, userID)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return 0, fmt.Errorf("rows iteration error: %w", err)
|
|
}
|
|
|
|
if len(ids) == 0 {
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return 0, fmt.Errorf("failed to commit: %w", err)
|
|
}
|
|
return 0, nil
|
|
}
|
|
|
|
// 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[])
|
|
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)
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return 0, fmt.Errorf("failed to commit: %w", err)
|
|
}
|
|
|
|
return len(ids), nil
|
|
}
|
|
|
|
// NotifyUnpaidOneMonth inserts admin_notifications for bookings that ended
|
|
// 30+ days ago with no completed payment. Runs daily.
|
|
// TODO: Notify affected user via email/SMS when SMTP is wired (E5).
|
|
func NotifyUnpaidOneMonth(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)
|
|
}
|
|
}()
|
|
|
|
rows, err := tx.Query(ctx, `
|
|
SELECT b.id, b.user_id
|
|
FROM bookings b
|
|
WHERE b.status NOT IN ('client_cancelled', 'we_cancelled')
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM payments p
|
|
WHERE p.booking_id = b.id AND p.status = 'completed'
|
|
)
|
|
AND b.end_time < NOW() - INTERVAL '30 days'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM admin_notifications an
|
|
WHERE an.booking_id = b.id AND an.reason = '1_month_no_pay'
|
|
)
|
|
`)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to query unpaid 1-month bookings: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
|
|
var ids, userIDs []string
|
|
for rows.Next() {
|
|
var id, userID string
|
|
if err := rows.Scan(&id, &userID); err != nil {
|
|
return 0, fmt.Errorf("failed to scan row: %w", err)
|
|
}
|
|
ids = append(ids, id)
|
|
userIDs = append(userIDs, userID)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return 0, fmt.Errorf("rows iteration error: %w", err)
|
|
}
|
|
|
|
if len(ids) == 0 {
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return 0, fmt.Errorf("failed to commit: %w", err)
|
|
}
|
|
return 0, nil
|
|
}
|
|
|
|
// 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[])
|
|
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)
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return 0, fmt.Errorf("failed to commit: %w", err)
|
|
}
|
|
|
|
return len(ids), nil
|
|
}
|
|
|
|
// TransitionDiscountCampaigns auto-transitions campaign statuses based on
|
|
// dates and redemption limits:
|
|
// - draft → active when start_date <= NOW()
|
|
// - active → completed when end_date < NOW() or max_redemptions reached
|
|
func TransitionDiscountCampaigns(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)
|
|
}
|
|
}()
|
|
|
|
result, err := tx.Exec(ctx, `
|
|
UPDATE discount_campaigns
|
|
SET status = 'active'
|
|
WHERE status = 'draft'
|
|
AND campaign_type = 'time_based'
|
|
AND start_date <= NOW()
|
|
`)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to activate campaigns: %w", err)
|
|
}
|
|
activated := result.RowsAffected()
|
|
|
|
result, err = tx.Exec(ctx, `
|
|
UPDATE discount_campaigns
|
|
SET status = 'completed'
|
|
WHERE status = 'active'
|
|
AND (
|
|
end_date < NOW()
|
|
OR (max_redemptions IS NOT NULL AND times_redeemed >= max_redemptions)
|
|
)
|
|
`)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to complete campaigns: %w", err)
|
|
}
|
|
completed := result.RowsAffected()
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return 0, fmt.Errorf("failed to commit: %w", err)
|
|
}
|
|
|
|
return int(activated + completed), nil
|
|
}
|
|
|
|
// CleanupExpiredVerificationCodes deletes expired verification codes and
|
|
// used codes older than 30 days.
|
|
func CleanupExpiredVerificationCodes(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)
|
|
}
|
|
}()
|
|
|
|
result, err := tx.Exec(ctx, `
|
|
DELETE FROM verification_codes
|
|
WHERE (used_at IS NOT NULL AND used_at < NOW() - INTERVAL '30 days')
|
|
OR (expires_at < NOW() AND used_at IS NULL)
|
|
`)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to cleanup verification codes: %w", err)
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return 0, fmt.Errorf("failed to commit: %w", err)
|
|
}
|
|
|
|
return int(result.RowsAffected()), nil
|
|
}
|
|
|
|
// CleanupExpiredRefreshTokens deletes expired refresh tokens and
|
|
// revoked tokens older than the shared RefreshTokenLifetime window (auth).
|
|
func CleanupExpiredRefreshTokens(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)
|
|
}
|
|
}()
|
|
|
|
// LOW 6 / finding 3: an expired refresh token whose family has no other
|
|
// live members is a family kill. Invalidate the family-alive cache
|
|
// (auth/jwt.go) for the affected families AFTER the commit so bound access
|
|
// tokens re-check the DB instead of riding the 30s cache TTL. The family
|
|
// ids are captured before the delete and invalidated after the commit: a
|
|
// pre-commit invalidation could race a concurrent VerifyToken that
|
|
// re-caches the still-present row as alive.
|
|
//
|
|
// RESIDUAL WINDOW (documented, Loop B finding 3): the invalidation is an
|
|
// in-memory cache operation that CANNOT be in the same transaction as the
|
|
// DELETE. A crash between the DELETE commit and the
|
|
// auth.InvalidateFamilyAliveBatch call leaves the cache warm for up to
|
|
// familyAliveCacheTTL, admitting a bound access token after its family was
|
|
// killed. The verify path closes this gap itself: verifyFamilyAlive
|
|
// (auth/jwt.go) re-validates an ALIVE cache verdict against the DB when it
|
|
// is within familyAliveRecheckGrace of its TTL, bounding the residual
|
|
// window to the grace (5s of 30s). The remaining window is a process crash
|
|
// exactly between the two statements — accepted and documented here.
|
|
rows, err := tx.Query(ctx, `
|
|
SELECT DISTINCT family_id FROM refresh_tokens
|
|
WHERE expires_at < NOW()
|
|
OR (revoked = TRUE AND created_at < NOW() - make_interval(days => $1))
|
|
`, int64(auth.RefreshTokenLifetime/(24*time.Hour)))
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to select expired refresh token families: %w", err)
|
|
}
|
|
var familyIDs []string
|
|
for rows.Next() {
|
|
var familyID sql.NullString
|
|
if err := rows.Scan(&familyID); err != nil {
|
|
rows.Close()
|
|
return 0, fmt.Errorf("failed to scan expired refresh token family: %w", err)
|
|
}
|
|
if familyID.Valid && familyID.String != "" {
|
|
familyIDs = append(familyIDs, familyID.String)
|
|
}
|
|
}
|
|
rows.Close()
|
|
if err := rows.Err(); err != nil {
|
|
return 0, fmt.Errorf("failed to iterate expired refresh token families: %w", err)
|
|
}
|
|
|
|
result, err := tx.Exec(ctx, `
|
|
DELETE FROM refresh_tokens
|
|
WHERE expires_at < NOW()
|
|
OR (revoked = TRUE AND created_at < NOW() - make_interval(days => $1))
|
|
`, int64(auth.RefreshTokenLifetime/(24*time.Hour)))
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to cleanup refresh tokens: %w", err)
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return 0, fmt.Errorf("failed to commit: %w", err)
|
|
}
|
|
|
|
auth.InvalidateFamilyAliveBatch(familyIDs)
|
|
|
|
return int(result.RowsAffected()), nil
|
|
}
|
|
|
|
// LondonDateString returns the Europe/London calendar date (YYYY-MM-DD) for
|
|
// the given instant. ApplyScheduledDefaultHours uses this so staged
|
|
// default-hours changes roll over at London midnight rather than UTC midnight:
|
|
// during BST a London date begins at 23:00 UTC the previous day, so using the
|
|
// UTC date would misdate the effective day inside the 00:00-01:00 BST window.
|
|
func LondonDateString(t time.Time) string {
|
|
return t.In(clock.London).Format("2006-01-02")
|
|
}
|
|
|
|
// ApplyScheduledDefaultHours applies any pending default hours changes that
|
|
// have reached their effective_date. Runs daily at 00:05 to catch midnight
|
|
// roll-overs even if the cron is slightly delayed.
|
|
func ApplyScheduledDefaultHours(ctx context.Context) (int, error) {
|
|
todayStr := LondonDateString(clock.Now())
|
|
|
|
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)
|
|
}
|
|
}()
|
|
|
|
// Find any changes where effective_date <= today (London) and not yet applied/cancelled
|
|
var rowID int
|
|
var hoursJSON string
|
|
err = tx.QueryRow(ctx, `
|
|
SELECT id, hours::text
|
|
FROM default_hours_scheduled_changes
|
|
WHERE effective_date <= $1::date
|
|
AND applied_at IS NULL
|
|
AND cancelled_at IS NULL
|
|
LIMIT 1
|
|
`, todayStr).Scan(&rowID, &hoursJSON)
|
|
if err != nil {
|
|
// No matching change — nothing to do
|
|
return 0, tx.Commit(ctx)
|
|
}
|
|
|
|
// Parse the staged hours
|
|
var hours []struct {
|
|
Weekday int `json:"weekday"`
|
|
StartTime string `json:"startTime"`
|
|
EndTime string `json:"endTime"`
|
|
IsOpen bool `json:"isOpen"`
|
|
}
|
|
if err := json.Unmarshal([]byte(hoursJSON), &hours); err != nil {
|
|
return 0, fmt.Errorf("failed to parse hours JSON for change %d: %w", rowID, err)
|
|
}
|
|
|
|
// Bulk-update working_hours with staged values
|
|
weekdays := make([]int, len(hours))
|
|
startTimes := make([]string, len(hours))
|
|
endTimes := make([]string, len(hours))
|
|
isOpenFlags := make([]bool, len(hours))
|
|
for i, h := range hours {
|
|
weekdays[i] = h.Weekday
|
|
startTimes[i] = h.StartTime
|
|
endTimes[i] = h.EndTime
|
|
isOpenFlags[i] = h.IsOpen
|
|
}
|
|
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE working_hours AS wh
|
|
SET start_time = v.start_time,
|
|
end_time = v.end_time,
|
|
is_open = v.is_open
|
|
FROM (
|
|
SELECT unnest($1::smallint[]) AS weekday,
|
|
unnest($2::time[]) AS start_time,
|
|
unnest($3::time[]) AS end_time,
|
|
unnest($4::boolean[]) AS is_open
|
|
) v
|
|
WHERE wh.weekday = v.weekday
|
|
`, weekdays, startTimes, endTimes, isOpenFlags); err != nil {
|
|
return 0, fmt.Errorf("failed to update working_hours for change %d: %w", rowID, err)
|
|
}
|
|
|
|
// Mark change as applied
|
|
if _, err := tx.Exec(ctx, `
|
|
UPDATE default_hours_scheduled_changes
|
|
SET applied_at = NOW()
|
|
WHERE id = $1
|
|
`, rowID); err != nil {
|
|
return 0, fmt.Errorf("failed to mark change %d as applied: %w", rowID, err)
|
|
}
|
|
|
|
// 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)
|
|
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)
|
|
}
|
|
|
|
if err := tx.Commit(ctx); err != nil {
|
|
return 0, fmt.Errorf("failed to commit: %w", err)
|
|
}
|
|
|
|
slog.Info("Applied scheduled default hours change", "change_id", rowID, "effective_date", todayStr)
|
|
return 1, nil
|
|
}
|