"""In-app scheduler — strangler-копия `app.services.scheduler` в scraper_kit (#2136).

Разница с боевым `app.services.scheduler`:

  1. Полная развязка от `app.*` — все продуктовые зависимости инжектируются через
     `SchedulerContext` (config/matcher/enrichment/session_factory/runs/shutdown),
     описанные Protocol'ами в `scraper_kit.contracts`. `grep "from app"` — пусто.

  2. **Registry-dispatch вместо copy-paste.** Боевой модуль держит 25+ почти
     одинаковых `trigger_*_run` (claim → new-session → create_task → close) плюс
     27-веточный `if/elif` в `scheduler_loop`. Здесь — единственный параметризованный
     путь `_dispatch()` + `HANDLERS: dict[str, Handler]`. Каждый source — одна запись
     реестра; общий claim/session/drain-boilerplate вынесен в `_dispatch`.

Kit-native (sweep-оркестраторы, уже перенесённые в `scraper_kit.orchestration.pipeline`)
регистрируются встроенно (`_default_kit_handlers`). Продуктовые джобы, тело которых
осталось в `app` (rosreestr_dkp / sber_index / deactivate_stale / *_backfill / …),
инжектируются извне как `Handler` через `build_registry(product_handlers=...)`.

Боевой рантайм (scraper-контейнер) по-прежнему крутит старый `app.services.scheduler` —
это COPY, не MOVE. Переключение — отдельный поздний strangler-шаг.

Критичная concurrency-логика (`_claim_run` advisory-lock + double-check, `reap_zombies`
порог, heartbeat, SIGTERM-drain) перенесена ДОСЛОВНО — тот же SQL, то же ветвление.
"""

from __future__ import annotations

import asyncio
import logging
import random
from collections.abc import Callable, Coroutine, Mapping
from dataclasses import dataclass, field
from datetime import UTC, datetime, time, timedelta
from typing import TYPE_CHECKING, Any

from sqlalchemy import text

from scraper_kit.orchestration import runs as _kit_runs
from scraper_kit.orchestration.pipeline import (
    run_avito_city_sweep,
    run_avito_full_load,
    run_avito_newbuilding_sweep,
    run_cian_city_sweep,
    run_cian_full_load,
    run_domclick_city_sweep,
    run_yandex_city_sweep,
)

if TYPE_CHECKING:
    from sqlalchemy.orm import Session

    from scraper_kit.contracts import (
        EnrichmentJobs,
        HouseMatcher,
        ScraperConfig,
        SessionFactory,
    )

logger = logging.getLogger(__name__)

# Loop interval — check каждую минуту
SCHEDULER_TICK_SEC = 60
ZOMBIE_THRESHOLD_HOURS = 6

# Бюджет дренажа детей < scheduler_main._DRAIN_TIMEOUT_S (100s) < docker grace (120s)
# (см. #1182 P2 — идентично боевому scheduler'у).
_CHILD_DRAIN_TIMEOUT_S = 80.0

# ── типы job/handler ─────────────────────────────────────────────────────────
# Job получает свежую сессию (открыта `_dispatch`), run_id, params и весь контекст
# (config/matcher/enrichment/runs) — чтобы иметь доступ к инжектированным зависимостям.
Job = Callable[["Session", int, dict[str, Any], "SchedulerContext"], Coroutine[Any, Any, None]]
# Pre-claim gate: вернуть False → пропустить claim (напр. cian: нет/протухли cookies).
PreClaim = Callable[["Session", dict[str, Any], "SchedulerContext"], Coroutine[Any, Any, bool]]
# Post-claim hook: сразу после успешного claim (напр. proxy_healthcheck sub-hourly reschedule).
PostClaim = Callable[["Session", int, dict[str, Any], "SchedulerContext"], None]


@dataclass(frozen=True)
class Handler:
    """Одна запись реестра source→обработчик.

    Инкапсулирует всё, что раньше размазывалось по индивидуальному `trigger_*_run`:
      - `job`        — асинхронное тело (SERP/backfill/enrichment). Общий boilerplate
                       (claim, свежая сессия, create_task, close, log) делает `_dispatch`.
      - `pre_claim`  — опциональный гейт ДО claim (cookie-проверка cian → defer).
      - `post_claim` — опциональный хук сразу ПОСЛЕ claim (sub-hourly reschedule).
      - `log_name`   — имя источника в логах ("scheduler: triggered <log_name> run_id=…").
    """

    job: Job
    log_name: str
    pre_claim: PreClaim | None = None
    post_claim: PostClaim | None = None


@dataclass
class SchedulerContext:
    """Инжектируемые зависимости + рантайм-состояние планировщика.

    Заменяет прямые импорты `app.core.config.settings` / `app.services.matching` /
    `app.services.house_imv_backfill` / `app.core.db.SessionLocal` /
    `app.services.scrape_runs` / `app.core.shutdown.shutdown_requested`.
    """

    config: ScraperConfig
    matcher: HouseMatcher
    enrichment: EnrichmentJobs
    session_factory: SessionFactory
    shutdown_requested: Callable[[], bool] = lambda: False
    # runs-модуль (create_run/update_heartbeat/mark_done/… ). По умолчанию — kit-копия
    # scraper_kit.orchestration.runs; в тестах подменяется recorder'ом.
    runs: Any = _kit_runs
    # #1182 P2: strong-ref'ы detached run-задач для graceful SIGTERM-drain'а.
    _inflight_tasks: set[asyncio.Task[None]] = field(default_factory=set)

    def spawn_tracked(self, coro: Coroutine[Any, Any, None]) -> asyncio.Task[None]:
        """create_task + регистрация в _inflight_tasks для graceful-drain'а (#1182 P2).

        strong-ref в set'е держит задачу до завершения (RUF006) и даёт scheduler_loop'у
        дождаться её на SIGTERM-drain'е; done-callback ретривит exception и убирает задачу.
        """
        task = asyncio.create_task(coro)
        self._inflight_tasks.add(task)

        def _on_done(t: asyncio.Task[None]) -> None:
            self._inflight_tasks.discard(t)
            if not t.cancelled():
                t.exception()

        task.add_done_callback(_on_done)
        return task

    async def drain_inflight(self) -> None:
        """Дождаться завершения detached run-задач перед teardown'ом (#1182 P2).

        Кооперативные дети дочекивают текущий unit + mark_done и резолвятся;
        некооперативные упираются в _CHILD_DRAIN_TIMEOUT_S и остаются на внешний
        hard-cancel fallback. Не busy-spin — один asyncio.wait.
        """
        pending = [t for t in self._inflight_tasks if not t.done()]
        if not pending:
            return
        logger.info("scheduler: draining %d in-flight run task(s) on shutdown", len(pending))
        _done, still = await asyncio.wait(pending, timeout=_CHILD_DRAIN_TIMEOUT_S)
        if still:
            logger.warning(
                "scheduler: %d task(s) did not drain in %.0fs — leaving for hard-cancel",
                len(still),
                _CHILD_DRAIN_TIMEOUT_S,
            )
        else:
            logger.info("scheduler: all in-flight run task(s) drained cleanly")


def compute_next_run_at(
    window_start_hour: int,
    window_end_hour: int,
    *,
    now: datetime | None = None,
    interval_days: int = 1,
) -> datetime:
    """Pick random datetime в window [start, end) UTC, через interval_days суток после now.

    interval_days задаёт каденс источника: 1 (default) = daily (back-compat), 7 = weekly.
    Берётся из schedule.default_params["interval_days"] вызывающим кодом (_claim_run /
    _defer_next_run_at); отсутствие ключа → 1 → прежнее ежедневное поведение.

    Если window_end_hour <= window_start_hour → cross-midnight window
    (например 22→3 → окно 22:00-23:59 ИЛИ 00:00-02:59).
    """
    now = now or datetime.now(tz=UTC)
    interval_days = max(1, int(interval_days))
    # Целевая дата = now + interval_days суток (interval_days=1 → завтра, как раньше).
    target = (now + timedelta(days=interval_days)).date()

    if window_end_hour > window_start_hour:
        # Обычное окно (например 2..5 → 02:00-04:59)
        start_seconds = window_start_hour * 3600
        end_seconds = window_end_hour * 3600
        rand_seconds = random.randint(start_seconds, end_seconds - 1)
        return datetime.combine(target, time(0, 0), tzinfo=UTC) + timedelta(seconds=rand_seconds)
    else:
        # Cross-midnight (22..3 → 22:00-23:59 + 00:00-02:59)
        # Длина окна = (24-start) + end часов
        total_seconds = ((24 - window_start_hour) + window_end_hour) * 3600
        rand_seconds = random.randint(0, total_seconds - 1)
        # Если rand попадает в первую часть (start..24)
        first_half = (24 - window_start_hour) * 3600
        if rand_seconds < first_half:
            # interval_days=1: текущая дата (если окно ещё не наступило сегодня) или next day.
            # interval_days>1: всегда целевая дата (стаггер на N суток вперёд).
            today_ok = interval_days == 1 and now.hour < window_start_hour
            base_date = now.date() if today_ok else target
            return datetime.combine(base_date, time(0, 0), tzinfo=UTC) + timedelta(
                seconds=window_start_hour * 3600 + rand_seconds
            )
        else:
            # Во второй части (0..end), целевого дня
            offset = rand_seconds - first_half
            return datetime.combine(target, time(0, 0), tzinfo=UTC) + timedelta(seconds=offset)


def has_running_run(db: Session, source: str) -> bool:
    """Есть ли активный run для source (status='running')."""
    row = db.execute(
        text(
            """
            SELECT 1 FROM scrape_runs
            WHERE source = :source AND status = 'running'
            LIMIT 1
            """
        ),
        {"source": source},
    ).fetchone()
    return row is not None


def reap_zombies(db: Session) -> int:
    """Mark scrape_runs as 'zombie' если heartbeat не обновлялся > ZOMBIE_THRESHOLD_HOURS hours."""
    zombie_interval = f"{ZOMBIE_THRESHOLD_HOURS} hours"
    result = db.execute(
        text(
            """
            UPDATE scrape_runs
            SET status = 'zombie', finished_at = NOW()
            WHERE status = 'running'
              AND (heartbeat_at IS NULL
                   OR heartbeat_at < NOW() - CAST(:interval AS interval))
            RETURNING id
            """
        ),
        {"interval": zombie_interval},
    )
    rows = result.fetchall()
    db.commit()
    if rows:
        logger.warning("scheduler: reaped %d zombie runs: %s", len(rows), [r.id for r in rows])
    return len(rows)


def _claim_run(db: Session, schedule_row: dict[str, Any], ctx: SchedulerContext) -> int | None:
    """Claim run: INSERT scrape_runs + UPDATE scrape_schedules next_run_at.

    Returns run_id или None если уже есть running run для этого source.
    Общий helper для всех source-handler'ов. Advisory-lock concurrency-логика перенесена
    ДОСЛОВНО из боевого scheduler'а (создание run — через инжектированный ctx.runs).
    """
    source = schedule_row["source"]
    if has_running_run(db, source):
        logger.info("scheduler: skip — already running for source=%s", source)
        return None

    # #750: атомарный claim через transaction-scoped advisory lock. Сериализует
    # конкурентные _claim_run одного source (redeploy-overlap старого+нового контейнера
    # / 2 реплики), иначе оба пройдут has_running_run-pre-check и запустят ДВОЙНОЙ sweep
    # → 2× rate к Avito/Cian → бан. pg_try_advisory_xact_lock НЕ блокирует: lock занят
    # другим тиком → false → return None. Лок авто-снимается на commit/rollback этой
    # транзакции (has_running_run выше — быстрый pre-check, обычно без взятия лока).
    got_lock = db.execute(
        text("SELECT pg_try_advisory_xact_lock(hashtext(:source))"),
        {"source": source},
    ).scalar()
    if not got_lock:
        logger.info("scheduler: skip — concurrent claim in progress for source=%s", source)
        return None

    # Double-checked под локом: конкурентный тик мог закоммитить running-run МЕЖДУ
    # pre-check и взятием лока (затем освободить lock на commit). Перепроверяем под
    # локом — иначе словим двойной run на узком окне.
    if has_running_run(db, source):
        logger.info("scheduler: skip — running appeared under lock for source=%s", source)
        db.rollback()  # освобождаем advisory lock (claim не состоялся)
        return None

    params = schedule_row.get("default_params") or {}
    run_id = ctx.runs.create_run(db, source=source, params=params)

    next_at = compute_next_run_at(
        schedule_row["window_start_hour"],
        schedule_row["window_end_hour"],
        interval_days=int(params.get("interval_days", 1)),
    )
    db.execute(
        text(
            """
            UPDATE scrape_schedules
            SET last_run_id = :run_id, last_run_at = NOW(),
                next_run_at = :next_at, updated_at = NOW()
            WHERE source = :source
            """
        ),
        {"run_id": run_id, "next_at": next_at, "source": source},
    )
    db.commit()
    logger.info(
        "scheduler: claimed run_id=%d source=%s next_run_at=%s",
        run_id,
        source,
        next_at.isoformat(),
    )
    return run_id


def _defer_next_run_at(db: Session, schedule_row: dict[str, Any]) -> None:
    """Сдвинуть next_run_at на следующее окно БЕЗ создания run (#1522).

    Используется когда pre_claim делает early-return (например cian: отсутствуют/протухли
    cookies) ДО _claim_run. Без этого get_due_schedules переотбирает schedule на каждом
    тике (SCHEDULER_TICK_SEC), и pre-check гоняется раз в минуту круглосуточно.
    """
    source = schedule_row["source"]
    params = schedule_row.get("default_params") or {}
    next_at = compute_next_run_at(
        schedule_row["window_start_hour"],
        schedule_row["window_end_hour"],
        interval_days=int(params.get("interval_days", 1)),
    )
    db.execute(
        text(
            """
            UPDATE scrape_schedules
            SET next_run_at = :next_at, updated_at = NOW()
            WHERE source = :source
            """
        ),
        {"next_at": next_at, "source": source},
    )
    db.commit()
    logger.info(
        "scheduler: deferred next_run_at source=%s next_run_at=%s (no run claimed)",
        source,
        next_at.isoformat(),
    )


def reschedule_after_minutes(*, param: str = "interval_minutes", default: int = 30) -> PostClaim:
    """Фабрика post_claim-хука: sub-hourly re-schedule сразу после claim (#2162).

    Боевой scheduler держит суточную гранулярность в compute_next_run_at, а
    proxy_healthcheck хочет каденс раз в N минут. Хук СРАЗУ после claim переопределяет
    next_run_at на now() + interval_minutes (source-специфично, не трогая shared
    compute_next_run_at). Если run ещё идёт на следующем тике — has_running_run в
    _claim_run вернёт None (skip) и next_run_at не сбросится → авто-throttle по факту.
    """

    def _hook(db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext) -> None:
        # source читаем из ЗАКЛЕЙМЛЕННОГО schedule — берём из params-owner через closure нельзя,
        # поэтому апдейтим по run_id → source (schedule owner). Проще: UPDATE по source из
        # scrape_runs текущего run_id, что эквивалентно WHERE source = <this source>.
        minutes = int(params.get(param, default))
        db.execute(
            text(
                """
                UPDATE scrape_schedules
                SET next_run_at = now() + make_interval(mins => CAST(:mins AS integer)),
                    updated_at = now()
                WHERE source = (SELECT source FROM scrape_runs WHERE id = CAST(:rid AS bigint))
                """
            ),
            {"mins": minutes, "rid": run_id},
        )
        db.commit()

    return _hook


# ── единственный параметризованный dispatch (заменяет 25 trigger_* + 27 if/elif) ──


async def _dispatch(
    handler: Handler,
    db: Session,
    schedule_row: dict[str, Any],
    ctx: SchedulerContext,
) -> int | None:
    """Общий путь claim→spawn для ЛЮБОГО source (весь boilerplate прежних trigger_*_run).

    1. Опциональный pre_claim-гейт (False → skip без claim).
    2. _claim_run (advisory-lock; None → уже running / concurrent claim).
    3. Опциональный post_claim-хук (sub-hourly reschedule).
    4. spawn detached run-задачи: свежая сессия → handler.job → close (+ exception-log).
    """
    if handler.pre_claim is not None:
        proceed = await handler.pre_claim(db, schedule_row, ctx)
        if not proceed:
            return None

    run_id = _claim_run(db, schedule_row, ctx)
    if run_id is None:
        return None

    params = schedule_row.get("default_params") or {}

    if handler.post_claim is not None:
        handler.post_claim(db, run_id, params, ctx)

    async def _run() -> None:
        run_db = ctx.session_factory()
        try:
            await handler.job(run_db, run_id, params, ctx)
        except Exception:
            logger.exception("scheduler: %s crashed run_id=%d", handler.log_name, run_id)
        finally:
            run_db.close()

    ctx.spawn_tracked(_run())
    logger.info("scheduler: triggered %s run_id=%d", handler.log_name, run_id)
    return run_id


def resolve_handler(source: str, registry: Mapping[str, Handler]) -> Handler | None:
    """Найти handler по source: exact-match, затем wildcard-префикс ("deactivate_stale_*").

    Единственная точка ветвления вместо 27 if/elif. Wildcard-ключ (кончается на "*")
    матчит по префиксу — покрывает семейство deactivate_stale_avito/yandex/cian одной
    записью реестра (как боевой `source.startswith("deactivate_stale_")`).
    """
    handler = registry.get(source)
    if handler is not None:
        return handler
    for key, h in registry.items():
        if key.endswith("*") and source.startswith(key[:-1]):
            return h
    return None


# ── kit-native sweep-обработчики (тело в scraper_kit.orchestration.pipeline) ──────
# Param-чтение идентично боевым trigger_*_run — та же семантика окон/дефолтов.


async def _job_avito_city_sweep(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    await run_avito_city_sweep(
        db,
        run_id=run_id,
        config=ctx.config,
        matcher=ctx.matcher,
        enrichment=ctx.enrichment,
        shutdown_requested=ctx.shutdown_requested,
        pages_per_anchor=int(params.get("pages_per_anchor", 3)),
        detail_top_n=int(params.get("detail_top_n", 20)),
        request_delay_sec=float(params.get("request_delay_sec", 7.0)),
        enrich_houses=bool(params.get("enrich_houses", True)),
        radius_m=int(params.get("radius_m", 1500)),
    )


async def _job_avito_newbuilding_sweep(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    await run_avito_newbuilding_sweep(
        db,
        run_id=run_id,
        config=ctx.config,
        matcher=ctx.matcher,
        shutdown_requested=ctx.shutdown_requested,
        pages=int(params.get("pages", 20)),
        request_delay_sec=float(params.get("request_delay_sec", 7.0)),
    )


async def _job_avito_full_load(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    _incremental_days = params.get("incremental_days")
    incremental_days = int(_incremental_days) if _incremental_days is not None else None
    await run_avito_full_load(
        db,
        run_id=run_id,
        config=ctx.config,
        matcher=ctx.matcher,
        shutdown_requested=ctx.shutdown_requested,
        price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
        concurrency=int(params.get("concurrency", 5)),
        request_delay_sec=float(params.get("request_delay_sec", 7.0)),
        secondary_only=bool(params.get("secondary_only", True)),
        resume_run_id=None,
        incremental_days=incremental_days,
    )


async def _job_avito_full_load_exhaustive(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    # incremental_days форсится в None → полный exhaustive обход (weekly cadence).
    await run_avito_full_load(
        db,
        run_id=run_id,
        config=ctx.config,
        matcher=ctx.matcher,
        shutdown_requested=ctx.shutdown_requested,
        price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
        concurrency=int(params.get("concurrency", 5)),
        request_delay_sec=float(params.get("request_delay_sec", 7.0)),
        secondary_only=bool(params.get("secondary_only", True)),
        resume_run_id=None,
        incremental_days=None,
    )


async def _job_yandex_city_sweep(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    await run_yandex_city_sweep(
        db,
        run_id=run_id,
        config=ctx.config,
        matcher=ctx.matcher,
        enrichment=ctx.enrichment,
        shutdown_requested=ctx.shutdown_requested,
        pages_per_anchor=int(params.get("pages_per_anchor", 2)),
        request_delay_sec=float(params.get("request_delay_sec", 9.0)),
        radius_m=int(params.get("radius_m", 1500)),
        enrich_address=bool(params.get("enrich_address", True)),
        segments=params.get("segments"),
    )


async def _job_cian_city_sweep(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    await run_cian_city_sweep(
        db,
        run_id=run_id,
        config=ctx.config,
        matcher=ctx.matcher,
        shutdown_requested=ctx.shutdown_requested,
        pages_per_anchor=int(params.get("pages_per_anchor", 3)),
        request_delay_sec=float(params.get("request_delay_sec", 5.0)),
        radius_m=int(params.get("radius_m", 1500)),
        detail_top_n=int(params.get("detail_top_n", 10)),
        enrich_houses=bool(params.get("enrich_houses", True)),
        newbuilding_only=bool(params.get("newbuilding_only", True)),
    )


async def _job_cian_full_load(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    await run_cian_full_load(
        db,
        run_id=run_id,
        config=ctx.config,
        matcher=ctx.matcher,
        shutdown_requested=ctx.shutdown_requested,
        price_cap_per_bucket=int(params.get("price_cap_per_bucket", 1400)),
        concurrency=int(params.get("concurrency", 5)),
        request_delay_sec=float(params.get("request_delay_sec", 4.0)),
        enrich_detail=bool(params.get("enrich_detail", False)),
        detail_top_n=int(params.get("detail_top_n", 0)),
        resume_run_id=None,
    )


async def _job_domclick_city_sweep(
    db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
    await run_domclick_city_sweep(
        db,
        run_id=run_id,
        config=ctx.config,
        matcher=ctx.matcher,
        shutdown_requested=ctx.shutdown_requested,
        city_id=int(params.get("city_id", 4)),
        rooms=params.get("rooms"),
        pages=int(params.get("pages_per_anchor", 5)),
        request_delay_sec=float(params.get("request_delay_sec", 6.0)),
    )


def _default_kit_handlers() -> dict[str, Handler]:
    """Встроенные kit-native sweep-обработчики (тело в scraper_kit.orchestration.pipeline)."""
    return {
        "avito_city_sweep": Handler(_job_avito_city_sweep, "avito_city_sweep"),
        "avito_newbuilding_sweep": Handler(_job_avito_newbuilding_sweep, "avito_newbuilding_sweep"),
        "avito_full_load": Handler(_job_avito_full_load, "avito_full_load"),
        "avito_full_load_exhaustive": Handler(
            _job_avito_full_load_exhaustive, "avito_full_load_exhaustive"
        ),
        "yandex_city_sweep": Handler(_job_yandex_city_sweep, "yandex_city_sweep"),
        "cian_city_sweep": Handler(_job_cian_city_sweep, "cian_city_sweep"),
        "cian_full_load": Handler(_job_cian_full_load, "cian_full_load"),
        "domclick_city_sweep": Handler(_job_domclick_city_sweep, "domclick_city_sweep"),
    }


def build_registry(product_handlers: Mapping[str, Handler] | None = None) -> dict[str, Handler]:
    """Собрать реестр source→Handler: kit-native sweeps + инжектированные продуктовые джобы.

    `product_handlers` — джобы, тело которых осталось в `app` (rosreestr_dkp / sber_index /
    deactivate_stale_* / *_backfill / cadastral_geo_match / house_* / proxy_healthcheck / …).
    Их строит app-side wiring и передаёт сюда. Продуктовые ключи МОГУТ переопределять
    kit-native (последнее слово за продуктом).

    Wildcard-ключи ("deactivate_stale_*") резолвятся `resolve_handler` по префиксу.
    """
    registry = dict(_default_kit_handlers())
    if product_handlers:
        registry.update(product_handlers)
    return registry


def get_due_schedules(db: Session) -> list[dict[str, Any]]:
    """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW())."""
    rows = (
        db.execute(
            text(
                """
            SELECT id, source, enabled, window_start_hour, window_end_hour,
                   default_params, last_run_id, last_run_at, next_run_at
            FROM scrape_schedules
            WHERE enabled = true
              AND (next_run_at IS NULL OR next_run_at <= NOW())
            """
            ),
        )
        .mappings()
        .all()
    )
    return [dict(r) for r in rows]


async def scheduler_loop(ctx: SchedulerContext, registry: Mapping[str, Handler]) -> None:
    """Бесконечный async loop — tick каждые SCHEDULER_TICK_SEC секунд.

    Структура (reap → get_due → dispatch → SIGTERM-drain) идентична боевому
    scheduler'у; 27-веточный if/elif заменён единственным `resolve_handler` + `_dispatch`.
    """
    logger.info("scheduler: started (tick=%ds)", SCHEDULER_TICK_SEC)
    # Initial sleep 30s чтобы дать FastAPI startup завершиться
    await asyncio.sleep(30)
    while True:
        # #1182 Phase 2: кооперативный SIGTERM-drain. Не reap'аем и не claim'аем
        # новые run'ы во время shutdown — даём текущему dispatch'у докатиться и выходим.
        if ctx.shutdown_requested():
            logger.info("scheduler: SIGTERM-drain — stop claiming new runs, exiting tick loop")
            break
        try:
            db = ctx.session_factory()
            try:
                # Reap zombies first
                reap_zombies(db)
                # Process due schedules
                due = get_due_schedules(db)
                for sch in due:
                    source = sch["source"]
                    handler = resolve_handler(source, registry)
                    if handler is None:
                        logger.warning("scheduler: unknown source=%s, skip", source)
                    else:
                        await _dispatch(handler, db, sch, ctx)

                    # #1182 Phase 2: после каждого dispatch'а проверяем drain — текущий
                    # run уже отпущен в свою asyncio-задачу (сам докоммитит/выйдет по
                    # своему checkpoint'у), а новые в этом тике не запускаем.
                    if ctx.shutdown_requested():
                        logger.info(
                            "scheduler: SIGTERM-drain — stop dispatch after source=%s", source
                        )
                        break
            finally:
                db.close()
        except Exception:
            logger.exception("scheduler: tick failed")
        # Промежуточная проверка перед 60s-сном: выходим сразу, не ждём целый тик.
        if ctx.shutdown_requested():
            logger.info("scheduler: SIGTERM-drain — exiting tick loop after dispatch")
            break
        await asyncio.sleep(SCHEDULER_TICK_SEC)

    # Tick-loop вышел только по SIGTERM-drain'у (иначе while True бесконечен): дожидаемся
    # detached run-задач (spawn_tracked) — пусть докоммитят текущий unit и сделают
    # mark_done, а не дадим asyncio.run() teardown'у хард-кансельнуть их mid-await (#1182).
    await ctx.drain_inflight()
    logger.info("scheduler: tick loop exited (in-flight drain complete)")


__all__ = [
    "SCHEDULER_TICK_SEC",
    "ZOMBIE_THRESHOLD_HOURS",
    "Handler",
    "SchedulerContext",
    "build_registry",
    "compute_next_run_at",
    "get_due_schedules",
    "has_running_run",
    "reap_zombies",
    "reschedule_after_minutes",
    "resolve_handler",
    "scheduler_loop",
]