CDC Subscriptions

Status: Available since v0.42.0 (CDC-01)
Requires: pg_ripple_http for Server-Sent Events delivery. PostgreSQL LISTEN/NOTIFY works without it.
SQL: pg_ripple.subscribe_cdc(), pg_ripple.unsubscribe_cdc(), pg_ripple.list_cdc_subscriptions()
HTTP: GET /subscribe/{id} (Server-Sent Events, via pg_ripple_http)
See also: Live SPARQL Subscriptions · Live Views

Overview

Change Data Capture (CDC) subscriptions let your application subscribe to a real-time stream of RDF triple changes — filtered by SPARQL pattern or SHACL shape — without polling the database.

When a matching triple is inserted or deleted, pg_ripple sends a PostgreSQL NOTIFY message on a named channel. Listeners receive a JSON payload describing the change. The pg_ripple_http companion service exposes these subscriptions as Server-Sent Events (SSE) streams for web and streaming applications.

sequenceDiagram
    participant App
    participant pg_ripple_http
    participant PostgreSQL
    participant pg_ripple_ext as pg_ripple extension

    App->>pg_ripple_http: GET /subscribe/{id}
    pg_ripple_http->>PostgreSQL: LISTEN pg_ripple_sub_{id}
    Note over App,pg_ripple_http: SSE connection established

    loop Triple Changes
        App->>PostgreSQL: INSERT triple (via SPARQL or SQL)
        pg_ripple_ext->>PostgreSQL: NOTIFY pg_ripple_sub_{id}, payload
        PostgreSQL->>pg_ripple_http: async notification
        pg_ripple_http->>App: event: row\ndata: {...}
    end

Creating a Subscription

-- Subscribe to all triple changes.
SELECT pg_ripple.create_subscription('my_feed');

-- Subscribe with a SPARQL pattern filter.
SELECT pg_ripple.create_subscription(
    'person_changes',
    filter_sparql := 'SELECT ?s ?p ?o WHERE { ?s a <https://schema.org/Person> ; ?p ?o }'
);

-- Subscribe with a SHACL shape filter.
SELECT pg_ripple.create_subscription(
    'shape_violations',
    filter_shape := '<https://shapes.example.org/PersonShape>'
);

Parameters

Returns TRUE if created, FALSE if a subscription with that name already exists.

Listening for Changes

-- Start listening.
LISTEN pg_ripple_cdc_my_feed;

-- Insert a triple.
SELECT pg_ripple.insert_triple(
    '<https://ex.org/alice>',
    '<https://schema.org/name>',
    '"Alice"'
);

-- In your application, receive notifications via pg_notify/asyncpg/etc.

Notification Payload

Each notification carries a JSON payload:

{
  "op": "add",
  "s": "<https://ex.org/alice>",
  "p": "<https://schema.org/name>",
  "o": "\"Alice\"",
  "g": ""
}

Listing Subscriptions

SELECT name, filter_sparql IS NOT NULL AS has_filter, created_at
FROM pg_ripple.list_subscriptions()
ORDER BY created_at;

Dropping a Subscription

-- Returns TRUE if removed, FALSE if not found.
SELECT pg_ripple.drop_subscription('my_feed');

WebSocket Access via pg_ripple_http

When the pg_ripple_http companion service is running, subscriptions are accessible as WebSocket endpoints:

ws://<host>:8080/ws/subscriptions/{name}

The service supports content negotiation via the Accept header:

• application/json (default) — JSON payload
• text/turtle — Turtle-serialized change notification
• application/ld+json — JSON-LD change notification

Integration Patterns

GraphRAG Pipeline

import asyncpg

async def watch_entity_changes():
    conn = await asyncpg.connect(dsn)
    await conn.execute("LISTEN pg_ripple_cdc_entity_changes")

    async for notification in conn.listen("pg_ripple_cdc_entity_changes"):
        payload = json.loads(notification.payload)
        # Re-embed entity on change.
        await update_embedding(payload["s"])

Live Dashboard

const ws = new WebSocket("ws://localhost:8080/ws/subscriptions/dashboard_feed");
ws.onmessage = (event) => {
  const change = JSON.parse(event.data);
  updateDashboard(change.op, change.s, change.p, change.o);
};

Underlying Tables

Related Functions

Further reading

• Blog: CDC and Knowledge Graphs — real-time event streaming from your knowledge graph