| """Notification / out-of-band confirmation endpoints — a torch-free router. |
| |
| Isolated from the monolithic `api.routes` (which imports the biometric models/torch) |
| so these endpoints and their API tests run in the lightweight, torch-free environment. |
| Paths, methods and request/response contracts are UNCHANGED from the previous |
| definitions in `api.routes`.""" |
|
|
| from __future__ import annotations |
|
|
| import time |
|
|
| from fastapi import APIRouter, Body |
| from fastapi.responses import JSONResponse |
|
|
| from api.services import audit_event, notifications, persist_prefs |
|
|
| router = APIRouter() |
|
|
|
|
| @router.post("/notify/prefs") |
| def notify_set_prefs(body: dict = Body(...)) -> dict: |
| r = notifications.set_prefs(body["user_id"], channels=body.get("channels"), |
| email=body.get("email"), phone=body.get("phone")) |
| persist_prefs() |
| return r |
|
|
|
|
| @router.get("/notify/prefs") |
| def notify_get_prefs(user_id: str) -> dict: |
| return notifications.get_prefs(user_id) |
|
|
|
|
| @router.post("/notify/subscribe") |
| def notify_subscribe(body: dict = Body(...)) -> dict: |
| """Store a Web Push subscription (from the browser) for the user.""" |
| return notifications.subscribe_push(body["user_id"], body["subscription"]) |
|
|
|
|
| @router.post("/notify/request") |
| def notify_request(body: dict = Body(...)) -> dict: |
| """Create a demo out-of-band payment confirmation (dynamic-linked to amount+merchant) |
| using the existing notification service, so the React OOB demo can exercise the |
| request -> respond -> status workflow without going through /wallet/pay.""" |
| return notifications.send_payment_confirmation( |
| body["user_id"], float(body.get("amount", 0.0)), |
| str(body.get("merchant", "")), now=time.time()) |
|
|
|
|
| @router.get("/notify/status") |
| def notify_status(confirmation_id: str) -> dict: |
| """Lifecycle of an OOB confirmation for UI polling: pending/approved/rejected/expired.""" |
| return {"confirmation_id": confirmation_id, |
| "status": notifications.confirmation_status(confirmation_id, now=time.time())} |
|
|
|
|
| @router.post("/notify/respond") |
| def notify_respond(body: dict = Body(...)): |
| """Approve/deny an OOB confirmation. Enforces terminal-state immutability: a |
| conflicting response after a terminal state returns HTTP 409 (oob_already_finalized) |
| and leaves the final state unchanged; a same-decision retry is idempotent.""" |
| r = notifications.respond(body["confirmation_id"], bool(body.get("approve", False)), |
| code=body.get("code"), now=time.time()) |
| if r.get("conflict"): |
| |
| audit_event("oob", "oob_conflict_rejected", |
| {"confirmation_id": r.get("confirmation_id"), "final_state": r.get("status")}) |
| return JSONResponse(status_code=409, content=r) |
| return r |
|
|
