# Request Flow ## Synchronous Request — Happy Path (REST → Core Banking) ``` External Client │ │ POST /api/v1/transactions │ Authorization: Bearer ▼ ┌─────────────────────────────────────────────────────────┐ │ gateway_api — Endpoint / Router │ │ │ │ 1. RequestIdPlug inject trace_id → conn assigns │ │ 2. CORSPlug set CORS headers │ │ 3. TransactionController.create/2 │ └───────────────────────────┬─────────────────────────────┘ │ MwRouter.Pipeline.call(context) ▼ ┌─────────────────────────────────────────────────────────┐ │ mw_router — Pipeline (Plug.Builder chain) │ │ │ │ Stage 1: MwAuth.Plug │ │ ├── detect "Bearer" prefix │ │ ├── MwAuth.JWT.verify(token) │ │ ├── load claims → context.user, context.roles │ │ └── reject 401 if invalid / expired │ │ │ │ Stage 2: MwRouter.RateLimiter │ │ ├── ex_rated token bucket per api_key │ │ └── reject 429 if bucket exhausted │ │ │ │ Stage 3: MwTransform.Plug (inbound) │ │ ├── validate JSON body against schema registry │ │ ├── map external field names → canonical message │ │ └── produce %MwKernel.Message{} struct │ │ │ │ Stage 4: MwRouter.RoutePlug │ │ ├── look up message.type in ETS route table │ │ ├── resolve → AdapterBanking │ │ └── check circuit breaker status (:fuse) │ │ │ │ Stage 5: MwRouter.Dispatcher.dispatch(message, adapter)│ │ └── calls AdapterBanking.send(state, message) │ └───────────────────────────┬─────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ adapter_banking │ │ │ │ 1. AdapterBanking.Transformer.to_iso8583(message) │ │ 2. AdapterBanking.Client.send(iso_payload) │ ──► Core Banking System │ 3. receive ISO 8583 response │ ◄── │ 4. AdapterBanking.Transformer.from_iso8583(response) │ │ 5. return %MwKernel.Message{status: :ok, ...} │ └───────────────────────────┬─────────────────────────────┘ │ {:ok, response_message} ▼ ┌─────────────────────────────────────────────────────────┐ │ mw_router — Pipeline continues │ │ │ │ Stage 6: MwTransform.Plug (outbound) │ │ └── map canonical response → client-facing JSON │ │ │ │ Stage 7: MwAudit.Plug │ │ ├── write audit event to DB (async, non-blocking) │ │ ├── emit :telemetry event (duration, status, adapter)│ │ └── broadcast to PubSub (live dashboard) │ └───────────────────────────┬─────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ gateway_api — Response │ │ │ │ HTTP 200 OK │ │ {"transaction_id": "…", "status": "approved", …} │ └─────────────────────────────────────────────────────────┘ ``` --- ## Asynchronous Flow — File Ingestion (SFTP → Data Warehouse) ``` SFTP Server │ │ (new file detected) ▼ ┌─────────────────────────────────────────────────────────┐ │ adapter_file — FileWatcher (GenServer, polls every 60s)│ │ │ │ 1. SFTPClient.list_new_files(config) │ │ 2. SFTPClient.download(file_path) │ │ 3. CsvParser.stream(file_contents) │ │ 4. emit rows as Broadway messages │ └───────────────────────────┬─────────────────────────────┘ │ Broadway.Messages ▼ ┌─────────────────────────────────────────────────────────┐ │ infra_queue — Broadway Pipeline (file_pipeline) │ │ │ │ Producer: adapter_file emits row batches │ │ Processor: MwTransform.map(row) → MwKernel.Message │ │ Batcher: collect N messages → adapter_dw.bulk_load │ │ DLQ: failed rows → dead_letter table in DB │ └───────────────────────────┬─────────────────────────────┘ │ bulk batch ▼ ┌─────────────────────────────────────────────────────────┐ │ adapter_dw — Batch Loader │ │ │ │ 1. BatchLoader.prepare(messages) │ │ 2. Client.bulk_insert(dw_payload) │ ──► Data Warehouse │ 3. acknowledge Broadway batch │ │ 4. emit completion event → PubSub │ └───────────────────────────┬─────────────────────────────┘ │ PubSub: "file_batch:completed" ▼ ┌─────────────────────────────────────────────────────────┐ │ gateway_ws — notification_channel │ │ │ │ Phoenix.PubSub subscriber pushes to connected clients: │ │ {"event": "batch_complete", "file": "…", "rows": 1200} │ └─────────────────────────────────────────────────────────┘ ``` --- ## WebSocket Connection Lifecycle ``` Mobile / Web Client │ │ wss://host/socket/websocket?token= ▼ ┌─────────────────────────────────────────────────────────┐ │ gateway_ws — UserSocket.connect/3 │ │ │ │ 1. MwAuth.JWT.verify(token) │ │ 2. assign user_id, tenant_id to socket │ │ 3. return {:ok, socket} or :error │ └───────────────────────────┬─────────────────────────────┘ │ │ channel join ▼ ┌─────────────────────────────────────────────────────────┐ │ TransactionChannel.join("transactions:" <> tx_id, …) │ │ │ │ 1. verify user owns this transaction │ │ 2. Phoenix.PubSub.subscribe(topic) │ │ 3. send current state snapshot to client │ └─────────────────────────────────────────────────────────┘ Later — when core banking callback arrives: │ ▼ ┌─────────────────────────────────────────────────────────┐ │ adapter_banking receives async status callback │ │ → Phoenix.PubSub.broadcast("transactions:" <> tx_id, │ │ %{status: "settled", …}) │ │ → TransactionChannel pushes to connected client │ └─────────────────────────────────────────────────────────┘ ``` --- ## Error Paths ### Authentication Failure ``` AuthPlug.call → MwAuth.JWT.verify returns {:error, :expired} → Pipeline halts (Plug.Conn.halt) → 401 {"error": "token_expired"} → Audit event written with status: :auth_failure ``` ### Circuit Breaker Open ``` RoutePlug.call → MwRouter.CircuitBreaker.check(:fuse, AdapterBanking) → returns {:error, :circuit_open} → Pipeline halts → 503 {"error": "downstream_unavailable", "retry_after": 30} → Audit event written with status: :circuit_open ``` ### Adapter Timeout ``` Dispatcher.dispatch → AdapterBanking.send → {:error, :timeout} → :fuse.melt(:adapter_banking) — record failure → if failures exceed threshold → circuit opens → 504 {"error": "gateway_timeout"} → Audit event written with status: :timeout → Telemetry event: adapter.timeout counter incremented ``` ### Broadway DLQ ``` Broadway processor raises exception on malformed CSV row → Broadway moves message to DLQ batch → DLQ batcher writes to dead_letter_queue DB table → Audit event written with source_file, row_number, error → Admin UI shows DLQ depth in dashboard ```