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