"""Data Queries — a REST call or a read-only SQL statement whose result rows
are used like a datasource's records: by page tables and charts, dashboards,
saved datasets (source_type "query", see app/crud/dataset_query.py) and form
choice fields (FormFieldDef.options_query_id).

Always run here on the server, never in the browser:

* {{param.NAME}} is filled from the caller's parameters (URL-encoded in a
  REST URL). A SQL query binds them as :NAME placeholders instead.
* {{secret.NAME}} is filled in only on the way out (URL, headers, body), and
  every secret value is masked in what comes back — the rows are shown to
  people (see app/crud/secrets.py).
* Results are cached per parameter set for cache_seconds, so a busy page
  doesn't call the API once per visitor.
* A REST URL must be http(s) and, unless DATA_QUERY_ALLOW_PRIVATE_HOSTS is
  set, may not point at this machine's own network.
"""

import asyncio
import ipaddress
import json
import re
import socket
import time
import uuid
from dataclasses import dataclass
from typing import Any
from urllib.parse import quote, urlsplit

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.core.tls import TRUSTSTORE_SSL_CONTEXT
from app.crud import secrets as secrets_crud
from app.crud.connectors import get_connector
from app.models.data_query import DataQuery

MAX_ROWS = 5000
MAX_BYTES = 5_000_000
TIMEOUT = 15.0
_CACHE_ENTRIES = 500
_PARAM = re.compile(r"\{\{\s*param\.([A-Za-z_][A-Za-z0-9_]*)\s*\}\}")


class DataQueryError(Exception):
    pass


@dataclass
class QueryResult:
    columns: list[str]
    rows: list[dict]
    cached: bool = False


# ---- Parameters ---------------------------------------------------------------------------------


def param_values(defs: list[dict], given: dict | None) -> dict[str, Any]:
    """The declared parameters' values: what the caller gave, else the
    default. Undeclared names are ignored; a required one left blank is an
    error."""
    given = given or {}
    values: dict[str, Any] = {}
    for p in defs:
        name = p["name"]
        value = given.get(name)
        if value is None or value == "":
            value = p.get("default")
        if (value is None or value == "") and p.get("required"):
            raise DataQueryError(f"The parameter {p.get('label') or name} is required.")
        values[name] = value
    return values


def _fill_params(text: str, values: dict[str, Any], *, encode: bool = False) -> str:
    def repl(m: re.Match) -> str:
        v = values.get(m.group(1))
        s = "" if v is None else str(v)
        return quote(s, safe="") if encode else s

    return _PARAM.sub(repl, text)


# ---- REST ---------------------------------------------------------------------------------------


def _check_host(url: str) -> None:
    parts = urlsplit(url)
    if parts.scheme not in ("http", "https") or not parts.hostname:
        raise DataQueryError("The URL must start with http:// or https://.")
    try:
        infos = socket.getaddrinfo(parts.hostname, parts.port or (443 if parts.scheme == "https" else 80))
    except socket.gaierror as exc:
        raise DataQueryError(f"Couldn't find the host {parts.hostname}.") from exc
    for info in infos:
        addr = ipaddress.ip_address(info[4][0])
        if addr.is_link_local or addr.is_multicast or addr.is_unspecified:
            raise DataQueryError("That address can't be called from a data query.")
        if (addr.is_private or addr.is_loopback) and not settings.data_query_allow_private_hosts:
            raise DataQueryError(
                "That address is on the server's own network. Set DATA_QUERY_ALLOW_PRIVATE_HOSTS=true to allow internal APIs."
            )


def _http_call(method: str, url: str, headers: dict[str, str], body: str | None) -> tuple[int, bytes]:
    """Sync, on a worker thread (see app/core/tls.py for why). Redirects are
    not followed — each hop would need its own address check."""
    with httpx.Client(timeout=TIMEOUT, verify=TRUSTSTORE_SSL_CONTEXT, follow_redirects=False) as client:
        with client.stream(method, url, headers=headers, content=body.encode("utf-8") if body else None) as resp:
            chunks, size = [], 0
            for chunk in resp.iter_bytes():
                size += len(chunk)
                if size > MAX_BYTES:
                    raise DataQueryError("The response is larger than 5 MB.")
                chunks.append(chunk)
            return resp.status_code, b"".join(chunks)


def _flatten(obj: dict, prefix: str = "", depth: int = 0) -> dict:
    out: dict[str, Any] = {}
    for k, v in obj.items():
        key = f"{prefix}{k}"
        if isinstance(v, dict) and depth < 2:
            out.update(_flatten(v, f"{key}.", depth + 1))
        else:
            out[key] = v
    return out


def extract_rows(payload: Any, rows_path: str | None) -> list[dict]:
    """The rows in a JSON response: the list at rows_path ("data.items"), or
    the whole response. Nested objects become "a.b" columns; a list of plain
    values becomes a single "value" column."""
    node = payload
    for part in [p for p in (rows_path or "").split(".") if p]:
        if isinstance(node, dict) and part in node:
            node = node[part]
        elif isinstance(node, list) and part.isdigit() and int(part) < len(node):
            node = node[int(part)]
        else:
            raise DataQueryError(f"The response has nothing at {rows_path}.")
    if node is None:
        return []
    items = node if isinstance(node, list) else [node]
    return [_flatten(item) if isinstance(item, dict) else {"value": item} for item in items[:MAX_ROWS]]


async def _run_rest(definition: dict, params: dict[str, Any], secret_values: dict[str, str]) -> list[dict]:
    method = (definition.get("method") or "GET").upper()
    if method not in ("GET", "POST"):
        raise DataQueryError("Only GET and POST requests are supported.")
    url = secrets_crud.fill_templates(_fill_params(definition.get("url") or "", params, encode=True), secret_values)
    headers = {
        str(k): secrets_crud.fill_templates(_fill_params(str(v), params), secret_values)
        for k, v in (definition.get("headers") or {}).items()
        if k
    }
    body = definition.get("body")
    if body:
        body = secrets_crud.fill_templates(_fill_params(body, params), secret_values)
        headers.setdefault("Content-Type", "application/json")
    headers.setdefault("Accept", "application/json")
    await asyncio.to_thread(_check_host, url)
    try:
        status, content = await asyncio.to_thread(_http_call, method, url, headers, body if method == "POST" else None)
    except httpx.HTTPError as exc:
        raise DataQueryError(f"The request failed: {exc}") from exc
    if status >= 400:
        raise DataQueryError(f"The API answered {status}: {content[:300].decode('utf-8', 'replace')}")
    try:
        payload = json.loads(content or b"null")
    except ValueError as exc:
        raise DataQueryError("The response isn't JSON.") from exc
    return extract_rows(payload, definition.get("rows_path"))


# ---- SQL ----------------------------------------------------------------------------------------


async def _run_sql(db: AsyncSession, tenant_id: uuid.UUID, connector_id: Any, definition: dict, params: dict[str, Any]) -> list[dict]:
    from app.crud.sql_providers import SqlError, placeholders, run_sql

    connector = await get_connector(db, tenant_id, connector_id, "database")
    if connector is None:
        raise DataQueryError("Choose an enabled Database connector for this query.")
    sql = definition.get("sql") or ""
    if not sql.strip():
        raise DataQueryError("Write the SQL statement to run.")
    config = await secrets_crud.connector_config(db, tenant_id, connector)
    bound = {name: params.get(name) for name in placeholders(sql)}
    try:
        result = await run_sql(config, sql, bound, write=False, max_rows=MAX_ROWS, timeout=TIMEOUT)
    except SqlError as exc:
        raise DataQueryError(str(exc)) from exc
    return result.rows


# ---- Running ------------------------------------------------------------------------------------


def _columns(rows: list[dict]) -> list[str]:
    seen: dict[str, None] = {}
    for r in rows:
        for k in r:
            seen.setdefault(k, None)
    return list(seen)


async def execute(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    kind: str,
    definition: dict,
    connector_id: Any,
    parameter_defs: list[dict],
    given: dict | None,
) -> QueryResult:
    """Runs a (saved or unsaved) query once, uncached."""
    params = param_values(parameter_defs, given)
    names = secrets_crud.referenced_names(definition)
    secret_values = await secrets_crud.load_values(db, tenant_id, names) if names else {}
    scrub = secrets_crud.Scrubber(secret_values)
    try:
        if kind == "rest":
            rows = await _run_rest(definition, params, secret_values)
        elif kind == "sql":
            rows = await _run_sql(db, tenant_id, connector_id, definition, params)
        else:
            raise DataQueryError(f"Unknown query kind {kind}.")
    except DataQueryError as exc:
        raise DataQueryError(scrub(str(exc))) from exc
    from app.crud.sql_providers import jsonable

    rows = scrub(jsonable(rows))
    return QueryResult(_columns(rows), rows)


_cache: dict[tuple, tuple[float, QueryResult]] = {}


def _cache_key(query: DataQuery, params: dict) -> tuple:
    return (str(query.tenant_id), str(query.id), query.updated_at.isoformat() if query.updated_at else "", json.dumps(params, sort_keys=True, default=str))


def clear_cache() -> None:
    _cache.clear()


async def run(db: AsyncSession, query: DataQuery, given: dict | None) -> QueryResult:
    """A saved query's rows, from the cache when a fresh copy is there."""
    params = param_values(query.parameters or [], given)
    key = _cache_key(query, params)
    now = time.monotonic()
    hit = _cache.get(key)
    if hit and hit[0] > now:
        return QueryResult(hit[1].columns, hit[1].rows, cached=True)
    result = await execute(db, query.tenant_id, query.kind, query.definition or {}, query.connector_id, query.parameters or [], params)
    if query.cache_seconds > 0:
        if len(_cache) >= _CACHE_ENTRIES:
            for k in sorted(_cache, key=lambda k: _cache[k][0])[: _CACHE_ENTRIES // 5]:
                _cache.pop(k, None)
        _cache[key] = (now + query.cache_seconds, result)
    return result


# ---- Access -------------------------------------------------------------------------------------


async def load(db: AsyncSession, tenant_id: uuid.UUID, query_id: Any) -> DataQuery | None:
    try:
        qid = uuid.UUID(str(query_id))
    except (ValueError, TypeError):
        return None
    return (await db.execute(select(DataQuery).where(DataQuery.tenant_id == tenant_id, DataQuery.id == qid))).scalar_one_or_none()


async def can_run(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, query: DataQuery, vendor: bool | None = None) -> bool:
    """Every internal user, unless the query names roles; a Portal user only
    through one of the named roles."""
    allowed = {str(r) for r in (query.role_ids or [])}
    if not allowed:
        if vendor is None:
            from app.models.identity import User

            vendor = user_id is None or (await db.execute(select(User.vendor_id).where(User.id == user_id))).scalar_one_or_none() is not None
        return not vendor
    from app.crud.access_policy import user_roles

    roles, admin = await user_roles(db, tenant_id, user_id)
    return admin or bool(roles & allowed)


async def run_for_user(db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID | None, query_id: Any, given: dict | None) -> QueryResult:
    """404/403-style errors as LookupError/PermissionError; DataQueryError
    when the query itself fails."""
    query = await load(db, tenant_id, query_id)
    if query is None:
        raise LookupError("Data query not found.")
    if not await can_run(db, tenant_id, user_id, query):
        raise PermissionError("Your role can't run this data query.")
    return await run(db, query, given)


def split_param_filters(query: DataQuery, filters: list[dict]) -> tuple[dict, list[dict]]:
    """A dataset filter "equals" on a declared parameter's name feeds that
    parameter (so a page's existing filter rows, {{tokens}} included, can
    drive the API); every other filter applies to the returned rows."""
    names = {p["name"] for p in (query.parameters or [])}
    params: dict[str, Any] = {}
    rest: list[dict] = []
    for f in filters:
        if f.get("field") in names and f.get("operator") in ("eq", "equals", "=") and f["field"] not in params:
            params[f["field"]] = f.get("value")
        else:
            rest.append(f)
    return params, rest