"""Alert evaluation (spec §140–141): match new events against `alerts.condition`
`{event_types, event_subtypes, min_importance, min_confidence, company, industries, countries, tags, metrics: {activity_score: {gt: 80}}}`
and record `alert_deliveries`. Channels: `web` (row only, the UI reads it) and `webhook` (POST JSON through the shared `fetch.Fetcher`
client after the SSRF guard, signed `X-CompanyAtlas-Signature: sha256=<hmac(body, alert id)>`). Email is stored as queued (no sender yet).

`evaluate_alerts(event_ids)` runs right after event creation; `alerts-sweep` catches up every 2 min via a watermark in settings_kv, and
evaluates metric-only alerts (no event) at most once per 24 h per alert.
"""
from __future__ import annotations

import hashlib
import hmac
import json
import logging
from datetime import UTC, datetime, timedelta
from typing import Any

from companyatlas.config import settings
from companyatlas.db import execute, fetch_all, fetch_val, jsonb, transaction
from companyatlas.fetch import BlockedDestination, Fetcher, validate_destination_async
from companyatlas.ids import new_id
from companyatlas.services.periodic import periodic

log = logging.getLogger(__name__)

WATERMARK_KEY = "alerts:last_event_created_at"
METRIC_ALERT_COOLDOWN_H = 24
_OPS = {"gt": lambda v, t: v > t, "gte": lambda v, t: v >= t, "lt": lambda v, t: v < t, "lte": lambda v, t: v <= t, "eq": lambda v, t: v == t}


def _lower_set(values: Any) -> set[str]:
    if not values:
        return set()
    if isinstance(values, str):
        values = [values]
    return {str(v).strip().lower() for v in values if str(v).strip()}


def metrics_match(cond: dict[str, Any], metrics: dict[str, float]) -> bool:
    spec = cond.get("metrics") or {}
    if not isinstance(spec, dict) or not spec:
        return True
    for metric, rule in spec.items():
        value = metrics.get(str(metric))
        if value is None:
            return False
        if isinstance(rule, dict):
            for op, threshold in rule.items():
                fn = _OPS.get(str(op))
                try:
                    if fn is None or not fn(float(value), float(threshold)):
                        return False
                except (TypeError, ValueError):
                    return False
        else:
            try:
                if float(value) < float(rule):
                    return False
            except (TypeError, ValueError):
                return False
    return True


def event_matches(alert: dict[str, Any], event: dict[str, Any], company: dict[str, Any], metrics: dict[str, float]) -> bool:
    """Pure matcher. `alert.company_id` pins a company; the condition narrows further."""
    cond = alert.get("condition") or {}
    if alert.get("company_id") and alert["company_id"] != event["company_id"]:
        return False
    if cond.get("company") and cond["company"] not in (company.get("slug"), company.get("id")):
        return False
    types = _lower_set(cond.get("event_types"))
    if types and str(event["event_type"]).lower() not in types and str(event["event_subtype"]).lower() not in types:
        return False
    subtypes = _lower_set(cond.get("event_subtypes"))
    if subtypes and str(event["event_subtype"]).lower() not in subtypes:
        return False
    if cond.get("min_importance") is not None and float(event.get("importance") or 0) < float(cond["min_importance"]):
        return False
    if cond.get("min_confidence") is not None and float(event.get("confidence") or 0) < float(cond["min_confidence"]):
        return False
    countries = {c.upper() for c in _lower_set(cond.get("countries"))}
    if countries and str(company.get("country") or "").upper() not in countries:
        return False
    industries = _lower_set(cond.get("industries"))
    if industries and not (industries & {str(i).lower() for i in (company.get("industries") or [])}):
        return False
    tags = _lower_set(cond.get("tags"))
    if tags and not (tags & {str(t).lower() for t in (event.get("tags") or [])}):
        return False
    return metrics_match(cond, metrics)


def sign(body: bytes, alert_id: str) -> str:
    return "sha256=" + hmac.new(alert_id.encode("utf-8"), body, hashlib.sha256).hexdigest()


def webhook_payload(alert: dict[str, Any], event: dict[str, Any] | None, company: dict[str, Any] | None, metrics: dict[str, float] | None = None) -> dict[str, Any]:
    return {
        "type": "event" if event else "metric",
        "alert": {"id": alert["id"], "name": alert.get("name"), "condition": alert.get("condition")},
        "company": {k: company.get(k) for k in ("id", "slug", "display_name", "canonical_domain", "country")} if company else None,
        "event": {k: event.get(k) for k in ("id", "event_type", "event_subtype", "importance", "confidence", "confidence_label", "title", "summary", "old_value",
                                           "new_value", "detected_at", "source_url", "surface", "origin", "status")} if event else None,
        "metrics": metrics or {},
        "delivered_at": datetime.now(UTC).isoformat(),
        "docs": f"{settings.site_url}/api",
    }


async def deliver_webhook(alert: dict[str, Any], payload: dict[str, Any]) -> tuple[str, str]:
    target = str(alert.get("target") or "").strip()
    if not target:
        return "failed", "no webhook target"
    try:
        await validate_destination_async(target)
    except BlockedDestination as exc:
        return "failed", f"blocked destination: {exc}"
    body = json.dumps(payload, default=str, ensure_ascii=False).encode("utf-8")
    headers = {"Content-Type": "application/json", "X-CompanyAtlas-Signature": sign(body, alert["id"]), "X-CompanyAtlas-Alert": alert["id"],
               "User-Agent": settings.user_agent}
    try:
        async with Fetcher(timeout_s=settings.webhook_timeout_s, http2=False, max_connections=4) as f:
            r = await f.client.post(target, content=body, headers=headers)
        if r.status_code < 300:
            return "sent", f"HTTP {r.status_code}"
        return "failed", f"HTTP {r.status_code}"
    except Exception as exc:  # noqa: BLE001
        return "failed", f"{exc.__class__.__name__}: {exc}"[:300]


async def _record(conn, alert: dict[str, Any], event_id: str | None, status: str, detail: str | None) -> None:  # type: ignore[no-untyped-def]
    await execute(conn, "insert into alert_deliveries (id, alert_id, event_id, channel, status, detail) values (:id, :a, :e, :ch, :st, :d)",
                  id=new_id("alert"), a=alert["id"], e=event_id, ch=alert.get("channel") or "web", st=status, d=detail)
    await execute(conn, "update alerts set last_fired_at = now() where id = :id", id=alert["id"])


async def evaluate_alerts(event_ids: list[str]) -> dict[str, int]:
    """Match the given events against all enabled alerts; deliver once per (alert, event)."""
    stats = {"events": 0, "matched": 0, "delivered": 0, "failed": 0}
    if not event_ids:
        return stats
    async with transaction() as conn:
        alerts = await fetch_all(conn, "select * from alerts where enabled")
        if not alerts:
            return stats
        events = await fetch_all(conn, """select e.*, co.slug, co.display_name, co.canonical_domain, co.country as company_country, co.industries as company_industries
                                          from events e join companies co on co.id = e.company_id where e.id = any(cast(:ids as text[])) and e.status in ('active', 'review')""",
                                 ids=event_ids)
        stats["events"] = len(events)
        metrics_cache: dict[str, dict[str, float]] = {}
        pending: list[tuple[dict[str, Any], dict[str, Any], dict[str, Any], dict[str, float]]] = []
        for ev in events:
            company = {"id": ev["company_id"], "slug": ev["slug"], "display_name": ev["display_name"], "canonical_domain": ev["canonical_domain"],
                       "country": ev["company_country"], "industries": ev["company_industries"]}
            for alert in alerts:
                cond = alert.get("condition") or {}
                needs_metrics = bool(cond.get("metrics"))
                if needs_metrics and ev["company_id"] not in metrics_cache:
                    metrics_cache[ev["company_id"]] = {r["metric"]: float(r["value"]) for r in await fetch_all(conn, "select metric, value from metrics_current where company_id = :c", c=ev["company_id"])}
                metrics = metrics_cache.get(ev["company_id"], {})
                if not event_matches(alert, ev, company, metrics):
                    continue
                dup = await fetch_val(conn, "select 1 from alert_deliveries where alert_id = :a and event_id = :e", a=alert["id"], e=ev["id"])
                if dup:
                    continue
                stats["matched"] += 1
                pending.append((alert, ev, company, metrics))
                if (alert.get("channel") or "web") != "webhook":
                    await _record(conn, alert, ev["id"], "delivered" if alert.get("channel") in (None, "web") else "queued", None)
                    stats["delivered"] += 1
    for alert, ev, company, metrics in pending:                # network outside the transaction
        if (alert.get("channel") or "web") != "webhook":
            continue
        status, detail = await deliver_webhook(alert, webhook_payload(alert, ev, company, metrics))
        async with transaction() as conn:
            await _record(conn, alert, ev["id"], status, detail)
        stats["delivered" if status == "sent" else "failed"] += 1
    return stats


async def evaluate_metric_alerts() -> dict[str, int]:
    """Alerts with only a metrics condition (no event filter) fire on the current metrics, at most once per cooldown."""
    stats = {"checked": 0, "fired": 0}
    fired: list[tuple[dict[str, Any], dict[str, Any], dict[str, float]]] = []
    async with transaction() as conn:
        alerts = await fetch_all(conn, """select * from alerts where enabled and jsonb_exists(condition, 'metrics') and coalesce(condition->'event_types', 'null'::jsonb) in ('null'::jsonb, '[]'::jsonb)
                                          and (last_fired_at is null or last_fired_at < :cutoff)""", cutoff=datetime.now(UTC) - timedelta(hours=METRIC_ALERT_COOLDOWN_H))
        for alert in alerts:
            stats["checked"] += 1
            cond = alert.get("condition") or {}
            if alert.get("company_id"):
                companies = await fetch_all(conn, "select id, slug, display_name, canonical_domain, country, industries from companies where id = :id", id=alert["company_id"])
            elif cond.get("company"):
                companies = await fetch_all(conn, "select id, slug, display_name, canonical_domain, country, industries from companies where slug = :s or id = :s", s=cond["company"])
            else:
                continue                                        # metric alerts must target one company
            for company in companies:
                metrics = {r["metric"]: float(r["value"]) for r in await fetch_all(conn, "select metric, value from metrics_current where company_id = :c", c=company["id"])}
                if not metrics or not metrics_match(cond, metrics):
                    continue
                detail = json.dumps({k: metrics.get(k) for k in (cond.get("metrics") or {})})
                if (alert.get("channel") or "web") == "webhook":
                    fired.append((alert, company, metrics))
                else:
                    await _record(conn, alert, None, "delivered", detail)
                stats["fired"] += 1
    for alert, company, metrics in fired:
        status, detail = await deliver_webhook(alert, webhook_payload(alert, None, company, metrics))
        async with transaction() as conn:
            await _record(conn, alert, None, status, detail)
    return stats


async def sweep(*, window_s: int | None = None) -> dict[str, int]:
    """Catch-up: evaluate events created since the watermark (bounded by the sweep window)."""
    window_s = window_s or settings.alerts_sweep_window_s
    async with transaction() as conn:
        raw = await fetch_val(conn, "select value from settings_kv where key = :k", k=WATERMARK_KEY)
        since = datetime.fromisoformat(str(raw).strip('"')) if raw else datetime.now(UTC) - timedelta(seconds=window_s)
        since = max(since, datetime.now(UTC) - timedelta(seconds=window_s))
        rows = await fetch_all(conn, "select id, created_at from events where created_at > :since and status in ('active', 'review') order by created_at limit 2000", since=since)
        if rows:
            await execute(conn, "insert into settings_kv (key, value, updated_at) values (:k, cast(:v as jsonb), now()) on conflict (key) do update set value = excluded.value, updated_at = now()",
                          k=WATERMARK_KEY, v=jsonb(rows[-1]["created_at"].isoformat()))
    stats = await evaluate_alerts([r["id"] for r in rows]) if rows else {"events": 0, "matched": 0, "delivered": 0, "failed": 0}
    stats.update({f"metric_{k}": v for k, v in (await evaluate_metric_alerts()).items()})
    return stats


@periodic("alerts-sweep", every_s=120, initial_delay_s=60)
async def alerts_sweep_task() -> None:
    stats = await sweep()
    if stats.get("matched") or stats.get("metric_fired"):
        log.info("alerts-sweep", extra=stats)


__all__ = ["deliver_webhook", "evaluate_alerts", "evaluate_metric_alerts", "event_matches", "metrics_match", "sign", "sweep", "webhook_payload"]