Stream typed provider state to Eve clients

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,
 };

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

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

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