# Event-Driven Architecture - Detailed Reference

This document contains detailed code examples and implementation patterns for event-driven architecture.

## Kafka Setup & Configuration

### Infrastructure Setup

**Docker Compose (Local Development):**

```yaml
# deployments/local/docker-compose.yml
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    networks:
      - microservices-network

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
    networks:
      - microservices-network

  schema-registry:
    image: confluentinc/cp-schema-registry:7.4.0
    depends_on:
      - kafka
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092
    networks:
      - microservices-network
```

**Kubernetes (Production):**

Use Confluent Operator or Strimzi operator for Kafka cluster management.

### KafkaJS Client Configuration

```typescript
// src/config/kafka.config.ts
import { Kafka, logLevel } from 'kafkajs';
import { logger } from '@goodgo/logger';

const kafkaConfig = {
  brokers: (process.env.KAFKA_BROKERS || 'localhost:9092').split(','),
  clientId: process.env.KAFKA_CLIENT_ID || process.env.SERVICE_NAME || 'unknown-service',
  retry: {
    initialRetryTime: 100,
    retries: 8,
    multiplier: 2,
    maxRetryTime: 30000,
  },
  logLevel: process.env.KAFKA_LOG_LEVEL === 'debug' ? logLevel.DEBUG : logLevel.INFO,
  logCreator: () => ({ level, log }) => {
    const { message, ...extra } = log;
    if (level >= logLevel.ERROR) {
      logger.error('Kafka Error', { message, ...extra });
    } else if (level >= logLevel.WARN) {
      logger.warn('Kafka Warning', { message, ...extra });
    } else {
      logger.debug('Kafka Log', { message, ...extra });
    }
  },
};

export const kafka = new Kafka(kafkaConfig);

// Producer instance (reuse across service)
export const producer = kafka.producer({
  idempotent: true,
  maxInFlightRequests: 1,
  transactionTimeout: 30000,
});

// Admin client for topic management
export const admin = kafka.admin();
```

### Environment Variables

```bash
# Kafka Configuration
KAFKA_BROKERS=localhost:9092
KAFKA_CLIENT_ID=my-service
KAFKA_LOG_LEVEL=info

# Schema Registry
SCHEMA_REGISTRY_URL=http://localhost:8081

# Consumer Configuration
KAFKA_CONSUMER_GROUP_ID=my-service-consumers
KAFKA_CONSUMER_SESSION_TIMEOUT=30000
KAFKA_CONSUMER_HEARTBEAT_INTERVAL=3000
```

### Connection Management

```typescript
// src/core/kafka/client.ts
import { producer, admin } from '../config/kafka.config';
import { logger } from '@goodgo/logger';

let isConnected = false;

export async function connectKafka(): Promise<void> {
  if (isConnected) {
    logger.warn('Kafka already connected');
    return;
  }

  try {
    await producer.connect();
    await admin.connect();
    isConnected = true;
    logger.info('Kafka connected successfully');
  } catch (error) {
    logger.error('Failed to connect to Kafka', { error });
    throw error;
  }
}

export async function disconnectKafka(): Promise<void> {
  if (!isConnected) {
    return;
  }

  try {
    await producer.disconnect();
    await admin.disconnect();
    isConnected = false;
    logger.info('Kafka disconnected');
  } catch (error) {
    logger.error('Error disconnecting from Kafka', { error });
    throw error;
  }
}

// Graceful shutdown handler
process.on('SIGTERM', async () => {
  await disconnectKafka();
  process.exit(0);
});
```

## Event Publishing Patterns

### Event Structure

```typescript
// src/types/event.types.ts
export interface BaseEvent {
  eventId: string;
  eventType: string;
  eventVersion: string;
  timestamp: string;
  source: string;
  correlationId?: string;
  traceId?: string;
  data: unknown;
}

export interface UserCreatedEvent extends BaseEvent {
  eventType: 'user.created';
  eventVersion: '1.0.0';
  data: {
    userId: string;
    email: string;
    name: string;
    createdAt: string;
  };
}
```

### Event Publisher Service

```typescript
// src/core/events/event-publisher.ts
import { producer } from '../config/kafka.config';
import { logger } from '@goodgo/logger';
import { v4 as uuidv4 } from 'uuid';
import type { BaseEvent } from '../../types/event.types';

export class EventPublisher {
  private serviceName: string;

  constructor(serviceName: string = process.env.SERVICE_NAME || 'unknown') {
    this.serviceName = serviceName;
  }

  async publish<T extends BaseEvent>(
    topic: string,
    event: Omit<T, 'eventId' | 'timestamp' | 'source'>,
    options?: {
      partitionKey?: string;
      headers?: Record<string, string>;
    }
  ): Promise<void> {
    const fullEvent: T = {
      ...event,
      eventId: uuidv4(),
      timestamp: new Date().toISOString(),
      source: this.serviceName,
    } as T;

    try {
      await producer.send({
        topic,
        messages: [
          {
            key: options?.partitionKey || fullEvent.eventId,
            value: JSON.stringify(fullEvent),
            headers: {
              'event-type': event.eventType,
              'event-version': event.eventVersion,
              'correlation-id': event.correlationId || '',
              ...options?.headers,
            },
          },
        ],
      });

      logger.info('Event published', {
        topic,
        eventType: event.eventType,
        eventId: fullEvent.eventId,
        correlationId: event.correlationId,
      });
    } catch (error) {
      logger.error('Failed to publish event', {
        topic,
        eventType: event.eventType,
        error,
      });
      throw error;
    }
  }

  async publishBatch<T extends BaseEvent>(
    events: Array<{
      topic: string;
      event: Omit<T, 'eventId' | 'timestamp' | 'source'>;
      partitionKey?: string;
    }>
  ): Promise<void> {
    const messages = events.map(({ topic, event, partitionKey }) => {
      const fullEvent: T = {
        ...event,
        eventId: uuidv4(),
        timestamp: new Date().toISOString(),
        source: this.serviceName,
      } as T;

      return {
        topic,
        messages: [
          {
            key: partitionKey || fullEvent.eventId,
            value: JSON.stringify(fullEvent),
            headers: {
              'event-type': event.eventType,
              'event-version': event.eventVersion,
              'correlation-id': event.correlationId || '',
            },
          },
        ],
      };
    });

    try {
      await producer.sendBatch({
        topicMessages: messages,
      });

      logger.info('Events published in batch', {
        count: events.length,
      });
    } catch (error) {
      logger.error('Failed to publish events batch', { error });
      throw error;
    }
  }
}

export const eventPublisher = new EventPublisher();
```

### Publishing Events from Services

```typescript
// src/modules/user/user.service.ts
import { eventPublisher } from '../../core/events/event-publisher';
import { logger } from '@goodgo/logger';
import type { UserCreatedEvent } from '../../types/event.types';

export class UserService {
  async createUser(data: CreateUserDto, correlationId?: string): Promise<User> {
    const user = await this.userRepository.create(data);

    try {
      await eventPublisher.publish<UserCreatedEvent>(
        'user.created',
        {
          eventType: 'user.created',
          eventVersion: '1.0.0',
          correlationId,
          data: {
            userId: user.id,
            email: user.email,
            name: user.name,
            createdAt: user.createdAt.toISOString(),
          },
        },
        {
          partitionKey: user.id,
        }
      );
    } catch (error) {
      logger.error('Failed to publish user.created event', { error, userId: user.id });
    }

    return user;
  }
}
```

### Transactional Publishing (Outbox Pattern)

```typescript
// src/core/events/outbox-pattern.ts
import { PrismaClient } from '@prisma/client';
import { EventPublisher } from './event-publisher';
import { logger } from '@goodgo/logger';

export class OutboxService {
  constructor(private prisma: PrismaClient) {}

  async addToOutbox(
    eventType: string,
    eventData: unknown,
    topic: string,
    correlationId?: string
  ): Promise<void> {
    await this.prisma.outboxEvent.create({
      data: {
        eventType,
        eventData: eventData as any,
        topic,
        correlationId,
        status: 'PENDING',
      },
    });
  }

  async processOutbox(eventPublisher: EventPublisher): Promise<void> {
    const events = await this.prisma.outboxEvent.findMany({
      where: { status: 'PENDING' },
      take: 100,
      orderBy: { createdAt: 'asc' },
    });

    for (const event of events) {
      try {
        await eventPublisher.publish(
          event.topic,
          {
            eventType: event.eventType,
            eventVersion: '1.0.0',
            correlationId: event.correlationId || undefined,
            data: event.eventData,
          },
          {
            partitionKey: event.id,
          }
        );

        await this.prisma.outboxEvent.update({
          where: { id: event.id },
          data: { status: 'PUBLISHED', publishedAt: new Date() },
        });
      } catch (error) {
        logger.error('Failed to process outbox event', { eventId: event.id, error });
        await this.prisma.outboxEvent.update({
          where: { id: event.id },
          data: { status: 'FAILED', retryCount: { increment: 1 } },
        });
      }
    }
  }
}

// Using outbox in transaction
export class UserService {
  constructor(
    private prisma: PrismaClient,
    private outboxService: OutboxService
  ) {}

  async createUserWithOutbox(data: CreateUserDto): Promise<User> {
    return await this.prisma.$transaction(async (tx) => {
      const user = await tx.user.create({ data });

      await this.outboxService.addToOutbox(
        'user.created',
        {
          userId: user.id,
          email: user.email,
          name: user.name,
        },
        'user.created'
      );

      return user;
    });
  }
}
```

## Event Consuming Patterns

### Event Consumer Service

```typescript
// src/core/events/event-consumer.ts
import { kafka } from '../config/kafka.config';
import { logger } from '@goodgo/logger';
import type { BaseEvent } from '../../types/event.types';

export interface EventHandler<T extends BaseEvent = BaseEvent> {
  handle(event: T): Promise<void>;
}

export class EventConsumer {
  private consumer: ReturnType<typeof kafka.consumer>;
  private handlers: Map<string, EventHandler[]> = new Map();

  constructor(
    groupId: string = process.env.KAFKA_CONSUMER_GROUP_ID || 'default-group'
  ) {
    this.consumer = kafka.consumer({
      groupId,
      sessionTimeout: 30000,
      heartbeatInterval: 3000,
    });
  }

  on<T extends BaseEvent>(eventType: string, handler: EventHandler<T>): void {
    if (!this.handlers.has(eventType)) {
      this.handlers.set(eventType, []);
    }
    this.handlers.get(eventType)!.push(handler as EventHandler);
  }

  async start(topics: string[]): Promise<void> {
    await this.consumer.connect();
    await this.consumer.subscribe({ topics, fromBeginning: false });

    await this.consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        try {
          const event: BaseEvent = JSON.parse(message.value?.toString() || '{}');
          const handlers = this.handlers.get(event.eventType) || [];

          if (handlers.length === 0) {
            logger.warn('No handler registered for event type', {
              eventType: event.eventType,
              topic,
            });
            return;
          }

          await Promise.all(handlers.map((handler) => handler.handle(event)));

          logger.debug('Event processed successfully', {
            eventType: event.eventType,
            eventId: event.eventId,
            topic,
            partition,
          });
        } catch (error) {
          logger.error('Error processing event', {
            topic,
            partition,
            offset: message.offset,
            error,
          });
          throw error;
        }
      },
    });

    logger.info('Event consumer started', { topics, groupId: this.consumer.groupId });
  }

  async stop(): Promise<void> {
    await this.consumer.disconnect();
    logger.info('Event consumer stopped');
  }
}
```

### Consumer Implementation

```typescript
// src/modules/notification/notification.consumer.ts
import { EventConsumer, EventHandler } from '../../core/events/event-consumer';
import { logger } from '@goodgo/logger';
import type { UserCreatedEvent } from '../../types/event.types';

class UserCreatedHandler implements EventHandler<UserCreatedEvent> {
  async handle(event: UserCreatedEvent): Promise<void> {
    logger.info('Processing user.created event', { eventId: event.eventId, userId: event.data.userId });

    await this.sendWelcomeEmail(event.data.email, event.data.name);
    await this.createNotificationPreferences(event.data.userId);
  }

  private async sendWelcomeEmail(email: string, name: string): Promise<void> {
    // Email sending logic
  }

  private async createNotificationPreferences(userId: string): Promise<void> {
    // Create default preferences
  }
}

export const notificationConsumer = new EventConsumer('notification-service');
notificationConsumer.on('user.created', new UserCreatedHandler());
await notificationConsumer.start(['user.created']);
```

### Dead Letter Queue (DLQ) Pattern

```typescript
// src/core/events/dlq-handler.ts
import { eventPublisher } from './event-publisher';
import { logger } from '@goodgo/logger';

export class DLQHandler {
  async sendToDLQ(
    originalEvent: BaseEvent,
    originalTopic: string,
    error: Error,
    retryCount: number
  ): Promise<void> {
    const dlqEvent = {
      ...originalEvent,
      eventType: `${originalEvent.eventType}.dlq`,
      data: {
        originalEvent,
        originalTopic,
        error: {
          message: error.message,
          stack: error.stack,
        },
        retryCount,
        dlqTimestamp: new Date().toISOString(),
      },
    };

    await eventPublisher.publish(`${originalTopic}.dlq`, dlqEvent);

    logger.error('Event sent to DLQ', {
      originalEventType: originalEvent.eventType,
      originalTopic,
      retryCount,
      error: error.message,
    });
  }
}

// Consumer with DLQ support
export class EventConsumerWithDLQ extends EventConsumer {
  private dlqHandler = new DLQHandler();
  private maxRetries = 3;

  async start(topics: string[]): Promise<void> {
    await this.consumer.connect();
    await this.consumer.subscribe({ topics, fromBeginning: false });

    await this.consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        let retryCount = 0;
        const maxAttempts = this.maxRetries + 1;

        while (retryCount < maxAttempts) {
          try {
            const event: BaseEvent = JSON.parse(message.value?.toString() || '{}');
            const handlers = this.handlers.get(event.eventType) || [];

            if (handlers.length === 0) {
              logger.warn('No handler registered for event type', {
                eventType: event.eventType,
                topic,
              });
              return;
            }

            await Promise.all(handlers.map((handler) => handler.handle(event)));
            return;
          } catch (error) {
            retryCount++;
            if (retryCount >= maxAttempts) {
              const event: BaseEvent = JSON.parse(message.value?.toString() || '{}');
              await this.dlqHandler.sendToDLQ(event, topic, error as Error, retryCount);
              return;
            }

            const delay = Math.min(1000 * Math.pow(2, retryCount), 30000);
            await new Promise((resolve) => setTimeout(resolve, delay));

            logger.warn('Retrying event processing', {
              topic,
              partition,
              offset: message.offset,
              retryCount,
              delay,
            });
          }
        }
      },
    });
  }
}
```

## Schema Versioning

### Schema Registry Setup

```typescript
// src/core/schema/schema-registry.ts
import { SchemaRegistry } from '@kafkajs/confluent-schema-registry';
import { logger } from '@goodgo/logger';

const registry = new SchemaRegistry({
  host: process.env.SCHEMA_REGISTRY_URL || 'http://localhost:8081',
});

export async function registerSchema(
  subject: string,
  schema: object
): Promise<number> {
  try {
    const { id } = await registry.register({ type: 'AVRO', schema: JSON.stringify(schema) });
    logger.info('Schema registered', { subject, id });
    return id;
  } catch (error) {
    logger.error('Failed to register schema', { subject, error });
    throw error;
  }
}

export async function getSchema(subject: string, version?: number) {
  try {
    const schema = await registry.getSchema(subject, version);
    return schema;
  } catch (error) {
    logger.error('Failed to get schema', { subject, version, error });
    throw error;
  }
}
```

### Avro Schema Definition

```json
// schemas/user.created.v1.avsc
{
  "type": "record",
  "name": "UserCreated",
  "namespace": "com.goodgo.events.user",
  "fields": [
    { "name": "userId", "type": "string" },
    { "name": "email", "type": "string" },
    { "name": "name", "type": "string" },
    { "name": "createdAt", "type": "string" }
  ]
}
```

### Schema Evolution

**Backward Compatible Changes (Safe):**
- Adding optional fields
- Removing fields (old consumers ignore)

**Forward Compatible Changes:**
- Removing optional fields
- Adding required fields (with defaults)

**Breaking Changes:**
- Change field type
- Remove required field
- Rename field

## Traefik Integration (SSE/WebSocket)

### SSE Endpoint Implementation

```typescript
// src/modules/events/events.controller.ts
import { Request, Response } from 'express';
import { kafka } from '../../config/kafka.config';
import { logger } from '@goodgo/logger';

export class EventsController {
  async streamEvents(req: Request, res: Response): Promise<void> {
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');
    res.setHeader('X-Accel-Buffering', 'no');

    const topic = req.query.topic as string;
    const eventType = req.query.eventType as string | undefined;

    if (!topic) {
      res.status(400).json({ success: false, error: { code: 'TOPIC_REQUIRED', message: 'Topic parameter required' } });
      return;
    }

    const consumer = kafka.consumer({
      groupId: `sse-${Date.now()}`,
    });

    try {
      await consumer.connect();
      await consumer.subscribe({ topic, fromBeginning: false });

      res.write(`data: ${JSON.stringify({ type: 'connected', topic })}\n\n`);

      await consumer.run({
        eachMessage: async ({ message }) => {
          try {
            const event = JSON.parse(message.value?.toString() || '{}');

            if (eventType && event.eventType !== eventType) {
              return;
            }

            res.write(`data: ${JSON.stringify(event)}\n\n`);
          } catch (error) {
            logger.error('Error processing SSE message', { error });
          }
        },
      });

      req.on('close', async () => {
        await consumer.disconnect();
        logger.info('SSE connection closed', { topic });
      });
    } catch (error) {
      logger.error('SSE connection error', { error });
      res.write(`data: ${JSON.stringify({ type: 'error', message: 'Connection failed' })}\n\n`);
      await consumer.disconnect();
    }
  }
}
```

### Traefik Routing Configuration

```yaml
# infra/traefik/dynamic/routes.yml
http:
  routers:
    events-sse:
      rule: "PathPrefix(`/api/v1/events/stream`)"
      entrypoints:
        - web
      service: events-service
      middlewares:
        - cors
        - compress
        - events-ratelimit

  services:
    events-service:
      loadBalancer:
        servers:
          - url: "http://events-service:5000"

  middlewares:
    events-ratelimit:
      rateLimit:
        average: 100
        period: 1m
        burst: 50
```

### Client Subscription Example

```typescript
// Client-side SSE subscription
const eventSource = new EventSource('/api/v1/events/stream?topic=user.created&eventType=user.created');

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log('Received event:', data);
};

eventSource.onerror = (error) => {
  console.error('SSE error:', error);
  eventSource.close();
};
```

## Error Handling & Resilience

### Event Publishing Errors

```typescript
import { retryWithBackoff } from '../resilience/retry';

async function publishWithRetry<T extends BaseEvent>(
  topic: string,
  event: T,
  maxRetries: number = 3
): Promise<void> {
  try {
    await retryWithBackoff(
      async () => {
        await eventPublisher.publish(topic, event);
      },
      maxRetries,
      1000
    );
  } catch (error) {
    logger.error('Failed to publish event after retries', {
      topic,
      eventType: event.eventType,
      eventId: event.eventId,
      error,
    });
    throw error;
  }
}
```

### Circuit Breaker for Kafka Operations

```typescript
import { createCircuitBreaker } from '../resilience/circuit-breaker';

const kafkaCircuitBreaker = createCircuitBreaker(
  async (topic: string, event: BaseEvent) => {
    await eventPublisher.publish(topic, event);
  },
  'kafka-producer',
  {
    timeout: 5000,
    errorThresholdPercentage: 50,
    resetTimeout: 30000,
  }
);

try {
  await kafkaCircuitBreaker.fire('user.created', event);
} catch (error) {
  if (kafkaCircuitBreaker.opened) {
    await storeEventForLaterProcessing(event);
  }
  throw error;
}
```

## Observability

### Event Logging

```typescript
import { logger } from '@goodgo/logger';

// Log when publishing
logger.info('Event published', {
  eventType: event.eventType,
  eventId: event.eventId,
  topic,
  correlationId: event.correlationId,
  traceId: event.traceId,
  source: event.source,
});

// Log when consuming
logger.info('Event consumed', {
  eventType: event.eventType,
  eventId: event.eventId,
  topic,
  partition,
  offset,
  processingTime: Date.now() - eventTimestamp,
});
```

### Metrics Collection

```typescript
// src/core/events/event-metrics.ts
import { Counter, Histogram, Gauge } from 'prom-client';

const eventsPublished = new Counter({
  name: 'events_published_total',
  help: 'Total number of events published',
  labelNames: ['topic', 'event_type'],
});

const eventsConsumed = new Counter({
  name: 'events_consumed_total',
  help: 'Total number of events consumed',
  labelNames: ['topic', 'event_type'],
});

const eventProcessingDuration = new Histogram({
  name: 'event_processing_duration_seconds',
  help: 'Event processing duration in seconds',
  labelNames: ['topic', 'event_type'],
});

const consumerLag = new Gauge({
  name: 'kafka_consumer_lag',
  help: 'Consumer lag per topic and partition',
  labelNames: ['topic', 'partition'],
});

// Usage in publisher
eventsPublished.inc({ topic, event_type: event.eventType });

// Usage in consumer
const timer = eventProcessingDuration.startTimer({ topic, event_type: event.eventType });
try {
  await handler.handle(event);
  eventsConsumed.inc({ topic, event_type: event.eventType });
} finally {
  timer();
}
```

### Distributed Tracing

```typescript
import { trace, context, SpanStatusCode } from '@opentelemetry/api';

const tracer = trace.getTracer('event-service');

const span = tracer.startSpan('event.publish', {
  attributes: {
    'event.type': event.eventType,
    'event.topic': topic,
    'event.id': event.eventId,
  },
});

try {
  await eventPublisher.publish(topic, event);
  span.setStatus({ code: SpanStatusCode.OK });
} catch (error) {
  span.setStatus({ code: SpanStatusCode.ERROR, message: error.message });
  throw error;
} finally {
  span.end();
}
```

### Health Checks

```typescript
// src/routes/health.routes.ts
import { admin } from '../config/kafka.config';

router.get('/health/kafka', async (req, res) => {
  try {
    await admin.listTopics();
    res.json({ status: 'healthy', kafka: 'connected' });
  } catch (error) {
    res.status(503).json({ status: 'unhealthy', kafka: 'disconnected', error: error.message });
  }
});
```

## Testing Patterns

### Unit Testing Publishers

```typescript
// src/core/events/__tests__/event-publisher.test.ts
import { EventPublisher } from '../event-publisher';
import { producer } from '../../config/kafka.config';

jest.mock('../../config/kafka.config');

describe('EventPublisher', () => {
  let publisher: EventPublisher;
  const mockSend = jest.fn();

  beforeEach(() => {
    publisher = new EventPublisher('test-service');
    (producer.send as jest.Mock) = mockSend;
    mockSend.mockResolvedValue({});
  });

  it('should publish event successfully', async () => {
    const event = {
      eventType: 'user.created',
      eventVersion: '1.0.0',
      data: { userId: '123' },
    };

    await publisher.publish('user.created', event);

    expect(mockSend).toHaveBeenCalledWith({
      topic: 'user.created',
      messages: expect.arrayContaining([
        expect.objectContaining({
          key: expect.any(String),
          value: expect.stringContaining('"eventType":"user.created"'),
        }),
      ]),
    });
  });
});
```

### Integration Testing with Test Containers

```typescript
// src/__tests__/integration/event-flow.e2e.ts
import { KafkaContainer, StartedKafkaContainer } from '@testcontainers/kafka';
import { EventPublisher } from '../../core/events/event-publisher';
import { EventConsumer } from '../../core/events/event-consumer';
import type { BaseEvent } from '../../types/event.types';

describe('Event Flow E2E', () => {
  let kafkaContainer: StartedKafkaContainer;
  let publisher: EventPublisher;
  let consumer: EventConsumer;

  beforeAll(async () => {
    kafkaContainer = await new KafkaContainer().start();
    process.env.KAFKA_BROKERS = kafkaContainer.getBootstrapServer();

    publisher = new EventPublisher('test-service');
    consumer = new EventConsumer('test-group');
  });

  afterAll(async () => {
    await kafkaContainer.stop();
  });

  it('should publish and consume event', async () => {
    const receivedEvents: BaseEvent[] = [];

    consumer.on('user.created', {
      handle: async (event) => {
        receivedEvents.push(event);
      },
    });

    await consumer.start(['user.created']);

    const event = {
      eventType: 'user.created',
      eventVersion: '1.0.0',
      data: { userId: '123', email: 'test@example.com' },
    };

    await publisher.publish('user.created', event);

    await new Promise((resolve) => setTimeout(resolve, 1000));

    expect(receivedEvents).toHaveLength(1);
    expect(receivedEvents[0].data.userId).toBe('123');
  });
});
```

### Mocking Kafka

```typescript
// src/__tests__/mocks/kafka.mock.ts
export const createMockKafka = () => {
  const messages: any[] = [];

  return {
    producer: () => ({
      connect: jest.fn(),
      send: jest.fn((params) => {
        messages.push(...params.messages);
        return Promise.resolve({});
      }),
      disconnect: jest.fn(),
    }),
    consumer: () => ({
      connect: jest.fn(),
      subscribe: jest.fn(),
      run: jest.fn(),
      disconnect: jest.fn(),
    }),
    admin: () => ({
      connect: jest.fn(),
      listTopics: jest.fn().mockResolvedValue([]),
      disconnect: jest.fn(),
    }),
    messages,
  };
};
```

## Complete Examples

### Example 1: User Created Event Flow

```typescript
// Auth Service publishes user.created event
// services/auth-service/src/modules/user/user.service.ts
export class UserService {
  async createUser(data: CreateUserDto): Promise<User> {
    const user = await this.userRepository.create(data);

    await eventPublisher.publish<UserCreatedEvent>(
      'user.created',
      {
        eventType: 'user.created',
        eventVersion: '1.0.0',
        data: {
          userId: user.id,
          email: user.email,
          name: user.name,
        },
      },
      { partitionKey: user.id }
    );

    return user;
  }
}

// Notification Service consumes and sends welcome email
// services/notification-service/src/modules/notification/notification.consumer.ts
notificationConsumer.on('user.created', {
  handle: async (event: UserCreatedEvent) => {
    await emailService.sendWelcomeEmail(event.data.email, event.data.name);
    await notificationPreferenceService.createDefaults(event.data.userId);
  },
});
```

### Example 2: Order Placed with Multiple Consumers

```typescript
// Order Service publishes order.placed
await eventPublisher.publish('order.placed', {
  eventType: 'order.placed',
  eventVersion: '1.0.0',
  data: {
    orderId: order.id,
    userId: order.userId,
    total: order.total,
    items: order.items,
  },
});

// Multiple consumers handle the event

// Payment Service: Process payment
paymentConsumer.on('order.placed', { handle: processPayment });

// Inventory Service: Reserve items
inventoryConsumer.on('order.placed', { handle: reserveItems });

// Notification Service: Send confirmation
notificationConsumer.on('order.placed', { handle: sendConfirmation });
```

### Example 3: SSE Endpoint with Filtering

```typescript
// Client subscribes to user events
const eventSource = new EventSource(
  '/api/v1/events/stream?topic=user.created&eventType=user.created',
  {
    headers: {
      Authorization: `Bearer ${token}`,
    },
  }
);

eventSource.addEventListener('user.created', (event) => {
  const data = JSON.parse(event.data);
  updateUI(data);
});
```
