Quickstart

This walkthrough starts a self-contained CDC stack with pre-seeded PostgreSQL and Neo4j data. It bootstraps denormalized documents into OpenSearch, then shows inserts, updates, relationship changes, and deletes propagating from each source.

This guide covers the capture pipeline. For live events over WebSocket and GraphQL, use the real-time demo and Real-time subscriptions.

The demo runs the engines standalone. A running standalone process cannot be attached to VentStream Cloud in place. To use centralized configuration, visibility, and lifecycle operations, create a managed pipeline and deploy a new managed agent.

The sample data uses an e-commerce domain:

Source	Target	What one document looks like
Postgres `shop.orders`	OS index `orders`	order + embedded customer (1:1) + line items (1:many)
Neo4j `Product`	OS index `products`	product + category + supplier→region (2 hops) + tags

Everything lives in demo/stack/: the docker-compose.yml, the seed data, and the two projection specs.

Prerequisites

Docker + Docker Compose v2 (docker compose version)
~4 GB free RAM for the containers
Free ports: 5544 (pg), 7474/7687 (neo4j), 9200 (opensearch), 5601 (dashboards, optional)

cd ventstream/demo/stack

Preflight: clean up and check port 9200

Start from a clean demo state so the seed counts and document IDs in this guide are deterministic:

docker compose --profile dashboards down -v --remove-orphans

This removes only the containers, networks, and volumes created by this Compose project. If you are intentionally resuming an earlier run, omit -v; the exact counts below may then differ if you changed or deleted sample data. The demo must have exclusive access to host port 9200. Check for another Docker container or local OpenSearch process before continuing:

docker ps --filter publish=9200 --format 'table {{.Names}}\t{{.Ports}}'
lsof -nP -iTCP:9200 -sTCP:LISTEN   # macOS and most developer workstations

Both checks should return no listener before you start the demo. On Docker Desktop, the demo container can sometimes appear to start even when a host process already owns localhost:9200. In that state, the verification curl commands may reach the wrong OpenSearch cluster. Stop the process shown by the checks above before continuing. For a Homebrew-managed OpenSearch, use brew services stop opensearch. For another Compose stack, run docker compose down from that stack’s directory. Do not kill an unrelated process without first identifying it.

1. Start the sources + target

--wait blocks until all three pass their healthchecks, so you don’t have to poll:

docker compose up -d --wait postgres neo4j opensearch

(Check status any time with docker compose ps.)

First boot pulls images (Postgres, Neo4j Enterprise, OpenSearch). Postgres seeds itself from seed/postgres.sql — 200 orders, 5 customers, line items, and the ventstream_shop publication.

2. Seed Neo4j + enable CDC

Postgres is already seeded. Neo4j needs two one-time commands — enable CDC, then load the catalog graph:

# Enable CDC (DIFF enrichment) on the default database. This is a
# per-database setting, NOT a server config. DIFF is the recommended
# mode for projections — see the Neo4j connector docs for why.
docker compose exec -T neo4j cypher-shell -u neo4j -p ventstream \
  "ALTER DATABASE neo4j SET OPTION txLogEnrichment 'DIFF';"

# Load the catalog (2,000 products, 4 categories, 20 suppliers, 3 regions).
docker compose exec -T neo4j cypher-shell -u neo4j -p ventstream \
  < seed/neo4j.cypher

3. Start the engines

The engines build from source via cargo-chef. Allow 5–10 minutes for the first build, depending on CPU and network speed; later runs are cached and start in seconds.

docker compose up -d --build engine-orders engine-products

Watch them bootstrap and switch to tailing. The demo enables debug logging so source progress, projection work, and sink batches are visible:

docker compose logs -f engine-orders

docker compose logs -f engine-products

Each engine moves through its lifecycle:

STARTING

Process up, source connection opening.

BOOTSTRAPPING

Snapshot scan — existing rows/nodes stream into OpenSearch.

TAILING

Steady state — live CDC changes flow as they happen.

4. Verify the initial load

curl -s 'http://localhost:9200/orders/_count'   | jq .count   # → 200
curl -s 'http://localhost:9200/products/_count' | jq .count   # → 2000

Inspect one denormalized order (Postgres source) — note the embedded customer and items:

curl -s 'http://localhost:9200/orders/_doc/shop.orders:%5B%22ord-0001%22%5D' | jq '._source'

Inspect one denormalized product (Neo4j source) — note the embedded category, supplier → region (2 hops), and tags:

curl -s 'http://localhost:9200/products/_search' -H 'content-type: application/json' -d '
  {"size":1,"query":{"term":{"id.keyword":"prod-2"}}}' | jq '.hits.hits[0]._source'

The Postgres doc _id is the fully-qualified table name plus the PK as a JSON array: shop.orders:["ord-0001"] (URL-encoded above), so a GET …/_doc/<id> works directly. The Neo4j doc _id is products_denormalized:<elementId>, so search by a field (above) rather than guessing the elementId.

5. Watch changes propagate

Keep the engine logs visible while you run these. Each change should appear in the log within ~1s and update OpenSearch.

Postgres
Neo4j

Update a row (1-hop)

docker compose exec -T postgres psql -U ventstream -d shop -c \
  "UPDATE shop.orders SET status='shipped', total=999.99 WHERE order_id='ord-0001';"

curl -s 'http://localhost:9200/orders/_doc/shop.orders:%5B%22ord-0001%22%5D' \
  | jq '._source | {status, total}'

Rename a customer (1:1 cascade) — every order for cust-002 recomposes:

docker compose exec -T postgres psql -U ventstream -d shop -c \
  "UPDATE shop.customers SET tier='platinum', name='Alan T. (VIP)' WHERE customer_id='cust-002';"

curl -s 'http://localhost:9200/orders/_doc/shop.orders:%5B%22ord-0001%22%5D' \
  | jq '._source.customer'

Add a line item (1:many cascade)

docker compose exec -T postgres psql -U ventstream -d shop -c \
  "INSERT INTO shop.order_items (item_id, order_id, sku, qty, price)
   VALUES ('item-0001-new', 'ord-0001', 'SKU-9999', 5, 49.99);"

curl -s 'http://localhost:9200/orders/_doc/shop.orders:%5B%22ord-0001%22%5D' \
  | jq '._source.items'

Delete a primary row — items first (FK), then the order:

docker compose exec -T postgres psql -U ventstream -d shop -c \
  "DELETE FROM shop.order_items WHERE order_id='ord-0001';
   DELETE FROM shop.orders WHERE order_id='ord-0001';"

# A GET by _id is real-time; _count lags ~1s on OpenSearch's refresh.
curl -s -o /dev/null -w '%{http_code}\n' \
  'http://localhost:9200/orders/_doc/shop.orders:%5B%22ord-0001%22%5D'   # → 404

Rename a hot shared node (bounded fan-out) — Category is a low-cardinality lookup (~500 products each). A property change on the node correctly cascades to every product in that category; hot-endpoint detection only filters relationship churn that would explode the fan-out.

docker compose exec -T neo4j cypher-shell -u neo4j -p ventstream \
  "MATCH (c:Category {id:'cat-electronics'}) SET c.name='Electronics & Gadgets';"

# Exact match via .keyword — a plain `match` would also hit the old
# "Electronics" through the shared "electronics" token.
curl -s 'http://localhost:9200/products/_search' -H 'content-type: application/json' -d '
  {"size":0,
   "query":{"term":{"category.id.keyword":"cat-electronics"}},
   "aggs":{"names":{"terms":{"field":"category.name.keyword"}}}}' \
  | jq '.aggregations.names.buckets'
# → [ { "key": "Electronics & Gadgets", "doc_count": 500 } ]

Multi-hop cascade (Supplier → Region) — move a supplier’s region; every product of that supplier recomposes 2 hops out:

docker compose exec -T neo4j cypher-shell -u neo4j -p ventstream \
  "MATCH (s:Supplier {id:'sup-1'})-[r:LOCATED_IN]->() DELETE r
   WITH s MATCH (reg:Region {id:'reg-apac'}) CREATE (s)-[:LOCATED_IN]->(reg);"

curl -s 'http://localhost:9200/products/_search' -H 'content-type: application/json' -d '
  {"size":1,"query":{"term":{"supplier.id.keyword":"sup-1"}}}' \
  | jq '.hits.hits[0]._source.supplier'   # → region.name = "Asia Pacific"

Delete a product

docker compose exec -T neo4j cypher-shell -u neo4j -p ventstream \
  "MATCH (p:Product {id:'prod-1'}) DETACH DELETE p;"

curl -s 'http://localhost:9200/products/_count' | jq .count   # → 1999

Stream it continuously (optional)

To see the pipeline move, fire updates fast and watch the engine react. We stream the statements into a single psql connection (a per-command docker exec would cap the rate at the container-exec overhead) and pace server-side with pg_sleep:

# terminal 1 — one update roughly every 100 ms over a single connection
while true; do
  printf "UPDATE shop.orders SET status=(ARRAY['pending','paid','shipped','delivered'])[1+floor(random()*4)], total=round((random()*1000)::numeric,2) WHERE order_id='ord-0002';\nSELECT pg_sleep(0.1);\n"
done | docker compose exec -T postgres psql -U ventstream -d shop -q

# terminal 2 — watch the engine: flush → ack → cursor advance
docker compose logs -f engine-orders

Or watch the document change in lockstep:

while true; do
  curl -s 'http://localhost:9200/orders/_doc/shop.orders:%5B%22ord-0002%22%5D' \
    | jq -c '._source | {status, total}'
  sleep 0.2
done

Ctrl-C to stop. Tune the pace with pg_sleep(0.1) — 0.02 for ~20 ms, or drop it entirely for max throughput. Each update recomputes only that one document; the dispatcher may combine nearby updates into one sink batch. Swap the WHERE to order_id='ord-'||lpad((1+floor(random()*200))::int::text,4,'0') to spread updates across all 200 orders.

6. Optional — inspection UIs

OpenSearch Dashboards

docker compose --profile dashboards up -d dashboards

Open http://localhost:5601 → Dev Tools:

GET orders/_search
GET products/_search

Neo4j Browser

Always on at http://localhost:7474 (user neo4j, password ventstream). Explore the source graph:

MATCH (p:Product)-[:IN_CATEGORY]->(c:Category) RETURN p, c LIMIT 25

7. Verify data flow

# Source counts
docker compose exec -T postgres psql -U ventstream -d shop -tc \
  "SELECT count(*) FROM shop.orders;"
docker compose exec -T neo4j cypher-shell -u neo4j -p ventstream \
  "MATCH (p:Product) RETURN count(p);"

# Target counts (match the source, minus anything you deleted)
curl -s localhost:9200/orders/_count   | jq .count
curl -s localhost:9200/products/_count | jq .count

# Is the Postgres replication slot active?
docker compose exec -T postgres psql -U ventstream -d shop -c \
  "SELECT slot_name, active FROM pg_replication_slots;"

# Any dead-lettered events? This should print nothing.
docker compose logs engine-orders engine-products \
  | grep 'metric="dlq.write"' || true

8. Teardown

# Keep your data — stop + remove containers/networks but KEEP volumes
# (fast re-run; engines resume from cursor/state, sources keep their data):
docker compose --profile dashboards down --remove-orphans

# Full reset — also drop Postgres data, Neo4j store, and engine state:
docker compose --profile dashboards down -v --remove-orphans

-v drops the volumes (Postgres data, Neo4j store, engine state) and the next up re-seeds from scratch — omit it to keep your data. --remove-orphans clears leftover containers from old runs so a re-run doesn’t hit name/port clashes.

Next steps

Author a projection

The orders and products specs, field by field.

How fan-out works

Why renaming a category updates 500 docs without exploding.

Use VentStream Cloud

Enroll managed agents and administer pipelines from the dashboard or CLI.

Deploy to Kubernetes

Ship the whole stack with the bundled manifests.

Get started

Concepts

Connectors

Guides

Engineering

VentStream Cloud

Deploy

Reference

Prerequisites

Preflight: clean up and check port 9200

1. Start the sources + target

2. Seed Neo4j + enable CDC

3. Start the engines

4. Verify the initial load

5. Watch changes propagate

Stream it continuously (optional)

6. Optional — inspection UIs

7. Verify data flow

8. Teardown

Next steps

Author a projection

How fan-out works

Use VentStream Cloud

Deploy to Kubernetes

​Prerequisites

​Preflight: clean up and check port 9200

​1. Start the sources + target

​2. Seed Neo4j + enable CDC

​3. Start the engines

​4. Verify the initial load

​5. Watch changes propagate

​Stream it continuously (optional)

​6. Optional — inspection UIs

​7. Verify data flow

​8. Teardown

​Next steps

Author a projection

How fan-out works

Use VentStream Cloud

Deploy to Kubernetes

Prerequisites

Preflight: clean up and check port 9200

1. Start the sources + target

2. Seed Neo4j + enable CDC

3. Start the engines

4. Verify the initial load

5. Watch changes propagate

Stream it continuously (optional)

6. Optional — inspection UIs

7. Verify data flow

8. Teardown

Next steps