Event-Driven — Senior chuqurlik

Idempotent producer + idempotent consumer

Ikki tomon ham himoyalanadi. Producer tomon: har xabarga barqaror eventId beriladi (retry’da o‘zgarmaydi). Consumer tomon — Inbox pattern: qabul qiluvchi ishlangan eventIdlarni bazada saqlaydi va yon ta’sirni o‘sha tranzaksiyada bajaradi. Shunda xabar 5 marta kelsa ham, ish bir marta bajariladi:

// INBOX (idempotent receiver): yon ta'sir + "ishlandi" belgisi BITTA tranzaksiyada
async handle(data) {
  await this.prisma.$transaction(async (tx) => {
    const seen = await tx.inbox.findUnique({ where: { id: data.eventId } });
    if (seen) return;                               // takror -> hech narsa qilma
    await tx.inbox.create({ data: { id: data.eventId } });   // belgilaymiz
    await tx.notification.create({ data: { ... } }); // asl ish
  });
  // tranzaksiya commit bo'lgach ack qilamiz
}

Kafka EOS (exactly-once semantics) — chegaralari bilan

Kafka haqiqiy "exactly-once"ni faqat o‘z ichida (Kafka’dan o‘qib → ishlab → Kafka’ga yozish) transaksiyalar va idempotent producer orqali beradi. Tashqi tizimga (Postgres, email) bu kafolat cho‘zilmaydi:

// Kafka EOS (exactly-once) — FAQAT Kafka ICHIDA (read -> process -> write atomik)
const producer = kafka.producer({ idempotent: true });          // dedup id
const consumer = kafka.consumer({ groupId: 'orders' });
await producer.transaction(async (txn) => {
  await txn.send({ topic: 'order.enriched', messages: [...] }); // yozuv
  await txn.sendOffsets({ consumerGroupId: 'orders', topics });  // offset birga commit
});
// MUHIM: bu tashqi DB/email'gacha cho'zilmaydi -> tashqi yon ta'sir uchun baribir
// at-least-once + idempotent receiver (inbox) kerak.

Tartibni qanday saqlaymiz

Kafka: bir xil key doim bir partisiyaga tushadi, partisiya ichida tartib kafolatlangan. RabbitMQ: consistent-hash exchange yoki single active consumer bilan per-key tartib. Asosi — single-writer prinsipi: bitta kalitni bir vaqtda faqat bitta consumer ishlasin.

// PER-KEY ordering: bir aggregate (userId) bo'yicha tartib saqlanishi shart
this.kafka.emit('order.events', {
  key: order.userId,        // SHU key -> doim BITTA partisiya -> tartiblangan
  value: { eventId, type: 'order.created', orderId: order.id },
});
// userId=42 ning barcha eventlari bitta partisiyada, ketma-ket ishlanadi.
// Boshqa userlar boshqa partisiyalarda parallel ketadi (masshtab + tartib birga).

Core darsda polling outbox (Cron) ko‘rdik — bu ishlaydi, lekin productionda kamchiliklari bor: qo‘shimcha DB yuki, kechikish (poll oralig‘i), tartib va tozalash muammolari.

Production usuli — CDC (Change Data Capture) / Debezium

Eng yetuk yechim: ma’lumotlar bazasining tranzaksiya jurnalini (WAL) kuzatish. Debezium Postgres WAL’ni o‘qiydi va outbox jadvaliga tushgan har yozuvni avtomatik Kafka’ga uzatadi — poller yo‘q, kechikish minimal, tartib jurnaldan tabiiy keladi.

// prisma/schema.prisma — outbox va inbox jadvallari
model Outbox {
  id        String   @id @default(uuid())
  aggregate String                    // "order"
  type      String                    // "order.created"
  payload   Json
  createdAt DateTime @default(now())
  @@index([createdAt])
}
model Inbox {                          // consumer tomonda idempotentlik
  id         String   @id             // eventId
  consumedAt DateTime @default(now())
}

// PRODUCTION usuli: Debezium CDC — Postgres WAL'ni o'qib, outbox'ni Kafka'ga uzatadi
// connector config (JSON) — poller YO'Q, kechikish past, tartib WAL'dan tabiiy
{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "database.hostname": "postgres",
  "table.include.list": "public.outbox",
  "transforms": "outbox",
  "transforms.outbox.type":
    "io.debezium.transforms.outbox.EventRouter",      // outbox -> topic router
  "transforms.outbox.route.by.field": "type"          // type -> topic nomi
}

To‘g‘ri retry strategiyasi

Kechiktirilgan retry (delayed retry): xato bo‘lsa, darhol emas — kutib qayta urinish, har safar oraliqni oshirib. RabbitMQda buni TTL + DLX bilan "retry navbatlari" orqali, Kafka’da esa alohida "retry topic"lar orqali quriladi.

// RabbitMQ kechiktirilgan retry: asosiy navbat -> retry navbati (TTL) -> qaytadi
// retry_queue arguments:
{
  'x-message-ttl': 5000,                       // 5s kut
  'x-dead-letter-exchange': 'main.exchange',   // TTL tugagach asosiyga qaytar
  'x-dead-letter-routing-key': 'order.created',
}

@EventPattern('order.created')
async handle(@Payload() data, @Ctx() ctx) {
  const ch = ctx.getChannelRef(); const msg = ctx.getMessage();
  const attempts = (msg.properties.headers['x-death']?.[0]?.count) ?? 0;
  try {
    await this.work(data);
    ch.ack(msg);
  } catch (e) {
    if (attempts >= 5) {
      ch.nack(msg, false, false);   // 5 urinish bo'ldi -> DLQ ("parking lot")
    } else {
      ch.publish('retry.exchange', 'order.created', msg.content);  // retry navbatiga
      ch.ack(msg);
    }
  }
}

Parking lot (DLQ) + replay

Belgilangan urinishdan keyin ham bo‘lmasa — xabar DLQga ("parking lot") suriladi, navbat bo‘shaydi. DLQ — chetga tashlangan joy emas, operatsion vosita: u monitoring ostida bo‘lishi va undan replay (qayta o‘ynatish) imkoniyati bo‘lishi shart. Idempotentlik (1-bo‘lim) replay’ni xavfsiz qiladi.

4 ta yadro tushuncha

• Partition — topic’ning bo‘lagi; parallellik va tartib birligi. Tartib faqat partisiya ichida.
• Consumer group — har partisiyani guruh ichidagi aniq bitta consumer o‘qiydi. Demak parallellik darajasi ≤ partisiyalar soni (ortiqcha consumer bekor turadi).
• Offset — consumer qayergacha o‘qiganini bildiruvchi belgi. Ishlangach commit qilsang → at-least-once; oldin commit qilsang → at-most-once.
• Consumer lag = oxirgi offset − commit qilingan offset. Tizim sog‘lig‘ining №1 metrikasi — o‘sib borsa, consumer yetishmayapti.

// NestJS Kafka consumer — CONSUMER GROUP va parallel ishlash
const app = await NestFactory.createMicroservice<MicroserviceOptions>(AppModule, {
  transport: Transport.KAFKA,
  options: {
    client: { brokers: ['localhost:9092'] },
    consumer: {
      groupId: 'inventory-group',         // bir guruh -> partisiyalar taqsimlanadi
      sessionTimeout: 30000,
    },
    run: {
      autoCommit: false,                   // QO'LDA commit -> at-least-once nazorati
      partitionsConsumedConcurrently: 3,   // 3 partisiyani parallel
    },
  },
});

Schema Registry va moslik (compatibility) rejimlari

Katta tizimlarda sxema markazlashgan Schema Registryda (Avro/Protobuf/JSON Schema) saqlanadi va har o‘zgarish moslikka tekshiriladi:

// Avro sxema + Schema Registry (Confluent) — kontraktni markazda boshqarish
{
  "type": "record", "name": "OrderCreated",
  "fields": [
    { "name": "eventId", "type": "string" },
    { "name": "orderId", "type": "string" },
    { "name": "total",   "type": "double" },
    { "name": "currency","type": ["null","string"], "default": null }  // YANGI: optional
  ]
}
// Registry compatibility = BACKWARD -> consumerlar avval yangilanadi.

Tolerant Reader pattern

Consumer sxemaga qat’iy bog‘lanmasin: faqat kerakli maydonlarni o‘qisin, noma’lum maydonlarni e’tiborsiz qoldirsin, yo‘q maydonga default bersin. Shunda producer additive o‘zgarish kiritsa, consumer sinmaydi:

// TOLERANT READER: noma'lum maydonlarni e'tiborsiz qoldir, yo'qiga default ber
function parseOrderCreated(raw: any): OrderCreated {
  return {
    eventId: raw.eventId,
    orderId: raw.orderId,
    total: raw.total,
    currency: raw.currency ?? 'UZS',   // eski eventda yo'q -> default
    // raw'dagi boshqa/yangi maydonlar -> shunchaki e'tiborsiz (xato bermaydi)
  };
}

Distributed tracing — async hop’lar bo‘ylab

Sirq: trace kontekstini (traceparent, W3C standart) xabar header’i orqali uzatish. OpenTelemetry produce’da inject, consume’da extract qiladi — natijada butun oqim (HTTP → Orders → broker → Notifications) bitta trace bo‘lib ko‘rinadi:

// DISTRIBUTED TRACING: trace kontekstini xabar header'i orqali uzatamiz
import { propagation, context, trace } from '@opentelemetry/api';

// PRODUCE — joriy trace'ni header'ga "inject" qilamiz
const headers: Record<string, string> = {};
propagation.inject(context.active(), headers);
this.bus.emit('order.created', { value: payload, headers });

// CONSUME — header'dan "extract" qilib, span ochamiz (oqim uzilmaydi)
const parentCtx = propagation.extract(context.active(), msg.properties.headers);
const span = trace.getTracer('notifications')
  .startSpan('handle order.created', {}, parentCtx);
// endi Jaeger'da: HTTP -> Orders -> broker -> Notifications BITTA trace

Kuzatiladigan metrikalar

• Consumer lag — №1 metrika; o‘ssa, consumer yetishmayapti (alert qo‘y).
• Processing latency va error rate — har consumer bo‘yicha.
• DLQ depth — DLQ’ga xabar tushsa darhol xabardor bo‘l.
• Throughput (msg/sek) — quvvat rejalashtirish uchun.

Backpressure — consumerni cho‘ktirmaslik

Producer consumerdan tez bo‘lsa, xabarlar to‘planadi. Consumer "hammasini birdan" olib cho‘kib qolmasligi uchun prefetch (QoS) bilan oqim cheklanadi:

// BACKPRESSURE (RabbitMQ): bir consumer bir vaqtda nechta unacked xabar ushlaydi
options: {
  // ...
  prefetchCount: 10,   // 10 dan ortiq tasdiqlanmagan xabar olmaydi -> consumer cho'kmaydi
  noAck: false,
}
// Kafka muqobili: max.poll.records + kerak bo'lsa pause()/resume()

Nazariyani ishlaydigan loyihada mustahkamlash — senior bo‘lishning yagona yo‘li. Quyida lab’ning yadro qismlari.

1 · Infratuzilma

# docker-compose.yml — to'liq lab uchun
services:
  rabbitmq:
    image: rabbitmq:3-management
    ports: ["5672:5672", "15672:15672"]
  postgres:
    image: postgres:16
    environment: { POSTGRES_PASSWORD: dev, POSTGRES_DB: shop }
    ports: ["5432:5432"]

2 · Integration test (testcontainers) — idempotentlikni isbotlash

Senior darajaning belgisi: EDAni test bilan kafolatlash. Testcontainers real broker ko‘taradi, sen esa "bir event 2 marta kelsa, ish 1 marta bajariladi"ni avtomatik tekshirasan:

// Integration test — testcontainers bilan REAL RabbitMQ ko'tarib tekshirish
import { RabbitMQContainer } from '@testcontainers/rabbitmq';

it('order.created -> email yuboriladi', async () => {
  const rmq = await new RabbitMQContainer().start();
  // ... app'ni rmq.getAmqpUrl() bilan ulaymiz
  await producer.emit('order.created', { eventId: 'e1', orderId: 'o1' });
  await waitFor(() => expect(emailSpy).toHaveBeenCalledWith('o1'));
  // idempotentlik: aynan shu eventni qayta yuboramiz
  await producer.emit('order.created', { eventId: 'e1', orderId: 'o1' });
  await waitFor(() => expect(emailSpy).toHaveBeenCalledTimes(1));  // 1 marta!
  await rmq.stop();
});

3 · Senior tayyorlik cheklisti

Quyidagilarning hammasiga "ha" deb javob bera olsang — bu mavzuda senior darajadasan: