hold-slayer/services/call_persistence.py

"""
Call Persistence — the data-access layer for calls and call flows.

Holds the on-hangup persistence hook plus the query/write functions
that both the REST handlers and the MCP tools call, so the two
surfaces can't drift. Every function takes an AsyncSession; callers
own the transaction (get_db for REST, session_scope for MCP/services).
"""

import asyncio
import logging
import uuid
from datetime import datetime

from sqlalchemy import desc, select
from sqlalchemy.ext.asyncio import AsyncSession

from db.database import (
    CallRecord,
    RecordingRecord,
    StoredCallFlow,
    TranscriptChunk,
    session_scope,
)
from models.call import ActiveCall, CallStatus
from models.call_flow import CallFlow, CallFlowStep

logger = logging.getLogger(__name__)


def flow_to_model(row: StoredCallFlow) -> CallFlow:
    """The one StoredCallFlow-row → CallFlow-model mapping."""
    return CallFlow(
        id=row.id,
        name=row.name,
        phone_number=row.phone_number,
        description=row.description or "",
        steps=[CallFlowStep(**s) for s in (row.steps or [])],
        tags=row.tags or [],
        notes=row.notes,
        avg_hold_time=row.avg_hold_time,
        success_rate=row.success_rate,
        last_used=row.last_used,
        times_used=row.times_used or 0,
    )


# ================================================================
# Call flows
# ================================================================

async def get_flow(session: AsyncSession, flow_id: str) -> StoredCallFlow | None:
    result = await session.execute(
        select(StoredCallFlow).where(StoredCallFlow.id == flow_id)
    )
    return result.scalar_one_or_none()


async def get_flow_by_number(
    session: AsyncSession, phone_number: str
) -> StoredCallFlow | None:
    result = await session.execute(
        select(StoredCallFlow).where(StoredCallFlow.phone_number == phone_number)
    )
    return result.scalar_one_or_none()


async def list_flows(session: AsyncSession) -> list[StoredCallFlow]:
    result = await session.execute(select(StoredCallFlow))
    return list(result.scalars().all())


async def create_flow(
    session: AsyncSession,
    flow_id: str,
    name: str,
    phone_number: str,
    steps: list[dict],
    description: str | None = None,
    tags: list[str] | None = None,
    notes: str | None = None,
) -> StoredCallFlow:
    row = StoredCallFlow(
        id=flow_id,
        name=name,
        phone_number=phone_number,
        description=description,
        steps=steps,
        tags=tags,
        notes=notes,
        last_verified=datetime.now(),
    )
    session.add(row)
    await session.flush()
    return row


# ================================================================
# Call history / records
# ================================================================

async def get_record(session: AsyncSession, call_id: str) -> CallRecord | None:
    result = await session.execute(
        select(CallRecord).where(CallRecord.id == call_id)
    )
    return result.scalar_one_or_none()


async def search_history(
    session: AsyncSession,
    number: str | None = None,
    number_contains: str | None = None,
    intent_contains: str | None = None,
    status: str | None = None,
    since: datetime | None = None,
    until: datetime | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[CallRecord]:
    stmt = select(CallRecord).order_by(desc(CallRecord.started_at))
    if number:
        stmt = stmt.where(CallRecord.remote_number == number)
    if number_contains:
        stmt = stmt.where(CallRecord.remote_number.contains(number_contains))
    if intent_contains:
        stmt = stmt.where(CallRecord.intent.icontains(intent_contains))
    if status:
        stmt = stmt.where(CallRecord.status == status)
    if since:
        stmt = stmt.where(CallRecord.started_at >= since)
    if until:
        stmt = stmt.where(CallRecord.started_at <= until)
    result = await session.execute(stmt.offset(offset).limit(limit))
    return list(result.scalars().all())


async def get_transcript_chunks(
    session: AsyncSession, call_id: str
) -> list[TranscriptChunk]:
    result = await session.execute(
        select(TranscriptChunk)
        .where(TranscriptChunk.call_id == call_id)
        .order_by(TranscriptChunk.seq)
    )
    return list(result.scalars().all())


async def latest_recording(
    session: AsyncSession, call_id: str
) -> RecordingRecord | None:
    result = await session.execute(
        select(RecordingRecord)
        .where(RecordingRecord.call_id == call_id)
        .order_by(desc(RecordingRecord.started_at))
    )
    return result.scalars().first()


async def persist_call_on_end(call: ActiveCall, final_status: CallStatus) -> None:
    """Insert a CallRecord and any transcript chunks for `call`.

    Wired into CallManager as its on_call_ended hook by the
    composition root in main.py. Retries briefly — losing the row
    means the call never happened as far as history is concerned, so
    the final failure logs at ERROR with the payload identifiers.
    """
    for attempt in range(3):
        try:
            await _write_call_record(call, final_status)
            return
        except Exception as e:
            if attempt == 2:
                logger.error(
                    f"Call record lost: id={call.id} number={call.remote_number} "
                    f"status={final_status.value}: {e}"
                )
                return
            await asyncio.sleep(2**attempt)


async def _write_call_record(call: ActiveCall, final_status: CallStatus) -> None:
    async with session_scope() as session:
        record = CallRecord(
            id=call.id,
            direction=call.direction,
            remote_number=call.remote_number,
            status=final_status.value,
            mode=call.mode.value,
            intent=call.intent,
            started_at=call.started_at,
            ended_at=datetime.now(),
            duration=int(call.duration),
            hold_time=int(call.hold_time),
            device_used=call.device,
            call_flow_id=call.call_flow_id,
            classification_timeline=[
                {
                    "timestamp": c.timestamp,
                    "audio_type": c.audio_type.value,
                    "confidence": c.confidence,
                }
                for c in call.classification_history
            ],
            metadata_={"services": list(call.services)},
        )
        session.add(record)

        # Each transcript chunk gets its own row with a sequence number
        # so the dashboard can render them in order with click-to-seek.
        for seq, text in enumerate(call.transcript_chunks):
            speaker = "unknown"
            payload = text
            if ":" in text:
                head, rest = text.split(":", 1)
                head = head.strip().lower()
                if head in {"caller", "agent", "receptionist", "caller_message"}:
                    speaker = head if head != "caller_message" else "caller"
                    payload = rest.strip()
            session.add(TranscriptChunk(
                id=f"tc_{uuid.uuid4().hex[:10]}",
                call_id=call.id,
                seq=seq,
                t_offset_ms=0,
                speaker=speaker,
                text=payload,
            ))