"""Reusable, parameterised SQL for the public/admin API.

Rules: user input is always bound (`:param`), never interpolated; sort keys go through whitelists; every list is bounded; the
company card is built in two phases (page of ids with an indexed ORDER BY, then one card query for those ids) so the per-company
lateral aggregates only run for the rows that are returned.
"""
from __future__ import annotations

from datetime import UTC, date, datetime, timedelta
from typing import Any

from fastapi import HTTPException
from sqlalchemy.ext.asyncio import AsyncConnection

from companyatlas.db import fetch_all, fetch_one, fetch_val
from companyatlas.taxonomy import EventType

# ------------------------------------------------------------------------------------------------ windows / params

WINDOWS: dict[str, timedelta] = {"24h": timedelta(hours=24), "7d": timedelta(days=7), "30d": timedelta(days=30), "90d": timedelta(days=90),
                                 "1y": timedelta(days=365)}
COUNT_CAP = 10_000          # `total` is exact up to this bound (keeps deep pagination cheap on the big tables)
MAX_LIMIT = 500


def now_utc() -> datetime:
    return datetime.now(UTC)


def window_start(window: str, default: str = "7d") -> datetime:
    return now_utc() - WINDOWS.get(window, WINDOWS[default])


def days_ago(days: int) -> datetime:
    return now_utc() - timedelta(days=days)


def parse_iso(value: str | None, name: str = "since") -> datetime | None:
    if not value:
        return None
    v = value.strip()
    if v.endswith(("Z", "z")):
        v = v[:-1] + "+00:00"
    try:
        dt = datetime.fromisoformat(v)
    except ValueError:
        try:
            dt = datetime.combine(date.fromisoformat(v), datetime.min.time())
        except ValueError as exc:
            raise HTTPException(status_code=422, detail=f"{name}: invalid ISO-8601 timestamp") from exc
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=UTC)
    return dt.astimezone(UTC)


def parse_bool(value: str | bool | None) -> bool | None:
    if value is None or value == "":
        return None
    if isinstance(value, bool):
        return value
    return value.strip().lower() in ("1", "true", "yes", "on")


def clamp(value: int, lo: int, hi: int) -> int:
    return max(lo, min(hi, value))


def csv_list(value: str | None, *, maxlen: int = 20) -> list[str]:
    if not value:
        return []
    return [x.strip() for x in value.split(",") if x.strip()][:maxlen]


# ------------------------------------------------------------------------------------------------ companies


async def get_company(conn: AsyncConnection, key: str) -> dict[str, Any] | None:
    key = (key or "").strip()
    if not key or len(key) > 200:
        return None
    return await fetch_one(conn, "select * from companies where slug = :k or id = :k limit 1", k=key)


async def require_company(conn: AsyncConnection, key: str) -> dict[str, Any]:
    row = await get_company(conn, key)
    if row is None:
        raise HTTPException(status_code=404, detail="company not found")
    return row


COMPANY_CARD_SQL = """
select c.id, c.slug, c.display_name, c.legal_name, c.canonical_domain, c.website, c.description, c.industries, c.industry_primary,
       c.country, c.hq_city, c.hq_region, c.public_company, c.ticker, c.exchange, c.founded_year, c.employees_band, c.logo_url,
       c.status, c.onboarding_status, c.onboarding_error, c.importance, c.tier, c.indexed, c.stats, c.last_event_at, c.last_observed_at,
       c.first_observed_at, c.discovered_at, c.created_at, c.updated_at, c.source_meta->'profile' as profile,
       mx.metrics, sc.sensors, sc.observations, sc.changes, sc.events, jc.jobs_open{sparkline_col}
from companies c
left join lateral (select jsonb_object_agg(m.metric, m.value) as metrics from metrics_current m where m.company_id = c.id) mx on true
left join lateral (select count(*) filter (where s.status <> 'retired') as sensors, coalesce(sum(s.observation_count), 0) as observations,
                          coalesce(sum(s.change_count), 0) as changes, coalesce(sum(s.event_count), 0) as events
                   from sensors s where s.company_id = c.id) sc on true
left join lateral (select count(*) as jobs_open from jobs j where j.company_id = c.id and j.status = 'open') jc on true
{sparkline_join}
where c.id = any(cast(:ids as text[]))
"""
SPARKLINE_COL = ", sp.sparkline"
SPARKLINE_JOIN = ("left join lateral (select array_agg(x.value order by x.day) as sparkline from (select day, value from metric_series ms "
                  "where ms.company_id = c.id and ms.metric = 'activity_score' and ms.day >= :spark_from order by day desc limit 30) x) sp on true")

COMPANY_SORTS: dict[str, str] = {
    "activity": "ma.value desc nulls last, c.importance desc, c.id",
    "events": "c.last_event_at desc nulls last, c.importance desc, c.id",
    "hiring": "mh.value desc nulls last, c.importance desc, c.id",
    "name": "c.display_name asc, c.id",
    "importance": "c.importance desc, c.display_name asc, c.id",
    "recent": "c.discovered_at desc, c.id",
    # only valid together with a `q` filter (binds :q); list endpoints switch to it automatically when q is present
    "relevance": ("greatest(coalesce(ts_rank(c.search, websearch_to_tsquery('simple', :q)), 0), similarity(c.display_name, :q), "
                  "similarity(c.canonical_domain, :q)) desc, c.importance desc, c.id"),
}
COMPANY_SORT_JOINS: dict[str, str] = {
    "activity": "left join metrics_current ma on ma.company_id = c.id and ma.metric = 'activity_score'",
    "hiring": "left join metrics_current mh on mh.company_id = c.id and mh.metric = 'hiring_momentum_30d'",
}


async def fetch_cards_by_ids(conn: AsyncConnection, ids: list[str], *, sparkline: bool = False) -> list[dict[str, Any]]:
    """Card rows for the given ids, in the given order."""
    if not ids:
        return []
    sql = COMPANY_CARD_SQL.format(sparkline_col=SPARKLINE_COL if sparkline else "", sparkline_join=SPARKLINE_JOIN if sparkline else "")
    params: dict[str, Any] = {"ids": ids}
    if sparkline:
        params["spark_from"] = (now_utc() - timedelta(days=30)).date()
    rows = await fetch_all(conn, sql, **params)
    by_id = {r["id"]: r for r in rows}
    return [by_id[i] for i in ids if i in by_id]


async def company_page_ids(conn: AsyncConnection, where: list[str], params: dict[str, Any], *, sort: str = "activity", limit: int = 25,
                           offset: int = 0, extra_join: str = "") -> tuple[list[str], int]:
    """Phase 1 of a company list: ids for the page + bounded total."""
    sort = sort if sort in COMPANY_SORTS else "activity"
    if sort == "relevance" and "q" not in params:
        sort = "activity"
    elif sort == "activity" and params.get("q"):
        sort = "relevance"
    join = " ".join(x for x in (COMPANY_SORT_JOINS.get(sort, ""), extra_join) if x)
    where_sql = (" where " + " and ".join(where)) if where else ""
    rows = await fetch_all(conn, f"select c.id from companies c {join}{where_sql} order by {COMPANY_SORTS[sort]} limit :limit offset :offset",
                           **params, limit=limit, offset=offset)
    total = await bounded_count(conn, f"from companies c {extra_join}{where_sql}", params)
    return [r["id"] for r in rows], total


async def bounded_count(conn: AsyncConnection, from_where_sql: str, params: dict[str, Any], cap: int = COUNT_CAP) -> int:
    val = await fetch_val(conn, f"select count(*) from (select 1 {from_where_sql} limit :cap) t", **params, cap=cap)
    return int(val or 0)


def company_filters(*, q: str | None = None, country: str | None = None, industry: str | None = None, tier: int | None = None,
                    public: bool | None = None, status: str | None = None, has_events: bool | None = None,
                    onboarding_status: str | None = None) -> tuple[list[str], dict[str, Any]]:
    where: list[str] = []
    params: dict[str, Any] = {}
    if q:
        q = q.strip()[:200]
        params["q"] = q
        if len(q) >= 3:
            where.append("(c.search @@ websearch_to_tsquery('simple', :q) or c.display_name % :q or c.canonical_domain % :q)")
        else:
            where.append("(c.display_name ilike :q_prefix or c.canonical_domain ilike :q_prefix)")
            params["q_prefix"] = q.replace("%", "").replace("_", "") + "%"
    if country:
        where.append("c.country = cast(:country as char(2))")
        params["country"] = country.strip().upper()[:2]
    if industry:
        where.append("cast(:industry as text) = any(c.industries)")
        params["industry"] = industry.strip()[:80]
    if tier is not None:
        where.append("c.tier = :tier")
        params["tier"] = tier
    if public is not None:
        where.append("c.public_company = :public")
        params["public"] = public
    if status:
        where.append("c.status = cast(:status as text)")
        params["status"] = status.strip().upper()[:40]
    if has_events is True:
        where.append("c.last_event_at is not null")
    elif has_events is False:
        where.append("c.last_event_at is null")
    if onboarding_status:
        where.append("c.onboarding_status = cast(:onboarding_status as text)")
        params["onboarding_status"] = onboarding_status.strip().lower()[:40]
    return where, params


# ------------------------------------------------------------------------------------------------ events

EVENT_SELECT = """
select e.id, e.company_id, e.sensor_id, e.change_id, e.cluster_id, e.surface, e.event_type, e.event_subtype, e.importance, e.confidence,
       e.confidence_label, e.title, e.summary, e.old_value, e.new_value, e.payload, e.entities, e.tags, e.detected_at, e.effective_at,
       e.published_at, e.source_url, e.snapshot_before, e.snapshot_after, e.language, e.origin, e.model_name, e.prompt_version, e.status,
       e.retracted_reason,
       c.slug as company_slug, c.display_name as company_display_name, c.canonical_domain as company_domain, c.country as company_country,
       c.logo_url as company_logo_url
from events e join companies c on c.id = e.company_id
"""
EVENT_SORTS: dict[str, str] = {"recent": "e.detected_at desc, e.id desc", "importance": "e.importance desc, e.detected_at desc, e.id desc"}
_EVENT_TYPES = {t.value for t in EventType}


def event_filters(*, company_id: str | None = None, event_type: str | None = None, event_subtype: str | None = None, country: str | None = None,
                  industry: str | None = None, since: datetime | None = None, until: datetime | None = None, min_importance: float | None = None,
                  min_confidence: float | None = None, q: str | None = None, surface: str | None = None, origin: str | None = None,
                  status: str | None = "active", event_types: list[str] | None = None,
                  event_subtypes: list[str] | None = None) -> tuple[list[str], dict[str, Any]]:
    where: list[str] = []
    params: dict[str, Any] = {}
    if status:
        where.append("e.status = cast(:e_status as text)")
        params["e_status"] = status
    if company_id:
        where.append("e.company_id = :e_company_id")
        params["e_company_id"] = company_id
    if event_type:
        where.append("e.event_type = cast(:e_type as text)")
        params["e_type"] = event_type.strip().upper()[:40]
    if event_types:
        where.append("e.event_type = any(cast(:e_types as text[]))")
        params["e_types"] = [t.upper()[:40] for t in event_types][:30]
    if event_subtype:
        where.append("e.event_subtype = cast(:e_subtype as text)")
        params["e_subtype"] = event_subtype.strip().upper()[:60]
    if event_subtypes:
        where.append("e.event_subtype = any(cast(:e_subtypes as text[]))")
        params["e_subtypes"] = [t.upper()[:60] for t in event_subtypes][:60]
    if country:
        where.append("c.country = cast(:e_country as char(2))")
        params["e_country"] = country.strip().upper()[:2]
    if industry:
        where.append("cast(:e_industry as text) = any(c.industries)")
        params["e_industry"] = industry.strip()[:80]
    if since is not None:
        where.append("e.detected_at >= :e_since")
        params["e_since"] = since
    if until is not None:
        where.append("e.detected_at < :e_until")
        params["e_until"] = until
    if min_importance is not None:
        where.append("e.importance >= :e_min_imp")
        params["e_min_imp"] = float(min_importance)
    if min_confidence is not None:
        where.append("e.confidence >= :e_min_conf")
        params["e_min_conf"] = float(min_confidence)
    if q:
        where.append("e.search @@ websearch_to_tsquery('english', :e_q)")
        params["e_q"] = q.strip()[:200]
    if surface:
        where.append("e.surface = cast(:e_surface as text)")
        params["e_surface"] = surface.strip().lower()[:40]
    if origin:
        where.append("e.origin = cast(:e_origin as text)")
        params["e_origin"] = origin.strip().lower()[:20]
    return where, params


async def fetch_events(conn: AsyncConnection, where: list[str], params: dict[str, Any], *, sort: str = "recent", limit: int = 50,
                       offset: int = 0) -> list[dict[str, Any]]:
    order = EVENT_SORTS.get(sort, EVENT_SORTS["recent"])
    where_sql = (" where " + " and ".join(where)) if where else ""
    return await fetch_all(conn, f"{EVENT_SELECT}{where_sql} order by {order} limit :limit offset :offset", **params,
                           limit=clamp(limit, 1, MAX_LIMIT), offset=max(0, offset))


async def count_events(conn: AsyncConnection, where: list[str], params: dict[str, Any]) -> int:
    where_sql = (" where " + " and ".join(where)) if where else ""
    return await bounded_count(conn, f"from events e join companies c on c.id = e.company_id{where_sql}", params)


async def fetch_event(conn: AsyncConnection, event_id: str) -> dict[str, Any] | None:
    return await fetch_one(conn, f"{EVENT_SELECT} where e.id = :id", id=event_id)


# ------------------------------------------------------------------------------------------------ metrics helpers


async def metric_series(conn: AsyncConnection, company_id: str, metric: str, days: int) -> list[dict[str, Any]]:
    return await fetch_all(conn, "select day, value, confidence from metric_series where company_id = :cid and metric = :m and day >= :d "
                                 "order by day asc limit :lim", cid=company_id, m=metric, d=days_ago(days).date(), lim=clamp(days, 1, 2000))


async def metric_values_at(conn: AsyncConnection, ids: list[str], metric: str, on_or_before: date) -> dict[str, float]:
    """Latest series value on/before a day per company (used for window deltas)."""
    if not ids:
        return {}
    rows = await fetch_all(conn, "select distinct on (company_id) company_id, value from metric_series where metric = :m and "
                                 "company_id = any(cast(:ids as text[])) and day <= :d order by company_id, day desc", m=metric, ids=ids, d=on_or_before)
    return {r["company_id"]: float(r["value"]) for r in rows}


async def settings_value(conn: AsyncConnection, key: str) -> Any:
    return await fetch_val(conn, "select value from settings_kv where key = :k", k=key)


__all__ = ["COMPANY_SORTS", "COUNT_CAP", "EVENT_SORTS", "MAX_LIMIT", "WINDOWS", "bounded_count", "clamp", "company_filters", "company_page_ids",
           "count_events", "csv_list", "days_ago", "event_filters", "fetch_cards_by_ids", "fetch_event", "fetch_events", "get_company",
           "metric_series", "metric_values_at", "now_utc", "parse_bool", "parse_iso", "require_company", "settings_value", "window_start"]