perf(scenes): lista serwisu od najnowszych prosto z indeksu

Ekran serwisu (/scenes?origin=tube:X&sort=release_date) dla serwisow,
ktorych scen nie ma wsrod najnowszych w katalogu (perverzija, hdporn.gg,
tubepornclassic, yesporn), sortowal wszystkie ich sceny: 9-19 s na
zimnym cache, 1-4 s na cieplym, 39 wejsc w 3 h.

Wzorem scene_tags.sort_date (GOON-64): data sortu powielona w
playback_sources i pilnowana wyzwalaczami (0048), wypelniona skryptem,
indeks (origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL
(0049). API dolacza DISTINCT ON zgodny z indeksem, bo 190 tys. scen ma
dwa zywe zrodla z tego samego serwisu. Po zmianie 0,03-0,24 s, takze
dalsze strony.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

alembic/versions/20260927_0048_playback_sources_sort_date.py

@@ -0,0 +1,86 @@
+"""Data sortowania sceny powielona w playback_sources (lista scen jednego serwisu).
+
+Ekran serwisu w apce (`/scenes?origin=tube:X&sort=release_date`) sortuje po dacie z
+`scenes`, a filtruje po `playback_sources`. Dla serwisow, ktorych scen nie ma wsrod
+najnowszych w katalogu (perverzija, hdporn.gg, tubepornclassic, yesporn), przejscie po
+ix_scenes_effective_date_desc nie znajduje strony w 1 s, wiec API bierze wszystkie id
+scen serwisu (20-115 tys.) i sortuje je po dacie. Na rozgrzanym cache 1-4 s, na zimnym
+9-19 s (log API 2026-09-27, 39 wejsc w 3 h). Ten sam problem co przy tagach (0042).
+
+`sort_date` = to samo wyrazenie co w 0042 i w `_effective_date_desc()`. Indeks
+(origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL z migracji 0049 daje od
+razu najnowsze sceny serwisu.
+
+Utrzymanie w bazie, bo do playback_sources pisze ingest, merge (przenosi zrodla miedzy
+scenami) i skrypty:
+  - BEFORE INSERT / UPDATE OF scene_id na playback_sources: data z wiersza sceny,
+  - wyzwalacz scen z 0042 poprawia teraz tez zrodla tej sceny.
+
+Kolumna nullable i bez DEFAULT, wiec ALTER jest natychmiastowy. Istniejace 4,4 mln
+wierszy wypelnia scripts/backfill_playback_sort_date.py, dopiero potem indeks (0049)
+i przelaczenie zapytania.
+
+Revision ID: 0048_playback_sort_date
+Revises: 0047_tag_vertical_counts
+Create Date: 2026-09-27
+"""
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0048_playback_sort_date"
+down_revision: str | None = "0047_tag_vertical_counts"
+branch_labels = None
+depends_on = None
+
+_SORT_DATE = "COALESCE({t}.release_date, ({t}.created_at AT TIME ZONE 'UTC')::date)"
+
+
+def upgrade() -> None:
+    op.execute("SET LOCAL lock_timeout = '5s'")
+    op.execute("ALTER TABLE playback_sources ADD COLUMN IF NOT EXISTS sort_date date")
+    op.execute(f"""
+        CREATE OR REPLACE FUNCTION playback_sources_fill_sort_date() RETURNS trigger
+        LANGUAGE plpgsql AS $$
+        BEGIN
+            SELECT {_SORT_DATE.format(t='s')} INTO NEW.sort_date
+              FROM scenes s WHERE s.id = NEW.scene_id;
+            RETURN NEW;
+        END $$
+    """)
+    op.execute("""
+        CREATE TRIGGER trg_playback_sources_fill_sort_date
+        BEFORE INSERT OR UPDATE OF scene_id ON playback_sources
+        FOR EACH ROW EXECUTE FUNCTION playback_sources_fill_sort_date()
+    """)
+    # Ta sama funkcja co w 0042, z dopisanym playback_sources. Wyzwalacz na scenes
+    # zostaje bez zmian.
+    op.execute(f"""
+        CREATE OR REPLACE FUNCTION scenes_propagate_sort_date() RETURNS trigger
+        LANGUAGE plpgsql AS $$
+        DECLARE d date := {_SORT_DATE.format(t='NEW')};
+        BEGIN
+            UPDATE scene_tags SET sort_date = d
+             WHERE scene_id = NEW.id AND sort_date IS DISTINCT FROM d;
+            UPDATE playback_sources SET sort_date = d
+             WHERE scene_id = NEW.id AND sort_date IS DISTINCT FROM d;
+            RETURN NULL;
+        END $$
+    """)
+
+
+def downgrade() -> None:
+    op.execute("SET LOCAL lock_timeout = '5s'")
+    op.execute(f"""
+        CREATE OR REPLACE FUNCTION scenes_propagate_sort_date() RETURNS trigger
+        LANGUAGE plpgsql AS $$
+        DECLARE d date := {_SORT_DATE.format(t='NEW')};
+        BEGIN
+            UPDATE scene_tags SET sort_date = d
+             WHERE scene_id = NEW.id AND sort_date IS DISTINCT FROM d;
+            RETURN NULL;
+        END $$
+    """)
+    op.execute("DROP TRIGGER IF EXISTS trg_playback_sources_fill_sort_date ON playback_sources")
+    op.execute("DROP FUNCTION IF EXISTS playback_sources_fill_sort_date()")
+    op.execute("ALTER TABLE playback_sources DROP COLUMN IF EXISTS sort_date")

alembic/versions/20260927_0049_playback_origin_sort_idx.py

@@ -0,0 +1,35 @@
+"""Indeks "najnowsze sceny serwisu" (kolumna z 0048).
+
+(origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL: lista scen serwisu,
+sort=release_date, idzie po tym indeksie od najnowszej i konczy po LIMIT, niezaleznie od
+tego, gdzie w calym katalogu leza sceny serwisu. Uzasadnienie i pomiary w 0048.
+
+Zakladac DOPIERO po wypelnieniu `sort_date` (scripts/backfill_playback_sort_date.py),
+z tego samego powodu co 0043. CONCURRENTLY w autocommit_block; nieudany zostawia indeks
+INVALID, wtedy DROP INDEX i ponownie.
+
+Revision ID: 0049_playback_origin_sort_idx
+Revises: 0048_playback_sort_date
+Create Date: 2026-09-27
+"""
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0049_playback_origin_sort_idx"
+down_revision: str | None = "0048_playback_sort_date"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.get_context().autocommit_block():
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_playback_sources_origin_sort "
+            "ON playback_sources (origin, sort_date DESC, scene_id DESC) WHERE dead_at IS NULL"
+        )
+
+
+def downgrade() -> None:
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_playback_sources_origin_sort")

app/api/scenes.py

@@ -759,7 +759,14 @@ def list_scenes(
         # stwierdzic: 148 s na produkcji (pomiar 2026-09-16). Zwieramy od razu.
         base = base.where(false() if origin else ~live_playback_exists())
 
-    if origin:
+    # Ekran serwisu od najnowszych: sceny serwisu bierzemy wprost z indeksu
+    # ix_playback_sources_origin_sort (data sortu powielona w zrodle, migracje 0048/0049)
+    # zamiast szukac ich w calym katalogu po dacie. Wtedy JOIN w galezi sortu zastepuje
+    # ponizszy EXISTS. Z fraza albo tagiem zostaja dotychczasowe sciezki.
+    origin_sort = bool(
+        sort == "release_date" and origin and ":" in origin and tag_sort_id is None and not q
+    )
+    if origin and not origin_sort:
         # Pelny origin z ekranu strony ('tube:hdporngg') porownujemy wprost: rownosc idzie
         # po ix_playback_sources_origin_live, a ILIKE '%...%' przeszukiwal kazde zrodlo
         # kazdej kandydujacej sceny (wyszukiwanie w obrebie strony 4,6 s, Sentry Slow DB
@@ -911,6 +918,21 @@ def list_scenes(
             ordered = base.join(
                 st, and_(st.scene_id == Scene.id, st.tag_id == tag_sort_id)
             ).order_by(st.sort_date.desc(), st.scene_id.desc())
+        elif origin_sort:
+            # Scena moze miec dwa zywe zrodla z tego samego serwisu (190 tys. par), a oba
+            # niosa te sama date, wiec DISTINCT ON zgodny z porzadkiem indeksu zbija je w
+            # jeden wiersz bez sortowania: skan dalej konczy sie po LIMIT. Pomiar i
+            # uzasadnienie w migracji 0048 (perverzija 13,7 s na zimnym cache).
+            ps_sort = (
+                select(PlaybackSource.scene_id, PlaybackSource.sort_date)
+                .where(PlaybackSource.origin == origin, PlaybackSource.dead_at.is_(None))
+                .distinct(PlaybackSource.sort_date, PlaybackSource.scene_id)
+                .order_by(PlaybackSource.sort_date.desc(), PlaybackSource.scene_id.desc())
+                .subquery("origin_sort")
+            )
+            ordered = base.join(ps_sort, ps_sort.c.scene_id == Scene.id).order_by(
+                ps_sort.c.sort_date.desc(), ps_sort.c.scene_id.desc()
+            )
         else:
             ordered = base.order_by(
                 _effective_date_desc(), Scene.created_at.desc(), Scene.id.desc()
@@ -1054,7 +1076,7 @@ def list_scenes(
                     Scene.id
                     == any_(bindparam("origin_scene_ids", origin_ids, type_=ARRAY(PG_UUID(as_uuid=True))))
                 )
-        if rows is None and origin and not q and sort in ("created_at", "release_date"):
+        if rows is None and origin and not q and not origin_sort and sort in ("created_at", "release_date"):
             walked = _search_walk_ids(session, ordered, _off, _want, _ORIGIN_LIST_WALK_BUDGET_MS)
             if walked is not None:
                 rows = _scenes_in_order(session, walked)

app/models/playback_source.py

@@ -8,9 +8,10 @@ nie tworzył duplikatów.
 from __future__ import annotations
 
 import uuid
-from datetime import datetime
+from datetime import date, datetime
 
 from sqlalchemy import (
+    Date,
     DateTime,
     ForeignKey,
     Integer,
@@ -66,3 +67,8 @@ class PlaybackSource(UUIDPKMixin, TimestampMixin, Base):
     API filtruje takie sources z listy. None = aktywny."""
 
     dead_reason: Mapped[str | None] = mapped_column(String(512), nullable=True)
+
+    # Data sortowania sceny (premiera albo dzien dodania), powielona tu pod indeks
+    # "najnowsze sceny serwisu" (migracje 0048 i 0049). Pisze ja WYZWALACZ w bazie,
+    # nie aplikacja: wartosc podana przy INSERT i tak zostanie nadpisana.
+    sort_date: Mapped[date | None] = mapped_column(Date)

scripts/backfill_playback_sort_date.py

@@ -0,0 +1,76 @@
+"""Wypelnia playback_sources.sort_date w istniejacych wierszach (migracja 0048).
+
+Kopia scripts/backfill_scene_tags_sort_date.py dla playback_sources: partiami po zakresach
+stron heapu, commit po kazdej partii, VACUUM co kilka partii, pauza oddaje dysk apce.
+Nowe wiersze w trakcie wypelnia juz wyzwalacz z 0048. Wznawialny: rusza tylko wiersze z
+sort_date IS NULL, na koncu dobiera resztki.
+
+  docker compose run -d --name backfill_ps_sort worker \
+      python3 -u /srv/scripts/backfill_playback_sort_date.py [stron_na_partie] [pauza_s]
+Po nim: alembic upgrade (0049, indeks) i dopiero wtedy kod, ktory z indeksu korzysta.
+"""
+from __future__ import annotations
+
+import sys
+import time
+
+from sqlalchemy import text
+
+from app.db import engine
+
+PAGES = int(sys.argv[1]) if len(sys.argv) > 1 else 1000
+PAUSE = float(sys.argv[2]) if len(sys.argv) > 2 else 10.0
+VACUUM_EVERY = 8
+SORT_DATE = "COALESCE(s.release_date, (s.created_at AT TIME ZONE 'UTC')::date)"
+
+
+def rel_pages() -> int:
+    with engine.connect() as c:
+        return int(c.execute(text(
+            "SELECT pg_relation_size('playback_sources') / current_setting('block_size')::int"
+        )).scalar())
+
+
+def vacuum(analyze: bool = False) -> None:
+    with engine.connect().execution_options(isolation_level="AUTOCOMMIT") as c:
+        c.execute(text("VACUUM (ANALYZE) playback_sources" if analyze else "VACUUM playback_sources"))
+
+
+t0 = time.time()
+total = 0
+page = 0
+batch = 0
+pages = rel_pages()
+print(f"playback_sources: {pages} stron, partia {PAGES} stron, pauza {PAUSE} s", flush=True)
+while page < pages:
+    tb = time.time()
+    with engine.begin() as c:
+        c.execute(text("SET LOCAL statement_timeout = '600s'"))
+        n = c.execute(text(f"""
+            UPDATE playback_sources ps SET sort_date = {SORT_DATE}
+              FROM scenes s
+             WHERE s.id = ps.scene_id AND ps.sort_date IS NULL
+               AND ps.ctid >= format('(%s,0)', CAST(:p0 AS int))::tid
+               AND ps.ctid <  format('(%s,0)', CAST(:p1 AS int))::tid
+        """), {"p0": page, "p1": page + PAGES}).rowcount
+    total += n
+    page += PAGES
+    batch += 1
+    if batch % VACUUM_EVERY == 0:
+        vacuum()
+        pages = rel_pages()  # tabela moze urosnac, zanim VACUUM zwolni miejsce
+    print(f"partia {batch}: strony do {page}/{pages}, {n} wierszy w {time.time() - tb:.1f} s, "
+          f"razem {total}, {time.time() - t0:.0f} s", flush=True)
+    if n:
+        time.sleep(PAUSE)
+
+with engine.begin() as c:
+    rest = c.execute(text(f"""
+        UPDATE playback_sources ps SET sort_date = {SORT_DATE}
+          FROM scenes s WHERE s.id = ps.scene_id AND ps.sort_date IS NULL
+    """)).rowcount
+print(f"resztki: {rest}", flush=True)
+vacuum(analyze=True)
+with engine.connect() as c:
+    left = c.execute(text("SELECT count(*) FROM playback_sources WHERE sort_date IS NULL")).scalar()
+print(f"KONIEC: {total + rest} wierszy, puste po wszystkim: {left}, {time.time() - t0:.0f} s", flush=True)