feat: publish & subscribe arbitrary custom tracks within a broadcast

doc/concept/layer/hang.md

@@ -83,6 +83,17 @@ The catalog is a JSON document published through the merge-patch helper (`@moq/j
 
 This keeps application-specific sections in the application layer while the base catalog stays generic.
 
+### Custom tracks
+
+A custom catalog section can carry its payload inline (for low-rate metadata), or it can reference a separate track in the same broadcast (for a stream of data, e.g. a `meta.json` track or an SCTE-35 event track). The relay treats such a track like any other; only the publisher and consumer give it meaning.
+
+The `@moq/publish` and `@moq/watch` components publish and subscribe to these tracks generically, with no per-application support. Each exposes a low-level track hook and re-exports `@moq/json` so the application encodes the payload itself:
+
+- **Publish**: `broadcast.publishTrack(name, serve)` runs `serve(track, effect)` per subscriber. For JSON, serve each track from a shared track-less `Json.Producer` (the same fan-out producer the catalog uses, seeding late joiners with the latest value). Advertise the track by writing your own catalog section with `broadcast.catalog.mutate(...)`.
+- **Watch**: `broadcast.subscribeTrack(name, priority, consume)` follows the active broadcast across reconnects. For JSON, wrap the track in a `Json.Consumer` inside `consume`. Read your section back from `broadcast.catalog` (unknown sections pass through the loose schema).
+
+So an application supports something like SCTE-35 entirely in its own code: publish an `scte35` section (and optionally a track) on one side, read it on the other, without hang, `@moq/publish`, or `@moq/watch` knowing anything about SCTE-35.
+
 ## Container
 
 The catalog also contains a `container` field for each rendition used to denote the encoding of each track.

doc/lib/js/@moq/publish.md

@@ -127,6 +127,34 @@ broadcast.video.device.set("screen");
 broadcast.name.set("bob.hang");
 ```
 
+### Custom tracks and catalog sections
+
+Beyond audio and video, you can publish arbitrary application tracks within the
+same broadcast (no separate broadcast needed). `publishTrack(name, serve)` runs
+`serve(track, effect)` for each subscriber; it rejects the built-in track names
+(catalog/audio/video). Encode the payload yourself with the re-exported
+`@moq/json`: a track-less `Json.Producer` is the same fan-out producer the catalog
+uses, seeding late joiners with the latest value.
+
+`publishTrack` does not touch the catalog; advertise the track by writing your own
+section to `broadcast.catalog` (the [catalog root](/concept/layer/hang#extensions)
+is a loose object, so any key passes through). This lets an app support something
+like an `scte35` section with no hang-specific support:
+
+```typescript
+import { Json } from "@moq/publish";
+
+const scte35 = new Json.Producer<{ splices: number[] }>({ initial: { splices: [] } });
+broadcast.publishTrack("scte35.json", (track, effect) => scte35.serve(track, effect));
+broadcast.catalog.mutate((c) => {
+    c.scte35 = { track: "scte35.json" };
+});
+scte35.update({ splices: [42] });
+```
+
+The component exposes everything via its `broadcast` property
+(`el.broadcast.publishTrack(...)`).
+
 ## Related Packages
 
 - **[@moq/watch](/lib/js/@moq/watch)** — Subscribe to and render MoQ broadcasts

doc/lib/js/@moq/watch.md

@@ -139,6 +139,39 @@ el.catalog = myCatalog;
 > `"msf"`) tears down the previous fetch loop, which clears `catalog`. Set the
 > catalog *after* switching to `"manual"`, not before.
 
+### Custom tracks and catalog sections
+
+A broadcast can carry arbitrary application tracks (for example a `meta.json`
+metadata track) alongside the media. An application advertises them in its own
+catalog section (the [catalog root](/concept/layer/hang#extensions) is a loose
+object, so unknown sections pass through to `broadcast.catalog`).
+`subscribeTrack(name, priority, consume)` follows the active broadcast across
+reconnects and runs `consume(track, effect)` each time it becomes active. Decode
+the payload yourself with the re-exported `@moq/json`:
+
+```typescript
+import { Json } from "@moq/watch";
+import { Signals } from "@moq/watch";
+
+// The app's own catalog section, read back from the loose catalog.
+const section = broadcast.catalog.peek()?.scte35;
+
+const scte35 = new Signals.Signal<{ splices: number[] } | undefined>(undefined);
+broadcast.subscribeTrack("scte35.json", 100, (track, effect) => {
+    const consumer = new Json.Consumer<{ splices: number[] }>(track);
+    effect.spawn(async () => {
+        for (;;) {
+            const next = await Promise.race([effect.cancel, consumer.next()]);
+            if (next === undefined) break;
+            scte35.set(next);
+        }
+    });
+});
+```
+
+The component exposes everything via its `broadcast` property
+(`el.broadcast.subscribeTrack(...)`).
+
 ## UI Overlay
 
 Import `@moq/watch/ui` for a Web Component overlay with buffering indicator, stats panel, and playback controls:

js/json/src/index.ts

@@ -1,3 +1,11 @@
+/**
+ * Snapshot/delta JSON publishing over MoQ tracks using RFC 7396 JSON Merge Patch. A
+ * {@link Producer} writes a JSON value to one track or fans it out to many; a {@link Consumer}
+ * reconstructs the value on the other side.
+ *
+ * @module
+ */
+
 export { Consumer } from "./consumer.ts";
 export { type Diff, deepEqual, diff, merge } from "./diff.ts";
 export { type Config, Producer } from "./producer.ts";

js/json/src/producer.test.ts

@@ -0,0 +1,67 @@
+import { expect, test } from "bun:test";
+import { Track } from "@moq/net";
+import { Effect } from "@moq/signals";
+import { Consumer } from "./consumer.ts";
+import { Producer } from "./producer.ts";
+
+test("a track-less producer seeds subscribers and fans out edits", async () => {
+	const source = new Producer<Record<string, unknown>>({ initial: {} });
+
+	// Edit before anyone subscribes: the value is retained, not lost.
+	source.mutate((v) => {
+		v.video = { renditions: {} };
+	});
+
+	const effect = new Effect();
+	const track = new Track("catalog.json");
+	source.serve(track, effect);
+	const consumer = new Consumer<Record<string, unknown>>(track);
+
+	// A new subscriber is seeded with the current value.
+	expect((await consumer.next())?.video).toEqual({ renditions: {} });
+
+	// An independent owner edits its own key; the subscriber sees it, the other key untouched.
+	source.mutate((v) => {
+		v.scte35 = { splices: [] };
+	});
+	const update = await consumer.next();
+	expect(update?.video).toEqual({ renditions: {} });
+	expect(update?.scte35).toEqual({ splices: [] });
+
+	effect.close();
+});
+
+test("a reconnecting subscriber is seeded with the full current value", async () => {
+	const source = new Producer<Record<string, unknown>>({ initial: {} });
+	source.mutate((v) => {
+		v.video = { renditions: {} };
+		v.scte35 = { splices: [] };
+	});
+
+	// The first subscription drains and ends...
+	const first = new Effect();
+	source.serve(new Track("catalog.json"), first);
+	first.close();
+
+	// ...and a fresh subscription still gets the current value, not nothing.
+	const effect = new Effect();
+	const track = new Track("catalog.json");
+	source.serve(track, effect);
+	const seeded = await new Consumer<Record<string, unknown>>(track).next();
+	expect(seeded?.video).toEqual({ renditions: {} });
+	expect(seeded?.scte35).toEqual({ splices: [] });
+
+	effect.close();
+});
+
+test("mutate without a value or initial throws", () => {
+	const source = new Producer<Record<string, unknown>>();
+	expect(() => source.mutate(() => {})).toThrow();
+});
+
+test("serve() throws on a track-bound producer", () => {
+	const producer = new Producer<Record<string, unknown>>(new Track("meta.json"));
+	const effect = new Effect();
+	expect(() => producer.serve(new Track("other"), effect)).toThrow();
+	effect.close();
+});

js/json/src/producer.ts

@@ -1,4 +1,5 @@
-import type * as Moq from "@moq/net";
+import * as Moq from "@moq/net";
+import type { Effect } from "@moq/signals";
 import type * as z from "zod/mini";
 
 import { deepEqual, diff } from "./diff.ts";
@@ -25,23 +26,72 @@ export interface Config<T> {
 	initial?: T;
 }
 
-/** Publishes a JSON value over a track, choosing snapshots and deltas automatically. */
+/**
+ * Publishes a JSON value as snapshots and deltas, chosen automatically.
+ *
+ * Construct it two ways:
+ *
+ * - **With a track** (`new Producer(track, config)`): writes directly to that one track.
+ * - **Without a track** (`new Producer(config)`): retains the value and fans it out to any number of
+ *   subscription tracks attached with {@link serve}, seeding late joiners with the current value.
+ *   This backs the hang catalog and is how an application publishes its own custom tracks.
+ */
 export class Producer<T> {
-	#track: Moq.Track;
 	#config: Config<T>;
 
+	// Leaf mode: writes snapshots/deltas straight to a single track.
+	#track?: Moq.Track;
 	#group?: Moq.Group;
 	#last?: unknown;
 	#groupBytes = 0;
 	#groupFrames = 0;
 
-	constructor(track: Moq.Track, config: Config<T> = {}) {
-		this.#track = track;
-		this.#config = config;
+	// Fan-out mode: retains the value and serves a child (leaf) Producer per subscriber.
+	#outputs?: Set<Producer<T>>;
+	#value?: T;
+
+	/** Create a track-less, fan-out producer; attach subscribers with {@link serve}. */
+	constructor(config?: Config<T>);
+	/** Create a producer that writes directly to `track`. */
+	constructor(track: Moq.Track, config?: Config<T>);
+	constructor(trackOrConfig?: Moq.Track | Config<T>, config: Config<T> = {}) {
+		if (trackOrConfig instanceof Moq.Track) {
+			this.#track = trackOrConfig;
+			this.#config = config;
+		} else {
+			this.#config = trackOrConfig ?? {};
+			this.#outputs = new Set();
+			this.#value = this.#config.initial;
+		}
+	}
+
+	/** The current value, or `undefined` if nothing has been published yet. */
+	get value(): T | undefined {
+		return this.#track ? (this.#last as T | undefined) : this.#value;
 	}
 
 	/** Publish a new value, emitting a snapshot or delta automatically. No-op if unchanged. */
 	update(value: T): void {
+		if (!this.#track) {
+			// Fan-out: retain the value and forward it to every subscriber. Isolate per-subscriber
+			// failures so one broken track (e.g. closed mid-update) doesn't stop the others.
+			this.#value = value;
+			for (const output of this.#outputs ?? []) {
+				try {
+					output.update(value);
+				} catch (err) {
+					this.#outputs?.delete(output);
+					try {
+						output.finish();
+					} catch {
+						// Already broken; nothing more to do.
+					}
+					console.warn("dropping failed json subscriber during fan-out", err);
+				}
+			}
+			return;
+		}
+
 		const valid = this.#config.schema ? this.#config.schema.parse(value) : value;
 
 		// Serialize once; parse it back to a normalized JSON value for diffing and comparison
@@ -57,7 +107,7 @@ export class Producer<T> {
 			this.#groupBytes += delta.length;
 			this.#groupFrames += 1;
 		} else {
-			this.#snapshot(snapshot);
+			this.#snapshot(this.#track, snapshot);
 		}
 
 		this.#last = json;
@@ -83,7 +133,7 @@ export class Producer<T> {
 	mutate(fn: (value: T) => void): void {
 		// Start from the last-published value, falling back to the configured initial value. We
 		// don't invent an empty object: mutating with nothing to start from is a usage error.
-		const base = this.#last ?? this.#config.initial;
+		const base = (this.#track ? this.#last : this.#value) ?? this.#config.initial;
 		if (base === undefined) {
 			throw new Error("mutate() requires a prior update() or `initial` in the config");
 		}
@@ -93,8 +143,35 @@ export class Producer<T> {
 		this.update(value);
 	}
 
-	/** Finish the track, closing any open group. */
+	/**
+	 * Serve a subscription request: seed the track with the current value, then forward updates.
+	 *
+	 * Only available on a track-less (fan-out) producer. The subscriber is removed and finished when
+	 * `effect` is cleaned up.
+	 */
+	serve(track: Moq.Track, effect: Effect): void {
+		if (!this.#outputs) {
+			throw new Error("serve() is only available on a track-less Producer");
+		}
+
+		const output = new Producer<T>(track, this.#config);
+		if (this.#value !== undefined) output.update(this.#value);
+
+		this.#outputs.add(output);
+		effect.cleanup(() => {
+			this.#outputs?.delete(output);
+			output.finish();
+		});
+	}
+
+	/** Finish: close the track (leaf) or finish every subscriber (fan-out). */
 	finish(): void {
+		if (!this.#track) {
+			for (const output of this.#outputs ?? []) output.finish();
+			this.#outputs?.clear();
+			return;
+		}
+
 		this.#group?.close();
 		this.#group = undefined;
 		this.#track.close();
@@ -117,11 +194,11 @@ export class Producer<T> {
 		return delta;
 	}
 
-	#snapshot(snapshot: Uint8Array): void {
+	#snapshot(track: Moq.Track, snapshot: Uint8Array): void {
 		// The previous group is complete; no more frames will be appended to it.
 		this.#group?.close();
 
-		const group = this.#track.appendGroup();
+		const group = track.appendGroup();
 		group.writeFrame(snapshot);
 		this.#groupBytes = snapshot.length;
 		this.#groupFrames = 1;

js/publish/package.json

@@ -21,7 +21,7 @@
 	"scripts": {
 		"build": "rimraf dist && vite build && tsc && bun ../common/package.ts",
 		"check": "tsc --noEmit",
-		"test": "bun test",
+		"test": "bun test --pass-with-no-tests",
 		"release": "bun ../common/release.ts"
 	},
 	"dependencies": {