import psycopg2 from psycopg2 import sql from psycopg2.extras import Json import json from app.core.config import settings import logging import time from typing import List import uuid # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') logger = logging.getLogger("database") AIVORA_PRODUCT_NAME = "Aivora" SUBSCRIPTION_ELIGIBILITY_SQL = ''' ( s."isActive" = TRUE OR LOWER(COALESCE(s."status", '')) IN ('active', 'trial', 'trialing') OR ( s."trialStart" IS NOT NULL AND s."trialEnd" IS NOT NULL AND NOW() >= s."trialStart" AND NOW() <= s."trialEnd" ) ) ''' def _plan_has_product(product_features, product_name: str) -> bool: if not isinstance(product_features, list): return False return any( isinstance(product, dict) and product.get("productName") == product_name and product.get("features") for product in product_features ) def get_db_connection(retries=3, delay=2): """ Establishes a database connection with retry logic. """ for attempt in range(retries): try: conn = psycopg2.connect( host=settings.DB_HOST, port=settings.DB_PORT, user=settings.DB_USER, password=settings.DB_PASSWORD, dbname=settings.DB_NAME, connect_timeout=10 ) return conn except psycopg2.OperationalError as e: logger.warning(f"Database connection attempt {attempt + 1} failed: {e}. Retrying in {delay}s...") if attempt < retries - 1: time.sleep(delay) else: logger.error("Max retries reached. Could not connect to the database.", exc_info=True) raise e except Exception as e: logger.error(f"Unexpected error connecting to database: {e}", exc_info=True) raise e def get_master_db_connection(retries=3, delay=2): """ Establishes a connection to the master database with retry logic. """ for attempt in range(retries): try: conn = psycopg2.connect( host=settings.DB_HOST, port=settings.DB_PORT, user=settings.DB_USER, password=settings.DB_PASSWORD, dbname=settings.MASTER_DB_NAME, connect_timeout=10 ) return conn except psycopg2.OperationalError as e: logger.warning(f"Master DB connection attempt {attempt + 1} failed: {e}. Retrying in {delay}s...") if attempt < retries - 1: time.sleep(delay) else: logger.error("Max retries reached. Could not connect to the master database.", exc_info=True) raise e except Exception as e: logger.error(f"Unexpected error connecting to master database: {e}", exc_info=True) raise e def bootstrap_tenant(tenant_id: str): """ Creates the schema for the tenant if it doesn't exist and creates the relevant tables within that schema. """ # Basic validation for schema name if not tenant_id.replace('_', '').replace('-', '').isalnum(): logger.error(f"Invalid tenant_id format: {tenant_id}") raise ValueError("Invalid tenant_id format. Must be alphanumeric, underscore, or hyphen.") logger.info(f"Bootstrapping tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: # Create extension cur.execute("CREATE EXTENSION IF NOT EXISTS vector") # Create schema cur.execute(sql.SQL("CREATE SCHEMA IF NOT EXISTS {}").format(sql.Identifier(tenant_id))) # Create strategist_knowledge_base table cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_knowledge_base ( id SERIAL PRIMARY KEY, content TEXT NOT NULL, embedding vector(3072) NOT NULL, metadata JSONB, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """).format(sql.Identifier(tenant_id))) # Migration: Ensure created_at exists and has default cur.execute(sql.SQL("ALTER TABLE {}.strategist_knowledge_base ADD COLUMN IF NOT EXISTS created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_knowledge_base ALTER COLUMN created_at SET DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("UPDATE {}.strategist_knowledge_base SET created_at = CURRENT_TIMESTAMP WHERE created_at IS NULL").format(sql.Identifier(tenant_id))) # Create strategist_summaries table cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_summaries ( id SERIAL PRIMARY KEY, ingestion_type TEXT NOT NULL, summary TEXT NOT NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """).format(sql.Identifier(tenant_id))) # Migration: Ensure created_at exists and has default cur.execute(sql.SQL("ALTER TABLE {}.strategist_summaries ADD COLUMN IF NOT EXISTS created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_summaries ALTER COLUMN created_at SET DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("UPDATE {}.strategist_summaries SET created_at = CURRENT_TIMESTAMP WHERE created_at IS NULL").format(sql.Identifier(tenant_id))) # Create strategist_settings table. # Settings are not summaries: one row per key, updated in place, so a # save cannot lose a concurrent save and the table does not grow # without bound. See app/services/tools_settings.py. cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_settings ( key TEXT PRIMARY KEY, value JSONB NOT NULL, updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """).format(sql.Identifier(tenant_id))) # Products extracted from the tenant's website, used by the # recommendation tool and the proactive endpoint. No embedding # column: retrieval reuses the knowledge base, which already has # this text embedded. related_keys holds each product's nearest # neighbours, computed once at ingestion so the proactive endpoint # is a lookup rather than a similarity search. raw is a reserve, # not part of the frontend contract (to_card never returns it): # the page's JSON-LD Product object verbatim when it had one, # otherwise its og:/product: meta tags -- fields extraction does # not surface today, kept so adding one later is a read rather # than a re-crawl of every tenant's site. cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_products ( id SERIAL PRIMARY KEY, product_key TEXT NOT NULL UNIQUE, name TEXT NOT NULL, description TEXT, image_url TEXT, -- Nullable: a source that exposes no product URLs at all -- would otherwise have every product rejected at -- validation, leaving the merchant with an empty catalog -- and no explanation. See migrate_products_table below. product_url TEXT, category TEXT, ctas JSONB NOT NULL DEFAULT '[]', options JSONB NOT NULL DEFAULT '[]', raw JSONB NOT NULL DEFAULT '{{}}', related_keys TEXT[] NOT NULL DEFAULT '{{}}', extracted_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, source_kind TEXT NOT NULL DEFAULT 'crawl', source_ref TEXT, external_id TEXT, brand TEXT, taxonomy_path TEXT[] NOT NULL DEFAULT '{{}}', taxonomy_source TEXT, raw_category TEXT, price_cents INT, price_max_cents INT, compare_at_cents INT, currency TEXT, on_sale BOOLEAN NOT NULL DEFAULT false, in_stock BOOLEAN NOT NULL DEFAULT true, status TEXT, attributes JSONB NOT NULL DEFAULT '[]', quality_score REAL, content_hash TEXT, record_hash TEXT, synced_at TIMESTAMP WITH TIME ZONE, -- Recommendation matching spec columns. tenant_relations is -- keyed by tenant so cross-tenant recs never collide; -- missing_fields flags an incomplete product for exclusion -- from serving without overloading status, which carries -- the source's own ACTIVE/DRAFT/ARCHIVED value. tenant_relations JSONB NOT NULL DEFAULT '{{}}', rating REAL, review_count INT, featured_rank INT, missing_fields TEXT[] NOT NULL DEFAULT '{{}}', price_reference_cents INT, fx_rate_used REAL, -- Attribute extraction (Phase 1). Nullable: null means -- "not yet extracted", not "extracted, and false". is_accessory BOOLEAN, price_tier TEXT, enriched_hash TEXT ) """).format(sql.Identifier(tenant_id))) # CREATE TABLE IF NOT EXISTS above won't add new columns to a # tenant's pre-existing table -- ensure them before the index # below needs source_kind/source_ref. for _col_name, _col_ddl in _PRODUCT_COLUMNS: cur.execute(sql.SQL( "ALTER TABLE {}.strategist_products ADD COLUMN IF NOT EXISTS {} " ).format(sql.Identifier(tenant_id), sql.Identifier(_col_name)) + sql.SQL(_col_ddl)) cur.execute(sql.SQL( "CREATE INDEX IF NOT EXISTS strategist_products_url_idx " "ON {}.strategist_products (product_url)" ).format(sql.Identifier(tenant_id))) cur.execute(sql.SQL( "CREATE INDEX IF NOT EXISTS strategist_products_category_idx " "ON {}.strategist_products (category)" ).format(sql.Identifier(tenant_id))) cur.execute(sql.SQL( "CREATE INDEX IF NOT EXISTS strategist_products_source_idx " "ON {}.strategist_products (source_kind, source_ref)" ).format(sql.Identifier(tenant_id))) cur.execute(sql.SQL( "CREATE INDEX IF NOT EXISTS strategist_products_status_idx " "ON {}.strategist_products (status, in_stock)" ).format(sql.Identifier(tenant_id))) # Pairing graph tables (Phase 2): candidate neighbor pairs, the # merchant's approve/reject decisions on them, and the embedding # cache that generates candidates. Kept off strategist_products so # a 3072-float vector never rides along on a catalogue read that # doesn't want it. for _table_name, _ddl in _PAIRING_TABLES: cur.execute(sql.SQL(_ddl).format(sql.Identifier(tenant_id))) cur.execute(sql.SQL( "CREATE INDEX IF NOT EXISTS strategist_neighbors_anchor_idx " "ON {}.strategist_product_neighbors (anchor_key, pair_type)" ).format(sql.Identifier(tenant_id))) # Create strategist_tickets table cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_tickets ( id SERIAL PRIMARY KEY, ticket_id TEXT UNIQUE NOT NULL, thread_id TEXT, user_name TEXT NOT NULL, heading TEXT NOT NULL, content TEXT NOT NULL, priority TEXT NOT NULL, category TEXT, status TEXT DEFAULT 'Open', email TEXT, contact_no TEXT, contact_medium TEXT, contact_details TEXT, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """).format(sql.Identifier(tenant_id))) # Migration: Ensure created_at exists and has default cur.execute(sql.SQL("ALTER TABLE {}.strategist_tickets ADD COLUMN IF NOT EXISTS created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_tickets ALTER COLUMN created_at SET DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("UPDATE {}.strategist_tickets SET created_at = CURRENT_TIMESTAMP WHERE created_at IS NULL").format(sql.Identifier(tenant_id))) # Migration: Ensure thread_id column exists cur.execute(sql.SQL("ALTER TABLE {}.strategist_tickets ADD COLUMN IF NOT EXISTS thread_id TEXT").format(sql.Identifier(tenant_id))) # Migration: Ensure category column exists cur.execute(sql.SQL("ALTER TABLE {}.strategist_tickets ADD COLUMN IF NOT EXISTS category TEXT").format(sql.Identifier(tenant_id))) # Create strategist_ticket_comments table cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_ticket_comments ( id SERIAL PRIMARY KEY, ticket_id TEXT NOT NULL REFERENCES {}.strategist_tickets(ticket_id) ON DELETE CASCADE, admin_name TEXT, comment TEXT NOT NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """).format(sql.Identifier(tenant_id), sql.Identifier(tenant_id))) # Migration: Ensure created_at exists and has default cur.execute(sql.SQL("ALTER TABLE {}.strategist_ticket_comments ADD COLUMN IF NOT EXISTS created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_ticket_comments ALTER COLUMN created_at SET DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) # Create strategist_feedback table cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_feedback ( id SERIAL PRIMARY KEY, thread_id TEXT NOT NULL, user_name TEXT, question TEXT NOT NULL, answer TEXT NOT NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """).format(sql.Identifier(tenant_id))) # Migration: Ensure metadata column exists cur.execute(sql.SQL("ALTER TABLE {}.strategist_feedback ADD COLUMN IF NOT EXISTS metadata JSONB DEFAULT '{{}}'").format(sql.Identifier(tenant_id))) # Create strategist_llm_usage table cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_llm_usage ( id SERIAL PRIMARY KEY, feature_name TEXT NOT NULL, model_name TEXT NOT NULL, prompt_tokens INTEGER NOT NULL, completion_tokens INTEGER NOT NULL, total_tokens INTEGER NOT NULL, thread_id TEXT, cumulative_prompt_tokens INTEGER DEFAULT 0, cumulative_completion_tokens INTEGER DEFAULT 0, metadata JSONB DEFAULT '{{}}', created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """).format(sql.Identifier(tenant_id))) # Migration: Ensure new usage columns exist cur.execute(sql.SQL("ALTER TABLE {}.strategist_llm_usage ADD COLUMN IF NOT EXISTS thread_id TEXT").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_llm_usage ADD COLUMN IF NOT EXISTS cumulative_prompt_tokens INTEGER DEFAULT 0").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_llm_usage ADD COLUMN IF NOT EXISTS cumulative_completion_tokens INTEGER DEFAULT 0").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_llm_usage ADD COLUMN IF NOT EXISTS metadata JSONB DEFAULT '{{}}'").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_llm_usage ADD COLUMN IF NOT EXISTS created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_llm_usage ALTER COLUMN created_at SET DEFAULT CURRENT_TIMESTAMP").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("UPDATE {}.strategist_llm_usage SET created_at = CURRENT_TIMESTAMP WHERE created_at IS NULL").format(sql.Identifier(tenant_id))) # Create strategist_thread_analytics table cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_thread_analytics ( thread_id TEXT PRIMARY KEY, sentiment_score FLOAT, intent TEXT, is_high_intent BOOLEAN, is_lead_qualified BOOLEAN, is_resolved BOOLEAN, escalation_needed BOOLEAN, positive_points TEXT, key_concerns TEXT, pain_point TEXT, feature_request TEXT, objection TEXT, cta_clicked BOOLEAN, summary TEXT, metadata JSONB, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, last_analyzed_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """).format(sql.Identifier(tenant_id))) # Migration: Ensure new analytics columns exist for existing tenants cur.execute(sql.SQL("ALTER TABLE {}.strategist_thread_analytics ADD COLUMN IF NOT EXISTS positive_points TEXT").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("ALTER TABLE {}.strategist_thread_analytics ADD COLUMN IF NOT EXISTS key_concerns TEXT").format(sql.Identifier(tenant_id))) # Create strategist_crawled_urls table cur.execute(sql.SQL(""" CREATE TABLE IF NOT EXISTS {}.strategist_crawled_urls ( id SERIAL PRIMARY KEY, source_id TEXT NOT NULL, url TEXT NOT NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """).format(sql.Identifier(tenant_id))) # Analytics Performance Indices cur.execute(sql.SQL("CREATE INDEX IF NOT EXISTS idx_usage_created_at ON {}.strategist_llm_usage(created_at)").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("CREATE INDEX IF NOT EXISTS idx_feedback_created_at ON {}.strategist_feedback(created_at)").format(sql.Identifier(tenant_id))) cur.execute(sql.SQL("CREATE INDEX IF NOT EXISTS idx_usage_thread ON {}.strategist_llm_usage(thread_id)").format(sql.Identifier(tenant_id))) ensure_thread_id_column(tenant_id, cur) conn.commit() logger.info(f"Successfully bootstrapped tenant: {tenant_id}") # Brand Persona Integration try: persona = fetch_brand_persona(tenant_id) if persona: logger.info(f"Brand persona found for {tenant_id}. Storing in strategist_summaries.") # Use a cursor from a new connection or handle existing connection carefully # Since we are inside bootstrap_tenant, we can use the existing conn if it's still open # but wait, conn is closed in finally. Let's do it before finally. with conn.cursor() as persona_cur: insert_query = sql.SQL(""" INSERT INTO {}.strategist_summaries (ingestion_type, summary) VALUES ('brand_persona', %s) """).format(sql.Identifier(tenant_id)) persona_cur.execute(insert_query, (persona,)) conn.commit() except Exception as pe: logger.warning(f"Persona aggregation failed during bootstrap for {tenant_id}: {pe}") return True except Exception as e: conn.rollback() logger.error(f"Error bootstrapping tenant {tenant_id}: {e}", exc_info=True) raise e finally: conn.close() def fetch_brand_persona(tenant_id: str): """ Fetches the latest brand analysis result and extracts tone and voice traits. Returns a string summarizing the brand persona. """ conn = get_db_connection() try: with conn.cursor() as cur: # Check if table exists cur.execute(""" SELECT EXISTS ( SELECT FROM information_schema.tables WHERE table_schema = %s AND table_name = 'brandforge_analyses' ) """, (tenant_id,)) if not cur.fetchone()[0]: return None # Fetch latest full_result query = sql.SQL(""" SELECT full_result FROM {}.brandforge_analyses ORDER BY created_at DESC LIMIT 1 """).format(sql.Identifier(tenant_id)) cur.execute(query) row = cur.fetchone() if not row or not row[0]: return None full_result = row[0] tone_list = full_result.get('tone', []) voice_list = full_result.get('voice', []) extracted_tone = [t.get('text') for t in tone_list if t.get('text')] extracted_voice = [v.get('text') for v in voice_list if v.get('text')] if not extracted_tone and not extracted_voice: return None persona_parts = [] if extracted_tone: persona_parts.append(f"Tone: {', '.join(extracted_tone)}") if extracted_voice: persona_parts.append(f"Voice: {', '.join(extracted_voice)}") return " | ".join(persona_parts) except Exception as e: logger.error(f"Error fetching brand persona for {tenant_id}: {e}") return None finally: conn.close() def ensure_thread_id_column(tenant_id: str, cur): """ Internal helper to add the thread_id column to the tickets table if missing. Used for self-healing legacy tenants. """ try: cur.execute(f"SELECT column_name FROM information_schema.columns WHERE table_schema = %s AND table_name = 'strategist_tickets' AND column_name = 'thread_id'", (tenant_id,)) if not cur.fetchone(): logger.info(f"Self-healing: Adding thread_id column to {tenant_id}.strategist_tickets") cur.execute(sql.SQL("ALTER TABLE {}.strategist_tickets ADD COLUMN thread_id TEXT").format(sql.Identifier(tenant_id))) except Exception as e: logger.warning(f"Failed to ensure thread_id column for {tenant_id}: {e}") def ensure_category_column(tenant_id: str, cur): """ Internal helper to add the category column to the tickets table if missing. Used for self-healing legacy tenants. """ try: cur.execute(f"SELECT column_name FROM information_schema.columns WHERE table_schema = %s AND table_name = 'strategist_tickets' AND column_name = 'category'", (tenant_id,)) if not cur.fetchone(): logger.info(f"Self-healing: Adding category column to {tenant_id}.strategist_tickets") cur.execute(sql.SQL("ALTER TABLE {}.strategist_tickets ADD COLUMN category TEXT").format(sql.Identifier(tenant_id))) except Exception as e: logger.warning(f"Failed to ensure category column for {tenant_id}: {e}") def insert_vector_data(tenant_id: str, content: str, embedding: list, metadata: dict): logger.info(f"Inserting vector data for tenant: {tenant_id}") # Quota Enforcement: Vector DB Space from app.services.infra.quotas import check_quota user_id = metadata.get("user_id") if isinstance(metadata, dict) else None metadata_size = len(json.dumps(metadata or {}).encode("utf-8")) content_size = len((content or "").encode("utf-8")) embedding_size = len(embedding or []) * 8 projected_row_size = max(1, content_size + metadata_size + embedding_size) check_quota( tenant_id, "storage_vector_capacity", user_id=user_id, requested_amount=projected_row_size, ) conn = get_db_connection() try: with conn.cursor() as cur: insert_query = sql.SQL(""" INSERT INTO {}.strategist_knowledge_base (content, embedding, metadata) VALUES (%s, %s, %s) """).format(sql.Identifier(tenant_id)) cur.execute(insert_query, (content, embedding, json.dumps(metadata))) conn.commit() # Historical Storage Tracking try: from app.services.integrations.firestore import log_storage_snapshot new_size = get_vector_db_size(tenant_id) log_storage_snapshot(tenant_id, new_size) except Exception as se: logger.warning(f"Failed to log storage snapshot for {tenant_id}: {se}") return True except Exception as e: conn.rollback() logger.error(f"Failed to insert vector data for tenant {tenant_id}: {e}", exc_info=True) raise e finally: conn.close() def insert_summary(tenant_id: str, ingestion_type: str, summary: str): logger.info(f"Inserting summary for tenant: {tenant_id}, type: {ingestion_type}") conn = get_db_connection() try: with conn.cursor() as cur: insert_query = sql.SQL(""" INSERT INTO {}.strategist_summaries (ingestion_type, summary) VALUES (%s, %s) """).format(sql.Identifier(tenant_id)) cur.execute(insert_query, (ingestion_type, summary)) conn.commit() return True except Exception as e: conn.rollback() logger.error(f"Failed to insert summary for tenant {tenant_id}: {e}", exc_info=True) raise e finally: conn.close() def insert_crawled_url(tenant_id: str, source_id: str, url: str): logger.info(f"Inserting crawled URL for tenant: {tenant_id}, source_id: {source_id}") conn = get_db_connection() try: with conn.cursor() as cur: insert_query = sql.SQL(""" INSERT INTO {}.strategist_crawled_urls (source_id, url) VALUES (%s, %s) """).format(sql.Identifier(tenant_id)) cur.execute(insert_query, (source_id, url)) conn.commit() return True except Exception as e: conn.rollback() logger.error(f"Failed to insert crawled URL for tenant {tenant_id}: {e}", exc_info=True) return False finally: conn.close() def search_vector_data(tenant_id: str, query_embedding: List[float], limit: int = 5): """ Performs cosine similarity search on the tenant's vector database. """ logger.info(f"Searching vectors for tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: # First check if the schema exists cur.execute(""" SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s """, (tenant_id,)) if not cur.fetchone(): logger.warning(f"Tenant schema {tenant_id} does not exist for search.") return [] query = sql.SQL(""" SELECT content, metadata, (embedding <=> CAST(%s AS vector)) as distance FROM {}.strategist_knowledge_base ORDER BY distance ASC LIMIT %s """).format(sql.Identifier(tenant_id)) # Convert list to string format '[v1,v2,...]' for robustness with pgvector vector_str = "[" + ",".join(map(str, query_embedding)) + "]" cur.execute(query, (vector_str, limit)) results = cur.fetchall() # Format results search_hits = [] for row in results: search_hits.append({ "content": row[0], "metadata": row[1], "score": 1 - float(row[2]) # Convert distance to similarity score }) return search_hits except Exception as e: logger.error(f"Failed to search vectors for tenant {tenant_id}: {e}", exc_info=True) raise e finally: conn.close() def settings_table_exists(tenant_id: str) -> bool: """Whether this tenant has been bootstrapped since strategist_settings landed.""" conn = get_db_connection() try: with conn.cursor() as cur: cur.execute( "SELECT EXISTS (SELECT FROM information_schema.tables " "WHERE table_schema = %s AND table_name = 'strategist_settings')", (tenant_id,), ) return cur.fetchone()[0] except Exception as e: logger.error(f"Could not check the settings table for {tenant_id}: {e}") return False finally: conn.close() def get_setting(tenant_id: str, key: str): """The stored value for a settings key, or None if there is none.""" conn = get_db_connection() try: with conn.cursor() as cur: cur.execute( "SELECT EXISTS (SELECT FROM information_schema.tables " "WHERE table_schema = %s AND table_name = 'strategist_settings')", (tenant_id,), ) if not cur.fetchone()[0]: return None cur.execute( sql.SQL("SELECT value FROM {}.strategist_settings WHERE key = %s") .format(sql.Identifier(tenant_id)), (key,)) row = cur.fetchone() return row[0] if row else None except Exception as e: logger.error(f"Could not read setting '{key}' for {tenant_id}: {e}") return None finally: conn.close() def update_setting(tenant_id: str, key: str, merge): """Read-modify-write a settings key under a row lock. `merge` receives the currently stored value (or None) and returns the value to store. It runs inside the transaction that holds the lock, so two concurrent saves cannot each read the old value and overwrite each other -- the second waits, then merges on top of the first one's result. The empty INSERT first is what makes the lock possible: SELECT FOR UPDATE locks nothing when the row does not exist yet. """ conn = get_db_connection() try: with conn.cursor() as cur: cur.execute( sql.SQL("INSERT INTO {}.strategist_settings (key, value) VALUES (%s, %s) " "ON CONFLICT (key) DO NOTHING") .format(sql.Identifier(tenant_id)), (key, Json({}))) cur.execute( sql.SQL("SELECT value FROM {}.strategist_settings WHERE key = %s FOR UPDATE") .format(sql.Identifier(tenant_id)), (key,)) row = cur.fetchone() current = row[0] if row and row[0] else None merged = merge(current) cur.execute( sql.SQL("UPDATE {}.strategist_settings SET value = %s, " "updated_at = CURRENT_TIMESTAMP WHERE key = %s") .format(sql.Identifier(tenant_id)), (Json(merged), key)) conn.commit() return merged except Exception: conn.rollback() logger.error(f"Could not save setting '{key}' for {tenant_id}", exc_info=True) raise finally: conn.close() def get_latest_summary(tenant_id: str, ingestion_type: str): logger.info(f"Fetching latest summary for tenant: {tenant_id}, type: {ingestion_type}") conn = get_db_connection() try: with conn.cursor() as cur: # First check if the schema exists to avoid relation missing errors cur.execute(""" SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s """, (tenant_id,)) if not cur.fetchone(): logger.warning(f"Tenant schema {tenant_id} does not exist.") return None query = sql.SQL(""" SELECT summary, created_at FROM {}.strategist_summaries WHERE ingestion_type = %s ORDER BY created_at DESC LIMIT 1 """).format(sql.Identifier(tenant_id)) cur.execute(query, (ingestion_type,)) result = cur.fetchone() return (result[0], result[1]) if result else (None, None) except Exception as e: logger.error(f"Error fetching summary: {e}", exc_info=True) return None, None finally: conn.close() def create_ticket(tenant_id: str, ticket_data: dict, thread_id: str = None): """ Inserts a new support ticket into the tenant-specific table. Self-healing: Automatically adds thread_id column if missing. """ logger.info(f"Creating ticket for tenant: {tenant_id}, thread: {thread_id}") # Quota Enforcement: Tickets from app.services.infra.quotas import check_quota, QuotaExceededError user_id = ticket_data.get("user_id") if isinstance(ticket_data, dict) else None check_quota(tenant_id, "total_tickets", user_id=user_id) conn = get_db_connection() try: with conn.cursor() as cur: query = sql.SQL(""" INSERT INTO {}.strategist_tickets (ticket_id, user_name, heading, content, priority, status, email, contact_no, contact_medium, thread_id, category) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id """).format(sql.Identifier(tenant_id)) try: cur.execute(query, ( ticket_data['ticket_id'], ticket_data['user_name'], ticket_data['heading'], ticket_data['content'], ticket_data['priority'], 'Open', ticket_data.get('email'), ticket_data.get('contact_no') or ticket_data.get('phone_number'), ticket_data.get('contact_medium') or ticket_data.get('any_other_contact_medium'), thread_id, ticket_data.get('category') or 'General enquiries' )) except Exception as e: if "column" in str(e).lower() and ("thread_id" in str(e).lower() or "category" in str(e).lower()): logger.warning(f"Self-healing: Adding missing column(s) to {tenant_id} during creation.") conn.rollback() with conn.cursor() as repair_cur: ensure_thread_id_column(tenant_id, repair_cur) ensure_category_column(tenant_id, repair_cur) conn.commit() # Retry cur.execute(query, ( ticket_data['ticket_id'], ticket_data['user_name'], ticket_data['heading'], ticket_data['content'], ticket_data['priority'], 'Open', ticket_data.get('email'), ticket_data.get('contact_no') or ticket_data.get('phone_number'), ticket_data.get('contact_medium') or ticket_data.get('any_other_contact_medium'), thread_id, ticket_data.get('category') or 'General enquiries' )) else: raise e new_id = cur.fetchone()[0] conn.commit() return new_id except QuotaExceededError: # Re-raise quota errors so the chat/API can handle them specifically raise except Exception as e: if conn: conn.rollback() logger.error(f"Error creating ticket for tenant {tenant_id}: {e}", exc_info=True) return None finally: conn.close() def get_ticket_by_thread(tenant_id: str, thread_id: str): """ Finds an existing ticket for a given conversation thread. Self-healing: Automatically adds thread_id column if missing. """ logger.info(f"Getting ticket for thread {thread_id} in tenant {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: query = sql.SQL("SELECT id, ticket_id, heading, content, priority, status, category FROM {}.strategist_tickets WHERE thread_id = %s LIMIT 1").format(sql.Identifier(tenant_id)) try: cur.execute(query, (thread_id,)) except Exception as e: if "column" in str(e).lower() and ("thread_id" in str(e).lower() or "category" in str(e).lower()): logger.warning(f"Self-healing: Adding missing column(s) to {tenant_id} during fetch-by-thread.") conn.rollback() with conn.cursor() as repair_cur: ensure_thread_id_column(tenant_id, repair_cur) ensure_category_column(tenant_id, repair_cur) conn.commit() # Retry cur.execute(query, (thread_id,)) else: raise e row = cur.fetchone() if row: return { "id": row[0], "ticket_id": row[1], "heading": row[2], "content": row[3], "priority": row[4], "status": row[5], "category": row[6] } return None except Exception as e: logger.error(f"Error fetching ticket by thread {thread_id}: {e}", exc_info=True) return None finally: conn.close() def update_ticket(tenant_id: str, ticket_id: str, update_data: dict): """ Updates an existing ticket with new details. """ logger.info(f"Updating ticket {ticket_id} for tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: query = sql.SQL(""" UPDATE {}.strategist_tickets SET heading = %s, content = %s, priority = %s, category = %s WHERE ticket_id = %s """).format(sql.Identifier(tenant_id)) cur.execute(query, ( update_data['heading'], update_data['content'], update_data['priority'], update_data.get('category'), ticket_id )) conn.commit() return True except Exception as e: logger.error(f"Failed to update ticket {ticket_id}: {e}", exc_info=True) conn.rollback() return False finally: conn.close() def update_ticket_status(tenant_id: str, ticket_id: str, status: str): """ Updates the status of an existing ticket. """ logger.info(f"Updating status of ticket {ticket_id} to '{status}' for tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: query = sql.SQL(""" UPDATE {}.strategist_tickets SET status = %s WHERE ticket_id = %s """).format(sql.Identifier(tenant_id)) cur.execute(query, (status, ticket_id)) conn.commit() return cur.rowcount > 0 except Exception as e: logger.error(f"Failed to update status for ticket {ticket_id}: {e}", exc_info=True) conn.rollback() return False finally: conn.close() def get_ticket(tenant_id: str, ticket_id: str): """ Retrieves a single ticket by its ticket_id. """ logger.info(f"Fetching ticket {ticket_id} for tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: # Check schema first cur.execute("SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s", (tenant_id,)) if not cur.fetchone(): return None query = sql.SQL("SELECT id, ticket_id, user_name, heading, content, priority, status, email, contact_no, contact_medium, contact_details, created_at, thread_id, category FROM {}.strategist_tickets WHERE ticket_id = %s").format(sql.Identifier(tenant_id)) try: cur.execute(query, (ticket_id,)) except Exception as e: if "column" in str(e).lower() and ("thread_id" in str(e).lower() or "category" in str(e).lower()): logger.warning(f"Self-healing: Adding missing column(s) to {tenant_id} during ticket fetch.") conn.rollback() with conn.cursor() as repair_cur: ensure_thread_id_column(tenant_id, repair_cur) ensure_category_column(tenant_id, repair_cur) conn.commit() # Retry cur.execute(query, (ticket_id,)) else: raise e row = cur.fetchone() if row: return { "id": row[0], "ticket_id": row[1], "user_name": row[2], "heading": row[3], "content": row[4], "priority": row[5], "status": row[6], "email": row[7], "contact_no": row[8], "contact_medium": row[9], "contact_details": row[10], "created_at": row[11].isoformat() if row[11] else None, "thread_id": row[12], "category": row[13] } return None except Exception as e: logger.error(f"Error fetching ticket {ticket_id} for tenant {tenant_id}: {e}", exc_info=True) return None finally: conn.close() def get_tickets(tenant_id: str): """ Retrieves all tickets for a specific tenant. Self-healing: If thread_id is missing (legacy tenant), it applies the migration and retries. """ logger.info(f"Fetching all tickets for tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: # Check schema first cur.execute("SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s", (tenant_id,)) if not cur.fetchone(): return [] try: query = sql.SQL("SELECT id, ticket_id, user_name, heading, content, priority, status, email, contact_no, contact_medium, contact_details, created_at, thread_id, category FROM {}.strategist_tickets ORDER BY created_at DESC").format(sql.Identifier(tenant_id)) cur.execute(query) except Exception as e: # Catch UndefinedColumn or similar schema-related errors if "column" in str(e).lower() and ("thread_id" in str(e).lower() or "category" in str(e).lower()): logger.warning(f"Detected missing column(s) in {tenant_id}. Applying migration...") conn.rollback() # Reset failed transaction with conn.cursor() as repair_cur: ensure_thread_id_column(tenant_id, repair_cur) ensure_category_column(tenant_id, repair_cur) conn.commit() # Retry the query cur.execute(query) else: raise e rows = cur.fetchall() tickets = [] for row in rows: tickets.append({ "id": row[0], "ticket_id": row[1], "user_name": row[2], "heading": row[3], "content": row[4], "priority": row[5], "status": row[6], "email": row[7], "contact_no": row[8], "contact_medium": row[9], "contact_details": row[10], "created_at": row[11].isoformat() if row[11] else None, "thread_id": row[12], "category": row[13] }) return tickets except Exception as e: logger.error(f"Error fetching tickets for tenant {tenant_id}: {e}", exc_info=True) return [] finally: conn.close() def add_ticket_comment(tenant_id: str, ticket_id: str, admin_name: str, comment: str): """ Inserts a new comment for a specific ticket. """ logger.info(f"Adding comment to ticket {ticket_id} for tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: query = sql.SQL(""" INSERT INTO {}.strategist_ticket_comments (ticket_id, admin_name, comment) VALUES (%s, %s, %s) RETURNING id, created_at """).format(sql.Identifier(tenant_id)) cur.execute(query, (ticket_id, admin_name, comment)) result = cur.fetchone() conn.commit() return { "id": result[0], "ticket_id": ticket_id, "admin_name": admin_name, "comment": comment, "created_at": result[1].isoformat() if result[1] else None } except Exception as e: logger.error(f"Failed to add comment to ticket {ticket_id} for tenant {tenant_id}: {e}", exc_info=True) conn.rollback() raise e finally: conn.close() def get_ticket_comments(tenant_id: str, ticket_id: str): """ Retrieves all comments for a specific ticket. """ logger.info(f"Fetching comments for ticket {ticket_id} in tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: # Check schema first cur.execute("SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s", (tenant_id,)) if not cur.fetchone(): return [] query = sql.SQL("SELECT id, ticket_id, admin_name, comment, created_at FROM {}.strategist_ticket_comments WHERE ticket_id = %s ORDER BY created_at ASC").format(sql.Identifier(tenant_id)) cur.execute(query, (ticket_id,)) rows = cur.fetchall() comments = [] for row in rows: comments.append({ "id": row[0], "ticket_id": row[1], "admin_name": row[2], "comment": row[3], "created_at": row[4].isoformat() if row[4] else None }) return comments except Exception as e: logger.error(f"Error fetching comments for ticket {ticket_id} in tenant {tenant_id}: {e}", exc_info=True) return [] finally: conn.close() def get_ticket_status_counts(tenant_id: str): """ Returns a dictionary of status counts for a tenant. """ logger.info(f"Fetching ticket status counts for tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: # Check schema first cur.execute("SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s", (tenant_id,)) if not cur.fetchone(): return {} query = sql.SQL("SELECT status, COUNT(*) FROM {}.strategist_tickets GROUP BY status").format(sql.Identifier(tenant_id)) cur.execute(query) rows = cur.fetchall() return {row[0] or "Unknown": row[1] for row in rows} except Exception as e: logger.error(f"Error fetching ticket status counts for tenant {tenant_id}: {e}", exc_info=True) return {} finally: conn.close() def insert_feedback(tenant_id: str, feedback_data: dict): """ Inserts user feedback into the tenant's strategist_feedback table. """ logger.info(f"Inserting feedback for tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: query = sql.SQL(""" INSERT INTO {}.strategist_feedback (thread_id, user_name, question, answer, metadata) VALUES (%s, %s, %s, %s, %s) RETURNING id """).format(sql.Identifier(tenant_id)) cur.execute(query, ( feedback_data['thread_id'], feedback_data.get('user_name'), feedback_data['question'], feedback_data['answer'], json.dumps(feedback_data.get('metadata', {})) )) feedback_db_id = cur.fetchone()[0] conn.commit() return feedback_db_id except Exception as e: logger.error(f"Failed to insert feedback for tenant {tenant_id}: {e}", exc_info=True) conn.rollback() raise e finally: conn.close() def delete_vector_data_by_source(tenant_id: str, source: str): """ Deletes all vector data for a specific tenant and source. """ logger.info(f"Deleting vector data for tenant: {tenant_id}, source: {source}") conn = get_db_connection() try: with conn.cursor() as cur: # Check schema first cur.execute("SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s", (tenant_id,)) if not cur.fetchone(): return 0 # Delete query targeting metadata source_id field delete_query = sql.SQL(""" DELETE FROM {}.strategist_knowledge_base WHERE metadata->>'source_id' = %s """).format(sql.Identifier(tenant_id)) cur.execute(delete_query, (source,)) deleted_count = cur.rowcount # Clean up strategist_crawled_urls delete_crawled_query = sql.SQL(""" DELETE FROM {}.strategist_crawled_urls WHERE source_id = %s """).format(sql.Identifier(tenant_id)) cur.execute(delete_crawled_query, (source,)) conn.commit() logger.info(f"Deleted {deleted_count} rows from strategist_knowledge_base for {tenant_id}, source: {source}") # Historical Storage Tracking if deleted_count > 0: try: from app.services.integrations.firestore import log_storage_snapshot new_size = get_vector_db_size(tenant_id) log_storage_snapshot(tenant_id, new_size) except Exception as se: logger.warning(f"Failed to log storage snapshot for {tenant_id} after deletion: {se}") return deleted_count except Exception as e: logger.error(f"Failed to delete vector data for tenant {tenant_id}: {e}", exc_info=True) conn.rollback() raise e finally: conn.close() def delete_vector_data_by_url_subpath(tenant_id: str, base_url: str): """ Deletes all vector data and crawled URLs for a base URL and its sub-paths. This is used to implement 'replacement' logic during re-analysis. """ logger.info(f"Deleting vector data and crawled URLs for tenant: {tenant_id}, base_url: {base_url}") conn = get_db_connection() try: with conn.cursor() as cur: # Pattern for sub-path matching url_pattern = base_url.rstrip('/') # 1. Delete from strategist_knowledge_base delete_kb_query = sql.SQL(""" DELETE FROM {}.strategist_knowledge_base WHERE metadata->>'ingestion_type' = 'website' AND (metadata->>'source' = %s OR metadata->>'source' LIKE %s) """).format(sql.Identifier(tenant_id)) cur.execute(delete_kb_query, (url_pattern, url_pattern + '/%')) kb_count = cur.rowcount # 2. Delete from strategist_crawled_urls delete_urls_query = sql.SQL(""" DELETE FROM {}.strategist_crawled_urls WHERE (url = %s OR url LIKE %s) """).format(sql.Identifier(tenant_id)) cur.execute(delete_urls_query, (url_pattern, url_pattern + '/%')) conn.commit() logger.info(f"Deleted {kb_count} rows from strategist_knowledge_base for {base_url} and its sub-paths.") return kb_count except Exception as e: if conn: conn.rollback() logger.error(f"Failed to delete by subpath for {base_url}: {e}", exc_info=True) raise e finally: if conn: conn.close() def get_source_ids_by_name(tenant_id: str, source_name: str, ingestion_type: str): """ Returns a list of source_ids matching a specific source name and type. Used for implementing replacement logic. """ conn = get_db_connection() try: with conn.cursor() as cur: # Check schema first cur.execute("SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s", (tenant_id,)) if not cur.fetchone(): return [] query = sql.SQL(""" SELECT DISTINCT metadata->>'source_id' FROM {}.strategist_knowledge_base WHERE metadata->>'source' = %s AND metadata->>'ingestion_type' = %s """).format(sql.Identifier(tenant_id)) cur.execute(query, (source_name, ingestion_type)) return [row[0] for row in cur.fetchall() if row[0]] except Exception as e: logger.error(f"Error fetching source_ids by name for {source_name}: {e}") return [] finally: conn.close() def insert_llm_usage(tenant_id: str, feature_name: str, model_name: str, prompt_tokens: int, completion_tokens: int, total_tokens: int, metadata: dict = None, thread_id: str = None): """ Persists LLM token usage data into the tenant's strategist_llm_usage table. Implements Delta-based tracking to avoid double-counting thread history. """ logger.info(f"[DB] Logging LLM usage for tenant: {tenant_id}, feature: {feature_name}, thread: {thread_id}") # Extract thread_id from metadata if not explicitly provided if not thread_id and metadata: thread_id = metadata.get("thread_id") conn = get_db_connection() try: with conn.cursor() as cur: # 1. Safety check: Ensure the table exists cur.execute(""" SELECT EXISTS ( SELECT FROM information_schema.tables WHERE table_schema = %s AND table_name = 'strategist_llm_usage' ) """, (tenant_id,)) if not cur.fetchone()[0]: logger.warning(f"[DB] Skipping usage log: Table strategist_llm_usage not found in {tenant_id}") return False last_cumulative_prompt = 0 last_cumulative_completion = 0 # 2. Fetch previous cumulative totals for delta calculation # We ONLY do delta tracking if a thread_id is present if thread_id: fetch_query = sql.SQL(""" SELECT cumulative_prompt_tokens, cumulative_completion_tokens FROM {}.strategist_llm_usage WHERE thread_id = %s AND feature_name = %s ORDER BY created_at DESC LIMIT 1 """).format(sql.Identifier(tenant_id)) cur.execute(fetch_query, (thread_id, feature_name)) row = cur.fetchone() if row: last_cumulative_prompt, last_cumulative_completion = row # 3. Calculate Deltas # If current < last, we assume it's a separate non-cumulative interaction or a reset, so delta = current if prompt_tokens < last_cumulative_prompt: delta_prompt = prompt_tokens delta_completion = completion_tokens else: delta_prompt = prompt_tokens - last_cumulative_prompt delta_completion = completion_tokens - last_cumulative_completion delta_total = delta_prompt + delta_completion # Ensure no negative logging due to clock drift or mixed models delta_prompt = max(0, delta_prompt) delta_completion = max(0, delta_completion) delta_total = max(0, delta_total) # 4. Insert Delta and Update Cumulative Reference insert_query = sql.SQL(""" INSERT INTO {}.strategist_llm_usage ( feature_name, model_name, prompt_tokens, completion_tokens, total_tokens, thread_id, cumulative_prompt_tokens, cumulative_completion_tokens, metadata, created_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP) """).format(sql.Identifier(tenant_id)) cur.execute(insert_query, ( feature_name, model_name, delta_prompt, delta_completion, delta_total, thread_id, prompt_tokens, # Store the current absolute total as the new cumulative reference completion_tokens, json.dumps(metadata or {}) )) conn.commit() logger.info(f"[DB] Logged Delta: {delta_total} tokens (Thread: {thread_id}, Feature: {feature_name})") return True except Exception as e: if conn: conn.rollback() logger.error(f"Failed to log LLM usage for tenant {tenant_id}: {e}") return False finally: if conn: conn.close() def get_llm_usage(tenant_id: str, limit: int = 100): logger.info(f"Fetching LLM usage for tenant: {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: query = sql.SQL("SELECT feature_name, model_name, prompt_tokens, completion_tokens, total_tokens, created_at FROM {}.strategist_llm_usage ORDER BY created_at DESC LIMIT %s").format(sql.Identifier(tenant_id)) cur.execute(query, (limit,)) rows = cur.fetchall() return [{ "feature": r[0], "model": r[1], "prompt_tokens": r[2], "completion_tokens": r[3], "total_tokens": r[4], "timestamp": r[5].isoformat() if r[5] else None } for r in rows] except Exception as e: logger.error(f"Error fetching usage for tenant {tenant_id}: {e}") return [] finally: conn.close() def upsert_thread_analytics(tenant_id: str, data: dict): """ Inserts or updates analytics for a specific thread. """ logger.info(f"Upserting thread analytics for thread {data.get('thread_id')} in tenant {tenant_id}") conn = get_db_connection() try: with conn.cursor() as cur: upsert_query = sql.SQL(""" INSERT INTO {}.strategist_thread_analytics ( thread_id, sentiment_score, intent, is_high_intent, is_lead_qualified, is_resolved, escalation_needed, positive_points, key_concerns, pain_point, feature_request, objection, cta_clicked, summary, metadata, last_analyzed_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, CURRENT_TIMESTAMP) ON CONFLICT (thread_id) DO UPDATE SET sentiment_score = EXCLUDED.sentiment_score, intent = EXCLUDED.intent, is_high_intent = EXCLUDED.is_high_intent, is_lead_qualified = EXCLUDED.is_lead_qualified, is_resolved = EXCLUDED.is_resolved, escalation_needed = EXCLUDED.escalation_needed, positive_points = EXCLUDED.positive_points, key_concerns = EXCLUDED.key_concerns, pain_point = EXCLUDED.pain_point, feature_request = EXCLUDED.feature_request, objection = EXCLUDED.objection, cta_clicked = EXCLUDED.cta_clicked, summary = EXCLUDED.summary, metadata = EXCLUDED.metadata, last_analyzed_at = CURRENT_TIMESTAMP """).format(sql.Identifier(tenant_id)) cur.execute(upsert_query, ( data['thread_id'], data.get('sentiment_score'), data.get('intent'), data.get('is_high_intent'), data.get('is_lead_qualified'), data.get('is_resolved'), data.get('escalation_needed'), data.get('positive_points'), data.get('key_concerns'), data.get('pain_point'), data.get('feature_request'), data.get('objection'), data.get('cta_clicked'), data.get('summary'), json.dumps(data.get('metadata', {})) )) conn.commit() return True except Exception as e: conn.rollback() logger.error(f"Failed to upsert thread analytics: {e}", exc_info=True) return False finally: conn.close() def get_unanalyzed_threads(tenant_id: str): """ Returns unique thread IDs that are present in feedback but not in analytics. """ conn = get_db_connection() try: with conn.cursor() as cur: query = sql.SQL(""" SELECT DISTINCT thread_id FROM {}.strategist_feedback WHERE thread_id NOT IN (SELECT thread_id FROM {}.strategist_thread_analytics) """).format(sql.Identifier(tenant_id), sql.Identifier(tenant_id)) cur.execute(query) return [row[0] for row in cur.fetchall()] except Exception as e: logger.error(f"Error fetching unanalyzed threads for tenant {tenant_id}: {e}", exc_info=True) return [] finally: conn.close() def get_knowledge_sources(tenant_id: str, ingestion_type: str): """ Returns a list of unique sources (filenames, URLs) for a given ingestion type. """ logger.info(f"Listing unique knowledge sources for tenant: {tenant_id}, type: {ingestion_type}") conn = get_db_connection() try: with conn.cursor() as cur: # Check schema first cur.execute("SELECT schema_name FROM information_schema.schemata WHERE schema_name = %s", (tenant_id,)) if not cur.fetchone(): return [] query = sql.SQL(""" SELECT metadata->>'source_id' as source_id, metadata->>'source' as source_name, MIN(created_at) as first_ingested_at, metadata FROM {}.strategist_knowledge_base WHERE metadata->>'ingestion_type' = %s GROUP BY 1, 2, 4 ORDER BY first_ingested_at DESC """).format(sql.Identifier(tenant_id)) cur.execute(query, (ingestion_type,)) rows = cur.fetchall() sources = [] for row in rows: dt = row[2] sources.append({ "source_id": row[0], "source": row[1], "ingested_date": dt.date().isoformat() if dt else None, "ingested_time": dt.time().strftime("%H:%M:%S") if dt else None, "created_at": dt.isoformat() if dt else None, "metadata": row[3] }) return sources except Exception as e: logger.error(f"Failed to fetch knowledge sources for tenant {tenant_id}: {e}", exc_info=True) return [] finally: conn.close() def get_eligible_tenants(): """ Returns tenant schema names eligible for scheduled analytics refresh. Active Aivora subscriptions from master are authoritative. The legacy schema scan remains as a fallback if the master lookup is unavailable. """ logger.info("Fetching list of eligible Strategist tenants") try: conn = get_master_db_connection() try: with conn.cursor() as cur: query = f''' SELECT o."tenantId"::text, p."productFeatures" FROM "Organisations" o JOIN "Subscriptions" s ON s."organisationId" = o."id" JOIN "Plans" p ON p."id" = s."planId" WHERE {SUBSCRIPTION_ELIGIBILITY_SQL} ORDER BY o."tenantId"::text ''' cur.execute(query) tenants = { tenant_id if str(tenant_id).startswith("org_") else f"org_{tenant_id}" for tenant_id, product_features in cur.fetchall() if _plan_has_product(product_features, AIVORA_PRODUCT_NAME) } if tenants: tenant_list = sorted(tenants) logger.info(f"Found {len(tenant_list)} active/trial Aivora tenants: {tenant_list}") return tenant_list logger.warning("No active/trial Aivora subscriptions found in master database.") finally: conn.close() except Exception as e: logger.error(f"Error fetching active Aivora tenants from master database: {e}", exc_info=True) logger.info("Falling back to tenant schemas that contain strategist_knowledge_base") conn = get_db_connection() try: with conn.cursor() as cur: query = """ SELECT DISTINCT table_schema FROM information_schema.tables WHERE table_name = 'strategist_knowledge_base' AND table_schema NOT IN ('public', 'information_schema', 'pg_catalog') """ cur.execute(query) tenants = [row[0] for row in cur.fetchall()] logger.info(f"Found {len(tenants)} eligible tenants: {tenants}") return tenants except Exception as e: logger.error(f"Error fetching eligible tenants: {e}", exc_info=True) return [] finally: conn.close() def fetch_organisation_profile(tenant_id: str): """Returns the tenant's Organisation record from the master DB, or None. Used for per-tenant email branding (business name, logo, website, email). """ raw_id = str(tenant_id).replace("org_", "") try: uuid.UUID(raw_id) except ValueError: logger.warning(f"Tenant ID '{tenant_id}' is not a valid UUID. Organisation lookup skipped.") return None conn = get_master_db_connection() try: with conn.cursor() as cur: # A tenant can have several Organisation rows (the product allows # re-registering a business name). Without an explicit order the # branding would flip between sends, so take the most recent. cur.execute( 'SELECT "name", "businessName", "image", "website", "email" ' 'FROM "Organisations" WHERE "tenantId"::text = %s ' 'ORDER BY "createdAt" DESC NULLS LAST LIMIT 1', (raw_id,), ) row = cur.fetchone() if not row: return None return {"name": row[0], "businessName": row[1], "image": row[2], "website": row[3], "email": row[4]} except Exception as e: logger.error(f"Failed to fetch organisation profile for {tenant_id}: {e}") return None finally: conn.close() def fetch_tenant_limits(tenant_id: str): """ Fetches real-time limits from the galaxiq_master database using the Organisation-Subscription-Plan path. """ logger.info(f"Fetching dynamic limits from master database for tenant: {tenant_id}") # 0. Early UUID Validation: Master DB expects UUID strings for tenantId raw_id = str(tenant_id).replace("org_", "") try: uuid.UUID(raw_id) except ValueError: logger.warning(f"Tenant ID '{tenant_id}' is not a valid UUID. Master limits lookup skipped.") return None, None conn = get_master_db_connection() try: with conn.cursor() as cur: # 1. Organisations table to get organisationId from tenantId org_query = 'SELECT "id" FROM "Organisations" WHERE "tenantId"::text = %s LIMIT 1' cur.execute(org_query, (raw_id,)) org_row = cur.fetchone() if not org_row: logger.warning(f"No Organisation found for tenantId {tenant_id}") return None, None org_id = org_row[0] # 2. An organisation can have multiple active/trial subscriptions for different products. # For Strategist/Aivora quota checks, select the newest eligible plan that includes Aivora. sub_query = f''' SELECT s."planId", p."productFeatures", p."updatedAt" FROM "Subscriptions" s JOIN "Plans" p ON p."id" = s."planId" WHERE s."organisationId" = %s AND {SUBSCRIPTION_ELIGIBILITY_SQL} ORDER BY s."createdAt" DESC NULLS LAST, p."updatedAt" DESC NULLS LAST ''' cur.execute(sub_query, (org_id,)) subscription_rows = cur.fetchall() if not subscription_rows: logger.warning(f"No active/trial Subscription found for organisationId {org_id}") return None, None for plan_id, product_features, updated_at in subscription_rows: if _plan_has_product(product_features, AIVORA_PRODUCT_NAME): return product_features, updated_at logger.warning(f"No active/trial Aivora plan found for organisationId {org_id}") return None, None except Exception as e: logger.error(f"Failed to fetch dynamic limits for tenant {tenant_id}: {e}") return None, None finally: conn.close() def get_ticket_count(tenant_id: str, start_date=None): """Returns the number of tickets created for the tenant, optionally since a start date.""" conn = get_db_connection() try: with conn.cursor() as cur: if start_date: query = sql.SQL("SELECT COUNT(*) FROM {}.strategist_tickets WHERE created_at >= %s").format(sql.Identifier(tenant_id)) cur.execute(query, (start_date,)) else: cur.execute(sql.SQL("SELECT COUNT(*) FROM {}.strategist_tickets").format(sql.Identifier(tenant_id))) return cur.fetchone()[0] except Exception as e: logger.error(f"Error getting ticket count for {tenant_id}: {e}") raise finally: conn.close() def get_vector_db_size(tenant_id: str): """Returns the total size of the knowledge base table in bytes.""" conn = get_db_connection() try: with conn.cursor() as cur: # We use pg_total_relation_size which includes indexes (crucial for vector db) cur.execute(sql.SQL("SELECT pg_total_relation_size('{0}.strategist_knowledge_base')").format(sql.Identifier(tenant_id))) return cur.fetchone()[0] except Exception as e: logger.error(f"Error getting vector db size for {tenant_id}: {e}") raise finally: conn.close() def get_catalogue_product_count(tenant_id: str) -> int: """How many products this tenant's catalogue holds, across every source. The BrandForge plan caps catalogue size, and a merchant thinks of that as one number rather than one per connected source -- so this counts the whole table rather than filtering by source_kind. A tenant who has never ingested anything has no table yet, which is zero products rather than an error. """ conn = get_db_connection() try: with conn.cursor() as cur: cur.execute("SELECT to_regclass(%s)", (f"{tenant_id}.strategist_products",)) if not cur.fetchone()[0]: return 0 cur.execute(sql.SQL("SELECT count(*) FROM {}.strategist_products") .format(sql.Identifier(tenant_id))) return int(cur.fetchone()[0]) finally: conn.close() def count_new_products(tenant_id: str, product_keys: list) -> int: """How many of these products the catalogue does not already hold. A re-sync or a re-uploaded file carries the same products every time. Those update rows rather than adding them, so counting the whole batch against a plan limit would refuse a merchant who is adding nothing at all -- and would refuse them harder the more often they refresh. """ if not product_keys: return 0 conn = get_db_connection() try: with conn.cursor() as cur: cur.execute("SELECT to_regclass(%s)", (f"{tenant_id}.strategist_products",)) if not cur.fetchone()[0]: return len(product_keys) cur.execute(sql.SQL( "SELECT count(*) FROM {}.strategist_products " "WHERE product_key = ANY(%s)" ).format(sql.Identifier(tenant_id)), (list(product_keys),)) already_held = int(cur.fetchone()[0]) return len(product_keys) - already_held finally: conn.close() def get_total_token_usage(tenant_id: str, start_date=None): """Returns the sum of total_tokens for the tenant, optionally since a start date.""" conn = get_db_connection() try: with conn.cursor() as cur: if start_date: query = sql.SQL("SELECT SUM(total_tokens) FROM {}.strategist_llm_usage WHERE created_at >= %s").format(sql.Identifier(tenant_id)) cur.execute(query, (start_date,)) else: cur.execute(sql.SQL("SELECT SUM(total_tokens) FROM {}.strategist_llm_usage").format(sql.Identifier(tenant_id))) row = cur.fetchone() return row[0] if row and row[0] else 0 except Exception as e: logger.error(f"Error getting token usage for {tenant_id}: {e}") raise finally: conn.close() def get_conversation_count(tenant_id: str, start_date=None): """Returns the number of unique threads that had activity [usage] since a start date.""" conn = get_db_connection() try: with conn.cursor() as cur: if start_date: query = sql.SQL("SELECT COUNT(DISTINCT thread_id) FROM {}.strategist_llm_usage WHERE created_at >= %s").format(sql.Identifier(tenant_id)) cur.execute(query, (start_date,)) else: cur.execute(sql.SQL("SELECT COUNT(DISTINCT thread_id) FROM {}.strategist_llm_usage").format(sql.Identifier(tenant_id))) return cur.fetchone()[0] except Exception as e: logger.error(f"Error getting conversation count for {tenant_id}: {e}") raise finally: conn.close() # Shared by bootstrap_tenant and migrate_pairing_tables so the DDL cannot drift # between the two -- see migrate_pairing_tables below for why both exist. Each # ddl is a plain triple-quoted string (not pre-wrapped in sql.SQL) so both call # sites can decide independently, via sql.SQL(ddl).format(...), how the schema # placeholder gets filled in. _PAIRING_TABLES = [ ("strategist_product_neighbors", """ CREATE TABLE IF NOT EXISTS {}.strategist_product_neighbors ( anchor_key TEXT NOT NULL, neighbor_key TEXT NOT NULL, pair_type TEXT NOT NULL, score REAL NOT NULL, -- Separate from score on purpose: a pair can be a strong -- relationship derived by a weak method. The approval queue -- gates on how much the derivation is trusted, not on how -- good the pair is, so collapsing these would make it -- incoherent. confidence REAL NOT NULL, source TEXT NOT NULL, -- What the merchant is shown when asked to approve. A bare -- 0.79 is not reviewable. reasons JSONB NOT NULL DEFAULT '[]', computed_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (anchor_key, neighbor_key, pair_type) ) """), ("strategist_pairing_decisions", """ CREATE TABLE IF NOT EXISTS {}.strategist_pairing_decisions ( anchor_key TEXT NOT NULL, neighbor_key TEXT NOT NULL, pair_type TEXT NOT NULL, decision TEXT NOT NULL, decided_by TEXT, decided_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (anchor_key, neighbor_key, pair_type) ) """), ("strategist_product_embeddings", """ CREATE TABLE IF NOT EXISTS {}.strategist_product_embeddings ( product_key TEXT PRIMARY KEY, content_hash TEXT, vector REAL[] NOT NULL, computed_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP ) """), ] _PRODUCT_COLUMNS = [ ("source_kind", "TEXT NOT NULL DEFAULT 'crawl'"), ("source_ref", "TEXT"), ("external_id", "TEXT"), ("brand", "TEXT"), ("taxonomy_path", "TEXT[] NOT NULL DEFAULT '{}'"), ("taxonomy_source", "TEXT"), ("raw_category", "TEXT"), ("price_cents", "INT"), ("price_max_cents", "INT"), ("compare_at_cents", "INT"), ("currency", "TEXT"), ("on_sale", "BOOLEAN NOT NULL DEFAULT false"), ("in_stock", "BOOLEAN NOT NULL DEFAULT true"), ("status", "TEXT"), ("attributes", "JSONB NOT NULL DEFAULT '[]'"), ("quality_score", "REAL"), ("content_hash", "TEXT"), ("record_hash", "TEXT"), ("synced_at", "TIMESTAMP WITH TIME ZONE"), ("tenant_relations", "JSONB NOT NULL DEFAULT '{}'"), ("rating", "REAL"), ("review_count", "INT"), ("featured_rank", "INT"), # Non-empty means the product is incomplete and must never be served. # Deliberately not an overload of status, which carries the source's own # ACTIVE/DRAFT/ARCHIVED value -- collapsing the two would make an archived # product indistinguishable from an unlinkable one. ("missing_fields", "TEXT[] NOT NULL DEFAULT '{}'"), ("price_reference_cents", "INT"), ("fx_rate_used", "REAL"), # Attribute extraction (Phase 1): filtered on, not read, so each gets its # own column rather than a JSONB lookup. Nullable -- null means "not yet # extracted", distinct from a definite false/value. ("is_accessory", "BOOLEAN"), ("price_tier", "TEXT"), ("enriched_hash", "TEXT"), ] def migrate_products_table(tenant_id: str) -> dict: """Bring an existing tenant's products table up to the catalog-sync shape. bootstrap_tenant uses CREATE TABLE IF NOT EXISTS, which will not alter a table that already holds crawled rows. source_kind defaults to 'crawl' so every pre-existing row is labelled correctly -- any other default would let a later sync delete them as stale. """ added = [] conn = get_db_connection() try: with conn.cursor() as cur: # ADD COLUMN IF NOT EXISTS succeeds silently whether or not it did # anything, so the pre-existing column set is checked up front -- # otherwise "added" (and the log line below) would always claim all # 19 columns, even on a no-op re-run of an already-migrated tenant. cur.execute(sql.SQL( "SELECT column_name FROM information_schema.columns " "WHERE table_schema = %s AND table_name = 'strategist_products'" ), (tenant_id,)) existing = {row[0] for row in cur.fetchall()} for name, ddl in _PRODUCT_COLUMNS: # ddl is built with sql.SQL() rather than folded into the # .format() call above: some defaults (e.g. taxonomy_path's # '{}') contain literal braces that .format() would otherwise # misparse as placeholders. stmt = sql.SQL( "ALTER TABLE {}.strategist_products ADD COLUMN IF NOT EXISTS {} " ).format(sql.Identifier(tenant_id), sql.Identifier(name)) + sql.SQL(ddl) cur.execute(stmt) if name not in existing: added.append(name) # A source that exposes no product URLs at all would otherwise have # every row rejected at validation, leaving the merchant with an # empty catalog and no explanation. DROP NOT NULL is idempotent, so # this stays safe on a tenant that has already been migrated. cur.execute(sql.SQL( "ALTER TABLE {}.strategist_products " "ALTER COLUMN product_url DROP NOT NULL" ).format(sql.Identifier(tenant_id))) cur.execute(sql.SQL( "CREATE INDEX IF NOT EXISTS strategist_products_source_idx " "ON {}.strategist_products (source_kind, source_ref)" ).format(sql.Identifier(tenant_id))) cur.execute(sql.SQL( "CREATE INDEX IF NOT EXISTS strategist_products_status_idx " "ON {}.strategist_products (status, in_stock)" ).format(sql.Identifier(tenant_id))) conn.commit() logger.info(f"Migrated products table for {tenant_id}: {len(added)} columns") return {"added": added} except Exception: conn.rollback() logger.error(f"Could not migrate products table for {tenant_id}", exc_info=True) raise finally: conn.close() def migrate_pairing_tables(tenant_id: str) -> dict: """Add the pairing graph tables to a tenant whose schema predates them. bootstrap_tenant uses CREATE TABLE IF NOT EXISTS, so a tenant provisioned before this phase never gets strategist_product_neighbors, strategist_pairing_decisions, or strategist_product_embeddings just by logging in again. Same reasoning as migrate_products_table above. """ created = [] conn = get_db_connection() try: with conn.cursor() as cur: # information_schema is checked up front, same reason as # migrate_products_table: CREATE TABLE IF NOT EXISTS succeeds # silently whether or not it did anything, so without this the # returned list would always claim all three tables, even on a # no-op re-run of an already-migrated tenant. cur.execute( "SELECT table_name FROM information_schema.tables " "WHERE table_schema = %s", (tenant_id,), ) existing = {row[0] for row in cur.fetchall()} for table_name, ddl in _PAIRING_TABLES: cur.execute(sql.SQL(ddl).format(sql.Identifier(tenant_id))) if table_name not in existing: created.append(table_name) cur.execute(sql.SQL( "CREATE INDEX IF NOT EXISTS strategist_neighbors_anchor_idx " "ON {}.strategist_product_neighbors (anchor_key, pair_type)" ).format(sql.Identifier(tenant_id))) conn.commit() logger.info(f"Migrated pairing tables for {tenant_id}: {len(created)} created") return {"created": created} except Exception: conn.rollback() logger.error(f"Could not migrate pairing tables for {tenant_id}", exc_info=True) raise finally: conn.close()