memby/server/internal/api/integration_tasks.go

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.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)
}