Cloudflare Workers

The @livestore/sync-cf package provides a comprehensive LiveStore sync provider for Cloudflare Workers. It uses Durable Objects for connectivity and, by default, persists events in the Durable Object’s own SQLite. You can optionally use Cloudflare D1 instead. Multiple transports are supported to fit different deployment scenarios.

Architecture

Key responsibilities:

• Worker: Routes sync requests to Durable Objects by storeId, handles auth validation
• Durable Object: Manages sync state, handles push/pull operations, maintains WebSocket connections
• Storage: Persists events in DO SQLite (default) or D1 (optional)

Installation

pnpm add @livestore/sync-cf

Transport modes

The sync provider supports three transport protocols, each optimized for different use cases:

WebSocket transport (Recommended)

Real-time bidirectional communication with automatic reconnection and live pull support.

client-ws.ts

import { makeWsSync } from '@livestore/sync-cf/client'

export const syncBackend = makeWsSync({
  url: 'wss://sync.example.com',
})

HTTP transport

HTTP-based sync with polling for live updates. Requires the enable_request_signal compatibility flag.

client-http.ts

import { makeHttpSync } from '@livestore/sync-cf/client'

export const syncBackend = makeHttpSync({
  url: 'https://sync.example.com',
  livePull: {
    pollInterval: 3000, // Poll every 3 seconds
  },
})

Durable Object RPC transport

Direct RPC communication between Durable Objects (internal use by @livestore/adapter-cloudflare).

client-do-rpc.ts

import type { CfTypes, SyncBackendRpcInterface } from '@livestore/sync-cf/cf-worker'
import { makeDoRpcSync } from '@livestore/sync-cf/client'

declare const state: CfTypes.DurableObjectState
declare const syncBackendDurableObject: CfTypes.DurableObjectStub<SyncBackendRpcInterface>

export const syncBackend = makeDoRpcSync({
  syncBackendStub: syncBackendDurableObject,
  durableObjectState: state,
  durableObjectContext: {
    bindingName: 'CLIENT_DO',
    durableObjectId: state.id.toString(),
  },
})

Client API reference

makeWsSync(options)

Creates a WebSocket-based sync backend client.

Options:

• url - WebSocket URL (supports ws/wss or http/https protocols)
• webSocketFactory? - Custom WebSocket implementation
• ping? - Ping configuration:
  • enabled?: boolean - Enable/disable ping (default: true)
  • requestTimeout?: Duration - Ping timeout (default: 10 seconds)
  • requestInterval?: Duration - Ping interval (default: 10 seconds)

Features:

• Real-time live pull
• Automatic reconnection
• Connection status tracking
• Ping/pong keep-alive

client-ws-options.ts

import { makeWsSync } from '@livestore/sync-cf/client'

export const syncBackend = makeWsSync({
  url: 'wss://sync.example.com',
  ping: {
    enabled: true,
    requestTimeout: 5000,
    requestInterval: 15000,
  },
})

makeHttpSync(options)

Creates an HTTP-based sync backend client with polling for live updates.

Options:

• url - HTTP endpoint URL
• headers? - Additional HTTP headers
• livePull? - Live pull configuration:
  • pollInterval?: Duration - Polling interval (default: 5 seconds)
• ping? - Ping configuration (same as WebSocket)

Features:

• HTTP request/response based
• Polling-based live pull
• Custom headers support
• Connection status via ping

client-http-options.ts

import { makeHttpSync } from '@livestore/sync-cf/client'

export const syncBackend = makeHttpSync({
  url: 'https://sync.example.com',
  headers: {
    Authorization: 'Bearer token',
    'X-Custom-Header': 'value',
  },
  livePull: {
    pollInterval: 2000, // Poll every 2 seconds
  },
})

makeDoRpcSync(options)

Creates a Durable Object RPC-based sync backend (for internal use).

Options:

• syncBackendStub - Durable Object stub implementing SyncBackendRpcInterface
• durableObjectState - The client Durable Object’s ctx (DurableObjectState); scopes live-pull routing to this instance so it resets on reconstruction
• durableObjectContext - Context for RPC callbacks:
  • bindingName - Wrangler binding name for the client DO
  • durableObjectId - Client Durable Object ID

Features:

• Direct RPC communication
• Real-time live pull via callbacks
• Hibernation support

handleSyncUpdateRpc(ctx, payload)

Handles the RPC callback for live-pull updates in Durable Objects. Pass the client DO’s ctx (DurableObjectState) so the update is routed to that instance’s live pull. The syncUpdateRpc method also receives the subscription’s storeId as a required trailing argument: a reconstructed (store-less) DO can re-boot its store from it and catch up rather than drop the update — see the Durable Object adapter’s recovery options.

client-do.ts env.ts schema.ts shared.ts

import { DurableObject } from 'cloudflare:workers'

import { type ClientDoWithRpcCallback, createStoreDoPromise } from '@livestore/adapter-cloudflare'
import { nanoid, type Store, type Unsubscribe } from '@livestore/livestore'
import { handleSyncUpdateRpc } from '@livestore/sync-cf/client'

import type { Env } from './env.ts'
import { schema, tables } from './schema.ts'
import { storeIdFromRequest } from './shared.ts'

type AlarmInfo = {
  isRetry: boolean
  retryCount: number
}

export class LiveStoreClientDO extends DurableObject<Env> implements ClientDoWithRpcCallback {
  override __DURABLE_OBJECT_BRAND: never = undefined as never

  private storeId: string | undefined
  private cachedStore: Store<typeof schema> | undefined
  private storeSubscription: Unsubscribe | undefined
  private readonly todosQuery = tables.todos.select()

  override async fetch(request: Request): Promise<Response> {
    // @ts-expect-error TODO remove casts once CF types are fixed in https://github.com/cloudflare/workerd/issues/4811
    this.storeId = storeIdFromRequest(request)

    const store = await this.getStore()
    await this.subscribeToStore()

    const todos = store.query(this.todosQuery)
    return new Response(JSON.stringify(todos, null, 2), {
      headers: { 'Content-Type': 'application/json' },
    })
  }

  private async getStore() {
    if (this.cachedStore !== undefined) {
      return this.cachedStore
    }

    const storeId = this.storeId ?? nanoid()

    const store = await createStoreDoPromise({
      schema,
      storeId,
      clientId: 'client-do',
      sessionId: nanoid(),
      durableObject: {
        // @ts-expect-error TODO remove once CF types are fixed in https://github.com/cloudflare/workerd/issues/4811
        ctx: this.ctx,
        env: this.env,
        bindingName: 'CLIENT_DO',
      },
      syncBackendStub: this.env.SYNC_BACKEND_DO.get(this.env.SYNC_BACKEND_DO.idFromName(storeId)),
      livePull: true,
    })

    this.cachedStore = store
    return store
  }

  private async subscribeToStore() {
    const store = await this.getStore()

    if (this.storeSubscription === undefined) {
      this.storeSubscription = store.subscribe(this.todosQuery, (todos: ReadonlyArray<typeof tables.todos.Type>) => {
        console.log(`todos for store (${this.storeId})`, todos)
      })
    }

    await this.ctx.storage.setAlarm(Date.now() + 1000)
  }

  override alarm(_alarmInfo?: AlarmInfo): void | Promise<void> {
    return this.subscribeToStore()
  }

  async syncUpdateRpc(payload: Uint8Array<ArrayBuffer>, storeId: string) {
    this.storeId = storeId
    await this.getStore()
    // @ts-expect-error TODO remove once CF types are fixed in https://github.com/cloudflare/workerd/issues/4811
    await handleSyncUpdateRpc(this.ctx, payload)
  }
}

import type { ClientDoWithRpcCallback } from '@livestore/adapter-cloudflare'
import type { CfTypes, SyncBackendRpcInterface } from '@livestore/sync-cf/cf-worker'

export type Env = {
  CLIENT_DO: CfTypes.DurableObjectNamespace<ClientDoWithRpcCallback>
  SYNC_BACKEND_DO: CfTypes.DurableObjectNamespace<SyncBackendRpcInterface>
  DB: CfTypes.D1Database
}

import { Events, makeSchema, Schema, State } from '@livestore/livestore'

export const tables = {
  todos: State.SQLite.table({
    name: 'todos',
    columns: {
      id: State.SQLite.text({ primaryKey: true }),
      text: State.SQLite.text({ default: '' }),
      completed: State.SQLite.boolean({ default: false }),
      deletedAt: State.SQLite.integer({ nullable: true, schema: Schema.DateFromMillis }),
    },
  }),
}

export const events = {
  todoCreated: Events.synced({
    name: 'v1.TodoCreated',
    schema: Schema.Struct({ id: Schema.String, text: Schema.String }),
  }),
  todoCompleted: Events.synced({
    name: 'v1.TodoCompleted',
    schema: Schema.Struct({ id: Schema.String }),
  }),
  todoUncompleted: Events.synced({
    name: 'v1.TodoUncompleted',
    schema: Schema.Struct({ id: Schema.String }),
  }),
  todoDeleted: Events.synced({
    name: 'v1.TodoDeleted',
    schema: Schema.Struct({
      id: Schema.String,
      deletedAt: Schema.DateFromString.check(Schema.isDateValid()),
    }),
  }),
  todoClearedCompleted: Events.synced({
    name: 'v1.TodoClearedCompleted',
    schema: Schema.Struct({ deletedAt: Schema.DateFromString.check(Schema.isDateValid()) }),
  }),
}

const materializers = State.SQLite.materializers(events, {
  'v1.TodoCreated': ({ id, text }) => tables.todos.insert({ id, text, completed: false }),
  'v1.TodoCompleted': ({ id }) => tables.todos.update({ completed: true }).where({ id }),
  'v1.TodoUncompleted': ({ id }) => tables.todos.update({ completed: false }).where({ id }),
  'v1.TodoDeleted': ({ id, deletedAt }) => tables.todos.update({ deletedAt }).where({ id }),
  'v1.TodoClearedCompleted': ({ deletedAt }) => tables.todos.update({ deletedAt }).where({ completed: true }),
})

const state = State.SQLite.makeState({ tables, materializers })

export const schema = makeSchema({ events, state })

import type { CfTypes } from '@livestore/sync-cf/cf-worker'

export const storeIdFromRequest = (request: CfTypes.Request) => {
  const url = new URL(request.url)
  const storeId = url.searchParams.get('storeId')

  if (storeId === null) {
    throw new Error('storeId is required in URL search params')
  }

  return storeId
}

Server API reference

makeDurableObject(options)

Creates a sync backend Durable Object class.

Options:

• onPush? - Callback for push events: (message, context) => void | Promise<void>
• onPushRes? - Callback for push responses: (message) => void | Promise<void>
• onPull? - Callback for pull requests: (message, context) => void | Promise<void>
• onPullRes? - Callback for pull responses: (message) => void | Promise<void>
• storage? - Storage engine: { _tag: 'do-sqlite' } | { _tag: 'd1', binding: string } (default: do-sqlite)
• enabledTransports? - Set of enabled transports: Set<'http' | 'ws' | 'do-rpc'>
• otel? - OpenTelemetry configuration:
  • baseUrl? - OTEL endpoint URL
  • serviceName? - Service name for traces

do-sync-backend.ts

import { makeDurableObject } from '@livestore/sync-cf/cf-worker'

const hasUserId = (p: unknown): p is { userId: string } =>
  typeof p === 'object' && p !== undefined && p !== null && 'userId' in p

export class SyncBackendDO extends makeDurableObject({
  onPush: async (message, { storeId, payload }) => {
    console.log(`Push to store ${storeId}:`, message.batch)

    // Custom business logic
    if (hasUserId(payload) === true) {
      await Promise.resolve()
    }
  },
  onPull: async (_message, { storeId }) => {
    console.log(`Pull from store ${storeId}`)
  },
  enabledTransports: new Set(['ws', 'http']), // Disable DO RPC
  otel: {
    baseUrl: 'https://otel.example.com',
    serviceName: 'livestore-sync',
  },
}) {}

makeWorker(options)

Creates a complete Cloudflare Worker for the sync backend.

Options:

• syncBackendBinding - Durable Object binding name defined in wrangler.toml
• validatePayload? - Payload validation function: (payload, context) => void | Promise<void>
• enableCORS? - Enable CORS headers (default: false)

makeWorker is a quick way to get started in simple demos. In most production workers you typically want to share routing logic with other endpoints, so prefer wiring your own fetch handler and call handleSyncRequest when you detect a sync request. A minimal example:

worker-minimal.ts env.ts

import type { CFWorker, CfTypes } from '@livestore/sync-cf/cf-worker'
import { handleSyncRequest, matchSyncRequest } from '@livestore/sync-cf/cf-worker'

import type { Env } from './env.ts'

export default {
  fetch: async (request: CfTypes.Request, env: Env, ctx: CfTypes.ExecutionContext) => {
    const searchParams = matchSyncRequest(request)

    if (searchParams !== undefined) {
      return handleSyncRequest({
        request,
        searchParams,
        env,
        ctx,
        syncBackendBinding: 'SYNC_BACKEND_DO',
      })
    }

    // Custom routes, assets, etc.
    return new Response('Not found', { status: 404 }) as unknown as CfTypes.Response
  },
} satisfies CFWorker<Env>

import type { CfTypes, SyncBackendRpcInterface } from '@livestore/sync-cf/cf-worker'

export interface Env {
  SYNC_BACKEND_DO: CfTypes.DurableObjectNamespace<SyncBackendRpcInterface>
}

worker-makeWorker.ts

import { makeWorker } from '@livestore/sync-cf/cf-worker'

export default makeWorker({
  syncBackendBinding: 'SYNC_BACKEND_DO',
  validatePayload: (payload, { storeId }) => {
    // Simple token-based guard at connection time
    const hasAuthToken = typeof payload === 'object' && payload !== null && 'authToken' in payload
    if (hasAuthToken === false) {
      throw new Error('Missing auth token')
    }
    if ((payload as any).authToken !== 'insecure-token-change-me') {
      throw new Error('Invalid auth token')
    }
    console.log(`Validated connection for store: ${storeId}`)
  },
  enableCORS: true,
})

handleSyncRequest(args)

Handles sync backend HTTP requests in custom workers.

Options:

• request - The incoming request
• searchParams - Parsed sync request parameters
• env - Worker environment
• ctx - Worker execution context
• syncBackendBinding - Durable Object binding name defined in wrangler.toml
• headers? - Response headers
• validatePayload? - Payload validation function

worker-handleSyncRequest.ts env.ts

import type { CFWorker, CfTypes } from '@livestore/sync-cf/cf-worker'
import { handleSyncRequest, matchSyncRequest } from '@livestore/sync-cf/cf-worker'

import type { Env } from './env.ts'

export default {
  fetch: async (request: CfTypes.Request, env: Env, ctx: CfTypes.ExecutionContext) => {
    const searchParams = matchSyncRequest(request)

    if (searchParams !== undefined) {
      return handleSyncRequest({
        request,
        searchParams,
        env,
        ctx,
        syncBackendBinding: 'SYNC_BACKEND_DO',
        headers: { 'X-Custom': 'header' },
        validatePayload: (payload, { storeId }) => {
          // Custom validation logic
          if (!(typeof payload === 'object' && payload !== null && 'authToken' in payload)) {
            throw new Error('Missing auth token')
          }
          console.log('Validating store', storeId)
        },
      })
    }

    return new Response('Not found', { status: 404 }) as unknown as CfTypes.Response
  },
} satisfies CFWorker<Env>

import type { CfTypes, SyncBackendRpcInterface } from '@livestore/sync-cf/cf-worker'

export interface Env {
  SYNC_BACKEND_DO: CfTypes.DurableObjectNamespace<SyncBackendRpcInterface>
}

matchSyncRequest(request)

Parses and validates sync request search parameters.

Returns the decoded search params or undefined if the request is not a LiveStore sync request.

match-sync.ts

import type { CfTypes } from '@livestore/sync-cf/cf-worker'
import { matchSyncRequest } from '@livestore/sync-cf/cf-worker'

declare const request: CfTypes.Request

const searchParams = matchSyncRequest(request)
if (searchParams !== undefined) {
  const { storeId, payload, transport } = searchParams
  console.log(`Sync request for store ${storeId} via ${transport}`)
  console.log(payload)
}

Configuration

Wrangler configuration

Configure your wrangler.toml for sync backend deployment (default: DO SQLite storage):

name = "livestore-sync"
main = "./src/worker.ts"
compatibility_date = "2025-05-07"
compatibility_flags = [
  "enable_request_signal", # Required for HTTP streaming
]

[[durable_objects.bindings]]
name = "SYNC_BACKEND_DO"
class_name = "SyncBackendDO"

[[migrations]]
tag = "v1"
new_sqlite_classes = ["SyncBackendDO"]

To use D1 instead of DO SQLite, add a D1 binding and reference it from makeDurableObject({ storage: { _tag: 'd1', binding: '...' } }):

[[d1_databases]]
binding = "DB"
database_name = "livestore-sync"
database_id = "your-database-id"

Environment variables

Required environment bindings:

env.ts

import type { CfTypes, SyncBackendRpcInterface } from '@livestore/sync-cf/cf-worker'

export interface Env {
  SYNC_BACKEND_DO: CfTypes.DurableObjectNamespace<SyncBackendRpcInterface>
}

Transport protocol details

LiveStore identifies sync requests purely by search parameters; the request path does not matter. Use matchSyncRequest(request) to detect sync traffic.

Required search parameters:

Param	Type	Required	Description
storeId	string	Yes	Target LiveStore identifier.
transport	'ws' | 'http'	Yes	Transport protocol selector.
payload	JSON (URI-encoded)	No	Arbitrary JSON used for auth/tenant routing; validated in validatePayload.

Examples (any path):

• WebSocket: https://sync.example.com?storeId=abc&transport=ws (must include Upgrade: websocket)
• HTTP: https://sync.example.com?storeId=abc&transport=http

Notes:

• For transport=ws, if the request is not a WebSocket upgrade, the backend returns 426 Upgrade Required.
• transport='do-rpc' is internal for Durable Object RPC and not exposed via URL parameters.

Data storage

By default, events are stored in the Durable Object’s SQLite with tables following the pattern:

eventlog_{PERSISTENCE_FORMAT_VERSION}_{storeId}

You can opt into D1 with the same table shape. The persistence format version is automatically managed and incremented when the storage schema changes.

Storage engines

• DO SQLite (default)
  • Pros: easiest deploy (no D1), data co-located with the DO, lowest latency
  • Cons: not directly inspectable outside the DO; operational tooling must go through the DO
• D1 (optional)
  • Pros: inspectable using D1 tools/clients; enables cross-store analytics outside DOs
  • Cons: extra hop, JSON response size considerations; requires D1 provisioning

Deployment

Deploy to Cloudflare Workers:

# Deploy the worker
npx wrangler deploy

# Create D1 database
npx wrangler d1 create livestore-sync

# Run migrations if needed
npx wrangler d1 migrations apply livestore-sync

Local development

Run locally with Wrangler:

# Start local development server
npx wrangler dev

# Access local D1 database
# Located at: .wrangler/state/d1/miniflare-D1DatabaseObject/XXX.sqlite

Examples

Basic WebSocket client

basic-ws-client.ts schema.ts

import { makeWorker } from '@livestore/adapter-web/worker'
import { makeWsSync } from '@livestore/sync-cf/client'

import { schema } from './schema.ts'

makeWorker({
  schema,
  sync: {
    backend: makeWsSync({
      url: 'wss://sync.example.com',
    }),
  },
})

import { Events, makeSchema, Schema, State } from '@livestore/livestore'

export const tables = {
  todos: State.SQLite.table({
    name: 'todos',
    columns: {
      id: State.SQLite.text({ primaryKey: true }),
      text: State.SQLite.text({ default: '' }),
      completed: State.SQLite.boolean({ default: false }),
      deletedAt: State.SQLite.integer({ nullable: true, schema: Schema.DateFromMillis }),
    },
  }),
}

export const events = {
  todoCreated: Events.synced({
    name: 'v1.TodoCreated',
    schema: Schema.Struct({ id: Schema.String, text: Schema.String }),
  }),
  todoCompleted: Events.synced({
    name: 'v1.TodoCompleted',
    schema: Schema.Struct({ id: Schema.String }),
  }),
  todoUncompleted: Events.synced({
    name: 'v1.TodoUncompleted',
    schema: Schema.Struct({ id: Schema.String }),
  }),
  todoDeleted: Events.synced({
    name: 'v1.TodoDeleted',
    schema: Schema.Struct({
      id: Schema.String,
      deletedAt: Schema.DateFromString.check(Schema.isDateValid()),
    }),
  }),
  todoClearedCompleted: Events.synced({
    name: 'v1.TodoClearedCompleted',
    schema: Schema.Struct({ deletedAt: Schema.DateFromString.check(Schema.isDateValid()) }),
  }),
}

const materializers = State.SQLite.materializers(events, {
  'v1.TodoCreated': ({ id, text }) => tables.todos.insert({ id, text, completed: false }),
  'v1.TodoCompleted': ({ id }) => tables.todos.update({ completed: true }).where({ id }),
  'v1.TodoUncompleted': ({ id }) => tables.todos.update({ completed: false }).where({ id }),
  'v1.TodoDeleted': ({ id, deletedAt }) => tables.todos.update({ deletedAt }).where({ id }),
  'v1.TodoClearedCompleted': ({ deletedAt }) => tables.todos.update({ deletedAt }).where({ completed: true }),
})

const state = State.SQLite.makeState({ tables, materializers })

export const schema = makeSchema({ events, state })

Custom worker with authentication

worker-auth.ts

import { makeDurableObject, makeWorker } from '@livestore/sync-cf/cf-worker'

export class SyncBackendDO extends makeDurableObject({
  onPush: async (message, { storeId }) => {
    // Log all sync events
    console.log(`Store ${storeId} received ${message.batch.length} events`)
  },
}) {}

const hasStoreAccess = (_userId: string, _storeId: string): boolean => true

export default makeWorker({
  syncBackendBinding: 'SYNC_BACKEND_DO',
  validatePayload: (payload, { storeId }) => {
    if (!(typeof payload === 'object' && payload !== null && 'userId' in payload)) {
      throw new Error('User ID required')
    }

    // Validate user has access to store
    if (hasStoreAccess((payload as any).userId as string, storeId) === false) {
      throw new Error('Unauthorized access to store')
    }
  },
  enableCORS: true,
})

Multi-Transport Setup

multi-transport.ts

import { makeDurableObject } from '@livestore/sync-cf/cf-worker'

type Transport = 'http' | 'ws' | 'do-rpc'

const getTransportFromContext = (ctx: unknown): Transport => {
  if (typeof ctx === 'object' && ctx !== null && 'transport' in (ctx as any)) {
    const t = (ctx as any).transport
    if (t === 'http' || t === 'ws' || t === 'do-rpc') return t
  }
  return 'http'
}

export class SyncBackendDO extends makeDurableObject({
  // Enable all transport modes
  enabledTransports: new Set<Transport>(['http', 'ws', 'do-rpc']),

  onPush: async (message, context) => {
    const transport = getTransportFromContext(context)
    console.log(`Push via ${transport}:`, message.batch.length)
  },
}) {}