message EventEnvelope {
  // Уникальный ID события (UUID)
  string id = 1;
  // Источник (например, "maniton.service.ledger")
  string source = 2;
  // Тип события (например, "maniton.ledger.v1.BalanceUpdated")
  string type = 3;
  // Время возникновения
  google.protobuf.Timestamp time = 4;
  // Сами данные (сериализованный Protobuf)
  bytes data = 5;
  // Контекст запроса для трассировки и идемпотентности
  RequestContext context = 6;
}

message RequestContext {
  string request_id = 1;        // UUID запроса
  string correlation_id = 2;    // UUID для связи запрос-ответ
  string idempotency_key = 3;   // Ключ идемпотентности
  string trace_id = 5;          // OpenTelemetry trace_id
  string user_id = 6;           // Инициатор (если есть)
  string client_ip = 8;
  Contour contour = 9;          // RF (РФ) или EXTERNAL (Внешний)
}

message KycStatusChangedEvent {
  RequestContext context = 1;
  string user_id = 2;
  KycStatus old_status = 3;
  KycStatus new_status = 4;
  google.protobuf.Timestamp changed_at = 5;
}

message BalanceUpdatedEvent {
  RequestContext context = 1;
  string subaccount_id = 2;
  Decimal old_balance = 3;
  Decimal new_balance = 4;
  string operation_id = 5;
  google.protobuf.Timestamp updated_at = 6;
}

message CfaIssuedEvent {
  RequestContext context = 1;
  string operation_id = 2;
  string instrument_id = 3;
  string holder_id = 4;
  MonetaryAmount amount = 5;
  string tx_hash = 6;
  google.protobuf.Timestamp issued_at = 7;
}

message TransactionReceiptReadyEvent {
  RequestContext context = 1;
  TransactionReceipt receipt = 2;
}

message TransactionReceipt {
  string tx_hash = 1;
  BesuTxStatus status = 2;
  uint64 block_number = 3;
  string block_hash = 4;
  uint64 gas_used = 5;
  google.protobuf.Timestamp mined_at = 6;
}

message FiatDepositedEvent {
  RequestContext context = 1;
  string user_id = 2;
  Money amount = 3;
  string payment_id = 4;
  google.protobuf.Timestamp deposited_at = 5;
}

message TradeExecutedEvent {
  RequestContext context = 1;
  string trade_id = 2;
  string buyer_id = 3;
  string seller_id = 4;
  string instrument_id = 5;
  Decimal quantity = 6;
  Decimal price = 7;
  Fee fee = 8;
  google.protobuf.Timestamp executed_at = 9;
}

message SecurityEvent {
  RequestContext context = 1;
  string event_type = 2;
  string user_id = 3;
  string ip = 4;
  map<string, string> details = 5;
  google.protobuf.Timestamp occurred_at = 6;
}

# Создание топика с партиционированием
kafka-topics.sh --create \
  --bootstrap-server localhost:9092 \
  --topic maniton.ledger.events.v1 \
  --partitions 10 \
  --replication-factor 3 \
  --config retention.ms=2592000000 \
  --config cleanup.policy=compact \
  --config segment.ms=86400000

import { Producer } from 'kafkajs';

const producer = kafka.producer({
  maxInFlightRequests: 1,
  idempotent: true,
  transactionTimeout: 30000,
  retries: {
    maxAttempts: 5,
    initialRetryTime: 100,
    maxRetryTime: 1000,
  },
});

await producer.connect();

// Отправка события
await producer.send({
  topic: 'maniton.ledger.events.v1',
  messages: [
    {
      key: operationId,
      value: Buffer.from(eventEnvelope.toBinary()),
      headers: {
        'event-type': 'OperationCreated',
        'content-type': 'application/protobuf',
      },
    },
  ],
});

import { Consumer } from 'kafkajs';

const consumer = kafka.consumer({
  groupId: 'ledger-service',
  maxWaitTimeInMs: 1000,
  maxBytesPerPartition: 1048576, // 1MB
  sessionTimeout: 30000,
  heartbeatInterval: 3000,
});

await consumer.subscribe({
  topic: 'maniton.cfa.events.v1',
  fromBeginning: false,
});

await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      const eventEnvelope = EventEnvelope.fromBinary(message.value);

      // Идемпотентность
      const existing = await this.operationRepository.findByIdempotencyKey(
        eventEnvelope.context.idempotencyKey
      );

      if (existing) {
        return; // Уже обработано
      }

      // Обработка события
      await this.handleEvent(eventEnvelope);
    },
  },
});

async handleEvent(event: EventEnvelope) {
  // Проверка идемпотентности
  const processed = await this.processedEventsRepository.findById(
    event.id
  );

  if (processed) {
    this.logger.warn('Event already processed', { eventId: event.id });
    return;
  }

  // Обработка события
  await this.processEvent(event);

  // Сохранение метки обработки
  await this.processedEventsRepository.save({
    eventId: event.id,
    processedAt: new Date(),
  });
}

async processEvent(event: EventEnvelope) {
  try {
    await this.handleEvent(event);
  } catch (error) {
    // Логирование ошибки
    this.logger.error('Event processing failed', {
      eventId: event.id,
      error: error.message,
      stack: error.stack,
    });

    // Отправка в Dead Letter Queue
    await this.dlqProducer.send({
      topic: 'maniton.dlq.v1',
      messages: [
        {
          key: event.id,
          value: Buffer.from(event.toBinary()),
          headers: {
            'original-topic': event.source,
            'error': error.message,
          },
        },
      ],
    });

    // Повторная попытка или ручной разбор
    throw error;
  }
}

function validateEventEnvelope(envelope: EventEnvelope): void {
  if (!envelope.id) {
    throw new Error('Event ID is required');
  }

  if (!envelope.type) {
    throw new Error('Event type is required');
  }

  if (!envelope.context) {
    throw new Error('Request context is required');
  }

  if (!envelope.context.requestId) {
    throw new Error('Request ID is required');
  }

  if (!envelope.context.idempotencyKey) {
    throw new Error('Idempotency key is required');
  }
}

this.logger.log('Event received', {
  eventId: event.id,
  type: event.type,
  source: event.source,
  userId: event.context?.userId,
  traceId: event.context?.traceId,
});

this.logger.log('Event processed', {
  eventId: event.id,
  type: event.type,
  duration: `${Date.now() - startTime}ms`,
});

import { Counter, Histogram, Gauge } from 'prom-client';

export const kafkaMessagesProducedTotal = new Counter({
  name: 'kafka_messages_produced_total',
  help: 'Total number of messages produced',
  labelNames: ['topic'],
});

export const kafkaMessagesConsumedTotal = new Counter({
  name: 'kafka_messages_consumed_total',
  help: 'Total number of messages consumed',
  labelNames: ['topic', 'consumer_group'],
});

export const kafkaConsumerLag = new Gauge({
  name: 'kafka_consumer_lag',
  help: 'Consumer lag by topic and partition',
  labelNames: ['topic', 'partition', 'consumer_group'],
});

export const kafkaMessageProcessingDuration = new Histogram({
  name: 'kafka_message_processing_duration_seconds',
  help: 'Time spent processing Kafka messages',
  labelNames: ['topic'],
  buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1],
});

groups:
  - name: kafka
    rules:
      - alert: KafkaConsumerLagHigh
        expr: kafka_consumer_lag > 1000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: 'Kafka consumer lag is high'
          description: 'Consumer lag is {{ $value }} for {{ $labels.topic }}'

      - alert: KafkaMessageProcessingSlow
        expr: histogram_quantile(0.95, kafka_message_processing_duration_seconds) > 1
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: 'Kafka message processing is slow'
          description: 'P95 processing time is {{ $value }}s'

      - alert: KafkaBrokerDown
        expr: up{job="kafka"} == 0
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: 'Kafka broker is down'

# Проверка consumer lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group ledger-service --describe

# Проверка позиций консьюмеров
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group ledger-service --members

# Проверка топиков
kafka-topics.sh --list --bootstrap-server localhost:9092

# Проверка сообщений
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic maniton.ledger.events.v1 --from-beginning

# Проверка консьюмеров
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --list