package api import ( "context" "errors" "fmt" "strings" "time" "github.com/ponzischeme89/memby/server/internal/mdblist" "github.com/ponzischeme89/memby/server/internal/scheduler" "github.com/ponzischeme89/memby/server/internal/store" ) // The work each external service does on a schedule. // // These are ordinary scheduler tasks carrying an Integration id, which is the whole of how // the integrations area gets a run history: a run is filed against the service as well as // against the job, and the console reads the scheduler's own table along the other axis. // There is no second store, no second retention job and no second place a piece of work // can be recorded as having failed — which was the point of asking whether the existing // infrastructure could answer before building anything. // // Every one of them begins by asking whether its integration is switched on, and reports // a *skipped* run rather than an error when it is not. That distinction is the feature: // "Memby did not run this" and "Memby ran this and it failed" look identical from a page // that only records failures, and an operator who has switched something off is entitled // to see the schedule quietly standing down rather than a red row every hour. // // The counters are the four in store.RunCounts and they mean the same thing everywhere: // processed is what the run looked at, changed is what it wrote, skipped is what it // deliberately passed over, failed is what went wrong without stopping the run. // ratingsRefreshBatch bounds one MDBList refresh run. // // It is a cost decision rather than a throughput one: MDBList is bought by the day, this // runs hourly, and the batch times the cadence is what the household spends on renewals // before a single television has asked for anything. Forty an hour renews a thousand-title // library about once a day while leaving most of the allowance for titles somebody is // actually looking at. const ratingsRefreshBatch = 40 // RegisterIntegrationTasks declares the background work belonging to external services. func (s *Server) RegisterIntegrationTasks(sched *scheduler.Scheduler) { if sched == nil { return } sched.Register(scheduler.Task{ ID: "sonarr-lifecycle", Name: "Series lifecycle scan", Group: "Sonarr and Radarr", Integration: integrationSonarr, Description: "Reads Sonarr's catalogue and records which shows have been added, " + "have returned or have been cancelled since the last reading.", Interval: 24 * time.Hour, Timeout: 10 * time.Minute, // It ran on start-up before it was a task and still should: the history it keeps // is a record of transitions, and a gateway that was down for a week has a week of // catching up to do before it can tell one from a first sighting. RunOnStart: true, Work: s.runSonarrLifecycleScan, }) sched.Register(scheduler.Task{ ID: "radarr-catalogue", Name: "Film catalogue refresh", Group: "Sonarr and Radarr", Integration: integrationRadarr, Description: "Re-reads Radarr's film catalogue, which is what the upcoming releases " + "row, the request pages and the Radarr-only film pages are all served from.", Interval: 6 * time.Hour, Timeout: 5 * time.Minute, Work: s.runRadarrCatalogueRefresh, }) sched.Register(scheduler.Task{ ID: "tracearr-import", Name: "Watch history import", Group: "Recommendations", Integration: integrationTracearr, Description: "Brings the household's watch history across from Tracearr, which is " + "what For You rows and watch-time summaries are built from.", // The cadence here only decides how often the question is asked. What is actually // due is decided by the persisted import stamps inside ImportIfDue, so a container // restarted three times in an evening still imports on the configured schedule // rather than three times over. Interval: 15 * time.Minute, Timeout: 30 * time.Minute, Work: s.runTracearrImport, }) sched.Register(scheduler.Task{ ID: "for-you-rebuild", Name: "Recommendation rebuild", Group: "Recommendations", Integration: integrationTracearr, Description: "Rebuilds every viewer's prepared For You rows from the imported " + "watch history. Runs once a day at the household's configured hour.", // Daily, and the hour is the operator's: the interval decides only how often the // question is asked, and the rebuild is skipped unless the household's chosen hour // has come round since the last one. Rebuilding at whatever time the container // happened to start is what it did before, which on a redeploy meant the heaviest // job in the gateway ran in the middle of the evening. Interval: time.Hour, Timeout: 30 * time.Minute, Work: s.runForYouRebuild, }) sched.Register(scheduler.Task{ ID: "mdblist-ratings-refresh", Name: "Ratings refresh", Group: "Ratings", Integration: integrationMDBList, Description: fmt.Sprintf( "Renews up to %d of the oldest stored review scores. Titles nobody has looked "+ "at yet are fetched by the warmer on demand rather than here.", ratingsRefreshBatch), Interval: time.Hour, Timeout: 10 * time.Minute, Work: s.runRatingsRefresh, }) // Not filed against any integration, deliberately. It touches all of them, so a run // per probe cycle under each service's history would bury the runs that say what that // service actually did under twelve reachability checks an hour. sched.Register(scheduler.Task{ ID: "integration-health", Name: "Integration health check", Group: "System", Description: "Asks each configured external service whether it is answering.", Interval: integrationProbeInterval, Timeout: 2 * time.Minute, RunOnStart: true, Work: s.runIntegrationHealthCheck, }) } // integrationOff is the outcome every integration task returns when its switch is off. // // A skipped run rather than an error or silence: silence is indistinguishable from a // scheduler that has stopped, and an error would put a red row in the console for a state // the operator chose. The sentence names the service so the row reads correctly in the // all-tasks table, where the integration column is not necessarily beside it. func integrationOff(name string) (scheduler.Outcome, error) { return scheduler.Outcome{Detail: name + " is switched off"}, nil } func (s *Server) runSonarrLifecycleScan(ctx context.Context) (scheduler.Outcome, error) { if !s.integrationEnabled(ctx, integrationSonarr) { return integrationOff("Sonarr") } result, err := s.scanSonarrLifecycle(ctx) if err != nil { return scheduler.Outcome{}, err } outcome := scheduler.Outcome{RunCounts: store.RunCounts{ Processed: result.Series, Changed: result.Changes, }} // Silent when nothing moved, which on a settled household is most days. See // scheduler.announce: a job that reports itself every day is one nobody reads. if result.Changes == 0 { return outcome, nil } parts := []string{plural(result.Changes, "change", "changes")} if result.Added > 0 { parts = append(parts, fmt.Sprintf("%d added", result.Added)) } if result.Cancelled > 0 { parts = append(parts, fmt.Sprintf("%d cancelled", result.Cancelled)) } if result.Notifications > 0 { parts = append(parts, plural(result.Notifications, "notification", "notifications")) } outcome.Detail = fmt.Sprintf("%s checked · %s", plural(result.Series, "series", "series"), strings.Join(parts, ", ")) return outcome, nil } func (s *Server) runRadarrCatalogueRefresh(ctx context.Context) (scheduler.Outcome, error) { if !s.integrationEnabled(ctx, integrationRadarr) { return integrationOff("Radarr") } // Straight to Radarr rather than through radarrMovieCatalogue: that reads the shared // cache first, and a refresh whose whole job is to replace the cache must not be // satisfied by it. movies, err := s.radarr.Movies(ctx) if err != nil { return scheduler.Outcome{}, fmt.Errorf("read Radarr catalogue: %w", err) } held, upcoming := 0, 0 now := time.Now() for _, movie := range movies { if movie.HasFile { held++ continue } // Upcoming is "monitored and not yet available", which is what the launcher's // releases row draws from — an unmonitored film Radarr is not chasing is not on // its way to anybody. if movie.Monitored { if movie.DigitalRelease == nil || movie.DigitalRelease.After(now) { upcoming++ } } } s.cacheRadarrMovies(ctx, movies) return scheduler.Outcome{ Detail: fmt.Sprintf("%s checked · %d in the library, %d still to come", plural(len(movies), "film", "films"), held, upcoming), RunCounts: store.RunCounts{Processed: len(movies), Changed: held, Skipped: upcoming}, }, nil } func (s *Server) runTracearrImport(ctx context.Context) (scheduler.Outcome, error) { if !s.integrationEnabled(ctx, integrationTracearr) { return integrationOff("Tracearr") } if s.forYou.Running() { // A rebuild started by an operator is already using it. Skipping is the right // answer rather than queueing: the next tick is fifteen minutes away and the // import is idempotent. return scheduler.Outcome{Detail: "a For You rebuild is already running"}, nil } result, imported, err := s.forYou.ImportIfDue( ctx, s.cfg.TracearrSyncInterval, s.cfg.TracearrFullInterval) if err != nil { return scheduler.Outcome{}, fmt.Errorf("Tracearr import: %w", err) } if !imported { // Not due. Silent, because this is asked four times an hour and answered "no" // almost every time. return scheduler.Outcome{}, nil } outcome := scheduler.Outcome{RunCounts: store.RunCounts{ Processed: result.Seen, Changed: result.Changed, }} if result.Seen == 0 && result.Changed == 0 { return outcome, nil } outcome.Detail = fmt.Sprintf("%s import · %s read, %d changed", result.Kind, plural(result.Seen, "session", "sessions"), result.Changed) return outcome, nil } // forYouRebuildDue reports whether the household's daily rebuild hour has come round // since the last one. // // Pure so the boundary can be tested: the two ways to get this wrong are both invisible // from a log — never rebuilding, and rebuilding on every tick — and both look like a // working scheduler from outside. A rebuild that has never happened is due, so a fresh // household does not wait until tomorrow evening for its first rows. func forYouRebuildDue(last *time.Time, now time.Time, hour int) bool { if hour < 0 || hour > 23 { hour = 0 } if now.Hour() < hour { return false } if last == nil { return true } // The comparison is against the local calendar day rather than "24 hours ago", the // rule every household-local boundary in Memby follows: a day containing a // daylight-saving change is 23 or 25 hours long, and subtracting hours would skip or // repeat a rebuild twice a year. previous := last.In(now.Location()) sameDay := previous.Year() == now.Year() && previous.YearDay() == now.YearDay() return !sameDay } func (s *Server) runForYouRebuild(ctx context.Context) (scheduler.Outcome, error) { if !s.integrationEnabled(ctx, integrationTracearr) { return integrationOff("Tracearr") } if s.forYou.Running() { return scheduler.Outcome{Detail: "a For You rebuild is already running"}, nil } now := time.Now().In(s.householdLocation()) state, err := s.store.TracearrImportState(ctx) if err != nil { return scheduler.Outcome{}, fmt.Errorf("read For You rebuild state: %w", err) } if !forYouRebuildDue(state.LastRebuildAt, now, s.cfg.ForYouRebuildHour) { return scheduler.Outcome{}, nil } result, err := s.forYou.RebuildAll(ctx, true) if err != nil { return scheduler.Outcome{}, fmt.Errorf("For You rebuild: %w", err) } if markErr := s.store.MarkForYouRebuild(ctx, now); markErr != nil { // The rebuild happened; failing the run over the stamp would have the console // report a failure for work that succeeded. It does mean the next tick tries // again, which is the safe direction to be wrong in. s.loggerFor(ctx).Warn("could not record For You rebuild", "error", markErr) } outcome := scheduler.Outcome{RunCounts: store.RunCounts{ Processed: result.Users, Changed: result.Built, Failed: result.Failed, }} if result.Users == 0 { return outcome, nil } outcome.Detail = fmt.Sprintf("%s processed · %d rebuilt", plural(result.Users, "viewer", "viewers"), result.Built) if result.Failed > 0 { outcome.Detail += fmt.Sprintf(", %d failed", result.Failed) } return outcome, nil } func (s *Server) runRatingsRefresh(ctx context.Context) (scheduler.Outcome, error) { settings, enabled := s.mdblistSettings(ctx) if !enabled || !s.integrationEnabled(ctx, integrationMDBList) { return integrationOff("MDBList") } keys, err := s.store.StaleRatingKeys( ctx, time.Now().Add(-ratingsRefreshInterval), ratingsRefreshBatch) if err != nil { return scheduler.Outcome{}, fmt.Errorf("read stale ratings: %w", err) } if len(keys) == 0 { return scheduler.Outcome{}, nil } counts := store.RunCounts{} var lastErr error for _, key := range keys { if ctx.Err() != nil { break } // The same daily allowance the on-demand warmer spends from, claimed the same way. // Two counters would let a quiet evening of browsing and a refresh run each spend a // full day's worth between them. if !s.claimRatingsBudget(time.Now()) { counts.Skipped++ continue } counts.Processed++ ratings, fetchErr := s.fetchAndStoreRatings(ctx, settings.APIKey, key) if fetchErr != nil { counts.Failed++ lastErr = fetchErr var apiErr *mdblist.APIError if errors.As(fetchErr, &apiErr) && (apiErr.StatusCode == 429 || apiErr.StatusCode == 402) { // The allowance is exhausted. Stopping the whole run is the point: the // remaining titles are still stale and will be the oldest next hour, and // hammering a provider that has just said no is how a key gets withdrawn. s.blockRatingsWarming(time.Now().Add(ratingsWarmBackoff)) break } continue } if len(ratings) == 0 { // A title MDBList has nothing for. Recorded rather than counted as changed: // the stamp moved so it is not re-fetched immediately, but nothing on a // television will look different. counts.Skipped++ continue } counts.Changed++ } outcome := scheduler.Outcome{RunCounts: counts} if counts.Processed == 0 && counts.Skipped == 0 { return outcome, nil } outcome.Detail = fmt.Sprintf("%s requested · %d updated, %d unavailable", plural(counts.Processed, "title", "titles"), counts.Changed, counts.Skipped) // A run that failed every title it asked for is a failed run, not a quiet one: that is // a wrong key or a provider that is down, and it belongs in the notification feed. A // run that lost a few titles among many is ordinary and stays a detail. if counts.Failed > 0 && counts.Failed == counts.Processed && lastErr != nil { return outcome, fmt.Errorf("every MDBList request failed: %w", lastErr) } if counts.Failed > 0 { outcome.Detail += fmt.Sprintf(", %d failed", counts.Failed) } return outcome, nil } // runIntegrationHealthCheck probes every configured service. // // A probe failure is not this task's failure: the whole job is to find out, and a red row // on the health check would say the gateway's own scheduler is broken when what is // actually broken is somebody's Sonarr. The verdict lives on the integration's row // instead, where it names the service. func (s *Server) runIntegrationHealthCheck(ctx context.Context) (scheduler.Outcome, error) { counts := store.RunCounts{} unreachable := []string{} for _, definition := range integrationCatalogue() { if definition.Probe == nil || !definition.Configured(s) { continue } if !definition.Enabled(s, ctx) { // A switched-off service is not probed at all. Probing one would be the // gateway continuing to call an API the operator has told it to stop calling, // which is exactly what the switch promises not to do. s.integrationHealth.forget(definition.ID) counts.Skipped++ continue } counts.Processed++ if state := s.probeIntegration(ctx, definition); !state.Reachable { counts.Failed++ unreachable = append(unreachable, definition.Name) } } outcome := scheduler.Outcome{RunCounts: counts} if len(unreachable) > 0 { outcome.Detail = "not answering: " + strings.Join(unreachable, ", ") } return outcome, nil } // plural is the "3 films" / "1 film" the run details are written in. The console prints // these sentences verbatim, so getting it wrong is visible on every row. func plural(count int, singular, many string) string { if count == 1 { return "1 " + singular } return fmt.Sprintf("%d %s", count, many) }