Electricsql

ElectricSQL

Overview

ElectricSQL is a sync engine that streams partial replicas of Postgres data to local clients via Shapes. It handles the read path — syncing rows from Postgres to the client in real-time using logical replication. Writes flow back through your existing API; Electric syncs the confirmed state back to all connected clients.

When to use: Real-time sync from Postgres to client apps, local-first architectures needing live data from Postgres, replacing polling with streaming updates, apps using TanStack DB with Electric collections, multi-client collaborative apps backed by Postgres.

When NOT to use: Non-Postgres databases, apps needing client-to-server sync built into the engine (Electric handles reads only), simple REST CRUD with no real-time needs, apps that don't benefit from local data.

Quick Reference

Common Mistakes

Delegation

If the local-first skill is available, delegate architecture decisions and sync engine comparison to it.

Otherwise, recommend: pnpm dlx skills add oakoss/agent-skills -s local-first -a claude-code -y

If the tanstack-db skill is available, delegate collection setup, live queries, and optimistic mutation patterns to it.

Otherwise, recommend: pnpm dlx skills add oakoss/agent-skills -s tanstack-db -a claude-code -y

If the tanstack-start skill is available, delegate shape proxy implementation with server functions to it.

Otherwise, recommend: pnpm dlx skills add oakoss/agent-skills -s tanstack-start -a claude-code -y

• Setup and deployment: Use Task agent for Docker and Postgres configuration
• Architecture review: Use Plan agent for auth proxy and write pattern design

References

• Setup, Docker, and Postgres configuration
• Shapes, ShapeStream, and React hooks
• Authentication and security patterns
• Write patterns from online to through-the-DB
• Observability and monitoring
• Sync performance optimization
• Error handling, retry patterns, txid flow, and control messages

Why Proxy

Never expose Electric directly to the internet. Electric has no built-in concept of users, roles, or permissions. Your API server must sit between clients and Electric to:

• Validate authentication tokens
• Inject where clauses that restrict data to the authenticated user
• Enforce per-table access rules
• Add audit logging

Client  -->  Your API (auth + where injection)  -->  Electric  -->  Postgres

Shape Proxy Pattern

Your API validates the request, adds security constraints, and proxies to Electric.

Express Proxy

import express from 'express';

const app = express();
const ELECTRIC_URL = process.env.ELECTRIC_URL ?? 'http://localhost:3000';

app.get('/api/shapes/items', authenticateUser, async (req, res) => {
  const userId = req.user.id;

  const url = new URL(`${ELECTRIC_URL}/v1/shape`);
  url.searchParams.set('table', 'items');
  url.searchParams.set('where', 'user_id = $1');
  url.searchParams.set('params[1]', userId);
  url.searchParams.set('columns', 'id,title,status,created_at');

  if (req.query.offset) {
    url.searchParams.set('offset', req.query.offset as string);
  }
  if (req.query.handle) {
    url.searchParams.set('handle', req.query.handle as string);
  }
  if (req.query.live) {
    url.searchParams.set('live', 'true');
  }

  const response = await fetch(url.toString());

  res.status(response.status);
  for (const [key, value] of response.headers.entries()) {
    res.setHeader(key, value);
  }

  const body = await response.text();
  res.send(body);
});

Hono Proxy

import { Hono } from 'hono';
import { jwt } from 'hono/jwt';

const app = new Hono();
const ELECTRIC_URL = process.env.ELECTRIC_URL ?? 'http://localhost:3000';

app.get(
  '/api/shapes/items',
  jwt({ secret: process.env.JWT_SECRET! }),
  async (c) => {
    const userId = c.get('jwtPayload').sub;

    const url = new URL(`${ELECTRIC_URL}/v1/shape`);
    url.searchParams.set('table', 'items');
    url.searchParams.set('where', 'user_id = $1');
    url.searchParams.set('params[1]', userId);

    const params = ['offset', 'handle', 'live'] as const;
    for (const param of params) {
      const value = c.req.query(param);
      if (value) url.searchParams.set(param, value);
    }

    const response = await fetch(url.toString());
    return new Response(response.body, {
      status: response.status,
      headers: response.headers,
    });
  },
);

TanStack Start Server Function Proxy

import { createServerFn } from '@tanstack/react-start';

const getItemsShape = createServerFn({ method: 'GET' })
  .validator(
    (input: { offset?: string; handle?: string; live?: string }) => input,
  )
  .handler(async ({ data, context }) => {
    const userId = context.user.id;
    const ELECTRIC_URL = process.env.ELECTRIC_URL ?? 'http://localhost:3000';

    const url = new URL(`${ELECTRIC_URL}/v1/shape`);
    url.searchParams.set('table', 'items');
    url.searchParams.set('where', 'user_id = $1');
    url.searchParams.set('params[1]', userId);

    if (data.offset) url.searchParams.set('offset', data.offset);
    if (data.handle) url.searchParams.set('handle', data.handle);
    if (data.live) url.searchParams.set('live', 'true');

    const response = await fetch(url.toString());
    return response;
  });

Gatekeeper Pattern

A middleware layer that checks permissions before allowing shape subscriptions. Useful when multiple shapes need different access rules.

import { type Request, type Response, type NextFunction } from 'express';

type ShapeConfig = {
  table: string;
  allowedRoles: string[];
  userFilter?: (userId: string) => Record<string, string>;
};

const SHAPE_CONFIGS: Record<string, ShapeConfig> = {
  items: {
    table: 'items',
    allowedRoles: ['user', 'admin'],
    userFilter: (userId) => ({
      where: 'user_id = $1',
      'params[1]': userId,
    }),
  },
  analytics: {
    table: 'analytics_events',
    allowedRoles: ['admin'],
  },
  public_posts: {
    table: 'posts',
    allowedRoles: ['user', 'admin'],
    userFilter: () => ({
      where: "visibility = 'public'",
    }),
  },
};

function shapeGatekeeper(shapeName: string) {
  return (req: Request, res: Response, next: NextFunction) => {
    const config = SHAPE_CONFIGS[shapeName];
    if (!config) {
      res.status(404).json({ error: 'Shape not found' });
      return;
    }

    if (!config.allowedRoles.includes(req.user.role)) {
      res.status(403).json({ error: 'Insufficient permissions' });
      return;
    }

    req.shapeConfig = config;
    next();
  };
}

Per-Table Security

Different tables often need different auth rules. Map each shape endpoint to specific access policies.

const ELECTRIC_URL = process.env.ELECTRIC_URL ?? 'http://localhost:3000';

async function proxyShape(
  table: string,
  extraParams: Record<string, string>,
  query: Record<string, string>,
): Promise<Response> {
  const url = new URL(`${ELECTRIC_URL}/v1/shape`);
  url.searchParams.set('table', table);

  for (const [key, value] of Object.entries(extraParams)) {
    url.searchParams.set(key, value);
  }
  for (const param of ['offset', 'handle', 'live']) {
    if (query[param]) url.searchParams.set(param, query[param]);
  }

  return fetch(url.toString());
}

app.get('/api/shapes/my-items', authenticateUser, async (req, res) => {
  const response = await proxyShape(
    'items',
    { where: 'user_id = $1', 'params[1]': req.user.id },
    req.query as Record<string, string>,
  );
  res.status(response.status).send(await response.text());
});

app.get(
  '/api/shapes/team-items',
  authenticateUser,
  requireRole('manager'),
  async (req, res) => {
    const response = await proxyShape(
      'items',
      { where: 'team_id = $1', 'params[1]': req.user.teamId },
      req.query as Record<string, string>,
    );
    res.status(response.status).send(await response.text());
  },
);

app.get(
  '/api/shapes/all-items',
  authenticateUser,
  requireRole('admin'),
  async (req, res) => {
    const response = await proxyShape(
      'items',
      {},
      req.query as Record<string, string>,
    );
    res.status(response.status).send(await response.text());
  },
);

Where Clause Injection

The server adds user-scoped filters so the client never controls what data it receives.

function buildUserShape(
  userId: string,
  table: string,
  additionalWhere?: string,
) {
  const params: Record<string, string> = {
    table,
    where: additionalWhere
      ? `user_id = $1 AND (${additionalWhere})`
      : 'user_id = $1',
    'params[1]': userId,
  };
  return params;
}

The client requests shapes through the proxy without specifying where clauses:

const stream = new ShapeStream({
  url: '/api/shapes/my-items',
  params: {},
});

The proxy injects user_id = $1 before forwarding to Electric. The client has no ability to bypass this filter.

Token Refresh in ShapeStream

ShapeStream connections are long-lived. Tokens expire during the connection lifetime. Use the onError callback to detect 401s and refresh:

import { FetchError, ShapeStream } from '@electric-sql/client';

function createAuthenticatedStream<T>(
  table: string,
  refreshToken: () => Promise<string>,
) {
  let currentToken = '';

  const stream = new ShapeStream<T>({
    url: '/api/shapes/' + table,
    params: { table },
    headers: {
      Authorization: async () => {
        if (!currentToken) {
          currentToken = await refreshToken();
        }
        return `Bearer ${currentToken}`;
      },
    },
    onError: async (error) => {
      if (error instanceof FetchError && error.status === 401) {
        currentToken = await refreshToken();
      }
    },
  });

  return stream;
}

function ItemList() {
  const { getAccessToken } = useAuth();

  const { data, isLoading } = useShape<Item>({
    url: '/api/shapes/items',
    params: { table: 'items' },
    headers: {
      Authorization: async () => `Bearer ${await getAccessToken()}`,
    },
  });

  if (isLoading) return <div>Loading...</div>;

  return (
    <ul>
      {data.map((item) => (
        <li key={item.id}>{item.title}</li>
      ))}
    </ul>
  );
}

Production Checklist

Error Handling

ShapeStream onError Callback

The onError handler controls retry behavior. Return an object to retry with modified params/headers, or return void to stop syncing.

import { ShapeStream, FetchError } from '@electric-sql/client';

const stream = new ShapeStream({
  url: '/api/shapes',
  params: { table: 'items' },
  onError: (error) => {
    if (error instanceof FetchError && error.status === 401) {
      return {
        headers: {
          Authorization: `Bearer ${getNewToken()}`,
        },
      };
    }

    if (error instanceof FetchError && error.status === 403) {
      console.error('Access denied, stopping sync');
      return;
    }

    return {};
  },
});

onError Return Types

The handler can also be async:

onError: async (error) => {
  if (error instanceof FetchError && error.status === 401) {
    const token = await refreshAccessToken();
    return {
      headers: { Authorization: `Bearer ${token}` },
    };
  }
  return {};
};

FetchError Handling for 401

Token refresh is the most common error recovery pattern:

import { ShapeStream, FetchError } from '@electric-sql/client';

let accessToken = await getAccessToken();

const stream = new ShapeStream({
  url: '/api/shapes',
  params: { table: 'items' },
  headers: {
    Authorization: `Bearer ${accessToken}`,
  },
  onError: async (error) => {
    if (error instanceof FetchError && error.status === 401) {
      accessToken = await refreshAccessToken();
      return {
        headers: { Authorization: `Bearer ${accessToken}` },
      };
    }
    console.error('Unrecoverable shape error:', error);
    return;
  },
});

Dynamic Auth Headers Alternative

Instead of handling 401 in onError, use a header function that always provides a fresh token:

const stream = new ShapeStream({
  url: '/api/shapes',
  params: { table: 'items' },
  headers: {
    Authorization: async () => {
      const token = await getAccessToken();
      return `Bearer ${token}`;
    },
  },
});

HTTP 409: Must-Refetch

A 409 response means the shape has been invalidated on the server. The client must discard its local state and re-sync from scratch with a new handle.

ShapeStream handles this automatically — it resets the offset and handle, then restarts the sync. If you subscribe to messages, you will see a must-refetch control message.

stream.subscribe((messages) => {
  for (const msg of messages) {
    if (msg.headers.control === 'must-refetch') {
      clearLocalCache();
    }
  }
});

Causes of 409:

• Server-side shape definition changed
• Electric restarted and lost shape state
• Shape compaction invalidated the previous handle

Automatic Backoff for 5xx and 429

ShapeStream includes built-in exponential backoff for transient errors:

• 5xx responses: Server errors trigger automatic retry with backoff
• 429 Too Many Requests: Rate limiting triggers automatic retry with backoff

No custom onError handling is needed for these cases. The stream reconnects automatically with increasing delays.

If you need to observe these retries:

const stream = new ShapeStream({
  url: '/api/shapes',
  params: { table: 'items' },
  onError: (error) => {
    if (error instanceof FetchError) {
      if (error.status >= 500 || error.status === 429) {
        console.warn(`Transient error ${error.status}, will auto-retry`);
        return {};
      }
    }
    return {};
  },
});

Control Messages

up-to-date

Signals that the client has received all current changes and is caught up with the server:

stream.subscribe((messages) => {
  for (const msg of messages) {
    if (msg.headers.control === 'up-to-date') {
      setIsSynced(true);
    }
  }
});

must-refetch

Shape invalidated. ShapeStream handles this internally by resetting and re-syncing. Subscribe to observe it:

stream.subscribe((messages) => {
  for (const msg of messages) {
    if (msg.headers.control === 'must-refetch') {
      setIsSynced(false);
    }
  }
});

snapshot-end

Marks the end of the initial snapshot batch. All rows from the initial sync have been delivered:

let snapshotComplete = false;

stream.subscribe((messages) => {
  for (const msg of messages) {
    if (msg.headers.control === 'snapshot-end') {
      snapshotComplete = true;
    }
  }
});

Transaction ID (txid) Handling

After a write, the server returns a txid so the client can confirm when Electric has synced the mutation. The txid must come from the same transaction as the mutation.

Correct: txid in Same Transaction

app.post('/api/todos', async (req, res) => {
  const result = await db.transaction(async (tx) => {
    const [todo] = await tx
      .insert(todos)
      .values({ title: req.body.title, userId: req.user.id })
      .returning();

    const [{ txid }] = await tx.execute<{ txid: string }>(
      sql`SELECT pg_current_xact_id()::text AS txid`,
    );

    return { todo, txid };
  });

  res.json(result);
});

Common txid Pitfalls

Querying txid in a separate transaction:

const [todo] = await db.insert(todos).values(data).returning();
const [{ txid }] = await db.execute(
  sql`SELECT pg_current_xact_id()::text AS txid`,
);

This returns the wrong txid because pg_current_xact_id() runs in a different transaction than the insert. The client will wait for a txid that does not correspond to its mutation.

Not awaiting txid confirmation on the client:

const result = await createTodo({ data: newTodo });

Without passing result.txid to the Electric collection's onInsert return, the client cannot confirm when the mutation is synced.

Client-Side txid Flow

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

const todoCollection = createCollection(
  electricCollectionOptions({
    id: 'todos',
    getKey: (row: Todo) => row.id,
    shapeOptions: { url: '/api/shapes/todos' },
    onInsert: async ({ transaction }) => {
      const newTodo = transaction.mutations[0].modified;
      const result = await createTodo({ data: newTodo });
      return { txid: result.txid };
    },
    onUpdate: async ({ transaction }) => {
      const changed = transaction.mutations[0].modified;
      const result = await updateTodo({ data: changed });
      return { txid: result.txid };
    },
  }),
);

Debug Logging for txid Flow

Track the full txid lifecycle to diagnose sync confirmation issues:

app.post('/api/todos', async (req, res) => {
  const result = await db.transaction(async (tx) => {
    const [todo] = await tx
      .insert(todos)
      .values({ title: req.body.title, userId: req.user.id })
      .returning();

    const [{ txid }] = await tx.execute<{ txid: string }>(
      sql`SELECT pg_current_xact_id()::text AS txid`,
    );

    console.log(`[txid] Mutation committed: txid=${txid}, todoId=${todo.id}`);
    return { todo, txid };
  });

  res.json(result);
});

On the client:

onInsert: async ({ transaction }) => {
  const newTodo = transaction.mutations[0].modified;
  const result = await createTodo({ data: newTodo });
  console.log(`[txid] Awaiting sync confirmation: txid=${result.txid}`);
  return { txid: result.txid };
},

Client-Side Sync Failure Tracking

Monitor error rates to detect persistent sync issues:

function createSyncMonitor(stream: ShapeStream) {
  let errorCount = 0;
  let lastErrorAt: Date | null = null;
  const ERROR_THRESHOLD = 5;
  const WINDOW_MS = 60_000;

  stream.subscribe(() => {
    errorCount = 0;
  });

  return {
    onError: (error: Error) => {
      errorCount++;
      lastErrorAt = new Date();

      if (errorCount >= ERROR_THRESHOLD) {
        console.error(`[sync] ${errorCount} errors in monitoring window`, {
          lastError: error.message,
          since: lastErrorAt,
        });
      }

      return {};
    },
    getStatus: () => ({
      errorCount,
      lastErrorAt,
      isHealthy: errorCount < ERROR_THRESHOLD,
    }),
  };
}

const monitor = createSyncMonitor(stream);
const stream = new ShapeStream({
  url: '/api/shapes',
  params: { table: 'items' },
  onError: monitor.onError,
});

Health Monitoring

Electric exposes a health endpoint for liveness checks.

curl http://localhost:3000/v1/health

Response returns 200 OK when Electric is connected to Postgres and ready to serve shapes.

Kubernetes Probes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: electric
spec:
  replicas: 2
  template:
    spec:
      containers:
        - name: electric
          image: electricsql/electric:latest
          ports:
            - containerPort: 3000
            - containerPort: 9090
          env:
            - name: DATABASE_URL
              valueFrom:
                secretKeyRef:
                  name: electric-secrets
                  key: database-url
            - name: ELECTRIC_PROMETHEUS_PORT
              value: '9090'
          livenessProbe:
            httpGet:
              path: /v1/health
              port: 3000
            initialDelaySeconds: 10
            periodSeconds: 15
            failureThreshold: 3
          readinessProbe:
            httpGet:
              path: /v1/health
              port: 3000
            initialDelaySeconds: 5
            periodSeconds: 10
          resources:
            requests:
              memory: '256Mi'
              cpu: '250m'
            limits:
              memory: '1Gi'
              cpu: '1000m'

Docker Compose Health Check

services:
  electric:
    image: electricsql/electric:latest
    environment:
      DATABASE_URL: postgresql://postgres:password@db:5432/app
      ELECTRIC_PROMETHEUS_PORT: 9090
    ports:
      - '3000:3000'
      - '9090:9090'
    healthcheck:
      test: ['CMD', 'curl', '-f', 'http://localhost:3000/v1/health']
      interval: 10s
      timeout: 5s
      retries: 3
      start_period: 10s
    depends_on:
      db:
        condition: service_healthy

Client-Side Observability

Tracking Sync Freshness

import { ShapeStream } from '@electric-sql/client';

type SyncMetrics = {
  lastSyncedAt: number | null;
  initialSyncDurationMs: number | null;
  reconnectionCount: number;
  totalRowsSynced: number;
};

function createMonitoredStream<T extends Record<string, unknown>>(
  url: string,
  table: string,
  where?: string,
): { stream: ShapeStream<T>; getMetrics: () => SyncMetrics } {
  const metrics: SyncMetrics = {
    lastSyncedAt: null,
    initialSyncDurationMs: null,
    reconnectionCount: 0,
    totalRowsSynced: 0,
  };

  const startTime = Date.now();
  let initialSyncComplete = false;

  const params: Record<string, string> = { table };
  if (where) params.where = where;

  const stream = new ShapeStream<T>({ url, params });

  stream.subscribe((messages) => {
    for (const msg of messages) {
      if ('key' in msg) {
        metrics.totalRowsSynced++;
      }

      if ('headers' in msg && msg.headers?.control === 'up-to-date') {
        metrics.lastSyncedAt = Date.now();
        if (!initialSyncComplete) {
          metrics.initialSyncDurationMs = Date.now() - startTime;
          initialSyncComplete = true;
        }
      }
    }
  });

  return { stream, getMetrics: () => ({ ...metrics }) };
}

Detecting Stale Shapes

function monitorStaleness(
  getMetrics: () => SyncMetrics,
  thresholdMs: number,
  onStale: (staleDurationMs: number) => void,
): () => void {
  const intervalId = setInterval(() => {
    const metrics = getMetrics();
    if (metrics.lastSyncedAt === null) return;

    const staleDuration = Date.now() - metrics.lastSyncedAt;
    if (staleDuration > thresholdMs) {
      onStale(staleDuration);
    }
  }, 5000);

  return () => clearInterval(intervalId);
}

const STALE_THRESHOLD_MS = 30_000;

const cleanup = monitorStaleness(
  getMetrics,
  STALE_THRESHOLD_MS,
  (staleDuration) => {
    console.warn(`Shape stale for ${Math.round(staleDuration / 1000)}s`);
    reportToTelemetry('shape_stale', { staleDuration });
  },
);

Reporting to Analytics

type TelemetryEvent = {
  event: string;
  properties: Record<string, unknown>;
  timestamp: number;
};

function reportToTelemetry(
  event: string,
  properties: Record<string, unknown>,
): void {
  const payload: TelemetryEvent = {
    event,
    properties,
    timestamp: Date.now(),
  };

  navigator.sendBeacon('/api/telemetry', JSON.stringify(payload));
}

function reportSyncMetrics(getMetrics: () => SyncMetrics): void {
  const metrics = getMetrics();
  reportToTelemetry('sync_metrics', {
    initialSyncDurationMs: metrics.initialSyncDurationMs,
    reconnectionCount: metrics.reconnectionCount,
    totalRowsSynced: metrics.totalRowsSynced,
    lastSyncedAt: metrics.lastSyncedAt,
  });
}

Prometheus Integration

Enable Prometheus metrics by setting the ELECTRIC_PROMETHEUS_PORT environment variable.

ELECTRIC_PROMETHEUS_PORT=9090

Prometheus Scrape Config

scrape_configs:
  - job_name: electric
    scrape_interval: 15s
    static_configs:
      - targets: ['electric:9090']
    metric_relabel_configs:
      - source_labels: [__name__]
        regex: 'electric_.*'
        action: keep

Key Metrics

Alerting Patterns

Prometheus Alerting Rules

groups:
  - name: electric_alerts
    rules:
      - alert: ElectricDown
        expr: up{job="electric"} == 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: 'Electric instance is down'

      - alert: HighReplicationLag
        expr: electric_replication_lag_bytes > 104857600
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: 'Replication lag exceeds 100MB'
          description: 'WAL lag is {{ $value | humanize1024 }}B'

      - alert: ConnectionSpike
        expr: |
          electric_connections_active
            > 2 * avg_over_time(electric_connections_active[1h])
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: 'Connection count spike detected'

      - alert: NoActiveShapes
        expr: electric_shapes_active == 0
        for: 10m
        labels:
          severity: info
        annotations:
          summary: 'No active shapes for 10 minutes'

      - alert: HighShapeCreationRate
        expr: rate(electric_shape_creation_total[5m]) > 10
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: 'Shapes being created at unusually high rate'

Stalled Sync Detection

Server-Side: Replication Slot Monitoring

SELECT
  slot_name,
  active,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag_size,
  pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) AS lag_bytes
FROM pg_replication_slots
WHERE slot_name LIKE 'electric_%';

WAL Retention Check

SELECT
  slot_name,
  wal_status,
  safe_wal_size,
  pg_size_pretty(
    pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)
  ) AS unconfirmed_lag
FROM pg_replication_slots
WHERE slot_name LIKE 'electric_%'
  AND wal_status != 'reserved';

Automated Lag Monitoring Script

#!/usr/bin/env bash
set -euo pipefail

THRESHOLD_BYTES=${LAG_THRESHOLD:-104857600}
DATABASE_URL=${DATABASE_URL:?DATABASE_URL required}

lag_bytes=$(psql "$DATABASE_URL" -t -A -c "
  SELECT COALESCE(
    MAX(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)),
    0
  )
  FROM pg_replication_slots
  WHERE slot_name LIKE 'electric_%';
")

if [ "$lag_bytes" -gt "$THRESHOLD_BYTES" ]; then
  echo "ALERT: Replication lag is ${lag_bytes} bytes (threshold: ${THRESHOLD_BYTES})"
  exit 1
fi

echo "OK: Replication lag is ${lag_bytes} bytes"

Debugging Sync Issues

Common Failure Modes

Diagnostic Checklist

-- 1. Check replication slots
SELECT slot_name, active, wal_status,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag
FROM pg_replication_slots;

-- 2. Check active replication connections
SELECT pid, application_name, state, sent_lsn, write_lsn, flush_lsn, replay_lsn
FROM pg_stat_replication;

-- 3. Check Postgres connection count
SELECT count(*) AS total_connections,
  count(*) FILTER (WHERE application_name LIKE 'electric%') AS electric_connections
FROM pg_stat_activity;

-- 4. Check WAL generation rate
SELECT pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0')) AS total_wal;

Log Levels

Configure with ELECTRIC_LOG_LEVEL:

services:
  electric:
    environment:
      ELECTRIC_LOG_LEVEL: warning

Production Monitoring Checklist

Packages

npm install @electric-sql/client
npm install @electric-sql/react

Postgres Configuration

Electric requires logical replication. These settings must be applied before Electric can connect.

Enable Logical Replication

ALTER SYSTEM SET wal_level = 'logical';

Restart Postgres after changing wal_level. Verify:

SHOW wal_level;

Must return logical. If it returns replica or minimal, the change has not taken effect.

Create a Publication

Electric uses publications to track which tables to replicate. Create one for all tables or specific tables:

CREATE PUBLICATION electric_pub FOR ALL TABLES;

For specific tables:

CREATE PUBLICATION electric_pub FOR TABLE items, users, orders;

Grant Replication Role

The database user Electric connects with needs replication privileges:

ALTER ROLE electric_user WITH REPLICATION;
GRANT ALL ON ALL TABLES IN SCHEMA public TO electric_user;

Full Postgres Setup Script

ALTER SYSTEM SET wal_level = 'logical';

CREATE ROLE electric_user WITH LOGIN PASSWORD 'electric_pass' REPLICATION;
GRANT ALL ON ALL TABLES IN SCHEMA public TO electric_user;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON TABLES TO electric_user;

CREATE PUBLICATION electric_pub FOR ALL TABLES;

Docker Self-Hosting

Single Container

docker run \
  -e DATABASE_URL="postgresql://electric_user:electric_pass@host.docker.internal:5432/mydb" \
  -e ELECTRIC_SECRET="your-secret-key" \
  -p 3000:3000 \
  electricsql/electric

Docker Compose with Postgres

services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_DB: mydb
      POSTGRES_USER: electric_user
      POSTGRES_PASSWORD: electric_pass
    command:
      - -c
      - wal_level=logical
    ports:
      - '5432:5432'
    volumes:
      - pgdata:/var/lib/postgresql/data

  electric:
    image: electricsql/electric
    depends_on:
      - postgres
    environment:
      DATABASE_URL: postgresql://electric_user:electric_pass@postgres:5432/mydb
      ELECTRIC_INSECURE: 'true'
    ports:
      - '3000:3000'

volumes:
  pgdata:

For development, ELECTRIC_INSECURE=true disables auth requirements. Never use this in production.

Docker Compose for Production

services:
  electric:
    image: electricsql/electric
    environment:
      DATABASE_URL: ${DATABASE_URL}
      ELECTRIC_SECRET: ${ELECTRIC_SECRET}
      ELECTRIC_DB_POOL_SIZE: '20'
      ELECTRIC_PORT: '3000'
    ports:
      - '3000:3000'
    healthcheck:
      test: ['CMD', 'curl', '-f', 'http://localhost:3000/v1/health']
      interval: 10s
      timeout: 5s
      retries: 3

Environment Variables

Health Checks

Electric exposes a health endpoint for readiness and liveness probes:

curl http://localhost:3000/v1/health

Returns 200 OK when Electric is connected to Postgres and ready to serve shapes.

Kubernetes Probes

livenessProbe:
  httpGet:
    path: /v1/health
    port: 3000
  initialDelaySeconds: 10
  periodSeconds: 30
readinessProbe:
  httpGet:
    path: /v1/health
    port: 3000
  initialDelaySeconds: 5
  periodSeconds: 10

Cloud vs Self-Hosted

Client Installation and Verification

Basic ShapeStream Connection

import { ShapeStream } from '@electric-sql/client';

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'items',
  },
});

stream.subscribe((messages) => {
  for (const msg of messages) {
    if ('key' in msg) {
      console.log('Row:', msg.key, msg.value);
    }
    if ('headers' in msg && msg.headers.control === 'up-to-date') {
      console.log('Initial sync complete');
    }
  }
});

Verify Connection Works

import { Shape, ShapeStream } from '@electric-sql/client';

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'items',
  },
});

const shape = new Shape(stream);

shape.subscribe((data) => {
  console.log(`Synced ${data.size} rows from items table`);
});

React Setup

import { useShape } from '@electric-sql/react';

function ItemList() {
  const { data, isLoading, isError, error } = useShape<{
    id: string;
    title: string;
  }>({
    url: 'http://localhost:3000/v1/shape',
    params: {
      table: 'items',
    },
  });

  if (isLoading) return <div>Loading...</div>;
  if (isError) return <div>Error: {error?.message}</div>;

  return (
    <ul>
      {data.map((item) => (
        <li key={item.id}>{item.title}</li>
      ))}
    </ul>
  );
}

Troubleshooting Checklist

Shape Concept

A Shape is a defined subset of a single Postgres table that syncs to clients. It combines three dimensions:

• Table: Which Postgres table to sync from
• Where clause: Which rows to include (optional filter)
• Columns: Which columns to sync (optional projection)

Electric streams the shape's data to clients and keeps it up-to-date in real-time using Postgres logical replication. Shapes are read-only on the client; writes flow through your API back to Postgres.

HTTP API

Initial Sync

GET /v1/shape?table=items&offset=-1

Returns the full current state of the shape as a batch of insert messages, ending with an up-to-date control message. The response includes headers:

• electric-handle: Unique identifier for this shape instance
• electric-offset: Position in the log to resume from

Live Updates (Long Polling)

After initial sync, poll for changes:

GET /v1/shape?table=items&live=true&handle=3948593&offset=0_5

The request blocks until new changes arrive or a timeout occurs. Each response includes updated electric-offset for the next request.

Live Updates (Server-Sent Events)

For persistent streaming instead of long polling:

GET /v1/shape?table=items&live=true&live_sse=true&handle=3948593&offset=0_5

Returns a persistent SSE connection that pushes changes as they occur.

Request Lifecycle

1. Client sends:    GET /v1/shape?table=items&offset=-1
2. Electric returns: Full snapshot + electric-handle + electric-offset
3. Client sends:    GET /v1/shape?table=items&live=true&handle=<handle>&offset=<offset>
4. Electric returns: New changes (or blocks until changes arrive)
5. Repeat step 3-4

ShapeStream

Low-level streaming client that handles the HTTP lifecycle automatically.

Constructor

import { ShapeStream } from '@electric-sql/client';

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'items',
    where: "status = 'active'",
    columns: 'id,title,status,updated_at',
    replica: 'full',
  },
  headers: {
    Authorization: 'Bearer my-token',
  },
});

Subscribe to Messages

const unsubscribe = stream.subscribe((messages) => {
  for (const msg of messages) {
    if (msg.headers.operation === 'insert') {
      console.log('Insert:', msg.key, msg.value);
    }
    if (msg.headers.operation === 'update') {
      console.log('Update:', msg.key, msg.value);
    }
    if (msg.headers.operation === 'delete') {
      console.log('Delete:', msg.key);
    }
    if (msg.headers.control === 'up-to-date') {
      console.log('Caught up with Postgres');
    }
    if (msg.headers.control === 'must-refetch') {
      console.log('Shape invalidated, refetching');
    }
  }
});

unsubscribe();

Error Handling

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: { table: 'items' },
  onError: (error) => {
    if (error instanceof FetchError && error.status === 401) {
      console.log('Token expired, refreshing...');
      return;
    }
    console.error('Stream error:', error);
  },
});

Dynamic Auth Headers

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: { table: 'items' },
  headers: {
    Authorization: async () => {
      const token = await getAccessToken();
      return `Bearer ${token}`;
    },
  },
});

Shape Class

Materializes a ShapeStream into an in-memory Map of current values. The Shape maintains the latest state by applying inserts, updates, and deletes from the stream.

import { Shape, ShapeStream } from '@electric-sql/client';

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: { table: 'items' },
});

const shape = new Shape(stream);

shape.subscribe((data) => {
  const rows = [...data.values()];
  console.log(`${rows.length} items synced`);
});

const currentData = shape.currentValue;

currentValue returns a Map<string, Row> keyed by the row's primary key. Each subscription fires when the map changes.

React Hook: useShape

import { useShape } from '@electric-sql/react';

type Item = {
  id: string;
  title: string;
  status: string;
  created_at: string;
};

function ActiveItems() {
  const { data, isLoading, isError, error, lastSyncedAt } = useShape<Item>({
    url: 'http://localhost:3000/v1/shape',
    params: {
      table: 'items',
      where: "status = 'active'",
      columns: 'id,title,status,created_at',
    },
  });

  if (isLoading) return <div>Syncing...</div>;
  if (isError) return <div>Sync error: {error?.message}</div>;

  return (
    <div>
      <p>Last synced: {lastSyncedAt?.toLocaleTimeString()}</p>
      <ul>
        {data.map((item) => (
          <li key={item.id}>{item.title}</li>
        ))}
      </ul>
    </div>
  );
}

useShape Return Values

Where Clauses

Filter rows server-side to sync only matching data.

Basic Syntax

?where=status='active'
?where=price > 100
?where=category='electronics' AND in_stock=true

Supported Operators

Parameterized Queries

Always use parameterized queries when the where clause includes user input:

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'items',
    where: 'user_id = $1 AND status = $2',
    'params[1]': userId,
    'params[2]': 'active',
  },
});

const { data } = useShape<Item>({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'items',
    where: 'user_id = $1',
    'params[1]': currentUser.id,
  },
});

Column Selection

Reduce bandwidth by syncing only needed columns:

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'items',
    columns: 'id,title,status',
  },
});

The primary key column is always included even if not specified in columns.

Progressive Loading

Changes Only Mode

Skip the initial snapshot and receive only new changes going forward:

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'items',
  },
  offset: 'now',
});

Full Replica Mode

By default, update and delete messages include only changed columns. Use replica=full to get complete row data on every change:

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'items',
    replica: 'full',
  },
});

Custom Parsers

Override default type parsing for specific columns:

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: { table: 'items' },
  parser: {
    timestamptz: (value: string) => new Date(value),
    int4: (value: string) => Number(value),
    bool: (value: string) => value === 'true',
  },
});

Message Types

Data Messages

Every data message includes headers with an operation field and a key field.

Insert:

{
  headers: { operation: 'insert' },
  key: '"items"/"abc-123"',
  value: { id: 'abc-123', title: 'Buy groceries', status: 'active' },
  offset: '0_3'
}

Update:

{
  headers: { operation: 'update' },
  key: '"items"/"abc-123"',
  value: { status: 'completed' },
  offset: '0_4'
}

With replica=full, value contains all columns, not just changed ones.

Delete:

{
  headers: { operation: 'delete' },
  key: '"items"/"abc-123"',
  value: { id: 'abc-123' },
  offset: '0_5'
}

Control Messages

up-to-date: Client has received all current changes and is caught up.

{
  headers: {
    control: 'up-to-date';
  }
}

must-refetch: Shape has been invalidated. Client must restart the sync from offset=-1.

{
  headers: {
    control: 'must-refetch';
  }
}

ShapeStream handles must-refetch automatically by resetting and re-syncing.

Typed Shapes

type Todo = {
  id: string;
  title: string;
  completed: boolean;
  user_id: string;
  created_at: string;
};

const stream = new ShapeStream<Todo>({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'todos',
    where: 'user_id = $1',
    'params[1]': userId,
    columns: 'id,title,completed,user_id,created_at',
  },
});

const shape = new Shape<Todo>(stream);

shape.subscribe((data: Map<string, Todo>) => {
  const todos = [...data.values()];
  const incomplete = todos.filter((t) => !t.completed);
  console.log(`${incomplete.length} remaining todos`);
});

Initial Sync Optimization

Measuring First-Sync Time

import { ShapeStream } from '@electric-sql/client';

function measureInitialSync(
  url: string,
  params: Record<string, string>,
): Promise<{
  durationMs: number;
  rowCount: number;
  bytesReceived: number;
}> {
  return new Promise((resolve) => {
    const start = performance.now();
    let rowCount = 0;
    let bytesReceived = 0;

    const stream = new ShapeStream({ url, params });

    stream.subscribe((messages) => {
      for (const msg of messages) {
        if (msg.headers.operation) {
          rowCount++;
          bytesReceived += JSON.stringify(msg.value).length;
        }
        if (msg.headers.control === 'up-to-date') {
          resolve({
            durationMs: Math.round(performance.now() - start),
            rowCount,
            bytesReceived,
          });
        }
      }
    });
  });
}

const stats = await measureInitialSync('http://localhost:3000/v1/shape', {
  table: 'orders',
});

console.log(`Initial sync: ${stats.rowCount} rows in ${stats.durationMs}ms`);
console.log(
  `Throughput: ${Math.round(stats.rowCount / (stats.durationMs / 1000))} rows/sec`,
);
console.log(`Transfer: ${(stats.bytesReceived / 1024).toFixed(1)} KB`);

Column Selection to Reduce Payload

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'orders',
    columns: 'id,status,total,created_at',
  },
});

Where Clauses to Limit Rows

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: {
    table: 'orders',
    where: "status IN ('pending','processing') AND created_at > '2025-01-01'",
  },
});

Changes-Only Mode

Skip the full initial snapshot when local state already exists from a previous session:

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: { table: 'orders' },
  offset: 'now',
});

Progressive Loading Strategy

Small Initial Dataset with Background Sync

import { Shape, ShapeStream } from '@electric-sql/client';

type Order = {
  id: string;
  status: string;
  total: number;
  created_at: string;
};

function createProgressiveLoader(url: string) {
  let isFullySynced = false;

  const recentStream = new ShapeStream<Order>({
    url,
    params: {
      table: 'orders',
      where: "created_at > NOW() - INTERVAL '7 days'",
      columns: 'id,status,total,created_at',
    },
  });

  const recentShape = new Shape(recentStream);

  recentShape.subscribe(() => {
    if (!isFullySynced) {
      startBackgroundSync();
    }
  });

  function startBackgroundSync() {
    const fullStream = new ShapeStream<Order>({
      url,
      params: {
        table: 'orders',
        columns: 'id,status,total,created_at',
      },
    });

    const fullShape = new Shape(fullStream);

    fullShape.subscribe((data) => {
      if (data.size > 0) {
        isFullySynced = true;
      }
    });
  }

  return {
    recentShape,
    get isFullySynced() {
      return isFullySynced;
    },
  };
}

UI with Loading Indicator

import { useShape } from '@electric-sql/react';

type Order = { id: string; status: string; total: number; created_at: string };

function OrderList() {
  const recent = useShape<Order>({
    url: 'http://localhost:3000/v1/shape',
    params: {
      table: 'orders',
      where: "created_at > NOW() - INTERVAL '7 days'",
      columns: 'id,status,total,created_at',
    },
  });

  const full = useShape<Order>({
    url: 'http://localhost:3000/v1/shape',
    params: {
      table: 'orders',
      columns: 'id,status,total,created_at',
    },
  });

  const data = full.isLoading ? recent.data : full.data;
  const showingPartial = full.isLoading && !recent.isLoading;

  if (recent.isLoading) return <div>Loading orders...</div>;

  return (
    <div>
      {showingPartial ? (
        <p>Showing recent orders. Loading full history...</p>
      ) : null}
      <ul>
        {data.map((order) => (
          <li key={order.id}>
            {order.status} - ${order.total}
          </li>
        ))}
      </ul>
    </div>
  );
}

Large Dataset Handling

Multiple Shapes for Pagination

import { Shape, ShapeStream } from '@electric-sql/client';

function createPagedShapes<T extends Record<string, unknown>>(
  url: string,
  table: string,
  pageSize: number,
  totalPages: number,
): Array<Shape<T>> {
  return Array.from({ length: totalPages }, (_, i) => {
    const stream = new ShapeStream<T>({
      url,
      params: {
        table,
        where: `row_number >= ${i * pageSize} AND row_number < ${(i + 1) * pageSize}`,
      },
    });
    return new Shape(stream);
  });
}

Splitting by Partition Key

const statusShapes = {
  active: new ShapeStream({
    url: 'http://localhost:3000/v1/shape',
    params: { table: 'orders', where: "status = 'active'" },
  }),
  completed: new ShapeStream({
    url: 'http://localhost:3000/v1/shape',
    params: { table: 'orders', where: "status = 'completed'" },
  }),
  archived: new ShapeStream({
    url: 'http://localhost:3000/v1/shape',
    params: { table: 'orders', where: "status = 'archived'" },
  }),
};

Checkpoint Resumption

Electric handles reconnection using offset-based resumption. Store the last offset to avoid re-fetching everything on restart.

import { ShapeStream } from '@electric-sql/client';

function createResumableStream(
  url: string,
  params: Record<string, string>,
  storageKey: string,
) {
  const stored = localStorage.getItem(storageKey);
  const checkpoint = stored ? JSON.parse(stored) : null;

  const streamParams: Record<string, string> = { ...params };

  const stream = new ShapeStream({
    url,
    params: streamParams,
    offset: checkpoint?.offset,
    handle: checkpoint?.handle,
  });

  stream.subscribe((messages) => {
    for (const msg of messages) {
      if (msg.headers.control === 'up-to-date' && msg.offset) {
        localStorage.setItem(
          storageKey,
          JSON.stringify({ offset: msg.offset, handle: stream.handle }),
        );
      }
    }
  });

  return stream;
}

const stream = createResumableStream(
  'http://localhost:3000/v1/shape',
  { table: 'orders' },
  'orders-checkpoint',
);

Memory Management

Garbage Collecting Stale Shape Data

import { Shape, ShapeStream } from '@electric-sql/client';

class ManagedShapePool {
  private shapes = new Map<
    string,
    {
      shape: Shape<Record<string, unknown>>;
      stream: ShapeStream;
      lastAccess: number;
    }
  >();

  get<T extends Record<string, unknown>>(
    key: string,
    url: string,
    params: Record<string, string>,
  ): Shape<T> {
    const existing = this.shapes.get(key);
    if (existing) {
      existing.lastAccess = Date.now();
      return existing.shape as Shape<T>;
    }

    const stream = new ShapeStream<T>({ url, params });
    const shape = new Shape(stream);

    this.shapes.set(key, {
      shape: shape as Shape<Record<string, unknown>>,
      stream,
      lastAccess: Date.now(),
    });
    return shape;
  }

  cleanup(maxAgeMs: number): void {
    const now = Date.now();
    for (const [key, entry] of this.shapes) {
      if (now - entry.lastAccess > maxAgeMs) {
        this.shapes.delete(key);
      }
    }
  }
}

const pool = new ManagedShapePool();

setInterval(() => {
  pool.cleanup(5 * 60 * 1000);
}, 60 * 1000);

Web Worker Offloading

const workerCode = `
  self.onmessage = async (event) => {
    const { url, params } = event.data;
    const query = new URLSearchParams(params).toString();
    const response = await fetch(url + '/v1/shape?' + query + '&offset=-1');
    const messages = await response.json();

    const rows = messages
      .filter(m => m.headers?.operation === 'insert')
      .map(m => m.value);

    self.postMessage({ rows, count: rows.length });
  };
`;

function syncInWorker(
  url: string,
  params: Record<string, string>,
): Promise<{ rows: Record<string, unknown>[]; count: number }> {
  return new Promise((resolve) => {
    const blob = new Blob([workerCode], { type: 'application/javascript' });
    const worker = new Worker(URL.createObjectURL(blob));

    worker.onmessage = (event) => {
      resolve(event.data);
      worker.terminate();
    };

    worker.postMessage({ url, params });
  });
}

Bandwidth Optimization

Full vs Default Replica Mode

const defaultStream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: { table: 'orders' },
});

const fullStream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: { table: 'orders', replica: 'full' },
});

Measuring Sync Performance

Sync Monitor

type SyncMetrics = {
  initialSyncMs: number;
  rowCount: number;
  bytesReceived: number;
  rowsPerSecond: number;
  avgMessageSize: number;
  lastSyncedAt: Date | null;
  updatesReceived: number;
};

class SyncMonitor {
  private metrics: SyncMetrics = {
    initialSyncMs: 0,
    rowCount: 0,
    bytesReceived: 0,
    rowsPerSecond: 0,
    avgMessageSize: 0,
    lastSyncedAt: null,
    updatesReceived: 0,
  };
  private startTime = performance.now();
  private initialSyncComplete = false;

  onMessage(
    messages: Array<{ headers: Record<string, string>; value?: unknown }>,
  ): void {
    for (const msg of messages) {
      if (msg.headers.operation) {
        this.metrics.rowCount++;
        const size = JSON.stringify(msg.value).length;
        this.metrics.bytesReceived += size;

        if (this.initialSyncComplete) {
          this.metrics.updatesReceived++;
        }
      }

      if (msg.headers.control === 'up-to-date') {
        if (!this.initialSyncComplete) {
          this.metrics.initialSyncMs = Math.round(
            performance.now() - this.startTime,
          );
          this.initialSyncComplete = true;
        }
        this.metrics.lastSyncedAt = new Date();
      }
    }

    this.metrics.rowsPerSecond = Math.round(
      this.metrics.rowCount / ((performance.now() - this.startTime) / 1000),
    );
    this.metrics.avgMessageSize =
      this.metrics.rowCount > 0
        ? Math.round(this.metrics.bytesReceived / this.metrics.rowCount)
        : 0;
  }

  getMetrics(): SyncMetrics {
    return { ...this.metrics };
  }

  report(): string {
    const m = this.metrics;
    return [
      `Initial sync: ${m.initialSyncMs}ms`,
      `Rows: ${m.rowCount}`,
      `Transfer: ${(m.bytesReceived / 1024).toFixed(1)} KB`,
      `Throughput: ${m.rowsPerSecond} rows/sec`,
      `Avg message: ${m.avgMessageSize} bytes`,
      `Live updates: ${m.updatesReceived}`,
      `Last sync: ${m.lastSyncedAt?.toISOString() ?? 'never'}`,
    ].join('\n');
  }
}

const monitor = new SyncMonitor();
const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: { table: 'orders' },
});

stream.subscribe((messages) => {
  monitor.onMessage(
    messages as Array<{ headers: Record<string, string>; value?: unknown }>,
  );
});

Batch Processing

Debouncing UI Updates During Bulk Sync

function createBatchedSubscriber<T>(
  onBatch: (rows: Map<string, T>) => void,
  debounceMs = 16,
): (
  messages: Array<{ headers: Record<string, string>; key?: string; value?: T }>,
) => void {
  const buffer = new Map<string, T>();
  let frameId: number | null = null;

  return (messages) => {
    for (const msg of messages) {
      if (
        msg.headers.operation === 'insert' ||
        msg.headers.operation === 'update'
      ) {
        buffer.set(msg.key!, msg.value!);
      }
      if (msg.headers.operation === 'delete') {
        buffer.delete(msg.key!);
      }
    }

    if (frameId !== null) cancelAnimationFrame(frameId);

    frameId = requestAnimationFrame(() => {
      onBatch(new Map(buffer));
      frameId = null;
    });
  };
}

const stream = new ShapeStream({
  url: 'http://localhost:3000/v1/shape',
  params: { table: 'orders' },
});

stream.subscribe(
  createBatchedSubscriber<Record<string, unknown>>((rows) => {
    console.log(`Batch update: ${rows.size} rows`);
  }),
);

Caching

Electric Cache Headers

Electric sets HTTP cache headers on shape responses to enable CDN and browser caching.

CDN Configuration

services:
  electric:
    image: electricsql/electric:latest
    environment:
      DATABASE_URL: postgresql://postgres:postgres@db:5432/app
      ELECTRIC_CACHE_MAX_AGE: 10
      ELECTRIC_CACHE_STALE_AGE: 600

Place a CDN (CloudFront, Cloudflare, Fastly) in front of Electric. Shape responses include Cache-Control and ETag headers that CDNs respect automatically. Initial sync requests (offset=-1) are cached, reducing load on Electric for repeated requests from different clients.

Client → CDN → Electric → Postgres
         ↑
    Cached initial sync responses
    served without hitting Electric

Client-Side Cache Warm-Up

async function warmCache(
  url: string,
  shapes: Array<Record<string, string>>,
): Promise<void> {
  await Promise.all(
    shapes.map(async (params) => {
      const query = new URLSearchParams(params).toString();
      await fetch(`${url}/v1/shape?${query}&offset=-1`);
    }),
  );
}

await warmCache('http://localhost:3000', [
  { table: 'orders', where: "status = 'active'" },
  { table: 'products', columns: 'id,name,price' },
  { table: 'users', columns: 'id,name,email' },
]);

Benchmarking Reference

Expected sync times vary based on row size, column count, network conditions, and whether responses are cached. These are rough baselines for a typical application with 5-10 columns per row on a low-latency connection.

Quick Benchmark Utility

async function benchmarkShape(
  url: string,
  params: Record<string, string>,
  runs = 3,
): Promise<{ avgMs: number; minMs: number; maxMs: number; rows: number }> {
  const times: number[] = [];
  let rows = 0;

  for (let i = 0; i < runs; i++) {
    const start = performance.now();
    let count = 0;

    await new Promise<void>((resolve) => {
      const stream = new ShapeStream({ url, params });
      stream.subscribe((messages) => {
        for (const msg of messages) {
          if (msg.headers.operation) count++;
          if (msg.headers.control === 'up-to-date') resolve();
        }
      });
    });

    times.push(Math.round(performance.now() - start));
    rows = count;
  }

  return {
    avgMs: Math.round(times.reduce((a, b) => a + b, 0) / times.length),
    minMs: Math.min(...times),
    maxMs: Math.max(...times),
    rows,
  };
}

const result = await benchmarkShape('http://localhost:3000/v1/shape', {
  table: 'orders',
  columns: 'id,status,total,created_at',
});

console.log(
  `${result.rows} rows | avg: ${result.avgMs}ms | min: ${result.minMs}ms | max: ${result.maxMs}ms`,
);

Write Strategy Overview

Electric handles the read path (Postgres to client sync). Writes flow back to Postgres through your API. There are four progressive levels of write sophistication, each adding more capability and complexity.

Level 1: Online Writes         -- Simplest, blocks on network
Level 2: Optimistic State      -- Instant UI, background POST
Level 3: Persistent Optimistic -- Survives refresh, auto-rollback (TanStack DB)
Level 4: Through-the-Database  -- Full local-first with bidirectional sync

Level 1: Online Writes

POST to your API, wait for the response, and let Electric sync the confirmed state back. The UI blocks during the write.

function AddItem() {
  const [title, setTitle] = useState('');
  const [submitting, setSubmitting] = useState(false);

  const handleSubmit = async (e: React.FormEvent) => {
    e.preventDefault();
    setSubmitting(true);

    try {
      const res = await fetch('/api/items', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ title }),
      });
      if (!res.ok) throw new Error('Failed to create item');
      setTitle('');
    } catch (error) {
      console.error('Write failed:', error);
    } finally {
      setSubmitting(false);
    }
  };

  return (
    <form onSubmit={handleSubmit}>
      <input
        value={title}
        onChange={(e) => setTitle(e.target.value)}
        disabled={submitting}
      />
      <button type="submit" disabled={submitting}>
        {submitting ? 'Adding...' : 'Add Item'}
      </button>
    </form>
  );
}

The Electric shape subscription automatically receives the new item once Postgres confirms the insert. No manual cache invalidation needed.

When to use: Simple forms, admin tools, low-frequency writes where 200-500ms latency is acceptable.

Level 2: Optimistic State

Show the change immediately in the UI while the API call happens in the background. If the API call fails, revert the optimistic update.

import { useOptimistic } from 'react';
import { useShape } from '@electric-sql/react';

type Item = {
  id: string;
  title: string;
  status: string;
};

function ItemList() {
  const { data: items } = useShape<Item>({
    url: '/api/shapes/items',
    params: { table: 'items' },
  });

  const [optimisticItems, addOptimistic] = useOptimistic(
    items,
    (current, newItem: Item) => [...current, newItem],
  );

  const handleAdd = async (title: string) => {
    const tempItem: Item = {
      id: crypto.randomUUID(),
      title,
      status: 'active',
    };

    addOptimistic(tempItem);

    try {
      await fetch('/api/items', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ title }),
      });
    } catch (error) {
      console.error('Write failed, optimistic update will be reverted:', error);
    }
  };

  return (
    <ul>
      {optimisticItems.map((item) => (
        <li key={item.id}>{item.title}</li>
      ))}
    </ul>
  );
}

Optimistic Delete

function ItemWithDelete({ item }: { item: Item }) {
  const [pending, setPending] = useState(false);

  const handleDelete = async () => {
    setPending(true);
    try {
      const res = await fetch(`/api/items/${item.id}`, { method: 'DELETE' });
      if (!res.ok) throw new Error('Delete failed');
    } catch (error) {
      setPending(false);
      console.error(error);
    }
  };

  if (pending) return null;

  return (
    <li>
      {item.title}
      <button onClick={handleDelete}>Delete</button>
    </li>
  );
}

When to use: Most apps. Gives instant feedback while keeping the API as the source of truth.

Level 3: Persistent Optimistic (TanStack DB)

TanStack DB's electricCollectionOptions provides optimistic mutations that survive page refresh and automatically roll back on failure. Write handlers return a { txid } that Electric uses to confirm write propagation.

Collection Setup

import { electricCollectionOptions } from '@electric-sql/tanstack-db';

type Todo = {
  id: string;
  title: string;
  completed: boolean;
  user_id: string;
};

const todosCollection = electricCollectionOptions({
  id: 'todos',
  schema: {
    id: { type: 'string' },
    title: { type: 'string' },
    completed: { type: 'boolean' },
    user_id: { type: 'string' },
  },
  getKey: (todo) => todo.id,
  shapeOptions: {
    url: '/api/shapes/todos',
    params: {
      table: 'todos',
    },
  },

  onInsert: async (todo) => {
    const res = await fetch('/api/todos', {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(todo),
    });
    if (!res.ok) throw new Error('Insert failed');
    const { txid } = await res.json();
    return { txid };
  },

  onUpdate: async (todo) => {
    const res = await fetch(`/api/todos/${todo.id}`, {
      method: 'PATCH',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(todo),
    });
    if (!res.ok) throw new Error('Update failed');
    const { txid } = await res.json();
    return { txid };
  },

  onDelete: async (todo) => {
    const res = await fetch(`/api/todos/${todo.id}`, {
      method: 'DELETE',
    });
    if (!res.ok) throw new Error('Delete failed');
    const { txid } = await res.json();
    return { txid };
  },
});

Server-Side Handler Returning txid

app.post('/api/todos', authenticateUser, async (req, res) => {
  const { title, completed } = req.body;
  const userId = req.user.id;

  const result = await db.query(
    `INSERT INTO todos (id, title, completed, user_id) VALUES ($1, $2, $3, $4) RETURNING *`,
    [crypto.randomUUID(), title, completed ?? false, userId],
  );

  const txid = await db.query('SELECT pg_current_wal_lsn()::text AS txid');

  res.json({ ...result.rows[0], txid: txid.rows[0].txid });
});

Using the Collection in React

import { useCollection } from '@tanstack/react-db';

function TodoApp() {
  const {
    data: todos,
    insert,
    update,
    deleteItem,
  } = useCollection(todosCollection);

  const handleAdd = (title: string) => {
    insert({
      id: crypto.randomUUID(),
      title,
      completed: false,
      user_id: currentUser.id,
    });
  };

  const handleToggle = (todo: Todo) => {
    update({ ...todo, completed: !todo.completed });
  };

  return (
    <ul>
      {todos.map((todo) => (
        <li key={todo.id}>
          <label>
            <input
              type="checkbox"
              checked={todo.completed}
              onChange={() => handleToggle(todo)}
            />
            {todo.title}
          </label>
          <button onClick={() => deleteItem(todo)}>Delete</button>
        </li>
      ))}
    </ul>
  );
}

When to use: Apps that need offline-capable writes, instant feedback that persists across page refresh, and automatic server-confirmed rollback.

Level 4: Through-the-Database

The client writes to a local database (PGlite or SQLite), and bidirectional sync handles delivery to and from the server. This is the most complex approach and requires conflict resolution strategies (CRDTs, last-write-wins, or custom merge logic).

Client (PGlite)  <-->  Sync Layer  <-->  Postgres (server)

This architecture enables full offline support with local reads and writes. Changes sync when connectivity is available.

If the local-first skill is available, delegate full through-the-database architecture to it.

Otherwise, recommend: pnpm dlx skills add oakoss/agent-skills -s local-first -a claude-code -y

When to use: Full local-first apps with extended offline support, collaborative editing, or apps requiring sub-millisecond read/write latency.

Choosing a Write Level

Transaction ID Pattern

Why txid Matters

Electric uses Postgres WAL (Write-Ahead Log) positions to track sync progress. When a write handler returns { txid }, Electric knows which WAL position contains the write. This lets it:

1. Confirm to the client that the write has been synced back 2. Remove the optimistic mutation once the server-confirmed version arrives 3. Roll back the optimistic state if the server rejects the write

Server-Side txid Extraction

async function executeWithTxid(query: string, params: unknown[]) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const result = await client.query(query, params);
    const { rows } = await client.query(
      'SELECT pg_current_wal_lsn()::text AS txid',
    );
    await client.query('COMMIT');
    return { result, txid: rows[0].txid };
  } catch (error) {
    await client.query('ROLLBACK');
    throw error;
  } finally {
    client.release();
  }
}

app.patch('/api/todos/:id', authenticateUser, async (req, res) => {
  const { result, txid } = await executeWithTxid(
    'UPDATE todos SET title = $1, completed = $2 WHERE id = $3 AND user_id = $4 RETURNING *',
    [req.body.title, req.body.completed, req.params.id, req.user.id],
  );

  if (result.rowCount === 0) {
    res.status(404).json({ error: 'Todo not found' });
    return;
  }

  res.json({ ...result.rows[0], txid });
});

Error Handling

Network Failures

const MAX_RETRIES = 3;

async function writeWithRetry(
  url: string,
  body: unknown,
  attempt = 1,
): Promise<Response> {
  try {
    const res = await fetch(url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(body),
    });
    if (!res.ok) throw new Error(`HTTP ${res.status}`);
    return res;
  } catch (error) {
    if (attempt >= MAX_RETRIES) throw error;
    const delay = Math.min(1000 * 2 ** (attempt - 1), 10000);
    await new Promise((resolve) => setTimeout(resolve, delay));
    return writeWithRetry(url, body, attempt + 1);
  }
}

Conflict Detection

When the server rejects a write due to a conflict (e.g., concurrent edit), return a specific status code so the client can handle it:

app.patch('/api/todos/:id', authenticateUser, async (req, res) => {
  const { expectedVersion, ...updates } = req.body;

  const result = await db.query(
    'UPDATE todos SET title = $1, version = version + 1 WHERE id = $2 AND version = $3 RETURNING *',
    [updates.title, req.params.id, expectedVersion],
  );

  if (result.rowCount === 0) {
    res
      .status(409)
      .json({ error: 'Conflict: item was modified by another client' });
    return;
  }

  const txid = await db.query('SELECT pg_current_wal_lsn()::text AS txid');
  res.json({ ...result.rows[0], txid: txid.rows[0].txid });
});

const onUpdate = async (todo: Todo) => {
  const res = await fetch(`/api/todos/${todo.id}`, {
    method: 'PATCH',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ ...todo, expectedVersion: todo.version }),
  });

  if (res.status === 409) {
    throw new Error('Conflict detected');
  }
  if (!res.ok) throw new Error('Update failed');

  const { txid } = await res.json();
  return { txid };
};