System Architecture

import { config, redisStreams, constant } from "@repo/config";
import { tradeFunction } from "./functions/tradeFunction.js";

const RedisStreams = redisStreams(config.REDIS_URL);
await RedisStreams.connect();

// Parallel processing configuration
const CONCURRENCY_LIMIT = 10; // Process up to 10 messages concurrently
const CONSUMER_GROUP = "engine-group";
const CONSUMER_NAME = `engine-${process.pid}-${Date.now()}`;

let activeTasks = 0;
const taskQueue: Array<() => Promise<void>> = [];

// Process a single message
async function processMessage(result: any) {
  try {
    await tradeFunction(result);
  } catch (error) {
    console.error("Error processing message:", error);
  }
}

// Worker function to process queued tasks
async function worker() {
  while (true) {
    if (taskQueue.length > 0 && activeTasks < CONCURRENCY_LIMIT) {
      const task = taskQueue.shift();
      if (task) {
        activeTasks++;
        task().finally(() => {
          activeTasks--;
        });
      }
    } else {
      await new Promise(resolve => setTimeout(resolve, 10));
    }
  }
}

// Start worker threads
const WORKER_COUNT = Math.min(CONCURRENCY_LIMIT, 5);
for (let i = 0; i < WORKER_COUNT; i++) {
  worker();
}

// Continuously consume messages from Redis Stream
while (true) {
  const result = await RedisStreams.readNextFromRedisStream(
    constant.redisStream,
    0, // Block indefinitely
    {
      consumerGroup: CONSUMER_GROUP,
      consumerName: CONSUMER_NAME
    }
  );
  
  if (result) {
    taskQueue.push(() => processMessage(result));
    console.log(`Queued message. Queue length: ${taskQueue.length}, Active: ${activeTasks}`);
  }
}

import { WebSocketServer } from "ws";
import { config, pubsubClient, constant } from "@repo/config";

// Initialize WebSocket server
const wss = new WebSocketServer({ port: config.WEBSOCKET_PORT });

// Connect to Redis Pub/Sub
const PubsubClient = pubsubClient(config.REDIS_URL);
await PubsubClient.connect();

wss.on("connection", async (socket) => {
  console.log("Client connected");

  // Subscribe to Redis Pub/Sub and forward to WebSocket
  await PubsubClient.subscriber(constant.pubsubKey, (data: any) => {
    socket.send(JSON.stringify(data));
  });

  socket.on("close", () => {
    console.log("Client disconnected");
  });
});

import WebSocket from "ws";
import { pubsubClient, config, redisClient, redisStreams, constant } from "@repo/config";

const ws = new WebSocket(config.BINANCE_WS_URL);
const PubsubClient = pubsubClient(config.REDIS_URL);
const RedisClient = redisClient(config.REDIS_URL);
const RedisStreams = redisStreams(config.REDIS_URL);

await PubsubClient.connect();
await RedisClient.connect();
await RedisStreams.connect();

const crypto_trades = ["ETH_USDC_PERP", "SOL_USDC_PERP", "BTC_USDC_PERP"];
let bidAskMap: Record<string, { bid: number; ask: number }> = {};

ws.on("open", function open() {
  // Subscribe to trading pairs
  crypto_trades.forEach((pair) => {
    ws.send(`{"method":"SUBSCRIBE","params":["trade.${pair}"],"id":4}`);
  });

  ws.on("message", async (data) => {
    const msg = JSON.parse(data.toString());
    
    // Calculate bid/ask with 0.5% spread
    const decimal = 4;
    const asset = msg.data.s.toString();
    const price = Math.floor(Number(msg.data.p) * 10 ** decimal);
    const bidValue = Math.floor((price - price * 0.005) * 10 ** decimal);
    const askValue = Math.floor((price + price * 0.005) * 10 ** decimal);
    
    // 1. Push to Redis queue for batch processing
    await RedisClient.pushData(constant.redisQueue, JSON.stringify(msg));
    
    // 2. Publish to WebSocket clients via Pub/Sub
    const BidAsk = { asset, bid: bidValue, ask: askValue };
    await PubsubClient.publish(constant.pubsubKey, JSON.stringify(BidAsk));
    
    // 3. Update in-memory price tracking
    price_updates.push({ asset, price, bidValue, askValue, decimal });
  });
});

// Send aggregated updates to Engine every 3 seconds
setInterval(async () => {
  await RedisStreams.addToRedisStream(
    constant.redisStream,
    { function: "pricePoller", message: JSON.stringify(price_updates) }
  );
}, 3000);

import { config, redisStreams, constant } from "@repo/config";
import { dbStorageFunction } from "./functions/dbStorageFunction.js";

// Connect to Redis Streams
const RedisStreams = redisStreams(config.REDIS_URL);
await RedisStreams.connect();

// Continuously process database operations
await RedisStreams.readRedisStream(constant.dbStorageStream, dbStorageFunction);

import { constant, config, redisClient } from "@repo/config";
import { timeScaleDB } from "@repo/timescaledb";

const db = timeScaleDB();
await db.connect();
await db.setupTimescale();

const RedisClient = redisClient(config.REDIS_URL);
await RedisClient.connect();

let BATCH_SIZE = 0;
const BATCH_LIMIT = 100;

while (true) {
  try {
    const msg = await RedisClient.popData(constant.redisQueue);
    
    if (msg) {
      const trade = JSON.parse(msg);
      
      // Convert timestamp to valid Date
      let timestamp = typeof trade.data.T === 'string' 
        ? parseInt(trade.data.T, 10) 
        : trade.data.T;
      
      // Handle microseconds to milliseconds conversion
      if (timestamp > 4102444800000) {
        timestamp = Math.floor(timestamp / 1000);
      }
      
      const time = new Date(timestamp);
      
      // Validate timestamp
      if (isNaN(time.getTime()) || time.getFullYear() < 2020 || time.getFullYear() > 2100) {
        console.error(`⚠️ Invalid timestamp: ${trade.data.T}`);
        continue;
      }
      
      const symbol = trade.data.s;
      const price = trade.data.p;
      const volume = trade.data.q;
      const trade_id = trade.data.t;
      const side = trade.data.m ? "sell" : "buy";

      // Insert into TimeScaleDB hypertable
      await db.getClient().query(
        `INSERT INTO trades (time, symbol, price, volume, trade_id, side)
        VALUES ($1, $2, $3, $4, $5, $6)
        ON CONFLICT DO NOTHING;`,
        [time, symbol, price, volume, trade_id, side]
      );

      BATCH_SIZE++;
    } else {
      await new Promise((resolve) => setTimeout(resolve, 100));
    }

    if (BATCH_SIZE >= BATCH_LIMIT) {
      BATCH_SIZE = 0;
    }
  } catch (err) {
    console.error("Error processing trade:", err);
  }
}