""" backend/memory/oracle_sync.py — Supabase ↔ Oracle DB dual-write synchronizer. Two public classes: MemorySynchronizer — Wraps a primary (Supabase) and a secondary (Oracle) MemoryBackend. Writes always go to both; Oracle failures are non-fatal (logged but not re-raised) because Oracle is a secondary replica, not the source of truth. OracleState — Singleton lifecycle manager hooked from main.py _on_startup(). Reads ORACLE_DB_* env vars, initialises OracleMemoryAdapter if they are set, and exposes oracle_adapter and memory_synchronizer as class-level attributes for use by other API modules. Design decisions: - IDs are generated once by MemorySynchronizer (uuid4) and passed to BOTH backends so keys stay consistent (BUG-1 fix). - Secondary write is wrapped in try/except so a primary success is never rolled back due to an Oracle failure (BUG-4 fix). - sync_update / sync_delete return a dict {'primary': bool, 'secondary': bool|None} so callers can distinguish which backend failed (BUG-8 fix). - OracleState checks oracle_adapter.is_connected, not the object's truthiness, to avoid initialising the synchronizer against an unconnected adapter (BUG-3 fix). """ from __future__ import annotations import logging import os import uuid from typing import Any, Dict, List, Optional from .memory_backend import MemoryBackend from .oracle_adapter import OracleMemoryAdapter _logger = logging.getLogger("memory.oracle_sync") # ── Synchronizer ───────────────────────────────────────────────────────────── class MemorySynchronizer: """ Dual-write coordinator: primary backend is authoritative, secondary is best-effort. Args: primary: Authoritative backend (Supabase). Errors propagate. secondary: Replica backend (Oracle). Errors are logged, never raised. """ def __init__(self, primary: MemoryBackend, secondary: OracleMemoryAdapter) -> None: self.primary = primary self.secondary = secondary # ── Writes ─────────────────────────────────────────────────────────────── async def sync_add_memory( self, user_id: str, memory_data: Dict[str, Any] ) -> str: """Insert into primary then Oracle. Returns the shared memory_id.""" shared_id = str(uuid.uuid4()) # single ID for both backends await self.primary.add_memory(user_id, memory_data, memory_id=shared_id) if self.secondary.is_connected: try: await self.secondary.add_memory( user_id, memory_data, memory_id=shared_id ) except Exception as exc: _logger.warning( "OracleSync.add_memory: Oracle write failed (non-fatal) — %s", exc ) return shared_id async def sync_update_memory( self, user_id: str, memory_id: str, new_data: Dict[str, Any] ) -> Dict[str, Optional[bool]]: """ Update both backends. Returns: {'primary': bool, 'secondary': bool | None} None means Oracle is not connected or the call was skipped. """ primary_ok = await self.primary.update_memory(user_id, memory_id, new_data) secondary_ok: Optional[bool] = None if self.secondary.is_connected: try: secondary_ok = await self.secondary.update_memory( user_id, memory_id, new_data ) except Exception as exc: _logger.warning( "OracleSync.update_memory: Oracle update failed — %s", exc ) secondary_ok = False return {"primary": primary_ok, "secondary": secondary_ok} async def sync_delete_memory( self, user_id: str, memory_id: str ) -> Dict[str, Optional[bool]]: """ Delete from both backends. Returns: {'primary': bool, 'secondary': bool | None} """ primary_ok = await self.primary.delete_memory(user_id, memory_id) secondary_ok: Optional[bool] = None if self.secondary.is_connected: try: secondary_ok = await self.secondary.delete_memory(user_id, memory_id) except Exception as exc: _logger.warning( "OracleSync.delete_memory: Oracle delete failed — %s", exc ) secondary_ok = False return {"primary": primary_ok, "secondary": secondary_ok} # ── Reads ──────────────────────────────────────────────────────────────── async def get_memory_from_primary( self, user_id: str, memory_id: str ) -> Optional[Dict[str, Any]]: return await self.primary.get_memory(user_id, memory_id) async def get_memory_from_secondary( self, user_id: str, memory_id: str ) -> Optional[Dict[str, Any]]: if not self.secondary.is_connected: return None return await self.secondary.get_memory(user_id, memory_id) async def list_memories_from_primary( self, user_id: str, limit: int = 100, offset: int = 0 ) -> List[Dict[str, Any]]: return await self.primary.list_memories(user_id, limit, offset) async def list_memories_from_secondary( self, user_id: str, limit: int = 100, offset: int = 0 ) -> List[Dict[str, Any]]: if not self.secondary.is_connected: return [] return await self.secondary.list_memories(user_id, limit, offset) # ── Lifecycle singleton ─────────────────────────────────────────────────────── class OracleState: """ Singleton lifecycle manager for the Oracle memory backend. Hooked from backend/main.py::_on_startup() — call await OracleState.initialize(). The synchronizer is wired lazily; if ORACLE_DB_* vars are absent the class silently stays disabled so the rest of the API is unaffected. Usage after startup: from memory.oracle_sync import OracleState adapter = OracleState.oracle_adapter # None if not configured sync = OracleState.memory_synchronizer # None if not configured """ oracle_adapter: Optional[OracleMemoryAdapter] = None memory_synchronizer: Optional[MemorySynchronizer] = None _initialized: bool = False @classmethod async def initialize(cls) -> None: if cls._initialized: return user = os.getenv("ORACLE_DB_USER") password = os.getenv("ORACLE_DB_PASSWORD") dsn = os.getenv("ORACLE_DB_DSN") if not all([user, password, dsn]): _logger.info( "OracleState: ORACLE_DB_* env vars not set — Oracle backend disabled." ) cls._initialized = True return adapter = OracleMemoryAdapter() try: await adapter.connect({"user": user, "password": password, "dsn": dsn}) except Exception as exc: _logger.warning( "OracleState: connect failed (%s) — Oracle backend disabled.", exc ) cls._initialized = True return cls.oracle_adapter = adapter # Wire up MemorySynchronizer only when a primary backend is available. # SupabaseMemoryBackend (backend/memory/supabase_backend.py) wraps the # existing state._sb client. If that module doesn't exist yet the # synchronizer stays None; oracle_adapter is still available for direct use. try: from memory.supabase_backend import SupabaseMemoryBackend # type: ignore[import] cls.memory_synchronizer = MemorySynchronizer( primary=SupabaseMemoryBackend(), secondary=cls.oracle_adapter ) _logger.info( "OracleState: MemorySynchronizer ready (Supabase→Oracle dual-write active)." ) except ImportError: _logger.info( "OracleState: supabase_backend not found — " "oracle_adapter available for direct use; synchronizer disabled." ) cls._initialized = True @classmethod async def shutdown(cls) -> None: """Call from _on_shutdown() to close the Oracle pool gracefully.""" if cls.oracle_adapter and cls.oracle_adapter.is_connected: await cls.oracle_adapter.disconnect() cls._initialized = False