Database Design | History Table | Case: Event Sourcing

Event Sourcing là mô hình lưu trữ trạng thái qua chuỗi sự kiện thay vì bản ghi hiện tại, sử dụng event store làm nguồn dữ liệu chính. Bạn hình dung Event Sourcing như là cách bạn ghi lại cuộc sống của bạn trong một cuốn nhật ký.

• Event Store (Table history_table): Đây là cuốn nhật ký. Mỗi trang ghi lại một sự kiện cụ thể, ví dụ: “Ngày 1: …, “Ngày 2: …. Bạn không xóa hay sửa trang cũ, chỉ thêm trang mới. Muốn biết mình có bao nhiêu tiền lúc này? Lật từng trang, cộng trừ từ đầu đến cuối.

create table history_table (
    event_id bigint identity(1,1) primary key,
    type varchar(50) check (type in ('recharge', 'withdraw', 'expire', 'transfer', 'behavior')),
    event_data nvarchar(max) check (isjson(event_data) = 1),
    event_timestamp datetime default getutcdate(),
    version int default 1,
    customer_id as try_cast(json_value(event_data, '$.customer_id') as int) persisted
);
create index idx_customer on history_table (customer_id

event_data (JSON): Nội dung chi tiết của trang (VD: nhận bao nhiêu point? đổi bao nhiêu point?").

• Projection (customer_points): Đây như tờ giấy nhỏ dán ở bìa nhật ký, ghi nhanh số điểm hiện tại. Tờ giấy này tiện để xem nhanh, nhưng nếu mất, vẫn có thể tính lại từ nhật ký (history_table).

create table customer_points (
    customer_id int primary key,
    points int default 0,
    last_updated datetime
);

create table customer_behavior (
    customer_id int,
    preference varchar(100),
    frequency int,
    primary key (customer_id, preference)
);

• Stored Procedure (ApplyEvent): Giống người thư ký. Khi có sự kiện mới, thư ký ghi vào nhật ký (history_table) và cập nhật tờ giấy nhỏ (customer_points) ngay lập tức, đảm bảo số điểm luôn đúng.

alter procedure applyevent
    @event_type varchar(50),
    @event_data nvarchar(max),
    @version int = 1
as
begin
    set nocount on;

    declare @customer_id int = try_cast(json_value(@event_data, '$.customer_id') as int);
    declare @points decimal(10,2) = try_cast(json_value(@event_data, '$.points') as decimal(10,2));
    declare @reason nvarchar(100) = json_value(@event_data, '$.reason');
    declare @direction nvarchar(10) = json_value(@event_data, '$.direction');
    declare @source nvarchar(50) = json_value(@event_data, '$.source');
    declare @expiry_date date = try_cast(json_value(@event_data, '$.expiry_date') as date);

    -- kiểm tra số dư trước khi trừ điểm
    if @event_type in ('withdraw', 'transfer', 'expire') and exists (
        select 1 from customer_points 
        where customer_id = @customer_id and points < @points
    )
    begin
        -- ghi nhận hành vi thất bại
        declare @json_event_data nvarchar(max);
        set @json_event_data = formatmessage(
            '{"customer_id": %s, "preference": "failed_transaction", "reason": "insufficient_points"}', 
            isnull(cast(@customer_id as nvarchar(20)), 'null')
        );
        exec applyevent 'behavior', @json_event_data, 1;
        return;
    end

    -- ghi log lịch sử giao dịch
    insert into history_table (type, event_data, version) 
    values (@event_type, @event_data, @version);

    -- xử lý điểm thưởng và cập nhật số dư
    if @event_type in ('recharge', 'withdraw', 'expire', 'transfer', 'adjustment')
    begin
        merge customer_points as target
        using (select 
                  @customer_id as customer_id, 
                  @points as points, 
                  @event_type as type,
                  @direction as direction,
                  @source as source,
                  @expiry_date as expiry_date
               ) as source
        on target.customer_id = source.customer_id
        when matched then
            update set points = case 
                when source.type = 'recharge' then target.points + source.points
                when source.type in ('withdraw', 'expire') then target.points - source.points
                when source.type = 'transfer' and source.direction = 'out' then target.points - source.points
                when source.type = 'transfer' and source.direction = 'in' then target.points + source.points
                when source.type = 'adjustment' then target.points + source.points
                else target.points
            end,
            last_updated = getutcdate()
        when not matched then
            insert (customer_id, points, expiry_date, last_updated)
            values (source.customer_id, source.points, source.expiry_date, getutcdate());
    end

    -- nếu là hoàn tiền do hết hàng, ghi nhận hành vi
    if @event_type = 'recharge' and @reason = 'out_of_stock_refunded'
    begin
        declare @refund_behavior nvarchar(max);
        set @refund_behavior = formatmessage(
            '{"customer_id": %s, "preference": "refund", "reason": "out_of_stock"}', 
            isnull(cast(@customer_id as nvarchar(20)), 'null')
        );
        exec applyevent 'behavior', @refund_behavior, 1;
    end

    -- xử lý điểm thưởng có ngày hết hạn
    if @event_type = 'expire'
    begin
        delete from customer_points 
        where customer_id = @customer_id 
              and expiry_date is not null 
              and expiry_date < getdate();
    end

    -- nếu điểm khuyến mãi hết hạn, ghi nhận hành vi
    if @event_type = 'expire' and @reason = 'promo_expired'
    begin
        declare @expire_behavior nvarchar(max);
        set @expire_behavior = formatmessage(
            '{"customer_id": %s, "preference": "expired_points", "reason": "promo_expired"}', 
            isnull(cast(@customer_id as nvarchar(20)), 'null')
        );
        exec applyevent 'behavior', @expire_behavior, 1;
    end

		-- nếu là hoàn tiền do lỗi hệ thống, ghi nhận hành vi
	if @event_type = 'recharge' and @reason = 'system_error_refunded'
	begin
		set @refund_behavior = formatmessage(
			'{"customer_id": %s, "preference": "refund", "reason": "system_error"}', 
			isnull(cast(@customer_id as nvarchar(20)), 'null')
		);
		exec applyevent 'behavior', @refund_behavior, 1;
	end


    -- ghi nhận hành vi khách hàng
    if @event_type = 'behavior'
    begin
        merge customer_behavior as target
        using (select 
                  @customer_id as customer_id, 
                  json_value(@event_data, '$.preference') as preference, 
                  1 as frequency
               ) as source
        on target.customer_id = source.customer_id 
           and target.preference = source.preference
        when matched then
            update set target.frequency = target.frequency + 1
        when not matched then
            insert (customer_id, preference, frequency)
            values (source.customer_id, source.preference, source.frequency);
    end
end;

• Replay Events: Nếu tờ giấy nhỏ bị rách, bạn mở nhật ký ra, đọc lại từ đầu, cộng trừ từng dòng để viết lại tờ giấy mới.
• Replay với Snapshot: như điểm dừng trong sổ: “Đến ngày 1/1, bạn có 1000 điểm”. Thay vì đọc hết sổ, máy chỉ đọc từ ảnh chụp đến nay, tiết kiệm thời gian. Giảm thời gian replay.

create table snapshots (
    customer_id int primary key,
    points int,
    last_event_id bigint,
    snapshot_timestamp datetime
);

create procedure replayevents
    @snapshot_id bigint = null
as
begin
    set nocount on;

    -- restore từ bảng snapshots
    insert into customer_points (customer_id, points, last_updated)
    select customer_id, points, snapshot_timestamp 
    from snapshots
    where last_event_id = coalesce(@snapshot_id, (select max(last_event_id) from snapshots));

    -- cập nhật lịch sử từ history_table
    merge customer_points as target
    using (
        select 
            try_cast(json_value(event_data, '$.customer_id') as int) as customer_id,
            sum(case 
                    when type = 'recharge' then try_cast(json_value(event_data, '$.points') as decimal(10,2))
                    when type in ('withdraw', 'expire') then -try_cast(json_value(event_data, '$.points') as decimal(10,2))
                    when type = 'transfer' and json_value(event_data, '$.direction') = 'out' then -try_cast(json_value(event_data, '$.points') as decimal(10,2))
                    when type = 'transfer' and json_value(event_data, '$.direction') = 'in' then try_cast(json_value(event_data, '$.points') as decimal(10,2))
               end) as total_points,
            max(event_timestamp) as last_updated
        from history_table
        where event_id > coalesce(@snapshot_id, 0)
        group by json_value(event_data, '$.customer_id')
    ) as source
    on target.customer_id = source.customer_id
    when matched then 
        update set 
            target.points = target.points + source.total_points,
            target.last_updated = source.last_updated
    when not matched then 
        insert (customer_id, points, last_updated)
        values (source.customer_id, source.total_points, source.last_updated);
end;

Test Case 1: TungTen mua sản phẩm bằng tiền USD

exec applyevent 'recharge', '{"customer_id": 1, "points": 240, "currency": "usd", "original_amount": 100}', 1;
select points from customer_points where customer_id = 1; -- kq mong muốn: 240 (true)

Test Case 2: TungTen dùng điểm thưởng đổi sản phẩm

exec applyevent 'withdraw', '{"customer_id": 1, "points": 80, "item": "mouse_laptop"}', 1;
select points from customer_points where customer_id = 1; -- kq mong muốn: 240 - 80 = 160 (true)

Test Case 3: TungTen tham gia event được cộng điểm

exec applyevent 'recharge', '{"customer_id": 1, "points": 50, "source": "offline_workshop", "event_name": "laptop_event_2025"}', 1;
select points from customer_points where customer_id = 1; -- kq mong muốn: 160 + 50 = 210 (true)

Test Case 4: TungTen trích điểm tặng bạn mình

exec applyevent 'transfer', '{"customer_id": 1, "points": 10, "direction": "out", "to": 2}', 1;
exec applyevent 'transfer', '{"customer_id": 2, "points": 10, "direction": "in", "from": 1}', 1;
select customer_id, points from customer_points where customer_id in (1, 2); 
--kq mong muốn: cust 1: 210 - 10 = 200, thêm cust 2 = 10 (true)

Test Case 5: Ghi nhận món hàng của TungTen mua và số lần để phục vụ phân tích sau này

exec applyevent 'behavior', '{"customer_id": 1, "preference": "mouse_laptop"}', 1;
select preference, frequency from customer_behavior where customer_id = 1; -- kq mong muốn: "mouse_laptop", frequency: 1, diem khong doi

Test Case 6: TungTen dùng điểm để đổi bộ combo sp nhưng hết hàng, hoàn lại điểm

exec applyevent 'withdraw', '{"customer_id": 1, "points": 150, "item": "combo_chuot_phim"}', 1;
exec applyevent 'recharge', '{"customer_id": 1, "points": 150, "reason": "out_of_stock_refunded"}', 1;
select points from customer_points where customer_id = 1; -- expected: 200 (200 - 150 + 150) (true)

Test Case 7: TungTen được tặng điểm nhân dịp sinh nhật nhưng chỉ dùng trong khoảng thời gian cho phép

exec applyevent 'recharge', '{"customer_id": 1, "points": 100, "source": "promo", "expiry_date": "2025-03-19"}', 1;
exec applyevent 'expire', '{"customer_id": 1, "points": 100, "reason": "promo_expired"}', 1;
select points from customer_points where customer_id = 1; -- expected: 200 (200 + 100 - 100) (true)

Test Case 8: TungTen bị trừ nhầm điểm, hoàn lại, giả sử tùng có 200 điểm và đang bị trừ nhầm 100 điểm, giờ cộng vào lại

exec applyevent 'recharge', '{"customer_id": 1, "points": 100, "reason": "system_error_refunded"}', 1;
select points from customer_points where customer_id = 1; -- expected: 300 (200 + 100) (true)

• Auditability (Khả Năng Kiểm Toán Toàn Diện): Mọi thay đổi trạng thái được ghi lại dưới dạng sự kiện bất biến trong history_table, cho phép kiểm tra lịch sử giao dịch đầy đủ, minh bạch.
• TungTen khiếu nại bị trừ nhầm điểm trong tháng 3/2025. Cửa hàng cần kiểm tra tất cả giao dịch của TungTen để xác minh.

-- xem toàn bộ lịch sử giao dịch
select event_id, type, event_data, event_timestamp
from history_table
where json_value(event_data, '$.customer_id') = '1'
order by event_id;

select 
    sum(
        case 
            when type = 'recharge' then cast(json_value(event_data, '$.points') as decimal(10,2))
            when type in ('withdraw', 'expire') then -cast(json_value(event_data, '$.points') as decimal(10,2))
            when type = 'transfer' and json_value(event_data, '$.direction') = 'out' then -cast(json_value(event_data, '$.points') as decimal(10,2))
            when type = 'transfer' and json_value(event_data, '$.direction') = 'in' then cast(json_value(event_data, '$.points') as decimal(10,2))
            when type = 'adjustment' then cast(json_value(event_data, '$.points') as decimal(10,2))
            else 0
        end
    ) as calculated_points
from history_table
where json_value(event_data, '$.customer_id') = '1';

• Rebuildability (Khả Năng Tái Tạo Trạng Thái): Có thể xóa trạng thái hiện tại (customer_points) và tái tạo lại từ history_table hoặc snapshot, đảm bảo dữ liệu luôn chính xác.
• Máy chủ chứa customer_points bị hỏng sau Black Friday, cửa hàng cần khôi phục điểm của TungTen từ dữ liệu lịch sử.

-- Xóa trạng thái hiện tại (giả lập hỏng dữ liệu)
TRUNCATE TABLE customer_points;
-- Replay từ snapshot tại event_id = 5
EXEC ReplayEventsFromSnapshot @snapshot_id = 5;
-- Kiểm tra điểm sau khi replay
SELECT points FROM customer_points WHERE customer_id = 1;

• Extensibility (Linh Hoạt Mở Rộng): Dễ thêm loại sự kiện mới hoặc projection mới mà không cần thay đổi cấu trúc chính.
• Cửa hàng muốn thêm tính năng “điểm thưởng sinh nhật” cho TungTen mà không sửa đổi hệ thống cũ.

-- Thêm sự kiện điểm thưởng sinh nhật
EXEC ApplyEvent 'recharge', '{"customer_id": 1, "points": 50, "source": "birthday_bonus"}', 1;
-- Kiểm tra điểm sau khi thêm
SELECT points FROM customer_points WHERE customer_id = 1;
-- Kiểm tra lịch sử có sự kiện mới không
SELECT event_id, type, event_data FROM history_table 
WHERE JSON_VALUE(event_data, '$.source') = 'birthday_bonus';

update lại store procedure để cập nhật trạng thái

• Analytics Support (Hỗ Trợ Phân Tích Dữ Liệu): Dữ liệu sự kiện trong history_table cho phép phân tích hành vi khách hàng và tối ưu chiến dịch.
• Performance Optimization (Hiệu Suất Cao Với Snapshot): Snapshot giảm thời gian replay, phù hợp với hệ thống lớn có hàng triệu sự kiện. Sau 1 năm, history_table có 1 triệu sự kiện, cửa hàng cần replay điểm của TungTen nhanh chóng.

-- Giả lập thêm snapshot mới tại event_id = 15 (sau Black Friday)
INSERT INTO snapshots (customer_id, points, last_event_id, snapshot_timestamp)
VALUES (1, 1120, 19, '2025-03-20 12:00:00');
-- Replay từ snapshot mới
TRUNCATE TABLE customer_points;
EXEC ReplayEventsFromSnapshot @snapshot_id = 19;
-- Kiểm tra điểm
SELECT points FROM customer_points WHERE customer_id = 1;