hold-slayer/main.py

"""
Hold Slayer Gateway — FastAPI Application Entry Point.

Your personal AI-powered telephony platform.
Navigates IVRs, waits on hold, and connects you when a human answers.

Usage:
    uvicorn main:app --host 0.0.0.0 --port 8000 --reload
    
    # Or directly:
    python main.py
"""

import logging
import sys
from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI
from fastapi.staticfiles import StaticFiles

from api import call_flows, call_history, calls, devices, routing, websocket
from api.deps import require_token
from config import Settings, get_settings
from core.gateway import AIPSTNGateway, build_sip_engine
from db.database import close_db, init_db
from mcp_server.server import create_mcp_server
from models.call import CallMode
from services.audio_classifier import AudioClassifier
from services.call_persistence import persist_call_on_create, persist_call_on_end
from services.hold_slayer import HoldSlayerService
from services.notification import NotificationService
from services.receptionist import ReceptionistService
from services.recording import RecordingService
from services.routing import RoutingService
from services.transcription import TranscriptionService
from services.tts import TTSService

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)-7s | %(name)s | %(message)s",
    datefmt="%H:%M:%S",
    stream=sys.stdout,
)
logger = logging.getLogger(__name__)


def _handle_db_error(exc: Exception) -> None:
    """Log a clear, human-readable database error and exit cleanly."""
    # Walk the exception chain to find the root asyncpg/psycopg cause
    cause = getattr(exc, "__cause__", None) or getattr(exc, "__context__", None)
    root = cause or exc
    root_type = type(root).__name__
    root_msg = str(root)

    if "InvalidPasswordError" in root_type or "password authentication failed" in root_msg:
        logger.critical(
            "\n"
            "❌  Database authentication failed — wrong password.\n"
            "    The password in DATABASE_URL does not match the PostgreSQL user.\n"
            "    Fix DATABASE_URL in your .env file and restart.\n"
            "    Default: DATABASE_URL=postgresql+asyncpg://holdslayer:changeme@localhost:5432/holdslayer"
        )
    elif "InvalidCatalogNameError" in root_type or "does not exist" in root_msg:
        logger.critical(
            "\n"
            "❌  Database does not exist.\n"
            "    Create it first:  createdb holdslayer\n"
            "    Or update DATABASE_URL in your .env file."
        )
    elif (
        "Connection refused" in root_msg
        or "could not connect" in root_msg.lower()
    ):
        logger.critical(
            "\n"
            "\u274c  Cannot reach PostgreSQL \u2014 connection refused.\n"
            "    Is PostgreSQL running? Check DATABASE_URL in your .env file."
        )
    elif (
        "nodename nor servname" in root_msg
        or "Name or service not known" in root_msg
    ):
        logger.critical(
            "\n"
            f"❌  Cannot resolve the database hostname.\n"
            f"    Check the host in DATABASE_URL in your .env file. (detail: {root_msg})"
        )
    else:
        logger.critical(
            f"\n❌  Database initialisation failed: {root_msg}\n"
            f"    Check DATABASE_URL in your .env file."
        )

    sys.exit(1)


def _check_startup_config(settings: Settings) -> None:
    """Refuse insecure or incomplete configurations before booting anything."""
    if not settings.database_url:
        logger.critical(
            "\n"
            "❌  DATABASE_URL is not set.\n"
            "    Add it to your .env file, e.g.:\n"
            "    DATABASE_URL=postgresql+asyncpg://holdslayer:<password>@localhost:5432/holdslayer"
        )
        sys.exit(1)

    token = settings.api_token.get_secret_value()
    if not token and settings.host not in ("127.0.0.1", "localhost", "::1"):
        logger.critical(
            "\n"
            "❌  API_TOKEN is not set but HOST binds beyond loopback "
            f"({settings.host}).\n"
            "    Every surface (REST, WebSocket, MCP make_call) would be open "
            "to the network.\n"
            "    Set API_TOKEN in .env (e.g. `openssl rand -hex 32`), or set "
            "HOST=127.0.0.1 for tokenless local development."
        )
        sys.exit(1)


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Startup: Initialize database, SIP engine, and services."""
    settings = get_settings()
    _check_startup_config(settings)

    # The MCP session manager lives in the mounted sub-app's lifespan;
    # without entering it, every /mcp request 500s.
    async with mcp_http_app.lifespan(app):
        # Initialize database
        logger.info("Initializing database...")
        try:
            await init_db()
        except Exception as e:
            _handle_db_error(e)

        # === Composition root ===
        # Build the gateway and every service here, wiring them by
        # constructor/registration — nothing constructs its own deps.
        gateway = AIPSTNGateway(
            settings=settings,
            on_call_created=persist_call_on_create,
            on_call_ended=persist_call_on_end,
        )

        classifier = AudioClassifier(settings.classifier)
        transcription = TranscriptionService(settings.speaches)
        tts = TTSService(settings.tts)
        routing_svc = RoutingService(gateway)
        recording_svc = RecordingService()
        receptionist = ReceptionistService(
            gateway,
            tts=tts,
            transcription=transcription,
            recording=recording_svc,
            routing=routing_svc,
        )
        gateway.attach_services(tts=tts)

        def launch_hold_slayer(call, sip_leg_id, call_flow_id):
            svc = HoldSlayerService(
                gateway=gateway,
                call_manager=gateway.call_manager,
                sip_engine=gateway.sip_engine,
                classifier=classifier,
                transcription=transcription,
                settings=settings,
                tts=tts,
            )
            gateway.spawn(
                svc.run(call, sip_leg_id, call_flow_id),
                name=f"holdslayer_{call.id}",
            )

        gateway.register_mode_handler(CallMode.HOLD_SLAYER, launch_hold_slayer)

        try:
            gateway.sip_engine = build_sip_engine(
                settings,
                gateway.media_pipeline,
                on_leg_state_change=gateway._on_sip_leg_state,
                on_device_registered=gateway._on_sip_device_registered,
                on_incoming_call=receptionist.on_inbound_call,
            )
        except Exception as e:
            logger.critical(f"\n❌  SIP engine failed to initialize:\n    {e}")
            sys.exit(1)

        await routing_svc.start()
        await gateway.start()
        app.state.gateway = gateway
        app.state.routing_service = routing_svc
        app.state.transcription_service = transcription

        notification_svc = NotificationService(gateway.event_bus, settings)
        await notification_svc.start()
        app.state.notification_service = notification_svc

        await recording_svc.start()
        app.state.recording_service = recording_svc

        logger.info("=" * 60)
        logger.info("🔥 Hold Slayer Gateway is LIVE")
        # Show a usable URL — 0.0.0.0 is the bind address, not a browser URL
        display_host = "localhost" if settings.host in ("0.0.0.0", "::") else settings.host
        # When launched via `uvicorn main:app --port XXXX`, the CLI --port arg
        # takes precedence over settings.port (which comes from .env).
        display_port = settings.port
        for i, arg in enumerate(sys.argv):
            if arg in ("--port", "-p") and i + 1 < len(sys.argv):
                try:
                    display_port = int(sys.argv[i + 1])
                except ValueError:
                    pass
        auth_state = "bearer token required" if settings.api_token.get_secret_value() else "auth disabled (loopback)"
        logger.info(f"   API:       http://{display_host}:{display_port}  [{auth_state}]")
        logger.info(f"   API Docs:  http://{display_host}:{display_port}/docs")
        logger.info(f"   WebSocket: ws://{display_host}:{display_port}/ws/events")
        logger.info(f"   MCP:       http://{display_host}:{display_port}/mcp/  (streamable HTTP)")
        logger.info("=" * 60)

        yield

        # Shutdown
        logger.info("Shutting down Hold Slayer Gateway...")
        await notification_svc.stop()
        await gateway.stop()
        await close_db()
        logger.info("Gateway shut down cleanly. 👋")


def _get_gateway_instance() -> AIPSTNGateway | None:
    """Lazy gateway resolver for MCP tools (set on app.state by the lifespan)."""
    return getattr(app.state, "gateway", None)


mcp = create_mcp_server(
    _get_gateway_instance,
    api_token=get_settings().api_token.get_secret_value(),
)
mcp_http_app = mcp.http_app(path="/")

app = FastAPI(
    title="Hold Slayer Gateway",
    description=(
        "🗡️ AI PSTN Gateway — Navigate IVRs, wait on hold, "
        "and connect you when a human answers.\n\n"
        "## Quick Start\n"
        "1. **POST /api/v1/calls/hold-slayer** — Launch the Hold Slayer\n"
        "2. **GET /api/v1/calls/{call_id}** — Check call status\n"
        "3. **WS /ws/events** — Real-time event stream\n"
        "4. **GET /api/v1/call-flows** — Manage stored IVR trees\n"
    ),
    version="0.1.0",
    lifespan=lifespan,
)

# === API Routes ===
# call_history must register before calls: both live under /api/v1/calls and
# calls' GET /{call_id} would otherwise capture the literal path "history".
_auth = [Depends(require_token)]
app.include_router(
    call_history.router, prefix="/api/v1/calls", tags=["Call History"], dependencies=_auth
)
app.include_router(calls.router, prefix="/api/v1/calls", tags=["Calls"], dependencies=_auth)
app.include_router(
    call_flows.router, prefix="/api/v1/call-flows", tags=["Call Flows"], dependencies=_auth
)
app.include_router(devices.router, prefix="/api/v1/devices", tags=["Devices"], dependencies=_auth)
app.include_router(routing.router, prefix="/api/v1/routing", tags=["Routing"], dependencies=_auth)
# WebSocket endpoints check the token themselves (query param or header)
app.include_router(websocket.router, prefix="/ws", tags=["WebSocket"])

# === MCP (streamable HTTP; clients connect to /mcp/ with the bearer token) ===
app.mount("/mcp", mcp_http_app)


@app.get("/api/v1/status", tags=["System"], dependencies=_auth)
async def api_status():
    """Gateway status summary for the dashboard header."""
    gateway = getattr(app.state, "gateway", None)
    if gateway:
        status = await gateway.status()
        return {
            "name": "Hold Slayer Gateway",
            "version": "0.1.0",
            "status": "running",
            "uptime": status["uptime"],
            "active_calls": status["active_calls"],
            "trunk": status["trunk"],
        }
    return {
        "name": "Hold Slayer Gateway",
        "version": "0.1.0",
        "status": "starting",
    }


@app.get("/health", tags=["System"])
async def health():
    """
    Health check. "healthy" means the gateway can actually do its job:
    real engine, registered trunk, reachable database. A mock engine or
    a failing dependency reports "degraded" with the reason visible.
    """
    from core.sip_engine import MockSIPEngine
    from db.database import session_scope

    gateway = getattr(app.state, "gateway", None)
    ready = gateway is not None and await gateway.sip_engine.is_ready()
    trunk_status = await gateway.sip_engine.get_trunk_status() if gateway else {"registered": False}
    engine_mode = (
        "mock" if gateway is None or isinstance(gateway.sip_engine, MockSIPEngine)
        else "sippy"
    )

    db_ok = False
    db_error = None
    try:
        from sqlalchemy import text
        async with session_scope() as session:
            await session.execute(text("SELECT 1"))
        db_ok = True
    except Exception as e:
        db_error = str(e)[:200]

    healthy = (
        ready
        and db_ok
        and engine_mode == "sippy"
        and trunk_status.get("registered", False)
    )

    checks = {
        "gateway": "ready" if gateway else "not initialized",
        "engine": engine_mode,
        "sip_engine": "ready" if ready else "not ready",
        "database": "ok" if db_ok else f"error: {db_error}",
        "sip_trunk": {
            "registered": trunk_status.get("registered", False),
            "host": trunk_status.get("host"),
            "reason": trunk_status.get("reason"),
        },
    }
    if gateway is not None:
        tts = getattr(gateway, "_tts", None)
        checks["tts"] = _availability(tts)
        transcription = getattr(app.state, "transcription_service", None)
        checks["stt"] = _availability(transcription)

    return {"status": "healthy" if healthy else "degraded", **checks}


def _availability(service) -> str:
    """Last-known reachability of an HTTP leaf service."""
    if service is None:
        return "not attached"
    available = getattr(service, "available", None)
    if available is None:
        return "unknown (no requests yet)"
    return "ok" if available else "unreachable"


# === Dashboard (built SvelteKit static, served at the root) ===
# Registered last: a "/" mount matches every path, so the API, WS,
# health, and MCP routes above must come first.
import os as _os  # noqa: E402

_dashboard_build = _os.path.join(_os.path.dirname(__file__), "dashboard", "build")
if _os.path.isdir(_dashboard_build):
    app.mount(
        "/",
        StaticFiles(directory=_dashboard_build, html=True),
        name="dashboard",
    )


if __name__ == "__main__":
    import uvicorn

    settings = get_settings()
    uvicorn.run(
        "main:app",
        host=settings.host,
        port=settings.port,
        reload=settings.debug,
        log_level=settings.log_level,
    )