Kafka Stream PII Sanitization Worker: High-Throughput Node.js Microservice

import { Kafka } from 'kafkajs';
import { sanitize } from '@privacyscrubber/sdk';

const kafka = new Kafka({ clientId: 'pii-scrubber-worker', brokers: ['localhost:9092'] });
const consumer = kafka.consumer({ groupId: 'ai-sanitization-group' });
const producer = kafka.producer();

await consumer.connect();
await producer.connect();
await consumer.subscribe({ topic: 'raw-user-events', fromBeginning: false });

await consumer.run({
  eachBatch: async ({ batch }) => {
    const sanitizedBatch = batch.messages.map((message) => {
      const rawString = message.value?.toString() || '';

      // In-memory sanitization (<0.45ms)
      const { scrubbedText } = sanitize(rawString, {
        profile: 'General',
        detectSecrets: true
      });

      return {
        key: message.key,
        value: scrubbedText
      };
    });

    // Produce clean batch to downstream AI ingestion topic
    await producer.send({
      topic: 'sanitized-ai-events',
      messages: sanitizedBatch
    });
  }
});

console.log('Kafka stream sanitization worker running active in-memory.');