Real-Time Anomaly Radar with Fastify, Kafka, ClickHouse, Rust→WASM, and WebGPU¶

// src/server.ts
import Fastify from "fastify";
import websocket from "@fastify/websocket";
import { Kafka } from "kafkajs";
import { ClickHouse } from "@clickhouse/client";
import * as wasm from "../anomaly-core/pkg/anomaly_core.js";

// Initialize Fastify
const app = Fastify({ 
  logger: { 
    level: 'info',
    transport: {
      target: 'pino-pretty'
    }
  } 
});

// Register WebSocket support
await app.register(websocket);

// Initialize Kafka
const kafka = new Kafka({ 
  brokers: ["localhost:9092"],
  clientId: "anomaly-radar"
});

const consumer = kafka.consumer({ groupId: "anomaly-consumers" });
const producer = kafka.producer();

// Initialize ClickHouse
const clickhouse = new ClickHouse({ 
  url: "http://localhost:8123",
  database: "anomaly_radar"
});

// Initialize WASM module
await wasm.init();

// Device statistics storage
const deviceStats = new Map<string, any>();

// WebSocket clients
const wsClients = new Set<any>();

// WebSocket endpoint
app.get("/ws", { websocket: true }, (connection, req) => {
  wsClients.add(connection);
  console.log("WebSocket client connected");

  connection.socket.on("close", () => {
    wsClients.delete(connection);
    console.log("WebSocket client disconnected");
  });
});

// REST API endpoints
app.get("/health", async (request, reply) => {
  return { status: "healthy", timestamp: new Date().toISOString() };
});

app.get("/devices", async (request, reply) => {
  try {
    const result = await clickhouse.query({
      query: "SELECT DISTINCT device FROM telemetry ORDER BY device",
      format: "JSONEachRow"
    });
    return await result.json();
  } catch (error) {
    reply.code(500).send({ error: "Failed to fetch devices" });
  }
});

app.get("/latest/:device", async (request, reply) => {
  const { device } = request.params as { device: string };
  try {
    const result = await clickhouse.query({
      query: `
        SELECT ts, device, value, score, is_anom 
        FROM anomalies 
        WHERE device = {device:String} 
        ORDER BY ts DESC 
        LIMIT 100
      `,
      query_params: { device },
      format: "JSONEachRow"
    });
    return await result.json();
  } catch (error) {
    reply.code(500).send({ error: "Failed to fetch latest data" });
  }
});

// Kafka consumer setup
await consumer.connect();
await producer.connect();
await consumer.subscribe({ topic: "telemetry", fromBeginning: true });

// Process messages
await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    try {
      const payload = JSON.parse(message.value?.toString() || "{}");
      const { ts, device, value } = payload;

      // Get or create device statistics
      if (!deviceStats.has(device)) {
        deviceStats.set(device, new wasm.SlidingStats(1000));
      }

      const stats = deviceStats.get(device);
      stats.add_value(value);

      // Calculate anomaly score
      const mean = stats.get_mean();
      const std = stats.get_std();
      const [zScore, isAnomaly] = wasm.zscore_anomaly(value, mean, std, 3.0);

      // Store in ClickHouse
      await clickhouse.insert({
        table: "telemetry",
        values: [{
          ts: new Date(ts),
          device,
          value,
          metadata: JSON.stringify({ partition, offset: message.offset })
        }],
        format: "JSONEachRow"
      });

      await clickhouse.insert({
        table: "anomalies",
        values: [{
          ts: new Date(ts),
          device,
          score: zScore,
          is_anom: isAnomaly ? 1 : 0,
          method: "zscore",
          metadata: JSON.stringify({ mean, std })
        }],
        format: "JSONEachRow"
      });

      // Broadcast to WebSocket clients
      const message = JSON.stringify({
        ts,
        device,
        value,
        score: zScore,
        is_anom: isAnomaly,
        mean,
        std
      });

      for (const client of wsClients) {
        if (client.socket.readyState === 1) { // WebSocket.OPEN
          client.socket.send(message);
        }
      }

      console.log(`Processed: ${device} = ${value} (anomaly: ${isAnomaly})`);
    } catch (error) {
      console.error("Error processing message:", error);
    }
  }
});

// Start server
const start = async () => {
  try {
    await app.listen({ port: 3000, host: "0.0.0.0" });
    console.log("Server running on http://localhost:3000");
  } catch (err) {
    app.log.error(err);
    process.exit(1);
  }
};

start();

// scripts/producer.ts
import { Kafka } from "kafkajs";

const kafka = new Kafka({ 
  brokers: ["localhost:9092"],
  clientId: "telemetry-producer"
});

const producer = kafka.producer();

// Gaussian noise generator
function gaussian(mean: number, std: number): number {
  const u = 1 - Math.random();
  const v = Math.random();
  const z = Math.sqrt(-2 * Math.log(u)) * Math.cos(2 * Math.PI * v);
  return mean + std * z;
}

// Generate synthetic telemetry data
function generateTelemetry(device: string): any {
  const baseValue = 50 + Math.sin(Date.now() / 10000) * 10;
  const noise = gaussian(0, 5);

  // Occasional spikes (2% probability)
  const spike = Math.random() < 0.02 ? gaussian(30, 10) : 0;

  const value = baseValue + noise + spike;

  return {
    ts: Date.now(),
    device,
    value: Math.round(value * 100) / 100,
    metadata: {
      source: "synthetic",
      version: "1.0"
    }
  };
}

async function startProducing() {
  await producer.connect();
  console.log("Producer connected");

  const devices = ["sensor-001", "sensor-002", "sensor-003", "sensor-004"];

  setInterval(async () => {
    for (const device of devices) {
      const telemetry = generateTelemetry(device);

      try {
        await producer.send({
          topic: "telemetry",
          messages: [{
            key: device,
            value: JSON.stringify(telemetry),
            timestamp: telemetry.ts.toString()
          }]
        });

        console.log(`Sent: ${device} = ${telemetry.value}`);
      } catch (error) {
        console.error("Error sending message:", error);
      }
    }
  }, 1000); // Send every second
}

startProducing().catch(console.error);

// src/monitoring.ts
import { ClickHouse } from "@clickhouse/client";

class PerformanceMonitor {
    private clickhouse: ClickHouse;
    private metrics: Map<string, number> = new Map();

    constructor(clickhouse: ClickHouse) {
        this.clickhouse = clickhouse;
    }

    async recordMetric(name: string, value: number) {
        this.metrics.set(name, value);

        await this.clickhouse.insert({
            table: "performance_metrics",
            values: [{
                ts: new Date(),
                metric: name,
                value: value
            }],
            format: "JSONEachRow"
        });
    }

    async getSystemHealth() {
        const queries = [
            "SELECT count() FROM telemetry WHERE ts > now() - INTERVAL 1 MINUTE",
            "SELECT count() FROM anomalies WHERE ts > now() - INTERVAL 1 MINUTE",
            "SELECT avg(score) FROM anomalies WHERE ts > now() - INTERVAL 1 MINUTE"
        ];

        const results = await Promise.all(
            queries.map(q => this.clickhouse.query({ query: q, format: "JSONEachRow" }))
        );

        return {
            telemetry_rate: await results[0].json(),
            anomaly_rate: await results[1].json(),
            avg_anomaly_score: await results[2].json()
        };
    }
}

// src/security.ts
import fastify from "fastify";
import rateLimit from "@fastify/rate-limit";
import helmet from "@fastify/helmet";

const app = fastify();

// Security headers
await app.register(helmet);

// Rate limiting
await app.register(rateLimit, {
    max: 100,
    timeWindow: '1 minute'
});

// Input validation
app.addSchema({
    $id: 'telemetry',
    type: 'object',
    properties: {
        device: { type: 'string', minLength: 1, maxLength: 50 },
        value: { type: 'number' },
        ts: { type: 'number' }
    },
    required: ['device', 'value', 'ts']
});

// tests/anomaly.test.ts
import { describe, it, expect } from 'vitest';
import * as wasm from '../anomaly-core/pkg/anomaly_core.js';

describe('Anomaly Detection', () => {
    it('should detect z-score anomalies', () => {
        const [score, isAnomaly] = wasm.zscore_anomaly(100, 50, 10, 3.0);
        expect(score).toBe(5.0);
        expect(isAnomaly).toBe(true);
    });

    it('should handle edge cases', () => {
        const [score, isAnomaly] = wasm.zscore_anomaly(50, 50, 0, 3.0);
        expect(score).toBe(0.0);
        expect(isAnomaly).toBe(false);
    });
});

// tests/integration.test.ts
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { Kafka } from 'kafkajs';

describe('Kafka Integration', () => {
    let kafka: Kafka;

    beforeAll(async () => {
        kafka = new Kafka({ brokers: ['localhost:9092'] });
    });

    it('should produce and consume messages', async () => {
        const producer = kafka.producer();
        const consumer = kafka.consumer({ groupId: 'test-group' });

        await producer.connect();
        await consumer.connect();

        await consumer.subscribe({ topic: 'test-topic' });

        // Produce message
        await producer.send({
            topic: 'test-topic',
            messages: [{ value: 'test message' }]
        });

        // Consume message
        const messages: any[] = [];
        await consumer.run({
            eachMessage: async ({ message }) => {
                messages.push(message.value?.toString());
            }
        });

        expect(messages).toContain('test message');
    });
});

// Essential anomaly detection patterns
const anomalyPatterns = {
    "kafka_streaming": "Kafka provides durable, replayable event streams",
    "clickhouse_analytics": "ClickHouse enables fast time-series analytics",
    "wasm_performance": "Rust→WASM enables high-performance anomaly detection",
    "webgpu_visualization": "WebGPU provides GPU-accelerated real-time rendering",
    "fastify_orchestration": "Fastify orchestrates the entire pipeline"
};

// Essential anomaly detection operations
// 1. Start Kafka consumer
await consumer.subscribe({ topic: "telemetry" });

// 2. Process messages with WASM
const [score, isAnomaly] = wasm.zscore_anomaly(value, mean, std, threshold);

// 3. Store in ClickHouse
await clickhouse.insert({ table: "anomalies", values: [data] });

// 4. Broadcast via WebSocket
wsClients.forEach(client => client.send(JSON.stringify(data)));

// 5. Render with WebGPU
renderPass.draw(data.length);