Event Sourcing

บริบท

service ของคุณพึ่ง event เป็นหลัก saga ร้อยขั้นตอนเข้าด้วยกันด้วย event และ CQRS projection ก็สร้างขึ้นจาก event แต่มีช่องโหว่ด้านความน่าเชื่อถือซ่อนอยู่ ปกติ service จะอัปเดต database ก่อนแล้วค่อย publish event แยกเป็นสอง step ถ้า commit การเปลี่ยนแปลงลง database สำเร็จแล้ว crash ก่อนจะ publish ระบบส่วนที่เหลือก็จะไม่มีวันรู้เรื่องนี้เลย state กับ event stream แยกจากกันแบบเงียบ ๆ สิ่งที่คุณต้องการคือวิธีอัปเดต state และ publish event ที่ผูกกันแน่น จนอย่างหนึ่งเกิดขึ้นโดยไม่มีอีกอย่างไม่ได้

ปัญหา

ปัญหา dual-write: การเขียนลงฐานข้อมูลและการ publish ไปยัง message broker เป็นสองการดำเนินงานที่คุณไม่สามารถห่อไว้ใน transaction เดียวได้ ไม่ว่าคุณจะทำอันไหนก่อน การ crash ระหว่างกลางก็ทิ้งให้ที่จัดเก็บทั้งสองไม่สอดคล้องกัน ยิ่งไปกว่านั้น การจัดเก็บ state แบบธรรมดาทิ้งประวัติศาสตร์ไป — เมื่อ row ถูกเขียนทับแล้ว คุณก็ไม่สามารถตอบได้อีกว่า “ยอดคงเหลือเมื่อวันอังคารที่แล้วเป็นเท่าไร?” หรือ “เรามาถึงสถานะนี้ได้อย่างไร?”

คุณจะจัดเก็บ state อย่างไรให้ event เป็นแหล่งความจริง (source of truth) การ publish ถูกรับประกัน และประวัติทั้งหมดถูกเก็บรักษาไว้?

วิธีแก้

ด้วย Event Sourcing คุณไม่เก็บ state ปัจจุบันเลย แต่เก็บลำดับของ domain event ที่สร้าง state นั้นขึ้นมา ไว้ใน event store แบบ append-only บัญชีหนึ่งจึงไม่ใช่ row ที่มีคอลัมน์ balance แต่เป็นรายการที่เรียงลำดับว่า AccountOpened, Deposited, Withdrawn, Deposited เวลาจะเอา state ปัจจุบัน ก็โหลด event ของ entity นั้นมาแล้ว fold หรือเล่นซ้ำให้กลายเป็น state ใน memory

เพราะ event คือหน่วยที่คุณ persist จึงไม่มี dual write: การ append event คือ การเปลี่ยน state นั่นเอง event store เดียวกันนี้ยังทำหน้าที่เป็น outbox ไปด้วย — service อื่นมา subscribe stream และได้รับทุก event ตามลำดับ โดยไม่มี step การ publish แยกต่างหากที่จะหายไป

flowchart LR
  Cmd[Command] --> Agg[Aggregate]
  Agg -->|append events| Store[(Event Store - append only)]
  Store -->|replay / fold| State[Current State]
  Store -->|subscribe| C1[CQRS Projection]
  Store -->|subscribe| C2[Other Service]
  Store -->|subscribe| C3[Audit / Analytics]

command append event; state ถูก fold จาก log และ log เดียวกันกระจายออกไปยังผู้ subscribe

ตัวอย่าง

ตัวอย่างนี้จำลองบัญชีธนาคาร command เป็นตัวผลิต event ส่วน state ปัจจุบันคำนวณจากการ fold รายการ event สังเกตว่าไม่มี setter ของ balance เลย ค่าจะขยับได้ด้วยการ apply event เท่านั้น

TypeScript

type Event =
  | { type: 'AccountOpened'; owner: string }
  | { type: 'Deposited'; amount: number }
  | { type: 'Withdrawn'; amount: number };

type Account = { owner: string; balance: number };

// Fold the log into current state.
function apply(state: Account, e: Event): Account {
  switch (e.type) {
    case 'AccountOpened': return { owner: e.owner, balance: 0 };
    case 'Deposited':     return { ...state, balance: state.balance + e.amount };
    case 'Withdrawn':     return { ...state, balance: state.balance - e.amount };
  }
}

const replay = (events: Event[]): Account =>
  events.reduce(apply, { owner: '', balance: 0 });

// A command validates against current state, then appends a new event.
async function withdraw(id: string, amount: number) {
  const state = replay(await store.load(id));
  if (state.balance < amount) throw new Error('insufficient funds');
  await store.append(id, { type: 'Withdrawn', amount }); // append = state change + publish
}

Python

from functools import reduce

def apply(state: dict, e: dict) -> dict:
    if e["type"] == "AccountOpened":
        return {"owner": e["owner"], "balance": 0}
    if e["type"] == "Deposited":
        return {**state, "balance": state["balance"] + e["amount"]}
    if e["type"] == "Withdrawn":
        return {**state, "balance": state["balance"] - e["amount"]}
    return state

def replay(events: list[dict]) -> dict:
    return reduce(apply, events, {"owner": "", "balance": 0})

# A command validates against current state, then appends a new event.
async def withdraw(account_id: str, amount: int) -> None:
    state = replay(await store.load(account_id))
    if state["balance"] < amount:
        raise ValueError("insufficient funds")
    await store.append(account_id, {"type": "Withdrawn", "amount": amount})  # append = state change + publish

Go

type Event struct {
    Type   string // "AccountOpened" | "Deposited" | "Withdrawn"
    Owner  string
    Amount int
}

type Account struct {
    Owner   string
    Balance int
}

// Fold the log into current state.
func Apply(s Account, e Event) Account {
    switch e.Type {
    case "AccountOpened":
        return Account{Owner: e.Owner, Balance: 0}
    case "Deposited":
        s.Balance += e.Amount
    case "Withdrawn":
        s.Balance -= e.Amount
    }
    return s
}

func Replay(events []Event) Account {
    var s Account
    for _, e := range events {
        s = Apply(s, e)
    }
    return s
}

// A command validates against current state, then appends a new event.
func Withdraw(ctx context.Context, id string, amount int) error {
    events, err := store.Load(ctx, id)
    if err != nil {
        return err
    }
    if Replay(events).Balance < amount {
        return errors.New("insufficient funds")
    }
    return store.Append(ctx, id, Event{Type: "Withdrawn", Amount: amount}) // append = state change + publish
}

Rust

enum Event {
    AccountOpened { owner: String },
    Deposited { amount: i64 },
    Withdrawn { amount: i64 },
}

#[derive(Default)]
struct Account {
    owner: String,
    balance: i64,
}

// Fold the log into current state.
fn apply(mut state: Account, e: &Event) -> Account {
    match e {
        Event::AccountOpened { owner } => Account { owner: owner.clone(), balance: 0 },
        Event::Deposited { amount } => { state.balance += amount; state }
        Event::Withdrawn { amount } => { state.balance -= amount; state }
    }
}

fn replay(events: &[Event]) -> Account {
    events.iter().fold(Account::default(), apply)
}

// A command validates against current state, then appends a new event.
async fn withdraw(id: &str, amount: i64) -> Result<(), Error> {
    let state = replay(&store::load(id).await?);
    if state.balance < amount {
        return Err(Error::InsufficientFunds);
    }
    store::append(id, Event::Withdrawn { amount }).await // append = state change + publish
}

ผลลัพธ์ที่ตามมา

สิ่งที่คุณได้รับ:

• การ publish event ที่น่าเชื่อถือ. การ append event คือการเขียน ดังนั้นจึงไม่มีช่องโหว่ dual-write — ผู้ subscribe ได้รับ event ที่เปลี่ยน state อย่างตรงเป๊ะ ตามลำดับ ทำให้ Event Sourcing เป็นแกนหลักตามธรรมชาติสำหรับ saga และ CQRS
• ได้ audit log ที่สมบูรณ์ ทุกการเปลี่ยนแปลงถูกเก็บไว้ แก้ไม่ได้ และอยู่ตลอดไป คุณจึงตอบได้ไม่ใช่แค่ว่า state เป็นอะไร แต่ตอบได้ด้วยว่ามาถึงจุดนั้นได้อย่างไรและเมื่อไร
• temporal query และการ replay. คุณสามารถสร้าง state ขึ้นใหม่ ณ ช่วงเวลาใดในอดีตก็ได้ และสร้าง read model ใหม่ได้ง่าย ๆ เพียงแค่เล่นประวัติซ้ำผ่าน projection ใหม่

สิ่งที่คุณต้องจ่าย:

• การ query เป็นเรื่องยาก. event store เก่งเรื่อง “ขอ event ของ entity หนึ่งให้หน่อย” แต่อ่อนเรื่อง “หาทุกบัญชีที่มียอดคงเหลือติดลบ” ดังนั้น Event Sourcing จึงเกือบจะจับคู่กับ CQRS เสมอ ซึ่งสร้าง read model ที่ query ได้จาก stream
• ต้นทุนการ replay. การ fold ประวัติ event ที่ยาวทุกครั้งที่โหลดมีราคาแพง ดังนั้นคุณจึงบันทึก snapshot เป็นระยะ และเล่นซ้ำเฉพาะ event หลังจากนั้น
• schema evolution และ eventual consistency event แก้ไม่ได้ การเปลี่ยนรูปร่าง event ตามเวลาจึงต้องทำ versioning และ upcast event เก่า ส่วนทุกอย่างที่สร้างจาก stream ไม่ว่าจะเป็น read model หรือ service อื่น ก็เป็น eventually consistent ทั้งหมด

เนื้อหาที่เกี่ยวข้อง

• CQRS — คู่หูมาตรฐานที่ทำให้ระบบ event-sourced query ได้
• Saga — saga พึ่งพา event ที่น่าเชื่อถือซึ่ง event store ส่งออกมา
• Database per Service — แต่ละ service เลือกใช้ Event Sourcing เป็นโมเดลจัดเก็บส่วนตัวของตัวเองได้

ข้อดี	ข้อแลกเปลี่ยน
audit log ครบสมบูรณ์ — รู้ว่า state เปลี่ยนยังไงและเมื่อไร	query current state ต้อง replay event — ช้ากว่า read current value
rebuild state ได้ทุกเวลา	event schema เปลี่ยนยาก — event เก่าต้อง compatible
เหมาะมากกับ CQRS	event store ขนาดใหญ่ต้องใช้ snapshotting
time-travel debugging — ดู state ณ เวลาใดก็ได้	conceptual overhead สูง ทีมต้องเปลี่ยน mental model

ข้อผิดพลาดที่พบบ่อย

Event Sourcing ทุก Service — ใช้ Event Sourcing โดยไม่มี audit หรือ time-travel requirement อาการ:

• service ง่าย ๆ ที่ไม่ต้องการ history ก็ใช้ event store
• complexity สูงโดยไม่ได้ประโยชน์
• ทีมใช้เวลา manage event schema มากกว่า build feature

Mutable Events — แก้ไข event ที่เกิดขึ้นแล้ว อาการ:

• แก้ event เพื่อ fix bug แทนที่จะสร้าง compensating event ใหม่
• audit log ไม่น่าเชื่อถืออีกต่อไป

💡 ตัวอย่างจากของจริง

Eventbrite:

• ใช้ Event Sourcing สำหรับ ticketing system
• ทุก ticket transaction เก็บเป็น event — ทำให้ audit และ refund ง่าย

Financial Systems (Goldman Sachs, etc.):

• regulatory requirement บังคับให้ต้องมี complete audit trail
• Event Sourcing ตอบโจทย์นี้โดยธรรมชาติ

🧩 event เป็นข้อเท็จจริงที่ไม่เปลี่ยนแปลง

event ที่เก็บไว้คือบันทึกของสิ่งที่เกิดขึ้นไปแล้ว เขียนในรูปอดีตกาลและไม่มีวันเปลี่ยน เวลาจะแก้ข้อผิดพลาด คุณ append event ชดเชยเข้าไปใหม่เพื่อกลับรายการ ไม่ใช่ไปแก้หรือลบ event เก่า การยึดมั่นว่า log เป็น append-only เท่านั้นคือสิ่งที่ทำให้ Event Sourcing มี audit trail และ publish ได้อย่างน่าเชื่อถือตั้งแต่แรก

ระบบที่เป็น event-sourced จัดเก็บอะไรจริง ๆ?

state ปัจจุบันเป็น row เดียวที่เปลี่ยนแปลงได้ต่อ entity ลำดับของ domain event ที่เรียงลำดับและเป็นแบบ append-only เฉพาะ event ล่าสุดของแต่ละ entity snapshot ของทุกตารางทุกคืน

state ปัจจุบันของ entity ได้มาอย่างไร?

โดยการอ่านคอลัมน์ balance ที่ cache ไว้ ด้วยการ fold หรือเล่น event ซ้ำตามลำดับ ซึ่งจะเริ่มจาก snapshot ก็ได้ โดยการ query ฐานข้อมูลฝั่งเขียนแยกต่างหาก ด้วยการไปถามจาก service อื่น

ทำไม Event Sourcing จึงแก้ปัญหา dual-write ได้?

ใช้ two-phase commit ข้าม database และ broker การ append event คือการเปลี่ยน state ดังนั้น state และ event stream จึงไม่มีวันแยกออกจากกันได้ ปิด message broker ทิ้งทั้งหมด เขียนลง broker ก่อน แล้วไม่เขียนลง database เลย

Event Sourcing เกือบจะจับคู่กับ pattern ใดเสมอ และเพราะอะไร?

API Composition เพื่อหลีกเลี่ยงการจัดเก็บ event Saga เพื่อแทนที่ compensation CQRS เพราะ event store อ่อนเรื่อง query เฉพาะกิจและต้องการ read model Database per Service เพื่อใช้ schema เดียวร่วมกัน

Submit

Mark as complete