diff --git a/backend/app/feedquery.py b/backend/app/feedquery.py index 1023ac2d..20665756 100644 --- a/backend/app/feedquery.py +++ b/backend/app/feedquery.py @@ -12,7 +12,7 @@ import binascii import json from datetime import date, datetime, timedelta, timezone -from sqlalchemy import BigInteger, Select, and_, cast, false, func, or_, select +from sqlalchemy import BigInteger, Select, and_, case, cast, false, func, or_, select from sqlalchemy.orm import Session, aliased from app.collation import collated @@ -265,7 +265,8 @@ def filtered_query( # 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. + # 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 @@ -543,6 +544,54 @@ def sort_keys(sort: str, seed: int, rank_expr=None, lang: str | None = None) -> 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: @@ -589,16 +638,22 @@ def keyset_predicate(keys: list[dict], cursor_id: str): return rec(0) -def encode_cursor(keys: list[dict], row) -> str: +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) - payload = json.dumps({"k": vals, "id": row.id}) + 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 decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str]: +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"] @@ -608,7 +663,8 @@ def decode_cursor(keys: list[dict], cursor: str) -> tuple[list, str]: datetime.fromisoformat(v) if (v is not None and k["kind"] == "dt") else v for k, v in zip(keys, raw) ] - return vals, payload["id"] + window = payload.get("w") + return vals, payload["id"], datetime.fromisoformat(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). diff --git a/backend/app/routes/feed.py b/backend/app/routes/feed.py index 4ae00d95..4e905e0c 100644 --- a/backend/app/routes/feed.py +++ b/backend/app/routes/feed.py @@ -19,8 +19,11 @@ from app.feedquery import ( filtered_query, keyset_predicate, order_by, + rank_window_start, serialize, sort_keys, + windowed_rank, + windowed_relevance_rows, ) from app.models import ( POSITION_MAX_S, @@ -109,30 +112,45 @@ def get_feed( db: Session = Depends(get_db), ) -> dict: query, rank_expr = filtered_query(db, user, **params) - keys = sort_keys(sort, seed, rank_expr, lang_for_user(user)) - - # Expose each key column on the row so we can build the next cursor from the last item. - query = query.add_columns(*[k["expr"].label(f"_k{i}") for i, k in enumerate(keys)]) - query = query.order_by(*order_by(keys)) + lang = lang_for_user(user) + keys = sort_keys(sort, seed, rank_expr, lang) + vals, cursor_id, window = None, None, None if cursor: # The read-model raises a plain ValueError on a malformed cursor; translate it to the # HTTP 400 here so feedquery stays web-framework-free. try: - vals, cursor_id = decode_cursor(keys, cursor) + vals, cursor_id, window = decode_cursor(keys, cursor) except ValueError: # `from None`: the ValueError is an expected bad-input signal, not an error to chain. raise refused(400, "invalid_cursor") from None + ranked = sort == "relevance" and rank_expr is not None + if ranked and not cursor: + window = rank_window_start(db, query) + if ranked and window is not None: + keys = sort_keys(sort, seed, windowed_rank(rank_expr, window), lang) + + # Expose each key column on the row so we can build the next cursor from the last item. + query = query.add_columns(*[k["expr"].label(f"_k{i}") for i, k in enumerate(keys)]) + if vals is not None: for k, v in zip(keys, vals): k["value"] = v - query = query.where(keyset_predicate(keys, cursor_id)) - - rows = db.execute(query.limit(limit + 1)).all() + if ranked and window is not None: + rows = windowed_relevance_rows(db, query, keys, window, cursor_id, limit + 1) + else: + query = query.order_by(*order_by(keys)) + if cursor_id is not None: + query = query.where(keyset_predicate(keys, cursor_id)) + rows = db.execute(query.limit(limit + 1)).all() has_more = len(rows) > limit page = rows[:limit] return { "items": [serialize(r) for r in page], - "next_cursor": encode_cursor(keys, page[-1]) if (has_more and page) else None, + "next_cursor": ( + encode_cursor(keys, page[-1], window if ranked else None) + if (has_more and page) + else None + ), "has_more": has_more, "limit": limit, } diff --git a/backend/tests/test_feed_relevance_window.py b/backend/tests/test_feed_relevance_window.py new file mode 100644 index 00000000..e8e34881 --- /dev/null +++ b/backend/tests/test_feed_relevance_window.py @@ -0,0 +1,114 @@ +"""Relevance ranks only the newest RANK_WINDOW matches (E31), DB-lane. + +A broad word ("a") matched most of the prod library and ts_rank read every match's tsvector: 6.8 s a page. +The ranking now covers the newest RANK_WINDOW matches and the rest follow newest first. The window's start +rides the cursor, so a video ingested mid-scroll cannot push a row behind the cursor and repeat it. +""" + +from datetime import datetime, timezone + +import pytest + +from app import feedquery +from app.models import Channel, Subscription, User, Video + + +def _video(vid: str, day: int, title: str) -> Video: + return Video( + id=vid, + channel_id="UCw", + published_at=datetime(2024, 1, day, tzinfo=timezone.utc), + title=title, + is_short=False, + live_status="none", + via_search=False, + via_explore=False, + ) + + +def _seed(db) -> User: + db.add(Channel(id="UCw", title="Window")) + user = User(email="window@example.com", email_verified=True) + db.add(user) + db.flush() + db.add(Subscription(user_id=user.id, channel_id="UCw", priority=0, hidden=False)) + # m1 is the oldest and the strongest match, m4 the strongest of the newest three; the rest tie on one + # title "foo" each (ts_rank counts occurrences, it does not normalise by length). + db.add_all( + [ + _video("m1", 1, "foo foo foo foo foo"), + _video("m2", 2, "foo and some other words here"), + _video("m3", 3, "foo plus some filler words"), + _video("m4", 4, "foo foo foo"), + _video("m5", 5, "foo with a longer title"), + _video("m6", 6, "foo with yet another title"), + ] + ) + db.commit() + return user + + +def _ids(client, cursor=None, limit=50): + url = f"/api/feed?q=foo&sort=relevance&limit={limit}" + if cursor: + url += f"&cursor={cursor}" + resp = client.get(url) + assert resp.status_code == 200, resp.text + body = resp.json() + return [it["id"] for it in body["items"]], body["next_cursor"] + + +def test_a_specific_word_ranks_every_match(db, as_user): + client = as_user(_seed(db)) + ids, _ = _ids(client) + assert ids[:2] == ["m1", "m4"], ids + + +def test_a_broad_word_ranks_the_newest_window_and_the_rest_follow_newest_first( + db, as_user, monkeypatch +): + monkeypatch.setattr(feedquery, "RANK_WINDOW", 3) + client = as_user(_seed(db)) + ids, _ = _ids(client) + assert ids[0] == "m4", ids # the strongest of the newest three + assert sorted(ids[:3]) == ["m4", "m5", "m6"], ids + assert ids[3:] == ["m3", "m2", "m1"], ids # past the window: newest first, m1 not ranked + + +def test_paging_keeps_the_window_page_one_ranked_against(db, as_user, monkeypatch): + monkeypatch.setattr(feedquery, "RANK_WINDOW", 3) + client = as_user(_seed(db)) + full, _ = _ids(client) + + first, cursor = _ids(client, limit=2) + assert first == full[:2] + # A newer match lands mid-scroll. Ranked against a recomputed window, m4 (shown on page one) would + # fall out of it to rank 0, sort behind the cursor and come back on page two. + db.add(_video("m7", 7, "foo arriving while the reader scrolls")) + db.commit() + + walked = list(first) + while cursor: + page, cursor = _ids(client, cursor, limit=2) + walked.extend(page) + assert len(walked) == len(set(walked)), walked + assert set(full) <= set(walked), walked + + +@pytest.mark.parametrize( + ("window", "limit"), + # 3: the page is read as its two halves (the window ranked, the rest newest first), and pages of 2 + # and 3 cross the window's edge mid-page and exactly at it. 2000: one ranked query. + [(3, 2), (3, 3), (2000, 2)], +) +def test_cursor_walk_matches_the_unpaged_order(db, as_user, monkeypatch, window, limit): + monkeypatch.setattr(feedquery, "RANK_WINDOW", window) + client = as_user(_seed(db)) + full, _ = _ids(client) + if window == 3: + assert full == ["m4", "m6", "m5", "m3", "m2", "m1"] + walked, cursor = _ids(client, limit=limit) + while cursor: + page, cursor = _ids(client, cursor, limit=limit) + walked.extend(page) + assert walked == full