diff --git a/internal/notifications/interest_hooks.go b/internal/notifications/interest_hooks.go index ef12e229..2262dcb4 100644 --- a/internal/notifications/interest_hooks.go +++ b/internal/notifications/interest_hooks.go @@ -39,12 +39,28 @@ func (p *interestTrackingProvider) ForUser(ctx context.Context, userID int) (use return store, err } tracked := &interestTrackingStore{UserStore: store, userID: userID, system: p.system, updater: p.system.Interest} - // Preserve the DeviceRegistry interface upgrade some callers probe for. - if registry, ok := store.(userstore.DeviceRegistry); ok { + // Preserve the interface upgrades callers probe for. Both are conditional + // on the backing store: advertising a capability it does not have would + // send callers down a fast path that can only fail. + registry, hasDevices := store.(userstore.DeviceRegistry) + rollup, hasRollup := store.(userstore.SeriesEpisodeRollupStore) + switch { + case hasDevices && hasRollup: + return &interestTrackingStoreWithDevicesAndRollup{ + interestTrackingStore: tracked, + DeviceRegistry: registry, + SeriesEpisodeRollupStore: rollup, + }, nil + case hasDevices: return &interestTrackingStoreWithDevices{ interestTrackingStore: tracked, DeviceRegistry: registry, }, nil + case hasRollup: + return &interestTrackingStoreWithRollup{ + interestTrackingStore: tracked, + SeriesEpisodeRollupStore: rollup, + }, nil } return tracked, nil } @@ -65,11 +81,48 @@ type interestTrackingStoreWithDevices struct { userstore.DeviceRegistry } +// interestTrackingStoreWithRollup adds the series-rollup capability only when +// the backing store actually has it. Unlike the other capabilities, this one +// cannot be forwarded unconditionally: the per-user SQLite backend has no +// catalog tables and genuinely cannot answer the query, and callers treat +// "implements the interface" as "can do this". Advertising it anyway would +// send every jellycompat series request down the fast path to fail, logging a +// warning each time before falling back — turning an expected capability +// absence into recurring noise on successful requests. +type interestTrackingStoreWithRollup struct { + *interestTrackingStore + userstore.SeriesEpisodeRollupStore +} + +type interestTrackingStoreWithDevicesAndRollup struct { + *interestTrackingStore + userstore.DeviceRegistry + userstore.SeriesEpisodeRollupStore +} + var _ userstore.SettingValueCompareAndSetter = (*interestTrackingStore)(nil) var _ userstore.SettingMutationTransactioner = (*interestTrackingStore)(nil) var _ userstore.SettingValueCompareAndSetter = (*interestTrackingStoreWithDevices)(nil) var _ userstore.SettingMutationTransactioner = (*interestTrackingStoreWithDevices)(nil) +// Optional store capabilities must survive the decorator. Embedding the +// UserStore interface promotes only that interface's methods, so each of these +// needs an explicit forward below; the assertions make a missing one a compile +// error instead of a silent production slowdown. +// +// SeriesEpisodeRollupStore is deliberately absent here: it is conditional on +// the backing store, so it lives on the wrapper types above rather than being +// forwarded unconditionally. +var _ userstore.WatchedBatchWriter = (*interestTrackingStore)(nil) +var _ userstore.VisibleHistoryAdder = (*interestTrackingStore)(nil) +var _ userstore.HistoryVisibilityStore = (*interestTrackingStore)(nil) +var _ userstore.WatchedBatchWriter = (*interestTrackingStoreWithDevices)(nil) +var _ userstore.VisibleHistoryAdder = (*interestTrackingStoreWithDevices)(nil) +var _ userstore.HistoryVisibilityStore = (*interestTrackingStoreWithDevices)(nil) +var _ userstore.SeriesEpisodeRollupStore = (*interestTrackingStoreWithRollup)(nil) +var _ userstore.SeriesEpisodeRollupStore = (*interestTrackingStoreWithDevicesAndRollup)(nil) +var _ userstore.DeviceRegistry = (*interestTrackingStoreWithDevicesAndRollup)(nil) + // WithPreferenceSettingsTransaction preserves the optional atomic-settings // capability of the wrapped store. Preference writes do not affect interest // signals, so the transaction can pass through unchanged; keeping the method @@ -298,6 +351,84 @@ func (s *interestTrackingStore) RemoveHistoryItems(ctx context.Context, profileI return err } +// --- Optional store capabilities. +// +// This decorator embeds the userstore.UserStore *interface*, which promotes +// only the methods declared on that interface. Any optional capability the +// backing store implements is invisible through the wrapper unless it is +// forwarded explicitly here — and because callers reach these via type +// assertion with a working fallback, a missing forward is silent: no error, no +// test failure, just a much slower path in production. Marking a series +// watched regressed exactly this way. When adding a new optional capability to +// userstore, forward it here and extend the compile-time assertions below. + +// MarkWatchedBatch forwards the transactional batch mark-watched write, whose +// whole purpose is that a large series commits as one unit. Falling through to +// the per-target loop would restore the partial-write window it removes. +func (s *interestTrackingStore) MarkWatchedBatch( + ctx context.Context, + profileID string, + targets []userstore.MarkWatchedTarget, + entries []userstore.WatchHistoryEntry, +) ([]userstore.WatchHistoryEntry, error) { + writer, ok := s.UserStore.(userstore.WatchedBatchWriter) + if !ok { + // The helper runs against the inner store, so this decorator's own + // MarkWatched hook never fires; queue here or the recompute is lost. + // A mid-loop error still leaves earlier targets written, so queue + // regardless of err — a redundant queue only costs one recompute, + // while a missing one leaves profile_series_interest stale until some + // unrelated mutation or the rebuild task touches the series. + written, err := userstore.MarkWatchedBatch(ctx, s.UserStore, profileID, targets, entries) + s.queueTargetMutations(profileID, targets) + return written, err + } + written, err := writer.MarkWatchedBatch(ctx, profileID, targets, entries) + // The batch write is one transaction: on error nothing landed, so there is + // nothing to recompute. + if err == nil { + s.queueItemMutations(profileID, written) + } + return written, err +} + +// queueItemMutations records one interest mutation per written entry, matching +// what the per-target path queues through MarkWatched. +func (s *interestTrackingStore) queueItemMutations(profileID string, entries []userstore.WatchHistoryEntry) { + for _, entry := range entries { + s.updater.QueueItemMutation(s.userID, profileID, entry.MediaItemID) + } +} + +// queueTargetMutations queues by requested target rather than written entry, +// for the non-transactional path where a partial write may have occurred. +func (s *interestTrackingStore) queueTargetMutations(profileID string, targets []userstore.MarkWatchedTarget) { + for _, target := range targets { + s.updater.QueueItemMutation(s.userID, profileID, target.MediaItemID) + } +} + +// AddVisibleHistory forwards the watermark-aware history insert. The generic +// fallback needs two round-trips (timestamp lookup, then insert) to do the +// same job. +func (s *interestTrackingStore) AddVisibleHistory(ctx context.Context, entry userstore.WatchHistoryEntry) (userstore.WatchHistoryEntry, error) { + adder, ok := s.UserStore.(userstore.VisibleHistoryAdder) + if !ok { + return userstore.AddVisibleHistory(ctx, s.UserStore, entry) + } + return adder.AddVisibleHistory(ctx, entry) +} + +// VisibleHistoryTimestamps forwards the batched watermark lookup; without it +// callers assume a single wall-clock timestamp for every item. +func (s *interestTrackingStore) VisibleHistoryTimestamps(ctx context.Context, profileID string, mediaItemIDs []string, at time.Time) (map[string]string, error) { + visibility, ok := s.UserStore.(userstore.HistoryVisibilityStore) + if !ok { + return userstore.VisibleHistoryTimestamps(ctx, s.UserStore, profileID, mediaItemIDs, at) + } + return visibility.VisibleHistoryTimestamps(ctx, profileID, mediaItemIDs, at) +} + func (s *interestTrackingStore) DeleteHistoryBySource(ctx context.Context, profileID string, mediaItemIDs []string, source userstore.WatchHistorySource) error { err := s.UserStore.DeleteHistoryBySource(ctx, profileID, mediaItemIDs, source) if err == nil { diff --git a/internal/notifications/interest_hooks_test.go b/internal/notifications/interest_hooks_test.go index 9396ea7b..28793d17 100644 --- a/internal/notifications/interest_hooks_test.go +++ b/internal/notifications/interest_hooks_test.go @@ -100,3 +100,231 @@ func TestInterestTrackingStorePreservesSettingCapabilities(t *testing.T) { t.Fatalf("wrapped receipt = %+v (%v)", receipt, err) } } + +// TestInterestTrackingStorePreservesWatchStateCapabilities pins the optional +// watch-state capabilities through the wrapper. These are type-asserted at +// their call sites (userstore.MarkWatchedBatch, AddVisibleHistory, +// VisibleHistoryTimestamps, jellycompat's rollup), and the decorator embeds +// the UserStore *interface* — which promotes only the methods in that +// interface. A capability the decorator does not forward explicitly is +// therefore invisible in production, and the caller silently degrades to its +// slow per-item fallback with no error and no test failure. +func TestInterestTrackingStorePreservesWatchStateCapabilities(t *testing.T) { + db, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if err := userdb.InitSchema(db); err != nil { + t.Fatalf("init schema: %v", err) + } + + inner := userdb.NewSQLiteUserStore(db) + provider := WrapUserStoreProvider( + preferenceTransactionTestProvider{store: inner}, + &System{}, + ) + wrapped, err := provider.ForUser(context.Background(), 1) + if err != nil { + t.Fatalf("ForUser: %v", err) + } + + // Every capability the bare store advertises must survive wrapping, + // otherwise the wrapped store is a silent performance downgrade. + for _, capability := range []struct { + name string + has func(userstore.UserStore) bool + }{ + {"WatchedBatchWriter", func(s userstore.UserStore) bool { + _, ok := s.(userstore.WatchedBatchWriter) + return ok + }}, + {"VisibleHistoryAdder", func(s userstore.UserStore) bool { + _, ok := s.(userstore.VisibleHistoryAdder) + return ok + }}, + {"HistoryVisibilityStore", func(s userstore.UserStore) bool { + _, ok := s.(userstore.HistoryVisibilityStore) + return ok + }}, + } { + if !capability.has(inner) { + t.Fatalf("test setup: bare store does not implement %s", capability.name) + } + if !capability.has(wrapped) { + t.Errorf("interest-tracking wrapper dropped %s; callers silently fall back to the per-item path", + capability.name) + } + } + + // SeriesEpisodeRollupStore is the exception: it must be advertised only + // when the backing store can actually answer it. The per-user SQLite store + // has no catalog tables, so claiming the capability would send every + // jellycompat series request down a fast path that can only fail and log. + if _, ok := userstore.UserStore(inner).(userstore.SeriesEpisodeRollupStore); ok { + t.Fatal("test setup: SQLite store unexpectedly implements SeriesEpisodeRollupStore") + } + if _, ok := wrapped.(userstore.SeriesEpisodeRollupStore); ok { + t.Error("wrapper advertises SeriesEpisodeRollupStore for a store that cannot perform the rollup") + } + + // The forwarded batch write must actually reach the backing store. + writer, ok := wrapped.(userstore.WatchedBatchWriter) + if !ok { + t.Fatal("wrapper dropped WatchedBatchWriter; cannot verify pass-through") + } + if err := wrapped.CreateProfile(context.Background(), userstore.Profile{ID: "p1", Name: "Test"}); err != nil { + t.Fatalf("CreateProfile: %v", err) + } + watchedAt := time.Date(2026, 8, 15, 12, 0, 0, 0, time.UTC).Format(time.RFC3339) + written, err := writer.MarkWatchedBatch(context.Background(), "p1", + []userstore.MarkWatchedTarget{ + {MediaItemID: "ep-1", DurationSeconds: 1200}, + {MediaItemID: "ep-2", DurationSeconds: 1500}, + }, + []userstore.WatchHistoryEntry{ + {ProfileID: "p1", MediaItemID: "ep-1", WatchedAt: watchedAt, DurationSeconds: 1200, Completed: true, Source: userstore.WatchHistorySourceManual}, + {ProfileID: "p1", MediaItemID: "ep-2", WatchedAt: watchedAt, DurationSeconds: 1500, Completed: true, Source: userstore.WatchHistorySourceManual}, + }, + ) + if err != nil { + t.Fatalf("MarkWatchedBatch through wrapper: %v", err) + } + if len(written) != 2 { + t.Fatalf("MarkWatchedBatch returned %d entries, want 2", len(written)) + } + for _, id := range []string{"ep-1", "ep-2"} { + progress, err := wrapped.GetProgress(context.Background(), "p1", id) + if err != nil || progress == nil || !progress.Completed { + t.Fatalf("GetProgress(%s) = %+v (%v), want completed", id, progress, err) + } + } +} + +// storeWithoutBatchWriter wraps a UserStore and hides the WatchedBatchWriter +// capability, standing in for a backend that has not implemented it. +type storeWithoutBatchWriter struct { + userstore.UserStore +} + +// TestInterestTrackingStoreQueuesMutationsOnBatchFallback covers the branch +// taken when the backing store lacks WatchedBatchWriter. The decorator hands +// the work to the generic helper against s.UserStore — the *inner* store — so +// the decorator's own MarkWatched hook never runs and nothing queues an +// interest recompute. Without this, marking a series watched on such a backend +// updates progress and history but leaves profile_series_interest stale until +// some unrelated mutation or the rebuild task happens to touch the series. +func TestInterestTrackingStoreQueuesMutationsOnBatchFallback(t *testing.T) { + db, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if err := userdb.InitSchema(db); err != nil { + t.Fatalf("init schema: %v", err) + } + + inner := storeWithoutBatchWriter{UserStore: userdb.NewSQLiteUserStore(db)} + if _, ok := userstore.UserStore(inner).(userstore.WatchedBatchWriter); ok { + t.Fatal("test setup: inner store must not implement WatchedBatchWriter") + } + + updater := &InterestUpdater{pending: map[interestMutation]int{}} + store := &interestTrackingStore{ + UserStore: inner, + userID: 1, + system: &System{}, + updater: updater, + } + if err := store.CreateProfile(context.Background(), userstore.Profile{ID: "p1", Name: "Test"}); err != nil { + t.Fatalf("CreateProfile: %v", err) + } + + watchedAt := time.Date(2026, 8, 15, 12, 0, 0, 0, time.UTC).Format(time.RFC3339) + if _, err := store.MarkWatchedBatch(context.Background(), "p1", + []userstore.MarkWatchedTarget{ + {MediaItemID: "ep-1", DurationSeconds: 1200}, + {MediaItemID: "ep-2", DurationSeconds: 1500}, + }, + []userstore.WatchHistoryEntry{ + {ProfileID: "p1", MediaItemID: "ep-1", WatchedAt: watchedAt, DurationSeconds: 1200, Completed: true, Source: userstore.WatchHistorySourceManual}, + {ProfileID: "p1", MediaItemID: "ep-2", WatchedAt: watchedAt, DurationSeconds: 1500, Completed: true, Source: userstore.WatchHistorySourceManual}, + }, + ); err != nil { + t.Fatalf("MarkWatchedBatch (fallback path): %v", err) + } + + updater.mu.Lock() + queued := make(map[string]struct{}, len(updater.pending)) + for mutation := range updater.pending { + queued[mutation.itemID] = struct{}{} + } + updater.mu.Unlock() + + for _, itemID := range []string{"ep-1", "ep-2"} { + if _, ok := queued[itemID]; !ok { + t.Errorf("no interest mutation queued for %s on the batch fallback path; interest goes stale", itemID) + } + } +} + +// rollupCapableStore is a UserStore that also implements the series rollup, +// standing in for the Postgres backend. +type rollupCapableStore struct { + userstore.UserStore + called bool +} + +func (s *rollupCapableStore) SeriesEpisodeWatchCounts(_ context.Context, _ string, seriesIDs []string) (map[string]userstore.SeriesWatchCounts, error) { + s.called = true + counts := make(map[string]userstore.SeriesWatchCounts, len(seriesIDs)) + for _, seriesID := range seriesIDs { + counts[seriesID] = userstore.SeriesWatchCounts{TotalEpisodes: 3, WatchedCount: 2} + } + return counts, nil +} + +// TestInterestTrackingStoreForwardsRollupWhenSupported is the other half of the +// conditional: a backend that can do the rollup must keep advertising it +// through the wrapper, and calls must reach it. Losing this would silently +// push jellycompat back onto the per-episode rollup for every series. +func TestInterestTrackingStoreForwardsRollupWhenSupported(t *testing.T) { + db, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if err := userdb.InitSchema(db); err != nil { + t.Fatalf("init schema: %v", err) + } + + inner := &rollupCapableStore{UserStore: userdb.NewSQLiteUserStore(db)} + provider := WrapUserStoreProvider(preferenceTransactionTestProvider{store: inner}, &System{}) + wrapped, err := provider.ForUser(context.Background(), 1) + if err != nil { + t.Fatalf("ForUser: %v", err) + } + + rollup, ok := wrapped.(userstore.SeriesEpisodeRollupStore) + if !ok { + t.Fatal("wrapper dropped SeriesEpisodeRollupStore for a store that supports it") + } + counts, err := rollup.SeriesEpisodeWatchCounts(context.Background(), "p1", []string{"series-1"}) + if err != nil { + t.Fatalf("SeriesEpisodeWatchCounts: %v", err) + } + if !inner.called { + t.Error("rollup call did not reach the backing store") + } + if counts["series-1"].WatchedCount != 2 { + t.Errorf("counts = %+v, want WatchedCount 2", counts["series-1"]) + } + + // The other capabilities must still survive alongside the rollup. + if _, ok := wrapped.(userstore.WatchedBatchWriter); !ok { + t.Error("rollup-capable wrapper dropped WatchedBatchWriter") + } + if _, ok := wrapped.(userstore.SettingValueCompareAndSetter); !ok { + t.Error("rollup-capable wrapper dropped SettingValueCompareAndSetter") + } +} diff --git a/internal/notifications/interest_updater.go b/internal/notifications/interest_updater.go index 33be6a51..f911cdfa 100644 --- a/internal/notifications/interest_updater.go +++ b/internal/notifications/interest_updater.go @@ -2,7 +2,6 @@ package notifications import ( "context" - "errors" "fmt" "log/slog" "sync" @@ -10,7 +9,6 @@ import ( "github.com/Silo-Server/silo-server/internal/access" "github.com/Silo-Server/silo-server/internal/userstore" - "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) @@ -127,18 +125,34 @@ func (u *InterestUpdater) flush(ctx context.Context) { profileID string seriesID string } + // Resolve every queued item to its series in one query. Marking or + // unmarking a whole series queues one mutation per episode, and those + // thousands of items collapse to a single series — resolving them + // one-at-a-time cost thousands of round-trips before the dedupe below + // could discard them. + itemIDs := make([]string, 0, len(batch)) + for mutation := range batch { + itemIDs = append(itemIDs, mutation.itemID) + } + seriesByItem, resolveErr := u.resolveSeriesIDs(ctx, itemIDs) + if resolveErr != nil { + u.logger.WarnContext(ctx, "interest series resolution failed", + "item_count", len(itemIDs), "error", resolveErr) + } + seen := make(map[recomputeKey]struct{}, len(batch)) for mutation, failures := range batch { if ctx.Err() != nil { return // shutting down; the rebuild task repairs anything dropped } - seriesID, ok, err := u.resolveSeriesID(ctx, mutation.itemID) - if err != nil { - u.logger.WarnContext(ctx, "interest series resolution failed", - "item_id", mutation.itemID, "error", err) + if resolveErr != nil { + // The lookup failed for the whole batch, so no item can be + // classified as "not a series". Requeue rather than silently + // dropping interest updates. requeue[mutation] = failures + 1 continue } + seriesID, ok := seriesByItem[mutation.itemID] if !ok { continue // movies and unknown items have no series interest } @@ -149,7 +163,7 @@ func (u *InterestUpdater) flush(ctx context.Context) { seen[key] = struct{}{} recomputeCtx, cancel := context.WithTimeout(ctx, interestRecomputeTime) - err = u.RecomputeSeries(recomputeCtx, mutation.userID, mutation.profileID, seriesID) + err := u.RecomputeSeries(recomputeCtx, mutation.userID, mutation.profileID, seriesID) cancel() if err != nil { u.logger.WarnContext(ctx, "interest recompute failed", @@ -178,26 +192,60 @@ func (u *InterestUpdater) flush(ctx context.Context) { u.mu.Unlock() } -// resolveSeriesID maps a media item ID (series, season, or episode) to its -// series content ID. Returns ok=false for movies and unknown items. -func (u *InterestUpdater) resolveSeriesID(ctx context.Context, itemID string) (string, bool, error) { - var seriesID string - err := u.pool.QueryRow(ctx, ` - SELECT series_id FROM episodes WHERE content_id = $1 - UNION ALL - SELECT series_id FROM seasons WHERE content_id = $1 - UNION ALL - SELECT content_id FROM media_items WHERE content_id = $1 AND type = 'series' - LIMIT 1`, - itemID, - ).Scan(&seriesID) - if errors.Is(err, pgx.ErrNoRows) { - return "", false, nil +// resolveSeriesIDs maps many media item IDs (series, season, or episode) to +// their series content IDs in one query. Items with no series — movies and +// unknown IDs — are absent from the result, which is what lets the caller keep +// treating "missing" as "no series interest". Semantics match resolveSeriesID +// per item; only the number of round-trips differs. +func (u *InterestUpdater) resolveSeriesIDs(ctx context.Context, itemIDs []string) (map[string]string, error) { + result := make(map[string]string, len(itemIDs)) + ids := make([]string, 0, len(itemIDs)) + seen := make(map[string]struct{}, len(itemIDs)) + for _, itemID := range itemIDs { + if itemID == "" { + continue + } + if _, dup := seen[itemID]; dup { + continue + } + seen[itemID] = struct{}{} + ids = append(ids, itemID) } + if len(ids) == 0 { + return result, nil + } + + // One branch per item kind, unioned, mirroring the single-item lookup. + rows, err := u.pool.Query(ctx, ` + SELECT content_id, series_id FROM episodes WHERE content_id = ANY($1::text[]) + UNION ALL + SELECT content_id, series_id FROM seasons WHERE content_id = ANY($1::text[]) + UNION ALL + SELECT content_id, content_id FROM media_items WHERE content_id = ANY($1::text[]) AND type = 'series'`, + ids, + ) if err != nil { - return "", false, err + return nil, err } - return seriesID, seriesID != "", nil + defer rows.Close() + for rows.Next() { + var itemID, seriesID string + if err := rows.Scan(&itemID, &seriesID); err != nil { + return nil, err + } + if seriesID == "" { + continue + } + // An ID matching more than one branch keeps the first row, matching + // the single-item query's UNION ALL ... LIMIT 1 ordering. + if _, exists := result[itemID]; !exists { + result[itemID] = seriesID + } + } + if err := rows.Err(); err != nil { + return nil, err + } + return result, nil } // RecomputeSeries rebuilds the (profile, series) interest rows from diff --git a/internal/notifications/interest_updater_resolve_test.go b/internal/notifications/interest_updater_resolve_test.go new file mode 100644 index 00000000..84466e4c --- /dev/null +++ b/internal/notifications/interest_updater_resolve_test.go @@ -0,0 +1,85 @@ +package notifications + +import ( + "context" + "os" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// TestResolveSeriesIDsBatchesLookups pins the batched item→series resolution +// used by the interest flush loop. Marking or unmarking a whole series queues +// one mutation per episode, and those all collapse to a single series; the +// previous per-item lookup issued one query each, which is what made a +// large-series unmark spend minutes in `episodes` lookups. +// +// Uses temp tables shadowing episodes/seasons/media_items, with the pool +// pinned to one connection so every query sees them. +func TestResolveSeriesIDsBatchesLookups(t *testing.T) { + dsn := os.Getenv("SILO_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("set SILO_TEST_DATABASE_URL to run the DB-backed series resolution test") + } + ctx := context.Background() + config, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatalf("parse db config: %v", err) + } + config.MaxConns = 1 + pool, err := pgxpool.NewWithConfig(ctx, config) + if err != nil { + t.Fatalf("connect db: %v", err) + } + t.Cleanup(pool.Close) + + for _, ddl := range []string{ + `CREATE TEMP TABLE episodes (content_id text PRIMARY KEY, series_id text NOT NULL)`, + `CREATE TEMP TABLE seasons (content_id text PRIMARY KEY, series_id text NOT NULL)`, + `CREATE TEMP TABLE media_items (content_id text PRIMARY KEY, type text NOT NULL)`, + } { + if _, err := pool.Exec(ctx, ddl); err != nil { + t.Fatalf("create temp table: %v", err) + } + } + if _, err := pool.Exec(ctx, ` + INSERT INTO episodes (content_id, series_id) VALUES ('ep-1','series-1'),('ep-2','series-1'),('ep-3','series-2'); + INSERT INTO seasons (content_id, series_id) VALUES ('season-1','series-1'); + INSERT INTO media_items (content_id, type) VALUES ('series-1','series'),('movie-1','movie'); + `); err != nil { + t.Fatalf("seed: %v", err) + } + + updater := &InterestUpdater{pool: pool} + + got, err := updater.resolveSeriesIDs(ctx, []string{ + "ep-1", "ep-2", "ep-3", "season-1", "series-1", "movie-1", "missing", "", "ep-1", + }) + if err != nil { + t.Fatalf("resolveSeriesIDs: %v", err) + } + + for itemID, wantSeries := range map[string]string{ + "ep-1": "series-1", + "ep-2": "series-1", + "ep-3": "series-2", + "season-1": "series-1", + "series-1": "series-1", + } { + if got[itemID] != wantSeries { + t.Errorf("resolveSeriesIDs[%s] = %q, want %q", itemID, got[itemID], wantSeries) + } + } + + // Movies, unknown IDs, and blanks must be absent rather than mapped: the + // flush loop reads "missing" as "no series interest" and skips them. + for _, itemID := range []string{"movie-1", "missing", ""} { + if seriesID, ok := got[itemID]; ok { + t.Errorf("resolveSeriesIDs[%q] = %q, want absent", itemID, seriesID) + } + } + + if empty, err := updater.resolveSeriesIDs(ctx, nil); err != nil || len(empty) != 0 { + t.Fatalf("resolveSeriesIDs(nil) = %v (%v), want empty", empty, err) + } +}