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()