"""Data-plane explorer: a web view over the Hive-partitioned Parquet lake."""

from __future__ import annotations

import json
import os
import re
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import duckdb

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "monitoring"))
from lake import flight_window, iter_parquet_files, parquet_reader  # noqa: E402

DATA_DIR = Path(os.environ.get("DATA_DIR", "./data"))
PORT = int(os.environ.get("EXPLORER_PORT", "8088"))
ROW_LIMIT = int(os.environ.get("ROW_LIMIT", "500"))
VIEW_REFRESH_S = float(os.environ.get("VIEW_REFRESH_S", "30"))
TREE_CACHE_S = float(os.environ.get("TREE_CACHE_S", "15"))
STATIC_DIR = Path(__file__).parent

_ALLOWED_START = re.compile(r"^\s*(SELECT|WITH|DESCRIBE|SUMMARIZE|SHOW)\b", re.IGNORECASE)
_FORBIDDEN = re.compile(
    r"\b(INSERT|UPDATE|DELETE|CREATE|DROP|ALTER|ATTACH|DETACH|COPY|EXPORT"
    r"|INSTALL|LOAD|SET|RESET|PRAGMA|CALL|IMPORT|VACUUM|CHECKPOINT)\b",
    re.IGNORECASE,
)

_lock = threading.Lock()
_con: duckdb.DuckDBPyConnection | None = None
_views_at = 0.0
_tree_cache: dict | None = None
_tree_at = 0.0
_last_tree_s = 0.0
_last_query_s = 0.0
_query_count = 0


def gate(sql: str) -> str | None:
    stripped = re.sub(r"--[^\n]*|/\*.*?\*/", " ", sql, flags=re.DOTALL).strip().rstrip(";")
    if not stripped:
        return "empty statement"
    if ";" in stripped:
        return "multiple statements are not allowed"
    if not _ALLOWED_START.match(stripped):
        return "only SELECT / WITH / DESCRIBE / SUMMARIZE / SHOW are allowed"
    if _FORBIDDEN.search(stripped):
        return "statement contains a forbidden keyword"
    return None


def _refresh_views() -> None:
    global _con, _views_at
    con = duckdb.connect()
    for ds in ("telemetry", "detections", "state"):
        reader = parquet_reader(DATA_DIR, ds)
        try:
            con.execute(f"CREATE OR REPLACE VIEW {ds} AS SELECT * FROM read_parquet({reader})")
        except duckdb.Error:
            pass
    with _lock:
        if _con is not None:
            _con.close()
        _con = con
        _views_at = time.monotonic()


def connect() -> duckdb.DuckDBPyConnection:
    if _con is None or time.monotonic() - _views_at > VIEW_REFRESH_S:
        _refresh_views()
    assert _con is not None
    return _con


def tree() -> dict:
    global _tree_cache, _tree_at, _last_tree_s
    now = time.monotonic()
    if _tree_cache is not None and now - _tree_at < TREE_CACHE_S:
        return _tree_cache
    started = now
    root: dict = {}
    total_bytes = 0
    total_files = 0
    for f in sorted(iter_parquet_files(DATA_DIR)):
        rel = f.relative_to(DATA_DIR)
        size = f.stat().st_size
        total_bytes += size
        total_files += 1
        node = root
        for part in rel.parts[:-1]:
            node = node.setdefault("children", {}).setdefault(part, {})
        leaf = node.setdefault("children", {}).setdefault(rel.parts[-1], {})
        leaf["bytes"] = size
        node = root
        node["bytes"] = node.get("bytes", 0) + size
        node["files"] = node.get("files", 0) + 1
        for part in rel.parts[:-1]:
            node = node["children"][part]
            node["bytes"] = node.get("bytes", 0) + size
            node["files"] = node.get("files", 0) + 1
    _last_tree_s = time.monotonic() - started
    _tree_cache = {"tree": root, "total_bytes": total_bytes, "total_files": total_files}
    _tree_at = now
    return _tree_cache


def run_query(sql: str) -> dict:
    global _last_query_s, _query_count
    reason = gate(sql)
    if reason:
        return {"error": f"rejected by read-only gate: {reason}"}
    started = time.monotonic()
    with _lock:
        con = connect()
        try:
            cur = con.sql(sql)
            columns = cur.columns
            rows = cur.fetchmany(ROW_LIMIT)
            _query_count += 1
            _last_query_s = time.monotonic() - started
            return {
                "columns": columns,
                "rows": [[repr(v) if isinstance(v, bytes) else v for v in row] for row in rows],
                "truncated": len(rows) == ROW_LIMIT,
            }
        except duckdb.Error as exc:
            return {"error": str(exc)}


def metrics_text() -> str:
    return "\n".join([
        "# HELP swarm_explorer_query_duration_seconds Wall time of the last SQL query",
        "# TYPE swarm_explorer_query_duration_seconds gauge",
        f"swarm_explorer_query_duration_seconds {_last_query_s:.4f}",
        "# HELP swarm_explorer_tree_duration_seconds Wall time of the last partition tree build",
        "# TYPE swarm_explorer_tree_duration_seconds gauge",
        f"swarm_explorer_tree_duration_seconds {_last_tree_s:.4f}",
        "# HELP swarm_explorer_queries_total Read-only queries served",
        "# TYPE swarm_explorer_queries_total counter",
        f"swarm_explorer_queries_total {_query_count}",
        "# HELP swarm_metric_flight_window Flight partitions in DuckDB views",
        "# TYPE swarm_metric_flight_window gauge",
        f"swarm_metric_flight_window {flight_window()}",
    ]) + "\n"


class Handler(BaseHTTPRequestHandler):
    def _send(self, code: int, body: bytes, ctype: str) -> None:
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def do_OPTIONS(self) -> None:  # noqa: N802
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()

    def _json(self, payload: dict, code: int = 200) -> None:
        self._send(code, json.dumps(payload, default=str).encode(), "application/json")

    def do_GET(self) -> None:  # noqa: N802
        if self.path in ("/", "/index.html"):
            self._send(200, (STATIC_DIR / "index.html").read_bytes(), "text/html; charset=utf-8")
        elif self.path == "/api/tree":
            self._json(tree())
        elif self.path == "/metrics":
            self._send(200, metrics_text().encode(), "text/plain; version=0.0.4")
        else:
            self._send(404, b"not found", "text/plain")

    def do_POST(self) -> None:  # noqa: N802
        if self.path != "/api/query":
            self._send(404, b"not found", "text/plain")
            return
        length = int(self.headers.get("Content-Length", "0"))
        try:
            sql = json.loads(self.rfile.read(length)).get("sql", "")
        except (json.JSONDecodeError, AttributeError):
            self._json({"error": "invalid request body"}, 400)
            return
        self._json(run_query(sql))

    def log_message(self, *_args: object) -> None:
        pass


def _view_loop() -> None:
    while True:
        try:
            _refresh_views()
        except Exception:
            pass
        time.sleep(VIEW_REFRESH_S)


if __name__ == "__main__":
    _refresh_views()
    threading.Thread(target=_view_loop, daemon=True).start()
    print(
        f"data-plane explorer on :{PORT}, views refresh every {VIEW_REFRESH_S}s, "
        f"last {flight_window()} flights, reading {DATA_DIR}"
    )
    ThreadingHTTPServer(("", PORT), Handler).serve_forever()