11.2K 144 987

Đã đăng vào thg 7 29, 3:13 CH 23 phút đọc

#1 Kiến trúc Event-Driven — Backend Golang

Tài liệu hướng dẫn cấu trúc source code cho nền tảng cộng đồng (kiểu DEV.to) mô tả trong note.md.

Hạng mục	Lựa chọn
Hình dạng hệ thống	Modular Monolith + Event Bus
Message broker	Kafka, kèm Outbox Pattern
Cơ sở dữ liệu	PostgreSQL
HTTP router	`chi`
Truy cập DB	`sqlc` + `pgx`
Module lõi	`identity`, `post`, `stats`

1. Vì sao Event-Driven cho dự án này

Nhìn vào bảng user_stats trong note.md:

total_post_views, following_tags_count, following_users_count,
followers_count, posts_count, bookmarks_count,
questions_count, answers_count

Đây là 8 con số denormalized — chúng là bản sao đã tính sẵn của dữ liệu nằm ở 8 nơi khác nhau. Mỗi con số đều cần được cập nhật khi có chuyện xảy ra ở một domain khác.

Cách làm thông thường (gọi hàm trực tiếp) sẽ dẫn tới thế này:

// ❌ Cách này sẽ giết chết codebase của bạn sau 6 tháng
func (s *PostService) Create(ctx context.Context, in CreatePostInput) error {
    post := s.repo.Insert(ctx, in)
    s.statsService.IncrementPostsCount(ctx, in.AuthorID)   // post → stats
    s.searchService.Index(ctx, post)                        // post → search
    s.notifyService.NotifyFollowers(ctx, post)              // post → notification
    s.feedService.FanOut(ctx, post)                         // post → feed
    return nil
}

Bốn vấn đề, tất cả đều nghiêm trọng:

post phải biết về sự tồn tại của 4 module khác. Thêm tính năng mới = sửa lại PostService.
Một dependency chết là chết hết. Elasticsearch sập → không đăng bài được, dù việc đăng bài chẳng liên quan gì tới search.
Người dùng phải chờ toàn bộ chuỗi. API trả về sau khi 4 việc phụ đã xong.
Không thể test post một mình. Muốn test phải mock cả 4 service.

Với Event-Driven, post chỉ làm đúng một việc và thông báo ra ngoài:

// ✅ post không biết ai đang nghe, và không cần biết
func (s *PostService) Create(ctx context.Context, in CreatePostInput) error {
    return s.tx.Do(ctx, func(tx pgx.Tx) error {
        post, err := s.repo.Insert(ctx, tx, in)
        if err != nil {
            return err
        }
        return s.outbox.Publish(ctx, tx, v1.PostCreated{
            PostID:   post.ID,
            AuthorID: post.AuthorID,
            Slug:     post.Slug,
        })
    })
}

Thêm tính năng mới = thêm một consumer mới, không đụng vào file này.

Đây không phải bữa trưa miễn phí. Bạn đánh đổi tính nhất quán tức thời lấy tính độc lập giữa các module. Sau khi API trả 201, posts_count có thể vẫn chưa tăng trong vài chục mili-giây. Hãy đọc kỹ mục 9.3 để biết chỗ nào chấp nhận được và chỗ nào không.

2. Ba luật bất di bất dịch

Mọi quy ước còn lại trong tài liệu này chỉ là hệ quả của ba luật sau.

Luật 1 — Module không bao giờ import module khác

// ❌ TUYỆT ĐỐI KHÔNG
import "github.com/you/project/internal/modules/stats"

// ✅ Được phép
import "github.com/you/project/internal/contracts/v1"      // hợp đồng chung
import "github.com/you/project/internal/platform/eventbus" // hạ tầng

Đồ thị phụ thuộc phải luôn là hình sao, không bao giờ có vòng lặp:

    identity ─┐
              ├──→ contracts ←── platform
        post ─┤
              │
       stats ─┘

Bạn có thể ép luật này bằng công cụ, xem mục 13.4.

Luật 2 — Event là hợp đồng, không phải lời gọi hàm

Publisher không biết ai đang nghe, không chờ kết quả, không quan tâm consumer thành công hay thất bại. Nếu bạn thấy mình muốn viết "phát event rồi chờ kết quả trả về", thì thứ bạn cần là một lời gọi hàm đồng bộ chứ không phải event — và nếu nó bắt buộc phải là lời gọi hàm đồng bộ thì hai module đó có lẽ nên là một.

Luật 3 — Ghi DB và phát event phải nằm trong cùng một transaction

// ❌ Sai kinh điển: hai hệ thống, không có transaction chung
tx.Commit()                          // DB xong
kafka.Publish(ctx, event)            // ← app crash ở ĐÚNG dòng này thì sao?

Nếu process chết giữa hai dòng: bài viết đã nằm trong DB, nhưng event không bao giờ được phát. posts_count sai vĩnh viễn, và không có gì trong log cho bạn biết điều đó. Lỗi này âm thầm và gần như không thể truy vết ngược.

Giải pháp là Outbox Pattern (mục 7): ghi event vào một bảng trong DB, cùng transaction với dữ liệu nghiệp vụ. Một tiến trình riêng đọc bảng đó và đẩy sang Kafka.

3. Cây thư mục đầy đủ

project/
├── cmd/                              # Ba binary, ba vòng đời khác nhau
│   ├── api/
│   │   └── main.go                   #   HTTP server — chỉ ghi DB + outbox
│   ├── worker/
│   │   └── main.go                   #   Kafka consumer — xử lý event
│   └── relay/
│       └── main.go                   #   Outbox relay — DB → Kafka
│
├── internal/
│   │
│   ├── contracts/                    # ★ HỢP ĐỒNG EVENT
│   │   ├── envelope.go               #   struct bao ngoài mọi event
│   │   ├── topics.go                 #   hằng số tên topic + event type
│   │   └── v1/
│   │       ├── user_events.go        #   UserRegistered, UserFollowed...
│   │       ├── post_events.go        #   PostCreated, PostViewed, PostUpvoted...
│   │       └── doc.go                #   quy tắc versioning
│   │
│   ├── modules/                      # ★ NGHIỆP VỤ
│   │   │
│   │   ├── identity/                 #   Đăng ký, đăng nhập, profile
│   │   │   ├── module.go             #     điểm lắp ráp duy nhất
│   │   │   ├── domain/
│   │   │   │   ├── user.go           #     entity thuần Go, không tag DB/JSON
│   │   │   │   └── errors.go         #     ErrEmailTaken, ErrInvalidCredentials...
│   │   │   ├── service/
│   │   │   │   ├── register.go
│   │   │   │   ├── login.go
│   │   │   │   └── service_test.go
│   │   │   ├── repository/
│   │   │   │   ├── repository.go     #     interface do service định nghĩa
│   │   │   │   ├── query.sql         #     SQL thuần cho sqlc
│   │   │   │   ├── postgres.go       #     impl, bọc quanh gen/
│   │   │   │   └── gen/              #     ⚠ sqlc sinh ra — KHÔNG sửa tay
│   │   │   ├── transport/
│   │   │   │   └── http/
│   │   │   │       ├── handler.go
│   │   │   │       ├── router.go
│   │   │   │       └── dto.go        #     request/response, tách khỏi domain
│   │   │   └── subscriber/           #     module này NGHE event gì
│   │   │       └── on_user_deleted.go
│   │   │
│   │   ├── post/                     #   Bài viết  (cấu trúc y hệt identity)
│   │   │   ├── module.go
│   │   │   ├── domain/
│   │   │   ├── service/
│   │   │   ├── repository/
│   │   │   ├── transport/http/
│   │   │   └── subscriber/
│   │   │
│   │   └── stats/                    #   ★ Module gần như CHỈ có consumer
│   │       ├── module.go
│   │       ├── repository/
│   │       │   ├── query.sql         #     toàn UPDATE ... = ... + 1
│   │       │   └── gen/
│   │       ├── subscriber/           #     ★ trái tim của module này
│   │       │   ├── on_post_created.go
│   │       │   ├── on_post_viewed.go
│   │       │   ├── on_user_followed.go
│   │       │   └── on_bookmark_added.go
│   │       └── transport/http/       #     chỉ đọc: GET /users/{id}/stats
│   │
│   └── platform/                     # ★ KỸ THUẬT — tuyệt đối không chứa nghiệp vụ
│       ├── eventbus/
│       │   ├── bus.go                #   interface Publisher / Subscriber
│       │   ├── kafka/
│       │   │   ├── publisher.go
│       │   │   └── consumer.go
│       │   └── inmem/
│       │       └── bus.go            #   impl cho unit test
│       ├── outbox/
│       │   ├── writer.go             #   ghi event trong transaction
│       │   ├── relay.go              #   poll bảng outbox → publish
│       │   └── query.sql
│       ├── idempotency/
│       │   ├── guard.go              #   chống xử lý trùng
│       │   └── query.sql
│       ├── postgres/
│       │   ├── pool.go
│       │   └── tx.go                 #   helper chạy trong transaction
│       ├── httpx/
│       │   ├── response.go           #   chuẩn hoá JSON trả về
│       │   ├── errors.go             #   map lỗi domain → HTTP status
│       │   └── middleware/
│       │       ├── auth.go
│       │       ├── request_id.go
│       │       └── recover.go
│       ├── logger/
│       │   └── logger.go             #   log/slog, có sẵn trong stdlib
│       └── config/
│           └── config.go
│
├── db/
│   └── migrations/
│       ├── 000001_create_users.up.sql
│       ├── 000002_create_user_stats.up.sql
│       ├── 000003_create_posts.up.sql
│       ├── 000004_create_outbox.up.sql
│       └── 000005_create_consumed_events.up.sql
│
├── deployments/
│   ├── docker-compose.yml            #   Postgres + Kafka + Kafka UI
│   └── Dockerfile
│
├── docs/
│   └── ARCHITECTURE.md               #   ← file này
│
├── sqlc.yaml
├── Makefile
├── go.mod
└── .env.example

4. Giải thích từng tầng

4.1 `cmd/` — vì sao tách ba binary

Binary	Nhiệm vụ	Vì sao tách riêng
`api`	Nhận HTTP, ghi DB + outbox	Scale theo lượng traffic người dùng
`worker`	Tiêu thụ event từ Kafka	Scale theo lượng event; một consumer chậm không làm chậm API
`relay`	Đọc outbox → publish Kafka	Chỉ chạy 1 instance hoặc phải có leader election

Ba tiến trình này có đặc tính vận hành hoàn toàn khác nhau: api cần độ trễ thấp, worker cần throughput, relay cần tính duy nhất. Nhét chung vào một binary nghĩa là bạn không thể tinh chỉnh cái nào cho ra hồn.

Trong môi trường dev, cả ba có thể chạy chung qua make dev để tiện — nhưng code phải tách sẵn, vì việc tách sau này khó hơn nhiều so với tách ngay từ đầu.

main.go chỉ làm 4 việc, không hơn: đọc config → mở kết nối → lắp ráp module → chạy và chờ tín hiệu dừng. Không có một dòng nghiệp vụ nào trong cmd/.

4.2 `internal/contracts/` — vì sao tập trung một chỗ

Đây là quyết định thiết kế then chốt và cũng dễ gây tranh cãi nhất.

Có hai trường phái:

Cách A — mỗi module tự sở hữu event của mình (modules/post/events/). Ưu điểm: quyền sở hữu rõ ràng. Nhược điểm: stats phải import .../modules/post/events — vẫn là import chéo module, vi phạm Luật 1 dù chỉ là struct dữ liệu.

Cách B — tập trung ở internal/contracts/ ← tài liệu này chọn cách này

Ưu điểm:

Không module nào import module nào. Luật 1 được giữ tuyệt đối và kiểm tra được tự động.
Toàn bộ ngôn ngữ giao tiếp của hệ thống nằm ở một chỗ. Người mới vào đọc contracts/ là hiểu hệ thống làm gì.
Khi tách microservice, chỉ cần copy nguyên package contracts sang service mới. Không có gì phải gỡ rối.

Nhược điểm (và cách sống chung):

contracts/ dễ trở thành bãi rác nếu không có kỷ luật. Quy tắc: chỉ chứa struct dữ liệu và hằng số. Không method, không logic, không import bất cứ package nội bộ nào. Nếu bạn thấy mình muốn viết một hàm trong contracts/, nó thuộc về module.

4.3 `internal/modules/` — bốn tầng bên trong

transport/http  →  service  →  repository  →  PostgreSQL
   (vào ra)       (nghiệp vụ)   (dữ liệu)
        ↓
   subscriber    →  service          (đường vào thứ hai: từ event)

Điểm quan trọng: service có hai đường vào — HTTP và event. Cả hai đều chỉ là "transport". Logic nghiệp vụ nằm ở service và không biết mình đang được gọi từ đâu.

Chiều phụ thuộc luôn hướng vào trong: transport biết service, service biết repository (qua interface), repository biết DB. Không bao giờ ngược lại.

Interface do bên gọi định nghĩa, không phải bên cài đặt. service định nghĩa nó cần gì từ repository:

// internal/modules/post/service/service.go
type Repository interface {
    Insert(ctx context.Context, tx pgx.Tx, p domain.Post) (domain.Post, error)
    FindBySlug(ctx context.Context, slug string) (domain.Post, error)
}

Nhờ vậy service test được mà không cần DB, và bạn có thể đổi cách lưu trữ mà không sửa nghiệp vụ.

4.4 `internal/modules/stats/` — module đặc biệt

stats gần như không có API ghi. Nó tồn tại để nghe event và cộng trừ counter. Toàn bộ "nghiệp vụ" của nó nằm trong subscriber/.

Đây chính là chỗ Event-Driven trả lại giá trị rõ nhất: tất cả 8 con số trong user_stats được duy trì bởi một module duy nhất, tách biệt hoàn toàn, mà không module nào khác biết nó tồn tại.

4.5 `internal/platform/` — ranh giới cần giữ sạch

Quy tắc kiểm tra: nếu bạn khởi động một dự án Go hoàn toàn khác và copy package này sang, nó có chạy được không? Nếu có → đúng chỗ. Nếu không → nó thuộc về modules/.

platform/ không được import modules/ hay contracts/.

5. Hợp đồng event

5.1 Quy ước đặt tên

<domain>.<entity>.<hành động ở thì quá khứ>.<version>

Ví dụ đúng	Ví dụ sai	Vì sao sai
`post.post.created.v1`	`post.create`	Thì hiện tại → đây là command, không phải event
`post.post.viewed.v1`	`post.increment_view`	Mô tả hành động của consumer, không phải sự kiện đã xảy ra
`social.user.followed.v1`	`stats.update_followers`	Publisher đang ra lệnh cho consumer — sai hoàn toàn

Thì quá khứ là bắt buộc. Event mô tả một việc đã xảy ra và không thể huỷ. Nếu tên event ở thì hiện tại, gần như chắc chắn bạn đang dùng event bus như một cách gọi hàm từ xa — và như thế thì mọi lợi ích của EDA đều mất, chỉ còn lại độ phức tạp.

Phép thử nhanh: nếu tên event nhắc tới việc consumer sẽ làm gì, thì tên đó sai. Publisher không được biết consumer làm gì.

5.2 Envelope

Mọi event đều được bọc trong một struct chung:

// internal/contracts/envelope.go
package contracts

import (
    "encoding/json"
    "time"
)

type Envelope struct {
    EventID       string          `json:"event_id"`       // UUID — khoá cho idempotency
    EventType     string          `json:"event_type"`     // "post.post.created.v1"
    AggregateID   string          `json:"aggregate_id"`   // ID thực thể → quyết định partition key
    OccurredAt    time.Time       `json:"occurred_at"`    // thời điểm sự việc xảy ra
    CorrelationID string          `json:"correlation_id"` // truy vết xuyên suốt một request
    Payload       json.RawMessage `json:"payload"`
}

Vì sao từng trường tồn tại:

EventID — consumer dùng nó để phát hiện message trùng. Không có trường này thì không thể idempotent.
AggregateID — dùng làm Kafka partition key. Mọi event của cùng một post rơi vào cùng partition ⇒ được xử lý đúng thứ tự. Bỏ qua trường này là nguồn gốc của những bug thứ tự cực khó tái hiện.
CorrelationID — gán ở HTTP middleware, truyền qua toàn bộ chuỗi event. Khi có sự cố, bạn grep một ID và thấy toàn bộ hành trình từ request tới consumer cuối cùng.
OccurredAt — thời điểm sự việc xảy ra, không phải thời điểm publish. Hai mốc này lệch nhau khi relay bị chậm, và consumer đôi khi cần biết mốc đầu.

5.3 Payload

// internal/contracts/v1/post_events.go
package v1

import "time"

const (
    TypePostCreated  = "post.post.created.v1"
    TypePostViewed   = "post.post.viewed.v1"
    TypePostUpvoted  = "post.post.upvoted.v1"
    TypePostDeleted  = "post.post.deleted.v1"
)

type PostCreated struct {
    PostID    string    `json:"post_id"`
    AuthorID  string    `json:"author_id"`
    Title     string    `json:"title"`
    Slug      string    `json:"slug"`
    ReadTime  int       `json:"read_time"`
    CreatedAt time.Time `json:"created_at"`
}

type PostViewed struct {
    PostID   string `json:"post_id"`
    AuthorID string `json:"author_id"`   // ← xem giải thích bên dưới
    ViewerID string `json:"viewer_id"`
}

Vì sao PostViewed mang theo AuthorID?

Vì stats cần cập nhật total_post_views của tác giả. Nếu payload không có AuthorID, consumer buộc phải query ngược sang bảng posts — tức là stats phải biết cấu trúc dữ liệu của post. Ranh giới module vỡ ngay lập tức.

Nguyên tắc chung: event phải chứa đủ dữ liệu để consumer làm việc mà không cần hỏi lại ai. Trùng lặp dữ liệu trong payload là cái giá rẻ để mua sự độc lập giữa các module. Đừng tiết kiệm ở chỗ này.

5.4 Versioning

Khi payload thay đổi theo hướng phá vỡ tương thích (xoá trường, đổi kiểu, đổi ngữ nghĩa):

Tạo contracts/v2/post_events.go với type mới post.post.created.v2
Publisher phát cả hai phiên bản trong một giai đoạn chuyển tiếp
Chuyển từng consumer sang v2
Khi không còn consumer nào nghe v1, ngừng phát v1

Không bao giờ sửa nghĩa của một event đã phát hành. Ngoài kia có thể còn message v1 đang nằm trong Kafka chờ được xử lý, và consumer đọc chúng bằng schema mới sẽ diễn giải sai dữ liệu — âm thầm, không báo lỗi.

Thêm trường có giá trị mặc định hợp lệ thì tương thích ngược, không cần lên version.

6. Event Bus

6.1 Interface

// internal/platform/eventbus/bus.go
package eventbus

import (
    "context"
    "github.com/you/project/internal/contracts"
)

type Publisher interface {
    Publish(ctx context.Context, topic string, e contracts.Envelope) error
}

type Handler func(ctx context.Context, e contracts.Envelope) error

type Subscriber interface {
    Subscribe(topic, group string, h Handler)
    Run(ctx context.Context) error
}

Interface nhỏ đến mức này là có chủ đích. Toàn bộ modules/ chỉ nhìn thấy hai phương thức trên; Kafka, NATS hay in-memory đều nằm sau nó. Đổi broker = viết một implementation mới, không sửa một dòng nghiệp vụ nào.

6.2 Hai implementation

Impl	Dùng khi	Đặc điểm
`kafka/`	Chạy thật, integration test	Bền, replay được, có consumer group
`inmem/`	Unit test	Chạy trong process, không cần Docker, gọi handler đồng bộ

Bản inmem là thứ giữ cho test suite của bạn chạy trong vài giây thay vì vài phút. Đừng bỏ qua nó.

6.3 Consumer group

Mỗi consumer đăng ký với một group riêng:

sub.Subscribe(contracts.TopicPost, "stats-service", statsHandler)
sub.Subscribe(contracts.TopicPost, "search-indexer", searchHandler)

Cùng một event được giao cho cả hai group (mỗi group nhận một bản sao), nhưng chỉ một instance trong mỗi group xử lý. Nhờ vậy bạn scale worker lên 5 pod mà posts_count không tăng gấp 5 lần.

7. Outbox Pattern

7.1 Vấn đề cần giải quyết

Bạn không thể có transaction chung giữa PostgreSQL và Kafka. Bất kỳ đoạn code nào có dạng "commit DB rồi publish" đều tồn tại một khoảnh khắc mà nếu process chết, dữ liệu và event sẽ lệch nhau vĩnh viễn.

7.2 Cách hoạt động

Ghi event vào một bảng trong chính DB đó, cùng transaction với dữ liệu nghiệp vụ. Một tiến trình riêng đọc bảng và đẩy đi.

-- db/migrations/000004_create_outbox.up.sql
CREATE TABLE outbox (
    id             BIGSERIAL PRIMARY KEY,
    event_id       UUID        NOT NULL UNIQUE,
    event_type     TEXT        NOT NULL,
    topic          TEXT        NOT NULL,
    aggregate_id   TEXT        NOT NULL,
    correlation_id TEXT,
    payload        JSONB       NOT NULL,
    occurred_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    published_at   TIMESTAMPTZ,          -- NULL = chưa gửi
    attempts       INT         NOT NULL DEFAULT 0,
    last_error     TEXT
);

-- Index một phần: chỉ đánh index các dòng chưa gửi.
-- Bảng có thể có hàng triệu dòng đã gửi, nhưng index luôn nhỏ.
CREATE INDEX idx_outbox_unpublished
    ON outbox (id) WHERE published_at IS NULL;

7.3 Phía ghi

// internal/platform/outbox/writer.go
func (w *Writer) Publish(ctx context.Context, tx pgx.Tx, topic, eventType, aggregateID string, payload any) error {
    body, err := json.Marshal(payload)
    if err != nil {
        return fmt.Errorf("marshal payload: %w", err)
    }
    _, err = tx.Exec(ctx, `
        INSERT INTO outbox (event_id, event_type, topic, aggregate_id, correlation_id, payload)
        VALUES ($1, $2, $3, $4, $5, $6)`,
        uuid.New(), eventType, topic, aggregateID, requestid.From(ctx), body)
    return err
}

Chú ý tham số tx pgx.Tx — bắt buộc truyền transaction vào. Đây là cách chúng ta dùng chính hệ thống kiểu của Go để ép Luật 3: không có transaction thì không gọi được hàm này, nên không ai vô tình publish ngoài transaction được.

7.4 Relay

// internal/platform/outbox/relay.go  (rút gọn)
func (r *Relay) tick(ctx context.Context) error {
    rows, err := r.db.Query(ctx, `
        SELECT id, event_id, event_type, topic, aggregate_id, correlation_id, payload, occurred_at
        FROM outbox
        WHERE published_at IS NULL
        ORDER BY id
        LIMIT 100
        FOR UPDATE SKIP LOCKED`)   // ← cho phép chạy nhiều relay an toàn
    ...
    for _, ev := range events {
        if err := r.pub.Publish(ctx, ev.Topic, ev.Envelope()); err != nil {
            r.markFailed(ctx, ev.ID, err)
            continue                              // lỗi một event không chặn các event khác
        }
        r.markPublished(ctx, ev.ID)
    }
    return nil
}

FOR UPDATE SKIP LOCKED là mấu chốt: nhiều instance relay chạy song song sẽ tự động chia nhau công việc thay vì tranh chấp hoặc gửi trùng.

ORDER BY id giữ thứ tự phát theo đúng thứ tự ghi.

7.5 Hệ quả: at-least-once

Nếu relay publish thành công nhưng chết trước khi kịp markPublished, event sẽ được gửi lại ở lần chạy sau. Đây là hành vi được thiết kế có chủ đích: thà gửi trùng còn hơn mất. Việc xử lý trùng thuộc trách nhiệm của consumer — mục tiếp theo.

7.6 Dọn dẹp

Bảng outbox sẽ phình ra vô hạn nếu không dọn. Xoá các dòng đã publish quá 7 ngày:

DELETE FROM outbox WHERE published_at < now() - INTERVAL '7 days';

Giữ lại 7 ngày để còn debug khi có sự cố.

8. Idempotency ở phía consumer

8.1 Vì sao bắt buộc

Kafka bảo đảm at-least-once. Cộng thêm cơ chế retry của relay, event sẽ bị giao trùng vào một ngày nào đó. Với một consumer làm việc posts_count = posts_count + 1, mỗi lần giao trùng là một con số sai.

Điều tệ nhất: sai lệch này không gây lỗi, không xuất hiện trong log. Bạn chỉ phát hiện khi có người dùng thắc mắc vì sao số bài viết của họ không khớp.

8.2 Bảng chống trùng

-- db/migrations/000005_create_consumed_events.up.sql
CREATE TABLE consumed_events (
    event_id     UUID        NOT NULL,
    consumer     TEXT        NOT NULL,   -- tên consumer group
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    PRIMARY KEY (event_id, consumer)
);

Khoá chính là cặp (event_id, consumer) chứ không phải riêng event_id — vì cùng một event được nhiều consumer group xử lý độc lập, và việc stats đã xử lý không được ngăn search xử lý.

8.3 Mẫu chuẩn cho mọi consumer

// internal/modules/stats/subscriber/on_post_created.go
package subscriber

func (s *Subscriber) OnPostCreated(ctx context.Context, e contracts.Envelope) error {
    var p v1.PostCreated
    if err := json.Unmarshal(e.Payload, &p); err != nil {
        // Payload hỏng — retry bao nhiêu lần cũng vô ích.
        // Trả lỗi vĩnh viễn để message đi thẳng vào DLQ.
        return fmt.Errorf("%w: unmarshal PostCreated: %v", eventbus.ErrPermanent, err)
    }

    return s.tx.Do(ctx, func(tx pgx.Tx) error {
        // 1. Đánh dấu đã xử lý. Nếu trùng → thoát êm, không làm gì.
        ok, err := s.guard.Mark(ctx, tx, e.EventID, ConsumerName)
        if err != nil {
            return err
        }
        if !ok {
            s.log.Debug("bỏ qua event trùng", "event_id", e.EventID)
            return nil
        }

        // 2. Việc thật — CÙNG transaction với bước 1.
        return s.repo.IncrementPostsCount(ctx, tx, p.AuthorID)
    })
}

Điểm sống còn: bước 1 và bước 2 phải nằm trong cùng một transaction. Nếu tách ra, sẽ có khoảnh khắc mà event đã được đánh dấu "đã xử lý" nhưng counter chưa tăng — và vì đã đánh dấu, lần retry sau sẽ bỏ qua nó. Bạn vừa tự tạo ra đúng cái bug mà idempotency sinh ra để ngăn chặn.

// internal/platform/idempotency/guard.go
// Trả về false nếu event này đã được consumer đó xử lý rồi.
func (g *Guard) Mark(ctx context.Context, tx pgx.Tx, eventID, consumer string) (bool, error) {
    tag, err := tx.Exec(ctx, `
        INSERT INTO consumed_events (event_id, consumer)
        VALUES ($1, $2)
        ON CONFLICT DO NOTHING`, eventID, consumer)
    if err != nil {
        return false, err
    }
    return tag.RowsAffected() == 1, nil
}

8.4 Khi nào không cần bảng này

Nếu thao tác vốn đã idempotent về mặt bản chất, bỏ qua guard cũng được:

-- Chạy 100 lần vẫn ra cùng kết quả → không cần guard
UPDATE users SET avatar_url = $2 WHERE id = $1;

-- Chạy 2 lần là sai ngay → BẮT BUỘC có guard
UPDATE user_stats SET posts_count = posts_count + 1 WHERE user_id = $1;

Phép thử: chạy hai lần liên tiếp có ra kết quả khác không? Có → cần guard.

9. Luồng dữ liệu đầu-cuối

9.1 Ví dụ: đăng bài viết

┌─ [api] ──────────────────────────────────────────────┐
│  POST /api/v1/posts                                  │
│    transport/http/handler.go                         │
│      → service.CreatePost()                          │
│          BEGIN TX                                    │
│            INSERT INTO posts                         │
│            INSERT INTO outbox (post.post.created.v1) │
│          COMMIT                     ← 201 trả về ĐÂY │
└──────────────────────────────────────────────────────┘
                        ↓  (bất đồng bộ)
┌─ [relay] ────────────────────────────────────────────┐
│  SELECT ... WHERE published_at IS NULL               │
│    FOR UPDATE SKIP LOCKED                            │
│  → kafka.Publish(topic=post, key=aggregate_id)       │
│  → UPDATE outbox SET published_at = now()            │
└──────────────────────────────────────────────────────┘
                        ↓
┌─ [worker] ───────────────────────────────────────────┐
│  group "stats-service" nhận post.post.created.v1     │
│    BEGIN TX                                          │
│      INSERT INTO consumed_events  ← trùng thì thoát  │
│      UPDATE user_stats SET posts_count += 1          │
│    COMMIT                                            │
└──────────────────────────────────────────────────────┘

9.2 Ví dụ: một event, nhiều consumer

Event post.post.viewed.v1 được ba group xử lý độc lập:

Consumer group	Việc làm	Nếu group này chết
`stats-service`	`user_stats.total_post_views += 1`	Hai group kia vẫn chạy bình thường
`post-counter`	`posts.views_count += 1`	—
`feed-ranker`	Cập nhật điểm xếp hạng	—

Đây là lợi ích cụ thể nhất của EDA: thêm consumer thứ tư không cần đụng vào module post, và một consumer sập không kéo theo cái nào khác.

9.3 Nhất quán cuối cùng (eventual consistency)

Sau khi API trả 201 Created, posts_count chưa tăng ngay. Độ trễ thực tế thường là 50–500ms (phụ thuộc chu kỳ poll của relay).

Đây là cái giá bạn trả cho kiến trúc này. Cần xử lý đúng ở hai chỗ:

Chấp nhận được — các con số thống kê, feed, thông báo, chỉ mục tìm kiếm. Người dùng không nhận ra 200ms.

KHÔNG chấp nhận được — những thứ phải nhất quán tức thời:

Kiểm tra trùng email khi đăng ký → dùng UNIQUE constraint trong cùng transaction, không qua event
Trừ tiền / trừ tồn kho → cùng transaction
Kiểm tra quyền truy cập → truy vấn trực tiếp, không đọc từ read model

Quy tắc: nếu đọc dữ liệu cũ vài trăm mili-giây dẫn tới quyết định sai, thì đừng dùng event cho việc đó.

Ở tầng UI, cách xử lý thông thường là optimistic update: tăng số hiển thị ngay tại client, để backend đồng bộ lại sau.

10. Lắp ráp module

Mỗi module có đúng một điểm lắp ráp, và đó là nơi duy nhất bên ngoài được phép biết tới:

// internal/modules/stats/module.go
package stats

type Module struct {
    sub *subscriber.Subscriber
    api *http.Handler
}

func New(db *pgxpool.Pool, guard *idempotency.Guard, log *slog.Logger) *Module {
    repo := repository.NewPostgres(db)
    return &Module{
        sub: subscriber.New(repo, guard, log),
        api: http.NewHandler(repo),
    }
}

// RegisterHTTP — chỉ các endpoint đọc
func (m *Module) RegisterHTTP(r chi.Router) {
    r.Get("/users/{id}/stats", m.api.GetUserStats)
}

// RegisterEvents — khai báo module này nghe gì.
// Đọc hàm này là biết ngay module tham gia vào hệ thống ra sao.
func (m *Module) RegisterEvents(bus eventbus.Subscriber) {
    const group = "stats-service"
    bus.Subscribe(contracts.TopicPost, group, eventbus.Route(map[string]eventbus.Handler{
        v1.TypePostCreated: m.sub.OnPostCreated,
        v1.TypePostViewed:  m.sub.OnPostViewed,
        v1.TypePostDeleted: m.sub.OnPostDeleted,
    }))
    bus.Subscribe(contracts.TopicSocial, group, eventbus.Route(map[string]eventbus.Handler{
        v1.TypeUserFollowed:   m.sub.OnUserFollowed,
        v1.TypeUserUnfollowed: m.sub.OnUserUnfollowed,
    }))
}

Và main.go chỉ còn là danh sách phẳng:

// cmd/worker/main.go
func main() {
    cfg := config.Load()
    db := postgres.MustConnect(cfg.DatabaseURL)
    log := logger.New(cfg.LogLevel)
    guard := idempotency.NewGuard()
    bus := kafka.NewConsumer(cfg.KafkaBrokers, log)

    modules := []interface{ RegisterEvents(eventbus.Subscriber) }{
        stats.New(db, guard, log),
        post.New(db, guard, log),
        identity.New(db, guard, log),
    }
    for _, m := range modules {
        m.RegisterEvents(bus)
    }

    ctx := signals.Context()          // huỷ khi nhận SIGINT/SIGTERM
    if err := bus.Run(ctx); err != nil {
        log.Error("worker dừng", "err", err)
        os.Exit(1)
    }
}

Không có dependency injection framework. Với quy mô này, một hàm New() viết tay dễ đọc và dễ debug hơn nhiều so với wire hay fx.

11. Xử lý lỗi, retry và DLQ

11.1 Phân loại lỗi — quyết định quan trọng nhất

// internal/platform/eventbus/errors.go
var (
    ErrPermanent = errors.New("permanent")  // retry vô ích → DLQ ngay
    ErrTransient = errors.New("transient")  // retry có thể thành công
)

Loại	Ví dụ	Xử lý
Transient	Mất kết nối DB, deadlock, Kafka timeout	Retry với exponential backoff
Permanent	JSON hỏng, thiếu trường bắt buộc, vi phạm ràng buộc nghiệp vụ	Vào DLQ ngay, không retry

Nhầm lẫn hai loại này gây hậu quả nặng theo hai hướng ngược nhau:

Coi lỗi permanent là transient → message hỏng bị retry vô hạn và chặn toàn bộ partition. Mọi event phía sau nó đứng im. Đây là sự cố production kinh điển của hệ Kafka.
Coi lỗi transient là permanent → mất event chỉ vì DB nấc một nhịp.

Khi không chắc chắn, mặc định coi là transient — nhưng phải có giới hạn số lần retry.

11.2 Chiến lược retry

lần 1: ngay lập tức
lần 2: sau 1 giây
lần 3: sau 5 giây
lần 4: sau 30 giây
lần 5: sau 2 phút
→ vẫn lỗi: đẩy vào DLQ, ghi log mức ERROR, bắn cảnh báo

11.3 Dead Letter Queue

Mỗi topic có một DLQ tương ứng: post.DLQ, social.DLQ.

Message vào DLQ mang theo metadata để chẩn đoán: original_topic, error_message, attempts, failed_at, consumer_group.

DLQ không có người xem thì vô dụng. Cần tối thiểu:

Cảnh báo khi DLQ có message mới (Slack/email)
Một lệnh CLI để phát lại sau khi đã sửa: make dlq-replay TOPIC=post

11.4 Quan sát hệ thống

Bốn chỉ số cần theo dõi ngay từ ngày đầu:

Chỉ số	Ngưỡng cảnh báo	Ý nghĩa khi vượt ngưỡng
Số dòng outbox chưa publish	> 1000	Relay chết hoặc Kafka không nhận
Consumer lag	> 10.000 message	Worker không theo kịp → cần scale
Số message trong DLQ	> 0	Có event hỏng cần người xử lý
Độ trễ event (`occurred_at` → xử lý xong)	p99 > 5s	Hệ thống đang tắc ở đâu đó

Chỉ số đầu tiên là quan trọng nhất và cũng dễ bị bỏ quên nhất: nếu relay chết âm thầm, API vẫn trả 200 bình thường trong khi không event nào được phát đi. Không có cảnh báo này, bạn chỉ phát hiện khi mọi con số thống kê đều đã sai.

12. Chiến lược kiểm thử

12.1 Ba tầng

Tầng	Phạm vi	Hạ tầng cần	Thời gian
Unit	`service` với repo giả + `eventbus/inmem`	Không	mili-giây
Consumer	Gọi thẳng handler với payload dựng sẵn	Postgres (testcontainers)	~giây
Integration	HTTP → outbox → relay → consumer	Postgres + Kafka	~chục giây

12.2 Test bắt buộc cho mọi consumer

Đây là test dễ bị quên nhất và cũng có giá trị cao nhất:

func TestOnPostCreated_LaIdempotent(t *testing.T) {
    ctx, db := setupTestDB(t)
    sub := newTestSubscriber(db)
    env := envelopeFor(v1.PostCreated{AuthorID: userID, PostID: postID})

    // Xử lý hai lần với CÙNG một event_id
    require.NoError(t, sub.OnPostCreated(ctx, env))
    require.NoError(t, sub.OnPostCreated(ctx, env))

    // posts_count phải là 1, không phải 2
    stats := fetchStats(t, db, userID)
    assert.Equal(t, 1, stats.PostsCount)
}

Mọi consumer đều phải có test này. Không có nó, bạn sẽ không biết idempotency hỏng cho tới khi Kafka giao trùng lần đầu trên production.

12.3 Test service không cần Kafka

func TestCreatePost_PhatEventPostCreated(t *testing.T) {
    bus := inmem.New()
    svc := service.New(fakeRepo{}, bus)

    _, err := svc.CreatePost(ctx, service.CreatePostInput{AuthorID: userID, Title: "Xin chào"})
    require.NoError(t, err)

    events := bus.Published()
    require.Len(t, events, 1)
    assert.Equal(t, v1.TypePostCreated, events[0].EventType)
}

Test này chạy trong vài mili-giây và trả lời đúng câu hỏi cần trả lời: service có phát đúng event không? Còn consumer xử lý event đó ra sao là chuyện của test khác.

13. Hạ tầng và công cụ

13.1 docker-compose

# deployments/docker-compose.yml
services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: secret
      POSTGRES_DB: community
    ports: ["5432:5432"]
    volumes: ["pgdata:/var/lib/postgresql/data"]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U app"]
      interval: 5s

  kafka:
    image: bitnami/kafka:3.7
    environment:
      KAFKA_CFG_NODE_ID: "0"
      KAFKA_CFG_PROCESS_ROLES: controller,broker
      KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: 0@kafka:9093
      KAFKA_CFG_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CFG_CONTROLLER_LISTENER_NAMES: CONTROLLER
    ports: ["9092:9092"]

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
    ports: ["8080:8080"]
    depends_on: [kafka]

volumes:
  pgdata:

Kafka ở chế độ KRaft — không cần Zookeeper. kafka-ui tại localhost:8080 cực kỳ hữu ích khi debug: bạn nhìn thấy message thật, consumer lag thật, DLQ thật.

13.2 sqlc

# sqlc.yaml
version: "2"
sql:
  - engine: postgresql
    schema: db/migrations
    queries: internal/modules/stats/repository/query.sql
    gen:
      go:
        package: gen
        out: internal/modules/stats/repository/gen
        sql_package: pgx/v5
  # lặp lại khối trên cho identity, post, outbox, idempotency

Ví dụ query cho stats:

-- internal/modules/stats/repository/query.sql

-- name: IncrementPostsCount :exec
UPDATE user_stats SET posts_count = posts_count + 1 WHERE user_id = $1;

-- name: DecrementPostsCount :exec
UPDATE user_stats SET posts_count = GREATEST(posts_count - 1, 0) WHERE user_id = $1;

-- name: IncrementTotalPostViews :exec
UPDATE user_stats SET total_post_views = total_post_views + 1 WHERE user_id = $1;

-- name: GetUserStats :one
SELECT * FROM user_stats WHERE user_id = $1;

GREATEST(x - 1, 0) là biện pháp phòng thủ: nếu một event xoá bị xử lý trùng do lỗi ở đâu đó, counter tệ nhất cũng chỉ dừng ở 0 chứ không xuống âm. Người dùng thấy 0 thì bối rối; thấy -3 thì mất niềm tin vào cả sản phẩm.

13.3 Makefile

.PHONY: dev up down migrate sqlc test lint

up:        ; docker compose -f deployments/docker-compose.yml up -d
down:      ; docker compose -f deployments/docker-compose.yml down
migrate:   ; migrate -path db/migrations -database "$(DATABASE_URL)" up
sqlc:      ; sqlc generate
test:      ; go test ./... -race -count=1
lint:      ; golangci-lint run

dev: up migrate
	go run ./cmd/api    & \
	go run ./cmd/relay  & \
	go run ./cmd/worker & \
	wait

Cờ -race trong make test là bắt buộc với hệ thống có nhiều goroutine như thế này.

13.4 Kiểm tra ranh giới tự động

Luật 1 chỉ có giá trị nếu nó được kiểm tra tự động. Con người sẽ quên; CI thì không.

Thêm vào .golangci.yml:

linters-settings:
  depguard:
    rules:
      modules-khong-import-cheo:
        files: ["**/internal/modules/**"]
        deny:
          - pkg: "github.com/you/project/internal/modules"
            desc: "Module không được import module khác — hãy dùng event qua internal/contracts"
      platform-phai-sach:
        files: ["**/internal/platform/**"]
        deny:
          - pkg: "github.com/you/project/internal/modules"
            desc: "platform là tầng kỹ thuật, không được biết tới nghiệp vụ"

Không có rule này, ranh giới module sẽ bị xói mòn dần trong vài tháng — thường bắt đầu bằng một lần "import tạm cho nhanh, sửa sau".

14. Checklist thêm module mới

Ví dụ thêm module social (follow / unfollow):

[ ] 1. Migration: db/migrations/00000X_create_follows.up.sql
[ ] 2. Contracts: thêm UserFollowed, UserUnfollowed vào internal/contracts/v1/user_events.go + hằng số type
[ ] 3. Topic: thêm TopicSocial vào internal/contracts/topics.go
[ ] 4. Khung module: internal/modules/social/ với domain/ service/ repository/ transport/http/ subscriber/
[ ] 5. Query + make sqlc
[ ] 6. Service phát event qua outbox, trong transaction
[ ] 7. HTTP handler + route trong module.go
[ ] 8. Consumer bên stats: on_user_followed.go cập nhật followers_count và following_users_count
[ ] 9. Đăng ký event trong stats/module.go → RegisterEvents
[ ] 10. Nối module vào cmd/api/main.go và cmd/worker/main.go
[ ] 11. Test: service phát đúng event + consumer idempotent
[ ] 12. make lint — xác nhận không có import chéo module

Một chi tiết đáng chú ý ở bước 8: sự kiện A follow B làm thay đổi thống kê của hai người — following_users_count của A và followers_count của B. Cả hai phải nằm trong cùng một transaction với consumed_events, nếu không việc retry giữa chừng sẽ làm lệch một trong hai con số.

15. Lộ trình mở rộng

15.1 Consumer mới, không đụng code cũ

Khi cần thêm tính năng, phần lớn trường hợp chỉ là viết thêm một consumer:

Tính năng	Nghe event	Việc làm
Tìm kiếm	`post.post.created.v1`	Đẩy vào Elasticsearch
Thông báo	`social.user.followed.v1`	Gửi email/push
Feed	`post.post.created.v1`	Fan-out vào feed người theo dõi
Gợi ý	`post.post.viewed.v1`	Cập nhật vector sở thích

Không file nào trong modules/post/ bị sửa. Đây là kiểm chứng thực tế cho thấy kiến trúc đang hoạt động đúng — nếu bạn phát hiện mình phải sửa post để thêm một trong các tính năng trên, hãy dừng lại và xem lại ranh giới.

15.2 Tách microservice khi cần

Nhờ Luật 1, việc tách một module ra thành service riêng là thao tác cơ học:

Tạo repo mới, copy nguyên internal/contracts/ và internal/platform/
Copy internal/modules/stats/
Trỏ tới database riêng (nếu muốn tách cả dữ liệu)
Bỏ stats.New(...) khỏi cmd/worker/main.go của monolith

Không sửa một dòng nghiệp vụ nào. Đây chính là lý do chúng ta chấp nhận độ phức tạp của event bus ngay từ đầu, dù hiện tại chỉ có một binary.

Chỉ nên tách khi có lý do vận hành cụ thể — cần scale riêng, cần ngôn ngữ khác, hoặc cần đội khác sở hữu. Tách vì "microservice nghe hiện đại hơn" là cách nhanh nhất để biến một hệ thống chạy tốt thành một hệ thống phân tán khó gỡ.

15.3 Rebuild lại số liệu thống kê

Sớm hay muộn user_stats cũng sẽ lệch — do bug, do lỗi vận hành, hoặc do một giai đoạn DLQ không ai xử lý. Hãy chuẩn bị sẵn từ đầu một job tính lại toàn bộ:

-- Nguồn sự thật vẫn là các bảng gốc; user_stats chỉ là bản sao đã tính sẵn.
UPDATE user_stats us
SET posts_count = (SELECT COUNT(*) FROM posts WHERE author_id = us.user_id);

Đây là lưới an toàn cuối cùng của kiến trúc này. Đừng để tới lúc cần mới đi viết nó.

Phụ lục — Thư viện đề xuất

Mục đích	Thư viện
HTTP router	`github.com/go-chi/chi/v5`
Postgres driver	`github.com/jackc/pgx/v5`
Sinh code SQL	`github.com/sqlc-dev/sqlc`
Migration	`github.com/golang-migrate/migrate/v4`
Kafka client	`github.com/twmb/franz-go`
UUID	`github.com/google/uuid`
Config	`github.com/caarlos0/env/v11`
Log	`log/slog` (thư viện chuẩn)
Test	`github.com/stretchr/testify` + `github.com/testcontainers/testcontainers-go`

Ghi chú về Kafka client: franz-go được ưu tiên hơn sarama vì API hiện đại hơn, ít bẫy hơn và hiệu năng tốt hơn. segmentio/kafka-go cũng là lựa chọn hợp lý nếu bạn thích API đơn giản.

Backend

Mục lục

1. Vì sao Event-Driven cho dự án này

2. Ba luật bất di bất dịch

Luật 1 — Module không bao giờ import module khác

Luật 2 — Event là hợp đồng, không phải lời gọi hàm

Luật 3 — Ghi DB và phát event phải nằm trong cùng một transaction

3. Cây thư mục đầy đủ

4. Giải thích từng tầng

4.1 cmd/ — vì sao tách ba binary

4.2 internal/contracts/ — vì sao tập trung một chỗ

4.3 internal/modules/ — bốn tầng bên trong

4.4 internal/modules/stats/ — module đặc biệt

4.5 internal/platform/ — ranh giới cần giữ sạch

5. Hợp đồng event

5.1 Quy ước đặt tên

5.2 Envelope

5.3 Payload

5.4 Versioning

6. Event Bus

6.1 Interface

6.2 Hai implementation

6.3 Consumer group

7. Outbox Pattern

7.1 Vấn đề cần giải quyết

7.2 Cách hoạt động

7.3 Phía ghi

7.4 Relay

7.5 Hệ quả: at-least-once

7.6 Dọn dẹp

8. Idempotency ở phía consumer

8.1 Vì sao bắt buộc

8.2 Bảng chống trùng

8.3 Mẫu chuẩn cho mọi consumer

8.4 Khi nào không cần bảng này

9. Luồng dữ liệu đầu-cuối

9.1 Ví dụ: đăng bài viết

9.2 Ví dụ: một event, nhiều consumer

9.3 Nhất quán cuối cùng (eventual consistency)

10. Lắp ráp module

11. Xử lý lỗi, retry và DLQ

11.1 Phân loại lỗi — quyết định quan trọng nhất

11.2 Chiến lược retry

11.3 Dead Letter Queue

11.4 Quan sát hệ thống

12. Chiến lược kiểm thử

12.1 Ba tầng

12.2 Test bắt buộc cho mọi consumer

12.3 Test service không cần Kafka

13. Hạ tầng và công cụ

13.1 docker-compose

13.2 sqlc

13.3 Makefile

13.4 Kiểm tra ranh giới tự động

14. Checklist thêm module mới

15. Lộ trình mở rộng

15.1 Consumer mới, không đụng code cũ

15.2 Tách microservice khi cần

15.3 Rebuild lại số liệu thống kê

Phụ lục — Thư viện đề xuất

Mục lục

4.1 `cmd/` — vì sao tách ba binary

4.2 `internal/contracts/` — vì sao tập trung một chỗ

4.3 `internal/modules/` — bốn tầng bên trong

4.4 `internal/modules/stats/` — module đặc biệt

4.5 `internal/platform/` — ranh giới cần giữ sạch