Blame for supabase_store.py

"""Accounts, chat history, and usage analytics in Supabase.

Supabase exposes every table through PostgREST, which is plain HTTPS + JSON, so
this module needs only the standard library (urllib). No public method raises:
each returns (result, error) and the app shows the error instead of crashing.
Writes that nothing waits for (chat rows, clicks, heartbeats) run on a
background thread so a slow database never delays the tutor.

Tables (created by supabase_schema.sql):
  anatomy_users     one row per account: user_id + email
  anatomy_messages  every chat message, in order, per user
  anatomy_sessions  one row per sign-in with first and last activity times
  anatomy_events    one row per button click
"""

from __future__ import annotations

import json
import logging
import os
import queue
import re
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone

log = logging.getLogger("anatomy.supabase")

USERS_TABLE = "anatomy_users"
MESSAGES_TABLE = "anatomy_messages"
SESSIONS_TABLE = "anatomy_sessions"
EVENTS_TABLE = "anatomy_events"

DEFAULT_TIMEOUT = 15.0
RETRIES = 2                      # attempts for network failures and 5xx replies
QUEUE_SIZE = 10_000              # background writes waiting for Supabase; newer ones are dropped (and logged) beyond this
EXISTS = "__user_id_exists__"    # error returned by create_user when the ID is taken

# Lower-case; letters, digits, dot, underscore, hyphen; 2 to 64 characters.
USER_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{1,63}$")
EMAIL_RE = re.compile(
    r"^[a-z0-9!#$%&'*+/=?^_`{|}~.-]+@"
    r"[a-z0-9](?:[a-z0-9-]*[a-z0-9])?"
    r"(?:\.[a-z0-9](?:[a-z0-9-]*[a-z0-9])?)*"
    r"\.[a-z]{2,}$"
)


def normalize_user_id(value: str | None) -> str:
    return (value or "").strip().lower()


def is_valid_user_id(value: str | None) -> bool:
    return USER_ID_RE.match(normalize_user_id(value)) is not None


def normalize_email(value: str | None) -> str:
    return (value or "").strip().lower()


def is_valid_email(value: str | None) -> bool:
    email = normalize_email(value)
    if not email or len(email) > 254 or ".." in email:
        return False
    local = email.split("@")[0]
    if local.startswith(".") or local.endswith("."):
        return False
    return EMAIL_RE.match(email) is not None


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def _env(*names: str) -> str:
    for name in names:
        value = os.environ.get(name, "")
        if value and value.strip():
            return value.strip()
    return ""


def _is_conflict(error: str | None) -> bool:
    text = error or ""
    return "HTTP 409" in text or "23505" in text


class SupabaseStore:
    """Thin PostgREST client scoped to this app's four tables."""

    def __init__(self, url: str | None = None, key: str | None = None, timeout: float | None = None) -> None:
        raw_url = (url if url is not None else _env("SUPABASE_URL", "SUPABASE_PROJECT_URL")).strip().rstrip("/")
        if raw_url and not raw_url.startswith(("http://", "https://")):
            raw_url = "https://" + raw_url
        if raw_url.endswith("/rest/v1"):          # tolerate a pasted REST endpoint
            raw_url = raw_url[: -len("/rest/v1")]
        self.url = raw_url
        self.key = key if key is not None else _env("SUPABASE_KEY", "SUPABASE_SERVICE_ROLE_KEY", "SUPABASE_SERVICE_KEY")
        try:
            self.timeout = float(timeout if timeout is not None else (_env("SUPABASE_TIMEOUT") or DEFAULT_TIMEOUT))
        except (TypeError, ValueError):
            self.timeout = DEFAULT_TIMEOUT
        self._queue: queue.Queue = queue.Queue(maxsize=QUEUE_SIZE)
        self._worker: threading.Thread | None = None
        self._worker_lock = threading.Lock()

    # ---- configuration -------------------------------------------------------
    @property
    def enabled(self) -> bool:
        return bool(self.url and self.key)

    def config_hint(self) -> str:
        """Why persistence is off (empty when it is on)."""
        missing = [name for name, value in (("SUPABASE_URL", self.url), ("SUPABASE_KEY", self.key)) if not value]
        return "Missing Space secret(s): " + ", ".join(missing) if missing else ""

    # ---- HTTP ----------------------------------------------------------------
    def _request(self, method: str, table: str, params: dict | None = None, body=None, prefer: str | None = None):
        """Return (ok, data, error). Never raises."""
        if not self.enabled:
            return False, None, self.config_hint()
        url = f"{self.url}/rest/v1/{table}"
        if params:
            url += "?" + urllib.parse.urlencode(params)
        headers = {"apikey": self.key, "Authorization": "Bearer " + self.key, "Accept": "application/json"}
        data = None
        if body is not None:
            data = json.dumps(body).encode("utf-8")
            headers["Content-Type"] = "application/json"
        if prefer:
            headers["Prefer"] = prefer

        error = None
        for attempt in range(RETRIES):
            request = urllib.request.Request(url, data=data, headers=headers, method=method)
            try:
                with urllib.request.urlopen(request, timeout=self.timeout) as response:
                    raw = response.read()
                if not raw:
                    return True, [], None
                try:
                    return True, json.loads(raw.decode("utf-8")), None
                except ValueError:
                    return True, [], None
            except urllib.error.HTTPError as err:
                detail = ""
                try:
                    detail = err.read().decode("utf-8", "ignore")[:400]
                except Exception:
                    pass
                error = f"Supabase {method} {table}: HTTP {err.code} {detail or err.reason}"
                if err.code < 500:
                    return False, None, error
            except Exception as err:                  # URLError, timeout, DNS, TLS
                error = f"Supabase {method} {table}: {type(err).__name__}: {err}"
            if method == "POST":                      # an insert is never repeated: the first attempt may have landed
                return False, None, error
            if attempt < RETRIES - 1:
                time.sleep(0.5 * (attempt + 1))
        return False, None, error

    def ping(self):
        """Connectivity and schema check. Returns (ok, error)."""
        ok, _, error = self._request("GET", USERS_TABLE, params={"select": "user_id", "limit": "1"})
        return ok, error

    # ---- accounts ------------------------------------------------------------
    def get_user(self, user_id: str):
        """Return (row_or_None, error)."""
        ok, data, error = self._request(
            "GET", USERS_TABLE,
            params={"select": "user_id,email,created_at,last_login_at", "user_id": "eq." + normalize_user_id(user_id), "limit": "1"},
        )
        if not ok:
            return None, error
        return (data[0] if isinstance(data, list) and data else None), None

    def create_user(self, user_id: str, email: str):
        """Insert a new account. Returns (row, error); error is EXISTS when the ID is taken."""
        now = now_iso()
        body = {"user_id": normalize_user_id(user_id), "email": normalize_email(email), "created_at": now, "last_login_at": now}
        ok, data, error = self._request("POST", USERS_TABLE, body=body, prefer="return=representation")
        if not ok:
            return None, (EXISTS if _is_conflict(error) else error)
        return (data[0] if isinstance(data, list) and data else body), None

    def record_login(self, user_id: str):
        ok, _, error = self._request(
            "PATCH", USERS_TABLE,
            params={"user_id": "eq." + normalize_user_id(user_id)},
            body={"last_login_at": now_iso()}, prefer="return=minimal",
        )
        return ok, error

    def delete_user(self, user_id: str):
        """Remove an account and everything stored for it (used by verify_supabase.py). Returns (ok, error)."""
        params = {"user_id": "eq." + normalize_user_id(user_id)}
        for table in (EVENTS_TABLE, SESSIONS_TABLE, MESSAGES_TABLE, USERS_TABLE):
            ok, _, error = self._request("DELETE", table, params=params, prefer="return=minimal")
            if not ok:
                return False, error
        return True, None

    # ---- chat history --------------------------------------------------------
    def load_messages(self, user_id: str, limit: int = 200):
        """The most recent `limit` messages of a user, oldest first. Returns (rows, error)."""
        ok, data, error = self._request(
            "GET", MESSAGES_TABLE,
            params={
                "select": "role,content,mode,topic,source,created_at",
                "user_id": "eq." + normalize_user_id(user_id),
                "order": "id.desc", "limit": str(int(limit)),
            },
        )
        if not ok:
            return [], error
        rows = data if isinstance(data, list) else []
        rows.reverse()
        return rows, None

    def append_messages(self, user_id: str, session_id: str, messages: list[dict]):
        """messages: [{'role', 'content', 'mode', 'topic', 'source'}]. Returns (ok, error)."""
        now = now_iso()
        rows = []
        for message in messages:
            if message.get("role") not in ("user", "assistant") or not isinstance(message.get("content"), str):
                continue
            # Every row carries the same keys: PostgREST rejects a batch whose objects differ.
            rows.append({
                "user_id": normalize_user_id(user_id), "session_id": session_id,
                "role": message["role"], "content": message["content"],
                "mode": message.get("mode"), "topic": message.get("topic"), "source": message.get("source"),
                "created_at": now,
            })
        if not rows:
            return True, None
        ok, _, error = self._request("POST", MESSAGES_TABLE, body=rows, prefer="return=minimal")
        return ok, error

    # ---- time on site --------------------------------------------------------
    def start_session(self, session_id: str, user_id: str):
        now = now_iso()
        ok, _, error = self._request(
            "POST", SESSIONS_TABLE,
            body={"session_id": session_id, "user_id": normalize_user_id(user_id), "started_at": now, "last_seen_at": now},
            prefer="return=minimal",
        )
        return ok, error

    def heartbeat(self, session_id: str):
        ok, _, error = self._request(
            "PATCH", SESSIONS_TABLE, params={"session_id": "eq." + session_id},
            body={"last_seen_at": now_iso()}, prefer="return=minimal",
        )
        return ok, error

    def end_session(self, session_id: str):
        now = now_iso()
        ok, _, error = self._request(
            "PATCH", SESSIONS_TABLE, params={"session_id": "eq." + session_id},
            body={"last_seen_at": now, "ended_at": now}, prefer="return=minimal",
        )
        return ok, error

    # ---- clicks --------------------------------------------------------------
    def log_event(self, user_id: str, session_id: str, event: str, detail: dict | None = None):
        ok, _, error = self._request(
            "POST", EVENTS_TABLE,
            body={"user_id": normalize_user_id(user_id), "session_id": session_id, "event": event,
                  "detail": detail or {}, "created_at": now_iso()},
            prefer="return=minimal",
        )
        return ok, error

    # ---- background writes ---------------------------------------------------
    def enqueue(self, method: str, *args) -> None:
        """Run a store method on the background thread, in order; failures are logged."""
        if not self.enabled:
            return
        try:
            self._queue.put_nowait((method, args))
        except queue.Full:
            log.warning("Supabase write queue is full; dropping %s", method)
            return
        with self._worker_lock:
            if self._worker is None or not self._worker.is_alive():
                self._worker = threading.Thread(target=self._run, name="supabase-writer", daemon=True)
                self._worker.start()

    def _run(self) -> None:
        while True:
            method, args = self._queue.get()
            try:
                ok, error = getattr(self, method)(*args)
                if not ok:
                    log.warning("%s failed: %s", method, error)
            except Exception:
                log.exception("%s failed", method)
            finally:
                self._queue.task_done()

    def flush(self, timeout: float = 3.0) -> bool:
        """Wait up to `timeout` seconds for queued writes to finish. Returns True when all of them have."""
        deadline = time.monotonic() + timeout
        with self._queue.all_tasks_done:          # Queue.join() with a timeout
            while self._queue.unfinished_tasks:
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    return False
                self._queue.all_tasks_done.wait(remaining)
        return True