# neon_client.py # Database client for Neon Postgres using asyncpg connection pool # # v2 (FIX F): # insert_call_log() previously reused the same placeholder ($7, # duration_seconds) both as an INT column value (total_duration_seconds) # AND inside `NOW() - INTERVAL '1 second' * $7`, where Postgres infers # the multiplication operand should be `double precision`. asyncpg's # prepared-statement planner then sees two conflicting inferred types # for the same parameter ("inconsistent types deduced for parameter $7 # ... double precision versus integer"), and every insert failed. # # Fix: compute `started_at` as a plain Python datetime # (utcnow() - timedelta(seconds=duration_seconds)) and pass it as its # own TIMESTAMPTZ parameter. duration_seconds is now used only once, # as a plain integer, with no SQL-side arithmetic/type inference. import os import json import logging import asyncpg import time from datetime import datetime, timedelta, timezone from typing import Optional, Dict, Any, List logger = logging.getLogger("neon_client") # Connection pool singleton _pool: Optional[asyncpg.Pool] = None # Cache dictionary: (company_id, line_token) -> (timestamp, config_dict) _config_cache: Dict[tuple, tuple] = {} async def init_db_pool(): """Initializes the asyncpg connection pool.""" global _pool if _pool is not None: return database_url = os.getenv("DATABASE_URL") if not database_url: logger.error("DATABASE_URL environment variable is missing. Database operations will fail.") return try: # Create pool with minimum 1 and maximum 5 connections _pool = await asyncpg.create_pool(dsn=database_url, min_size=1, max_size=5) logger.info("Neon database connection pool established.") except Exception as e: logger.exception(f"Failed to create Neon connection pool: {e}") async def close_db_pool(): """Closes the connection pool.""" global _pool if _pool is not None: await _pool.close() _pool = None logger.info("Neon database connection pool closed.") async def fetch_tenant_config(company_id: str, line_token: Optional[str] = None) -> Optional[Dict[str, Any]]: """ Fetches the tenant configurations for a company, verified by company_id and optional line_token. Normalizes the caller details list structure on the fly. """ global _pool, _config_cache # Check cache hit (60s TTL) cache_key = (company_id, line_token) if cache_key in _config_cache: cached_time, cached_val = _config_cache[cache_key] if time.time() - cached_time < 60.0: logger.info(f"Returning cached tenant config for company_id={company_id}") return cached_val if _pool is None: await init_db_pool() if _pool is None: return None query = """ SELECT c.id AS company_id, c.name AS company_name, c.name_ur AS company_name_ur, c.overview, c.overview_ur, c.is_active AS company_active, ac.agent_name, ac.agent_name_ur, ac.agent_role, ac.agent_role_ur, ac.gender AS agent_gender, ac.system_instruction_rules, ac.caller_details_to_collect, ac.rag_enabled, ac.rag_similarity_threshold, ac.id AS agent_config_id, pl.id AS phone_line_id, pl.line_token, pl.is_enabled AS line_active FROM companies c LEFT JOIN phone_lines pl ON pl.company_id = c.id AND pl.is_enabled = true LEFT JOIN agent_configs ac ON ac.id = pl.agent_config_id WHERE c.id = $1 AND c.is_active = true AND ($2::text IS NULL OR pl.line_token = $2) LIMIT 1 """ try: async with _pool.acquire() as conn: row = await conn.fetchrow(query, company_id, line_token) if not row: return None config = dict(row) # Normalize caller details configuration list raw_details = config.get("caller_details_to_collect") normalized_details = [] if raw_details: if isinstance(raw_details, str): try: raw_details = json.loads(raw_details) except Exception: pass if isinstance(raw_details, list): for item in raw_details: if isinstance(item, str): label = item.replace("_", " ").title() normalized_details.append({ "key": item, "label": label, "label_ur": label, "type": "text" }) elif isinstance(item, dict) and "key" in item: normalized_details.append({ "key": item["key"], "label": item.get("label", item["key"].replace("_", " ").title()), "label_ur": item.get("label_ur", item.get("label", item["key"].replace("_", " ").title())), "type": item.get("type", "text") }) config["caller_details_to_collect"] = normalized_details _config_cache[cache_key] = (time.time(), config) return config except Exception as e: logger.error(f"Error querying tenant config: {e}") return None async def insert_call_log( call_id: str, company_id: str, phone_line_id: Optional[int], full_transcript: str, extracted_caller_data: Dict[str, Any], duration_seconds: int, status: str = "completed" ) -> bool: """ Inserts or updates a call record inside Neon database. FIX F: started_at is now computed in Python as a plain TIMESTAMPTZ parameter ($8), rather than derived in SQL via `NOW() - INTERVAL '1 second' * $7`. This avoids asyncpg/Postgres deducing two conflicting types (integer vs double precision) for the same placeholder when duration_seconds is also bound to the INT column total_duration_seconds ($7). """ global _pool if _pool is None: await init_db_pool() if _pool is None: return False query = """ INSERT INTO call_logs ( id, company_id, phone_line_id, direction, status, full_transcript_history, extracted_caller_data, total_duration_seconds, started_at, ended_at ) VALUES ( $1, $2, $3, 'inbound', $4, $5, $6, $7, $8, NOW() ) ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, full_transcript_history = EXCLUDED.full_transcript_history, extracted_caller_data = EXCLUDED.extracted_caller_data, total_duration_seconds = EXCLUDED.total_duration_seconds, ended_at = EXCLUDED.ended_at """ try: extracted_json = json.dumps(extracted_caller_data) duration_int = int(duration_seconds) started_at = datetime.now(timezone.utc) - timedelta(seconds=duration_int) async with _pool.acquire() as conn: await conn.execute( query, call_id, company_id, phone_line_id, status, full_transcript, extracted_json, duration_int, started_at, ) logger.info(f"Successfully saved call log to database: {call_id}") return True except Exception as e: logger.error(f"Failed to insert call log to database: {e}") return False