peter
3c16012524
ruff gains RUF100, PLW1510 and flake8-bugbear, every hit fixed rather than ignored. B008 stays on with FastAPI's Depends/Query declared immutable. basedpyright moves to standard mode, pytest to --strict-markers, and a hand run of pytest now requires the database unless SIFTLODE_TEST_REQUIRE_DB=0. PLW1510 found a real bug: the three ffprobe probes in plex/stream.py never looked at the exit status, so a probe that failed was read as the file's truth (an HDR source sent down the HW-safe path, a multi-audio file collapsed to one track, a subtitle offset taken from garbage) and the warning each except branch promises never fired. They now run with check=True.
684 lines
32 KiB
Python
684 lines
32 KiB
Python
"""Feed read-model: the pure query-building and row-serialization layer behind the feed.
|
|
|
|
Extracted from `routes/feed.py` so the read-model has one home instead of living inside the
|
|
router module — the feed route handlers and the playlist-detail query share these helpers (the
|
|
projection, the filter builder, and keyset paging). Nothing here defines a route or reaches for
|
|
request state; it's all `(db, user, params) -> query`, and a malformed cursor raises a plain
|
|
`ValueError` for the route handler to translate (the tsquery sanitizer lives in `app.utils`).
|
|
"""
|
|
|
|
import base64
|
|
import binascii
|
|
import json
|
|
from datetime import date, datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import BigInteger, Select, and_, case, cast, false, func, or_, select
|
|
from sqlalchemy.orm import Session, aliased
|
|
|
|
from app.collation import collated
|
|
from app.models import (
|
|
LIVE_OR_UPCOMING,
|
|
BlockedChannel,
|
|
Channel,
|
|
ChannelTag,
|
|
ExploredChannel,
|
|
Playlist,
|
|
PlaylistItem,
|
|
SearchFind,
|
|
Subscription,
|
|
Tag,
|
|
User,
|
|
Video,
|
|
VideoState,
|
|
)
|
|
from app.utils import iso, like_escape, to_tsquery_str
|
|
|
|
|
|
def _channel_url(channel_id: str, handle: str | None) -> str:
|
|
if handle and handle.startswith("@"):
|
|
return f"https://www.youtube.com/{handle}"
|
|
return f"https://www.youtube.com/channel/{channel_id}"
|
|
|
|
|
|
def _saved_expr(user: User):
|
|
"""A correlated EXISTS labelled `saved`: is this video in the user's Watch later list?
|
|
(Watch later is the built-in playlist that replaced the old per-video 'saved' status.)"""
|
|
return (
|
|
select(1)
|
|
.select_from(PlaylistItem)
|
|
.join(Playlist, Playlist.id == PlaylistItem.playlist_id)
|
|
.where(
|
|
Playlist.user_id == user.id,
|
|
Playlist.kind == "watch_later",
|
|
PlaylistItem.video_id == Video.id,
|
|
)
|
|
.exists()
|
|
.label("saved")
|
|
)
|
|
|
|
|
|
def feed_columns(user: User, state) -> list:
|
|
"""The shared feed-row projection (joined Channel fields + per-user watch state +
|
|
`saved`), consumed by both the feed query and the playlist-detail query so a new column
|
|
is added in exactly one place. `state` is the caller's aliased VideoState (outer-joined)."""
|
|
return [
|
|
Video.id,
|
|
Video.title,
|
|
Video.channel_id,
|
|
Channel.title.label("channel_title"),
|
|
Channel.thumbnail_url.label("channel_thumbnail"),
|
|
Channel.handle.label("channel_handle"),
|
|
Video.published_at,
|
|
Video.thumbnail_url,
|
|
Video.duration_seconds,
|
|
Video.view_count,
|
|
Video.is_short,
|
|
Video.live_status,
|
|
func.coalesce(state.status, "new").label("status"),
|
|
func.coalesce(state.position_seconds, 0).label("position_seconds"),
|
|
_saved_expr(user),
|
|
]
|
|
|
|
|
|
def serialize(row) -> dict:
|
|
return {
|
|
"id": row.id,
|
|
"title": row.title,
|
|
"channel_id": row.channel_id,
|
|
"channel_title": row.channel_title,
|
|
"channel_thumbnail": row.channel_thumbnail,
|
|
"channel_url": _channel_url(row.channel_id, row.channel_handle),
|
|
"published_at": iso(row.published_at),
|
|
"thumbnail_url": row.thumbnail_url,
|
|
"duration_seconds": row.duration_seconds,
|
|
"view_count": row.view_count,
|
|
"is_short": row.is_short,
|
|
"live_status": row.live_status,
|
|
"status": row.status or "new",
|
|
"position_seconds": row.position_seconds or 0,
|
|
"saved": bool(getattr(row, "saved", False)),
|
|
"watch_url": f"https://www.youtube.com/watch?v={row.id}",
|
|
}
|
|
|
|
|
|
def _tag_visible(user: User):
|
|
"""A ChannelTag row this user may see: a shared system tag (user_id NULL) or their own."""
|
|
return or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id)
|
|
|
|
|
|
def tags_by_category(db: Session, tag_ids: list[int]) -> dict[str, list[int]]:
|
|
"""Group tag ids by their category (language/topic/other). The single source for the category split
|
|
shared by the feed's tag WHERE (filtered_query) and the facet/count summary (feed_summary), so the
|
|
two can't disagree on which selected tag belongs to which category."""
|
|
if not tag_ids:
|
|
return {} # skip the pointless `WHERE id IN ()` round-trip on the no-tag-filter hot path
|
|
out: dict[str, list[int]] = {}
|
|
for tid, cat in db.execute(select(Tag.id, Tag.category).where(Tag.id.in_(tag_ids))).all():
|
|
out.setdefault(cat, []).append(tid)
|
|
return out
|
|
|
|
|
|
def not_blocked_cond(user: User):
|
|
"""Per-user channel blocklist: never show this user videos from a channel they blocked
|
|
(applies to feed, Library, explore, the channel page and the Home resume rail alike).
|
|
|
|
A shared expression rather than a clause each surface writes for itself. Blocking does not touch
|
|
`video_states`, so a half-watched video from a newly blocked channel survives with its resume
|
|
position — and the one query that copied only PART of this predicate set is exactly how that
|
|
video came back as the landing page's suggestion after the user had said "never again"."""
|
|
return Video.channel_id.notin_(
|
|
select(BlockedChannel.channel_id).where(BlockedChannel.user_id == user.id)
|
|
)
|
|
|
|
|
|
def filtered_query(
|
|
db: Session,
|
|
user: User,
|
|
*,
|
|
tags: list[int],
|
|
tag_modes: dict[str, str],
|
|
untagged: bool = False,
|
|
channel_ids: list[str],
|
|
q: str | None,
|
|
min_duration: int | None,
|
|
max_duration: int | None,
|
|
max_age_days: int | None,
|
|
published_after: date | None,
|
|
published_before: date | None,
|
|
show_normal: bool,
|
|
include_shorts: bool,
|
|
include_live: bool,
|
|
show: str,
|
|
scope: str = "my",
|
|
library_source: str = "organic",
|
|
exclude_tag_category: str | None = None,
|
|
) -> tuple[Select, object]:
|
|
"""Build the feed query (joins + all WHERE filters), shared by /feed and /feed/count.
|
|
Returns the column-bearing select plus the priority/relevance rank expression for sorting.
|
|
|
|
`scope="my"` (default) restricts the feed to the user's own non-hidden subscriptions.
|
|
`scope="all"` shows every video in the shared catalog (any user's ingested channels);
|
|
the subscription is then only LEFT-joined so per-channel priority still resolves for
|
|
channels the user happens to be subscribed to. Per-user watch state stays private in
|
|
either mode via the VideoState outer join."""
|
|
state = aliased(VideoState)
|
|
status_expr = func.coalesce(state.status, "new")
|
|
|
|
query = select(*feed_columns(user, state)).join(Channel, Channel.id == Video.channel_id)
|
|
|
|
if scope == "all":
|
|
# Whole shared catalog; subscription is optional (only for priority sort).
|
|
query = query.outerjoin(
|
|
Subscription,
|
|
and_(
|
|
Subscription.channel_id == Video.channel_id,
|
|
Subscription.user_id == user.id,
|
|
),
|
|
)
|
|
else:
|
|
# "Mine": the user's own videos = their non-hidden subscriptions OR videos they
|
|
# surfaced via their own live YouTube search. Both are LEFT-joined so the Source
|
|
# filter below can include either or both, and the subscription row stays available
|
|
# for the priority sort.
|
|
query = query.outerjoin(
|
|
Subscription,
|
|
and_(
|
|
Subscription.channel_id == Video.channel_id,
|
|
Subscription.user_id == user.id,
|
|
Subscription.hidden.is_(False),
|
|
),
|
|
).outerjoin(
|
|
SearchFind,
|
|
and_(SearchFind.video_id == Video.id, SearchFind.user_id == user.id),
|
|
)
|
|
|
|
# The state-gated views (watched/hidden/in_progress) only ever return videos that HAVE a
|
|
# video_states row, so an INNER join lets the planner drive from the user's own small video_states
|
|
# set (~100 rows) instead of scanning all videos and outer-joining (PERF-16). status is NOT NULL on
|
|
# a present row (server_default 'new'), so the raw-column filters below match the coalesce form
|
|
# exactly. 'unwatched'/'all' include row-less 'new' videos, so they keep the outer join.
|
|
state_required = show in ("watched", "hidden", "in_progress")
|
|
join_state = query.join if state_required else query.outerjoin
|
|
query = join_state(state, and_(state.video_id == Video.id, state.user_id == user.id))
|
|
|
|
# Hide videos the maintenance job has flagged as currently unplayable (deleted/private/
|
|
# abandoned). They're either deleted after a grace period or unhidden if they recover;
|
|
# either way they shouldn't show in any feed view meanwhile.
|
|
query = query.where(Video.unavailable_since.is_(None))
|
|
|
|
# Explored videos are PER-EXPLORER private: a via_explore video (ingested when someone
|
|
# opened an un-subscribed channel's page) is visible only to users who explored or
|
|
# subscribe to its channel, so one user's curiosity never leaks into everyone else's
|
|
# catalog. Applied in BOTH scopes; the channel page reaches its ephemeral videos by
|
|
# asking for scope=all + channel_ids (the channel is in this accessible set).
|
|
accessible_explore = (
|
|
select(Subscription.channel_id).where(Subscription.user_id == user.id)
|
|
).union(select(ExploredChannel.channel_id).where(ExploredChannel.user_id == user.id))
|
|
query = query.where(
|
|
or_(
|
|
Video.via_explore.is_(False),
|
|
Video.channel_id.in_(accessible_explore),
|
|
)
|
|
)
|
|
|
|
query = query.where(not_blocked_cond(user))
|
|
|
|
# Source filter. In the shared Library ("all" scope) it uses the GLOBAL via_search flag
|
|
# (search-discovered by anyone). In "my" scope it's the user's OWN provenance:
|
|
# "organic" = your subscriptions, "search" = videos you found via your own search,
|
|
# "all" = both. Default "organic" keeps the Mine feed = subscriptions only.
|
|
if scope == "all":
|
|
if library_source == "organic":
|
|
query = query.where(Video.via_search.is_(False))
|
|
elif library_source == "search":
|
|
query = query.where(Video.via_search.is_(True))
|
|
# "all" → both organic and search-discovered videos
|
|
else:
|
|
subscribed = Subscription.channel_id.isnot(None)
|
|
searched = SearchFind.video_id.isnot(None)
|
|
if library_source == "search":
|
|
query = query.where(searched)
|
|
elif library_source == "all":
|
|
query = query.where(or_(subscribed, searched))
|
|
else: # "organic" (default): subscriptions only
|
|
query = query.where(subscribed)
|
|
|
|
if channel_ids:
|
|
# OR across the picked channels; every other filter still applies on top.
|
|
query = query.where(Video.channel_id.in_(channel_ids))
|
|
if min_duration is not None:
|
|
query = query.where(Video.duration_seconds >= min_duration)
|
|
if max_duration is not None:
|
|
query = query.where(Video.duration_seconds <= max_duration)
|
|
if max_age_days is not None:
|
|
cutoff = datetime.now(timezone.utc).timestamp() - max_age_days * 86400
|
|
query = query.where(Video.published_at >= datetime.fromtimestamp(cutoff, tz=timezone.utc))
|
|
if published_after is not None:
|
|
start = datetime.combine(published_after, datetime.min.time(), tzinfo=timezone.utc)
|
|
query = query.where(Video.published_at >= start)
|
|
if published_before is not None:
|
|
end = datetime.combine(
|
|
published_before, datetime.min.time(), tzinfo=timezone.utc
|
|
) + timedelta(days=1)
|
|
query = query.where(Video.published_at < end)
|
|
# Full-text relevance search over the weighted search document (title > creator keywords +
|
|
# the queries that surfaced the video > description; see Video.search_vector). YouTube-like
|
|
# "fuzzy" matching: word-order-independent, multi-word AND, prefix on the word being typed,
|
|
# accent-insensitive (unaccent_simple). The channel name still matches as a substring so
|
|
# channel searches work. `rank_expr` (weight-aware ts_rank) drives the "relevance" sort, over the
|
|
# newest RANK_WINDOW matches only (see there).
|
|
rank_expr = None
|
|
if q:
|
|
# Resolve the channels whose NAME matches up front (there are only a few hundred channels, so
|
|
# this is a tiny standalone scan) and match videos by channel_id. ORing an unindexed
|
|
# `unaccent(channels.title) ILIKE '%q%'` straight into the per-video predicate defeated the
|
|
# search_vector GIN — the planner detoasted every non-matching tsvector to test the `@@`
|
|
# (~2.6s/search on the prod-sized set); channel_id.in_(...) keeps the GIN as a BitmapOr branch
|
|
# (PERF-12). Empty match -> false() so the OR collapses to the text match alone. The query
|
|
# is a literal inside the pattern: `_` in `my_channel` is not a wildcard.
|
|
like = func.unaccent(f"%{like_escape(q)}%")
|
|
matching_channel_ids = (
|
|
db.execute(select(Channel.id).where(func.unaccent(Channel.title).ilike(like)))
|
|
.scalars()
|
|
.all()
|
|
)
|
|
channel_match = (
|
|
Video.channel_id.in_(matching_channel_ids) if matching_channel_ids else false()
|
|
)
|
|
ts_str = to_tsquery_str(q)
|
|
if ts_str:
|
|
tsq = func.to_tsquery("public.unaccent_simple", ts_str)
|
|
query = query.where(or_(Video.search_vector.op("@@")(tsq), channel_match))
|
|
# Scale ts_rank (a float4) to an integer score so the keyset cursor compares it
|
|
# EXACTLY — a raw float4 vs the float8 cursor value mismatches on round-trip and
|
|
# breaks paging (the same page repeats). 1e6 granularity is ample; ties break on
|
|
# published_at then id.
|
|
rank_expr = cast(func.ts_rank(Video.search_vector, tsq) * 1000000, BigInteger)
|
|
else:
|
|
# No usable tokens (e.g. only punctuation): fall back to a plain substring match.
|
|
query = query.where(or_(func.unaccent(Video.title).ilike(like), channel_match))
|
|
|
|
if tags:
|
|
# AND across tag categories (e.g. language AND topic narrows), OR within a category — unless that
|
|
# category's mode is "and" (its per-category any/all toggle), which requires a channel to carry
|
|
# ALL its selected tags. ⚠️ feed_summary.passes() reimplements this exact rule in Python for the
|
|
# sidebar count/facets — keep the two in lockstep (equivalence-tested by
|
|
# test_feed_summary_matches_count_plus_facets).
|
|
by_category = tags_by_category(db, tags)
|
|
# Facet counting drops the category being counted so its own chips don't zero each
|
|
# other out (standard drill-down faceting: a category's count ignores its own
|
|
# selections but still honours the other categories' filters).
|
|
if exclude_tag_category:
|
|
by_category.pop(exclude_tag_category, None)
|
|
visible = _tag_visible(user)
|
|
for category, ids in by_category.items():
|
|
if tag_modes.get(category, "or") == "and" and len(ids) > 1:
|
|
sub = (
|
|
select(ChannelTag.channel_id)
|
|
.where(ChannelTag.tag_id.in_(ids), visible)
|
|
.group_by(ChannelTag.channel_id)
|
|
.having(func.count(func.distinct(ChannelTag.tag_id)) == len(set(ids)))
|
|
)
|
|
else:
|
|
sub = select(ChannelTag.channel_id).where(ChannelTag.tag_id.in_(ids), visible)
|
|
query = query.where(Video.channel_id.in_(sub))
|
|
|
|
# "No tag" filter (item B): keep only channels the USER has not personally tagged in the channel
|
|
# manager — i.e. channels with no ChannelTag row owned by this user (user_id == me, source "user").
|
|
# System/auto tags (user_id NULL) don't count: this is "channels I haven't tagged yet". It's a plain
|
|
# channel-level narrowing (like channel_ids/date), NOT a tag-category selection, so feed_summary
|
|
# leaves it applied in the base pass and needs no mirror in passes().
|
|
if untagged:
|
|
query = query.where(
|
|
Video.channel_id.notin_(
|
|
select(ChannelTag.channel_id).where(ChannelTag.user_id == user.id)
|
|
)
|
|
)
|
|
|
|
# Content type: Normal / Shorts / Live·Upcoming as a union of enabled types.
|
|
# Explicit Watched/Saved/Hidden views show every type so nothing goes missing.
|
|
explicit_view = show in ("watched", "hidden")
|
|
if not explicit_view:
|
|
type_clauses = []
|
|
if show_normal:
|
|
type_clauses.append(
|
|
and_(Video.is_short.is_(False), Video.live_status.notin_(LIVE_OR_UPCOMING))
|
|
)
|
|
if include_shorts:
|
|
type_clauses.append(Video.is_short.is_(True))
|
|
if include_live:
|
|
type_clauses.append(Video.live_status.in_(LIVE_OR_UPCOMING))
|
|
query = query.where(or_(*type_clauses) if type_clauses else false())
|
|
|
|
if show == "unwatched":
|
|
query = query.where(status_expr.notin_(("watched", "hidden")))
|
|
elif show == "in_progress":
|
|
# Started but not finished: a stored resume position, not yet watched/hidden. INNER-joined
|
|
# (state_required) so the raw state columns are safe and the planner can use video_states.
|
|
query = query.where(
|
|
and_(state.position_seconds > 0, state.status.notin_(("watched", "hidden")))
|
|
)
|
|
elif show == "watched":
|
|
query = query.where(state.status == "watched")
|
|
elif show == "hidden":
|
|
query = query.where(state.status == "hidden")
|
|
else: # all
|
|
query = query.where(status_expr != "hidden")
|
|
|
|
return query, rank_expr
|
|
|
|
|
|
def feed_summary(db: Session, user: User, **params) -> dict:
|
|
"""Single-pass replacement for /feed/count + 3x /facets (PERF-13): the sidebar's total video count
|
|
and every tag's distinct-channel facet count, derived in Python from ONE grouped query instead of
|
|
four full passes over the subscribed set on every filter change.
|
|
|
|
The one query is the base filter with the TAG predicate removed, grouped by channel: it yields every
|
|
channel that has a matching video, with its video count. From that ~few-hundred-row map we compute:
|
|
- count = sum of the video counts of the channels that pass the SELECTED tag filter;
|
|
- facets = per tag, the number of distinct channels that carry it AND pass the drill-down filter
|
|
(the tag's own category dropped, unless it's a conjunctive topic-AND — mirroring the
|
|
old /facets exclude_tag_category logic exactly).
|
|
Tag matching is channel-based, identical to filtered_query's tag block (AND across categories, OR
|
|
within, a category in "and" mode requires ALL its selected tags)."""
|
|
tag_modes = params.get("tag_modes", {})
|
|
|
|
# ONE pass: per-channel video counts for the base filter, the positive tag filter stripped so every
|
|
# candidate channel is present to facet over. The "No tag" (untagged) filter is a plain channel-level
|
|
# narrowing, so it STAYS applied here (not stripped) and the facets naturally drill down within it.
|
|
base, _rank = filtered_query(db, user, **{**params, "tags": []})
|
|
ch_col = base.with_only_columns(Video.channel_id).subquery()
|
|
channel_counts: dict[str, int] = dict(
|
|
db.execute(select(ch_col.c.channel_id, func.count()).group_by(ch_col.c.channel_id)).all()
|
|
)
|
|
if not channel_counts:
|
|
return {"count": 0, "facets": {}}
|
|
|
|
# channel -> its visible tag ids, plus each tag's category (for grouping the facet output).
|
|
visible = or_(ChannelTag.user_id.is_(None), ChannelTag.user_id == user.id)
|
|
channel_tags: dict[str, set[int]] = {}
|
|
tag_category: dict[int, str] = {}
|
|
for cid, tid, cat in db.execute(
|
|
select(ChannelTag.channel_id, ChannelTag.tag_id, Tag.category)
|
|
.join(Tag, Tag.id == ChannelTag.tag_id)
|
|
.where(visible, ChannelTag.channel_id.in_(channel_counts.keys()))
|
|
).all():
|
|
channel_tags.setdefault(cid, set()).add(tid)
|
|
tag_category[tid] = cat
|
|
|
|
# Selected tags grouped by category (read from Tag via the shared helper, so a selected tag that no
|
|
# base channel carries still filters everything out for its category — matching the SQL `IN (empty)`
|
|
# behaviour).
|
|
selected_by_cat: dict[str, set[int]] = {
|
|
cat: set(ids) for cat, ids in tags_by_category(db, params.get("tags") or []).items()
|
|
}
|
|
|
|
def passes(ch_tags: set[int], exclude_cat: str | None) -> bool:
|
|
# The Python mirror of filtered_query's SQL tag block (AND across categories, OR within,
|
|
# topic-AND requires all selected) — keep in lockstep; see the ⚠️ note there. (The "No tag"
|
|
# filter is applied in the base query above, so it needs no mirror here.)
|
|
for cat, sel in selected_by_cat.items():
|
|
if cat == exclude_cat:
|
|
continue
|
|
if tag_modes.get(cat, "or") == "and" and len(sel) > 1:
|
|
if (
|
|
not sel <= ch_tags
|
|
): # conjunctive: channel must carry ALL this category's selected tags
|
|
return False
|
|
elif not (sel & ch_tags): # disjunctive: channel must carry ANY selected tag
|
|
return False
|
|
return True
|
|
|
|
count = sum(
|
|
n for cid, n in channel_counts.items() if passes(channel_tags.get(cid, set()), None)
|
|
)
|
|
|
|
facets: dict[int, int] = {}
|
|
for category in ("language", "topic", "other"):
|
|
# Drop this category's own selections so its chips keep independent counts; a conjunctive
|
|
# topic-AND keeps them (each remaining chip narrows to channels that ALSO have the selected).
|
|
exclude = None if tag_modes.get(category, "or") == "and" else category
|
|
for ch_tags in channel_tags.values():
|
|
if not passes(ch_tags, exclude):
|
|
continue
|
|
for tid in ch_tags:
|
|
if tag_category.get(tid) == category:
|
|
facets[tid] = facets.get(tid, 0) + 1
|
|
return {"count": count, "facets": facets}
|
|
|
|
|
|
def feed_years(db: Session, user: User, **params) -> list[dict]:
|
|
"""Per-year video counts for the current filter context (item 8 / E8·S2): powers the year control's
|
|
quick chips + the "more…" per-year-counts dialog. Every OTHER filter (tags, channels, content type,
|
|
search, watch state, scope) still applies, but the publish-date narrowing itself is DROPPED —
|
|
published_after/before and max_age_days — so the year list is the full set of years available and
|
|
switching years stays a live drill-down (the same principle as the tag facets dropping their own
|
|
category). One grouped query, newest year first."""
|
|
base, _rank = filtered_query(
|
|
db,
|
|
user,
|
|
**{**params, "published_after": None, "published_before": None, "max_age_days": None},
|
|
)
|
|
pub = base.with_only_columns(Video.published_at.label("pub")).subquery()
|
|
# Bucket by UTC year to stay in lockstep with the year filter's boundaries (published_after/before
|
|
# are pinned to UTC in filtered_query). published_at is timestamptz, so a bare extract('year', …)
|
|
# would use the DB session timezone and, on a non-UTC session, bucket a video near a year boundary
|
|
# into a different year than the filter includes — the count and the drill-down feed would disagree.
|
|
year = func.extract("year", func.timezone("UTC", pub.c.pub))
|
|
rows = db.execute(
|
|
select(year.label("y"), func.count()).group_by(year).order_by(year.desc())
|
|
).all()
|
|
return [{"year": int(y), "count": n} for y, n in rows if y is not None]
|
|
|
|
|
|
# --- Keyset (cursor) pagination ---------------------------------------------
|
|
#
|
|
# Each sort is described as an ordered list of key columns (expression + direction
|
|
# + whether it can be NULL + a JSON round-trip kind), always with Video.id as the
|
|
# final, unique tiebreaker. This drives BOTH the ORDER BY and the keyset WHERE so
|
|
# they can never drift apart — the prerequisite for correct cursor paging. Keyset
|
|
# replaces OFFSET/LIMIT: deep scrolling no longer scans-and-discards skipped rows,
|
|
# and the page boundary is stable even as the scheduler ingests new videos mid-scroll
|
|
# (OFFSET would shift and duplicate/skip rows).
|
|
def sort_keys(sort: str, seed: int, rank_expr=None, lang: str | None = None) -> list[dict]:
|
|
"""`lang` orders the title sorts in the reader's alphabet (`collation`); the other keys ignore it."""
|
|
pub = {"expr": Video.published_at, "asc": False, "nullable": False, "kind": "dt"}
|
|
title = collated(Video.title, lang)
|
|
table: dict[str, list[dict]] = {
|
|
"newest": [pub],
|
|
"oldest": [{**pub, "asc": True}],
|
|
"views": [{"expr": Video.view_count, "asc": False, "nullable": True, "kind": "num"}],
|
|
"views_asc": [{"expr": Video.view_count, "asc": True, "nullable": True, "kind": "num"}],
|
|
"duration_desc": [
|
|
{"expr": Video.duration_seconds, "asc": False, "nullable": True, "kind": "num"}
|
|
],
|
|
"duration_asc": [
|
|
{"expr": Video.duration_seconds, "asc": True, "nullable": True, "kind": "num"}
|
|
],
|
|
"title": [{"expr": title, "asc": True, "nullable": True, "kind": "str"}],
|
|
"title_desc": [{"expr": title, "asc": False, "nullable": True, "kind": "str"}],
|
|
"subscribers": [
|
|
{"expr": Channel.subscriber_count, "asc": False, "nullable": True, "kind": "num"}
|
|
],
|
|
"subscribers_asc": [
|
|
{"expr": Channel.subscriber_count, "asc": True, "nullable": True, "kind": "num"}
|
|
],
|
|
# Your per-channel priority (set in the channel manager), newest first within a tier.
|
|
# coalesce keeps it null-safe in "all" scope where unsubscribed channels have no row.
|
|
"priority": [
|
|
{
|
|
"expr": func.coalesce(Subscription.priority, 0),
|
|
"asc": False,
|
|
"nullable": False,
|
|
"kind": "num",
|
|
},
|
|
pub,
|
|
],
|
|
"priority_asc": [
|
|
{
|
|
"expr": func.coalesce(Subscription.priority, 0),
|
|
"asc": True,
|
|
"nullable": False,
|
|
"kind": "num",
|
|
},
|
|
pub,
|
|
],
|
|
# Deterministic per-seed shuffle: the same seed yields the same total order, so
|
|
# the cursor stays valid across pages of one shuffled run.
|
|
"shuffle": [
|
|
{
|
|
"expr": func.md5(func.concat(Video.id, str(seed))),
|
|
"asc": True,
|
|
"nullable": False,
|
|
"kind": "str",
|
|
}
|
|
],
|
|
}
|
|
# Relevance (search-result ranking) only exists when there's a query to rank against;
|
|
# without `rank_expr` it falls through to newest. Ties break by newest, then id.
|
|
if rank_expr is not None:
|
|
table["relevance"] = [
|
|
{"expr": rank_expr, "asc": False, "nullable": False, "kind": "num"},
|
|
pub,
|
|
]
|
|
return table.get(sort) or table["newest"]
|
|
|
|
|
|
# Relevance ranks only the newest RANK_WINDOW matches (E31): ts_rank reads every match's tsvector, and
|
|
# a broad word matches most of the library ("a" ranked 193k rows in 6.8 s on prod, against 1.1 s to just
|
|
# find them). The rows past the window keep rank 0, so they follow the ranked ones newest first; a word
|
|
# that broad has no meaningful relevance order anyway, and a specific one fits the window whole.
|
|
RANK_WINDOW = 2000
|
|
|
|
|
|
def rank_window_start(db: Session, query: Select) -> datetime | None:
|
|
"""The published_at the ranked window starts at: the RANK_WINDOW-th newest match, None when fewer
|
|
match (rank them all). The newest-first walk stops after RANK_WINDOW matches (40 ms for "a")."""
|
|
return db.scalar(
|
|
query.with_only_columns(Video.published_at)
|
|
.order_by(Video.published_at.desc())
|
|
.offset(RANK_WINDOW - 1)
|
|
.limit(1)
|
|
)
|
|
|
|
|
|
def windowed_rank(rank_expr, since: datetime | None):
|
|
"""`rank_expr` inside the window, 0 past it. CASE evaluates ts_rank only for the rows inside."""
|
|
if since is None:
|
|
return rank_expr
|
|
return case((Video.published_at >= since, rank_expr), else_=0)
|
|
|
|
|
|
def windowed_relevance_rows(
|
|
db: Session, query: Select, keys: list[dict], window: datetime, cursor_id: str | None, n: int
|
|
) -> list:
|
|
"""Up to `n` rows of the windowed relevance order, read as its two halves: the window ranked, then
|
|
the rest newest first (every window row sorts before every row past it). As one ORDER BY the
|
|
planner computes the CASE for every match and sorts them all (2.7 s for "a" on prod); split, each
|
|
half rides an index (80 ms + 5 ms). With a cursor, `keys` carry its values (keyset_predicate)."""
|
|
pub_key = keys[1] # relevance: [rank, published_at]
|
|
rows: list = []
|
|
if cursor_id is None or pub_key["value"] >= window:
|
|
inside = query.where(Video.published_at >= window).order_by(*order_by(keys))
|
|
if cursor_id is not None:
|
|
inside = inside.where(keyset_predicate(keys, cursor_id))
|
|
rows = list(db.execute(inside.limit(n)).all())
|
|
if len(rows) < n:
|
|
newest = [pub_key] # past the window every rank is 0: the order is newest first
|
|
rest = query.where(Video.published_at < window).order_by(*order_by(newest))
|
|
if cursor_id is not None:
|
|
rest = rest.where(keyset_predicate(newest, cursor_id))
|
|
rows += db.execute(rest.limit(n - len(rows))).all()
|
|
return rows
|
|
|
|
|
|
def order_by(keys: list[dict]):
|
|
cols = []
|
|
for k in keys:
|
|
c = k["expr"].asc() if k["asc"] else k["expr"].desc()
|
|
cols.append(c.nulls_last() if k["nullable"] else c)
|
|
cols.append(Video.id.asc()) # unique final tiebreaker
|
|
return cols
|
|
|
|
|
|
def keyset_predicate(keys: list[dict], cursor_id: str):
|
|
"""WHERE clause selecting rows strictly AFTER the cursor in the sort order.
|
|
|
|
Lexicographic over the key columns then id, NULL-aware for `nulls_last`: a NULL
|
|
sorts after every non-NULL value, so a cursor sitting on a non-NULL value must
|
|
also pull in the NULL group, and equality treats NULL == NULL.
|
|
|
|
Built recursively so that for a NON-nullable leading key we can prepend a non-strict
|
|
range bound (`col <= v` for desc / `col >= v` for asc) as an AND conjunct — logically
|
|
a no-op over the OR of strict-after branches, but it lets the planner bound the index
|
|
scan instead of scanning from the top and filtering every earlier row (PERF-15: deep
|
|
`newest` pages went 51.8k -> ~0.5k buffers). A NULLABLE key can't take that bound (the
|
|
NULL group sorts after every value and `col <= v` would drop it), so it keeps the pure
|
|
NULL-aware OR form."""
|
|
|
|
def eq(k):
|
|
return k["expr"].is_(None) if k["value"] is None else k["expr"] == k["value"]
|
|
|
|
def after(k):
|
|
if k["value"] is None:
|
|
return false() # nulls are last; nothing sorts strictly after them
|
|
cmp = k["expr"] > k["value"] if k["asc"] else k["expr"] < k["value"]
|
|
return or_(cmp, k["expr"].is_(None)) if k["nullable"] else cmp
|
|
|
|
def rec(i: int):
|
|
if i == len(keys):
|
|
return Video.id > cursor_id # all keys tied -> break by the unique id
|
|
k = keys[i]
|
|
tail = or_(after(k), and_(eq(k), rec(i + 1)))
|
|
if not k["nullable"] and k["value"] is not None:
|
|
bound = k["expr"] >= k["value"] if k["asc"] else k["expr"] <= k["value"]
|
|
return and_(bound, tail)
|
|
return tail
|
|
|
|
return rec(0)
|
|
|
|
|
|
def encode_cursor(keys: list[dict], row, window: datetime | None = None) -> str:
|
|
"""`window` is the relevance window's start: the cursor carries it, so every page ranks against the
|
|
window page one did. Recomputed per page, a video ingested mid-scroll would push the window's oldest
|
|
row to rank 0, behind the cursor, and the reader would see it twice."""
|
|
vals = []
|
|
for i, k in enumerate(keys):
|
|
v = getattr(row, f"_k{i}")
|
|
vals.append(v.isoformat() if (v is not None and k["kind"] == "dt") else v)
|
|
body: dict = {"k": vals, "id": row.id}
|
|
if window is not None:
|
|
body["w"] = window.isoformat()
|
|
payload = json.dumps(body)
|
|
return base64.urlsafe_b64encode(payload.encode()).decode()
|
|
|
|
|
|
def _cursor_dt(v):
|
|
"""A cursor's timestamp: aware or refused. The server writes only aware ones (timestamptz), and
|
|
windowed_relevance_rows compares them in Python, where a naive one beside an aware one raises — a
|
|
500 for a hand-made cursor instead of the 400 (E31 review 2)."""
|
|
if v is None:
|
|
return None
|
|
dt = datetime.fromisoformat(v)
|
|
if dt.tzinfo is None:
|
|
raise ValueError("naive cursor timestamp")
|
|
return dt
|
|
|
|
|
|
def decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str, datetime | None]:
|
|
try:
|
|
payload = json.loads(base64.urlsafe_b64decode(cursor.encode()))
|
|
raw = payload["k"]
|
|
if len(raw) != len(keys):
|
|
raise ValueError("key arity mismatch")
|
|
vals = [_cursor_dt(v) if k["kind"] == "dt" else v for k, v in zip(keys, raw, strict=True)]
|
|
for k, v in zip(keys, vals, strict=True):
|
|
if v is None and not k["nullable"]:
|
|
raise ValueError("null value for a non-nullable key")
|
|
window = payload.get("w")
|
|
return vals, payload["id"], _cursor_dt(window) if window else None
|
|
except (binascii.Error, json.JSONDecodeError, KeyError, ValueError, TypeError) as exc:
|
|
# Framework-free: the route handler translates this into an HTTP 400 so the
|
|
# read-model stays web-layer-agnostic (see routes/feed.get_feed).
|
|
raise ValueError("Invalid feed cursor") from exc
|