diff --git a/api/main.py b/api/main.py index fdd9303..173d27c 100644 --- a/api/main.py +++ b/api/main.py @@ -13,7 +13,7 @@ hands off responsibility to the message queue. This separation of concerns is fundamental to scalable microservices architecture. """ -from fastapi import FastAPI, Depends, HTTPException, status +from fastapi import FastAPI, Depends, HTTPException, status, WebSocket, WebSocketDisconnect from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel @@ -23,6 +23,8 @@ import json import uuid import time import logging +import redis +import threading # ══════════════════════════════════════════════════════════════════════════════ # CONFIGURATION @@ -85,6 +87,66 @@ def get_kafka_producer(): raise return kafka_producer +# ══════════════════════════════════════════════════════════════════════════════ +# REDIS & WEBSOCKET PUBSUB +# ══════════════════════════════════════════════════════════════════════════════ + +redis_client = None +connected_clients = set() +pubsub_thread = None + +def get_redis_client(): + """Get or create Redis client""" + global redis_client + if redis_client is None: + try: + redis_client = redis.Redis( + host="redis", + port=6379, + decode_responses=True + ) + redis_client.ping() + logger.info("[REDIS] Connected to Redis") + except Exception as e: + logger.warning(f"[REDIS] Connection failed: {e}") + redis_client = None + raise + return redis_client + +def start_pubsub_listener(): + """Start background thread to listen for fraud results on Redis pub/sub""" + global pubsub_thread + + def listen_and_broadcast(): + try: + redis_conn = get_redis_client() + pubsub = redis_conn.pubsub() + pubsub.subscribe("fraud_results") + logger.info("[PUBSUB] Listening to 'fraud_results' channel") + + for message in pubsub.listen(): + if message["type"] == "message": + try: + data = json.loads(message["data"]) + # Broadcast to all connected WebSocket clients + for client in list(connected_clients): + try: + if not client.client_state.value == 0: # Check if still connected + client.send_json(data) + except Exception as e: + logger.warning(f"[PUBSUB] Failed to send to client: {e}") + try: + connected_clients.discard(client) + except: + pass + except Exception as e: + logger.error(f"[PUBSUB] Failed to parse message: {e}") + except Exception as e: + logger.error(f"[PUBSUB] Listener error: {e}") + + pubsub_thread = threading.Thread(target=listen_and_broadcast, daemon=True) + pubsub_thread.start() + # ══════════════════════════════════════════════════════════════════════════════ # AUTHENTICATION # ══════════════════════════════════════════════════════════════════════════════ @@ -211,6 +273,34 @@ async def get_demo_token(user_id: str): logger.info(f"[JWT GENERATION] Generated token for user={user_id}") return {"access_token": token, "token_type": "bearer"} +@app.websocket("/ws/fraud-results") +async def websocket_fraud_results(websocket: WebSocket): + """ + WebSocket endpoint for real-time fraud detection results. + + Clients connect and receive live updates when fraud detection completes. + Results are streamed from Redis pub/sub channel via background listener. + """ + await websocket.accept() + connected_clients.add(websocket) + logger.info(f"[WEBSOCKET] Client connected. Total clients: {len(connected_clients)}") + + try: + # Keep connection alive while client is connected + while True: + # Wait for any client message (optional heartbeat/ping) + data = await websocket.receive_text() + logger.debug(f"[WEBSOCKET] Received from client: {data}") + except WebSocketDisconnect: + connected_clients.discard(websocket) + logger.info(f"[WEBSOCKET] Client disconnected. Total clients: {len(connected_clients)}") + except Exception as e: + logger.error(f"[WEBSOCKET] Error: {e}") + try: + connected_clients.discard(websocket) + except: + pass + # ══════════════════════════════════════════════════════════════════════════════ # STARTUP / SHUTDOWN # ══════════════════════════════════════════════════════════════════════════════ @@ -224,6 +314,13 @@ async def startup_event(): logger.info(f"Topic: {KAFKA_TOPIC}") logger.info("Listening for authenticated transaction requests...") logger.info("="*80) + + # Start Redis pub/sub listener for WebSocket broadcasts + try: + get_redis_client() + start_pubsub_listener() + except Exception as e: + logger.warning(f"[STARTUP] Redis pub/sub not available: {e}") @app.on_event("shutdown") async def shutdown_event(): diff --git a/consumer/decision.py b/consumer/decision.py index 437d246..f685173 100644 --- a/consumer/decision.py +++ b/consumer/decision.py @@ -122,8 +122,12 @@ class DecisionEngine: return decision def store_result(self, transaction_id: str, decision: str, - layer1_score: float, layer2_score: float = None) -> None: - """Store decision result in Redis for retrieval""" + layer1_score: float, layer2_score: float = None, + transaction_data: dict = None) -> None: + """Store decision result in Redis and publish to pub/sub channel""" + import json + from datetime import datetime + key = f"result:{transaction_id}" value = { "decision": decision, @@ -136,3 +140,25 @@ class DecisionEngine: self.redis_client.setex(key, 3600, str(value)) logger.info(f"[RESULT STORAGE] Stored decision in Redis: {key} = {decision}") + + # Publish to pub/sub channel for real-time WebSocket updates + try: + # Build message with transaction details for frontend + message = { + "transaction_id": transaction_id, + "decision": decision, + "fraud_score": round(layer2_score, 4) if layer2_score is not None else round(layer1_score, 4), + "layer1_score": round(layer1_score, 4), + "timestamp": datetime.utcnow().isoformat() + "Z" + } + + # Add transaction data if provided (amount, merchant) + if transaction_data: + message["amount"] = transaction_data.get("amount") + message["merchant"] = transaction_data.get("merchant") + + # Publish to Redis pub/sub channel + self.redis_client.publish("fraud_results", json.dumps(message)) + logger.info(f"[RESULT STORAGE] Published result to pub/sub channel: {transaction_id[:8]}... → {decision}") + except Exception as e: + logger.error(f"[RESULT STORAGE] Failed to publish to pub/sub: {str(e)}") diff --git a/consumer/pipeline.py b/consumer/pipeline.py index f95cdd6..b146c09 100644 --- a/consumer/pipeline.py +++ b/consumer/pipeline.py @@ -193,7 +193,8 @@ def process_transaction(transaction, components): transaction_id=transaction_id, decision=decision, layer1_score=layer1_score, - layer2_score=layer2_score + layer2_score=layer2_score, + transaction_data=transaction # Pass full transaction for pub/sub message ) # ───────────────────────────────────────────────────────────────────────── diff --git a/frontend/src/App.css b/frontend/src/App.css index f5fa690..5eb44d3 100644 --- a/frontend/src/App.css +++ b/frontend/src/App.css @@ -345,3 +345,170 @@ font-size: 0.8rem; } } + +/* Consumer Status Card */ +.consumer-card { + border-top: 4px solid var(--primary); + background: linear-gradient(135deg, #f8fafc 0%, #f1f5f9 100%); +} + +.consumer-content { + display: flex; + flex-direction: column; + gap: 1rem; +} + +.connection-status { + display: flex; + justify-content: center; +} + +.connection-status .status-badge { + font-size: 0.875rem; + padding: 0.5rem 1rem; +} + +.connection-status .status-badge.connected { + background: var(--success); + color: white; +} + +.connection-status .status-badge.disconnected { + background: var(--error); + color: white; +} + +.empty-state { + text-align: center; + padding: 2rem 1rem; + color: var(--text-light); +} + +.empty-state p { + margin: 0.5rem 0; +} + +.consumer-status-details { + display: flex; + flex-direction: column; + gap: 0.75rem; + background: white; + padding: 1rem; + border-radius: 0.375rem; +} + +.status-row { + display: flex; + justify-content: space-between; + align-items: center; + padding: 0.5rem 0; + border-bottom: 1px solid var(--border); +} + +.status-row:last-child { + border-bottom: none; +} + +.status-row label { + font-weight: 600; + color: var(--text); + font-size: 0.875rem; + min-width: 120px; +} + +.status-row span { + color: var(--text-light); + font-size: 0.875rem; + text-align: right; + flex: 1; + margin-left: 1rem; +} + +/* Pagination Controls */ +.pagination-controls { + display: flex; + flex-direction: column; + gap: 1rem; + padding: 1rem; + background: var(--bg); + border-radius: 0.375rem; + margin-bottom: 1rem; +} + +.pagination-info { + text-align: center; + font-size: 0.875rem; + color: var(--text-light); + font-weight: 500; +} + +.pagination-buttons { + display: flex; + align-items: center; + justify-content: center; + gap: 1rem; +} + +.btn-pagination { + display: flex; + align-items: center; + gap: 0.5rem; + background: var(--primary); + color: white; + padding: 0.5rem 1rem; + border: none; + border-radius: 0.375rem; + font-size: 0.875rem; + font-weight: 500; + cursor: pointer; + transition: all 0.2s ease; + min-width: 100px; + justify-content: center; +} + +.btn-pagination:hover:not(:disabled) { + background: var(--primary-dark); + transform: translateY(-2px); + box-shadow: 0 4px 12px rgba(59, 130, 246, 0.3); +} + +.btn-pagination:disabled { + opacity: 0.5; + cursor: not-allowed; +} + +.page-indicator { + font-size: 0.875rem; + font-weight: 600; + color: var(--text); + min-width: 120px; + text-align: center; +} + +.page-size-selector { + display: flex; + align-items: center; + justify-content: center; + gap: 0.75rem; +} + +.page-size-selector label { + font-size: 0.875rem; + font-weight: 500; + color: var(--text); +} + +.select-input { + padding: 0.5rem; + border: 1px solid var(--border); + border-radius: 0.375rem; + font-size: 0.875rem; + cursor: pointer; + background: white; +} + +.select-input:focus { + outline: none; + border-color: var(--primary); + box-shadow: 0 0 0 3px rgba(59, 130, 246, 0.1); +} diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index fc2a987..02f5867 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1,5 +1,5 @@ -import React, { useState } from 'react'; -import { AlertCircle, CheckCircle, Clock, Lock, Send } from 'lucide-react'; +import React, { useState, useEffect, useRef } from 'react'; +import { AlertCircle, CheckCircle, Clock, Lock, Send, Wifi, WifiOff, ChevronLeft, ChevronRight } from 'lucide-react'; import axios from 'axios'; import './App.css'; @@ -28,6 +28,15 @@ interface Result { timestamp: string; } +interface FraudResult { + transaction_id: string; + decision: string; + fraud_score: number; + amount?: number; + merchant?: string; + timestamp: string; +} + // Use localhost when browser accesses from host machine const API_BASE_URL = 'http://localhost:8000'; @@ -46,9 +55,72 @@ const App: React.FC = () => { }); const [results, setResults] = useState([]); + const [consumerStatus, setConsumerStatus] = useState(null); + const [wsConnected, setWsConnected] = useState(false); const [loading, setLoading] = useState(false); const [error, setError] = useState(null); const [success, setSuccess] = useState(null); + + // Pagination state + const [currentPage, setCurrentPage] = useState(1); + const [pageSize, setPageSize] = useState(25); + + const wsRef = useRef(null); + const reconnectTimeoutRef = useRef | null>(null); + + // WebSocket connection management + useEffect(() => { + const connectWebSocket = () => { + try { + const wsUrl = 'ws://localhost:8000/ws/fraud-results'; + const ws = new WebSocket(wsUrl); + + ws.onopen = () => { + console.log('[WEBSOCKET] Connected to fraud results stream'); + setWsConnected(true); + }; + + ws.onmessage = (event) => { + try { + const data = JSON.parse(event.data) as FraudResult; + console.log('[WEBSOCKET] Received fraud result:', data); + setConsumerStatus(data); + } catch (e) { + console.error('[WEBSOCKET] Failed to parse message:', e); + } + }; + + ws.onerror = (error) => { + console.error('[WEBSOCKET] Error:', error); + setWsConnected(false); + }; + + ws.onclose = () => { + console.log('[WEBSOCKET] Disconnected, attempting reconnect...'); + setWsConnected(false); + // Exponential backoff reconnection: 1s → 2s → 4s → 8s + const delay = Math.min(1000 * Math.pow(2, 3), 8000); + reconnectTimeoutRef.current = setTimeout(connectWebSocket, delay); + }; + + wsRef.current = ws; + } catch (e) { + console.error('[WEBSOCKET] Connection failed:', e); + setWsConnected(false); + } + }; + + connectWebSocket(); + + return () => { + if (wsRef.current) { + wsRef.current.close(); + } + if (reconnectTimeoutRef.current) { + clearTimeout(reconnectTimeoutRef.current); + } + }; + }, []); // Generate JWT token const generateToken = async () => { @@ -173,6 +245,19 @@ const App: React.FC = () => { }); }; + // Pagination helpers + const totalPages = Math.ceil(results.length / pageSize); + const paginatedResults = results.slice( + (currentPage - 1) * pageSize, + currentPage * pageSize + ); + + const getFraudScoreColor = (score: number): string => { + if (score < 0.3) return '#10b981'; // Green - low risk + if (score < 0.7) return '#f59e0b'; // Yellow - medium risk + return '#ef4444'; // Red - high risk + }; + return (
@@ -181,6 +266,88 @@ const App: React.FC = () => {
+ {/* Consumer Status Panel */} +
+
+
+ {wsConnected ? ( + + ) : ( + + )} +

Real-Time Consumer Pipeline

+
+
+
+
+ + {wsConnected ? '● Connected' : '● Disconnected'} + +
+ + {consumerStatus ? ( +
+
+ + {consumerStatus.transaction_id.substring(0, 12)}... +
+ {consumerStatus.amount && ( +
+ + ${consumerStatus.amount.toFixed(2)} +
+ )} + {consumerStatus.merchant && ( +
+ + {consumerStatus.merchant} +
+ )} +
+ +
+ {(consumerStatus.fraud_score * 100).toFixed(1)}% +
+
+
+ +
+ {consumerStatus.decision} +
+
+
+ + {new Date(consumerStatus.timestamp).toLocaleString()} +
+
+ ) : ( +
+

Waiting for fraud detection results...

+

+ Submit transactions to see results +

+
+ )} +
+
+ {/* Authentication Section */}
@@ -308,8 +475,58 @@ const App: React.FC = () => {

Transaction Results

+ + {/* Pagination Controls */} +
+
+ + Showing {paginatedResults.length > 0 ? (currentPage - 1) * pageSize + 1 : 0}– + {Math.min(currentPage * pageSize, results.length)} of {results.length} transactions + +
+ +
+ + + + Page {currentPage} of {totalPages || 1} + + + +
+ +
+ + +
+
+ + {/* Results List */}
- {results.map((result, idx) => ( + {paginatedResults.map((result, idx) => (