Drizzle Helper

Use the optional Drizzle helper to check sync schemas against database row types and wrap mutation handlers.

Install and import

Install Drizzle when you want the optional helper:

pnpm add drizzle-orm@rc

Import from the Drizzle entrypoint:

import { $type, applyOpsWithDrizzle, defineAccount, defineCollection } from 'valtio-sync/drizzle'
import { valtioSync } from 'valtio-sync/server'

Type-checked schema definitions

The Drizzle entrypoint provides schema definition wrappers that check your Zod field map against a Drizzle table's selected row shape:

import { $type, defineAccount, defineCollection, serverOnly } from "valtio-sync/drizzle";
import type { infer } from "valtio-sync/schema";
import { z } from "zod";
import { accountTable, todosTable } from "./db/schema";

export const account = defineAccount({
  dbType: $type<typeof accountTable>(),
  fields: {
    theme: z.enum(["light", "dark"]).default("light"),
  },
});

export const todos = defineCollection({
  dbType: $type<typeof todosTable>(),
  fields: {
    userId: serverOnly(),
    serverVersion: serverOnly(),
    id: z.string(),
    title: z.string().default(""),
    completed: z.boolean().default(false),
  },
});

export const Todo = todos.recordSchema;
export type Todo = infer<typeof todos>;

The table imports in this example refer to your application's Drizzle schema.

Exclude server-only columns

Use serverOnly() for columns that must not be part of the synced record, such as userId and serverVersion above.

Every key in the table's $inferSelect must appear in fields. Fields marked serverOnly() are excluded from:

• The inferred record type.
• Runtime validation.
• Serialization.
• Patch handling.

Use serverOnly() explicitly. An ordinary z.never() is still treated as a normal field schema.

recordSchema excludes server-only columns. It keeps the synced fields' defaults and transforms, along with any definition-level refine callback.

Match database field types

The dbType marker is compile-time only. At runtime the wrappers create the same schema definitions as valtio-sync/schema.

Field keys must exactly match typeof table.$inferSelect, and each Zod output type must be assignable to the matching Drizzle selected value type. Narrower schemas are allowed, such as a Zod enum for a string column. Wider schemas are rejected, such as a nullable Zod field for a non-null Drizzle column.

Mutation handlers

applyOpsWithDrizzle returns handlers for valtioSync. For each mutation, it:

• Uses a transaction when the database provides transaction.
• Runs optional authorization and conflict checks.
• Calls your mutation handler.
• Writes a sync event row.

The database and authentication helpers below are application code that you must provide.

const handlers = applyOpsWithDrizzle({
  db,
  syncEvents: {
    write: async ({ tx, ctx, collection, recordId, op }) => {
      const [event] = await tx
        .insert(syncEvents)
        .values({
          userId: ctx.user.id,
          collection,
          recordId,
          op,
        })
        .returning({ seq: syncEvents.seq })

      return event.seq
    },
  },
  authorize: async ({ ctx, collection, op }) => {
    await assertCanSync(ctx.user, collection, op)
  },
  checkConflict: async ({ tx, ctx, collection, op }) => {
    await assertFreshBaseVersion(tx, ctx.user.id, collection, op)
  },
  handlers: {
    todos: {
      readChanges: async ({ ctx, since }) => readTodoChanges(ctx.user.id, since),
      create: async ({ tx, ctx, record }) => {
        const row = await insertTodo(tx, ctx.user.id, record)
        return { record: row }
      },
      update: async ({ tx, ctx, op, patch }) => {
        const row = await updateTodo(tx, ctx.user.id, op.id, patch)
        return { record: row }
      },
      delete: async ({ tx, ctx, op }) => {
        await deleteTodo(tx, ctx.user.id, op.id)
        return {}
      },
    },
  },
})

const syncServer = valtioSync({
  schema: { account, todos },
  getContext: async (request) => ({ user: await requireUser(request) }),
  handlers,
})

export const POST = syncServer.handle

Choose a server version

If a mutation handler omits serverVersion, the helper uses the sequence returned by syncEvents.write. Return a specific serverVersion when your table has its own per-record version.

syncEvents.write inserts the event and returns its seq. The database can generate this value with an identity or serial column.

A global sequence that always increases works when readChanges filters by account or user and reads events with seq > since.

Reserve a sequence before inserting

If your application reserves the sequence before inserting the event, use the compatibility shape:

syncEvents: {
  table: syncEvents,
  nextSeq: async ({ tx, ctx }) => reserveNextSeq(tx, ctx.user.id),
  toRow: ({ ctx, collection, recordId, op, seq }) => ({
    userId: ctx.user.id,
    seq,
    collection,
    recordId,
    op,
  }),
}

The helper expects a Drizzle-like db with transaction and an insert(table).values(row) shape for the compatibility path. If transaction is unavailable, the callback runs directly against db.

Return remote changes

Read operations are passed through unchanged. Your readChanges and readSnapshot handlers must turn database rows into CollectionChanges.

If you provide readChanges, handle since: null as the first read on a new device. Return changes.mode: "snapshot" when the remaining events cannot reconstruct the complete collection state.

Sync event retention

The Drizzle helper writes events; your app decides when to remove old ones. See Server API for the retention model and how to track active clients.

Store and index events

Keep sync_events small. It is a change feed with limited retention, not a permanent audit log:

sync_events
  user_id or account_id
  seq
  collection
  record_id
  op
  created_at

Index it for the way readChanges reads:

(user_id, seq)

or, for account-scoped apps:

(account_id, seq)

For upserts, the event can point to the current application row. For deletes, the event itself is the tombstone until it is pruned.

Record how far cleanup has reached

Store the last removed sequence for each user or account:

sync_retention
  user_id or account_id
  pruned_through_seq

When a cleanup job deletes events through sequence 100, update pruned_through_seq to 100 in the same job. If a client later asks for changes from before that floor, return an authoritative snapshot:

readChanges: async ({ ctx, since }) => {
  const floor = await readRetainedFloorSeq(ctx.user.id)

  if (since !== null && since < floor) {
    return {
      serverSeq: await readLatestSeq(ctx.user.id),
      changes: {
        mode: 'snapshot',
        upserted: (await readAllTodos(ctx.user.id)).map((row) => ({
          id: row.id,
          serverVersion: row.version,
          record: row,
        })),
        deleted: [],
      },
    }
  }

  return readTodoChanges(ctx.user.id, since)
}

This example delegates since: null to readTodoChanges. That function must return enough data to initialize a new device, as described in First sync on a device.