feat(playlists): gocron-based per-user scheduler

Builds the per-user daily-at-03:00-local scheduler for #392 Half B. Uses github.com/go-co-op/gocron/v2 with WithLocation(userTZ) for each user's job. Hourly reconciliation pass keeps the in-memory job set in sync with the active-users query. Start sequence: clear stale in_flight rows; register a daily job for every active user at their stored timezone; fire a one-shot runOnce to catch up missed schedules during downtime; start gocron. Stop drains and shuts down the gocron loop. Refresh(ctx, userID) removes the user's existing job (if any) and registers a fresh one at their current timezone — wired into the PUT /api/me/timezone handler and new-user registration in the next commit. Server struct gains PlaylistScheduler; main.go constructs it after the eventbus and threads it through to api.Mount. The existing StartSystemPlaylistCron call stays for one more commit so we don't have a window where no scheduler is running; Task 3 deletes it. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-13 11:57:13 -04:00
parent 230da7bdcb
commit 46c8edfa82
8 changed files with 339 additions and 6 deletions
@@ -0,0 +1,246 @@
+// Per-user system-playlist scheduler (#392 Half B). Replaces the
+// 24h-anchored cron loop in system_cron.go (deleted in the same arc).
+//
+// Each active user gets one gocron daily job at 03:00 in their stored
+// timezone. The scheduler reconciles its in-memory job set with the
+// "active users" query once per hour so newly-active / no-longer-
+// active users are picked up without restarts. New-user registration
+// and PUT /api/me/timezone both call Refresh() so changes take effect
+// synchronously without waiting for the hourly pass.
+//
+// Scheduler ownership: cmd/minstrel/main.go constructs one instance
+// at startup, calls Start, and stops it on graceful shutdown. The
+// server.Server struct holds a pointer so the API layer can call
+// Refresh from handlers.
+
+package playlists
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/go-co-op/gocron/v2"
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5/pgtype"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"git.fabledsword.com/bvandeusen/minstrel/internal/db/dbq"
+)
+
+// dailyRebuildHour is the local-time hour at which each user's system
+// playlists rebuild. 03:00 is well past most "today" listening and
+// before morning use; matches the operator framing in the design doc.
+const dailyRebuildHour = 3
+
+// reconciliationInterval drives the hourly pass that brings the
+// in-memory job set in sync with the live "active users" query.
+const reconciliationInterval = time.Hour
+
+// Scheduler owns one gocron daily job per active user. Refresh and
+// the hourly reconciliation pass mutate s.jobs under s.mu.
+type Scheduler struct {
+	gocron  gocron.Scheduler
+	pool    *pgxpool.Pool
+	logger  *slog.Logger
+	dataDir string
+
+	mu   sync.Mutex
+	jobs map[pgtype.UUID]uuid.UUID // user_id → gocron job id
+}
+
+// NewScheduler builds an idle scheduler. Caller must invoke Start
+// before any builds fire.
+func NewScheduler(pool *pgxpool.Pool, logger *slog.Logger, dataDir string) (*Scheduler, error) {
+	g, err := gocron.NewScheduler()
+	if err != nil {
+		return nil, fmt.Errorf("init gocron: %w", err)
+	}
+	return &Scheduler{
+		gocron:  g,
+		pool:    pool,
+		logger:  logger,
+		dataDir: dataDir,
+		jobs:    map[pgtype.UUID]uuid.UUID{},
+	}, nil
+}
+
+// Start clears any stale in_flight rows from a previous crash,
+// registers a daily-3am job for every active user at their stored
+// timezone, fires a one-shot runOnce to catch up missed schedules,
+// and starts the gocron loop.
+//
+// ctx scopes BuildSystemPlaylists calls fired by the jobs. Cancel
+// the parent context to drain in-progress builds on shutdown.
+func (s *Scheduler) Start(ctx context.Context) error {
+	q := dbq.New(s.pool)
+	if err := q.ClearStaleSystemPlaylistInFlight(ctx); err != nil {
+		s.logger.Warn("scheduler: clear stale in_flight failed", "err", err)
+		// Non-fatal — TryClaimSystemPlaylistRun will block any
+		// concurrent build attempts; stale rows just stay stuck.
+	}
+
+	users, err := q.ListActiveUsersWithTimezones(ctx)
+	if err != nil {
+		return fmt.Errorf("list active users: %w", err)
+	}
+
+	s.mu.Lock()
+	for _, u := range users {
+		if err := s.registerLocked(ctx, u.ID, u.Timezone); err != nil {
+			s.logger.Warn("scheduler: register at startup failed",
+				"user_id", uuidStringPL(u.ID), "err", err)
+		}
+	}
+	s.mu.Unlock()
+
+	// One-shot startup catch-up: build for every active user right
+	// now, regardless of their scheduled time. Mirrors the existing
+	// system_cron.go startup runOnce so users who missed yesterday's
+	// schedule (process down) get caught up.
+	go s.runStartupCatchUp(ctx, users)
+
+	// Hourly reconciliation pass.
+	if _, err := s.gocron.NewJob(
+		gocron.DurationJob(reconciliationInterval),
+		gocron.NewTask(func() {
+			s.reconcile(ctx)
+		}),
+	); err != nil {
+		return fmt.Errorf("register reconciliation job: %w", err)
+	}
+
+	s.gocron.Start()
+	s.logger.Info("scheduler: started", "users", len(users))
+	return nil
+}
+
+// Refresh removes the user's existing job (if any) and registers a
+// fresh one at their current timezone. Safe to call from any
+// goroutine. Returns an error only on DB lookup failure; gocron-side
+// errors are logged but not propagated since they don't represent
+// recoverable conditions for the caller.
+func (s *Scheduler) Refresh(ctx context.Context, userID pgtype.UUID) error {
+	q := dbq.New(s.pool)
+	user, err := q.GetUserByID(ctx, userID)
+	if err != nil {
+		return fmt.Errorf("get user: %w", err)
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.registerLocked(ctx, userID, user.Timezone)
+}
+
+// registerLocked replaces the user's job entry. Caller must hold s.mu.
+func (s *Scheduler) registerLocked(ctx context.Context, userID pgtype.UUID, tz string) error {
+	if existing, ok := s.jobs[userID]; ok {
+		if err := s.gocron.RemoveJob(existing); err != nil {
+			s.logger.Warn("scheduler: remove existing job failed",
+				"user_id", uuidStringPL(userID), "err", err)
+		}
+		delete(s.jobs, userID)
+	}
+	loc := validateTimezoneOrUTC(tz)
+	job, err := s.gocron.NewJob(
+		gocron.DailyJob(1, gocron.NewAtTimes(
+			gocron.NewAtTime(dailyRebuildHour, 0, 0),
+		)),
+		gocron.NewTask(func() {
+			if err := BuildSystemPlaylists(ctx, s.pool, s.logger, userID, time.Now(), s.dataDir); err != nil {
+				s.logger.Warn("scheduler: build failed",
+					"user_id", uuidStringPL(userID), "err", err)
+			}
+		}),
+		gocron.WithLocation(loc),
+	)
+	if err != nil {
+		return fmt.Errorf("register daily job: %w", err)
+	}
+	s.jobs[userID] = job.ID()
+	return nil
+}
+
+// reconcile syncs the in-memory job set with the live active-users
+// query. Newly-active users get jobs; no-longer-active users have
+// their jobs removed. Runs hourly.
+func (s *Scheduler) reconcile(ctx context.Context) {
+	q := dbq.New(s.pool)
+	users, err := q.ListActiveUsersWithTimezones(ctx)
+	if err != nil {
+		s.logger.Warn("scheduler: reconcile list failed", "err", err)
+		return
+	}
+
+	active := make(map[pgtype.UUID]string, len(users))
+	for _, u := range users {
+		active[u.ID] = u.Timezone
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	// Add users newly active.
+	for id, tz := range active {
+		if _, ok := s.jobs[id]; !ok {
+			if err := s.registerLocked(ctx, id, tz); err != nil {
+				s.logger.Warn("scheduler: reconcile add failed",
+					"user_id", uuidStringPL(id), "err", err)
+			}
+		}
+	}
+
+	// Remove users no longer active.
+	for id := range s.jobs {
+		if _, ok := active[id]; !ok {
+			if jobID, exists := s.jobs[id]; exists {
+				if err := s.gocron.RemoveJob(jobID); err != nil {
+					s.logger.Warn("scheduler: reconcile remove failed",
+						"user_id", uuidStringPL(id), "err", err)
+				}
+				delete(s.jobs, id)
+			}
+		}
+	}
+}
+
+// runStartupCatchUp fires BuildSystemPlaylists once for every active
+// user at startup, regardless of scheduled time. Matches the
+// behavior of the previous system_cron.go startup runOnce so users
+// whose scheduled fire was missed during downtime get caught up.
+func (s *Scheduler) runStartupCatchUp(ctx context.Context, users []dbq.ListActiveUsersWithTimezonesRow) {
+	for _, u := range users {
+		if err := BuildSystemPlaylists(ctx, s.pool, s.logger, u.ID, time.Now(), s.dataDir); err != nil {
+			s.logger.Warn("scheduler: startup catch-up build failed",
+				"user_id", uuidStringPL(u.ID), "err", err)
+		}
+	}
+}
+
+// Stop drains gocron and stops the scheduler loop.
+func (s *Scheduler) Stop() {
+	if err := s.gocron.Shutdown(); err != nil {
+		s.logger.Warn("scheduler: shutdown failed", "err", err)
+	}
+}
+
+// validateTimezone returns the parsed Location for tz, or an error.
+// Callers should use this when accepting input from clients.
+func validateTimezone(tz string) (*time.Location, error) {
+	if tz == "" {
+		return nil, fmt.Errorf("timezone is empty")
+	}
+	return time.LoadLocation(tz)
+}
+
+// validateTimezoneOrUTC returns the parsed Location for tz, or
+// time.UTC if parsing fails. Used internally by the scheduler so
+// corrupted DB values don't crash the process.
+func validateTimezoneOrUTC(tz string) *time.Location {
+	loc, err := validateTimezone(tz)
+	if err != nil {
+		return time.UTC
+	}
+	return loc
+}
@@ -0,0 +1,51 @@
+package playlists
+
+import (
+	"testing"
+)
+
+func TestValidateTimezone_Valid(t *testing.T) {
+	cases := []string{"UTC", "America/New_York", "Europe/London", "Asia/Tokyo"}
+	for _, tz := range cases {
+		t.Run(tz, func(t *testing.T) {
+			loc, err := validateTimezone(tz)
+			if err != nil {
+				t.Fatalf("validateTimezone(%q): %v", tz, err)
+			}
+			if loc == nil {
+				t.Fatalf("validateTimezone(%q): nil location", tz)
+			}
+		})
+	}
+}
+
+func TestValidateTimezone_Invalid(t *testing.T) {
+	cases := []string{"", "Not/A/Zone", "GMT+5", "EST5EDT-not-iana"}
+	for _, tz := range cases {
+		t.Run(tz, func(t *testing.T) {
+			_, err := validateTimezone(tz)
+			if err == nil {
+				t.Fatalf("validateTimezone(%q): expected error, got nil", tz)
+			}
+		})
+	}
+}
+
+func TestValidateTimezoneOrUTC_FallbackToUTC(t *testing.T) {
+	// validateTimezoneOrUTC returns UTC for any unparseable input.
+	// Used by the scheduler when reading stored timezones — corrupted
+	// DB values should not crash; they should fall back to UTC.
+	loc := validateTimezoneOrUTC("Not/A/Zone")
+	if loc.String() != "UTC" {
+		t.Errorf("expected UTC fallback for invalid input, got %s", loc)
+	}
+
+	// Valid input returns the named location, not UTC.
+	loc = validateTimezoneOrUTC("America/New_York")
+	if loc.String() == "UTC" {
+		t.Errorf("valid input incorrectly fell back to UTC")
+	}
+	if loc.String() != "America/New_York" {
+		t.Errorf("expected America/New_York, got %s", loc)
+	}
+}