scripts/smoke_supabase.py

"""Supabase smoke check for schema and bucket availability.

This script reads only from environment variables. It intentionally avoids
hardcoded project keys and does not upload user PDFs.
"""
from __future__ import annotations

import asyncio
import os
from pathlib import Path

import httpx
from dotenv import load_dotenv


REQUIRED_TABLES = [
    "profiles",
    "submissions",
    "reviews",
    "feedback",
    "artifacts",
    "pipeline_runs",
    "pipeline_steps",
    "events",
    "provider_calls",
    "evidence_items",
]

REQUIRED_VIEWS = ["submission_usage_summary"]


async def main() -> int:
    load_dotenv(Path(__file__).resolve().parents[1] / ".env")
    url = os.getenv("SUPABASE_URL", "").rstrip("/")
    key = os.getenv("SUPABASE_SERVICE_ROLE_KEY", "")
    bucket = os.getenv("SUPABASE_STORAGE_BUCKET", "paper-review-artifacts")
    if not url or not key:
        print("Missing SUPABASE_URL or SUPABASE_SERVICE_ROLE_KEY")
        return 2

    headers = {"apikey": key, "Authorization": f"Bearer {key}"}
    ok = True
    async with httpx.AsyncClient(base_url=url, headers=headers, timeout=20.0) as client:
        for table in REQUIRED_TABLES:
            response = await client.get(f"/rest/v1/{table}", params={"select": "*", "limit": "1"})
            status_ok = response.status_code == 200
            ok = ok and status_ok
            print(f"TABLE {table}: {response.status_code}")

        for view in REQUIRED_VIEWS:
            response = await client.get(f"/rest/v1/{view}", params={"select": "*", "limit": "1"})
            status_ok = response.status_code == 200
            ok = ok and status_ok
            print(f"VIEW {view}: {response.status_code}")

        response = await client.get("/storage/v1/bucket")
        if response.status_code == 200:
            bucket_ids = [item.get("id") for item in response.json()]
            bucket_ok = bucket in bucket_ids
            ok = ok and bucket_ok
            print(f"BUCKET {bucket}: {bucket_ok}")
        else:
            ok = False
            print(f"BUCKET_CHECK: {response.status_code}")

    return 0 if ok else 1


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))