siftlode/backend/tests/test_feed_relevance_window.py

"""Relevance ranks only the newest RANK_WINDOW matches, 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.
"""

import base64
import json
from datetime import datetime, timezone

import pytest

from app import feedquery
from app.models import User
from app.catalog.models import Channel, Subscription, 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


@pytest.mark.parametrize(
    "payload",
    [
        {"k": [5, None], "id": "m1", "w": "2024-01-01T00:00:00+00:00"},  # null published_at
        {"k": [5, "2024-01-03T00:00:00+00:00"], "id": "m1", "w": "2024-01-01T00:00:00"},  # naive w
        {"k": [5, "2024-01-03T00:00:00"], "id": "m1", "w": "2024-01-01T00:00:00+00:00"},  # naive k
    ],
)
def test_a_hand_made_relevance_cursor_is_refused_not_a_crash(db, as_user, payload):
    # The page halves compare the cursor's timestamp with the window's in Python: a null or a naive
    # one raised there, past the route's 400 — a 500.
    client = as_user(_seed(db))
    cursor = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode()
    resp = client.get(f"/api/feed?q=foo&sort=relevance&limit=2&cursor={cursor}")
    assert resp.status_code == 400, resp.text