server-26/drb-c2-core/app/internal/alerter.py

"""
Alert dispatch engine.

Loads enabled alert rules from Firestore and checks each one against the call's
talkgroup ID, tags, and transcript.  On a match:
  1. Creates an AlertEvent document in Firestore.
  2. Optionally POSTs a Discord webhook message if the rule has one configured.

Raw transcript text is withheld from both by default -- see _snippet_allowed
and server-26#85.

Never raises — failures are logged as warnings so the pipeline always completes.
"""
import uuid
from datetime import datetime, timezone
from typing import Optional
from app.config import settings
from app.internal.logger import logger
from app.internal import firestore as fstore


async def check_and_dispatch(
    call_id: str,
    node_id: str,
    talkgroup_id: Optional[int],
    talkgroup_name: Optional[str],
    tags: list[str],
    transcript: Optional[str],
) -> None:
    """
    Check all enabled alert rules and fire events for any that match this call.
    """
    try:
        # Scoped to the call's own org — an unscoped query here would let an
        # alert rule created by one org fire (and POST its Discord webhook)
        # on another org's radio traffic. org_id is resolved from the call
        # doc rather than threaded through as a new parameter, since every
        # caller of check_and_dispatch already has call_id and the call doc
        # is the single source of truth for a call's org once mqtt_handler.py
        # / upload.py have stamped it. None only for a call from a node that
        # predates tenancy and hasn't been through the backfill script yet —
        # such calls fall back to the pre-tenancy behaviour of checking
        # every rule regardless of org.
        call_doc = await fstore.doc_get("calls", call_id)
        org_id = (call_doc or {}).get("org_id")
        if org_id is not None:
            rules = await fstore.collection_list("alert_rules", enabled=True, org_id=org_id)
        else:
            rules = await fstore.collection_list("alert_rules", enabled=True)
    except Exception as e:
        logger.warning(f"Alerter: could not load rules: {e}")
        return

    # Loop-invariant: every rule here belongs to the same org, so the opt-in is
    # resolved once rather than per match.
    snippet_allowed = await _snippet_allowed(org_id)

    for rule in rules:
        matched_keywords = _match_rule(rule, talkgroup_id, tags, transcript)
        if not matched_keywords:
            continue

        alert_id = str(uuid.uuid4())
        snippet = _snippet(transcript) if snippet_allowed else None
        now = datetime.now(timezone.utc).isoformat()
        event = {
            "alert_id": alert_id,
            "org_id": org_id,
            "rule_id": rule.get("rule_id", ""),
            "rule_name": rule.get("name", ""),
            "call_id": call_id,
            "node_id": node_id,
            "talkgroup_id": talkgroup_id,
            "talkgroup_name": talkgroup_name or "",
            "matched_keywords": matched_keywords,
            "transcript_snippet": snippet,
            "triggered_at": now,
            "acknowledged": False,
        }

        try:
            await fstore.doc_set("alert_events", alert_id, event, merge=False)
            logger.info(
                f"Alert fired: rule='{rule.get('name')}' call={call_id} "
                f"keywords={matched_keywords}"
            )
        except Exception as e:
            logger.warning(f"Alerter: could not save alert event: {e}")
            continue

        webhook_url = rule.get("discord_webhook")
        if webhook_url:
            await _post_webhook(webhook_url, rule.get("name", ""), talkgroup_name, matched_keywords, snippet)


async def _snippet_allowed(org_id: Optional[str]) -> bool:
    """
    Whether raw transcript text may be attached to an alert (server-26#85).

    Two gates, both of which must be open:

    1. ``settings.alert_transcript_snippet_enabled`` -- the operator switch,
       default False, set from the environment and unreachable from the app.
    2. ``alert_snippet_opt_in`` on the org document -- the customer's own
       explicit, contractual opt-in.

    Gate 1 exists because gate 2 alone is not a real control: the frontend
    reads and (per the Firestore rules, not ``auth.py``) can write org state
    directly from the browser, so an org owner could otherwise opt themselves
    into receiving person names lifted from live public-safety traffic. Board
    minutes #42 suppress names on every surface until E&O is bound.

    Fails CLOSED on any error, and on a call with no org (a pre-tenancy node
    that has not been backfilled), because the cost of wrongly withholding a
    snippet is a less informative alert and the cost of wrongly emitting one
    is unrecallable disclosure to a third party.
    """
    if not settings.alert_transcript_snippet_enabled:
        return False
    if not org_id:
        return False
    try:
        org = await fstore.doc_get("organizations", org_id)
    except Exception as e:
        logger.warning(
            f"Alerter: could not read snippet opt-in for org={org_id}, "
            f"withholding transcript: {e}"
        )
        return False
    return bool((org or {}).get("alert_snippet_opt_in"))


def _match_rule(
    rule: dict,
    talkgroup_id: Optional[int],
    tags: list[str],
    transcript: Optional[str],
) -> list[str]:
    """Return list of matched keywords/reasons, or empty list if no match."""
    matched: list[str] = []

    # Talkgroup ID match
    rule_tg_ids = rule.get("talkgroup_ids", [])
    if rule_tg_ids and talkgroup_id is not None and talkgroup_id in rule_tg_ids:
        matched.append(f"talkgroup:{talkgroup_id}")

    # Keyword match against tags + transcript
    rule_keywords = [kw.lower() for kw in rule.get("keywords", [])]
    for kw in rule_keywords:
        if kw in tags:
            matched.append(kw)
        elif transcript and kw in transcript.lower():
            matched.append(kw)

    return matched


def _snippet(transcript: Optional[str], max_len: int = 200) -> Optional[str]:
    if not transcript:
        return None
    return transcript[:max_len] + ("…" if len(transcript) > max_len else "")


async def _post_webhook(
    url: str,
    rule_name: str,
    talkgroup_name: Optional[str],
    matched_keywords: list[str],
    snippet: Optional[str],
) -> None:
    try:
        import httpx
        tg_label = talkgroup_name or "Unknown"
        kw_str = ", ".join(matched_keywords)
        body = (
            f"**Alert: {rule_name}**\n"
            f"Talkgroup: {tg_label}\n"
            f"Matched: {kw_str}"
        )
        if snippet:
            body += f"\n> {snippet}"
        async with httpx.AsyncClient(timeout=5.0) as client:
            await client.post(url, json={"content": body})
    except Exception as e:
        logger.warning(f"Alerter: Discord webhook POST failed: {e}")