api/public_snapshot.py

"""Writer idempotente per lo snapshot pubblico operativo.

Lo snapshot è un DTO sanificato e singleton: viene aggiornato dal backend con
service_role e letto dall'endpoint pubblico senza esporre task, sessioni o log.
Un errore di persistenza non deve impedire l'avvio del servizio.
"""
from __future__ import annotations

import asyncio
import logging
import os
from typing import Any

from .state import _agent_tasks, _loop_registry

try:
    from supabase import create_client
except ImportError:  # pragma: no cover - deployment dependency guard
    create_client = None

_snapshot_client: Any | None = None
_snapshot_client_initialized = False


def get_snapshot_client() -> Any | None:
    """Return one stable Supabase client for the canonical snapshot project.

    Unlike the general round-robin pool, this client never changes project
    between a write and a read. Dedicated variables can be used when the
    snapshot lives in a separate project; otherwise project A is canonical.
    """
    global _snapshot_client, _snapshot_client_initialized
    if _snapshot_client_initialized:
        return _snapshot_client
    _snapshot_client_initialized = True
    if create_client is None:
        _logger.warning("BOOT: public snapshot skipped — supabase package unavailable")
        return None
    url = os.getenv("PUBLIC_SNAPSHOT_SUPABASE_URL") or os.getenv("SUPABASE_URL")
    key = (
        os.getenv("PUBLIC_SNAPSHOT_SUPABASE_SERVICE_ROLE_KEY")
        or os.getenv("SUPABASE_SERVICE_ROLE_KEY")
        or os.getenv("SUPABASE_KEY")
    )
    if not url or not key:
        _logger.warning("BOOT: public snapshot skipped — canonical Supabase URL/key unavailable")
        return None
    try:
        _snapshot_client = create_client(url, key)
        _logger.info("BOOT: canonical Supabase snapshot client initialized")
    except Exception as exc:
        _logger.warning("BOOT: canonical Supabase snapshot client init failed type=%s", type(exc).__name__)
    return _snapshot_client
from .version import RUNTIME_VERSION

_logger = logging.getLogger("agente_ai.public_snapshot")

_ACTIVE_STATUSES = {"RUNNING", "IN_PROGRESS", "EXECUTING", "PROCESSING"}


def _snapshot_row() -> dict[str, Any]:
    tasks = list(_agent_tasks.values())
    return {
        "singleton": True,
        "service_status": "operational",
        "active_sessions": len(_loop_registry),
        "queued_tasks": sum(1 for task in tasks if str(task.get("status", "")).upper() == "QUEUED"),
        "in_progress_tasks": sum(1 for task in tasks if str(task.get("status", "")).upper() in _ACTIVE_STATUSES),
        "app_version": RUNTIME_VERSION,
    }


async def write_public_dashboard_snapshot() -> bool:
    """Upsert the singleton public snapshot without blocking application startup.

    Returns ``False`` on missing Supabase or a schema/permission error. The
    public status route remains available with its degraded response while the
    service continues booting; the exact exception is kept in backend logs.
    """
    client = get_snapshot_client()
    if client is None:
        _logger.warning("BOOT: public snapshot skipped — Supabase client unavailable")
        return False

    row = _snapshot_row()

    def operation() -> None:
        client.table("public_dashboard_snapshot").upsert(row, on_conflict="singleton").execute()

    last_error: Exception | None = None
    for attempt in range(1, 4):
        try:
            await asyncio.to_thread(operation)
            _logger.info(
                "BOOT: public snapshot upserted status=%s sessions=%d queued=%d in_progress=%d version=%s",
                row["service_status"], row["active_sessions"], row["queued_tasks"],
                row["in_progress_tasks"], row["app_version"],
            )
            return True
        except Exception as exc:
            last_error = exc
            if attempt < 3:
                await asyncio.sleep(2)

    error_code = getattr(last_error, "code", None) or getattr(last_error, "status_code", None)
    _logger.warning(
        "BOOT: public snapshot upsert failed after retries type=%s code=%s; public status remains degraded",
        type(last_error).__name__ if last_error else "unknown",
        str(error_code)[:32] if error_code is not None else "unknown",
    )
    return False


async def refresh_public_dashboard_snapshot() -> bool:
    """Alias for future heartbeat refreshes; intentionally idempotent."""
    return await write_public_dashboard_snapshot()