SQLite Litestream Clean Architecture

SQLite Litestream

SQLite Litestream Clean Architecture WAL Replication S3 Backup Disaster Recovery Point-in-time Single Server Edge Computing Production

Litestream Setup

# === Litestream Installation and Config ===

# Install
# curl -fsSL https://github.com/benbjohnson/litestream/releases/download/v0.3.13/litestream-v0.3.13-linux-amd64.tar.gz | tar xz
# sudo mv litestream /usr/local/bin/

# litestream.yml
# dbs:
#   - path: /data/app.db
#     replicas:
#       - type: s3
#         bucket: my-backup-bucket
#         path: app.db
#         region: ap-southeast-1
#         retention: 720h        # 30 days
#         sync-interval: 1s
#         snapshot-interval: 4h

# Start replication
# litestream replicate -config litestream.yml

# Restore database
# litestream restore -config litestream.yml -o /data/app.db

# Point-in-time restore
# litestream restore -config litestream.yml -o /data/app.db -timestamp 2024-01-15T10:30:00Z

# Systemd service
# [Unit]
# Description=Litestream Replication
# After=network.target
#
# [Service]
# Type=simple
# ExecStart=/usr/local/bin/litestream replicate -config /etc/litestream.yml
# Restart=always
# RestartSec=5
# Environment=AWS_ACCESS_KEY_ID=xxx
# Environment=AWS_SECRET_ACCESS_KEY=xxx
#
# [Install]
# WantedBy=multi-user.target

from dataclasses import dataclass

@dataclass
class ReplicaConfig:
    destination: str
    type: str
    retention: str
    sync_interval: str
    cost_gb_month: float
    latency: str

replicas = [
    ReplicaConfig("AWS S3", "s3", "30 days", "1s", 0.023, "~100ms"),
    ReplicaConfig("Google Cloud Storage", "gcs", "30 days", "1s", 0.020, "~100ms"),
    ReplicaConfig("Azure Blob Storage", "abs", "30 days", "1s", 0.018, "~100ms"),
    ReplicaConfig("S3-compatible (MinIO)", "s3", "30 days", "1s", 0.0, "~10ms (local)"),
    ReplicaConfig("Backblaze B2", "s3", "30 days", "1s", 0.005, "~150ms"),
]

print("=== Replica Destinations ===")
for r in replicas:
    print(f"  [{r.destination}] Type: {r.type}")
    print(f"    Retention: {r.retention} | Sync: {r.sync_interval}")
    print(f"    Cost: /GB/month | Latency: {r.latency}")

Clean Architecture

# === Clean Architecture with SQLite ===

# Project Structure:
# app/
# ├── domain/          # Entity & Business Rules
# │   ├── entity.py
# │   └── repository.py  # Interface
# ├── usecase/         # Application Logic
# │   └── user_usecase.py
# ├── infrastructure/  # SQLite Implementation
# │   └── sqlite_repo.py
# ├── delivery/        # HTTP / CLI
# │   └── http_handler.py
# └── main.py

# domain/entity.py
# @dataclass
# class User:
#     id: int
#     name: str
#     email: str
#     created_at: datetime

# domain/repository.py (Interface)
# from abc import ABC, abstractmethod
# class UserRepository(ABC):
#     @abstractmethod
#     def get_by_id(self, user_id: int) -> User: ...
#     @abstractmethod
#     def create(self, user: User) -> User: ...
#     @abstractmethod
#     def list_all(self) -> list[User]: ...

# infrastructure/sqlite_repo.py
# class SQLiteUserRepository(UserRepository):
#     def __init__(self, db_path: str):
#         self.conn = sqlite3.connect(db_path)
#         self._create_table()
#
#     def get_by_id(self, user_id: int) -> User:
#         row = self.conn.execute(
#             "SELECT id, name, email, created_at FROM users WHERE id = ?",
#             (user_id,)).fetchone()
#         return User(*row) if row else None
#
#     def create(self, user: User) -> User:
#         self.conn.execute(
#             "INSERT INTO users (name, email) VALUES (?, ?)",
#             (user.name, user.email))
#         self.conn.commit()
#         return user

# usecase/user_usecase.py
# class UserUseCase:
#     def __init__(self, repo: UserRepository):
#         self.repo = repo
#     def get_user(self, user_id: int) -> User:
#         return self.repo.get_by_id(user_id)
#     def create_user(self, name: str, email: str) -> User:
#         user = User(id=0, name=name, email=email, created_at=datetime.now())
#         return self.repo.create(user)

@dataclass
class ArchLayer:
    layer: str
    responsibility: str
    depends_on: str
    example: str

layers = [
    ArchLayer("Entity (Domain)", "Business objects and rules",
        "Nothing (innermost layer)",
        "User, Product, Order dataclass"),
    ArchLayer("Repository Interface", "Data access contract",
        "Entity only",
        "UserRepository ABC with get, create, list methods"),
    ArchLayer("Use Case", "Application business logic",
        "Entity + Repository Interface",
        "CreateUserUseCase, GetOrderUseCase"),
    ArchLayer("Infrastructure", "Concrete implementations",
        "Entity + Repository Interface",
        "SQLiteUserRepo, S3FileStorage"),
    ArchLayer("Delivery", "HTTP/CLI/gRPC handlers",
        "Use Case",
        "FastAPI routes, Click CLI commands"),
]

print("=== Clean Architecture Layers ===")
for l in layers:
    print(f"  [{l.layer}]")
    print(f"    Responsibility: {l.responsibility}")
    print(f"    Depends on: {l.depends_on}")
    print(f"    Example: {l.example}")

Production Deployment

# === Production Checklist ===

@dataclass
class CheckItem:
    category: str
    item: str
    config: str
    importance: str

checklist = [
    CheckItem("SQLite", "WAL mode enabled",
        "PRAGMA journal_mode=WAL;", "Critical"),
    CheckItem("SQLite", "Busy timeout set",
        "PRAGMA busy_timeout=5000;", "Critical"),
    CheckItem("SQLite", "Foreign keys enabled",
        "PRAGMA foreign_keys=ON;", "High"),
    CheckItem("SQLite", "Synchronous mode",
        "PRAGMA synchronous=NORMAL;", "High"),
    CheckItem("Litestream", "Replication running",
        "systemctl status litestream", "Critical"),
    CheckItem("Litestream", "Retention configured",
        "retention: 720h (30 days)", "High"),
    CheckItem("Litestream", "Restore tested",
        "litestream restore -o test.db monthly", "Critical"),
    CheckItem("Monitoring", "Replication lag alert",
        "Alert if no WAL segments for 5 minutes", "High"),
    CheckItem("Monitoring", "Database size tracking",
        "Track file size, warn at 80% disk", "Medium"),
    CheckItem("Backup", "S3 versioning enabled",
        "aws s3api put-bucket-versioning", "High"),
    CheckItem("Security", "Database file permissions",
        "chmod 600 app.db; chown app:app app.db", "High"),
    CheckItem("Performance", "Connection pooling",
        "Single writer, multiple readers via WAL", "High"),
]

print("=== Production Checklist ===")
for c in checklist:
    print(f"  [{c.importance}] {c.category}: {c.item}")
    print(f"    Config: {c.config}")

# Sizing guide
sizing = {
    "< 1 GB database": "Single SQLite + Litestream, any VPS",
    "1-10 GB database": "SQLite + Litestream, SSD VPS, 4GB+ RAM",
    "10-50 GB database": "Consider PostgreSQL migration",
    "> 50 GB database": "Definitely PostgreSQL or distributed DB",
    "< 100 writes/sec": "SQLite handles well",
    "100-1000 writes/sec": "SQLite with WAL, tune busy_timeout",
    "> 1000 writes/sec": "Need PostgreSQL or similar",
}

print(f"\n\nSizing Guide:")
for k, v in sizing.items():
    print(f"  [{k}]: {v}")

เคล็ดลับ

• WAL: เปิด WAL mode เสมอ ให้ Read ไม่ Block Write
• Restore Test: ทดสอบ Restore ทุกเดือน ไม่ใช่รอตอนเกิดปัญหา
• Single Writer: SQLite รองรับ Single Writer เท่านั้น ใช้ Mutex ป้องกัน
• S3 Lifecycle: ตั้ง S3 Lifecycle Rule ลบ Backup เก่าอัตโนมัติ
• Clean Arch: แยก Repository Interface ออกจาก Implementation เปลี่ยน DB ง่าย

Litestream คืออะไร

Open Source Replicate SQLite S3 Cloud Storage Real-time WAL Write-Ahead Log Point-in-time Recovery Embedded Single Server Edge Computing

สรุป

SQLite Litestream Clean Architecture WAL Replication S3 Backup Point-in-time Recovery Repository Pattern Single Server Edge Production