"""The chain reconciler — what makes a chain a guarantee, not an event.

The executor already spawns a job's chained successors the moment it
finishes. That path is a latency optimisation and it has a hole exactly
the size of *"the event didn't happen"*: a deploy that cancelled the
parent mid-run, a worker OOM between the write-back and the spawn, a
Redis blip, a row an operator chained up two minutes ago whose parent had
already run this morning.

This job closes that hole by asking a question about **state** rather than
remembering an event: *is there a chained job whose parent has moved on
without it?* Two timestamps answer it, the condition self-clears once the
child runs, and it therefore converges regardless of how the fast path
was missed. See ``scheduled_jobs/chain.py`` for the predicate.

Cadence is 5 minutes rather than hourly because it advances **one hop per
tick** on purpose — firing a grandchild inside the same tick as its
parent would invert the very ordering the chain exists to establish. A
deploy landing in the middle of a seven-deep cloud chain therefore costs
at most ~35 minutes of catch-up, against a nightly window measured in
hours.
"""
from __future__ import annotations

import logging
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.scheduled_jobs.chain import plan_advances
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

CHAIN_ADVANCE_HANDLER = "scheduler.chain_advance"

#: Per-tick ceiling on how many successors this job will fire. A ceiling
#: exists because every advance takes a concurrency slot: a mass re-chain
#: (an operator wiring up thirteen cloud accounts at once, or a migration
#: seeding a fleet of chains) would otherwise release the whole fleet in
#: one tick and reproduce the pre-dawn pile-up this work set out to fix.
#: Anything over the cap simply waits for the next tick — it is still
#: overdue then, because the predicate is stateless.
MAX_ADVANCES_PER_TICK = 12


@register_scheduled_job(
    key=CHAIN_ADVANCE_HANDLER,
    display_name="任务链推进对账",
    description=(
        "每 5 分钟找出「上游已经跑过、自己却还没跟上」的 chained 任务并触发它。"
        "执行器在上游结束时会直接拉起下游 —— 那是快路径,只管延迟;这条任务"
        "才是保证:它比较的是两个时间戳而不是记住一个事件,所以无论快路径是"
        "怎么漏掉的(发布打断、worker OOM、Redis 抖动、刚刚才配好的依赖)都会"
        "收敛,而且条件在下游跑过之后自动消失。"
        "**每跳只推进一层** —— 在同一跳里把孙子也拉起来,恰好会破坏这套机制"
        "要建立的那个顺序。单次触发上限 12 个,防止一次批量改链把整个车队"
        "同时放出去、重演凌晨的排队。"
    ),
)
async def run_chain_advance(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.core import background
    from app.scheduled_jobs.service import execute_job

    try:
        cap = int(config.get("max_advances", MAX_ADVANCES_PER_TICK))
    except (TypeError, ValueError):
        cap = MAX_ADVANCES_PER_TICK
    cap = max(1, min(cap, 64))
    dry_run = bool(config.get("dry_run", False))

    overdue = await plan_advances(db)
    fired: list[dict[str, str]] = []
    for advance in overdue[:cap]:
        if not dry_run:
            background.spawn(
                execute_job(advance.job_id, triggered_by="chain_advance"),
                name=f"chain-advance-{advance.job_name}",
            )
        fired.append(
            {
                "job": advance.job_name,
                "after": advance.parent_name,
                "reason": advance.reason,
            },
        )

    if fired:
        logger.info(
            "chain_advance released %d overdue successor(s)%s: %s",
            len(fired),
            " (dry run)" if dry_run else "",
            ", ".join(f"{f['job']} after {f['after']}" for f in fired),
        )

    return {
        "overdue": len(overdue),
        "advanced": len(fired),
        "deferred_over_cap": max(0, len(overdue) - cap),
        "dry_run": dry_run,
        # Bounded on purpose: the console reads the full picture from
        # ``GET /scheduled-jobs/chains``, and a run summary that grew with
        # the fleet would bloat the ledger this job is meant to keep
        # moving.
        "sample": fired[:10],
    }