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, connector role grants, RLS policies). When a prerequisite is missing, the subscription fails in one of two ways, and they need opposite debugging — do not assume the silent one:

• Subscription rejected (CHANNEL_ERROR, error object usually empty) → the table is not in the publication, or the subscription role has no SELECT on it.
• Subscribed but no events arrive → the table's RLS filters the subscription role out, or the table was added less than ~10 seconds ago.

Also note that the realtime module is not enabled in every environment; check first (see the first step of Database-side setup).

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. The examples below use the public.todos table to show how to enable, inspect, and remove table listening.

First confirm the capability is enabled in this environment — no code will receive events without it:

SELECT EXISTS (
  SELECT 1 FROM pg_namespace WHERE nspname = 'realtime'
) AS has_realtime_schema;

Continue only when it returns true; false means the environment has not been enabled yet — ask the server side to enable it rather than debugging the client against this page.

Terminology

Term	Description
publication	The PG logical replication publication object. Realtime uses it to decide which tables to listen to. The default name is cloudbase_realtime (the name is tenant-configurable; if your environment uses a different name, follow that configuration)
connector role	One per database, cloudbase_realtime_admin, created by the server, responsible for reading the WAL and running the poller
subscription role	The role in the client JWT: anon / authenticated / service_role. RLS checks are based on it

1. Enable listening on a table

The minimum setup is two statements:

-- 1. Add the table to the publication (when the publication already exists)
ALTER PUBLICATION cloudbase_realtime ADD TABLE public.todos;

-- 2. Grant the connector role access (optional when that role already owns the table; kept here because it is idempotent)
GRANT SELECT ON public.todos TO "cloudbase_realtime_admin";

First-time setup (publication does not exist yet)

CREATE PUBLICATION has no IF NOT EXISTS; when the publication does not exist, use:

CREATE PUBLICATION cloudbase_realtime FOR TABLE public.todos;

Idempotent script

Works for both first-time and incremental setup, safe to re-run:

DO $$
BEGIN
  IF NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = 'cloudbase_realtime') THEN
    CREATE PUBLICATION cloudbase_realtime;
  END IF;

  IF NOT EXISTS (
    SELECT 1 FROM pg_publication_tables
    WHERE pubname = 'cloudbase_realtime'
      AND schemaname = 'public'
      AND tablename  = 'todos'
  ) THEN
    EXECUTE 'ALTER PUBLICATION cloudbase_realtime ADD TABLE public.todos';
  END IF;
END $$;

-- GRANT / REPLICA IDENTITY are naturally idempotent, safe to re-run
GRANT SELECT ON public.todos TO "cloudbase_realtime_admin";

-- Optional: makes UPDATE / DELETE carry the **full previous row**.
-- Without it old_record still carries the primary key; it is empty only for tables with no primary key.
-- The cost is that the whole old row enters the WAL and the replication stream,
-- so enable it per table when you actually need the old values.
ALTER TABLE public.todos REPLICA IDENTITY FULL;

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.todos TO anon, authenticated;

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

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

3. Inspect current listening

Check which tables are in the publication:

SELECT schemaname, tablename
FROM pg_publication_tables
WHERE pubname = 'cloudbase_realtime'
ORDER BY 1, 2;

Check currently active subscriptions (realtime view):

SELECT entity, filters, created_at
FROM realtime.subscription
ORDER BY created_at DESC;

4. Remove listening

Remove a single table (most common):

ALTER PUBLICATION cloudbase_realtime DROP TABLE public.todos;

After removal, the poller notices in the next cycle and automatically stops pushing changes for that table; existing client subscriptions do not disconnect immediately, but no longer receive events for the table.

Turn off all listening entirely:

DROP PUBLICATION cloudbase_realtime;

Use with caution

DROP PUBLICATION removes all tables in the publication. To re-enable, follow the first-time setup flow again. Do not manually delete replication slots (such as cloudbase_realtime_replication_slot); they are managed by the server.

Troubleshooting

Symptom	Check
Subscription rejected (CHANNEL_ERROR, error object empty)	① Is the table in the publication (see Inspect current listening); ② does the subscription role (anon / authenticated) have SELECT on the table
Subscribed but no events arrive	① do RLS policies allow the subscription role; ② has it been at least ~10 seconds since the table was added
old_record is empty for UPDATE / DELETE	The table has no primary key and is not set to REPLICA IDENTITY FULL; a table with a primary key carries the key even without it
Changes take effect only after ~10 seconds	Normal. The poller periodically picks up publication changes; no restart needed