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/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/emby"
	"github.com/ponzischeme89/memby/server/internal/foryou"
	"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/radarr"
	"github.com/ponzischeme89/memby/server/internal/recommend"
	"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
	}

	logLevel := 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); err != nil {
		log.Error("fatal", "error", err)
		os.Exit(1)
	}
}

func run(log *slog.Logger, events *logging.Buffer) 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,
	)
	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"))
	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,
		})
	}

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

	// 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
	}
	go server.WatchMaintenance(ctx, 30*time.Second)
	// One probe per gateway, not per TV: the answer is the same for the whole house.
	go server.WatchEmbyReachability(ctx, cfg.EmbyHealthInterval)
	// One Sonarr catalogue reading per day records lifecycle changes for the household and
	// materialises cancellation notifications for every known viewer.
	go server.WatchSonarrLifecycle(ctx, 24*time.Hour)

	if err := server.LoadUpdatePolicy(ctx); err != nil {
		return err
	}
	go server.WatchUpdatePolicy(ctx, 60*time.Second)

	go syncer.Schedule(ctx, cfg.SyncInterval)
	if cfg.SyncOnStart {
		go func() {
			if _, err := syncer.Sync(ctx, "incremental", "startup"); err != nil {
				log.Warn("startup sync failed", "error", err)
			}
		}()
	}
	if forYouService != nil {
		go forYouService.Schedule(
			ctx,
			cfg.TracearrSyncInterval,
			cfg.TracearrFullInterval,
			cfg.ForYouRebuildHour,
		)
		go func() {
			importCtx, cancel := context.WithTimeout(ctx, cfg.SyncTimeout)
			// Only if one is actually owed. An unconditional startup import made every
			// redeploy or container bounce a fresh pass over Tracearr's history.
			if _, _, err := forYouService.ImportIfDue(
				importCtx, cfg.TracearrSyncInterval, cfg.TracearrFullInterval,
			); err != nil {
				cancel()
				log.Warn("startup Tracearr import failed", "error", err)
				return
			}
			cancel()
			// Only an algorithm-version change warrants startup work. Normal dirty
			// profiles wait for the daily off-peak rebuild.
			rebuildCtx, rebuildCancel := context.WithTimeout(ctx, cfg.SyncTimeout)
			defer rebuildCancel()
			if err := forYouService.RebuildOutdated(rebuildCtx); err != nil {
				log.Warn("startup outdated For You rebuild failed", "error", err)
			}
		}()
	}
	go pruneAnalytics(ctx, st, cfg.AnalyticsRetention, log)

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

	go sweepIdleSessions(ctx, st, cfg.SessionIdleExpiry, log)

	errCh := make(chan error, 1)
	go func() {
		log.Info("gateway ready",
			"listen", cfg.ListenAddr,
			"emby", cfg.EmbyURL,
			"protocol", api.ProtocolVersion,
			"sync_every", cfg.SyncInterval,
		)
		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
}

// pruneAnalytics keeps raw engagement and journey events inside their retention window.
func pruneAnalytics(ctx context.Context, st *store.Store, retention time.Duration, log *slog.Logger) {
	if retention <= 0 {
		return
	}
	ticker := time.NewTicker(24 * time.Hour)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			removed, err := st.PruneRowEvents(ctx, retention)
			if err != nil {
				log.Warn("analytics prune failed", "error", err)
				continue
			}
			if removed > 0 {
				log.Info("pruned row events", "count", removed)
			}
		}
	}
}

// sweepIdleSessions retires gateway tokens that have not been used in a long time, so a
// TV that was factory-reset does not leave a live Emby token in the database forever.
func sweepIdleSessions(ctx context.Context, st *store.Store, idle time.Duration, log *slog.Logger) {
	ticker := time.NewTicker(6 * time.Hour)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			removed, err := st.DeleteIdleSessions(ctx, idle)
			if err != nil {
				log.Warn("session sweep failed", "error", err)
				continue
			}
			if removed > 0 {
				log.Info("retired idle sessions", "count", removed)
			}
		}
	}
}