server-26/drb-c2-core/app/routers/calls.py

from datetime import datetime, timezone, timedelta
from fastapi import APIRouter, BackgroundTasks, HTTPException, Query, Depends
from pydantic import BaseModel
from typing import Optional
from app.internal import firestore as fstore
from app.internal.auth import (
    require_admin_token,
    require_service_or_firebase_token,
    resolve_caller_org_id,
    reprocess_limiter,
)
from app.internal.storage import gcs_uri_for_call, with_playback_url


class TranscriptUpdate(BaseModel):
    transcript: str

router = APIRouter(prefix="/calls", tags=["calls"])


@router.get("")
async def list_calls(
    node_id: Optional[str] = Query(None),
    status: Optional[str] = Query(None),
    system_id: Optional[str] = Query(None),
    decoded: dict = Depends(require_service_or_firebase_token),
):
    filters = {}
    if node_id:
        filters["node_id"] = node_id
    if status:
        filters["status"] = status
    if system_id:
        filters["system_id"] = system_id
    org_id = await resolve_caller_org_id(decoded)
    if org_id is not None:  # service key / platform admin stay unrestricted
        filters["org_id"] = org_id
    calls = await fstore.collection_list("calls", **filters)
    # audio_url is not stored — it's a short-lived signed link minted per read.
    return [with_playback_url(c) for c in calls]


@router.get("/search")
async def search_calls(
    limit: int = Query(50, ge=1, le=200),
    cursor: Optional[str] = Query(None, description="started_at of the last row of the previous page"),
    system_id: Optional[str] = Query(None),
    node_id: Optional[str] = Query(None),
    talkgroup_id: Optional[int] = Query(None),
    link: str = Query("any", pattern="^(any|orphan|linked)$"),
    transcript: str = Query("any", pattern="^(any|yes|no)$"),
    q: Optional[str] = Query(None, description="case-insensitive substring of the transcript"),
    decoded: dict = Depends(require_admin_token),
):
    """
    Paged, filterable call archive — the backend for the /calls page.

    `GET /calls` returns every call in one unordered shot, which is fine for a
    node's handful of active calls and useless as an archive: no order, no
    paging, no way to find the orphans. This route is the archive read.

    Only the org scope and the started_at ordering go to Firestore, because
    that pair is the one composite index that exists (infra/firestore/
    firestore.indexes.json). Every other filter runs in Python over a bounded
    window, the same shape admin.py's correlation debug route uses — adding a
    composite index per filter combination would be a worse trade than reading
    10x the page and discarding most of it.

    `window_exhausted` says the scan hit its cap before filling the page, so an
    empty result means "not in this window", not "none exist".
    """
    org_id = await resolve_caller_org_id(decoded)
    if org_id is None:
        # resolve_caller_org_id lets platform admins see every org (server-26#4).
        # A new browse surface shouldn't widen that, so fall back to the
        # caller's own org claim when they have one.
        org_id = decoded.get("org_id")
    if not org_id:
        raise HTTPException(403, "No organization scope for this caller.")

    window = max(limit * 10, 200)
    rows = await fstore.collection_where(
        "calls",
        [("org_id", "==", org_id)],
        order_by=[("started_at", "DESCENDING")],
        limit_to=window,
        start_after={"started_at": cursor} if cursor else None,
    )

    needle = (q or "").strip().lower()

    def _keep(c: dict) -> bool:
        if system_id and c.get("system_id") != system_id:
            return False
        if node_id and c.get("node_id") != node_id:
            return False
        if talkgroup_id is not None and c.get("talkgroup_id") != talkgroup_id:
            return False
        linked = bool(c.get("incident_ids") or c.get("incident_id"))
        if link == "orphan" and linked:
            return False
        if link == "linked" and not linked:
            return False
        text = c.get("transcript_corrected") or c.get("transcript") or ""
        if transcript == "yes" and not text:
            return False
        if transcript == "no" and text:
            return False
        if needle and needle not in text.lower():
            return False
        return True

    matches = [c for c in rows if _keep(c)]
    page = matches[:limit]

    # Cursor advances over the SCANNED window, not the filtered page — otherwise
    # a page whose last match sits early in the window would re-scan everything
    # after it on the next request and loop forever on a sparse filter.
    next_cursor = None
    if len(rows) == window:
        last_scanned = rows[-1].get("started_at")
        next_cursor = last_scanned.isoformat() if hasattr(last_scanned, "isoformat") else last_scanned

    return {
        "calls": [with_playback_url(c) for c in page],
        "next_cursor": next_cursor,
        "scanned": len(rows),
        "matched": len(matches),
        "window_exhausted": len(rows) == window,
    }


@router.get("/{call_id}")
async def get_call(call_id: str, decoded: dict = Depends(require_service_or_firebase_token)):
    call = await fstore.doc_get("calls", call_id)
    if not call:
        raise HTTPException(404, f"Call '{call_id}' not found.")
    org_id = await resolve_caller_org_id(decoded)
    if org_id is not None and call.get("org_id") != org_id:
        raise HTTPException(404, f"Call '{call_id}' not found.")
    return with_playback_url(call)


@router.post("/{call_id}/reprocess")
async def reprocess_call(
    call_id: str,
    background_tasks: BackgroundTasks,
    _: dict = Depends(require_admin_token),
):
    """
    Re-run the full intelligence pipeline (transcription -> extraction ->
    correlation) for a call. Admin-only (SAAS_PLAN.md B2c) — this was
    previously gated only by "any valid Firebase token", which meant any
    signed-in viewer could loop it and burn the owner's OpenAI/Gemini
    credits (DEFERRED.md, calls.py:42). The rate limiter below is a second
    guard against the same thing happening from a compromised/careless
    admin session, not the primary fix.
    """
    call = await fstore.doc_get("calls", call_id)
    if not call:
        raise HTTPException(404, f"Call '{call_id}' not found.")
    reprocess_limiter.check(call_id)

    from app.routers.upload import _run_intelligence_pipeline

    gcs_uri = gcs_uri_for_call(call)

    background_tasks.add_task(
        _run_intelligence_pipeline,
        call_id=call_id,
        node_id=call.get("node_id"),
        system_id=call.get("system_id"),
        talkgroup_id=call.get("talkgroup_id"),
        talkgroup_name=call.get("talkgroup_name"),
        gcs_uri=gcs_uri,
    )
    return {"ok": True, "call_id": call_id}


@router.post("/close-stale")
async def close_stale_calls(
    older_than_minutes: int = Query(30, ge=1, le=1440, description="Close active calls started more than this many minutes ago."),
    dry_run: bool = Query(False, description="If true, return what would be closed without writing."),
    _: dict = Depends(require_admin_token),
):
    """
    Find and close calls stuck in 'active' status — e.g. because a node rebooted
    before sending an end-call event.  Returns the list of affected call IDs.
    """
    cutoff = datetime.now(timezone.utc) - timedelta(minutes=older_than_minutes)
    active_calls = await fstore.collection_list("calls", status="active")

    stale = []
    for call in active_calls:
        started_raw = call.get("started_at")
        if not started_raw:
            continue
        if isinstance(started_raw, datetime):
            started = started_raw if started_raw.tzinfo else started_raw.replace(tzinfo=timezone.utc)
        else:
            try:
                started = datetime.fromisoformat(str(started_raw).replace("Z", "+00:00"))
            except Exception:
                continue
        if started < cutoff:
            stale.append(call)

    if not dry_run:
        now_iso = datetime.now(timezone.utc).isoformat()
        for call in stale:
            await fstore.doc_set("calls", call["call_id"], {
                "status": "ended",
                "ended_at": now_iso,
            })

    return {
        "dry_run": dry_run,
        "older_than_minutes": older_than_minutes,
        "count": len(stale),
        "call_ids": [c["call_id"] for c in stale],
    }


@router.patch("/{call_id}/transcript")
async def patch_transcript(
    call_id: str,
    body: TranscriptUpdate,
    background_tasks: BackgroundTasks,
    _: dict = Depends(require_admin_token),
):
    """Overwrite a call's transcript and re-run intelligence extraction."""
    from app.internal.feature_flags import resolve_flags

    call = await fstore.doc_get("calls", call_id)
    if not call:
        raise HTTPException(404, f"Call '{call_id}' not found.")

    # This route is destructive before it is constructive: it wipes the call's
    # tags, severity, location, units and embedding and unlinks it from every
    # incident, on the promise that re-extraction will rebuild all of it. With
    # correlation off that promise cannot be kept, and the call would be left
    # permanently blank and orphaned while the route still answered 200.
    # Refuse before the first write rather than half-run (server-26#76).
    _, flag = await resolve_flags(call.get("system_id"))
    if not flag("correlation_enabled"):
        raise HTTPException(
            409,
            "Correlation is disabled, so the re-extraction this correction depends on "
            "cannot run. The transcript was not changed. Enable correlation and retry.",
        )

    # Save user correction as transcript_corrected; leave original transcript intact.
    # Clear stale intelligence fields so re-extraction runs fresh.
    await fstore.doc_set("calls", call_id, {
        "transcript_corrected": body.transcript,
        "tags": [],
        "severity": "unknown",
        "location": None,
        "units": [],
        "vehicles": [],
        "embedding": None,
    })
    # server-26#96/#114 review: doc_set(merge=True) can only ADD/overwrite keys
    # in a nested map, never remove one, so the fields above get cleared but a
    # prior `scenes` map would survive re-extraction forever. A call corrected
    # from 3 scenes down to 1 would keep scenes.1/scenes.2 with pre-correction
    # transcripts and incident_ids -- corrupting the exact per-scene tally #96
    # exists to make trustworthy, and re-feeding stale text into #114's
    # summarizer fix if a stale scene's incident_id still names a real
    # incident. Must be a real delete, not a merge over an empty map.
    await fstore.doc_update("calls", call_id, {"scenes": fstore.DELETE_FIELD})

    # Unlink from ALL current incidents so re-correlation starts clean.
    # Handles both old single incident_id and new incident_ids list.
    old_ids: list[str] = call.get("incident_ids") or (
        [call["incident_id"]] if call.get("incident_id") else []
    )
    for old_incident_id in old_ids:
        old_incident = await fstore.doc_get("incidents", old_incident_id)
        if old_incident:
            remaining = [c for c in (old_incident.get("call_ids") or []) if c != call_id]
            if remaining:
                await fstore.doc_set("incidents", old_incident_id, {
                    "call_ids": remaining,
                    "summary_stale": True,
                })
            else:
                await fstore.doc_set("incidents", old_incident_id, {
                    "call_ids": [],
                    "status": "resolved",
                    "resolved_at": datetime.now(timezone.utc).isoformat(),
                    "summary_stale": True,
                })
    await fstore.doc_set("calls", call_id, {"incident_ids": [], "incident_id": None})

    # Learn from the correction: diff original → corrected and add new tokens to vocabulary
    system_id = call.get("system_id")
    original_text = call.get("transcript_corrected") or call.get("transcript") or ""
    if system_id and original_text and flag("vocabulary_learning_enabled"):
        from app.internal.vocabulary_learner import learn_from_correction
        await learn_from_correction(system_id, original_text, body.transcript)

    from app.routers.upload import _run_extraction_pipeline
    background_tasks.add_task(
        _run_extraction_pipeline,
        call_id=call_id,
        node_id=call.get("node_id"),
        system_id=call.get("system_id"),
        talkgroup_id=call.get("talkgroup_id"),
        talkgroup_name=call.get("talkgroup_name"),
        transcript=body.transcript,
        segments=call.get("segments"),
        preserve_transcript_correction=True,
    )
    return {"ok": True, "call_id": call_id}