Postgres CDC

Listen to PostgreSQL table changes. Bind presence / postgres_changes listeners before subscribe().

Prerequisite

The samples on this page only work after the business database is prepared as described in Database-side setup (publication, RLS policies, and required grants). If you skip this, the subscription succeeds but no events arrive, which is easy to mistake for an SDK issue.

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

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

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

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

channel.subscribe((status, err) => {
  if (status === "SUBSCRIBED") {
    console.log("ready for database changes");
  }
  if (status === "CHANNEL_ERROR") {
    console.error(err);
  }
});

filter accepts either a raw string or postgresChangesFilter(). Both produce the same wire format:

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

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

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

Operator	String	Builder	Meaning
eq	id=eq.1	.eq("id", 1)	equal
neq	id=neq.1	.neq("id", 1)	not equal
lt / lte / gt / gte	age=gte.18	.gte("age", 18)	comparison
in	status=in.(active,pending)	.in("status", ["active", "pending"])	in list
like / ilike	title=like.%foo%	.like("title", "%foo%")	pattern match
is	deleted_at=is.null	.is("deleted_at", null)	IS null/true/false
match / imatch	title=match.^foo	.match("title", "^foo")	POSIX regex
isdistinct	value=isdistinct.1	.isDistinct("value", 1)	NULL-safe inequality

Negate with a not. prefix on strings, or .not(column, operator, value) on the builder. Combine conditions with commas (AND), or chain builder calls.

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

Use select to receive a subset of columns and shrink the payload:

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

To delay SUBSCRIBED until the server confirms the CDC subscription, set postgres_changes_options.wait:

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

Note

Realtime evaluates filters server-side over a single table's WAL. There is no resource embedding (!inner) and no or() grouping. Use % (not *) as the like / ilike wildcard.

Database-side setup

postgres_changes uses logical replication on the business database.

1. Add the table to the publication

The server reads changes from the cloudbase_realtime publication (the name is tenant-configurable; if your environment uses a different name, follow that configuration). Tables that are not added never emit events:

ALTER PUBLICATION cloudbase_realtime ADD TABLE public.messages;

Confirm which tables are included:

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

2. Grant table access and enable RLS

Change events are checked against the business table's own RLS policies for each subscriber:

-- Grant SELECT to roles
GRANT SELECT ON public.messages TO anon, authenticated;

-- Enable row-level security
ALTER TABLE public.messages ENABLE ROW LEVEL SECURITY;

-- Only deliver rows that belong to the current user
CREATE POLICY messages_select ON public.messages
  FOR SELECT TO authenticated
  USING (owner_id = auth.uid());