pg-tide Relay: Hub-and-Spoke Integration

Available since: v0.52.0 (originally pg-trickle relay, migrated to pg-tide in v0.93.0; current bridge semantics refreshed in v0.127.0)

Requires for relay: pg_tide ≥ 0.33.0 (relay process and outbox/inbox); pg_ripple ≥ 0.127.0

Requires for IVM views: pg_trickle ≥ 0.46.0 (IVM only)

Note: pg-trickle v0.46.0 extracted the relay, outbox, and inbox subsystem into the standalone pg_tide extension (trickle-labs/pg-tide). After v0.46.0, pg_trickle provides IVM only. All relay functionality described here requires pg_tide.

What this integration does

Imagine you have IoT sensor readings arriving via Kafka, customer records coming in from a CRM webhook, and order data flowing through NATS — all using different field names, different identifiers for the same real-world entities, and different data shapes. Downstream teams want a clean, unified, enriched view of this data pushed to their own systems in real time, without any of them having to understand the messy source schemas.

This is the hub-and-spoke pattern. pg-tide acts as the transport network: its relay CLI pulls data in from any source (Kafka, NATS, webhooks) and pushes enriched data out to any sink. pg_ripple acts as the intelligent hub in the middle: it turns the incoming JSON into a knowledge graph, runs inference rules to derive new facts, validates data quality with SHACL, and serializes the enriched results back to JSON or JSON-LD for downstream consumers. The output format is flexible — simple decoded JSON (via raw CDC triggers) or shaped JSON-LD (via framing triggers) — depending on what each downstream consumer needs.

The whole pipeline runs inside a single PostgreSQL database. Both extensions share the same transaction context, so data moves from inbox to triplestore to outbox without ever leaving the database process.

                          ┌────────────────────────────────┐
                          │         pg-ripple hub           │
                          │   (PostgreSQL + pg_ripple ext)  │
    INBOUND               │                                │               OUTBOUND
    ───────               │  ┌──────────┐  ┌───────────┐  │               ────────
                          │  │ Datalog  │  │  SHACL    │  │
  ┌──────────┐  relay     │  │ inference│  │ validation│  │     relay    ┌──────────┐
  │  Kafka   │──reverse──▶│  └────┬─────┘  └─────┬─────┘  │──forward──▶│  NATS    │
  │ (orders) │            │       │              │        │             │ (events) │
  └──────────┘            │  ┌────▼──────────────▼─────┐  │             └──────────┘
                          │  │                         │  │
  ┌──────────┐  relay     │  │   RDF Triple Store      │  │     relay    ┌──────────┐
  │  NATS    │──reverse──▶│  │   (VP tables, HTAP)     │──│──forward──▶│  Webhook  │
  │(sensors) │            │  │                         │  │             │ (API)     │
  └──────────┘            │  └────▲──────────────▲─────┘  │             └──────────┘
                          │       │              │        │
  ┌──────────┐  relay     │  ┌────┴─────┐  ┌────┴─────┐  │     relay    ┌──────────┐
  │ Webhook  │──reverse──▶│  │owl:sameAs│  │ SPARQL   │  │──forward──▶│  Kafka    │
  │ (CRM)    │            │  │ linking  │  │federation│  │             │(enriched)│
  └──────────┘            │  └──────────┘  └──────────┘  │             └──────────┘
                          │                                │
                          │  pg-tide stream tables         │
                          │  (inbox → transform → outbox)  │
                          └────────────────────────────────┘

The data flow through the hub has five stages:

Ingest — pg-tide relay reverse mode delivers raw JSON events into inbox tables.
Transform — a trigger converts the JSON into RDF triples and loads them into the triplestore.
Enrich — Datalog inference rules derive new facts (alerts, entity links, risk scores).
Validate — SHACL shapes enforce data quality before anything leaves the hub.
Distribute — pg-tide relay forward mode pushes enriched, validated JSON events to any number of sinks.

Prerequisites

Relay examples require pg_tide and pg_ripple in the same database. Install pg_trickle as well only when you also use pg_ripple's IVM-backed live views. If both companion extensions are installed, create pg_tide before pg_trickle so pgtrickle.attach_outbox() can call tide.outbox_create():

CREATE EXTENSION pg_tide;      -- relay, outbox, inbox (trickle-labs/pg-tide ≥ 0.33.0)
CREATE EXTENSION pg_trickle;   -- optional IVM only (trickle-labs/pg-trickle ≥ 0.46.0)
CREATE EXTENSION pg_ripple;    -- RDF triple store (≥ 0.127.0)

If pg_tide is not installed, relay-dependent features are unavailable. pg_ripple degrades gracefully: SPARQL views and IVM continue to work (they require pg_trickle, not pg_tide), but outbox/inbox operations raise a descriptive error:

ERROR: pg_tide extension is not installed;
      install pg_tide from https://github.com/trickle-labs/pg-tide
       then run: CREATE EXTENSION pg_tide

If pg_trickle is not installed, IVM-backed SPARQL, Datalog, CONSTRUCT, DESCRIBE, ASK, and framing views are unavailable:

ERROR: pg_trickle is not installed -- SPARQL views require pg_trickle

Call pg_ripple.relay_available() or pg_ripple.pg_tide_available() before calling relay functions. Call pg_ripple.pg_trickle_available() before calling IVM-backed view functions.

A worked example: IoT sensor hub

The following walkthrough builds a complete hub that ingests temperature readings from IoT sensors on a Kafka topic (iot.sensors), detects anomalies using inference rules, and publishes enriched alerts back to Kafka on a separate topic (iot.alerts). The pipeline is intentionally symmetrical: the same broker and format convention on both ends, with pg_ripple doing the enrichment in the middle. Each step shows both the SQL to run and what the data looks like at that point.

Step 1 — Pull sensor events from Kafka

The relay process runs outside PostgreSQL and continuously polls configured sources. You tell it what to poll and where to write the results using tide.relay_set_inbox_v2(). Here we subscribe to the iot.sensors Kafka topic and direct its events into a table called sensor_inbox. Enriched alerts will flow back out on the iot.alerts topic on the same broker — same system, both ends:

SELECT tide.relay_set_inbox_v2(jsonb_build_object(
  'name',   'sensor-readings',
  'inbox',  'sensor_inbox',
  'source', 'kafka',
  'config', jsonb_build_object(
    'brokers', '${env:KAFKA_BROKERS}',
    'topic',   'iot.sensors'
  )
));

The ${env:KAFKA_BROKERS} syntax tells the relay to expand an environment variable at runtime. Pipeline configs are stored in the database as JSONB, but sensitive values like broker addresses can reference environment variables this way — the actual value stays in the relay process's environment and never needs to be stored in plaintext in the database.

Each time the relay receives a message from Kafka, it inserts a row into sensor_inbox. The original Kafka message payload arrives as a JSONB column. A typical row looks like this:

{
  "event_id": "kafka:iot.sensors:0:42",
  "event_type": "sensor_reading",
  "payload": {
    "device": "sensor-7",
    "temp": 22.5,
    "unit": "°C",
    "ts": "2026-04-28T10:00:00Z"
  }
}

Step 2 — Convert JSON events to RDF triples

Raw JSON has no standard semantics. Two sensor vendors might both call their field "temp" but mean very different things. By converting to RDF we attach well-defined, globally unique meanings to each field — in this case using the SAREF IoT ontology.

The pg_ripple.json_to_ntriples_and_load() function (v0.52.0+) does this in one call. Its context parameter works like a JSON-LD @context: it maps the incoming JSON field names to the full predicate IRIs you want to store. The relay delivers plain JSON; the IRI mapping is applied at load time inside PostgreSQL, so the original message is never modified.

A trigger on sensor_inbox fires for every inserted row:

CREATE OR REPLACE FUNCTION transform_sensor_to_rdf()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
    PERFORM pg_ripple.json_to_ntriples_and_load(
        payload     => NEW.payload,
        subject_iri => 'https://example.org/observation/' || NEW.event_id,
        type_iri    => 'https://saref.etsi.org/core/Measurement',
        context     => '{
            "@vocab":  "https://saref.etsi.org/core/",
            "device":  "https://saref.etsi.org/core/measurementMadeBy",
            "temp":    "https://saref.etsi.org/core/hasValue",
            "ts":      "https://saref.etsi.org/core/hasTimestamp",
            "unit":    "https://qudt.org/schema/qudt/unit"
        }'::jsonb
    );
    RETURN NEW;
END;
$$;

CREATE TRIGGER sensor_to_rdf
AFTER INSERT ON sensor_inbox
FOR EACH ROW EXECUTE FUNCTION transform_sensor_to_rdf();

The context object supports two resolution mechanisms:

@vocab — a default IRI prefix applied to every unmapped key. Any field not explicitly listed gets expanded to https://saref.etsi.org/core/{field}.
Explicit entries — override specific keys with the exact IRI you want, regardless of the @vocab default.

Nested JSON objects become blank nodes. Arrays produce one triple per element. null values are silently skipped. This means you can point the relay at an arbitrary third-party JSON event and describe the entire mapping in one JSONB literal — no code changes needed when a vendor renames a field, only a context update.

After the trigger fires, those five triples exist in the triplestore — one rdf:type triple (from type_iri) and one for each of the four data fields in the source JSON. If you queried them back as JSON-LD immediately after load, you would see:

{
  "@context": {
    "saref": "https://saref.etsi.org/core/",
    "xsd": "http://www.w3.org/2001/XMLSchema#",
    "qudt": "https://qudt.org/schema/qudt/"
  },
  "@id": "https://example.org/observation/kafka:iot.sensors:0:42",
  "@type": "saref:Measurement",
  "saref:measurementMadeBy": {
    "@id": "https://example.org/device/sensor-7"
  },
  "saref:hasValue": {
    "@value": "22.5",
    "@type": "xsd:decimal"
  },
  "saref:hasTimestamp": {
    "@value": "2026-04-28T10:00:00Z",
    "@type": "xsd:dateTime"
  },
  "qudt:unit": "°C"
}

This is the data at rest inside the triplestore, expressed as JSON-LD. The @type and @id fields carry the full semantic meaning from the SAREF ontology, so any consumer that understands SAREF can correctly interpret the reading.

Step 3 — Add inference rules to detect anomalies

Datalog rules let you express facts that can be derived from the stored triples. Rather than writing triggers for every business rule, you declare the rules once and pg_ripple materialises the inferred triples automatically.

The rules below fire an alert whenever a measurement exceeds 40°C and link devices across sources that share a serial number (entity resolution):

SELECT pg_ripple.load_rules(
    rules    => $$
        % Derive an alert for any observation above the threshold.
        % Two triples are inferred per matching observation:
        %   <obs> ex:tempAlert <device>          — links observation to device
        %   <obs> ex:alertType "high_temperature" — records the alert category
        % Storing the alert type as a literal triple keeps the meaning inside
        % the triplestore rather than hardcoding it in downstream trigger code.
        ex:tempAlert(Obs, Device) :-
            saref:measurementMadeBy(Obs, Device),
            saref:hasValue(Obs, Val),
            Val > 40.0.

        ex:alertType(Obs, "high_temperature") :-
            saref:hasValue(Obs, Val),
            Val > 40.0.

        % Link two devices if they share a serial number, even if they
        % appear under different identifiers in different source systems.
        owl:sameAs(D1, D2) :-
            schema:serialNumber(D1, SN),
            schema:serialNumber(D2, SN),
            D1 \= D2.
    $$,
    rule_set => 'sensor_enrichment'
);

When a 45°C reading arrives from sensor-7, these rules materialise two new triples — inferred facts that did not exist in the raw data:

<https://example.org/observation/kafka:iot.sensors:0:99>
    <https://example.org/tempAlert>
    <https://example.org/device/sensor-7> .

<https://example.org/observation/kafka:iot.sensors:0:99>
    <https://example.org/alertType>
    "high_temperature" .

Step 4 — Enforce data quality with SHACL

Before enriched data leaves the hub, SHACL shapes act as a quality gate. Any observation that lacks a measurementMadeBy link or a timestamp will fail validation and be flagged rather than silently forwarded to downstream consumers:

SELECT pg_ripple.load_shacl($$
    @prefix sh:    <http://www.w3.org/ns/shacl#> .
    @prefix saref: <https://saref.etsi.org/core/> .
    @prefix xsd:   <http://www.w3.org/2001/XMLSchema#> .
    @prefix ex:    <https://example.org/> .

    ex:ObservationShape a sh:NodeShape ;
        sh:targetClass saref:Measurement ;
        sh:property [
            sh:path saref:measurementMadeBy ;
            sh:minCount 1 ;
            sh:maxCount 1 ;
            sh:message "Every measurement must reference exactly one device." ;
        ] ;
        sh:property [
            sh:path saref:hasTimestamp ;
            sh:minCount 1 ;
            sh:datatype xsd:dateTime ;
            sh:message "Every measurement must have an xsd:dateTime timestamp." ;
        ] .
$$);

Step 5 — Route enriched events to downstream consumers

Now that observations are stored, enriched, and validated, we set up the outbound pipeline. We create a bridge table to hold outbound events, subscribe to the inferred alert triples, and configure relay forward pipelines to deliver them wherever they are needed.

-- The bridge table holds the outbound alert payloads before the trigger
-- publishes them to a pg_tide outbox.
CREATE TABLE enriched_events (
    id         BIGSERIAL PRIMARY KEY,
    event_type TEXT NOT NULL,
    payload    JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT now()
);

-- Declare which triples the subscription should watch for.
-- Only triples where the predicate is ex:tempAlert will pass through.
SELECT pg_ripple.create_subscription(
    name          => 'high-temp-alerts',
    filter_sparql => 'FILTER(?p = <https://example.org/tempAlert>)'
);

-- Wire the subscription to the outbox table.
-- This installs a trigger on the VP delta table for ex:tempAlert; whenever a
-- matching triple lands, the trigger decodes it and writes a row to
-- enriched_events. That is the only writer to enriched_events in this pipeline.
SELECT pg_ripple.enable_cdc_bridge_trigger(
    name      => 'high-temp-alerts',
    predicate => 'https://example.org/tempAlert',
    outbox    => 'enriched_events'
);

-- Reformat the raw decoded triple into plain JSON before the row is committed.
-- The frame template is a static JSONB literal: the @context maps short
-- property names to the stored SAREF predicate IRIs, and the property slots
-- ({}) tell pg_ripple which predicates to pull. Only the @id is dynamic —
-- it scopes the generated CONSTRUCT query to this specific observation.
-- The final unwrap + strip step removes JSON-LD structural keywords so the
-- consumer receives the same plain JSON as before, but the output shape is
-- now fully described by the frame rather than hand-assembled field by field.
CREATE OR REPLACE FUNCTION reformat_alert_payload()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
DECLARE
    framed JSONB;
    item   JSONB;
BEGIN
    framed := pg_ripple.export_jsonld_framed(
        frame => '{
            "@context": {
                "device": "https://saref.etsi.org/core/measurementMadeBy",
                "temp":   "https://saref.etsi.org/core/hasValue",
                "unit":   "https://qudt.org/schema/qudt/unit",
                "ts":     "https://saref.etsi.org/core/hasTimestamp",
                "alert":  "https://example.org/alertType"
            },
            "@type":  "https://saref.etsi.org/core/Measurement",
            "device": {},
            "temp":   {},
            "unit":   {},
            "ts":     {},
            "alert":  {}
        }'::jsonb || jsonb_build_object('@id', NEW.payload ->> 'subject')
    );

    -- @context lives at the top level of the framed document and is discarded
    -- automatically when we unwrap @graph. @type and @id sit on the node
    -- itself; strip them with the jsonb - operator to get plain JSON.
    item := (framed -> '@graph' -> 0) - '@type' - '@id';

    IF item IS NOT NULL THEN
        NEW.payload := item;
    END IF;

    RETURN NEW;
END;
$$;

CREATE TRIGGER reformat_alert
BEFORE INSERT ON enriched_events
FOR EACH ROW EXECUTE FUNCTION reformat_alert_payload();

-- Create a pg_tide outbox so the relay can poll it.
-- pg_tide manages all outbox messages in tide.tide_outbox_messages;
-- triggers call tide.outbox_publish() instead of inserting into a bridge table.
SELECT tide.outbox_create(
    'enriched-events',
    p_retention_hours  := 24,
    p_inline_threshold := 0
);

-- Install a publish trigger so the relay picks up enriched_events rows.
CREATE OR REPLACE FUNCTION bridge_alert_to_tide_outbox()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
    PERFORM tide.outbox_publish(
        'enriched-events',
        jsonb_build_object(
            'subject',   pg_ripple.decode_id(NEW.s),
            'predicate', pg_ripple.decode_id(TG_ARGV[0]::bigint),
            'object',    pg_ripple.decode_id(NEW.o),
            'graph',     pg_ripple.decode_id(NEW.g)
        ),
        '{}'::jsonb   -- headers
    );
    RETURN NEW;
END;
$$;

Now configure the relay to forward those events back to Kafka on the iot.alerts topic — the same broker the raw sensor readings came from:

-- Publish enriched alerts to iot.alerts on the same Kafka broker
SELECT tide.relay_set_outbox_v2(jsonb_build_object(
  'name',      'alerts-to-kafka',
  'outbox',    'enriched-events',
  'sink_type', 'kafka',
  'config',    jsonb_build_object(
    'brokers', '${env:KAFKA_BROKERS}',
    'topic',   'iot.alerts'
  )
));

The same outbox can fan out to additional sinks without any changes to the pg_ripple side — just register extra relay_set_outbox_v2 pipelines:

-- Also push to NATS for real-time dashboard consumers (optional)
SELECT tide.relay_set_outbox_v2(jsonb_build_object(
  'name',      'alerts-to-nats',
  'outbox',    'enriched-events',
  'sink_type', 'nats',
  'config',    jsonb_build_object(
    'url',     'nats://nats:4222',
    'subject', 'iot.alerts.{event_type}'
  )
));

-- Or to a partner API via webhook (optional)
SELECT tide.relay_set_outbox_v2(jsonb_build_object(
  'name',      'alerts-to-partner',
  'outbox',    'enriched-events',
  'sink_type', 'webhook',
  'config',    jsonb_build_object(
    'url',    'https://partner.example.com/events',
    'method', 'POST'
  )
));

The reformat_alert_payload trigger produces the same plain JSON as the previous jsonb_build_object version, but the entire output shape is now described by the static frame template — including the alert field. The ex:alertType literal triple was inferred by the Datalog rule in Step 3, so the frame simply maps the short name "alert" to that predicate and picks it up like any other property; there is nothing to hardcode in the trigger. The || appends only the dynamic @id to scope the generated CONSTRUCT query to this specific observation. Two steps then strip the JSON-LD structural keywords: -> '@graph' -> 0 unwraps the node and discards the top-level @context; - '@type' - '@id' removes the remaining keywords from the node. A consumer reading from iot.alerts sees a plain JSON document it can process without any knowledge of RDF or the triplestore:

{
  "device": "sensor-7",
  "temp":   45.2,
  "unit":   "°C",
  "ts":     "2026-04-28T10:00:00Z",
  "alert":  "high_temperature"
}

Compare this with the original inbound message from Step 1: same field names (device, temp, unit, ts), one new alert field derived by the inference rules. The values here come from a different observation (the 45.2°C reading that triggered the alert), but the shape is identical. The triplestore and Datalog engine are invisible to both the producer and the consumer — they see only ordinary JSON.

Choosing a CDC bridge approach

Step 5 used enable_cdc_bridge_trigger (Approach A below) — the simplest option, with latency under 10 ms. Two other approaches offer different trade-offs for different data paths in the same hub. All three write to the same enriched_events outbox table; only the mechanism that detects new triples and writes outbox rows differs.

Approach A — Trigger bridge (lowest latency, < 10 ms)

The simplest approach: a trigger on VP delta tables fires in the same transaction that inserted the triple. The triple is decoded from its internal integer representation and written directly to the bridge table before the transaction commits. Nothing can slip through — the data is in both places or neither:

CREATE OR REPLACE FUNCTION _pg_ripple.bridge_to_outbox()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
    INSERT INTO enriched_events (event_type, payload)
    VALUES (
        TG_OP,
        jsonb_build_object(
            'subject',   pg_ripple.decode_id(NEW.s),
            'predicate', pg_ripple.decode_id(TG_ARGV[0]::bigint),
            'object',    pg_ripple.decode_id(NEW.o),
            'graph',     pg_ripple.decode_id(NEW.g)
        )
    );
    RETURN NEW;
END;
$$;

The pg_ripple.enable_cdc_bridge_trigger() call in Step 5 installs exactly this pattern automatically — you do not need to write the trigger function by hand. The raw {subject, predicate, object, graph} payload is useful on its own, but you can also reshape it before it leaves the outbox:

Plain JSON via framing — add a BEFORE INSERT trigger that calls export_jsonld_framed() with the desired @context, unwraps @graph, and strips @type and @id with the jsonb - operator, as Step 5 does with reformat_alert_payload(). This is more declarative than building the payload manually with jsonb_build_object — changing an outbound property name is a one-line @context edit rather than a code change.
Full JSON-LD — store the framed document directly without stripping (see Outbound framing).

Best for: High-priority alerts, strict transactional guarantees. Trade-off: decode_id() is called once per row, which adds overhead on high-volume write paths. For filtering — forwarding only alerts and not every measurement — you need an extra WHERE condition in the trigger.

Approach B — Background worker bridge (best throughput, 50–500 ms)

The pg_ripple background worker (enabled since v0.52.0) wakes up when it receives CDC NOTIFY events, collects a batch of them, decodes all the integer dictionary IDs in a single bulk SPI call, and inserts the decoded rows into the bridge table in one go. The amortised cost per triple is much lower than the trigger approach, and the batch size and flush interval are configurable.

This approach adds a small latency window (the batch collection time), but for high-volume enriched-data streams — thousands of triples per second — the throughput improvement is significant. Use this as the default for bulk data paths.

Best for: Bulk enriched data, high-volume streams. Trade-off: Adds configurable milliseconds of latency.

Approach C — Named subscription with SPARQL CONSTRUCT view (most flexible)

Named subscriptions (v0.42.0+) let you attach a SPARQL FILTER expression that controls exactly which triples are bridged. Only the triples that match the filter expression will ever touch the outbox, which is useful when the inference rules produce many intermediate triples that you do not want to forward:

-- Only bridge the final alert triples, not intermediate inference steps
SELECT pg_ripple.create_subscription(
    name          => 'alerts',
    filter_sparql => 'FILTER(?p = <https://example.org/tempAlert>)'
);

Combine this with export_jsonld_framed() to shape the outbound payload into exactly the JSON structure the downstream consumer expects. A JSON-LD frame is a template you write once that describes the desired nesting and field names. pg_ripple translates it to a SPARQL CONSTRUCT query internally, executes it, applies the W3C embedding algorithm, and compacts the result with your @context:

SELECT pg_ripple.export_jsonld_framed(
    frame => '{
        "@context": {
            "ex":     "https://example.org/",
            "schema": "https://schema.org/",
            "saref":  "https://saref.etsi.org/core/",
            "xsd":    "http://www.w3.org/2001/XMLSchema#",
            "alertLevel": "ex:alertLevel",
            "name":       "schema:name",
            "latestTemp": "ex:latestTemp"
        },
        "@type": "ex:Alert",
        "alertLevel": {},
        "name": {},
        "latestTemp": {}
    }'::jsonb
);

The framed output is a clean, nested JSON-LD document ready to publish through the pg_tide outbox:

{
  "@context": {
    "ex":     "https://example.org/",
    "schema": "https://schema.org/",
    "xsd":    "http://www.w3.org/2001/XMLSchema#",
    "alertLevel": "ex:alertLevel",
    "name":       "schema:name",
    "latestTemp": "ex:latestTemp"
  },
  "@graph": [
    {
      "@id":       "https://example.org/device/sensor-7",
      "@type":     "ex:Alert",
      "alertLevel": "HIGH",
      "name":       "Boiler Room Sensor 7",
      "latestTemp": { "@value": "45.2", "@type": "xsd:decimal" }
    }
  ]
}

Use jsonld_frame_to_sparql(frame => ...) to inspect the generated CONSTRUCT query before running the full export — this is useful for performance tuning.

Best for: Complex SPARQL-shaped payloads, scheduled reports, ad-hoc shapes. Trade-off: Polling-based unless combined with a LISTEN/NOTIFY wake-up.

Which approach to use

In practice you will use all three for different data paths in the same hub:

Data path	Mechanism	Typical latency
High-priority alerts	Approach A — trigger bridge	< 10 ms
Bulk enriched data	Approach B — background worker	50–500 ms
Shaped reports / views	Approach C — SPARQL CONSTRUCT	Cron-driven

Common patterns

Unifying records from multiple sources

A real hub almost always receives data about the same real-world entities from multiple systems — a customer appears in the CRM as crm:C1 and in the ERP as erp:A1, but they are the same person. Datalog owl:sameAs rules detect these overlaps from shared attributes (email address, serial number, phone number) and create linking triples that allow downstream SPARQL queries to treat both records as one:

% Link CRM and ERP records that share an email address
owl:sameAs(CrmCust, ErpAcct) :-
    crm:emailAddress(CrmCust, E),
    erp:contact_email(ErpAcct, E).

Once those owl:sameAs triples are materialised, pg_ripple's OWL RL canonicalisation ensures that any query for crm:C1 will transparently find data originally stored under erp:A1 as well.

Speaking a common language to downstream consumers

Each source system uses its own property names. Your CRM calls it crm:customerName; the ERP calls it erp:accountTitle. Rather than requiring every downstream consumer to understand every source vocabulary, Datalog rules project everything onto a single shared ontology (here, Schema.org):

% Both CRM and ERP names map to schema:name
schema:name(X, V)  :- crm:customerName(X, V).
schema:name(X, V)  :- erp:accountTitle(X, V).

% Both email fields map to schema:email
schema:email(X, V) :- crm:emailAddress(X, V).
schema:email(X, V) :- erp:contact_email(X, V).

Downstream consumers now only need to understand Schema.org. Source schema changes are isolated to a single Datalog rule update in the hub — no downstream changes required.

Rich JSON-LD for event-driven downstream consumers

Here is what a fully enriched and shaped customer record looks like by the time it reaches a downstream consumer via the relay. Notice how all the messy source-system vocabulary has been replaced with clean Schema.org terms, and the document is self-describing thanks to the @context:

{
  "@context": {
    "schema": "https://schema.org/",
    "ex":     "https://example.org/",
    "xsd":    "http://www.w3.org/2001/XMLSchema#"
  },
  "@id":   "https://example.org/customer/C1",
  "@type": "schema:Customer",
  "schema:name":  "Jane Doe",
  "schema:email": "jane@example.com",
  "ex:riskScore": { "@value": "0.87", "@type": "xsd:decimal" },
  "ex:highValueCustomer": true,
  "owl:sameAs": [
    { "@id": "https://erp.example.com/accounts/A1" },
    { "@id": "https://support.example.com/tickets/T9" }
  ],
  "_relay_dedup_key": "ripple:4200042"
}

The owl:sameAs array shows the entity resolution result — this one record links the CRM, ERP, and support-ticket identities together. The ex:riskScore was derived by a Datalog rule from order history data.

The _relay_dedup_key is a convention for idempotent delivery: its value is derived from pg_ripple's internal statement ID (the i column in VP tables), so downstream consumers can detect and discard duplicates even if the relay restarts and replays part of the outbox. Including it is optional — the worked example above omits it because the plain-JSON reformat trigger does not add it, but it is straightforward to include via '_relay_dedup_key', 'ripple:' || NEW.i in the jsonb_build_object call.

JSON-LD mapping: inbound context and outbound framing

JSON-LD mapping is the mechanism that lets pg_ripple act as a true semantic bridge: arbitrary JSON goes in, canonicalised knowledge graph triples are stored, and shaped JSON-LD comes out. The inbound and outbound sides each have a dedicated function.

Inbound — `json_to_ntriples_and_load()` with a context map

When the relay delivers a raw JSON event, pg_ripple.json_to_ntriples_and_load() converts it to RDF triples in one step. The context parameter is a JSONB object that works like a JSON-LD @context:

Incoming JSON                    context map                     stored triples
─────────────                    ───────────                     ──────────────
{ "temp": 45.2, "device": ... }  "temp"  → saref:hasValue   →   <obs> saref:hasValue "45.2"^^xsd:decimal
                                 "device"→ saref:madeby      →   <obs> saref:madeby <device-7>
                                 @vocab  → saref:             →   unmapped keys get saref: prefix

It supports:

@vocab — default IRI prefix for all keys not explicitly listed.
Explicit key-to-IRI mappings — override specific fields with exact predicate IRIs.
Nested objects — become blank nodes with their own predicates resolved through the same context.
Arrays — produce one triple per element.
null values — silently skipped.

The context is stored only in the trigger definition, not in the triplestore. When a source vendor renames a field, you update the context JSONB; no triples need to change.

-- Full context example for a heterogeneous event shape
PERFORM pg_ripple.json_to_ntriples_and_load(
    payload     => NEW.payload,
    subject_iri => 'https://example.org/event/' || NEW.event_id,
    type_iri    => 'https://schema.org/Event',
    context     => '{
        "@vocab":      "https://schema.org/",
        "ts":          "https://schema.org/startDate",
        "location":    "https://schema.org/location",
        "description": "https://schema.org/description",
        "external_id": "https://example.org/externalId"
    }'::jsonb
);

Outbound — `export_jsonld_framed()` with a frame template

On the way out, pg_ripple.export_jsonld_framed() (v0.17.0+) shapes the flat RDF into whatever nested JSON structure the downstream consumer expects. A frame is a JSON template that describes the desired structure; pg_ripple handles everything else:

Translates the frame into a SPARQL CONSTRUCT query.
Executes the query against the triplestore.
Applies the W3C JSON-LD 1.1 embedding algorithm to produce nested nodes.
Compacts IRI strings using the frame's @context.

stored triples               frame template                  outbound JSON-LD
──────────────               ──────────────                  ───────────────
<device> schema:name "X"     "@type": "schema:Device"        { "@type": "Device",
<device> ex:temp 45.2        "name": {},                →      "name": "X",
<device> ex:alert "HIGH"     "temp": {},                       "temp": 45.2,
                             "alert": {}                        "alert": "HIGH" }

Use this inside the outbox bridge trigger to produce consumer-ready JSON-LD:

-- Write a framed JSON-LD event to the outbox every time an alert triple lands
CREATE OR REPLACE FUNCTION bridge_alert_to_outbox()
RETURNS TRIGGER LANGUAGE plpgsql AS $$
DECLARE
    framed JSONB;
BEGIN
    framed := pg_ripple.export_jsonld_framed(
        frame => '{
            "@context": {
                "schema": "https://schema.org/",
                "ex":     "https://example.org/",
                "xsd":    "http://www.w3.org/2001/XMLSchema#",
                "name":       "schema:name",
                "latestTemp": "ex:latestTemp",
                "alertLevel": "ex:alertLevel"
            },
            "@type": "ex:Alert",
            "name": {},
            "latestTemp": {},
            "alertLevel": {}
        }'::jsonb
    );

    INSERT INTO enriched_events (event_type, payload)
    VALUES ('alert', framed);

    RETURN NEW;
END;
$$;

The downstream consumer receives a document shaped exactly to the frame — with short, readable property names from the @context, nested objects where the frame requests them, and a self-describing @context block so it can be parsed without any knowledge of pg_ripple's internals.

If the downstream consumer expects plain JSON rather than JSON-LD, strip the structural keywords before inserting into the outbox:

-- @context is at the top level; -> '@graph' -> 0 discards it automatically.
-- @type and @id live on the node; chain - to remove them in one expression.
item := (framed -> '@graph' -> 0) - '@type' - '@id';

INSERT INTO enriched_events (event_type, payload)
VALUES ('alert', item);

The jsonb - operator accepts either a single key (- '@type') or a text array (- ARRAY['@type','@id']) to remove several keywords at once.

Debugging the frame translation

Before running export_jsonld_framed() in production, use jsonld_frame_to_sparql() to see the SPARQL CONSTRUCT query that will be generated. This is useful for verifying that the frame matches your stored triple shapes and for identifying any missing patterns before they cause silent empty results:

SELECT pg_ripple.jsonld_frame_to_sparql(
    frame => '{
        "@context": { "schema": "https://schema.org/" },
        "@type":    "schema:Device",
        "schema:name": {}
    }'::jsonb
);

Symmetric round-trip

The two functions form a symmetric pair: json_to_ntriples_and_load() maps field names from the source JSON vocabulary to RDF predicates on the way in; export_jsonld_framed() maps those same predicates back to the field names and nested structure the consumer needs on the way out. You can use different @context definitions for different consumers — the triplestore is the stable canonical representation in the middle, and the vocabularies at each edge are entirely configurable.

Deployment

pg_tide and pg_ripple run in the same PostgreSQL instance. The pg-tide relay process is separate and stateless. A single relay instance handles both directions — inbound pipelines (external source → inbox table) and outbound pipelines (outbox table → external sink) run in the same process. You do not need separate relay binaries for each direction.

For high availability, run two or three relay instances pointing at the same PostgreSQL database. PostgreSQL advisory locks elect exactly one owner per pipeline — if one instance dies, another acquires its pipelines on the next discovery interval.

The relay only needs one environment variable to start: the database URL. All pipeline configuration is registered in the database via SQL (tide.relay_set_outbox_v2() / tide.relay_set_inbox_v2()), and the relay reads it on startup and hot-reloads it when you make changes — no restart required.

Sensitive values like broker addresses can use ${env:VAR} placeholders inside the JSONB config. The relay expands them from its own process environment at runtime, so credentials never need to be stored in the database.

# docker-compose.yml sketch
services:
  postgres:
    image: ghcr.io/trickle-labs/pg-ripple:latest
    # pg_ripple, pg_trickle, pg_tide, PostGIS, and pgvector all pre-installed

  relay:
    image: ghcr.io/trickle-labs/pg-tide:0.33.0
    environment:
      PG_TIDE_POSTGRES_URL: postgres://relay:pw@postgres/hub
      # All pipeline config (topics, subjects, poll intervals) lives in the DB.
      # Broker addresses can use ${env:VAR} refs — set them here as env vars.
      # Register pipelines with tide.relay_set_outbox_v2() / tide.relay_set_inbox_v2().
      KAFKA_BROKERS: kafka:9092   # expanded by ${env:KAFKA_BROKERS} in pipeline config
      NATS_URL: nats://nats:4222  # similarly available via ${env:NATS_URL}
    ports:
      - "9090:9090"   # Prometheus metrics + /health endpoint

  pg-ripple-http:
    image: pg-ripple-http:latest
    environment:
      DATABASE_URL: postgres://ripple:pw@postgres/hub
    ports:
      - "8080:8080"
    # SPARQL endpoint for ad-hoc queries from dashboards and tools

  kafka:
    image: redpandadata/redpanda:latest

  nats:
    image: nats:latest
    command: ["-js"]   # JetStream enabled for durable subscriptions

For Kubernetes deployments, the relay's /health endpoint integrates with readiness probes. See Kubernetes & Helm for a full Helm chart example.

Things to watch out for

Outbox growing faster than it drains (backpressure)

When a single inbound event triggers many inferred triples — for example, an owl:sameAs merge that touches hundreds of related facts — the outbox can accumulate rows faster than the relay drains them. Three controls help:

Use pg_tide's retention drain (tide.outbox_create(p_retention_hours => N)) to cap outbox size and drop the oldest rows once a maximum depth is reached.
Use the relay's /health endpoint as a Kubernetes readiness probe. When the relay falls behind, it signals not-ready, letting the cluster apply back-pressure to inbound sources.
Use pg_ripple's source column to bridge only explicit triples (source = 0) and suppress inferred triples (source = 1) from a particular outbox, reducing volume without changing the inference rules.

Keeping consumers in sync after rule changes

When you update a Datalog rule or SHACL shape — adding a new derived property, changing a threshold — downstream consumers are receiving a different data shape than before. Manage this with:

Version the subject template in the relay outbox configuration: ripple.v2.enriched.{type} rather than ripple.enriched.{type}.
Include a @context version field in outbound JSON-LD payloads so consumers can detect schema changes programmatically.
Use the relay's full-refresh mode to re-snapshot the entire outbox after a rule change, ensuring consumers that missed the transition catch up.

Bidirectional CRM ⇄ ERP Walkthrough (v0.77.0)

Available since: v0.77.0

v0.77.0 adds the generic bidirectional integration primitives that make it possible to build a two-way relay between any two systems through pg_ripple. This walkthrough uses a CRM ⇄ ERP scenario as the concrete example.

Topology

  CRM (source of truth       ERP (source of truth
   for email/phone)           for tax ID/status)
       │                           │
       │  pg-tide relay            │  pg-tide relay
       ▼                           ▼
  <urn:source:crm>           <urn:source:erp>
       │                           │
       └──────────┬────────────────┘
                  ▼
           pg_ripple hub
           (named graphs)
                  │
          conflict policies
          (source_priority)
                  │
          resolved projection
                  │
    ┌─────────────┴─────────────┐
    ▼                           ▼
  CRM outbox               ERP outbox
  (pg-tide)                (pg-tide)

Step 1: Register mappings with default graph IRIs

-- CRM mapping: contacts
SELECT pg_ripple.register_json_mapping(
    'crm_contact',
    '{"@context": {
        "ex:name":    "http://schema.org/name",
        "ex:email":   "http://schema.org/email",
        "ex:phone":   "http://schema.org/telephone"
    }}'::jsonb,
    default_graph_iri   => '<urn:source:crm>',
    timestamp_path      => '$.lastModified',
    iri_template        => 'https://crm.example.com/contacts/{id}'
);

-- ERP mapping: employees
SELECT pg_ripple.register_json_mapping(
    'erp_employee',
    '{"@context": {
        "ex:name":   "http://schema.org/name",
        "ex:taxId":  "http://schema.org/taxID",
        "ex:status": "http://schema.org/employmentType"
    }}'::jsonb,
    default_graph_iri   => '<urn:source:erp>',
    timestamp_path      => '$.updatedAt',
    iri_template        => 'https://erp.example.com/employees/{id}'
);

Step 2: Declare conflict resolution policies

-- Email: CRM is authoritative
SELECT pg_ripple.register_conflict_policy(
    'http://schema.org/email',
    'source_priority',
    '{"order": ["<urn:source:crm>", "<urn:source:erp>"]}'::jsonb
);

-- Name: last-write wins across both systems
SELECT pg_ripple.register_conflict_policy(
    'http://schema.org/name',
    'latest_wins'
);

-- Tax ID: ERP is sole source; reject any CRM attempt
SELECT pg_ripple.register_conflict_policy(
    'http://schema.org/taxID',
    'reject_on_conflict'
);

Step 3: Ingest payloads

-- Ingest a CRM contact update (diff mode: derives per-triple timestamps)
SELECT pg_ripple.ingest_json(
    '{"ex:name": "Alice Smith", "ex:email": "alice@crm.example.com"}'::jsonb,
    'https://crm.example.com/contacts/42',
    'crm_contact',
    mode => 'diff'
);

-- Ingest the matching ERP employee (upsert mode: replaces sh:maxCount 1 fields)
SELECT pg_ripple.ingest_json(
    '{"ex:name": "Alice Smith", "ex:taxId": "123-45-6789"}'::jsonb,
    'https://erp.example.com/employees/E-999',
    'erp_employee',
    mode => 'upsert'
);

Step 4: Cross-source entity linking (BIDI-REF-01)

After the ERP confirms Alice is the same person as CRM contact 42:

-- Record the linkback: CRM hub IRI ↔ ERP employee IRI
SELECT pg_ripple.record_linkback(
    'event-uuid-from-outbox'::uuid,
    target_iri => 'https://erp.example.com/employees/E-999'
);

This writes <https://crm.example.com/contacts/42> owl:sameAs <https://erp.example.com/employees/E-999> into the target graph and flushes any buffered events.

Step 5: Loop-safe subscriptions (BIDI-LOOP-01)

When creating subscriptions, use exclude_graphs to prevent echo loops:

-- CRM subscription: exclude CRM's own graph to prevent loop
SELECT pg_ripple.create_subscription(
    'crm_contact_sync',
    outbox_table => 'pg_ripple_outbox.crm_contact_events'
    -- Note: exclude_graphs and propagation_depth are set via the
    -- _pg_ripple.subscriptions table after creation:
    -- UPDATE _pg_ripple.subscriptions
    --   SET exclude_graphs = ARRAY['<urn:source:crm>'],
    --       propagation_depth = 1
    -- WHERE name = 'crm_contact_sync';
);

Step 6: Observe per-graph metrics (BIDI-OBS-01)

SELECT graph_iri, triple_count, last_write_at, conflicts_total
FROM pg_ripple.graph_stats()
ORDER BY triple_count DESC;

Outbound event wire format (BIDI-WIRE-01)

Every event emitted to an outbox table follows this shape:

{
  "version": "1.0",
  "event_id": "550e8400-e29b-41d4-a716-446655440000",
  "subscription": "crm_contact_sync",
  "subject": "https://crm.example.com/contacts/42",
  "source_graph": "urn:source:crm",
  "after": {
    "http://schema.org/name": "Alice Smith",
    "http://schema.org/email": "alice@crm.example.com"
  },
  "base": {
    "http://schema.org/name": "Alice"
  },
  "subject_resolved": true
}

The base object carries the previous values for changed predicates, enabling compare-and-swap (CAS) safety at the receiving system. See assert_cas(event, actual) for a built-in CAS helper.

The full JSON Schema is at docs/src/operations/event-schema-v1.json.