"""Ultron master DB logic for the jarvis-tools service (Railway).
Full Supabase SQL (read + write). Creds from env SUPABASE_URL +
SUPABASE_DB_PASSWORD. Seatbelt on drop/truncate/where-less delete. Isolated from
the memory service (psycopg only — no ML deps).
"""
from __future__ import annotations

import os
import re

_DANGER = re.compile(r"\b(drop|truncate)\b", re.I)


def _creds():
    """(host, port, user, password). Prefers the IPv4 pooler when
    SUPABASE_POOLER_HOST is set (required from Railway — the direct
    db.<ref>.supabase.co host is IPv6-only and unreachable there). Falls back to
    the direct connection otherwise (fine from a Mac/VPS)."""
    url = os.environ.get("SUPABASE_URL", "")
    pw = os.environ.get("SUPABASE_DB_PASSWORD", "")
    if not (url and pw):
        return None
    m = re.search(r"//([^.]+)\.", url)
    if not m:
        return None
    ref = m.group(1)
    pooler = os.environ.get("SUPABASE_POOLER_HOST", "")
    if pooler:
        return (pooler, int(os.environ.get("SUPABASE_POOLER_PORT", "5432")),
                f"postgres.{ref}", pw)
    return (f"db.{ref}.supabase.co", 5432, "postgres", pw)


def configured() -> bool:
    return _creds() is not None


def _connect():
    import psycopg
    host, port, user, pw = _creds()
    return psycopg.connect(
        f"host={host} port={port} user={user} password={pw} "
        f"dbname=postgres sslmode=require", connect_timeout=15, autocommit=True)


def _needs_confirm(sql: str) -> bool:
    s = sql.strip()
    if _DANGER.search(s):
        return True
    if re.match(r"(?is)^\s*(delete|update)\b", s) and \
            not re.search(r"\bwhere\b", s, re.I):
        return True
    return False


def query(sql: str, confirm: str = "") -> str:
    sql = str(sql).strip()
    if not sql:
        return "ultron_query failed: no SQL."
    if _needs_confirm(sql) and str(confirm).lower() != "yes":
        return ("That statement is irreversible (drop/truncate or a WHERE-less "
                "delete/update). Re-send with confirm='yes' to proceed.")
    with _connect() as conn:
        cur = conn.execute(sql)
        if cur.description:
            cols = [d.name for d in cur.description]
            rows = cur.fetchmany(500)
            if not rows:
                return "(0 rows)"
            head = " | ".join(cols)
            body = "\n".join(
                " | ".join("" if v is None else str(v) for v in r) for r in rows)
            more = "\n…(first 500 rows)" if len(rows) == 500 else ""
            out = f"{head}\n{'-' * len(head)}\n{body}{more}"
            return out[:5000] + (" …[truncated]" if len(out) > 5000 else "")
        return f"OK — {cur.rowcount} row(s) affected."


def schema(table: str = "") -> str:
    table = str(table).strip()
    with _connect() as conn:
        if not table:
            tabs = [r[0] for r in conn.execute(
                "select table_name from information_schema.tables "
                "where table_schema='public' order by 1").fetchall()]
            lines = []
            for t in tabs:
                try:
                    n = conn.execute(f'select count(*) from "{t}"').fetchone()[0]
                except Exception:  # noqa: BLE001
                    n = "?"
                lines.append(f"- {t} ({n} rows)")
            return "Tables:\n" + "\n".join(lines)
        cols = conn.execute(
            "select column_name, data_type from information_schema.columns "
            "where table_schema='public' and table_name=%s "
            "order by ordinal_position", (table,)).fetchall()
        if not cols:
            return f"No table named {table!r}."
        return f"{table} columns:\n" + "\n".join(f"- {c} ({t})" for c, t in cols)