diff --git a/internal/access/resolver_test.go b/internal/access/resolver_test.go index 6568ed21..7b6e05b7 100644 --- a/internal/access/resolver_test.go +++ b/internal/access/resolver_test.go @@ -99,6 +99,9 @@ func (s stubStore) ListHistory(context.Context, string, int, int) ([]userstore.W func (s stubStore) ListCompletedHistory(context.Context, userstore.CompletedHistoryQuery) ([]userstore.WatchHistoryEntry, error) { panic("unused") } +func (s stubStore) ListCompletedHistoryItems(context.Context, userstore.CompletedHistoryItemQuery) ([]userstore.CompletedHistoryItem, error) { + panic("unused") +} func (s stubStore) RemoveHistoryItems(context.Context, string, []string, time.Time) error { panic("unused") } diff --git a/internal/api/handlers/catalog_resources.go b/internal/api/handlers/catalog_resources.go index 3a22b155..b15e1ccf 100644 --- a/internal/api/handlers/catalog_resources.go +++ b/internal/api/handlers/catalog_resources.go @@ -243,7 +243,7 @@ func (h *CatalogResourceHandler) HandleGetSeasons(w http.ResponseWriter, r *http } var userData *catalog.SeasonUserData if hasProgressMap { - userData = aggregateUserDataFromProgress(episodes, progressMap) + userData = catalog.EpisodeRollupUserData(episodes, progressMap) } sr := h.items.toSeasonResponseFromEpisodes(r, id, s, episodes, userData) resp = append(resp, sr) diff --git a/internal/api/handlers/items.go b/internal/api/handlers/items.go index 08e3c0d8..2b2b3948 100644 --- a/internal/api/handlers/items.go +++ b/internal/api/handlers/items.go @@ -1370,12 +1370,11 @@ func (h *ItemsHandler) getLeafUserData(r *http.Request, contentID string, itemTy return nil } - progress, err := store.GetProgress(r.Context(), profileID, contentID) - if err != nil || progress == nil { + progress, err := userstore.GetProgressWithCompletedHistory(r.Context(), store, profileID, contentID) + if err != nil { return nil } - - return leafUserDataFromProgress(*progress) + return leafUserDataFromProgress(progress) } // listLeafUserData batch-fetches watch progress for the given content IDs in a @@ -1386,19 +1385,23 @@ func (h *ItemsHandler) listLeafUserData(r *http.Request, contentIDs []string) ma return nil } - progressMap, err := store.ListProgressByMediaItems(r.Context(), profileID, contentIDs) + progressMap, err := userstore.ListProgressWithCompletedHistory(r.Context(), store, profileID, contentIDs) if err != nil { return nil } result := make(map[string]*catalog.SeasonUserData, len(progressMap)) for contentID, progress := range progressMap { - result[contentID] = leafUserDataFromProgress(progress) + progressCopy := progress + result[contentID] = leafUserDataFromProgress(&progressCopy) } return result } -func leafUserDataFromProgress(progress userstore.WatchProgress) *catalog.SeasonUserData { +func leafUserDataFromProgress(progress *userstore.WatchProgress) *catalog.SeasonUserData { + if progress == nil { + return nil + } return &catalog.SeasonUserData{ PositionSeconds: progress.PositionSeconds, DurationSeconds: progress.DurationSeconds, @@ -1467,7 +1470,7 @@ func (h *ItemsHandler) getAggregateUserData(r *http.Request, episodes []*models. if err != nil { return nil } - return aggregateUserDataFromProgress(episodes, progressMap) + return catalog.EpisodeRollupUserData(episodes, progressMap) } func (h *ItemsHandler) progressMapForEpisodes(r *http.Request, episodes []*models.Episode) (map[string]userstore.WatchProgress, bool) { @@ -1475,7 +1478,8 @@ func (h *ItemsHandler) progressMapForEpisodes(r *http.Request, episodes []*model if !ok { return nil, false } - progressMap, err := h.listProgressForEpisodeIDs(r.Context(), store, profileID, episodeContentIDs(episodes)) + episodeIDs := episodeContentIDs(episodes) + progressMap, err := h.listProgressForEpisodeIDs(r.Context(), store, profileID, episodeIDs) if err != nil { return nil, false } @@ -1490,7 +1494,7 @@ func (h *ItemsHandler) listProgressForEpisodeIDs(ctx context.Context, store user if end > len(episodeIDs) { end = len(episodeIDs) } - chunk, err := store.ListProgressByMediaItems(ctx, profileID, episodeIDs[start:end]) + chunk, err := userstore.ListProgressWithCompletedHistory(ctx, store, profileID, episodeIDs[start:end]) if err != nil { return nil, err } @@ -1501,39 +1505,6 @@ func (h *ItemsHandler) listProgressForEpisodeIDs(ctx context.Context, store user return progressMap, nil } -func aggregateUserDataFromProgress(episodes []*models.Episode, progressMap map[string]userstore.WatchProgress) *catalog.SeasonUserData { - if len(episodes) == 0 { - return nil - } - - var watchedCount int - var inProgressCount int - for _, ep := range episodes { - if ep == nil { - continue - } - progress, ok := progressMap[ep.ContentID] - if !ok { - continue - } - if progress.Completed { - watchedCount++ - continue - } - if progress.PositionSeconds > 0 { - inProgressCount++ - } - } - - unplayedCount := len(episodes) - watchedCount - return &catalog.SeasonUserData{ - WatchedCount: watchedCount, - UnplayedCount: unplayedCount, - InProgressCount: inProgressCount, - Played: watchedCount == len(episodes), - } -} - func episodeContentIDs(episodes []*models.Episode) []string { ids := make([]string, 0, len(episodes)) seen := make(map[string]struct{}, len(episodes)) diff --git a/internal/api/handlers/items_user_data_test.go b/internal/api/handlers/items_user_data_test.go index 7d26c2d1..af7502f2 100644 --- a/internal/api/handlers/items_user_data_test.go +++ b/internal/api/handlers/items_user_data_test.go @@ -3,6 +3,7 @@ package handlers import ( "context" "errors" + "net/http" "net/http/httptest" "strconv" "testing" @@ -75,6 +76,53 @@ func TestGetLeafUserDataReturnsAudiobookProgress(t *testing.T) { } } +func TestGetLeafUserDataUsesCompletedHistoryWhenProgressMissing(t *testing.T) { + store := newPlaybackTestStore(t) + addCompletedHistoryForUserDataTest(t, store, "movie-history-only") + handler := &ItemsHandler{storeProvider: testUserStoreProvider{store: store}} + req := authorizedUserDataRequest() + + userData := handler.getLeafUserData(req, "movie-history-only", "movie") + if userData == nil { + t.Fatal("movie user data = nil, want history-backed watched state") + } + if !userData.Played { + t.Fatalf("Played = false, want true from completed history") + } + if userData.PositionSeconds != 0 || userData.IsInProgress { + t.Fatalf("history-only user data = %+v, want no resume position", userData) + } +} + +func TestGetLeafUserDataPreservesResumeWhenCompletedHistoryExists(t *testing.T) { + store := newPlaybackTestStore(t) + if err := store.SetProgressAt( + context.Background(), + "profile-1", + "movie-rewatch", + 600, + 7200, + false, + time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC), + ); err != nil { + t.Fatalf("seed progress: %v", err) + } + addCompletedHistoryForUserDataTest(t, store, "movie-rewatch") + handler := &ItemsHandler{storeProvider: testUserStoreProvider{store: store}} + req := authorizedUserDataRequest() + + userData := handler.getLeafUserData(req, "movie-rewatch", "movie") + if userData == nil { + t.Fatal("movie user data = nil, want progress-backed user data") + } + if !userData.Played { + t.Fatalf("Played = false, want true from completed history") + } + if userData.PositionSeconds != 600 || userData.DurationSeconds != 7200 || !userData.IsInProgress { + t.Fatalf("resume fields = %+v, want in-progress resume preserved", userData) + } +} + func TestGetAggregateUserDataReturnsNilWhenProgressBatchFails(t *testing.T) { store := &failingBatchProgressStore{} handler := &ItemsHandler{storeProvider: testUserStoreProvider{store: store}} @@ -97,6 +145,36 @@ func TestGetAggregateUserDataReturnsNilWhenProgressBatchFails(t *testing.T) { } } +func TestGetAggregateUserDataCountsCompletedHistory(t *testing.T) { + store := newPlaybackTestStore(t) + if err := store.SetProgressAt( + context.Background(), + "profile-1", + "episode-progress-complete", + 0, + 1800, + true, + time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC), + ); err != nil { + t.Fatalf("seed progress: %v", err) + } + addCompletedHistoryForUserDataTest(t, store, "episode-history-complete") + handler := &ItemsHandler{storeProvider: testUserStoreProvider{store: store}} + req := authorizedUserDataRequest() + + userData := handler.getAggregateUserData(req, []*models.Episode{ + {ContentID: "episode-progress-complete"}, + {ContentID: "episode-history-complete"}, + {ContentID: "episode-unplayed"}, + }) + if userData == nil { + t.Fatal("aggregate user data = nil, want counts") + } + if userData.WatchedCount != 2 || userData.UnplayedCount != 1 || userData.Played { + t.Fatalf("aggregate user data = %+v, want two watched and one unplayed", userData) + } +} + type failingBatchProgressStore struct { userstore.UserStore calls int @@ -117,3 +195,24 @@ func (s *failingBatchProgressStore) ListProgressByMediaItems( } return progress, nil } + +func authorizedUserDataRequest() *http.Request { + req := httptest.NewRequest("GET", "/items/movie-1", nil) + ctx := apimw.SetClaims(req.Context(), &auth.Claims{UserID: 1}) + ctx = apimw.SetProfileID(ctx, "profile-1") + return req.WithContext(ctx) +} + +func addCompletedHistoryForUserDataTest(t *testing.T, store userstore.UserStore, mediaItemID string) { + t.Helper() + if err := store.AddHistory(context.Background(), userstore.WatchHistoryEntry{ + ProfileID: "profile-1", + MediaItemID: mediaItemID, + WatchedAt: "2026-05-04T12:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceTrakt, + }); err != nil { + t.Fatalf("seed completed history: %v", err) + } +} diff --git a/internal/api/handlers/user_state.go b/internal/api/handlers/user_state.go index c947b614..c8ee4ece 100644 --- a/internal/api/handlers/user_state.go +++ b/internal/api/handlers/user_state.go @@ -123,7 +123,7 @@ func resolveItemUserStatesWithOptions( } } - progressMap, err := store.ListProgressByMediaItems(ctx, profileID, progressIDs) + progressMap, err := userstore.ListProgressWithCompletedHistory(ctx, store, profileID, progressIDs) if err != nil { return nil, err } diff --git a/internal/api/handlers/user_state_test.go b/internal/api/handlers/user_state_test.go index 9ff974b3..c10684da 100644 --- a/internal/api/handlers/user_state_test.go +++ b/internal/api/handlers/user_state_test.go @@ -6,6 +6,7 @@ import ( "time" "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/userstore" ) func TestResolveItemUserStatesIncludesCompletedEbookReaderProgress(t *testing.T) { @@ -94,3 +95,32 @@ func TestResolveItemUserStatesExcludesHiddenEbookReaderProgress(t *testing.T) { t.Fatalf("ebook updated after hidden_before = %+v, want played again", states["ebook-reread"]) } } + +func TestResolveItemUserStatesIncludesCompletedHistory(t *testing.T) { + ctx := context.Background() + store := newProfileTestStore(t) + addCompletedHistoryForUserDataTest(t, store, "movie-history-only") + items := []*models.MediaItem{ + {ContentID: "movie-history-only", Type: "movie", Title: "Imported Movie"}, + } + + states, err := resolveItemUserStates(ctx, store, "profile-1", nil, items) + if err != nil { + t.Fatalf("resolveItemUserStates: %v", err) + } + if states["movie-history-only"] == nil || !states["movie-history-only"].Played { + t.Fatalf("history-only movie state = %+v, want played", states["movie-history-only"]) + } +} + +func TestAllEpisodesCompletedIncludesCompletedHistory(t *testing.T) { + episodes := []*models.Episode{{ContentID: "episode-progress"}, {ContentID: "episode-history"}} + progress := map[string]userstore.WatchProgress{ + "episode-progress": {MediaItemID: "episode-progress", Completed: true}, + "episode-history": {MediaItemID: "episode-history", Completed: true}, + } + + if !allEpisodesCompleted(episodes, progress) { + t.Fatal("allEpisodesCompleted = false, want completed from progress plus history") + } +} diff --git a/internal/catalog/user_data_rollup.go b/internal/catalog/user_data_rollup.go new file mode 100644 index 00000000..a58d71a8 --- /dev/null +++ b/internal/catalog/user_data_rollup.go @@ -0,0 +1,44 @@ +package catalog + +import ( + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/userstore" +) + +// EpisodeRollupUserData computes aggregate watch state for a season or series +// from pre-fetched per-episode progress. Completed history should already be +// folded into progressMap by the caller's userstore helper. +func EpisodeRollupUserData(episodes []*models.Episode, progressMap map[string]userstore.WatchProgress) *SeasonUserData { + if len(episodes) == 0 { + return &SeasonUserData{} + } + + watchedCount := 0 + inProgressCount := 0 + totalEpisodes := 0 + for _, ep := range episodes { + if ep == nil { + continue + } + totalEpisodes++ + progress, ok := progressMap[ep.ContentID] + if ok && progress.Completed { + watchedCount++ + continue + } + if ok && progress.PositionSeconds > 0 { + inProgressCount++ + } + } + if totalEpisodes == 0 { + return &SeasonUserData{} + } + + unplayedCount := totalEpisodes - watchedCount + return &SeasonUserData{ + WatchedCount: watchedCount, + UnplayedCount: unplayedCount, + InProgressCount: inProgressCount, + Played: watchedCount == totalEpisodes, + } +} diff --git a/internal/jellycompat/content_direct.go b/internal/jellycompat/content_direct.go index f8e0f882..de834d68 100644 --- a/internal/jellycompat/content_direct.go +++ b/internal/jellycompat/content_direct.go @@ -503,15 +503,8 @@ func (s *directContentService) GetItemDetail(ctx context.Context, session *Sessi if s.storeProvider != nil { store, storeErr := s.storeProvider.ForUser(ctx, session.StreamAppUserID) if storeErr == nil { - progress, _ := store.GetProgress(ctx, session.ProfileID, contentID) - if progress != nil { - result.UserData = &catalog.SeasonUserData{ - PositionSeconds: progress.PositionSeconds, - DurationSeconds: progress.DurationSeconds, - Played: progress.Completed, - IsInProgress: progress.PositionSeconds > 0, - } - } + progress, _ := userstore.GetProgressWithCompletedHistory(ctx, store, session.ProfileID, contentID) + result.UserData = seasonUserDataFromProgress(progress) // A series never has a progress row of its own, so roll watch // state up from its episodes (mirrors applySeasonUserData) to @@ -519,8 +512,9 @@ func (s *directContentService) GetItemDetail(ctx context.Context, session *Sessi if result.UserData == nil && strings.EqualFold(result.Type, "series") && s.episodeRepo != nil { if episodesBySeries, epErr := s.episodeRepo.ListBySeriesIDs(ctx, []string{contentID}); epErr == nil { episodes := episodesBySeries[contentID] - progressMap := chunkedProgressByMediaItems(ctx, store, session.ProfileID, modelEpisodeContentIDs(episodes)) - result.UserData = seriesUserDataFromEpisodes(episodes, progressMap) + episodeIDs := modelEpisodeContentIDs(episodes) + progressMap := chunkedProgressByMediaItems(ctx, store, session.ProfileID, episodeIDs) + result.UserData = catalog.EpisodeRollupUserData(episodes, progressMap) } } } @@ -646,7 +640,7 @@ func (s *directContentService) ListEpisodes(ctx context.Context, session *Sessio for _, ep := range episodes { episodeIDs = append(episodeIDs, ep.ContentID) } - if progressEntries, progressErr := store.ListProgressByMediaItems(ctx, session.ProfileID, episodeIDs); progressErr == nil { + if progressEntries, progressErr := userstore.ListProgressWithCompletedHistory(ctx, store, session.ProfileID, episodeIDs); progressErr == nil { progressMap = progressEntries } } @@ -661,7 +655,8 @@ func (s *directContentService) ListEpisodes(ctx context.Context, session *Sessio ue := modelEpisodeToUpstream(ep, seriesID) s.presignEpisode(ctx, &ue) if progress, ok := progressMap[ep.ContentID]; ok { - ue.UserData = seasonUserDataFromProgress(progress) + progressCopy := progress + ue.UserData = seasonUserDataFromProgress(&progressCopy) } result = append(result, ue) } @@ -710,7 +705,7 @@ func (s *directContentService) enrichListItemsUserData(ctx context.Context, sess contentIDs = append(contentIDs, item.ContentID) } } - progressMap, err := store.ListProgressByMediaItems(ctx, session.ProfileID, contentIDs) + progressMap, err := userstore.ListProgressWithCompletedHistory(ctx, store, session.ProfileID, contentIDs) if err != nil { return } @@ -718,7 +713,8 @@ func (s *directContentService) enrichListItemsUserData(ctx context.Context, sess for i := range items { progress, ok := progressMap[items[i].ContentID] if ok { - items[i].UserData = seasonUserDataFromProgress(progress) + progressCopy := progress + items[i].UserData = seasonUserDataFromProgress(&progressCopy) } } @@ -778,42 +774,11 @@ func (s *directContentService) enrichSeriesListUserData(ctx context.Context, ses continue } if episodes, ok := episodesBySeries[items[i].ContentID]; ok { - items[i].UserData = seriesUserDataFromEpisodes(episodes, progressMap) + items[i].UserData = catalog.EpisodeRollupUserData(episodes, progressMap) } } } -// seriesUserDataFromEpisodes computes WatchedCount/UnplayedCount/ -// InProgressCount/Played for a whole series from a pre-fetched progressMap. -// Pure function — no I/O. Counting semantics match the native API's series -// rollup (all episodes including specials; in-progress = started but not -// completed). -func seriesUserDataFromEpisodes(episodes []*models.Episode, progressMap map[string]userstore.WatchProgress) *catalog.SeasonUserData { - watched := 0 - unplayed := 0 - inProgress := 0 - for _, ep := range episodes { - if ep == nil { - continue - } - progress, ok := progressMap[ep.ContentID] - if ok && progress.Completed { - watched++ - continue - } - if ok && progress.PositionSeconds > 0 { - inProgress++ - } - unplayed++ - } - return &catalog.SeasonUserData{ - WatchedCount: watched, - UnplayedCount: unplayed, - InProgressCount: inProgress, - Played: unplayed == 0 && len(episodes) > 0, - } -} - // modelEpisodeContentIDs returns the non-empty content ids of the given episodes. func modelEpisodeContentIDs(episodes []*models.Episode) []string { ids := make([]string, 0, len(episodes)) @@ -834,7 +799,7 @@ func chunkedProgressByMediaItems(ctx context.Context, store userstore.UserStore, const chunkSize = 500 result := make(map[string]userstore.WatchProgress, len(mediaItemIDs)) for start := 0; start < len(mediaItemIDs); start += chunkSize { - chunk, err := store.ListProgressByMediaItems(ctx, profileID, mediaItemIDs[start:min(start+chunkSize, len(mediaItemIDs))]) + chunk, err := userstore.ListProgressWithCompletedHistory(ctx, store, profileID, mediaItemIDs[start:min(start+chunkSize, len(mediaItemIDs))]) if err != nil { continue } @@ -864,7 +829,7 @@ func (s *directContentService) enrichSeasonUserData(ctx context.Context, session episodeIDs = append(episodeIDs, ep.ContentID) } } - progressMap, err := store.ListProgressByMediaItems(ctx, session.ProfileID, episodeIDs) + progressMap, err := userstore.ListProgressWithCompletedHistory(ctx, store, session.ProfileID, episodeIDs) if err != nil { return } @@ -874,23 +839,9 @@ func (s *directContentService) enrichSeasonUserData(ctx context.Context, session // applySeasonUserData computes WatchedCount/UnplayedCount/Played for a season // using a pre-fetched progressMap. Pure function — no I/O. func applySeasonUserData(season *upstreamSeason, episodes []*models.Episode, progressMap map[string]userstore.WatchProgress) { - watched := 0 - unplayed := 0 - for _, ep := range episodes { - if ep == nil { - continue - } - progress, ok := progressMap[ep.ContentID] - if ok && progress.Completed { - watched++ - } else { - unplayed++ - } - } - season.UserData = &catalog.SeasonUserData{ - WatchedCount: watched, - UnplayedCount: unplayed, - Played: unplayed == 0 && len(episodes) > 0, + season.UserData = catalog.EpisodeRollupUserData(episodes, progressMap) + if season.UserData == nil { + season.UserData = &catalog.SeasonUserData{} } } @@ -905,7 +856,7 @@ func (s *directContentService) batchProgressForEpisodes(ctx context.Context, ses if err != nil { return map[string]userstore.WatchProgress{} } - progressMap, err := store.ListProgressByMediaItems(ctx, session.ProfileID, episodeIDs) + progressMap, err := userstore.ListProgressWithCompletedHistory(ctx, store, session.ProfileID, episodeIDs) if err != nil || progressMap == nil { return map[string]userstore.WatchProgress{} } @@ -914,9 +865,10 @@ func (s *directContentService) batchProgressForEpisodes(ctx context.Context, ses // enrichEpisodeUserData adds user data for a single episode. func (s *directContentService) enrichEpisodeUserData(ctx context.Context, session *Session, ep *upstreamEpisode) { - if progressMap, err := s.progressMapForContentIDs(ctx, session, []string{ep.ContentID}); err == nil { + if progressMap, err := s.progressForContentIDs(ctx, session, []string{ep.ContentID}); err == nil { if progress, ok := progressMap[ep.ContentID]; ok { - ep.UserData = seasonUserDataFromProgress(progress) + progressCopy := progress + ep.UserData = seasonUserDataFromProgress(&progressCopy) } } } @@ -928,15 +880,22 @@ func (s *directContentService) userStore(ctx context.Context, session *Session) return s.storeProvider.ForUser(ctx, session.StreamAppUserID) } -func (s *directContentService) progressMapForContentIDs(ctx context.Context, session *Session, contentIDs []string) (map[string]userstore.WatchProgress, error) { +func (s *directContentService) progressForContentIDs(ctx context.Context, session *Session, contentIDs []string) (map[string]userstore.WatchProgress, error) { store, err := s.userStore(ctx, session) if err != nil { return nil, err } - return store.ListProgressByMediaItems(ctx, session.ProfileID, contentIDs) + progressMap, err := userstore.ListProgressWithCompletedHistory(ctx, store, session.ProfileID, contentIDs) + if err != nil { + return nil, err + } + return progressMap, nil } -func seasonUserDataFromProgress(progress userstore.WatchProgress) *catalog.SeasonUserData { +func seasonUserDataFromProgress(progress *userstore.WatchProgress) *catalog.SeasonUserData { + if progress == nil { + return nil + } return &catalog.SeasonUserData{ PositionSeconds: progress.PositionSeconds, DurationSeconds: progress.DurationSeconds, diff --git a/internal/jellycompat/content_direct_test.go b/internal/jellycompat/content_direct_test.go index d8f79cde..9bdd3cc4 100644 --- a/internal/jellycompat/content_direct_test.go +++ b/internal/jellycompat/content_direct_test.go @@ -241,6 +241,9 @@ func (s *progressCountingStore) ListHistory(context.Context, string, int, int) ( func (s *progressCountingStore) ListCompletedHistory(context.Context, userstore.CompletedHistoryQuery) ([]userstore.WatchHistoryEntry, error) { panic("unused") } +func (s *progressCountingStore) ListCompletedHistoryItems(context.Context, userstore.CompletedHistoryItemQuery) ([]userstore.CompletedHistoryItem, error) { + return nil, nil +} func (s *progressCountingStore) RemoveHistoryItems(context.Context, string, []string, time.Time) error { panic("unused") } diff --git a/internal/jellycompat/mapping.go b/internal/jellycompat/mapping.go index b3ab98c3..f650f02d 100644 --- a/internal/jellycompat/mapping.go +++ b/internal/jellycompat/mapping.go @@ -548,10 +548,11 @@ func userDataDTO(itemID string, data *catalog.SeasonUserData, isFavorite bool, p if progress != nil { pos := clampResumeSeconds(progress.PositionSeconds, progress.DurationSeconds) + played := dto.Played || progress.Completed dto.PlaybackPositionTicks = secondsToTicks(pos) - dto.PlayedPercentage = playedPercentage(pos, progress.DurationSeconds, progress.Completed) - dto.Played = progress.Completed - if progress.Completed { + dto.PlayedPercentage = playedPercentage(pos, progress.DurationSeconds, played) + dto.Played = played + if played { dto.PlayCount = 1 } dto.LastPlayedDate = progress.UpdatedAt diff --git a/internal/jellycompat/mapping_userdata_test.go b/internal/jellycompat/mapping_userdata_test.go index 944192f7..d87537d3 100644 --- a/internal/jellycompat/mapping_userdata_test.go +++ b/internal/jellycompat/mapping_userdata_test.go @@ -92,6 +92,24 @@ func TestUserDataDTOProgressCompletedZeros(t *testing.T) { } } +func TestUserDataDTOProgressDoesNotClearPlayedData(t *testing.T) { + data := &catalog.SeasonUserData{Played: true} + progress := &upstreamProgress{ + MediaItemID: "x", + PositionSeconds: 600.0, + DurationSeconds: 1290.0, + Completed: false, + } + + dto := userDataDTO("item-4", data, false, progress) + if !dto.Played { + t.Fatalf("Played = false, want aggregate played state preserved") + } + if dto.PlayCount != 1 { + t.Fatalf("PlayCount = %d, want watched count preserved", dto.PlayCount) + } +} + func TestUserDataDTOProgressRewatchKeepsPlayedAndPosition(t *testing.T) { progress := &upstreamProgress{ MediaItemID: "x", diff --git a/internal/jellycompat/series_userdata_test.go b/internal/jellycompat/series_userdata_test.go index 557bbaa6..5b484bb4 100644 --- a/internal/jellycompat/series_userdata_test.go +++ b/internal/jellycompat/series_userdata_test.go @@ -3,6 +3,7 @@ package jellycompat import ( "testing" + "github.com/Silo-Server/silo-server/internal/catalog" "github.com/Silo-Server/silo-server/internal/models" "github.com/Silo-Server/silo-server/internal/userstore" ) @@ -62,13 +63,20 @@ func TestSeriesUserDataFromEpisodes(t *testing.T) { wantPlayed: false, }, { - name: "nil episodes skipped", + name: "nil episodes do not count as unplayed", episodes: []*models.Episode{nil, ep("a"), nil}, progress: map[string]userstore.WatchProgress{ "a": {Completed: true}, }, - wantWatched: 1, - wantPlayed: true, + wantWatched: 1, + wantUnplayed: 0, + wantPlayed: true, + }, + { + name: "all nil episodes", + episodes: []*models.Episode{nil, nil}, + progress: map[string]userstore.WatchProgress{}, + wantPlayed: false, }, { name: "zero-position progress row is not in-progress", @@ -84,7 +92,7 @@ func TestSeriesUserDataFromEpisodes(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - got := seriesUserDataFromEpisodes(tc.episodes, tc.progress) + got := catalog.EpisodeRollupUserData(tc.episodes, tc.progress) if got.WatchedCount != tc.wantWatched { t.Errorf("WatchedCount = %d, want %d", got.WatchedCount, tc.wantWatched) } @@ -109,3 +117,17 @@ func TestModelEpisodeContentIDs(t *testing.T) { t.Errorf("modelEpisodeContentIDs = %v, want [a b]", got) } } + +func TestSeriesUserDataFromEpisodesIncludesCompletedHistory(t *testing.T) { + got := catalog.EpisodeRollupUserData( + []*models.Episode{ep("progress-complete"), ep("history-complete"), ep("unplayed")}, + map[string]userstore.WatchProgress{ + "progress-complete": {Completed: true}, + "history-complete": {Completed: true}, + }, + ) + + if got.WatchedCount != 2 || got.UnplayedCount != 1 || got.Played { + t.Fatalf("series user data = %+v, want two watched and one unplayed", got) + } +} diff --git a/internal/jellycompat/userdata_direct.go b/internal/jellycompat/userdata_direct.go index 42315717..7c4bb3c3 100644 --- a/internal/jellycompat/userdata_direct.go +++ b/internal/jellycompat/userdata_direct.go @@ -222,7 +222,8 @@ func (s *directUserDataService) ListProgressByMediaItems(ctx context.Context, se return nil, fmt.Errorf("open user store: %w", err) } - progressMap, err := store.ListProgressByMediaItems(ctx, session.ProfileID, mediaItemIDs) + mediaItemIDs = normalizeContentIDs(mediaItemIDs) + progressMap, err := userstore.ListProgressWithCompletedHistory(ctx, store, session.ProfileID, mediaItemIDs) if err != nil { return nil, fmt.Errorf("list progress by media items: %w", err) } @@ -241,7 +242,7 @@ func (s *directUserDataService) GetProgress(ctx context.Context, session *Sessio return nil, fmt.Errorf("open user store: %w", err) } - progress, err := store.GetProgress(ctx, session.ProfileID, contentID) + progress, err := userstore.GetProgressWithCompletedHistory(ctx, store, session.ProfileID, contentID) if err != nil { return nil, fmt.Errorf("get progress: %w", err) } diff --git a/internal/jellycompat/userdata_direct_test.go b/internal/jellycompat/userdata_direct_test.go new file mode 100644 index 00000000..7d89ec99 --- /dev/null +++ b/internal/jellycompat/userdata_direct_test.go @@ -0,0 +1,114 @@ +package jellycompat + +import ( + "context" + "database/sql" + "net/url" + "testing" + + "github.com/Silo-Server/silo-server/internal/models" + "github.com/Silo-Server/silo-server/internal/userdb" + "github.com/Silo-Server/silo-server/internal/userstore" +) + +type compatTestUserStoreProvider struct { + store userstore.UserStore +} + +func (p compatTestUserStoreProvider) ForUser(context.Context, int) (userstore.UserStore, error) { + return p.store, nil +} + +func (p compatTestUserStoreProvider) Close() error { + return nil +} + +func TestDirectUserDataServiceProgressUsesCompletedHistory(t *testing.T) { + store := newJellycompatUserStore(t) + addCompletedHistoryForJellycompatTest(t, store, "movie-history-only") + service := &directUserDataService{storeProvider: compatTestUserStoreProvider{store: store}} + session := &Session{StreamAppUserID: 1, ProfileID: "profile-1"} + + progress, err := service.GetProgress(context.Background(), session, "movie-history-only") + if err != nil { + t.Fatalf("GetProgress: %v", err) + } + if progress == nil || !progress.Completed { + t.Fatalf("GetProgress = %+v, want synthetic completed progress", progress) + } + + progressMap, err := service.ListProgressByMediaItems(context.Background(), session, []string{"movie-history-only"}) + if err != nil { + t.Fatalf("ListProgressByMediaItems: %v", err) + } + if progressMap["movie-history-only"] == nil || !progressMap["movie-history-only"].Completed { + t.Fatalf("ListProgressByMediaItems = %+v, want completed history overlay", progressMap) + } +} + +func TestBrowseItemsPlayedFilterUsesCompletedHistory(t *testing.T) { + store := newJellycompatUserStore(t) + addCompletedHistoryForJellycompatTest(t, store, "movie-history-only") + browse := &stubBrowseSource{ + items: []*models.MediaItem{ + {ContentID: "movie-history-only", Type: "movie", Title: "Imported"}, + {ContentID: "movie-unplayed", Type: "movie", Title: "Unplayed"}, + }, + total: 2, + } + service := newDirectContentServiceForTest(browse, compatTestUserStoreProvider{store: store}) + session := &Session{StreamAppUserID: 1, ProfileID: "profile-1"} + + playedParams := url.Values{} + playedParams.Set("is_played", "true") + played, err := service.BrowseItems(context.Background(), session, playedParams) + if err != nil { + t.Fatalf("BrowseItems played: %v", err) + } + if len(played.Items) != 1 || played.Items[0].ContentID != "movie-history-only" { + t.Fatalf("played filter items = %+v, want history-only movie", played.Items) + } + + unplayedParams := url.Values{} + unplayedParams.Set("is_played", "false") + unplayed, err := service.BrowseItems(context.Background(), session, unplayedParams) + if err != nil { + t.Fatalf("BrowseItems unplayed: %v", err) + } + if len(unplayed.Items) != 1 || unplayed.Items[0].ContentID != "movie-unplayed" { + t.Fatalf("unplayed filter items = %+v, want only unplayed movie", unplayed.Items) + } +} + +func newJellycompatUserStore(t *testing.T) userstore.UserStore { + t.Helper() + 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("InitSchema: %v", err) + } + store := userdb.NewSQLiteUserStore(db) + if err := store.CreateProfile(context.Background(), userstore.Profile{ID: "profile-1", Name: "Profile"}); err != nil { + t.Fatalf("CreateProfile: %v", err) + } + return store +} + +func addCompletedHistoryForJellycompatTest(t *testing.T, store userstore.UserStore, mediaItemID string) { + t.Helper() + if err := store.AddHistory(context.Background(), userstore.WatchHistoryEntry{ + ProfileID: "profile-1", + MediaItemID: mediaItemID, + WatchedAt: "2026-05-04T12:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceTrakt, + }); err != nil { + t.Fatalf("AddHistory: %v", err) + } +} diff --git a/internal/recommendations/signals_test.go b/internal/recommendations/signals_test.go index ae8f54ee..5fc0cf41 100644 --- a/internal/recommendations/signals_test.go +++ b/internal/recommendations/signals_test.go @@ -134,6 +134,34 @@ func (s *fakeSignalStore) ListCompletedHistory(_ context.Context, query userstor return filtered[query.Offset:end], nil } +func (s *fakeSignalStore) ListCompletedHistoryItems(_ context.Context, query userstore.CompletedHistoryItemQuery) ([]userstore.CompletedHistoryItem, error) { + latest := map[string]userstore.CompletedHistoryItem{} + for _, entry := range s.history { + if entry.ProfileID != query.ProfileID || !entry.Completed { + continue + } + if len(query.MediaItemIDs) > 0 && !slices.Contains(query.MediaItemIDs, entry.MediaItemID) { + continue + } + if len(query.IncludeSources) > 0 && !slices.Contains(query.IncludeSources, entry.Source) { + continue + } + if slices.Contains(query.ExcludeSources, entry.Source) { + continue + } + current := latest[entry.MediaItemID] + if current.MediaItemID != "" && current.WatchedAt >= entry.WatchedAt { + continue + } + latest[entry.MediaItemID] = userstore.CompletedHistoryItem{MediaItemID: entry.MediaItemID, WatchedAt: entry.WatchedAt} + } + items := make([]userstore.CompletedHistoryItem, 0, len(latest)) + for _, item := range latest { + items = append(items, item) + } + return items, nil +} + func (s *fakeSignalStore) GetProfile(context.Context, string) (*userstore.Profile, error) { return s.profile, nil } diff --git a/internal/userdb/progress.go b/internal/userdb/progress.go index f34d6aec..4b0bf93b 100644 --- a/internal/userdb/progress.go +++ b/internal/userdb/progress.go @@ -24,13 +24,23 @@ func UpdateProgress(db *sql.DB, profileID, mediaItemID string, position, duratio return nil } now := nowUTC() + completed := false + if duration > 0 && position/duration > userstore.WatchedFraction(thresholds.WatchedPct) { + completed = true + position = 0 // match MarkWatched() — completed rows hold no resume point + } // Mirrors the Postgres pgstore UpdateProgress: `completed` is a one-way // watched latch; position resets to 0 on completion so a rewatch // heartbeat on a completed row re-enters Continue Watching through plain // MAX while the watched flag survives. query := ` INSERT INTO watch_progress (profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) - VALUES (?, ?, ?, ?, ?, ?) + SELECT ?, ?, ?, ?, ?, ` + visibleTimestampSQL + ` + FROM (SELECT 1) seed + LEFT JOIN hidden_history_items hhi + ON hhi.profile_id = ? + AND hhi.media_item_id = ? + WHERE true ON CONFLICT(profile_id, media_item_id) DO UPDATE SET position_seconds = CASE WHEN excluded.completed = 1 THEN 0 ELSE MAX(excluded.position_seconds, watch_progress.position_seconds) END, @@ -39,23 +49,20 @@ func UpdateProgress(db *sql.DB, profileID, mediaItemID string, position, duratio THEN 1 ELSE watch_progress.completed END, updated_at = excluded.updated_at ` - completed := false - if duration > 0 && position/duration > userstore.WatchedFraction(thresholds.WatchedPct) { - completed = true - position = 0 // match MarkWatched() — completed rows hold no resume point - } - _, err := db.Exec(query, profileID, mediaItemID, position, duration, completed, now) + _, err := db.Exec(query, profileID, mediaItemID, position, duration, completed, now, now, profileID, mediaItemID) if err != nil { return fmt.Errorf("updating progress: %w", err) } return nil } -// SetProgress bypasses the forward-only guard (for rewatches/explicit seek). -// It unconditionally sets the position to the given value. The completed flag -// stays a one-way watched latch: only ClearProgress/ClearProgressBatch (mark -// unwatched) release it. +// SetProgress bypasses the forward-only guard (for rewatches/explicit seek) +// after the min-resume threshold. The completed flag stays a one-way watched +// latch: only ClearProgress/ClearProgressBatch (mark unwatched) release it. func SetProgress(db *sql.DB, profileID, mediaItemID string, position, duration float64, thresholds userstore.ProgressThresholds) error { + if duration > 0 && position > 0 && position/duration < userstore.MinResumeFraction(thresholds.MinResumePct) { + return nil + } now := nowUTC() completed := false if duration > 0 && position/duration > userstore.WatchedFraction(thresholds.WatchedPct) { @@ -64,14 +71,19 @@ func SetProgress(db *sql.DB, profileID, mediaItemID string, position, duration f } query := ` INSERT INTO watch_progress (profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) - VALUES (?, ?, ?, ?, ?, ?) + SELECT ?, ?, ?, ?, ?, ` + visibleTimestampSQL + ` + FROM (SELECT 1) seed + LEFT JOIN hidden_history_items hhi + ON hhi.profile_id = ? + AND hhi.media_item_id = ? + WHERE true ON CONFLICT(profile_id, media_item_id) DO UPDATE SET position_seconds = excluded.position_seconds, duration_seconds = excluded.duration_seconds, completed = watch_progress.completed OR excluded.completed, updated_at = excluded.updated_at ` - _, err := db.Exec(query, profileID, mediaItemID, position, duration, completed, now) + _, err := db.Exec(query, profileID, mediaItemID, position, duration, completed, now, now, profileID, mediaItemID) if err != nil { return fmt.Errorf("setting progress: %w", err) } @@ -166,14 +178,19 @@ func MarkWatched(db *sql.DB, profileID, mediaItemID string, duration float64) er now := nowUTC() query := ` INSERT INTO watch_progress (profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) - VALUES (?, ?, 0, ?, 1, ?) + SELECT ?, ?, 0, ?, 1, ` + visibleTimestampSQL + ` + FROM (SELECT 1) seed + LEFT JOIN hidden_history_items hhi + ON hhi.profile_id = ? + AND hhi.media_item_id = ? + WHERE true ON CONFLICT(profile_id, media_item_id) DO UPDATE SET position_seconds = 0, duration_seconds = excluded.duration_seconds, completed = 1, updated_at = excluded.updated_at ` - _, err := db.Exec(query, profileID, mediaItemID, duration, now) + _, err := db.Exec(query, profileID, mediaItemID, duration, now, now, profileID, mediaItemID) if err != nil { return fmt.Errorf("marking watched: %w", err) } @@ -194,9 +211,7 @@ func ClearProgress(db *sql.DB, profileID, mediaItemID string) error { } // MarkProgressBatch marks every (profile, media_item_id) pair as completed in a -// single transaction. SQLite has no UNNEST, so each row goes through the same -// MarkWatched UPSERT but inside one BEGIN/COMMIT — still much cheaper than -// per-call autocommit. +// single SQLite statement. func MarkProgressBatch(db *sql.DB, profileID string, mediaItemIDs []string, updatedAt time.Time) error { mediaItemIDs = compactText(mediaItemIDs) if len(mediaItemIDs) == 0 { @@ -205,28 +220,39 @@ func MarkProgressBatch(db *sql.DB, profileID string, mediaItemIDs []string, upda if updatedAt.IsZero() { updatedAt = time.Now().UTC() } - tx, err := db.Begin() - if err != nil { - return fmt.Errorf("begin mark progress batch: %w", err) - } - defer tx.Rollback() updatedAtText := updatedAt.UTC().Format(time.RFC3339) - for _, mediaItemID := range mediaItemIDs { - if _, err := tx.Exec(` - INSERT INTO watch_progress (profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) - VALUES (?, ?, 0, 0, 1, ?) - ON CONFLICT(profile_id, media_item_id) DO UPDATE SET - completed = 1, - position_seconds = 0, - updated_at = excluded.updated_at - WHERE watch_progress.completed != 1 - OR watch_progress.updated_at < excluded.updated_at - `, profileID, mediaItemID, updatedAtText); err != nil { - return fmt.Errorf("mark progress batch row: %w", err) - } + targetValues := make([]string, len(mediaItemIDs)) + args := make([]any, 0, len(mediaItemIDs)+4) + for i, mediaItemID := range mediaItemIDs { + targetValues[i] = "(?)" + args = append(args, mediaItemID) } - if err := tx.Commit(); err != nil { - return fmt.Errorf("commit mark progress batch: %w", err) + args = append(args, updatedAtText, updatedAtText, profileID, profileID) + if _, err := db.Exec(` + WITH target(media_item_id) AS ( + VALUES `+strings.Join(targetValues, ",")+` + ), + visible AS ( + SELECT + t.media_item_id, + `+visibleTimestampSQL+` AS updated_at + FROM target t + LEFT JOIN hidden_history_items hhi + ON hhi.profile_id = ? + AND hhi.media_item_id = t.media_item_id + ) + INSERT INTO watch_progress (profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) + SELECT ?, media_item_id, 0, 0, 1, updated_at + FROM visible + WHERE true + ON CONFLICT(profile_id, media_item_id) DO UPDATE SET + completed = 1, + position_seconds = 0, + updated_at = excluded.updated_at + WHERE watch_progress.completed != 1 + OR watch_progress.updated_at < excluded.updated_at + `, args...); err != nil { + return fmt.Errorf("marking progress batch: %w", err) } return nil } @@ -492,6 +518,35 @@ func AddHistory(db *sql.DB, entry WatchHistoryEntry) error { return nil } +func AddVisibleHistory(db *sql.DB, entry WatchHistoryEntry) (WatchHistoryEntry, error) { + if entry.ID == "" { + entry.ID = generateUUID() + } + if entry.WatchedAt == "" { + entry.WatchedAt = nowUTC() + } + if entry.Source == "" { + entry.Source = userstore.WatchHistorySourceLegacy + } + identityJSON, err := json.Marshal(entry.Identity) + if err != nil { + return entry, fmt.Errorf("marshaling watch identity: %w", err) + } + if err := db.QueryRow(` + INSERT INTO watch_history (id, profile_id, media_item_id, watched_at, duration_seconds, completed, source, watch_identity) + SELECT ?, ?, ?, `+visibleTimestampSQL+`, ?, ?, ?, ? + FROM (SELECT 1) seed + LEFT JOIN hidden_history_items hhi + ON hhi.profile_id = ? + AND hhi.media_item_id = ? + WHERE true + RETURNING watched_at + `, entry.ID, entry.ProfileID, entry.MediaItemID, entry.WatchedAt, entry.WatchedAt, entry.DurationSeconds, entry.Completed, entry.Source, string(identityJSON), entry.ProfileID, entry.MediaItemID).Scan(&entry.WatchedAt); err != nil { + return entry, fmt.Errorf("adding visible history entry: %w", err) + } + return entry, nil +} + func AddHistoryIfMissing(db *sql.DB, entry WatchHistoryEntry) (bool, error) { if entry.WatchedAt == "" { entry.WatchedAt = nowUTC() @@ -574,49 +629,15 @@ func ListCompletedHistory(db *sql.DB, query userstore.CompletedHistoryQuery) ([] if limit <= 0 || limit > 500 { limit = 500 } - args := []any{query.ProfileID} - includeSourceFilter := "" - if len(query.IncludeSources) > 0 { - placeholders := make([]string, 0, len(query.IncludeSources)) - for _, source := range query.IncludeSources { - placeholders = append(placeholders, "?") - args = append(args, string(source)) - } - includeSourceFilter = " AND h.source IN (" + strings.Join(placeholders, ",") + ")" - } - sourceFilter := "" - if len(query.ExcludeSources) > 0 { - placeholders := make([]string, 0, len(query.ExcludeSources)) - for _, source := range query.ExcludeSources { - placeholders = append(placeholders, "?") - args = append(args, string(source)) - } - sourceFilter = " AND h.source NOT IN (" + strings.Join(placeholders, ",") + ")" - } - mediaFilter := "" - if len(query.MediaItemIDs) > 0 { - placeholders := make([]string, 0, len(query.MediaItemIDs)) - for _, mediaItemID := range query.MediaItemIDs { - placeholders = append(placeholders, "?") - args = append(args, mediaItemID) - } - mediaFilter = " AND h.media_item_id IN (" + strings.Join(placeholders, ",") + ")" - } + filters, args := completedHistoryFilterSQL(query.ProfileID, query.MediaItemIDs, query.IncludeSources, query.ExcludeSources) args = append(args, limit, query.Offset) rows, err := db.Query(` SELECT h.id, h.profile_id, h.media_item_id, h.watched_at, h.duration_seconds, h.completed, h.source, h.watch_identity FROM watch_history h WHERE h.profile_id = ? AND h.completed = 1 - `+includeSourceFilter+sourceFilter+mediaFilter+` - AND NOT EXISTS ( - SELECT 1 - FROM hidden_history_items hhi - WHERE hhi.profile_id = h.profile_id - AND hhi.media_item_id = h.media_item_id - AND h.watched_at <= hhi.hidden_before - ) - ORDER BY h.watched_at ASC + `+filters+completedHistoryVisibleSQL+` + ORDER BY h.watched_at ASC, h.id ASC LIMIT ? OFFSET ? `, args...) if err != nil { @@ -646,6 +667,89 @@ func ListCompletedHistory(db *sql.DB, query userstore.CompletedHistoryQuery) ([] return results, nil } +func ListCompletedHistoryItems(db *sql.DB, query userstore.CompletedHistoryItemQuery) ([]userstore.CompletedHistoryItem, error) { + filters, args := completedHistoryFilterSQL(query.ProfileID, query.MediaItemIDs, query.IncludeSources, query.ExcludeSources) + rows, err := db.Query(` + SELECT h.media_item_id, MAX(h.watched_at) + FROM watch_history h + WHERE h.profile_id = ? + AND h.completed = 1 + `+filters+completedHistoryVisibleSQL+` + GROUP BY h.media_item_id + ORDER BY h.media_item_id ASC`, + args..., + ) + if err != nil { + return nil, fmt.Errorf("listing completed history items: %w", err) + } + defer rows.Close() + + var results []userstore.CompletedHistoryItem + for rows.Next() { + var item userstore.CompletedHistoryItem + if err := rows.Scan(&item.MediaItemID, &item.WatchedAt); err != nil { + return nil, fmt.Errorf("scanning completed history item: %w", err) + } + results = append(results, item) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating completed history items: %w", err) + } + return results, nil +} + +const completedHistoryVisibleSQL = ` + AND NOT EXISTS ( + SELECT 1 + FROM hidden_history_items hhi + WHERE hhi.profile_id = h.profile_id + AND hhi.media_item_id = h.media_item_id + AND h.watched_at <= hhi.hidden_before + )` + +const visibleTimestampSQL = ` + CASE + WHEN hhi.hidden_before IS NOT NULL AND ? <= hhi.hidden_before + THEN strftime('%Y-%m-%dT%H:%M:%SZ', hhi.hidden_before, '+1 second') + ELSE ? + END` + +func completedHistoryFilterSQL( + profileID string, + mediaItemIDs []string, + includeSources []userstore.WatchHistorySource, + excludeSources []userstore.WatchHistorySource, +) (string, []any) { + args := []any{profileID} + var filters strings.Builder + if len(includeSources) > 0 { + placeholders := make([]string, 0, len(includeSources)) + for _, source := range includeSources { + placeholders = append(placeholders, "?") + args = append(args, string(source)) + } + filters.WriteString(" AND h.source IN (" + strings.Join(placeholders, ",") + ")") + } + if len(excludeSources) > 0 { + placeholders := make([]string, 0, len(excludeSources)) + for _, source := range excludeSources { + placeholders = append(placeholders, "?") + args = append(args, string(source)) + } + filters.WriteString(" AND h.source NOT IN (" + strings.Join(placeholders, ",") + ")") + } + mediaItemIDs = compactText(mediaItemIDs) + if len(mediaItemIDs) > 0 { + placeholders := make([]string, 0, len(mediaItemIDs)) + for _, mediaItemID := range mediaItemIDs { + placeholders = append(placeholders, "?") + args = append(args, mediaItemID) + } + filters.WriteString(" AND h.media_item_id IN (" + strings.Join(placeholders, ",") + ")") + } + return filters.String(), args +} + func RemoveHistoryItems(db *sql.DB, profileID string, mediaItemIDs []string, removedAt time.Time) error { mediaItemIDs = compactText(mediaItemIDs) if len(mediaItemIDs) == 0 { @@ -662,35 +766,63 @@ func RemoveHistoryItems(db *sql.DB, profileID string, mediaItemIDs []string, rem defer tx.Rollback() removedAtText := removedAt.UTC().Format(time.RFC3339) - for _, mediaItemID := range mediaItemIDs { - if _, err := tx.Exec(` - INSERT INTO hidden_history_items (profile_id, media_item_id, hidden_before, updated_at) - VALUES (?, ?, ?, ?) - ON CONFLICT(profile_id, media_item_id) DO UPDATE SET - hidden_before = CASE - WHEN excluded.hidden_before > hidden_history_items.hidden_before - THEN excluded.hidden_before - ELSE hidden_history_items.hidden_before - END, - updated_at = excluded.updated_at - `, profileID, mediaItemID, removedAtText, removedAtText); err != nil { - return fmt.Errorf("upserting hidden history item: %w", err) - } + targetValues := make([]string, len(mediaItemIDs)) + watermarkArgs := make([]any, 0, len(mediaItemIDs)+5) + for i, mediaItemID := range mediaItemIDs { + targetValues[i] = "(?)" + watermarkArgs = append(watermarkArgs, mediaItemID) + } + watermarkArgs = append(watermarkArgs, removedAtText, removedAtText, profileID, profileID, removedAtText) + if _, err := tx.Exec(` + WITH target(media_item_id) AS ( + VALUES `+strings.Join(targetValues, ",")+` + ), + watermark AS ( + SELECT + t.media_item_id, + CASE + WHEN MAX(h.watched_at) IS NOT NULL AND MAX(h.watched_at) > ? + THEN MAX(h.watched_at) + ELSE ? + END AS hidden_before + FROM target t + LEFT JOIN watch_history h + ON h.profile_id = ? + AND h.media_item_id = t.media_item_id + GROUP BY t.media_item_id + ) + INSERT INTO hidden_history_items (profile_id, media_item_id, hidden_before, updated_at) + SELECT ?, media_item_id, hidden_before, ? + FROM watermark + WHERE true + ON CONFLICT(profile_id, media_item_id) DO UPDATE SET + hidden_before = CASE + WHEN excluded.hidden_before > hidden_history_items.hidden_before + THEN excluded.hidden_before + ELSE hidden_history_items.hidden_before + END, + updated_at = excluded.updated_at + `, watermarkArgs...); err != nil { + return fmt.Errorf("upserting hidden history items: %w", err) } placeholders := make([]string, len(mediaItemIDs)) - args := make([]any, 0, len(mediaItemIDs)+2) + args := make([]any, 0, len(mediaItemIDs)+1) args = append(args, profileID) for i, mediaItemID := range mediaItemIDs { placeholders[i] = "?" args = append(args, mediaItemID) } - args = append(args, removedAtText) if _, err := tx.Exec(` DELETE FROM watch_history WHERE profile_id = ? AND media_item_id IN (`+strings.Join(placeholders, ",")+`) - AND watched_at <= ? + AND watched_at <= ( + SELECT hhi.hidden_before + FROM hidden_history_items hhi + WHERE hhi.profile_id = watch_history.profile_id + AND hhi.media_item_id = watch_history.media_item_id + ) `, args...); err != nil { return fmt.Errorf("deleting removed history rows: %w", err) } @@ -750,6 +882,66 @@ func historyIsHidden(db *sql.DB, profileID, mediaItemID, watchedAt string) (bool return exists, nil } +func VisibleHistoryTimestamps(db *sql.DB, profileID string, mediaItemIDs []string, at time.Time) (map[string]string, error) { + mediaItemIDs = compactText(mediaItemIDs) + result := make(map[string]string, len(mediaItemIDs)) + if len(mediaItemIDs) == 0 { + return result, nil + } + if at.IsZero() { + at = time.Now().UTC() + } + targetValues := make([]string, len(mediaItemIDs)) + args := make([]any, 0, len(mediaItemIDs)+1) + for i, mediaItemID := range mediaItemIDs { + targetValues[i] = "(?)" + args = append(args, mediaItemID) + } + args = append(args, profileID) + rows, err := db.Query(` + WITH target(media_item_id) AS ( + VALUES `+strings.Join(targetValues, ",")+` + ) + SELECT t.media_item_id, hhi.hidden_before + FROM target t + LEFT JOIN hidden_history_items hhi + ON hhi.media_item_id = t.media_item_id + AND hhi.profile_id = ? + `, args...) + if err != nil { + return nil, fmt.Errorf("listing visible history timestamps: %w", err) + } + defer rows.Close() + + for rows.Next() { + var mediaItemID string + var hiddenBefore sql.NullString + if err := rows.Scan(&mediaItemID, &hiddenBefore); err != nil { + return nil, fmt.Errorf("scanning visible history timestamp: %w", err) + } + result[mediaItemID] = visibleTimestampAfterHiddenString(at, hiddenBefore) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating visible history timestamps: %w", err) + } + return result, nil +} + +func visibleTimestampAfterHiddenString(at time.Time, hiddenBefore sql.NullString) string { + timestamp := at.UTC().Format(time.RFC3339) + if !hiddenBefore.Valid { + return timestamp + } + hiddenAt, err := time.Parse(time.RFC3339, hiddenBefore.String) + if err != nil { + return timestamp + } + if at.UTC().After(hiddenAt) { + return timestamp + } + return hiddenAt.UTC().Add(time.Second).Format(time.RFC3339) +} + func compactText(values []string) []string { if len(values) == 0 { return nil diff --git a/internal/userdb/progress_test.go b/internal/userdb/progress_test.go index bfad8483..774b89cb 100644 --- a/internal/userdb/progress_test.go +++ b/internal/userdb/progress_test.go @@ -177,6 +177,132 @@ func TestListCompletedHistoryAppliesScopedFilters(t *testing.T) { } } +func TestListCompletedHistoryItemsAppliesScopedFilters(t *testing.T) { + db, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + defer db.Close() + + if err := InitSchema(db); err != nil { + t.Fatalf("InitSchema: %v", err) + } + + entries := []userstore.WatchHistoryEntry{ + { + ProfileID: "profile-1", + MediaItemID: "movie-history-only", + WatchedAt: "2026-04-25T12:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceTrakt, + }, + { + ProfileID: "profile-1", + MediaItemID: "movie-hidden", + WatchedAt: "2026-04-25T12:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceSimkl, + }, + { + ProfileID: "profile-1", + MediaItemID: "movie-future-hidden", + WatchedAt: "2026-04-25T12:10:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceTrakt, + }, + { + ProfileID: "profile-2", + MediaItemID: "movie-other-profile", + WatchedAt: "2026-04-25T12:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceTrakt, + }, + { + ProfileID: "profile-1", + MediaItemID: "movie-incomplete", + WatchedAt: "2026-04-25T12:00:00Z", + DurationSeconds: 7200, + Completed: false, + Source: userstore.WatchHistorySourceTrakt, + }, + { + ProfileID: "profile-1", + MediaItemID: "movie-playback", + WatchedAt: "2026-04-25T12:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourcePlayback, + }, + } + for _, entry := range entries { + if err := AddHistory(db, entry); err != nil { + t.Fatalf("AddHistory(%s): %v", entry.MediaItemID, err) + } + } + if err := RemoveHistoryItems(db, "profile-1", []string{"movie-hidden"}, time.Date(2026, 4, 25, 12, 5, 0, 0, time.UTC)); err != nil { + t.Fatalf("RemoveHistoryItems: %v", err) + } + if err := RemoveHistoryItems(db, "profile-1", []string{"movie-future-hidden"}, time.Date(2026, 4, 25, 12, 5, 0, 0, time.UTC)); err != nil { + t.Fatalf("RemoveHistoryItems(future): %v", err) + } + + items, err := ListCompletedHistoryItems(db, userstore.CompletedHistoryItemQuery{ + ProfileID: "profile-1", + MediaItemIDs: []string{ + "movie-history-only", + "movie-hidden", + "movie-future-hidden", + "movie-other-profile", + "movie-incomplete", + "movie-missing", + }, + }) + if err != nil { + t.Fatalf("ListCompletedHistoryItems: %v", err) + } + if len(items) != 1 || items[0].MediaItemID != "movie-history-only" || items[0].WatchedAt != "2026-04-25T12:00:00Z" { + t.Fatalf("ListCompletedHistoryItems = %v, want movie-history-only with latest watched_at", items) + } + + if err := AddHistory(db, userstore.WatchHistoryEntry{ + ProfileID: "profile-1", + MediaItemID: "movie-future-hidden", + WatchedAt: "2026-04-25T12:11:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourcePlayback, + }); err != nil { + t.Fatalf("AddHistory(future replacement): %v", err) + } + items, err = ListCompletedHistoryItems(db, userstore.CompletedHistoryItemQuery{ + ProfileID: "profile-1", + MediaItemIDs: []string{"movie-future-hidden"}, + }) + if err != nil { + t.Fatalf("ListCompletedHistoryItems(future replacement): %v", err) + } + if len(items) != 1 || items[0].MediaItemID != "movie-future-hidden" || items[0].WatchedAt != "2026-04-25T12:11:00Z" { + t.Fatalf("ListCompletedHistoryItems(future replacement) = %v, want movie-future-hidden with latest watched_at", items) + } + + items, err = ListCompletedHistoryItems(db, userstore.CompletedHistoryItemQuery{ + ProfileID: "profile-1", + MediaItemIDs: []string{"movie-history-only", "movie-playback"}, + IncludeSources: []userstore.WatchHistorySource{userstore.WatchHistorySourceTrakt, userstore.WatchHistorySourcePlayback}, + ExcludeSources: []userstore.WatchHistorySource{userstore.WatchHistorySourcePlayback}, + }) + if err != nil { + t.Fatalf("ListCompletedHistoryItems(source filters): %v", err) + } + if len(items) != 1 || items[0].MediaItemID != "movie-history-only" { + t.Fatalf("ListCompletedHistoryItems(source filters) = %v, want [movie-history-only]", items) + } +} + func TestMarkProgressBatch_CompactsDirtyInput(t *testing.T) { db, err := sql.Open("sqlite3", ":memory:") if err != nil { diff --git a/internal/userdb/sqlitestore.go b/internal/userdb/sqlitestore.go index 20009d67..c34575a8 100644 --- a/internal/userdb/sqlitestore.go +++ b/internal/userdb/sqlitestore.go @@ -103,6 +103,10 @@ func (s *SQLiteUserStore) AddHistory(_ context.Context, entry userstore.WatchHis return AddHistory(s.db, entry) } +func (s *SQLiteUserStore) AddVisibleHistory(_ context.Context, entry userstore.WatchHistoryEntry) (userstore.WatchHistoryEntry, error) { + return AddVisibleHistory(s.db, entry) +} + func (s *SQLiteUserStore) AddHistoryIfMissing(_ context.Context, entry userstore.WatchHistoryEntry) (bool, error) { return AddHistoryIfMissing(s.db, entry) } @@ -115,6 +119,14 @@ func (s *SQLiteUserStore) ListCompletedHistory(_ context.Context, query userstor return ListCompletedHistory(s.db, query) } +func (s *SQLiteUserStore) ListCompletedHistoryItems(_ context.Context, query userstore.CompletedHistoryItemQuery) ([]userstore.CompletedHistoryItem, error) { + return ListCompletedHistoryItems(s.db, query) +} + +func (s *SQLiteUserStore) VisibleHistoryTimestamps(_ context.Context, profileID string, mediaItemIDs []string, at time.Time) (map[string]string, error) { + return VisibleHistoryTimestamps(s.db, profileID, mediaItemIDs, at) +} + func (s *SQLiteUserStore) RemoveHistoryItems(_ context.Context, profileID string, mediaItemIDs []string, removedAt time.Time) error { return RemoveHistoryItems(s.db, profileID, mediaItemIDs, removedAt) } diff --git a/internal/userstore/pgstore/progress.go b/internal/userstore/pgstore/progress.go index e372836f..0e92ab9e 100644 --- a/internal/userstore/pgstore/progress.go +++ b/internal/userstore/pgstore/progress.go @@ -2,6 +2,7 @@ package pgstore import ( "context" + "database/sql" "encoding/json" "fmt" "strings" @@ -54,7 +55,7 @@ func (s *PostgresUserStore) UpdateProgress(ctx context.Context, profileID, media if duration > 0 && position > 0 && position/duration < userstore.MinResumeFraction(thresholds.MinResumePct) { return nil } - now := nowUTC() + now := time.Now().UTC() completed := false if duration > 0 && position/duration > userstore.WatchedFraction(thresholds.WatchedPct) { completed = true @@ -66,8 +67,22 @@ func (s *PostgresUserStore) UpdateProgress(ctx context.Context, profileID, media // Watching through plain GREATEST (stored position is 0) while the // watched flag survives. _, err := s.pool.Exec(ctx, ` + WITH visible AS ( + SELECT + CASE + WHEN hhi.hidden_before IS NOT NULL AND $7::timestamptz <= hhi.hidden_before + THEN hhi.hidden_before + interval '1 second' + ELSE $7::timestamptz + END AS updated_at + FROM (SELECT 1) seed + LEFT JOIN user_history_hidden_items hhi + ON hhi.user_id = $1 + AND hhi.profile_id = $2 + AND hhi.media_item_id = $3 + ) INSERT INTO user_watch_progress (user_id, profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7) + SELECT $1, $2, $3, $4, $5, $6, updated_at + FROM visible ON CONFLICT(user_id, profile_id, media_item_id) DO UPDATE SET position_seconds = CASE WHEN excluded.completed THEN 0 ELSE GREATEST(excluded.position_seconds, user_watch_progress.position_seconds) END, @@ -83,19 +98,34 @@ func (s *PostgresUserStore) UpdateProgress(ctx context.Context, profileID, media return nil } +// SetProgress bypasses the forward-only guard after the min-resume threshold. func (s *PostgresUserStore) SetProgress(ctx context.Context, profileID, mediaItemID string, position, duration float64, thresholds userstore.ProgressThresholds) error { if duration > 0 && position > 0 && position/duration < userstore.MinResumeFraction(thresholds.MinResumePct) { return nil } - now := nowUTC() + now := time.Now().UTC() completed := false if duration > 0 && position/duration > userstore.WatchedFraction(thresholds.WatchedPct) { completed = true position = 0 // match MarkWatched() — completed rows hold no resume point } _, err := s.pool.Exec(ctx, ` + WITH visible AS ( + SELECT + CASE + WHEN hhi.hidden_before IS NOT NULL AND $7::timestamptz <= hhi.hidden_before + THEN hhi.hidden_before + interval '1 second' + ELSE $7::timestamptz + END AS updated_at + FROM (SELECT 1) seed + LEFT JOIN user_history_hidden_items hhi + ON hhi.user_id = $1 + AND hhi.profile_id = $2 + AND hhi.media_item_id = $3 + ) INSERT INTO user_watch_progress (user_id, profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) - VALUES ($1, $2, $3, $4, $5, $6, $7) + SELECT $1, $2, $3, $4, $5, $6, updated_at + FROM visible ON CONFLICT(user_id, profile_id, media_item_id) DO UPDATE SET position_seconds = excluded.position_seconds, duration_seconds = excluded.duration_seconds, @@ -171,11 +201,11 @@ func (s *PostgresUserStore) SetProgressIfNewer(ctx context.Context, profileID, m INSERT INTO user_watch_progress (user_id, profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT(user_id, profile_id, media_item_id) DO UPDATE SET - position_seconds = EXCLUDED.position_seconds, - duration_seconds = EXCLUDED.duration_seconds, - completed = EXCLUDED.completed, - updated_at = EXCLUDED.updated_at - WHERE EXCLUDED.updated_at > user_watch_progress.updated_at`, + position_seconds = EXCLUDED.position_seconds, + duration_seconds = EXCLUDED.duration_seconds, + completed = user_watch_progress.completed OR EXCLUDED.completed, + updated_at = EXCLUDED.updated_at + WHERE EXCLUDED.updated_at > user_watch_progress.updated_at`, s.userID, profileID, mediaItemID, position, duration, completed, updatedAt.UTC(), ) if err != nil { @@ -189,10 +219,24 @@ func (s *PostgresUserStore) MarkWatched(ctx context.Context, profileID, mediaIte duration = 0 } - now := nowUTC() + now := time.Now().UTC() _, err := s.pool.Exec(ctx, ` + WITH visible AS ( + SELECT + CASE + WHEN hhi.hidden_before IS NOT NULL AND $5::timestamptz <= hhi.hidden_before + THEN hhi.hidden_before + interval '1 second' + ELSE $5::timestamptz + END AS updated_at + FROM (SELECT 1) seed + LEFT JOIN user_history_hidden_items hhi + ON hhi.user_id = $1 + AND hhi.profile_id = $2 + AND hhi.media_item_id = $3 + ) INSERT INTO user_watch_progress (user_id, profile_id, media_item_id, position_seconds, duration_seconds, completed, updated_at) - VALUES ($1, $2, $3, 0, $4, TRUE, $5) + SELECT $1, $2, $3, 0, $4, TRUE, updated_at + FROM visible ON CONFLICT(user_id, profile_id, media_item_id) DO UPDATE SET position_seconds = 0, duration_seconds = excluded.duration_seconds, @@ -230,10 +274,27 @@ func (s *PostgresUserStore) MarkProgressBatch(ctx context.Context, profileID str updatedAt = time.Now().UTC() } _, err := s.pool.Exec(ctx, ` + WITH target(media_item_id) AS ( + SELECT unnest($3::text[]) + ), + visible AS ( + SELECT + t.media_item_id, + CASE + WHEN hhi.hidden_before IS NOT NULL AND $4::timestamptz <= hhi.hidden_before + THEN hhi.hidden_before + interval '1 second' + ELSE $4::timestamptz + END AS updated_at + FROM target t + LEFT JOIN user_history_hidden_items hhi + ON hhi.user_id = $1 + AND hhi.profile_id = $2 + AND hhi.media_item_id = t.media_item_id + ) INSERT INTO user_watch_progress (user_id, profile_id, media_item_id, completed, position_seconds, duration_seconds, updated_at) - SELECT $1, $2, mid, TRUE, 0, 0, $4 - FROM unnest($3::text[]) AS mid + SELECT $1, $2, media_item_id, TRUE, 0, 0, updated_at + FROM visible ON CONFLICT (user_id, profile_id, media_item_id) DO UPDATE SET completed = TRUE, position_seconds = 0, @@ -488,6 +549,48 @@ func (s *PostgresUserStore) AddHistory(ctx context.Context, entry userstore.Watc return nil } +func (s *PostgresUserStore) AddVisibleHistory(ctx context.Context, entry userstore.WatchHistoryEntry) (userstore.WatchHistoryEntry, error) { + if entry.ID == "" { + entry.ID = generateUUID() + } + if entry.WatchedAt == "" { + entry.WatchedAt = nowUTC() + } + if entry.Source == "" { + entry.Source = userstore.WatchHistorySourceLegacy + } + identityJSON, err := json.Marshal(entry.Identity) + if err != nil { + return entry, fmt.Errorf("marshaling watch identity: %w", err) + } + var watchedAt time.Time + if err := s.pool.QueryRow(ctx, ` + WITH visible AS ( + SELECT + CASE + WHEN hhi.hidden_before IS NOT NULL AND $5::timestamptz <= hhi.hidden_before + THEN hhi.hidden_before + interval '1 second' + ELSE $5::timestamptz + END AS watched_at + FROM (SELECT 1) seed + LEFT JOIN user_history_hidden_items hhi + ON hhi.user_id = $2 + AND hhi.profile_id = $3 + AND hhi.media_item_id = $4 + ) + INSERT INTO user_watch_history (id, user_id, profile_id, media_item_id, watched_at, duration_seconds, completed, source, watch_identity) + SELECT $1, $2, $3, $4, watched_at, $6, $7, $8, $9 + FROM visible + RETURNING watched_at`, + entry.ID, s.userID, entry.ProfileID, entry.MediaItemID, entry.WatchedAt, + entry.DurationSeconds, entry.Completed, entry.Source, string(identityJSON), + ).Scan(&watchedAt); err != nil { + return entry, fmt.Errorf("adding visible history entry: %w", err) + } + entry.WatchedAt = timeToString(watchedAt) + return entry, nil +} + func (s *PostgresUserStore) AddHistoryIfMissing(ctx context.Context, entry userstore.WatchHistoryEntry) (bool, error) { if entry.WatchedAt == "" { entry.WatchedAt = nowUTC() @@ -560,15 +663,7 @@ func (s *PostgresUserStore) ListCompletedHistory(ctx context.Context, query user if limit <= 0 || limit > 500 { limit = 500 } - sources := make([]string, 0, len(query.ExcludeSources)) - for _, source := range query.ExcludeSources { - sources = append(sources, string(source)) - } - includeSources := make([]string, 0, len(query.IncludeSources)) - for _, source := range query.IncludeSources { - includeSources = append(includeSources, string(source)) - } - mediaItemIDs := compactMediaItemIDs(query.MediaItemIDs) + includeSources, excludeSources, mediaItemIDs := completedHistoryFilterArgs(query.MediaItemIDs, query.IncludeSources, query.ExcludeSources) rows, err := s.pool.Query(ctx, ` SELECT h.id, h.profile_id, h.media_item_id, h.watched_at, h.duration_seconds, h.completed, h.source, h.watch_identity::text FROM user_watch_history h @@ -578,17 +673,10 @@ func (s *PostgresUserStore) ListCompletedHistory(ctx context.Context, query user AND (cardinality($3::text[]) = 0 OR h.source = ANY($3::text[])) AND (cardinality($4::text[]) = 0 OR h.source <> ALL($4::text[])) AND (cardinality($5::text[]) = 0 OR h.media_item_id = ANY($5::text[])) - AND NOT EXISTS ( - SELECT 1 - FROM user_history_hidden_items hhi - WHERE hhi.user_id = h.user_id - AND hhi.profile_id = h.profile_id - AND hhi.media_item_id = h.media_item_id - AND h.watched_at <= hhi.hidden_before - ) - ORDER BY h.watched_at ASC + `+completedHistoryVisibleSQL+` + ORDER BY h.watched_at ASC, h.id ASC LIMIT $6 OFFSET $7`, - s.userID, query.ProfileID, includeSources, sources, mediaItemIDs, limit, query.Offset, + s.userID, query.ProfileID, includeSources, excludeSources, mediaItemIDs, limit, query.Offset, ) if err != nil { return nil, fmt.Errorf("listing completed history: %w", err) @@ -609,6 +697,106 @@ func (s *PostgresUserStore) ListCompletedHistory(ctx context.Context, query user return results, nil } +func (s *PostgresUserStore) ListCompletedHistoryItems(ctx context.Context, query userstore.CompletedHistoryItemQuery) ([]userstore.CompletedHistoryItem, error) { + includeSources, excludeSources, mediaItemIDs := completedHistoryFilterArgs(query.MediaItemIDs, query.IncludeSources, query.ExcludeSources) + rows, err := s.pool.Query(ctx, ` + SELECT h.media_item_id, MAX(h.watched_at) + FROM user_watch_history h + WHERE h.user_id = $1 + AND h.profile_id = $2 + AND h.completed = true + AND (cardinality($3::text[]) = 0 OR h.source = ANY($3::text[])) + AND (cardinality($4::text[]) = 0 OR h.source <> ALL($4::text[])) + AND (cardinality($5::text[]) = 0 OR h.media_item_id = ANY($5::text[])) + `+completedHistoryVisibleSQL+` + GROUP BY h.media_item_id + ORDER BY h.media_item_id ASC`, + s.userID, query.ProfileID, includeSources, excludeSources, mediaItemIDs, + ) + if err != nil { + return nil, fmt.Errorf("listing completed history items: %w", err) + } + defer rows.Close() + + var results []userstore.CompletedHistoryItem + for rows.Next() { + var item userstore.CompletedHistoryItem + var watchedAt time.Time + if err := rows.Scan(&item.MediaItemID, &watchedAt); err != nil { + return nil, fmt.Errorf("scanning completed history item: %w", err) + } + item.WatchedAt = timeToString(watchedAt) + results = append(results, item) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating completed history items: %w", err) + } + return results, nil +} + +func (s *PostgresUserStore) VisibleHistoryTimestamps(ctx context.Context, profileID string, mediaItemIDs []string, at time.Time) (map[string]string, error) { + mediaItemIDs = compactMediaItemIDs(mediaItemIDs) + result := make(map[string]string, len(mediaItemIDs)) + if len(mediaItemIDs) == 0 { + return result, nil + } + if at.IsZero() { + at = time.Now().UTC() + } + rows, err := s.pool.Query(ctx, ` + SELECT t.media_item_id, hhi.hidden_before + FROM unnest($3::text[]) AS t(media_item_id) + LEFT JOIN user_history_hidden_items hhi + ON hhi.user_id = $1 + AND hhi.profile_id = $2 + AND hhi.media_item_id = t.media_item_id`, + s.userID, profileID, mediaItemIDs, + ) + if err != nil { + return nil, fmt.Errorf("listing visible history timestamps: %w", err) + } + defer rows.Close() + + for rows.Next() { + var mediaItemID string + var hiddenBefore sql.NullTime + if err := rows.Scan(&mediaItemID, &hiddenBefore); err != nil { + return nil, fmt.Errorf("scanning visible history timestamp: %w", err) + } + result[mediaItemID] = visibleTimestampAfterHiddenTime(at, hiddenBefore) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterating visible history timestamps: %w", err) + } + return result, nil +} + +const completedHistoryVisibleSQL = ` + AND NOT EXISTS ( + SELECT 1 + FROM user_history_hidden_items hhi + WHERE hhi.user_id = h.user_id + AND hhi.profile_id = h.profile_id + AND hhi.media_item_id = h.media_item_id + AND h.watched_at <= hhi.hidden_before + )` + +func completedHistoryFilterArgs( + mediaItemIDs []string, + includeSources []userstore.WatchHistorySource, + excludeSources []userstore.WatchHistorySource, +) ([]string, []string, []string) { + include := make([]string, 0, len(includeSources)) + for _, source := range includeSources { + include = append(include, string(source)) + } + exclude := make([]string, 0, len(excludeSources)) + for _, source := range excludeSources { + exclude = append(exclude, string(source)) + } + return include, exclude, compactMediaItemIDs(mediaItemIDs) +} + func (s *PostgresUserStore) RemoveHistoryItems( ctx context.Context, profileID string, @@ -630,8 +818,23 @@ func (s *PostgresUserStore) RemoveHistoryItems( defer tx.Rollback(ctx) if _, err := tx.Exec(ctx, ` + WITH target(media_item_id) AS ( + SELECT unnest($3::text[]) + ), + watermark AS ( + SELECT + t.media_item_id, + GREATEST($4::timestamptz, COALESCE(MAX(h.watched_at), $4::timestamptz)) AS hidden_before + FROM target t + LEFT JOIN user_watch_history h + ON h.user_id = $1 + AND h.profile_id = $2 + AND h.media_item_id = t.media_item_id + GROUP BY t.media_item_id + ) INSERT INTO user_history_hidden_items (user_id, profile_id, media_item_id, hidden_before, updated_at) - SELECT $1, $2, unnest($3::text[]), $4, $4 + SELECT $1, $2, media_item_id, hidden_before, $4 + FROM watermark ON CONFLICT (user_id, profile_id, media_item_id) DO UPDATE SET hidden_before = GREATEST(user_history_hidden_items.hidden_before, EXCLUDED.hidden_before), updated_at = EXCLUDED.updated_at @@ -640,12 +843,16 @@ func (s *PostgresUserStore) RemoveHistoryItems( } if _, err := tx.Exec(ctx, ` - DELETE FROM user_watch_history - WHERE user_id = $1 - AND profile_id = $2 - AND media_item_id = ANY($3::text[]) - AND watched_at <= $4 - `, s.userID, profileID, mediaItemIDs, removedAt.UTC()); err != nil { + DELETE FROM user_watch_history h + USING user_history_hidden_items hhi + WHERE h.user_id = $1 + AND h.profile_id = $2 + AND h.media_item_id = ANY($3::text[]) + AND hhi.user_id = h.user_id + AND hhi.profile_id = h.profile_id + AND hhi.media_item_id = h.media_item_id + AND h.watched_at <= hhi.hidden_before + `, s.userID, profileID, mediaItemIDs); err != nil { return fmt.Errorf("deleting removed history rows: %w", err) } @@ -699,6 +906,17 @@ func (s *PostgresUserStore) historyIsHidden( return exists, nil } +func visibleTimestampAfterHiddenTime(at time.Time, hiddenBefore sql.NullTime) string { + if at.IsZero() { + at = time.Now().UTC() + } + at = at.UTC() + if !hiddenBefore.Valid || at.After(hiddenBefore.Time) { + return timeToString(at) + } + return timeToString(hiddenBefore.Time.UTC().Add(time.Second)) +} + func compactMediaItemIDs(mediaItemIDs []string) []string { result := make([]string, 0, len(mediaItemIDs)) seen := make(map[string]struct{}, len(mediaItemIDs)) diff --git a/internal/userstore/progress_helpers.go b/internal/userstore/progress_helpers.go new file mode 100644 index 00000000..bc5fd015 --- /dev/null +++ b/internal/userstore/progress_helpers.go @@ -0,0 +1,204 @@ +package userstore + +import ( + "context" + "strings" + "time" +) + +type HistoryVisibilityStore interface { + VisibleHistoryTimestamps(ctx context.Context, profileID string, mediaItemIDs []string, at time.Time) (map[string]string, error) +} + +type VisibleHistoryAdder interface { + AddVisibleHistory(ctx context.Context, entry WatchHistoryEntry) (WatchHistoryEntry, error) +} + +func AddVisibleHistory(ctx context.Context, store UserStore, entry WatchHistoryEntry) (WatchHistoryEntry, error) { + if adder, ok := store.(VisibleHistoryAdder); ok { + return adder.AddVisibleHistory(ctx, entry) + } + entryTimes, err := VisibleHistoryTimestamps(ctx, store, entry.ProfileID, []string{entry.MediaItemID}, parseHistoryTimestamp(entry.WatchedAt)) + if err != nil { + return entry, err + } + if entryTime := entryTimes[entry.MediaItemID]; entryTime != "" { + entry.WatchedAt = entryTime + } + if err := store.AddHistory(ctx, entry); err != nil { + return entry, err + } + return entry, nil +} + +func VisibleHistoryTimestamps(ctx context.Context, store UserStore, profileID string, mediaItemIDs []string, at time.Time) (map[string]string, error) { + mediaItemIDs = compactHistoryMediaItemIDs(mediaItemIDs) + result := make(map[string]string, len(mediaItemIDs)) + if len(mediaItemIDs) == 0 { + return result, nil + } + if visibilityStore, ok := store.(HistoryVisibilityStore); ok { + return visibilityStore.VisibleHistoryTimestamps(ctx, profileID, mediaItemIDs, at) + } + timestamp := at.UTC().Format(time.RFC3339) + if at.IsZero() { + timestamp = time.Now().UTC().Format(time.RFC3339) + } + for _, mediaItemID := range mediaItemIDs { + result[mediaItemID] = timestamp + } + return result, nil +} + +func parseHistoryTimestamp(value string) time.Time { + if value == "" { + return time.Time{} + } + parsed, err := time.Parse(time.RFC3339, value) + if err != nil { + return time.Time{} + } + return parsed +} + +// CompletedHistoryItemMap returns the latest completed-history item row for a +// scoped item query. Lookup failures degrade to an empty map so user-data +// enrichment can keep returning progress rows. +func CompletedHistoryItemMap(ctx context.Context, store UserStore, query CompletedHistoryItemQuery) map[string]CompletedHistoryItem { + result := map[string]CompletedHistoryItem{} + if store == nil || query.ProfileID == "" { + return result + } + query.MediaItemIDs = compactHistoryMediaItemIDs(query.MediaItemIDs) + if len(query.MediaItemIDs) == 0 { + return result + } + items, err := store.ListCompletedHistoryItems(ctx, query) + if err != nil { + return result + } + for _, item := range items { + if item.MediaItemID != "" { + result[item.MediaItemID] = item + } + } + return result +} + +// GetProgressWithCompletedHistory returns normal progress overlaid with +// completed history for callers that present a single item's played state. +func GetProgressWithCompletedHistory(ctx context.Context, store UserStore, profileID, mediaItemID string) (*WatchProgress, error) { + mediaItemID = strings.TrimSpace(mediaItemID) + if store == nil || profileID == "" || mediaItemID == "" { + return nil, nil + } + progress, err := store.GetProgress(ctx, profileID, mediaItemID) + if err != nil { + return nil, err + } + if progress != nil && progress.Completed { + return progress, nil + } + completed := CompletedHistoryItemMap(ctx, store, CompletedHistoryItemQuery{ + ProfileID: profileID, + MediaItemIDs: []string{mediaItemID}, + })[mediaItemID] + if completed.MediaItemID == "" { + return progress, nil + } + if progress == nil { + return &WatchProgress{ + ProfileID: profileID, + MediaItemID: mediaItemID, + Completed: true, + UpdatedAt: completed.WatchedAt, + }, nil + } + progress.Completed = true + if timestampAfter(completed.WatchedAt, progress.UpdatedAt) { + progress.UpdatedAt = completed.WatchedAt + } + return progress, nil +} + +// ListProgressWithCompletedHistory returns progress for mediaItemIDs with +// completed history folded into the map. History is only queried for IDs that +// are not already completed by a progress row. +func ListProgressWithCompletedHistory(ctx context.Context, store UserStore, profileID string, mediaItemIDs []string) (map[string]WatchProgress, error) { + mediaItemIDs = compactHistoryMediaItemIDs(mediaItemIDs) + if store == nil || profileID == "" || len(mediaItemIDs) == 0 { + return map[string]WatchProgress{}, nil + } + progressMap, err := store.ListProgressByMediaItems(ctx, profileID, mediaItemIDs) + if err != nil { + return nil, err + } + if progressMap == nil { + progressMap = map[string]WatchProgress{} + } + + candidates := make([]string, 0, len(mediaItemIDs)) + for _, mediaItemID := range mediaItemIDs { + if progress, ok := progressMap[mediaItemID]; ok && progress.Completed { + continue + } + candidates = append(candidates, mediaItemID) + } + if len(candidates) == 0 { + return progressMap, nil + } + + completed := CompletedHistoryItemMap(ctx, store, CompletedHistoryItemQuery{ + ProfileID: profileID, + MediaItemIDs: candidates, + }) + for mediaItemID, completedItem := range completed { + if progress, ok := progressMap[mediaItemID]; ok { + progress.Completed = true + if timestampAfter(completedItem.WatchedAt, progress.UpdatedAt) { + progress.UpdatedAt = completedItem.WatchedAt + } + progressMap[mediaItemID] = progress + continue + } + progressMap[mediaItemID] = WatchProgress{ + ProfileID: profileID, + MediaItemID: mediaItemID, + Completed: true, + UpdatedAt: completedItem.WatchedAt, + } + } + return progressMap, nil +} + +func compactHistoryMediaItemIDs(mediaItemIDs []string) []string { + result := make([]string, 0, len(mediaItemIDs)) + seen := make(map[string]struct{}, len(mediaItemIDs)) + for _, mediaItemID := range mediaItemIDs { + mediaItemID = strings.TrimSpace(mediaItemID) + if mediaItemID == "" { + continue + } + if _, ok := seen[mediaItemID]; ok { + continue + } + seen[mediaItemID] = struct{}{} + result = append(result, mediaItemID) + } + return result +} + +func timestampAfter(left, right string) bool { + if left == "" { + return false + } + if right == "" { + return true + } + leftTime, leftErr := time.Parse(time.RFC3339, left) + rightTime, rightErr := time.Parse(time.RFC3339, right) + if leftErr == nil && rightErr == nil { + return leftTime.After(rightTime) + } + return left > right +} diff --git a/internal/userstore/progress_helpers_test.go b/internal/userstore/progress_helpers_test.go new file mode 100644 index 00000000..d590b20f --- /dev/null +++ b/internal/userstore/progress_helpers_test.go @@ -0,0 +1,46 @@ +package userstore_test + +import ( + "context" + "database/sql" + "testing" + + "github.com/Silo-Server/silo-server/internal/userdb" + "github.com/Silo-Server/silo-server/internal/userstore" +) + +func TestGetProgressWithCompletedHistoryCarriesHistoryTimestamp(t *testing.T) { + db, err := sql.Open("sqlite3", ":memory:") + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + defer db.Close() + if err := userdb.InitSchema(db); err != nil { + t.Fatalf("InitSchema: %v", err) + } + store := userdb.NewSQLiteUserStore(db) + if err := store.CreateProfile(context.Background(), userstore.Profile{ID: "profile-1", Name: "Profile"}); err != nil { + t.Fatalf("CreateProfile: %v", err) + } + if err := store.AddHistory(context.Background(), userstore.WatchHistoryEntry{ + ProfileID: "profile-1", + MediaItemID: "movie-history-only", + WatchedAt: "2026-05-04T12:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceTrakt, + }); err != nil { + t.Fatalf("AddHistory: %v", err) + } + + progress, err := userstore.GetProgressWithCompletedHistory(context.Background(), store, "profile-1", "movie-history-only") + if err != nil { + t.Fatalf("GetProgressWithCompletedHistory: %v", err) + } + if progress == nil || !progress.Completed { + t.Fatalf("progress = %+v, want synthetic completed progress", progress) + } + if progress.UpdatedAt != "2026-05-04T12:00:00Z" { + t.Fatalf("UpdatedAt = %q, want history watched_at", progress.UpdatedAt) + } +} diff --git a/internal/userstore/store.go b/internal/userstore/store.go index 28da107d..b176f3c1 100644 --- a/internal/userstore/store.go +++ b/internal/userstore/store.go @@ -36,6 +36,7 @@ type UserStore interface { AddHistoryIfMissing(ctx context.Context, entry WatchHistoryEntry) (bool, error) ListHistory(ctx context.Context, profileID string, limit, offset int) ([]WatchHistoryEntry, error) ListCompletedHistory(ctx context.Context, query CompletedHistoryQuery) ([]WatchHistoryEntry, error) + ListCompletedHistoryItems(ctx context.Context, query CompletedHistoryItemQuery) ([]CompletedHistoryItem, error) RemoveHistoryItems(ctx context.Context, profileID string, mediaItemIDs []string, removedAt time.Time) error DeleteHistoryBySource(ctx context.Context, profileID string, mediaItemIDs []string, source WatchHistorySource) error ListHomeDismissals(ctx context.Context, profileID, surface string) ([]HomeItemDismissal, error) diff --git a/internal/userstore/storetest/suite.go b/internal/userstore/storetest/suite.go index e1bc0363..d9964e48 100644 --- a/internal/userstore/storetest/suite.go +++ b/internal/userstore/storetest/suite.go @@ -475,6 +475,120 @@ func testProgress(t *testing.T, newStore func(t *testing.T) userstore.UserStore) t.Fatalf("GetProgress(after new watch) = %+v, want completed progress", wp) } + if err := store.CreateProfile(ctx, userstore.Profile{ID: "p2", Name: "Other"}); err != nil { + t.Fatalf("CreateProfile(p2): %v", err) + } + for _, entry := range []userstore.WatchHistoryEntry{ + { + ProfileID: "p1", + MediaItemID: "movie-history-only", + DurationSeconds: 7200, + Completed: true, + WatchedAt: "2026-03-23T12:05:00Z", + Source: userstore.WatchHistorySourceTrakt, + }, + { + ProfileID: "p1", + MediaItemID: "movie-hidden-history", + DurationSeconds: 7200, + Completed: true, + WatchedAt: "2026-03-23T12:05:00Z", + Source: userstore.WatchHistorySourceSimkl, + }, + { + ProfileID: "p1", + MediaItemID: "movie-future-hidden", + DurationSeconds: 7200, + Completed: true, + WatchedAt: "2026-03-23T12:10:00Z", + Source: userstore.WatchHistorySourceTrakt, + }, + { + ProfileID: "p2", + MediaItemID: "movie-other-profile", + DurationSeconds: 7200, + Completed: true, + WatchedAt: "2026-03-23T12:05:00Z", + Source: userstore.WatchHistorySourceTrakt, + }, + { + ProfileID: "p1", + MediaItemID: "movie-incomplete-history", + DurationSeconds: 7200, + Completed: false, + WatchedAt: "2026-03-23T12:05:00Z", + Source: userstore.WatchHistorySourceTrakt, + }, + } { + if err := store.AddHistory(ctx, entry); err != nil { + t.Fatalf("AddHistory(%s): %v", entry.MediaItemID, err) + } + } + if err := store.RemoveHistoryItems(ctx, "p1", []string{"movie-hidden-history"}, time.Date(2026, 3, 23, 12, 6, 0, 0, time.UTC)); err != nil { + t.Fatalf("RemoveHistoryItems(movie-hidden-history): %v", err) + } + if err := store.RemoveHistoryItems(ctx, "p1", []string{"movie-future-hidden"}, time.Date(2026, 3, 23, 12, 6, 0, 0, time.UTC)); err != nil { + t.Fatalf("RemoveHistoryItems(movie-future-hidden): %v", err) + } + completedItems, err := store.ListCompletedHistoryItems(ctx, userstore.CompletedHistoryItemQuery{ + ProfileID: "p1", + MediaItemIDs: []string{ + "movie-1", + "movie-history-only", + "movie-hidden-history", + "movie-future-hidden", + "movie-other-profile", + "movie-incomplete-history", + }, + }) + if err != nil { + t.Fatalf("ListCompletedHistoryItems: %v", err) + } + completedSet := map[string]userstore.CompletedHistoryItem{} + for _, item := range completedItems { + completedSet[item.MediaItemID] = item + } + if completedSet["movie-1"].MediaItemID == "" || completedSet["movie-history-only"].MediaItemID == "" { + t.Fatalf("ListCompletedHistoryItems = %v, want movie-1 and movie-history-only", completedItems) + } + for _, id := range []string{"movie-hidden-history", "movie-future-hidden", "movie-other-profile", "movie-incomplete-history"} { + if completedSet[id].MediaItemID != "" { + t.Fatalf("ListCompletedHistoryItems included %s: %v", id, completedItems) + } + } + if err := store.AddHistory(ctx, userstore.WatchHistoryEntry{ + ProfileID: "p1", + MediaItemID: "movie-future-hidden", + DurationSeconds: 7200, + Completed: true, + WatchedAt: "2026-03-23T12:11:00Z", + Source: userstore.WatchHistorySourcePlayback, + }); err != nil { + t.Fatalf("AddHistory(movie-future-hidden newer): %v", err) + } + completedItems, err = store.ListCompletedHistoryItems(ctx, userstore.CompletedHistoryItemQuery{ + ProfileID: "p1", + MediaItemIDs: []string{"movie-future-hidden"}, + }) + if err != nil { + t.Fatalf("ListCompletedHistoryItems(movie-future-hidden newer): %v", err) + } + if len(completedItems) != 1 || completedItems[0].MediaItemID != "movie-future-hidden" || completedItems[0].WatchedAt != "2026-03-23T12:11:00Z" { + t.Fatalf("ListCompletedHistoryItems(movie-future-hidden newer) = %v, want movie-future-hidden with latest watched_at", completedItems) + } + traktItems, err := store.ListCompletedHistoryItems(ctx, userstore.CompletedHistoryItemQuery{ + ProfileID: "p1", + MediaItemIDs: []string{"movie-1", "movie-history-only"}, + IncludeSources: []userstore.WatchHistorySource{userstore.WatchHistorySourcePlayback, userstore.WatchHistorySourceTrakt}, + ExcludeSources: []userstore.WatchHistorySource{userstore.WatchHistorySourcePlayback}, + }) + if err != nil { + t.Fatalf("ListCompletedHistoryItems(source filters): %v", err) + } + if len(traktItems) != 1 || traktItems[0].MediaItemID != "movie-history-only" { + t.Fatalf("ListCompletedHistoryItems(source filters) = %v, want [movie-history-only]", traktItems) + } + // Manual watched state helpers. if err := store.MarkWatched(ctx, "p1", "movie-3", 5400); err != nil { t.Fatalf("MarkWatched: %v", err) diff --git a/internal/userstore/types.go b/internal/userstore/types.go index e8d34ac1..2dc74bd4 100644 --- a/internal/userstore/types.go +++ b/internal/userstore/types.go @@ -155,6 +155,18 @@ type CompletedHistoryQuery struct { Offset int } +type CompletedHistoryItemQuery struct { + ProfileID string + MediaItemIDs []string + IncludeSources []WatchHistorySource + ExcludeSources []WatchHistorySource +} + +type CompletedHistoryItem struct { + MediaItemID string + WatchedAt string +} + // Favorite represents a favorited media item. type Favorite struct { ProfileID string diff --git a/internal/watchstate/mark_played_batch_test.go b/internal/watchstate/mark_played_batch_test.go index fb9beac8..2facc520 100644 --- a/internal/watchstate/mark_played_batch_test.go +++ b/internal/watchstate/mark_played_batch_test.go @@ -20,16 +20,3 @@ func TestMarkPlayedBatch_SingleUpsert(t *testing.T) { t.Fatalf("expected completed flag set; got:\n%s", sql) } } - -func TestMarkUnplayedBatch_BatchedUpdate(t *testing.T) { - sql, _ := buildMarkUnplayedBatchSQL() - if !strings.Contains(sql, "UPDATE user_watch_progress") { - t.Fatalf("expected UPDATE user_watch_progress; got:\n%s", sql) - } - if !strings.Contains(sql, "completed = FALSE") { - t.Fatalf("expected completed = FALSE; got:\n%s", sql) - } - if !strings.Contains(sql, "media_item_id = ANY($3::text[])") { - t.Fatalf("expected ANY(text[]) batch filter; got:\n%s", sql) - } -} diff --git a/internal/watchstate/service.go b/internal/watchstate/service.go index 8d662481..d415e979 100644 --- a/internal/watchstate/service.go +++ b/internal/watchstate/service.go @@ -60,7 +60,7 @@ func (s *Service) RecordManualMarkUnwatched(ctx context.Context, userID int, pro } func (s *Service) RecordManualMarkUnwatchedWithResult(ctx context.Context, userID int, profileID string, targetIDs []string) (ManualMarkResult, error) { - return s.recordMarkUnwatched(ctx, userID, profileID, targetIDs, userstore.WatchHistorySourceManual) + return s.recordMarkUnwatched(ctx, userID, profileID, targetIDs) } func (s *Service) RecordPlaybackStop( @@ -86,6 +86,9 @@ func (s *Service) RecordPlaybackStop( if err != nil { return result, err } + if watchedAt.IsZero() { + watchedAt = time.Now().UTC() + } if err := store.SetProgress(ctx, profileID, targetID, position, duration, thresholds); err != nil { return result, err } @@ -105,7 +108,8 @@ func (s *Service) RecordPlaybackStop( Source: userstore.WatchHistorySourcePlayback, } s.applyStableIdentity(ctx, &entry) - if err := store.AddHistory(ctx, entry); err != nil { + entry, err = userstore.AddVisibleHistory(ctx, store, entry) + if err != nil { return result, err } result.Completed = entry.Completed @@ -145,6 +149,26 @@ func (s *Service) RecordImportedWatchWithSource( return s.addImportedHistoryIfMissingWithSource(ctx, store, profileID, targetID, duration, completed, watchedAt, source) } +func (s *Service) RecordImportedWatchIfNewerWithSource( + ctx context.Context, + userID int, + profileID, targetID string, + duration, position float64, + completed bool, + updatedAt time.Time, + watchedAt *time.Time, + source userstore.WatchHistorySource, +) (bool, error) { + store, err := s.storeForUser(ctx, userID) + if err != nil { + return false, err + } + if _, err := store.SetProgressIfNewer(ctx, profileID, targetID, position, duration, completed, updatedAt); err != nil { + return false, err + } + return s.addImportedHistoryIfMissingWithSource(ctx, store, profileID, targetID, duration, completed, watchedAt, source) +} + func (s *Service) RecordImportedHistory( ctx context.Context, userID int, @@ -226,7 +250,7 @@ func (s *Service) RecordJellycompatMarkPlayed(ctx context.Context, userID int, p } func (s *Service) RecordJellycompatMarkUnplayed(ctx context.Context, userID int, profileID, targetID string) error { - _, err := s.recordMarkUnwatched(ctx, userID, profileID, []string{targetID}, userstore.WatchHistorySourceJellycompat) + _, err := s.recordMarkUnwatched(ctx, userID, profileID, []string{targetID}) return err } @@ -238,11 +262,10 @@ func (s *Service) RecordJellycompatMarkPlayedBatch(ctx context.Context, userID i return s.recordMarkWatchedBatch(ctx, userID, profileID, targetIDs, watchedAt, userstore.WatchHistorySourceJellycompat) } -// RecordJellycompatMarkUnplayedBatch clears progress and deletes -// jellycompat-sourced history entries for all targets in a single statement -// each (audit 2026-05-01 §2.7). +// RecordJellycompatMarkUnplayedBatch hides prior visible history and clears +// progress for all targets in a single store operation. func (s *Service) RecordJellycompatMarkUnplayedBatch(ctx context.Context, userID int, profileID string, targetIDs []string) error { - return s.recordMarkUnwatchedBatch(ctx, userID, profileID, targetIDs, userstore.WatchHistorySourceJellycompat) + return s.recordMarkUnwatchedBatch(ctx, userID, profileID, targetIDs) } func (s *Service) storeForUser(ctx context.Context, userID int) (userstore.UserStore, error) { @@ -271,7 +294,9 @@ func (s *Service) recordMarkWatched( if err != nil { return ManualMarkResult{}, err } - entryTime := formatWatchedAt(watchedAt) + if watchedAt.IsZero() { + watchedAt = time.Now().UTC() + } result := ManualMarkResult{Entries: make([]userstore.WatchHistoryEntry, 0, len(targets))} for _, target := range targets { if err := store.MarkWatched(ctx, profileID, target.MediaItemID, target.DurationSeconds); err != nil { @@ -281,13 +306,14 @@ func (s *Service) recordMarkWatched( ID: uuid.NewString(), ProfileID: profileID, MediaItemID: target.MediaItemID, - WatchedAt: entryTime, + WatchedAt: formatWatchedAt(watchedAt), DurationSeconds: target.DurationSeconds, Completed: true, Source: source, } s.applyStableIdentity(ctx, &histEntry) - if err := store.AddHistory(ctx, histEntry); err != nil { + histEntry, err = userstore.AddVisibleHistory(ctx, store, histEntry) + if err != nil { return result, err } result.Entries = append(result.Entries, histEntry) @@ -300,22 +326,16 @@ func (s *Service) recordMarkUnwatched( userID int, profileID string, targetIDs []string, - source userstore.WatchHistorySource, ) (ManualMarkResult, error) { store, err := s.storeForUser(ctx, userID) if err != nil { return ManualMarkResult{}, err } - result, err := s.completedHistoryForTargets(ctx, store, profileID, targetIDs, source) + result, err := s.completedHistoryForTargets(ctx, store, profileID, targetIDs, []userstore.WatchHistorySource{userstore.WatchHistorySourceManual}) if err != nil { return ManualMarkResult{}, err } - for _, targetID := range targetIDs { - if err := store.ClearProgress(ctx, profileID, targetID); err != nil { - return result, err - } - } - return result, store.DeleteHistoryBySource(ctx, profileID, targetIDs, source) + return result, store.RemoveHistoryItems(ctx, profileID, targetIDs, time.Now().UTC()) } func (s *Service) completedHistoryForTargets( @@ -323,23 +343,55 @@ func (s *Service) completedHistoryForTargets( store userstore.UserStore, profileID string, targetIDs []string, - source userstore.WatchHistorySource, + includeSources []userstore.WatchHistorySource, ) (ManualMarkResult, error) { if len(targetIDs) == 0 { return ManualMarkResult{}, nil } - entries, err := store.ListCompletedHistory(ctx, userstore.CompletedHistoryQuery{ - ProfileID: profileID, - MediaItemIDs: targetIDs, - IncludeSources: []userstore.WatchHistorySource{ - source, - }, - Limit: len(targetIDs) * 20, - }) - if err != nil { - return ManualMarkResult{}, err + const pageSize = 500 + var entries []userstore.WatchHistoryEntry + for offset := 0; ; offset += pageSize { + page, err := store.ListCompletedHistory(ctx, userstore.CompletedHistoryQuery{ + ProfileID: profileID, + MediaItemIDs: targetIDs, + IncludeSources: includeSources, + Limit: pageSize, + Offset: offset, + }) + if err != nil { + return ManualMarkResult{}, err + } + entries = append(entries, page...) + if len(page) < pageSize { + break + } } - return ManualMarkResult{Entries: entries}, nil + return ManualMarkResult{Entries: representativeHistoryEntries(targetIDs, entries)}, nil +} + +func representativeHistoryEntries(targetIDs []string, entries []userstore.WatchHistoryEntry) []userstore.WatchHistoryEntry { + if len(targetIDs) == 0 || len(entries) == 0 { + return nil + } + latestByTarget := make(map[string]userstore.WatchHistoryEntry, len(targetIDs)) + for _, entry := range entries { + current, ok := latestByTarget[entry.MediaItemID] + if !ok || entry.WatchedAt > current.WatchedAt || (entry.WatchedAt == current.WatchedAt && entry.ID > current.ID) { + latestByTarget[entry.MediaItemID] = entry + } + } + result := make([]userstore.WatchHistoryEntry, 0, len(latestByTarget)) + seen := make(map[string]struct{}, len(targetIDs)) + for _, targetID := range targetIDs { + if _, ok := seen[targetID]; ok { + continue + } + seen[targetID] = struct{}{} + if entry, ok := latestByTarget[targetID]; ok { + result = append(result, entry) + } + } + return result } func (s *Service) recordMarkWatchedBatch( @@ -366,17 +418,16 @@ func (s *Service) recordMarkWatchedBatch( // Strategy A (audit 2026-05-01 §2.7): batch the progress upsert because it // powers hot Continue-Watching queries. History inserts stay per-target so // per-episode stable-identity resolution still applies. - entryTime := formatWatchedAt(watchedAt) for _, targetID := range targetIDs { histEntry := userstore.WatchHistoryEntry{ ProfileID: profileID, MediaItemID: targetID, - WatchedAt: entryTime, + WatchedAt: formatWatchedAt(watchedAt), Completed: true, Source: source, } s.applyStableIdentity(ctx, &histEntry) - if err := store.AddHistory(ctx, histEntry); err != nil { + if _, err := userstore.AddVisibleHistory(ctx, store, histEntry); err != nil { return err } } @@ -388,7 +439,6 @@ func (s *Service) recordMarkUnwatchedBatch( userID int, profileID string, targetIDs []string, - source userstore.WatchHistorySource, ) error { if len(targetIDs) == 0 { return nil @@ -397,10 +447,7 @@ func (s *Service) recordMarkUnwatchedBatch( if err != nil { return err } - if err := store.ClearProgressBatch(ctx, profileID, targetIDs, time.Now().UTC()); err != nil { - return err - } - return store.DeleteHistoryBySource(ctx, profileID, targetIDs, source) + return store.RemoveHistoryItems(ctx, profileID, targetIDs, time.Now().UTC()) } // buildMarkPlayedBatchSQL returns the upsert that marks every media_item_id in @@ -420,25 +467,6 @@ func buildMarkPlayedBatchSQL() (string, []any) { OR user_watch_progress.updated_at < EXCLUDED.updated_at`, nil } -// buildMarkUnplayedBatchSQL returns the update that clears the completed flag -// and resets position to 0 for every media_item_id in $3 for a given -// (user, profile). Pairs with the jellycompat unplayed-batch path; the matching -// history-row deletion uses DeleteHistoryBySource which already takes a slice. -// -// The `completed = TRUE OR position_seconds <> 0` predicate clears partially- -// watched rows in addition to fully-completed ones — the prior single-item -// ClearProgress path DELETE-d unconditionally, so any non-default state must -// be cleared (otherwise "mark unplayed" leaves resume position untouched). -// Skip rows already in the target state to avoid pointless writes. -func buildMarkUnplayedBatchSQL() (string, []any) { - return ` - UPDATE user_watch_progress - SET completed = FALSE, position_seconds = 0, updated_at = $4 - WHERE user_id = $1 AND profile_id = $2 - AND media_item_id = ANY($3::text[]) - AND (completed = TRUE OR position_seconds <> 0)`, nil -} - func (s *Service) addImportedHistoryIfMissing( ctx context.Context, store userstore.UserStore, diff --git a/internal/watchstate/service_test.go b/internal/watchstate/service_test.go index f81095c6..d3a7a9c5 100644 --- a/internal/watchstate/service_test.go +++ b/internal/watchstate/service_test.go @@ -188,6 +188,35 @@ func TestManualMarkWatchedAddsEpisodeIdentity(t *testing.T) { } } +func TestManualMarkWatchedPreservesVisibleWatchedAt(t *testing.T) { + store, db := newTestUserStore(t) + defer db.Close() + + watchedAt := time.Date(2026, 4, 25, 12, 0, 0, 0, time.UTC) + service := NewService(testStoreProvider{store: store}) + err := service.RecordManualMarkWatched( + context.Background(), + 1, + "profile-1", + []LeafWatchTarget{{MediaItemID: "movie-1", DurationSeconds: 7200}}, + watchedAt, + ) + if err != nil { + t.Fatalf("RecordManualMarkWatched: %v", err) + } + + history, err := store.ListHistory(context.Background(), "profile-1", 10, 0) + if err != nil { + t.Fatalf("ListHistory: %v", err) + } + if len(history) != 1 { + t.Fatalf("history len = %d, want 1", len(history)) + } + if history[0].WatchedAt != "2026-04-25T12:00:00Z" { + t.Fatalf("history watched_at = %q, want caller watchedAt", history[0].WatchedAt) + } +} + func TestIdentityLookupFailureDoesNotBlockHistory(t *testing.T) { store, db := newTestUserStore(t) defer db.Close() @@ -296,6 +325,338 @@ func TestStableIdentityResolverResolvesSeasonZeroSpecial(t *testing.T) { } } +func TestManualMarkUnwatchedSuppressesImportedHistoryButReturnsManualHistory(t *testing.T) { + store, db := newTestUserStore(t) + defer db.Close() + + if err := store.CreateProfile(context.Background(), userstore.Profile{ID: "profile-1", Name: "Profile"}); err != nil { + t.Fatalf("CreateProfile: %v", err) + } + if err := store.SetProgressAt( + context.Background(), + "profile-1", + "movie-1", + 0, + 7200, + true, + time.Date(2026, 5, 4, 12, 0, 0, 0, time.UTC), + ); err != nil { + t.Fatalf("SetProgressAt: %v", err) + } + if err := store.AddHistory(context.Background(), userstore.WatchHistoryEntry{ + ID: "trakt-history-1", + ProfileID: "profile-1", + MediaItemID: "movie-1", + WatchedAt: "2026-05-04T12:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceTrakt, + Identity: userstore.WatchIdentity{ + StableType: "movie", + ProviderIDs: map[string]string{"tmdb": "603"}, + }, + }); err != nil { + t.Fatalf("AddHistory: %v", err) + } + if err := store.AddHistory(context.Background(), userstore.WatchHistoryEntry{ + ID: "simkl-history-1", + ProfileID: "profile-1", + MediaItemID: "movie-1", + WatchedAt: "2026-05-04T13:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceSimkl, + Identity: userstore.WatchIdentity{ + StableType: "movie", + ProviderIDs: map[string]string{"tmdb": "603"}, + }, + }); err != nil { + t.Fatalf("AddHistory: %v", err) + } + if err := store.AddHistory(context.Background(), userstore.WatchHistoryEntry{ + ID: "manual-history-1", + ProfileID: "profile-1", + MediaItemID: "movie-1", + WatchedAt: "2026-05-04T14:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceManual, + Identity: userstore.WatchIdentity{ + StableType: "movie", + ProviderIDs: map[string]string{"tmdb": "603"}, + }, + }); err != nil { + t.Fatalf("AddHistory: %v", err) + } + + service := NewService(testStoreProvider{store: store}) + result, err := service.RecordManualMarkUnwatchedWithResult(context.Background(), 1, "profile-1", []string{"movie-1"}) + if err != nil { + t.Fatalf("RecordManualMarkUnwatchedWithResult: %v", err) + } + if len(result.Entries) != 1 || result.Entries[0].Source != userstore.WatchHistorySourceManual { + t.Fatalf("unwatch result entries = %+v, want only manual history for outbound sync", result.Entries) + } + + progress, err := store.GetProgress(context.Background(), "profile-1", "movie-1") + if err != nil { + t.Fatalf("GetProgress: %v", err) + } + if progress != nil { + t.Fatalf("progress after unwatch = %+v, want nil", progress) + } + completedItems, err := store.ListCompletedHistoryItems(context.Background(), userstore.CompletedHistoryItemQuery{ + ProfileID: "profile-1", + MediaItemIDs: []string{"movie-1"}, + }) + if err != nil { + t.Fatalf("ListCompletedHistoryItems: %v", err) + } + if len(completedItems) != 0 { + t.Fatalf("completed items after unwatch = %v, want empty", completedItems) + } +} + +func TestManualMarkUnwatchedReturnsOneOutboundEntryPerTarget(t *testing.T) { + store, db := newTestUserStore(t) + defer db.Close() + createWatchstateProfile(t, store) + + for _, entry := range []userstore.WatchHistoryEntry{ + { + ID: "manual-history-older", + ProfileID: "profile-1", + MediaItemID: "movie-1", + WatchedAt: "2026-05-04T12:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceManual, + Identity: userstore.WatchIdentity{ + StableType: "movie", + ProviderIDs: map[string]string{"tmdb": "603"}, + }, + }, + { + ID: "manual-history-newer", + ProfileID: "profile-1", + MediaItemID: "movie-1", + WatchedAt: "2026-05-04T13:00:00Z", + DurationSeconds: 7200, + Completed: true, + Source: userstore.WatchHistorySourceManual, + Identity: userstore.WatchIdentity{ + StableType: "movie", + ProviderIDs: map[string]string{"tmdb": "603"}, + }, + }, + } { + if err := store.AddHistory(context.Background(), entry); err != nil { + t.Fatalf("AddHistory(%s): %v", entry.ID, err) + } + } + + service := NewService(testStoreProvider{store: store}) + result, err := service.RecordManualMarkUnwatchedWithResult(context.Background(), 1, "profile-1", []string{"movie-1"}) + if err != nil { + t.Fatalf("RecordManualMarkUnwatchedWithResult: %v", err) + } + if len(result.Entries) != 1 { + t.Fatalf("unwatch result entries = %+v, want one representative entry", result.Entries) + } + if result.Entries[0].ID != "manual-history-newer" { + t.Fatalf("representative history id = %q, want newest manual history", result.Entries[0].ID) + } +} + +func TestManualMarkWatchedAfterHiddenWatermarkIsVisible(t *testing.T) { + store, db := newTestUserStore(t) + defer db.Close() + createWatchstateProfile(t, store) + + hiddenBefore := time.Now().UTC().Add(time.Second).Format(time.RFC3339) + if _, err := db.Exec(` + INSERT INTO hidden_history_items (profile_id, media_item_id, hidden_before, updated_at) + VALUES (?, ?, ?, ?)`, + "profile-1", + "movie-1", + hiddenBefore, + hiddenBefore, + ); err != nil { + t.Fatalf("seed hidden watermark: %v", err) + } + + service := NewService(testStoreProvider{store: store}) + result, err := service.RecordManualMarkWatchedWithResult( + context.Background(), + 1, + "profile-1", + []LeafWatchTarget{{MediaItemID: "movie-1", DurationSeconds: 7200}}, + time.Now().UTC(), + ) + if err != nil { + t.Fatalf("RecordManualMarkWatchedWithResult: %v", err) + } + if len(result.Entries) != 1 { + t.Fatalf("result entries = %+v, want one history entry", result.Entries) + } + if result.Entries[0].WatchedAt <= hiddenBefore { + t.Fatalf("history watched_at = %q, want after hidden_before %q", result.Entries[0].WatchedAt, hiddenBefore) + } + + progress, err := store.GetProgress(context.Background(), "profile-1", "movie-1") + if err != nil { + t.Fatalf("GetProgress: %v", err) + } + if progress == nil || !progress.Completed { + t.Fatalf("progress = %+v, want visible completed progress", progress) + } + history, err := store.ListHistory(context.Background(), "profile-1", 10, 0) + if err != nil { + t.Fatalf("ListHistory: %v", err) + } + if len(history) != 1 || history[0].WatchedAt <= hiddenBefore { + t.Fatalf("history = %+v, want visible history after hidden watermark %q", history, hiddenBefore) + } +} + +func TestImportedWatchIfNewerDoesNotOverwriteNewerResume(t *testing.T) { + store, db := newTestUserStore(t) + defer db.Close() + createWatchstateProfile(t, store) + if err := store.SetProgressAt( + context.Background(), + "profile-1", + "movie-1", + 1200, + 7200, + false, + time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC), + ); err != nil { + t.Fatalf("SetProgressAt: %v", err) + } + + watchedAt := time.Date(2026, 5, 4, 12, 0, 0, 0, time.UTC) + service := NewService(testStoreProvider{store: store}) + created, err := service.RecordImportedWatchIfNewerWithSource( + context.Background(), + 1, + "profile-1", + "movie-1", + 7200, + 0, + true, + watchedAt, + &watchedAt, + userstore.WatchHistorySourceTrakt, + ) + if err != nil { + t.Fatalf("RecordImportedWatchIfNewerWithSource: %v", err) + } + if !created { + t.Fatal("created = false, want imported history row recorded") + } + progress, err := store.GetProgress(context.Background(), "profile-1", "movie-1") + if err != nil { + t.Fatalf("GetProgress: %v", err) + } + if progress == nil || progress.Completed || progress.PositionSeconds != 1200 { + t.Fatalf("progress after older import = %+v, want newer resume preserved", progress) + } +} + +func TestImportedWatchIfNewerCompletesOlderResume(t *testing.T) { + store, db := newTestUserStore(t) + defer db.Close() + createWatchstateProfile(t, store) + if err := store.SetProgressAt( + context.Background(), + "profile-1", + "movie-1", + 1200, + 7200, + false, + time.Date(2026, 5, 4, 12, 0, 0, 0, time.UTC), + ); err != nil { + t.Fatalf("SetProgressAt: %v", err) + } + + watchedAt := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) + service := NewService(testStoreProvider{store: store}) + created, err := service.RecordImportedWatchIfNewerWithSource( + context.Background(), + 1, + "profile-1", + "movie-1", + 7200, + 0, + true, + watchedAt, + &watchedAt, + userstore.WatchHistorySourceSimkl, + ) + if err != nil { + t.Fatalf("RecordImportedWatchIfNewerWithSource: %v", err) + } + if !created { + t.Fatal("created = false, want imported history row recorded") + } + progress, err := store.GetProgress(context.Background(), "profile-1", "movie-1") + if err != nil { + t.Fatalf("GetProgress: %v", err) + } + if progress == nil || !progress.Completed || progress.PositionSeconds != 0 { + t.Fatalf("progress after newer import = %+v, want completed projection", progress) + } +} + +func TestImportedWatchIfNewerSuppressesHiddenOlderWatch(t *testing.T) { + store, db := newTestUserStore(t) + defer db.Close() + createWatchstateProfile(t, store) + + hiddenBefore := time.Date(2026, 5, 5, 12, 0, 0, 0, time.UTC) + if err := store.RemoveHistoryItems(context.Background(), "profile-1", []string{"movie-1"}, hiddenBefore); err != nil { + t.Fatalf("RemoveHistoryItems: %v", err) + } + watchedAt := time.Date(2026, 5, 4, 12, 0, 0, 0, time.UTC) + service := NewService(testStoreProvider{store: store}) + created, err := service.RecordImportedWatchIfNewerWithSource( + context.Background(), + 1, + "profile-1", + "movie-1", + 7200, + 0, + true, + watchedAt, + &watchedAt, + userstore.WatchHistorySourceTrakt, + ) + if err != nil { + t.Fatalf("RecordImportedWatchIfNewerWithSource: %v", err) + } + if created { + t.Fatal("created = true, want hidden imported history skipped") + } + progress, err := store.GetProgress(context.Background(), "profile-1", "movie-1") + if err != nil { + t.Fatalf("GetProgress: %v", err) + } + if progress != nil { + t.Fatalf("progress after hidden import = %+v, want nil", progress) + } + completedItems, err := store.ListCompletedHistoryItems(context.Background(), userstore.CompletedHistoryItemQuery{ + ProfileID: "profile-1", + MediaItemIDs: []string{"movie-1"}, + }) + if err != nil { + t.Fatalf("ListCompletedHistoryItems: %v", err) + } + if len(completedItems) != 0 { + t.Fatalf("completed items = %v, want hidden import skipped", completedItems) + } +} + func newTestUserStore(t *testing.T) (userstore.UserStore, *sql.DB) { t.Helper() db, err := sql.Open("sqlite3", ":memory:") @@ -308,3 +669,10 @@ func newTestUserStore(t *testing.T) (userstore.UserStore, *sql.DB) { } return userdb.NewSQLiteUserStore(db), db } + +func createWatchstateProfile(t *testing.T, store userstore.UserStore) { + t.Helper() + if err := store.CreateProfile(context.Background(), userstore.Profile{ID: "profile-1", Name: "Profile"}); err != nil { + t.Fatalf("CreateProfile: %v", err) + } +} diff --git a/internal/watchsync/service.go b/internal/watchsync/service.go index b67d6376..fd5b2da5 100644 --- a/internal/watchsync/service.go +++ b/internal/watchsync/service.go @@ -36,7 +36,7 @@ type mediaMatcher interface { } type watchStateImporter interface { - RecordImportedHistoryWithSource(ctx context.Context, userID int, profileID, targetID string, duration float64, completed bool, watchedAt *time.Time, source userstore.WatchHistorySource) (bool, error) + RecordImportedWatchIfNewerWithSource(ctx context.Context, userID int, profileID, targetID string, duration, position float64, completed bool, updatedAt time.Time, watchedAt *time.Time, source userstore.WatchHistorySource) (bool, error) } const ( @@ -993,13 +993,15 @@ func (s *Service) ImportWatched( continue } duration, _ := s.mediaDuration(ctx, match.MediaItemID) - created, err := s.watchState.RecordImportedHistoryWithSource( + created, err := s.watchState.RecordImportedWatchIfNewerWithSource( ctx, conn.UserID, conn.ProfileID, match.MediaItemID, duration, + 0, true, + *row.LastWatchedAt, row.LastWatchedAt, historySourceForProvider(importer), ) diff --git a/internal/watchsync/service_test.go b/internal/watchsync/service_test.go index 0121b756..92d64e52 100644 --- a/internal/watchsync/service_test.go +++ b/internal/watchsync/service_test.go @@ -803,8 +803,29 @@ func (noOpWatchState) RecordImportedHistoryWithSource( return false, nil } +func (noOpWatchState) RecordImportedWatchIfNewerWithSource( + context.Context, + int, + string, + string, + float64, + float64, + bool, + time.Time, + *time.Time, + userstore.WatchHistorySource, +) (bool, error) { + return false, nil +} + type recordingWatchState struct { - sources []userstore.WatchHistorySource + sources []userstore.WatchHistorySource + updatedAt []time.Time + watchedAt []*time.Time + completed []bool + positions []float64 + durations []float64 + targetIDs []string } func (s *recordingWatchState) RecordImportedHistoryWithSource( @@ -821,6 +842,28 @@ func (s *recordingWatchState) RecordImportedHistoryWithSource( return true, nil } +func (s *recordingWatchState) RecordImportedWatchIfNewerWithSource( + _ context.Context, + _ int, + _ string, + targetID string, + duration float64, + position float64, + completed bool, + updatedAt time.Time, + watchedAt *time.Time, + source userstore.WatchHistorySource, +) (bool, error) { + s.sources = append(s.sources, source) + s.updatedAt = append(s.updatedAt, updatedAt) + s.watchedAt = append(s.watchedAt, watchedAt) + s.completed = append(s.completed, completed) + s.positions = append(s.positions, position) + s.durations = append(s.durations, duration) + s.targetIDs = append(s.targetIDs, targetID) + return true, nil +} + func TestServiceStartsAndPollsDeviceAuth(t *testing.T) { repo := newServiceFakeRepo() provider := &authProviderStub{} @@ -1277,6 +1320,53 @@ func TestServiceImportWatchedUsesProviderHistorySource(t *testing.T) { if len(watchState.sources) != 1 || watchState.sources[0] != userstore.WatchHistorySourceSimkl { t.Fatalf("recorded sources = %+v, want simkl", watchState.sources) } + if len(watchState.targetIDs) != 1 || watchState.targetIDs[0] != "movie-1" { + t.Fatalf("recorded target ids = %+v, want movie-1", watchState.targetIDs) + } + if len(watchState.completed) != 1 || !watchState.completed[0] { + t.Fatalf("recorded completed flags = %+v, want true", watchState.completed) + } + if len(watchState.positions) != 1 || watchState.positions[0] != 0 { + t.Fatalf("recorded positions = %+v, want 0", watchState.positions) + } + if len(watchState.updatedAt) != 1 || !watchState.updatedAt[0].Equal(watchedAt) { + t.Fatalf("recorded updated_at = %+v, want %v", watchState.updatedAt, watchedAt) + } + if len(watchState.watchedAt) != 1 || watchState.watchedAt[0] == nil || !watchState.watchedAt[0].Equal(watchedAt) { + t.Fatalf("recorded watched_at = %+v, want %v", watchState.watchedAt, watchedAt) + } +} + +func TestServiceImportWatchedSkipsRowsWithoutLastWatchedAt(t *testing.T) { + repo := newServiceFakeRepo() + provider := watchedImporterStub{ + rows: []RemoteWatch{{ + Provider: "trakt", + Kind: historyimport.KindMovie, + Title: "Inception", + Year: 2010, + }}, + } + watchState := &recordingWatchState{} + service := NewService(repo, NewRegistry()). + WithMatcher(matchedMatcherStub{mediaItemID: "movie-1"}). + WithWatchState(watchState) + + result, err := service.ImportWatched(context.Background(), Connection{ + ID: "conn-1", + Provider: "trakt", + UserID: 7, + ProfileID: "profile-1", + }, ServerConfig{}, provider) + if err != nil { + t.Fatalf("ImportWatched: %v", err) + } + if result.Found != 1 || result.Imported != 0 { + t.Fatalf("result = %+v, want found row skipped with no import", result) + } + if len(watchState.targetIDs) != 0 { + t.Fatalf("watch state calls = %+v, want none", watchState.targetIDs) + } } func TestServiceImportWatchedPersistsBatchCursorsAndWarnings(t *testing.T) {