Neo4j CDC Concepts, Change Identifiers, Event Semantics, Capture Scope, and Consumer Checkpoints

// OPTIONAL — self-managed Enterprise 2026.07.1, run against system database.
CYPHER 25
ALTER DATABASE neo4j SET OPTION txLogEnrichment "FULL";
SHOW DATABASES YIELD name, options
WHERE name = 'neo4j'
RETURN name, options;
// Expected invariant: options contains txLogEnrichment: "FULL".

// OPTIONAL — Enterprise/Aura supported CDC database.
// Transaction A: capture an exclusive starting cursor.
CYPHER 25
CALL db.cdc.current() YIELD id, txCommitTime
RETURN id, txCommitTime;

// Transaction B: make one AtlasMart change.
CYPHER 25
MERGE (p:Product {productId:'P-2001'})
SET p.name='Trail Camera Pro', p.price=189.00, p.labTag='ch20';

// Transaction C: use the id from Transaction A as $from.
CYPHER 25
CALL db.cdc.query($from, [
  {select:'n', labels:['Product'], operation:'c'},
  {select:'n', labels:['Product'], operation:'u'}
])
YIELD id, txId, seq, metadata, event
RETURN id, txId, seq, metadata.txCommitTime AS committedAt,
       event.operation AS op, event.keys AS keys,
       event.state.before AS before, event.state.after AS after;

CYPHER 25
CALL db.cdc.query($from, [
  {select:'n', labels:['Product'], operation:'u', changesTo:['price']},
  {select:'r', type:'CONTAINS', operation:'c'}
])
YIELD id, txId, seq, metadata, event
RETURN id, txId, seq, metadata.txCommitTime AS committedAt, event;

// OPTIONAL Enterprise RBAC example. db.cdc.query exposes all matching database changes
// rather than being reduced to the caller's ordinary entity-level graph visibility.
CYPHER 25
GRANT ACCESS ON DATABASE neo4j TO atlasmart_cdc_reader;
GRANT EXECUTE PROCEDURE db.cdc.query ON DBMS TO atlasmart_cdc_reader;
GRANT EXECUTE BOOSTED PROCEDURE db.cdc.query ON DBMS TO atlasmart_cdc_reader;

#!/usr/bin/env python3
"""AtlasMart Chapter 20 deterministic CDC simulator — Python standard library only."""
import argparse, json, os, sys
from pathlib import Path

EVENTS = [
 {"epoch":"atlasmart-source-v1","ordinal":0,"id":"sim-2000","txId":500,"seq":0,"entity":"Product","key":"P-2001","op":"c","after":{"name":"Trail Camera Pro","price":189.0,"schemaVersion":1}},
 {"epoch":"atlasmart-source-v1","ordinal":1,"id":"sim-2001","txId":500,"seq":1,"entity":"Product","key":"P-2002","op":"c","after":{"name":"Field Battery","price":49.0,"schemaVersion":1}},
 {"epoch":"atlasmart-source-v1","ordinal":2,"id":"sim-2002","txId":502,"seq":0,"entity":"Customer","key":"C-2001","op":"c","after":{"name":"Mina Rahimi","tier":"GOLD","schemaVersion":1}},
 {"epoch":"atlasmart-source-v1","ordinal":3,"id":"sim-2003","txId":503,"seq":0,"entity":"Order","key":"O-2001","op":"c","after":{"customerId":"C-2001","status":"PAID","schemaVersion":1}},
 {"epoch":"atlasmart-source-v1","ordinal":4,"id":"sim-2004","txId":505,"seq":0,"entity":"Product","key":"P-2001","op":"u","after":{"name":"Trail Camera Pro","price":179.0,"schemaVersion":2,"currency":"USD"}},
 {"epoch":"atlasmart-source-v1","ordinal":5,"id":"sim-2005","txId":506,"seq":0,"entity":"Product","key":"P-2002","op":"d","after":None},
]
SNAPSHOT = {
 "Product":{"P-2001":{"name":"Trail Camera Pro","price":179.0,"schemaVersion":2,"currency":"USD"}},
 "Customer":{"C-2001":{"name":"Mina Rahimi","tier":"GOLD","schemaVersion":1}},
 "Order":{"O-2001":{"customerId":"C-2001","status":"PAID","schemaVersion":1}},
}
STATE = Path('.atlasmart_ch20_state.json')
CHECKPOINT = Path('.atlasmart_ch20_checkpoint.json')

def load(path, default):
    return json.loads(path.read_text()) if path.exists() else default

def save(path, obj):
    tmp = path.with_suffix(path.suffix + '.tmp')
    tmp.write_text(json.dumps(obj, indent=2, sort_keys=True))
    os.replace(tmp, path)

def reset():
    for p in (STATE, CHECKPOINT):
        if p.exists(): p.unlink()
    print('RESET state=empty checkpoint=none')

def durable_apply(state, ev):
    # Receipt and business-state mutation are persisted together in one state document.
    if ev['id'] in state['receipts']:
        return 'duplicate'
    bucket = state['entities'].setdefault(ev['entity'], {})
    if ev['op'] == 'd': bucket.pop(ev['key'], None)
    else: bucket[ev['key']] = ev['after']
    state['receipts'].append(ev['id'])
    save(STATE, state)
    return 'applied'

def consume(drop=None, duplicate=None, crash_after=None):
    state = load(STATE, {'receipts':[], 'entities':{}})
    cp = load(CHECKPOINT, {'epoch':'atlasmart-source-v1','ordinal':-1,'id':None})
    if cp['epoch'] != 'atlasmart-source-v1':
        print('GAP epoch-mismatch -> BACKFILL_REQUIRED'); return 3
    stream = [e for e in EVENTS if e['ordinal'] > cp['ordinal'] and e['id'] != drop]
    if duplicate:
        match = next((e for e in stream if e['id'] == duplicate), None)
        if match: stream.insert(stream.index(match)+1, dict(match))
    expected = cp['ordinal'] + 1
    for ev in stream:
        # An at-least-once source may replay an event that the target already applied.
        # A duplicate older than the next expected ordinal is harmless; a replay at the
        # expected ordinal advances the checkpoint after we verify the receipt exists.
        if ev['id'] in state['receipts']:
            if ev['ordinal'] > expected:
                print(f'GAP expectedOrdinal={expected} got={ev["ordinal"]} -> BACKFILL_REQUIRED')
                return 4
            print(f'DUPLICATE id={ev["id"]} txId={ev["txId"]} seq={ev["seq"]} key={ev["key"]}')
            if ev['ordinal'] == expected:
                cp = {'epoch':ev['epoch'], 'ordinal':ev['ordinal'], 'id':ev['id']}
                save(CHECKPOINT, cp)
                expected = ev['ordinal'] + 1
            continue
        if ev['ordinal'] != expected:
            print(f'GAP expectedOrdinal={expected} got={ev["ordinal"]} -> BACKFILL_REQUIRED')
            return 4
        result = durable_apply(state, ev)
        print(f'{result.upper()} id={ev["id"]} txId={ev["txId"]} seq={ev["seq"]} key={ev["key"]}')
        if crash_after == ev['id']:
            print('CRASH after durable target apply, before checkpoint advance')
            return 9
        # Offset/checkpoint is advanced only after downstream state is durable.
        cp = {'epoch':ev['epoch'], 'ordinal':ev['ordinal'], 'id':ev['id']}
        save(CHECKPOINT, cp)
        expected = ev['ordinal'] + 1
    print(f'CHECKPOINT ordinal={cp["ordinal"]} id={cp["id"]}')
    return 0

def backfill():
    state = {'receipts':['BACKFILL@sim-current'], 'entities':SNAPSHOT}
    save(STATE, state)
    latest = EVENTS[-1]
    save(CHECKPOINT, {'epoch':latest['epoch'],'ordinal':latest['ordinal'],'id':latest['id']})
    print('BACKFILL entities=3 products=1 customers=1 orders=1')
    print(f'CHECKPOINT ordinal={latest["ordinal"]} id={latest["id"]}')

def show():
    print(json.dumps({'checkpoint':load(CHECKPOINT,None),'state':load(STATE,None)},indent=2,sort_keys=True))

p=argparse.ArgumentParser()
sub=p.add_subparsers(dest='cmd',required=True)
sub.add_parser('reset'); sub.add_parser('backfill'); sub.add_parser('show')
c=sub.add_parser('consume'); c.add_argument('--drop'); c.add_argument('--duplicate'); c.add_argument('--crash-after')
a=p.parse_args()
if a.cmd=='reset': reset()
elif a.cmd=='backfill': backfill()
elif a.cmd=='show': show()
elif a.cmd=='consume': sys.exit(consume(a.drop,a.duplicate,a.crash_after))

# 1) Normal at-least-once consumer run
$ python cdc_sim.py reset
RESET state=empty checkpoint=none
$ python cdc_sim.py consume
APPLIED id=sim-2000 txId=500 seq=0 key=P-2001
APPLIED id=sim-2001 txId=500 seq=1 key=P-2002
APPLIED id=sim-2002 txId=502 seq=0 key=C-2001
APPLIED id=sim-2003 txId=503 seq=0 key=O-2001
APPLIED id=sim-2004 txId=505 seq=0 key=P-2001
APPLIED id=sim-2005 txId=506 seq=0 key=P-2002
CHECKPOINT ordinal=5 id=sim-2005

# Notice txId jumps 500 -> 502 and 503 -> 505. That is NOT our gap detector.
# The simulator's ordinal is synthetic teaching metadata; real CDC uses opaque cursor validity/retention evidence.