Documentation
¶
Overview ¶
Package scheduler owns the periodic deep-analysis tick and its sub-workers (recent-history replay + recently-added sweep).
Responsibilities:
- Schedule a periodic deep-analysis run on a fixed Go-duration interval (default 24h), matching the fleet docker-*-scheduler convention: one pass at startup (when the last run is older than one interval) plus a time.Ticker every interval thereafter. The pass is a safety net over the real-time WebSocket listener, so a drifting wall-clock start hour is immaterial; using an interval rather than an absolute HH:MM boundary means the app reads no local wall-clock time (no TZ / time/tzdata dependency).
- Fan out per-item work across a bounded worker pool with a circuit breaker that aborts the pass after a threshold of consecutive per-item failures.
- Persist the last-run marker through api.Cache so a cold restart does not double-run the analysis.
Stable contracts preserved (keep these exact: Loki alerts grep the log strings and the on-disk cache schema depends on the field names):
- WARN slog keys ("scheduler: aborting history processing after consecutive failures", "scheduler: failed to fetch history", "scheduler: failed to fetch sections", "scheduler: deep analysis already in progress, skipping") byte-for-byte identical.
- INFO slog keys ("scheduler enabled", "scheduled deep analysis starting", "running initial deep analysis", "deep analysis completed", "scheduler: processing recently added episode", "scheduler stopped") identical.
- /config/cache.json schema unchanged — LastSchedulerRun reads and writes go through api.Cache and are tagged by the concrete internal/cache package.
Consumer note: scheduler depends on api.PlexReader, api.Cache, api.UserLookup, and a Syncer interface satisfied by *sync.Syncer (declared locally to keep this package testable without importing internal/sync, which would create a visible dependency direction only used in one place). In practice main.go wires the concrete *sync.Syncer through.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type CacheSaver ¶
type CacheSaver func() error
CacheSaver is the narrow persistence sink the scheduler needs: a single "flush the cache to disk" call invoked at the end of each deep-analysis tick. Deliberately separate from api.Cache (which deliberately excludes file-system concerns) so the scheduler can trigger a disk flush without the api.Cache consumers needing to know about the persistence path. *cache.Cache satisfies this via a trivial closure in the composition root.
type Config ¶
type Config struct {
Ignore api.IgnoreChecker // library skip rules; nil means "never skip"
Interval time.Duration // deep-analysis cadence; <=0 means disabled
Enable bool // scheduler on/off
}
Config captures the subset of application configuration the Scheduler actually reads. Decoupling from the full main.config keeps the package boundary clean and lets tests construct a Scheduler without mimicking the app's full env-var surface.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler owns the deep-analysis tick and its workers. Safe for concurrent Run calls, but the intended shape is a single Run goroutine per process.
Concurrent Run invocations collapse their overlapping deep-analysis triggers onto a single in-flight run via singleflight.Group. Within one Run goroutine the initial catch-up and scheduled ticks are already sequential, so the dedup only matters when Run is driven from more than one goroutine. The runner goroutine that loses the dedup race still logs a WARN with the "scheduler: deep analysis already in progress, skipping" key so Loki alerts keyed on that string continue to fire.
func New ¶
func New(cfg Config, reader api.PlexReader, c api.Cache, lookup api.UserLookup, userClient api.UserClientFunc, s Syncer, saveCache CacheSaver) *Scheduler
New constructs a Scheduler with the given collaborators. saveCache may be nil in tests that don't exercise the disk-flush path.
func (*Scheduler) Run ¶
Run is the outer scheduler loop: it runs a deep-analysis pass at startup when the cache's LastSchedulerRun marker is absent or older than one Interval, then runs one every Interval via a time.Ticker. Returns when the context is cancelled. A disabled scheduler (Enable=false or Interval<=0) returns immediately.
type Syncer ¶
type Syncer interface {
ChangeTracksForEpisode(ctx context.Context, userClient api.PlexReadWriter, userID string, reference *streams.Episode, trigger string)
ProcessNewOrUpdatedEpisodeAllUsers(ctx context.Context, episode *streams.Episode, trigger string)
}
Syncer is the narrow interface the scheduler needs from the sync package: a per-user track-apply call plus the multi-user fan-out. The ignore-library/ignore-label skip checks live on api.IgnoreChecker (injected via Config.Ignore) rather than on the Syncer, so overlapping event/scheduler paths share one decision point instead of the three duplicated implementations that existed previously. *sync.Syncer satisfies this. Declared here (rather than imported) to keep scheduler independent of internal/sync for test ergonomics.
Source Files