From 0df723e4c19d0daf203c3f05fb9556880483c99b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 15 Jul 2026 01:37:58 +0100 Subject: [PATCH] Stream typed provider state to Eve clients --- src/hermodr-daemon.cjs | 53 +++++++++++++ src/hermodr-state-stream.cjs | 116 +++++++++++++++++++++++++++++ test/hermodr-catalog.test.cjs | 17 +++++ test/hermodr-state-stream.test.cjs | 72 ++++++++++++++++++ 4 files changed, 258 insertions(+) create mode 100644 src/hermodr-state-stream.cjs create mode 100644 test/hermodr-state-stream.test.cjs diff --git a/src/hermodr-daemon.cjs b/src/hermodr-daemon.cjs index d4563c5..602f017 100644 --- a/src/hermodr-daemon.cjs +++ b/src/hermodr-daemon.cjs @@ -9,6 +9,7 @@ const path = require("path"); const { defineOdinDocuments } = require("./odin/documents.cjs"); const { createIdunnRudpHealthPublisher, publishIdunnRudpHealth } = require("./odin/idunn-rudp.cjs"); const { parseArgs } = require("./odin/utils.cjs"); +const { HermodrStateStreamRegistry } = require("./hermodr-state-stream.cjs"); const repoRoot = path.resolve(__dirname, ".."); const projectsRoot = path.resolve(repoRoot, ".."); @@ -191,6 +192,10 @@ function rejectRemovedOptions(options) { } function createHermodrBridge(options) { + const stateStreams = new HermodrStateStreamRegistry({ + pollIntervalMs: options.statePollIntervalMs ?? 1_000, + staleAfterMs: options.stateStaleAfterMs ?? 10_000, + }); return { async handle(request, response) { const url = new URL(request.url || "/", `http://${request.headers.host || "localhost"}`); @@ -264,6 +269,24 @@ function createHermodrBridge(options) { return; } + if (request.method === "GET" && url.pathname.startsWith("/hermodr/state/")) { + const source = normalizeStateBindingSource({ + providerId: decodeURIComponent(url.pathname.slice("/hermodr/state/".length)), + sourceId: stringOption(url.searchParams.get("sourceId"), ""), + schemaId: stringOption(url.searchParams.get("schemaId"), ""), + }); + if (!source) { + writeJson(response, 400, { ok: false, error: "providerId, sourceId, and schemaId are required for a typed state stream." }); + return; + } + openStateEventStream(request, response, stateStreams, source, async () => { + const resolved = await readVisibleDocument(options, source); + if (!resolved) throw new Error(`Provider '${source.providerId}' did not expose ${source.schemaId} at ${source.documentId}.`); + return resolved.document; + }); + return; + } + if (request.method === "POST" && url.pathname === "/hermodr/commands/sleipnir/input-mapping") { const body = await readJsonBody(request); writeJson(response, 202, await publishSleipnirMapping(options, body)); @@ -303,6 +326,34 @@ async function readCatalog(options) { return createBrowserCatalog(providerAdvertisements, options); } +function normalizeStateBindingSource(value) { + const providerId = String(value?.providerId || "").trim(); + const sourceId = String(value?.sourceId || "").trim(); + const schemaId = String(value?.schemaId || "").trim(); + if (!providerId || !sourceId || !schemaId) return null; + const schemaName = schemaId.replace(/\.v\d+$/, ""); + const prefix = `${schemaName}:`; + const documentId = sourceId.startsWith(prefix) ? sourceId.slice(prefix.length) : sourceId; + if (!documentId) return null; + return { providerId, sourceId, schemaId, documentId }; +} + +function openStateEventStream(request, response, registry, source, read) { + writeCors(response, 200); + response.setHeader("Content-Type", "text/event-stream; charset=utf-8"); + response.setHeader("Cache-Control", "no-cache, no-transform"); + response.setHeader("Connection", "keep-alive"); + response.flushHeaders?.(); + const release = registry.acquire(source, read, event => { + response.write(`id: ${event.sequence}\n`); + response.write(`event: ${event.type}\n`); + response.write(`data: ${JSON.stringify(event)}\n\n`); + }); + const close = () => release(); + request.once("close", close); + response.once("close", close); +} + function createBrowserCatalog(providerAdvertisements, options = {}) { const surfaces = providerAdvertisements.flatMap((provider) => { const advertised = provider.surfaces?.length ? provider.surfaces : [{}]; @@ -1385,7 +1436,9 @@ if (require.main === module) { module.exports = { createBrowserCatalog, + createHermodrBridge, findProviderCdnRoute, + normalizeStateBindingSource, normalizeProviderAdvertisement, surfaceRecordKeys, }; diff --git a/src/hermodr-state-stream.cjs b/src/hermodr-state-stream.cjs new file mode 100644 index 0000000..9f0c0fb --- /dev/null +++ b/src/hermodr-state-stream.cjs @@ -0,0 +1,116 @@ +"use strict"; + +const { EventEmitter } = require("node:events"); + +class HermodrStateStream extends EventEmitter { + constructor(source, read, options = {}) { + super(); + this.source = Object.freeze({ ...source }); + this.read = read; + this.pollIntervalMs = options.pollIntervalMs ?? 1_000; + this.staleAfterMs = options.staleAfterMs ?? 10_000; + this.schedule = options.schedule ?? ((callback, delay) => setTimeout(callback, delay)); + this.cancel = options.cancel ?? clearTimeout; + this.now = options.now ?? Date.now; + this.sequence = 0; + this.lastValue = undefined; + this.lastFingerprint = ""; + this.lastSuccessAt = 0; + this.stale = false; + this.running = false; + this.timer = null; + } + + start() { + if (this.running) return; + this.running = true; + void this.poll(); + } + + stop() { + this.running = false; + if (this.timer !== null) this.cancel(this.timer); + this.timer = null; + } + + async poll() { + if (!this.running) return; + try { + const value = await this.read(); + const fingerprint = stableFingerprint(value); + const recovered = this.stale; + const changed = fingerprint !== this.lastFingerprint; + this.lastSuccessAt = this.now(); + this.stale = false; + if (changed || recovered) { + this.lastValue = value; + this.lastFingerprint = fingerprint; + this.publish(recovered ? "reconnected" : this.sequence === 0 ? "snapshot" : "update", value); + } + } catch (error) { + const elapsed = this.lastSuccessAt ? this.now() - this.lastSuccessAt : this.staleAfterMs; + if (!this.stale && elapsed >= this.staleAfterMs) { + this.stale = true; + this.publish("stale", undefined, error); + } + } finally { + if (this.running) this.timer = this.schedule(() => void this.poll(), this.pollIntervalMs); + } + } + + publish(type, value, error) { + const event = { + type, + sequence: ++this.sequence, + observedAt: new Date(this.now()).toISOString(), + source: this.source, + ...(value === undefined ? {} : { value }), + ...(error ? { error: error instanceof Error ? error.message : String(error) } : {}), + }; + this.emit("event", event); + } +} + +class HermodrStateStreamRegistry { + constructor(options = {}) { + this.options = options; + this.streams = new Map(); + } + + acquire(source, read, listener) { + const key = stateSourceKey(source); + let entry = this.streams.get(key); + if (!entry) { + const stream = new HermodrStateStream(source, read, this.options); + entry = { stream, subscribers: 0 }; + this.streams.set(key, entry); + } + entry.subscribers += 1; + entry.stream.on("event", listener); + entry.stream.start(); + let released = false; + return () => { + if (released) return; + released = true; + entry.stream.off("event", listener); + entry.subscribers -= 1; + if (entry.subscribers === 0) { + entry.stream.stop(); + this.streams.delete(key); + } + }; + } +} + +function stateSourceKey(source) { + return [source.providerId, source.schemaId, source.documentId].map(value => String(value || "")).join("\u001f"); +} + +function stableFingerprint(value) { + return JSON.stringify(value, (_key, item) => { + if (!item || typeof item !== "object" || Array.isArray(item)) return item; + return Object.fromEntries(Object.entries(item).sort(([left], [right]) => left.localeCompare(right))); + }); +} + +module.exports = { HermodrStateStream, HermodrStateStreamRegistry, stateSourceKey }; diff --git a/test/hermodr-catalog.test.cjs b/test/hermodr-catalog.test.cjs index b4d3d62..456ad5d 100644 --- a/test/hermodr-catalog.test.cjs +++ b/test/hermodr-catalog.test.cjs @@ -7,6 +7,7 @@ const { createBrowserCatalog, findProviderCdnRoute, normalizeProviderAdvertisement, + normalizeStateBindingSource, surfaceRecordKeys, } = require("../src/hermodr-daemon.cjs"); @@ -42,3 +43,19 @@ test("CDN routing follows the asset URI provider instead of a product name", () assert.equal(route.providerId, "fixture.game"); assert.equal(route.endpoint, "rudp://127.0.0.1:3000"); }); + +test("state binding sources select typed records without interpreting pointer paths", () => { + assert.deepEqual(normalizeStateBindingSource({ + providerId: "voidbot.swarm", + sourceId: "voidbot.swarm_state_snapshot:voidbot-swarm", + schemaId: "voidbot.swarm_state_snapshot.v1", + pointerId: "summary.globalHeat", + }), { + providerId: "voidbot.swarm", + sourceId: "voidbot.swarm_state_snapshot:voidbot-swarm", + schemaId: "voidbot.swarm_state_snapshot.v1", + documentId: "voidbot-swarm", + }); + assert.equal(normalizeStateBindingSource({ providerId: "voidbot.swarm", sourceId: "", schemaId: "schema.v1" }), null); + assert.equal(normalizeStateBindingSource({ providerId: "voidbot.swarm", sourceId: "record", schemaId: "" }), null); +}); diff --git a/test/hermodr-state-stream.test.cjs b/test/hermodr-state-stream.test.cjs new file mode 100644 index 0000000..03d221a --- /dev/null +++ b/test/hermodr-state-stream.test.cjs @@ -0,0 +1,72 @@ +"use strict"; + +const assert = require("node:assert/strict"); +const test = require("node:test"); +const { HermodrStateStream, HermodrStateStreamRegistry } = require("../src/hermodr-state-stream.cjs"); + +const source = { providerId: "fixture", schemaId: "fixture.heat.v1", documentId: "swarm" }; + +test("state stream emits ordered provider snapshots and updates without rewriting documents", async () => { + let now = 1_000; + let document = { heat: 1.2, nested: { owner: "provider" } }; + const initialDocument = document; + const stream = new HermodrStateStream(source, async () => document, { + now: () => now, + schedule: () => null, + }); + const events = []; + stream.on("event", event => events.push(event)); + stream.running = true; + + await stream.poll(); + await stream.poll(); + document = { heat: 0.15, nested: { owner: "provider" } }; + now += 1_000; + await stream.poll(); + + assert.deepEqual(events.map(event => [event.type, event.sequence]), [["snapshot", 1], ["update", 2]]); + assert.strictEqual(events[0].value, initialDocument); + assert.deepEqual(events[1].value, document); +}); + +test("state stream reports stale once and reconnected with the current provider document", async () => { + let now = 5_000; + let failure = false; + const document = { heat: 1.2 }; + const stream = new HermodrStateStream(source, async () => { + if (failure) throw new Error("provider unavailable"); + return document; + }, { now: () => now, staleAfterMs: 100, schedule: () => null }); + const events = []; + stream.on("event", event => events.push(event)); + stream.running = true; + + await stream.poll(); + failure = true; + now += 100; + await stream.poll(); + await stream.poll(); + failure = false; + now += 100; + await stream.poll(); + + assert.deepEqual(events.map(event => [event.type, event.sequence]), [ + ["snapshot", 1], ["stale", 2], ["reconnected", 3], + ]); + assert.deepEqual(events[2].value, document); +}); + +test("registry multiplexes subscribers onto one provider stream and releases it after the last client", () => { + const registry = new HermodrStateStreamRegistry({ schedule: () => null }); + const read = async () => ({ heat: 1.2 }); + const releaseFirst = registry.acquire(source, read, () => {}); + const releaseSecond = registry.acquire(source, read, () => {}); + + assert.equal(registry.streams.size, 1); + assert.equal([...registry.streams.values()][0].subscribers, 2); + releaseFirst(); + assert.equal(registry.streams.size, 1); + releaseFirst(); + releaseSecond(); + assert.equal(registry.streams.size, 0); +});