CDC

Subscribe to Change Data Capture streams from db.cdc.postgres and db.cdc.sqlite sources. List configured sources, open a stream, and receive row-level change events over a channel. The API is driver-neutral: both kinds return the same source info and the same change events, and differ only in the capabilities they publish.

Loading

local cdc = require("cdc")

list_sources

List the configured CDC sources the caller is allowed to see:

local sources, err = cdc.list_sources()
if err then return nil, err end
for _, s in ipairs(sources) do
    print(s.id, s.kind, s.state, s.capabilities.before_images)
end

Sources the caller lacks cdc.source on are omitted rather than reported as an error.

Returns: table, error

source

Retrieve one source by its registry entry ID or replication slot name:

local info, err = cdc.source("app:pg_cdc")
if err then return nil, err end
if info == nil then
    -- no such source
end

Returns: table, error (source info, or nil if not found)

stream

Open a change stream on a source. The returned cdc.Stream exposes a channel that delivers change events:

local stream, err = cdc.stream("app:pg_cdc", {
    tables = { "public.users", "public.orders" },
    ops    = { "insert", "update" },
    buffer = 128,
})
if err then return nil, err end

-- The caller owns stream until close(), release(), or task cleanup.

Unknown option keys are rejected with errors.INVALID. Table names are matched case-insensitively against both the qualified relation and the bare table name. Snapshot rows are filtered by tables only; ops applies to live changes.

A stream receives a snapshot when either opts.snapshot is true or the source entry's snapshot field is set; snapshot rows arrive first with op = "snapshot", then the stream continues into live changes with no gap. opts.after is reserved for drivers that resume from a cursor — every driver shipped today returns errors.INVALID ("cdc operation is not supported by this source") for it, including db.cdc.postgres when it reports capture_resume.

Filters narrow delivery only. Access to a source is granted by the cdc.subscribe permission, never by a filter.

Returns: Stream, error

The Lua delivery channel has a separate fixed capacity of 64. The buffer option controls the PostgreSQL source subscription, not that channel.

Stream Methods

channel

Return the channel that receives change events. The first call subscribes to the source and yields; subsequent calls return the same channel. The first call can return a subscription error. Channel :receive() returns value, true for a change or nil, false when the stream ends:

local stream, stream_err = cdc.stream("app:pg_cdc")
if stream_err then return nil, stream_err end
local ch, subscribe_err = stream:channel()
if subscribe_err then
    stream:close()
    return nil, subscribe_err
end

while true do
    local change, ok = ch:receive()
    if not ok then break end

    if change.op == "snapshot" then
        seed_row(change.table, change.after)
    elseif change.op == "insert" then
        handle_new_user(change.table, change.after)
    elseif change.op == "update" then
        handle_update(change.table, change.before, change.after)
    elseif change.op == "delete" then
        handle_delete(change.table, change.before)
    end
end

local _, close_err = stream:close()
if close_err then return nil, close_err end

The stream is lazy: construct it, then call channel() before generating the writes it should observe. This is live observation, not replay of changes made before the subscription.

When a source terminates a stream with a failure, the channel delivers an error value before it closes. receive is an alias for channel.

close

Stop the subscription and release the stream. The method is idempotent, and the runtime also closes the stream at the end of the task scope. release is an alias for close.

local _, err = stream:close()
if err then return nil, err end

Change Event

Each message received on the channel is a change table:

before and after are row maps keyed by column name.

Source Info

cdc.source and each entry of cdc.list_sources return the same record:

Branch on capabilities rather than on kind:

local info = cdc.source("app:changes")
if not info.capabilities.before_images then
    -- before is not a guaranteed full row image; keep your own last-known state
end

See CDC sources for field semantics.

Permissions

A denied action returns errors.PERMISSION_DENIED.

Errors

See Error Handling for working with errors.

See Also

• Change Data Capture - Source configuration and capabilities
• Channel - Channel semantics
• Database - SQL database services