minstrel/internal/playlists/scheduler.go

// 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"
	"git.fabledsword.com/bvandeusen/minstrel/internal/eventbus"
	"git.fabledsword.com/bvandeusen/minstrel/internal/taste"
)

// 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
	bus     *eventbus.Bus // #968: announce rebuilds so clients self-refresh

	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. bus may be nil (tests); rebuild events are
// then skipped.
func NewScheduler(
	pool *pgxpool.Pool,
	logger *slog.Logger,
	dataDir string,
	bus *eventbus.Bus,
) (*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,
		bus:     bus,
		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.
//
// gocron v2 doesn't expose a per-job location option — WithLocation is
// a SchedulerOption that affects every job. To get per-user timezones
// we use a cron expression with the CRON_TZ= prefix, which the
// underlying robfig/cron parser honors. The expression "CRON_TZ=$tz
// 0 3 * * *" fires at 03:00 in the named zone every day.
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)
	}
	// Fall back to UTC if the stored tz is unparseable so a corrupted
	// DB row doesn't error out registration.
	resolvedTZ := "UTC"
	if _, err := validateTimezone(tz); err == nil {
		resolvedTZ = tz
	}
	cronExpr := fmt.Sprintf("CRON_TZ=%s 0 %d * * *", resolvedTZ, dailyRebuildHour)
	job, err := s.gocron.NewJob(
		gocron.CronJob(cronExpr, false /* withSeconds */),
		gocron.NewTask(func() {
			s.rebuildUserDaily(ctx, userID, time.Now())
		}),
	)
	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 {
		s.rebuildUserDaily(ctx, u.ID, time.Now())
	}
}

// rebuildUserDaily recomputes the user's taste profile, then rebuilds their
// system playlists + You-might-like rows. Taste runs first so the playlist
// build can read a fresh profile (phase-2 consumption); both steps are
// best-effort and a failure in one is logged without blocking the other.
func (s *Scheduler) rebuildUserDaily(ctx context.Context, userID pgtype.UUID, now time.Time) {
	if err := taste.BuildTasteProfile(ctx, s.pool, s.logger, userID, taste.DefaultConfig()); err != nil {
		s.logger.Warn("scheduler: taste profile rebuild failed",
			"user_id", uuidStringPL(userID), "err", err)
	}
	if err := BuildSystemPlaylists(ctx, s.pool, s.logger, userID, now, s.dataDir); err != nil {
		s.logger.Warn("scheduler: build failed",
			"user_id", uuidStringPL(userID), "err", err)
		return
	}
	// #968: tell the user's connected clients their home content was
	// regenerated, so a tab/app left open across the rebuild refreshes its
	// system-playlist + You-might-like views (and a stale active queue can
	// re-pull) instead of serving yesterday's snapshot until a manual reload.
	s.publishRebuilt(userID)
}

// publishRebuilt broadcasts a user-scoped "system playlists rebuilt" event.
// No-op when the bus is nil (test construction).
func (s *Scheduler) publishRebuilt(userID pgtype.UUID) {
	if s.bus == nil {
		return
	}
	s.bus.Publish(eventbus.Event{
		Kind:   "playlist.system_rebuilt",
		UserID: uuidStringPL(userID),
		Data:   map[string]any{},
	})
}

// 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
}