perf(api): liczniki +N ulubionych z cache i indeksem, podobne sceny z najrzadszych tagow

- /favorites i /favorites/studios: liczniki w cache procesu (swiezy 5 min, potem
  odswiezanie w tle), mark_seen zeruje od razu; parametr counts=false dla ekranow,
  ktore chca tylko stan gwiazdki
- licznik studiow idzie indeksem ix_scenes_studio_effective_date_desc (migracja 0038)
  i konczy po 200 scenach na studio: 10-25 s -> kilkanascie ms
- okno licznika sortuje tak jak lista (data premiery albo dzien dodania), wczesniej
  rozjechane od 0033
- similar_scenes: pula z 4 najrzadszych tagow zamiast grupowania po wszystkich
  (do 5 s -> ponizej 0,4 s)
- enrich-tags: tagi jednym zapytaniem po slugach i jeden INSERT
- /scene-favorites: serializer wsadowy

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

alembic/versions/20260915_0038_scenes_studio_effective_date_idx.py

@@ -0,0 +1,37 @@
+"""Indeks pod liste scen studia i licznik "+N" ulubionych studiow.
+
+Oba sortuja po dacie premiery albo, gdy jej nie ma, dniu trafienia do katalogu (to samo
+wyrazenie co ix_scenes_effective_date_desc z 0033), w obrebie jednego studia. Bez tego
+Postgres czytal WSZYSTKIE sceny studia (Brazzers ma ich dziesiatki tysiecy), sprawdzal
+kazdej playback i sortowal, a licznik dla 209 ulubionych studiow trwal 10-25 s. Z tym
+indeksem idzie po kolei i konczy po 200 widocznych scenach na studio.
+
+CONCURRENTLY w autocommit_block, jak w 0035. Nieudany CONCURRENTLY zostawia indeks
+INVALID; wtedy DROP INDEX i ponownie.
+
+Revision ID: 0038_scenes_studio_effdate_idx
+Revises: 0037_clip_feed
+Create Date: 2026-09-15
+"""
+from __future__ import annotations
+
+from alembic import op
+
+revision = "0038_scenes_studio_effdate_idx"
+down_revision: str | None = "0037_clip_feed"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.get_context().autocommit_block():
+        op.execute(
+            "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_scenes_studio_effective_date_desc "
+            "ON scenes (studio_id, COALESCE(release_date, (created_at AT TIME ZONE 'UTC')::date) DESC, "
+            "created_at DESC, id DESC) WHERE studio_id IS NOT NULL"
+        )
+
+
+def downgrade() -> None:
+    with op.get_context().autocommit_block():
+        op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_scenes_studio_effective_date_desc")

app/api/favorites.py

@@ -21,18 +21,21 @@ Endpointy (studios):
 """
 from __future__ import annotations
 
+import logging
+import threading
+import time
 import uuid
 from datetime import UTC, datetime
 from typing import Annotated
 
-from fastapi import APIRouter, Depends, HTTPException, Path, status
+from fastapi import APIRouter, Depends, HTTPException, Path, Query, status
 from pydantic import BaseModel
 from sqlalchemy import exists, func, select
 from sqlalchemy.orm import Session
 
 from app.api.device import get_device_id
 from app.auth import require_api_key
-from app.db import get_session
+from app.db import get_session, session_scope
 from app.models.favorite_movie import FavoriteMovie
 from app.models.favorite_performer import FavoritePerformer
 from app.models.favorite_site import FavoriteSite
@@ -42,6 +45,8 @@ from app.models.performer import Performer
 from app.models.scene import Scene, ScenePerformer
 from app.models.studio import Studio
 
+log = logging.getLogger(__name__)
+
 router = APIRouter(
     prefix="/favorites", tags=["favorites"], dependencies=[Depends(require_api_key)]
 )
@@ -61,55 +66,71 @@ def _new_counts(session: Session, device_id: str, *, kind: str) -> dict:
     """Policz per-favorite "+N nowych" = sceny created_at > last_seen_at, ale TYLKO wśród
     tych które user zobaczy na liście: te same filtry widoczności co list_scenes (żywy
     playback + blacklist + stub) i to samo okno (top-_FAVORITES_PAGE_CAP pod sortem
-    release_date desc), z pominięciem backfillu. Liczy w SQL (count(*) FILTER), zwraca
+    `release_date` z list_scenes, czyli data premiery albo dzień dodania), z pominięciem
-    {group_id: n}, bez streamowania N*200 wierszy do Pythona.
+    backfillu. Liczy w SQL (count(*) FILTER), zwraca {group_id: n}.
 
     kind="performer": grupuje po ScenePerformer.performer_id, join favorite_performers.
     kind="studio":    grupuje po Scene.studio_id, join favorite_studios. Studia potrzebują
                       odsiewu stub (mogą nie mieć performera); performerki nie (mają).
     """
-    from sqlalchemy import and_, func
+    from sqlalchemy import and_, func, true
 
-    from app.api.scenes import blacklist_clauses, live_playback_exists, stub_exclusion_clause
+    from app.api.scenes import (
+        _effective_date_desc,
+        blacklist_clauses,
+        live_playback_exists,
+        stub_exclusion_clause,
+    )
 
     clauses = [live_playback_exists(), *blacklist_clauses(session, device_id)]
+    list_order = (_effective_date_desc(), Scene.created_at.desc(), Scene.id.desc())
 
-    if kind == "performer":
+    if kind == "studio":
-        gid = ScenePerformer.performer_id
+        # Po kolei indeksem ix_scenes_studio_effective_date_desc (migracja 0038) i stop po
-        last_seen = FavoritePerformer.last_seen_at
+        # 200 widocznych scenach na studio. Okno row_number() czytało wszystkie sceny
+        # wszystkich ulubionych studiów (324 tys. u najcięższego urządzenia, 10-25 s).
+        clauses.append(stub_exclusion_clause())
+        top = (
+            select(Scene.created_at.label("created_at"), Scene.backfill.label("backfill"))
+            .where(Scene.studio_id == FavoriteStudio.studio_id, *clauses)
+            .order_by(*list_order)
+            .limit(_FAVORITES_PAGE_CAP)
+            .lateral("top")
+        )
+        rows = session.execute(
+            select(
+                FavoriteStudio.studio_id,
+                func.count().filter(
+                    and_(
+                        top.c.backfill.is_(False),
+                        top.c.created_at > FavoriteStudio.last_seen_at,
+                    )
+                ),
+            )
+            .select_from(FavoriteStudio)
+            .join(top, true())
+            .where(FavoriteStudio.device_id == device_id)
+            .group_by(FavoriteStudio.studio_id)
+        ).all()
+        return {gid_val: int(n) for gid_val, n in rows}
+
+    # Performerki: sceny idą przez scene_performers, więc indeks po dacie nie pomoże i
+    # trzeba przeczytać całą filmografię. To jest drogie (6-15 s przy ~1000 ulubionych),
+    # dlatego wynik siedzi w cache, patrz _cached_new_counts.
     base = (
         select(
-                gid.label("gid"),
+            ScenePerformer.performer_id.label("gid"),
             Scene.created_at.label("created_at"),
             Scene.backfill.label("backfill"),
-                last_seen.label("last_seen"),
+            FavoritePerformer.last_seen_at.label("last_seen"),
         )
         .select_from(FavoritePerformer)
         .join(ScenePerformer, ScenePerformer.performer_id == FavoritePerformer.performer_id)
         .join(Scene, Scene.id == ScenePerformer.scene_id)
         .where(FavoritePerformer.device_id == device_id)
     )
-        partition = ScenePerformer.performer_id
-    else:
-        clauses.append(stub_exclusion_clause())
-        gid = Scene.studio_id
-        last_seen = FavoriteStudio.last_seen_at
-        base = (
-            select(
-                gid.label("gid"),
-                Scene.created_at.label("created_at"),
-                Scene.backfill.label("backfill"),
-                last_seen.label("last_seen"),
-            )
-            .select_from(FavoriteStudio)
-            .join(Scene, Scene.studio_id == FavoriteStudio.studio_id)
-            .where(FavoriteStudio.device_id == device_id)
-        )
-        partition = Scene.studio_id
-
     rn = func.row_number().over(
-        partition_by=partition,
+        partition_by=ScenePerformer.performer_id, order_by=list_order
-        order_by=(Scene.release_date.desc().nullslast(), Scene.created_at.desc()),
     ).label("rn")
     inner = base.add_columns(rn).where(*clauses).subquery()
     rows = session.execute(
@@ -186,6 +207,94 @@ def _merged_new_counts(session: Session, device_id: str, *, kind: str) -> dict:
     return counts
 
 
+# Cache liczników "+N" w pamięci procesu API (uvicorn chodzi jednym procesem). Liczenie
+# dla urządzeń z setkami ulubionych trwa sekundy, a apka woła /favorites przy wejściu na
+# listę scen, ulubione i każdą performerkę. Liczniki zmieniają się wolno (ingest), więc:
+# świeży wynik oddajemy od razu, starszy niż _COUNTS_FRESH_SEC też od razu i liczymy
+# nowy w tle, a dopiero starszy niż _COUNTS_MAX_AGE_SEC liczymy na miejscu.
+# mark_seen zeruje licznik w cache, żeby badge zgasł natychmiast.
+_COUNTS_FRESH_SEC = 300.0
+_COUNTS_MAX_AGE_SEC = 6 * 3600.0
+_counts_cache: dict[tuple[str, str], tuple[float, dict]] = {}
+_counts_seen: dict[tuple[str, str], dict] = {}  # gid -> kiedy oznaczono jako obejrzane
+_counts_locks: dict[tuple[str, str], threading.Lock] = {}
+_counts_guard = threading.Lock()
+
+
+def _counts_lock(key: tuple[str, str]) -> threading.Lock:
+    with _counts_guard:
+        lock = _counts_locks.get(key)
+        if lock is None:
+            lock = _counts_locks[key] = threading.Lock()
+        return lock
+
+
+def _store_counts(key: tuple[str, str], started: float, counts: dict) -> None:
+    with _counts_guard:
+        # mark_seen w trakcie liczenia: wynik policzono ze starym last_seen_at
+        for gid, at in _counts_seen.get(key, {}).items():
+            if at >= started:
+                counts[gid] = 0
+        _counts_seen.pop(key, None)
+        _counts_cache[key] = (time.monotonic(), counts)
+        stale = [k for k, (at, _) in _counts_cache.items() if time.monotonic() - at > _COUNTS_MAX_AGE_SEC]
+        for k in stale:
+            _counts_cache.pop(k, None)
+            _counts_locks.pop(k, None)
+
+
+def _compute_counts(key: tuple[str, str]) -> None:
+    device_id, kind = key
+    started = time.monotonic()
+    with session_scope() as session:
+        counts = _merged_new_counts(session, device_id, kind=kind)
+    _store_counts(key, started, counts)
+
+
+def _refresh_in_background(key: tuple[str, str]) -> None:
+    lock = _counts_lock(key)
+    if not lock.acquire(blocking=False):
+        return  # już ktoś liczy
+
+    def run() -> None:
+        try:
+            _compute_counts(key)
+        except Exception:  # noqa: BLE001
+            log.exception("favorites: odświeżenie liczników w tle nie wyszło %s", key)
+        finally:
+            lock.release()
+
+    threading.Thread(target=run, name="fav-counts", daemon=True).start()
+
+
+def _cached_new_counts(device_id: str, *, kind: str) -> dict:
+    key = (device_id, kind)
+    hit = _counts_cache.get(key)
+    if hit is not None:
+        age = time.monotonic() - hit[0]
+        if age < _COUNTS_MAX_AGE_SEC:
+            if age > _COUNTS_FRESH_SEC:
+                _refresh_in_background(key)
+            return dict(hit[1])
+    # Brak w cache: liczymy teraz. Lock, żeby równoległe zapytania z kilku ekranów
+    # nie liczyły tego samego naraz, drugie poczeka i weźmie gotowy wynik.
+    with _counts_lock(key):
+        hit = _counts_cache.get(key)
+        if hit is not None and time.monotonic() - hit[0] < _COUNTS_FRESH_SEC:
+            return dict(hit[1])
+        _compute_counts(key)
+    return dict(_counts_cache[key][1])
+
+
+def _mark_counts_seen(device_id: str, kind: str, gid: uuid.UUID) -> None:
+    key = (device_id, kind)
+    with _counts_guard:
+        _counts_seen.setdefault(key, {})[gid] = time.monotonic()
+        hit = _counts_cache.get(key)
+        if hit is not None and hit[1].get(gid):
+            hit[1][gid] = 0
+
+
 class FavoriteOut(BaseModel):
     performer_id: uuid.UUID
     canonical_name: str
@@ -206,6 +315,10 @@ class FavoriteListOut(BaseModel):
 def list_favorites(
     session: Annotated[Session, Depends(get_session)],
     device_id: Annotated[str, Depends(get_device_id)],
+    counts: bool = Query(
+        default=True,
+        description="false: bez liczników +N (ekrany, które chcą tylko wiedzieć, czy zapalić gwiazdkę)",
+    ),
 ) -> FavoriteListOut:
     rows = session.execute(
         select(FavoritePerformer, Performer)
@@ -220,7 +333,7 @@ def list_favorites(
     # _job_refresh_taxonomy_counts), ta sama definicja co przed (sceny z żywym
     # playback). Wcześniej grouped count z EXISTS playback per-request. Migracja 0019.
     scene_counts: dict = {perf.id: perf.scene_count for _, perf in rows}
-    new_counts = _merged_new_counts(session, device_id, kind="performer")
+    new_counts = _cached_new_counts(device_id, kind="performer") if counts else {}
 
     items: list[FavoriteOut] = []
     new_total = 0
@@ -297,6 +410,7 @@ def mark_seen(
         raise HTTPException(status_code=404, detail="not in favorites")
     fav.last_seen_at = datetime.now(UTC)
     session.commit()
+    _mark_counts_seen(device_id, "performer", performer_id)
     return SeenOut(performer_id=performer_id, last_seen_at=fav.last_seen_at)
 
 
@@ -327,6 +441,10 @@ class FavoriteStudioListOut(BaseModel):
 def list_favorite_studios(
     session: Annotated[Session, Depends(get_session)],
     device_id: Annotated[str, Depends(get_device_id)],
+    counts: bool = Query(
+        default=True,
+        description="false: bez liczników +N i filmów (sam stan gwiazdki)",
+    ),
 ) -> FavoriteStudioListOut:
     rows = session.execute(
         select(FavoriteStudio, Studio)
@@ -336,6 +454,24 @@ def list_favorite_studios(
     ).all()
     if not rows:
         return FavoriteStudioListOut(items=[], total=0, new_total=0)
+    if not counts:
+        return FavoriteStudioListOut(
+            items=[
+                FavoriteStudioOut(
+                    studio_id=st.id,
+                    name=st.name,
+                    slug=st.slug,
+                    network=st.network,
+                    scene_count=st.scene_count,
+                    new_count=0,
+                    last_seen_at=fav.last_seen_at,
+                    created_at=fav.created_at,
+                )
+                for fav, st in rows
+            ],
+            total=len(rows),
+            new_total=0,
+        )
 
     # scene_count: zdenormalizowany Studio.scene_count (refresh w tle, migracja 0019).
     scene_counts: dict = {st.id: st.scene_count for _, st in rows}
@@ -362,7 +498,7 @@ def list_favorite_studios(
             .group_by(Movie.studio_id)
         ).all()
     }
-    new_counts = _merged_new_counts(session, device_id, kind="studio")
+    new_counts = _cached_new_counts(device_id, kind="studio")
 
     items: list[FavoriteStudioOut] = []
     new_total = 0
@@ -440,6 +576,7 @@ def mark_studio_seen(
         raise HTTPException(status_code=404, detail="not in favorites")
     fav.last_seen_at = datetime.now(UTC)
     session.commit()
+    _mark_counts_seen(device_id, "studio", studio_id)
     return SeenStudioOut(studio_id=studio_id, last_seen_at=fav.last_seen_at)
 
 

app/api/scene_favorites.py

@@ -16,7 +16,7 @@ from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.api.device import get_device_id
-from app.api.scenes import _build_scene_out
+from app.api.scenes import _build_scenes_out_batch
 from app.api.schemas import SceneOut
 from app.auth import require_api_key
 from app.db import get_session
@@ -54,7 +54,11 @@ def list_scene_favorites(
         )
         .all()
     )
-    items = [_build_scene_out(session, scene, device_id=device_id) for scene, _ in rows]
+    # Wsadowo jak listy scen: ekran ulubionych rysuje ten sam kafelek, a pojedyncze
+    # _build_scene_out robilo 7 zapytan na scene (setki ulubionych = sekundy).
+    items = _build_scenes_out_batch(
+        session, [scene for scene, _ in rows], light=True, device_id=device_id
+    )
     return SceneFavoriteListOut(items=items, total=len(items))
 
 

app/api/scenes.py

@@ -9,7 +9,7 @@ from typing import Annotated
 
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from pydantic import BaseModel
-from sqlalchemy import Date, distinct, exists, false, func, literal_column, select
+from sqlalchemy import Date, and_, distinct, exists, false, func, literal_column, select, union
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -681,12 +681,35 @@ def similar_scenes(
     # Fill z tagów gdy performerów mało/brak (anonimowe tube sceny), inaczej pomijamy,
     # bo szeroki tag skanuje setki tysięcy scen.
     if tag_ids and len(cand_ids) < limit * 4:
+        # Pula z NAJRZADSZYCH tagów (najwięcej mówią o scenie), po _TAG_SEED_ROWS scen na
+        # tag, a wspólne tagi liczymy tylko dla tej puli. Grupowanie po wszystkich tagach
+        # sceny naraz czytało całe 'amateur' + 'teen-18' + 'handjob' (600 tys. wierszy,
+        # do 5 s). Sceny, które łączą z celem wyłącznie szerokie tagi, i tak są słabo
+        # podobne, więc ich pominięcie nic nie psuje.
+        _TAG_SEED_TAGS, _TAG_SEED_ROWS = 4, 1500
+        tag_sizes = dict(
+            session.execute(select(Tag.id, Tag.scene_count).where(Tag.id.in_(tag_ids))).all()
+        )
+        live_tags = [t for t in tag_ids if (tag_sizes.get(t) or 0) > 0] or list(tag_ids)
+        seed_tags = sorted(live_tags, key=lambda t: tag_sizes.get(t) or 0)[:_TAG_SEED_TAGS]
+        pool = union(
+            *[
+                select(SceneTag.scene_id)
+                .where(SceneTag.tag_id == t, SceneTag.scene_id != scene_id)
+                .limit(_TAG_SEED_ROWS)
+                for t in seed_tags
+            ]
+        ).subquery("pool")
+        shared = func.count(distinct(SceneTag.tag_id))
         rows = session.execute(
-            select(SceneTag.scene_id, func.count(distinct(SceneTag.tag_id)).label("c"))
+            select(pool.c.scene_id, shared.label("c"))
-            .where(SceneTag.tag_id.in_(tag_ids), SceneTag.scene_id != scene_id)
+            .join(
-            .group_by(SceneTag.scene_id)
+                SceneTag,
-            .having(func.count(distinct(SceneTag.tag_id)) >= 2)
+                and_(SceneTag.scene_id == pool.c.scene_id, SceneTag.tag_id.in_(tag_ids)),
-            .order_by(func.count(distinct(SceneTag.tag_id)).desc())
+            )
+            .group_by(pool.c.scene_id)
+            .having(shared >= 2)
+            .order_by(shared.desc())
             .limit(_TAG_FILL_CAP)
         ).all()
         for sid, c in rows:
@@ -1462,23 +1485,41 @@ def enrich_tags_from_tube(
     # (scene_id, tag_id), drugi flush dostaje UniqueViolation (GOON-H, 4 events
     # w 10h mimo wcześniejszego seen_tag_ids fix). ON CONFLICT skip'uje silently.
     from sqlalchemy.dialects.postgresql import insert as pg_insert
-    added = 0
+
-    seen_tag_ids: set = set()
+    from app.normalize.tag_stoplist import is_junk_tag
-    for name in tag_names:
+
-        norm = NormalizedTag(name=name, slug=slugify(name), external_id=None)
+    # Istniejące tagi jednym zapytaniem po slugach. resolve_tag robi 1-4 zapytania na
-        tag = resolve_tag(session, norm=norm)
+    # tag (Sentry N+1 przy 20-40 tagach ze strony), więc woła się go już tylko dla
-        if tag is None or tag.id in seen_tag_ids:
+    # nowych nazw i dla slugów z końcówką litera+"2", które kanonizuje po swojemu.
-            continue
+    slugs = {slugify(n) for n in tag_names}
-        seen_tag_ids.add(tag.id)
+    direct = [
-        stmt = (
+        s for s in slugs
-            pg_insert(SceneTag.__table__)
+        if s and len(s) <= 120 and not is_junk_tag(s)
-            .values(scene_id=scene_id, tag_id=tag.id, source_id=None)
+        and not (len(s) > 1 and s[-1] == "2" and s[-2].isalpha())
-            .on_conflict_do_nothing(index_elements=["scene_id", "tag_id"])
+    ]
+    by_slug = (
+        {t.slug: t for t in session.execute(select(Tag).where(Tag.slug.in_(direct))).scalars()}
+        if direct else {}
     )
-        result = session.execute(stmt)
+    tag_ids: list = []
-        # rowcount == 1 gdy faktycznie wstawiony, 0 gdy ON CONFLICT skip
+    for name in tag_names:
-        if result.rowcount and result.rowcount > 0:
+        slug = slugify(name)
-            added += 1
+        tag = by_slug.get(slug)
+        if tag is None:
+            tag = resolve_tag(session, norm=NormalizedTag(name=name, slug=slug, external_id=None))
+        if tag is not None and tag.id not in tag_ids:
+            tag_ids.append(tag.id)
+    added = 0
+    if tag_ids:
+        # Jeden INSERT; sortowane id, żeby dwa równoległe wywołania dla tej samej sceny
+        # zakładały blokady w tej samej kolejności.
+        result = session.execute(
+            pg_insert(SceneTag.__table__)
+            .values([{"scene_id": scene_id, "tag_id": t, "source_id": None} for t in sorted(tag_ids)])
+            .on_conflict_do_nothing(index_elements=["scene_id", "tag_id"])
+            .returning(SceneTag.__table__.c.tag_id)
+        )
+        added = len(result.fetchall())
     session.commit()
     return EnrichTagsOut(scene_id=scene_id, added=added, tube_used=sitetag, tags=tag_names)