From 695722c480187d54e9d24345b5d4ac304e26b696 Mon Sep 17 00:00:00 2001 From: George Lambert Date: Fri, 11 Sep 2026 15:17:21 -0400 Subject: [PATCH] Initial import of verae-archive-aggregator from zapier monorepo --- NATS.md | 8 +++++ README.md | 29 +++++++++++++++++++ SUMMARY.md | 11 +++++++ package.json | 8 +++++ src/aggregate.js | 66 ++++++++++++++++++++++++++++++++++++++++++ test/aggregate.test.js | 44 ++++++++++++++++++++++++++++ 6 files changed, 166 insertions(+) create mode 100644 NATS.md create mode 100644 README.md create mode 100644 SUMMARY.md create mode 100644 package.json create mode 100644 src/aggregate.js create mode 100644 test/aggregate.test.js diff --git a/NATS.md b/NATS.md new file mode 100644 index 0000000..6a73642 --- /dev/null +++ b/NATS.md @@ -0,0 +1,8 @@ +# NATS — verae-archive-aggregator + +| Direction | Address | Peer | Body | +|-----------|---------|------|------| +| OUT | `verae.archive.query` | all WORM | `{ correlationId, sha256, tenantId, kinds[], traceId }` | +| IN | `verae.archive.reply.` | WORM hits | `{ archiveId, sha256, records[] }` | + +Stop collecting at `WAIT_ARCHIVE_MS`. Silent archives are listed in `silent[]`, not errors. diff --git a/README.md b/README.md new file mode 100644 index 0000000..b44e251 --- /dev/null +++ b/README.md @@ -0,0 +1,29 @@ +# verae-archive-aggregator + +When a wait path sets `includeAttached`, this process broadcasts `verae.archive.query` and **merges** replies into the job JSON. Silent bloom misses omit slices; the chain receipt still succeeds. + +**Forgejo:** https://git.georgelambert.org/marchon/verae-archive-aggregator +**Catalog:** https://zapier.georgelambert.org/packages/verae-archive-aggregator/README.pdf +**NATS:** https://zapier.georgelambert.org/docs-master/modules/verae-archive-aggregator/NATS.pdf + +## Run + +```bash +cd packages/verae-archive-aggregator +npm test +``` + +## Addresses + +| Direction | Subject | +|-----------|---------| +| out | `verae.archive.query` `{ correlationId, sha256, tenantId, kinds[] }` | +| in | `verae.archive.reply.` (hits only) | + +## Depends on + +- middleware wait path (`includeAttached` / `includeTree`) +- one or more archive-worm and/or tree-node replicas +- NATS (private) + +Timeout: missing archives omit their records; do not fail the seal. diff --git a/SUMMARY.md b/SUMMARY.md new file mode 100644 index 0000000..d6951ab --- /dev/null +++ b/SUMMARY.md @@ -0,0 +1,11 @@ +# verae-archive-aggregator + +**Job:** If `includeAttached`, fan out archive query and merge replies into the job JSON. + +**Expects:** wait-path flag from middleware-http; replies on `verae.archive.reply.`. + +**Sends:** `verae.archive.query` to all WORM nodes. + +**Timeout:** missing archives omit slices; chain receipt still succeeds. + +**Test:** `npm test` — one silent bloom miss + one hit; merge seal + metadata-attach. diff --git a/package.json b/package.json new file mode 100644 index 0000000..56b82c5 --- /dev/null +++ b/package.json @@ -0,0 +1,8 @@ +{ + "name": "verae-archive-aggregator", + "version": "0.1.0", + "type": "module", + "description": "Aggregate bloom-filtered WORM archive replies into job JSON", + "scripts": { "test": "node --test test/*.test.js" }, + "engines": { "node": ">=20" } +} diff --git a/src/aggregate.js b/src/aggregate.js new file mode 100644 index 0000000..cc6c6d0 --- /dev/null +++ b/src/aggregate.js @@ -0,0 +1,66 @@ +/** + * Fan-out archive query and merge replies until timeout. + * @module aggregate + */ + +/** + * @param {object} opts + * @param {string} opts.sha256 + * @param {string} opts.tenantId + * @param {string[]} [opts.kinds] + * @param {Array<{ archiveId: string, query: (sha: string) => object[]|null }>} opts.archives + * @param {number} [opts.timeoutMs=50] unused in in-process mock (archives are sync) + * @returns {{ archivesQueried: boolean, archiveReplies: number, records: object[], silent: string[] }} + */ +export function aggregateAttached({ sha256, tenantId, kinds = ['publicMeta', 'privateMeta', 'file'], archives }) { + const records = []; + const silent = []; + for (const arch of archives) { + const hit = arch.query(sha256); + if (hit == null) { + silent.push(arch.archiveId); + continue; + } + for (const rec of hit) { + if (!kinds.length || kinds.includes(rec.kind)) records.push(rec); + } + } + return { + sha256, + tenantId, + archivesQueried: true, + archiveReplies: archives.length - silent.length, + silent, + records, + }; +} + +/** + * Merge chain receipts with archive records. + * @param {object[]} chainReceipts + * @param {object[]} archiveRecords + */ +export function mergeReceipts(chainReceipts, archiveRecords) { + const extra = archiveRecords.map((r) => { + if (r.kind === 'tree') { + return { + kind: 'tree-leaf', + archiveId: r.archiveId, + attachedAt: r.storedAt, + itemizedOnMainChain: false, + merkleRoot: r.record?.merkleRoot, + proof: r.record?.proof, + leafIndex: r.record?.leafIndex, + chainSealJobId: r.record?.chainSealJobId, + ...r.record, + }; + } + return { + kind: r.kind === 'file' ? 'file-attach' : 'metadata-attach', + archiveId: r.archiveId, + attachedAt: r.storedAt, + ...r.record, + }; + }); + return [...chainReceipts, ...extra]; +} diff --git a/test/aggregate.test.js b/test/aggregate.test.js new file mode 100644 index 0000000..8027f6e --- /dev/null +++ b/test/aggregate.test.js @@ -0,0 +1,44 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { WormArchive } from '../../verae-archive-worm/src/archive.js'; +import { aggregateAttached, mergeReceipts } from '../src/aggregate.js'; + +describe('aggregateAttached', () => { + it('silent miss + one hit', () => { + const a = new WormArchive('archive-a'); + const b = new WormArchive('archive-b'); + const sha = 'c'.repeat(64); + b.put({ sha256: sha, kind: 'publicMeta', record: { publicMetadata: { k: 1 } } }); + const out = aggregateAttached({ sha256: sha, tenantId: 't1', archives: [a, b] }); + assert.equal(out.archiveReplies, 1); + assert.deepEqual(out.silent, ['archive-a']); + assert.equal(out.records[0].record.publicMetadata.k, 1); + }); + + it('merges seal + attach receipts', () => { + const seal = [{ kind: 'seal', timestamp: '2026-01-01T00:00:00Z', certificate: 'x' }]; + const recs = [ + { kind: 'publicMeta', archiveId: 'archive-b', storedAt: '2026-01-02T00:00:00Z', record: { publicMetadata: { n: 2 } } }, + ]; + const merged = mergeReceipts(seal, recs); + assert.equal(merged.length, 2); + assert.equal(merged[0].kind, 'seal'); + assert.equal(merged[1].kind, 'metadata-attach'); + }); + + it('merges tree-leaf receipts from bulk summaries', () => { + const seal = [{ kind: 'seal', of: 'merkleRoot', sha256: 'aa'.repeat(32) }]; + const recs = [ + { + kind: 'tree', + archiveId: 'tree-node-east', + storedAt: '2026-01-02T00:00:00Z', + record: { merkleRoot: 'aa'.repeat(32), leafIndex: 1, proof: [] }, + }, + ]; + const merged = mergeReceipts(seal, recs); + assert.equal(merged[1].kind, 'tree-leaf'); + assert.equal(merged[1].itemizedOnMainChain, false); + assert.equal(merged[1].archiveId, 'tree-node-east'); + }); +});