Mỗi khi bạn CẬP NHẬT dữ liệu, bạn đang XÓA lịch sử: Giải pháp Chrono Temporal cho PostgreSQL

Trong một dự án gần đây, chúng tôi đã gặp phải một vấn đề hóc búa khi một khách hàng khiếu nại về hóa đơn. Vấn đề tưởng chừng đơn giản: khách hàng muốn biết gói đăng ký họ đang sử dụng tại thời điểm bị trừ tiền là gì.

Tuy nhiên, khi kiểm tra cơ sở dữ liệu, chúng tôi... bó tay. Cơ sở dữ liệu chỉ lưu trữ trạng thái hiện tại. Gói dịch vụ của khách hàng đã được nâng cấp hai lần kể từ lúc đó. Các giá trị cũ đã biến mất vĩnh viễn.

Chúng tôi không làm gì bất thường cả; bảng dữ liệu có các cột created_at và updated_at như mọi dự án khác. Nhưng những cột đó chỉ cho bạn biết khi nào dòng dữ liệu thay đổi, chứ không cho biết nó trông như thế nào trước khi thay đổi.

Tôi liên tục gặp phải các biến thể của vấn đề này:

• "Giá sản phẩm này vào ngày Black Friday năm ngoái là bao nhiêu?"
• "Quyền hạn của nhân viên này khi thực hiện hành động đó là gì?"
• "Hợp đồng này nói gì trước khi được sửa đổi?"

Câu trả lời luôn là: chúng tôi không biết.

Tại sao các giải pháp truyền thống gây đau đầu

Hầu hết các đội phát triển xử lý vấn đề này theo một trong ba cách, và không cách nào thực sự lý tưởng.

1. Bảng nhật ký kiểm toán (Audit log table): Một bảng riêng ghi lại mọi thay đổi dưới dạng sự kiện. Cách này tốt cho việc tuân thủ quy định nhưng cực kỳ tồi tệ để truy vấn. Việc tái tạo trạng thái tại một thời điểm trong quá khứ đòi hỏi phải phát lại một chuỗi sự kiện, vừa phức tạp vừa chậm chạp.
2. Xóa mềm với phân loại phiên bản: Thêm các cột is_current, version, superseded_at vào bảng chính. Nó nhanh chóng trở nên lộn xộn, làm ô nhiễm các truy vấn của bạn, và mỗi lập trình viên trong đội sẽ thực hiện nó theo một cách hơi khác nhau.
3. Chấp nhận mất mát: Cách tiếp cận phổ biến nhất. Bạn tự nhủ sẽ sửa sau, nhưng bạn không bao giờ làm vậy.

Điểm chung của cả ba cách tiếp cận là chúng thiếu một mô hình dữ liệu sạch sẽ và dễ truy vấn cho chính thời gian.

Mô hình Song thời gian (Bitemporal Modeling)

Có một khái niệm đã được thiết lập tốt trong lý thuyết cơ sở dữ liệu gọi là mô hình song thời gian (bitemporal modeling). Nó theo dõi hai trục thời gian độc lập cho mọi bản ghi:

• Thời gian hiệu lực (Valid time): Khi sự việc là đúng trong thế giới thực.
• Thời gian giao dịch (Transaction time): Khi bạn ghi nhận nó vào cơ sở dữ liệu.

Hầu hết các hệ thống kiểm tra chỉ theo dõi thời gian giao dịch. Chrono-temporal theo dõi cả hai, điều này cho phép bạn trả lời các câu hỏi như "Vào ngày 1 tháng 1, chúng ta đã biết điều gì về những gì đúng vào tháng 7 năm ngoái?". Điều này cực kỳ quan trọng đối với việc tuân thủ, sửa chữa quy hồi và gỡ lỗi.

Ý tưởng cốt lõi rất đơn giản: thay vì đột biến bản ghi tại chỗ (in-place mutation), bạn lưu trữ các bản chụp nhanh (snapshots) được phân phiên bản với các khoảng thời gian:

Thay vì thay đổi bản ghi, bạn lưu trữ các bản chụp có phạm vi thời gian:

• valid_to = NULL → Bản ghi hiện tại.
• Cập nhật = Đóng bản ghi hiện tại + Chèn phiên bản mới.
• Dữ liệu cũ không bao giờ bị phá hủy.

Giới thiệu Chrono Temporal

Chrono Temporal là một lớp dữ liệu thời gian (temporal data layer) dành cho PostgreSQL cho phép bạn truy vấn dữ liệu tại bất kỳ thời điểm nào, theo dõi thay đổi và tính toán sự khác biệt (diff).

Tôi đã quyết định xây dựng nó như một thư viện Python tái sử dụng thay vì phải giải quyết từ đầu ở mỗi dự án. Các mục tiêu bao gồm:

• Tích hợp vào bất kỳ dự án PostgreSQL hiện có nào mà không cần thay đổi kiến trúc.
• API Python async sạch sẽ dựa trên SQLAlchemy 2.0.
• Hỗ trợ truy vấn du hành thời gian, lịch sử đầy đủ và tính năng diff sẵn có.
• Phân phối như một gói PyPI chuẩn.

Bạn có thể cài đặt dễ dàng:

pip install chrono-temporal

Mô hình cốt lõi

Mọi thứ được lưu trữ trong một bảng duy nhất temporal_records:

class TemporalRecord(Base):
    __tablename__ = "temporal_records"

    id = Column(Integer, primary_key=True)
    entity_type = Column(String(100), nullable=False)  # ví dụ: "user", "product"
    entity_id = Column(String(255), nullable=False)    # ID của thực thể
    valid_from = Column(DateTime(timezone=True), nullable=False)
    valid_to = Column(DateTime(timezone=True), nullable=True)  # NULL = đang hoạt động
    data = Column(JSON, nullable=False)                # Toàn bộ dữ liệu
    notes = Column(Text, nullable=True)

Sự kết hợp entity_type + entity_id xác định thực thể của bạn. Trường data là một payload JSON, hoàn toàn linh hoạt, hoạt động với mọi thực thể trong hệ thống mà không cần thay đổi lược đồ (schema).

API Dịch vụ

Lớp TemporalService bao bọc tất cả logic truy vấn thời gian:

from chrono_temporal import TemporalService, TemporalRecordCreate

svc = TemporalService(session)

# Tạo bản ghi có phiên bản
await svc.create(TemporalRecordCreate(
    entity_type="user",
    entity_id="user_001",
    valid_from=datetime(2024, 1, 1, tzinfo=timezone.utc),
    data={"name": "Daniel", "plan": "free", "email": "[email protected]"}
))

# Truy vấn du hành thời gian — nó trông như thế nào vào tháng 3/2024?
records = await svc.get_at_point_in_time(
    "user", "user_001",
    datetime(2024, 3, 1, tzinfo=timezone.utc)
)

# Lịch sử đầy đủ — mọi phiên bản từng tồn tại
history = await svc.get_history("user", "user_001")

# Diff — cái gì đã thay đổi giữa hai ngày?
diff = await svc.get_diff(
    "user", "user_001",
    datetime(2024, 1, 1, tzinfo=timezone.utc),
    datetime(2025, 7, 1, tzinfo=timezone.utc),
)
print(diff["changed"])   # {"plan": {"from": "free", "to": "pro"}}
print(diff["unchanged"]) # ["name", "email"]

Ví dụ thực tế: Lịch sử giá sản phẩm

Để thấy cách nó kết hợp với logic kinh doanh thực tế, dưới đây là một dịch vụ lịch sử giá hoàn chỉnh được xây dựng dựa trên chrono-temporal:

class PriceHistoryService:
    def __init__(self, db):
        self.temporal = TemporalService(db)

    async def update_price(self, product_id, new_price, effective_from=None, reason=None):
        """Cập nhật giá — đóng bản ghi hiện tại và tạo phiên bản mới."""
        effective_from = effective_from or datetime.now(timezone.utc)

        current = (await self.temporal.get_current(ENTITY_TYPE, product_id))[0]
        await self.temporal.close_record(current.id, effective_from)

        return await self.temporal.create(
            TemporalRecordCreate(
                entity_type=ENTITY_TYPE,
                entity_id=product_id,
                valid_from=effective_from,
                data={**current.data, "price": new_price},
                notes=reason or f"Price changed from {current.data['price']} to {new_price}",
            )
        )

    async def get_price_at(self, product_id, as_of):
        """Giá là bao nhiêu vào một ngày cụ thể?"""
        records = await self.temporal.get_at_point_in_time(
            ENTITY_TYPE, product_id, as_of
        )
        return records[0] if records else None

Ứng dụng thực tiễn

Mô hình này có thể áp dụng ở bất cứ đâu mà dữ liệu của bạn có một lịch sử ý nghĩa:

• Hóa đơn đăng ký: Gói dịch vụ nào đang hoạt động trong thời gian có khoản phí bị tranh chấp?
• Định giá thương mại điện tử: Lịch sử giá đầy đủ, giá thấp nhất/nhất từng có, so sánh Black Friday.
• Hệ thống nhân sự: Lịch sử lương, thay đổi vai trò, chuyển bộ phận.
• Hợp đồng và pháp lý: Mỗi lần sửa đổi, điều khoản nói gì tại thời điểm ký kết.
• Tuân thủ: Lịch trình kiểm tra bất biến, có thể bảo vệ trước cơ quan quản lý.

So sánh với các phương pháp khác

• So với Event Sourcing: Event sourcing lưu trữ các sự kiện và dẫn xuất trạng thái bằng cách phát lại chúng. Chrono-temporal lưu trữ các snapshot trực tiếp, giúp truy vấn thời điểm trở nên đơn giản và không cần công cụ phát lại.
• So với Temporal Tables (mở rộng PostgreSQL): temporal_tables xử lý thời gian giao dịch tự động ở cấp cơ sở dữ liệu, nhưng truy vấn nó từ Python rất dài dòng. Chrono-temporal cung cấp một API Python sạch sẽ và xử lý cả thời gian hiệu lực lẫn thời gian giao dịch một cách rõ ràng.
• So với việc tự xây dựng: Bạn có thể tự xây dựng nó. Hầu hết các đội đều làm vậy, một lần, thiếu nhất quán, với các lỗi tinh tế trong các điều kiện biên. Chrono-temporal có bộ kiểm thử đầy đủ bao gồm các trường hợp cạnh.

Kết luận

Chrono Temporal là một thư viện mã nguồn mở mới (19 ngày tuổi) và có sẵn trên PyPI. Lộ trình phát triển trong tương lai bao gồm hỗ trợ Django ORM, endpoint tóm tắt dòng thời gian và phiên bản đám mây được lưu trữ.

Nếu bạn đang gặp phải các trường hợp sử dụng trên, tôi thực sự muốn nghe xem nó phù hợp hoặc không phù hợp với tình huống của bạn như thế nào.

Cài đặt: pip install chrono-temporal GitHub: github.com/Daniel7303/chrono-temporal-api-framework Demo: https://chrono-temporal-ap.onrender.com/docs#/

Thư viện được xây dựng với Python 3.11, SQLAlchemy 2.0, asyncpg, Pydantic 2, FastAPI và PostgreSQL.