Build a Streaming Data Pipeline

// src/pipeline/streaming.ts — Real-time event pipeline with windowed aggregation and DLQ
import { Redis } from "ioredis";
import { pool } from "../db";
import { randomBytes } from "node:crypto";

const redis = new Redis(process.env.REDIS_URL!);

interface PipelineEvent {
  id: string;
  type: string;
  source: string;
  timestamp: number;
  data: Record<string, any>;
}

interface PipelineStage {
  name: string;
  type: "transform" | "filter" | "aggregate" | "enrich" | "sink";
  handler: (event: PipelineEvent, context: PipelineContext) => Promise<PipelineEvent | PipelineEvent[] | null>;
  config: Record<string, any>;
}

interface PipelineContext {
  pipelineId: string;
  stageIndex: number;
  retryCount: number;
  metadata: Record<string, any>;
}

interface WindowAggregation {
  windowId: string;
  windowType: "tumbling" | "sliding" | "session";
  windowSize: number;         // milliseconds
  key: string;
  values: Map<string, number>;
  startTime: number;
  endTime: number;
  eventCount: number;
}

const pipelines = new Map<string, PipelineStage[]>();
const windows = new Map<string, WindowAggregation>();

// Register a pipeline
export function registerPipeline(id: string, stages: PipelineStage[]): void {
  pipelines.set(id, stages);
}

// Ingest events into the pipeline
export async function ingest(pipelineId: string, events: PipelineEvent[]): Promise<{
  accepted: number; rejected: number; processingTimeMs: number;
}> {
  const start = Date.now();
  const stages = pipelines.get(pipelineId);
  if (!stages) throw new Error(`Pipeline '${pipelineId}' not found`);

  let accepted = 0;
  let rejected = 0;

  // Check backpressure
  const queueSize = await redis.llen(`pipeline:queue:${pipelineId}`);
  if (queueSize > 100000) {
    // Apply backpressure — reject with retry-after
    throw new BackpressureError(`Pipeline queue full: ${queueSize} events pending`);
  }

  for (const event of events) {
    try {
      await processEvent(pipelineId, stages, event);
      accepted++;
    } catch (error: any) {
      // Send to dead letter queue
      await sendToDLQ(pipelineId, event, error.message);
      rejected++;
    }
  }

  // Update metrics
  await redis.hincrby(`pipeline:metrics:${pipelineId}`, "accepted", accepted);
  await redis.hincrby(`pipeline:metrics:${pipelineId}`, "rejected", rejected);

  return { accepted, rejected, processingTimeMs: Date.now() - start };
}

async function processEvent(
  pipelineId: string,
  stages: PipelineStage[],
  event: PipelineEvent
): Promise<void> {
  let currentEvents: PipelineEvent[] = [event];

  for (let i = 0; i < stages.length; i++) {
    const stage = stages[i];
    const nextEvents: PipelineEvent[] = [];

    for (const evt of currentEvents) {
      const ctx: PipelineContext = { pipelineId, stageIndex: i, retryCount: 0, metadata: {} };

      try {
        const result = await stage.handler(evt, ctx);
        if (result === null) continue;  // filtered out
        if (Array.isArray(result)) nextEvents.push(...result);
        else nextEvents.push(result);
      } catch (error: any) {
        // Retry with backoff
        let retried = false;
        for (let retry = 1; retry <= 3; retry++) {
          await sleep(retry * 100);
          try {
            ctx.retryCount = retry;
            const result = await stage.handler(evt, ctx);
            if (result !== null) {
              if (Array.isArray(result)) nextEvents.push(...result);
              else nextEvents.push(result);
            }
            retried = true;
            break;
          } catch {}
        }
        if (!retried) throw error;
      }
    }

    currentEvents = nextEvents;
    if (currentEvents.length === 0) break;
  }
}

// Windowed aggregation
export async function aggregateWindow(
  windowType: WindowAggregation["windowType"],
  windowSizeMs: number,
  event: PipelineEvent,
  groupKey: string,
  valueField: string,
  aggregation: "count" | "sum" | "avg" | "min" | "max"
): Promise<{ windowComplete: boolean; result?: Record<string, number> }> {
  const windowStart = Math.floor(event.timestamp / windowSizeMs) * windowSizeMs;
  const windowId = `${groupKey}:${windowStart}`;

  const key = `window:${windowId}`;
  const group = event.data[groupKey] || "default";
  const value = event.data[valueField] || 1;

  // Update window in Redis
  await redis.hincrby(key, `${group}:count`, 1);
  await redis.hincrbyfloat(key, `${group}:sum`, value);
  await redis.expire(key, Math.ceil(windowSizeMs / 1000) * 2);
  await redis.hincrby(key, "totalEvents", 1);

  // Check if window is complete (tumbling window)
  const windowEnd = windowStart + windowSizeMs;
  if (Date.now() >= windowEnd) {
    // Flush window
    const data = await redis.hgetall(key);
    const result: Record<string, number> = {};

    for (const [k, v] of Object.entries(data)) {
      if (k === "totalEvents") continue;
      const [groupName, metric] = k.split(":");
      if (aggregation === "count" && metric === "count") result[groupName] = parseInt(v);
      if (aggregation === "sum" && metric === "sum") result[groupName] = parseFloat(v);
      if (aggregation === "avg" && metric === "sum") {
        const count = parseInt(data[`${groupName}:count`] || "1");
        result[groupName] = parseFloat(v) / count;
      }
    }

    await redis.del(key);
    return { windowComplete: true, result };
  }

  return { windowComplete: false };
}

// Dead letter queue
async function sendToDLQ(pipelineId: string, event: PipelineEvent, error: string): Promise<void> {
  await redis.rpush(`pipeline:dlq:${pipelineId}`, JSON.stringify({
    event, error, failedAt: new Date().toISOString(),
  }));
  await redis.hincrby(`pipeline:metrics:${pipelineId}`, "dlqSize", 1);
}

// Replay DLQ events
export async function replayDLQ(pipelineId: string, limit: number = 100): Promise<{ replayed: number; failed: number }> {
  let replayed = 0, failed = 0;
  for (let i = 0; i < limit; i++) {
    const item = await redis.lpop(`pipeline:dlq:${pipelineId}`);
    if (!item) break;
    const { event } = JSON.parse(item);
    try {
      await ingest(pipelineId, [event]);
      replayed++;
    } catch {
      failed++;
      await redis.rpush(`pipeline:dlq:${pipelineId}`, item);  // put back
    }
  }
  return { replayed, failed };
}

// Pipeline monitoring
export async function getMetrics(pipelineId: string): Promise<{
  eventsAccepted: number; eventsRejected: number;
  dlqSize: number; queueDepth: number; throughputPerSecond: number;
}> {
  const stats = await redis.hgetall(`pipeline:metrics:${pipelineId}`);
  const queueDepth = await redis.llen(`pipeline:queue:${pipelineId}`);

  return {
    eventsAccepted: parseInt(stats.accepted || "0"),
    eventsRejected: parseInt(stats.rejected || "0"),
    dlqSize: parseInt(stats.dlqSize || "0"),
    queueDepth,
    throughputPerSecond: 0,
  };
}

function sleep(ms: number): Promise<void> {
  return new Promise((r) => setTimeout(r, ms));
}

export class BackpressureError extends Error {
  constructor(message: string) { super(message); this.name = "BackpressureError"; }
}