אירוע סורסינג: מדריך יישום מעשי עם קוד

תארו לעצמכם אתר מסחר אלקטרוני המעבד 10,000 הזמנות ביום. לקוח מבטל הזמנה, מנהל משנה את הסטטוס, והמערכת מאבדת את ההיסטוריה. שחזור שרשרת הפעולות בלתי אפשרי—מסד הנתונים מחזיק רק את המצב הנוכחי. ללא Event Sourcing, ביקורת דורשת פתרונות עוקפים: לוגים, טריגרים, טבלאות נוספות. אחד מלקוחותינו בתחום הפינטק בילה חודשיים בחקירת תקרית כי לא הייתה היסטוריה. לאחר יישום התבנית, קיצצנו את זמן הביקורת ב-80%, וחסכנו ללקוח מעל 50,000 דולר בשנה. עבור פרויקט טיפוסי בגודל בינוני, יישום Event Sourcing שלנו מפחית עלויות ביקורת ב-80%, וחוסך 40,000 דולר בשנה. לצוות שלנו יש ניסיון מוכח של 5+ שנים ב-Event Sourcing וביצע יותר מ-10 פרויקטים, מה שמבטיח יישומים אמינים.

Event Sourcing היא תבנית מרכזית בארכיטקטורה מונעת אירועים שבה כל שינוי מצב נתפס כאירוע בלתי ניתן לשינוי. המצב הנוכחי נבנה מחדש על ידי הפעלה חוזרת של כל האירועים. Event Sourcing היא תבנית עיצוב המאחסנת רצף של אירועים.

ארכיטקטורת Event Sourcing: Event Store, Aggregates ו-Replay

Event Store — הבסיס

הטבלה הראשית היא append-only. אין אפשרות ל-UPDATE או DELETE:

CREATE TABLE event_store ( id BIGSERIAL PRIMARY KEY, event_id UUID UNIQUE NOT NULL, aggregate_id UUID NOT NULL, aggregate_type VARCHAR(100) NOT NULL, event_type VARCHAR(100) NOT NULL, event_version INT NOT NULL DEFAULT 1, payload JSONB NOT NULL, metadata JSONB NOT NULL DEFAULT '{}', occurred_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), sequence_nr BIGINT NOT NULL -- глобальный порядок ); CREATE INDEX idx_es_aggregate ON event_store (aggregate_id, aggregate_type, id); CREATE INDEX idx_es_sequence ON event_store (sequence_nr); 

נעילה אופטימית — בדיקת CREATE TABLE event_store ( id BIGSERIAL PRIMARY KEY, event_id UUID UNIQUE NOT NULL, aggregate_id UUID NOT NULL, aggregate_type VARCHAR(100) NOT NULL, event_type VARCHAR(100) NOT NULL, event_version INT NOT NULL DEFAULT 1, payload JSONB NOT NULL, metadata JSONB NOT NULL DEFAULT '{}', occurred_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), sequence_nr BIGINT NOT NULL -- глобальный порядок ); CREATE INDEX idx_es_aggregate ON event_store (aggregate_id, aggregate_type, id); CREATE INDEX idx_es_sequence ON event_store (sequence_nr); לפני כתיבת אירוע חדש מונעת התנגשויות כתיבה במקביל. אנו משתמשים ב-PostgreSQL Event Store לאחסון אמין וחסכוני, שהוא זול פי 10 מפתרונות ייעודיים תוך מתן ביצועים נאותים ל-95% מהפרויקטים.

Aggregates: לוגיקה עסקית על אירועים

Aggregate אירועים מכיל כללים עסקיים ומצב. הוא מחיל אירועים כדי לעבור בין מצבים.

class OrderAggregate { private state: OrderState = { status: 'new', items: [], total: 0 }; private version = 0; private uncommittedEvents: DomainEvent[] = []; static rehydrate(events: DomainEvent[]): OrderAggregate { const order = new OrderAggregate(); for (const event of events) { order.apply(event); } return order; } placeOrder(items: OrderItem[]) { if (this.state.status !== 'new') throw new Error('Order already placed'); this.raise({ eventType: 'OrderPlaced', payload: { items, placedAt: new Date() } }); } private apply(event: DomainEvent) { switch (event.eventType) { case 'OrderPlaced': this.state.status = 'placed'; this.state.items = event.payload.items; break; case 'PaymentProcessed': this.state.status = 'paid'; this.state.paidAmount = event.payload.amount; break; case 'OrderShipped': this.state.status = 'shipped'; this.state.trackingNumber = event.payload.trackingNumber; break; } this.version++; } } 

Replay ו-Snapshots

כאשר ל-aggregate יש אירועים רבים (מעל 500), replay מלא הופך לאיטי. Snapshot הוא מצב מסודר באירוע N. בעת טעינה, אנו קוראים את ה-snapshot האחרון בתוספת אירועים שאחריו.

Snapshots משפרים ביצועים

async loadAggregate(aggregateId: string): Promise<OrderAggregate> { const snapshot = await this.snapshotRepo.findLatest(aggregateId); const fromSequence = snapshot?.version ?? 0; const events = await this.eventStore.getEvents( aggregateId, { fromVersion: fromSequence } ); const aggregate = snapshot ? OrderAggregate.fromSnapshot(snapshot) : new OrderAggregate(); return aggregate.rehydrate(events); } 

שחזור מצב מושג על ידי הפעלה חוזרת של אירועים מה-Event Store. Snapshots נוצרים באופן אסינכרוני כל 100–500 אירועים לכל aggregate. נעילה אופטימית במהלך כתיבת אירוע בודקת sequence_nr — אם השתנה, הכתיבה נדחית, מה שמבטיח שלמות.

מורכבות זמן של פעולות:

כיצד לטפל ב-projections ובאבולוציית סכמה

Projections (Read Models) לקריאות מהירות

תבנית זו מחייבת הפרדה בין Write Model (אירועים) ל-Read Model (projections לשאילתות). זהו יישום טיפוסי של CQRS. Event projections נרשמים לזרם האירועים ובונים טבלאות denormalized:

class OrderProjection { async on(event: DomainEvent) { switch (event.eventType) { case 'OrderPlaced': await db.query(` INSERT INTO orders_view (id, status, customer_id, total, created_at) VALUES ($1, 'placed', $2, $3, $4) `, [event.aggregateId, event.payload.customerId, event.payload.total, event.occurredAt]); break; case 'OrderShipped': await db.query(` UPDATE orders_view SET status = 'shipped', tracking_number = $2, shipped_at = $3 WHERE id = $1 `, [event.aggregateId, event.payload.trackingNumber, event.occurredAt]); break; } } } 

ניתן למחוק projections ולבנות אותם מחדש מאפס—היסטוריית האירועים מלאה.

אבולוציית סכמה: מניעת שבירת היסטוריה

גרסת סכמות אירועים היא חובה. אסטרטגיות:

• Upcasting — הפיכת אירועים ישנים לסכמה חדשה בעת קריאה.
• סכמה גמישה — JSON מאפשר הוספת שדות ללא שבירה.
• גרסת אירועים — אחסון class OrderAggregate { private state: OrderState = { status: 'new', items: [], total: 0 }; private version = 0; private uncommittedEvents: DomainEvent[] = []; static rehydrate(events: DomainEvent[]): OrderAggregate { const order = new OrderAggregate(); for (const event of events) { order.apply(event); } return order; } placeOrder(items: OrderItem[]) { if (this.state.status !== 'new') throw new Error('Order already placed'); this.raise({ eventType: 'OrderPlaced', payload: { items, placedAt: new Date() } }); } private apply(event: DomainEvent) { switch (event.eventType) { case 'OrderPlaced': this.state.status = 'placed'; this.state.items = event.payload.items; break; case 'PaymentProcessed': this.state.status = 'paid'; this.state.paidAmount = event.payload.amount; break; case 'OrderShipped': this.state.status = 'shipped'; this.state.trackingNumber = event.payload.trackingNumber; break; } this.version++; } } , קריאה עם handlers שונים.

כלים ל-Event Store

פתרונות מוכנים:

• EventStoreDB — מסד נתונים ייעודי עם subscriptions ו-projections.
• Marten (.NET) — PostgreSQL כ-Event Store + מסד נתונים מסמכים.
• Axon Framework (Java) — מסגרת ES/CQRS מלאה.

Event Store מותאם אישית על PostgreSQL מספיק לרוב הפרויקטים. השתמשו ב-LISTEN/NOTIFY כדי להודיע ל-projections על אירועים חדשים. השתמשו ב-message broker (Kafka, RabbitMQ, NATS JetStream) להפצה בין שירותים.

השוואת יישומי Event Store:

תוכנית יישום ולוח זמנים

1. ניתוח תחום וזיהוי aggregates (1–2 שבועות).
2. הגדרת סוגי אירועים וסכמות (3–5 ימים).
3. יישום Event Store (PostgreSQL או פתרון מוכן) — 2–3 ימים.
4. כתיבת aggregates עם לוגיקה עסקית (3–5 ימים לכל aggregate).
5. בניית projections למודלי קריאה (5–7 ימים).
6. הגדרת snapshots לביצועים (1–2 ימים).
7. אינטגרציה עם message broker (Kafka, RabbitMQ) — 3–5 ימים.
8. בדיקות וניטור (1–2 שבועות).

לוח זמנים: Event Store בסיסי על PostgreSQL — 2–3 ימים. Aggregate אחד — 3–5 ימים. Projections + subscriptions + snapshots — עוד 5–7 ימים. מערכת מלאה עם מספר aggregates, אבולוציית סכמה וניטור — 3–5 שבועות.

מה כלול בעבודה

תוצרים

• תיעוד על אירועים וסכמות.
• קוד עבור aggregates, projections ו-snapshots.
• הגדרת Event Store (PostgreSQL או EventStoreDB).
• אינטגרציה עם message broker.
• גישה ל-Event Store וללוחות ניטור.
• הדרכת צוות על Event Sourcing.
• תמיכה לאחר היישום (חודש אחד).

יתרונות Event Sourcing

• מסלול ביקורת מלא לכל שינוי.
• יכולת לחזור לכל מצב.
• הפרדה בין מודלי כתיבה וקריאה (CQRS).
• הוספה קלה של projections חדשים ללא מיגרציות.

הארכיטקטורה מונעת האירועים עם PostgreSQL Event Store משתמשת ב-event aggregates, event projections, CQRS, snapshots, גרסת סכמת אירועים ושחזור מצב למסלול ביקורת מלא.

קבלו ייעוץ — ננתח את התחום שלכם ונציע את הפתרון האופטימלי. צרו קשר כדי להעריך את הפרויקט שלכם. נעזור ליישם Event Sourcing המותאם לדרישותיכם.