This skill builds on db-core and db-core/collection-setup. Read those first.

Custom Adapter Authoring

Each call to an adapter's sync() function starts a sync run. The run owns the callbacks and resources installed by that call until its returned cleanup ends them. A sync run may make several backend requests or open a longer-lived provider session, so do not use “request” or “session” as a synonym for the run.

Setup

import { createCollection } from '@tanstack/db'
import type { CollectionConfig } from '@tanstack/db'

interface MyItem {
  id: string
  name: string
}

interface BackendEvent<T> {
  type: 'insert' | 'update' | 'delete'
  id: string
  data: T
}

function myBackendCollectionOptions<T extends object>(config: {
  endpoint: string
  getKey: (item: T) => string
}): CollectionConfig<T, string> {
  return {
    getKey: config.getKey,
    sync: {
      sync: ({ begin, write, commit, markReady, fail }) => {
        let isInitialSyncComplete = false
        const bufferedEvents: Array<BackendEvent<T>> = []
        const initialSyncAbort = new AbortController()

        // 1. Subscribe to real-time events FIRST
        const unsubscribe = myWebSocket.subscribe(config.endpoint, (event) => {
          if (!isInitialSyncComplete) {
            bufferedEvents.push(event)
            return
          }
          begin()
          write({ type: event.type, key: event.id, value: event.data })
          commit()
        })

        // 2. Fetch initial data
        void fetch(config.endpoint, { signal: initialSyncAbort.signal })
          .then(async (res) => {
            const items = await res.json()
            begin()
            for (const item of items) {
              write({ type: 'insert', value: item })
            }
            commit()

            // 3. Process buffered events
            isInitialSyncComplete = true
            for (const event of bufferedEvents) {
              begin()
              write({ type: event.type, key: event.id, value: event.data })
              commit()
            }

            // 4. Signal that a usable snapshot exists
            markReady()
          })
          .catch((error) => {
            if (initialSyncAbort.signal.aborted) return
            console.error('Initial sync failed:', error)
            // This catch runs only before this run's markReady(). After it,
            // report problems with reportError() and keep the rows. Do not
            // read collection.status: a reload keeps the previous status.
            fail(error)
          })

        // 5. Return cleanup function
        return () => {
          initialSyncAbort.abort()
          unsubscribe()
        }
      },
      rowUpdateMode: 'partial',
    },
    onInsert: async ({ transaction }) => {
      const response = await fetch(config.endpoint, {
        method: 'POST',
        body: JSON.stringify(transaction.mutations[0].modified),
      })
      await waitForServerObservation(response)
    },
    onUpdate: async ({ transaction }) => {
      const mut = transaction.mutations[0]
      const response = await fetch(`${config.endpoint}/${mut.key}`, {
        method: 'PATCH',
        body: JSON.stringify(mut.changes),
      })
      await waitForServerObservation(response)
    },
    onDelete: async ({ transaction }) => {
      const response = await fetch(
        `${config.endpoint}/${transaction.mutations[0].key}`,
        {
          method: 'DELETE',
        },
      )
      await waitForServerObservation(response)
    },
  }
}

Core Patterns

ChangeMessage format

// Insert
write({ type: 'insert', value: item })

// Update (partial — only changed fields)
write({ type: 'update', key: itemId, value: partialItem })

// Update (full row replacement)
write({ type: 'update', key: itemId, value: fullItem })
// Set rowUpdateMode: "full" in sync config

// Delete
write({ type: 'delete', key: itemId, value: item })

On-demand sync with loadSubset

import { parseLoadSubsetOptions } from '@tanstack/db'

syncMode: 'on-demand',
sync: {
  sync: ({ begin, write, commit, markReady, collection }) => {
    const stopSync = subscribeToBackendChanges()
    markReady()

    return {
      cleanup: stopSync,
      loadSubset: async (options) => {
        const { filters, sorts, limit } = parseLoadSubsetOptions(options)
        const items = await api.items.list({
          filters,
          sorts,
          limit,
          offset: options.offset,
          // Translate cursor.whereFrom/whereCurrent expressions for your API.
          cursor: translateCursorExpressions(options.cursor),
        })

        begin()
        for (const item of items) {
          const key = collection.config.getKey(item)
          write(
            collection.has(key)
              ? { type: 'update', key, value: item }
              : { type: 'insert', value: item },
          )
        }
        commit()
      },
    }
  },
  rowUpdateMode: 'full',
}

sync() returns the handlers in a SyncConfigRes object. loadSubset() must write fetched rows through begin() → write() → commit() and resolve void (or return true for an immediate synchronous result); it does not return the fetched rows. parseLoadSubsetOptions() returns only filters, sorts, and limit. Read offset and cursor from the original options. cursor contains query expressions (whereFrom and whereCurrent), not an opaque backend cursor; translate or combine those expressions for your API. Return unloadSubset only when loadSubset creates an ongoing resource, such as a per-subset server subscription, that must be released. A Collection subscription installs its logical owner before calling loadSubset. A synchronous throw fails the load but keeps that owner; when the owner is released, core calls unloadSubset for the failed load too. Make unloadSubset safe after partial setup. A must-refetch can call loadSubset again with the same options. Replay first detaches and releases the old acquisition lease, then starts a fresh acquisition if the logical demand and replay are still current. Core releases the current acquisition when the demand ends.

Managing optimistic state duration

Mutation handlers must not resolve until server changes have synced back to the collection. Five strategies:

Refetch (simplest): await collection.utils.refetch()
Transaction ID: await a utility that observes the mutation's transaction ID in the sync stream, such as Electric's await collection.utils.awaitTxId(txid)
ID-based tracking: await specific record ID appearing in sync stream
Version/timestamp: wait until sync stream catches up to mutation time
Provider method: await backend.waitForPendingWrites()

Persisted sync metadata

The metadata API on the sync config allows adapters to store per-row and per-collection metadata that persists across sync transactions. This is useful for tracking resume tokens, cursors, LSNs, or other adapter-specific state.

The metadata object is available on the sync config argument alongside begin, write, and commit. Core supplies it at runtime, but its public type is optional, so strict TypeScript code must guard it or assert its presence. Without persistence the metadata is in-memory only and does not survive reloads. With persistence, it is durable across sessions.

sync: ({ begin, write, commit, markReady, fail, metadata }) => {
  if (!metadata) throw new Error('Sync metadata API is unavailable')

  // Row metadata: store per-row state (e.g. server version, ETag)
  metadata.row.get(key) // => unknown | undefined
  metadata.row.set(key, { version: 3, etag: 'abc' })
  metadata.row.delete(key)

  // Collection metadata: store per-collection state (e.g. resume cursor)
  metadata.collection.get('cursor') // => unknown | undefined
  metadata.collection.set('cursor', 'token_abc123')
  metadata.collection.delete('cursor')
  metadata.collection.list() // => [{ key: 'cursor', value: 'token_abc123' }]
  metadata.collection.list('resume') // filter by prefix
}

Row metadata writes are tied to the current transaction. Deleting a row also deletes its metadata. An insert sets metadata from message.metadata. A metadata-less insert deletes stale metadata unless metadata.row.set() already queued an explicit value for that key in the same transaction; that queued value wins.

Collection metadata writes staged before truncate() are preserved and commit atomically with the truncate transaction.

Typical usage — resume token:

sync: ({ begin, write, commit, markReady, metadata }) => {
  if (!metadata) throw new Error('Sync metadata API is unavailable')

  const lastCursor = metadata.collection.get('cursor') as string | undefined

  const stream = subscribeFromCursor(lastCursor)
  stream.on('data', (batch) => {
    begin()
    for (const item of batch.items) {
      write({ type: item.type, key: item.id, value: item.data })
    }
    metadata.collection.set('cursor', batch.cursor)
    commit()
  })

  stream.on('ready', () => markReady())
  stream.on('initial-error', (error) => fail(error))
  return () => stream.close()
}

Expression parsing for predicate push-down

import {
  parseWhereExpression,
  parseOrderByExpression,
  extractSimpleComparisons,
} from '@tanstack/db'

// In loadSubset or queryFn:
const comparisons = extractSimpleComparisons(options.where)
// Returns: [{ field: ['name'], operator: 'eq', value: 'John' }]

const orderBy = parseOrderByExpression(options.orderBy)
// Returns: [{ field: ['created_at'], direction: 'desc', nulls: 'last' }]

Common Mistakes

CRITICAL Defining loadSubset beside sync()

Wrong:

sync: {
  sync: ({ markReady }) => markReady(),
  loadSubset: async () => fetch('/items').then((response) => response.json()),
}

Correct: return { loadSubset, cleanup } from sync() and apply loaded rows with the sync transaction primitives, as shown above. Add unloadSubset when each loaded subset owns a resource that must be released.

CRITICAL Not calling markReady() in sync implementation

Wrong:

sync: ({ begin, write, commit }) => {
  fetchData().then((items) => {
    begin()
    items.forEach((item) => write({ type: 'insert', value: item }))
    commit()
    // forgot markReady()!
  })
}

Correct:

sync: ({ begin, write, commit, markReady }) => {
  fetchData().then((items) => {
    begin()
    items.forEach((item) => write({ type: 'insert', value: item }))
    commit()
    markReady()
  })
}

markReady() transitions the collection to "ready" status. Without it, live queries never resolve and useLiveSuspenseQuery hangs forever in Suspense.

If initial sync fails before it produces a usable snapshot and you stop retrying, call fail(error) instead. The status becomes error, readiness waits reject with the supplied cause, and dependent live queries move to error. A later markReady() in the same sync run does nothing; only a new sync run can reach ready. Use collection.reload() to start a new sync run.

After the first markReady(), the rows stay. Report a sync problem with reportError(error) while you keep trying; it sets the node's failureReason. Call reportHealthy() when sync works again; it does nothing when nothing is wrong. When you stop trying, call fail(error): after readiness the node shows an error, and only reload() clears it. reportError() and reportHealthy() throw before the first markReady(). Apps read these reports from collection.health and from each live query's health tree. While a retry is in progress before readiness, report nothing.

A throwing sync function crashes the Collection (status: 'crashed'). Reserve that for broken local state, not for network failures.

Source: docs/guides/collection-options-creator.md

HIGH Race condition: subscribing after initial fetch

Wrong:

sync: ({ begin, write, commit, markReady }) => {
  fetchAll().then((data) => {
    writeAll(data)
    subscribe(onChange) // changes during fetch are LOST
    markReady()
  })
}

Correct:

sync: ({ begin, write, commit, markReady }) => {
  const buffer = []
  subscribe((event) => {
    if (!ready) {
      buffer.push(event)
      return
    }
    begin()
    write(event)
    commit()
  })
  fetchAll().then((data) => {
    writeAll(data)
    ready = true
    buffer.forEach((e) => {
      begin()
      write(e)
      commit()
    })
    markReady()
  })
}

Subscribe to real-time events before fetching initial data. Buffer events during the fetch, then replay them after the initial sync completes.

Source: docs/guides/collection-options-creator.md

HIGH write() called without begin()

Wrong:

onMessage((event) => {
  write({ type: event.type, key: event.id, value: event.data })
  commit()
})

Correct:

onMessage((event) => {
  begin()
  write({ type: event.type, key: event.id, value: event.data })
  commit()
})

Sync data must be written within a transaction (begin → write → commit). Calling write() without begin() throws NoPendingSyncTransactionWriteError.

Source: packages/db/src/collection/sync.ts:110

HIGH Inserting a different value for an existing synced key

An insert for an existing synced key is normalized to an update only when the value is unchanged. A different value throws DuplicateKeySyncError, including for plain custom configs with no utils.

Emit an update, or delete/truncate the old row before inserting the new one.

Tension: Simplicity vs. Correctness in Sync

Getting-started simplicity (localOnly, eager mode) conflicts with production correctness (on-demand sync, race condition prevention, proper markReady handling). Agents optimizing for quick setup tend to skip buffering, markReady, and cleanup functions.

See also: db-core/collection-setup/SKILL.md — for built-in adapter patterns to model after.

@tanstack/db

db-core/custom-adapter

Custom Adapter Authoring#

Setup#

Core Patterns#

ChangeMessage format#

On-demand sync with loadSubset#

Managing optimistic state duration#

Persisted sync metadata#

Expression parsing for predicate push-down#

Common Mistakes#

CRITICAL Defining loadSubset beside sync()#

CRITICAL Not calling markReady() in sync implementation#

HIGH Race condition: subscribing after initial fetch#

HIGH write() called without begin()#

HIGH Inserting a different value for an existing synced key#

Tension: Simplicity vs. Correctness in Sync#

Custom Adapter Authoring

Setup

Core Patterns

ChangeMessage format

On-demand sync with loadSubset

Managing optimistic state duration

Persisted sync metadata

Expression parsing for predicate push-down

Common Mistakes

CRITICAL Defining loadSubset beside sync()

CRITICAL Not calling markReady() in sync implementation

HIGH Race condition: subscribing after initial fetch

HIGH write() called without begin()

HIGH Inserting a different value for an existing synced key

Tension: Simplicity vs. Correctness in Sync