feat(search): live YouTube search backend

Add a live YouTube search that materialises results into the shared catalog so
they render with the normal feed cards + in-app player and gain per-user state.

- YouTubeClient.search_videos(): search.list (100 units), embeddable-only, returns
  flat stubs + nextPageToken; surfaces liveBroadcastContent for live filtering.
- routes/search.py GET /api/search/youtube: require_human + per-user daily cap
  (search_daily_limit_per_user, default 70) + can_spend pre-check (429 on either);
  drops live/upcoming, upserts channel stubs (channels.list) + video stubs, enriches
  (videos.list), runs the youtube.com/shorts probe, then excludes Shorts/live and
  returns feed cards in relevance order with the YouTube pageToken as the cursor.
- Provenance: videos.via_search / channels.from_search (migration 0028) flag
  search-discovered rows; the feed hides them from the Library (scope=all) by default
  via exclude_search_discovered, leaving the Mine feed untouched.
- quota.actions_today() counts a user's per-action events today for the cap; only the
  search.list call is attributed VIDEOS_SEARCH so the counter is exactly 1 per search.

backend/alembic/versions/0028_live_search_provenance.py

@@ -0,0 +1,46 @@
+"""live YouTube search: provenance flags
+
+Revision ID: 0028_live_search_provenance
+Revises: 0027_message_e2ee
+Create Date: 2026-06-29
+
+Adds provenance flags so videos (and channels) that entered the catalog only because a live
+YouTube search surfaced them can be told apart from organically-synced content and hidden from
+the Library by default:
+  - videos.via_search      — this video first arrived via search
+  - channels.from_search   — this channel exists only because search surfaced one of its videos
+Both default false (every existing row is organic), indexed for the feed's exclude filter.
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0028_live_search_provenance"
+down_revision: Union[str, None] = "0027_message_e2ee"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "videos",
+        sa.Column(
+            "via_search", sa.Boolean(), nullable=False, server_default="false"
+        ),
+    )
+    op.create_index("ix_videos_via_search", "videos", ["via_search"])
+    op.add_column(
+        "channels",
+        sa.Column(
+            "from_search", sa.Boolean(), nullable=False, server_default="false"
+        ),
+    )
+    op.create_index("ix_channels_from_search", "channels", ["from_search"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_channels_from_search", table_name="channels")
+    op.drop_column("channels", "from_search")
+    op.drop_index("ix_videos_via_search", table_name="videos")
+    op.drop_column("videos", "via_search")

backend/app/config.py

@@ -71,6 +71,9 @@ class Settings(BaseSettings):
     youtube_api_proxy: str = ""
     # Daily quota budget (units). Default leaves headroom under the 10,000/day free limit.
     quota_daily_budget: int = 9000
+    # Per-user cap on live YouTube searches per day. search.list costs 100 units each, so the
+    # shared budget only affords ~80-90/day total — this stops one user draining it. Admin-tunable.
+    search_daily_limit_per_user: int = 70
     # Recent-first backfill: how far back to fetch on the first pass per channel.
     backfill_recent_max_videos: int = 100
     backfill_recent_max_days: int = 365

backend/app/main.py

@@ -39,6 +39,7 @@ from app.routes import (
     playlists,
     quota,
     scheduler as scheduler_routes,
+    search as search_routes,
     setup as setup_routes,
     sync,
     tags,
@@ -113,6 +114,7 @@ app.include_router(auth.router)
 app.include_router(sync.router)
 app.include_router(tags.router)
 app.include_router(feed.router)
+app.include_router(search_routes.router)
 app.include_router(me.router)
 app.include_router(notifications.router)
 app.include_router(messages.router)

backend/app/models.py

@@ -175,6 +175,12 @@ class Channel(Base, TimestampMixin):
     backfill_done: Mapped[bool] = mapped_column(
         Boolean, default=False, server_default="false"
     )
+    # True when this channel only entered the catalog because a live YouTube search
+    # surfaced one of its videos (nobody subscribes to it / it has no organic upload sync).
+    # Lets the Library hide search-only "noise" by default.
+    from_search: Mapped[bool] = mapped_column(
+        Boolean, default=False, server_default="false", index=True
+    )
 
     videos: Mapped[list["Video"]] = relationship(back_populates="channel")
     subscriptions: Mapped[list["Subscription"]] = relationship(back_populates="channel")
@@ -238,6 +244,12 @@ class Video(Base, TimestampMixin):
     shorts_probed: Mapped[bool] = mapped_column(
         Boolean, default=False, server_default="false", index=True
     )
+    # 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
+    )
     # none | live | upcoming | premiere | was_live
     live_status: Mapped[str] = mapped_column(
         String(16), default="none", server_default="none", index=True

backend/app/quota.py

@@ -36,6 +36,7 @@ class QuotaAction:
     VIDEOS_BACKFILL_FULL = "videos_backfill_full"
     VIDEOS_ENRICH = "videos_enrich"
     VIDEOS_LOOKUP = "videos_lookup"
+    VIDEOS_SEARCH = "videos_search"
     CHANNELS_DISCOVER = "channels_discover"
     CHANNELS_SUBSCRIBE = "channels_subscribe"
     CHANNELS_UNSUBSCRIBE = "channels_unsubscribe"
@@ -109,6 +110,26 @@ def can_spend(db: Session, units: int) -> bool:
     return remaining_today(db) >= units
 
 
+def actions_today(db: Session, user_id: int, action: str) -> int:
+    """How many quota events of `action` this user has logged so far in the current Pacific
+    day — for per-user, per-action daily caps (e.g. the live-search limit). Counts events,
+    not units, so it only works for actions charged exactly once per user action."""
+    from sqlalchemy import func, select  # local import: keep the module's import head lean
+
+    return (
+        db.scalar(
+            select(func.count())
+            .select_from(QuotaEvent)
+            .where(
+                QuotaEvent.user_id == user_id,
+                QuotaEvent.action == action,
+                QuotaEvent.created_at >= pacific_day_start_utc(),
+            )
+        )
+        or 0
+    )
+
+
 def record_usage(db: Session, units: int) -> None:
     """Atomically add `units` to today's counter (upsert) and log an attribution event."""
     if units <= 0:

backend/app/routes/feed.py

@@ -124,6 +124,7 @@ def _filtered_query(
     include_live: bool,
     show: str,
     scope: str = "my",
+    exclude_search_discovered: bool = True,
     exclude_tag_category: str | None = None,
 ) -> tuple[Select, object]:
     """Build the feed query (joins + all WHERE filters), shared by /feed and /feed/count.
@@ -171,6 +172,12 @@ def _filtered_query(
     # either way they shouldn't show in any feed view meanwhile.
     query = query.where(Video.unavailable_since.is_(None))
 
+    # Hide videos that only entered the catalog via a live YouTube search (search-discovered
+    # "noise") from the Library by default. Only meaningful in "all" scope: "my" already
+    # restricts to subscribed channels, where a once-searched video is legitimately the user's.
+    if scope == "all" and exclude_search_discovered:
+        query = query.where(Video.via_search.is_(False))
+
     if channel_id:
         query = query.where(Video.channel_id == channel_id)
     if min_duration is not None:
@@ -271,6 +278,7 @@ def _feed_params(
     include_live: bool = False,
     show: str = "unwatched",
     scope: str = "my",
+    exclude_search_discovered: bool = True,
 ) -> dict:
     return {
         "tags": tags,
@@ -287,6 +295,7 @@ def _feed_params(
         "include_live": include_live,
         "show": show,
         "scope": scope,
+        "exclude_search_discovered": exclude_search_discovered,
     }
 
 

backend/app/routes/search.py

@@ -0,0 +1,204 @@
+"""Live YouTube search.
+
+Searches YouTube directly (search.list) and materialises the results into the shared catalog
+so they render with the normal feed cards + in-app player and gain per-user state (watch /
+save / resume). search.list costs **100 quota units per call**, so it's the most expensive
+read by far: it's gated behind a per-user daily cap + a budget pre-check, and only real users
+(never the shared demo account) may call it.
+
+Ingested results are flagged (videos.via_search / channels.from_search) so the Library can hide
+this search-discovered "noise" by default. Shorts and currently-live/upcoming videos are never
+ingested this way (Shorts are confirmed by the youtube.com/shorts probe; live/upcoming are
+dropped from the search response up front)."""
+from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import and_, select
+from sqlalchemy.orm import Session, aliased
+
+from app import quota, sysconfig
+from app.auth import require_human
+from app.db import get_db
+from app.models import Channel, User, Video, VideoState
+from app.routes.feed import _serialize, feed_columns
+from app.sync.subscriptions import apply_channel_details
+from app.sync.videos import _insert_stubs, apply_video_details, parse_dt
+from app.youtube.client import YouTubeClient, YouTubeError
+from app.youtube.shorts import make_client, probe_is_short
+
+router = APIRouter(prefix="/api/search", tags=["search"])
+
+# search.list costs this many quota units per page.
+SEARCH_COST = 100
+# live_status values we never surface from a live search (was_live VODs are real content).
+_LIVE_HIDDEN = ("live", "upcoming")
+
+
+def _classify_shorts(db: Session, videos: list[Video]) -> None:
+    """Confirm/deny Shorts for the freshly-ingested videos so none slip into the feed.
+
+    Mirrors the scheduler's run_shorts_classification, but scoped to just these rows and run
+    synchronously (the search is an explicit, awaited action). Non-candidates (too long, no
+    duration, or live) are finalised without a probe; the rest hit youtube.com/shorts (no
+    quota). Undetermined probes are left for the scheduler to retry."""
+    probe_max = sysconfig.get_int(db, "shorts_probe_max_seconds")
+    to_probe: list[Video] = []
+    for v in videos:
+        if v.shorts_probed:
+            continue
+        if (
+            v.live_status != "none"
+            or v.duration_seconds is None
+            or v.duration_seconds > probe_max
+        ):
+            v.is_short = False
+            v.shorts_probed = True
+        else:
+            to_probe.append(v)
+    if to_probe:
+        with make_client() as client:
+
+            def work(v: Video):
+                return v, probe_is_short(client, v.id)
+
+            with ThreadPoolExecutor(max_workers=16) as pool:
+                for v, result in pool.map(work, to_probe):
+                    if result is None:
+                        continue
+                    v.is_short = result
+                    v.shorts_probed = True
+    db.commit()
+
+
+def _serialize_ids(db: Session, user: User, ids: list[str]) -> list[dict]:
+    """Serialize the given video ids with the shared feed projection (joined channel + this
+    user's watch state + saved), preserving the input (search relevance) order."""
+    if not ids:
+        return []
+    state = aliased(VideoState)
+    rows = db.execute(
+        select(*feed_columns(user, state))
+        .join(Channel, Channel.id == Video.channel_id)
+        .outerjoin(state, and_(state.video_id == Video.id, state.user_id == user.id))
+        .where(Video.id.in_(ids))
+    ).all()
+    by_id = {r.id: _serialize(r) for r in rows}
+    return [by_id[i] for i in ids if i in by_id]
+
+
+@router.get("/youtube")
+def search_youtube(
+    q: str | None = None,
+    cursor: str | None = None,
+    user: User = Depends(require_human),
+    db: Session = Depends(get_db),
+) -> dict:
+    """One page of live YouTube search results, materialised into the catalog. `cursor` is the
+    YouTube nextPageToken (each page is another 100-unit search)."""
+    term = (q or "").strip()
+    if not term:
+        raise HTTPException(status_code=400, detail="A search term is required.")
+
+    # Guard the shared quota BEFORE spending: per-user daily cap, then today's budget.
+    daily_limit = sysconfig.get_int(db, "search_daily_limit_per_user")
+    used = quota.actions_today(db, user.id, quota.QuotaAction.VIDEOS_SEARCH)
+    if used >= daily_limit:
+        raise HTTPException(
+            status_code=429,
+            detail=f"You've reached today's YouTube search limit ({daily_limit}). Try again tomorrow.",
+        )
+    if not quota.can_spend(db, SEARCH_COST):
+        raise HTTPException(
+            status_code=429,
+            detail="The shared YouTube quota for today is used up. Try again tomorrow.",
+        )
+
+    with YouTubeClient(db, user) as yt:
+        try:
+            with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH):
+                page = yt.search_videos(term, page_token=cursor)
+        except YouTubeError as exc:
+            raise HTTPException(status_code=502, detail=f"YouTube search failed: {exc}")
+
+        # Drop currently-live/upcoming results (search can't filter them and we never ingest
+        # them this way) and anything missing a channel id.
+        candidates = [
+            it
+            for it in page["items"]
+            if it["live_broadcast"] == "none" and it["channel_id"]
+        ]
+        ids = [it["id"] for it in candidates]
+        if not ids:
+            return {
+                "items": [],
+                "next_cursor": page["next_page_token"],
+                "has_more": bool(page["next_page_token"]),
+            }
+
+        # 1) Channels are the FK target — create stubs for unknown ones (flagged from_search),
+        #    then enrich the new ones (channels.list, 1 unit). Stubs survive an enrich failure.
+        channel_ids = list({it["channel_id"] for it in candidates})
+        existing = set(
+            db.scalars(select(Channel.id).where(Channel.id.in_(channel_ids))).all()
+        )
+        new_channel_ids = [c for c in channel_ids if c not in existing]
+        if new_channel_ids:
+            titles = {it["channel_id"]: it["channel_title"] for it in candidates}
+            for cid in new_channel_ids:
+                db.add(Channel(id=cid, title=titles.get(cid), from_search=True))
+            db.commit()
+            try:
+                with quota.attribute(user.id, quota.QuotaAction.CHANNELS_DISCOVER):
+                    details = yt.get_channels(new_channel_ids)
+                apply_channel_details(db, details)
+                db.commit()
+            except YouTubeError:
+                db.rollback()  # details can be backfilled later; the stubs remain
+
+        # 2) Video stubs, flagged via_search. Existing rows keep their flags (ON CONFLICT DO
+        #    NOTHING), so an organically-synced video stays organic.
+        _insert_stubs(
+            db,
+            [
+                {
+                    "id": it["id"],
+                    "channel_id": it["channel_id"],
+                    "title": it["title"],
+                    "published_at": parse_dt(it["published_at"]),
+                    "thumbnail_url": it["thumbnail_url"],
+                    "via_search": True,
+                }
+                for it in candidates
+            ],
+        )
+
+        # 3) Enrich (videos.list, 1 unit/50) so cards have duration/stats and we can judge live.
+        videos = db.scalars(select(Video).where(Video.id.in_(ids))).all()
+        try:
+            with quota.attribute(user.id, quota.QuotaAction.VIDEOS_ENRICH):
+                detail_by_id = {it["id"]: it for it in yt.get_videos(ids)}
+        except YouTubeError:
+            detail_by_id = {}
+        now = datetime.now(timezone.utc)
+        for v in videos:
+            item = detail_by_id.get(v.id)
+            if item is None:
+                continue
+            apply_video_details(v, item)
+            v.enriched_at = now
+        db.commit()
+
+        # 4) Confirm/deny Shorts (no quota) so none enter via search.
+        _classify_shorts(db, videos)
+
+    # 5) Exclude Shorts and still-live results; keep the search's relevance order.
+    keep = {
+        v.id for v in videos if not v.is_short and v.live_status not in _LIVE_HIDDEN
+    }
+    ordered = [vid for vid in ids if vid in keep]
+    return {
+        "items": _serialize_ids(db, user, ordered),
+        "next_cursor": page["next_page_token"],
+        "has_more": bool(page["next_page_token"]),
+    }

backend/app/sysconfig.py

@@ -39,6 +39,7 @@ SPECS: tuple[ConfigSpec, ...] = (
     # --- Quota (shared YouTube Data API daily budget) ---
     ConfigSpec("quota_daily_budget", "int", "quota", "quota_daily_budget", min=1, max=1_000_000),
     ConfigSpec("backfill_quota_reserve", "int", "quota", "backfill_quota_reserve", min=0, max=1_000_000),
+    ConfigSpec("search_daily_limit_per_user", "int", "quota", "search_daily_limit_per_user", min=0, max=100_000),
     # --- Backfill (recent-first first pass per channel) ---
     ConfigSpec("backfill_recent_max_videos", "int", "backfill", "backfill_recent_max_videos", min=1, max=100_000),
     ConfigSpec("backfill_recent_max_days", "int", "backfill", "backfill_recent_max_days", min=1, max=36_500),

backend/app/youtube/client.py

@@ -351,6 +351,49 @@ class YouTubeClient:
         """Delete a whole playlist on YouTube. 50 units."""
         self._write("DELETE", "playlists", params={"id": playlist_id})
 
+    def search_videos(
+        self, q: str, page_token: str | None = None, order: str = "relevance"
+    ) -> dict:
+        """Live YouTube search (search.list) for embeddable videos. **100 quota units per
+        call** — by far the most expensive read, so callers must gate it (per-user cap +
+        budget pre-check). Public read: prefers the API key over OAuth.
+
+        Returns one page (up to 50) as ``{"items": [...], "next_page_token": str|None}``
+        where each item is a flat dict ready for stub insertion. `liveBroadcastContent` is
+        surfaced so the caller can drop currently-live/upcoming results (search can't filter
+        those out for us). Snippet carries no duration/stats — the caller enriches via
+        videos.list afterwards."""
+        params = {
+            "part": "snippet",
+            "q": q,
+            "type": "video",
+            "maxResults": 50,
+            "order": order,
+            "videoEmbeddable": "true",
+            "safeSearch": "none",
+        }
+        if page_token:
+            params["pageToken"] = page_token
+        data = self._get("search", params, cost=100)
+        items: list[dict] = []
+        for it in data.get("items", []):
+            vid = it.get("id", {}).get("videoId")
+            if not vid:
+                continue
+            sn = it.get("snippet", {})
+            items.append(
+                {
+                    "id": vid,
+                    "channel_id": sn.get("channelId"),
+                    "channel_title": sn.get("channelTitle"),
+                    "title": sn.get("title"),
+                    "published_at": sn.get("publishedAt"),
+                    "thumbnail_url": best_thumbnail(sn.get("thumbnails")),
+                    "live_broadcast": sn.get("liveBroadcastContent") or "none",
+                }
+            )
+        return {"items": items, "next_page_token": data.get("nextPageToken")}
+
     def get_videos(self, video_ids: list[str]) -> list[dict]:
         items: list[dict] = []
         for batch in _chunks(video_ids, 50):