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 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.
    """
    try:
        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,
                ))
    except Exception as e:
        logger.warning(f"Could not persist call {call.id}: {e}")