Spaces:
Sleeping
Sleeping
| """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 ------------------------------------------------------- | |
| 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 | |