Plex backend pass (#9): watch_sync bugs + backend dedup

Closes the Plex #9 backend backlog (frontend was the prior follow-up). Behavior-
neutral cleanup + two two-way-sync bug fixes; PW2 verified a non-bug.

fix(plex): watch-sync PW1 pagination + PW3 lost-update guard
- PW1 watch_history: the history is a GLOBAL cross-account feed read as a single
  500-row page, then filtered to the owner — on a busy family server the owner's
  recent views can sit past page 1 and be missed. Now paginate (desc) until the
  since-cutoff, bounded by max_pages (daily reconcile is the backstop; logs if hit).
- PW3 push_state_to_plex: it flagged synced_to_plex=True unconditionally after a
  push, so a local edit landing between the push and the flag-set was marked clean
  and never pushed (lost update). Capture the row's freshest timestamp when the
  push is scheduled (_push_watch) and only settle the flag if the row is unchanged.
- PW2 was flagged as "accountID hardcoded to 1" but is NOT a bug: on a PMS account 1
  is reserved for the owner/admin and an owner link always holds the admin token —
  added a clarifying comment so it isn't re-flagged.

chore(plex): backend dedup
- PC2 unified_library + facets shared a ~13-field filter Query signature + p-dict →
  one LibraryFilters dataclass injected via Depends(); as_dict() feeds the builders.
- PS-C1 sync.py collection upsert dup (resync_collection ≈ _sync_collections) →
  _upsert_collection().
- PS-C2 sync.py 8-field filterable-metadata block dup (_apply_item ≈ _sync_shows) →
  _apply_facet_fields().
- PW-C1 _repush_dirty read duration from the mirrored PlexItem.duration_s instead of
  a per-row plex.metadata() round-trip.
- PW-C2 rk→id map built inline in two places → _rk_to_id() helper.

backend/app/plex/client.py

@@ -217,27 +217,42 @@ class PlexClient:
         return self._get("/accounts").get("Account", []) or []
 
     def watch_history(
-        self, min_viewed_at: int = 0, account_id: int | None = None, size: int = 500
+        self, min_viewed_at: int = 0, account_id: int | None = None, size: int = 500, max_pages: int = 40
     ) -> list[dict]:
-        """The most recent watch-history rows (ratingKey / viewedAt / accountID / type), newest
-        first — works without Plex Pass. Filtered client-side to `account_id` and to rows at/after
-        `min_viewed_at` (epoch seconds): the efficient "what changed since T" feed. Sorted desc, so
-        we stop at the first row older than the cutoff."""
-        mc = self._get(
-            "/status/sessions/history/all",
-            params={
-                "sort": "viewedAt:desc",
-                "X-Plex-Container-Start": 0,
-                "X-Plex-Container-Size": size,
-            },
-        )
-        out = []
-        for r in mc.get("Metadata", []) or []:
-            if int(r.get("viewedAt") or 0) < min_viewed_at:
+        """The watch-history rows (ratingKey / viewedAt / accountID / type) newer than `min_viewed_at`
+        (epoch seconds), newest first — works without Plex Pass. The efficient "what changed since T"
+        feed, filtered client-side to `account_id`.
+
+        The history is a GLOBAL feed across all server accounts, so paginate (sorted desc) until we
+        reach a row at/older than the cutoff: on a busy family server the owner's recent views can sit
+        past the first page behind other accounts' rows. `max_pages` bounds the scan (the daily full
+        reconcile is the backstop for anything beyond it)."""
+        out: list[dict] = []
+        for page in range(max_pages):
+            mc = self._get(
+                "/status/sessions/history/all",
+                params={
+                    "sort": "viewedAt:desc",
+                    "X-Plex-Container-Start": page * size,
+                    "X-Plex-Container-Size": size,
+                },
+            )
+            rows = mc.get("Metadata", []) or []
+            reached_cutoff = False
+            for r in rows:
+                if int(r.get("viewedAt") or 0) < min_viewed_at:
+                    reached_cutoff = True
+                    break
+                if account_id is not None and int(r.get("accountID") or -1) != account_id:
+                    continue
+                out.append(r)
+            if reached_cutoff or len(rows) < size:
                 break
-            if account_id is not None and int(r.get("accountID") or -1) != account_id:
-                continue
-            out.append(r)
+        else:
+            log.warning(
+                "watch_history hit the %d-page scan cap (since=%s); reconcile will catch the tail",
+                max_pages, min_viewed_at,
+            )
         return out
 
     def on_deck(self) -> list[dict]: