chatkit/backend/app/storage.py

"""Raw-file storage on Supabase Storage (object storage).

Bytes live in a Storage bucket; only metadata + the object path are kept in the
``raw_files`` table. Uploads are best-effort: if storage isn't configured (e.g.
local dev or SQLite mode), uploads are skipped and an empty path is returned so
the rest of the flow keeps working.
"""

from __future__ import annotations

from functools import lru_cache
from typing import Optional

from starlette.concurrency import run_in_threadpool

from .config import get_settings


def storage_enabled() -> bool:
    s = get_settings()
    return bool(s.supabase_url and s.supabase_service_role_key)


@lru_cache
def _client():
    from supabase import create_client

    s = get_settings()
    return create_client(s.supabase_url, s.supabase_service_role_key)


def _ensure_bucket(bucket: str):
    client = _client()
    try:
        existing = {b.name for b in client.storage.list_buckets()}
        if bucket not in existing:
            client.storage.create_bucket(bucket, options={"public": False})
    except Exception:
        # Bucket may already exist or listing may be restricted; uploads will surface real errors.
        pass


def _upload_sync(path: str, data: bytes, content_type: str) -> str:
    s = get_settings()
    bucket = s.supabase_storage_bucket
    _ensure_bucket(bucket)
    client = _client()
    client.storage.from_(bucket).upload(
        path,
        data,
        {"content-type": content_type or "application/octet-stream", "upsert": "true"},
    )
    return path


async def upload_raw_file(
    quiz_id: int, data_type: str, file_name: str, data: bytes, content_type: str
) -> Optional[str]:
    """Upload bytes and return the storage object path, or None if disabled."""
    if not storage_enabled():
        return None
    safe_name = file_name.replace("/", "_") or "file"
    path = f"quiz_{quiz_id}/{data_type}/{safe_name}"
    return await run_in_threadpool(_upload_sync, path, data, content_type)


def _signed_url_sync(path: str, expires_in: int) -> Optional[str]:
    s = get_settings()
    client = _client()
    res = client.storage.from_(s.supabase_storage_bucket).create_signed_url(path, expires_in)
    # supabase-py has used both 'signedURL' and 'signedUrl' across versions
    if isinstance(res, dict):
        return res.get("signedURL") or res.get("signedUrl") or res.get("signed_url")
    return None


async def create_signed_url(storage_path: str, expires_in: int = 3600) -> Optional[str]:
    """Return a short-lived download URL for a stored object, or None if unavailable."""
    if not storage_path or not storage_enabled():
        return None
    try:
        return await run_in_threadpool(_signed_url_sync, storage_path, expires_in)
    except Exception:
        return None