python-services/neon_client.py

# 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