siftlode/backend/app/discover/pins.py

"""The comfort shelf: pin a thing you can always fall back to, and have it be one click from playing.

Five kinds are pinnable — a video, a YouTube channel, a Plex show, a Plex movie, a playlist — because
comfort media is almost always "put on X", a series or a channel, and only sometimes one particular
video. `models.ComfortPin` carries that union as five nullable FKs under a `num_nonnulls(...) = 1`
check; this module is where the union is spoken about in one vocabulary, so no call site invents a
second one.

Every shelf card carries a resolved `next`: the thing that actually plays when you click the tile —
a show's in-progress-or-next episode, a channel's newest unwatched video, a playlist's first
unwatched item. That buys the property the whole surface exists for: one click, no intermediate page,
no second decision. Channels and playlists are resolved for the whole shelf at once (`_Batch`), so the
cost does not grow a query or two per pin.

Domain module, so nothing here raises HTTP. `resolve_ref` returns None for "no such thing, or not
yours" and the route turns that into its 404 — core never imports routes.
"""

from dataclasses import dataclass, field

from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.discover import cards, queries
from app.models import Channel, ComfortPin, PlexItem, PlexShow, Playlist, User, Video
from app.plex import rollup
from app.plex.cards import leaf_card, show_card
from app.plex.lookups import by_rating_key

# kind -> the ComfortPin column that holds it. The client's vocabulary and the schema's, in one place.
PIN_COLUMNS: dict[str, str] = {
    "video": "video_id",
    "channel": "channel_id",
    "plex_item": "plex_item_id",
    "plex_show": "plex_show_id",
    "playlist": "playlist_id",
}
PIN_KINDS = tuple(PIN_COLUMNS)


def resolve_ref(db: Session, user: User, kind: str, ref: str) -> tuple[str, object] | None:
    """Translate the client's (kind, ref) into the (column, value) the row stores, or None.

    The client speaks the ids the rest of the API speaks: a video id, a channel id, a Plex
    `rating_key`, a playlist id. Plex's internal integer keys never leave the backend, so the
    translation belongs here rather than in five call sites.

    The COLUMN half of every answer comes from `PIN_COLUMNS`; only the LOOKUP differs per kind
    (a direct `db.get`, a rating-key translation, an int parse plus an ownership check). Before the
    E24 review the branches spelled the column names out again and `PIN_COLUMNS` had no readers at
    all — a constant that called itself the one vocabulary while nothing spoke it."""
    col = PIN_COLUMNS.get(kind)
    if col is None:
        return None
    if kind == "video":
        return (col, ref) if db.get(Video, ref) is not None else None
    if kind == "channel":
        return (col, ref) if db.get(Channel, ref) is not None else None
    if kind == "plex_item":
        it = by_rating_key(db, PlexItem, ref)
        return (col, it.id) if it is not None else None
    if kind == "plex_show":
        sh = by_rating_key(db, PlexShow, ref)
        return (col, sh.id) if sh is not None else None
    try:
        pid = int(ref)
    # silent-ok: validating a reference the client supplied. "Not an integer" is answered as "no
    # such pin target", which is the same answer this function gives for one that does not exist.
    except ValueError:
        return None
    pl = db.get(Playlist, pid)
    # A playlist is per-user, unlike the other four (which are shared catalog rows whose PRIVACY
    # lives in the state tables). Someone else's list must read as "no such thing", not as a 403 —
    # the same reason `owned_or_404` 404s rather than 403s.
    return (col, pl.id) if pl is not None and pl.user_id == user.id else None


def reorder(db: Session, user: User, ids: list[int]) -> None:
    """Put the listed pins in the listed order, WITHIN the slots they already hold.

    The client sends the order it can SEE, which is not always the whole shelf: with Plex off the Home
    rail hides the Plex pins, and a stale tab may not know a pin added elsewhere. Numbering only the
    listed ids 0..n-1 collided with every pin left out — two tiles on one position, and the hidden
    ones jumped about when they came back. So the pins left out keep their slots, and the listed ones
    are dealt into the slots the listed set occupies. Ids that are not this user's, or repeated, are
    ignored rather than failing a stale replay wholesale. Flushes; the route commits."""
    # Locked: the rows it leaves alone are part of the answer too. Two overlapping reorders each
    # skipped the rows that already matched in their own snapshot and interleaved into two pins on
    # one slot; the second now waits and deals from the first one's result.
    current = list_pins(db, user, lock=True)
    owned = {p.id for p in current}
    listed: list[int] = []
    for i in ids:
        if i in owned and i not in listed:
            listed.append(i)
    chosen = set(listed)
    queue = iter(listed)
    order = [next(queue) if p.id in chosen else p.id for p in current]
    by_id = {p.id: p for p in current}
    for position, pin_id in enumerate(order):
        if by_id[pin_id].position != position:
            by_id[pin_id].position = position
    db.flush()


def list_pins(db: Session, user: User, *, lock: bool = False) -> list[ComfortPin]:
    stmt = (
        select(ComfortPin)
        .where(ComfortPin.user_id == user.id)
        .order_by(ComfortPin.position.asc(), ComfortPin.id.asc())
    )
    return list(db.execute(stmt.with_for_update() if lock else stmt).scalars().all())


def next_position(db: Session, user: User) -> int:
    """New pins land at the end of the shelf.

    `max(position) + 1`, not the row count. Positions are never renumbered on removal — `remove_pin`
    deletes the row, and every FK cascade silently removes a pin when the thing it points at goes
    away — so past the first hole a count under-counts and would drop the new pin into the MIDDLE of
    the shelf, or onto a position another pin already holds (nothing in the schema forbids that).
    Same idiom as `routes/playlists._next_playlist_position`."""
    return (
        db.scalar(
            select(func.coalesce(func.max(ComfortPin.position), -1)).where(
                ComfortPin.user_id == user.id
            )
        )
        + 1
    )


@dataclass
class _Batch:
    """What the shelf resolves for every pinned channel and playlist in one go."""

    channels: dict[str, Channel] = field(default_factory=dict)
    channel_summary: dict[str, tuple[int, dict | None]] = field(default_factory=dict)
    playlists: dict[int, Playlist] = field(default_factory=dict)
    playlist_summary: dict[int, tuple[int, dict | None]] = field(default_factory=dict)

    @classmethod
    def load(cls, db: Session, user: User, pins: list[ComfortPin]) -> "_Batch":
        ch_ids = [p.channel_id for p in pins if p.channel_id is not None]
        pl_ids = [p.playlist_id for p in pins if p.playlist_id is not None]
        b = cls()
        if ch_ids:
            b.channels = {
                c.id: c for c in db.scalars(select(Channel).where(Channel.id.in_(ch_ids)))
            }
            b.channel_summary = queries.channels_summary(db, user, ch_ids)
        if pl_ids:
            b.playlists = {
                p.id: p for p in db.scalars(select(Playlist).where(Playlist.id.in_(pl_ids)))
            }
            b.playlist_summary = queries.playlists_summary(db, user, pl_ids)
        return b


def _shelf_card(db: Session, user: User, pin: ComfortPin, batch: _Batch) -> dict | None:
    """One pin as a unified card with its resolved `next`. None when the pinned row has gone (the FK
    cascade removes the pin itself, so this is only the window between a delete and a refetch)."""
    card: dict | None = None
    nxt: dict | None = None

    if pin.video_id is not None:
        row = queries.yt_video_row(db, user, pin.video_id)
        if row is None:
            return None
        card = cards.from_video(row)
        nxt = dict(card)  # a video plays itself

    elif pin.channel_id is not None:
        ch = batch.channels.get(pin.channel_id)
        if ch is None:
            return None
        count, row = batch.channel_summary.get(ch.id, (0, None))
        card = cards.from_channel(ch, count)
        nxt = cards.from_video(row) if row else None

    elif pin.plex_item_id is not None:
        it = db.get(PlexItem, pin.plex_item_id)
        if it is None:
            return None
        st = rollup.item_state_map(db, user.id, [it.id]).get(it.id)
        card = cards.from_plex(leaf_card(it, st))
        nxt = dict(card)  # a movie or a single episode plays itself

    elif pin.plex_show_id is not None:
        sh = db.get(PlexShow, pin.plex_show_id)
        if sh is None:
            return None
        status = rollup.state_map(db, user.id, [sh.id]).get(sh.id, "new")
        card = cards.from_plex(show_card(sh, status))
        ep = queries.next_episode(db, user.id, sh.id)
        nxt = cards.from_plex(ep) if ep else None

    elif pin.playlist_id is not None:
        pl = batch.playlists.get(pin.playlist_id)
        if pl is None:
            return None
        count, row = batch.playlist_summary.get(pl.id, (0, None))
        card = cards.from_playlist(pl, count, row.get("thumbnail_url") if row else None)
        nxt = cards.from_video(row) if row else None

    if card is None:
        return None
    # A COPY, never the card itself: a self-playing pin (a video, a movie) would otherwise make
    # `card["next"] is card`, which serializes as a circular reference and 500s the whole shelf. The
    # copy is taken before `pin_id`/`next` are attached, so `next` stays a plain playable card.
    card["pin_id"] = pin.id
    card["next"] = nxt
    return card


_PLEX_COLUMNS = ("plex_item_id", "plex_show_id")


def shelf(db: Session, user: User, *, include_plex: bool = True) -> list[dict]:
    """The whole shelf, in the user's own order.

    `include_plex=False` is the Home surface with the deployment's Plex toggle off (the owner's call,
    2026-09-19): the rails beside it go YouTube-only, and a Plex tile there would offer a Play that
    lands on a gated module. The pins themselves are kept — the pin menus still read the full shelf
    (`GET /pins`), and switching Plex back on brings the tiles back in their old slots
    (`reorder` keeps a hidden pin's place)."""
    out = []
    pins = list_pins(db, user)
    if not include_plex:
        pins = [p for p in pins if all(getattr(p, c) is None for c in _PLEX_COLUMNS)]
    batch = _Batch.load(db, user, pins)
    for pin in pins:
        card = _shelf_card(db, user, pin, batch)
        if card is not None:
            out.append(card)
    return out


def playables(shelf_cards: list[dict]) -> list[dict]:
    """The `next` of every shelf entry that has one — what the hero may draw from. A pinned channel
    with nothing unwatched left, or an empty playlist, simply is not an answer right now.

    Each carries `pinned_from`: the pin it was resolved from. A show's next EPISODE is not itself
    pinned, so without it the hero's menu looked the episode up, missed, and offered "Pin" under a
    kicker that says "From your shelf" — and taking it inserted a second pin. With it, the hero's
    pin row acts on the shelf entry the card came from (the owner's call, 2026-09-19: "Unpin")."""
    return [
        {
            **c["next"],
            "pinned_from": {k: c[k] for k in ("source", "kind", "id", "title")},
        }
        for c in shelf_cards
        if c.get("next")
    ]