Initial import of verae-archive-aggregator from zapier monorepo

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.<correlationId>` | WORM hits | `{ archiveId, sha256, records[] }` |
+
+Stop collecting at `WAIT_ARCHIVE_MS`. Silent archives are listed in `silent[]`, not errors.

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.<correlationId>`.
+
+**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.

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" }
+}

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];
+}

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');
+  });
+});