| 1 | """ |
| 2 | REST routes for the notification routing module. |
| 3 | |
| 4 | Two surfaces: |
| 5 | |
| 6 | /customers/{customer_code}/notification_routes |
| 7 | /customers/{customer_code}/notification_dispatch_log |
| 8 | admin/analyst CRUD + read-only audit view, used by the CoPilot |
| 9 | frontend's per-customer Notifications tab. |
| 10 | |
| 11 | /notifications/dispatch |
| 12 | called by Talon (NanoClaw) after every successful investigation. |
| 13 | Walks the customer's routes, fires each match, logs each |
| 14 | outcome, returns a per-route result list. Best-effort — Talon |
| 15 | does not retry, and a failure here MUST NOT fail the upstream |
| 16 | investigation. |
| 17 | """ |
| 18 | |
| 19 | from __future__ import annotations |
| 20 | |
| 21 | from fastapi import APIRouter |
| 22 | from fastapi import Depends |
| 23 | from fastapi import Security |
| 24 | from loguru import logger |
| 25 | from sqlalchemy.ext.asyncio import AsyncSession |
| 26 | |
| 27 | from app.auth.models.users import User |
| 28 | from app.auth.utils import AuthHandler |
| 29 | from app.db.db_session import get_db |
| 30 | from app.notifications.schema.notifications import DispatchLogListResponse |
| 31 | from app.notifications.schema.notifications import DispatchRequest |
| 32 | from app.notifications.schema.notifications import DispatchResponse |
| 33 | from app.notifications.schema.notifications import NotificationRouteCreate |
| 34 | from app.notifications.schema.notifications import NotificationRouteListResponse |
| 35 | from app.notifications.schema.notifications import NotificationRouteRead |
| 36 | from app.notifications.schema.notifications import NotificationRouteResponse |
| 37 | from app.notifications.schema.notifications import NotificationRouteUpdate |
| 38 | from app.notifications.schema.notifications import ShuffleAppListResponse |
| 39 | from app.notifications.schema.notifications import ShuffleIntegrationCreate |
| 40 | from app.notifications.schema.notifications import ShuffleIntegrationListResponse |
| 41 | from app.notifications.schema.notifications import ShuffleIntegrationRead |
| 42 | from app.notifications.schema.notifications import ShuffleIntegrationResponse |
| 43 | from app.notifications.schema.notifications import ShuffleIntegrationUpdate |
| 44 | from app.notifications.schema.notifications import ShuffleOrgListResponse |
| 45 | from app.notifications.schema.notifications import ShuffleVerifyResponse |
| 46 | from app.notifications.services import notifications as svc |
| 47 | |
| 48 | notifications_router = APIRouter() |
| 49 | |
| 50 | |
| 51 | # --------------------------------------------------------------------------- |
| 52 | # Per-customer route CRUD |
| 53 | # --------------------------------------------------------------------------- |
| 54 | |
| 55 | |
| 56 | @notifications_router.get( |
| 57 | "/customers/{customer_code}/notification_routes", |
| 58 | response_model=NotificationRouteListResponse, |
| 59 | description="List notification routes for a customer.", |
| 60 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 61 | ) |
| 62 | async def list_routes_route( |
| 63 | customer_code: str, |
| 64 | session: AsyncSession = Depends(get_db), |
| 65 | ) -> NotificationRouteListResponse: |
| 66 | routes = await svc.list_routes(customer_code, session) |
| 67 | return NotificationRouteListResponse( |
| 68 | success=True, |
| 69 | message=f"{len(routes)} route(s) retrieved", |
| 70 | routes=[NotificationRouteRead.from_orm(r) for r in routes], |
| 71 | ) |
| 72 | |
| 73 | |
| 74 | @notifications_router.post( |
| 75 | "/customers/{customer_code}/notification_routes", |
| 76 | response_model=NotificationRouteResponse, |
| 77 | description="Create a new notification route for a customer.", |
| 78 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 79 | ) |
| 80 | async def create_route_route( |
| 81 | customer_code: str, |
| 82 | payload: NotificationRouteCreate, |
| 83 | session: AsyncSession = Depends(get_db), |
| 84 | current_user: User = Depends(AuthHandler().get_current_user), |
| 85 | ) -> NotificationRouteResponse: |
| 86 | logger.info(f"User {current_user.id} creating notification route " f"for customer {customer_code}") |
| 87 | route = await svc.create_route( |
| 88 | customer_code=customer_code, |
| 89 | payload=payload, |
| 90 | created_by=getattr(current_user, "username", None) or str(current_user.id), |
| 91 | session=session, |
| 92 | ) |
| 93 | return NotificationRouteResponse( |
| 94 | success=True, |
| 95 | message="Route created", |
| 96 | route=NotificationRouteRead.from_orm(route), |
| 97 | ) |
| 98 | |
| 99 | |
| 100 | @notifications_router.patch( |
| 101 | "/customers/{customer_code}/notification_routes/{route_id}", |
| 102 | response_model=NotificationRouteResponse, |
| 103 | description="Update an existing notification route. Only fields included in the body are modified.", |
| 104 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 105 | ) |
| 106 | async def update_route_route( |
| 107 | customer_code: str, |
| 108 | route_id: int, |
| 109 | payload: NotificationRouteUpdate, |
| 110 | session: AsyncSession = Depends(get_db), |
| 111 | ) -> NotificationRouteResponse: |
| 112 | route = await svc.update_route(route_id, customer_code, payload, session) |
| 113 | return NotificationRouteResponse( |
| 114 | success=True, |
| 115 | message="Route updated", |
| 116 | route=NotificationRouteRead.from_orm(route), |
| 117 | ) |
| 118 | |
| 119 | |
| 120 | @notifications_router.delete( |
| 121 | "/customers/{customer_code}/notification_routes/{route_id}", |
| 122 | description="Delete a notification route. Dispatch log entries for the route are retained.", |
| 123 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 124 | ) |
| 125 | async def delete_route_route( |
| 126 | customer_code: str, |
| 127 | route_id: int, |
| 128 | session: AsyncSession = Depends(get_db), |
| 129 | ) -> dict: |
| 130 | await svc.delete_route(route_id, customer_code, session) |
| 131 | return {"success": True, "message": "Route deleted"} |
| 132 | |
| 133 | |
| 134 | # --------------------------------------------------------------------------- |
| 135 | # Dispatch log (read-only audit) |
| 136 | # --------------------------------------------------------------------------- |
| 137 | |
| 138 | |
| 139 | @notifications_router.get( |
| 140 | "/customers/{customer_code}/notification_dispatch_log", |
| 141 | response_model=DispatchLogListResponse, |
| 142 | description="Recent notification dispatch attempts for a customer (newest first, capped at 100).", |
| 143 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 144 | ) |
| 145 | async def list_dispatch_log_route( |
| 146 | customer_code: str, |
| 147 | session: AsyncSession = Depends(get_db), |
| 148 | ) -> DispatchLogListResponse: |
| 149 | entries = await svc.list_dispatch_log(customer_code, session, limit=100) |
| 150 | return DispatchLogListResponse( |
| 151 | success=True, |
| 152 | message=f"{len(entries)} entry/entries retrieved", |
| 153 | entries=entries, |
| 154 | ) |
| 155 | |
| 156 | |
| 157 | # --------------------------------------------------------------------------- |
| 158 | # Deployment-scoped Shuffle helpers (Phase 3) |
| 159 | # --------------------------------------------------------------------------- |
| 160 | |
| 161 | |
| 162 | @notifications_router.get( |
| 163 | "/notifications/shuffle/orgs", |
| 164 | response_model=ShuffleOrgListResponse, |
| 165 | description=( |
| 166 | "List every Shuffle org the deployment's admin Bearer key can see. " |
| 167 | "Used by the integration form's org picker so admins choose from a " |
| 168 | "dropdown instead of pasting Org-Ids. Not customer-scoped — each " |
| 169 | "org is later attached to a specific customer via a " |
| 170 | "customer_shuffle_integration row." |
| 171 | ), |
| 172 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 173 | ) |
| 174 | async def list_shuffle_orgs_route( |
| 175 | session: AsyncSession = Depends(get_db), |
| 176 | ) -> ShuffleOrgListResponse: |
| 177 | orgs = await svc.list_orgs(session) |
| 178 | return ShuffleOrgListResponse( |
| 179 | success=True, |
| 180 | message=f"{len(orgs)} org(s) retrieved", |
| 181 | orgs=orgs, |
| 182 | ) |
| 183 | |
| 184 | |
| 185 | # --------------------------------------------------------------------------- |
| 186 | # Per-customer Shuffle integrations (Phase 2) |
| 187 | # --------------------------------------------------------------------------- |
| 188 | |
| 189 | |
| 190 | @notifications_router.get( |
| 191 | "/customers/{customer_code}/shuffle_integrations", |
| 192 | response_model=ShuffleIntegrationListResponse, |
| 193 | description="List Shuffle integrations (per-customer Org-Id rows) for a customer.", |
| 194 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 195 | ) |
| 196 | async def list_shuffle_integrations_route( |
| 197 | customer_code: str, |
| 198 | session: AsyncSession = Depends(get_db), |
| 199 | ) -> ShuffleIntegrationListResponse: |
| 200 | integrations = await svc.list_shuffle_integrations(customer_code, session) |
| 201 | return ShuffleIntegrationListResponse( |
| 202 | success=True, |
| 203 | message=f"{len(integrations)} integration(s) retrieved", |
| 204 | integrations=[ShuffleIntegrationRead.from_orm(i) for i in integrations], |
| 205 | ) |
| 206 | |
| 207 | |
| 208 | @notifications_router.post( |
| 209 | "/customers/{customer_code}/shuffle_integrations", |
| 210 | response_model=ShuffleIntegrationResponse, |
| 211 | description="Create a new Shuffle integration for a customer (records the customer's Shuffle Org-Id).", |
| 212 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 213 | ) |
| 214 | async def create_shuffle_integration_route( |
| 215 | customer_code: str, |
| 216 | payload: ShuffleIntegrationCreate, |
| 217 | session: AsyncSession = Depends(get_db), |
| 218 | current_user: User = Depends(AuthHandler().get_current_user), |
| 219 | ) -> ShuffleIntegrationResponse: |
| 220 | logger.info(f"User {current_user.id} adding Shuffle integration " f"({payload.display_name}) for customer {customer_code}") |
| 221 | integration = await svc.create_shuffle_integration( |
| 222 | customer_code=customer_code, |
| 223 | payload=payload, |
| 224 | created_by=getattr(current_user, "username", None) or str(current_user.id), |
| 225 | session=session, |
| 226 | ) |
| 227 | return ShuffleIntegrationResponse( |
| 228 | success=True, |
| 229 | message="Integration created", |
| 230 | integration=ShuffleIntegrationRead.from_orm(integration), |
| 231 | ) |
| 232 | |
| 233 | |
| 234 | @notifications_router.patch( |
| 235 | "/customers/{customer_code}/shuffle_integrations/{integration_id}", |
| 236 | response_model=ShuffleIntegrationResponse, |
| 237 | description="Update an existing Shuffle integration. Only fields included in the body are modified.", |
| 238 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 239 | ) |
| 240 | async def update_shuffle_integration_route( |
| 241 | customer_code: str, |
| 242 | integration_id: int, |
| 243 | payload: ShuffleIntegrationUpdate, |
| 244 | session: AsyncSession = Depends(get_db), |
| 245 | ) -> ShuffleIntegrationResponse: |
| 246 | integration = await svc.update_shuffle_integration(integration_id, customer_code, payload, session) |
| 247 | return ShuffleIntegrationResponse( |
| 248 | success=True, |
| 249 | message="Integration updated", |
| 250 | integration=ShuffleIntegrationRead.from_orm(integration), |
| 251 | ) |
| 252 | |
| 253 | |
| 254 | @notifications_router.delete( |
| 255 | "/customers/{customer_code}/shuffle_integrations/{integration_id}", |
| 256 | description="Delete a Shuffle integration. Refused if any notification routes reference it.", |
| 257 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 258 | ) |
| 259 | async def delete_shuffle_integration_route( |
| 260 | customer_code: str, |
| 261 | integration_id: int, |
| 262 | session: AsyncSession = Depends(get_db), |
| 263 | ) -> dict: |
| 264 | await svc.delete_shuffle_integration(integration_id, customer_code, session) |
| 265 | return {"success": True, "message": "Integration deleted"} |
| 266 | |
| 267 | |
| 268 | @notifications_router.get( |
| 269 | "/customers/{customer_code}/shuffle_integrations/{integration_id}/apps", |
| 270 | response_model=ShuffleAppListResponse, |
| 271 | description=( |
| 272 | "Fetch the Shuffle app catalog scoped to this customer's org. Used " |
| 273 | "by the route form's app picker so admins can pick from a list " |
| 274 | "instead of hand-typing UUIDs." |
| 275 | ), |
| 276 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 277 | ) |
| 278 | async def list_shuffle_apps_route( |
| 279 | customer_code: str, |
| 280 | integration_id: int, |
| 281 | session: AsyncSession = Depends(get_db), |
| 282 | ) -> ShuffleAppListResponse: |
| 283 | apps = await svc.list_apps_for_integration(integration_id, customer_code, session) |
| 284 | return ShuffleAppListResponse( |
| 285 | success=True, |
| 286 | message=f"{len(apps)} app(s) retrieved", |
| 287 | apps=apps, |
| 288 | ) |
| 289 | |
| 290 | |
| 291 | @notifications_router.get( |
| 292 | "/customers/{customer_code}/shuffle_integrations/{integration_id}/verify", |
| 293 | response_model=ShuffleVerifyResponse, |
| 294 | description="Probe Shuffle with the integration's Org-Id to confirm the connector is reachable and the org is valid.", |
| 295 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 296 | ) |
| 297 | async def verify_shuffle_integration_route( |
| 298 | customer_code: str, |
| 299 | integration_id: int, |
| 300 | session: AsyncSession = Depends(get_db), |
| 301 | ) -> ShuffleVerifyResponse: |
| 302 | result = await svc.verify_integration(integration_id, customer_code, session) |
| 303 | return ShuffleVerifyResponse(**result) |
| 304 | |
| 305 | |
| 306 | # --------------------------------------------------------------------------- |
| 307 | # Dispatch — called by Talon after each investigation |
| 308 | # --------------------------------------------------------------------------- |
| 309 | |
| 310 | |
| 311 | @notifications_router.post( |
| 312 | "/notifications/dispatch", |
| 313 | response_model=DispatchResponse, |
| 314 | description=( |
| 315 | "Walk the customer's notification routes for the given trigger and " |
| 316 | "severity, dispatch each match, and log each outcome. Idempotent — " |
| 317 | "re-dispatching the same (customer, alert, route, trigger) is a no-op. " |
| 318 | "Talon calls this after writing back an investigation report." |
| 319 | ), |
| 320 | dependencies=[Security(AuthHandler().require_any_scope("admin", "analyst"))], |
| 321 | ) |
| 322 | async def dispatch_route( |
| 323 | payload: DispatchRequest, |
| 324 | session: AsyncSession = Depends(get_db), |
| 325 | ) -> DispatchResponse: |
| 326 | logger.info( |
| 327 | f"Notification dispatch requested for customer {payload.customer_code} " |
| 328 | f"alert {payload.alert_id} trigger {payload.trigger.value} " |
| 329 | f"severity {payload.severity_assessment.value}", |
| 330 | ) |
| 331 | return await svc.dispatch(payload, session) |