Postgres CDC

监听 PostgreSQL 表变更。presence / postgres_changes 监听必须在 subscribe() 之前绑定。

前置条件

本页代码能跑通的前提是业务库已配置好 数据库侧准备（publication、RLS 策略、必需的 grant）。这些没做的话，订阅会成功但收不到任何事件，容易误判成 SDK 问题。

const channel = realtime.channel("db-changes");

channel.on("postgres_changes", { event: "*", schema: "public" }, (payload) => {
  console.log("public schema 变更", payload);
});

channel.on(
  "postgres_changes",
  { event: "INSERT", schema: "public", table: "messages" },
  (payload) => {
    console.log("新增", payload.new);
  }
);

channel.on(
  "postgres_changes",
  {
    event: "UPDATE",
    schema: "public",
    table: "users",
    filter: "username=eq.Realtime",
  },
  (payload) => {
    console.log("更新", payload.new, payload.old);
  }
);

channel.subscribe((status, err) => {
  if (status === "SUBSCRIBED") {
    console.log("开始接收数据库变更");
  }
  if (status === "CHANNEL_ERROR") {
    console.error(err);
  }
});

filter 可以是原始字符串，也可以用 postgresChangesFilter() 构建，二者线格式相同：

import { postgresChangesFilter } from "@cloudbase/js-sdk/realtime-js";

// 字符串
{ event: "UPDATE", schema: "public", table: "users", filter: "id=eq.1" }

// Builder
{
  event: "UPDATE",
  schema: "public",
  table: "users",
  filter: postgresChangesFilter().eq("id", 1),
}

运算符	字符串	Builder	含义
eq	id=eq.1	.eq("id", 1)	等于
neq	id=neq.1	.neq("id", 1)	不等于
lt / lte / gt / gte	age=gte.18	.gte("age", 18)	比较
in	status=in.(active,pending)	.in("status", ["active", "pending"])	属于列表
like / ilike	title=like.%foo%	.like("title", "%foo%")	模式匹配
is	deleted_at=is.null	.is("deleted_at", null)	IS null/true/false
match / imatch	title=match.^foo	.match("title", "^foo")	POSIX 正则
isdistinct	value=isdistinct.1	.isDistinct("value", 1)	NULL 安全不等于

取反：字符串加 not. 前缀，或使用 .not(column, operator, value)。多个条件用逗号（AND）连接，Builder 则链式调用。

channel.on(
  "postgres_changes",
  {
    event: "UPDATE",
    schema: "public",
    table: "orders",
    filter: postgresChangesFilter()
      .gt("amount", 100)
      .not("status", "in", ["draft", "archived"]),
  },
  (payload) => console.log(payload)
);

用 select 只接收部分列，减小载荷：

channel.on(
  "postgres_changes",
  {
    event: "*",
    schema: "public",
    table: "users",
    select: ["id", "first_name"],
  },
  (payload) => {
    // payload.new 仅含 { id, first_name }
    console.log(payload);
  }
);

需要等服务端确认 CDC 订阅就绪后再触发 SUBSCRIBED 时，设置 postgres_changes_options.wait：

const channel = realtime.channel("db-changes", {
  config: {
    postgres_changes_options: { wait: true, timeout: 15000 },
  },
});

提示

Realtime 在服务端按单表 WAL 过滤，没有资源嵌套（!inner）和 or() 分组。like / ilike 通配符使用 % 而不是 *。

数据库侧准备

postgres_changes 走的是业务库的逻辑复制。

1. 把表加入 publication

服务端从 publication cloudbase_realtime 读取变更（名称由租户配置决定，若你的环境指定了其他名称以配置为准）。没有加进去的表，变更完全不触发：

ALTER PUBLICATION cloudbase_realtime ADD TABLE public.messages;

确认已加入的表：

SELECT schemaname, tablename
FROM pg_publication_tables
WHERE pubname = 'cloudbase_realtime';

2. 业务表授权 + 开启 RLS

变更事件是按业务表自己的 RLS 策略逐订阅者校验的：

-- 让角色有读权限
GRANT SELECT ON public.messages TO anon, authenticated;

-- 开启行级安全
ALTER TABLE public.messages ENABLE ROW LEVEL SECURITY;

-- 只让用户收到自己的数据
CREATE POLICY messages_select ON public.messages
  FOR SELECT TO authenticated
  USING (owner_id = auth.uid());