call-center/services/routing_service/app.py

from __future__ import annotations

import json

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

from services.shared.core import Role, new_id, utc_now_iso
from services.shared.db import engine, get_session
from services.shared.models import HealthResponse, QueueCreate, QueueOut
from services.shared.security import require_roles
from services.shared.sql_init import init_sql_schema
from services.shared.sql_models import IvrSessionRow, Queue, RoutingCounter

app = FastAPI(title="routing-service", version="1.1.0")

init_sql_schema()

_AGENTS = {
    "voice": ["operator_a", "operator_b", "operator_c"],
    "telegram": ["operator_t1", "operator_t2"],
    "webchat": ["operator_w1"],
    "email": ["operator_e1"],
}


def _db_backend_name() -> str:
    return engine.url.get_backend_name()


def _next_agent_index(session, *, channel: str, agent_count: int) -> int:
    if _db_backend_name() != "postgresql":
        counter = session.execute(
            select(RoutingCounter).where(RoutingCounter.channel == channel)
        ).scalar_one_or_none()
        if not counter:
            counter = RoutingCounter(channel=channel, counter=0)
            session.add(counter)
            session.flush()

        idx = counter.counter % agent_count
        counter.counter += 1
        return idx

    counter_value = session.execute(
        text(
            """
            INSERT INTO routing_counters(channel, counter)
            VALUES (:channel, 1)
            ON CONFLICT (channel)
            DO UPDATE SET counter = routing_counters.counter + 1
            RETURNING counter
            """
        ),
        {"channel": channel},
    ).scalar_one()
    return (int(counter_value) - 1) % agent_count


def _parse_rules_json(raw_rules_json: str | None) -> list[dict]:
    if not raw_rules_json:
        return []
    try:
        data = json.loads(raw_rules_json)
    except (TypeError, json.JSONDecodeError):
        return []

    if isinstance(data, list):
        return [item for item in data if isinstance(item, dict)]

    if isinstance(data, dict):
        nested_rules = data.get("rules")
        if isinstance(nested_rules, list):
            return [item for item in nested_rules if isinstance(item, dict)]
        if data:
            return [data]
        return []

    return []


def _to_out(row: Queue) -> QueueOut:
    return QueueOut(
        queue_id=row.queue_id,
        name=row.name,
        description=row.description,
        rules=_parse_rules_json(row.rules_json),
        created_at=row.created_at,
    )


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


@app.post("/queues", response_model=QueueOut)
def create_queue(payload: QueueCreate, _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR))) -> QueueOut:
    session = get_session()
    try:
        row = Queue(
            queue_id=new_id("que"),
            name=payload.name,
            description=payload.description,
            rules_json=json.dumps([r.model_dump() for r in payload.rules], ensure_ascii=False),
            created_at=utc_now_iso(),
        )
        session.add(row)
        session.commit()
        session.refresh(row)
        return _to_out(row)
    finally:
        session.close()


@app.get("/queues", response_model=list[QueueOut])
def list_queues(
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> list[QueueOut]:
    session = get_session()
    try:
        rows = session.execute(select(Queue).order_by(Queue.id.desc())).scalars().all()
        return [_to_out(r) for r in rows]
    finally:
        session.close()


@app.patch("/queues/{queue_id}/rules", response_model=QueueOut)
def update_rules(
    queue_id: str,
    payload: QueueCreate,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> QueueOut:
    session = get_session()
    try:
        row = session.execute(select(Queue).where(Queue.queue_id == queue_id)).scalar_one_or_none()
        if not row:
            raise HTTPException(status_code=404, detail="Queue not found")
        row.name = payload.name
        row.description = payload.description
        row.rules_json = json.dumps([r.model_dump() for r in payload.rules], ensure_ascii=False)
        session.commit()
        return _to_out(row)
    finally:
        session.close()


@app.post("/queues/{queue_id}/route")
def route_interaction(
    queue_id: str,
    channel: str,
    priority: int = 3,
    ivr_session_id: str | None = None,
    _: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> dict:
    session = get_session()
    try:
        original_queue = session.execute(select(Queue).where(Queue.queue_id == queue_id)).scalar_one_or_none()
        if not original_queue:
            raise HTTPException(status_code=404, detail="Queue not found")

        resolved_queue_id = queue_id
        ivr_flow_id = None
        ivr_outcome_code = None
        if ivr_session_id:
            ivr_session = session.execute(
                select(IvrSessionRow).where(IvrSessionRow.session_id == ivr_session_id)
            ).scalar_one_or_none()
            if not ivr_session:
                raise HTTPException(status_code=404, detail="IVR session not found")
            if ivr_session.status != "completed":
                raise HTTPException(status_code=400, detail="IVR session is not completed")
            if ivr_session.queue_id != queue_id:
                raise HTTPException(status_code=400, detail="IVR session does not belong to the requested queue")
            resolved_queue_id = ivr_session.resolved_queue_id or queue_id
            ivr_flow_id = ivr_session.flow_id
            ivr_outcome_code = ivr_session.outcome_code

        queue = session.execute(select(Queue).where(Queue.queue_id == resolved_queue_id)).scalar_one_or_none()
        if not queue:
            raise HTTPException(status_code=404, detail="Queue not found")

        agents = _AGENTS.get(channel)
        if not agents:
            raise HTTPException(status_code=400, detail="No agents for channel")

        idx = _next_agent_index(session, channel=channel, agent_count=len(agents))

        sla_seconds = 30
        for rule in _parse_rules_json(queue.rules_json):
            if rule.get("channel") == channel and int(rule.get("priority", 3)) == priority:
                sla_seconds = int(rule.get("sla_seconds", 30))
                break

        session.commit()
        response = {
            "queue_id": resolved_queue_id,
            "channel": channel,
            "priority": priority,
            "assignee": agents[idx],
            "sla_seconds": sla_seconds,
        }
        if ivr_session_id:
            response.update(
                {
                    "original_queue_id": queue_id,
                    "resolved_queue_id": resolved_queue_id,
                    "ivr_flow_id": ivr_flow_id,
                    "ivr_outcome_code": ivr_outcome_code,
                    "ivr_session_id": ivr_session_id,
                }
            )
        return response
    finally:
        session.close()


@app.delete("/queues/{queue_id}")
def delete_queue(
    queue_id: str,
    _: dict = Depends(require_roles(Role.ADMIN)),
) -> dict:
    session = get_session()
    try:
        row = session.execute(select(Queue).where(Queue.queue_id == queue_id)).scalar_one_or_none()
        if not row:
            raise HTTPException(status_code=404, detail="Queue not found")
        session.delete(row)
        session.commit()
        return {"queue_id": queue_id, "deleted": True}
    finally:
        session.close()