fix(notifications): restore store capabilities and batch series resolution (#647)
* fix(notifications): restore store capabilities and batch series resolution Marking or unmarking a large series spent minutes in the interest-tracking layer. Two independent defects, both in internal/notifications. The decorator embeds the userstore.UserStore *interface*, which promotes only that interface's methods. Every optional capability the backing store implements was therefore invisible through the wrapper, and because callers reach these by type assertion with a working fallback, the loss was silent: no error, no test failure, just the slow path. cmd/silo wraps the provider unconditionally when notifications are enabled, so in production userstore.MarkWatchedBatch's assertion failed and #645's transactional batch write never ran. AddVisibleHistory, VisibleHistoryTimestamps, and the jellycompat series rollup were degraded the same way. Forward all four capabilities explicitly, and add compile-time assertions so a future capability is a build error rather than a silent slowdown. Separately, the interest flush resolved each queued item to its series with one query apiece, then deduped the results. A whole-series mark queues one mutation per episode, so thousands of lookups collapsed to a single series after paying for all of them. Resolve the batch in one query and dedupe from that; the single-item resolveSeriesID had no other callers and is removed rather than left to drift. Measured on the dev server against a 6,375-episode series: mark 5.2s -> ~0.9s unmark 116s -> ~8.6s episodes index scans 18.5M -> 19,298 Fixes #646. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(notifications): queue interest recomputes on the batch fallback path Review catch (CodeRabbit on #647). When the backing store lacks WatchedBatchWriter, the forward handed the work to the generic helper against s.UserStore — the inner store — so this decorator's own MarkWatched hook never fired and nothing queued an interest recompute. Marking a series watched on such a backend updated progress and history but left profile_series_interest stale until an unrelated mutation or the rebuild task touched the series. Queue by requested target on that path, and do so even when the helper returns an error: the fallback is a per-target loop, so a mid-loop failure still leaves earlier targets written. A redundant queue costs one recompute; a missing one is silent staleness. The transactional path keeps queuing from written entries only on success, because on error nothing landed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(notifications): advertise the series rollup only when the store has it Review catch (Codex on #647). Forwarding SeriesEpisodeWatchCounts unconditionally made the wrapper always satisfy SeriesEpisodeRollupStore, even over the per-user SQLite backend, which has no catalog tables and cannot answer the query. Callers read "implements the interface" as "can do this", so every jellycompat series detail and browse would enter the fast path, take the error, log "series watch rollup query failed", and only then fall back — turning an expected capability absence into recurring warning noise on requests that succeeded. Make the capability conditional on the backing store, the way DeviceRegistry already is, via wrapper types composed in ForUser. The remaining capabilities stay unconditional: those have real generic fallbacks and every store can perform them. Tests cover both directions — a SQLite-backed wrapper must not advertise the rollup, and a rollup-capable store must keep it (and reach it) through the wrapper. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Quick
@@ -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 {
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user