// CloudEvents standard format
interface CloudEvent<T> {
  // Required attributes
  specversion: "1.0";
  id: string;          // Unique event identifier
  source: string;      // Event source URI
  type: string;        // Event type
  
  // Optional attributes
  datacontenttype?: string;
  dataschema?: string;
  subject?: string;
  time?: string;       // RFC 3339 timestamp
  
  // Extension attributes
  [key: string]: unknown;
  
  // Event data
  data: T;
}

// Example: Order Created Event
interface OrderCreatedData {
  orderId: string;
  customerId: string;
  items: OrderItem[];
  totalAmount: number;
  currency: string;
}

const orderCreatedEvent: CloudEvent<OrderCreatedData> = {
  specversion: "1.0",
  id: "A234-1234-1234",
  source: "/orders/service",
  type: "com.example.order.created",
  datacontenttype: "application/json",
  time: "2026-02-12T13:45:00Z",
  subject: "order-12345",
  data: {
    orderId: "order-12345",
    customerId: "customer-789",
    items: [
      { productId: "prod-1", quantity: 2, price: 29.99 },
      { productId: "prod-2", quantity: 1, price: 49.99 }
    ],
    totalAmount: 109.97,
    currency: "USD"
  }
};

// schemas/order-events.avsc
{
  "type": "record",
  "name": "OrderCreated",
  "namespace": "com.example.orders",
  "fields": [
    {"name": "orderId", "type": "string"},
    {"name": "customerId", "type": "string"},
    {"name": "items", "type": {"type": "array", "items": {
      "type": "record",
      "name": "OrderItem",
      "fields": [
        {"name": "productId", "type": "string"},
        {"name": "quantity", "type": "int"},
        {"name": "price", "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2}}
      ]
    }}},
    {"name": "totalAmount", "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2}},
    {"name": "currency", "type": "string"},
    {"name": "createdAt", "type": {"type": "long", "logicalType": "timestamp-millis"}}
  ]
}

// src/kafka/producer.ts
import { Kafka, Producer, Partitioners, CompressionTypes } from "kafkajs";
import { SchemaRegistry, SchemaType } from "@kafkajs/confluent-schema-registry";

interface ProducerConfig {
  brokers: string[];
  clientId: string;
  schemaRegistryUrl: string;
}

export class EventProducer {
  private kafka: Kafka;
  private producer: Producer;
  private registry: SchemaRegistry;
  private schemaCache: Map<string, number> = new Map();

  constructor(config: ProducerConfig) {
    this.kafka = new Kafka({
      clientId: config.clientId,
      brokers: config.brokers,
      retry: {
        initialRetryTime: 100,
        retries: 8,
        maxRetryTime: 30000,
        factor: 2
      }
    });

    this.producer = this.kafka.producer({
      createPartitioner: Partitioners.DefaultPartitioner,
      allowAutoTopicCreation: false,
      transactionTimeout: 30000,
      idempotent: true
    });

    this.registry = new SchemaRegistry({ host: config.schemaRegistryUrl });
  }

  async connect(): Promise<void> {
    await this.producer.connect();
    console.log("Producer connected to Kafka");
  }

  async publishEvent<T>(
    topic: string,
    key: string,
    event: CloudEvent<T>,
    schemaSubject: string
  ): Promise<void> {
    // Get or register schema
    let schemaId = this.schemaCache.get(schemaSubject);
    if (!schemaId) {
      const schema = await this.registry.getLatestSchemaId(schemaSubject);
      schemaId = schema;
      this.schemaCache.set(schemaSubject, schemaId);
    }

    // Encode with Avro
    const encodedValue = await this.registry.encode(schemaId, event);

    await this.producer.send({
      topic,
      compression: CompressionTypes.LZ4,
      messages: [
        {
          key: Buffer.from(key),
          value: encodedValue,
          headers: {
            "ce-specversion": event.specversion,
            "ce-id": event.id,
            "ce-source": event.source,
            "ce-type": event.type,
            "ce-time": event.time || new Date().toISOString()
          }
        }
      ]
    });
  }

  // Transactional publishing for exactly-once semantics
  async publishEventsTransactionally<T>(
    events: Array<{ topic: string; key: string; event: CloudEvent<T>; schema: string }>
  ): Promise<void> {
    const transaction = await this.producer.transaction();

    try {
      for (const { topic, key, event, schema } of events) {
        const schemaId = await this.registry.getLatestSchemaId(schema);
        const encodedValue = await this.registry.encode(schemaId, event);

        await transaction.send({
          topic,
          messages: [{ key: Buffer.from(key), value: encodedValue }]
        });
      }

      await transaction.commit();
    } catch (error) {
      await transaction.abort();
      throw error;
    }
  }

  async disconnect(): Promise<void> {
    await this.producer.disconnect();
  }
}

// src/kafka/consumer.ts
import { Kafka, Consumer, EachMessagePayload } from "kafkajs";
import { SchemaRegistry } from "@kafkajs/confluent-schema-registry";

interface ConsumerConfig {
  brokers: string[];
  groupId: string;
  clientId: string;
  schemaRegistryUrl: string;
}

type EventHandler<T> = (event: CloudEvent<T>, metadata: EventMetadata) => Promise<void>;

interface EventMetadata {
  topic: string;
  partition: number;
  offset: string;
  timestamp: string;
  headers: Record<string, string>;
}

export class EventConsumer {
  private kafka: Kafka;
  private consumer: Consumer;
  private registry: SchemaRegistry;
  private handlers: Map<string, EventHandler<unknown>> = new Map();

  constructor(config: ConsumerConfig) {
    this.kafka = new Kafka({
      clientId: config.clientId,
      brokers: config.brokers
    });

    this.consumer = this.kafka.consumer({
      groupId: config.groupId,
      sessionTimeout: 30000,
      heartbeatInterval: 3000,
      maxBytesPerPartition: 1048576,
      retry: { retries: 5 }
    });

    this.registry = new SchemaRegistry({ host: config.schemaRegistryUrl });
  }

  registerHandler<T>(eventType: string, handler: EventHandler<T>): void {
    this.handlers.set(eventType, handler as EventHandler<unknown>);
  }

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

    await this.consumer.run({
      eachMessage: async (payload: EachMessagePayload) => {
        await this.handleMessage(payload);
      }
    });
  }

  private async handleMessage(payload: EachMessagePayload): Promise<void> {
    const { topic, partition, message } = payload;
    
    try {
      // Decode Avro message
      const decodedValue = await this.registry.decode(message.value!);
      
      // Extract CloudEvents headers
      const headers: Record<string, string> = {};
      for (const [key, value] of Object.entries(message.headers || {})) {
        headers[key] = value?.toString() || "";
      }

      const eventType = headers["ce-type"];
      const handler = this.handlers.get(eventType);

      if (!handler) {
        console.warn(`No handler registered for event type: ${eventType}`);
        return;
      }

      const metadata: EventMetadata = {
        topic,
        partition,
        offset: message.offset,
        timestamp: message.timestamp,
        headers
      };

      await handler(decodedValue, metadata);
      
    } catch (error) {
      console.error("Error processing message:", error);
      // Implement dead letter queue logic here
      await this.sendToDeadLetterQueue(topic, message, error);
    }
  }

  private async sendToDeadLetterQueue(
    originalTopic: string,
    message: unknown,
    error: Error
  ): Promise<void> {
    // Publish to DLQ for manual inspection
    const dlqTopic = `${originalTopic}.dlq`;
    // Implementation details...
  }

  async disconnect(): Promise<void> {
    await this.consumer.disconnect();
  }
}