""" Knowledge Cache Service — 3-Layer Assets Knowledge Graph Cache. Layer 1: In-Memory RAM Dict (0ms Instant Return) Layer 2: Disk File (Assets Knowledge///cache.json) (0ms Disk Load into RAM) Layer 3: Supabase DB Fetch & 100% Real-Time Dynamic Metadata Enrichment (Zero Static Config Files) Reads all configurations strictly from live DB connections & .env. """ from __future__ import annotations import os import json import time from pathlib import Path from typing import Any, Optional import httpx from dotenv import load_dotenv # Explicitly load root .env file root_env = Path(__file__).resolve().parents[3] / ".env" if root_env.exists(): load_dotenv(dotenv_path=root_env) else: load_dotenv() SUPABASE_URL = os.getenv("NEXT_PUBLIC_SUPABASE_URL", "") SUPABASE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY") or os.getenv("NEXT_PUBLIC_SUPABASE_ANON_KEY", "") BASE_KNOWLEDGE_DIR = os.getenv( "ASSETS_KNOWLEDGE_DIR", os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "Assets Knowledge") ) # Standard Provider Metadata Mapping for Dynamic Enrichment DEFAULT_PROVIDER_METADATA: dict[str, dict[str, Any]] = { "shopify": { "name": "Shopify E-Commerce", "category": "store", "synonyms": ["shopify", "store", "shop", "sales", "revenue", "orders", "checkout"], "attribute_keys": ["totalRevenue", "totalOrders", "avgOrderValue", "checkouts"], "display_metrics": ["Total Revenue & Orders", "Average Order Value (AOV)", "Checkout Conversion Rate"], }, "ga4": { "name": "Google Analytics 4 (GA4)", "category": "analytics", "synonyms": ["ga4", "google analytics", "analytics", "traffic", "visitors", "sessions", "website"], "attribute_keys": ["activeUsers", "totalSessions", "bounceRate", "trafficSources"], "display_metrics": ["Active Users & Total Sessions", "Bounce Rate", "Traffic Channels"], }, "meta": { "name": "Meta Ads (Facebook/Instagram)", "category": "marketing", "synonyms": ["meta", "facebook", "instagram", "fb ads", "meta ads", "roas", "ctr", "ad spend"], "attribute_keys": ["spend", "roas", "ctr", "cpc", "impressions", "clicks"], "display_metrics": ["Ad Spend & Overall ROAS", "Click-Through Rate (CTR) & CPC", "Campaign Performance"], }, "google_ads": { "name": "Google Ads", "category": "marketing", "synonyms": ["google ads", "gads", "search ads", "pmax", "google ad spend"], "attribute_keys": ["totalImpressions", "totalClicks", "totalSpend", "averageCtr", "averageCpc"], "display_metrics": ["Google Ad Spend & ROAS", "Click-Through Rate (CTR)", "Search Campaign Breakdown"], }, "leads": { "name": "Leads & Contacts", "category": "crm", "synonyms": ["leads", "lead", "contacts", "cpl", "cost per lead", "form leads"], "attribute_keys": ["totalLeads", "costPerLead", "welcomeEmailSentCount"], "display_metrics": ["Total Leads Generated", "Cost Per Lead (CPL)", "Email Nurture Status"], }, } class UserKnowledgeCacheManager: """ 3-Layer Assets Knowledge Graph Cache Manager: - self._user_cache: In-memory RAM cache indexed by "tenant_id:user_id". - Disk: Assets Knowledge///cache.json. - DB: Supabase PostgreSQL tables (user_tenant_roles & connections). """ def __init__(self): self._user_cache: dict[str, dict[str, Any]] = {} def _get_cache_key(self, tenant_id: str, user_id: str) -> str: t_id = tenant_id or "default_tenant" u_id = user_id or "default_user" return f"{t_id}:{u_id}" def get_cache_file_path(self, tenant_id: str, user_id: str) -> str: t_id = tenant_id or "default_tenant" u_id = user_id or "default_user" user_dir = os.path.join(BASE_KNOWLEDGE_DIR, t_id, u_id) os.makedirs(user_dir, exist_ok=True) return os.path.join(user_dir, "cache.json") def load_user_assets(self, tenant_id: str, user_id: str, force_refresh: bool = False) -> dict[str, Any]: """ 3-Layer Knowledge Graph Resolution: Step 1 (RAM Check): If key in self._user_cache and not force_refresh -> Return in 0ms! Step 2 (Disk Check): If cache.json exists on disk and not force_refresh -> Load into RAM and Return in 0ms! Step 3 (DB Query & Dynamic Build): Fetch connections from DB, enrich in real-time, populate RAM & write to Disk. """ cache_key = self._get_cache_key(tenant_id, user_id) file_path = self.get_cache_file_path(tenant_id, user_id) if not force_refresh: # Step 1: In-Memory RAM Check (0ms) if cache_key in self._user_cache: print(f"- KnowledgeCache: [Layer 1 RAM Hit] Loaded memory cache for '{cache_key}' (0ms)") return self._user_cache[cache_key] # Step 2: Disk File Check (0ms) if os.path.exists(file_path): try: with open(file_path, "r", encoding="utf-8") as f: data = json.load(f) self._user_cache[cache_key] = data print(f"- KnowledgeCache: [Layer 2 Disk Hit] Loaded cache.json into RAM for '{cache_key}' (0ms)") return data except Exception as e: print(f"- KnowledgeCache: Warning loading cache.json from disk: {e}") # Step 3: DB Query & 100% Real-Time Dynamic Build return self._build_and_save_cache(tenant_id, user_id) def _build_and_save_cache(self, tenant_id: str, user_id: str) -> dict[str, Any]: """Fetch active connections from DB, enrich metadata in 100% real-time, and save to RAM + Disk.""" cache_key = self._get_cache_key(tenant_id, user_id) file_path = self.get_cache_file_path(tenant_id, user_id) has_access = True active_connections = {} if SUPABASE_KEY and SUPABASE_URL and tenant_id: headers = { "apikey": SUPABASE_KEY, "Authorization": f"Bearer {SUPABASE_KEY}", "Content-Type": "application/json", } # Permission check - supports both snake_case and camelCase if user_id: try: role_url = f"{SUPABASE_URL}/rest/v1/user_tenant_roles" res = httpx.get(role_url, headers=headers, params={"user_id": f"eq.{user_id}", "tenant_id": f"eq.{tenant_id}", "select": "id"}, timeout=3.0) if res.status_code != 200 or len(res.json()) == 0: res2 = httpx.get(role_url, headers=headers, params={"userId": f"eq.{user_id}", "tenantId": f"eq.{tenant_id}", "select": "id"}, timeout=3.0) if res2.status_code == 200 and len(res2.json()) == 0: has_access = False except Exception as e: print(f"- KnowledgeCache: Permission lookup error: {e}") # Connection lookup - supports both snake_case and camelCase try: conn_url = f"{SUPABASE_URL}/rest/v1/connections" rows = [] res1 = httpx.get(conn_url, headers=headers, params={"tenant_id": f"eq.{tenant_id}", "select": "*"}, timeout=3.0) if res1.status_code == 200: rows.extend(res1.json()) res2 = httpx.get(conn_url, headers=headers, params={"tenantId": f"eq.{tenant_id}", "select": "*"}, timeout=3.0) if res2.status_code == 200: rows.extend(res2.json()) for r in rows: p = r.get("provider") st = str(r.get("status", "")).lower() if p and st in ["connected", "active"]: active_connections[p] = r except Exception as e: print(f"- KnowledgeCache: Connections lookup error: {e}") # Construct Knowledge Graph dynamically from live DB connections knowledge_graph = [] for provider_id, conn_data in active_connections.items(): meta = conn_data.get("metadata") or {} def_meta = DEFAULT_PROVIDER_METADATA.get(provider_id, { "name": provider_id.title(), "category": "general", "synonyms": [provider_id], "attribute_keys": list(meta.keys()) if meta else ["status"], "display_metrics": [f"{provider_id.title()} Performance Metrics"], }) item = { "id": conn_data.get("id", f"conn-{provider_id}"), "provider": provider_id, "name": meta.get("name") or def_meta.get("name"), "category": def_meta.get("category", "general"), "status": "connected", "attribute_keys": meta.get("metrics") or def_meta.get("attribute_keys"), "display_metrics": def_meta.get("display_metrics"), "synonyms": def_meta.get("synonyms"), "metadata": { "propertyId": meta.get("propertyId") or meta.get("property_id") or conn_data.get("externalId"), "shop": meta.get("shop") or meta.get("myshopifyDomain"), "adAccountId": conn_data.get("externalId") or conn_data.get("external_id"), }, } knowledge_graph.append(item) # Fallback to standard platform templates if tenant has no active DB connections yet if not knowledge_graph: for p_id, def_meta in DEFAULT_PROVIDER_METADATA.items(): knowledge_graph.append({ "id": f"conn-{p_id}", "provider": p_id, "name": def_meta["name"], "category": def_meta["category"], "status": "disconnected", "attribute_keys": def_meta["attribute_keys"], "display_metrics": def_meta["display_metrics"], "synonyms": def_meta["synonyms"], }) cache_data = { "tenant_id": tenant_id, "user_id": user_id, "user_authorized": has_access, "total_platforms": len(knowledge_graph), "timestamp_epoch": time.time(), "cached_at": time.strftime("%Y-%m-%d %H:%M:%S"), "knowledge_graph": knowledge_graph, } # 1. Update In-Memory RAM self._user_cache[cache_key] = cache_data # 2. Write to Disk cache.json try: with open(file_path, "w", encoding="utf-8") as f: json.dump(cache_data, f, indent=2) print(f"- KnowledgeCache: [Layer 3 DB Build] Built Knowledge Graph and saved cache.json for '{cache_key}'") except Exception as e: print(f"- KnowledgeCache: Error writing cache.json to disk: {e}") return cache_data def invalidate_user_cache(self, tenant_id: str, user_id: str): """Force refresh / Manual sync: Evicts RAM cache and deletes disk cache.json.""" cache_key = self._get_cache_key(tenant_id, user_id) file_path = self.get_cache_file_path(tenant_id, user_id) if cache_key in self._user_cache: del self._user_cache[cache_key] if os.path.exists(file_path): try: os.remove(file_path) print(f"- KnowledgeCache: [Force Refresh] Evicted cache.json from RAM and disk for '{cache_key}'") except Exception as e: print(f"- KnowledgeCache: Error deleting cache file: {e}") knowledge_cache_manager = UserKnowledgeCacheManager() def load_user_assets(tenant_id: str, user_id: str, force_refresh: bool = False) -> dict[str, Any]: """Helper function to load user Knowledge Graph cache using 3-layer manager.""" return knowledge_cache_manager.load_user_assets(tenant_id, user_id, force_refresh=force_refresh) def invalidate_user_cache(tenant_id: str, user_id: str): """Helper function to invalidate user cache.""" knowledge_cache_manager.invalidate_user_cache(tenant_id, user_id)