electric-schema-shapes

此技能基于 electric-shapes。请先阅读该技能，了解 ShapeStream 配置。

Electric — 架构与 Shapes

设置

设计表时，请牢记每个 shape 只同步一张表。对于跨表数据，请使用多个 shape，并在客户端进行联接。

-- Schema designed for Electric shapes
CREATE TABLE todos (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  org_id UUID NOT NULL,
  text TEXT NOT NULL,
  completed BOOLEAN DEFAULT false,
  created_at TIMESTAMPTZ DEFAULT now()
);

ALTER TABLE todos REPLICA IDENTITY FULL;

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

const todoStream = new ShapeStream({
  url: '/api/todos', // Proxy sets: table=todos, where=org_id=$1
})

核心模式

使用多个 shape 处理跨表数据

// Each shape syncs one table — join client-side
const todoStream = new ShapeStream({ url: '/api/todos' })
const userStream = new ShapeStream({ url: '/api/users' })

// With TanStack DB, use .join() in live queries:
// q.from({ todo: todoCollection })
//   .join({ user: userCollection }, ({ todo, user }) => eq(todo.userId, user.id))

选择副本模式

// Default: only changed columns sent on update
const stream = new ShapeStream({ url: '/api/todos' })

// Full: all columns + old_value on updates (more bandwidth, needed for diffs)
const stream = new ShapeStream({
  url: '/api/todos',
  params: { replica: 'full' },
})

用于乐观写入的后端 txid 握手

在与变更操作相同的事务中调用 pg_current_xact_id()::xid::text。如果在事务外查询，你会得到不同的 txid，客户端将永远无法完成协调。

// API endpoint — txid MUST be in the same transaction as the INSERT
app.post('/api/todos', async (req, res) => {
  const client = await pool.connect()
  try {
    await client.query('BEGIN')
    const result = await client.query(
      'INSERT INTO todos (id, text, org_id) VALUES ($1, $2, $3) RETURNING id',
      [crypto.randomUUID(), req.body.text, req.body.orgId]
    )
    const txResult = await client.query(
      'SELECT pg_current_xact_id()::xid::text AS txid'
    )
    await client.query('COMMIT')
    // txid accepts number | bigint | `${bigint}`
    res.json({ id: result.rows[0].id, txid: parseInt(txResult.rows[0].txid) })
  } finally {
    client.release()
  }
})

// Client awaits txid before dropping optimistic state
await todoCollection.utils.awaitTxId(txid)

常见错误

高危：设计跨多张表的 shape

错误：

const stream = new ShapeStream({
  url: '/api/data',
  params: {
    table: 'todos JOIN users ON todos.user_id = users.id',
  },
})

正确：

const todoStream = new ShapeStream({ url: '/api/todos' })
const userStream = new ShapeStream({ url: '/api/users' })

Shape 只能对应单张表。跨表数据需要使用多个 shape，并通过 TanStack DB 实时查询在客户端进行联接。

来源：AGENTS.md:104-105

中危：在 WHERE 中使用枚举列时未将其转换为文本

错误：

// Proxy route
originUrl.searchParams.set('where', "status IN ('active', 'done')")

正确：

originUrl.searchParams.set('where', "status::text IN ('active', 'done')")

WHERE 子句中的枚举类型需要显式使用 ::text 转换。否则，查询可能失败或返回意外结果。

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

高危：未为乐观写入设置 txid 握手

错误：

// Backend: just INSERT, return id
app.post('/api/todos', async (req, res) => {
  const result = await db.query(
    'INSERT INTO todos (text) VALUES ($1) RETURNING id',
    [req.body.text]
  )
  res.json({ id: result.rows[0].id })
})

正确：

// Backend: INSERT and return txid in same transaction
app.post('/api/todos', async (req, res) => {
  const client = await pool.connect()
  try {
    await client.query('BEGIN')
    const result = await client.query(
      'INSERT INTO todos (text) VALUES ($1) RETURNING id',
      [req.body.text]
    )
    const txResult = await client.query(
      'SELECT pg_current_xact_id()::xid::text AS txid'
    )
    await client.query('COMMIT')
    res.json({ id: result.rows[0].id, txid: parseInt(txResult.rows[0].txid) })
  } finally {
    client.release()
  }
})

如果没有 txid，在 Electric 的同步版本到达之前，乐观状态就会被丢弃，导致界面闪烁。客户端使用 awaitTxId(txid) 保留乐观状态，直到真实数据完成同步。

来源：AGENTS.md:116-119

另请参阅：electric-shapes/SKILL.md — Shapes 不可变；动态筛选条件需要新的 ShapeStream 实例。 另请参阅：electric-orm/SKILL.md — 架构设计会同时影响 shapes（读取）和 ORM 查询（写入）。

版本

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