perf(jobs): partial indexes for the enrich/maintenance scanners + boolean cleanup (E2 S6, PERF-17/18/3)

Two scheduler jobs full-seq-scanned the 257k-row videos table every run:
- PERF-17 enrich_pending (`WHERE enriched_at IS NULL LIMIT N`, every few minutes):
  44,680 buffers even with 0 pending -> a partial ix_videos_enrich_pending makes
  it a 1-buffer index scan (empty when nothing pending).
- PERF-18 maintenance revalidate (`WHERE unavailable_since IS NULL ORDER BY
  last_checked_at NULLS FIRST LIMIT N`, ~300 pages/run): 44,754 buffers + top-N
  sort -> a partial ix_videos_recheck_order in that exact order = a 42-buffer
  index scan. The order_by is rewritten to `asc().nulls_first()` (equivalent to
  the old `(IS NULL) DESC, ASC`) so it rides the index; the unused plain
  ix_videos_last_checked_at is dropped.

PERF-3: drop the never-picked boolean btrees (is_short, via_explore) and slim the
useful ones to partials over the queried subset (via_search=true, NOT
shorts_probed, unavailable_since IS NOT NULL) — less write amplification on the
biggest table, same lookups. Model: those 6 columns lose their inline index=True
(the partials are migration-defined, like the search_vector GIN); live_status
left alone.

Measured on the prod-sized localdev reseed: enrich 44,680 -> 1 buffer (25.9 ->
0.02 ms), maintenance 44,754 -> 42 buffers (35.5 -> 0.17 ms); all 11 CONCURRENTLY
index ops on 257k rows in 0.75s. Gate green (346 pytest incl. drift + downgrade
round-trip).

backend/alembic/versions/0069_job_index_hygiene.py

@@ -0,0 +1,90 @@
+"""scheduler-job partial indexes + boolean-index cleanup (PERF-17/18/3)
+
+Two scheduler jobs full-seq-scanned the 257k-row videos table every run:
+  - PERF-17: enrich_pending (`WHERE enriched_at IS NULL LIMIT N`, every few minutes) — 44.6k buffers /
+    run even when nothing is pending. A partial index over just the pending rows makes it a 1-buffer
+    (empty) index scan.
+  - PERF-18: maintenance revalidate (`WHERE unavailable_since IS NULL ORDER BY last_checked_at NULLS
+    FIRST LIMIT N`, ~300 pages/run) — 44.7k buffers/page; ix_videos_last_checked_at had idx_scan=0. A
+    partial index in that exact order serves it as an index scan (~36 buffers/page) and replaces the
+    unused plain one.
+
+PERF-3: the videos table also carried plain btrees on low-cardinality boolean columns the planner never
+picks (a `col IS false` majority predicate can't ride a btree). Drop the never-useful ones (is_short,
+via_explore) and slim the rest to partial indexes over the small/shrinking subset the job queries target
+(via_search=true, NOT shorts_probed, unavailable_since IS NOT NULL) — less write amplification on the
+biggest table, same lookups. live_status is left alone (a separate finding).
+
+All CONCURRENTLY (autocommit_block) — downtime ~0; IF (NOT) EXISTS keeps the drift test's
+downgrade->re-upgrade idempotent.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "0069_job_index_hygiene"
+down_revision: Union[str, None] = "0068_published_at_not_null"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    with op.get_context().autocommit_block():
+        # PERF-17: enrich-pending partial (empty when nothing is pending -> 1-buffer scan).
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_enrich_pending "
+            "ON videos (id) WHERE enriched_at IS NULL"
+        )
+        # PERF-18: maintenance revalidate order, as a partial in the exact WHERE + ORDER BY shape.
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_recheck_order "
+            "ON videos (last_checked_at ASC NULLS FIRST) WHERE unavailable_since IS NULL"
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_last_checked_at")
+        # PERF-3: drop the never-picked boolean btrees; slim the rest to partials over the target subset.
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_is_short")
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_via_explore")
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_via_search")
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_via_search "
+            "ON videos (id) WHERE via_search"
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_shorts_probed")
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_shorts_probed "
+            "ON videos (id) WHERE NOT shorts_probed"
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_unavailable_since")
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_unavailable_since "
+            "ON videos (unavailable_since) WHERE unavailable_since IS NOT NULL"
+        )
+
+
+def downgrade() -> None:
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_enrich_pending")
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_recheck_order")
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_last_checked_at "
+            "ON videos (last_checked_at)"
+        )
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_is_short ON videos (is_short)"
+        )
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_via_explore ON videos (via_explore)"
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_via_search")
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_via_search ON videos (via_search)"
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_shorts_probed")
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_shorts_probed ON videos (shorts_probed)"
+        )
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_videos_unavailable_since")
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_videos_unavailable_since "
+            "ON videos (unavailable_since)"
+        )

backend/app/models.py

@@ -285,18 +285,22 @@ class Video(Base, TimestampMixin):
     # folds YouTube's own relevance judgement into the local search: a video YouTube returned
     # for "ti amo magyarul" becomes findable by that query locally even if its title lacks it.
     search_terms: Mapped[str | None] = mapped_column(Text)
+    # No btree here or on the flag columns below: migration 0069 (PERF-3) dropped the never-picked
+    # boolean btrees and replaced the useful ones with partial indexes over the target subset
+    # (via_search=true, NOT shorts_probed, unavailable_since IS NOT NULL) — migration-defined, like the
+    # search_vector GIN.
     is_short: Mapped[bool] = mapped_column(
-        Boolean, default=False, server_default="false", index=True
+        Boolean, default=False, server_default="false"
     )
     # Whether the youtube.com/shorts/<id> probe has run for this video.
     shorts_probed: Mapped[bool] = mapped_column(
-        Boolean, default=False, server_default="false", index=True
+        Boolean, default=False, server_default="false"
     )
     # True when this video first entered the catalog via a live YouTube search (not an
     # organic channel/playlist sync). Used to hide search-discovered "noise" from the
     # Library by default; left False by the organic ingest paths.
     via_search: Mapped[bool] = mapped_column(
-        Boolean, default=False, server_default="false", index=True
+        Boolean, default=False, server_default="false"
     )
     # True when this video entered the catalog because a user EXPLORED its (un-subscribed)
     # channel. Like via_search it's hidden from the shared Library by default, but it's
@@ -304,7 +308,7 @@ class Video(Base, TimestampMixin):
     # explored or subscribe to its channel (a gate in the feed query), so one user's
     # curiosity never floods everyone else's catalog. Left False by the organic ingest paths.
     via_explore: Mapped[bool] = mapped_column(
-        Boolean, default=False, server_default="false", index=True
+        Boolean, default=False, server_default="false"
     )
     # none | live | upcoming | premiere | was_live
     live_status: Mapped[str] = mapped_column(
@@ -320,11 +324,14 @@ class Video(Base, TimestampMixin):
     # (least-recently-checked first); unavailable_since marks a video as currently
     # unplayable (hidden from the feed immediately, hard-deleted after a grace period);
     # unavailable_reason records why (deleted|private|paywalled|abandoned|stuck_live).
+    # last_checked_at: no plain btree — migration 0069 (PERF-18) replaced it with the partial
+    # ix_videos_recheck_order (last_checked_at NULLS FIRST WHERE unavailable_since IS NULL) that the
+    # revalidate job's exact ORDER BY rides; unavailable_since is a partial WHERE IS NOT NULL (PERF-3).
     last_checked_at: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), index=True
+        DateTime(timezone=True)
     )
     unavailable_since: Mapped[datetime | None] = mapped_column(
-        DateTime(timezone=True), index=True
+        DateTime(timezone=True)
     )
     unavailable_reason: Mapped[str | None] = mapped_column(String(24))
 

backend/app/sync/maintenance.py

@@ -194,10 +194,10 @@ def _revalidate_rolling(db: Session, yt: YouTubeClient) -> dict:
             db.execute(
                 select(Video)
                 .where(Video.unavailable_since.is_(None))
-                .order_by(
+                # NULLS FIRST = never-checked first, then least-recently-checked; equivalent to the old
-                    Video.last_checked_at.is_(None).desc(),  # never-checked first
+                # `(last_checked_at IS NULL) DESC, last_checked_at ASC` but in the exact shape of the
-                    Video.last_checked_at.asc(),
+                # partial ix_videos_recheck_order (migration 0069, PERF-18) so it rides that index.
-                )
+                .order_by(Video.last_checked_at.asc().nulls_first())
                 .limit(min(50, batch - checked))
             )
             .scalars()