""" supabase_store.py — email-keyed persistence for MyeAI (Myeloma AI). Stores each participant's 16 intake answers and their full follow-up chat in Supabase, keyed by email address, so that a returning participant is restored exactly where they left off. DESIGN NOTES ------------ * No new Python dependencies. Supabase's REST API is PostgREST, which is plain HTTPS + JSON, so this module talks to it with the standard library only (``urllib``). Adding the ``supabase`` pip package would pull in httpx/pydantic /gotrue/realtime and risk exactly the kind of version conflict that requirements.txt already warns about for this Space. * Nothing here ever raises into the Gradio callbacks. Every public method returns ``(result, error_message)`` and the app degrades gracefully: if Supabase is unreachable or unconfigured, the chatbot still works, it just does not persist. * Two tables (see supabase_schema.sql): - ``p3_sessions`` one row per email; the resumable state (answers, convo, chat transcript, follow-up counter). - ``p3_messages`` append-only log of every chat turn, kept for analysis and as a safety net (a session row is overwritten when a participant restarts, the message log is never lost). """ from __future__ import annotations import json import os import re import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone DEFAULT_SESSIONS_TABLE = "p3_sessions" DEFAULT_MESSAGES_TABLE = "p3_messages" # Returned as the error from save_session when another tab or device has # already moved the stored conversation past the revision we hold. CONFLICT = "__revision_conflict__" DEFAULT_TIMEOUT = 20.0 _RETRIES = 2 # total attempts for transient (network / 5xx) failures # Deliberately permissive but structural: local@domain.tld _EMAIL_RE = re.compile( r"^[A-Za-z0-9!#$%&'*+/=?^_`{|}~.-]+@" r"[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?" r"(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?)*" r"\.[A-Za-z]{2,}$" ) # ---------------------------------------------------------------------- # Email helpers # ---------------------------------------------------------------------- def normalize_email(value): """Lower-case, trimmed form used as the primary key. Also tolerates the common paste artefacts ```` and ``mailto:``. """ email = (value or "").strip() if email.lower().startswith("mailto:"): email = email[7:] email = email.strip().strip("<>").strip() return email.lower() def is_valid_email(value): email = normalize_email(value) if not email or len(email) > 254 or ".." in email: return False local = email.split("@")[0] if not local or local.startswith(".") or local.endswith("."): return False return _EMAIL_RE.match(email) is not None # ---------------------------------------------------------------------- # User ID helpers # ---------------------------------------------------------------------- # Study IDs vary a lot between sites, so this is permissive: it only insists on # a leading letter or digit and a sane length, and rejects control characters. _USER_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._@#/+-]{0,63}$") def normalize_user_id(value): """Trimmed form, with runs of internal whitespace collapsed to one space.""" return re.sub(r"\s+", " ", (value or "").strip()) def is_valid_user_id(value): return _USER_ID_RE.match(normalize_user_id(value)) is not None def user_ids_match(a, b): """Compare two User IDs. Case- and spacing-insensitive, so a participant is not locked out by typing 'p3-014' instead of 'P3-014'.""" return normalize_user_id(a).casefold() == normalize_user_id(b).casefold() def _now_iso(): return datetime.now(timezone.utc).isoformat() def _answered(answers): """How many of the intake questions actually have an answer.""" if not isinstance(answers, (list, tuple)): return 0 return sum(1 for a in answers if a is not None) def _missing_column(err, column): """True when PostgREST rejected the request because `column` is absent. Lets a project created before the hardening columns were added keep working instead of failing every save. """ text = (err or "").lower() if column.lower() not in text: return False return ("does not exist" in text or "could not find" in text or "schema cache" in text) def _env(*names): for n in names: v = os.environ.get(n, "") if v and v.strip(): return v.strip() return "" # ---------------------------------------------------------------------- # Store # ---------------------------------------------------------------------- class SupabaseStore: """Thin PostgREST client scoped to this app's two tables.""" def __init__(self, url=None, key=None, sessions_table=None, messages_table=None, timeout=None): raw_url = url if url is not None else _env( "SUPABASE_URL", "SUPABASE_PROJECT_URL") raw_url = (raw_url or "").strip().rstrip("/") if raw_url and not raw_url.startswith(("http://", "https://")): raw_url = "https://" + raw_url # Tolerate someone pasting the full REST endpoint. if raw_url.endswith("/rest/v1"): 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", "SUPABASE_ANON_KEY", ) self.sessions_table = ( sessions_table or _env("SUPABASE_SESSIONS_TABLE") or DEFAULT_SESSIONS_TABLE) self.messages_table = ( messages_table or _env("SUPABASE_MESSAGES_TABLE") or DEFAULT_MESSAGES_TABLE) try: self.timeout = float(timeout if timeout is not None else (_env("SUPABASE_TIMEOUT") or DEFAULT_TIMEOUT)) except (TypeError, ValueError): self.timeout = DEFAULT_TIMEOUT # ---- configuration ---- @property def enabled(self): return bool(self.url and self.key) def config_hint(self): """Human-readable reason persistence is off (empty string when on).""" if self.url and self.key: return "" missing = [] if not self.url: missing.append("SUPABASE_URL") if not self.key: missing.append("SUPABASE_KEY") return ("Saving is off — missing Space secret(s): " + ", ".join(missing)) # ---- low-level HTTP ---- def _request(self, method, table, params=None, body=None, prefer=None): """Return (ok, data, error). Never raises.""" if not self.enabled: return False, None, self.config_hint() url = "{}/rest/v1/{}".format(self.url, table) if params: url = url + "?" + urllib.parse.urlencode(params) data = None headers = { "apikey": self.key, "Authorization": "Bearer " + self.key, "Accept": "application/json", } if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" if prefer: headers["Prefer"] = prefer last_err = None for attempt in range(_RETRIES): req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: raw = resp.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 e: detail = "" try: detail = e.read().decode("utf-8", "ignore")[:400] except Exception: pass last_err = "Supabase {} {} -> HTTP {} {}".format( method, table, e.code, detail or e.reason) if 500 <= e.code < 600 and attempt < _RETRIES - 1: time.sleep(0.5 * (attempt + 1)) continue return False, None, last_err except Exception as e: # URLError, timeout, DNS, TLS last_err = "Supabase {} {} -> {}: {}".format( method, table, type(e).__name__, e) if attempt < _RETRIES - 1: time.sleep(0.5 * (attempt + 1)) continue return False, None, last_err return False, None, last_err # ---- health ---- def ping(self): """Cheap connectivity + schema check. Returns (ok, error).""" ok, _, err = self._request( "GET", self.sessions_table, params={"select": "email", "limit": "1"}) return ok, err # ---- sessions ---- def load_session(self, email): """Return (row_or_None, error). row_or_None is None when unseen.""" email = normalize_email(email) if not email: return None, "No email supplied." ok, data, err = self._request( "GET", self.sessions_table, params={"select": "*", "email": "eq." + email, "limit": "1"}, ) if not ok: return None, err if isinstance(data, list) and data: return data[0], None return None, None def ensure_session(self, email, user_id=None): """Create the row if this email is new. Returns (row_or_None, error).""" email = normalize_email(email) if not email: return None, "No email supplied." row, err = self.load_session(email) if err: return None, err if row is not None: return row, None now = _now_iso() body = {"email": email, "created_at": now, "updated_at": now} if user_id is not None: body["user_id"] = user_id prefer = "return=representation,resolution=merge-duplicates" ok, data, err = self._request( # merge-duplicates makes a concurrent double-submit harmless. "POST", self.sessions_table, body=body, prefer=prefer) if not ok and "user_id" in body and _missing_column(err, "user_id"): # A project that has not re-run supabase_schema.sql yet. Create the # session without the User ID rather than refusing to save at all. body.pop("user_id", None) ok, data, err = self._request( "POST", self.sessions_table, body=body, prefer=prefer) if not ok: return None, err if isinstance(data, list) and data: return data[0], None return None, None def save_session(self, email, answers=None, convo=None, chat_display=None, followups=None, user_id=None, expected_revision=None): """Update the supplied fields for this email, inserting if absent. Only fields that are not None are written, so a mid-chat save does not clobber the stored intake answers. If ``expected_revision`` is given, the write only lands when the stored revision still matches -- otherwise another tab or device has moved the conversation on and we must not overwrite it. Pass None to write unconditionally (used for answers-only saves, which cannot lose a conversation). Returns (ok, revision, error). ``error`` is CONFLICT when the write was refused because the stored revision had moved on. """ email = normalize_email(email) if not email: return False, expected_revision, "No email supplied." payload = {"updated_at": _now_iso()} if answers is not None: payload["answers"] = answers payload["answers_count"] = _answered(answers) if convo is not None: payload["convo"] = convo if chat_display is not None: payload["chat_display"] = chat_display if followups is not None: payload["followups"] = int(followups) if user_id is not None: payload["user_id"] = user_id params = {"email": "eq." + email} guarded = expected_revision is not None if guarded: params["revision"] = "eq." + str(int(expected_revision)) payload["revision"] = int(expected_revision) + 1 ok, data, err = self._request( "PATCH", self.sessions_table, params=params, body=payload, prefer="return=representation", ) if not ok: # A project created before the hardening columns existed: drop them # and retry rather than refusing to save at all. for optional in ("answers_count", "user_id"): if not ok and _missing_column(err, optional): payload.pop(optional, None) ok, data, err = self._request( "PATCH", self.sessions_table, params=params, body=payload, prefer="return=representation") if not ok and guarded and _missing_column(err, "revision"): retry_ok, _unused, retry_err = self.save_session( email, answers=answers, convo=convo, chat_display=chat_display, followups=followups, user_id=user_id, expected_revision=None) # The column does not exist, so there is no revision to report. # Hand back the caller's own value rather than None, which would # otherwise end up in the page's revision state. return retry_ok, expected_revision, retry_err if not ok: return False, expected_revision, err if isinstance(data, list) and data: return True, self._revision_of(data[0], expected_revision), None # Nothing matched: either the row is genuinely absent, or the revision # guard rejected the write. These need opposite responses. if guarded: existing, load_err = self.load_session(email) if load_err: return False, expected_revision, load_err if existing is not None: if self._already_applied(existing, payload, expected_revision): # This exact write is what is stored: it committed and only # its acknowledgement was lost (a compare-and-set PATCH can # never match its own guard a second time, so a retry lands # here). Report the success it actually was, otherwise the # session would treat its own write as a rival's and refuse # every later save. return True, self._revision_of(existing, expected_revision), None # A genuine second writer. Hand back the revision the caller # already holds, NOT the winner's: adopting the winner's would # let this stale session's very next write match the guard and # overwrite the newer conversation. return False, expected_revision, CONFLICT payload["email"] = email payload.setdefault("created_at", _now_iso()) ok, data, err = self._request( "POST", self.sessions_table, body=payload, prefer="return=representation,resolution=merge-duplicates", ) if not ok: return False, expected_revision, err new_rev = expected_revision if isinstance(data, list) and data: new_rev = self._revision_of(data[0], expected_revision) return True, new_rev, None def save_answers_monotonic(self, email, answers): """Write intake answers, but never replace a more complete set. The questionnaire auto-saves on every click, and Gradio can deliver those events out of order — a snapshot taken at question 10 must not be allowed to land after the finished 16-answer set and wipe six answers. Two guards, both needed: * ``answers_count`` — never replace a more complete set of answers. * ``updated_at`` — never land on top of a write made after this one was built. Without it, Start Over (which resets answers_count to 0, and so admits anything) could be silently undone by an auto-save still in flight, restoring answers the participant just erased. Never inserts: the row already exists, created at sign-in. Returns (ok, error). """ email = normalize_email(email) if not email: return False, "No email supplied." count = _answered(answers) now = _now_iso() ok, _, err = self._request( "PATCH", self.sessions_table, params={"email": "eq." + email, "answers_count": "lte." + str(count), "updated_at": "lte." + now}, body={"answers": answers, "answers_count": count, "updated_at": now}, prefer="return=minimal", ) if not ok and _missing_column(err, "answers_count"): # Older schema without the completeness column: keep the recency # guard, which needs no new column. ok, _, err = self._request( "PATCH", self.sessions_table, params={"email": "eq." + email, "updated_at": "lte." + now}, body={"answers": answers, "updated_at": now}, prefer="return=minimal", ) return (True, None) if ok else (False, err) @staticmethod def _already_applied(existing, payload, expected_revision): """True when the stored row IS the write we just attempted. Deliberately exact rather than a heuristic. The stored revision must be precisely the one this write would have produced, and every field it wrote must already hold the value it sent. Anything looser — "the stored conversation is a prefix of mine", say — cannot tell a lost acknowledgement from a second tab, and would let a stale session silently undo a Start Over performed elsewhere. """ try: stored_revision = int(existing.get("revision")) except (AttributeError, TypeError, ValueError): return False if stored_revision != int(expected_revision) + 1: return False for key in ("answers", "convo", "chat_display", "followups", "user_id"): if key in payload and existing.get(key) != payload[key]: return False return True @staticmethod def _revision_of(row, fallback): try: return int(row.get("revision")) except (AttributeError, TypeError, ValueError): return fallback def delete_session(self, email): """Remove the resumable state for an email (the message log is kept).""" email = normalize_email(email) if not email: return False, "No email supplied." ok, _, err = self._request( "DELETE", self.sessions_table, params={"email": "eq." + email}) return (True, None) if ok else (False, err) # ---- append-only message log ---- def append_messages(self, email, messages): """messages: [{'role','content','turn'(optional)}]. Returns (ok, error).""" email = normalize_email(email) if not email or not messages: return True, None now = _now_iso() rows = [] for m in messages: role = m.get("role") content = m.get("content") if role not in ("user", "assistant") or not isinstance(content, str): continue try: turn = int(m["turn"]) if m.get("turn") is not None else None except (TypeError, ValueError): turn = None # Every row must carry an identical key set: PostgREST rejects a # batch whose objects differ ("All object keys must match"), which # would silently drop the whole batch since logging is best-effort. rows.append({"email": email, "role": role, "content": content, "turn": turn, "created_at": now}) if not rows: return True, None ok, _, err = self._request( "POST", self.messages_table, body=rows, prefer="return=minimal") return (True, None) if ok else (False, err) def delete_messages(self, email): """Remove an email's transcript log. Only used by verify_supabase.py.""" email = normalize_email(email) if not email: return False, "No email supplied." ok, _, err = self._request( "DELETE", self.messages_table, params={"email": "eq." + email}) return (True, None) if ok else (False, err) def load_messages(self, email, limit=1000): """Full chronological turn log for an email. Returns (rows, error).""" email = normalize_email(email) if not email: return [], "No email supplied." ok, data, err = self._request( "GET", self.messages_table, params={"select": "*", "email": "eq." + email, "order": "id.asc", "limit": str(int(limit))}, ) if not ok: return [], err return (data if isinstance(data, list) else []), None