hold-slayer/db/database.py

"""
Database connection and session management.

PostgreSQL via asyncpg + SQLAlchemy async.
"""

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from pathlib import Path

from sqlalchemy import (
    JSON,
    Boolean,
    Column,
    DateTime,
    Float,
    Integer,
    String,
    Text,
    func,
)
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from config import get_settings


class Base(DeclarativeBase):
    """SQLAlchemy declarative base for all ORM models."""

    pass


# ============================================================
# ORM Models
# ============================================================


class CallRecord(Base):
    __tablename__ = "call_records"

    id = Column(String, primary_key=True)
    direction = Column(String, nullable=False)  # inbound / outbound
    remote_number = Column(String, index=True, nullable=False)
    status = Column(String, nullable=False)  # completed / missed / failed / active / on_hold
    mode = Column(String, nullable=False)  # direct / hold_slayer / ai_assisted
    intent = Column(Text)  # What the user wanted (for hold_slayer)
    started_at = Column(DateTime, default=func.now())
    ended_at = Column(DateTime, nullable=True)
    duration = Column(Integer, default=0)  # seconds
    hold_time = Column(Integer, default=0)  # seconds spent on hold
    device_used = Column(String)
    recording_path = Column(String, nullable=True)
    summary = Column(Text, nullable=True)
    action_items = Column(JSON, nullable=True)
    sentiment = Column(String, nullable=True)
    call_flow_id = Column(String, nullable=True)  # which flow was used
    classification_timeline = Column(JSON, nullable=True)  # [{time, type, confidence}, ...]
    metadata_ = Column("metadata", JSON, nullable=True)

    def __repr__(self) -> str:
        return f"<CallRecord {self.id} {self.remote_number} {self.status}>"


class StoredCallFlow(Base):
    __tablename__ = "call_flows"

    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)
    phone_number = Column(String, index=True, nullable=False)
    description = Column(Text)
    steps = Column(JSON, nullable=False)  # Serialized list[CallFlowStep]
    last_verified = Column(DateTime, nullable=True)
    avg_hold_time = Column(Integer, nullable=True)
    success_rate = Column(Float, nullable=True)
    times_used = Column(Integer, default=0)
    last_used = Column(DateTime, nullable=True)
    notes = Column(Text, nullable=True)
    tags = Column(JSON, default=list)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())

    def __repr__(self) -> str:
        return f"<StoredCallFlow {self.id} {self.phone_number}>"


class Device(Base):
    __tablename__ = "devices"

    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)  # "Office SIP Phone"
    type = Column(String, nullable=False)  # sip_phone / cell / tablet / softphone
    sip_uri = Column(String, nullable=True)  # sip:robert@gateway.helu.ca
    phone_number = Column(String, nullable=True)  # For PSTN devices
    priority = Column(Integer, default=10)  # Routing priority (lower = higher priority)
    is_online = Column(Boolean, default=False, nullable=False)
    capabilities = Column(JSON, default=list)  # ["voice", "video", "sms"]
    dnd = Column(Boolean, default=False, nullable=False)
    last_seen = Column(DateTime, nullable=True)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())

    def __repr__(self) -> str:
        return f"<Device {self.id} {self.name} ({self.type})>"


class RoutingRuleRecord(Base):
    __tablename__ = "routing_rules"

    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)
    priority = Column(Integer, default=100, nullable=False)  # lower runs first
    enabled = Column(Boolean, default=True, nullable=False)
    match = Column(JSON, nullable=False)  # caller_pattern, dnis, time_range, days
    action = Column(JSON, nullable=False)  # {type, ...}
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())

    def __repr__(self) -> str:
        return f"<RoutingRule {self.id} {self.name} p={self.priority}>"


class TranscriptChunk(Base):
    __tablename__ = "transcript_chunks"

    id = Column(String, primary_key=True)
    call_id = Column(String, index=True, nullable=False)
    seq = Column(Integer, nullable=False)
    t_offset_ms = Column(Integer, default=0)  # offset from call start
    speaker = Column(String, default="unknown")  # caller / agent / receptionist / unknown
    text = Column(Text, nullable=False)
    confidence = Column(Float, nullable=True)
    created_at = Column(DateTime, default=func.now())

    def __repr__(self) -> str:
        return f"<TranscriptChunk {self.call_id}#{self.seq}>"


class RecordingRecord(Base):
    __tablename__ = "recordings"

    id = Column(String, primary_key=True)
    call_id = Column(String, index=True, nullable=False)
    path = Column(String, nullable=False)
    format = Column(String, default="wav")
    duration_s = Column(Float, default=0.0)
    size_bytes = Column(Integer, default=0)
    channels = Column(Integer, default=1)
    started_at = Column(DateTime, default=func.now())
    ended_at = Column(DateTime, nullable=True)

    def __repr__(self) -> str:
        return f"<Recording {self.id} call={self.call_id} {self.path}>"


# ============================================================
# Engine & Session
# ============================================================

_engine = None
_session_factory = None


def get_engine():
    """Get or create the async engine."""
    global _engine
    if _engine is None:
        settings = get_settings()
        _engine = create_async_engine(
            settings.database_url,
            echo=settings.debug,
            pool_size=10,
            max_overflow=20,
        )
    return _engine


def get_session_factory() -> async_sessionmaker[AsyncSession]:
    """Get or create the session factory."""
    global _session_factory
    if _session_factory is None:
        _session_factory = async_sessionmaker(
            get_engine(),
            class_=AsyncSession,
            expire_on_commit=False,
        )
    return _session_factory


@asynccontextmanager
async def session_scope() -> AsyncIterator[AsyncSession]:
    """A commit-on-success session — the one session-lifecycle convention.

    REST handlers get it via the get_db dependency; services and MCP
    tools use it directly.
    """
    factory = get_session_factory()
    async with factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


async def get_db() -> AsyncIterator[AsyncSession]:
    """FastAPI dependency: yield an async database session."""
    async with session_scope() as session:
        yield session


# The autogenerated baseline revision — a schema created by the old
# create_all path is identical to it, so such databases are stamped
# here and then migrated forward like any other.
_BASELINE_REVISION = "1173a71329ed"


def _upgrade_to_head(connection) -> None:
    from alembic import command
    from alembic.config import Config
    from sqlalchemy import inspect

    cfg = Config(str(Path(__file__).resolve().parent.parent / "alembic.ini"))
    cfg.attributes["connection"] = connection

    inspector = inspect(connection)
    if not inspector.has_table("alembic_version") and inspector.has_table("call_records"):
        command.stamp(cfg, _BASELINE_REVISION)
    command.upgrade(cfg, "head")


async def init_db():
    """Bring the schema to Alembic head (tests create tables directly)."""
    engine = get_engine()
    async with engine.begin() as conn:
        await conn.run_sync(_upgrade_to_head)


async def close_db():
    """Close the database engine."""
    global _engine, _session_factory
    if _engine is not None:
        await _engine.dispose()
        _engine = None
        _session_factory = None