feat(plex): 'last watched' browse sort + auto-apply in the in-progress view (S3)

Add a 'lastwatched' sort key to the unified Plex browse. Movies rank by
coalesce(progress_updated_at, watched_at); shows by max(coalesce(...)) across
their episodes (rollup.resumed_at, now broadened from progress-only so watched
episodes count too). Never-watched titles sink via nullslast.

Frontend auto-applies lastwatched:desc when entering the in-progress watch-state
view and restores the pre-entry sort on exit; a manual sort change while there
still wins. i18n HU+EN. Backend tests cover both union legs + mixed nullslast.

backend/app/plex/rollup.py

@@ -43,8 +43,10 @@ def rollup(cards: list[dict]) -> dict:
 def agg_subquery(db: Session, user_id: int, include_resumed: bool = False):
     """Per-show episode counts for a user: total, watched, in-progress. Reused to (a) filter the show
     grid by aggregate watch-state, (b) compute each card's badge, and (c) drive the Discover
-    continue-watching rail. `include_resumed` adds `resumed_at` = max(progress_updated_at) across the
-    show's episodes, so the rail can order shows by how recently they were last progressed."""
+    continue-watching rail, and (d) the browse "last watched" sort. `include_resumed` adds `resumed_at`
+    = max(coalesce(progress_updated_at, watched_at)) across the show's episodes — i.e. how recently ANY
+    episode was last touched (resumed OR marked watched) — so callers can order shows by recency. A
+    fully-watched show (no resume progress) still gets a timestamp from its episodes' watched_at."""
     ps = aliased(PlexState)
     cols = [
         PlexItem.show_id.label("show_id"),
@@ -63,7 +65,9 @@ def agg_subquery(db: Session, user_id: int, include_resumed: bool = False):
         ).label("inprog"),
     ]
     if include_resumed:
-        cols.append(func.max(ps.progress_updated_at).label("resumed_at"))
+        cols.append(
+            func.max(func.coalesce(ps.progress_updated_at, ps.watched_at)).label("resumed_at")
+        )
     return (
         select(*cols)
         .outerjoin(ps, and_(ps.item_id == PlexItem.id, ps.user_id == user_id))

backend/app/routes/plex/browse.py

@@ -138,6 +138,9 @@ def unified_library(
                 cast(null(), Integer).label("season_count"),
                 func.coalesce(st.position_seconds, 0).label("position_seconds"),
                 m_status.label("status"),
+                # "last watched" sort key: when the movie was last resumed, else when it was marked
+                # watched. NULL for never-touched movies → nullslast sinks them (see the sort dict).
+                func.coalesce(st.progress_updated_at, st.watched_at).label("last_watched"),
                 (
                     func.ts_rank(vector_col(PlexItem, lang), tsq)
                     if tsq is not None
@@ -159,7 +162,7 @@ def unified_library(
             mq = mq.where(recheck(PlexItem, field, tsq, lang))
         selects.append(mq)
     if want_shows and grid_field:
-        agg = rollup.agg_subquery(db, user.id).subquery()
+        agg = rollup.agg_subquery(db, user.id, include_resumed=True).subquery()
         tot = func.coalesce(agg.c.total, 0)
         wat = func.coalesce(agg.c.watched, 0)
         inp = func.coalesce(agg.c.inprog, 0)
@@ -179,6 +182,9 @@ def unified_library(
                 PlexShow.child_count.label("season_count"),
                 literal(0).label("position_seconds"),
                 s_status.label("status"),
+                # Show leg's "last watched": max(resume/watched time) across the show's episodes (see
+                # rollup.agg_subquery include_resumed). NULL for a never-watched show → nullslast.
+                agg.c.resumed_at.label("last_watched"),
                 (
                     func.ts_rank(vector_col(PlexShow, lang), tsq)
                     if tsq is not None
@@ -212,6 +218,7 @@ def unified_library(
                 "rating": u.c.rating,
                 "duration": u.c.duration_s,
                 "release": u.c.rel,
+                "lastwatched": u.c.last_watched,
             }.get(sort, u.c.added_at)
             order = [(col.asc() if sort_dir == "asc" else col.desc()).nullslast()]
         order.append(u.c.rk.desc())

backend/tests/test_plex_last_watched_sort.py

@@ -0,0 +1,158 @@
+"""S3: the "last watched" browse sort (`sort=lastwatched`). Movies rank by
+coalesce(progress_updated_at, watched_at); shows by max(coalesce(...)) across their episodes
+(rollup.resumed_at). Never-watched titles sink to the bottom (nullslast). DB lane (`db` fixture) —
+skipped without Postgres.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+from fastapi.testclient import TestClient
+
+import app.state as state
+from app.auth import current_user
+from app.db import get_db
+from app.main import app
+from app.models import PlexItem, PlexLibrary, PlexShow, PlexState, User
+
+NOW = datetime(2024, 6, 1, tzinfo=timezone.utc)
+
+
+def _seed(db):
+    lib_m = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
+    lib_s = PlexLibrary(plex_key="2", title="Shows", kind="show", enabled=True)
+    db.add_all([lib_m, lib_s])
+    db.flush()
+
+    # Movies: newest resume, a watched-only (coalesce falls back to watched_at), an older resume, and a
+    # never-touched one.
+    mo_new = PlexItem(rating_key="mo_new", library_id=lib_m.id, kind="movie", title="Newest Resume")
+    mo_watched = PlexItem(
+        rating_key="mo_watched", library_id=lib_m.id, kind="movie", title="Watched Only"
+    )
+    mo_old = PlexItem(rating_key="mo_old", library_id=lib_m.id, kind="movie", title="Older Resume")
+    mo_never = PlexItem(rating_key="mo_never", library_id=lib_m.id, kind="movie", title="Never")
+
+    # Shows: a recently-resumed episode, a watched-only episode, and a never-watched show.
+    sh_recent = PlexShow(
+        rating_key="sh_recent", library_id=lib_s.id, title="Recent Show", child_count=1
+    )
+    sh_watched = PlexShow(
+        rating_key="sh_watched", library_id=lib_s.id, title="Watched Show", child_count=1
+    )
+    sh_never = PlexShow(
+        rating_key="sh_never", library_id=lib_s.id, title="Never Show", child_count=1
+    )
+    db.add_all([mo_new, mo_watched, mo_old, mo_never, sh_recent, sh_watched, sh_never])
+    db.flush()
+
+    ep_recent = PlexItem(
+        rating_key="sh_recent_e1",
+        library_id=lib_s.id,
+        kind="episode",
+        title="E1",
+        show_id=sh_recent.id,
+        season_number=1,
+        episode_number=1,
+    )
+    ep_watched = PlexItem(
+        rating_key="sh_watched_e1",
+        library_id=lib_s.id,
+        kind="episode",
+        title="E1",
+        show_id=sh_watched.id,
+        season_number=1,
+        episode_number=1,
+    )
+    db.add_all([ep_recent, ep_watched])
+
+    user = User(email="lastwatched@example.com")
+    db.add(user)
+    db.flush()
+
+    db.add_all(
+        [
+            PlexState(
+                user_id=user.id,
+                item_id=mo_new.id,
+                status="new",
+                position_seconds=100,
+                progress_updated_at=NOW,
+            ),
+            PlexState(
+                user_id=user.id,
+                item_id=mo_watched.id,
+                status="watched",
+                watched_at=NOW - timedelta(hours=1),
+            ),
+            PlexState(
+                user_id=user.id,
+                item_id=mo_old.id,
+                status="new",
+                position_seconds=100,
+                progress_updated_at=NOW - timedelta(hours=2),
+            ),
+            # Show episodes.
+            PlexState(
+                user_id=user.id,
+                item_id=ep_recent.id,
+                status="new",
+                position_seconds=50,
+                progress_updated_at=NOW - timedelta(minutes=30),
+            ),
+            PlexState(
+                user_id=user.id,
+                item_id=ep_watched.id,
+                status="watched",
+                watched_at=NOW - timedelta(minutes=90),
+            ),
+        ]
+    )
+    db.commit()
+    return user
+
+
+def _get(db, user, monkeypatch, path):
+    monkeypatch.setattr(state, "_configured_cache", True)
+    app.dependency_overrides[get_db] = lambda: db
+    app.dependency_overrides[current_user] = lambda: user
+    try:
+        return TestClient(app).get(path)
+    finally:
+        app.dependency_overrides.clear()
+
+
+def _ids(resp):
+    return [i["id"] for i in resp.json()["items"]]
+
+
+def test_movies_ordered_by_last_watched_desc(db, monkeypatch):
+    user = _seed(db)
+    resp = _get(
+        db, user, monkeypatch, "/api/plex/library?scope=movie&sort=lastwatched&sort_dir=desc"
+    )
+    assert resp.status_code == 200, resp.text
+    # Newest resume, then the watched-only (coalesce → watched_at), then the older resume, then never.
+    assert _ids(resp) == ["mo_new", "mo_watched", "mo_old", "mo_never"]
+
+
+def test_shows_ordered_by_last_watched_desc(db, monkeypatch):
+    user = _seed(db)
+    resp = _get(
+        db, user, monkeypatch, "/api/plex/library?scope=show&sort=lastwatched&sort_dir=desc"
+    )
+    assert resp.status_code == 200, resp.text
+    # Recent resume first, then the watched-only show, then the never-watched one (nullslast).
+    assert _ids(resp) == ["sh_recent", "sh_watched", "sh_never"]
+
+
+def test_last_watched_mixed_scope_nullslast(db, monkeypatch):
+    user = _seed(db)
+    resp = _get(
+        db, user, monkeypatch, "/api/plex/library?scope=both&sort=lastwatched&sort_dir=desc"
+    )
+    assert resp.status_code == 200, resp.text
+    ids = _ids(resp)
+    # Global recency order across both legs: mo_new (NOW) > sh_recent (−30m) > mo_watched (−1h) >
+    # sh_watched (−90m) > mo_old (−2h); the two never-watched titles sink to the bottom.
+    assert ids[:5] == ["mo_new", "sh_recent", "mo_watched", "sh_watched", "mo_old"]
+    assert set(ids[5:]) == {"mo_never", "sh_never"}