12.5K 160 1.1K

Đã đăng vào thg 8 13, 4:48 CH 18 phút đọc

#10 Bước 6 — Consumer, Idempotency và DLQ Phần 2

5.5 `transport/http` — endpoint đọc

internal/modules/stats/transport/http/handler.go:

package http

import (
	"errors"
	"log/slog"
	"net/http"
	"time"

	"github.com/go-chi/chi/v5"
	"github.com/google/uuid"

	"github.com/chuongtd/community/internal/modules/stats/domain"
	"github.com/chuongtd/community/internal/platform/httpx"
)

// StatsResponse liệt kê TƯỜNG MINH từng trường trả ra.
// Không trả thẳng domain.UserStats — lý lẽ ở bước 4 §14.2.
type StatsResponse struct {
	UserID              string    `json:"user_id"`
	PostsCount          int32     `json:"posts_count"`
	TotalPostViews      int32     `json:"total_post_views"`
	FollowersCount      int32     `json:"followers_count"`
	FollowingUsersCount int32     `json:"following_users_count"`
	BookmarksCount      int32     `json:"bookmarks_count"`
	UpdatedAt           time.Time `json:"updated_at"`
}

type Reader interface {
	GetUserStats(ctx context.Context, userID uuid.UUID) (domain.UserStats, error)
}

type Handler struct {
	repo Reader
	log  *slog.Logger
}

func NewHandler(repo Reader, log *slog.Logger) *Handler {
	return &Handler{repo: repo, log: log}
}

func (h *Handler) GetUserStats(w http.ResponseWriter, r *http.Request) {
	id, err := uuid.Parse(chi.URLParam(r, "id"))
	if err != nil {
		httpx.Fail(w, r, http.StatusBadRequest, "invalid_id", "id không phải UUID")
		return
	}

	st, err := h.repo.GetUserStats(r.Context(), id)
	switch {
	case errors.Is(err, domain.ErrStatsNotFound):
		// 404 ở đây có hai nghĩa khác nhau — xem ghi chú bên dưới.
		httpx.Fail(w, r, http.StatusNotFound, "not_found", "chưa có thống kê cho người dùng này")
		return
	case err != nil:
		h.log.Error("stats: đọc thất bại", "err", err, "user_id", id)
		httpx.Fail(w, r, http.StatusInternalServerError, "internal", "lỗi hệ thống")
		return
	}

	httpx.JSON(w, http.StatusOK, StatsResponse{
		UserID:              st.UserID.String(),
		PostsCount:          st.PostsCount,
		TotalPostViews:      st.TotalPostViews,
		FollowersCount:      st.FollowersCount,
		FollowingUsersCount: st.FollowingUsersCount,
		BookmarksCount:      st.BookmarksCount,
		UpdatedAt:           st.UpdatedAt,
	})
}

404 ở đây nghĩa là gì — và vì sao nó không sai

Endpoint này trả 404 trong hai tình huống rất khác nhau:

Người dùng không tồn tại

Người dùng vừa đăng ký xong, event chưa được xử lý (dưới một giây)

Trường hợp 2 là nhất quán cuối cùng nhìn từ phía client, và ta không thể xoá nó — đó là cái giá đã cố ý trả ở bước 4 §18.2 để giữ Luật 1.

Cách xử lý thực tế, theo thứ tự nên xét:

Cách Đánh đổi

Trả 404, client hiển thị số 0 Đơn giản nhất, và đúng: giá trị thật sắp tới cũng là 0

COALESCE ở tầng đọc → luôn trả 8 số 0 Không phân biệt được "user không tồn tại" với "chưa kịp xử lý"

Chặn: identity tạo dòng đồng bộ Vi phạm Luật 1. Đã bác ở bước 4 §18.1

Tài liệu này chọn cách đầu. Điều quan trọng là client phải biết 404 ở endpoint này không có nghĩa "người này không tồn tại" — nên thông báo lỗi viết là "chưa có thống kê", không phải "không tìm thấy người dùng".

Cách	Đánh đổi
Trả 404, client hiển thị số 0	Đơn giản nhất, và đúng: giá trị thật sắp tới cũng là 0
`COALESCE` ở tầng đọc → luôn trả 8 số 0	Không phân biệt được "user không tồn tại" với "chưa kịp xử lý"
Chặn: `identity` tạo dòng đồng bộ	Vi phạm Luật 1. Đã bác ở bước 4 §18.1

6. `cmd/worker/main.go`

New-Item -ItemType Directory -Force cmd/worker

package main

import (
	"context"
	"log/slog"
	"os"
	"os/signal"
	"syscall"
	"time"

	"golang.org/x/sync/errgroup"

	"github.com/chuongtd/community/internal/modules/stats"
	"github.com/chuongtd/community/internal/platform/cleanup"
	"github.com/chuongtd/community/internal/platform/config"
	"github.com/chuongtd/community/internal/platform/eventbus"
	"github.com/chuongtd/community/internal/platform/eventbus/kafka"
	"github.com/chuongtd/community/internal/platform/idempotency"
	"github.com/chuongtd/community/internal/platform/logger"
	"github.com/chuongtd/community/internal/platform/postgres"
)

func main() {
	if err := run(); err != nil {
		slog.Error("worker dừng vì lỗi", "err", err)
		os.Exit(1)
	}
}

func run() error {
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	if err := cfg.ValidateWorker(); err != nil {
		return err
	}

	log := logger.New(cfg.LogLevel)
	slog.SetDefault(log)

	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	pgCfg := postgres.DefaultConfig(cfg.DatabaseURL, "community-worker")
	pgCfg.MaxConns = 10 // ngân sách kết nối: bước 2 §3.2
	pool, err := postgres.Connect(ctx, pgCfg)
	if err != nil {
		return err
	}
	defer pool.Close()

	// Publisher dùng cho DLQ. Cùng cấu hình acks=all với relay (§4.2).
	pub, err := kafka.NewPublisher(kafka.Config{
		Brokers:        cfg.KafkaBrokers,
		ClientID:       "community-worker",
		ProduceTimeout: cfg.KafkaProduceTimeout,
	})
	if err != nil {
		return err
	}
	defer pub.Close()

	bus := kafka.NewConsumer(kafka.ConsumerConfig{
		Brokers:     cfg.KafkaBrokers,
		ClientID:    "community-worker",
		MaxAttempts: cfg.WorkerMaxAttempts,
		PollMax:     cfg.WorkerPollMax,
	}, pub, log)

	txm := postgres.NewTxManager(pool)
	guard := idempotency.NewGuard()

	// Danh sách module. Thêm module mới = thêm một dòng.
	modules := []interface {
		RegisterEvents(eventbus.Subscriber)
	}{
		stats.New(pool, txm, guard, log),
	}
	for _, m := range modules {
		m.RegisterEvents(bus)
	}

	g, ctx := errgroup.WithContext(ctx)
	g.Go(func() error { return bus.Run(ctx) })
	g.Go(func() error {
		return cleanup.Run(ctx, pool, log, cleanup.Config{Every: time.Hour})
	})
	return g.Wait()
}

Thêm vào config.go:

	// cmd/worker.
	WorkerMaxAttempts int `env:"WORKER_MAX_ATTEMPTS" envDefault:"4"`
	WorkerPollMax     int `env:"WORKER_POLL_MAX" envDefault:"500"`

// ValidateWorker kiểm những gì riêng cmd/worker cần.
func (c Config) ValidateWorker() error {
	if len(c.KafkaBrokers) == 0 {
		return errors.New("config: cmd/worker cần KAFKA_BROKERS")
	}
	if c.WorkerMaxAttempts <= 0 {
		return errors.New("config: WORKER_MAX_ATTEMPTS phải lớn hơn 0")
	}
	if c.WorkerPollMax <= 0 {
		return errors.New("config: WORKER_POLL_MAX phải lớn hơn 0")
	}
	return nil
}

Và lắp stats vào cmd/api/main.go để có endpoint đọc — chỉ một dòng, và nó không cần Kafka:

	identity.New(pool, txm, tokens, tokens, log).RegisterHTTP(r)
	stats.New(pool, txm, idempotency.NewGuard(), log).RegisterHTTP(r)   // ← THÊM

guard truyền vào cmd/api nhưng không bao giờ được dùng — vì api chỉ gọi RegisterHTTP, còn guard chỉ nằm trong đường RegisterEvents. Trông hơi thừa nhưng đây là cái giá rẻ nhất trong ba lựa chọn: tách Module thành ReadModule/EventModule là bịa thêm hai kiểu cho một vấn đề không có; để New nhận nil rồi tin rằng không ai gọi sai là mời gọi panic. Một tham số không dùng, tường minh, dễ hiểu hơn cả hai.

7. `platform/cleanup` — trả nợ hai job dọn dẹp

Hai bảng hạ tầng lớn lên mãi nếu không dọn. Bước 5 §13.3 và bước 3 §12 đều hoãn việc này tới bước 6. cmd/worker là chỗ đúng: nó là tiến trình chạy dài, không phục vụ request nào, nên một ticker ở đây không ảnh hưởng ai.

internal/platform/cleanup/cleanup.go:

// Package cleanup dọn định kỳ hai bảng hạ tầng.
package cleanup

import (
	"context"
	"log/slog"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
)

type Config struct {
	Every time.Duration

	// OutboxRetention: giữ event ĐÃ GỬI để còn điều tra (ARCHITECTURE §7.6).
	OutboxRetention time.Duration

	// ConsumedRetention PHẢI dài hơn retention của Kafka. Xem ghi chú dưới.
	ConsumedRetention time.Duration
}

func (c *Config) setDefaults() {
	if c.Every <= 0 {
		c.Every = time.Hour
	}
	if c.OutboxRetention <= 0 {
		c.OutboxRetention = 7 * 24 * time.Hour
	}
	if c.ConsumedRetention <= 0 {
		c.ConsumedRetention = 30 * 24 * time.Hour
	}
}

func Run(ctx context.Context, pool *pgxpool.Pool, log *slog.Logger, cfg Config) error {
	cfg.setDefaults()

	t := time.NewTicker(cfg.Every)
	defer t.Stop()

	for {
		select {
		case <-ctx.Done():
			return nil
		case <-t.C:
		}

		// ⚠️ CHỈ xoá dòng đã gửi. Dòng dead_at là event chưa từng rời khỏi
		// hệ thống — xoá tự động là mất dữ liệu thật (bước 5 §13.3).
		tag, err := pool.Exec(ctx, `
			DELETE FROM outbox
			WHERE published_at IS NOT NULL AND published_at < $1`,
			time.Now().Add(-cfg.OutboxRetention))
		if err != nil {
			log.Error("cleanup: dọn outbox thất bại", "err", err)
		} else if n := tag.RowsAffected(); n > 0 {
			log.Info("cleanup: đã dọn outbox", "rows", n)
		}

		tag, err = pool.Exec(ctx, `
			DELETE FROM consumed_events WHERE processed_at < $1`,
			time.Now().Add(-cfg.ConsumedRetention))
		if err != nil {
			log.Error("cleanup: dọn consumed_events thất bại", "err", err)
		} else if n := tag.RowsAffected(); n > 0 {
			log.Info("cleanup: đã dọn consumed_events", "rows", n)
		}
	}
}

⚠️ Vì sao consumed_events giữ 30 ngày mà Kafka chỉ 7

Bước 3 §6.3 đã nêu con số; đây là lý do đầy đủ, và nó là một trong những chỗ dễ tự bắn vào chân nhất.

Quy tắc: thời gian giữ consumed_events phải dài hơn khoảng thời gian Kafka còn có thể giao lại một message.

Nếu giữ 3 ngày trong khi Kafka giữ 7: một consumer phải reset offset về đầu topic (chuyện xảy ra khi sửa bug hoặc rebuild) sẽ đọc lại đủ 7 ngày. Với những event của ngày thứ 4–7, guard không còn bản ghi nào → nó nói "lần đầu" → mọi counter cộng thêm một lần nữa. Bạn vừa dùng một cơ chế chống-trùng để tạo ra dữ liệu trùng.

30 ngày cho khoảng đệm rộng mà bảng vẫn nhỏ — mỗi dòng vài chục byte.

Và nếu bạn tăng retention.ms của Kafka lên 30 ngày, phải tăng con số này lên trước. Thứ tự quan trọng: tăng vế phòng thủ trước, tăng vế rủi ro sau.

Thêm vào Makefile và dev.ps1:

worker:                   ## Chạy Kafka consumer
	go run ./cmd/worker

    "worker" { go run ./cmd/worker }

8. Chạy thử đầu-cuối

8.1 Chuẩn bị

.\dev.ps1 up
.\dev.ps1 topics
sqlc generate
go build ./...

8.2 Bốn terminal

Terminal	Lệnh
1	`.\dev.ps1 api`
2	`.\dev.ps1 relay`
3	`.\dev.ps1 worker`
4	gọi HTTP

Terminal 3 phải in:

{"level":"INFO","msg":"kafka: consumer bắt đầu","group":"stats-service","topics":["identity"]}
{"level":"INFO","msg":"kafka: nhận partition","group":"stats-service","parts":{"identity":[0,1,2]}}

Dòng thứ hai là bằng chứng consumer group đã hoạt động: nó nhận cả ba partition vì đang là thành viên duy nhất.

Nếu bảng outbox còn event từ các bước trước, worker sẽ xử lý chúng ngay bây giờ — ConsumeResetOffset(AtStart()) cho group mới đọc lại từ đầu topic (§4.4). Đó là hành vi đúng, và nó cũng là lần đầu bạn thấy guard làm việc thật.

8.3 Đăng ký

$body = @{
    username = "hoangnh6"
    name     = "Nguyễn Huy Hoàng"
    email    = "hoangnh6@example.com"
    password = "matkhau-rat-dai-va-an-toan"
} | ConvertTo-Json

$res = Invoke-RestMethod -Uri http://localhost:8000/auth/register `
    -Method Post -ContentType "application/json; charset=utf-8" `
    -Body ([Text.Encoding]::UTF8.GetBytes($body)) `
    -Headers @{ "X-Request-ID" = "thu-nghiem-worker-001" }

$userId = $res.user.id
$userId

8.4 Theo dõi sợi chỉ chạy qua ba tiến trình

Terminal 2 (relay) — trong vòng 1 giây:

{"level":"INFO","msg":"outbox: đã gửi","count":1,"topics":1}

Terminal 3 (worker) — ngay sau đó, không có log INFO nào (xử lý thành công thì im lặng). Xác nhận bằng database:

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "SELECT event_id, consumer, processed_at FROM consumed_events ORDER BY processed_at DESC LIMIT 3;"

               event_id               |    consumer    |         processed_at
--------------------------------------+----------------+------------------------------
 9c1e....-....-....-....-............ | stats-service  | 2026-08-13 09:31:22.61+07

8.5 ★ Số liệu đã tự xuất hiện

Invoke-RestMethod -Uri "http://localhost:8000/users/$userId/stats"

user_id               : 3f2a...
posts_count           : 0
total_post_views      : 0
followers_count       : 0
following_users_count : 0
bookmarks_count       : 0
updated_at            : 2026-08-13T09:31:22.61+07:00

Dòng này không do bất kỳ dòng code nào trong identity tạo ra. Nó có mặt vì stats nghe được một event và tự quyết định làm gì. Đó là toàn bộ điều mà năm bài trước đang xây tới.

Kiểm chứng Luật 1 bằng trình biên dịch, không bằng niềm tin:

go list -deps ./internal/modules/identity/... | Select-String "modules/stats"
go list -deps ./internal/modules/stats/...    | Select-String "modules/identity"

Cả hai lệnh không ra dòng nào. Hai module hoàn toàn không biết nhau, nhưng dữ liệu vẫn chảy được — qua contracts, thứ mà cả hai đều phụ thuộc và không ai sở hữu.

Đáng đưa hai lệnh này vào CI. ARCHITECTURE.md §13.4 nói về ý này; giờ nó đã có đối tượng thật để kiểm.

9. Ba bài diễn tập sự cố

9.1 ★ Giao trùng — bài quan trọng nhất

Đây là bài chứng minh giá trị của cả mục 3. Ta buộc Kafka giao lại bằng cách xoá offset của group:

# Worker phải DỪNG trước, nếu không lệnh reset bị từ chối.
# Ctrl+C ở terminal 3.

docker compose -f deployments/docker-compose.yml exec kafka /opt/kafka/bin/kafka-consumer-groups.sh `
    --bootstrap-server localhost:9092 --group stats-service `
    --reset-offsets --to-earliest --topic identity --execute

Ghi lại trạng thái trước:

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "SELECT count(*) AS so_dong, max(updated_at) AS cap_nhat_cuoi FROM user_stats;"

Chạy lại worker (.\dev.ps1 worker). Nó đọc lại toàn bộ topic từ đầu. Log sẽ đầy:

{"level":"DEBUG","msg":"stats: bỏ qua event trùng","event_id":"9c1e..."}

(Cần LOG_LEVEL=debug để thấy dòng này.)

So lại:

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "SELECT count(*) AS so_dong, max(updated_at) AS cap_nhat_cuoi FROM user_stats;"

so_dong không đổi. cap_nhat_cuoi không đổi. Mọi event đã được xử lý lại, và không có gì thay đổi — đó chính là định nghĩa của idempotent.

Vì sao bài này quan trọng hơn nó trông

Với EnsureUserStats, ON CONFLICT DO NOTHING một mình đã đủ, nên bài diễn tập này chưa cho thấy guard cứu được gì. Giá trị của nó là kiểm chứng đường ống: xác nhận event_id đi qua bốn chặng (outbox → Kafka → envelope → consumed_events) mà không bị đổi ở đâu.

Nếu event_id bị sinh lại ở một chặng nào đó — ví dụ relay tạo UUID mới thay vì dùng UUID trong bảng — guard sẽ không bao giờ phát hiện trùng, và bài diễn tập này là cách duy nhất thấy được điều đó. Ngày module post thêm posts_count + 1, đường ống đã được chứng minh sẵn.

Muốn thấy hậu quả thật ngay bây giờ: comment tạm dòng if !ok { return nil }, chạy lại bài này với một handler tăng counter, và xem con số nhân đôi.

Đừng quên đưa LOG_LEVEL về info sau khi xong.

9.2 Message độc → DLQ

Chèn một envelope hỏng thẳng vào topic. Payload không giải mã được thành v1.UserRegistered hợp lệ:

$bad = '{"event_id":"11111111-1111-1111-1111-111111111111","event_type":"identity.user.registered.v1","aggregate_id":"x","occurred_at":"2026-08-13T00:00:00Z","payload":{"user_id":"khong-phai-uuid"}}'

$bad | docker compose -f deployments/docker-compose.yml exec -T kafka /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic identity

Log worker:

{"level":"WARN","msg":"lỗi vĩnh viễn, vào DLQ ngay","event_id":"11111111-...","err":"permanent: stats: user_id không phải UUID (\"khong-phai-uuid\")..."}
{"level":"ERROR","msg":"kafka: đã chuyển vào DLQ","dlq_topic":"identity.DLQ","attempts":0,...}

Hai điều đáng thấy:

attempts: 0 — không thử lại lần nào. eventbus.Permanent từ bước 2 §6.1 làm việc: một UUID sai định dạng thì thử lại 4 lần vẫn sai.
Worker tiếp tục chạy bình thường. Đăng ký một user mới ngay lúc này và nó vẫn được xử lý — partition không bị chặn.

Xem message trong DLQ, kèm header chẩn đoán:

docker compose -f deployments/docker-compose.yml exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic identity.DLQ --from-beginning --property print.headers=true

Hoặc Kafka UI → Topics → identity.DLQ → Messages. Header dlq_error, dlq_original_offset, dlq_consumer_group cho bạn đủ thông tin để biết ai hỏng ở đâu vì sao — không phải đi đoán.

9.3 Worker chết giữa lô

# Trong lúc worker đang xử lý, nhấn Ctrl+C ở terminal 3.

Kiểm tra offset đã commit so với offset cuối của topic:

docker compose -f deployments/docker-compose.yml exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group stats-service --describe

GROUP          TOPIC     PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
stats-service  identity  0          4               4               0
stats-service  identity  1          2               2               0
stats-service  identity  2          3               3               0

LAG = 0 nghĩa là mọi thứ đã xử lý xong. Nếu bạn tắt đúng lúc đang xử lý dở, LAG sẽ dương — và khi bật lại, phần chưa xong được giao lại (có thể kèm vài message đã xử lý, mà guard sẽ khử).

LAG là chỉ số quan trọng nhất của tầng consumer. Lag tăng đều nghĩa là worker không theo kịp hoặc đã chết; lag = 0 nghĩa là hệ thống đang bắt kịp thời gian thực.

10. Kiểm chứng bằng test

10.1 Guard — cần Postgres thật

internal/platform/idempotency/guard_test.go:

func TestMarkLanDauTraTrueLanSauTraFalse(t *testing.T) {
	pool := testPool(t)          // helper từ bước 2 §8.3
	txm := postgres.NewTxManager(pool)
	g := idempotency.NewGuard()
	id := uuid.NewString()

	// Lần đầu.
	require.NoError(t, txm.Do(context.Background(), func(tx pgx.Tx) error {
		ok, err := g.Mark(context.Background(), tx, id, "test-consumer")
		require.NoError(t, err)
		require.True(t, ok)
		return nil
	}))

	// Lần hai, cùng consumer.
	require.NoError(t, txm.Do(context.Background(), func(tx pgx.Tx) error {
		ok, err := g.Mark(context.Background(), tx, id, "test-consumer")
		require.NoError(t, err)
		require.False(t, ok, "lần thứ hai phải bị chặn")
		return nil
	}))
}

func TestHaiConsumerKhacNhauDeuXuLyDuoc(t *testing.T) {
	pool := testPool(t)
	txm := postgres.NewTxManager(pool)
	g := idempotency.NewGuard()
	id := uuid.NewString()

	for _, consumer := range []string{"stats-service", "search-indexer"} {
		require.NoError(t, txm.Do(context.Background(), func(tx pgx.Tx) error {
			ok, err := g.Mark(context.Background(), tx, id, consumer)
			require.NoError(t, err)
			require.True(t, ok, "%s phải được xử lý độc lập", consumer)
			return nil
		}))
	}
}

// ★ Test này canh giữ chính lỗi thiết kế mà §3.2 cảnh báo.
func TestRollbackXoaLuonDauDaDanh(t *testing.T) {
	pool := testPool(t)
	txm := postgres.NewTxManager(pool)
	g := idempotency.NewGuard()
	id := uuid.NewString()
	boom := errors.New("việc thật thất bại")

	err := txm.Do(context.Background(), func(tx pgx.Tx) error {
		ok, err := g.Mark(context.Background(), tx, id, "test-consumer")
		require.NoError(t, err)
		require.True(t, ok)
		return boom // ROLLBACK
	})
	require.ErrorIs(t, err, boom)

	// Sau rollback, event phải được coi như CHƯA xử lý.
	require.NoError(t, txm.Do(context.Background(), func(tx pgx.Tx) error {
		ok, err := g.Mark(context.Background(), tx, id, "test-consumer")
		require.NoError(t, err)
		require.True(t, ok, "rollback phải xoá dấu, nếu không event bị bỏ rơi vĩnh viễn")
		return nil
	}))
}

Test thứ ba là test có giá trị nhất trong nhóm. Nó khẳng định điều làm nên toàn bộ ý nghĩa của việc truyền tx: dấu "đã xử lý" và công việc thật cùng sống cùng chết. Nếu ai đó "tối ưu" Mark để dùng pool thay vì tx, chỉ test này đỏ — hai test trên vẫn xanh.

10.2 `TestHaiConsumerKhacNhauDeuXuLyDuoc` canh giữ điều gì

Nó là bản dịch của bước 3 §6.1 thành code. Nếu ai đó đổi khoá chính thành PRIMARY KEY (event_id) — nghe hợp lý, ngắn hơn, "vẫn chống trùng được" — test này đỏ ngay. Không có nó, hậu quả là consumer group thứ hai âm thầm bỏ qua mọi event mà group thứ nhất đã xử lý, và không có gì báo lỗi.

10.3 Subscriber — không cần Kafka, không cần Postgres

Ba interface khai ở §5.2 cho phép test thuần bộ nhớ:

type fakeGuard struct {
	seen map[string]bool
}

func (f *fakeGuard) Mark(_ context.Context, _ pgx.Tx, eventID, _ string) (bool, error) {
	if f.seen[eventID] {
		return false, nil
	}
	if f.seen == nil {
		f.seen = map[string]bool{}
	}
	f.seen[eventID] = true
	return true, nil
}

type fakeRepo struct{ ensured []uuid.UUID }

func (f *fakeRepo) EnsureUserStats(_ context.Context, _ pgx.Tx, id uuid.UUID) error {
	f.ensured = append(f.ensured, id)
	return nil
}

// fakeTx gọi fn với tx = nil. Được, vì các fake không dùng tới nó.
type fakeTx struct{}

func (fakeTx) Do(ctx context.Context, fn func(pgx.Tx) error) error { return fn(nil) }

func TestEventTrungChiGhiMotLan(t *testing.T) {
	repo := &fakeRepo{}
	sub := subscriber.New(fakeTx{}, &fakeGuard{}, repo, testLogger())

	userID := uuid.NewString()
	env := mustEnvelope(t, v1.TypeUserRegistered, userID, v1.UserRegistered{
		UserID: userID, Username: "a", Email: "a@b.c",
	})

	require.NoError(t, sub.OnUserRegistered(context.Background(), env))
	require.NoError(t, sub.OnUserRegistered(context.Background(), env)) // giao lại

	require.Len(t, repo.ensured, 1, "event trùng không được ghi lần thứ hai")
}

func TestPayloadHongLaLoiVinhVien(t *testing.T) {
	sub := subscriber.New(fakeTx{}, &fakeGuard{}, &fakeRepo{}, testLogger())

	env := mustEnvelope(t, v1.TypeUserRegistered, "agg", map[string]string{
		"user_id": "khong-phai-uuid",
	})

	err := sub.OnUserRegistered(context.Background(), env)
	require.Error(t, err)
	require.True(t, eventbus.IsPermanent(err),
		"phải là lỗi vĩnh viễn, nếu không message độc sẽ chặn partition")
}

Test thứ hai chặn đúng cái poison message ở §1.3. Nếu ai đó sửa eventbus.Permanent(...) thành fmt.Errorf(...) — trông vô hại — một message có user_id sai sẽ bị thử lại 4 lần rồi vẫn vào DLQ, nên hậu quả nhẹ. Nhưng nếu cùng lúc MaxAttempts bị đặt thành 0 để "thử lại vô hạn cho chắc", partition tắc vĩnh viễn. Test này giữ nửa đầu của hàng rào đó.

10.4 Route bỏ qua event lạ

func TestEventTypeLaKhongLamGayConsumer(t *testing.T) {
	repo := &fakeRepo{}
	sub := subscriber.New(fakeTx{}, &fakeGuard{}, repo, testLogger())

	h := eventbus.Route(map[string]eventbus.Handler{
		v1.TypeUserRegistered: sub.OnUserRegistered,
	})

	env := mustEnvelope(t, "identity.user.banned.v99", "agg", map[string]string{})

	require.NoError(t, h(context.Background(), env),
		"event lạ phải được bỏ qua êm, không phải lỗi")
	require.Empty(t, repo.ensured)
}

Test này bảo vệ lời hứa lớn nhất của kiến trúc: thêm event mới không bao giờ làm hỏng consumer cũ (bước 2 §6.3). Nó nhỏ và trông tầm thường, nhưng nếu Route bị sửa thành trả lỗi cho type lạ, mọi consumer trong hệ thống sẽ gãy vào ngày ai đó thêm một event — và nguyên nhân nằm ở module khác với nơi phát sinh sự cố.

10.5 Chạy

go test ./internal/... -race -count=1

11. Vận hành

11.1 Ba chỉ số của tầng consumer

# 1. Consumer lag — chỉ số số 1. Cảnh báo khi > 10.000 (ARCHITECTURE §11.4).
docker compose -f deployments/docker-compose.yml exec kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --all-groups

# 2. DLQ có gì chưa. Cảnh báo khi > 0.
docker compose -f deployments/docker-compose.yml exec kafka /opt/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --bootstrap-server localhost:9092 --topic identity.DLQ

-- 3. Consumer có đang làm việc không (nhìn từ phía database).
SELECT consumer, count(*) AS da_xu_ly, max(processed_at) AS gan_nhat
FROM consumed_events GROUP BY consumer ORDER BY gan_nhat DESC;

Câu thứ ba trả lời được câu hỏi mà lag không trả lời được: "consumer còn sống nhưng có đang làm gì không?" Một consumer bị kẹt trong vòng rebalance (§4.5) vẫn hiện là thành viên của group, vẫn có lag, nhưng gan_nhat sẽ đứng im. Đó là dấu hiệu rõ nhất của chính sự cố khó chẩn đoán nhất trong tài liệu này.

11.2 Phát lại message từ DLQ

Sau khi đã sửa nguyên nhân — thứ tự này quan trọng, giống bước 5 §13.2:

# Xem trước đã.
docker compose -f deployments/docker-compose.yml exec kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic identity.DLQ --from-beginning --property print.headers=true --max-messages 10

# Bơm ngược về topic gốc. Value giữ nguyên byte nên copy thẳng là được —
# đó là lý do §4.6 không bọc thêm lớp nào quanh Value.
docker compose -f deployments/docker-compose.yml exec kafka bash -c "\
  /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic identity.DLQ --from-beginning --timeout-ms 5000 2>/dev/null | \
  /opt/kafka/bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic identity"

⚠️ Lệnh trên mất Key (console consumer không in key theo mặc định, và console producer không đọc key trừ khi bật parse.key). Mất Key nghĩa là message được rải sang partition khác → mất bảo đảm thứ tự. Chấp nhận được cho việc phát lại thủ công một lần; nếu bạn phát lại thường xuyên, hãy viết một cmd/dlq-replay nhỏ bằng Go dùng lại kafka.Publisher và giữ nguyên cả Key lẫn header. Việc đó nằm trong §12.

11.3 Bao nhiêu worker?

Khác với relay, worker scale được thoải mái — và đây là điểm khác biệt cốt lõi giữa hai tiến trình:

	`cmd/relay`	`cmd/worker`
Thứ tự do đâu bảo đảm	`ORDER BY id` với một instance (bước 5 phụ lục A)	Kafka partition — mỗi partition chỉ một consumer trong group
Số instance nên chạy	1	Tới bằng số partition (3)
Thêm instance thì sao	Nguy cơ hỏng thứ tự	Group tự rebalance, chia partition

Chạy worker thứ hai và xem group tự chia việc:

# Terminal 5
.\dev.ps1 worker

Log của cả hai worker sẽ in nhả partition rồi nhận partition, và ba partition được chia đôi (2 + 1). Đây là consumer group hoạt động đúng như thiết kế — và cũng là lý do OnPartitionsRevoked ở §4.3 phải commit trước khi nhả.

Worker thứ tư sẽ nằm không, vì chỉ có 3 partition. Số partition là trần song song của một consumer group; muốn hơn thì phải tăng partition (tăng được, không giảm được — bước 1 §6).

12. Những thứ cố ý chưa có

Chưa có	Vì sao	Sẽ làm ở
Module `post` (bảng `posts`, `000006`)	Bước tiếp theo tự nhiên. Nó sẽ là bài kiểm tra thật của guard, vì `posts_count + 1` không tự idempotent	Bước 7
`cmd/dlq-replay`	Cần Go để giữ Key và header (§11.2). Chỉ đáng viết khi DLQ thật sự có khách	Khi cần
Xử lý song song nhiều partition	Bản tuần tự đủ cho ~2.000 event/giây. Song song đổi bảo đảm thứ tự, nên phải là quyết định có số đo (Phụ lục A)	Khi có số đo
Retry topic	Ngân sách 700ms đủ cho lỗi tạm thời hiện có. Chưa có handler nào gọi dịch vụ ngoài (Phụ lục B)	Khi có handler gọi API ngoài
Prometheus metrics, `/healthz` cho worker	Cần chọn hệ metric — quyết định riêng. Ba câu ở §11.1 trả lời cùng câu hỏi	Khi dựng giám sát
`identity/subscriber/`	`identity` chưa cần nghe event của ai. Cây thư mục ở `ARCHITECTURE.md` §3 có sẵn chỗ, chưa có nội dung	Khi có nhu cầu thật
Dockerfile cho ba binary	Đóng gói cả ba cùng lúc, một `Dockerfile` nhiều target	Khi triển khai

13. Checklist hoàn thành

[ ] sqlc generate sạch; gen có MarkConsumed, EnsureUserStats, GetUserStats
[ ] go build ./... và go vet ./... sạch
[ ] go test ./internal/... -race -count=1 xanh
[ ] .\dev.ps1 worker in consumer bắt đầu và nhận partition
[ ] Đăng ký user → consumed_events có một dòng với consumer = 'stats-service'
[ ] GET /users/{id}/stats trả 200 với 8 counter bằng 0
[ ] go list -deps xác nhận identity ⇎ stats không import nhau
[ ] Diễn tập §9.1: reset offset → đọc lại toàn bộ → user_stats không đổi
[ ] Diễn tập §9.2: envelope hỏng vào identity.DLQ với attempts: 0, worker vẫn chạy
[ ] Diễn tập §9.3: kafka-consumer-groups --describe cho LAG = 0
[ ] Chạy worker thứ hai → thấy nhả partition / nhận partition ở cả hai
[ ] DisableAutoCommit và BlockRebalanceOnPoll đều có mặt trong consumer.go

Cấu trúc thư mục lúc này:

project/
├── cmd/
│   ├── api/           (+ stats.RegisterHTTP)
│   ├── relay/
│   ├── worker/        ← MỚI
│   └── smoketest/
├── internal/
│   ├── contracts/
│   ├── modules/
│   │   ├── identity/
│   │   └── stats/                  ← MỚI (trọn module)
│   │       ├── module.go
│   │       ├── domain/stats.go
│   │       ├── repository/
│   │       ├── subscriber/         ← trái tim module này
│   │       └── transport/http/
│   └── platform/
│       ├── cleanup/                ← MỚI
│       ├── idempotency/guard.go    ← MỚI
│       └── eventbus/kafka/
│           ├── publisher.go
│           └── consumer.go         ← MỚI
└── db/migrations/     (không thêm gì)

Commit:

git add .
git commit -m "feat(worker): consumer group, idempotency guard, DLQ và module stats"

Phụ lục A — Xử lý song song nhiều partition

EachPartition ở §4.3 xử lý tuần tự: partition 0 xong mới sang 1. Với lô 500 record và mỗi record 2ms là 1 giây — chấp nhận được, nhưng có thể nhanh gấp ba.

Vì thứ tự chỉ được bảo đảm trong phạm vi một partition, chạy các partition song song là an toàn:

var g errgroup.Group
var mu sync.Mutex
var done []*kgo.Record

fetches.EachPartition(func(p kgo.FetchTopicPartition) {
	g.Go(func() error {
		var local []*kgo.Record
		for _, rec := range p.Records {
			if err := c.handleRecord(ctx, group, routes[rec.Topic], rec); err != nil {
				break // dừng ĐÚNG partition này, các partition khác chạy tiếp
			}
			local = append(local, rec)
		}
		mu.Lock()
		done = append(done, local...)
		mu.Unlock()
		return nil
	})
})
_ = g.Wait()   // PHẢI chờ xong hết mới được commit

Ba điều bắt buộc, thiếu một là sai:

Tuần tự bên trong một partition. Song song trong partition là phá bỏ đúng thứ tự mà aggregate_id làm partition key đã mua được ở bước 5 §6.3.
g.Wait() trước CommitRecords. Commit khi còn goroutine đang chạy là commit cho việc chưa xong.
Ngân sách thời gian không đổi. Song song không nới RebalanceTimeout; nó chỉ làm trường hợp xấu nhất bằng partition chậm nhất thay vì tổng.

Đổi lại: khó debug hơn, và một partition chậm không còn hiện rõ trong log. Chỉ làm khi có số đo cho thấy vòng lặp tuần tự là nút thắt.

Phụ lục B — Retry topic: khi 700ms không đủ

Ngân sách ở §4.5 chỉ đủ cho lỗi tạm thời rất ngắn. Ngày bạn có handler gọi dịch vụ ngoài — gửi mail, gọi payment — và dịch vụ đó sập 10 phút, mọi message trong khoảng đó sẽ vào DLQ dù chẳng có gì hỏng.

Cách giải chuẩn: chuyển thời gian chờ ra khỏi vòng lặp poll.

identity ──lỗi tạm thời──→ identity.retry.5s ──→ (consumer riêng, chờ rồi bơm lại)
                                    │ vẫn lỗi
                                    ↓
                          identity.retry.1m ──→ ...
                                    │ vẫn lỗi
                                    ↓
                              identity.DLQ

Mỗi topic retry có consumer riêng, và consumer đó được phép chờ lâu vì nó không giữ partition của topic chính. Consumer chính không bao giờ chặn quá 700ms.

Cái giá phải nói rõ: message đi qua retry topic mất thứ tự so với topic gốc. Message A vào retry rồi quay lại sẽ tới sau message B của cùng aggregate. Với việc như gửi mail thì không sao; với việc phụ thuộc thứ tự (đổi tên rồi xoá) thì đây là lỗi.

Nên: retry topic cho handler có tác dụng phụ ra ngoài; retry tại chỗ cho handler chỉ ghi database. Hai loại handler khác nhau, hai cơ chế khác nhau — đừng dùng một cơ chế cho cả hai.

Phụ lục C — Tính lại số liệu từ đầu

user_stats là read model: mọi con số trong đó đều suy ra được từ bảng gốc. Nghĩa là khi số liệu lệch — vì một bug, vì một event bị bỏ qua, vì ai đó sửa tay database — bạn không phải đi tìm chỗ lệch, chỉ cần tính lại.

Hai cách, và cách thứ hai đáng tin hơn:

1. Tính lại từ bảng gốc. RecalculatePostsCount ở bước 3 §9.5 (đang comment vì chưa có bảng posts) đọc thẳng nguồn sự thật:

UPDATE user_stats us
SET posts_count = (SELECT count(*) FROM posts WHERE author_id = us.user_id),
    updated_at  = now();

Nhanh, chính xác, nhưng chỉ dùng được với counter có bảng gốc. total_post_views không có bảng gốc nào — không ai lưu từng lượt xem.

2. Phát lại toàn bộ event. Đây là lúc bạn nhận được cổ tức lớn nhất của kiến trúc event-driven:

# Xoá read model và lịch sử chống trùng của riêng consumer này.
docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "
  TRUNCATE user_stats;
  DELETE FROM consumed_events WHERE consumer = 'stats-service';"

# Đọc lại từ đầu topic.
docker compose -f deployments/docker-compose.yml exec kafka /opt/kafka/bin/kafka-consumer-groups.sh `
    --bootstrap-server localhost:9092 --group stats-service `
    --reset-offsets --to-earliest --all-topics --execute

Bật worker lại và số liệu được dựng lại từ chính chuỗi event đã tạo ra nó. Không có bước nào phải đoán.

Hai giới hạn phải biết trước khi tin vào cách này:

Chỉ dựng lại được phần lịch sử Kafka còn giữ (7 ngày). Muốn dựng lại từ ngày đầu tiên thì topic phải để retention.ms = -1, hoặc phát lại từ bảng outbox — nhưng outbox cũng chỉ giữ 7 ngày. Nếu khả năng dựng lại toàn bộ là yêu cầu thật, đó là một quyết định về retention phải ra trước, không phải sau khi cần tới.
DELETE FROM consumed_events WHERE consumer = ... phải đúng một consumer. Xoá cả bảng là mở cửa cho mọi consumer khác xử lý trùng (§3.4).

Bước tiếp theo

Xương sống kiến trúc đã xong. Từ đây mọi thứ là lặp lại đúng mẫu này, không sửa hạ tầng nữa.

Bước 7 — Module post là bước tự nhiên, và nó là bài kiểm tra thật đầu tiên của mọi thứ vừa xây:

Migration 000006_create_posts + module post đầy đủ (CRUD, slug, read_time)
Phát post.post.created.v1 và post.post.deleted.v1 qua outbox
stats.OnPostCreated gọi IncrementPostsCount — posts_count + 1 không tự idempotent, nên đây là lần đầu guard thật sự cứu bạn khỏi một con số sai
Mở lại RecalculatePostsCount đã comment từ bước 3

Ba việc sẽ chứng minh kiến trúc đứng được: thêm một module mà không sửa platform/, thêm một consumer mà không sửa stats cũ, và thêm một topic mà không sửa relay.

Nếu cả ba đều đúng, bạn có một modular monolith thật — thứ có thể tách thành microservice sau này bằng cách đổi eventbus, không phải viết lại nghiệp vụ.

Backend

#10 Bước 6 — Consumer, Idempotency và DLQ Phần 2

5.5 `transport/http` — endpoint đọc

404 ở đây nghĩa là gì — và vì sao nó không sai

6. `cmd/worker/main.go`

7. `platform/cleanup` — trả nợ hai job dọn dẹp

⚠️ Vì sao `consumed_events` giữ 30 ngày mà Kafka chỉ 7

8. Chạy thử đầu-cuối

8.1 Chuẩn bị

8.2 Bốn terminal

8.3 Đăng ký

8.4 Theo dõi sợi chỉ chạy qua ba tiến trình

8.5 ★ Số liệu đã tự xuất hiện

9. Ba bài diễn tập sự cố

9.1 ★ Giao trùng — bài quan trọng nhất

Vì sao bài này quan trọng hơn nó trông

9.2 Message độc → DLQ

9.3 Worker chết giữa lô

10. Kiểm chứng bằng test

10.1 Guard — cần Postgres thật

10.2 `TestHaiConsumerKhacNhauDeuXuLyDuoc` canh giữ điều gì

10.3 Subscriber — không cần Kafka, không cần Postgres

10.4 Route bỏ qua event lạ

10.5 Chạy

11. Vận hành

11.1 Ba chỉ số của tầng consumer

11.2 Phát lại message từ DLQ

11.3 Bao nhiêu worker?

12. Những thứ cố ý chưa có

13. Checklist hoàn thành

Phụ lục A — Xử lý song song nhiều partition

Phụ lục B — Retry topic: khi 700ms không đủ

Phụ lục C — Tính lại số liệu từ đầu

Bước tiếp theo

Mục lục

5.5 transport/http — endpoint đọc

404 ở đây nghĩa là gì — và vì sao nó không sai

6. cmd/worker/main.go

7. platform/cleanup — trả nợ hai job dọn dẹp

⚠️ Vì sao consumed_events giữ 30 ngày mà Kafka chỉ 7

8. Chạy thử đầu-cuối

8.1 Chuẩn bị

8.2 Bốn terminal

8.3 Đăng ký

8.4 Theo dõi sợi chỉ chạy qua ba tiến trình

8.5 ★ Số liệu đã tự xuất hiện

9. Ba bài diễn tập sự cố

9.1 ★ Giao trùng — bài quan trọng nhất

Vì sao bài này quan trọng hơn nó trông

9.2 Message độc → DLQ

9.3 Worker chết giữa lô

10. Kiểm chứng bằng test

10.1 Guard — cần Postgres thật

10.2 TestHaiConsumerKhacNhauDeuXuLyDuoc canh giữ điều gì

10.3 Subscriber — không cần Kafka, không cần Postgres

10.4 Route bỏ qua event lạ

10.5 Chạy

11. Vận hành

11.1 Ba chỉ số của tầng consumer

11.2 Phát lại message từ DLQ

11.3 Bao nhiêu worker?

12. Những thứ cố ý chưa có

13. Checklist hoàn thành

Phụ lục A — Xử lý song song nhiều partition

Phụ lục B — Retry topic: khi 700ms không đủ

Phụ lục C — Tính lại số liệu từ đầu

Bước tiếp theo

Mục lục

5.5 `transport/http` — endpoint đọc

6. `cmd/worker/main.go`

7. `platform/cleanup` — trả nợ hai job dọn dẹp

⚠️ Vì sao `consumed_events` giữ 30 ngày mà Kafka chỉ 7

10.2 `TestHaiConsumerKhacNhauDeuXuLyDuoc` canh giữ điều gì