backend/persistence.py

"""Persistence wrappers: Cloudflare R2 (objects) + Supabase (logging). Phase 5.

Both degrade to safe no-ops when their credentials are absent, so the pipeline runs
end-to-end locally. Tables match backend/db/schema.sql.
"""
from __future__ import annotations

import logging
import threading
import time
import uuid
from typing import Any, Callable, Optional

from .config import settings

logger = logging.getLogger(__name__)

# The supabase client is httpx-based and shared across the event-loop thread (job status
# updates) and the run_in_executor worker thread (pipeline logging). Serialize all access
# and retry transient transport errors so best-effort logging is actually reliable on the
# (slower) deployed box, where unsynchronized/stale connections intermittently dropped writes.
_SB_LOCK = threading.Lock()
_SB_RETRIES = 3


def _sb_call(fn: Callable[[], Any]) -> Any:
    last: Exception | None = None
    with _SB_LOCK:
        for i in range(_SB_RETRIES):
            try:
                return fn()
            except Exception as e:  # noqa: BLE001  (transient transport/connection errors)
                last = e
                time.sleep(0.3 * (i + 1))
    raise last  # type: ignore[misc]


# ---------------------------------------------------------------------------
# Object storage — Supabase Storage by default (no extra account), R2 if configured.
# ---------------------------------------------------------------------------
class ObjectStore:
    def __init__(self, supabase_client=None) -> None:
        self._r2 = None
        self._sb = supabase_client          # reuse the Supabase service_role client
        self._bucket = settings.storage_bucket
        if settings.r2_enabled:
            try:
                import boto3  # type: ignore
                self._r2 = boto3.client(
                    "s3", endpoint_url=settings.r2_endpoint,
                    aws_access_key_id=settings.r2_key_id,
                    aws_secret_access_key=settings.r2_secret,
                )
                logger.info("object store: Cloudflare R2 (bucket=%s)", settings.r2_bucket)
            except Exception as e:  # noqa: BLE001
                logger.warning("R2 init failed: %s", e)
        elif self._sb is not None:
            logger.info("object store: Supabase Storage (bucket=%s)", self._bucket)

    def upload(self, key: str, data: bytes, content_type: str) -> str:
        """Store bytes, return a public url. Falls back to a local placeholder if neither
        backend is configured (the frontend then shows its gradient placeholder)."""
        # 1) Cloudflare R2
        if self._r2 is not None:
            self._r2.put_object(Bucket=settings.r2_bucket, Key=key, Body=data,
                                ContentType=content_type)
            ep = (settings.r2_endpoint or "").rstrip("/")
            return f"{ep}/{settings.r2_bucket}/{key}"
        # 2) Supabase Storage (via the service_role client; bucket must be public)
        if self._sb is not None:
            try:
                _sb_call(lambda: self._sb.storage.from_(self._bucket).upload(
                    key, data, {"content-type": content_type, "upsert": "true"}))
                url = self._sb.storage.from_(self._bucket).get_public_url(key)
                return url if isinstance(url, str) else key
            except Exception as e:  # noqa: BLE001
                logger.warning("supabase storage upload failed: %s", e)
        # 3) local placeholder
        return f"local://uploads/{key}"


# ---------------------------------------------------------------------------
# Supabase logging
# ---------------------------------------------------------------------------
class SupabaseLogger:
    def __init__(self) -> None:
        self._client = None
        if settings.supabase_enabled:
            try:
                from supabase import create_client  # type: ignore
                self._client = create_client(settings.supabase_url, settings.supabase_key)
                logger.info("Supabase logging enabled")
            except Exception as e:  # noqa: BLE001
                logger.warning("Supabase init failed, logging disabled: %s", e)

    @property
    def enabled(self) -> bool:
        return self._client is not None

    def _insert(self, table: str, rows: list[dict[str, Any]]) -> None:
        if self._client is None or not rows:
            return
        try:
            _sb_call(lambda: self._client.table(table).insert(rows).execute())
        except Exception as e:  # noqa: BLE001
            logger.warning("supabase insert into %s failed: %s", table, e)

    # ---- jobs ----
    def upsert_job(self, row: dict[str, Any]) -> None:
        if self._client is None:
            return
        try:
            _sb_call(lambda: self._client.table("jobs").upsert(row).execute())
        except Exception as e:  # noqa: BLE001
            logger.warning("supabase upsert job failed: %s", e)

    # ---- inference_logs (one row per classified segment) ----
    def log_segments(self, rows: list[dict[str, Any]]) -> None:
        self._insert("inference_logs", rows)

    # ---- review_queue (uncertain predictions) ----
    def log_review(self, rows: list[dict[str, Any]]) -> None:
        self._insert("review_queue", rows)

    # ---- health_snapshots (per-upload aggregate) ----
    def log_snapshot(self, row: dict[str, Any]) -> None:
        self._insert("health_snapshots", [row])

    # ---- reads (Phase 6 review/tracker + Phase 7 dashboard) ----
    def _select(self, table: str, build) -> list[dict[str, Any]]:
        if self._client is None:
            return []
        try:
            return _sb_call(lambda: build(self._client.table(table).select("*")).execute().data) or []
        except Exception as e:  # noqa: BLE001
            logger.warning("supabase select from %s failed: %s", table, e)
            return []

    def review_queue(self, limit: int = 50) -> list[dict[str, Any]]:
        return self._select("review_queue", lambda q: q.eq("status", "pending")
                            .order("created_at", desc=True).limit(limit))

    def reef_locations(self) -> list[dict[str, Any]]:
        return self._select("reef_locations", lambda q: q.order("name"))

    def snapshots(self, reef_id: str) -> list[dict[str, Any]]:
        return self._select("health_snapshots", lambda q: q.eq("reef_location_id", reef_id)
                            .order("snapshot_time"))

    def recent_logs(self, limit: int = 5000) -> list[dict[str, Any]]:
        return self._select("inference_logs", lambda q: q.order("ts", desc=True).limit(limit))

    def confirm_label(self, review_id: str, label: str, labeled_by: str = "admin") -> bool:
        if self._client is None:
            return False
        try:
            row = _sb_call(lambda: self._client.table("review_queue").select("*").eq("id", review_id).execute().data)
            if not row:
                return False
            r = row[0]
            _sb_call(lambda: self._client.table("human_labels").insert({
                "review_queue_id": review_id, "image_id": r.get("image_id"),
                "segment_id": r.get("segment_id"), "confirmed_label": label,
                "labeled_by": labeled_by,
            }).execute())
            _sb_call(lambda: self._client.table("review_queue").update({"status": "confirmed"}).eq("id", review_id).execute())
            return True
        except Exception as e:  # noqa: BLE001
            logger.warning("confirm_label failed: %s", e)
            return False


def new_request_id() -> str:
    return str(uuid.uuid4())


supabase = SupabaseLogger()
r2 = ObjectStore(supabase_client=supabase._client)  # exported as `r2` for pipeline import