Consuming events

สิ่งที่จะสร้าง

services/notification/cmd/main.go และ services/notification/internal/notifier/notifier.go — เซอร์วิส Notification คือ kafka.Consumer ใน consumer group ของตัวเองชื่อ "notification" ที่อ่าน topic "payments" ซึ่ง Process & Publish → เติม event payment.succeeded/payment.failed ไว้แล้ว โครงเหมือนกับ saga consumer ของ Order ที่ The Saga Handler → สร้าง

ต่างกันตรงที่ Handle ของ saga อัปเดต row ใน database ส่วน Handle ของ Notification ทำสิ่งที่ยังไม่มีเซอร์วิสไหนในคอร์สนี้เคยทำ — เอาผลการชำระเงินที่เพิ่ง consume มาจาก Kafka ไป enqueue เป็น send-job บน queue notification.send ของ RabbitMQ ที่ Exchanges & Queues → declare ไว้ โดยใช้ pkg/amqp.Client.Publish แบบไม่แก้อะไรเลย

นี่คือ process เดียวในทั้งระบบที่พูดทั้งสอง broker พร้อมกัน — Kafka ขาเข้า RabbitMQ ขาออก — และบทนี้คือจุดที่สถาปัตยกรรมสอง broker ที่ Architecture → วาดไว้ได้รันจริงตั้งแต่ต้นจนจบ ส่วน worker ที่ดึง job เหล่านั้นกลับออกมาจาก notification.send แล้ว “ส่ง” ออกไปจริง ๆ คือบทถัดไป The Send Worker → บทนี้หยุดตรงวินาทีที่ job อยู่บน queue อย่างปลอดภัย

ทำไม

ประเด็นทั้งหมดของ Notification คือคำถามที่สถาปัตยกรรมตอบไว้ก่อนจะมีโค้ดสักบรรทัด: ทำไม ต้อง consume event จาก Kafka แค่เพื่อส่งต่อให้ RabbitMQ ทันที แทนที่จะส่งอีเมลตรงนั้นเลยใน Kafka handler?

เพราะ “ผลการชำระเงินเกิดขึ้นแล้ว” กับ “ส่ง notification อันนี้” เป็น message คนละชนิดกันจริง ๆ และระบบนี้มีเครื่องมือที่เหมาะกับแต่ละอันอยู่แล้ว

event payment.succeeded บน Kafka คือ fact ที่ทนทาน replay ได้ และ saga ของ Order กับ Notification กับ เซอร์วิสตัวถัดไปที่จะเพิ่มเข้ามา ต่างอ่านได้แยกกันโดยไม่มีตัวไหนรู้จักตัวอื่น ส่วนการส่งอีเมลให้ลูกค้าคือ job ที่ต้องเกิดขึ้นครั้งเดียว โดย worker ตัวที่ว่าง พร้อม ack รายตัว retry และ dead-letter queue เมื่อ email provider ล่ม

Exchanges & Queues → ตั้งชื่อการแบ่งนี้ไว้ชัดแล้ว: log ที่ replay ได้ของ Kafka ตอบคำถาม “ทุกฝ่ายที่สนใจเห็น fact นี้แล้วหรือยัง” ส่วน work queue ของ RabbitMQ ตอบ “ยังมี job ที่ต้องทำครั้งเดียวค้างอยู่ไหม” Notification คือรอยต่อที่ fact แปลงร่างเป็น job

การ enqueue แทนที่จะส่ง inline ยังทำให้ Kafka consumer เร็ว Producer & Consumer → พูดไว้แล้วว่า Consumer.Run จะ commit offset หลัง Handle return เท่านั้น ดังนั้นถ้า Handle ค้างรอ email provider ที่ช้า consumer ของ topic "payments" ทั้งตัวจะติดค้างตามไปด้วย และถ้า provider ล่ม offset ก็หยุดเดินทั้งหมด แล้ว redeliver event เดิมซ้ำทุกครั้งที่ restart

การโยนงานให้ RabbitMQ แทนทำให้ Handle return ในระดับไมโครวินาที offset commit ทันที และทุกส่วนที่ช้าและพังได้ของการส่ง notification ย้ายไปอยู่ใน process แยก (worker) ที่อ่าน queue แยก ซึ่งกลไก ack/retry/dead-letter ของ RabbitMQ ที่ Acks, Retry & Dead Letters → สร้างไว้คือเครื่องมือที่เหมาะกับงานนั้นพอดี

และ consumer group ของ Notification เป็นของตัวเอง: NewConsumer(brokers, "notification", "payments") saga ของ Order อ่าน topic "payments" เดียวกันภายใต้ group "order" The Saga Handler → ส่วน Notification อ่าน topic เดียวกันนั้นภายใต้ group "notification" กฎของ Topics, Partitions & Consumer Groups → คือสิ่งที่ทำให้ปลอดภัย: consumer ที่อยู่คนละ group จะได้อ่าน topic ทั้งหมดครบถ้วนและเป็นอิสระต่อกัน

Notification จึงเห็น event payment.* ทุกตัวที่ saga เห็น และ progress, lag หรือ downtime ของ group ฝั่งหนึ่งไม่กระทบอีกฝั่งเลย saga ยืนยันคำสั่งซื้อ ส่วน Notification ส่งอีเมลแจ้งลูกค้าเรื่องเดียวกัน — fact เดียวกันที่มีคน consume สองครั้ง โดยสองเซอร์วิสที่ไม่มีวันรู้ว่าอีกตัวมีอยู่ นี่เป็นครั้งแรกที่คอร์สนี้มี consumer group สอง ตัวรันจริงบน topic เดียวมาพิสูจน์กฎนั้น แทนที่จะพูดลอย ๆ

ข้อดีข้อเสีย

Enqueue RabbitMQ job จาก Kafka handler เทียบกับ ส่ง notification แบบ inline ใน handler เดียวกัน

• Pros: Kafka consumer ยังเร็วและ offset commit ไม่เคยรอ email/SMS provider ภายนอก ดังนั้น provider ที่ช้าหรือล่มไม่ทำให้การ consume "payments" ติดค้างหรือ trigger การ redeliver ของ Kafka ไม่จบ การส่งจริงได้การรับประกัน ack, retry, และ dead-letter ต่อ message ของ RabbitMQ แทน model “reprocess ทั้ง event” ของ Kafka ซึ่งเหมาะกับ “การส่งครั้งนี้ล้มเหลว ลองใหม่ / วางไว้ก่อน” มากกว่ามาก และ worker ที่ส่งจริง scale เป็นอิสระแบบ competing consumers Exchanges & Queues → — เพิ่ม worker ตอน provider ช้าโดยไม่แตะฝั่ง Kafka เลย
• Cons: ต้องรัน broker ตัวที่สอง และมี hop เพิ่มอีกขั้นที่ต้องตามรอย — notification ตอนนี้ไหลผ่าน Kafka → Notification → RabbitMQ → worker แทนที่จะเป็น Kafka → handler → provider และเพราะทั้งสอง hop เป็น at-least-once (Kafka redeliver ขาเข้า, RabbitMQ redeliver ขาออก) payment.succeeded ที่ซ้ำตัวเดียวกลายเป็น send-job ที่ซ้ำได้ ดังนั้น worker ต้อง idempotent — ต้นทุนที่ The Send Worker → ต้องรับมือ เหมือนกับ consumer ทุกตัวในคอร์สนี้

Consumer group "notification" เฉพาะของตัวเอง เทียบกับ ใช้ saga consumer ของ Order ให้ส่ง notification ด้วย

• Pros: decouple สมบูรณ์ — Notification กับ saga อ่าน fact เดียวกันโดยไม่มีโค้ดหรือการประสานงานร่วมกันเลย ดังนั้น Notification จะ lag, restart, หรือ deploy เป็นอิสระได้โดยไม่เคยกระทบว่าคำสั่งซื้อจะไปถึง CONFIRMED/CANCELLED หรือไม่ และการเพิ่ม Notification ไม่ทำให้ saga, Payment, หรือ Order เสียอะไรเลย — ไม่มีโค้ดในตัวไหนเปลี่ยนเพื่อให้เซอร์วิสนี้เกิดขึ้น
• Cons: สอง group อิสระต่าง deserialize event payment.* ทุกตัวจาก topic — การทำงานซ้ำคงที่เล็ก ๆ เทียบกับ consumer เดียวที่ทำทั้ง update state และ notification ในรอบเดียว ซึ่งในสเกลนี้เล็กน้อยมาก แต่คือ trade ที่ตรงไปตรงมาแลกกับการ decouple

ติดตั้ง

1. services/notification/internal/notifier/notifier.go

// Package notifier is the Notification service's core: it consumes payment
// result events from Kafka and enqueues one send-job per event onto the
// notification.send RabbitMQ work queue, instead of delivering inline. The
// worker (a separate process) drains that queue and does the actual send.
package notifier

import (
  "context"
  "encoding/json"
  "fmt"

  "github.com/avetavos/shopmicro/pkg/amqp"
  "github.com/avetavos/shopmicro/pkg/events"
)

// result is the payload Payment publishes on every payment.* event — the
// same shape as processor.Result over in the Payment service. Notification
// only needs to read it, so it declares its own local copy rather than
// importing Payment's internal package.
type result struct {
  OrderID     string `json:"order_id"`
  AmountCents int64  `json:"amount_cents"`
  Reason      string `json:"reason,omitempty"`
}

// SendJob is the body of a notification.send message: everything the worker
// needs to deliver one notification, and nothing it doesn't. The worker
// (next lesson) unmarshals exactly this shape.
type SendJob struct {
  OrderID     string `json:"order_id"`
  Outcome     string `json:"outcome"` // "succeeded" or "failed"
  AmountCents int64  `json:"amount_cents"`
  Message     string `json:"message"`
}

// Notifier turns each payment result event into a notification.send job.
type Notifier struct {
  amqp *amqp.Client
}

// New returns a Notifier that enqueues jobs through client.
func New(client *amqp.Client) *Notifier {
  return &Notifier{amqp: client}
}

// Handle implements the Consumer.Run handle signature. It ignores every
// event.Type except "payment.succeeded"/"payment.failed", builds a
// customer-facing SendJob from the payment result, and publishes it to the
// notification.send queue. It never delivers anything itself — returning
// nil commits the Kafka offset the instant the job is safely on the queue.
func (n *Notifier) Handle(ctx context.Context, e events.Event) error {
  if e.Type != "payment.succeeded" && e.Type != "payment.failed" {
    return nil
  }

  var r result
  if err := json.Unmarshal(e.Payload, &r); err != nil {
    return fmt.Errorf("notifier: unmarshal payment result: %w", err)
  }

  job := SendJob{OrderID: r.OrderID, AmountCents: r.AmountCents}
  if e.Type == "payment.succeeded" {
    job.Outcome = "succeeded"
    job.Message = fmt.Sprintf("Your order %s is confirmed — payment of %d cents went through.", r.OrderID, r.AmountCents)
  } else {
    job.Outcome = "failed"
    job.Message = fmt.Sprintf("Your order %s could not be completed: %s.", r.OrderID, r.Reason)
  }

  body, err := json.Marshal(job)
  if err != nil {
    return fmt.Errorf("notifier: marshal send-job: %w", err)
  }

  if err := n.amqp.Publish(ctx, "notification.send", body); err != nil {
    return fmt.Errorf("notifier: enqueue send-job for order %s: %w", r.OrderID, err)
  }

  return nil
}

บันทึกไฟล์นี้เป็น services/notification/internal/notifier/notifier.go มีบางจุดที่ควรพูดถึง:

• Handle มี signature func(context.Context, events.Event) error ตรงกับที่ Consumer.Run ต้องการพอดี — รูปแบบเดียวกับ Processor.Handle ของ Payment และ Handler.Handle ของ saga การต่อสายเข้า consumer.Run จึงไม่ต้องใช้ adapter เลย ตัว filter — ignore อะไรก็ตามที่ไม่ใช่ผลการชำระเงิน return nil เพื่อให้ offset ยัง commit — คือ pattern สองบรรทัดเดียวกับที่ Kafka handler ทุกตัวในคอร์สนี้เปิดหัวด้วย
• handler นี้ unmarshal e.Payload ต่างจาก saga handler ที่ไม่ทำ saga ต้องการแค่ order id กับ success flag ซึ่งอยู่บน envelope ทั้งคู่ ส่วน Notification ต้องใช้ จำนวนเงิน ไปประกอบ message ที่ส่งให้ลูกค้า และค่านั้นอยู่ใน amount_cents ของ payload เท่านั้น ส่วน Reason มีค่าเฉพาะบน payment.failed เพราะ Payment mark ไว้เป็น omitempty ซึ่งก็ตรงกับ branch ล้มเหลวที่อ่านค่านี้พอดี
• ผู้รับเป็นการทำให้ง่ายอย่างตั้งใจ event ผลการชำระเงินพก order_id กับจำนวนเงิน แต่ไม่มีอีเมลหรือเบอร์ของลูกค้า — ดังนั้น SendJob อธิบาย อะไรเกิดขึ้นกับคำสั่งซื้อไหน และปล่อยเรื่อง ติดต่อใครอย่างไร ไว้ให้ worker เซอร์วิส Notification จริง ๆ จะ resolve ข้อมูลติดต่อของลูกค้าตรงนี้ (lookup ด้วย order_id หรือ customer_id ที่พกมากับ event) แบบเดียวกับที่ Process & Publish → พูดชัดว่าเซอร์วิส Payment จริงจะต่อ card network จริง กลไกสอง broker ที่ module นี้สอนเหมือนกันทั้งสองแบบ

2. services/notification/cmd/main.go

// Command notification runs the Notification service: a Kafka consumer in
// its own "notification" group reading the "payments" topic, whose handler
// enqueues a send-job onto the notification.send RabbitMQ queue for each
// payment result. The worker that drains that queue is a separate binary
// (cmd/worker). This process speaks both brokers: Kafka in, RabbitMQ out.
package main

import (
  "context"
  "log"
  "os"
  "os/signal"
  "strings"
  "syscall"

  "github.com/avetavos/shopmicro/pkg/amqp"
  "github.com/avetavos/shopmicro/pkg/config"
  "github.com/avetavos/shopmicro/pkg/kafka"
  "github.com/avetavos/shopmicro/services/notification/internal/notifier"
)

func main() {
  ctx, cancel := context.WithCancel(context.Background())
  defer cancel()

  brokers := strings.Split(config.Get("KAFKA_BROKERS", "localhost:9092"), ",")
  rabbitURL := config.Get("RABBITMQ_URL", "amqp://shopmicro:shopmicro@localhost:5672/")

  amqpClient, err := amqp.Connect(rabbitURL)
  if err != nil {
    log.Fatalf("notification: connect to rabbitmq: %v", err)
  }
  defer amqpClient.Close()

  if err := amqpClient.DeclareTopology(); err != nil {
    log.Fatalf("notification: declare rabbitmq topology: %v", err)
  }

  n := notifier.New(amqpClient)

  consumer := kafka.NewConsumer(brokers, "notification", "payments")
  defer consumer.Close()

  go func() {
    if err := consumer.Run(ctx, n.Handle); err != nil {
      log.Printf("notification: consumer stopped: %v", err)
    }
  }()

  log.Println("notification: consumer started, group=notification topic=payments")

  stop := make(chan os.Signal, 1)
  signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
  <-stop

  log.Println("notification: shutting down")
  cancel()
}

บันทึกไฟล์นี้เป็น services/notification/cmd/main.go รูปแบบเป็น composition root เดียวกับที่ทุกเซอร์วิสในคอร์สนี้มี: สร้าง dependency เริ่ม loop ระยะยาวใน goroutine แล้ว shutdown บน SIGINT/SIGTERM สิ่งเดียวที่ใหม่คือไฟล์นี้สร้าง client สอง ตัว — kafka.Consumer กับ amqp.Client — แล้วเรียก amqpClient.DeclareTopology() ตอน startup

call นั้น idempotent Exchanges & Queues → การ declare queue notification.send พร้อม dead-letter setup ซ้ำจึงไม่ทำอะไรเลยถ้า worker หรือ run ก่อนหน้า declare ไว้แล้ว ปลอดภัยที่ทั้งเซอร์วิส Notification และ worker จะ declare topology เดียวกันแยกกัน ตัวไหนเริ่มก่อนก็ชนะไป ส่วนอีกตัวกลายเป็น no-op

consumer.Run block จนกว่า ctx จะโดน cancel จึงต้องรันเป็น goroutine ตอน shutdown cancel() จะปลด block ให้ แล้ว amqpClient.Close() กับ consumer.Close() ที่ defer ไว้จะคืน connection ของทั้งสอง broker

ตรวจสอบผล

ยก Postgres, Kafka, และ RabbitMQ ขึ้นมา — Notification ต้องการทั้งสามตัว — แล้วรัน Catalog, Order, gateway, และ Payment แบบเดียวกับที่ The Saga Handler → ทิ้งไว้:

cd deploy/compose && docker compose up -d postgres kafka rabbitmq

go run ./services/catalog/cmd

go run ./services/order/cmd

go run ./gateway/cmd

go run ./services/payment/cmd

เริ่มเซอร์วิส Notification แต่ ยัง ไม่ต้องเริ่ม worker เพื่อให้ job ที่ enqueue เข้าไปค้างอยู่บน queue ให้คุณเห็นชัด ๆ:

go run ./services/notification/cmd

notification: consumer started, group=notification topic=payments

สร้างคำสั่งซื้อขนาดเล็ก ต่ำกว่า limit $5,000 ของ Payment:

curl -s -X POST localhost:8080/v1/orders \
  -H 'Content-Type: application/json' \
  -d '{"customer_id":"cust-1","items":[{"product_id":"8f14e45f-ceea-4c9d-b2a5-0c1e3f4a9b21","quantity":2}]}'

ภายในไม่กี่วินาที — ช่วงที่ outbox relay publish order.created, Payment publish payment.succeeded แล้ว Notification consume ต่อ — จะมี send-job หนึ่งอันค้างอยู่บน notification.send ยืนยันได้ใน management UI ที่ http://localhost:15672 (login shopmicro / shopmicro) → Queues and Streams → notification.send โดย Messages Ready ต้องเป็น 1 คลิกที่ queue เปิด Get messages แล้วอ่าน body ได้เลย (ตั้ง Requeue: Yes ไว้ เพื่อไม่ให้ consume job ทิ้งไป):

{"order_id":"3a7c9e21-1e4d-4b8a-9c6e-2f8b1d5a7c90","outcome":"succeeded","amount_cents":2598,"message":"Your order 3a7c9e21-1e4d-4b8a-9c6e-2f8b1d5a7c90 is confirmed — payment of 2598 cents went through."}

job อันนี้เกิดจาก event ของ Kafka และตอนนี้รออยู่บน queue ของ RabbitMQ เท่ากับพิสูจน์ hop ข้ามสอง broker แล้ว สังเกตว่า notification.send มี message ready ค้างไว้โดยไม่มี consumer มาดึงออก ไม่มีอะไรหายระหว่างที่ worker ยัง down เพราะ durable queue เก็บ job ไว้จนกว่าจะมี worker มา ack ตรงตามการรับประกันที่ Exchanges & Queues → วางไว้ ทีนี้ลองวางคำสั่งซื้อขนาดใหญ่เพื่อดูว่า message ฝั่งล้มเหลวก็เข้า queue เหมือนกัน:

curl -s -X POST localhost:8080/v1/products \
  -H 'Content-Type: application/json' \
  -d '{"name":"Server Rack","description":"42U enterprise rack","price_cents":600000,"stock":5}'

curl -s -X POST localhost:8080/v1/orders \
  -H 'Content-Type: application/json' \
  -d '{"customer_id":"cust-1","items":[{"product_id":"7b23f5a1-4c9d-4e8a-b2a5-1e3f4a9b21c8","quantity":1}]}'

notification.send ตอนนี้แสดง 2 messages ready; body ของอันที่สองพก "outcome":"failed" และ message ที่ลงท้ายด้วย could not be completed: amount exceeds limit. — handler เดียวกัน branch บน e.Type หยุด Notification ด้วย Ctrl-C:

notification: shutting down

แล้วยืนยันว่า module ยังคง build ผ่าน:

go build ./...

ไม่มี output แปลว่าสำเร็จ

ตรวจสอบความเข้าใจ:

• saga ของ Order และ Notification ต่าง consume topic "payments" บรรทัดไหนในแต่ละเซอร์วิสที่รับประกันว่าทั้งคู่เห็น payment event ทุกตัว อย่างเป็นอิสระ และอะไรจะพังถ้าสองเซอร์วิสใช้ค่าเดียวกัน?
• ทำไม Handle ถึง enqueue RabbitMQ job แล้ว return แทนที่จะส่ง notification เอง? อะไรกันแน่ที่จะติดค้าง ถ้าส่งแบบ inline แล้ว email provider ค้างไป 30 วินาที?
• send-job ทิ้ง notification.send ไว้พร้อม message สองอันที่ ready และไม่มี consumer message พวกนั้นจะอยู่ที่ไหนถ้า notification.send ถูก declare เป็น non-durable แล้ว broker restart?
• Handle unmarshal e.Payload แต่ Handler.Handle ของ saga ไม่เคยทำ ทำไม Notification ต้องการ payload ในเมื่อ saga ไม่ต้องการ?

สรุป

Notifier.Handle ใน services/notification/internal/notifier/notifier.go เป็น Kafka handler ที่มีรูปแบบเดียวกับตัวอื่นทุกตัวในคอร์สนี้ — filter หา payment.succeeded/payment.failed, return nil สำหรับทุกอย่างที่เหลือ — แต่แทนที่จะแตะ database กลับสร้าง SendJob ที่จะส่งให้ลูกค้าแล้วเรียก amqp.Client.Publish(ctx, "notification.send", body) เท่ากับเปลี่ยน fact ของ Kafka ให้กลายเป็น job ของ RabbitMQ

services/notification/cmd/main.go คือ process เดียวในระบบนี้ที่ถือทั้ง kafka.Consumer (group "notification" ของตัวเองบน "payments") และ amqp.Client โดย declare RabbitMQ topology แบบ idempotent ตอน startup แล้ว shutdown ทั้งคู่ผ่าน context ที่ cancel ร่วมกัน

เพราะเลือก enqueue แทนการส่ง inline Kafka offset จึง commit ในระดับไมโครวินาที และทุกส่วนที่ช้าและพังได้ของการส่งถูกดันไปอยู่ใน queue แยก ยิ่งกว่านั้น เพราะ "notification" เป็น consumer group ของตัวเอง เซอร์วิสนี้จึงอ่าน payment stream เดียวกับ saga ของ Order โดยไม่มี coupling ระหว่างกันเลย นี่คือครั้งแรกที่คอร์สนี้มีสอง group รันจริงบน topic เดียวมาพิสูจน์เรื่องนั้น

curl ผ่าน gateway วาง job จริงสองอันบน notification.send มองเห็นได้และ durable ใน management UI ทั้งที่ยังไม่มี worker รันมา consume เลย บทถัดไป The Send Worker → คือ binary แยกที่ดึง queue นั้น “ส่ง” notification แต่ละอัน แล้ว ack ทิ้ง — เอากลไก ack/retry/dead-letter จาก Acks, Retry & Dead Letters → มาใช้จริง