"""Certificate inventory service — status derivation, probe-target
building, probed-cert upsert, and the query/manual-entry surface the
router + MCP tools call.

Cron orchestration (the sweep loop, the expiry-alert fan-out) lives in
``app.scheduled_jobs.handlers``; the reusable pieces live here so the
router, MCP tools, and tests share one implementation.
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING

from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.cert_probe import DEFAULT_PORT, ProbedCert
from app.domains.cmdb.models import (
    CERT_SOURCE_MANUAL,
    CERT_SOURCE_PROBED,
    CERT_STATUS_EXPIRED,
    CERT_STATUS_EXPIRING,
    CERT_STATUS_UNREACHABLE,
    CERT_STATUS_VALID,
    CmdbCertificate,
    CmdbDnsRecord,
    CmdbDomain,
)

if TYPE_CHECKING:
    # Annotation-only. ``cert_owner_resolver`` is imported lazily inside
    # ``list_certificates`` to keep the module import graph acyclic, so the
    # name was previously undefined for every annotation consumer
    # (``typing.get_type_hints``, IDE/mypy resolution) — silent only
    # because ``from __future__ import annotations`` defers evaluation.
    from app.domains.cmdb.cert_owner_resolver import CertOwnerResolution

# Hostnames serving these record types are worth a 443 probe. NS / MX /
# TXT / SRV / CAA don't terminate TLS for the zone host, so probing them
# yields nothing useful.
PROBEABLE_RECORD_TYPES = frozenset({"A", "AAAA", "CNAME"})

# A cert within this many days of ``not_after`` is surfaced as
# ``expiring`` on the row's coarse status field. (The expiry-alert cron
# computes its own finer 30/7/1 buckets independently.)
EXPIRING_WINDOW_DAYS = 30


def derive_cert_status(not_after: datetime | None, now: datetime | None = None) -> str:
    """Coarse lifecycle status from ``not_after``."""
    if not_after is None:
        return CERT_STATUS_VALID  # nothing to warn on; treat as benign
    now = now or datetime.now(UTC)
    if not_after.tzinfo is None:
        not_after = not_after.replace(tzinfo=UTC)
    if now >= not_after:
        return CERT_STATUS_EXPIRED
    if not_after - now <= timedelta(days=EXPIRING_WINDOW_DAYS):
        return CERT_STATUS_EXPIRING
    return CERT_STATUS_VALID


# Written onto ``last_error`` for a probed cert whose host has left the probe
# target set. Cleared automatically by the next successful probe (the reachable
# branch of ``upsert_probed_certificate`` sets ``last_error = None``), so it can
# never stick after the DNS record comes back.
PROBE_TARGET_GONE_ERROR = (
    "probe target no longer in DNS (record disabled or removed) — "
    "certificate data is frozen at the last probe"
)


async def mark_orphaned_probe_targets(
    db: AsyncSession, target_hosts: set[str], now: datetime | None = None,
) -> int:
    """Flag probed certs whose host is no longer a probe target.

    ``build_probe_targets`` reads live DNS: a record that is disabled or
    removed at the provider correctly leaves the target set, and the sweep
    then never touches that certificate again. Nothing used to say so, and the
    stored ``status`` is only written at probe time — so the row became a
    zombie: frozen lifecycle, no signal, still counted against
    ``domain_certificate_health``.

    Production 2026-08-18 (this is not a broken prober — the prober is right):

      * ``tem.pingcap.net`` / ``tem-test.pingcap.net`` — live DNSPod record
        ``status='disable'``; last probed 2026-06-26, cert expired 06-27, still
        displaying ``expiring`` 52 days later.
      * ``do3.pingcap.net`` — no live DNS record at all (only soft-deleted
        legacy rows from the retired 腾讯云 / UCloud integrations).
      * ``download.pingcap.org`` — DNSPod record disabled AND deleted.

    Deliberately does NOT change ``status`` or ``not_after``: whether the
    endpoint is really gone (retire the row) or the DNS change was a mistake
    (re-enable it) is a human call. This only makes the condition visible —
    ``last_error`` explains it on the row, and the sweep's summary counts it.
    Returns the number of rows newly flagged.
    """
    now = now or datetime.now(UTC)
    rows = list((await db.execute(
        select(CmdbCertificate).where(and_(
            CmdbCertificate.deleted_at.is_(None),
            CmdbCertificate.source == CERT_SOURCE_PROBED,
            CmdbCertificate.probe_host.isnot(None),
        ))
    )).scalars().all())
    flagged = 0
    for row in rows:
        host = (row.probe_host or "").lower()
        if host in target_hosts:
            continue
        if row.last_error == PROBE_TARGET_GONE_ERROR:
            continue  # already flagged; don't churn updated_at every night
        row.last_error = PROBE_TARGET_GONE_ERROR
        flagged += 1
    return flagged


def effective_cert_status(
    row: CmdbCertificate, now: datetime | None = None,
) -> str:
    """Freshness-aware read of a certificate's coarse status.

    ``status`` is computed by :func:`derive_cert_status` at WRITE time (probe
    / sync / manual add) and never re-derived, so a host that stops being
    probed keeps whatever status its last probe produced — forever. Production
    2026-08-18: ``tem.pingcap.net``, ``tem-test.pingcap.net`` and
    ``do3.pingcap.net`` all still read ``expiring`` while their certificate
    had expired **52 days earlier**; their last probe was in June while every
    other host was probed that morning. The stored value contradicted the
    row's own ``not_after``.

    The lifecycle half of the status is a pure function of ``not_after``, so
    it is derived on read here. ``unreachable`` is NOT overwritten: that value
    records the *probe outcome* ("we could not verify this endpoint"), which is
    a different question from "has the certificate expired" — collapsing the
    two would hide a broken probe behind a lifecycle answer.

    Mirror of ``efficiency_service._effective_triage``: the stored column
    stays the source of truth for writers, and readers get the fresh view.
    Keep this in sync with :func:`effective_cert_status_matches`, which
    answers the same question for a filter.
    """
    if row.status == CERT_STATUS_UNREACHABLE:
        return CERT_STATUS_UNREACHABLE
    if row.not_after is None:
        return row.status
    return derive_cert_status(row.not_after, now)


def effective_cert_status_matches(
    row: CmdbCertificate, wanted: str, now: datetime | None = None,
) -> bool:
    """Whether ``row`` matches a ``status`` filter under the effective read.

    Filtering must use the same answer the row DISPLAYS, or "show me the
    expiring ones" returns a list whose rows say ``expired`` (and misses the
    stale-status rows that are the whole reason the effective read exists).
    Certificate volume is a few hundred rows and the list endpoint already
    filters owner in memory for the same reason, so this stays in Python
    rather than growing a SQL twin that could drift.
    """
    return effective_cert_status(row, now) == wanted


def fqdn_for(name: str, zone_name: str) -> str | None:
    """Resolve a DNS record (RR ``name`` + ``zone_name``) to a probeable
    FQDN. Returns ``None`` for wildcard records (can't probe ``*``)."""
    name = (name or "").strip().rstrip(".")
    zone = (zone_name or "").strip().rstrip(".")
    if not zone:
        return None
    if name in ("", "@", zone):
        return zone
    if name.startswith("*"):
        return None
    if name.endswith("." + zone):
        return name
    return f"{name}.{zone}"


def probed_dedup_key(host: str, port: int) -> str:
    return f"{CERT_SOURCE_PROBED}:{host}:{port}"


async def build_probe_targets(db: AsyncSession) -> list[dict]:
    """Build the de-duplicated list of ``{host, port, environment}`` probe
    targets from enabled-tenant DNS records + domains.

    Owner is intentionally NOT inherited here. Certificate ownership is
    resolved LIVE from the DNS record / zone domain at read / alert time
    (see ``cert_owner_resolver.CertOwnerResolver``); copying an owner onto
    the probed cert would re-introduce the snapshot drift the resolver
    exists to eliminate. ``environment`` is still inherited (it is an
    admin-assigned lifecycle classification, not a person reference).
    """
    from app.domains.cmdb.service import _active_cloud_integration_ids

    active = await _active_cloud_integration_ids(db)
    if not active:
        return []

    # Map zone_name → domain (for environment inheritance).
    domains = list((await db.execute(
        select(CmdbDomain).where(and_(
            CmdbDomain.deleted_at.is_(None),
            CmdbDomain.integration_id.in_(active),
        ))
    )).scalars().all())
    domain_by_name: dict[str, CmdbDomain] = {d.domain_name: d for d in domains}

    targets: dict[str, dict] = {}

    def _add(host: str | None, environment: str | None) -> None:
        if not host:
            return
        host = host.lower()
        existing = targets.get(host)
        if existing is None:
            targets[host] = {
                "host": host,
                "port": DEFAULT_PORT,
                "environment": environment,
            }
            return
        # Merge: prefer a non-null environment from any source.
        if existing["environment"] is None and environment is not None:
            existing["environment"] = environment

    # Apex domains.
    for dom in domains:
        _add(dom.domain_name, dom.environment)

    # DNS records (A/AAAA/CNAME, enabled, not deleted).
    records = list((await db.execute(
        select(CmdbDnsRecord).where(and_(
            CmdbDnsRecord.deleted_at.is_(None),
            CmdbDnsRecord.status == "enable",
            CmdbDnsRecord.record_type.in_(PROBEABLE_RECORD_TYPES),
            CmdbDnsRecord.integration_id.in_(active),
        ))
    )).scalars().all())
    for rec in records:
        host = fqdn_for(rec.name, rec.zone_name)
        if host is None:
            continue
        dom = domain_by_name.get(rec.zone_name)
        env = dom.environment if dom else None
        _add(host, env)

    return list(targets.values())


async def upsert_probed_certificate(
    db: AsyncSession,
    probed: ProbedCert,
    *,
    environment: str | None,
    now: datetime | None = None,
) -> tuple[CmdbCertificate | None, str]:
    """Insert/update a probed certificate row.

    Returns ``(row, action)`` where action is ``created`` / ``updated`` /
    ``unreachable`` / ``skipped``. An unreachable probe only marks an
    EXISTING row (never invents one) and never overwrites ``not_after``,
    so an owner is never falsely told a cert lapsed when Helix simply
    couldn't reach the host.
    """
    now = now or datetime.now(UTC)
    dedup_key = probed_dedup_key(probed.host, probed.port)
    row = (await db.execute(
        select(CmdbCertificate).where(CmdbCertificate.dedup_key == dedup_key)
    )).scalar_one_or_none()

    if not probed.ok:
        if row is None:
            return None, "skipped"
        row.status = CERT_STATUS_UNREACHABLE
        row.last_error = probed.error
        row.last_probed_at = now
        # Do NOT clear ``deleted_at`` on the unreachable path: an admin may
        # have explicitly retired this probed cert, and a failed probe is no
        # evidence the cert is back. We still record the unreachable status
        # for audit, but a soft-deleted row stays out of the active list.
        # (The reachable branch below intentionally un-retires, since a live
        # successful probe IS evidence the cert is still served.)
        return row, "unreachable"

    status = derive_cert_status(probed.not_after, now)
    if row is None:
        # No owner snapshot — ownership is live-resolved from DNS/domain.
        row = CmdbCertificate(
            id=uuid.uuid4(),
            dedup_key=dedup_key,
            source=CERT_SOURCE_PROBED,
            probe_host=probed.host,
            probe_port=probed.port,
            environment=environment,
            first_seen_at=now,
        )
        db.add(row)
        action = "created"
    else:
        action = "updated"
        # Resurrect a previously-swept row that came back.
        row.deleted_at = None

    row.common_name = probed.common_name
    row.sans = probed.sans
    row.issuer = probed.issuer
    row.serial_number = probed.serial_number
    row.fingerprint_sha256 = probed.fingerprint_sha256
    row.not_before = probed.not_before
    row.not_after = probed.not_after
    row.key_algorithm = probed.key_algorithm
    row.key_bits = probed.key_bits
    row.signature_algorithm = probed.signature_algorithm
    row.status = status
    row.last_error = None
    row.last_probed_at = now
    return row, action


# ── Query / read surface (router + MCP) ─────────────────────────────


async def list_certificates(
    db: AsyncSession,
    *,
    status: str | None = None,
    source: str | None = None,
    owner_person_id: str | None = None,
    environment: str | None = None,
    search: str | None = None,
    owner_ids_scope: list[str] | None = None,
    page: int = 1,
    page_size: int = 50,
) -> tuple[list[CmdbCertificate], int, dict[str, CertOwnerResolution]]:
    """List non-deleted certificates with optional filters + scope.

    Ownership is resolved LIVE (DNS record → zone domain → network owner)
    rather than read from a stored column, so both the ``owner_person_id``
    filter and ``owner_ids_scope`` (employee/manager scope) match against
    the RESOLVED effective owner. Because resolution can't be expressed in
    SQL, owner filtering + pagination happen in memory over the rows that
    pass the SQL-expressible filters (status/source/environment/search).
    Certificate volume is small (hundreds), so this is cheap.

    Returns ``(rows, total, resolution_by_cert_id)`` — pass the resolution
    map to the response builder so it can surface 本人 + source + 域名负责人
    without re-resolving.
    """
    from app.domains.cmdb.cert_owner_resolver import CertOwnerResolver
    from app.domains.cmdb.network_notifications import resolve_network_owner_id

    conds = [CmdbCertificate.deleted_at.is_(None)]
    # ``status`` is deliberately NOT filtered in SQL — see
    # ``effective_cert_status``: the stored column can contradict the row's own
    # ``not_after`` when probing stopped, and a SQL filter would disagree with
    # what the row renders. Applied in memory below, next to the owner filter
    # that is in memory for the same "can't be expressed in SQL" reason.
    if source:
        conds.append(CmdbCertificate.source == source)
    if environment:
        conds.append(CmdbCertificate.environment == environment)
    if search:
        like = f"%{search.strip()}%"
        conds.append(or_(
            CmdbCertificate.common_name.ilike(like),
            CmdbCertificate.probe_host.ilike(like),
            CmdbCertificate.issuer.ilike(like),
        ))

    # An empty explicit scope can short-circuit before any work.
    if owner_ids_scope is not None and not owner_ids_scope:
        return [], 0, {}

    all_rows = list((await db.execute(
        select(CmdbCertificate).where(and_(*conds))
        # NULLS LAST emulation (MySQL/TiDB reject NULLS LAST): order by
        # is-null first, then ascending not_after so the soonest-expiring
        # certs come first.
        .order_by(
            CmdbCertificate.not_after.is_(None),
            CmdbCertificate.not_after.asc(),
        )
    )).scalars().all())

    network_owner_id = await resolve_network_owner_id(db)
    resolver = await CertOwnerResolver.build(db, network_owner_id=network_owner_id)
    resolutions: dict[str, CertOwnerResolution] = {
        str(r.id): resolver.resolve(r) for r in all_rows
    }

    scope_set = set(owner_ids_scope) if owner_ids_scope is not None else None
    filtered: list[CmdbCertificate] = []
    for r in all_rows:
        if status and not effective_cert_status_matches(r, status):
            continue
        res = resolutions[str(r.id)]
        eff = res.owner_person_id
        if owner_person_id and str(eff or "") != str(owner_person_id):
            continue
        if scope_set is not None and str(eff or "") not in scope_set:
            continue
        filtered.append(r)

    total = len(filtered)
    page = max(1, page)
    page_size = max(1, min(200, page_size))
    start = (page - 1) * page_size
    page_rows = filtered[start:start + page_size]
    # Trim the resolution map to the rows we hand back.
    page_res = {str(r.id): resolutions[str(r.id)] for r in page_rows}
    return page_rows, total, page_res


async def get_certificate(db: AsyncSession, cert_id: str) -> CmdbCertificate | None:
    row = await db.get(CmdbCertificate, cert_id)
    if row is None or row.deleted_at is not None:
        return None
    return row


async def create_manual_certificate(
    db: AsyncSession,
    *,
    common_name: str,
    not_after: datetime | None,
    not_before: datetime | None = None,
    issuer: str | None = None,
    sans: list[str] | None = None,
    owner_person_id: str | None = None,
    environment: str | None = None,
    remark: str | None = None,
    actor_person_id: str | None = None,
    now: datetime | None = None,
) -> CmdbCertificate:
    now = now or datetime.now(UTC)
    row = CmdbCertificate(
        id=uuid.uuid4(),
        dedup_key=f"{CERT_SOURCE_MANUAL}:{uuid.uuid4()}",
        source=CERT_SOURCE_MANUAL,
        common_name=common_name,
        sans=sans or [],
        issuer=issuer,
        not_before=not_before,
        not_after=not_after,
        status=derive_cert_status(not_after, now),
        # An owner supplied at manual-entry time is a deliberate override.
        owner_person_id=owner_person_id,
        owner_is_manual=owner_person_id is not None,
        environment=environment,
        remark=remark,
        first_seen_at=now,
        last_modified_by_person_id=actor_person_id,
        last_modified_at=now,
    )
    db.add(row)
    return row


async def annotate_certificate(
    db: AsyncSession,
    cert: CmdbCertificate,
    *,
    fields_set: set[str],
    owner_person_id: str | None = None,
    environment: str | None = None,
    remark: str | None = None,
    actor_person_id: str | None = None,
    trace_id: str | None = None,
    now: datetime | None = None,
) -> CmdbCertificate:
    """Apply an admin PATCH to a certificate row.

    Only fields present in ``fields_set`` are written (so an omitted
    field is left unchanged, while an explicit ``null`` clears it). An
    owner change is recorded to ``cmdb_network_owner_history``.
    """
    from app.domains.cmdb.owner_history import (
        CHANGE_REASON_ADMIN_PATCH,
        RESOURCE_TYPE_CERTIFICATE,
        record_owner_change,
    )

    now = now or datetime.now(UTC)
    if "owner_person_id" in fields_set:
        # Only a deliberate override counts as the effective owner; the
        # previous owner for history purposes is whatever override was in
        # effect (a dormant legacy snapshot is not an "owner" here).
        prev_owner = cert.owner_person_id if cert.owner_is_manual else None
        new_owner = str(owner_person_id) if owner_person_id else None
        if prev_owner != new_owner:
            await record_owner_change(
                db,
                resource_type=RESOURCE_TYPE_CERTIFICATE,
                resource_id=str(cert.id),
                previous_owner_person_id=prev_owner,
                new_owner_person_id=new_owner,
                actor_person_id=actor_person_id,
                change_reason=CHANGE_REASON_ADMIN_PATCH,
                trace_id=trace_id,
            )
        # Setting an owner is an override; clearing it (null) hands the
        # cert back to live DNS/domain resolution.
        cert.owner_person_id = new_owner
        cert.owner_is_manual = new_owner is not None
    if "environment" in fields_set:
        cert.environment = environment
    if "remark" in fields_set:
        cert.remark = remark
    cert.last_modified_by_person_id = actor_person_id
    cert.last_modified_at = now
    return cert