memby/server/internal/library/syncer.go

// Package library imports Emby's catalogue into Postgres so the gateway can answer from
// its own copy instead of asking Emby on every request.
//
// Two shapes of import:
//
//   - **full** — page through everything, then delete whatever the pass did not touch.
//     Run once to seed, and again whenever the library has been reorganised.
//   - **incremental** — ask Emby only for items changed since the last successful run.
//     Cheap enough to run hourly, which is what new episodes need; films appearing
//     weekly are picked up by the same pass.
//
// Only shared metadata is imported (EnableUserData=false). Watched state, favourites and
// resume positions are per-user and stay live.
package library

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/ponzischeme89/memby/server/internal/emby"
	"github.com/ponzischeme89/memby/server/internal/store"
)

const (
	// pageSize balances round trips against Emby's response size. 500 items of metadata
	// is roughly a megabyte of JSON.
	pageSize = 500

	// syncFields is everything the gateway serves or filters on. Images are requested as
	// tags only — the artwork itself is proxied on demand.
	//
	// ProviderIds is here for external ratings: it is what a title is called at MDBList,
	// and reading it from the import means a home row resolves forty cards from one
	// indexed query instead of asking Emby about each of them.
	syncFields = "Genres,Studios,Overview,Taglines,ProductionYear,CommunityRating,OfficialRating," +
		"RunTimeTicks,SeriesName,PrimaryImageAspectRatio,DateCreated,PremiereDate,People," +
		"CollectionName,MediaStreams,RecursiveItemCount,ProviderIds"

	syncImageTypes = "Backdrop,Primary,Logo,Thumb"
	syncItemTypes  = "Movie,Series,Episode"
)

// ErrNoCredentials means nothing has ever signed in and no service account is set, so
// there is no way to talk to Emby on the library's behalf.
var ErrNoCredentials = errors.New("library: no emby credentials available for sync")

type Syncer struct {
	emby  *emby.Client
	store *store.Store
	log   *slog.Logger

	// serviceCred is the optional configured account. When empty, the newest TV session
	// is borrowed instead.
	serviceCred emby.Credentials

	mu        sync.Mutex
	running   bool
	afterSync func(Result)
}

func NewSyncer(embyClient *emby.Client, st *store.Store, serviceCred emby.Credentials, log *slog.Logger) *Syncer {
	return &Syncer{emby: embyClient, store: st, serviceCred: serviceCred, log: log}
}

// SetAfterSync installs the inexpensive invalidation callback used by derived data.
func (s *Syncer) SetAfterSync(callback func(Result)) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.afterSync = callback
}

// Result summarises one import.
type Result struct {
	Kind       string        `json:"kind"`
	Seen       int           `json:"seen"`
	Upserted   int           `json:"upserted"`
	Changed    int           `json:"changed"`
	Removed    int           `json:"removed"`
	Duration   time.Duration `json:"-"`
	DurationMs int64         `json:"durationMs"`
}

// Running reports whether an import is in flight, so the admin page can disable its
// buttons and the scheduler can skip a tick.
func (s *Syncer) Running() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.running
}

// Sync runs an import. kind is "full" or "incremental"; an incremental run with no prior
// successful sync silently upgrades itself to a full one, because there is no watermark
// to work from.
func (s *Syncer) Sync(ctx context.Context, kind, trigger string) (Result, error) {
	s.mu.Lock()
	if s.running {
		s.mu.Unlock()
		return Result{}, errors.New("library: a sync is already running")
	}
	s.running = true
	s.mu.Unlock()
	defer func() {
		s.mu.Lock()
		s.running = false
		s.mu.Unlock()
	}()

	cred, err := s.credentials(ctx)
	if err != nil {
		return Result{}, err
	}

	var since *time.Time
	if kind == "incremental" {
		since, err = s.store.LastSuccessfulSyncAt(ctx)
		if err != nil {
			return Result{}, err
		}
		if since == nil {
			s.log.Info("no previous sync; upgrading to a full import")
			kind = "full"
		}
	}

	startedAt := time.Now().UTC()
	runID, err := s.store.StartSyncRun(ctx, kind, trigger)
	if err != nil {
		return Result{}, err
	}

	s.log.Info("library sync started", "kind", kind, "trigger", trigger)
	result, syncErr := s.run(ctx, cred, kind, since, startedAt)
	result.Kind = kind
	result.Duration = time.Since(startedAt)
	result.DurationMs = result.Duration.Milliseconds()

	record := store.SyncRun{
		Status:        store.SyncStatusSuccess,
		ItemsSeen:     result.Seen,
		ItemsUpserted: result.Upserted,
		ItemsRemoved:  result.Removed,
	}
	if syncErr != nil {
		record.Status = store.SyncStatusFailed
		record.Error = syncErr.Error()
	}
	// Always record the outcome, even when the caller's context died mid-import.
	if err := s.store.FinishSyncRun(context.WithoutCancel(ctx), runID, record); err != nil {
		s.log.Error("could not record sync run", "error", err)
	}

	if syncErr != nil {
		return result, syncErr
	}
	s.log.Info("library sync finished",
		"kind", kind, "trigger", trigger, "seen", result.Seen,
		"upserted", result.Upserted, "changed", result.Changed, "removed", result.Removed,
		"duration", result.Duration.Round(time.Millisecond))
	s.mu.Lock()
	afterSync := s.afterSync
	s.mu.Unlock()
	if afterSync != nil {
		afterSync(result)
	}
	return result, nil
}

func (s *Syncer) run(
	ctx context.Context,
	cred emby.Credentials,
	kind string,
	since *time.Time,
	syncedAt time.Time,
) (Result, error) {
	var result Result

	for startIndex := 0; ; startIndex += pageSize {
		params := url.Values{
			"IncludeItemTypes":       {syncItemTypes},
			"Recursive":              {"true"},
			"SortBy":                 {"DateCreated"},
			"SortOrder":              {"Ascending"},
			"StartIndex":             {strconv.Itoa(startIndex)},
			"Limit":                  {strconv.Itoa(pageSize)},
			"Fields":                 {syncFields},
			"ImageTypeLimit":         {"1"},
			"EnableImages":           {"true"},
			"EnableImageTypes":       {syncImageTypes},
			"EnableTotalRecordCount": {"false"},
			// The imported copy is shared by every user, so it must not carry one
			// user's watched/favourite state.
			"EnableUserData": {"false"},
		}
		if since != nil {
			// Emby returns items created or edited after this instant. Overlap by a
			// minute so an item saved during the previous run is not missed.
			params.Set("MinDateLastSaved", since.Add(-time.Minute).UTC().Format(time.RFC3339))
		}

		page, err := s.emby.Items(ctx, cred, params)
		if err != nil {
			return result, fmt.Errorf("library: fetch page at %d: %w", startIndex, err)
		}
		if len(page.Items) == 0 {
			break
		}

		items := make([]store.LibraryItem, 0, len(page.Items))
		for _, raw := range page.Items {
			if item, ok := toLibraryItem(raw); ok {
				items = append(items, item)
			}
		}
		changed, err := s.store.UpsertLibraryItems(ctx, items, syncedAt)
		if err != nil {
			return result, err
		}

		result.Seen += len(page.Items)
		result.Upserted += len(items)
		result.Changed += int(changed)
		s.log.Info("library sync progress",
			"kind", kind,
			"seen", result.Seen,
			"upserted", result.Upserted,
			"changed", result.Changed,
		)

		if len(page.Items) < pageSize {
			break
		}
	}

	// Only a full pass has seen everything, so only a full pass may delete.
	if kind == "full" {
		removed, err := s.store.DeleteLibraryItemsBefore(ctx, syncedAt)
		if err != nil {
			return result, err
		}
		result.Removed = int(removed)
	}
	return result, nil
}

// EmbyCredentials is how the ingest worker borrows the same account the scheduled import
// uses. One rule for "who does the gateway talk to Emby as" rather than two, so a
// household with a service account configured never has an event-driven read appear in
// somebody's Emby history as their television.
func (s *Syncer) EmbyCredentials(ctx context.Context) (emby.Credentials, error) {
	return s.credentials(ctx)
}

// credentials prefers the configured service account and otherwise borrows the most
// recent TV session.
func (s *Syncer) credentials(ctx context.Context) (emby.Credentials, error) {
	if s.serviceCred.Token != "" && s.serviceCred.UserID != "" {
		return s.serviceCred, nil
	}
	sess, err := s.store.NewestSession(ctx)
	if err != nil {
		if errors.Is(err, store.ErrNotFound) {
			return emby.Credentials{}, ErrNoCredentials
		}
		return emby.Credentials{}, err
	}
	return emby.Credentials{
		UserID:   sess.EmbyUserID,
		Token:    sess.EmbyToken,
		DeviceID: "memby-gateway-sync",
		// The token is borrowed from a television, but the import is the gateway's own
		// work and must not appear in Emby's device list as that set.
		Gateway: true,
	}, nil
}

// Find searches Emby for a term and imports whatever comes back, answering with the
// payloads as they were stored.
//
// The catalogue is a copy, and a copy is only ever as fresh as the last import — up to
// MEMBY_SYNC_INTERVAL behind, an hour by default. That is invisible to a viewer, whose
// rows are read from Emby live, and very visible to an operator, who cannot pin a film to
// the hero until the gateway has heard of it. So the picker asks Emby itself, and what it
// finds is *adopted* rather than merely displayed: an id that came back from here resolves
// through LibraryItemsByID immediately, which is what the policy validation and the hero
// row both read, so nothing downstream has to know the title arrived early.
//
// It imports with exactly the fields a scheduled pass uses. Writing a thinner payload
// would leave an adopted title missing People, MediaStreams and ProviderIds until Emby
// next reported it changed — which for a film nobody edits again is never.
func (s *Syncer) Find(ctx context.Context, term string, limit int) ([]json.RawMessage, error) {
	trimmed := strings.TrimSpace(term)
	if trimmed == "" || limit <= 0 {
		return nil, nil
	}
	cred, err := s.credentials(ctx)
	if err != nil {
		return nil, err
	}
	page, err := s.emby.Items(ctx, cred, url.Values{
		"SearchTerm":             {trimmed},
		"IncludeItemTypes":       {syncItemTypes},
		"Recursive":              {"true"},
		"Limit":                  {strconv.Itoa(limit)},
		"Fields":                 {syncFields},
		"ImageTypeLimit":         {"1"},
		"EnableImages":           {"true"},
		"EnableImageTypes":       {syncImageTypes},
		"EnableTotalRecordCount": {"false"},
		"EnableUserData":         {"false"},
	})
	if err != nil {
		return nil, fmt.Errorf("library: search emby: %w", err)
	}
	items := make([]store.LibraryItem, 0, len(page.Items))
	found := make([]json.RawMessage, 0, len(page.Items))
	for _, raw := range page.Items {
		item, ok := toLibraryItem(raw)
		if !ok {
			continue
		}
		items = append(items, item)
		found = append(found, item.Payload)
	}
	if len(items) == 0 {
		return nil, nil
	}
	// Stamped now, like any other import. A full pass sweeps rows carrying a stamp older
	// than the pass itself, so a title adopted while one is running is never its victim.
	if _, err := s.store.UpsertLibraryItems(ctx, items, time.Now().UTC()); err != nil {
		return nil, err
	}
	s.log.Info("library search adopted titles", "term", trimmed, "found", len(found))
	return found, nil
}

// disabledSyncPoll is how often a switched-off schedule wakes to ask whether it still is.
// A setting an operator has just changed must not need a restart, which is the same reason
// the Emby reachability probe keeps ticking slowly while it is off.
const disabledSyncPoll = 5 * time.Minute

// Schedule runs an incremental import on an interval until ctx is cancelled.
//
// It is reconciliation now rather than discovery. Where the *arr webhooks are configured, a
// file is in the catalogue within a minute of Sonarr or Radarr putting it there and this
// pass exists for what they do not manage — media dropped in by hand, a title edited in
// Emby, a webhook that never arrived because the gateway was down. Where they are not, it
// is still the only thing that notices anything, which is why the interval is the
// operator's rather than a constant.
//
// interval is a function rather than a value because it is read every cycle: an operator
// who has just lengthened the sweep must see that take effect without restarting the
// container. Zero means switched off, and this keeps waking to ask.
func (s *Syncer) Schedule(
	ctx context.Context, interval func() time.Duration, paused ...func() bool,
) {
	if interval == nil {
		s.log.Info("library auto-sync disabled")
		return
	}
	s.log.Info("library auto-sync scheduled", "interval", durationLabel(interval()))
	for {
		wait := interval()
		disabled := wait <= 0
		if disabled {
			wait = disabledSyncPoll
		}
		timer := time.NewTimer(wait)
		select {
		case <-ctx.Done():
			timer.Stop()
			return
		case <-timer.C:
		}
		timer.Stop()
		if disabled {
			continue
		}
		if len(paused) > 0 && paused[0] != nil && paused[0]() {
			s.log.Debug("skipping scheduled sync; quiet time is active")
			continue
		}
		if s.Running() {
			s.log.Info("skipping scheduled sync; one is already running")
			continue
		}
		if _, err := s.Sync(ctx, "incremental", "schedule"); err != nil {
			if errors.Is(err, ErrNoCredentials) {
				// Nobody has signed in yet. Not worth an error-level log every hour.
				s.log.Info("skipping scheduled sync; no credentials yet")
				continue
			}
			s.log.Error("scheduled sync failed", "error", err)
		}
	}
}

func durationLabel(value time.Duration) string {
	if value <= 0 {
		return "off"
	}
	return value.String()
}

// syncItem mirrors the Emby fields promoted to columns.
type syncItem struct {
	ID              string   `json:"Id"`
	Name            string   `json:"Name"`
	Type            string   `json:"Type"`
	SeriesID        string   `json:"SeriesId"`
	SeriesName      string   `json:"SeriesName"`
	ProductionYear  *int     `json:"ProductionYear"`
	CommunityRating *float64 `json:"CommunityRating"`
	Genres          []string `json:"Genres"`
	Studios         []struct {
		Name string `json:"Name"`
	} `json:"Studios"`
	DateCreated *time.Time `json:"DateCreated"`
}

// toLibraryItem flattens the columns Postgres filters on while keeping the payload whole.
func toLibraryItem(raw json.RawMessage) (store.LibraryItem, bool) {
	var parsed syncItem
	if err := json.Unmarshal(raw, &parsed); err != nil || parsed.ID == "" {
		return store.LibraryItem{}, false
	}

	studios := make([]string, 0, len(parsed.Studios))
	for _, studio := range parsed.Studios {
		if name := strings.TrimSpace(studio.Name); name != "" {
			studios = append(studios, name)
		}
	}
	genres := make([]string, 0, len(parsed.Genres))
	for _, genre := range parsed.Genres {
		if g := strings.TrimSpace(genre); g != "" {
			genres = append(genres, g)
		}
	}

	return store.LibraryItem{
		ID:              parsed.ID,
		Type:            parsed.Type,
		Name:            parsed.Name,
		SeriesID:        parsed.SeriesID,
		SeriesName:      parsed.SeriesName,
		ProductionYear:  parsed.ProductionYear,
		CommunityRating: parsed.CommunityRating,
		Genres:          genres,
		Studios:         studios,
		DateCreated:     parsed.DateCreated,
		SearchText:      searchText(parsed),
		Payload:         raw,
	}, true
}

// searchText is what full-text search matches against. Series name is included so
// searching a show finds its episodes.
func searchText(parsed syncItem) string {
	parts := []string{parsed.Name}
	if parsed.SeriesName != "" && !strings.EqualFold(parsed.SeriesName, parsed.Name) {
		parts = append(parts, parsed.SeriesName)
	}
	if parsed.ProductionYear != nil {
		parts = append(parts, strconv.Itoa(*parsed.ProductionYear))
	}
	parts = append(parts, parsed.Genres...)
	// Studios too, so "A24" or "Pixar" finds a shelf's worth of titles. Existing rows
	// keep their old text until the next full import rewrites them.
	for _, studio := range parsed.Studios {
		if studio.Name != "" {
			parts = append(parts, studio.Name)
		}
	}
	return strings.Join(parts, " ")
}