Initial import of verae-archive-aggregator from zapier monorepo
This commit is contained in:
commit
969151d2b4
6 changed files with 166 additions and 0 deletions
8
NATS.md
Normal file
8
NATS.md
Normal file
|
|
@ -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.
|
||||
29
README.md
Normal file
29
README.md
Normal file
|
|
@ -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.<correlationId>` (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.
|
||||
11
SUMMARY.md
Normal file
11
SUMMARY.md
Normal file
|
|
@ -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.
|
||||
8
package.json
Normal file
8
package.json
Normal file
|
|
@ -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" }
|
||||
}
|
||||
66
src/aggregate.js
Normal file
66
src/aggregate.js
Normal file
|
|
@ -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];
|
||||
}
|
||||
44
test/aggregate.test.js
Normal file
44
test/aggregate.test.js
Normal file
|
|
@ -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');
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue