"""Monthly evidence pre-compile — collect the audit season's evidence early.

An auditor asking for "everyone who held production cloud access in May"
should not be waiting on a query in November. This job runs the
catalogue's company-wide questions once per month over the closed month
and freezes the bundles, so the evidence for a period exists **from
within that period** rather than being reconstructed under deadline.

Which queries run is derived, not listed: any catalogue entry whose
required parameters are empty is company-wide and therefore
pre-compilable. Person- and object-scoped questions (``person_footprint``,
``revocation_proof``, ``object_lifecycle``) need a subject and stay
on-demand — a hand-maintained list here would drift from the catalogue
the first time someone adds a query.

Control coverage comes from the same place: each definition already
declares the ``control_keys`` it evidences, so the run summary reports
which controls now have month-stamped evidence without a second mapping
table to keep in sync.
"""
from __future__ import annotations

import calendar
from datetime import date, datetime, timedelta
from typing import Any
from zoneinfo import ZoneInfo

from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.auditor.evidence_query.catalog import EVIDENCE_QUERIES
from app.domains.auditor.evidence_query.service import (
    GENERATED_VIA_SCHEDULED,
    run_evidence_query,
)
from app.scheduled_jobs.registry import register_scheduled_job
from app.scheduled_jobs.slo import SLO_DAILY_DIGEST

# The month a period NAMES is a calendar month in the business's own
# timezone, not in UTC. The cron fires at Asia/Shanghai so a scheduled
# run agrees either way, but a MANUAL trigger in the CST small hours is
# still the previous day in UTC — and on the 1st that silently picks the
# wrong month. Same constant, same reason, as ai_spending_handlers.
_PERIOD_TZ = ZoneInfo("Asia/Shanghai")


def precompilable_queries() -> list[Any]:
    """Catalogue entries that need no subject to be meaningful.

    Derived from the declared parameter list rather than hard-coded: a
    list here would silently stop covering a query the moment someone
    registers a new one.
    """
    return [
        definition for definition in EVIDENCE_QUERIES
        if not any(param.required for param in definition.params)
    ]


def closed_month_window(today: date) -> tuple[str, str]:
    """The previous whole calendar month, as ISO dates.

    Deliberately the *closed* month: running over the current month
    would freeze a partial picture and stamp it with a hash, and the
    next run would differ for no reason other than time passing —
    which is exactly the signal the hash is supposed to reserve for
    "the underlying data changed".
    """
    # Stepping back one day from the 1st lands in the previous month
    # regardless of its length; ``calendar`` then gives that month's
    # final day.
    last_of_prev = today.replace(day=1) - timedelta(days=1)
    start = last_of_prev.replace(day=1)
    end = last_of_prev.replace(
        day=calendar.monthrange(last_of_prev.year, last_of_prev.month)[1],
    )
    return start.isoformat(), end.isoformat()


@register_scheduled_job(
    key="auditor.evidence_precompile",
    display_name="审计证据月度预编译",
    description=(
        "每月 2 号 03:20(Asia/Shanghai)把证据查询目录里所有「无必填参数」的"
        "公司级问题按**上一个完整自然月**跑一遍并冻结成 bundle,供审计季直接取用 —— "
        "证据在所属周期内就已存在,而不是等到审计季在 deadline 下现场重建。\n"
        "跑哪些查询是从目录的参数声明**推导**的,不是写死清单:任何必填参数为空的条目"
        "都是公司级问题。需要主体的问题(单人足迹 / 回收证明 / 对象生命周期)保持按需触发。\n"
        "只读:不调用任何外部接口、不调用模型,只对 Helix 自己的台账做确定性查询。\n"
        "config={\"month\":\"YYYY-MM\"} 可覆写窗口(补跑历史月份);"
        "{\"query_keys\":[...]} 可限定只跑指定查询。"
    ),
    slo=SLO_DAILY_DIGEST,
    weight="heavy",
)
async def run_evidence_precompile(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    """Freeze one bundle per company-wide query for the closed month."""

    override_month = config.get("month")
    if override_month:
        year, month = (int(part) for part in str(override_month).split("-"))
        start = date(year, month, 1)
        end = date(year, month, calendar.monthrange(year, month)[1])
        period_start, period_end = start.isoformat(), end.isoformat()
    else:
        # Local date, not UTC. The cron fires at 03:20 Asia/Shanghai, so
        # the two agree on a scheduled run — but a MANUAL trigger between
        # 00:00 and 08:00 CST is still the previous day in UTC, and on
        # the 1st that silently pre-compiles the wrong month.
        period_start, period_end = closed_month_window(
            datetime.now(_PERIOD_TZ).date(),
        )

    known = {d.key for d in precompilable_queries()}
    only = set(config.get("query_keys") or [])
    # A typo'd key would otherwise select nothing, compile nothing, and
    # record ``ok`` — a month with no evidence reported as a healthy run.
    unknown_keys = sorted(only - known)
    if unknown_keys:
        raise RuntimeError(
            "evidence pre-compile: config.query_keys names "
            f"{', '.join(unknown_keys)}, which is not a company-wide "
            f"query. Known: {', '.join(sorted(known))}"
        )
    definitions = [
        d for d in precompilable_queries()
        if not only or d.key in only
    ]

    compiled: list[dict[str, Any]] = []
    skipped: list[dict[str, str]] = []
    failed: list[dict[str, str]] = []
    controls: set[str] = set()

    for definition in definitions:
        params: dict[str, Any] = {}
        declared = {p.name for p in definition.params}
        if "period_start" in declared:
            params["period_start"] = period_start
        if "period_end" in declared:
            params["period_end"] = period_end
        # Point-in-time questions get the month's final instant rather
        # than a range: "who held admin in May" is answered as of the
        # end of May, not as an interval.
        if "at" in declared and "period_start" not in declared:
            params["at"] = period_end

        # Re-running a month must not double its evidence. The handler is
        # re-triggerable by hand (that is what ``config.month`` is for)
        # and the runner retries, so without this a backfill silently
        # stacks a second identical bundle per query and the pack section
        # renders each exhibit twice.
        existing = await _existing_bundle(db, definition.key, params)
        if existing is not None:
            skipped.append({
                "query_key": definition.key,
                "bundle_id": str(existing.id),
                "content_hash": existing.content_hash,
            })
            continue

        try:
            async with db.begin_nested():
                bundle = await run_evidence_query(
                    db,
                    query_key=definition.key,
                    params=params,
                    actor_user_id=None,
                    generated_via=GENERATED_VIA_SCHEDULED,
                )
        except Exception as exc:  # noqa: BLE001
            # The savepoint above means this query's partial writes are
            # already discarded and the session is usable again. Without
            # it, a DB-level error would poison the session and every
            # remaining query would fail with PendingRollbackError,
            # misreported as nine independent source failures.
            failed.append({"query_key": definition.key, "error": str(exc)[:200]})
            continue
        # Commit each bundle as it lands. This remains deliberately outside
        # the recoverable-query block: a failed commit leaves the outer
        # session unusable, so swallowing it and continuing would turn one
        # transaction failure into misleading per-query results.
        await db.commit()
        compiled.append({
            "query_key": definition.key,
            "bundle_id": str(bundle.id),
            "row_count": bundle.row_count,
            "content_hash": bundle.content_hash,
        })
        controls.update(definition.control_keys)

    if failed:
        # Raise so the job records ``failed`` rather than ``ok``: a run
        # that quietly reports success while an exhibit is missing is
        # the failure mode that makes a monthly control look healthy for
        # months. The successful bundles are committed above, so raising
        # costs the month nothing beyond the queries that actually
        # failed.
        raise RuntimeError(
            f"evidence pre-compile: {len(failed)} query/queries failed for "
            f"{period_start}..{period_end}: "
            + "; ".join(f"{f['query_key']}={f['error']}" for f in failed)
        )

    return {
        "period_start": period_start,
        "period_end": period_end,
        "queries_compiled": len(compiled),
        "queries_already_present": len(skipped),
        "controls_evidenced": sorted(controls),
        "bundles": compiled,
        "model_calls": 0,
    }


__all__ = [
    "closed_month_window",
    "precompilable_queries",
    "run_evidence_precompile",
]


async def _existing_bundle(
    db: AsyncSession,
    query_key: str,
    params: dict[str, Any],
) -> Any:
    """This month's scheduled bundle for this query, if one exists.

    Matched on (query, version, generated_via, canonical params) rather
    than on a unique constraint: the bundle table deliberately allows
    repeats, because an auditor re-asking the same question and getting
    a second bundle with a different hash IS the finding. Only the
    SCHEDULED producer needs to be idempotent, and only for the exact
    question it asks.

    The params, NOT the ``period_start`` column. The first version
    compared ``period_start.date()`` against the month's first day and
    so never matched a point-in-time query: those are asked ``at`` the
    month's LAST day, and ``_period_bounds`` writes that single instant
    into both period columns. ``privileged_role_roster`` therefore
    duplicated on every retry while the other eight skipped correctly —
    and the test that was supposed to cover this used a period query, so
    it stayed green. The params are what identify the question; compare
    those.
    """
    from sqlalchemy import select  # noqa: PLC0415

    from app.domains.auditor.evidence_query.catalog import (  # noqa: PLC0415
        get_query,
    )
    from app.domains.auditor.evidence_query.contracts import (  # noqa: PLC0415
        canonical_params,
    )
    from app.domains.auditor.models import (  # noqa: PLC0415
        AuditorEvidenceBundle,
    )

    definition = get_query(query_key)
    if definition is None:
        return None
    wanted = canonical_params(params)
    rows = (await db.execute(
        select(AuditorEvidenceBundle).where(
            AuditorEvidenceBundle.query_key == query_key,
            AuditorEvidenceBundle.query_version == definition.version,
            AuditorEvidenceBundle.generated_via == GENERATED_VIA_SCHEDULED,
        )
    )).scalars().all()
    for row in rows:
        if canonical_params(row.params or {}) == wanted:
            return row
    return None