11.1K 143 952

Đã đăng vào khoảng 3 giờ trước 28 phút đọc

#4 Bước 3 — Database Migrations & Code Gen (sqlc)

Tài liệu thực hành. Mỗi migration chạy xong đều kiểm chứng được ngay bằng một câu SQL.

Yêu cầu: đã hoàn thành Bước 1 và Bước 2 Thời gian: 50–70 phút Kiến trúc tổng thể: ARCHITECTURE.md

Mục tiêu

Việc	Kết quả
4 migration	`users`, `user_stats`, `outbox`, `consumed_events`
`sqlc.yaml`	Cấu hình sinh code cho 4 package
`sqlc generate`	Code Go type-safe, kiểm tra SQL ngay lúc biên dịch

Hai bảng đầu là dữ liệu nghiệp vụ lấy từ note.md. Hai bảng sau là hạ tầng — chúng không phục vụ tính năng nào cho người dùng, mà giữ cho Luật 3 và idempotency thành hiện thực.

Kết thúc bước này, code Go đã đọc ghi được cả bốn bảng. Chưa có service, chưa có HTTP, chưa có consumer.

1. Cài công cụ

go install github.com/sqlc-dev/sqlc/cmd/sqlc@latest
go install -tags 'postgres' github.com/golang-migrate/migrate/v4/cmd/migrate@latest

Hai binary này nằm ở $(go env GOPATH)\bin. Nếu thư mục đó chưa có trong PATH:

# Kiểm tra
sqlc version
migrate -version

# Nếu báo "not recognized" — thêm vĩnh viễn rồi mở lại terminal
$gobin = "$(go env GOPATH)\bin"
[Environment]::SetEnvironmentVariable("PATH", "$env:PATH;$gobin", "User")

<details> <summary>Không muốn cài? Chạy qua Docker</summary>

# sqlc
docker run --rm -v "${PWD}:/src" -w /src sqlc/sqlc generate

# migrate
docker run --rm -v "${PWD}/db/migrations:/migrations" --network host `
  migrate/migrate -path=/migrations `
  -database "postgres://app:secret@localhost:5432/community?sslmode=disable" up

Chậm hơn một chút nhưng không phụ thuộc PATH. Trên Windows, --network host không hoạt động như Linux — dùng host.docker.internal thay cho localhost trong DSN.

</details>

Vì sao -tags 'postgres' khi cài migrate?

golang-migrate hỗ trợ hàng chục loại database, mỗi loại là một driver. Không có build tag, binary sinh ra sẽ kèm tất cả — nặng hơn và kéo theo dependency bạn không dùng. Tag postgres chỉ biên dịch driver bạn cần.

Nếu quên tag, migrate chạy nhưng báo unknown driver postgres — thông báo lỗi rất dễ gây hiểu nhầm là DSN sai.

2. Quy ước migration

New-Item -ItemType Directory -Force db/migrations

Quy tắc đặt tên

{số thứ tự 6 chữ số}_{mô_tả_bằng_snake_case}.{up|down}.sql

Mỗi migration là một cặp:

File	Nội dung
`000001_create_users.up.sql`	Tạo bảng
`000001_create_users.down.sql`	Xoá đúng những gì `.up` đã tạo

Vì sao phải có file .down? golang-migrate chạy được mà không cần nó, nhưng khi đó bạn mất khả năng rollback. Ở môi trường dev, lỗi thường gặp là: viết migration sai → chạy → nhận ra sai → muốn lùi lại một bước nhưng không lùi được, đành down -v xoá sạch database.

File .down mất 30 giây để viết. Lần đầu cần tới nó, nó tiết kiệm cho bạn cả buổi.

Ba luật của migration

Luật 1 — Migration đã chạy trên môi trường khác thì không bao giờ sửa nữa.

golang-migrate ghi vào bảng schema_migrations rằng version 3 đã áp dụng. Bạn sửa nội dung file 000003 rồi chạy migrate up — không có gì xảy ra, vì công cụ chỉ nhìn số version. Máy bạn có schema mới, server có schema cũ, cả hai đều tưởng mình đang ở version 3. Sai lệch này không có cách nào phát hiện tự động.

Cần đổi gì → tạo migration mới.

Luật 2 — Một migration làm một việc.

Migration đổi 5 bảng cùng lúc mà hỏng ở bảng thứ 3 sẽ để lại trạng thái dở dang, khó gỡ.

Luật 3 — Migration phải chạy được trên database đã có dữ liệu.

ALTER TABLE ... ADD COLUMN x INT NOT NULL sẽ thất bại nếu bảng đã có dòng nào, vì các dòng cũ không có giá trị cho x. Phải có DEFAULT, hoặc tách ba bước: thêm cột nullable → điền dữ liệu → thêm ràng buộc.

3. `000001` — users

db/migrations/000001_create_users.up.sql:

CREATE TABLE users (
    id            UUID         PRIMARY KEY DEFAULT gen_random_uuid(),
    name          VARCHAR(255) NOT NULL,
    username      VARCHAR(50)  NOT NULL,
    email         VARCHAR(255) NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    avatar_url    VARCHAR(255),
    reputation    INT          NOT NULL DEFAULT 0,
    created_at    TIMESTAMPTZ  NOT NULL DEFAULT now(),
    updated_at    TIMESTAMPTZ  NOT NULL DEFAULT now(),

    CONSTRAINT users_username_key UNIQUE (username),
    CONSTRAINT users_email_key    UNIQUE (email),
    CONSTRAINT users_reputation_non_negative CHECK (reputation >= 0)
);

COMMENT ON COLUMN users.email IS 'Luôn lưu dạng chữ thường — chuẩn hoá ở tầng service';

db/migrations/000001_create_users.down.sql:

DROP TABLE IF EXISTS users;

3.1 `gen_random_uuid()` — không cần extension

Từ PostgreSQL 13, gen_random_uuid() có sẵn trong core. Các hướng dẫn cũ bảo bạn CREATE EXTENSION pgcrypto — không còn cần nữa (bước 1 dùng PostgreSQL 16).

Việc để DB tự sinh id mặc định không có nghĩa là ứng dụng không được tự sinh. Ở §9.3 bạn sẽ thấy service tự tạo UUID trước khi INSERT — vì nó cần id để dựng event ngay trong cùng transaction. DEFAULT chỉ là lưới an toàn cho những lần INSERT thủ công bằng psql.

3.2 `TIMESTAMPTZ`, không bao giờ `TIMESTAMP`

note.md ghi TIMESTAMP. Đây là một trong những cái bẫy đắt nhất của PostgreSQL, nên tôi đổi có chủ đích:

Kiểu	Lưu gì	Chuyện gì xảy ra
`TIMESTAMP`	Thời điểm không có múi giờ	`10:00` — nhưng 10 giờ ở đâu? Không ai biết
`TIMESTAMPTZ`	Thời điểm tuyệt đối	Lưu chuẩn UTC, tự đổi theo múi giờ của phiên

TIMESTAMP trông vô hại tới ngày server đổi múi giờ, hoặc bạn thêm một server ở vùng khác, hoặc mùa hè đổi giờ. Lúc đó dữ liệu cũ im lặng lệch một giờ và không có cách nào biết dòng nào đã lệch.

TIMESTAMPTZ tốn đúng số byte như TIMESTAMP (8 byte). Không có lý do gì để chọn cái kia.

3.3 Email phân biệt hoa thường — cái bẫy kinh điển

UNIQUE (email) trong PostgreSQL là phân biệt hoa thường. Nghĩa là:

nguyen@example.com
Nguyen@Example.com    ← DB coi là hai người khác nhau

Người dùng đăng ký bằng chữ thường, hôm sau đăng nhập gõ hoa chữ đầu (điện thoại tự viết hoa), và nhận "email không tồn tại".

Ba cách xử lý:

Cách	Đánh giá
Chuẩn hoá về chữ thường ở service trước khi ghi ← chọn cách này	Đơn giản, hiển nhiên, không phụ thuộc extension
`CREATE EXTENSION citext` rồi dùng kiểu `CITEXT`	Chạy đúng nhưng thêm một extension, và `sqlc` cần override kiểu
`CREATE UNIQUE INDEX ON users (lower(email))`	Chạy đúng, nhưng mọi truy vấn phải viết `WHERE lower(email) = ...`, quên một chỗ là index không được dùng

Cách 1 thắng vì nó không có cái bẫy nào ẩn phía sau. Đổi lại, tầng service bắt buộc phải gọi strings.ToLower — dòng COMMENT ON COLUMN ở trên tồn tại để nhắc điều đó cho người đọc schema về sau.

3.4 `VARCHAR(255)` — giữ hay đổi?

Tôi giữ nguyên theo note.md. Trong PostgreSQL, VARCHAR(n) và TEXT lưu trữ giống hệt nhau, không khác gì về hiệu năng; VARCHAR(n) chỉ thêm một phép kiểm tra độ dài.

Điểm đáng lưu ý: giới hạn độ dài là ràng buộc nghiệp vụ, và ràng buộc nghiệp vụ thì hay đổi. Nới VARCHAR(50) lên VARCHAR(100) cần một ALTER TABLE; còn nếu dùng TEXT + kiểm tra ở tầng service thì chỉ cần deploy code.

Với password_hash thì nên cân nhắc thật: bcrypt cho ra 60 ký tự, nhưng nếu mai này đổi sang Argon2id thì chuỗi dài hơn. VARCHAR(255) vẫn dư, nên tạm ổn — chỉ cần nhớ rằng con số này là một quyết định, không phải mặc định.

4. `000002` — user_stats

db/migrations/000002_create_user_stats.up.sql:

CREATE TABLE user_stats (
    user_id               UUID        PRIMARY KEY
                                      REFERENCES users(id) ON DELETE CASCADE,
    total_post_views      INT         NOT NULL DEFAULT 0,
    following_tags_count  INT         NOT NULL DEFAULT 0,
    following_users_count INT         NOT NULL DEFAULT 0,
    followers_count       INT         NOT NULL DEFAULT 0,
    posts_count           INT         NOT NULL DEFAULT 0,
    bookmarks_count       INT         NOT NULL DEFAULT 0,
    questions_count       INT         NOT NULL DEFAULT 0,
    answers_count         INT         NOT NULL DEFAULT 0,
    updated_at            TIMESTAMPTZ NOT NULL DEFAULT now(),

    -- Lưới an toàn: counter âm là dấu hiệu logic sai, phải nổ ngay
    CONSTRAINT user_stats_non_negative CHECK (
        total_post_views      >= 0 AND
        following_tags_count  >= 0 AND
        following_users_count >= 0 AND
        followers_count       >= 0 AND
        posts_count           >= 0 AND
        bookmarks_count       >= 0 AND
        questions_count       >= 0 AND
        answers_count         >= 0
    )
);

COMMENT ON TABLE user_stats IS
    'Bản sao đã tính sẵn. Nguồn sự thật là các bảng gốc. Cập nhật qua event consumer.';

db/migrations/000002_create_user_stats.down.sql:

DROP TABLE IF EXISTS user_stats;

4.1 `user_id` vừa là khoá chính vừa là khoá ngoại

Không có cột id riêng. Quan hệ với users là một-một, nên user_id làm khoá chính luôn:

Không thể có hai dòng thống kê cho cùng một người — DB đảm bảo, không cần code kiểm tra
Ít hơn một index so với thiết kế có id riêng
ON DELETE CASCADE — xoá user thì thống kê tự biến mất, không để lại rác

4.2 `CHECK (>= 0)` phối hợp với `GREATEST` như thế nào

Hai lớp phòng thủ cho cùng một vấn đề, và chúng phục vụ hai mục đích khác nhau:

-- Lớp 1 — trong query: không bao giờ để xuống dưới 0
UPDATE user_stats SET posts_count = GREATEST(posts_count - 1, 0) WHERE user_id = $1;

-- Lớp 2 — trong schema: nếu có query nào QUÊN GREATEST, nó nổ ngay
CONSTRAINT user_stats_non_negative CHECK (posts_count >= 0)

Lớp 1 xử lý luồng bình thường: event xoá bị giao trùng thì counter dừng ở 0 chứ không xuống âm.

Lớp 2 bắt lỗi lập trình. Một query mới viết thiếu GREATEST sẽ làm CHECK thất bại — và bạn thấy ngay khi chạy test, chứ không phải phát hiện sau ba tháng khi có người dùng báo hồ sơ của họ hiện -2 bài viết.

Nhưng CHECK thất bại làm consumer lỗi thì sao? Đúng, và đó là điều mong muốn. Lỗi này là permanent — retry bao nhiêu lần cũng vậy — nên event sẽ vào DLQ và bắn cảnh báo. Sai số liệu ầm ĩ tốt hơn sai số liệu im lặng, vì bạn còn sửa được.

4.3 Dòng `user_stats` được tạo lúc nào?

Đây là câu hỏi phải trả lời trước khi viết consumer, vì UPDATE trên dòng không tồn tại không báo lỗi — nó chỉ ảnh hưởng 0 dòng và trả về thành công.

Chọn: module stats tự tạo dòng của mình khi nghe event identity.user.registered.v1, bằng query EnsureUserStats ở §9.5.

Lý do là Luật 1: user_stats là bảng của stats, nên chỉ stats được ghi vào đó. Phương án ngược lại — để identity ghi cả hai bảng trong một transaction — nghe hợp lý hơn về mặt dữ liệu, nhưng buộc identity phải import stats, và bảng user_stats từ đó có hai module cùng ghi.

Khoá ngoại REFERENCES users(id) vẫn luôn thoả mãn: event chỉ rời bảng outbox sau khi transaction tạo user đã commit, nên tới lúc consumer chạy thì dòng users chắc chắn đã có.

Vì sao không dùng upsert (INSERT ... ON CONFLICT DO UPDATE) trong mỗi lần tăng counter? Vì như thế thì mỗi handler đều phải mang theo trách nhiệm tạo dòng, và không handler nào biết user sắp có hay sẽ không bao giờ có. Một chỗ tạo dòng — handler OnUserRegistered — rõ ràng hơn nhiều.

Đính chính: bản đầu của mục này chọn phương án "identity tạo cả hai dòng trong một transaction". Khi viết Register thật ở bước 4 §18, mâu thuẫn với Luật 1 lộ ra và tôi đã sửa lại. Đổi lại là chấp nhận nhất quán cuối cùng: có một khoảng dưới một giây mà user tồn tại còn dòng thống kê thì chưa — đúng loại dữ liệu mà đánh đổi này sinh ra để phục vụ.

5. `000003` — outbox

db/migrations/000003_create_outbox.up.sql:

CREATE TABLE outbox (
    id             BIGSERIAL    PRIMARY KEY,
    event_id       UUID         NOT NULL,
    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,
    attempts       INT          NOT NULL DEFAULT 0,
    last_error     TEXT,

    CONSTRAINT outbox_event_id_key UNIQUE (event_id)
);

-- 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, index vẫn nhỏ như lúc đầu.
CREATE INDEX idx_outbox_unpublished
    ON outbox (id)
    WHERE published_at IS NULL;

-- Phục vụ job dọn dẹp định kỳ
CREATE INDEX idx_outbox_published_at
    ON outbox (published_at)
    WHERE published_at IS NOT NULL;

COMMENT ON TABLE outbox IS
    'Event chờ đẩy sang Kafka. Ghi CÙNG transaction với dữ liệu nghiệp vụ (ARCHITECTURE.md §7).';

db/migrations/000003_create_outbox.down.sql:

DROP TABLE IF EXISTS outbox;

5.1 Index một phần — vì sao quan trọng

Relay chỉ quan tâm các dòng published_at IS NULL. Ở trạng thái khoẻ mạnh, số dòng đó là vài chục, trong khi bảng có thể chứa hàng triệu dòng đã gửi.

Loại index	Kích thước sau 10 triệu event
`CREATE INDEX ON outbox (id)`	Hàng trăm MB, lớn dần mãi
`CREATE INDEX ON outbox (id) WHERE published_at IS NULL`	Vài KB, không lớn thêm

Index một phần chỉ chứa các dòng khớp điều kiện. Dòng nào được đánh dấu published_at sẽ tự rời khỏi index. Truy vấn của relay vì thế chạy nhanh như nhau bất kể bảng to cỡ nào.

5.2 ⚠️ Không dùng "id lớn nhất đã xử lý" làm con trỏ

Đây là lỗi thiết kế làm mất event vĩnh viễn, và nó chỉ xuất hiện khi có nhiều request đồng thời — nên máy dev gần như không bao giờ tái hiện được.

Cách làm sai — nghe rất hợp lý:

-- ❌ SAI: nhớ id cuối cùng, lần sau đọc từ đó trở đi
SELECT * FROM outbox WHERE id > :last_seen_id ORDER BY id LIMIT 100;

Vì sao sai: số thứ tự được cấp lúc INSERT, nhưng dòng chỉ hiện ra lúc COMMIT. Hai thao tác đó cách nhau một khoảng thời gian.

Thời điểm  Transaction A          Transaction B          Relay nhìn thấy
─────────────────────────────────────────────────────────────────────────
t1         INSERT → nhận id=5
t2                                INSERT → nhận id=6
t3                                COMMIT                  id=6
t4                                                        đọc, đặt last_seen=6
t5         COMMIT                                         id=5 ← ĐÃ BỊ BỎ QUA

Ở t4, dòng id=5 chưa hiện ra vì A chưa commit. Relay đọc thấy id=6, ghi nhớ last_seen = 6. Đến t5 dòng id=5 xuất hiện, nhưng relay sẽ không bao giờ đọc lại vùng id <= 6 nữa.

Event đó biến mất vĩnh viễn. Không có lỗi, không có log, không có cách nào phát hiện.

Cách đúng — lọc theo trạng thái, không theo vị trí:

-- ✅ ĐÚNG: một dòng chỉ rời khỏi tầm nhìn khi ĐÃ được đánh dấu gửi
SELECT * FROM outbox
WHERE published_at IS NULL
ORDER BY id
LIMIT 100
FOR UPDATE SKIP LOCKED;

Dòng id=5 chỉ đơn giản là chưa hiện ở lượt trước; lượt sau nó vẫn thoả published_at IS NULL và được xử lý bình thường.

Đánh đổi: A commit sau B nên id=5 được publish sau id=6, dù số nhỏ hơn. Điều này chấp nhận được — hai transaction chạy song song thì "thứ tự đúng" giữa chúng vốn không có định nghĩa. Thứ tự thật sự cần đảm bảo là giữa các event của cùng một thực thể, và điều đó do aggregate_id làm partition key lo (xem ARCHITECTURE.md §5.2). Nhiều event ghi trong cùng một transaction thì luôn giữ đúng thứ tự tương đối, vì chúng hiện ra cùng lúc.

5.3 `FOR UPDATE SKIP LOCKED`

Cho phép chạy nhiều instance relay mà không giẫm chân nhau:

FOR UPDATE — khoá các dòng đọc được cho tới hết transaction
SKIP LOCKED — dòng nào đang bị relay khác khoá thì bỏ qua, không xếp hàng chờ

Không có SKIP LOCKED, relay thứ hai sẽ chặn tại chỗ chờ relay thứ nhất xong — chạy hai instance mà thông lượng vẫn như một.

5.4 `JSONB` hay `TEXT` cho payload?

JSONB — vì nó truy vấn được, và điều đó cực kỳ có ích lúc điều tra sự cố:

-- Tìm mọi event chưa gửi của một tác giả
SELECT id, event_type, occurred_at
FROM outbox
WHERE published_at IS NULL
  AND payload->>'author_id' = '...';

Với TEXT bạn phải LIKE '%author_id%' — chậm và không đáng tin.

JSONB cũng kiểm tra tính hợp lệ của JSON ngay lúc INSERT: payload hỏng bị chặn tại DB thay vì trở thành message rác trong Kafka.

Đánh đổi duy nhất: JSONB không giữ thứ tự khoá và không giữ khoảng trắng — điều này không quan trọng với ta, vì consumer giải mã bằng tên trường chứ không so sánh chuỗi thô.

5.5 `BIGSERIAL` — đủ lớn chưa?

BIGSERIAL là BIGINT + sequence, tối đa khoảng 9,2 × 10¹⁸. Với 1 triệu event mỗi ngày, bạn dùng hết sau khoảng 25 tỷ năm.

(Cú pháp hiện đại tương đương là BIGINT GENERATED ALWAYS AS IDENTITY. Hai cách hoạt động như nhau; BIGSERIAL ngắn gọn hơn và vẫn được hỗ trợ đầy đủ.)

Điều không an toàn là SERIAL (INT, tối đa ~2,1 tỷ) — với 1 triệu event/ngày thì hết sau chưa tới 6 năm, và lúc hết thì mọi INSERT đều thất bại.

6. `000004` — consumed_events

db/migrations/000004_create_consumed_events.up.sql:

CREATE TABLE consumed_events (
    event_id     UUID        NOT NULL,
    consumer     TEXT        NOT NULL,
    processed_at TIMESTAMPTZ NOT NULL DEFAULT now(),

    PRIMARY KEY (event_id, consumer)
);

-- Phục vụ job dọn dẹp định kỳ
CREATE INDEX idx_consumed_events_processed_at
    ON consumed_events (processed_at);

COMMENT ON TABLE consumed_events IS
    'Chống xử lý trùng. Kafka giao at-least-once nên event SẼ bị giao lại (ARCHITECTURE.md §8).';

db/migrations/000004_create_consumed_events.down.sql:

DROP TABLE IF EXISTS consumed_events;

6.1 Vì sao khoá chính là cặp `(event_id, consumer)`

Cùng một event được nhiều consumer group xử lý độc lập:

post.post.created.v1  ─┬─→  stats-service    (tăng posts_count)
                       ├─→  search-indexer   (đẩy vào Elasticsearch)
                       └─→  feed-fanout      (đẩy vào feed người theo dõi)

Nếu khoá chính chỉ là event_id, group nào ghi trước sẽ chặn hai group còn lại — chúng thấy "đã xử lý rồi" và bỏ qua. Hai phần ba công việc biến mất trong im lặng.

Cặp (event_id, consumer) cho phép mỗi group có bản ghi riêng của mình.

6.2 Khoá chính đã là index — không cần thêm

PRIMARY KEY (event_id, consumer) tự tạo một unique index trên đúng cặp cột đó. Truy vấn duy nhất mà consumer thực hiện là INSERT ... ON CONFLICT DO NOTHING, và nó dùng chính index này.

Index processed_at thêm vào chỉ để job dọn dẹp không phải quét toàn bảng. Không có nó, lệnh DELETE ... WHERE processed_at < ... sẽ đọc từng dòng của một bảng nhiều triệu dòng.

6.3 Giữ dữ liệu bao lâu?

Bảng này lớn lên mãi nếu không dọn, nhưng dọn quá sớm thì phá vỡ chính thứ nó sinh ra để bảo vệ.

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

Cấu hình	Giá trị
Kafka `retention.ms`	7 ngày (mặc định)
Giữ `consumed_events`	30 ngày

Vì sao dư nhiều tới vậy: một consumer bị lỗi và phải reset offset về đầu topic sẽ đọc lại toàn bộ 7 ngày message. Nếu consumed_events chỉ giữ 3 ngày, những event của ngày thứ 4 đến thứ 7 sẽ được xử lý lại lần nữa — mọi counter cộng thêm một lần nữa.

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

7. Chạy migration

$env:DATABASE_URL = "postgres://app:secret@localhost:5432/community?sslmode=disable"
migrate -path db/migrations -database $env:DATABASE_URL up

1/u create_users (18.2ms)
2/u create_user_stats (12.7ms)
3/u create_outbox (15.1ms)
4/u create_consumed_events (9.8ms)

7.1 Kiểm chứng

docker compose -f deployments/docker-compose.yml exec postgres `
  psql -U app -d community -c "\dt"

              List of relations
 Schema |      Name         | Type  | Owner
--------+-------------------+-------+-------
 public | consumed_events   | table | app
 public | outbox            | table | app
 public | schema_migrations | table | app
 public | user_stats        | table | app
 public | users             | table | app

Kiểm tra index đã tạo đúng:

docker compose -f deployments/docker-compose.yml exec postgres `
  psql -U app -d community -c "\di idx_*"

 public | idx_consumed_events_processed_at | index | app | consumed_events
 public | idx_outbox_published_at          | index | app | outbox
 public | idx_outbox_unpublished           | index | app | outbox

Thử ràng buộc CHECK có thật sự chặn:

docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c @"
INSERT INTO users (name, username, email, password_hash)
VALUES ('Kiểm tra', 'kiemtra', 'kiemtra@example.com', 'x') RETURNING id;
"@

Lấy id vừa trả về rồi thử ghi counter âm:

INSERT INTO user_stats (user_id, posts_count) VALUES ('<id vừa lấy>', -1);
-- ERROR: new row for relation "user_stats" violates check constraint "user_stats_non_negative"

Thấy lỗi đó là đúng. Dọn lại:

DELETE FROM users WHERE username = 'kiemtra';   -- user_stats tự xoá theo CASCADE

7.2 Các lệnh `migrate` cần nhớ

Lệnh	Tác dụng
`migrate ... up`	Chạy hết migration chưa áp dụng
`migrate ... down 1`	Lùi lại một bước
`migrate ... version`	Xem version hiện tại
`migrate ... force N`	Ép version về N (dùng khi kẹt `dirty`)
`migrate ... drop -f`	⚠️ Xoá toàn bộ bảng

7.3 ⚠️ Trạng thái `dirty` — chắc chắn bạn sẽ gặp

Khi một migration lỗi giữa chừng, golang-migrate đánh dấu database là dirty và từ chối chạy tiếp:

error: Dirty database version 3. Fix and force version.

Đây không phải hỏng hóc — đó là cơ chế bảo vệ. Công cụ không biết migration số 3 đã chạy được bao nhiêu phần, nên nó dừng lại thay vì đoán.

Cách gỡ:

# 1. Xem migration số 3 thực sự đã làm được gì
docker compose -f deployments/docker-compose.yml exec postgres psql -U app -d community -c "\dt"

# 2a. Nếu KHÔNG có gì được tạo → lùi version về 2, sửa file, chạy lại
migrate -path db/migrations -database $env:DATABASE_URL force 2
migrate -path db/migrations -database $env:DATABASE_URL up

# 2b. Nếu ĐÃ tạo xong hết → chỉ cần xác nhận version 3 là đúng
migrate -path db/migrations -database $env:DATABASE_URL force 3

force chỉ sửa con số trong bảng schema_migrations; nó không đụng gì tới schema thật. Bạn phải tự nhìn xem schema đang ở trạng thái nào rồi mới quyết định ép về số mấy. Ép sai số sẽ tạo ra sai lệch âm thầm giữa cái DB có và cái golang-migrate tưởng nó có.

Ở môi trường dev, khi không chắc chắn thì làm lại từ đầu là an toàn nhất:

migrate -path db/migrations -database $env:DATABASE_URL drop -f
migrate -path db/migrations -database $env:DATABASE_URL up

8. Cấu hình `sqlc.yaml`

Tạo sqlc.yaml ở thư mục gốc:

version: "2"
sql:
  # ── platform/outbox ────────────────────────────────────────
  - engine: "postgresql"
    schema: "db/migrations"
    queries: "internal/platform/outbox/query.sql"
    gen:
      go:
        package: "gen"
        out: "internal/platform/outbox/gen"
        sql_package: "pgx/v5"
        emit_pointers_for_null_types: true
        emit_empty_slices: true
        emit_interface: false
        overrides: &pgtypes
          - db_type: "uuid"
            go_type: "github.com/google/uuid.UUID"
          - db_type: "pg_catalog.timestamptz"
            go_type: "time.Time"

  # ── platform/idempotency ───────────────────────────────────
  - engine: "postgresql"
    schema: "db/migrations"
    queries: "internal/platform/idempotency/query.sql"
    gen:
      go:
        package: "gen"
        out: "internal/platform/idempotency/gen"
        sql_package: "pgx/v5"
        emit_pointers_for_null_types: true
        emit_empty_slices: true
        emit_interface: false
        overrides: *pgtypes

  # ── modules/identity ───────────────────────────────────────
  - engine: "postgresql"
    schema: "db/migrations"
    queries: "internal/modules/identity/repository/query.sql"
    gen:
      go:
        package: "gen"
        out: "internal/modules/identity/repository/gen"
        sql_package: "pgx/v5"
        emit_pointers_for_null_types: true
        emit_empty_slices: true
        emit_interface: false
        overrides: *pgtypes

  # ── modules/stats ──────────────────────────────────────────
  - 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"
        emit_pointers_for_null_types: true
        emit_empty_slices: true
        emit_interface: false
        overrides: *pgtypes

8.1 Vì sao bốn khối riêng thay vì một

Mỗi package sinh ra một bộ code độc lập. Điều này chủ đích trùng khớp với ranh giới module ở Luật 1:

identity có kiểu User của riêng nó
stats có kiểu UserStat của riêng nó
Không package nào import package kia

Nếu dùng một khối chung, mọi module sẽ import cùng một package gen — và ngày identity đổi kiểu một cột, stats biên dịch lỗi dù nó chẳng liên quan gì. Đó chính là kiểu ràng buộc mà kiến trúc này sinh ra để tránh.

8.2 Neo YAML `&pgtypes` / `*pgtypes`

&pgtypes đặt tên cho khối overrides, *pgtypes dùng lại nó. Đây là cú pháp YAML chuẩn, không phải tính năng riêng của sqlc.

Không có nó, bạn phải chép khối override bốn lần — và ngày sửa một dòng, chắc chắn sẽ quên một chỗ.

8.3 Từng tuỳ chọn nghĩa là gì

Tuỳ chọn	Không có nó	Có nó
`sql_package: pgx/v5`	Sinh code cho `database/sql`	Dùng thẳng `pgx`, khớp với bước 2
`overrides: uuid`	`pgtype.UUID` (struct có `.Bytes`, `.Valid`)	`uuid.UUID` — dùng thẳng được
`overrides: timestamptz`	`pgtype.Timestamptz`	`time.Time`
`emit_pointers_for_null_types`	Cột NULL → kiểu wrapper	Cột NULL → con trỏ `string`, `time.Time`
`emit_empty_slices`	Query `:many` không có kết quả → `nil`	→ slice rỗng, `for range` an toàn
`emit_interface: false`	Sinh thêm interface `Querier`	Không sinh — xem lý do dưới

Vì sao tắt emit_interface?

sqlc có thể sinh sẵn interface Querier chứa mọi query. Nhưng ARCHITECTURE.md §4.3 quy định interface do bên gọi định nghĩa: service khai báo đúng những phương thức nó cần.
// Service chỉ cần ba việc này — interface phản ánh đúng thế
type Repository interface {
    Insert(ctx context.Context, tx pgx.Tx, u domain.User) (domain.User, error)
    FindByEmail(ctx context.Context, email string) (domain.User, error)
    Exists(ctx context.Context, username string) (bool, error)
}
Interface hẹp thì fake trong test chỉ vài dòng. Còn Querier với 20 phương thức buộc bạn phải cài đủ 20 cái, dù test chỉ dùng một.

8.4 `pgtype.UUID` khó dùng đến mức nào

Đây là lý do thật sự của khối overrides:

// ❌ Không override — mỗi lần đọc id là một đoạn dài dòng
var id pgtype.UUID
_ = row.Scan(&id)
if !id.Valid {
    return errors.New("id null")
}
u, err := uuid.FromBytes(id.Bytes[:])

// ✅ Có override
var id uuid.UUID
_ = row.Scan(&id)

Nhân đoạn trên với số lần bạn chạm vào một UUID trong toàn dự án — đó là hàng trăm dòng code không mang lại giá trị nào.

9. Viết query

9.1 Cú pháp `sqlc`

Mỗi query có một dòng chú thích khai báo tên và kiểu trả về:

-- name: TênHàmGo :kiểu

Kiểu	Hàm Go trả về	Dùng khi
`:one`	`(Row, error)`	Chắc chắn đúng một dòng
`:many`	`([]Row, error)`	Danh sách
`:exec`	`error`	INSERT/UPDATE/DELETE, không cần biết ảnh hưởng mấy dòng
`:execrows`	`(int64, error)`	Cần biết ảnh hưởng mấy dòng

:execrows là mấu chốt của cơ chế idempotency — xem §9.2.

9.2 `platform/idempotency`

New-Item -ItemType Directory -Force internal/platform/idempotency

internal/platform/idempotency/query.sql:

-- name: MarkConsumed :execrows
-- Trả về 1 nếu đây là lần xử lý ĐẦU TIÊN, 0 nếu event đã được xử lý rồi.
-- Consumer dựa vào con số này để quyết định có làm việc thật hay bỏ qua.
INSERT INTO consumed_events (event_id, consumer)
VALUES ($1, $2)
ON CONFLICT (event_id, consumer) DO NOTHING;

-- name: WasConsumed :one
-- Chỉ dùng để chẩn đoán. KHÔNG dùng để kiểm tra trước khi ghi:
-- giữa lúc kiểm tra và lúc ghi có khoảng trống cho race condition.
SELECT EXISTS (
    SELECT 1 FROM consumed_events WHERE event_id = $1 AND consumer = $2
);

-- name: DeleteConsumedBefore :execrows
DELETE FROM consumed_events WHERE processed_at < $1;

Vì sao INSERT ... ON CONFLICT chứ không phải SELECT rồi INSERT

Cách nghĩ tự nhiên là: kiểm tra đã xử lý chưa, chưa thì làm.
// ❌ SAI — có khoảng trống giữa hai lệnh
if exists, _ := q.WasConsumed(ctx, id, consumer); exists {
    return nil
}
// ← hai worker cùng chạy tới đây cùng lúc, cả hai đều thấy "chưa xử lý"
q.MarkConsumed(ctx, id, consumer)
doWork()   // ← chạy HAI lần
INSERT ... ON CONFLICT DO NOTHING là một lệnh nguyên tử. Database bảo đảm chỉ đúng một trong hai worker nhận được rowsAffected = 1; worker kia nhận 0 và bỏ qua.

Đây là ví dụ điển hình của nguyên tắc: để database xử lý tranh chấp đồng thời, đừng tự làm ở tầng ứng dụng. Nó có khoá và có transaction; code Go thì không.

9.3 `platform/outbox`

New-Item -ItemType Directory -Force internal/platform/outbox

internal/platform/outbox/query.sql:

-- name: EnqueueEvent :exec
-- Gọi TRONG transaction của nghiệp vụ. Đây là toàn bộ ý nghĩa của Outbox Pattern.
INSERT INTO outbox (
    event_id, event_type, topic, aggregate_id, correlation_id, payload, occurred_at
) VALUES ($1, $2, $3, $4, $5, $6, $7);

-- name: FetchUnpublished :many
-- Lọc theo TRẠNG THÁI, không theo vị trí con trỏ — xem tài liệu §5.2.
-- SKIP LOCKED cho phép chạy nhiều relay song song.
SELECT id, event_id, event_type, topic, aggregate_id,
       correlation_id, payload, occurred_at, attempts
FROM outbox
WHERE published_at IS NULL
ORDER BY id
LIMIT $1
FOR UPDATE SKIP LOCKED;

-- name: MarkPublished :exec
-- Đánh dấu cả lô bằng một lệnh thay vì N lệnh.
UPDATE outbox SET published_at = now() WHERE id = ANY($1::bigint[]);

-- name: MarkFailed :exec
UPDATE outbox SET attempts = attempts + 1, last_error = $2 WHERE id = $1;

-- name: CountUnpublished :one
-- Chỉ số giám sát quan trọng nhất của hệ thống (ARCHITECTURE.md §11.4).
SELECT count(*) FROM outbox WHERE published_at IS NULL;

-- name: DeletePublishedBefore :execrows
DELETE FROM outbox WHERE published_at IS NOT NULL AND published_at < $1;

Vì sao MarkPublished nhận một mảng?

Relay đọc 100 dòng mỗi lượt. Đánh dấu từng dòng một là 100 lượt đi về DB; = ANY($1::bigint[]) gộp thành một.

sqlc đọc $1::bigint[] và sinh ra tham số ids []int64 — đúng kiểu Go bạn muốn, không phải tự ghép chuỗi.

9.4 `modules/identity` (users)

New-Item -ItemType Directory -Force internal/modules/identity/repository

internal/modules/identity/repository/query.sql:

-- name: CreateUser :one
-- id do ứng dụng sinh, KHÔNG dùng DEFAULT gen_random_uuid():
-- service cần biết id ngay để dựng event trong cùng transaction.
INSERT INTO users (id, name, username, email, password_hash, avatar_url)
VALUES ($1, $2, $3, $4, $5, $6)
RETURNING *;

-- name: GetUserByID :one
SELECT * FROM users WHERE id = $1;

-- name: GetUserByEmail :one
-- email luôn được service chuẩn hoá về chữ thường trước khi gọi (§3.3).
SELECT * FROM users WHERE email = $1;

-- name: GetUserByUsername :one
SELECT * FROM users WHERE username = $1;

-- name: UpdateUserProfile :one
UPDATE users
SET name       = $2,
    avatar_url = $3,
    updated_at = now()
WHERE id = $1
RETURNING *;

9.5 `modules/stats` (user_stats)

New-Item -ItemType Directory -Force internal/modules/stats/repository

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

-- name: EnsureUserStats :exec
-- Đường chính: consumer OnUserRegistered của module stats gọi hàm này (§4.3).
-- ON CONFLICT khiến việc event bị giao trùng trở nên vô hại.
INSERT INTO user_stats (user_id) VALUES ($1) ON CONFLICT (user_id) DO NOTHING;

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

-- ─── posts_count ────────────────────────────────────────────
-- name: IncrementPostsCount :exec
UPDATE user_stats
SET posts_count = posts_count + 1, updated_at = now()
WHERE user_id = $1;

-- name: DecrementPostsCount :exec
-- GREATEST chặn counter xuống âm khi event xoá bị giao trùng (§4.2).
UPDATE user_stats
SET posts_count = GREATEST(posts_count - 1, 0), updated_at = now()
WHERE user_id = $1;

-- ─── total_post_views ───────────────────────────────────────
-- name: IncrementTotalPostViews :exec
UPDATE user_stats
SET total_post_views = total_post_views + 1, updated_at = now()
WHERE user_id = $1;

-- ─── followers / following ──────────────────────────────────
-- name: IncrementFollowersCount :exec
UPDATE user_stats
SET followers_count = followers_count + 1, updated_at = now()
WHERE user_id = $1;

-- name: DecrementFollowersCount :exec
UPDATE user_stats
SET followers_count = GREATEST(followers_count - 1, 0), updated_at = now()
WHERE user_id = $1;

-- name: IncrementFollowingUsersCount :exec
UPDATE user_stats
SET following_users_count = following_users_count + 1, updated_at = now()
WHERE user_id = $1;

-- name: DecrementFollowingUsersCount :exec
UPDATE user_stats
SET following_users_count = GREATEST(following_users_count - 1, 0), updated_at = now()
WHERE user_id = $1;

-- ─── bookmarks ──────────────────────────────────────────────
-- name: IncrementBookmarksCount :exec
UPDATE user_stats
SET bookmarks_count = bookmarks_count + 1, updated_at = now()
WHERE user_id = $1;

-- name: DecrementBookmarksCount :exec
UPDATE user_stats
SET bookmarks_count = GREATEST(bookmarks_count - 1, 0), updated_at = now()
WHERE user_id = $1;

-- ─── Tính lại toàn bộ — lưới an toàn cuối cùng ──────────────
-- name: RecalculatePostsCount :exec
-- Dùng khi số liệu lệch (ARCHITECTURE.md §15.3). Nguồn sự thật là bảng gốc.
UPDATE user_stats us
SET posts_count = (SELECT count(*) FROM posts WHERE author_id = us.user_id),
    updated_at  = now()
WHERE us.user_id = $1;

⚠️ RecalculatePostsCount tham chiếu bảng posts — bảng này chưa tồn tại.

sqlc phân tích SQL dựa trên schema và sẽ báo lỗi relation "posts" does not exist. Hãy tạm comment query này lại (thêm -- vào mọi dòng) và mở lại ở bước 4, khi migration 000005_create_posts đã có.

Tôi để nó ở đây thay vì bỏ hẳn, vì đây là ví dụ trực tiếp cho thấy sqlc bắt lỗi lúc sinh code, không phải lúc chạy. Với database/sql viết tay, một query sai tên bảng chỉ nổ khi có người dùng thật gọi tới nó.

Vì sao mỗi counter một query riêng, không gộp thành một query động?

Vì tên cột không thể là tham số trong SQL. Muốn động thì phải ghép chuỗi — mở cửa cho SQL injection và mất luôn kiểm tra lúc biên dịch.

Có một cách gộp hợp lệ: một UPDATE nhận đủ 8 delta và cộng tất cả cùng lúc (posts_count = posts_count + sqlc.arg(posts_delta), ...). Nó ngắn hơn, nhưng mỗi lần gọi đều ghi lại cả 8 cột và chỗ gọi trở nên khó đọc (IncrementPostsCount(userID) rõ ràng hơn UpdateStats(userID, 0,0,0,0,1,0,0,0)). Với số lượng counter cố định như ở đây, viết tường minh đáng giá hơn.

10. Sinh code và đọc kết quả

sqlc generate

Không có output nghĩa là thành công. Kiểm tra:

Get-ChildItem -Recurse -Filter "*.go" internal/platform/outbox, internal/platform/idempotency, internal/modules | Select-Object FullName

internal/modules/identity/repository/gen/db.go
internal/modules/identity/repository/gen/models.go
internal/modules/identity/repository/gen/query.sql.go
internal/modules/stats/repository/gen/db.go
internal/modules/stats/repository/gen/models.go
internal/modules/stats/repository/gen/query.sql.go
internal/platform/idempotency/gen/db.go
internal/platform/idempotency/gen/models.go
internal/platform/idempotency/gen/query.sql.go
internal/platform/outbox/gen/db.go
internal/platform/outbox/gen/models.go
internal/platform/outbox/gen/query.sql.go

10.1 Ba file, ba vai trò

File	Nội dung
`db.go`	Interface `DBTX` và hàm `New(db DBTX) *Queries`
`models.go`	Struct tương ứng với mọi bảng trong schema
`query.sql.go`	Một hàm Go cho mỗi query bạn viết

10.2 `DBTX` — chi tiết làm mọi thứ khớp vào nhau

Mở internal/platform/outbox/gen/db.go:

type DBTX interface {
	Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
	Query(context.Context, string, ...interface{}) (pgx.Rows, error)
	QueryRow(context.Context, string, ...interface{}) pgx.Row
}

func New(db DBTX) *Queries { return &Queries{db: db} }

Cả *pgxpool.Pool lẫn pgx.Tx đều thoả interface này. Đó chính là mảnh ghép nối TxManager ở bước 2 với code vừa sinh ra:

// Trong transaction — dùng cho mọi thao tác ghi
err := txm.Do(ctx, func(tx pgx.Tx) error {
    q := gen.New(tx)                       // ← truyền tx
    if err := q.CreateUser(ctx, ...); err != nil {
        return err
    }
    return q.EnqueueEvent(ctx, ...)        // ← CÙNG transaction. Đây là Luật 3.
})

// Ngoài transaction — chấp nhận được với thao tác chỉ đọc
q := gen.New(pool)                          // ← truyền pool
user, err := q.GetUserByID(ctx, id)

Không có DBTX, bạn sẽ phải sinh hai bộ hàm — một cho pool, một cho transaction. sqlc giải quyết bằng một interface ba phương thức.

10.3 Vì sao `models.go` có cả bảng không liên quan

Mở internal/modules/identity/repository/gen/models.go — bạn sẽ thấy cả Outbox và ConsumedEvent, dù identity chẳng dùng tới.

Lý do: sqlc sinh struct cho mọi bảng trong schema, và cả bốn khối cấu hình đều trỏ tới cùng thư mục db/migrations.

Điều này không gây hại, và thực ra còn hợp với kiến trúc:

Type không dùng thì không tốn gì lúc chạy
Mỗi package tự chủ hoàn toàn — identity.User và một User ở package khác là hai kiểu độc lập, không package nào phải import package kia

Đúng thứ Luật 1 mong muốn: module chia sẻ schema, không chia sẻ kiểu Go.

10.4 Đọc thử code sinh ra

Get-Content internal/platform/idempotency/gen/query.sql.go

const markConsumed = `-- name: MarkConsumed :execrows
INSERT INTO consumed_events (event_id, consumer)
VALUES ($1, $2)
ON CONFLICT (event_id, consumer) DO NOTHING
`

type MarkConsumedParams struct {
	EventID  uuid.UUID
	Consumer string
}

func (q *Queries) MarkConsumed(ctx context.Context, arg MarkConsumedParams) (int64, error) {
	result, err := q.db.Exec(ctx, markConsumed, arg.EventID, arg.Consumer)
	if err != nil {
		return 0, err
	}
	return result.RowsAffected(), nil
}

Ba điều đáng chú ý:

uuid.UUID, không phải pgtype.UUID — override ở §8 có hiệu lực
(int64, error) — :execrows cho ra đúng con số mà logic idempotency cần
Query có 2 tham số trở lên → sinh struct ...Params — gọi bằng tên trường nên không thể truyền nhầm thứ tự

Xác nhận biên dịch được:

go build ./...

10.5 ⚠️ Không bao giờ sửa tay file trong `gen/`

Mỗi lần sqlc generate sẽ ghi đè toàn bộ. Sửa tay thì lần chạy sau mất sạch, và mất một cách âm thầm.

Cần thêm logic → viết ở file khác trong package cha (repository/postgres.go), gọi vào gen.

Nên thêm một dòng vào .gitignore? Không. Hãy commit code sinh ra:

Người mới clone repo go build được ngay, không cần cài sqlc
Diff của gen/ trong pull request cho thấy rõ thay đổi SQL ảnh hưởng tới code Go thế nào — thường lộ ra những thay đổi phá vỡ tương thích mà nhìn file .sql không thấy
CI không phải cài thêm công cụ

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

Test này chứng minh cả bốn bảng và code sinh ra hoạt động cùng nhau trong một transaction — chính là hình dáng của mọi thao tác ghi ở các bước sau.

Tạo internal/platform/outbox/outbox_integration_test.go:

package outbox_test

import (
	"context"
	"encoding/json"
	"os"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	idem "github.com/yourname/community/internal/platform/idempotency/gen"
	obox "github.com/yourname/community/internal/platform/outbox/gen"
	"github.com/yourname/community/internal/platform/postgres"
	users "github.com/yourname/community/internal/modules/identity/repository/gen"
)

func setup(t *testing.T) (context.Context, *postgres.TxManager, *pgxpool.Pool) {
	t.Helper()

	dsn := os.Getenv("TEST_DATABASE_URL")
	if dsn == "" {
		dsn = "postgres://app:secret@localhost:5432/community?sslmode=disable"
	}
	ctx := context.Background()
	pool, err := postgres.Connect(ctx, postgres.DefaultConfig(dsn, "community-test"))
	if err != nil {
		t.Skipf("bỏ qua: cần Postgres đang chạy (%v)", err)
	}
	t.Cleanup(pool.Close)
	return ctx, postgres.NewTxManager(pool), pool
}

// Đây là hình dáng của MỌI thao tác ghi trong hệ thống:
// dữ liệu nghiệp vụ và event đi vào DB cùng một lúc, hoặc không cái nào cả.
func TestGhiUserVaOutboxTrongCungTransaction(t *testing.T) {
	ctx, txm, pool := setup(t)

	userID := uuid.New()
	eventID := uuid.New()
	username := "user_" + userID.String()[:8]

	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, userID)
		_, _ = pool.Exec(context.Background(), `DELETE FROM outbox WHERE event_id = $1`, eventID)
	})

	err := txm.Do(ctx, func(tx pgx.Tx) error {
		u, err := users.New(tx).CreateUser(ctx, users.CreateUserParams{
			ID:           userID,
			Name:         "Nguyễn Huy Hoàng",
			Username:     username,
			Email:        username + "@example.com",
			PasswordHash: "hash",
		})
		if err != nil {
			return err
		}

		payload, err := json.Marshal(map[string]string{"user_id": u.ID.String()})
		if err != nil {
			return err
		}

		return obox.New(tx).EnqueueEvent(ctx, obox.EnqueueEventParams{
			EventID:     eventID,
			EventType:   "identity.user.registered.v1",
			Topic:       "identity",
			AggregateID: u.ID.String(),
			Payload:     payload,
			OccurredAt:  time.Now().UTC(),
		})
	})
	require.NoError(t, err)

	var n int
	require.NoError(t, pool.QueryRow(ctx,
		`SELECT count(*) FROM outbox WHERE event_id = $1 AND published_at IS NULL`, eventID).Scan(&n))
	assert.Equal(t, 1, n, "event phải nằm trong outbox và chưa được gửi")
}

// Nếu test này đỏ, Luật 3 đã bị phá: sẽ có ngày dữ liệu ghi thành công
// nhưng event không bao giờ được phát.
func TestLoiNghiepVuKeoTheoOutboxRollback(t *testing.T) {
	ctx, txm, pool := setup(t)

	userID := uuid.New()
	eventID := uuid.New()

	err := txm.Do(ctx, func(tx pgx.Tx) error {
		if err := obox.New(tx).EnqueueEvent(ctx, obox.EnqueueEventParams{
			EventID:     eventID,
			EventType:   "identity.user.registered.v1",
			Topic:       "identity",
			AggregateID: userID.String(),
			Payload:     []byte(`{}`),
			OccurredAt:  time.Now().UTC(),
		}); err != nil {
			return err
		}
		// Vi phạm khoá ngoại: user này không tồn tại
		_, err := tx.Exec(ctx, `INSERT INTO user_stats (user_id) VALUES ($1)`, uuid.New())
		return err
	})

	require.Error(t, err)
	assert.True(t, postgres.IsForeignKeyViolation(err), "phải là lỗi khoá ngoại, nhận: %v", err)

	var n int
	require.NoError(t, pool.QueryRow(ctx,
		`SELECT count(*) FROM outbox WHERE event_id = $1`, eventID).Scan(&n))
	assert.Equal(t, 0, n, "outbox phải rollback theo")
}

func TestMarkConsumed_LanDauLa1_LanSauLa0(t *testing.T) {
	ctx, _, pool := setup(t)

	eventID := uuid.New()
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), `DELETE FROM consumed_events WHERE event_id = $1`, eventID)
	})

	q := idem.New(pool)
	arg := idem.MarkConsumedParams{EventID: eventID, Consumer: "stats-service"}

	first, err := q.MarkConsumed(ctx, arg)
	require.NoError(t, err)
	assert.EqualValues(t, 1, first, "lần đầu phải ghi được")

	second, err := q.MarkConsumed(ctx, arg)
	require.NoError(t, err)
	assert.EqualValues(t, 0, second, "lần hai phải bị bỏ qua — đây là cốt lõi của idempotency")

	// Consumer group khác vẫn phải xử lý được cùng event đó
	other, err := q.MarkConsumed(ctx, idem.MarkConsumedParams{EventID: eventID, Consumer: "search-indexer"})
	require.NoError(t, err)
	assert.EqualValues(t, 1, other, "group khác phải có bản ghi riêng")
}

func TestCounterKhongXuongAm(t *testing.T) {
	ctx, _, pool := setup(t)

	userID := uuid.New()
	username := "cnt_" + userID.String()[:8]
	t.Cleanup(func() {
		_, _ = pool.Exec(context.Background(), `DELETE FROM users WHERE id = $1`, userID)
	})

	_, err := pool.Exec(ctx, `
		INSERT INTO users (id, name, username, email, password_hash)
		VALUES ($1, 'Test', $2, $3, 'hash')`, userID, username, username+"@example.com")
	require.NoError(t, err)
	_, err = pool.Exec(ctx, `INSERT INTO user_stats (user_id) VALUES ($1)`, userID)
	require.NoError(t, err)

	// Giảm khi đang ở 0: GREATEST giữ nó tại 0 thay vì cho xuống -1
	_, err = pool.Exec(ctx, `
		UPDATE user_stats SET posts_count = GREATEST(posts_count - 1, 0) WHERE user_id = $1`, userID)
	require.NoError(t, err)

	var n int
	require.NoError(t, pool.QueryRow(ctx,
		`SELECT posts_count FROM user_stats WHERE user_id = $1`, userID).Scan(&n))
	assert.Equal(t, 0, n)

	// Còn UPDATE trực tiếp xuống âm phải bị CHECK constraint chặn
	_, err = pool.Exec(ctx, `UPDATE user_stats SET posts_count = -1 WHERE user_id = $1`, userID)
	assert.Error(t, err, "CHECK constraint phải chặn giá trị âm")
}

Chạy:

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

ok  github.com/yourname/community/internal/contracts                0.29s
ok  github.com/yourname/community/internal/platform/eventbus/inmem  0.27s
ok  github.com/yourname/community/internal/platform/outbox          1.12s
ok  github.com/yourname/community/internal/platform/postgres        0.96s

Test thứ hai (TestLoiNghiepVuKeoTheoOutboxRollback) là test có giá trị nhất của cả bước này: nó chứng minh Luật 3 hoạt động. Khi nào nó còn xanh, bạn còn chắc chắn rằng không có kịch bản nào dữ liệu ghi thành công mà event bị mất.

12. Bảo trì hai bảng hạ tầng

Cả outbox và consumed_events đều lớn lên vô hạn. Chưa cần job tự động ở giai đoạn này, nhưng nên biết sẵn hai câu lệnh:

-- Giữ 7 ngày để còn điều tra khi có sự cố
DELETE FROM outbox
WHERE published_at IS NOT NULL AND published_at < now() - INTERVAL '7 days';

-- Giữ 30 ngày — phải DÀI HƠN Kafka retention (§6.3)
DELETE FROM consumed_events
WHERE processed_at < now() - INTERVAL '30 days';

Cả hai đã có query tương ứng (DeletePublishedBefore, DeleteConsumedBefore), nên bước sau chỉ cần một goroutine chạy định kỳ.

Với bảng rất lớn, đừng DELETE một phát. Một lệnh xoá 10 triệu dòng giữ khoá lâu và làm phình transaction log. Chia lô:
DELETE FROM outbox
WHERE id IN (
    SELECT id FROM outbox
    WHERE published_at IS NOT NULL AND published_at < $1
    LIMIT 10000
);
Lặp tới khi rowsAffected = 0. Đây cũng là lý do DeletePublishedBefore khai :execrows — để vòng lặp biết lúc nào dừng.

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

Chưa có	Vì sao	Sẽ làm ở
Bảng `posts`	Nằm ngoài danh sách 4 migration của bước này. Nó thuộc về module `post`, nên tạo cùng lúc với module đó sẽ giữ được mạch: một migration ↔ một chủ sở hữu	Bước dựng module `post` (`000005`)
`outbox/writer.go`	Query đã có; phần bọc Go cần `contracts.Envelope` ghép vào	Bước 4
`outbox/relay.go`	Cần thêm `Publisher` của Kafka mới chạy được	Bước 5
`idempotency/guard.go`	`MarkConsumed` đã có, nhưng lớp bọc chỉ có ý nghĩa khi đã có consumer thật gọi tới	Bước 6
Job dọn dẹp	Chưa có gì để dọn	Bước 6
`RecalculatePostsCount`	Cần bảng `posts` (đang comment lại)	Bước dựng module `post`

Về việc thiếu bảng posts: note.md có mô tả bảng này, và internal/contracts/v1/post_events.go ở bước 2 đã khai báo event PostCreated. Chúng chưa gặp nhau vì migration cho posts nằm ngoài phạm vi bốn file của bước 3.

Không có gì hỏng: event là hợp đồng, nó tồn tại độc lập với bảng. Chỉ khi viết PostService ở bước 4 thì mới cần bảng thật — và lúc đó 000005_create_posts sẽ được tạo cùng module.

14. Checklist hoàn thành

[ ] migrate ... version trả về 4 (không kèm chữ dirty)
[ ] \dt liệt kê đủ 5 bảng (4 bảng + schema_migrations)
[ ] \di idx_* liệt kê đủ 3 index
[ ] Mỗi migration có đủ cặp .up.sql và .down.sql (8 file)
[ ] sqlc generate chạy không lỗi
[ ] 12 file .go được sinh trong 4 thư mục gen/
[ ] go build ./... thành công
[ ] go test ./internal/... -race -count=1 — tất cả ok, không SKIP
[ ] RecalculatePostsCount đang được comment (chờ bảng posts)

Thử rollback rồi chạy lại để chắc chắn file .down viết đúng:

migrate -path db/migrations -database $env:DATABASE_URL down -all
migrate -path db/migrations -database $env:DATABASE_URL up
migrate -path db/migrations -database $env:DATABASE_URL version

Nếu down -all báo lỗi, một file .down.sql nào đó viết thiếu. Sửa ngay bây giờ — ngày bạn thật sự cần rollback thì thường là ngày tệ nhất để phát hiện nó hỏng.

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

project/
├── cmd/smoketest/main.go
├── db/migrations/
│   ├── 000001_create_users.up.sql / .down.sql
│   ├── 000002_create_user_stats.up.sql / .down.sql
│   ├── 000003_create_outbox.up.sql / .down.sql
│   └── 000004_create_consumed_events.up.sql / .down.sql
├── deployments/docker-compose.yml
├── docs/
│   ├── ARCHITECTURE.md
│   ├── 01-khoi-tao-du-an-va-ha-tang.md
│   ├── 02-nen-mong-platform.md
│   └── 03-migrations-va-sqlc.md
├── internal/
│   ├── contracts/            (bước 2)
│   ├── modules/
│   │   ├── identity/repository/{query.sql, gen/}
│   │   └── stats/repository/{query.sql, gen/}
│   └── platform/
│       ├── eventbus/         (bước 2)
│       ├── postgres/         (bước 2)
│       ├── idempotency/{query.sql, gen/}
│       └── outbox/{query.sql, gen/, outbox_integration_test.go}
├── sqlc.yaml
├── go.mod
└── go.sum

Thêm hai lệnh vào Makefile:

migrate:                  ## Chạy migration
	migrate -path db/migrations -database "$(DATABASE_URL)" up

sqlc:                     ## Sinh lại code từ SQL
	sqlc generate

Commit:

git add .
git commit -m "feat(db): migration cho users, user_stats, outbox, consumed_events + sqlc"

Bước tiếp theo

Bước 4 — Dựng module identity: bọc code gen vừa sinh thành outbox.Writer (nhận tx pgx.Tx, ghi contracts.Envelope); dựng module identity hoàn chỉnh từ HTTP tới repository — đăng ký, đăng nhập, băm mật khẩu, cấp JWT; và phát event identity.user.registered.v1 vào bảng outbox.

Đó là bước đầu tiên bạn thấy một event thật đi từ HTTP request vào tới bảng outbox — và cũng là lúc bốn bảng vừa tạo ở đây được dùng thật.

Backend

Mục tiêu

Mục lục

1. Cài công cụ

2. Quy ước migration

Quy tắc đặt tên

Ba luật của migration

3. 000001 — users

3.1 gen_random_uuid() — không cần extension

3.2 TIMESTAMPTZ, không bao giờ TIMESTAMP

3.3 Email phân biệt hoa thường — cái bẫy kinh điển

3.4 VARCHAR(255) — giữ hay đổi?

4. 000002 — user_stats

4.1 user_id vừa là khoá chính vừa là khoá ngoại

4.2 CHECK (>= 0) phối hợp với GREATEST như thế nào

4.3 Dòng user_stats được tạo lúc nào?

5. 000003 — outbox

5.1 Index một phần — vì sao quan trọng

5.2 ⚠️ Không dùng "id lớn nhất đã xử lý" làm con trỏ

5.3 FOR UPDATE SKIP LOCKED

5.4 JSONB hay TEXT cho payload?

5.5 BIGSERIAL — đủ lớn chưa?

6. 000004 — consumed_events

6.1 Vì sao khoá chính là cặp (event_id, consumer)

6.2 Khoá chính đã là index — không cần thêm

6.3 Giữ dữ liệu bao lâu?

7. Chạy migration

7.1 Kiểm chứng

7.2 Các lệnh migrate cần nhớ

7.3 ⚠️ Trạng thái dirty — chắc chắn bạn sẽ gặp

8. Cấu hình sqlc.yaml

8.1 Vì sao bốn khối riêng thay vì một

8.2 Neo YAML &pgtypes / *pgtypes

8.3 Từng tuỳ chọn nghĩa là gì

8.4 pgtype.UUID khó dùng đến mức nào

9. Viết query

9.1 Cú pháp sqlc

9.2 platform/idempotency

Vì sao INSERT ... ON CONFLICT chứ không phải SELECT rồi INSERT

9.3 platform/outbox

9.4 modules/identity (users)

9.5 modules/stats (user_stats)

10. Sinh code và đọc kết quả

10.1 Ba file, ba vai trò

10.2 DBTX — chi tiết làm mọi thứ khớp vào nhau

10.3 Vì sao models.go có cả bảng không liên quan

10.4 Đọc thử code sinh ra

10.5 ⚠️ Không bao giờ sửa tay file trong gen/

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

12. Bảo trì hai bảng hạ tầng

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

14. Checklist hoàn thành

Bước tiếp theo

Mục lục

3. `000001` — users

3.1 `gen_random_uuid()` — không cần extension

3.2 `TIMESTAMPTZ`, không bao giờ `TIMESTAMP`

3.4 `VARCHAR(255)` — giữ hay đổi?

4. `000002` — user_stats

4.1 `user_id` vừa là khoá chính vừa là khoá ngoại

4.2 `CHECK (>= 0)` phối hợp với `GREATEST` như thế nào

4.3 Dòng `user_stats` được tạo lúc nào?

5. `000003` — outbox

5.3 `FOR UPDATE SKIP LOCKED`

5.4 `JSONB` hay `TEXT` cho payload?

5.5 `BIGSERIAL` — đủ lớn chưa?

6. `000004` — consumed_events

6.1 Vì sao khoá chính là cặp `(event_id, consumer)`

7.2 Các lệnh `migrate` cần nhớ

7.3 ⚠️ Trạng thái `dirty` — chắc chắn bạn sẽ gặp

8. Cấu hình `sqlc.yaml`

8.2 Neo YAML `&pgtypes` / `*pgtypes`

8.4 `pgtype.UUID` khó dùng đến mức nào

9.1 Cú pháp `sqlc`

9.2 `platform/idempotency`

Vì sao `INSERT ... ON CONFLICT` chứ không phải `SELECT` rồi `INSERT`

9.3 `platform/outbox`

9.4 `modules/identity` (users)

9.5 `modules/stats` (user_stats)

10.2 `DBTX` — chi tiết làm mọi thứ khớp vào nhau

10.3 Vì sao `models.go` có cả bảng không liên quan

10.5 ⚠️ Không bao giờ sửa tay file trong `gen/`