Event Sourcing — Senior chuqurlik

Uchta strategiya

• Weak/tolerant schema: yangi maydonlarni ixtiyoriy (default bilan) qo‘shish; o‘quvchi yo‘q maydonni e’tiborsiz qoldiradi. Eng oddiy, lekin faqat qo‘shish uchun.
• Versiyalangan eventlar: OrderCreated.v1, OrderCreated.v2 — yangi shakl yangi versiya sifatida.
• Upcasting: o‘qishda eski eventni yangi shaklga aylantirish (saqlangan event tegilmaydi).

// Eski event (v1) va yangi (v2): manzil oddiy string edi -> obyektga aylandi
type OrderCreatedV1 = { id: string; address: string };
type OrderCreatedV2 = { id: string; address: { city: string; street: string } };

// UPCASTER: O'QISHDA eski eventni yangi shaklga aylantiradi
// (saqlangan event O'ZGARMAYDI — faqat o'qiyotganda transformatsiya)
function upcast(e: StoredEvent): DomainEvent {
  if (e.type === "OrderCreated.v1") {
    return {
      type: "OrderCreated.v2",
      data: { id: e.data.id, address: parseAddress(e.data.address) },  // string -> obyekt
    };
  }
  return e;   // allaqachon yangi versiya
}

Yechim — expectedVersion bilan append

Aggregat yuklanganda uning joriy versiyasini bilamiz. Append qilganda "men 5-versiyani ko‘rgandim, 6 ga yozmoqchiman" deymiz. Agar oraliqda kimdir 6 ni yozib qo‘ygan bo‘lsa — yozuv rad etiladi (reload + retry). Postgres’da buni PRIMARY KEY (aggregate_id, version) tabiiy beradi:

// Postgres event store — optimistik concurrency PRIMARY KEY orqali
// CREATE TABLE events (aggregate_id UUID, version INT, type TEXT, data JSONB,
//   PRIMARY KEY (aggregate_id, version));   <- bir version ikki marta yozilmaydi

async append(aggregateId: string, expectedVersion: number, events: DomainEvent[]) {
  try {
    await this.prisma.$transaction(
      events.map((e, i) =>
        this.prisma.event.create({
          data: {
            aggregateId,
            version: expectedVersion + i + 1,   // 6, 7, ... (kutilgan + 1 dan)
            type: e.type,
            data: e as any,
          },
        }),
      ),
    );
  } catch (err) {
    if (isUniqueViolation(err))                 // (aggregate_id, version) band -> kimdir yozdi
      throw new ConcurrencyException("Concurrent modification — reload & retry");
    throw err;
  }
}

Strategiyalar

• Qachon: har N eventda (masalan, 100), yoki vaqt bo‘yicha, yoki yuklashda "juda ko‘p event qoldi" deb aniqlanganda.
• Qayerda: alohida snapshot ombori (event store’dan ajratilgan) — eventlarni bulg‘amaydi.
• Versiyalash: snapshot — bu aggregat holatining seriyalashtirilgan shakli; holat sxemasi o‘zgarsa, eski snapshot o‘qib bo‘lmaydi.

// SNAPSHOT siyosati — bu OPTIMIZATSIYA, hech qachon haqiqat manbai emas
async save(order: Order) {
  await this.appendEvents(order);
  if (order.version % 100 === 0) {                // siyosat: har 100 eventda
    await this.snapshots.save(order.id, {
      version: order.version,
      state: order.toSnapshot(),
      schemaVersion: 3,                           // snapshot SXEMASI ham versiyalanadi
    });
  }
}

async load(id: string): Promise<Order> {
  const snap = await this.snapshots.getLatest(id);
  // snapshot sxemasi eskirgan/mos kelmasa -> tashlab, eventlardan TO'LIQ qur
  const usable = snap && snap.schemaVersion === 3;
  const order = usable ? Order.fromSnapshot(snap) : new Order();
  const from = usable ? snap.version : 0;
  order.loadFromHistory(await this.eventStore.getEvents(id, { after: from }));
  return order;
}

Yechim — crypto-shredding

PII (shaxsiy ma’lumot)ni eventda shifrlangan holda saqla, kalitni esa alohida (har foydalanuvchi uchun). "Unutish" — eventni emas, kalitni o‘chirish: shifrlangan ma’lumot endi tiklanmaydi (amalda o‘chirilgan), lekin event jurnalining tuzilishi buzilmaydi:

// CRYPTO-SHREDDING (GDPR): append-only event'ni o'chirib bo'lmaydi.
// Yechim: PII'ni har foydalanuvchi uchun alohida kalit bilan SHIFRLAYMIZ.
function storeEvent(e: any, userKey: Buffer) {
  return { ...e, email: encrypt(e.email, userKey) };   // PII shifrlangan saqlanadi
}

// "Unutilish huquqi" (right to be forgotten):
await keyStore.delete(userId);   // foydalanuvchi kalitini O'CHIRAMIZ
// -> eventlar joyida qoladi (append-only buzilmaydi),
//    lekin shifrlangan PII endi HECH QACHON o'qilmaydi = amalda o'chirilgan

Process Manager nima

U eventlarni tinglaydi, o‘z holatini yuritadi (jarayon qaysi bosqichda) va navbatdagi commandni chiqaradi. Ya’ni: event kiradi → holat yangilanadi → command chiqadi.

// PROCESS MANAGER: eventlarga reaksiya qilib, O'Z holatini yuritadi va COMMAND chiqaradi
@EventsHandler(OrderPlacedEvent)
export class OrderProcessManager {
  constructor(private commands: CommandBus) {}

  async handle(e: OrderPlacedEvent) {
    // 1-qadam: inventarni rezerv qil
    await this.commands.execute(new ReserveInventoryCommand(e.orderId, e.items));
  }
}

@EventsHandler(InventoryReservedEvent)
export class OnInventoryReserved {
  async handle(e: InventoryReservedEvent) {
    // 2-qadam: to'lovni amalga oshir
    await this.commands.execute(new ChargePaymentCommand(e.orderId));
  }
}
// muvaffaqiyatsizlikda -> KOMPENSATSIYA (masalan, ReleaseInventory) -> 13-mavzu Saga

Masshtabda projection — catch-up subscription

Ko‘p eventli tizimda projectionlar eventlarni catch-up subscription orqali ketma-ket o‘qiydi: checkpoint’dan boshlab, yangilarini olib, read modelni yangilaydi (9-mavzu Deep: idempotency + checkpoint). Bu projectionlarni event oqimi bilan moslashtiradi va rebuild’ni mumkin qiladi.

To‘liq Event Sourcing arxitekturasi — yig‘ma ko‘rinish

Senior tayyorlik cheklisti