memby/server/cmd/memby-server/main.go

// Command memby-server is the Memby gateway: one HTTP service in front of Emby that
// owns auth, caching and the shaping of TV screens, so the Android client can stay thin.
package main

import (
	"context"
	"encoding/json"
	"errors"
	"flag"
	"fmt"
	"log/slog"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"

	"github.com/ponzischeme89/memby/server/internal/adminevents"
	"github.com/ponzischeme89/memby/server/internal/api"
	"github.com/ponzischeme89/memby/server/internal/bazarr"
	"github.com/ponzischeme89/memby/server/internal/buildinfo"
	"github.com/ponzischeme89/memby/server/internal/cache"
	"github.com/ponzischeme89/memby/server/internal/config"
	"github.com/ponzischeme89/memby/server/internal/credits"
	"github.com/ponzischeme89/memby/server/internal/emby"
	"github.com/ponzischeme89/memby/server/internal/foryou"
	"github.com/ponzischeme89/memby/server/internal/integrations"
	"github.com/ponzischeme89/memby/server/internal/library"
	"github.com/ponzischeme89/memby/server/internal/logging"
	"github.com/ponzischeme89/memby/server/internal/mdblist"
	"github.com/ponzischeme89/memby/server/internal/notify"
	"github.com/ponzischeme89/memby/server/internal/radarr"
	"github.com/ponzischeme89/memby/server/internal/recommend"
	"github.com/ponzischeme89/memby/server/internal/runtimestats"
	"github.com/ponzischeme89/memby/server/internal/scheduler"
	"github.com/ponzischeme89/memby/server/internal/sonarr"
	"github.com/ponzischeme89/memby/server/internal/store"
	"github.com/ponzischeme89/memby/server/internal/tracearr"
)

func main() {
	// A distroless image has no shell or curl, so the container's healthcheck re-runs
	// this binary with -healthcheck and it probes itself over the loopback interface.
	healthcheck := flag.Bool("healthcheck", false, "probe the local /healthz endpoint and exit")
	flag.Parse()
	if *healthcheck {
		if err := probeHealth(); err != nil {
			os.Stderr.WriteString(err.Error() + "\n")
			os.Exit(1)
		}
		return
	}

	// A variable rather than a fixed level: the console can move it at runtime, which is
	// the only way turning debug on is any use — a restart to watch something happen
	// restarts the thing being watched.
	logLevel := &slog.LevelVar{}
	logLevel.Set(logging.ParseLevel(os.Getenv("MEMBY_LOG_LEVEL")))
	logCapacity := logging.ParseCapacity(os.Getenv("MEMBY_LOG_BUFFER_CAPACITY"), 5_000)
	logFormat := logging.ParseFormat(os.Getenv("MEMBY_LOG_FORMAT"))
	logHistoryPath := strings.TrimSpace(os.Getenv("MEMBY_LOG_HISTORY_PATH"))
	if logHistoryPath == "" {
		logHistoryPath = "/data/logs/events.jsonl"
	}
	log, events, err := logging.NewPersistentBuffered(
		os.Stdout, logLevel, logCapacity, logFormat, logHistoryPath,
	)
	if err != nil {
		os.Stderr.WriteString("open persistent log history: " + err.Error() + "\n")
		os.Exit(1)
	}
	defer events.Close()
	// Every line names the build that wrote it. A gateway is deployed from a working
	// tree, often while a television is running an older app, so "which server said
	// this" is a real question that a reader should never have to scroll for.
	log = log.With("version", buildinfo.Version())

	if err := run(log, events, logLevel); err != nil {
		log.Error("fatal", "error", err)
		os.Exit(1)
	}
}

func run(log *slog.Logger, events *logging.Buffer, logLevel *slog.LevelVar) error {
	cfg, err := config.Load()
	if err != nil {
		return err
	}

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	// Postgres may still be starting when compose brings us up; retry briefly rather
	// than crash-looping the container.
	st, err := openStore(ctx, cfg.DatabaseURL, log)
	if err != nil {
		return err
	}
	defer st.Close()

	if err := st.Migrate(ctx); err != nil {
		return err
	}

	ca, err := cache.Open(cfg.RedisURL)
	if err != nil {
		return err
	}
	defer ca.Close()
	if err := ca.Ping(ctx); err != nil {
		return err
	}

	// A run interrupted by a restart is still marked "running" in the database; clear
	// those before anything reads the sync history.
	if err := st.MarkStaleRunsFailed(ctx); err != nil {
		log.Warn("could not clear interrupted sync runs", "error", err)
	}

	embyClient := emby.New(
		cfg.EmbyURL, cfg.EmbyPublicURL, cfg.ClientName, cfg.GatewayClientName,
		cfg.UpstreamTimeout,
	)
	embyClient.SetMediaURL(cfg.EmbyMediaURL)
	mdblistClient := mdblist.New(mdblist.DefaultBaseURL, cfg.UpstreamTimeout)
	var sonarrClient *sonarr.Client
	if cfg.SonarrURL != "" {
		sonarrClient = sonarr.New(cfg.SonarrURL, cfg.SonarrAPIKey, cfg.UpstreamTimeout)
		log.Info("sonarr integration enabled", "url", cfg.SonarrURL)
	}
	var radarrClient *radarr.Client
	if cfg.RadarrURL != "" {
		radarrClient = radarr.New(cfg.RadarrURL, cfg.RadarrAPIKey, cfg.UpstreamTimeout)
		log.Info("radarr integration enabled", "url", cfg.RadarrURL)
	}
	// Bazarr gets its own timeout rather than UpstreamTimeout: a manual search queries
	// live subtitle providers and legitimately takes tens of seconds, and cutting it short
	// reads to the viewer as "no subtitles found" rather than as a timeout.
	var bazarrClient *bazarr.Client
	if cfg.BazarrURL != "" {
		bazarrClient = bazarr.New(cfg.BazarrURL, cfg.BazarrAPIKey, cfg.BazarrTimeout)
		log.Info("bazarr integration enabled", "url", cfg.BazarrURL)
	}

	recommender := recommend.NewEngine(embyClient, log.With("component", "recommendations"))
	if cfg.RecommendationWeights != "" {
		_ = json.Unmarshal([]byte(cfg.RecommendationWeights), &recommender.WeightedConfig)
	}
	// Candidates come from the imported library when one exists, which keeps the
	// recommendation rebuild off Emby entirely.
	recommender.Library = st
	recommender.Behavior = st
	var tracearrClient *tracearr.Client
	if cfg.TracearrURL != "" {
		tracearrClient = tracearr.New(
			cfg.TracearrURL,
			cfg.TracearrAPIKey,
			cfg.TracearrServerID,
			cfg.UpstreamTimeout,
		)
		recommender.Tracearr = tracearrClient
		log.Info("tracearr recommendation signals enabled", "url", cfg.TracearrURL)
	}

	syncer := library.NewSyncer(embyClient, st, emby.Credentials{
		UserID:   cfg.SyncUserID,
		Token:    cfg.SyncAPIKey,
		DeviceID: "memby-gateway-sync",
		Gateway:  true,
	}, log.With("component", "library"))

	// Sonarr and Radarr are what put files on disk, so they are what the catalogue learns
	// from. The worker is built whenever either webhook is configured; with neither token
	// set both hooks 404 and nothing here ever has anything to do, so it is not started.
	var ingester *library.Ingester
	if cfg.SonarrWebhookToken != "" || cfg.RadarrWebhookToken != "" {
		ingester = &library.Ingester{
			Store:       st,
			Emby:        embyClient,
			Credentials: syncer.EmbyCredentials,
			Log:         log.With("component", "library-ingest"),
			Settle:      cfg.IngestSettleDelay,
		}
	}
	var forYouService *foryou.Service
	if tracearrClient != nil {
		forYouService = foryou.New(
			st, tracearrClient, recommender, log.With("component", "for-you"),
			cfg.ForYouMinRebuildAge, cfg.ForYouRefreshInterval,
		)
		forYouService.ConfigureTimeContext(cfg.SonarrLocation)
		forYouService.ConfigureHouseholdUsers(embyClient, emby.Credentials{
			UserID: cfg.SyncUserID, Token: cfg.SyncAPIKey,
			DeviceID: "memby-for-you-builder", DeviceName: "MbyGateway For You",
			Gateway: true,
		})
	}

	// Credits discovery. Deliberately built here rather than inside the API server, because
	// it owns a long-running worker and a queue whose lifetime is the process's — the server
	// only reads its markers and feeds it the two live signals it already receives.
	creditsLoad := credits.NewPlaybackLoad()
	var creditsService *credits.Service
	if cfg.CreditsEnabled {
		sampler := &credits.Sampler{Binary: cfg.CreditsFFmpeg}
		var detector credits.Detector
		if sampler.Available() {
			detector = &credits.VisualDetector{Sampler: sampler}
		} else {
			// No decoder in the image is a deliberate deployment, not a fault. The subsystem
			// still runs and still writes markers, on the household's own stop positions —
			// which cost no media access at all and, on a well-watched show, agree more
			// closely than any single reading of the picture.
			log.Warn("credits: ffmpeg not found; running on behavioural evidence only")
		}
		database := credits.Postgres{Store: st}
		creditsConfig := credits.Config{
			PrefetchEpisodes:    cfg.CreditsPrefetchEpisodes,
			MaxPrefetchEpisodes: cfg.CreditsMaxPrefetch,
			QueueLimit:          cfg.CreditsQueueLimit,
			RetryCooldown:       credits.DefaultConfig().RetryCooldown,
			StrongWindow:        credits.DefaultConfig().StrongWindow,
			UsefulWindow:        credits.DefaultConfig().UsefulWindow,
			WeakWindow:          credits.DefaultConfig().WeakWindow,
		}
		if saved, found, settingsErr := st.CreditsSettings(ctx); settingsErr != nil {
			log.Warn("credits settings unavailable; using environment defaults", "error", settingsErr)
		} else if found {
			creditsConfig.PrefetchEpisodes = saved.PrefetchEpisodes
			creditsConfig.MaxPrefetchEpisodes = saved.MaxPrefetch
			creditsConfig.QueueLimit = saved.CandidateLimit
			creditsConfig.RetryCooldown = time.Duration(saved.RetryHours) * time.Hour
		}
		creditsConfig = credits.NormaliseConfig(creditsConfig)
		creditsService = credits.New(credits.Deps{
			Repository: database,
			History:    database,
			Resolver: credits.NewEmbyResolver(embyClient, emby.Credentials{
				UserID: cfg.SyncUserID, Token: cfg.SyncAPIKey,
				DeviceID: "memby-credits", DeviceName: "MbyGateway Credits",
				Gateway: true,
			}),
			Source:    &credits.TracearrSource{DB: database, Cfg: creditsConfig},
			Detector:  detector,
			Behaviour: database,
			Load:      creditsLoad,
			Log:       log,
			Config:    creditsConfig,
		})
		// media_url is on this line rather than the ready line because credits detection is
		// the only thing that uses it, and because reading it back is the only way an
		// operator can tell that a scan is taking the short path. Where it equals EmbyURL
		// the scans go out the public way, which on a split host is the whole cost.
		log.Info("credits detection enabled",
			"prefetch", creditsConfig.PrefetchEpisodes,
			"queue_limit", creditsConfig.QueueLimit,
			"visual", detector != nil,
			"media_url", cfg.EmbyMediaURL)
	}

	// The administrative event bus, the integration dispatcher that subscribes to it and
	// the scheduler that publishes into it are built before the server, because the server
	// takes all three: a handler that could not publish would have to check for nil at
	// every call site, which is exactly how an event comes to be silently dropped.
	adminBus := adminevents.New(st, log)
	// The notification service is built before both the dispatcher and the server, because
	// both write into it: the dispatcher records what it posted to Discord, and the server
	// registers the in-app and broadcast providers on it. The store is its recorder, which
	// is the whole audit trail.
	notifier := notify.New(st, log)
	dispatcher := integrations.New(st, log, adminBus, notifier)
	adminBus.AddSink(dispatcher)
	sched := scheduler.New(st, log, adminBus)

	server := api.New(cfg, api.Deps{
		Emby:        embyClient,
		Store:       st,
		Cache:       ca,
		Recommender: recommender,
		ForYou:      forYouService,
		Sonarr:      sonarrClient,
		Radarr:      radarrClient,
		Bazarr:      bazarrClient,
		MDBList:     mdblistClient,
		Credits:     creditsService,
		CreditsLoad: creditsLoad,
		Syncer:      syncer,
		Ingester:    ingester,
		Log:         log,
		Events:      events,

		AdminEvents:  adminBus,
		Scheduler:    sched,
		Integrations: dispatcher,
		Notify:       notifier,
		LogLevel:     logLevel,
	})
	if err := server.LoadQuietTime(ctx); err != nil {
		return err
	}
	sched.SetPaused(server.ActivityPaused)
	// The ranker calls Tracearr directly on the rebuild path, so it needs the operator's
	// switch too — read at call time, because an integration switched off in the console
	// must stop being called without a restart. It is wired here rather than where the
	// engine is built because the switch belongs to the server, which does not exist yet
	// at that point.
	recommender.TracearrAllowed = server.TracearrEnabled
	dispatcher.SetPaused(server.ActivityPaused)
	dispatcher.Start(ctx)
	if creditsService != nil {
		creditsService.SetPaused(server.ActivityPaused)
		runtimestats.Go("Credits detection", "Credits detection", func() { creditsService.Run(ctx) })
	}
	if ingester != nil {
		// Quiet time is honoured here rather than at the hook: the webhook is recorded
		// whatever the hour, and this is what waits.
		ingester.Paused = server.ActivityPaused
		// The news follows the scan rather than the webhook, so the banner can say a title
		// is there rather than that it is coming. Installed here for the same reason
		// SetAfterSync is: library stays ignorant of what an alert is.
		ingester.Announce = server.AnnounceLibraryIngest
		runtimestats.Go("Library ingest", "Library sync", func() { ingester.Run(ctx) })
	}

	// Registration is separate from construction so the task list reads as a declaration
	// of what the gateway does in the background rather than as more wiring in here.
	server.RegisterHousekeeping(sched)
	server.RegisterCreditsTasks(sched)
	server.RegisterWatchTimeTasks(sched)
	server.RegisterRequestTasks(sched)
	// Sonarr's daily catalogue reading, Radarr's film refresh, the Tracearr import and the
	// For You rebuild all used to be goroutines with tickers of their own here. They are
	// scheduler tasks now, which is what gives the console's integrations area a run
	// history without a second store — and what lets an operator run any of them by hand.
	server.RegisterIntegrationTasks(sched)
	sched.Start(ctx)

	// Installed after the server exists, because both halves of a finished import are
	// its business: derived data has to be invalidated, and the TVs told the catalogue
	// moved. Kept off the syncer itself so library stays ignorant of the API.
	syncer.SetAfterSync(func(result library.Result) {
		afterCtx, cancel := context.WithTimeout(context.Background(), cfg.RecommendTimeout)
		defer cancel()
		server.AnnounceLibrarySync(afterCtx, result)
		if forYouService == nil || (result.Changed == 0 && result.Removed == 0) {
			return
		}
		forYouService.MarkAllDirty(afterCtx)
	})

	if err := server.LoadMaintenance(ctx); err != nil {
		return err
	}
	// The operator's runtime amendments to what this container was started with. Loaded
	// before the watchers below, because two of them read it on their first tick.
	if err := server.LoadGatewaySettings(ctx); err != nil {
		return err
	}
	if err := server.LoadMetadataHeroSettings(ctx); err != nil {
		return err
	}
	// Every long-running worker below is started through runtimestats.Go rather than with
	// a bare `go`, so the admin console can name what is running instead of reporting a
	// count of goroutines nobody can interpret. The name is the worker's identity across
	// restarts and appears verbatim in the console, so it is wording rather than a symbol.
	runtimestats.Go("Gateway settings watcher", "Gateway settings", func() {
		server.WatchGatewaySettings(ctx, 30*time.Second)
	})
	runtimestats.Go("Maintenance watcher", "Gateway settings", func() {
		server.WatchMaintenance(ctx, 30*time.Second)
	})
	runtimestats.Go("Quiet time watcher", "Gateway settings", func() {
		server.WatchQuietTime(ctx, 30*time.Second)
	})
	// One probe per gateway, not per TV: the answer is the same for the whole house.
	runtimestats.Go("Emby health probe", "Emby", func() { server.WatchEmbyReachability(ctx) })
	// The trend ring behind the console's runtime page. A single instantaneous goroutine
	// or heap figure cannot show a leak; this is what makes one visible.
	runtimestats.Go("Runtime sampler", "Gateway API", func() { runtimestats.StartSampling(ctx) })

	if err := server.LoadUpdatePolicy(ctx); err != nil {
		return err
	}
	runtimestats.Go("App update policy watcher", "Gateway settings", func() {
		server.WatchUpdatePolicy(ctx, 60*time.Second)
	})

	runtimestats.Go("Library sync schedule", "Library sync", func() {
		syncer.Schedule(ctx, server.LibrarySyncInterval, server.ActivityPaused)
	})
	if cfg.SyncOnStart {
		runtimestats.Go("Startup library sync", "Library sync", func() {
			if server.ActivityPaused() {
				return
			}
			if _, err := syncer.Sync(ctx, "incremental", "startup"); err != nil {
				log.Warn("startup sync failed", "error", err)
			}
		})
	}
	if forYouService != nil {
		// The import and the daily rebuild are scheduler tasks now — see
		// RegisterIntegrationTasks — so what is left here is the one piece of start-up work
		// neither of them is: profiles produced by an older algorithm. Only a version change
		// warrants it, and normal dirty profiles still wait for the daily off-peak rebuild.
		runtimestats.Go("Startup For You migration", "Recommendations", func() {
			if server.ActivityPaused() {
				return
			}
			rebuildCtx, cancel := context.WithTimeout(ctx, cfg.SyncTimeout)
			defer cancel()
			if err := forYouService.RebuildOutdated(rebuildCtx); err != nil {
				log.Warn("startup outdated For You rebuild failed", "error", err)
			}
		})
	}
	httpServer := &http.Server{
		Addr:              cfg.ListenAddr,
		Handler:           server.Routes(),
		ReadHeaderTimeout: 10 * time.Second,
		// No WriteTimeout: image proxying streams bodies of unpredictable size.
		IdleTimeout: 60 * time.Second,
	}

	errCh := make(chan error, 1)
	runtimestats.Go("HTTP listener", "HTTP server", func() {
		log.Info("gateway ready",
			"listen", cfg.ListenAddr,
			"emby", cfg.EmbyURL,
			"protocol", api.ProtocolVersion,
			"sync_every", cfg.SyncInterval,
		)
		// Announced here rather than beside the log line above, so the feed cannot record
		// a start that then failed to bind to its port.
		server.AnnounceServerStart(buildinfo.Version())
		if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			errCh <- err
		}
	})

	select {
	case err := <-errCh:
		return err
	case <-ctx.Done():
		log.Info("shutting down")
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		return httpServer.Shutdown(shutdownCtx)
	}
}

// probeHealth is the container healthcheck: hit our own /healthz over loopback.
func probeHealth() error {
	addr := os.Getenv("MEMBY_LISTEN_ADDR")
	if addr == "" {
		addr = ":8080"
	}
	// Wildcard listen addresses still mean "connect to localhost" from in here.
	if idx := strings.LastIndex(addr, ":"); idx >= 0 {
		addr = "127.0.0.1" + addr[idx:]
	}

	client := &http.Client{Timeout: 3 * time.Second}
	resp, err := client.Get("http://" + addr + "/healthz")
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("healthz returned %d", resp.StatusCode)
	}
	return nil
}

func openStore(ctx context.Context, databaseURL string, log *slog.Logger) (*store.Store, error) {
	var lastErr error
	for attempt := range 10 {
		st, err := store.Open(ctx, databaseURL)
		if err == nil {
			return st, nil
		}
		lastErr = err
		log.Warn("waiting for postgres", "attempt", attempt+1, "error", err)
		select {
		case <-ctx.Done():
			return nil, ctx.Err()
		case <-time.After(2 * time.Second):
		}
	}
	return nil, lastErr
}