const trades = $({ t3: { sym: 'AAPL', qty: 500 } }), got = []
trades.sink({ apply: b => got.push(b) })
trades.t3.qty.update(380)  // a bare write = a synchronous batch of one
print(`seq ${got[0].seq}: ${show(got[0].rows[0])} — delivered before update() returned`)
// →
trades.t3.qty.update(380)  // Object.is-equal at the leaf
print(`nothing emitted: still ${got.length} batch, seq still ${runtime().seq}`)
// →

const rows = $([{ sym: 'AAPL', qty: 200 }, { sym: 'NVDA', qty: 90 }]); let fired = 0
rows.sink({ apply: b => print(`at batch close: the sink fires once (seq ${b.seq}) · fired ${++fired}`) })
batch(() => {
  rows.insert({ sym: 'MSFT', qty: 40 })
  print(`mid-batch: rowCount() → ${rows.rowCount()}, the read sees the write · sinks fired: ${fired}`)
})
// →
// →

const rows = $([{ qty: 50 }, { qty: 120 }, { qty: 300 }])
const big = rows.filter(r => r.qty > 100), n = big.length()  // height 1, height 2
const name = { [rows[node].id]: 'source', [big[node].id]: 'filter', [n[node].id]: 'length' }
const hook = runtime().onCommit(c => print(`seq ${c.seq} settled ${c.nodes.map(x => name[x.id]).join(' → ')}, each once`))
big.sink({ apply: () => print(`the sink fires after settle: n already reads ${n[value]}`) })
rows.get(0).set('qty', 500); hook.dispose()  // 50 → 500 joins big; n 2 → 3
// →
// →

const rows = $([{ qty: 1 }, { qty: 2 }]), was = rows.rowCount(); let sibling = 0
rows.sink({ apply() { throw new Error('boom') } })
rows.sink({ apply: () => sibling++ })
try { rows.insert({ qty: 3 }) } catch (e) { print(`${e.constructor.name} after the drain: ${e.errors.length} sink failed (${e.errors[0].message})`) }
print(`state intact: rowCount ${was} → ${rows.rowCount()}, the write committed · the sibling sink still fired ${sibling}×`)
// →
// →

const a = $({ n: 1 }), b = $({ n: 0 }), log = []
a.sink({ apply: c => { log.push(`seq ${c.seq}: a.n = ${a.n[value]}`); b.n.update(a.n[value] * 2) } })
b.sink({ apply: c => log.push(`seq ${c.seq}: b.n = ${b.n[value]}`) })
a.n.update(21)  // an effect that writes
print(`${log.join(' · ')} — the effect's write became the NEXT commit`)
// →

const mine = Symbol('peer-A'), replica = $({}), calls = { mine: 0, other: 0 }
replica.sink({ origin: mine, apply: () => calls.mine++ })  // declares what it must not hear
replica.sink({ apply: b => { calls.other++; calls.echo = b.origin === mine } })
ingest(replica, [{ t: 'add', k: 'a', v: { n: 1 } }], { origin: mine })
print(`peer-A's own sink: ${calls.mine} applies, skipped · the other sink: ${calls.other} apply, batch.origin === mine: ${calls.echo}`)
// →

const rows = $([{ qty: 1 }, { qty: 2 }]), got = {}
const sink = { init: s => (got.init = s.size), apply: b => (got.first ??= b.seq) }
batch(() => { rows.insert({ qty: 3 }); rows.sink(sink); rows.insert({ qty: 4 }) })  // attached mid-batch
print(`init saw ${got.init} rows — the settled batch (seq ${runtime().seq}), both inserts · applies so far: ${got.first ?? 'none'}`)
rows.get(0).set('qty', 9)
print(`apply first fired at seq ${got.first} — the NEXT commit`)
// →
// →

const rows = $([{ qty: 1 }]), t = $({ t: { qty: 500 } }), got = []
rows.sink({ apply: b => got.push(b) }); t.sink({ apply: b => got.push(b) })
batch(() => { const k = rows.insert({ qty: 2 }); rows.get(k).remove() })  // add + remove, one batch
print(`add + remove annihilated: ${got.length} batches delivered, no delta`)
batch(() => { t.t.qty.update(1); t.t.qty.update(2); t.t.qty.update(3) })  // three writes, one key
print(`${got.length} batch, ${got[0].rows.length} delta: ${show(got[0].rows[0])} — first prev, last value`)
// →
// →

const q = () => [[{ n: 1 }], [{ n: 2 }]], commits = { sync: 0, co: 0 }  // two chunks that settle in one tick
const feed = (c = q()) => ({ [Symbol.asyncIterator]: () => ({ next: () => Promise.resolve({ done: !c.length, value: c.shift() }) }) })
const sync = fromAsync(runtime(), feed()), co = fromAsync(runtime(), feed(), { coalesce: 'microtask' })
handleFor(sync.source).sink({ apply: () => commits.sync++ }); handleFor(co.source).sink({ apply: () => commits.co++ })
while (sync.status() === 'pending' || co.status() === 'pending') await new Promise(r => setTimeout(r))
print(`default: ${q().length} chunks → ${commits.sync} commits · coalesce('microtask'): ${commits.co} commit · same ${co.source.rowCount()} rows each`)
// →

const t = $({ t3: { meta: { tag: 'old', book: null } } }), got = []; t.sink({ apply: b => got.push(...b.rows) })
t.t3.meta.tag.remove()  // nested remove: first-class (pathDelete)
t.t9.meta.tag.remove()  // absent target: idempotent no-op
t.t3.meta.book.depth.update(3)  // null intermediate: vivified (pathCopy)
print(`${show(got[0])} · deleted: ${got[0].deleted} · row.meta = ${JSON.stringify(got[0].row.meta)}`)
print(`${got.length} deltas for three writes — the absent-target remove emitted nothing · ${show(got[1])}, vivified from null`)
// →
// →

const src = $({})
ingest(src, [{ t: 'add', k: '\x00key', v: { px: NaN, at: undefined } }])
const row = src.get('\x00key')[value]
print(`${src.rowCount()} row · key ${JSON.stringify('\x00key')} live: ${row !== undefined} — a NUL byte is just a key`)
print(`px is NaN: ${Number.isNaN(row.px)} · row owns 'at': ${'at' in row}, at === undefined: ${row.at === undefined} — both first-class`)
// →
// →