Convert call routes and session runtime to async PostgreSQL

Author: lakshyashishir

backend/app/api/routes/call.py

@@ -2,11 +2,13 @@
 from typing import Optional
 
 import httpx
-from fastapi import APIRouter, HTTPException, Request
+from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import Response
 from pydantic import BaseModel, Field
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
+from app.database import get_db
 from app.integrations.phone import normalize_phone_number
 from app.integrations.twilio_client import (
     TwilioConfigError,
@@ -15,8 +17,9 @@
     start_outbound_call,
 )
 from app.models import CompleteResponse, Session, Worker, new_id, utc_now_iso
+from app.services import store
 from app.services.session_runtime import complete_session, create_session, require_session, update_session
-from app.services.store import call_session_index, sessions, workers
+from app.services.store import call_session_index
 
 router = APIRouter(tags=["call"])
 
@@ -35,23 +38,23 @@ class TwilioStatusResponse(BaseModel):
     status: Optional[str] = None
 
 
-def _find_worker_by_phone(phone_number: str) -> Optional[Worker]:
+async def _find_worker_by_phone(db: AsyncSession, phone_number: str) -> Optional[Worker]:
     normalized = normalize_phone_number(phone_number)
-    for worker in workers.values():
-        if normalize_phone_number(worker.phone_number or "") == normalized:
-            return worker
-    return None
+    if not normalized:
+        return None
+    return await store.get_worker_by_phone(db, normalized)
 
 
-def _upsert_worker(
+async def _upsert_worker(
+    db: AsyncSession,
     *,
     phone_number: Optional[str],
     worker_name: Optional[str],
     specialization: Optional[str],
     experience_years: int,
 ) -> Worker:
     normalized_phone = normalize_phone_number(phone_number)
-    existing = _find_worker_by_phone(normalized_phone or "") if normalized_phone else None
+    existing = await _find_worker_by_phone(db, normalized_phone or "") if normalized_phone else None
     if existing:
         updates = {}
         if worker_name:
@@ -61,7 +64,7 @@ def _upsert_worker(
         updates["experience_years"] = max(existing.experience_years, experience_years)
         if updates:
             updated = existing.model_copy(update=updates)
-            workers[existing.id] = updated
+            await store.update_worker(db, updated)
             return updated
         return existing
 
@@ -73,7 +76,7 @@ def _upsert_worker(
         phone_number=normalized_phone,
         created_at=utc_now_iso(),
     )
-    workers[worker.id] = worker
+    await store.create_worker(db, worker)
     return worker
 
 
@@ -94,15 +97,20 @@ def _build_greeting_text(session: Session) -> str:
 
 
 @router.post("/calls/twilio/start")
-async def start_twilio_screening_call(payload: TwilioCallStartRequest):
-    worker = _upsert_worker(
+async def start_twilio_screening_call(
+    payload: TwilioCallStartRequest,
+    db: AsyncSession = Depends(get_db),
+):
+    worker = await _upsert_worker(
+        db,
         phone_number=payload.phone_number,
         worker_name=payload.worker_name,
         specialization=payload.specialization,
         experience_years=payload.experience_years,
     )
     assignment = (payload.assignment or settings.twilio_default_assignment).strip()
-    session, first_question = create_session(
+    session, first_question = await create_session(
+        db,
         worker,
         assignment,
         interview_mode="call",
@@ -122,7 +130,8 @@ async def start_twilio_screening_call(payload: TwilioCallStartRequest):
     except httpx.HTTPError as exc:
         raise HTTPException(status_code=502, detail=f"Twilio call start failed: {exc}") from exc
 
-    session = update_session(
+    session = await update_session(
+        db,
         session.id,
         external_call_id=outbound.get("sid"),
         external_call_status=outbound.get("status") or "queued",
@@ -143,45 +152,46 @@ async def start_twilio_screening_call(payload: TwilioCallStartRequest):
 
 
 @router.api_route("/calls/twilio/twiml", methods=["GET", "POST"])
-async def twilio_twiml(request: Request) -> Response:
+async def twilio_twiml(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+) -> Response:
     query = dict(request.query_params)
     body = dict(await request.form()) if request.method == "POST" else {}
     payload = {**query, **body}
     session_id = payload.get("session_id")
     if not session_id:
         raise HTTPException(status_code=400, detail="session_id is required")
 
-    session = require_session(str(session_id))
+    session = await require_session(db, str(session_id))
     return _twiml_response(_build_greeting_text(session))
 
 
 @router.api_route("/calls/twilio/incoming", methods=["GET", "POST"])
-async def twilio_incoming_webhook(request: Request) -> Response:
-    """
-    Incoming voice webhook used when a caller dials your Twilio number directly.
-
-    Creates a call-mode session on-the-fly and responds with TwiML that
-    speaks the phone-assessment greeting and the first interview question.
-    """
+async def twilio_incoming_webhook(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+) -> Response:
     query = dict(request.query_params)
     body = dict(await request.form()) if request.method == "POST" else {}
     payload: dict[str, object] = {**query, **body}
 
-    # Twilio sends From (caller) + CallSid on voice webhooks.
     from_number = payload.get("From") or payload.get("from") or payload.get("Caller")
     call_sid = extract_twilio_sid(payload)
     if not from_number:
         raise HTTPException(status_code=400, detail="From number is required")
 
-    worker = _upsert_worker(
+    worker = await _upsert_worker(
+        db,
         phone_number=str(from_number),
         worker_name="श्रमिक",
         specialization=None,
         experience_years=0,
     )
 
     assignment = settings.twilio_default_assignment.strip()
-    session, _ = create_session(
+    session, _ = await create_session(
+        db,
         worker,
         assignment,
         interview_mode="call",
@@ -193,14 +203,16 @@ async def twilio_incoming_webhook(request: Request) -> Response:
     )
 
     if call_sid:
-        # Enables /calls/twilio/status to resolve the session even without session_id query params.
         call_session_index[call_sid] = session.id
 
     return _twiml_response(_build_greeting_text(session))
 
 
 @router.api_route("/calls/twilio/status", methods=["GET", "POST"], response_model=TwilioStatusResponse)
-async def twilio_status_callback(request: Request) -> TwilioStatusResponse:
+async def twilio_status_callback(
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+) -> TwilioStatusResponse:
     body = await request.form() if request.method == "POST" else {}
     payload = {**dict(request.query_params), **dict(body)}
     call_sid = extract_twilio_sid(payload)
@@ -212,7 +224,7 @@ async def twilio_status_callback(request: Request) -> TwilioStatusResponse:
     resolved_session_id = None
     if session_id:
         try:
-            require_session(str(session_id))
+            await require_session(db, str(session_id))
             resolved_session_id = str(session_id)
         except HTTPException:
             resolved_session_id = None
@@ -226,18 +238,20 @@ async def twilio_status_callback(request: Request) -> TwilioStatusResponse:
     except ValueError:
         parsed_duration = None
 
-    session = update_session(
+    current = await require_session(db, resolved_session_id)
+    session = await update_session(
+        db,
         resolved_session_id,
-        external_call_id=call_sid or require_session(resolved_session_id).external_call_id,
-        external_call_status=str(status) if status else require_session(resolved_session_id).external_call_status,
-        call_duration_seconds=parsed_duration or require_session(resolved_session_id).call_duration_seconds,
-        latest_call_recording_url=str(recording_url) if recording_url else require_session(resolved_session_id).latest_call_recording_url,
+        external_call_id=call_sid or current.external_call_id,
+        external_call_status=str(status) if status else current.external_call_status,
+        call_duration_seconds=parsed_duration or current.call_duration_seconds,
+        latest_call_recording_url=str(recording_url) if recording_url else current.latest_call_recording_url,
     )
     if call_sid:
         call_session_index[call_sid] = session.id
 
     terminal_statuses = {"completed", "busy", "failed", "no-answer", "canceled"}
     if (session.external_call_status or "").lower() in terminal_statuses and session.status == "live":
-        session = complete_session(session.id, locale="hi")
+        session = await complete_session(db, session.id, locale="hi")
 
     return TwilioStatusResponse(ok=True, session_id=session.id, status=session.external_call_status)

backend/app/db_models.py

@@ -12,6 +12,7 @@ class WorkerDB(Base):
     name: Mapped[str] = mapped_column(String(80), nullable=False)
     specialization: Mapped[str] = mapped_column(String(120), nullable=False)
     experience_years: Mapped[int] = mapped_column(Integer, nullable=False)
+    phone_number: Mapped[str] = mapped_column(String(30), nullable=True)
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
         default=lambda: datetime.now(timezone.utc),
@@ -43,3 +44,10 @@ class SessionDB(Base):
     integrity_events: Mapped[dict] = mapped_column(JSON, default=list)
     current_phase: Mapped[str] = mapped_column(String(50), default="intro")
     self_ratings: Mapped[dict] = mapped_column(JSON, default=dict)
+    interview_mode: Mapped[str] = mapped_column(String(20), default="web", nullable=True)
+    call_provider: Mapped[str] = mapped_column(String(30), nullable=True)
+    call_phone_number: Mapped[str] = mapped_column(String(30), nullable=True)
+    external_call_id: Mapped[str] = mapped_column(String(100), nullable=True)
+    external_call_status: Mapped[str] = mapped_column(String(30), nullable=True)
+    call_duration_seconds: Mapped[int] = mapped_column(Integer, nullable=True)
+    latest_call_recording_url: Mapped[str] = mapped_column(Text, nullable=True)

backend/app/models.py

@@ -18,6 +18,7 @@ class Worker(BaseModel):
     name: str
     specialization: str
     experience_years: int
+    phone_number: Optional[str] = None
     created_at: str
 
 
@@ -124,6 +125,13 @@ class Session(BaseModel):
     phase0_profile: Dict[str, Any] = Field(default_factory=dict)
     phase0_completed: bool = False
     portfolio_enrichment: List[PortfolioItem] = Field(default_factory=list)
+    interview_mode: Optional[str] = "web"
+    call_provider: Optional[str] = None
+    call_phone_number: Optional[str] = None
+    external_call_id: Optional[str] = None
+    external_call_status: Optional[str] = None
+    call_duration_seconds: Optional[int] = None
+    latest_call_recording_url: Optional[str] = None
 
 
 class PriorWorkMediaRequest(BaseModel):

backend/app/services/session_runtime.py

@@ -1,4 +1,5 @@
 from fastapi import HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.agents.screening_logic import choose_opening_question, finalize_session, run_agent_turn
 from app.models import (
@@ -9,10 +10,11 @@
     new_id,
     utc_now_iso,
 )
-from app.services.store import sessions
+from app.services import store
 
 
-def create_session(
+async def create_session(
+    db: AsyncSession,
     worker: Worker,
     assignment: str,
     *,
@@ -26,7 +28,6 @@ def create_session(
     first_question = choose_opening_question(
         worker.name,
         assignment,
-        interview_mode=interview_mode,
         locale=locale,
     )
     session = Session(
@@ -48,26 +49,27 @@ def create_session(
         external_call_id=external_call_id,
         external_call_status=external_call_status,
     )
-    sessions[session.id] = session
+    await store.create_session(db, session)
     return session, first_question
 
 
-def require_session(session_id: str) -> Session:
-    session = sessions.get(session_id)
+async def require_session(db: AsyncSession, session_id: str) -> Session:
+    session = await store.get_session(db, session_id)
     if not session:
         raise HTTPException(status_code=404, detail="Session not found")
     return session
 
 
-def append_turn(
+async def append_turn(
+    db: AsyncSession,
     session_id: str,
     worker_text: str,
     *,
     locale: str = "en",
     rubric_tag: str | None = None,
     acoustic_confidence: float | None = None,
 ) -> tuple[Session, TurnResponse]:
-    session = require_session(session_id)
+    session = await require_session(db, session_id)
     if session.status != "live":
         raise HTTPException(status_code=400, detail="Session already completed")
 
@@ -95,7 +97,7 @@ def append_turn(
             "live_score": new_score,
         }
     )
-    sessions[session_id] = updated
+    await store.update_session(db, updated)
     return updated, TurnResponse(
         ai_question=result["ai_reply"],
         coach_note="",
@@ -105,17 +107,17 @@ def append_turn(
     )
 
 
-def complete_session(session_id: str, *, locale: str = "en") -> Session:
-    session = require_session(session_id)
+async def complete_session(db: AsyncSession, session_id: str, *, locale: str = "en") -> Session:
+    session = await require_session(db, session_id)
     if session.status == "completed":
         return session
     completed = finalize_session(session, locale)
-    sessions[session_id] = completed
+    await store.update_session(db, completed)
     return completed
 
 
-def update_session(session_id: str, **updates: object) -> Session:
-    session = require_session(session_id)
+async def update_session(db: AsyncSession, session_id: str, **updates: object) -> Session:
+    session = await require_session(db, session_id)
     updated = session.model_copy(update=updates)
-    sessions[session_id] = updated
+    await store.update_session(db, updated)
     return updated