DEV Community

木头人
木头人

Posted on

Ditch Electron: Building a Local-First Sync Engine with Turso & SQLModel

#electrobun #robyn #turso #htmx

Part 4 of the ERTH Architecture Series: Implementing offline-first database schemas, UUIDv7 indexing, and cloud-edge SQLite replication.

The Cover Page

In the third part of this series, we established a Zero-Trust shield around our Python backend (Robyn) using in-memory Opaque Tokens.

Our application is responsive, self-healing, and secure. But it is still stateless. If the user closes the application, all data disappears.

To turn our assistant into a true personal database, we need persistence. But we don't want a heavy database engine like PostgreSQL running on the user's laptop, nor do we want a simple offline SQLite file that isolates data on a single machine.

We want a Local-First Sync Engine:

  1. 0.1ms latency: All reads and writes must happen instantly on local disk.
  2. Offline-first: The app must work fully on an airplane or subway without internet.
  3. Seamless Cloud Sync: When network is restored, data must replicate bidirectionally to the cloud.
  4. Multi-device roaming: Sync across laptops, tablets, and phones without data conflicts.

In this fourth post, we will build this layer using Turso (libSQL) and SQLModel, leveraging UUIDv7 and Tombstone deletion patterns to handle offline concurrency.

The Synchronization Architecture

Our Local-First database architecture relies on Turso's embedded replica synchronization:

Turso Sync Architecture

To ensure this works smoothly without database corruption, we must redesign our schemas using two core concepts: UUIDv7 and Tombstones.

Concept 1: Ditch Autoincrement IDs, Use UUIDv7

In single-machine apps, INTEGER PRIMARY KEY AUTOINCREMENT is standard. But in local-first apps, you may write data on your phone and laptop concurrently while offline. If both assign ID=5, they will collide and corrupt the synchronization channel when you reconnect.

To prevent collisions, we must generate decentralized, unique primary keys. We select UUIDv7 (RFC 9562):

  • Why not UUIDv4? UUIDv4 is completely random. In database engines like SQLite, inserting random strings forces index B-Trees to split continuously, slowing down I/O performance.
  • Why UUIDv7? UUIDv7 combines a millisecond timestamp with random bytes. This means it is globally unique (no collisions) but monotonically increasing over time. SQLite B-Tree indexes can insert new keys sequentially, preserving high-speed write performance.

Concept 2: The Tombstone Deletion Pattern

If client A deletes a record physically (DELETE FROM todos WHERE id=5) while offline, client B doesn't know it was deleted. During sync, the cloud sees that client B has task 5 but client A does not, and it will "resurrect" the deleted task.

To avoid these "ghost records," we must never physically delete data. Instead, we use Tombstones:

  • We add an is_deleted integer flag.
  • Deletions update is_deleted=1 and update the updated_at timestamp.
  • The sync engine propagates this update. When reading lists, we query WHERE is_deleted=0.

Step 1: Implementing the SQLModel Layer in Python

We install SQLModel, which combines SQLAlchemy's database mapping with Pydantic's data validation. Here is our complete database initialization and operations layer in Python:

# backend/db.py
import os
import time
import uuid
from sqlmodel import Field, SQLModel, create_engine, Session, select

# Configuration Eviction: Pull database connection string from environment
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///local_edge.db")

connect_args = {}
if DATABASE_URL.startswith("sqlite"):
    connect_args = {"check_same_thread": False}

engine = create_engine(DATABASE_URL, connect_args=connect_args)

class Todo(SQLModel, table=True):
    """Data contract model for tasks"""
    __tablename__ = "todos"

    id: str = Field(primary_key=True)
    title: str
    is_completed: int = Field(default=0)
    is_deleted: int = Field(default=0)  # Tombstone flag
    created_at: int
    updated_at: int

def generate_uuidv7() -> str:
    """Generate an RFC 9562 compatible UUIDv7 string"""
    timestamp_ms = int(time.time() * 1000)
    timestamp_bytes = timestamp_ms.to_bytes(6, byteorder='big')
    rand_bytes = bytearray(os.urandom(10))

    # Structure version (7)
    rand_a = int.from_bytes(rand_bytes[0:2], byteorder='big') & 0x0FFF
    time_hi_and_version = (7 << 12) | rand_a

    # Structure variant (2)
    rand_b = int.from_bytes(rand_bytes[2:4], byteorder='big') & 0x3FFF
    clk_seq_and_variant = 0x8000 | rand_b
    node = rand_bytes[4:10]

    val = uuid.UUID(fields=(
        int.from_bytes(timestamp_bytes[0:4], byteorder='big'),
        int.from_bytes(timestamp_bytes[4:6], byteorder='big'),
        time_hi_and_version,
        clk_seq_and_variant >> 8,
        clk_seq_and_variant & 0xFF,
        int.from_bytes(node, byteorder='big')
    ))
    return str(val)

async def init_db():
    """Verify local schema and apply updates at startup"""
    SQLModel.metadata.create_all(engine)

async def get_active_todos() -> list:
    """Fetch active todos sorted by creation time"""
    with Session(engine) as session:
        statement = select(Todo).where(Todo.is_deleted == 0).order_by(Todo.created_at.desc())
        return [todo.model_dump() for todo in session.exec(statement).all()]

async def add_todo(title: str) -> dict:
    """Persist a new todo item"""
    todo_id = generate_uuidv7()
    now = int(time.time() * 1000)
    todo = Todo(
        id=todo_id, title=title, is_completed=0, is_deleted=0, created_at=now, updated_at=now
    )
    with Session(engine) as session:
        session.add(todo)
        session.commit()
        session.refresh(todo)
        return todo.model_dump()

async def soft_delete_todo(todo_id: str) -> bool:
    """Soft delete using a tombstone flag"""
    with Session(engine) as session:
        todo = session.get(Todo, todo_id)
        if not todo or todo.is_deleted == 1:
            return False
        todo.is_deleted = 1
        todo.updated_at = int(time.time() * 1000)
        session.add(todo)
        session.commit()
        return True
Enter fullscreen mode Exit fullscreen mode

Step 2: Handoff from Local-Only to Cloud Sync

Because we decoupled the database path using Configuration Eviction (reading DATABASE_URL from environment variables), our database can transition instantly from local SQLite to a remote Turso cloud database without changing any Python source code.

For local development and offline use, the system defaults to a local file database:

# Falls back to sqlite:///local_edge.db
uv run python app.py
Enter fullscreen mode Exit fullscreen mode

When the user logs in and establishes a network sync connection to their Turso edge cluster, we spin up the backend by injecting the LibSQL connection credentials:

# Format: sqlite+libsql://<remote_URL>?auth_token=<YOUR_TOKEN>
env DATABASE_URL="sqlite+libsql://erth-assistant-username.turso.io?auth_token=eyJhbGci..." uv run python app.py
Enter fullscreen mode Exit fullscreen mode

Under the hood, SQLModel automatically recognizes the sqlite+libsql driver, connects to the cloud via WebSockets (Hrana protocol), and synchronization begins silently in the background. If the network drops, it falls back to the local replica, preserving 0.1ms query performance.

What’s Next?

Our desktop app now handles persistence, self-healing, and secure cloud synchronization. Now it's time to build the frontend.

But we want to avoid Node.js build complexity. We don't want Webpack, Vite, or React/Vue configurations to slow down frontend-backend iteration.

In our final post (Part 5), we will use HTMX over Robyn to achieve Zero-Build server-driven UI rendering directly inside our ElectroBun desktop window.

📖 Read the Full Book on Leanpub (Includes a free 5-chapter preview edition!)

👉 Explore the open-source code on GitHub

Stay tuned for Part 5!

Top comments (0)