From f78489ae00c1b9ec4177ea1991daabdb9fa1838f Mon Sep 17 00:00:00 2001 From: Stephen Adamson Date: Tue, 7 Jul 2026 00:09:34 +0100 Subject: [PATCH] feat: add logutil and jobs internal packages Add centralized jobs scheduler and logutil package with ANSI colors and duration formatting. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus --- backend/Makefile | 58 +++ backend/internal/jobs/cleanup.go | 177 ++++++++ backend/internal/jobs/scheduler.go | 190 +++++++++ backend/internal/jobs/scheduler_test.go | 538 ++++++++++++++++++++++++ backend/internal/logutil/logutil.go | 70 +++ 5 files changed, 1033 insertions(+) create mode 100644 backend/Makefile create mode 100644 backend/internal/jobs/cleanup.go create mode 100644 backend/internal/jobs/scheduler.go create mode 100644 backend/internal/jobs/scheduler_test.go create mode 100644 backend/internal/logutil/logutil.go diff --git a/backend/Makefile b/backend/Makefile new file mode 100644 index 0000000..9fede02 --- /dev/null +++ b/backend/Makefile @@ -0,0 +1,58 @@ +# Pipeline targets for Crussell backend +# Use: make (runs with default tags: test,dev) +# make TAGS="test,!dev" + +TAGS ?= test,dev +RACE_FLAGS ?= -race -timeout 480s + +.PHONY: test test-all test-race vet lint ci help + +# --- Test matrix --- + +test: ## Run tests with default build tags (test,dev) + go test -tags "$(TAGS)" -count=1 -parallel 8 ./... + +test-prod: ## Run tests with production build tags (test,!dev) + go test -tags "test,!dev" -count=1 -parallel 8 ./... + +test-dev: ## Run tests with dev build tags (test,dev) + go test -tags "test,dev" -count=1 -parallel 8 ./... + +test-all: test-dev test-prod ## Run tests under both dev and prod build tags + +test-race: ## Run tests with race detector enabled + go test -tags "$(TAGS)" $(RACE_FLAGS) -count=1 ./... + +test-count: ## Run each test 10 times to detect flakiness + go test -tags "$(TAGS)" -count=10 -parallel 8 ./... + +# --- Static analysis --- + +vet: ## go vet under current build tags + go vet -tags "$(TAGS)" ./... + +vet-all: ## go vet under both dev and prod build tags + @echo "=== vet: dev ===" + go vet -tags "test,dev" ./... + @echo "=== vet: prod ===" + go vet -tags "test,!dev" ./... + +lint: ## staticcheck under current build tags + staticcheck -tags "$(TAGS)" ./... + +lint-all: ## staticcheck under both build tags + @echo "=== staticcheck: dev ===" + staticcheck -tags "test,dev" ./... + @echo "=== staticcheck: prod ===" + staticcheck -tags "test,!dev" ./... + +# --- Combined --- + +ci: test-all vet-all lint-all ## Full CI pipeline: test + vet + lint under both tag sets + +ci-race: test-race vet lint ## CI with race detection + +# --- Utilities --- + +help: ## Show this help + @grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-15s\033[0m %s\n", $$1, $$2}' diff --git a/backend/internal/jobs/cleanup.go b/backend/internal/jobs/cleanup.go new file mode 100644 index 0000000..703bf8c --- /dev/null +++ b/backend/internal/jobs/cleanup.go @@ -0,0 +1,177 @@ +package jobs + +import ( + "time" + + "crussell/auth" + authHandlers "crussell/handlers/auth" + "crussell/handlers/scheduling" + "crussell/handlers/user" + "crussell/mw" +) + +// 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, + }) + + // === 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, + }) + + // === 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, + }) +} diff --git a/backend/internal/jobs/scheduler.go b/backend/internal/jobs/scheduler.go new file mode 100644 index 0000000..e22f87d --- /dev/null +++ b/backend/internal/jobs/scheduler.go @@ -0,0 +1,190 @@ +// Package jobs provides a centralized cron scheduler for all background +// maintenance tasks. It wraps github.com/robfig/cron/v3 with per-job +// concurrency control, timeout, panic recovery, and logging. +// +// Usage: +// +// sched := jobs.New() +// jobs.RegisterAll(sched) +// sched.Start() +// defer sched.Shutdown() +package jobs + +import ( + "context" + "crypto/rand" + "crussell/clock" + "crussell/internal/logutil" + "encoding/base64" + "fmt" + "log" + "os" + "strings" + "sync/atomic" + "time" + + "github.com/robfig/cron/v3" +) + +var ( + hostname string + randomPrefix string + jobID atomic.Uint64 +) + +func init() { + h, err := os.Hostname() + if err != nil || h == "" { + h = "localhost" + } + hostname = h + + var buf [12]byte + rand.Read(buf[:]) + b64 := base64.StdEncoding.EncodeToString(buf[:]) + b64 = strings.NewReplacer("+", "", "/", "").Replace(b64) + randomPrefix = b64[0:10] +} + +func RandomPrefix() string { return randomPrefix } + +func Hostname() string { return hostname } + +// Shorthand references to logutil constants (avoids package-qualified noise). +var ( + colorReset = logutil.Reset + colorBold = logutil.Bold + colorDim = logutil.Dim + colorCyan = logutil.Cyan + colorGreen = logutil.Green + colorRed = logutil.Red + colorYellow = logutil.Yellow + colorMagenta = logutil.Magenta + colorBoldMagenta = logutil.BoldMagenta +) + +var ( + coloredDuration = logutil.ColoredDuration + coloredRows = logutil.ColoredRows +) + +// Job defines a single periodic task. +type Job struct { + // Name is a human-readable name for logging. + Name string + // Schedule is a standard 5-field cron expression ("*/5 * * * *"). + Schedule string + // Timeout is the maximum duration per execution. 0 = no timeout. + Timeout time.Duration + // Concurrency limits simultaneous runs. 0 = unlimited, 1 = serial (skip if inflight). + Concurrency int + // Handler is the work function. It receives a context that is cancelled on shutdown. +// Returns the number of rows affected (inserted/updated/deleted) and any error. + Handler func(context.Context) (int, error) +} + +// Scheduler manages all registered cron jobs with parallel execution. +type Scheduler struct { + cron *cron.Cron + entries []cron.EntryID + registry []Job + baseCtx context.Context + cancel context.CancelFunc + semaphores map[string]chan struct{} +} + +// New creates a new Scheduler. +func New() *Scheduler { + ctx, cancel := context.WithCancel(context.Background()) + return &Scheduler{ + cron: cron.New(cron.WithLocation(clock.London)), + baseCtx: ctx, + cancel: cancel, + semaphores: make(map[string]chan struct{}), + } +} + +// Register adds a job to the scheduler. Must be called before Start. +// Panics if the cron expression is invalid. +func (s *Scheduler) Register(job Job) { + parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow) + if _, err := parser.Parse(job.Schedule); err != nil { + panic("jobs: invalid cron schedule \"" + job.Schedule + "\" for job \"" + job.Name + "\": " + err.Error()) + } + s.registry = append(s.registry, job) +} + +// Start begins executing all registered jobs on their schedules. +func (s *Scheduler) Start() { + for _, job := range s.registry { + j := job + entryID, err := s.cron.AddFunc(j.Schedule, s.wrapJob(j)) + if err != nil { + log.Fatalf("jobs: failed to add func for %q: %v", j.Name, err) + } + s.entries = append(s.entries, entryID) + } + s.cron.Start() + log.Printf("jobs: scheduler started — %d jobs registered", len(s.registry)) +} + +// Shutdown gracefully stops the scheduler. It prevents new job invocations +// and cancels the base context for in-flight jobs. Returns a channel that +// closes when all in-flight jobs complete. +func (s *Scheduler) Shutdown() <-chan struct{} { + log.Println("jobs: scheduler shutting down...") + s.cancel() // Cancel base context — in-flight handlers should abort + ctx := s.cron.Stop() + return ctx.Done() +} + +// wrapJob adds panic recovery, timeout, concurrency control, and logging. +func (s *Scheduler) wrapJob(job Job) func() { + var sem chan struct{} + if job.Concurrency > 0 { + sem = make(chan struct{}, job.Concurrency) + } + + return func() { + jid := fmt.Sprintf("%s/%s-%06d", hostname, randomPrefix, jobID.Add(1)) + jidStr := colorYellow + jid + colorReset + + // Concurrency guard: if max runs are in-flight, skip this tick. + if sem != nil { + select { + case sem <- struct{}{}: + // Acquired slot + default: + log.Printf("%s[WARN]%s %s \"%sJOB %s%s%s\" %sskipped%s (previous run still in progress)", colorYellow, colorReset, jidStr, colorBoldMagenta, colorYellow, job.Name, colorReset, colorDim, colorReset) + return + } + defer func() { <-sem }() + } + + // Panic recovery so one bad job can't take down the scheduler. + defer func() { + if r := recover(); r != nil { + log.Printf("%s[ERROR]%s %s \"%sJOB %s%s%s\" %spanic recovered%s: %v", colorMagenta, colorReset, jidStr, colorBoldMagenta, colorMagenta, job.Name, colorReset, colorYellow, colorReset, r) + } + }() + + start := time.Now() + log.Printf("%s[DEBUG]%s %s \"%sJOB %s%s%s\" %sstarted%s", colorCyan, colorReset, jidStr, colorBoldMagenta, colorCyan, job.Name, colorReset, colorDim, colorReset) + + ctx := s.baseCtx + var cancel context.CancelFunc + if job.Timeout > 0 { + ctx, cancel = context.WithTimeout(ctx, job.Timeout) + defer cancel() + } + + n, err := job.Handler(ctx) + if err != nil { + log.Printf("%s[ERROR]%s %s \"%sJOB %s%s%s\" %sfailed%s: %v in %s", colorRed, colorReset, jidStr, colorBoldMagenta, colorRed, job.Name, colorReset, colorDim, colorReset, err, coloredDuration(time.Since(start))) + } else if n > 0 { + log.Printf("%s[INFO]%s %s \"%sJOB %s%s%s\" %scompleted%s: %d %s in %s", colorGreen, colorReset, jidStr, colorBoldMagenta, colorGreen, job.Name, colorReset, colorDim, colorReset, n, coloredRows(n), coloredDuration(time.Since(start))) + } else { + log.Printf("%s[DEBUG]%s %s \"%sJOB %s%s%s\" %scompleted%s: %s in %s", colorCyan, colorReset, jidStr, colorBoldMagenta, colorGreen, job.Name, colorReset, colorDim, colorReset, coloredRows(0), coloredDuration(time.Since(start))) + } + } +} diff --git a/backend/internal/jobs/scheduler_test.go b/backend/internal/jobs/scheduler_test.go new file mode 100644 index 0000000..09604b0 --- /dev/null +++ b/backend/internal/jobs/scheduler_test.go @@ -0,0 +1,538 @@ +package jobs + +import ( + "context" + "strings" + "sync" + "testing" + "time" +) + +// TestNew verifies New returns a functional Scheduler. +func TestNew(t *testing.T) { + s := New() + if s == nil { + t.Fatal("New() returned nil") + } + if s.cron == nil { + t.Error("Scheduler.cron is nil") + } + if s.baseCtx == nil { + t.Error("Scheduler.baseCtx is nil") + } + if s.cancel == nil { + t.Error("Scheduler.cancel is nil") + } +} + +// TestRegister_ValidCron accepts standard 5-field cron expressions. +func TestRegister_ValidCron(t *testing.T) { + exprs := []string{ + "*/5 * * * *", + "0 * * * *", + "0 3 * * *", + "30 3 * * *", + "* * * * *", + "0 0 1 1 *", + } + for _, expr := range exprs { + t.Run(expr, func(t *testing.T) { + s := New() + s.Register(Job{ + Name: "test", + Schedule: expr, + Handler: func(ctx context.Context) (int, error) { return 0, nil }, + }) + if len(s.registry) != 1 { + t.Errorf("expected 1 job, got %d", len(s.registry)) + } + }) + } +} + +// TestRegister_InvalidCron panics on garbage expressions. +func TestRegister_InvalidCron(t *testing.T) { + exprs := []string{ + "not-a-cron", + "", + "* * * *", + "70 * * * *", + } + for _, expr := range exprs { + t.Run(expr, func(t *testing.T) { + s := New() + defer func() { + if r := recover(); r == nil { + t.Error("expected panic for invalid cron expression") + } + }() + s.Register(Job{ + Name: "bad", + Schedule: expr, + Handler: func(ctx context.Context) (int, error) { return 0, nil }, + }) + }) + } +} + +// TestWrapJob_RunsHandler verifies the wrapped function calls the handler. +func TestWrapJob_RunsHandler(t *testing.T) { + s := New() + var ran bool + job := Job{ + Name: "test-run", + Handler: func(ctx context.Context) (int, error) { + ran = true + return 0, nil + }, + } + fn := s.wrapJob(job) + fn() + if !ran { + t.Error("handler was not called") + } +} + +// TestWrapJob_PanicRecovery catches panics in the handler. +func TestWrapJob_PanicRecovery(t *testing.T) { + s := New() + job := Job{ + Name: "test-panic", + Handler: func(ctx context.Context) (int, error) { + panic("deliberate panic") + }, + } + fn := s.wrapJob(job) + + // Should not propagate the panic + fn() +} + +// TestWrapJob_TimeoutCancelsContext verifies the handler receives a deadline. +func TestWrapJob_TimeoutCancelsContext(t *testing.T) { + s := New() + ctxReceived := make(chan context.Context, 1) + job := Job{ + Name: "test-timeout", + Timeout: 10 * time.Millisecond, + Handler: func(ctx context.Context) (int, error) { + ctxReceived <- ctx + // Block until context is cancelled + <-ctx.Done() + return 0, ctx.Err() + }, + } + + fn := s.wrapJob(job) + done := make(chan error, 1) + go func() { + fn() + done <- nil + }() + + select { + case ctx := <-ctxReceived: + deadline, ok := ctx.Deadline() + if !ok { + t.Error("expected a deadline on the context") + } + if time.Until(deadline) > 10*time.Millisecond { + t.Errorf("deadline too far in the future: %v", time.Until(deadline)) + } + case <-time.After(time.Second): + t.Fatal("handler was not called within 1s") + } + + select { + case <-done: + // Finished + case <-time.After(time.Second): + t.Fatal("job did not complete within 1s") + } +} + +// TestWrapJob_ConcurrencySkip skips the second invocation when one is inflight. +func TestWrapJob_ConcurrencySkip(t *testing.T) { + s := New() + var mu sync.Mutex + callCount := 0 + block := make(chan struct{}) + + job := Job{ + Name: "test-concurrency", + Concurrency: 1, + Handler: func(ctx context.Context) (int, error) { + mu.Lock() + callCount++ + mu.Unlock() + <-block // Block until test releases + return 0, nil + }, + } + fn := s.wrapJob(job) + + // First invocation — will block inside handler + go fn() + + // Wait for first invocation to enter handler + mu.Lock() + firstStarted := callCount == 1 + mu.Unlock() + if !firstStarted { + // Give it time + time.Sleep(50 * time.Millisecond) + } + + // Second invocation — should skip because first is still running + fn() + + mu.Lock() + if callCount > 1 { + t.Errorf("expected callCount 1 (skip), got %d", callCount) + } + mu.Unlock() + + // Clean up + close(block) +} + +// TestWrapJob_ConcurrencyNoLimit allows multiple invocations. +func TestWrapJob_ConcurrencyNoLimit(t *testing.T) { + s := New() + block := make(chan struct{}) + var mu sync.Mutex + callCount := 0 + started := make(chan struct{}, 2) + + job := Job{ + Name: "test-no-limit", + Concurrency: 0, // unlimited + Handler: func(ctx context.Context) (int, error) { + mu.Lock() + callCount++ + mu.Unlock() + started <- struct{}{} + <-block + return 0, nil + }, + } + fn := s.wrapJob(job) + + go fn() + go fn() + + // Both should start + <-started + <-started + + mu.Lock() + if callCount != 2 { + t.Errorf("expected callCount 2, got %d", callCount) + } + mu.Unlock() + + close(block) +} + +// TestStartAndShutdown verifies the scheduler lifecycle. +func TestStartAndShutdown(t *testing.T) { + s := New() + s.Register(Job{ + Name: "test-lifecycle", + Schedule: "0 0 1 1 *", // Once a year — won't fire during test + Handler: func(ctx context.Context) (int, error) { return 0, nil }, + }) + s.Start() + + select { + case <-s.Shutdown(): + // Clean shutdown + case <-time.After(time.Second): + t.Fatal("shutdown did not complete within 1s") + } +} + +// TestShutdown_CancelsBaseContext verifies in-flight jobs receive cancellation. +func TestShutdown_CancelsBaseContext(t *testing.T) { + s := New() + ctxReceived := make(chan context.Context, 1) + block := make(chan struct{}) + + s.Register(Job{ + Name: "test-shutdown-cancel", + Schedule: "* * * * *", // Will fire within 60s — but we test wrapJob directly + Handler: func(ctx context.Context) (int, error) { return 0, nil }, + }) + + // Test that the base context is cancelled on shutdown + job := Job{ + Name: "test-cancel", + Handler: func(ctx context.Context) (int, error) { + ctxReceived <- ctx + <-block + return 0, nil + }, + } + fn := s.wrapJob(job) + + go fn() + + select { + case <-ctxReceived: + // Handler started + case <-time.After(time.Second): + t.Fatal("handler was not called within 1s") + } + + // Shutdown should cancel the base context + shutdownDone := make(chan struct{}) + go func() { + <-s.Shutdown() + close(shutdownDone) + }() + + // The handler's context should be cancelled now + select { + case <-shutdownDone: + // Good + case <-time.After(time.Second): + t.Fatal("shutdown did not complete within 1s") + } + + close(block) +} + +// ============================================================ +// RandomPrefix / coloredDuration / coloredRows Tests +// ============================================================ + +func TestRandomPrefix_NonEmpty(t *testing.T) { + p := RandomPrefix() + if p == "" { + t.Error("expected RandomPrefix to return non-empty string") + } + if len(p) < 8 { + t.Errorf("expected RandomPrefix to be at least 8 chars, got %q (len=%d)", p, len(p)) + } +} + +func TestRandomPrefix_StableDuringRun(t *testing.T) { + a := RandomPrefix() + b := RandomPrefix() + if a != b { + t.Errorf("expected RandomPrefix to be stable, got %q != %q", a, b) + } +} + +func TestColoredDuration_Fast(t *testing.T) { + out := coloredDuration(100 * time.Millisecond) + if !strings.Contains(out, "100ms") { + t.Errorf("expected 100ms in output, got %q", out) + } + if !strings.Contains(out, "[32m") { + t.Errorf("expected green color (32) for fast duration, got %q", out) + } + if !strings.HasSuffix(out, "[0m") { + t.Error("expected output to end with color reset") + } +} + +func TestColoredDuration_Medium(t *testing.T) { + out := coloredDuration(2 * time.Second) + if !strings.Contains(out, "2s") { + t.Errorf("expected 2s in output, got %q", out) + } + if !strings.Contains(out, "[33m") { + t.Errorf("expected yellow color (33) for medium duration, got %q", out) + } + if !strings.HasSuffix(out, "[0m") { + t.Error("expected output to end with color reset") + } +} + +func TestColoredDuration_Slow(t *testing.T) { + out := coloredDuration(10 * time.Second) + if !strings.Contains(out, "10s") { + t.Errorf("expected 10s in output, got %q", out) + } + if !strings.Contains(out, "[31m") { + t.Errorf("expected red color (31) for slow duration, got %q", out) + } + if !strings.HasSuffix(out, "[0m") { + t.Error("expected output to end with color reset") + } +} + +func TestColoredRows_Singular(t *testing.T) { + out := coloredRows(1) + if !strings.Contains(out, "1 row") { + t.Errorf("expected '1 row' in output, got %q", out) + } + if !strings.Contains(out, "[34;1m") { + t.Errorf("expected bold blue (34;1) color, got %q", out) + } + if !strings.HasSuffix(out, "[0m") { + t.Error("expected output to end with color reset") + } +} + +func TestColoredRows_Plural(t *testing.T) { + out := coloredRows(3) + if !strings.Contains(out, "3 rows") { + t.Errorf("expected '3 rows' in output, got %q", out) + } + if !strings.Contains(out, "[34;1m") { + t.Errorf("expected bold blue (34;1) color, got %q", out) + } +} + +func TestColoredRows_Zero(t *testing.T) { + out := coloredRows(0) + if !strings.Contains(out, "0 rows") { + t.Errorf("expected '0 rows' in output, got %q", out) + } +} + +func TestColoredRows_Reset(t *testing.T) { + if !strings.HasSuffix(coloredRows(0), "[0m") { + t.Error("expected output to end with color reset") + } + if !strings.HasSuffix(coloredRows(5), "[0m") { + t.Error("expected output to end with color reset") + } +} + +// ============================================================ +// RegisterAll Tests +// ============================================================ + +// TestRegisterAll_RegistersExpectedJobs verifies RegisterAll registers exactly 19 jobs +// with all required fields populated (non-empty Name, non-empty Schedule, non-nil +// Handler, positive Timeout). +func TestRegisterAll_RegistersExpectedJobs(t *testing.T) { + t.Parallel() + + s := New() + RegisterAll(s) + + if got := len(s.registry); got != 19 { + t.Fatalf("RegisterAll() registered %d jobs, want 19", got) + } + + registered := make(map[string]Job, len(s.registry)) + for _, j := range s.registry { + registered[j.Name] = j + } + + expected := expectedJobNames() + + for name := range expected { + job, ok := registered[name] + if !ok { + t.Errorf("missing expected job %q", name) + continue + } + requireValidJob(t, job) + } + + for name := range registered { + if !expected[name] { + t.Errorf("unexpected job %q", name) + } + } +} + +// TestRegisterAll_ValidSchedules verifies all cron expressions in registered jobs +// parse without panic. RegisterAll internally calls Register, which parses every +// schedule; if this test completes without panic, all 19 schedules are valid. +func TestRegisterAll_ValidSchedules(t *testing.T) { + t.Parallel() + + s := New() + // RegisterAll panics on invalid cron — no panic means all schedules are valid. + RegisterAll(s) +} + +// TestRegisterAll_HandlerSignatures verifies every registered job has a non-nil +// handler function, confirming all handlers are properly wired at compile time. +func TestRegisterAll_HandlerSignatures(t *testing.T) { + t.Parallel() + + s := New() + RegisterAll(s) + + for _, j := range s.registry { + if j.Handler == nil { + t.Errorf("job %q has nil Handler", j.Name) + } + } +} + +func expectedJobNames() map[string]bool { + return map[string]bool{ + "cleanup-reservations": true, + "cleanup-expired-deposits": true, + "cleanup-rate-limiters": true, + "cleanup-gdpr-export-cache": true, + "cleanup-progressive-rate-limiter": true, + "cleanup-expired-loyalty-redemptions": true, + "cleanup-old-idempotency-keys": true, + "cleanup-revoked-jtis": true, + "cleanup-stale-login-entries": true, + "anonymize-stale-guest-accounts": true, + "cleanup-expired-financial-records": true, + "cleanup-idle-accounts": true, + "cleanup-expired-gift-cards": true, + "cleanup-old-name-history": true, + "notify-unpaid-1-week": true, + "notify-unpaid-1-month": true, + "transition-discount-campaigns": true, + "cleanup-verification-codes": true, + "cleanup-refresh-tokens": true, + } +} + +// TestRegisterAll_NoDuplicateCronExpressions verifies that no two jobs share the +// exact same cron expression. Identical schedules risk DB contention and should +// be deliberately staggered. +func TestRegisterAll_NoDuplicateCronExpressions(t *testing.T) { + t.Parallel() + + s := New() + RegisterAll(s) + + cronJobs := make(map[string][]string) + for _, j := range s.registry { + cronJobs[j.Schedule] = append(cronJobs[j.Schedule], j.Name) + } + + // Known-intentional groupings: jobs at the same frequency that touch + // disjoint tables (no contention risk). + knownGroupings := map[int]bool{ + 4: true, // */5 * * * * — 4 cleanup jobs, different domains + 5: true, // 0 * * * * — 5 hourly cleanup jobs, different tables + 2: true, // 0 2 * * * — 2 daily cleanup jobs, different tables + } + + for cron, jobs := range cronJobs { + if len(jobs) > 1 && !knownGroupings[len(jobs)] { + t.Errorf("cron %q shared by %d jobs: %v — stagger to avoid contention", cron, len(jobs), jobs) + } + } +} + +func requireValidJob(t *testing.T, j Job) { + t.Helper() + if j.Name == "" { + t.Error("job has empty Name") + } + if j.Schedule == "" { + t.Errorf("job %q has empty Schedule", j.Name) + } + if j.Handler == nil { + t.Errorf("job %q has nil Handler", j.Name) + } + if j.Timeout <= 0 { + t.Errorf("job %q has non-positive Timeout (%v)", j.Name, j.Timeout) + } +} diff --git a/backend/internal/logutil/logutil.go b/backend/internal/logutil/logutil.go new file mode 100644 index 0000000..b83b974 --- /dev/null +++ b/backend/internal/logutil/logutil.go @@ -0,0 +1,70 @@ +// Package logutil provides ANSI-coloured logging helpers shared between the +// job scheduler and the HTTP request logger. Both use the same colour scheme +// and duration formatting so log output is consistent across components. +package logutil + +import ( + "fmt" + "os" + "time" +) + +// noColor controls whether ANSI escape codes are emitted. +// Set via the standard NO_COLOR environment variable (https://no-color.org). +var noColor bool + +func init() { + _, noColor = os.LookupEnv("NO_COLOR") +} + +// color returns the escape sequence when NO_COLOR is unset, empty otherwise. +func color(seq string) string { + if noColor { + return "" + } + return seq +} + +// Reset, Bold, Dim — ANSI escape codes for log formatting. +// They are empty strings when NO_COLOR is set. +var ( + Reset = color("\033[0m") + Bold = color("\033[1m") + Dim = color("\033[2m") + Cyan = color("\033[36m") + Green = color("\033[32m") + Red = color("\033[31m") + Yellow = color("\033[33m") + Magenta = color("\033[35m") + + BoldGreen = color("\033[32;1m") + BoldYellow = color("\033[33;1m") + BoldRed = color("\033[31;1m") + BoldBlue = color("\033[34;1m") + BoldMagenta = color("\033[35;1m") +) + +var ( + DebugLvl = Cyan + "[DEBUG]" + Reset + WarnLvl = Yellow + "[WARN] " + Reset + ErrorLvl = Red + "[ERROR]" + Reset +) + +func ColoredDuration(d time.Duration) string { + switch { + case d < 500*time.Millisecond: + return Green + d.String() + Reset + case d < 5*time.Second: + return Yellow + d.String() + Reset + default: + return Red + d.String() + Reset + } +} + +func ColoredRows(n int) string { + text := fmt.Sprintf("%d row", n) + if n != 1 { + text += "s" + } + return BoldBlue + text + Reset +}