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 } // 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) } // 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() }