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/hex"
	"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/cache"
	"github.com/ponzischeme89/memby/server/internal/config"
	"github.com/ponzischeme89/memby/server/internal/emby"
	"github.com/ponzischeme89/memby/server/internal/library"
	"github.com/ponzischeme89/memby/server/internal/logging"
	"github.com/ponzischeme89/memby/server/internal/recommend"
	"github.com/ponzischeme89/memby/server/internal/sonarr"
	"github.com/ponzischeme89/memby/server/internal/store"
)

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"))
	log := logging.New(os.Stdout, logLevel)

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

func run(log *slog.Logger) 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.UpstreamTimeout)
	retiredSessions, err := st.TrimSessionsToLimit(ctx, cfg.MaxClientsPerUser)
	if err != nil {
		return err
	}
	for _, sess := range retiredSessions {
		_ = ca.Delete(ctx, cache.SessionKey(hex.EncodeToString(sess.TokenHash)))
		if err := embyClient.Logout(ctx, emby.Credentials{
			UserID: sess.EmbyUserID, Token: sess.EmbyToken,
			DeviceID: sess.DeviceID, DeviceName: sess.DeviceName,
		}); err != nil {
			log.Warn("could not retire excess emby session",
				"username", sess.Username,
				"device", sess.DeviceName,
				"error", err,
			)
		}
	}
	if len(retiredSessions) > 0 {
		log.Info("enforced per-user device allowance",
			"retired_sessions", len(retiredSessions),
			"max_clients_per_user", cfg.MaxClientsPerUser,
		)
	}
	var sonarrClient *sonarr.Client
	if cfg.SonarrURL != "" {
		sonarrClient = sonarr.New(cfg.SonarrURL, cfg.SonarrAPIKey, cfg.UpstreamTimeout)
		log.Info("sonarr integration enabled", "url", cfg.SonarrURL)
	}

	recommender := recommend.NewEngine(embyClient, log)
	// Candidates come from the imported library when one exists, which keeps the
	// recommendation rebuild off Emby entirely.
	recommender.Library = st

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

	server := api.New(cfg, api.Deps{
		Emby:        embyClient,
		Store:       st,
		Cache:       ca,
		Recommender: recommender,
		Sonarr:      sonarrClient,
		Syncer:      syncer,
		Log:         log,
	})

	if err := server.LoadMaintenance(ctx); err != nil {
		return err
	}
	go server.WatchMaintenance(ctx, 30*time.Second)

	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)
			}
		}()
	}
	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("server ready",
			"listen", cfg.ListenAddr,
			"emby", cfg.EmbyURL,
			"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"
	}
	// ":8080" and "0.0.0.0:8080" both 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 row events inside their retention window. The admin page
// aggregates at read time, so nothing survives the prune — deliberately, since this is
// tuning telemetry rather than a permanent record of what anyone watched.
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)
			}
		}
	}
}