Electric Adapter Reference

Install

pnpm add @tanstack/electric-db-collection @tanstack/react-db

Required Config

import { createCollection } from '@tanstack/react-db'
import { electricCollectionOptions } from '@tanstack/electric-db-collection'

const collection = createCollection(
  electricCollectionOptions({
    shapeOptions: { url: '/api/todos' },
    getKey: (item) => item.id,
  }),
)

• shapeOptions -- ElectricSQL ShapeStream config; url is the proxy URL to Electric
• getKey -- extracts unique key from each item

Optional Config

Sync Modes

• eager loads the full shape before use.
• on-demand loads only the subsets requested by live queries.
• progressive loads the requested subset first, then completes the full sync in the background.

With SQLite persistence, progressive resume keeps hydrated rows visible while the full Electric stream resumes. On-demand subset snapshots use a bounded refresh wait, so a stalled native long-poll refresh does not block loading forever.

Three Sync Strategies

1. Txid Return (Recommended)

Handler returns { txid }. Client waits for that txid in the Electric stream.

onInsert: async ({ transaction }) => {
  const response = await api.todos.create(transaction.mutations[0].modified)
  return { txid: response.txid }
},

2. awaitMatch (Custom Match)

Use when txids are unavailable. Import isChangeMessage to match on message content.

import { isChangeMessage } from "@tanstack/electric-db-collection"

onInsert: async ({ transaction, collection }) => {
  const newItem = transaction.mutations[0].modified
  await api.todos.create(newItem)
  await collection.utils.awaitMatch(
    (message) =>
      isChangeMessage(message) &&
      message.headers.operation === "insert" &&
      message.value.text === newItem.text,
    5000 // timeout ms, defaults to 3000
  )
},

3. Simple Timeout (Prototyping)

onInsert: async ({ transaction }) => {
  await api.todos.create(transaction.mutations[0].modified)
  await new Promise((resolve) => setTimeout(resolve, 2000))
},

Utility Methods (collection.utils)

• awaitTxId(txid, timeout?) -- wait for txid in Electric stream; default timeout 5s
• awaitMatch(matchFn, timeout?) -- wait for message matching predicate; default timeout 3000ms

Helper Exports

import {
  isChangeMessage,
  isControlMessage,
} from '@tanstack/electric-db-collection'
// isChangeMessage(msg) -- true for insert/update/delete
// isControlMessage(msg) -- true for up-to-date/must-refetch

generateTxId Backend Pattern

The txid must be queried inside the same Postgres transaction as the mutation.

async function generateTxId(tx: any): Promise<number> {
  const result = await tx`SELECT pg_current_xact_id()::xid::text as txid`
  const txid = result[0]?.txid
  if (txid === undefined) throw new Error('Failed to get transaction ID')
  return parseInt(txid, 10)
}

async function createTodo(data) {
  let txid!: number
  const result = await sql.begin(async (tx) => {
    txid = await generateTxId(tx) // INSIDE the transaction
    const [todo] = await tx`INSERT INTO todos ${tx(data)} RETURNING *`
    return todo
  })
  return { todo: result, txid }
}

Querying txid outside the transaction produces a mismatched txid -- awaitTxId stalls indefinitely.

Schema vs Parser: Two Separate Paths

When using Electric with a schema, data enters the collection via two independent paths:

1. Sync path — Electric's ShapeStream applies the parser from shapeOptions. The schema is NOT applied to synced data.
2. Mutation path — insert() and update() run through the collection schema. The parser is not involved.

For types that need transformation (e.g., timestamptz), you need BOTH configured:

const todosCollection = createCollection(
  electricCollectionOptions({
    schema: z.object({
      id: z.string(),
      text: z.string(),
      completed: z.boolean(), // Electric auto-parses bools
      created_at: z.coerce.date(), // mutation path: coerce string → Date
    }),
    shapeOptions: {
      url: '/api/todos',
      parser: {
        timestamptz: (value: string) => new Date(value), // sync path: parse incoming strings
      },
    },
    getKey: (item) => item.id,
  }),
)

Postgres → Electric type handling

Note: z.coerce.date() is Zod-specific. Other StandardSchema libraries have their own coercion patterns.

Proxy Route

Electric collections connect to a proxy URL (shapeOptions.url), not directly to Electric. Your app server must forward shape requests to Electric, passing through the Electric protocol query params.

The proxy route must:

1. Accept GET requests at the URL you specify in shapeOptions.url
2. Forward all query parameters (these are Electric protocol params like offset, handle, live, etc.)
3. Proxy the response (SSE stream) back to the client
4. Optionally add authentication headers or filter params

Implementation depends on your framework — use createServerFn in TanStack Start, API routes in Next.js, loader in Remix, etc. See the @electric-sql/client skills for proxy route examples:

npx @electric-sql/client intent list

Electric Client Skills

For deeper Electric-specific guidance (ShapeStream config, shape filtering, etc.), load the Electric client's built-in skills:

npx @electric-sql/client intent list

Debug Logging

localStorage.debug = 'ts/db:electric'

Complete Example

Always use a schema — types are inferred automatically, avoiding generic placement confusion.

import { createCollection } from '@tanstack/react-db'
import { electricCollectionOptions } from '@tanstack/electric-db-collection'
import { z } from 'zod'

const todoSchema = z.object({
  id: z.string(),
  text: z.string().min(1),
  completed: z.boolean(),
  created_at: z.coerce.date(),
})

const todosCollection = createCollection(
  electricCollectionOptions({
    id: 'todos',
    schema: todoSchema,
    getKey: (item) => item.id,
    shapeOptions: {
      url: '/api/todos',
      params: { table: 'todos' },
      parser: {
        timestamptz: (value: string) => new Date(value), // sync path
      },
    },
    onInsert: async ({ transaction }) => {
      const response = await api.todos.create(transaction.mutations[0].modified)
      return { txid: response.txid }
    },
    onUpdate: async ({ transaction }) => {
      const { original, changes } = transaction.mutations[0]
      const response = await api.todos.update({
        where: { id: original.id },
        data: changes,
      })
      return { txid: response.txid }
    },
    onDelete: async ({ transaction }) => {
      const response = await api.todos.delete(transaction.mutations[0].key)
      return { txid: response.txid }
    },
  }),
)