call-center/services/ivr_service/app.py

from __future__ import annotations

import json
import os

from fastapi import Depends, FastAPI, HTTPException, Query
from sqlalchemy import select

from services.shared.core import Role, new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.event_bus import append_outbox_event, event_bus_enabled
from services.shared.models import (
    HealthResponse,
    IvrDtmfIn,
    IvrFlowCreate,
    IvrFlowOut,
    IvrFlowUpdate,
    IvrSessionOut,
    IvrSessionStartIn,
)
from services.shared.security import get_actor, require_roles
from services.shared.sql_init import init_sql_schema
from services.shared.sql_models import InteractionTimeline, IvrFlowRow, IvrSessionRow, VoiceEventRow

app = FastAPI(title="ivr-service", version="1.0.0")

init_sql_schema()


def _load_flow_doc(raw: str) -> dict:
    return json.loads(raw or "{}")


def _to_flow_out(row: IvrFlowRow) -> IvrFlowOut:
    return IvrFlowOut(
        flow_id=row.flow_id,
        name=row.name,
        description=row.description,
        channel="voice",
        queue_id=row.queue_id,
        version=row.version,
        is_active=row.is_active,
        entry_node_id=row.entry_node_id,
        flow_json=_load_flow_doc(row.flow_json),
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _to_session_out(row: IvrSessionRow) -> IvrSessionOut:
    return IvrSessionOut(
        session_id=row.session_id,
        call_id=row.call_id,
        interaction_id=row.interaction_id,
        flow_id=row.flow_id,
        queue_id=row.queue_id,
        current_node_id=row.current_node_id,
        entered_digits=json.loads(row.entered_digits_json or "[]"),
        status=row.status,
        outcome_code=row.outcome_code,
        resolved_queue_id=row.resolved_queue_id,
        resolved_queue_code=row.resolved_queue_code,
        completed_at=row.completed_at,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _trusted_runtime_subjects() -> set[str]:
    raw = os.getenv("IVR_RUNTIME_TRUSTED_SERVICE_SUBJECTS", "svc:asterisk-bridge")
    return {item.strip() for item in raw.split(",") if item.strip()}


def _is_trusted_runtime_actor(actor: dict) -> bool:
    return (
        actor.get("auth_source") == "service"
        and str(actor.get("sub") or "").strip() in _trusted_runtime_subjects()
    )


def _require_runtime_actor(actor: dict = Depends(get_actor)) -> dict:
    if actor.get("auth_source") == "service":
        if _is_trusted_runtime_actor(actor):
            return actor
        raise HTTPException(status_code=403, detail="Untrusted service subject")
    if actor.get("role") in {Role.ADMIN.value, Role.SUPERVISOR.value}:
        return actor
    raise HTTPException(status_code=403, detail="Insufficient role")


def _validate_digit(value: str) -> str:
    digit = value.strip()
    if len(digit) != 1 or digit not in "0123456789":
        raise HTTPException(status_code=400, detail="Digit must be a single character 0-9")
    return digit


def _validate_flow_document(entry_node_id: str, flow_doc: dict) -> dict[str, dict]:
    nodes = flow_doc.get("nodes")
    if not isinstance(nodes, list) or not nodes:
        raise HTTPException(status_code=400, detail="flow_json.nodes must be a non-empty list")

    node_map: dict[str, dict] = {}
    for node in nodes:
        if not isinstance(node, dict):
            raise HTTPException(status_code=400, detail="Each flow node must be an object")
        node_id = str(node.get("node_id") or "").strip()
        if not node_id:
            raise HTTPException(status_code=400, detail="Each flow node requires node_id")
        if node_id in node_map:
            raise HTTPException(status_code=400, detail=f"Duplicate node_id: {node_id}")
        node_map[node_id] = node

    if entry_node_id not in node_map:
        raise HTTPException(status_code=400, detail="entry_node_id must reference an existing node")

    for node_id, node in node_map.items():
        is_terminal = bool(node.get("is_terminal"))
        options = node.get("options", [])
        prompt_audio_key = node.get("prompt_audio_key")
        prompt_sequence = node.get("prompt_sequence")
        if not isinstance(options, list):
            raise HTTPException(status_code=400, detail=f"Node {node_id} options must be a list")
        if prompt_audio_key is not None and not str(prompt_audio_key).strip():
            raise HTTPException(status_code=400, detail=f"Node {node_id} prompt_audio_key must be non-empty when set")
        if prompt_sequence is not None:
            if not isinstance(prompt_sequence, list) or not prompt_sequence:
                raise HTTPException(status_code=400, detail=f"Node {node_id} prompt_sequence must be a non-empty list")
            for index, prompt in enumerate(prompt_sequence, start=1):
                if not isinstance(prompt, dict):
                    raise HTTPException(
                        status_code=400,
                        detail=f"Node {node_id} prompt_sequence item {index} must be an object",
                    )
                prompt_key = str(prompt.get("prompt_audio_key") or "").strip()
                if not prompt_key:
                    raise HTTPException(
                        status_code=400,
                        detail=f"Node {node_id} prompt_sequence item {index} requires prompt_audio_key",
                    )
                prompt_text = prompt.get("prompt_text")
                if prompt_text is not None and not str(prompt_text).strip():
                    raise HTTPException(
                        status_code=400,
                        detail=f"Node {node_id} prompt_sequence item {index} prompt_text must be non-empty when set",
                    )
                prompt_language = prompt.get("language")
                if prompt_language is not None and not str(prompt_language).strip():
                    raise HTTPException(
                        status_code=400,
                        detail=f"Node {node_id} prompt_sequence item {index} language must be non-empty when set",
                    )

        seen_digits: set[str] = set()
        for option in options:
            if not isinstance(option, dict):
                raise HTTPException(status_code=400, detail=f"Node {node_id} option must be an object")
            digit = _validate_digit(str(option.get("digit") or ""))
            if digit in seen_digits:
                raise HTTPException(status_code=400, detail=f"Duplicate digit {digit} in node {node_id}")
            seen_digits.add(digit)
            target_node_id = str(option.get("target_node_id") or "").strip()
            if target_node_id not in node_map:
                raise HTTPException(status_code=400, detail=f"Option target {target_node_id} is missing")

        for ref_name in ("invalid_target_node_id", "no_input_target_node_id"):
            target = node.get(ref_name)
            if target is not None and str(target).strip() not in node_map:
                raise HTTPException(status_code=400, detail=f"{ref_name} for node {node_id} is invalid")

        if is_terminal:
            if not str(node.get("outcome_code") or "").strip():
                raise HTTPException(status_code=400, detail=f"Terminal node {node_id} requires outcome_code")
            if not str(node.get("resolved_queue_id") or "").strip():
                raise HTTPException(status_code=400, detail=f"Terminal node {node_id} requires resolved_queue_id")
            if not str(node.get("resolved_queue_code") or "").strip():
                raise HTTPException(status_code=400, detail=f"Terminal node {node_id} requires resolved_queue_code")
        elif not options:
            raise HTTPException(status_code=400, detail=f"Node {node_id} requires at least one option")

    return node_map


def _deactivate_other_flows(session, queue_id: str, *, keep_flow_id: str | None = None) -> None:
    rows = session.execute(
        select(IvrFlowRow).where(
            IvrFlowRow.queue_id == queue_id,
            IvrFlowRow.channel == "voice",
            IvrFlowRow.is_active.is_(True),
        )
    ).scalars().all()
    now = utc_now_iso()
    for row in rows:
        if keep_flow_id and row.flow_id == keep_flow_id:
            continue
        row.is_active = False
        row.updated_at = now


def _get_flow(session, flow_id: str) -> IvrFlowRow:
    row = session.execute(select(IvrFlowRow).where(IvrFlowRow.flow_id == flow_id)).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="IVR flow not found")
    return row


def _get_session_row(session, session_id: str) -> IvrSessionRow:
    row = session.execute(select(IvrSessionRow).where(IvrSessionRow.session_id == session_id)).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="IVR session not found")
    return row


def _push_timeline(session, interaction_id: str | None, action: str, metadata: dict | None = None) -> None:
    if not interaction_id:
        return
    session.add(
        InteractionTimeline(
            interaction_id=interaction_id,
            timestamp=utc_now_iso(),
            action=action,
            metadata_json=json.dumps(metadata or {}, ensure_ascii=False),
        )
    )


def _emit_ivr_completed_event(session, row: IvrSessionRow, *, terminal_node_id: str) -> None:
    session.add(
        VoiceEventRow(
            event_id=new_id("vev"),
            event_type="ivr.completed",
            call_id=row.call_id,
            interaction_id=row.interaction_id,
            payload_json=json.dumps(
                {
                    "session_id": row.session_id,
                    "flow_id": row.flow_id,
                    "outcome_code": row.outcome_code,
                    "resolved_queue_id": row.resolved_queue_id,
                    "digits": json.loads(row.entered_digits_json or "[]"),
                    "terminal_node_id": terminal_node_id,
                },
                ensure_ascii=False,
            ),
            created_at=utc_now_iso(),
        )
    )


def _load_flow_runtime(session, row: IvrSessionRow | IvrFlowRow) -> tuple[IvrFlowRow, dict[str, dict]]:
    flow = row if isinstance(row, IvrFlowRow) else _get_flow(session, row.flow_id)
    flow_doc = _load_flow_doc(flow.flow_json)
    node_map = _validate_flow_document(flow.entry_node_id, flow_doc)
    return flow, node_map


def _complete_session(
    session,
    row: IvrSessionRow,
    *,
    next_node: dict,
    next_node_id: str,
    entered_digits: list[str],
    completed_at: str,
) -> None:
    row.status = "completed"
    row.outcome_code = str(next_node.get("outcome_code") or "").strip() or None
    row.resolved_queue_id = str(next_node.get("resolved_queue_id") or "").strip() or None
    row.resolved_queue_code = str(next_node.get("resolved_queue_code") or "").strip() or None
    row.completed_at = completed_at
    _emit_ivr_completed_event(session, row, terminal_node_id=next_node_id)
    _push_timeline(
        session,
        row.interaction_id,
        "ivr.completed",
        {
            "session_id": row.session_id,
            "flow_id": row.flow_id,
            "outcome_code": row.outcome_code,
            "resolved_queue_id": row.resolved_queue_id,
        },
    )
    if event_bus_enabled():
        append_outbox_event(
            session,
            event_type="ivr.completed",
            producer_service="ivr-service",
            entity_type="ivr_session",
            entity_id=row.session_id,
            correlation_id=row.interaction_id or row.call_id,
            payload={
                "event": "ivr.completed",
                "call_id": row.call_id,
                "interaction_id": row.interaction_id,
                "session_id": row.session_id,
                "flow_id": row.flow_id,
                "outcome_code": row.outcome_code,
                "resolved_queue_id": row.resolved_queue_id,
                "digits": entered_digits,
                "terminal_node_id": next_node_id,
            },
        )


def _session_response(row: IvrSessionRow, current_node: dict | None) -> dict:
    return {
        "session": _to_session_out(row).model_dump(),
        "current_node": current_node,
        "completed": row.status == "completed",
    }


@app.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
    return HealthResponse(status="ok", service="ivr-service")


@app.post("/ivr/flows", response_model=IvrFlowOut)
def create_flow(
    payload: IvrFlowCreate,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> IvrFlowOut:
    session = get_session()
    try:
        _validate_flow_document(payload.entry_node_id, payload.flow_json)
        if payload.is_active:
            _deactivate_other_flows(session, payload.queue_id)
        now = utc_now_iso()
        row = IvrFlowRow(
            flow_id=new_id("ivr"),
            name=payload.name,
            description=payload.description,
            channel="voice",
            queue_id=payload.queue_id,
            version=1,
            is_active=payload.is_active,
            entry_node_id=payload.entry_node_id,
            flow_json=json.dumps(payload.flow_json, ensure_ascii=False),
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        return _to_flow_out(row)
    finally:
        session.close()


@app.get("/ivr/flows", response_model=list[IvrFlowOut])
def list_flows(
    queue_id: str | None = None,
    active_only: bool = False,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> list[IvrFlowOut]:
    session = get_session()
    try:
        stmt = select(IvrFlowRow).order_by(IvrFlowRow.id.desc())
        if queue_id:
            stmt = stmt.where(IvrFlowRow.queue_id == queue_id)
        if active_only:
            stmt = stmt.where(IvrFlowRow.is_active.is_(True))
        rows = session.execute(stmt).scalars().all()
        return [_to_flow_out(row) for row in rows]
    finally:
        session.close()


@app.get("/ivr/flows/{flow_id}", response_model=IvrFlowOut)
def get_flow(
    flow_id: str,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> IvrFlowOut:
    session = get_session()
    try:
        return _to_flow_out(_get_flow(session, flow_id))
    finally:
        session.close()


@app.patch("/ivr/flows/{flow_id}", response_model=IvrFlowOut)
def update_flow(
    flow_id: str,
    payload: IvrFlowUpdate,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> IvrFlowOut:
    session = get_session()
    try:
        row = _get_flow(session, flow_id)
        flow_doc = _load_flow_doc(row.flow_json)
        changed = False

        if payload.name is not None and payload.name != row.name:
            row.name = payload.name
            changed = True
        if payload.description is not None and payload.description != row.description:
            row.description = payload.description
            changed = True

        candidate_entry = payload.entry_node_id if payload.entry_node_id is not None else row.entry_node_id
        if payload.flow_json is not None:
            flow_doc = payload.flow_json
            changed = True
        _validate_flow_document(candidate_entry, flow_doc)

        if candidate_entry != row.entry_node_id:
            row.entry_node_id = candidate_entry
            changed = True
        if payload.flow_json is not None:
            row.flow_json = json.dumps(flow_doc, ensure_ascii=False)
        if payload.is_active is not None and payload.is_active != row.is_active:
            row.is_active = payload.is_active
            changed = True

        if row.is_active:
            _deactivate_other_flows(session, row.queue_id, keep_flow_id=row.flow_id)

        if changed:
            row.version += 1
            row.updated_at = utc_now_iso()

        session.commit()
        session.refresh(row)
        return _to_flow_out(row)
    finally:
        session.close()


@app.post("/ivr/flows/{flow_id}/activate", response_model=IvrFlowOut)
def activate_flow(
    flow_id: str,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> IvrFlowOut:
    session = get_session()
    try:
        row = _get_flow(session, flow_id)
        _deactivate_other_flows(session, row.queue_id, keep_flow_id=row.flow_id)
        row.is_active = True
        row.updated_at = utc_now_iso()
        session.commit()
        session.refresh(row)
        return _to_flow_out(row)
    finally:
        session.close()


@app.post("/ivr/sessions/start")
def start_session(
    payload: IvrSessionStartIn,
    _: dict = Depends(_require_runtime_actor),
) -> dict:
    session = get_session()
    try:
        flow = session.execute(
            select(IvrFlowRow).where(
                IvrFlowRow.queue_id == payload.queue_id,
                IvrFlowRow.channel == "voice",
                IvrFlowRow.is_active.is_(True),
            )
        ).scalar_one_or_none()
        if not flow:
            raise HTTPException(status_code=404, detail="No active IVR flow for queue")

        _, node_map = _load_flow_runtime(session, flow)
        now = utc_now_iso()
        row = IvrSessionRow(
            session_id=new_id("ivs"),
            call_id=payload.call_id,
            interaction_id=payload.interaction_id,
            flow_id=flow.flow_id,
            queue_id=payload.queue_id,
            current_node_id=flow.entry_node_id,
            entered_digits_json="[]",
            status="active",
            outcome_code=None,
            resolved_queue_id=None,
            resolved_queue_code=None,
            completed_at=None,
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        _push_timeline(
            session,
            payload.interaction_id,
            "ivr.session.started",
            {"session_id": row.session_id, "flow_id": flow.flow_id, "queue_id": payload.queue_id},
        )
        session.commit()
        session.refresh(row)
        return _session_response(row, node_map[row.current_node_id])
    finally:
        session.close()


@app.post("/ivr/sessions/{session_id}/dtmf")
def process_dtmf(
    session_id: str,
    payload: IvrDtmfIn,
    _: dict = Depends(_require_runtime_actor),
) -> dict:
    session = get_session()
    try:
        row = _get_session_row(session, session_id)
        if row.status != "active":
            raise HTTPException(status_code=409, detail="IVR session is not active")

        _, node_map = _load_flow_runtime(session, row)
        current_node = node_map[row.current_node_id]
        digit = _validate_digit(payload.digit)
        entered_digits = json.loads(row.entered_digits_json or "[]")

        next_node_id = None
        matched = False
        for option in current_node.get("options", []):
            if str(option.get("digit")) == digit:
                next_node_id = str(option["target_node_id"])
                matched = True
                entered_digits.append(digit)
                break

        if not matched:
            fallback_target = current_node.get("invalid_target_node_id")
            next_node_id = str(fallback_target).strip() if fallback_target else row.current_node_id

        next_node = node_map[next_node_id]
        now = utc_now_iso()
        row.current_node_id = next_node_id
        row.entered_digits_json = json.dumps(entered_digits, ensure_ascii=False)
        row.updated_at = now

        _push_timeline(
            session,
            row.interaction_id,
            "ivr.step.completed",
            {
                "session_id": row.session_id,
                "digit": digit,
                "matched": matched,
                "current_node_id": next_node_id,
            },
        )

        if next_node.get("is_terminal"):
            _complete_session(
                session,
                row,
                next_node=next_node,
                next_node_id=next_node_id,
                entered_digits=entered_digits,
                completed_at=now,
            )

        session.commit()
        session.refresh(row)
        return _session_response(row, next_node)
    finally:
        session.close()


@app.post("/ivr/sessions/{session_id}/no-input")
def process_no_input(
    session_id: str,
    _: dict = Depends(_require_runtime_actor),
) -> dict:
    session = get_session()
    try:
        row = _get_session_row(session, session_id)
        if row.status != "active":
            raise HTTPException(status_code=409, detail="IVR session is not active")

        _, node_map = _load_flow_runtime(session, row)
        current_node = node_map[row.current_node_id]
        next_node_id = str(current_node.get("no_input_target_node_id") or "").strip() or row.current_node_id
        next_node = node_map[next_node_id]
        entered_digits = json.loads(row.entered_digits_json or "[]")
        now = utc_now_iso()

        row.current_node_id = next_node_id
        row.updated_at = now

        _push_timeline(
            session,
            row.interaction_id,
            "ivr.step.no_input",
            {
                "session_id": row.session_id,
                "current_node_id": next_node_id,
            },
        )

        if next_node.get("is_terminal"):
            _complete_session(
                session,
                row,
                next_node=next_node,
                next_node_id=next_node_id,
                entered_digits=entered_digits,
                completed_at=now,
            )

        session.commit()
        session.refresh(row)
        return _session_response(row, next_node)
    finally:
        session.close()


@app.get("/ivr/sessions/{session_id}", response_model=IvrSessionOut)
def get_session_detail(
    session_id: str,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> IvrSessionOut:
    session = get_session()
    try:
        return _to_session_out(_get_session_row(session, session_id))
    finally:
        session.close()


@app.get("/ivr/sessions", response_model=list[IvrSessionOut])
def list_sessions(
    call_id: str | None = None,
    interaction_id: str | None = None,
    status: str | None = None,
    limit: int = Query(default=100, ge=1, le=500),
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> list[IvrSessionOut]:
    session = get_session()
    try:
        stmt = select(IvrSessionRow).order_by(IvrSessionRow.id.desc())
        if call_id:
            stmt = stmt.where(IvrSessionRow.call_id == call_id)
        if interaction_id:
            stmt = stmt.where(IvrSessionRow.interaction_id == interaction_id)
        if status:
            stmt = stmt.where(IvrSessionRow.status == status)
        rows = session.execute(stmt.limit(max(limit, 1))).scalars().all()
        return [_to_session_out(row) for row in rows]
    finally:
        session.close()


@app.post("/ivr/sessions/{session_id}/abandon", response_model=IvrSessionOut)
def abandon_session(
    session_id: str,
    _: dict = Depends(_require_runtime_actor),
) -> IvrSessionOut:
    session = get_session()
    try:
        row = _get_session_row(session, session_id)
        if row.status != "active":
            raise HTTPException(status_code=409, detail="IVR session is not active")
        row.status = "abandoned"
        row.updated_at = utc_now_iso()
        _push_timeline(
            session,
            row.interaction_id,
            "ivr.abandoned",
            {"session_id": row.session_id, "flow_id": row.flow_id},
        )
        session.commit()
        session.refresh(row)
        return _to_session_out(row)
    finally:
        session.close()