electric-shapes

Electric — Shape 流式传输

设置

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

const stream = new ShapeStream({
  url: '/api/todos', // Your proxy route, NOT direct Electric URL
  // Built-in parsers auto-handle: bool, int2, int4, float4, float8, json, jsonb
  // Add custom parsers for other types (see references/type-parsers.md)
  parser: {
    timestamptz: (date: string) => new Date(date),
  },
})

const shape = new Shape(stream)

shape.subscribe(({ rows }) => {
  console.log('synced rows:', rows)
})

// Wait for initial sync
const rows = await shape.rows

核心模式

使用 WHERE 子句和位置参数筛选行

const stream = new ShapeStream({
  url: '/api/todos',
  params: {
    table: 'todos',
    where: 'user_id = $1 AND status = $2',
    params: { '1': userId, '2': 'active' },
  },
})

选择特定列（必须包含主键）

const stream = new ShapeStream({
  url: '/api/todos',
  params: {
    table: 'todos',
    columns: ['id', 'title', 'status'], // PK required
  },
})

在 snake_case 和 camelCase 之间映射列名

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

const stream = new ShapeStream({
  url: '/api/todos',
  columnMapper: snakeCamelMapper(),
})
// DB column "created_at" arrives as "createdAt" in client
// WHERE clauses auto-translate: "createdAt" → "created_at"

使用重试处理错误

const stream = new ShapeStream({
  url: '/api/todos',
  onError: (error) => {
    console.error('sync error', error)
    return {} // Return {} to retry; returning void stops the stream
  },
})

有关 401 错误时刷新身份验证令牌的信息，请参阅 electric-proxy-auth/SKILL.md。

从已存储的偏移量恢复

const stream = new ShapeStream({
  url: '/api/todos',
  offset: storedOffset, // Both offset AND handle required
  handle: storedHandle,
})

在更新时获取包含旧值的副本

const stream = new ShapeStream({
  url: '/api/todos',
  params: {
    table: 'todos',
    replica: 'full', // Sends unchanged columns + old_value on updates
  },
})

常见错误

严重：从 onError 返回 void 会永久停止同步

错误：

const stream = new ShapeStream({
  url: '/api/todos',
  onError: (error) => {
    console.error('sync error', error)
    // Returning nothing = stream stops forever
  },
})

正确：

const stream = new ShapeStream({
  url: '/api/todos',
  onError: (error) => {
    console.error('sync error', error)
    return {} // Return {} to retry
  },
})

onError 返回 undefined 表示永久停止数据流。至少返回 {} 以重试，或者返回 { headers, params }，以更新后的值重试。

来源：packages/typescript-client/src/client.ts:409-418

高：使用列时未包含主键

错误：

const stream = new ShapeStream({
  url: '/api/todos',
  params: {
    table: 'todos',
    columns: ['title', 'status'],
  },
})

正确：

const stream = new ShapeStream({
  url: '/api/todos',
  params: {
    table: 'todos',
    columns: ['id', 'title', 'status'],
  },
})

服务端会返回 400 错误。columns 列表必须始终包含主键列。

来源：website/docs/guides/shapes.md

高：设置 offset 时未设置用于恢复的 handle

错误：

new ShapeStream({
  url: '/api/todos',
  offset: storedOffset,
})

正确：

new ShapeStream({
  url: '/api/todos',
  offset: storedOffset,
  handle: storedHandle,
})

会抛出 MissingShapeHandleError。要从已存储的位置恢复数据流，必须同时提供 offset 和 handle。

来源：packages/typescript-client/src/client.ts:1997-2003

高：在 WHERE 子句中使用非确定性函数

错误：

const stream = new ShapeStream({
  url: '/api/events',
  params: {
    table: 'events',
    where: 'start_time > now()',
  },
})

正确：

const stream = new ShapeStream({
  url: '/api/events',
  params: {
    table: 'events',
    where: 'start_time > $1',
    params: { '1': new Date().toISOString() },
  },
})

服务端会拒绝包含 now()、random()、count() 等非确定性函数的 WHERE 子句。请使用静态值或位置参数。

来源：packages/sync-service/lib/electric/replication/eval/env/known_functions.ex

高：未解析自定义 Postgres 类型

错误：

const stream = new ShapeStream({
  url: '/api/events',
})
// createdAt will be string "2024-01-15T10:30:00.000Z", not a Date

正确：

const stream = new ShapeStream({
  url: '/api/events',
  parser: {
    timestamptz: (date: string) => new Date(date),
    timestamp: (date: string) => new Date(date),
  },
})

Electric 会自动解析 bool、int2、int4、float4、float8、json、jsonb 和 int8（→ BigInt）。其他类型都会以字符串形式传入——请为 timestamptz、date、numeric 等类型添加自定义解析器。完整列表请参阅 references/type-parsers.md。

来源：AGENTS.md:300-308

中：在 params 中使用保留参数名

错误：

const stream = new ShapeStream({
  url: '/api/todos',
  params: {
    table: 'todos',
    cursor: 'abc', // Reserved!
    offset: '0', // Reserved!
  },
})

正确：

const stream = new ShapeStream({
  url: '/api/todos',
  params: {
    table: 'todos',
    page_cursor: 'abc',
    page_offset: '0',
  },
})

会抛出 ReservedParamError。cursor、handle、live、offset、cache-buster 以及所有带 subset__* 前缀的参数名，均由 Electric 协议保留。

来源：packages/typescript-client/src/client.ts:1984-1985

中：修改运行中数据流的形状选项

错误：

const stream = new ShapeStream({
  url: '/api/todos',
  params: { table: 'todos', where: "status = 'active'" },
})
// Later...
stream.options.params.where = "status = 'done'" // No effect!

正确：

// Create a new stream with different params
const newStream = new ShapeStream({
  url: '/api/todos',
  params: { table: 'todos', where: "status = 'done'" },
})

每个订阅中的 Shape 都是不可变的。更改运行中数据流的 params 不会产生效果。若要使用不同的筛选条件，请创建新的 ShapeStream 实例。

来源：AGENTS.md:106

参考资料

• WHERE 子句支持的类型和函数
• 内置类型解析器

另请参阅：electric-proxy-auth/SKILL.md — Shape URL 必须指向代理路由，而不能直接指向 Electric。 另请参阅：electric-debugging/SKILL.md — onError 语义和退避机制对于诊断同步问题至关重要。

版本

目标版本：@electric-sql/client v1.5.10。