siftlode/backend/app/routes/search.py

"""Live YouTube search.

Searches YouTube directly 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). The
source is admin-selectable (``search_source``): **scrape** (InnerTube, the website's own
endpoint, zero API quota — the default) or **api** (search.list, **100 quota units per page**).
Either way it's gated behind a per-user daily cap, the API source additionally behind a budget
pre-check, and only real users (never the shared demo account) may call it. The result ids are
enriched through the cheap shared videos.list path (1 unit / 50) regardless of source.

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 pydantic import BaseModel
from sqlalchemy import and_, delete, func, or_, select, update
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session, aliased

from app import logs, quota, sysconfig
from app.auth import require_human
from app.db import get_db
from app.models import (
    LIVE_OR_UPCOMING,
    BlockedChannel,
    Channel,
    Playlist,
    PlaylistItem,
    SearchFind,
    Subscription,
    User,
    Video,
    VideoState,
)
from app.feedquery import feed_columns, serialize
from app.sync.subscriptions import apply_channel_details
from app.sync.videos import insert_stubs, apply_video_details, parse_dt
from app.youtube import search_scrape
from app.youtube.client import YouTubeClient, YouTubeError
from app.youtube.shorts import make_client, probe_is_short

log = logs.get_logger(__name__)
router = APIRouter(prefix="/api/search", tags=["search"])

# search.list costs this many quota units per page.
SEARCH_COST = 100


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.enriched_at is None:
            # Enrichment failed (proxy blip) or YouTube omitted the row — duration is unknown,
            # so DON'T finalise shorts_probed off a null duration; leave it for the scheduler
            # (which only classifies enriched rows) to retry. Otherwise a real Short would leak.
            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]


def _ingest_candidates(
    db: Session, yt: YouTubeClient, user: User, candidates: list[dict]
) -> list[str]:
    """Materialise one page of search candidates into the catalog (channel stubs + enrich,
    video stubs flagged via_search, enrich, Shorts classification) and return the KEEPABLE
    video ids — non-Short, non-live — in the input relevance order."""
    ids = [it["id"] for it in candidates]
    if not ids:
        return []

    # 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:
            # The stubs remain and the details can be backfilled later — but until they are, those
            # channels sit in the results with no thumbnail and no subscriber count, looking like
            # channels that have neither.
            log.warning(
                "the channels found by a search could not be enriched",
                extra=logs.event("search.channel_enrich_failed", channels=len(new_channel_ids)),
            )
            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.
    #    published_at is NOT NULL (migration 0068); the scrape source exposes no publish date
    #    (only a relative "3 years ago"), so fall back to now() to satisfy the constraint — the
    #    enrich in step 3 overwrites it with the video's real snippet date.
    now = datetime.now(timezone.utc)
    insert_stubs(
        db,
        [
            {
                "id": it["id"],
                "channel_id": it["channel_id"],
                "title": it["title"],
                "published_at": parse_dt(it["published_at"]) or now,
                "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:
        # Every candidate then stays unenriched, and an unenriched video cannot be judged live or
        # Short — so it is dropped. A working search returns nothing and reads as "no results".
        log.warning(
            "the videos found by a search could not be enriched",
            extra=logs.event("search.video_enrich_failed", candidates=len(ids)),
        )
        detail_by_id = {}
    enriched_at = datetime.now(
        timezone.utc
    )  # when THIS enrich ran (distinct from the stub placeholder)
    for v in videos:
        item = detail_by_id.get(v.id)
        if item is None:
            continue
        apply_video_details(v, item)
        v.enriched_at = enriched_at
    db.commit()

    # 4) Confirm/deny Shorts (no quota) so none enter via search.
    _classify_shorts(db, videos)

    keep = {
        v.id
        for v in videos
        if v.enriched_at is not None and not v.is_short and v.live_status not in LIVE_OR_UPCOMING
    }
    return [vid for vid in ids if vid in keep]


def _new_first(db: Session, user: User, ids: list[str]) -> list[str]:
    """Reorder results so genuinely-new discoveries come first and things the user already has
    (a subscribed channel, or a video they've watched / in-progress / saved / playlisted) sink
    to the back — keeping the relevance order within each group."""
    if not ids:
        return ids
    sub_ch = select(Subscription.channel_id).where(Subscription.user_id == user.id)
    has_state = (
        select(VideoState.id)
        .where(VideoState.video_id == Video.id, VideoState.user_id == user.id)
        .exists()
    )
    in_playlist = (
        select(PlaylistItem.id)
        .join(Playlist, Playlist.id == PlaylistItem.playlist_id)
        .where(PlaylistItem.video_id == Video.id, Playlist.user_id == user.id)
        .exists()
    )
    have = set(
        db.scalars(
            select(Video.id).where(
                Video.id.in_(ids),
                or_(Video.channel_id.in_(sub_ch), has_state, in_playlist),
            )
        ).all()
    )
    return [i for i in ids if i not in have] + [i for i in ids if i in have]


@router.get("/youtube")
def search_youtube(
    q: str | None = None,
    cursor: str | None = None,
    limit: int = 20,
    user: User = Depends(require_human),
    db: Session = Depends(get_db),
) -> dict:
    """Live YouTube search results materialised into the catalog. `limit` is how many results to
    gather (the count selector); with the free scrape source we page through continuations until
    that many are collected, so there's no manual "load more". The API source costs 100 units per
    page, so it returns a single page (≤50) regardless. `cursor` lets the caller fetch further."""
    term = (q or "").strip()
    if not term:
        raise HTTPException(status_code=400, detail="A search term is required.")
    limit = max(1, min(limit, 100))

    # Source: "scrape" (InnerTube, no API quota) or "api" (search.list, 100 units/page).
    source = sysconfig.get_str(db, "search_source")

    # Per-user daily cap applies to both sources (anti-abuse); with scrape it's the only guard
    # since the search itself spends no quota.
    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.",
        )
    # The 100-unit budget pre-check only applies to the API source.
    if source == "api" and 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.",
        )

    # Channels this user has blocked are dropped before ingestion — never shown, never stored.
    blocked = set(
        db.scalars(select(BlockedChannel.channel_id).where(BlockedChannel.user_id == user.id)).all()
    )

    collected: list[str] = []
    next_cursor = cursor
    MAX_PAGES = 12  # scrape safety bound
    with YouTubeClient(db, user) as yt:
        for _ in range(MAX_PAGES):
            try:
                if source == "api":
                    with quota.attribute(user.id, quota.QuotaAction.VIDEOS_SEARCH):
                        page = yt.search_videos(term, page_token=next_cursor)
                else:
                    page = search_scrape.search(term, continuation=next_cursor)
            except (YouTubeError, search_scrape.ScrapeError) as exc:
                if collected:
                    break  # keep what we already gathered
                # 422, not 502: this is a JSON endpoint, and the FE treats 502/503/504 as a gateway
                # dropout (auto-retry + "connection lost" banner — lib/api/core RETRIABLE_GATEWAY),
                # which hides the real message and burns a second quota-costed search. Upstream failure
                # on a JSON endpoint = 422 + message; reserve 5xx for the raw byte-proxies. (B-core-1)
                raise HTTPException(
                    status_code=422, detail=f"YouTube search failed: {exc}"
                ) from exc

            # Drop currently-live/upcoming, channel-less, and blocked-channel results.
            candidates = [
                it
                for it in page["items"]
                if it["live_broadcast"] == "none"
                and it["channel_id"]
                and it["channel_id"] not in blocked
            ]
            # Ingest (DB writes + enrich) runs outside the page-fetch try above; guard it too so a
            # write/shape failure degrades to a 422 with a message on this JSON endpoint instead of a
            # bare 500 hidden behind the FE's generic "search failed" (same rationale as B-core-1).
            try:
                collected.extend(_ingest_candidates(db, yt, user, candidates))
            except Exception as exc:  # noqa: BLE001 — any ingest failure, not just YouTubeError
                db.rollback()
                # The term is NOT in this line. A search string is exactly what the privacy
                # policy keeps out of the log unless the user turned Teljes/TRACE on for
                # themselves, and `redact_text` does not cover it — it scrubs addresses and
                # credential params, not what someone typed into the search box. What the
                # ERROR line carries instead is the shape of the failure, which is what tells
                # an ingest bug from an upstream one; the term follows at TRACE, where the
                # consent that allows it is the only thing that emits the record at all.
                log.exception(
                    "live search ingest failed",
                    extra=logs.event(
                        "search.live.ingest_failed",
                        source=source,
                        candidates=len(candidates),
                        collected=len(collected),
                    ),
                )
                log.log(
                    logs.TRACE,
                    "the term the ingest failed on",
                    # log-pii-ok: TRACE is the consented level, and this is what it is for
                    extra=logs.event("search.live.term", term=term),
                )
                if collected:
                    break  # keep what earlier pages already gathered
                raise HTTPException(
                    status_code=422, detail=f"YouTube search failed: {exc}"
                ) from exc
            next_cursor = page["next_page_token"]
            # The API source pages cost 100 units each — never auto-page; one page per request.
            if source == "api" or len(collected) >= limit or not next_cursor:
                break

    # Count this as ONE per-user search against the daily cap. The scrape source spends no
    # quota, so it never logs an event on its own; the API source already logged exactly one
    # (via record_usage — it never auto-pages). Logging once here, after a search that produced
    # at least one page (a total failure raises 502 above and never reaches this), keeps the cap
    # counting user searches instead of internal continuation pages.
    if source != "api":
        quota.log_action(user.id, quota.QuotaAction.VIDEOS_SEARCH)

    ordered = collected[:limit]

    if ordered:
        # Record these as THIS user's search finds (Mine feed / Source filter), and fold the
        # query into each result's search_terms so local full-text search inherits YouTube's
        # relevance. Idempotent; the generated search_vector updates automatically.
        db.execute(
            pg_insert(SearchFind)
            .values([{"user_id": user.id, "video_id": vid} for vid in ordered])
            .on_conflict_do_nothing(index_elements=["user_id", "video_id"])
        )
        db.execute(
            update(Video)
            .where(Video.id.in_(ordered))
            .where(func.coalesce(Video.search_terms, "").notilike(f"%{term}%"))
            .values(
                search_terms=func.trim(
                    func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term)
                )
            )
        )
        db.commit()

    ordered = _new_first(db, user, ordered)
    return {
        "items": _serialize_ids(db, user, ordered),
        "next_cursor": next_cursor,
        "has_more": bool(next_cursor),
        "source": source,
    }


class ClearSearchIn(BaseModel):
    video_ids: list[str] = []


@router.post("/clear")
def clear_search(
    body: ClearSearchIn,
    user: User = Depends(require_human),
    db: Session = Depends(get_db),
) -> dict:
    """Discard the given live-search results "as if never added": drop this user's search-finds
    for them, then hard-delete the now-orphaned ones — a via_search video nobody KEPT (no watch
    state, not playlisted, channel not subscribed) and that no other user still has as a search
    find. Kept videos and organically-owned ones are left untouched."""
    ids = body.video_ids
    if not ids:
        return {"removed": 0}
    db.execute(
        delete(SearchFind).where(SearchFind.user_id == user.id, SearchFind.video_id.in_(ids))
    )
    db.commit()
    has_state = select(VideoState.id).where(VideoState.video_id == Video.id).exists()
    in_playlist = select(PlaylistItem.id).where(PlaylistItem.video_id == Video.id).exists()
    subscribed = select(Subscription.id).where(Subscription.channel_id == Video.channel_id).exists()
    other_find = select(SearchFind.id).where(SearchFind.video_id == Video.id).exists()
    removed = db.execute(
        delete(Video).where(
            Video.id.in_(ids),
            Video.via_search.is_(True),
            ~has_state,
            ~in_playlist,
            ~subscribed,
            ~other_find,
        )
    ).rowcount
    db.commit()
    return {"removed": removed}