Phase 3: Redis Streams Scaling

import Redis from 'ioredis';
 
/**
 * THE PRODUCER: Edge Ingestion Layer
 * Duty: Accept validated data and fire it into the Redis Stream as fast as possible.
 * It does not care about the database, disk I/O, or worker status.
 */
export class StreamProducer {
  private static redis: Redis;
  private static readonly STREAM_KEY = 'ingestion:gps_stream';
 
  static initialize() {
    // In production, configure connection pooling and cluster nodes
    this.redis = new Redis(process.env.REDIS_URL || 'redis://127.0.0.1:6379');
    console.log('[Redis] Producer connected to stream pipeline.');
  }
 
  /**
   * Pushes a GPS payload to the stream.
   * O(1) time complexity.
   */
  static async publishGpsPoint(payload: any): Promise<void> {
    if (!this.redis) this.initialize();
 
    try {
      // XADD: Append to stream
      // MAXLEN ~ 1000000: Crucial for Memory Control. It caps the stream size to approx 1M records
      // '*': Auto-generate Redis Stream ID (Timestamp-Sequence)
      await this.redis.xadd(
        this.STREAM_KEY,
        'MAXLEN', '~', 1000000,
        '*',
        'data', JSON.stringify(payload) // The JSON Tax (Bottleneck for Phase 4)
      );
    } catch (error) {
      console.error('[StreamProducer] Failed to inject payload:', error);
      // In a strict Absolute Control environment, fallback strategies (e.g., local file buffer)
      // would be implemented here if Redis is temporarily unreachable.
    }
  }
 
  static async shutdown() {
    await this.redis?.quit();
  }
}

import Redis from 'ioredis';
import { Database } from '../db.js'; // Assuming Phase 1 DB layer
 
/**
 * THE CONSUMER: Parallel Processing Logic
 * Duty: Pull events from the stream, process them, and acknowledge (ACK).
 * If it crashes, unacknowledged events remain in the PEL (Pending Entries List).
 */
export class StreamWorker {
  private redis: Redis;
  private readonly STREAM_KEY = 'ingestion:gps_stream';
  private readonly GROUP_NAME = 'fleet_db_workers';
  private consumerName: string;
  private isRunning: boolean = false;
 
  constructor(consumerId: string) {
    this.redis = new Redis(process.env.REDIS_URL || 'redis://127.0.0.1:6379');
    this.consumerName = `worker-${consumerId}`;
  }
 
  async start() {
    this.isRunning = true;
    await this.createConsumerGroup();
    console.log(`[Worker: ${this.consumerName}] Ready and polling for events...`);
    this.poll();
  }
 
  /**
   * Initializes the Consumer Group.
   * If the group already exists, Redis throws a 'BUSYGROUP' error which we safely ignore.
   */
  private async createConsumerGroup() {
    try {
      // Create group, start reading from '$' (only new messages arriving after group creation)
      // For absolute data retention, we could use '0' to read from the very beginning.
      await this.redis.xgroup('CREATE', this.STREAM_KEY, this.GROUP_NAME, '$', 'MKSTREAM');
    } catch (error: any) {
      if (!error.message.includes('BUSYGROUP')) {
        console.error('[Worker] Fatal error creating consumer group:', error);
        process.exit(1);
      }
    }
  }
 
  /**
   * The Infinite Polling Loop.
   * Uses XREADGROUP with BLOCK to prevent CPU thrashing while waiting for data.
   */
  private async poll() {
    while (this.isRunning) {
      try {
        // BLOCK 2000: Wait up to 2 seconds for new data.
        // COUNT 100: Fetch up to 100 messages at once for Batch Insertion efficiency.
        // '>': Fetch messages that have NEVER been delivered to ANY consumer in this group.
        const response = await this.redis.xreadgroup(
          'GROUP', this.GROUP_NAME, this.consumerName,
          'COUNT', 100,
          'BLOCK', 2000,
          'STREAMS', this.STREAM_KEY, '>'
        );
 
        if (response) {
          const stream = response[0];
          const messages = stream[1]; // Array of [messageId, [field1, value1, ...]]
 
          if (messages.length > 0) {
            await this.processBatch(messages);
          }
        }
      } catch (error) {
        console.error(`[Worker: ${this.consumerName}] Polling error:`, error);
        // Sleep briefly to prevent tight-looping on connection failures
        await new Promise(res => setTimeout(res, 1000));
      }
    }
  }
 
  /**
   * Core processing and acknowledgment.
   */
  private async processBatch(messages: any[]) {
    const messageIds: string[] = [];
    
    // In a real high-performance setup, we parse and build a bulk INSERT query here.
    for (const message of messages) {
      const msgId = message[0];
      const fields = message[1]; // ['data', '{"lat":33.1...}']
      
      const rawData = fields[1];
      const payload = JSON.parse(rawData);
 
      // Simulate Database Write (Using the DB Layer from Phase 1)
      await Database.query(
        `INSERT INTO fleet_locations (device_id, lat, lng, speed, timestamp) VALUES ($1, $2, $3, $4, $5)`,
        [payload.deviceId, payload.lat, payload.lng, payload.speed, payload.timestamp]
      );
 
      messageIds.push(msgId);
    }
 
    // Acknowledge the batch. This removes them from the Pending Entries List (PEL).
    if (messageIds.length > 0) {
      await this.redis.xack(this.STREAM_KEY, this.GROUP_NAME, ...messageIds);
    }
  }
 
  stop() {
    this.isRunning = false;
    this.redis.quit();
  }
}