Feature/predictive rpc monitoring 576

src/components/providers/binanceProvider.ts

@@ -0,0 +1,24 @@
+import axios from 'axios';
+
+import { USE_MOCKS }
+  from '../config/env';
+
+import { getMockPrice }
+  from '../mocks/mockBinance';
+
+export async function getPrice(
+  symbol: string,
+) {
+  if (USE_MOCKS) {
+    return getMockPrice(
+      symbol,
+    );
+  }
+
+  const response =
+    await axios.get(
+      `https://api.binance.com/api/v3/ticker/price?symbol=${symbol}`,
+    );
+
+  return response.data;
+}

src/components/providers/vtpassProvider.ts

@@ -0,0 +1,35 @@
+import axios from 'axios';
+
+import { USE_MOCKS }
+  from '../config/env';
+
+import {
+  mockPurchaseAirtime,
+} from '../mocks/mockVtpass';
+
+export async function purchaseAirtime(
+  phone: string,
+  amount: number,
+) {
+  if (USE_MOCKS) {
+    return mockPurchaseAirtime(
+      phone,
+      amount,
+    );
+  }
+
+  return axios.post(
+    process.env.VTPASS_URL!,
+    {
+      phone,
+      amount,
+    },
+    {
+      headers: {
+        Authorization:
+          process.env
+            .VTPASS_API_KEY!,
+      },
+    },
+  );
+}

src/config/env.ts

@@ -0,0 +1,3 @@
+export const USE_MOCKS =
+  process.env.USE_MOCKS ===
+  'true';

src/config/logger.ts

@@ -1,5 +1,28 @@
 import winston from 'winston';
 
+
+const {
+  combine,
+  timestamp,
+  errors,
+  json,
+  colorize,
+  printf,
+} = winston.format;
+
+const consoleFormat = printf(
+  ({
+    level,
+    message,
+    timestamp,
+    stack,
+  }) => {
+    return `${timestamp} [${level}]: ${
+      stack || message
+    }`;
+  },
+);
+
 import { HttpLogTransport }
   from '../transport/httpLogTransport';
 
@@ -23,12 +46,43 @@ export const logger =
   winston.createLogger({
     level: 'info',
 
-    format:
-      winston.format.combine(
-        winston.format.timestamp(),
+    format: combine(
+      timestamp(),
+      errors({ stack: true }),
+      json(),
+    ),
 
-        winston.format.json(),
-      ),
+    defaultMeta: {
+      service: 'backend-api',
+    },
+
+    transports: [
+      // Error Logs
+      new winston.transports.File({
+        filename:
+          'logs/error.log',
+        level: 'error',
+      }),
 
-    transports,
-  });
+      // Combined Logs
+      new winston.transports.File({
+        filename:
+          'logs/combined.log',
+      }),
+    ],
+  });
+
+if (
+  process.env.NODE_ENV !==
+  'production'
+) {
+  logger.add(
+    new winston.transports.Console({
+      format: combine(
+        colorize(),
+        timestamp(),
+        consoleFormat,
+      ),
+    }),
+  );
+}

src/middleware/requestLogger.ts

@@ -0,0 +1,18 @@
+import morgan from 'morgan';
+
+import { logger }
+  from '../config/logger';
+
+const stream = {
+  write: (message: string) => {
+    logger.info(
+      message.trim(),
+    );
+  },
+};
+
+export const requestLogger =
+  morgan(
+    ':method :url :status :response-time ms',
+    { stream },
+  );

src/mocks/mockBinance.ts

@@ -0,0 +1,20 @@
+import {
+  randomFloat,
+} from '../utils/random';
+
+export async function getMockPrice(
+  symbol: string,
+) {
+  return {
+    symbol,
+
+    price: randomFloat(
+      25000,
+      70000,
+      2,
+    ),
+
+    timestamp:
+      Date.now(),
+  };
+}

src/mocks/mockVtpass.ts

@@ -0,0 +1,28 @@
+import {
+  randomChoice,
+} from '../utils/random';
+
+export async function mockPurchaseAirtime(
+  phone: string,
+  amount: number,
+) {
+  return {
+    success: true,
+
+    phone,
+
+    amount,
+
+    transactionId:
+      crypto.randomUUID(),
+
+    provider: randomChoice([
+      'MTN',
+      'Airtel',
+      'Glo',
+      '9mobile',
+    ]),
+
+    status: 'successful',
+  };
+}

src/network/__tests__/test_polling.py

@@ -0,0 +1,63 @@
+import pytest
+import asyncio
+import time
+from src.network.polling import RegionalPollingEngine
+
+class MockResponse:
+    def __init__(self, status, json_data):
+        self.status = status
+        self._json_data = json_data
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        pass
+
+    async def json(self):
+        return self._json_data
+
+@pytest.mark.asyncio
+async def test_polling_engine_enforces_2500ms_timeout_cutoff(monkeypatch):
+    """
+    Asserts that if an endpoint hangs indefinitely, the engine drops it at 2500ms
+    and returns a clean failure object without stalling other tasks.
+    """
+    # 1. Setup a normal fast region alongside an un-responsive lagging region
+    endpoints = {
+        "FAST-NODE": "https://fast.node/api",
+        "LAGGING-NODE": "https://laggy.node/api"
+    }
+
+    engine = RegionalPollingEngine(endpoints)
+
+    # 2. Mock aiohttp.ClientSession.get to simulate varying endpoint performance
+    async def mock_get(self, url, **kwargs):
+        if "fast" in url:
+            return MockResponse(200, {"metrics": "nominal"})
+        elif "laggy" in url:
+            # Simulate a 10-second backend hang, well beyond our 2.5s cutoff
+            await asyncio.sleep(10.0)
+            return MockResponse(200, {"metrics": "delayed"})
+        return MockResponse(404, {})
+
+    monkeypatch.setattr("aiohttp.ClientSession.get", mock_get)
+
+    start_time = time.monotonic()
+    results = await engine.poll_all_regions_concurrently()
+    duration = time.monotonic() - start_time
+
+    # 3. Assertions and Structural Checks
+    # The overall run should finish shortly after the 2.5s limit, not waiting 10s
+    assert duration < 3.0, f"The execution pool stalled! Total duration took too long: {duration}s"
+
+    fast_node_res = next(r for r in results if r["region"] == "FAST-NODE")
+    laggy_node_res = next(r for r in results if r["region"] == "LAGGING-NODE")
+
+    # Fast node should return successfully
+    assert fast_node_res["status"] == "SUCCESS"
+    assert fast_node_res["payload"]["metrics"] == "nominal"
+
+    # Lagging node should be caught and dropped cleanly by the timeout guard
+    assert laggy_node_res["status"] == "TIMEOUT"
+    assert "2500ms threshold" in laggy_node_res["error"]

src/network/__tests__/test_rpc_supervisor.py

@@ -0,0 +1,38 @@
+import pytest
+import asyncio
+from src.network.nonce_tracker import PredictiveRPCSupervisor, HorizonNodeProfile
+
+class MockResponseContext:
+    def __init__(self, status):
+        self.status = status
+    async def __aenter__(self): return self
+    async def __aexit__(self, exc_type, exc_val, exc_tb): pass
+
+@pytest.mark.asyncio
+async def test_rpc_supervisor_preemptive_failover_logic(monkeypatch):
+    """
+    Asserts that the supervisor instantly shifts traffic to secondary backup nodes
+    if the primary node's health drops below acceptable bounds.
+    """
+    primary = [{"name": "primary-node-01", "url": "https://primary.stellar.org"}]
+    fallback = [{"name": "backup-node-01", "url": "https://backup.stellar.org"}]
+
+    supervisor = PredictiveRPCSupervisor(primary, fallback)
+    assert supervisor.get_active_endpoint_url() == "https://primary.stellar.org"
+
+    # Mock client session requests to simulate primary node failure and backup success
+    async def mock_get(self, url, **kwargs):
+        if "primary" in url:
+            # Simulate network timeout block
+            await asyncio.sleep(2.0)
+            return MockResponseContext(504)
+        return MockResponseContext(200)
+
+    monkeypatch.setattr("aiohttp.ClientSession.get", mock_get)
+
+    # Trigger predictive health evaluation cycles
+    await supervisor.run_predictive_ping_cycle()
+
+    # The supervisor should notice the primary node timeout and shift to the backup node
+    assert supervisor.active_node.is_healthy is False
+    assert supervisor.get_active_endpoint_url() == "https://backup.stellar.org"