// 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 } // 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", }, nil } // Schedule runs an incremental import on an interval until ctx is cancelled. // // New episodes tend to land through the day and films weekly; an hourly incremental pass // covers both without ever asking Emby for the whole catalogue again. func (s *Syncer) Schedule(ctx context.Context, interval time.Duration) { if interval <= 0 { s.log.Info("library auto-sync disabled") return } ticker := time.NewTicker(interval) defer ticker.Stop() s.log.Info("library auto-sync scheduled", "interval", interval.String()) for { select { case <-ctx.Done(): return case <-ticker.C: 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) } } } } // 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, " ") }