| 1 | """ |
| 2 | Notification routing service — CRUD for routes, the dispatch loop, and |
| 3 | a read-only view over the dispatch log. |
| 4 | |
| 5 | The dispatch loop is the heart of the module. It's called via |
| 6 | `POST /notifications/dispatch` (Talon's after-investigation hook) and |
| 7 | walks every enabled route for the customer, filters by trigger and |
| 8 | severity, formats the message body per channel, calls the appropriate |
| 9 | dispatcher, and records the outcome in `notification_dispatch_log`. The |
| 10 | log row is what gives us idempotency — re-dispatching the same |
| 11 | (customer, alert, route, trigger) is a no-op. |
| 12 | """ |
| 13 | |
| 14 | from __future__ import annotations |
| 15 | |
| 16 | from datetime import datetime |
| 17 | from typing import List |
| 18 | from typing import Optional |
| 19 | |
| 20 | from fastapi import HTTPException |
| 21 | from loguru import logger |
| 22 | from sqlalchemy import desc |
| 23 | from sqlalchemy import select |
| 24 | from sqlalchemy import update |
| 25 | from sqlalchemy.exc import IntegrityError |
| 26 | from sqlalchemy.ext.asyncio import AsyncSession |
| 27 | |
| 28 | from app.connectors.utils import get_connector_info_from_db |
| 29 | from app.db.universal_models import CustomerNotificationRoute |
| 30 | from app.db.universal_models import CustomerShuffleIntegration |
| 31 | from app.db.universal_models import NotificationDispatchLog |
| 32 | from app.notifications.schema.notifications import SEVERITY_ORDER |
| 33 | from app.notifications.schema.notifications import DispatchOutcome |
| 34 | from app.notifications.schema.notifications import DispatchRequest |
| 35 | from app.notifications.schema.notifications import DispatchResponse |
| 36 | from app.notifications.schema.notifications import DispatchStatus |
| 37 | from app.notifications.schema.notifications import NotificationChannel |
| 38 | from app.notifications.schema.notifications import NotificationRouteCreate |
| 39 | from app.notifications.schema.notifications import NotificationRouteUpdate |
| 40 | from app.notifications.schema.notifications import NotificationTrigger |
| 41 | from app.notifications.schema.notifications import ShuffleApp |
| 42 | from app.notifications.schema.notifications import ShuffleIntegrationCreate |
| 43 | from app.notifications.schema.notifications import ShuffleIntegrationUpdate |
| 44 | from app.notifications.schema.notifications import ShuffleOrg |
| 45 | from app.notifications.services.dispatchers import dispatch_shuffle |
| 46 | from app.notifications.services.dispatchers import ( |
| 47 | list_shuffle_apps as shuffle_apps_client, |
| 48 | ) |
| 49 | from app.notifications.services.dispatchers import ( |
| 50 | list_shuffle_orgs as shuffle_orgs_client, |
| 51 | ) |
| 52 | from app.notifications.services.dispatchers import ( |
| 53 | verify_shuffle_org as verify_shuffle_org_client, |
| 54 | ) |
| 55 | |
| 56 | # Name of the Shuffle row in CoPilot's connectors table. The |
| 57 | # `connector_url` (Shuffle base URL) and `connector_api_key` (admin |
| 58 | # Bearer token) are read fresh on every dispatch so a key rotation |
| 59 | # takes effect without restarting the backend. |
| 60 | _SHUFFLE_CONNECTOR_NAME = "Shuffle" |
| 61 | |
| 62 | |
| 63 | async def _get_shuffle_connector(session: AsyncSession) -> tuple[str, str]: |
| 64 | """Fetch (base_url, api_key) for the Shuffle connector. Raises |
| 65 | HTTPException if the connector row is missing or unconfigured — |
| 66 | surfaces a clear 4xx in the dispatch endpoint instead of a generic |
| 67 | 500 when an admin forgets to configure Shuffle.""" |
| 68 | info = await get_connector_info_from_db(_SHUFFLE_CONNECTOR_NAME, session) |
| 69 | if not info: |
| 70 | raise HTTPException( |
| 71 | status_code=503, |
| 72 | detail=( |
| 73 | "Shuffle connector is not configured in CoPilot. " |
| 74 | "Add the Shuffle connector with a valid API key before " |
| 75 | "creating Shuffle-channel notification routes." |
| 76 | ), |
| 77 | ) |
| 78 | api_key = info.get("connector_api_key") or "" |
| 79 | base_url = info.get("connector_url") or "https://shuffler.io" |
| 80 | if not api_key: |
| 81 | raise HTTPException( |
| 82 | status_code=503, |
| 83 | detail="Shuffle connector is configured but has no API key set.", |
| 84 | ) |
| 85 | return (base_url, api_key) |
| 86 | |
| 87 | |
| 88 | # --------------------------------------------------------------------------- |
| 89 | # CRUD |
| 90 | # --------------------------------------------------------------------------- |
| 91 | |
| 92 | |
| 93 | async def list_routes(customer_code: str, session: AsyncSession) -> List[CustomerNotificationRoute]: |
| 94 | """All routes for a customer, newest-first. UI list source.""" |
| 95 | result = await session.execute( |
| 96 | select(CustomerNotificationRoute) |
| 97 | .where(CustomerNotificationRoute.customer_code == customer_code) |
| 98 | .order_by(desc(CustomerNotificationRoute.created_at)), |
| 99 | ) |
| 100 | return result.scalars().all() |
| 101 | |
| 102 | |
| 103 | async def get_route(route_id: int, customer_code: str, session: AsyncSession) -> CustomerNotificationRoute: |
| 104 | """Single route, scoped by customer to keep the tenant boundary |
| 105 | explicit at lookup time.""" |
| 106 | result = await session.execute( |
| 107 | select(CustomerNotificationRoute).where( |
| 108 | CustomerNotificationRoute.id == route_id, |
| 109 | CustomerNotificationRoute.customer_code == customer_code, |
| 110 | ), |
| 111 | ) |
| 112 | route = result.scalars().first() |
| 113 | if not route: |
| 114 | raise HTTPException(status_code=404, detail="Route not found") |
| 115 | return route |
| 116 | |
| 117 | |
| 118 | async def create_route( |
| 119 | customer_code: str, |
| 120 | payload: NotificationRouteCreate, |
| 121 | created_by: Optional[str], |
| 122 | session: AsyncSession, |
| 123 | ) -> CustomerNotificationRoute: |
| 124 | # Shuffle-channel sanity check: the integration must exist AND |
| 125 | # belong to the same customer. Pydantic validators caught the "is |
| 126 | # the field present" question; this catches the cross-tenant version. |
| 127 | if payload.channel == NotificationChannel.SHUFFLE: |
| 128 | await _ensure_integration_belongs_to_customer(payload.shuffle_integration_id, customer_code, session) |
| 129 | |
| 130 | route = CustomerNotificationRoute( |
| 131 | customer_code=customer_code, |
| 132 | name=payload.name, |
| 133 | trigger=payload.trigger.value, |
| 134 | channel=payload.channel.value, |
| 135 | destination=payload.destination, |
| 136 | min_severity=payload.min_severity.value, |
| 137 | format_template=payload.format_template, |
| 138 | enabled=payload.enabled, |
| 139 | created_by=created_by, |
| 140 | shuffle_integration_id=payload.shuffle_integration_id, |
| 141 | shuffle_app_id=payload.shuffle_app_id, |
| 142 | shuffle_app_name=payload.shuffle_app_name, |
| 143 | ) |
| 144 | session.add(route) |
| 145 | await session.commit() |
| 146 | await session.refresh(route) |
| 147 | return route |
| 148 | |
| 149 | |
| 150 | async def update_route( |
| 151 | route_id: int, |
| 152 | customer_code: str, |
| 153 | payload: NotificationRouteUpdate, |
| 154 | session: AsyncSession, |
| 155 | ) -> CustomerNotificationRoute: |
| 156 | route = await get_route(route_id, customer_code, session) |
| 157 | |
| 158 | # Pydantic v1 vs v2 parity — exclude_unset returns only the fields |
| 159 | # the client actually sent so a PATCH that omits `enabled` doesn't |
| 160 | # accidentally re-flag it. |
| 161 | data = payload.model_dump(exclude_unset=True) |
| 162 | |
| 163 | # If the PATCH switches the channel to Shuffle (or re-points an |
| 164 | # existing Shuffle route at a different integration), the new |
| 165 | # integration must belong to the same customer. |
| 166 | new_integration_id = data.get("shuffle_integration_id", route.shuffle_integration_id) |
| 167 | new_channel = data.get("channel") |
| 168 | if hasattr(new_channel, "value"): |
| 169 | new_channel_value = new_channel.value |
| 170 | else: |
| 171 | new_channel_value = new_channel or route.channel |
| 172 | if new_channel_value == NotificationChannel.SHUFFLE.value and new_integration_id: |
| 173 | await _ensure_integration_belongs_to_customer(new_integration_id, customer_code, session) |
| 174 | |
| 175 | for field, value in data.items(): |
| 176 | # Enums: write the underlying string into the DB column. |
| 177 | if hasattr(value, "value"): |
| 178 | value = value.value |
| 179 | setattr(route, field, value) |
| 180 | route.updated_at = datetime.utcnow() |
| 181 | |
| 182 | await session.commit() |
| 183 | await session.refresh(route) |
| 184 | return route |
| 185 | |
| 186 | |
| 187 | async def delete_route(route_id: int, customer_code: str, session: AsyncSession) -> None: |
| 188 | route = await get_route(route_id, customer_code, session) |
| 189 | await session.delete(route) |
| 190 | await session.commit() |
| 191 | |
| 192 | |
| 193 | # --------------------------------------------------------------------------- |
| 194 | # Shuffle integrations (Phase 2) |
| 195 | # --------------------------------------------------------------------------- |
| 196 | |
| 197 | |
| 198 | async def _ensure_integration_belongs_to_customer( |
| 199 | integration_id: int, |
| 200 | customer_code: str, |
| 201 | session: AsyncSession, |
| 202 | ) -> CustomerShuffleIntegration: |
| 203 | """Tenant-boundary check for Shuffle integration references. |
| 204 | |
| 205 | Used at route create/update time. Without this, a malicious or |
| 206 | typo'd `shuffle_integration_id` could silently route customer A's |
| 207 | notifications through customer B's Shuffle org. Failing closed with |
| 208 | a 400 is the right answer — the route never persists. |
| 209 | """ |
| 210 | result = await session.execute( |
| 211 | select(CustomerShuffleIntegration).where( |
| 212 | CustomerShuffleIntegration.id == integration_id, |
| 213 | CustomerShuffleIntegration.customer_code == customer_code, |
| 214 | ), |
| 215 | ) |
| 216 | integration = result.scalars().first() |
| 217 | if not integration: |
| 218 | raise HTTPException( |
| 219 | status_code=400, |
| 220 | detail=( |
| 221 | f"Shuffle integration {integration_id} not found for " |
| 222 | f"customer {customer_code}. Cross-tenant references are " |
| 223 | f"refused — create the integration on the target customer first." |
| 224 | ), |
| 225 | ) |
| 226 | return integration |
| 227 | |
| 228 | |
| 229 | async def list_shuffle_integrations(customer_code: str, session: AsyncSession) -> List[CustomerShuffleIntegration]: |
| 230 | result = await session.execute( |
| 231 | select(CustomerShuffleIntegration) |
| 232 | .where(CustomerShuffleIntegration.customer_code == customer_code) |
| 233 | .order_by(desc(CustomerShuffleIntegration.created_at)), |
| 234 | ) |
| 235 | return result.scalars().all() |
| 236 | |
| 237 | |
| 238 | async def get_shuffle_integration(integration_id: int, customer_code: str, session: AsyncSession) -> CustomerShuffleIntegration: |
| 239 | return await _ensure_integration_belongs_to_customer(integration_id, customer_code, session) |
| 240 | |
| 241 | |
| 242 | async def create_shuffle_integration( |
| 243 | customer_code: str, |
| 244 | payload: ShuffleIntegrationCreate, |
| 245 | created_by: Optional[str], |
| 246 | session: AsyncSession, |
| 247 | ) -> CustomerShuffleIntegration: |
| 248 | integration = CustomerShuffleIntegration( |
| 249 | customer_code=customer_code, |
| 250 | display_name=payload.display_name, |
| 251 | shuffle_org_id=payload.shuffle_org_id, |
| 252 | enabled=payload.enabled, |
| 253 | created_by=created_by, |
| 254 | ) |
| 255 | session.add(integration) |
| 256 | await session.commit() |
| 257 | await session.refresh(integration) |
| 258 | return integration |
| 259 | |
| 260 | |
| 261 | async def update_shuffle_integration( |
| 262 | integration_id: int, |
| 263 | customer_code: str, |
| 264 | payload: ShuffleIntegrationUpdate, |
| 265 | session: AsyncSession, |
| 266 | ) -> CustomerShuffleIntegration: |
| 267 | integration = await _ensure_integration_belongs_to_customer(integration_id, customer_code, session) |
| 268 | data = payload.model_dump(exclude_unset=True) |
| 269 | for field, value in data.items(): |
| 270 | setattr(integration, field, value) |
| 271 | integration.updated_at = datetime.utcnow() |
| 272 | await session.commit() |
| 273 | await session.refresh(integration) |
| 274 | return integration |
| 275 | |
| 276 | |
| 277 | async def delete_shuffle_integration(integration_id: int, customer_code: str, session: AsyncSession) -> None: |
| 278 | integration = await _ensure_integration_belongs_to_customer(integration_id, customer_code, session) |
| 279 | # Refuse if any routes still reference this integration — better to |
| 280 | # surface the dependency than silently leave routes pointing at a |
| 281 | # missing FK that the dispatch loop will then have to skip. |
| 282 | result = await session.execute( |
| 283 | select(CustomerNotificationRoute).where(CustomerNotificationRoute.shuffle_integration_id == integration_id), |
| 284 | ) |
| 285 | referencing = result.scalars().all() |
| 286 | if referencing: |
| 287 | names = ", ".join(r.name for r in referencing[:5]) |
| 288 | raise HTTPException( |
| 289 | status_code=409, |
| 290 | detail=( |
| 291 | f"Integration is referenced by {len(referencing)} route(s) " |
| 292 | f"({names}{'…' if len(referencing) > 5 else ''}). Delete " |
| 293 | f"or re-point those routes first." |
| 294 | ), |
| 295 | ) |
| 296 | await session.delete(integration) |
| 297 | await session.commit() |
| 298 | |
| 299 | |
| 300 | async def list_apps_for_integration( |
| 301 | integration_id: int, |
| 302 | customer_code: str, |
| 303 | session: AsyncSession, |
| 304 | ) -> List[ShuffleApp]: |
| 305 | """Fetch the Shuffle app catalog scoped to this customer's org. |
| 306 | |
| 307 | Used by the route form's app picker. Roundtrip is short (Shuffle |
| 308 | returns the catalog quickly) and the result is small, so we don't |
| 309 | cache — fresh data on every form open is fine for v1. |
| 310 | """ |
| 311 | integration = await _ensure_integration_belongs_to_customer(integration_id, customer_code, session) |
| 312 | base_url, api_key = await _get_shuffle_connector(session) |
| 313 | ok, apps_raw, error = await shuffle_apps_client( |
| 314 | base_url=base_url, |
| 315 | api_key=api_key, |
| 316 | org_id=integration.shuffle_org_id, |
| 317 | ) |
| 318 | if not ok: |
| 319 | raise HTTPException( |
| 320 | status_code=502, |
| 321 | detail=f"Failed to fetch apps from Shuffle: {error}", |
| 322 | ) |
| 323 | # Forward only the fields the UI needs; ignore extra metadata that |
| 324 | # Shuffle returns (versioning, ownership info, internal ids). |
| 325 | apps: List[ShuffleApp] = [] |
| 326 | for raw in apps_raw: |
| 327 | if not isinstance(raw, dict): |
| 328 | continue |
| 329 | if not raw.get("id") or not raw.get("name"): |
| 330 | continue |
| 331 | apps.append( |
| 332 | ShuffleApp( |
| 333 | id=str(raw.get("id")), |
| 334 | name=str(raw.get("name")), |
| 335 | description=raw.get("description"), |
| 336 | large_image=raw.get("large_image"), |
| 337 | ), |
| 338 | ) |
| 339 | return apps |
| 340 | |
| 341 | |
| 342 | async def verify_integration(integration_id: int, customer_code: str, session: AsyncSession) -> dict: |
| 343 | integration = await _ensure_integration_belongs_to_customer(integration_id, customer_code, session) |
| 344 | base_url, api_key = await _get_shuffle_connector(session) |
| 345 | ok, app_count, error = await verify_shuffle_org_client( |
| 346 | base_url=base_url, |
| 347 | api_key=api_key, |
| 348 | org_id=integration.shuffle_org_id, |
| 349 | ) |
| 350 | return { |
| 351 | "success": ok, |
| 352 | "message": "Shuffle integration reachable" if ok else "Shuffle integration check failed", |
| 353 | "org_id": integration.shuffle_org_id, |
| 354 | "app_count": app_count, |
| 355 | "error": error, |
| 356 | } |
| 357 | |
| 358 | |
| 359 | async def list_orgs(session: AsyncSession) -> List[ShuffleOrg]: |
| 360 | """List every Shuffle org the deployment's admin Bearer can see. |
| 361 | |
| 362 | Used by the integration form's org-picker dropdown so admins pick |
| 363 | a real org instead of pasting a UUID. Not customer-scoped — the |
| 364 | caller's auth gate (admin/analyst scope) is the only access check; |
| 365 | each org is then attached to a specific customer via the |
| 366 | integration row at create time. |
| 367 | """ |
| 368 | base_url, api_key = await _get_shuffle_connector(session) |
| 369 | ok, orgs_raw, error = await shuffle_orgs_client(base_url=base_url, api_key=api_key) |
| 370 | if not ok: |
| 371 | raise HTTPException( |
| 372 | status_code=502, |
| 373 | detail=f"Failed to fetch orgs from Shuffle: {error}", |
| 374 | ) |
| 375 | # Forward only the fields the UI needs. Shuffle's full org payload |
| 376 | # carries internal billing/users/region state we don't want leaking |
| 377 | # through. |
| 378 | orgs: List[ShuffleOrg] = [] |
| 379 | for raw in orgs_raw: |
| 380 | if not isinstance(raw, dict): |
| 381 | continue |
| 382 | if not raw.get("id") or not raw.get("name"): |
| 383 | continue |
| 384 | # `creator_org` is set on sub-orgs to the parent's UUID and |
| 385 | # empty/None on top-level orgs. We forward it as-is so the UI |
| 386 | # can render a "(sub-org)" hint without re-querying. |
| 387 | creator_org = raw.get("creator_org") |
| 388 | if creator_org in ("", "PARENT_ORG_ID"): # ignore placeholder fixtures |
| 389 | creator_org = None |
| 390 | orgs.append( |
| 391 | ShuffleOrg( |
| 392 | id=str(raw.get("id")), |
| 393 | name=str(raw.get("name")), |
| 394 | description=raw.get("description") or None, |
| 395 | role=raw.get("role") or None, |
| 396 | creator_org=creator_org, |
| 397 | ), |
| 398 | ) |
| 399 | return orgs |
| 400 | |
| 401 | |
| 402 | # --------------------------------------------------------------------------- |
| 403 | # Dispatch log (read-only) |
| 404 | # --------------------------------------------------------------------------- |
| 405 | |
| 406 | |
| 407 | async def list_dispatch_log( |
| 408 | customer_code: str, |
| 409 | session: AsyncSession, |
| 410 | limit: int = 100, |
| 411 | ) -> List[NotificationDispatchLog]: |
| 412 | """Recent dispatch history for a customer. Defaults to 100 rows |
| 413 | so the audit-log tab in the UI loads quickly even for noisy |
| 414 | customers.""" |
| 415 | result = await session.execute( |
| 416 | select(NotificationDispatchLog) |
| 417 | .where(NotificationDispatchLog.customer_code == customer_code) |
| 418 | .order_by(desc(NotificationDispatchLog.dispatched_at)) |
| 419 | .limit(limit), |
| 420 | ) |
| 421 | return result.scalars().all() |
| 422 | |
| 423 | |
| 424 | # --------------------------------------------------------------------------- |
| 425 | # Dispatch — the core loop Talon invokes |
| 426 | # --------------------------------------------------------------------------- |
| 427 | |
| 428 | |
| 429 | def _severity_meets(report_severity: str, route_min: str) -> bool: |
| 430 | """Inclusive severity comparison. |
| 431 | |
| 432 | A route with `min_severity="High"` fires when the report is High or |
| 433 | Critical. SEVERITY_ORDER is sorted ascending — a higher index = more |
| 434 | severe. |
| 435 | """ |
| 436 | try: |
| 437 | return SEVERITY_ORDER.index(report_severity) >= SEVERITY_ORDER.index(route_min) |
| 438 | except ValueError: |
| 439 | # Unknown severity string — fail closed. Better to drop a |
| 440 | # notification than fire it on bad input. |
| 441 | logger.warning( |
| 442 | f"Unknown severity in routing comparison " f"(report={report_severity!r}, route_min={route_min!r}); " f"skipping route.", |
| 443 | ) |
| 444 | return False |
| 445 | |
| 446 | |
| 447 | def _trigger_applies(report_trigger: str, route_trigger: str) -> bool: |
| 448 | """Decide whether a route's trigger matches the dispatch event type. |
| 449 | |
| 450 | Triggers represent the kind of event that caused the dispatch |
| 451 | (currently only `investigation_complete`, with more event types |
| 452 | arriving as we add hooks for analyst review / IOC enrichment / |
| 453 | scheduled sweeps). Severity filtering lives in `min_severity`, |
| 454 | not here — this function is purely an event-type equality check. |
| 455 | |
| 456 | Routes with stale `severity_critical_or_high` values from earlier |
| 457 | schemas are treated as `investigation_complete` so they keep |
| 458 | firing instead of being silently filtered out. |
| 459 | """ |
| 460 | if route_trigger == "severity_critical_or_high": |
| 461 | # Backward compat: legacy enum value, treat as the catch-all |
| 462 | # event type so existing routes don't go dark on upgrade. |
| 463 | return report_trigger == NotificationTrigger.INVESTIGATION_COMPLETE.value |
| 464 | return route_trigger == report_trigger |
| 465 | |
| 466 | |
| 467 | def _format_default_body(req: DispatchRequest) -> str: |
| 468 | """Plain default formatter when a route has no `format_template`. |
| 469 | |
| 470 | Markdown-ish but readable in both Slack and email — both channels |
| 471 | render this acceptably without extra structure. Phase 4 swaps for |
| 472 | per-channel templates. |
| 473 | """ |
| 474 | parts = [ |
| 475 | f"*AI investigation complete* — severity: *{req.severity_assessment.value}*", |
| 476 | "", |
| 477 | f"Customer: `{req.customer_code}`", |
| 478 | f"Alert: #{req.alert_id}" + (f" — {req.alert_name}" if req.alert_name else ""), |
| 479 | "", |
| 480 | req.summary.strip(), |
| 481 | ] |
| 482 | if req.report_url: |
| 483 | parts.extend(["", f"Full report: {req.report_url}"]) |
| 484 | return "\n".join(parts) |
| 485 | |
| 486 | |
| 487 | def _render_body(route: CustomerNotificationRoute, req: DispatchRequest) -> str: |
| 488 | """Apply the route's `format_template` if set, else fall back. |
| 489 | |
| 490 | Phase 1's templating is intentionally minimal — `{{ variable }}` |
| 491 | substitution only, no Jinja control flow. Real Jinja can come in |
| 492 | Phase 4 when the per-channel templates land. |
| 493 | """ |
| 494 | if not route.format_template: |
| 495 | return _format_default_body(req) |
| 496 | |
| 497 | body = route.format_template |
| 498 | substitutions = { |
| 499 | "{{customer_code}}": req.customer_code, |
| 500 | "{{alert_id}}": str(req.alert_id), |
| 501 | "{{alert_name}}": req.alert_name or "", |
| 502 | "{{severity}}": req.severity_assessment.value, |
| 503 | "{{summary}}": req.summary, |
| 504 | "{{report_url}}": req.report_url or "", |
| 505 | } |
| 506 | for token, value in substitutions.items(): |
| 507 | body = body.replace(token, value) |
| 508 | return body |
| 509 | |
| 510 | |
| 511 | async def _record_log( |
| 512 | session: AsyncSession, |
| 513 | *, |
| 514 | customer_code: str, |
| 515 | alert_id: int, |
| 516 | route_id: int, |
| 517 | trigger: str, |
| 518 | status: str, |
| 519 | error_message: Optional[str], |
| 520 | latency_ms: Optional[int], |
| 521 | payload_preview: Optional[str], |
| 522 | shuffle_execution_id: Optional[str] = None, |
| 523 | ) -> bool: |
| 524 | """Record a dispatch outcome. Returns False ONLY when the dispatch |
| 525 | has already been recorded as `sent` — i.e. a true idempotency hit |
| 526 | against a successful prior dispatch. Returns True in all other |
| 527 | cases, including overwriting a previous failed/skipped attempt |
| 528 | with the new result so retries land cleanly. |
| 529 | |
| 530 | Idempotency model: |
| 531 | - One row per (customer_code, alert_id, route_id, trigger) tuple |
| 532 | (enforced by a unique index) |
| 533 | - If the existing row's status is `sent`, refuse the new write |
| 534 | (caller treats as "already done, skip") |
| 535 | - If the existing row's status is `failed`/`skipped`, overwrite |
| 536 | with the new outcome — a previous failure must not block a |
| 537 | retry |
| 538 | - If no row exists yet, insert a fresh one |
| 539 | """ |
| 540 | # Pre-flight: check whether a row already exists for this |
| 541 | # (customer, alert, route, trigger) tuple. Doing the check up front |
| 542 | # lets us update-in-place when needed — avoids the rollback path |
| 543 | # whose `session.rollback()` expires every loaded object in the |
| 544 | # session (route, integrations, etc.) and breaks subsequent |
| 545 | # attribute access in async context. |
| 546 | result = await session.execute( |
| 547 | select(NotificationDispatchLog).where( |
| 548 | NotificationDispatchLog.customer_code == customer_code, |
| 549 | NotificationDispatchLog.alert_id == alert_id, |
| 550 | NotificationDispatchLog.route_id == route_id, |
| 551 | NotificationDispatchLog.trigger == trigger, |
| 552 | ), |
| 553 | ) |
| 554 | existing = result.scalars().first() |
| 555 | |
| 556 | if existing is not None and existing.status == "sent": |
| 557 | # True idempotency hit — don't overwrite a successful dispatch. |
| 558 | return False |
| 559 | |
| 560 | if existing is not None: |
| 561 | # Previous failed/skipped attempt — overwrite it so the log |
| 562 | # reflects the latest outcome and the retry path is clean. |
| 563 | existing.status = status |
| 564 | existing.error_message = error_message |
| 565 | existing.latency_ms = latency_ms |
| 566 | existing.payload_preview = payload_preview[:500] if payload_preview else None |
| 567 | existing.shuffle_execution_id = shuffle_execution_id |
| 568 | existing.dispatched_at = datetime.utcnow() |
| 569 | await session.commit() |
| 570 | return True |
| 571 | |
| 572 | # No prior record — insert fresh. |
| 573 | log = NotificationDispatchLog( |
| 574 | customer_code=customer_code, |
| 575 | alert_id=alert_id, |
| 576 | route_id=route_id, |
| 577 | trigger=trigger, |
| 578 | status=status, |
| 579 | error_message=error_message, |
| 580 | latency_ms=latency_ms, |
| 581 | payload_preview=payload_preview[:500] if payload_preview else None, |
| 582 | shuffle_execution_id=shuffle_execution_id, |
| 583 | ) |
| 584 | session.add(log) |
| 585 | try: |
| 586 | await session.commit() |
| 587 | return True |
| 588 | except IntegrityError: |
| 589 | # Race: another concurrent dispatch slipped in between our |
| 590 | # SELECT and INSERT. Roll back, treat as idempotency hit. The |
| 591 | # caller's outcome will be `skipped` and the route's attrs |
| 592 | # will be expired — but the caller has already cached them |
| 593 | # into locals so this is safe. |
| 594 | await session.rollback() |
| 595 | return False |
| 596 | |
| 597 | |
| 598 | async def dispatch(req: DispatchRequest, session: AsyncSession) -> DispatchResponse: |
| 599 | """Walk the customer's routes, fire each match, log each outcome. |
| 600 | |
| 601 | Idempotency is enforced at the log table — we attempt the insert |
| 602 | *before* calling the provider, so a re-dispatch sees the existing |
| 603 | row and short-circuits without sending. (The cost is one wasted |
| 604 | INSERT in the race case, which is fine.) |
| 605 | """ |
| 606 | routes = await list_routes(req.customer_code, session) |
| 607 | |
| 608 | matched_routes = [ |
| 609 | r |
| 610 | for r in routes |
| 611 | if r.enabled and _trigger_applies(req.trigger.value, r.trigger) and _severity_meets(req.severity_assessment.value, r.min_severity) |
| 612 | ] |
| 613 | |
| 614 | outcomes: List[DispatchOutcome] = [] |
| 615 | sent = failed = skipped = 0 |
| 616 | |
| 617 | # Shuffle dispatch needs the deployment's connector creds. We fetch |
| 618 | # them once per dispatch call (before the per-route loop) so a |
| 619 | # whole batch shares one DB read. Only fetch if at least one |
| 620 | # matched route uses Shuffle — early-out keeps zero-route customers |
| 621 | # from touching the connectors table. |
| 622 | shuffle_creds: Optional[tuple[str, str]] = None |
| 623 | shuffle_creds_error: Optional[str] = None |
| 624 | if any(r.channel == NotificationChannel.SHUFFLE.value for r in matched_routes): |
| 625 | try: |
| 626 | shuffle_creds = await _get_shuffle_connector(session) |
| 627 | except HTTPException as e: |
| 628 | # Connector misconfigured — the dispatch endpoint surfaces |
| 629 | # the helper's 503 to ad-hoc callers, but for a batch we'd |
| 630 | # rather mark each route as failed in the log than abort |
| 631 | # the whole loop. |
| 632 | logger.warning(f"Shuffle connector unavailable: {e.detail}") |
| 633 | shuffle_creds_error = str(e.detail) |
| 634 | |
| 635 | for route in matched_routes: |
| 636 | # Cache every route attribute we'll need into locals UP FRONT. |
| 637 | # Once we cross any `await` (let alone any rollback) the route |
| 638 | # SQLAlchemy state can be expired and a synchronous attribute |
| 639 | # access then triggers an implicit refresh query — which in |
| 640 | # AsyncSession throws MissingGreenlet. Caching here means the |
| 641 | # rest of the loop is plain-Python access on locals. |
| 642 | route_id = route.id |
| 643 | route_name = route.name |
| 644 | route_channel = route.channel |
| 645 | route_destination = route.destination |
| 646 | route_shuffle_app_id = route.shuffle_app_id |
| 647 | route_shuffle_integration_id = route.shuffle_integration_id |
| 648 | |
| 649 | body = _render_body(route, req) |
| 650 | body_preview = body[:500] |
| 651 | |
| 652 | latency_ms: Optional[int] = None |
| 653 | result_status = "sent" |
| 654 | error_message: Optional[str] = None |
| 655 | shuffle_execution_id: Optional[str] = None |
| 656 | |
| 657 | try: |
| 658 | if route_channel == NotificationChannel.SHUFFLE.value: |
| 659 | # Shuffle hosted MCP. Fire-and-record — we POST to |
| 660 | # /api/v1/apps/{app_id}/mcp with the deployment's admin |
| 661 | # Bearer + the customer's Org-Id, capture the |
| 662 | # execution_id, and consider the dispatch "sent" on |
| 663 | # HTTP 200. We do NOT poll for the downstream app's |
| 664 | # terminal state. |
| 665 | if shuffle_creds is None: |
| 666 | result_status = "failed" |
| 667 | error_message = shuffle_creds_error or "Shuffle connector unavailable" |
| 668 | latency_ms = 0 |
| 669 | elif not route_shuffle_app_id: |
| 670 | result_status = "failed" |
| 671 | error_message = "Route has no shuffle_app_id (data integrity issue)" |
| 672 | latency_ms = 0 |
| 673 | else: |
| 674 | integration = await session.get(CustomerShuffleIntegration, route_shuffle_integration_id) |
| 675 | if not integration or integration.customer_code != req.customer_code: |
| 676 | # Defense-in-depth: we already enforce tenant |
| 677 | # isolation at create/update time, but a hand- |
| 678 | # edited row could still slip through. Refusing |
| 679 | # at dispatch time prevents cross-tenant leaks. |
| 680 | result_status = "failed" |
| 681 | error_message = ( |
| 682 | "Route's shuffle_integration is missing or belongs to a " "different customer; refusing to dispatch." |
| 683 | ) |
| 684 | latency_ms = 0 |
| 685 | elif not integration.enabled: |
| 686 | result_status = "skipped" |
| 687 | error_message = "Shuffle integration is disabled" |
| 688 | latency_ms = 0 |
| 689 | else: |
| 690 | base_url, api_key = shuffle_creds |
| 691 | # Cache integration attrs too — same reason as |
| 692 | # the route caching above. |
| 693 | integration_org_id = integration.shuffle_org_id |
| 694 | # Shuffle's input_text is natural language. We |
| 695 | # prepend a "send to {destination}" hint so the |
| 696 | # Shuffle app agent knows where to deliver, and |
| 697 | # follow with the formatted body. |
| 698 | if route_destination: |
| 699 | input_text = f"Send to {route_destination}: {body}" |
| 700 | else: |
| 701 | input_text = body |
| 702 | ( |
| 703 | result_status, |
| 704 | error_message, |
| 705 | latency_ms, |
| 706 | shuffle_execution_id, |
| 707 | ) = await dispatch_shuffle( |
| 708 | base_url=base_url, |
| 709 | api_key=api_key, |
| 710 | org_id=integration_org_id, |
| 711 | app_id=route_shuffle_app_id, |
| 712 | input_text=input_text, |
| 713 | ) |
| 714 | else: |
| 715 | # Unknown channel — preserved as a failure rather than |
| 716 | # silently dropped so a misconfigured row surfaces in |
| 717 | # the dispatch log. |
| 718 | result_status = "failed" |
| 719 | error_message = f"Unsupported channel: {route_channel}" |
| 720 | latency_ms = None |
| 721 | except Exception as e: # noqa: BLE001 — best-effort, never raise |
| 722 | logger.exception(f"Dispatcher raised for route {route_id}: {e!r}") |
| 723 | result_status = "failed" |
| 724 | error_message = f"Dispatcher exception: {type(e).__name__}: {e}" |
| 725 | |
| 726 | # Record (or update) the dispatch outcome. _record_log handles |
| 727 | # the retry-after-failure case in-place so a previous failed |
| 728 | # row doesn't block a new attempt — the only way we get back |
| 729 | # `False` here is a true idempotency hit on a previously-sent |
| 730 | # dispatch. |
| 731 | recorded = await _record_log( |
| 732 | session, |
| 733 | customer_code=req.customer_code, |
| 734 | alert_id=req.alert_id, |
| 735 | route_id=route_id, |
| 736 | trigger=req.trigger.value, |
| 737 | status=result_status, |
| 738 | error_message=error_message, |
| 739 | latency_ms=latency_ms, |
| 740 | payload_preview=body_preview, |
| 741 | shuffle_execution_id=shuffle_execution_id, |
| 742 | ) |
| 743 | |
| 744 | if not recorded: |
| 745 | skipped += 1 |
| 746 | outcomes.append( |
| 747 | DispatchOutcome( |
| 748 | route_id=route_id, |
| 749 | route_name=route_name, |
| 750 | channel=route_channel, |
| 751 | status=DispatchStatus.SKIPPED, |
| 752 | error_message="Already dispatched (idempotency)", |
| 753 | latency_ms=None, |
| 754 | ), |
| 755 | ) |
| 756 | continue |
| 757 | |
| 758 | # Maintain denorm columns for the UI list. Cheaper than joining |
| 759 | # the log table on every render. We do this via an explicit |
| 760 | # UPDATE statement rather than mutating the loaded route |
| 761 | # object, so route's expiration state can't bite us. |
| 762 | if result_status == "sent": |
| 763 | sent += 1 |
| 764 | await session.execute( |
| 765 | update(CustomerNotificationRoute) |
| 766 | .where(CustomerNotificationRoute.id == route_id) |
| 767 | .values( |
| 768 | dispatch_count=CustomerNotificationRoute.dispatch_count + 1, |
| 769 | last_dispatched_at=datetime.utcnow(), |
| 770 | ), |
| 771 | ) |
| 772 | # Bump the integration's last_used_at on a successful |
| 773 | # Shuffle dispatch — gives the integration list a "fired |
| 774 | # 2h ago" signal without a join against the log. |
| 775 | if route_channel == NotificationChannel.SHUFFLE.value and route_shuffle_integration_id: |
| 776 | await session.execute( |
| 777 | update(CustomerShuffleIntegration) |
| 778 | .where(CustomerShuffleIntegration.id == route_shuffle_integration_id) |
| 779 | .values(last_used_at=datetime.utcnow()), |
| 780 | ) |
| 781 | await session.commit() |
| 782 | elif result_status == "skipped": |
| 783 | skipped += 1 |
| 784 | else: |
| 785 | failed += 1 |
| 786 | |
| 787 | outcomes.append( |
| 788 | DispatchOutcome( |
| 789 | route_id=route_id, |
| 790 | route_name=route_name, |
| 791 | channel=route_channel, |
| 792 | status=DispatchStatus(result_status), |
| 793 | error_message=error_message, |
| 794 | latency_ms=latency_ms, |
| 795 | shuffle_execution_id=shuffle_execution_id, |
| 796 | ), |
| 797 | ) |
| 798 | |
| 799 | return DispatchResponse( |
| 800 | success=True, |
| 801 | message=(f"Dispatched {sent} of {len(matched_routes)} matching route(s) " f"for customer {req.customer_code} alert {req.alert_id}"), |
| 802 | routes_matched=len(matched_routes), |
| 803 | dispatched=sent, |
| 804 | skipped=skipped, |
| 805 | failed=failed, |
| 806 | outcomes=outcomes, |
| 807 | ) |