"""APScheduler job definitions dla worker'a (M5).

Domyślny harmonogram:
  - tpdb           co 6h, delta od ostatniego successful run
  - stashdb        co 6h, delta
  - performer-driven, co 12h, top-N performerów z bazy (auto-discovers nowe sceny przez
    ALL_DIRECT_SCRAPERS, 25 tube'ów per-tube HTTP scraping)
  - performer-continuous, tick co N sekund, 1 performer per tick (ORDER BY last_searched_at)

Konfigurację (interwały, włącz/wyłącz) można nadpisać przez env (`GOON_SCHED_*`),
patrz `app/scheduler/config.py`.

Uwaga: APScheduler in-process (BlockingScheduler), wystarczy dla self-hosted single
worker. Dla multi-worker trzebaby Redis/SQLAlchemy job store + distributed lock.
"""
from __future__ import annotations

import logging
from datetime import datetime, timezone
from typing import Any

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.interval import IntervalTrigger

from app.connectors import get_movie_connectors
from app.connectors.stashdb import StashDBConnector
from app.connectors.tpdb import TPDBConnector
from app.ingest import ingest_from_connector, ingest_movies_from_connector
from app.scheduler.browse_latest import run_browse_latest
from app.scheduler.performer_driven import run_continuous_one_at_a_time, run_performer_driven

log = logging.getLogger(__name__)


# Stała "epoka" dla IntervalTrigger.start_date, kotwica siatki fire-times.
# Bez start_date APScheduler liczy next_run_time = add_job_time + interval, więc każdy
# restart workera (a tych jest dużo, manual deploys, OOM, obraz przebudowany) odsuwa
# kolejny fire o pełen interval. Bug-reporty 2026-05-19 (`93d3c485` "brak freshporno")
# i 2026-05-23 (`2fbf1c73` "Czemu nie ma nowych filmów?") to dokładnie ten case:
# worker restartowany 15× w ciągu 3 dni → movie_ingest (24h) nigdy nie odpalił po
# 2026-05-20 05:29.
#
# Ze stałym start_date w przeszłości next_run_time leży na siatce co N godzin od tej
# kotwicy → restart workera nie zmienia kiedy następny fire. 05:00 UTC = 07:00 PL,
# niski ruch, bez kolizji z ręcznymi deployami w godzinach pracy.
INTERVAL_ANCHOR = datetime(2026, 1, 1, 5, 0, tzinfo=timezone.utc)


# Hard-timeout dla jobów robiących zewnętrzne HTTP (tpdb/stashdb/performer-driven).
# Bez tego zawis connectora bez własnego timeoutu blokował job na wiele godzin, a
# `max_instances=1` blokował KOLEJNE fire'y do restartu workera (incident 2026-06-02:
# 6 runów wisiało 8.7h od wspólnego anchora 05:00). Po timeoucie job WRACA (slot się
# zwalnia → następny fire leci); osierocony wątek dożywa do restartu (jak movie-ingest),
# a jego 'running' DB-row sprząta periodic reaper (_job_reap_stuck).
_JOB_TIMEOUT_SEC = 1800  # 30 min, healthy tpdb/stashdb delta to minuty (po SQL-phash), performer-driven top-N ~10-20 min.


def _run_with_timeout(fn, *, label: str, timeout_sec: int = _JOB_TIMEOUT_SEC) -> None:
    from concurrent.futures import ThreadPoolExecutor
    from concurrent.futures import TimeoutError as FutureTimeout

    ex = ThreadPoolExecutor(max_workers=1)
    try:
        fut = ex.submit(fn)
        try:
            fut.result(timeout=timeout_sec)
        except FutureTimeout:
            log.error(
                "[scheduler] %s HUNG > %ds — zwalniam slot, orphan thread dożyje do restartu",
                label, timeout_sec,
            )
        except Exception:
            log.exception("[scheduler] %s job failed", label)
    finally:
        ex.shutdown(wait=False)


def _job_tpdb() -> None:
    log.info("[scheduler] tpdb delta starting")
    _run_with_timeout(lambda: ingest_from_connector(TPDBConnector(), use_delta=True), label="tpdb")


def _job_stashdb() -> None:
    log.info("[scheduler] stashdb delta starting")
    _run_with_timeout(lambda: ingest_from_connector(StashDBConnector(), use_delta=True), label="stashdb")


def _job_performer_driven(top_n: int) -> None:
    log.info("[scheduler] performer-driven top-%d starting", top_n)
    _run_with_timeout(
        lambda: run_performer_driven(top_n=top_n, per_performer_limit=50),
        label="performer-driven",
    )


def _job_reap_stuck() -> None:
    """Periodic reaper, czyści ingest_runs wiszące w 'running' >2h (zombie po zawisach
    connectorów / kill mid-run). Startup-only reaper nie łapał ich gdy worker długo żył
    (incident 2026-06-02: zombie wisiały 8.7h). Delayed import, unika cyklu z worker.py."""
    try:
        from app.scheduler.worker import reap_stuck_ingest_runs

        reaped = reap_stuck_ingest_runs()
        if reaped:
            log.warning("[scheduler] periodic reaper: %d stuck ingest_runs", reaped)
    except Exception:
        log.exception("[scheduler] periodic reaper failed")


def _job_deep_crawl(pages_per_run: int) -> None:
    """Deep-crawl pełnych katalogów browse-tube'ów (Faza 2a, ingest-all). Round-robin
    po sitetagu, wznawialny kursor (app/scheduler/deep_crawl.py). Hard-timeout 1h."""
    log.info("[scheduler] deep-crawl starting (pages_per_run=%d)", pages_per_run)
    from app.scheduler.deep_crawl import run_deep_crawl

    _run_with_timeout(
        lambda: run_deep_crawl(pages_per_run=pages_per_run),
        label="deep-crawl",
        timeout_sec=3600,
    )


def _job_browse_latest(max_pages: int) -> None:
    """Browse-latest, scrap newest scenes z rich-metadata tubes (shyfap + ...).
    Komplementarny do performer-driven: forward-fill (new scenes) vs backward (known performers).
    """
    log.info("[scheduler] browse-latest starting (max_pages=%d)", max_pages)
    try:
        run_browse_latest(max_pages=max_pages)
    except Exception:
        log.exception("[scheduler] browse-latest job failed")


def _job_movie_ingest() -> None:
    """Movies ingest, paradisehill (primary) + dooplay mirrory.

    Paradisehill jako primary daje canonical movie record (title + year + studio).
    Mirrory dooplay (mangoporn/streamporn/pandamovies) doklejają playback sources
    z native-friendly origins (mangoporn:luluvid, :voe, etc.), `extract_stream_from_hoster`
    rozwiązuje je do bezpośredniego stream URL → mobile gra natywnie zamiast WebView.

    Matching mirror→primary movie idzie przez `resolve_movie` (title+year+studio
    similarity). Każdy connector osobny IngestRun + delta od ostatniego success.

    Kolejność: paradisehill FIRST (żeby mirrory miały do czego się przykleić),
    potem mirrory. Pojedynczy failed connector NIE zatrzymuje pozostałych,
    każdy w osobnym try/except.

    HARD TIMEOUT per-connector (bug-report 2026-05-30 "ingest znów się zawiesił"):
    sam try/except chroni przed *wyjątkiem*, ale NIE przed *hangiem* (CPU-bound
    ReDoS na patologicznej stronie / thread-stall), wtedy jeden mirror blokuje
    resztę i mangoporn (jedyny z realnym new-content) nigdy nie startuje.
    Każdy connector leci w osobnym wątku z `future.result(timeout)`; po
    przekroczeniu logujemy i idziemy dalej (osierocony wątek dożywa do restartu
    workera, OK, bo loop się odblokowuje). Healthy run ~50s, cap 6 min = zapas.
    """
    from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout

    PER_CONNECTOR_TIMEOUT = 360  # sekundy

    for name, cls in get_movie_connectors():
        log.info("[scheduler] movie ingest %s starting", name)
        try:
            ex = ThreadPoolExecutor(max_workers=1)
            fut = ex.submit(ingest_movies_from_connector, cls(), use_delta=True)
            try:
                fut.result(timeout=PER_CONNECTOR_TIMEOUT)
            except FutureTimeout:
                log.error(
                    "[scheduler] movie ingest %s HUNG > %ds — skip, kolejka leci dalej",
                    name, PER_CONNECTOR_TIMEOUT,
                )
            finally:
                # shutdown(wait=False): nie blokuj na join osieroconego wątku.
                ex.shutdown(wait=False)
        except Exception:
            log.exception("[scheduler] movie ingest %s failed", name)


def _job_refresh_taxonomy_counts() -> None:
    """Przelicza denormalizowane scene_count na tags/performers/studios.

    Hot-path /tags|/performers|/studios|/favorites czyta gotową kolumnę zamiast
    agregować 6.3M scene_tags per-request (~4.3s → <20ms). Patrz migracja 0019 +
    app/scheduler/taxonomy_counts.py.
    """
    log.info("[scheduler] taxonomy counts refresh starting")
    try:
        from app.scheduler.taxonomy_counts import refresh_taxonomy_counts
        changed = refresh_taxonomy_counts()
        log.info("[scheduler] taxonomy counts refresh done: %s", changed)
    except Exception:
        log.exception("[scheduler] taxonomy counts refresh failed")


def _job_site_icons() -> None:
    """Loga stron do zakladki Sites (favicon -> PNG 64x64 w app/static/site-icons).
    Patrz scripts/site_icons.py; pliki mlodsze niz 14 dni zostaja."""
    from scripts.site_icons import live_sitetags, refresh_site_icons

    try:
        stats = refresh_site_icons(live_sitetags())
        log.info("[scheduler] site-icons: %s", stats)
    except Exception:  # noqa: BLE001
        log.exception("[scheduler] site-icons failed")


def _job_refresh_sxyprn_thumbs(batch: int = 1200) -> None:
    """Odświeża wygasłe sxyprn miniaturki z żywych stron /post/<id> (bug 2026-06-10).

    sxyprn/trafficdeposit thumbnaile są podpisane czasowo i rotują (token wygasa po
    ~tygodniach → 404), ale strona post żyje i ma świeży poster (`og:image`). Search/
    listingi NIE re-surfaceują starych postów, więc jedyna droga to per-post page fetch.
    Bierzemy `batch` najdawniej-aktualizowanych źródeł, cykl po całym katalogu co kilka
    dni (mieści się w oknie wygaśnięcia). Patrz scripts/refresh_sxyprn_thumbs.py.
    """
    log.info("[scheduler] sxyprn thumb refresh starting (batch=%d)", batch)

    def _run() -> None:
        from sqlalchemy import text

        from app.db import session_scope
        from scripts.refresh_sxyprn_thumbs import refresh_batch

        with session_scope() as session:
            rows = session.execute(
                text(
                    "SELECT id, page_url FROM playback_sources "
                    "WHERE origin='tube:sxyprncom' AND dead_at IS NULL "
                    "ORDER BY updated_at ASC LIMIT :n"
                ).bindparams(n=batch)
            ).all()
        refreshed, dead, untouched = refresh_batch(list(rows))
        log.info(
            "[scheduler] sxyprn thumb refresh done: refreshed=%d dead=%d untouched=%d (of %d)",
            refreshed, dead, untouched, len(rows),
        )

    _run_with_timeout(_run, label="sxyprn-thumb-refresh")


def _job_bulk_dedup_performers() -> None:
    """Pair-wise dedup po performer overlap, safety net dla duplikatów które
    resolver-time scoring nie złapał.

    Use case (bug-report 2026-05-20, "brak Brazzers Exxtra po 15-05"):
    freshporno scrape przed fixem release_date tworzył duplicate scenes zamiast
    PS-merge do canonical TPDB scen. Resolver scoring miał score >0.92 (auto)
    z release_date, ale BEZ release_date wagi się przesuwały i wpadało w review/new.

    Bulk_dedup performers strategy iteruje per performer, robi pair-wise scoring
    dla wszystkich scen tego performera, łapie duplicate-y które ingest-time
    resolver pominął (np. gdy 2 sceny tej samej title+performer ale różny release_date).
    Auto-merge gdy score≥0.92, pending merge_candidate gdy 0.75-0.92.
    """
    log.info("[scheduler] bulk_dedup performers starting")
    try:
        from app.scheduler.bulk_dedup import run_bulk_dedup
        # cross_source_only=True: bez tego flag pairwise generuje N²/2 par na płodnego
        # performera, materializowane w listę → worker OOM-killed co 12h (6GB RSS na
        # 7.6GB boxie, 2026-06-06), ubijając przy okazji równoległe tpdb/stashdb/ingesty.
        # Flag zawęża do cross-source kandydatów (TPDB↔StashDB) z pre-filtrem candidate.
        # Timeout-wrap jak tpdb/stashdb, job nie ma własnego hard-timeoutu.
        _run_with_timeout(
            lambda: run_bulk_dedup(strategy="performers", dry_run=False, cross_source_only=True),
            label="bulk-dedup-performers",
        )
        log.info("[scheduler] bulk_dedup performers done")
    except Exception:
        log.exception("[scheduler] bulk_dedup performers failed")


def _job_thumb_asset_dedup() -> None:
    """Scala same-video-różne-tytuły dupy hdporn.gg/fullmovies.xxx po asset-id miniatury
    (reports 205b17d9/5a2944cb). bulk_dedup tego nie łapie (różne tytuły, brak phash);
    re-ingesty pod nowymi tytułami → dupy odrastają, stąd cyklicznie."""
    log.info("[scheduler] thumb-asset dedup starting")
    try:
        from app.scheduler.thumb_dedup import run_thumb_asset_dedup
        # run_thumb_asset_dedup loguje liczby/errory wewnętrznie; _run_with_timeout
        # połyka return, więc nie kapturujemy.
        _run_with_timeout(
            lambda: run_thumb_asset_dedup(commit=True),
            label="thumb-asset-dedup",
        )
        log.info("[scheduler] thumb-asset dedup done")
    except Exception:
        log.exception("[scheduler] thumb-asset dedup failed")


def _job_phash_backfill(batch: int) -> None:
    """Dolicza phash scenom, które mają miniaturkę, ale nie mają odcisku. Kolejkę
    planujemy raz (`plan_phash_backfill`), tu tylko mielimy paczkami, bo to pobrania
    z cudzych CDN-ów. Gdy kolejka pusta, job jest no-opem i nic nie kosztuje."""
    try:
        from app.scheduler.phash_backfill import run_phash_backfill

        out = _run_with_timeout(
            lambda: run_phash_backfill(batch=batch), label="phash-backfill"
        )
        if out and out.get("taken"):
            log.info("[scheduler] phash backfill %s", out)
    except Exception:
        log.exception("[scheduler] phash backfill failed")


def _job_duration_plan() -> None:
    """Dopisuje do kolejki sceny, ktore od wczoraj przyszly bez dlugosci. Scrapery
    czesci tubow jej nie zbieraja przy ingescie, wiec bez tego zaleglosc odrastalaby
    z kazda doba. `ON CONFLICT DO NOTHING`, wiec powtorzenie nic nie psuje."""
    try:
        from app.scheduler.duration_backfill import plan_duration_backfill

        added = _run_with_timeout(
            lambda: plan_duration_backfill(), label="duration-plan"
        )
        log.info("[scheduler] duration plan done (%s)", added)
    except Exception:
        log.exception("[scheduler] duration plan failed")


def _job_duration_backfill(batch: int) -> None:
    """Dolicza dlugosc scenom, ktore jej nie maja. Bez dlugosci scena wypada z feedu,
    bo `duration_sec >= 60` nie jest prawda dla NULL. Kolejke planujemy raz
    (`plan_duration_backfill`), tu tylko mielimy paczkami, bo to pobrania cudzych stron.
    Pusta kolejka = no-op, wiec nic nie kosztuje."""
    try:
        from app.scheduler.duration_backfill import run_duration_backfill

        out = _run_with_timeout(
            lambda: run_duration_backfill(batch=batch), label="duration-backfill"
        )
        if out and out.get("taken"):
            log.info("[scheduler] duration backfill %s", out)
    except Exception:
        log.exception("[scheduler] duration backfill failed")


def _job_phash_blacklist() -> None:
    """Wyłapuje phashe, które przestały cokolwiek znaczyć (zaślepka / czarna klatka /
    intro studia) i kasuje je z fingerprintów. Audyt 2026-07-27: 73 takie wartości przy
    4375 scenach, rekordzistka dzielona przez 892 sceny o 892 różnych tytułach.
    Cyklicznie, bo tube potrafi zacząć serwować zaślepkę w dowolnym momencie, a wartość
    raz wpisana na blacklistę zostaje tam na stałe (inaczej odrastałaby przy ingeście)."""
    log.info("[scheduler] phash blacklist starting")
    try:
        from app.scheduler.phash_blacklist import run_phash_blacklist

        _run_with_timeout(run_phash_blacklist, label="phash-blacklist")
        log.info("[scheduler] phash blacklist done")
    except Exception:
        log.exception("[scheduler] phash blacklist failed")


def _job_tag_tail_trim() -> None:
    """Przycina ogon tagow rule34video (booru: nazwy postaci, serii, autorow).
    356 scen wnosilo 1112 roznych tagow do wspolnej listy, 602 uzyte dokladnie raz.
    Zostawia tagi z >=3 scen i najwyzej 15 na scene; tagi uzywane takze poza
    rule34video sa chronione, bo naleza do wspolnego slownika."""
    log.info("[scheduler] tag tail trim starting")
    try:
        from app.scheduler.tag_tail_trim import run

        stats = run()
        log.info("[scheduler] tag tail trim done: %s", stats)
    except Exception:
        log.exception("[scheduler] tag tail trim failed")


def _job_junk_performers() -> None:
    """Kasuje kategorie i studia podszywające się pod performerów („Creampie",
    „Natural tits", „Brazzers"). Audyt 2026-07-27: 277 rekordów, 33 784 przypisania.
    Odrastają, bo tube-search dokleja performera po jednym tokenie z zapytania, a
    resolve_performer tworzy go z dowolnej nazwy podanej przez tube."""
    log.info("[scheduler] junk performers starting")
    try:
        from app.scheduler.junk_performers import run_junk_performer_cleanup

        _run_with_timeout(run_junk_performer_cleanup, label="junk-performers")
        log.info("[scheduler] junk performers done")
    except Exception:
        log.exception("[scheduler] junk performers failed")


def _job_title_duration_dedup() -> None:
    """Scal missing-merge dupy (ten sam performer + identyczny tytuł + długość) których
    bulk_dedup nie łapie (tube re-scrape / cross-tube np. porn00 vs xnxx, reports
    28fe8181/32df33b1). playback-only = to co user widzi. Re-ingesty → dupy odrastają."""
    log.info("[scheduler] title-duration dedup starting")
    try:
        from app.scheduler.title_duration_dedup import run_title_duration_dedup

        _run_with_timeout(
            lambda: run_title_duration_dedup(playback_only=True, commit=True),
            label="title-duration-dedup",
        )
        log.info("[scheduler] title-duration dedup done")
    except Exception:
        log.exception("[scheduler] title-duration dedup failed")


def _job_studio_title_dedup() -> None:
    """Scal kopie tej samej sceny z tubow, ktore zapisuja tytul kazdy po swojemu (to samo
    studio + ten sam rdzen tytulu, report bb1a357d). title_duration_dedup wymaga
    identycznego tytulu, a phash nie lapie, bo miniatury sa z roznych klatek."""
    log.info("[scheduler] studio-title dedup starting")
    try:
        from app.scheduler.studio_title_dedup import run_studio_title_dedup

        _run_with_timeout(
            lambda: run_studio_title_dedup(days=30, commit=True),
            label="studio-title-dedup",
        )
        log.info("[scheduler] studio-title dedup done")
    except Exception:
        log.exception("[scheduler] studio-title dedup failed")


def _job_ingest_watchdog(
    max_age_hours: int, search_max_age_hours: int, movie_max_age_hours: int
) -> None:
    """Per-origin freshness watchdog → Sentry. Obejmuje sceny (browse/search) ORAZ
    filmy (od 2026-08-03; wcześniej filmy nie były pokryte i streamporn.vip stał 23 dni
    niezauważony). Globalny monitor (jeden Source 'tube-scraper') tego nie łapie;
    pojedynczy origin może zamarznąć przy success-runie (report 14f3a655)."""
    try:
        from app.scheduler.ingest_watchdog import run_ingest_freshness_watchdog

        res = run_ingest_freshness_watchdog(
            max_age_hours=max_age_hours,
            search_max_age_hours=search_max_age_hours,
            movie_max_age_hours=movie_max_age_hours,
        )
        if res["stale"]:
            log.warning("[scheduler] ingest-watchdog: %d stale origin(s)", len(res["stale"]))
    except Exception:
        log.exception("[scheduler] ingest-watchdog failed")


def _job_hetzner_monitor() -> None:
    """Hetzner Cloud bandwidth monitor, alert do Sentry przy progach % included_traffic.
    No-op gdy brak HETZNER_API_TOKEN/SERVER_ID w env (loguje że wyłączony)."""
    try:
        from app.scheduler.hetzner_monitor import run_hetzner_bandwidth_check

        run_hetzner_bandwidth_check()
    except Exception:
        log.exception("[scheduler] hetzner-monitor failed")


def _job_source_stats() -> None:
    """Przelicz source_stats (ranking źródeł na Sites screen), freshness/richness z DB
    + health z telemetrii odtwarzania (playback_events) lub proxy. Ciężki agregat."""
    try:
        from app.scheduler.source_stats import run_source_stats

        _run_with_timeout(run_source_stats, label="source-stats")
    except Exception:
        log.exception("[scheduler] source-stats failed")


def _job_performer_continuous(refresh_after_days: int) -> None:
    """Continuous worker, 1 performer per tick, ORDER BY last_searched_at NULLS FIRST.

    Per tick: full search across ~25 tubeów (per_performer_limit=None). Tick zajmuje
    ~50-80s. Interval ustawiony na 15s + max_instances=1 + coalesce=True znaczy że
    real rate to max(15s, tick_duration), efektywnie ~1 perf/50-80s.
    """
    try:
        run_continuous_one_at_a_time(
            refresh_after_days=refresh_after_days,
            per_performer_limit=None,  # full coverage all tubes
        )
    except Exception:
        log.exception("[scheduler] performer-continuous failed")


def _job_tpdb_movie_enrich(batch: int) -> None:
    """Wzbogaca filmy metadanymi z TPDB /movies: obsada, kategorie (tagi), studio,
    reżyser + kanoniczny UUID do dedupu mirrorów. Batch najświeższych grywalnych
    filmów bez obsady/studia per run. paradisehill (primary) prawie nie ma obsady,
    więc to domyka największą lukę. Patrz app/enrich/tpdb_movies.py."""
    log.info("[scheduler] tpdb-movie-enrich starting (batch=%d)", batch)
    try:
        from app.db import session_scope
        from app.enrich.tpdb_movies import run_tpdb_movie_enrich

        _run_with_timeout(
            lambda: run_tpdb_movie_enrich(session_scope, limit=batch),
            label="tpdb-movie-enrich",
        )
        log.info("[scheduler] tpdb-movie-enrich done")
    except Exception:
        log.exception("[scheduler] tpdb-movie-enrich failed")


def _job_clips_latest(pages: int) -> None:
    """Klipy do feedu Swipe: globalne „najnowsze" RedGifs (patrz clips_ingest)."""
    from app.scheduler.clips_ingest import run_redgifs_latest

    try:
        _run_with_timeout(lambda: run_redgifs_latest(pages=pages), label="clips-latest")
    except Exception:
        log.exception("[scheduler] clips-latest failed")


def _job_clips_maintenance() -> None:
    """Feed Swipe: decay afiniczności, retencja zdarzeń, reaper martwych plików."""
    from app.scheduler.clips_ingest import run_clips_maintenance

    try:
        _run_with_timeout(run_clips_maintenance, label="clips-maintenance")
    except Exception:
        log.exception("[scheduler] clips-maintenance failed")


def build_scheduler(cfg: dict[str, Any]) -> BlockingScheduler:
    """Buduje scheduler na podstawie cfg dictu.

    cfg keys:
      tpdb_hours: int | None        (None = wyłączony)
      stashdb_hours: int | None
      performer_driven_hours: int | None
      performer_driven_top_n: int
      performer_continuous_seconds: int | None
      performer_continuous_refresh_days: int
    """
    # misfire_grace_time: domyslnie 1 s. O 05:00 UTC (wspolny anchor) startuje kilkanascie
    # zadan naraz i dzienny movie-ingest dostawal watek 1,1 s po czasie, wiec APScheduler go
    # POMIJAL ("was missed by 0:00:01"), dzien w dzien. Filmy staly 4-8 dni, zanim watchdog
    # to pokazal (2026-09-19). Godzina zapasu; coalesce i max_instances=1 i tak zbijaja
    # zaleglosci w jedno uruchomienie.
    sched = BlockingScheduler(timezone="UTC", job_defaults={"misfire_grace_time": 3600})

    sched.add_job(
        _job_site_icons,
        IntervalTrigger(hours=24, start_date=INTERVAL_ANCHOR),
        id="site-icons",
        replace_existing=True,
        max_instances=1,
        coalesce=True,
    )

    if cfg.get("tpdb_hours"):
        sched.add_job(
            _job_tpdb,
            IntervalTrigger(hours=cfg["tpdb_hours"], start_date=INTERVAL_ANCHOR),
            id="tpdb",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: tpdb every %dh", cfg["tpdb_hours"])

    if cfg.get("stashdb_hours"):
        sched.add_job(
            _job_stashdb,
            IntervalTrigger(hours=cfg["stashdb_hours"], start_date=INTERVAL_ANCHOR),
            id="stashdb",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: stashdb every %dh", cfg["stashdb_hours"])

    if cfg.get("performer_driven_hours"):
        top_n = cfg.get("performer_driven_top_n") or 20
        sched.add_job(
            lambda: _job_performer_driven(top_n),
            IntervalTrigger(hours=cfg["performer_driven_hours"], start_date=INTERVAL_ANCHOR),
            id="performer_driven",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info(
            "scheduler: performer-driven every %dh (top_n=%d)",
            cfg["performer_driven_hours"],
            top_n,
        )

    if cfg.get("browse_latest_hours"):
        max_pages = cfg.get("browse_latest_max_pages") or 5
        sched.add_job(
            lambda: _job_browse_latest(max_pages),
            IntervalTrigger(hours=cfg["browse_latest_hours"], start_date=INTERVAL_ANCHOR),
            id="browse_latest",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info(
            "scheduler: browse-latest every %dh (max_pages=%d)",
            cfg["browse_latest_hours"], max_pages,
        )

    if cfg.get("clips_hours"):
        clips_pages = cfg.get("clips_latest_pages") or 5
        sched.add_job(
            lambda: _job_clips_latest(clips_pages),
            IntervalTrigger(hours=cfg["clips_hours"], start_date=INTERVAL_ANCHOR),
            id="clips_latest",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: clips-latest every %dh (pages=%d)", cfg["clips_hours"], clips_pages)

    if cfg.get("clips_maintenance_hours"):
        sched.add_job(
            _job_clips_maintenance,
            IntervalTrigger(hours=cfg["clips_maintenance_hours"], start_date=INTERVAL_ANCHOR),
            id="clips_maintenance",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: clips-maintenance every %dh", cfg["clips_maintenance_hours"])

    if cfg.get("bulk_dedup_hours"):
        sched.add_job(
            _job_bulk_dedup_performers,
            IntervalTrigger(hours=cfg["bulk_dedup_hours"], start_date=INTERVAL_ANCHOR),
            id="bulk_dedup_performers",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: bulk-dedup performers every %dh", cfg["bulk_dedup_hours"])

    if cfg.get("tpdb_movie_enrich_hours"):
        _tme_batch = cfg.get("tpdb_movie_enrich_batch") or 200
        sched.add_job(
            lambda: _job_tpdb_movie_enrich(_tme_batch),
            IntervalTrigger(hours=cfg["tpdb_movie_enrich_hours"], start_date=INTERVAL_ANCHOR),
            id="tpdb_movie_enrich",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info(
            "scheduler: tpdb-movie-enrich every %dh (batch=%d)",
            cfg["tpdb_movie_enrich_hours"], _tme_batch,
        )

    if cfg.get("thumb_dedup_hours"):
        sched.add_job(
            _job_thumb_asset_dedup,
            IntervalTrigger(hours=cfg["thumb_dedup_hours"], start_date=INTERVAL_ANCHOR),
            id="thumb_asset_dedup",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: thumb-asset dedup every %dh", cfg["thumb_dedup_hours"])

    if cfg.get("phash_blacklist_hours"):
        sched.add_job(
            _job_phash_blacklist,
            IntervalTrigger(hours=cfg["phash_blacklist_hours"], start_date=INTERVAL_ANCHOR),
            id="phash_blacklist",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: phash blacklist every %dh", cfg["phash_blacklist_hours"])

    if cfg.get("duration_backfill_minutes"):
        sched.add_job(
            _job_duration_plan,
            IntervalTrigger(hours=24, start_date=INTERVAL_ANCHOR),
            id="duration_plan",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: duration plan every 24h")
    if cfg.get("duration_backfill_minutes"):
        _dur_batch = int(cfg.get("duration_backfill_batch") or 300)
        sched.add_job(
            _job_duration_backfill,
            IntervalTrigger(
                minutes=cfg["duration_backfill_minutes"], start_date=INTERVAL_ANCHOR
            ),
            args=[_dur_batch],
            id="duration_backfill",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info(
            "scheduler: duration backfill every %dmin (batch=%d)",
            cfg["duration_backfill_minutes"],
            _dur_batch,
        )

    if cfg.get("phash_backfill_minutes"):
        _batch = int(cfg.get("phash_backfill_batch") or 400)
        sched.add_job(
            _job_phash_backfill,
            IntervalTrigger(
                minutes=cfg["phash_backfill_minutes"], start_date=INTERVAL_ANCHOR
            ),
            args=[_batch],
            id="phash_backfill",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info(
            "scheduler: phash backfill every %dm, batch %d",
            cfg["phash_backfill_minutes"], _batch,
        )

    if cfg.get("tag_tail_trim_hours"):
        sched.add_job(
            _job_tag_tail_trim,
            IntervalTrigger(hours=cfg["tag_tail_trim_hours"], start_date=INTERVAL_ANCHOR),
            id="tag_tail_trim",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: tag tail trim every %dh", cfg["tag_tail_trim_hours"])

    if cfg.get("junk_performers_hours"):
        sched.add_job(
            _job_junk_performers,
            IntervalTrigger(hours=cfg["junk_performers_hours"], start_date=INTERVAL_ANCHOR),
            id="junk_performers",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: junk performers every %dh", cfg["junk_performers_hours"])

    if cfg.get("title_dedup_hours"):
        sched.add_job(
            _job_title_duration_dedup,
            IntervalTrigger(hours=cfg["title_dedup_hours"], start_date=INTERVAL_ANCHOR),
            id="title_duration_dedup",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: title-duration dedup every %dh", cfg["title_dedup_hours"])

    if cfg.get("studio_title_dedup_hours"):
        sched.add_job(
            _job_studio_title_dedup,
            IntervalTrigger(hours=cfg["studio_title_dedup_hours"], start_date=INTERVAL_ANCHOR),
            id="studio_title_dedup",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: studio-title dedup every %dh", cfg["studio_title_dedup_hours"])

    if cfg.get("ingest_watchdog_hours"):
        wd_max_age = cfg.get("ingest_watchdog_max_age_hours") or 48
        wd_search_max_age = cfg.get("ingest_watchdog_search_max_age_hours") or 168
        wd_movie_max_age = cfg.get("ingest_watchdog_movie_max_age_hours") or 72
        sched.add_job(
            lambda: _job_ingest_watchdog(wd_max_age, wd_search_max_age, wd_movie_max_age),
            IntervalTrigger(hours=cfg["ingest_watchdog_hours"], start_date=INTERVAL_ANCHOR),
            id="ingest_watchdog",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info(
            "scheduler: ingest-watchdog every %dh (browse=%dh, search=%dh, movies=%dh)",
            cfg["ingest_watchdog_hours"], wd_max_age, wd_search_max_age, wd_movie_max_age,
        )

    if cfg.get("hetzner_monitor_hours"):
        sched.add_job(
            _job_hetzner_monitor,
            IntervalTrigger(hours=cfg["hetzner_monitor_hours"], start_date=INTERVAL_ANCHOR),
            id="hetzner_monitor",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: hetzner-monitor every %dh", cfg["hetzner_monitor_hours"])

    if cfg.get("source_stats_hours"):
        sched.add_job(
            _job_source_stats,
            IntervalTrigger(hours=cfg["source_stats_hours"], start_date=INTERVAL_ANCHOR),
            id="source_stats",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: source-stats every %dh", cfg["source_stats_hours"])

    if cfg.get("movie_ingest_hours"):
        sched.add_job(
            _job_movie_ingest,
            IntervalTrigger(hours=cfg["movie_ingest_hours"], start_date=INTERVAL_ANCHOR),
            id="movie_ingest",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: movie-ingest every %dh", cfg["movie_ingest_hours"])

    if cfg.get("taxonomy_counts_hours"):
        sched.add_job(
            _job_refresh_taxonomy_counts,
            IntervalTrigger(hours=cfg["taxonomy_counts_hours"], start_date=INTERVAL_ANCHOR),
            id="taxonomy_counts",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: taxonomy-counts refresh every %dh", cfg["taxonomy_counts_hours"])

    if cfg.get("performer_continuous_seconds"):
        refresh_days = cfg.get("performer_continuous_refresh_days") or 30
        seconds = cfg["performer_continuous_seconds"]
        sched.add_job(
            lambda: _job_performer_continuous(refresh_days),
            IntervalTrigger(seconds=seconds),
            id="performer_continuous",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info(
            "scheduler: performer-continuous every %ds (refresh_after=%dd)",
            seconds, refresh_days,
        )

    if cfg.get("deep_crawl_hours"):
        pages = cfg.get("deep_crawl_pages_per_run") or 60
        sched.add_job(
            lambda: _job_deep_crawl(pages),
            IntervalTrigger(hours=cfg["deep_crawl_hours"], start_date=INTERVAL_ANCHOR),
            id="deep_crawl",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: deep-crawl every %dh (%d pages/run)", cfg["deep_crawl_hours"], pages)

    # Periodic reaper, czyści zombie 'running' runy co godzinę. Domyślnie ZAWSZE on
    # (cfg.get(...,1)), bo startup-only reaper nie łapie zawisów gdy worker długo żyje.
    reap_hours = cfg.get("reap_stuck_hours", 1)
    if reap_hours:
        sched.add_job(
            _job_reap_stuck,
            IntervalTrigger(hours=reap_hours, start_date=INTERVAL_ANCHOR),
            id="reap_stuck",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: reap-stuck every %dh", reap_hours)

    # sxyprn thumbnail refresh, WYŁĄCZONE (default 0). Token trafficdeposit jest
    # bucketowany godzinowo i ważny ~1h (weryfikacja 2026-06-10: stored ts=11:00 martwy
    # o 12:27, aktualny ts=13:00 żyje). Przechowywanie URL-i jest bezcelowe, wygasają
    # w godzinę, więc periodyczny refresh tylko wali w sxyprn na darmo. Działające
    # thumbnaile sxyprn wymagają ON-DEMAND resolve przy serwowaniu (proxy fetch post
    # page → bieżący og:image, cache ~45min). Job zostaje w kodzie ale domyślnie off.
    sxyprn_hours = cfg.get("sxyprn_thumb_refresh_hours", 0)
    if sxyprn_hours:
        batch = cfg.get("sxyprn_thumb_refresh_batch", 1200)
        sched.add_job(
            lambda: _job_refresh_sxyprn_thumbs(batch),
            IntervalTrigger(hours=sxyprn_hours, start_date=INTERVAL_ANCHOR),
            id="sxyprn_thumb_refresh",
            replace_existing=True,
            max_instances=1,
            coalesce=True,
        )
        log.info("scheduler: sxyprn-thumb-refresh every %dh (batch=%d)", sxyprn_hours, batch)

    return sched


DEFAULT_CONFIG: dict[str, Any] = {
    "tpdb_hours": 6,
    "stashdb_hours": 6,
    "sxyprn_thumb_refresh_hours": 0,  # off, token ~1h TTL, refresh bezcelowy (patrz register_jobs)
    "sxyprn_thumb_refresh_batch": 1200,
    "performer_driven_hours": 12,
    "performer_driven_top_n": 20,
    # Browse-latest, newest scenes z rich-metadata tubes. Co 6h (4×/dobę) × ~100
    # scen/tube/run łapie świeże sceny których performera jeszcze nie znamy (newcomerki
    # → canonical ingest dorobi potem). NB: ten DEFAULT_CONFIG jest poglądowy, realnie
    # run_forever() bierze interwały z Settings (config.py: sched_browse_latest_hours=6).
    "browse_latest_hours": 6,
    "browse_latest_max_pages": 5,
    # Movies, paradisehill + dooplay mirrory. Raz dziennie wystarczy (sites rosną
    # wolniej niż tube'y). Najwazniejsze: mirrory dorzucają native-friendly playback
    # sources do paradisehill movies → mobile gra natywnie zamiast WebView.
    "movie_ingest_hours": 24,
    # Continuous worker: tick co 15s, ale max_instances=1 + coalesce sprawia że
    # efektywny rate = max(15s, tick_duration). Tick z full coverage (25 tubes) ~50-80s,
    # więc realnie ~1 perf/60s. Przy 14.7k performerów = ~10 dni full sweep + refresh
    # każdego co 30 dni.
    "performer_continuous_seconds": 15,
    "performer_continuous_refresh_days": 30,
    # Taxonomy scene_count refresh, denormalizacja liczników dla /tags|/performers|
    # /studios|/favorites. Co 3h: counts do tego stale, dla sortu "popular" bez znaczenia.
    "taxonomy_counts_hours": 3,
    # Periodic reaper zombie 'running' runów, co 1h (próg 'running'>2h w funkcji).
    "reap_stuck_hours": 1,
    # Deep-crawl pełnych katalogów browse-tube'ów (Faza 2a). Co 1h, 60 stron/run,
    # round-robin po tube'ach. Backfill katalogów (porndoe ~62k itd.) przez dni.
    "deep_crawl_hours": 1,
    "deep_crawl_pages_per_run": 60,
    # Ingest freshness watchdog, per-origin alert do Sentry, co 6h. Browse próg 48h,
    # search (performer-driven) próg 7d (nierówna kadencja).
    "ingest_watchdog_hours": 6,
    "ingest_watchdog_max_age_hours": 48,
    "ingest_watchdog_search_max_age_hours": 168,
    # Hetzner Cloud bandwidth monitor, co 6h, alert Sentry przy progach % included.
    "hetzner_monitor_hours": 6,
    # Title+duration dedup, missing-merge tube-dupy, co 12h, playback-only.
    "title_dedup_hours": 12,
}