verify_supabase.py

"""One-command check that a Supabase project is set up for the Creighton Anatomy Learning Platform.

Run it after creating the tables with supabase_schema.sql:

    export SUPABASE_URL="https://YOUR-PROJECT-REF.supabase.co"
    export SUPABASE_KEY="YOUR-SERVICE-ROLE-KEY"
    python3 verify_supabase.py

It creates a throwaway account (verify-<timestamp>), exercises every operation
the app performs (account creation, sign-in checks, chat history, session
timing, click events), reads everything back, and deletes the account again.
It touches no real user data. Exit code 0 means every check passed.
"""

from __future__ import annotations

import sys
import time
import uuid

from supabase_store import EXISTS, SupabaseStore, is_valid_email, is_valid_user_id

PASS, FAIL = "  PASS  ", "  FAIL  "
failures: list[str] = []


def check(name: str, ok: bool, detail: str = "") -> bool:
    print((PASS if ok else FAIL) + name + (f"   [{detail}]" if detail and not ok else ""))
    if not ok:
        failures.append(name)
    return ok


def main() -> int:
    store = SupabaseStore()

    print("\n1. Configuration")
    if not check("SUPABASE_URL and SUPABASE_KEY are set", store.enabled, store.config_hint()):
        print('\nSet them first:\n  export SUPABASE_URL="https://YOUR-PROJECT-REF.supabase.co"\n  export SUPABASE_KEY="YOUR-SERVICE-ROLE-KEY"')
        return 2
    print(f"       url = {store.url}")
    print(f"       key = {store.key[:10]}...{store.key[-4:]}  ({len(store.key)} chars)")

    print("\n2. Connectivity and schema")
    ok, error = store.ping()
    if not check("can read anatomy_users", ok, str(error)):
        print("\n   -> 'relation ... does not exist': run supabase_schema.sql in Supabase Studio > SQL Editor."
              "\n   -> 'permission denied' or 'JWT': use the service_role key (Project Settings > API), not the anon key.")
        return 1
    for table in ("anatomy_messages", "anatomy_sessions", "anatomy_events"):
        ok, _, error = store._request("GET", table, params={"select": "*", "limit": "1"})
        check(f"can read {table}", ok, str(error))

    user_id = f"verify-{int(time.time())}"
    email = f"{user_id}@example.invalid"
    print(f"\n3. Accounts (throwaway user {user_id})")
    check("validators accept the test user", is_valid_user_id(user_id) and is_valid_email(email))
    row, error = store.get_user(user_id)
    check("unknown user ID is reported as absent", row is None and error is None, str(error))
    row, error = store.create_user(user_id, email)
    check("create_user stores the account", error is None and row is not None and row.get("email") == email, str(error))
    _, error = store.create_user(user_id, "someone.else@example.invalid")
    check("create_user refuses a taken user ID", error == EXISTS, str(error))
    row, error = store.get_user(user_id)
    check("get_user returns the account with its email", row is not None and row.get("email") == email, str(error))
    ok, error = store.record_login(user_id)
    check("record_login updates last_login_at", ok, str(error))

    session_id = uuid.uuid4().hex
    print("\n4. Time on site")
    ok, error = store.start_session(session_id, user_id)
    check("start_session inserts a session row", ok, str(error))
    time.sleep(1.1)
    ok, error = store.heartbeat(session_id)
    check("heartbeat updates last_seen_at", ok, str(error))
    ok, error = store.end_session(session_id)
    check("end_session sets ended_at", ok, str(error))
    ok, data, error = store._request("GET", "anatomy_sessions", params={"select": "started_at,last_seen_at,ended_at", "session_id": "eq." + session_id})
    session_row = data[0] if ok and data else {}
    check("session row has started_at < last_seen_at and ended_at",
          bool(session_row) and session_row["started_at"] < session_row["last_seen_at"] and session_row["ended_at"],
          str(error or session_row))
    ok, data, error = store._request("GET", "anatomy_time_per_user", params={"select": "seconds_on_site,sessions", "user_id": "eq." + user_id})
    check("anatomy_time_per_user view reports the session", ok and data and data[0]["sessions"] == 1 and data[0]["seconds_on_site"] >= 1, str(error or data))

    print("\n5. Clicks")
    for event, detail in (("sign_in", {}), ("ask", {"chars": 12}), ("ask", {"chars": 30}), ("browse_slide", {"lecture": 7, "slide": 3})):
        ok, error = store.log_event(user_id, session_id, event, detail)
        check(f"log_event stores '{event}'", ok, str(error))
    ok, data, error = store._request("GET", "anatomy_button_clicks", params={"select": "event,clicks", "user_id": "eq." + user_id, "order": "event.asc"})
    counts = {r["event"]: r["clicks"] for r in (data or [])} if ok else {}
    check("anatomy_button_clicks view counts clicks per button", counts == {"ask": 2, "browse_slide": 1, "sign_in": 1}, str(error or counts))

    print("\n6. Chat history")
    ok, error = store.append_messages(user_id, session_id, [
        {"role": "user", "content": "Which nerve roots form the brachial plexus?", "mode": "ask", "topic": "brachial plexus", "source": None},
        {"role": "assistant", "content": "C5 to T1 (Lecture 7, slide 3).\n\n---\n**Source:** Course lecture slides: Lecture 7 (Brachial Plexus), slide 3.",
         "mode": "ask", "topic": "brachial plexus", "source": "slides"},
    ])
    check("append_messages stores a question and its answer", ok, str(error))
    ok, error = store.append_messages(user_id, session_id, [{"role": "user", "content": "Quiz me on: brachial plexus", "mode": "quiz", "topic": "brachial plexus", "source": None}])
    check("append_messages stores a later message", ok, str(error))
    rows, error = store.load_messages(user_id)
    check("load_messages returns all messages oldest first",
          error is None and [r["role"] for r in rows] == ["user", "assistant", "user"] and rows[1]["source"] == "slides",
          str(error or [r.get("role") for r in rows]))
    rows, error = store.load_messages(user_id, limit=2)
    check("load_messages(limit) keeps the most recent messages in order",
          error is None and [r["role"] for r in rows] == ["assistant", "user"], str(error or rows))

    print("\n7. Background writer")
    store.enqueue("log_event", user_id, session_id, "sign_out", {})
    check("queued write finishes", store.flush(10.0))
    ok, data, error = store._request("GET", "anatomy_events", params={"select": "event", "user_id": "eq." + user_id, "event": "eq.sign_out"})
    check("queued event is stored", ok and len(data or []) == 1, str(error or data))

    print("\n8. Cleanup")
    ok, error = store.delete_user(user_id)
    check("delete_user removes the account and its data", ok, str(error))
    row, error = store.get_user(user_id)
    check("account is gone", row is None and error is None, str(error))

    print()
    if failures:
        print(f"{len(failures)} check(s) failed: " + ", ".join(failures))
        return 1
    print("All checks passed. Supabase is ready for the app.")
    return 0


if __name__ == "__main__":
    sys.exit(main())