diff --git a/.github/instructions/snyk_rules.instructions.md b/.github/instructions/snyk_rules.instructions.md deleted file mode 100644 index eef0634..0000000 --- a/.github/instructions/snyk_rules.instructions.md +++ /dev/null @@ -1,14 +0,0 @@ ---- -alwaysApply: true -always_on: true -trigger: always_on -applyTo: "**" -description: Snyk Security At Inception ---- - -# Project security best practices - -- Always run snyk_code_scan tool for new first party code that is generated in a Snyk-supported language. -- If any security issues are found based on newly introduced or modified code or dependencies, attempt to fix the issues using the results context from Snyk. -- Rescan the code after fixing the issues to ensure that the issues were fixed and that there are no newly introduced issues. -- Repeat this process until no new issues are found. diff --git a/README.md b/README.md index 8c70d9a..a9cb40b 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,20 @@ This will start: - Frontend: http://localhost:3000 - Backend API: http://localhost:3005 (health endpoint: `/health`) +Local development with Docker (hot reload) + +Use the development compose file to run both services with live reload and all required env vars configured: + +```bash +docker compose -f docker-compose.dev.yml up --build +``` + +This will start: + +- Frontend (dev server): http://localhost:3000 +- Backend API (dev): http://localhost:3005 +- WebSocket (dev): ws://localhost:8080 + Local development (without Docker) - Frontend: @@ -47,6 +61,46 @@ Tests - Frontend E2E: run Cypress after starting the frontend with `npm run cypress:open` for local `npm run cypress:run` for CI/CD pipelines - Backend unit test run in `server/market-trading-service` with `npm run test` (Jest), `npm run test:watch`, `npm run test:coverage` also available if working on test and see the coverage +## Assumptions & Trade-offs + +### Assumptions + +- Data model: simplified ticker domain (price, change, volume, 24h high/low); no corporate actions, splits, multi-currency, or latency-sensitive guarantees. + +- Simulation: price history and live ticks are synthetic for UX validation, not financial accuracy. + +- Environment: single-node backend with in-memory storage is acceptable for this challenge; no cross-process persistence required. + +- Client: no auth is required to view market data. + +- Contracts: WebSocket message shapes are minimal and stable: `{"type":"connected","payload":{"clientId"}}` and `{"type":"data","payload":{"ticker"}}`. REST endpoints are reachable for initial bootstrap/fallback. + +- CI/CD via github actions to run test on PR + +- Used Code Rabbit to review + +### Trade-offs + +- Real-time delivery vs simplicity: WebSocket for live updates plus a one-time REST bootstrap for fast first paint (slight duplication accepted for responsiveness). + +- Subscription scope: initial subscribe-to-all for clarity; future optimization could subscribe only to visible/selected symbols. + +- Update cadence: immediate per-tick broadcasts, no batching/debouncing; simple but more frames under heavy load. Batching window (e.g., 50–150 ms) can be added later. + +- State storage: in-memory repository and subscription registry (per-process); not horizontally scalable without a shared store/pub-sub or sticky sessions. + +- Consistency vs responsiveness: values rounded for display and updated on each `data` frame; suited for UI, not for reconciliation. + +- Error handling: if REST fails, fall back to mock data; if WS drops, auto-reconnect and keep last-known data. Prioritizes resilience for demos. + +- Client architecture: Next.js app with a WebSocket Provider context and React Query; SSR of live data is out of scope to keep the real-time logic client-side and testable. + +- Testing: REST fallback and provider no-op defaults keep unit tests stable; deep WS integration tests are limited and can be added with a small WS mock. + +- Selection model: selection tracked by symbol and derived from the latest ticker list to ensure live updates without stale references. + +- Some of the Bonus features aren't covered due to time constrain. + License This project is available under the repository LICENSE file. diff --git a/client/trading-dashboard/app/(dashboard)/dashboard/page.tsx b/client/trading-dashboard/app/(dashboard)/dashboard/page.tsx index 1884d1d..c0e40e2 100644 --- a/client/trading-dashboard/app/(dashboard)/dashboard/page.tsx +++ b/client/trading-dashboard/app/(dashboard)/dashboard/page.tsx @@ -37,6 +37,8 @@ const TradingDashboard = () => { }); }; + + const handleSelectTicker = (ticker: Ticker) => { setSelectedTicker(ticker); setSidebarOpen(false); // Close sidebar on mobile after selection diff --git a/client/trading-dashboard/app/Providers.tsx b/client/trading-dashboard/app/Providers.tsx index 6da20df..ef6d2a4 100644 --- a/client/trading-dashboard/app/Providers.tsx +++ b/client/trading-dashboard/app/Providers.tsx @@ -3,6 +3,7 @@ import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; import { ReactQueryDevtools } from "@tanstack/react-query-devtools"; import { useState } from "react"; +import { WebSocketProvider } from "@/providers/WebSocketProvider"; export default function Providers({ children }: { children: React.ReactNode }) { const [queryClient] = useState( @@ -19,7 +20,7 @@ export default function Providers({ children }: { children: React.ReactNode }) { return ( - {children} + {children} ); diff --git a/client/trading-dashboard/components/Footer.tsx b/client/trading-dashboard/components/Footer.tsx index 069e0e7..7e24a0c 100644 --- a/client/trading-dashboard/components/Footer.tsx +++ b/client/trading-dashboard/components/Footer.tsx @@ -1,5 +1,3 @@ -import { BarChart3 } from "lucide-react"; - const Footer = () => { const year = new Date().getFullYear(); return ( diff --git a/client/trading-dashboard/components/TickerGrid.tsx b/client/trading-dashboard/components/TickerGrid.tsx index ed2dffb..a24336d 100644 --- a/client/trading-dashboard/components/TickerGrid.tsx +++ b/client/trading-dashboard/components/TickerGrid.tsx @@ -3,7 +3,7 @@ import TickerCard from "./TickerCard"; import { Ticker } from "@/types"; import { useQuery } from "@tanstack/react-query"; import { API_BASE_URL, MOCK_TICKERS } from "@/constants"; -import { LoadingSpinner } from "./ui/LoadingSpinner"; +import { useWebSocketContext } from "@/providers/WebSocketProvider"; async function getPosts(): Promise { const res = await fetch(`${API_BASE_URL}/tickers`); @@ -18,14 +18,21 @@ async function getPosts(): Promise { return tickers; } const TickerGrid = () => { + const { tickers: wsTickers } = useWebSocketContext(); + const useFallback = !wsTickers || wsTickers.length === 0; const { data, error } = useQuery({ queryKey: ["tickers"], queryFn: getPosts, - refetchInterval: 1000, + refetchInterval: useFallback ? 1000 : false, placeholderData: MOCK_TICKERS, + enabled: useFallback, }); - if (error) { + const tickersToShow: Ticker[] = useFallback + ? data ?? MOCK_TICKERS + : wsTickers; + + if (error && useFallback) { console.error("Error fetching tickers:", error); } @@ -45,9 +52,9 @@ const TickerGrid = () => { data-testid="ticker-grid" className="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-3 gap-6" > - {data && - data.length > 0 && - data?.map((ticker) => ( + {tickersToShow && + tickersToShow.length > 0 && + tickersToShow.map((ticker) => ( ))} diff --git a/client/trading-dashboard/constants/index.ts b/client/trading-dashboard/constants/index.ts index 85174ee..dded089 100644 --- a/client/trading-dashboard/constants/index.ts +++ b/client/trading-dashboard/constants/index.ts @@ -1,9 +1,18 @@ // All constants import { Ticker } from "@/types"; -export const API_BASE_URL = - `${process.env.NEXT_PUBLIC_MARKET_TRADING_URL}/api` || - "http://localhost:3005/api"; +// Ensure sane defaults even when env vars are missing at build time +const PUBLIC_API_BASE = + process.env.NEXT_PUBLIC_MARKET_TRADING_URL && + process.env.NEXT_PUBLIC_MARKET_TRADING_URL.trim() !== "" + ? process.env.NEXT_PUBLIC_MARKET_TRADING_URL + : "http://localhost:3005"; +export const API_BASE_URL = `${PUBLIC_API_BASE}/api`; + +export const WS_URL = + process.env.NEXT_PUBLIC_WS_URL && process.env.NEXT_PUBLIC_WS_URL.trim() !== "" + ? process.env.NEXT_PUBLIC_WS_URL + : "ws://localhost:8080"; export const MOCK_TICKERS: Ticker[] = [ { diff --git a/client/trading-dashboard/example.env b/client/trading-dashboard/example.env index 2537c4c..b93f6a4 100644 --- a/client/trading-dashboard/example.env +++ b/client/trading-dashboard/example.env @@ -1,2 +1,3 @@ NEXT_PUBLIC_BASE_URL=http://localhost:3000 -NEXT_PUBLIC_MARKET_TRADING_URL=http://localhost:3005 \ No newline at end of file +NEXT_PUBLIC_MARKET_TRADING_URL=http://localhost:3005 +NEXT_PUBLIC_WS_URL=ws://localhost:8080 \ No newline at end of file diff --git a/client/trading-dashboard/hooks/useTradingData.ts b/client/trading-dashboard/hooks/useTradingData.ts index c0b41d2..07afd93 100644 --- a/client/trading-dashboard/hooks/useTradingData.ts +++ b/client/trading-dashboard/hooks/useTradingData.ts @@ -1,18 +1,23 @@ -import { useState, useRef, useEffect } from "react"; +import { useState, useRef, useEffect, useMemo } from "react"; import { useQuery } from "@tanstack/react-query"; import { Ticker, HistoricalData } from "@/types"; import { API_BASE_URL, MOCK_TICKERS } from "@/constants"; import { generateMockHistory } from "@/lib/utils"; +import { useWebSocketContext } from "@/providers/WebSocketProvider"; export const useTradingData = () => { - const [selectedTicker, setSelectedTicker] = useState(null); + const [selectedSymbol, setSelectedSymbol] = useState(null); const [chartDays, setChartDays] = useState(7); const [error, setError] = useState(null); const priceStatus = useRef<{ [key: string]: "up" | "down" | "neutral" }>({}); const previousTickers = useRef([]); - // Fetch tickers with automatic polling every 2 seconds for tickers - const { data: tickers = [], isLoading: tickersLoading } = useQuery({ + const { tickers: wsTickers, priceStatus: wsPriceStatus } = + useWebSocketContext(); + + // REST fallback only if WS not yet providing data + const useFallback = !wsTickers || wsTickers.length === 0; + const { data: restTickers = [], isLoading: restLoading } = useQuery({ queryKey: ["tickers"], queryFn: async () => { try { @@ -29,12 +34,21 @@ export const useTradingData = () => { return MOCK_TICKERS; } }, - refetchInterval: 2000, + enabled: useFallback, + refetchInterval: useFallback ? 2000 : false, staleTime: 0, }); - // Track price changes for animations + const tickers: Ticker[] = useFallback ? restTickers : wsTickers; + + // Track price changes for animations (fallback when WS not in use) useEffect(() => { + if (!useFallback) { + // When using WS, price status is provided by the provider + priceStatus.current = wsPriceStatus; + return; + } + if (previousTickers.current.length > 0 && tickers.length > 0) { tickers.forEach((newTicker) => { const oldTicker = previousTickers.current.find( @@ -50,14 +64,20 @@ export const useTradingData = () => { }); } previousTickers.current = tickers; - }, [tickers]); + }, [tickers, useFallback, wsPriceStatus]); - // Set initial selected ticker + // Set initial selected symbol useEffect(() => { - if (!selectedTicker && tickers.length > 0) { - setSelectedTicker(tickers[0]); + if (!selectedSymbol && tickers.length > 0) { + setSelectedSymbol(tickers[0].symbol); } - }, [tickers, selectedTicker]); + }, [tickers, selectedSymbol]); + + // Derive the selected ticker from the latest tickers list so it always stays fresh + const selectedTicker: Ticker | null = useMemo(() => { + if (!selectedSymbol) return null; + return tickers.find((t) => t.symbol === selectedSymbol) ?? null; + }, [tickers, selectedSymbol]); // Fetch chart data when ticker or days change const { data: chartData = [], isLoading: chartLoading } = useQuery({ @@ -83,11 +103,14 @@ export const useTradingData = () => { return { tickers, selectedTicker, - setSelectedTicker, + setSelectedTicker: (ticker: Ticker) => setSelectedSymbol(ticker.symbol), chartData, chartDays, setChartDays, - loading: { tickers: tickersLoading, chart: chartLoading }, + loading: { + tickers: useFallback ? restLoading : false, + chart: chartLoading, + }, error, priceStatus: priceStatus.current, }; diff --git a/client/trading-dashboard/package-lock.json b/client/trading-dashboard/package-lock.json index a222087..43c7184 100644 --- a/client/trading-dashboard/package-lock.json +++ b/client/trading-dashboard/package-lock.json @@ -17,6 +17,7 @@ "next": "16.0.3", "react": "19.2.0", "react-dom": "19.2.0", + "react-use-websocket": "^4.13.0", "recharts": "^3.4.1", "tailwind-merge": "^3.4.0" }, @@ -5684,7 +5685,6 @@ "integrity": "sha512-rRqJg/6gd538VHvR3PSrdRBb/1Vy2YfzHqzvbhGIQpDRKIa4FgV/54b5Q1xYSxOOwKvjXweS26E0Q+nAMwp2pQ==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "ansi-colors": "^4.1.1", "strip-ansi": "^6.0.1" @@ -10972,6 +10972,12 @@ } } }, + "node_modules/react-use-websocket": { + "version": "4.13.0", + "resolved": "https://registry.npmjs.org/react-use-websocket/-/react-use-websocket-4.13.0.tgz", + "integrity": "sha512-anMuVoV//g2N76Wxqvqjjo1X48r9Np3y1/gMl7arX84tAPXdy5R7sB5lO5hvCzQRYjqXwV8XMAiEBOUbyrZFrw==", + "license": "MIT" + }, "node_modules/recharts": { "version": "3.4.1", "resolved": "https://registry.npmjs.org/recharts/-/recharts-3.4.1.tgz", diff --git a/client/trading-dashboard/package.json b/client/trading-dashboard/package.json index edf58ec..1092235 100644 --- a/client/trading-dashboard/package.json +++ b/client/trading-dashboard/package.json @@ -23,6 +23,7 @@ "next": "16.0.3", "react": "19.2.0", "react-dom": "19.2.0", + "react-use-websocket": "^4.13.0", "recharts": "^3.4.1", "tailwind-merge": "^3.4.0" }, diff --git a/client/trading-dashboard/providers/WebSocketProvider.tsx b/client/trading-dashboard/providers/WebSocketProvider.tsx new file mode 100644 index 0000000..406d096 --- /dev/null +++ b/client/trading-dashboard/providers/WebSocketProvider.tsx @@ -0,0 +1,213 @@ +"use client"; + +import React, { + createContext, + useContext, + useEffect, + useMemo, + useRef, + useState, +} from "react"; +import useWebSocket, { ReadyState } from "react-use-websocket"; +import { API_BASE_URL, WS_URL, MOCK_TICKERS } from "@/constants"; +import type { Ticker } from "@/types"; + +type PriceStatus = "up" | "down" | "neutral"; + +interface WebSocketContextValue { + connected: boolean; + clientId?: string; + tickers: Ticker[]; + priceStatus: Record; + subscribe: (symbols: string[]) => void; + unsubscribe: (symbols: string[]) => void; +} + +const WebSocketContext = createContext( + undefined +); + +export const WebSocketProvider: React.FC<{ children: React.ReactNode }> = ({ + children, +}) => { + const [clientId, setClientId] = useState(); + const [tickersMap, setTickersMap] = useState>({}); + const [priceStatus, setPriceStatus] = useState>( + {} + ); + const subscribed = useRef>(new Set()); + + // Keep previous prices to determine direction + const prevPrices = useRef>({}); + + const { sendJsonMessage, readyState, lastMessage } = useWebSocket(WS_URL, { + shouldReconnect: () => true, + reconnectAttempts: 10, + reconnectInterval: 2000, + onOpen: () => { + // After open, if we already have initial list, resubscribe + const symbols = Array.from(subscribed.current); + if (symbols.length) { + sendJsonMessage({ type: "subscribe", payload: { symbols } }); + } + }, + onError: (e) => { + // Non-fatal; UI will still show initial REST data + console.error("WebSocket error:", e); + }, + }); + + const connected = readyState === ReadyState.OPEN; + + // Initial bootstrap: fetch tickers once, populate map, and subscribe to all + useEffect(() => { + let cancelled = false; + const bootstrap = async () => { + try { + const res = await fetch(`${API_BASE_URL}/tickers`); + if (!res.ok) throw new Error("Failed to fetch initial tickers"); + const json = await res.json(); + const initialTickers: Ticker[] = json?.data?.tickers ?? MOCK_TICKERS; + + if (cancelled) return; + + setTickersMap((prev) => { + const next: Record = { ...prev }; + for (const t of initialTickers) { + next[t.symbol] = t; + prevPrices.current[t.symbol] = t.price; + } + return next; + }); + + const symbols = initialTickers.map((t) => t.symbol.toUpperCase()); + symbols.forEach((s) => subscribed.current.add(s)); + + // Subscribe via WS when available (or once connected) + sendJsonMessage({ type: "subscribe", payload: { symbols } }); + } catch (e) { + console.error("Failed to bootstrap tickers:", e); + // Populate with mock to keep UI alive + setTickersMap((prev) => { + const next: Record = { ...prev }; + for (const t of MOCK_TICKERS) { + next[t.symbol] = t; + prevPrices.current[t.symbol] = t.price; + } + return next; + }); + } + }; + + bootstrap(); + return () => { + cancelled = true; + }; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, []); + + // Handle incoming WS messages + useEffect(() => { + if (!lastMessage?.data) return; + try { + const data = JSON.parse(lastMessage.data as string); + const { type, payload } = data || {}; + + if (type === "connected" && payload?.clientId) { + setClientId(payload.clientId); + } + + if (type === "data" && payload?.ticker) { + const incoming: Ticker = normalizeTicker(payload.ticker); + + setTickersMap((prev) => { + const prevTicker = prev[incoming.symbol]; + const next: Record = { + ...prev, + [incoming.symbol]: incoming, + }; + + // Determine price direction for animation/status + const oldPrice = + prevTicker?.price ?? prevPrices.current[incoming.symbol]; + if (typeof oldPrice === "number") { + if (incoming.price > oldPrice) { + setPriceStatus((ps) => ({ ...ps, [incoming.symbol]: "up" })); + } else if (incoming.price < oldPrice) { + setPriceStatus((ps) => ({ ...ps, [incoming.symbol]: "down" })); + } + } + prevPrices.current[incoming.symbol] = incoming.price; + return next; + }); + } + } catch (e) { + // ignore malformed messages + } + }, [lastMessage]); + + const tickers = useMemo(() => Object.values(tickersMap), [tickersMap]); + + const subscribe = (symbols: string[]) => { + const list = symbols.map((s) => s.toUpperCase()); + list.forEach((s) => subscribed.current.add(s)); + sendJsonMessage({ type: "subscribe", payload: { symbols: list } }); + }; + + const unsubscribe = (symbols: string[]) => { + const list = symbols.map((s) => s.toUpperCase()); + list.forEach((s) => subscribed.current.delete(s)); + sendJsonMessage({ type: "unsubscribe", payload: { symbols: list } }); + }; + + const value = useMemo( + () => ({ + connected, + clientId, + tickers, + priceStatus, + subscribe, + unsubscribe, + }), + [connected, clientId, tickers, priceStatus] + ); + + return ( + + {children} + + ); +}; + +function normalizeTicker(t: any): Ticker { + return { + symbol: String(t.symbol), + name: String(t.name), + price: Number(t.price), + previousClose: Number(t.previousClose ?? t.price), + change: Number(t.change), + changePercent: Number(t.changePercent), + volume: Number(t.volume), + high24h: Number(t.high24h), + low24h: Number(t.low24h), + lastUpdate: + typeof t.lastUpdate === "string" + ? t.lastUpdate + : new Date(t.lastUpdate).toISOString(), + } as Ticker; +} + +export const useWebSocketContext = () => { + const ctx = useContext(WebSocketContext); + if (!ctx) { + // Provide a safe fallback when not wrapped by provider (e.g., unit tests) + return { + connected: false, + tickers: [] as Ticker[], + priceStatus: {} as Record, + subscribe: () => {}, + unsubscribe: () => {}, + } as WebSocketContextValue; + } + return ctx; +}; diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml new file mode 100644 index 0000000..c3d4642 --- /dev/null +++ b/docker-compose.dev.yml @@ -0,0 +1,34 @@ +services: + server: + image: node:20-alpine + working_dir: /app + ports: + - "3005:3005" + - "8080:8080" + environment: + - PORT=3005 + - NODE_ENV=development + - ALLOWED_ORIGINS=http://localhost:3000 + - PRICE_UPDATE_INTERVAL=2000 + - WS_PORT=8080 + volumes: + - ./server/market-trading-service:/app + - /app/node_modules + command: sh -c "npm install && npm run dev" + + client: + image: node:20-alpine + working_dir: /app + ports: + - "3000:3000" + environment: + - NODE_ENV=development + - NEXT_TELEMETRY_DISABLED=1 + - NEXT_PUBLIC_MARKET_TRADING_URL=http://localhost:3005 + - NEXT_PUBLIC_WS_URL=ws://localhost:8080 + volumes: + - ./client/trading-dashboard:/app + - /app/node_modules + depends_on: + - server + command: sh -c "npm install && npm run dev" diff --git a/docker-compose.yml b/docker-compose.yml index 51e2ce9..57a6fee 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,4 +1,3 @@ -version: "3.8" services: server: build: @@ -6,10 +5,13 @@ services: dockerfile: Dockerfile ports: - "3005:3005" + - "8080:8080" environment: - PORT=3005 - NODE_ENV=production - ALLOWED_ORIGINS=http://localhost:3000 + - PRICE_UPDATE_INTERVAL=2000 + - WS_PORT=8080 restart: unless-stopped client: @@ -22,6 +24,7 @@ services: - NODE_ENV=production - NEXT_TELEMETRY_DISABLED=1 - NEXT_PUBLIC_MARKET_TRADING_URL=http://localhost:3005 + - NEXT_PUBLIC_WS_URL=ws://localhost:8080 depends_on: - server restart: unless-stopped diff --git a/docs/architecture.md b/docs/architecture.md index f2fde6d..fc745de 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -10,7 +10,7 @@ graph LR end subgraph Server - Backend["Node.js Backend
localhost:8080"] + Backend["Node.js Backend
localhost:3005"] DataSource["(Simulated Data Source)"] end @@ -30,7 +30,7 @@ Clean Code: Code is organized into logical modules with clear responsibilities. Scalability: The decoupled nature of the services allows them to be scaled independently. The use of WebSockets provides an efficient, low-latency communication channel for real-time data. ## Server Architecture - +``` ├── server/market-trading-service/ ├── src/ │ ├── core/ # Business logic & domain @@ -60,23 +60,19 @@ Scalability: The decoupled nature of the services allows them to be scaled indep ├── tsconfig.json └── .env ├── client/trading-dashboard/ - +``` ## Client Architecture Nextjs with Typescript, Tailwind CSS ### Data Fetching -+` -+- Using React Query (TanStack Query) as we need to implement polling -+- No state management library, using Context if needed -+` +- Using React Query (TanStack Query) as we need to implement polling +- No state management library, using Context if needed ### Chart -+` -+- Using Recharts which is leveraging D3.js components for easy integration into React -+` +- Using Recharts which is leveraging D3.js components for easy integration into React ### Unit Test diff --git a/server/market-trading-service/package-lock.json b/server/market-trading-service/package-lock.json index 2841d2a..7e454e3 100644 --- a/server/market-trading-service/package-lock.json +++ b/server/market-trading-service/package-lock.json @@ -13,7 +13,8 @@ "dotenv": "^17.2.3", "express": "^5.1.0", "helmet": "^8.1.0", - "module-alias": "^2.2.3" + "module-alias": "^2.2.3", + "ws": "^8.18.3" }, "devDependencies": { "@types/cors": "^2.8.19", @@ -22,6 +23,7 @@ "@types/module-alias": "^2.0.4", "@types/node": "^24.10.1", "@types/supertest": "^6.0.3", + "@types/ws": "^8.18.1", "jest": "^30.2.0", "nodemon": "^3.1.11", "supertest": "^7.1.4", @@ -61,7 +63,6 @@ "integrity": "sha512-e7jT4DxYvIDLk1ZHmU/m/mB19rex9sv0c2ftBtjSBv+kVM/902eh0fINUzD7UwLLNR+jU585GxUJ8/EBfAM5fw==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@babel/code-frame": "^7.27.1", "@babel/generator": "^7.28.5", @@ -1403,7 +1404,6 @@ "integrity": "sha512-GNWcUTRBgIRJD5zj+Tq0fKOJ5XZajIiBroOF0yvj2bSU1WvNdYS/dn9UxwsujGW4JX06dnHyjV2y9rRaybH0iQ==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "undici-types": "~7.16.0" } @@ -1486,6 +1486,16 @@ "@types/superagent": "^8.1.0" } }, + "node_modules/@types/ws": { + "version": "8.18.1", + "resolved": "https://registry.npmjs.org/@types/ws/-/ws-8.18.1.tgz", + "integrity": "sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@types/yargs": { "version": "17.0.34", "resolved": "https://registry.npmjs.org/@types/yargs/-/yargs-17.0.34.tgz", @@ -2101,7 +2111,6 @@ } ], "license": "MIT", - "peer": true, "dependencies": { "baseline-browser-mapping": "^2.8.25", "caniuse-lite": "^1.0.30001754", @@ -3706,7 +3715,6 @@ "integrity": "sha512-F26gjC0yWN8uAA5m5Ss8ZQf5nDHWGlN/xWZIh8S5SRbsEKBovwZhxGd6LJlbZYxBgCYOtreSUyb8hpXyGC5O4A==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@jest/core": "30.2.0", "@jest/types": "30.2.0", @@ -5786,7 +5794,6 @@ "integrity": "sha512-f0FFpIdcHgn8zcPSbf1dRevwt047YMnaiJM3u2w2RewrB+fob/zePZcrOyQoLMMO7aBIddLcQIEK5dYjkLnGrQ==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@cspotcode/source-map-support": "^0.8.0", "@tsconfig/node10": "^1.0.7", @@ -5876,7 +5883,6 @@ "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", "dev": true, "license": "Apache-2.0", - "peer": true, "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" @@ -6178,6 +6184,27 @@ "node": "^14.17.0 || ^16.13.0 || >=18.0.0" } }, + "node_modules/ws": { + "version": "8.18.3", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.18.3.tgz", + "integrity": "sha512-PEIGCY5tSlUt50cqyMXfCzX+oOPqN0vuGqWzbcJ2xvnkzkq46oOpz7dQaTDBdfICb4N14+GARUDw2XV2N4tvzg==", + "license": "MIT", + "engines": { + "node": ">=10.0.0" + }, + "peerDependencies": { + "bufferutil": "^4.0.1", + "utf-8-validate": ">=5.0.2" + }, + "peerDependenciesMeta": { + "bufferutil": { + "optional": true + }, + "utf-8-validate": { + "optional": true + } + } + }, "node_modules/y18n": { "version": "5.0.8", "resolved": "https://registry.npmjs.org/y18n/-/y18n-5.0.8.tgz", diff --git a/server/market-trading-service/package.json b/server/market-trading-service/package.json index 2bf83d3..212eec1 100644 --- a/server/market-trading-service/package.json +++ b/server/market-trading-service/package.json @@ -19,7 +19,8 @@ "dotenv": "^17.2.3", "express": "^5.1.0", "helmet": "^8.1.0", - "module-alias": "^2.2.3" + "module-alias": "^2.2.3", + "ws": "^8.18.3" }, "devDependencies": { "@types/cors": "^2.8.19", @@ -28,6 +29,7 @@ "@types/module-alias": "^2.0.4", "@types/node": "^24.10.1", "@types/supertest": "^6.0.3", + "@types/ws": "^8.18.1", "jest": "^30.2.0", "nodemon": "^3.1.11", "supertest": "^7.1.4", diff --git a/server/market-trading-service/src/container/Container.ts b/server/market-trading-service/src/container/Container.ts index 58634fe..ccaab64 100644 --- a/server/market-trading-service/src/container/Container.ts +++ b/server/market-trading-service/src/container/Container.ts @@ -1,13 +1,14 @@ -import { TickerRepository } from "../infrastructure/repositories/TickerRepository"; -import { PriceSimulator } from "../core/services/PriceSimulator"; -import { MarketDataService } from "../core/services/MarketDataService"; -import { TickerController } from "../api/controllers/TickerController"; -import { ErrorHandler } from "../api/middleware/ErrorHandler"; +import { TickerRepository } from "@/infrastructure/repositories/TickerRepository"; +import { PriceSimulator } from "@/core/services/PriceSimulator"; +import { MarketDataService } from "@/core/services/MarketDataService"; +import { TickerController } from "@/api/controllers/TickerController"; +import { ErrorHandler } from "@/api/middleware/ErrorHandler"; import { ITickerRepository, IPriceSimulator, IMarketDataService, -} from "../core/types"; +} from "@/core/types"; +import { WebSocketManager } from "@/infrastructure/websocket/WebSocketManager"; export class Container { private services: Map = new Map(); @@ -61,4 +62,8 @@ export class Container { getErrorHandler(): ErrorHandler { return this.get("ErrorHandler"); } + + createWebSocketManager(port: number): WebSocketManager { + return new WebSocketManager(port, this.getMarketDataService()); + } } diff --git a/server/market-trading-service/src/core/services/MarketDataService.ts b/server/market-trading-service/src/core/services/MarketDataService.ts index 1ffe7ab..a6ca7cd 100644 --- a/server/market-trading-service/src/core/services/MarketDataService.ts +++ b/server/market-trading-service/src/core/services/MarketDataService.ts @@ -7,6 +7,8 @@ import { } from "@/core/types"; export class MarketDataService implements IMarketDataService { + private subscribers: Map void>> = + new Map(); constructor( private repository: ITickerRepository, private priceSimulator: IPriceSimulator @@ -25,7 +27,10 @@ export class MarketDataService implements IMarketDataService { tickers.forEach((ticker) => { this.priceSimulator.start(ticker, (updatedTicker) => { + // Persist updated ticker state this.repository.update(updatedTicker); + // Notify any active subscribers (e.g., WebSocket clients) so they receive real-time updates + this.notifySubscribers(updatedTicker); }); }); } @@ -40,6 +45,39 @@ export class MarketDataService implements IMarketDataService { return this.priceSimulator.generateHistoricalData(ticker, days); } + subscribeToTicker( + symbol: string, + callback: (ticker: ITicker) => void + ): string { + symbol = symbol.toUpperCase(); + + if (!this.subscribers.has(symbol)) { + this.subscribers.set(symbol, new Map()); + } + + const callbackId = Math.random().toString(36).substring(7); + this.subscribers.get(symbol)!.set(callbackId, callback); + + return callbackId; + } + + unsubscribeFromTicker(symbol: string, callbackId: string): void { + const callbacks = this.subscribers.get(symbol.toUpperCase()); + if (callbacks) { + callbacks.delete(callbackId); + if (callbacks.size === 0) { + this.subscribers.delete(symbol); + } + } + } + + notifySubscribers(ticker: ITicker): void { + const callbacks = this.subscribers.get(ticker.symbol); + if (callbacks) { + callbacks.forEach((callback) => callback(ticker)); + } + } + stopSimulation(): void { return this.priceSimulator.stopAll(); } diff --git a/server/market-trading-service/src/core/types/index.ts b/server/market-trading-service/src/core/types/index.ts index faa8f9c..c0128f1 100644 --- a/server/market-trading-service/src/core/types/index.ts +++ b/server/market-trading-service/src/core/types/index.ts @@ -1,3 +1,5 @@ +// WebSocket Types +import type { WebSocket as WsWebSocket } from "ws"; // Domain Types export interface Ticker { symbol: string; @@ -43,12 +45,16 @@ export interface IMarketDataService { getHistoricalData(symbol: string, days: number): Promise; startSimulation(): Promise; stopSimulation(): void; + subscribeToTicker(symbol: string, callback: (ticker: Ticker) => void): string; + unsubscribeFromTicker(symbol: string, callbackId: string): void; } // Configuration export interface AppConfig { port: number; env: string; + priceUpdateInterval: number; + wsPort: number; } // Ticker related types @@ -78,3 +84,15 @@ export interface TickerJSON { } export type Interval = "hourly" | "daily" | "15min"; + +export interface WSMessage { + type: "subscribe" | "unsubscribe" | "ping" | "pong" | "error" | "data"; + payload?: any; +} + +export interface WSClient { + id: string; + ws: WsWebSocket; + subscriptions: Set; + isAlive: boolean; +} diff --git a/server/market-trading-service/src/index.ts b/server/market-trading-service/src/index.ts index 23a9d24..3ab0349 100644 --- a/server/market-trading-service/src/index.ts +++ b/server/market-trading-service/src/index.ts @@ -80,7 +80,9 @@ async function startServer() { throw error; } - console.log("Price simulation started"); + // Start WebSocket server + container.createWebSocketManager(config.wsPort); + console.log(`WebSocket Server running on port ${config.wsPort}`); const shutdown = (signal: string) => { console.log(`\n⚠️ ${signal} received, shutting down gracefully...`); diff --git a/server/market-trading-service/src/infrastructure/config/Config.ts b/server/market-trading-service/src/infrastructure/config/Config.ts index 445becb..4e68b2b 100644 --- a/server/market-trading-service/src/infrastructure/config/Config.ts +++ b/server/market-trading-service/src/infrastructure/config/Config.ts @@ -6,8 +6,12 @@ dotenv.config(); export class Config { static get(): AppConfig { return { + wsPort: parseInt(process.env.WS_PORT || "8080"), port: parseInt(process.env.PORT || "3005"), env: process.env.NODE_ENV || "development", + priceUpdateInterval: parseInt( + process.env.PRICE_UPDATE_INTERVAL || "2000" + ), }; } } diff --git a/server/market-trading-service/src/infrastructure/websocket/WebSocketManager.ts b/server/market-trading-service/src/infrastructure/websocket/WebSocketManager.ts new file mode 100644 index 0000000..8f82378 --- /dev/null +++ b/server/market-trading-service/src/infrastructure/websocket/WebSocketManager.ts @@ -0,0 +1,190 @@ +import WebSocket, { WebSocketServer } from "ws"; +import { WSClient, WSMessage, Ticker, IMarketDataService } from "@/core/types"; + +export class WebSocketManager { + private wss: WebSocketServer; + private clients: Map = new Map(); + private marketDataService: IMarketDataService; + // Track callback IDs so we can properly unsubscribe and avoid memory leaks per client + private subscriptionIds: Map> = new Map(); + + constructor(port: number, marketDataService: IMarketDataService) { + this.marketDataService = marketDataService; + this.wss = new WebSocketServer({ port }); + this.initialize(); + } + + private initialize(): void { + this.wss.on("connection", (ws: WebSocket) => { + const clientId = this.generateClientId(); + const client: WSClient = { + id: clientId, + ws: ws as any, + subscriptions: new Set(), + isAlive: true, + }; + + this.clients.set(clientId, client); + console.log(`Client connected: ${clientId}`); + + ws.send( + JSON.stringify({ + type: "connected", + payload: { clientId }, + }) + ); + + ws.on("message", (data: WebSocket.Data) => { + this.handleMessage(client, data.toString()); + }); + + ws.on("close", () => { + this.handleDisconnect(clientId); + }); + + ws.on("error", (error) => { + console.error(`WebSocket error for client ${clientId}:`, error); + }); + + ws.on("pong", () => { + client.isAlive = true; + }); + }); + + // Heartbeat + setInterval(() => { + this.clients.forEach((client) => { + if (!client.isAlive) { + client.ws.terminate(); + this.clients.delete(client.id); + return; + } + client.isAlive = false; + client.ws.ping(); + }); + }, 30000); + } + + private handleMessage(client: WSClient, message: string): void { + try { + const msg: WSMessage = JSON.parse(message); + + switch (msg.type) { + case "subscribe": + this.handleSubscribe(client, msg.payload?.symbols || []); + break; + case "unsubscribe": + this.handleUnsubscribe(client, msg.payload?.symbols || []); + break; + case "ping": + client.ws.send(JSON.stringify({ type: "pong" })); + break; + } + } catch (error) { + client.ws.send( + JSON.stringify({ + type: "error", + payload: { message: "Invalid message format" }, + }) + ); + } + } + + private async handleSubscribe( + client: WSClient, + symbols: string[] + ): Promise { + for (const symbol of symbols) { + const ticker = await this.marketDataService.getTicker(symbol); + if (ticker) { + client.subscriptions.add(symbol.toUpperCase()); + + // Send initial data + client.ws.send( + JSON.stringify({ + type: "data", + payload: { ticker }, + }) + ); + + // Subscribe to updates + const callbackId = this.marketDataService.subscribeToTicker( + symbol, + (updatedTicker) => { + if (client.subscriptions.has(updatedTicker.symbol)) { + client.ws.send( + JSON.stringify({ + type: "data", + payload: { ticker: updatedTicker }, + }) + ); + } + } + ); + if (!this.subscriptionIds.has(client.id)) { + this.subscriptionIds.set(client.id, new Map()); + } + this.subscriptionIds + .get(client.id)! + .set(symbol.toUpperCase(), callbackId); + } + } + + client.ws.send( + JSON.stringify({ + type: "subscribed", + payload: { symbols: Array.from(client.subscriptions) }, + }) + ); + } + + private handleUnsubscribe(client: WSClient, symbols: string[]): void { + symbols.forEach((symbol) => { + client.subscriptions.delete(symbol.toUpperCase()); + const perClient = this.subscriptionIds.get(client.id); + const upper = symbol.toUpperCase(); + if (perClient && perClient.has(upper)) { + const callbackId = perClient.get(upper)!; + this.marketDataService.unsubscribeFromTicker(upper, callbackId); + perClient.delete(upper); + } + }); + + client.ws.send( + JSON.stringify({ + type: "unsubscribed", + payload: { symbols }, + }) + ); + } + + private handleDisconnect(clientId: string): void { + console.log(`Client disconnected: ${clientId}`); + // Clean up any active subscriptions for this client + const perClient = this.subscriptionIds.get(clientId); + if (perClient) { + perClient.forEach((callbackId, symbol) => { + this.marketDataService.unsubscribeFromTicker(symbol, callbackId); + }); + this.subscriptionIds.delete(clientId); + } + this.clients.delete(clientId); + } + + private generateClientId(): string { + return Math.random().toString(36).substring(2, 15); + } + + broadcast(ticker: Ticker): void { + const message = JSON.stringify({ + type: "data", + payload: { ticker }, + }); + + this.clients.forEach((client) => { + if (client.subscriptions.has(ticker.symbol)) { + client.ws.send(message); + } + }); + } +}