package store import ( "context" "encoding/json" "fmt" "strings" "time" "github.com/jackc/pgx/v5" ) // LibraryItem is one imported Emby item. Payload is Emby's JSON verbatim; the flat // columns exist only so Postgres can filter and rank without opening the JSON. type LibraryItem struct { ID string Type string Name string SeriesID string SeriesName string ProductionYear *int CommunityRating *float64 Genres []string Studios []string DateCreated *time.Time SearchText string Payload json.RawMessage } // LibraryStats is what the admin page shows about the imported library. type LibraryStats struct { Total int64 `json:"total"` ByType map[string]int64 `json:"byType"` LastSynced *time.Time `json:"lastSynced"` } // UpsertLibraryItems writes a batch, refreshing synced_at on every row it touches, and // returns how many recommendation-relevant payloads were inserted or actually changed. // // synced_at doubles as the mark-and-sweep marker: a full import stamps everything it // sees, then deletes whatever kept an older stamp. func (s *Store) UpsertLibraryItems(ctx context.Context, items []LibraryItem, syncedAt time.Time) (int64, error) { if len(items) == 0 { return 0, nil } batch := &pgx.Batch{} for _, item := range items { batch.Queue(` WITH previous AS MATERIALIZED ( SELECT type, name, series_id, series_name, production_year, community_rating, genres, studios, date_created, payload->'RunTimeTicks' AS runtime_ticks, payload->'MediaStreams' AS media_streams, payload->'Container' AS container FROM library_items WHERE id = $1 ), upserted AS ( INSERT INTO library_items ( id, type, name, series_id, series_name, production_year, community_rating, genres, studios, date_created, search_text, payload, synced_at ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12::jsonb,$13) ON CONFLICT (id) DO UPDATE SET type = EXCLUDED.type, name = EXCLUDED.name, series_id = EXCLUDED.series_id, series_name = EXCLUDED.series_name, production_year = EXCLUDED.production_year, community_rating = EXCLUDED.community_rating, genres = EXCLUDED.genres, studios = EXCLUDED.studios, date_created = EXCLUDED.date_created, search_text = EXCLUDED.search_text, payload = EXCLUDED.payload, synced_at = EXCLUDED.synced_at RETURNING 1 ) SELECT NOT EXISTS (SELECT 1 FROM previous) OR EXISTS ( SELECT 1 FROM previous WHERE ROW( type, name, series_id, series_name, production_year, community_rating, genres, studios, date_created, runtime_ticks, media_streams, container ) IS DISTINCT FROM ROW( $2::text, $3::text, $4::text, $5::text, $6::int, $7::real, $8::text[], $9::text[], $10::timestamptz, $12::jsonb->'RunTimeTicks', $12::jsonb->'MediaStreams', $12::jsonb->'Container' ) ) FROM upserted`, item.ID, item.Type, item.Name, item.SeriesID, item.SeriesName, item.ProductionYear, item.CommunityRating, item.Genres, item.Studios, item.DateCreated, item.SearchText, string(item.Payload), syncedAt) } results := s.pool.SendBatch(ctx, batch) defer results.Close() var changed int64 for range items { var recommendationChanged bool if err := results.QueryRow().Scan(&recommendationChanged); err != nil { return changed, fmt.Errorf("store: upsert library items: %w", err) } if recommendationChanged { changed++ } } return changed, nil } // DeleteLibraryItemsBefore removes anything a full import did not touch — items deleted // from Emby since the last run. func (s *Store) DeleteLibraryItemsBefore(ctx context.Context, cutoff time.Time) (int64, error) { tag, err := s.pool.Exec(ctx, `DELETE FROM library_items WHERE synced_at < $1`, cutoff) if err != nil { return 0, fmt.Errorf("store: prune library: %w", err) } return tag.RowsAffected(), nil } // NamedItem is the least a caller can be told about a catalogue row and still identify // it: what it is called and, where the library knows, when it came out. type NamedItem struct { ID string Name string Year int } // LibraryItemsByName finds catalogue rows by title, case-insensitively. // // The comparison that decides the answer is not this one: the caller normalises both // sides (punctuation and spacing are where an *arr and Emby actually differ) and picks by // year. This is the narrowing query — a handful of rows out of twenty thousand — so that // the matching rule can stay a pure function with one definition. func (s *Store) LibraryItemsByName(ctx context.Context, itemType, name string) ([]NamedItem, error) { trimmed := strings.TrimSpace(name) if trimmed == "" { return nil, nil } rows, err := s.pool.Query(ctx, ` SELECT id, name, COALESCE(production_year, 0) FROM library_items WHERE type = $1 AND lower(name) = lower($2) LIMIT 50`, itemType, trimmed) if err != nil { return nil, fmt.Errorf("store: library items by name: %w", err) } defer rows.Close() out := []NamedItem{} for rows.Next() { var item NamedItem if err := rows.Scan(&item.ID, &item.Name, &item.Year); err != nil { return nil, err } out = append(out, item) } return out, rows.Err() } // DeleteLibraryItem removes one item and anything derived from it. // // The credits marker goes with it, and that is the point of doing this in one place: the // marker table is keyed on the item id and nothing else prunes it, so a title deleted from // the library would otherwise leave a Skip Credits position behind for a file that no // longer exists — and if that id were ever reused, in front of the wrong programme. // // Deleting a series takes its episodes with it, because Emby's own hierarchy is the only // thing that made those rows meaningful. func (s *Store) DeleteLibraryItem(ctx context.Context, itemID string) (int64, error) { if strings.TrimSpace(itemID) == "" { return 0, nil } tag, err := s.pool.Exec(ctx, `DELETE FROM library_items WHERE id = $1 OR series_id = $1`, itemID) if err != nil { return 0, fmt.Errorf("store: delete library item: %w", err) } if _, err := s.pool.Exec(ctx, `DELETE FROM credits_markers WHERE item_id = $1 OR series_id = $1`, itemID); err != nil { return 0, fmt.Errorf("store: delete credits markers: %w", err) } return tag.RowsAffected(), nil } // SearchLibrary answers from the imported library rather than Emby. // // Full-text match first, with a trailing ILIKE so partial words ("sever") still hit // before someone finishes typing on a remote. func (s *Store) SearchLibrary(ctx context.Context, term string, limit int) ([]json.RawMessage, error) { trimmed := strings.TrimSpace(term) if trimmed == "" { return nil, nil } rows, err := s.pool.Query(ctx, ` SELECT payload FROM library_items WHERE search_tsv @@ plainto_tsquery('simple', $1) OR search_text ILIKE '%' || $1 || '%' ORDER BY ts_rank(search_tsv, plainto_tsquery('simple', $1)) DESC, (lower(name) = lower($1)) DESC, community_rating DESC NULLS LAST, name ASC LIMIT $2`, trimmed, limit) if err != nil { return nil, fmt.Errorf("store: search library: %w", err) } return collectPayloads(rows) } // LibraryContainsProviderIDs reports which external catalogue ids are already represented in // Emby. ProviderIds stays inside the imported payload because it is not a ranking field; // request autocomplete is the one place that needs to compare it with Radarr/Sonarr. func (s *Store) LibraryContainsProviderIDs( ctx context.Context, provider string, ids []int, ) (map[int]bool, error) { found := map[int]bool{} if len(ids) == 0 { return found, nil } values := make([]string, 0, len(ids)) for _, id := range ids { if id > 0 { values = append(values, fmt.Sprint(id)) } } rows, err := s.pool.Query(ctx, ` SELECT DISTINCT (payload->'ProviderIds'->>$1)::int FROM library_items WHERE payload->'ProviderIds'->>$1 = ANY($2::text[])`, provider, values) if err != nil { return nil, fmt.Errorf("store: library provider ids: %w", err) } defer rows.Close() for rows.Next() { var id int if err := rows.Scan(&id); err != nil { return nil, err } found[id] = true } return found, rows.Err() } // LibraryProviderItemIDs is LibraryContainsProviderIDs with the answer the request page // needs: not only whether Emby has the title, but which item it is. // // That id is what lets a request that has finally downloaded stop being a status card and // become something a viewer can press. Ordering by item_id keeps the answer stable when a // household holds the same title twice — a duplicate import, or a remake sharing an id in // bad metadata — so a card does not point at a different copy between two reads. func (s *Store) LibraryProviderItemIDs( ctx context.Context, provider string, ids []int, ) (map[int]string, error) { found := map[int]string{} if len(ids) == 0 { return found, nil } values := make([]string, 0, len(ids)) for _, id := range ids { if id > 0 { values = append(values, fmt.Sprint(id)) } } if len(values) == 0 { return found, nil } rows, err := s.pool.Query(ctx, ` SELECT DISTINCT ON ((payload->'ProviderIds'->>$1)::int) (payload->'ProviderIds'->>$1)::int, id FROM library_items WHERE payload->'ProviderIds'->>$1 = ANY($2::text[]) ORDER BY (payload->'ProviderIds'->>$1)::int, id`, provider, values) if err != nil { return nil, fmt.Errorf("store: library provider item ids: %w", err) } defer rows.Close() for rows.Next() { var ( id int itemID string ) if err := rows.Scan(&id, &itemID); err != nil { return nil, err } found[id] = itemID } return found, rows.Err() } // LibraryCandidates returns unwatched-agnostic candidates in the given genres, for the // recommendation engine. User state is applied by the caller, which is the only place // that knows it. func (s *Store) LibraryCandidates(ctx context.Context, genres []string, limit int) ([]json.RawMessage, error) { if len(genres) == 0 { return nil, nil } rows, err := s.pool.Query(ctx, ` SELECT payload FROM library_items WHERE type IN ('Movie', 'Series') AND genres && $1 ORDER BY community_rating DESC NULLS LAST, date_created DESC NULLS LAST LIMIT $2`, genres, limit) if err != nil { return nil, fmt.Errorf("store: library candidates: %w", err) } return collectPayloads(rows) } // AllRecommendationCandidates returns the complete Movie/Series catalogue for an // offline For You rebuild. The resulting per-user pool is deliberately over-provisioned // so a short runtime filter still has enough ranked titles to fill the TV row. func (s *Store) AllRecommendationCandidates(ctx context.Context) ([]json.RawMessage, error) { rows, err := s.pool.Query(ctx, ` SELECT payload FROM library_items WHERE type IN ('Movie', 'Series') ORDER BY community_rating DESC NULLS LAST, date_created DESC NULLS LAST`) if err != nil { return nil, fmt.Errorf("store: all recommendation candidates: %w", err) } return collectPayloads(rows) } func (s *Store) LibraryItemsByID( ctx context.Context, ids []string, ) ([]json.RawMessage, error) { if len(ids) == 0 { return []json.RawMessage{}, nil } rows, err := s.pool.Query(ctx, ` SELECT payload FROM library_items WHERE id = ANY($1)`, ids) if err != nil { return nil, fmt.Errorf("store: library items by id: %w", err) } return collectPayloads(rows) } // CuratedCandidates filters the imported catalogue for a server-authored shelf. Arrays // are matched case-insensitively because Emby studio capitalisation is not consistent. func (s *Store) CuratedCandidates( ctx context.Context, itemTypes, genres, studios []string, limit int, ) ([]json.RawMessage, error) { if len(itemTypes) == 0 || (len(genres) == 0 && len(studios) == 0) { return nil, nil } rows, err := s.pool.Query(ctx, ` SELECT payload FROM library_items WHERE type = ANY($1) AND ( cardinality($2::text[]) = 0 OR EXISTS ( SELECT 1 FROM unnest(genres) AS genre WHERE lower(genre) = ANY($2) ) ) AND ( cardinality($3::text[]) = 0 OR EXISTS ( SELECT 1 FROM unnest(studios) AS studio WHERE lower(studio) = ANY($3) ) ) ORDER BY community_rating DESC NULLS LAST, date_created DESC NULLS LAST LIMIT $4`, itemTypes, lowerStrings(genres), lowerStrings(studios), limit) if err != nil { return nil, fmt.Errorf("store: curated candidates: %w", err) } return collectPayloads(rows) } // LibraryGenres returns every genre with enough catalogue depth to make a shelf feel // intentional. The recommendation engine still applies per-user seen filtering and may // drop a shelf afterwards when too few unseen titles remain. func (s *Store) LibraryGenres( ctx context.Context, itemTypes []string, minItems int, ) ([]string, error) { if len(itemTypes) == 0 { return nil, nil } if minItems < 1 { minItems = 1 } rows, err := s.pool.Query(ctx, ` SELECT genre, count(DISTINCT item.id) AS item_count FROM library_items AS item CROSS JOIN LATERAL unnest(item.genres) AS genre WHERE item.type = ANY($1) AND btrim(genre) <> '' GROUP BY genre HAVING count(DISTINCT item.id) >= $2 ORDER BY item_count DESC, lower(genre) ASC`, itemTypes, minItems) if err != nil { return nil, fmt.Errorf("store: library genres: %w", err) } defer rows.Close() out := []string{} for rows.Next() { var genre string var count int if err := rows.Scan(&genre, &count); err != nil { return nil, err } out = append(out, genre) } return out, rows.Err() } // SeriesRef is the minimum needed to link an outside catalogue's show — Sonarr's, in // practice — to the Emby series the library holds, so a card built from that catalogue // can open the show's own page. Year is 0 when Emby does not know it. type SeriesRef struct { ID string Name string Year int LogoTag string } // SeriesRefs lists every imported series. The catalogue is a household's, not a // provider's: a few hundred rows of three short columns, which is why this reads them // all rather than querying per title. func (s *Store) SeriesRefs(ctx context.Context) ([]SeriesRef, error) { rows, err := s.pool.Query(ctx, ` SELECT id, name, COALESCE(production_year, 0), COALESCE(payload->'ImageTags'->>'Logo', '') FROM library_items WHERE type = 'Series'`) if err != nil { return nil, fmt.Errorf("store: series refs: %w", err) } defer rows.Close() out := []SeriesRef{} for rows.Next() { var ref SeriesRef if err := rows.Scan(&ref.ID, &ref.Name, &ref.Year, &ref.LogoTag); err != nil { return nil, err } out = append(out, ref) } return out, rows.Err() } func lowerStrings(values []string) []string { out := make([]string, 0, len(values)) for _, value := range values { out = append(out, strings.ToLower(strings.TrimSpace(value))) } return out } func (s *Store) LibraryStats(ctx context.Context) (LibraryStats, error) { stats := LibraryStats{ByType: map[string]int64{}} rows, err := s.pool.Query(ctx, `SELECT type, count(*) FROM library_items GROUP BY type`) if err != nil { return stats, fmt.Errorf("store: library stats: %w", err) } defer rows.Close() for rows.Next() { var itemType string var count int64 if err := rows.Scan(&itemType, &count); err != nil { return stats, err } stats.ByType[itemType] = count stats.Total += count } if err := rows.Err(); err != nil { return stats, err } var lastSynced *time.Time if err := s.pool.QueryRow(ctx, `SELECT max(synced_at) FROM library_items`).Scan(&lastSynced); err != nil { return stats, err } stats.LastSynced = lastSynced return stats, nil } func collectPayloads(rows pgx.Rows) ([]json.RawMessage, error) { defer rows.Close() out := []json.RawMessage{} for rows.Next() { var payload []byte if err := rows.Scan(&payload); err != nil { return nil, err } out = append(out, json.RawMessage(payload)) } return out, rows.Err() }