DavidL72Code/UMB_Sustainable_Chatbot
0
1"""Supabase-backed storage for the staff dashboard.2 3Talks to PostgREST and GoTrue over HTTPS with the standard library, so the4deployment does not need a Postgres driver in the image.5 6Two rules shape this module:7 * Nothing here may break a chat response. Every network call is wrapped, and8 writes are dispatched on a background thread so a slow or unreachable9 Supabase never adds latency to a user's answer.10 * Ordinary chats leave no transcript. Only numbers go to chat_metrics; the11 question and answer are written only when an answer is flagged for review.12"""13 14from __future__ import annotations15 16import json17import os18import threading19import urllib.error20import urllib.parse21import urllib.request22from typing import Any, Optional23 24_TIMEOUT_SECONDS = float(os.getenv("SUPABASE_TIMEOUT_SECONDS", "8"))25 26 27def _auth_error_message(detail: str) -> str:28 """The human-readable reason from a Supabase auth error body, if present.29 30 Supabase returns e.g. {"msg": "User already registered"} or31 {"error_description": "..."}. Without this the caller can only say32 "could not create that account", which hides whether the address is33 already taken, signups are disabled, or the password failed policy.34 """35 try:36 parsed = json.loads(detail)37 except Exception:38 return ""39 if not isinstance(parsed, dict):40 return ""41 for field in ("msg", "message", "error_description", "error"):42 value = parsed.get(field)43 if isinstance(value, str) and value.strip():44 return value.strip()[:200]45 return ""46 47 48class SupabaseStore:49 def __init__(50 self,51 url: str = "",52 service_key: str = "",53 anon_key: str = "",54 timeout: float = _TIMEOUT_SECONDS,55 ) -> None:56 self.url = (url or os.getenv("SUPABASE_URL", "")).strip().rstrip("/")57 self.service_key = (service_key or os.getenv("SUPABASE_SERVICE_ROLE_KEY", "")).strip()58 self.anon_key = (anon_key or os.getenv("SUPABASE_ANON_KEY", "")).strip()59 self.timeout = timeout60 self._warned = False61 62 @property63 def enabled(self) -> bool:64 """True when writes and reads should go to Supabase."""65 return bool(self.url and self.service_key)66 67 @property68 def auth_enabled(self) -> bool:69 """True when employees sign in through Supabase Auth."""70 return bool(self.url and self.anon_key)71 72 # -- plumbing ----------------------------------------------------------73 def _request(74 self,75 method: str,76 path: str,77 *,78 key: str,79 body: Optional[dict | list] = None,80 headers: Optional[dict[str, str]] = None,81 ) -> Optional[Any]:82 request_headers = {83 "apikey": key,84 "Authorization": f"Bearer {key}",85 "Content-Type": "application/json",86 }87 request_headers.update(headers or {})88 self.last_error = ""89 data = json.dumps(body).encode("utf-8") if body is not None else None90 request = urllib.request.Request(91 f"{self.url}{path}", data=data, headers=request_headers, method=method92 )93 try:94 with urllib.request.urlopen(request, timeout=self.timeout) as response:95 raw = response.read().decode("utf-8", errors="replace")96 return json.loads(raw) if raw.strip() else True97 except urllib.error.HTTPError as exc:98 detail = exc.read().decode("utf-8", errors="replace")[:300]99 self.last_error = _auth_error_message(detail)100 self._warn(f"Supabase {method} {path} failed ({exc.code}): {detail}")101 except Exception as exc:102 self.last_error = ""103 self._warn(f"Supabase {method} {path} failed: {type(exc).__name__}: {exc}")104 return None105 106 def _warn(self, message: str) -> None:107 # De-duplicate by message rather than silencing everything after the108 # first failure. The old rule meant one early hiccup hid every later109 # error for the life of the process, so a failing signup left no trace110 # in the logs at all.111 seen = getattr(self, "_warned_messages", None)112 if seen is None:113 seen = self._warned_messages = set()114 if message not in seen:115 seen.add(message)116 print(message, flush=True)117 118 def _insert_async(self, table: str, row: dict) -> None:119 if not self.enabled:120 return121 122 def worker() -> None:123 self._request(124 "POST",125 f"/rest/v1/{table}",126 key=self.service_key,127 body=[row],128 headers={"Prefer": "return=minimal,resolution=merge-duplicates"},129 )130 131 threading.Thread(target=worker, name=f"supabase-{table}", daemon=True).start()132 133 # -- writes ------------------------------------------------------------134 def record_chat_metrics(self, row: dict) -> None:135 """One content-free row per answer."""136 self._insert_async("chat_metrics", row)137 138 def record_flagged_chat(self, row: dict) -> None:139 """Full transcript, written only for answers flagged for review."""140 self._insert_async("flagged_chats", row)141 142 def record_audit_event(self, row: dict) -> None:143 self._insert_async("admin_audit_events", row)144 145 # -- reads -------------------------------------------------------------146 def ping(self) -> bool:147 """Run the cheapest possible real query, for keepalive health checks.148 149 Supabase pauses free-tier projects after 7 days without activity, so150 something has to touch the database on a schedule. This asks for a151 single id and discards it: enough to be a genuine Postgres round-trip,152 small enough to sit behind an uptime monitor hitting it every 5153 minutes. Returns False when the project is unreachable so the caller154 can fail the health check rather than report a false green.155 """156 if not self.enabled:157 return False158 query = urllib.parse.urlencode({"select": "id", "limit": 1})159 result = self._request("GET", f"/rest/v1/chat_metrics?{query}", key=self.service_key)160 return isinstance(result, list)161 162 def fetch_flagged_chats(self, limit: int = 50) -> list[dict]:163 if not self.enabled:164 return []165 # flagged_chats.id references chat_metrics.id, so PostgREST can embed166 # the numbers alongside the transcript in one request.167 query = urllib.parse.urlencode(168 {169 "select": "*,chat_metrics(*)",170 "order": "created_at.desc",171 "limit": max(1, min(limit, 500)),172 }173 )174 result = self._request("GET", f"/rest/v1/flagged_chats?{query}", key=self.service_key)175 return result if isinstance(result, list) else []176 177 def fetch_chat_metrics(self, limit: int = 200) -> list[dict]:178 if not self.enabled:179 return []180 query = urllib.parse.urlencode(181 {"select": "*", "order": "created_at.desc", "limit": max(1, min(limit, 2000))}182 )183 result = self._request("GET", f"/rest/v1/chat_metrics?{query}", key=self.service_key)184 return result if isinstance(result, list) else []185 186 def fetch_daily_metrics(self, days: int = 30) -> list[dict]:187 if not self.enabled:188 return []189 query = urllib.parse.urlencode(190 {"select": "*", "order": "day.desc", "limit": max(1, min(days, 365))}191 )192 result = self._request("GET", f"/rest/v1/daily_metrics?{query}", key=self.service_key)193 return result if isinstance(result, list) else []194 195 def fetch_audit_events(self, limit: int = 50) -> list[dict]:196 if not self.enabled:197 return []198 query = urllib.parse.urlencode(199 {"select": "*", "order": "created_at.desc", "limit": max(1, min(limit, 500))}200 )201 result = self._request("GET", f"/rest/v1/admin_audit_events?{query}", key=self.service_key)202 return result if isinstance(result, list) else []203 204 def mark_reviewed(self, event_id: str, username: str, note: str = "") -> bool:205 if not self.enabled or not event_id:206 return False207 query = urllib.parse.urlencode({"id": f"eq.{event_id}"})208 body = {"reviewed_by": username, "reviewed_at": "now()", "review_note": note or None}209 result = self._request(210 "PATCH",211 f"/rest/v1/flagged_chats?{query}",212 key=self.service_key,213 body=body,214 headers={"Prefer": "return=minimal"},215 )216 return result is not None217 218 # -- visitor accounts and history --------------------------------------219 #220 # These use the visitor's own access token, never the service role key, so221 # the row-level policies decide what each visitor can see. Staff tables are222 # never touched here, and these tables are never read with the service role.223 def visitor_sign_up(self, email: str, password: str) -> Optional[dict]:224 if not self.auth_enabled or not email or not password:225 return None226 result = self._request(227 "POST", "/auth/v1/signup", key=self.anon_key,228 body={"email": email, "password": password},229 )230 return result if isinstance(result, dict) else None231 232 def visitor_sign_in(self, email: str, password: str) -> Optional[dict]:233 """Return the visitor's session, including the access token."""234 if not self.auth_enabled or not email or not password:235 return None236 result = self._request(237 "POST", "/auth/v1/token?grant_type=password", key=self.anon_key,238 body={"email": email, "password": password},239 )240 if not isinstance(result, dict) or not result.get("access_token"):241 return None242 return result243 244 def visitor_recover(self, email: str, redirect_to: str = "") -> bool:245 """Ask Supabase to email a password reset link.246 247 Returns True whenever the request was accepted. The caller must not248 expose whether the address actually has an account.249 """250 if not self.auth_enabled or not email:251 return False252 path = "/auth/v1/recover"253 if redirect_to:254 path += "?" + urllib.parse.urlencode({"redirect_to": redirect_to})255 return self._request("POST", path, key=self.anon_key, body={"email": email}) is not None256 257 def visitor_update_password(self, access_token: str, password: str) -> Optional[dict]:258 """Set a new password using the recovery token from the emailed link.259 260 Returns the updated user so the caller can open a full session; the261 recovery token is already a valid one.262 """263 if not self.auth_enabled or not access_token or not password:264 return None265 result = self._as_visitor(266 "PUT", "/auth/v1/user", access_token, body={"password": password}267 )268 return result if isinstance(result, dict) else None269 270 def visitor_refresh(self, refresh_token: str) -> Optional[dict]:271 """Exchange a refresh token so a visitor is not signed out every hour."""272 if not self.auth_enabled or not refresh_token:273 return None274 result = self._request(275 "POST", "/auth/v1/token?grant_type=refresh_token", key=self.anon_key,276 body={"refresh_token": refresh_token},277 )278 if not isinstance(result, dict) or not result.get("access_token"):279 return None280 return result281 282 def _as_visitor(283 self,284 method: str,285 path: str,286 access_token: str,287 body: Optional[dict | list] = None,288 extra_headers: Optional[dict[str, str]] = None,289 ) -> Optional[Any]:290 if not self.auth_enabled or not access_token:291 return None292 headers = {"Authorization": f"Bearer {access_token}"}293 headers.update(extra_headers or {})294 # The apikey stays the public anon key; the bearer token is the295 # visitor's, so PostgREST evaluates the policies as that visitor.296 return self._request(method, path, key=self.anon_key, body=body, headers=headers)297 298 def list_visitor_conversations(self, access_token: str, limit: int = 30) -> list[dict]:299 query = urllib.parse.urlencode(300 {"select": "*", "order": "updated_at.desc", "limit": max(1, min(limit, 200))}301 )302 result = self._as_visitor("GET", f"/rest/v1/visitor_conversations?{query}", access_token)303 return result if isinstance(result, list) else []304 305 def create_visitor_conversation(self, access_token: str, user_id: str, title: str) -> Optional[str]:306 result = self._as_visitor(307 "POST", "/rest/v1/visitor_conversations", access_token,308 body=[{"user_id": user_id, "title": (title or "New chat")[:120]}],309 extra_headers={"Prefer": "return=representation"},310 )311 if isinstance(result, list) and result:312 return str(result[0].get("id", "")) or None313 return None314 315 def append_visitor_messages(self, access_token: str, rows: list[dict]) -> bool:316 if not rows:317 return False318 result = self._as_visitor(319 "POST", "/rest/v1/visitor_messages", access_token,320 body=rows, extra_headers={"Prefer": "return=minimal"},321 )322 return result is not None323 324 # A session keeps its most recent turns rather than growing without limit.325 # Opening a long chat should not pull thousands of rows into the browser,326 # and the useful part of a conversation is its recent end. Counts user and327 # assistant rows together, so 200 is roughly 100 exchanges.328 VISITOR_MESSAGE_CAP = 200329 330 def fetch_visitor_messages(331 self, access_token: str, conversation_id: str, limit: int = 0332 ) -> list[dict]:333 """The most recent messages of a session, oldest first.334 335 Fetched newest-first with a limit and then reversed, so the cap keeps336 the end of the conversation. Ordering ascending with a limit would keep337 the beginning and silently hide everything the visitor last said.338 """339 cap = max(1, limit or self.VISITOR_MESSAGE_CAP)340 query = urllib.parse.urlencode({341 "select": "*",342 "conversation_id": f"eq.{conversation_id}",343 "order": "created_at.desc",344 "limit": cap,345 })346 result = self._as_visitor("GET", f"/rest/v1/visitor_messages?{query}", access_token)347 if not isinstance(result, list):348 return []349 return list(reversed(result))350 351 def visitor_session_is_full(self, access_token: str, conversation_id: str) -> bool:352 """True when a session has reached VISITOR_MESSAGE_CAP saved messages.353 354 Asks for at most cap ids rather than a count so one round trip answers355 it. A full session stops accepting new saves; the visitor is asked to356 start a new one instead of having older turns silently dropped.357 """358 query = urllib.parse.urlencode({359 "select": "id",360 "conversation_id": f"eq.{conversation_id}",361 "limit": self.VISITOR_MESSAGE_CAP,362 })363 result = self._as_visitor("GET", f"/rest/v1/visitor_messages?{query}", access_token)364 if not isinstance(result, list):365 return False366 return len(result) >= self.VISITOR_MESSAGE_CAP367 368 def touch_visitor_conversation(self, access_token: str, conversation_id: str) -> None:369 query = urllib.parse.urlencode({"id": f"eq.{conversation_id}"})370 self._as_visitor(371 "PATCH", f"/rest/v1/visitor_conversations?{query}", access_token,372 body={"updated_at": "now()"}, extra_headers={"Prefer": "return=minimal"},373 )374 375 def delete_visitor_conversation(self, access_token: str, conversation_id: str) -> bool:376 """A visitor can always delete their own history."""377 query = urllib.parse.urlencode({"id": f"eq.{conversation_id}"})378 result = self._as_visitor(379 "DELETE", f"/rest/v1/visitor_conversations?{query}", access_token,380 extra_headers={"Prefer": "return=minimal"},381 )382 return result is not None383 384 # -- auth --------------------------------------------------------------385 def sign_in(self, email: str, password: str) -> Optional[dict]:386 """Verify one employee against Supabase Auth.387 388 Returns the user record on success and None otherwise. Staff are389 managed from the Supabase dashboard, so adding or removing an employee390 needs no restart and no redeploy.391 """392 if not self.auth_enabled or not email or not password:393 return None394 result = self._request(395 "POST",396 "/auth/v1/token?grant_type=password",397 key=self.anon_key,398 body={"email": email, "password": password},399 )400 if not isinstance(result, dict):401 return None402 user = result.get("user")403 if not isinstance(user, dict):404 return None405 if not self.is_staff(user):406 print(407 f"Rejected dashboard sign-in for {user.get('email', 'unknown')}: "408 "the account is not marked as staff.",409 flush=True,410 )411 return None412 return user413 414 @staticmethod415 def is_staff(user: dict) -> bool:416 """Only accounts explicitly marked as staff may reach the dashboard.417 418 Supabase projects allow public email signup by default, and visitor419 accounts may later share this project, so a valid login is not by420 itself proof of employment. app_metadata is writable only with the421 service role key, so a user cannot grant themselves this role.422 """423 metadata = user.get("app_metadata")424 if not isinstance(metadata, dict):425 return False426 if str(metadata.get("role", "")).strip().lower() == "staff":427 return True428 roles = metadata.get("roles")429 if isinstance(roles, list):430 return any(str(role).strip().lower() == "staff" for role in roles)431 return False432