api/global_state_sync_optimized.py

"""
global_state_sync_optimized.py — Sincronizzazione Selettiva Ottimizzata per Workload Separati

Architettura Separazione Workload:
  A: Analytics/Cache/Read-Heavy (NO sincronizzazione, solo letture)
  B: Sync/State/Transazioni (PRIMARY — sincronizzazione completa)
  C: Memory/RAG/Embeddings (sincronizzazione memoria e embeddings)
  D: Audit/Logging/Compliance (sincronizzazione audit-only, append-only)

Strategia Sincronizzazione:
  1. B ← A: Nessuna (A è read-only, non genera stato)
  2. B ← C: Sincronizzazione memoria/skill (bidirezionale)
  3. B ← D: Sincronizzazione audit (unidirezionale, D → B)
  4. C ← B: Sincronizzazione stato globale (unidirezionale, B → C)
  5. D ← B: Sincronizzazione eventi (unidirezionale, B → D)

Benefici:
  - Riduce traffico di sincronizzazione (solo tabelle rilevanti)
  - Evita conflitti di write (ogni nodo scrive su tabelle specifiche)
  - Mantiene audit immutabile (D append-only)
  - Massimizza throughput (parallelizzazione per nodo)
"""

import os
import asyncio
import logging
from typing import Optional, Dict, List, Any, Literal
from datetime import datetime, timedelta
from enum import Enum
import json

_logger = logging.getLogger("global_state_sync_optimized")

# ── Enumerazione Nodi ──────────────────────────────────────────────────────
class SupabaseNode(str, Enum):
    A = "A"  # Analytics/Cache
    B = "B"  # PRIMARY (Sync/State)
    C = "C"  # Memory/RAG
    D = "D"  # Audit/Logging


# ── Configurazione Nodi ───────────────────────────────────────────────────
SUPABASE_CONFIG = {
    "A": {
        "url": os.getenv("SUPABASE_URL_A", ""),
        "key": os.getenv("SUPABASE_KEY_A", ""),
        "role": "Analytics/Cache",
        "sync_enabled": False,  # A non sincronizza (read-only)
    },
    "B": {
        "url": os.getenv("SUPABASE_URL", ""),  # PRIMARY
        "key": os.getenv("SUPABASE_KEY", ""),
        "role": "Sync/State (PRIMARY)",
        "sync_enabled": True,
    },
    "C": {
        "url": os.getenv("SUPABASE_URL_C", ""),
        "key": os.getenv("SUPABASE_KEY_C", ""),
        "role": "Memory/RAG",
        "sync_enabled": True,
    },
    "D": {
        "url": os.getenv("SUPABASE_URL_D", ""),
        "key": os.getenv("SUPABASE_KEY_D", ""),
        "role": "Audit/Logging",
        "sync_enabled": True,
    },
}

# ── Mapping Tabelle → Nodi ─────────────────────────────────────────────────
TABLE_NODE_MAPPING = {
    # Nodo B (PRIMARY) — Stato Globale
    "cluster_state": "B",
    "global_state": "B",
    "sync_state": "B",
    "agent_state": "B",
    "daemon_status": "B",
    
    # Nodo C — Memoria e RAG
    "agent_memory": "C",
    "skill_memory": "C",
    "embeddings": "C",
    "conversations": "C",
    "rag_index": "C",
    "vector_store": "C",
    
    # Nodo D — Audit e Logging
    "audit_events": "D",
    "audit_log": "D",
    "compliance_log": "D",
    "event_log": "D",
    "activity_log": "D",
}

# ── Configurazione Sincronizzazione ────────────────────────────────────────
SYNC_RULES = {
    # (source_node, target_node): [tabelle da sincronizzare]
    ("B", "C"): ["agent_memory", "skill_memory", "cluster_state"],  # B → C: stato globale
    ("C", "B"): ["agent_memory", "skill_memory"],  # C → B: memoria aggiornata
    ("B", "D"): ["audit_events", "event_log"],  # B → D: eventi
    ("D", "B"): [],  # D → B: nessuna (audit è append-only)
}

SYNC_ENABLED = os.getenv("GLOBAL_STATE_SYNC_ENABLED", "true").lower() == "true"
SYNC_INTERVAL_SECONDS = int(os.getenv("SYNC_INTERVAL_SECONDS", "30"))


# ── Client Supabase ───────────────────────────────────────────────────────
class SupabaseClient:
    """Client per accedere a un singolo database Supabase."""
    
    def __init__(self, url: str, key: str, node_id: str):
        self.url = url
        self.key = key
        self.node_id = node_id
        self.base_url = f"{url}/rest/v1" if url else None
    
    async def query(self, table: str, filters: Optional[Dict] = None) -> List[Dict]:
        """Esegue una query SELECT su una tabella."""
        if not self.base_url:
            return []
        
        import httpx
        
        url = f"{self.base_url}/{table}"
        headers = {
            "apikey": self.key,
            "Authorization": f"Bearer {self.key}",
            "Content-Type": "application/json",
        }
        
        try:
            async with httpx.AsyncClient() as client:
                response = await client.get(url, headers=headers, timeout=10.0)
                if response.status_code == 200:
                    return response.json()
                else:
                    _logger.warning(f"Supabase {self.node_id} query failed: {response.status_code}")
                    return []
        except Exception as exc:
            _logger.error(f"Supabase {self.node_id} error: {exc}")
            return []
    
    async def insert(self, table: str, data: Dict) -> bool:
        """Inserisce un record in una tabella."""
        if not self.base_url:
            return False
        
        import httpx
        
        url = f"{self.base_url}/{table}"
        headers = {
            "apikey": self.key,
            "Authorization": f"Bearer {self.key}",
            "Content-Type": "application/json",
        }
        
        try:
            async with httpx.AsyncClient() as client:
                response = await client.post(url, json=data, headers=headers, timeout=10.0)
                return response.status_code in (200, 201)
        except Exception as exc:
            _logger.error(f"Supabase {self.node_id} insert error: {exc}")
            return False


# ── Sincronizzazione Selettiva ─────────────────────────────────────────────
class SelectiveSyncManager:
    """Gestisce la sincronizzazione selettiva tra nodi."""
    
    def __init__(self):
        self.clients = {}
        self._enabled = SYNC_ENABLED
        self._last_sync = {}
        
        for node_id, config in SUPABASE_CONFIG.items():
            if config["url"] and config["key"]:
                self.clients[node_id] = SupabaseClient(config["url"], config["key"], node_id)
    
    async def sync_all(self) -> Dict[str, Any]:
        """Esegue la sincronizzazione completa tra tutti i nodi."""
        if not self._enabled or not self.clients:
            return {"ok": False, "error": "Sync disabled or no clients configured"}
        
        results = {
            "ok": True,
            "synced_at": datetime.now().isoformat(),
            "syncs": [],
        }
        
        # Esegui sincronizzazioni secondo le regole
        for (source, target), tables in SYNC_RULES.items():
            if not tables:
                continue  # Salta se nessuna tabella da sincronizzare
            
            source_client = self.clients.get(source)
            target_client = self.clients.get(target)
            
            if not source_client or not target_client:
                continue
            
            sync_result = await self._sync_nodes(source_client, target_client, tables)
            results["syncs"].append(sync_result)
        
        return results
    
    async def _sync_nodes(
        self,
        source: SupabaseClient,
        target: SupabaseClient,
        tables: List[str],
    ) -> Dict[str, Any]:
        """Sincronizza tabelle specifiche da source a target."""
        result = {
            "source": source.node_id,
            "target": target.node_id,
            "tables": tables,
            "synced_count": 0,
            "errors": [],
        }
        
        for table in tables:
            try:
                # Leggi da source
                rows = await source.query(table)
                if not rows:
                    continue
                
                # Scrivi su target (upsert logic)
                for row in rows:
                    success = await target.insert(table, row)
                    if success:
                        result["synced_count"] += 1
                    else:
                        result["errors"].append(f"Failed to sync {table} row")
            
            except Exception as exc:
                result["errors"].append(f"Error syncing {table}: {str(exc)}")
                _logger.error(f"Sync error {source.node_id} → {target.node_id} ({table}): {exc}")
        
        return result
    
    async def sync_memory(self, session_id: str) -> Dict[str, Any]:
        """Sincronizza memoria per una sessione specifica (B ↔ C)."""
        if not self._enabled:
            return {"ok": False}
        
        source_client = self.clients.get("B")
        target_client = self.clients.get("C")
        
        if not source_client or not target_client:
            return {"ok": False, "error": "B or C not configured"}
        
        try:
            # Leggi memoria da B
            rows = await source_client.query("agent_memory", {"session_id": session_id})
            
            # Scrivi su C
            for row in rows:
                await target_client.insert("agent_memory", row)
            
            return {
                "ok": True,
                "session_id": session_id,
                "synced_rows": len(rows),
            }
        except Exception as exc:
            _logger.error(f"Memory sync error: {exc}")
            return {"ok": False, "error": str(exc)}
    
    async def log_audit_event(self, event: Dict) -> bool:
        """Registra un evento di audit su D (append-only)."""
        if not self._enabled:
            return False
        
        target_client = self.clients.get("D")
        if not target_client:
            return False
        
        event["timestamp"] = datetime.now().isoformat()
        return await target_client.insert("audit_events", event)


# ── Scheduler Sincronizzazione ─────────────────────────────────────────────
class SyncScheduler:
    """Scheduler per sincronizzazione periodica."""
    
    def __init__(self, manager: SelectiveSyncManager):
        self.manager = manager
        self._running = False
    
    async def start(self):
        """Avvia il scheduler di sincronizzazione."""
        self._running = True
        _logger.info(f"Starting sync scheduler (interval: {SYNC_INTERVAL_SECONDS}s)")
        
        while self._running:
            try:
                result = await self.manager.sync_all()
                if result["ok"]:
                    _logger.debug(f"Sync completed: {len(result['syncs'])} operations")
                else:
                    _logger.warning(f"Sync failed: {result.get('error')}")
            except Exception as exc:
                _logger.error(f"Sync scheduler error: {exc}")
            
            await asyncio.sleep(SYNC_INTERVAL_SECONDS)
    
    def stop(self):
        """Ferma il scheduler."""
        self._running = False
        _logger.info("Sync scheduler stopped")


# ── Istanza Globale ───────────────────────────────────────────────────────
_sync_manager = SelectiveSyncManager()
_sync_scheduler = SyncScheduler(_sync_manager)


# ── Funzioni Pubbliche ────────────────────────────────────────────────────
async def get_unified_memory(session_id: str) -> Dict[str, Any]:
    """Recupera memoria unificata per una sessione."""
    return await _sync_manager.sync_memory(session_id)


async def sync_all() -> Dict[str, Any]:
    """Esegue sincronizzazione completa."""
    return await _sync_manager.sync_all()


async def log_audit(event: Dict) -> bool:
    """Registra un evento di audit."""
    return await _sync_manager.log_audit_event(event)


async def start_scheduler():
    """Avvia il scheduler di sincronizzazione."""
    await _sync_scheduler.start()


def stop_scheduler():
    """Ferma il scheduler."""
    _sync_scheduler.stop()