packages/forkhorse-gatekeeper/src/feed.ts

1import type { ProjectState } from "../../forkhorse-dashboard/src/contracts.ts";
2import type { SidebarBatch } from "./ui-types.ts";
3
4type Receiver = { receive(batch: SidebarBatch): Promise<void>; dispose(): void };
5type Sink = { receiver: Receiver; busy: boolean; pending?: SidebarBatch };
6
7/** Host-owned state holder; never expose this object as an iframe capability. */
8export class SnapshotFeed {
9  #batch: SidebarBatch;
10  #sink?: Sink;
11  #timer?: ReturnType<typeof setTimeout>;
12  #pending?: ProjectState;
13  #wait?: { resolve(batch: SidebarBatch): void; reject(error: Error): void };
14  #waitTimer?: ReturnType<typeof setTimeout>;
15  #closed = false;
16
17  constructor(initial: ProjectState) {
18    this.#batch = { version: 0, state: structuredClone(initial) };
19  }
20
21  snapshot(): SidebarBatch {
22    this.#assertOpen();
23    return structuredClone(this.#batch);
24  }
25
26  /** Coalesce bursts at 250ms. A slow receiver retains at most one latest snapshot. */
27  publish(state: ProjectState): void {
28    this.#assertOpen();
29    if (state.project.id !== this.#batch.state.project.id)
30      throw new Error("A sidebar session is bound to one project.");
31    this.#pending = structuredClone(state);
32    this.#timer ??= setTimeout(() => {
33      this.#timer = undefined;
34      this.#batch = { version: this.#batch.version + 1, state: this.#pending! };
35      this.#pending = undefined;
36      const wait = this.#wait;
37      this.#clearWait();
38      wait?.resolve(this.snapshot());
39      if (this.#sink) {
40        this.#sink.pending = this.snapshot();
41        this.#send(this.#sink);
42      }
43    }, 250);
44  }
45
46  subscribe(receiver: Receiver): () => void {
47    this.#assertOpen();
48    this.#sink?.receiver.dispose();
49    const sink: Sink = { receiver, busy: false, pending: this.snapshot() };
50    this.#sink = sink;
51    this.#send(sink);
52    return () => {
53      if (this.#sink === sink) this.#sink = undefined;
54      sink.pending = undefined;
55      receiver.dispose();
56    };
57  }
58
59  waitForChanges(after: number): Promise<SidebarBatch> {
60    this.#assertOpen();
61    if (!Number.isSafeInteger(after) || after < 0 || after > this.#batch.version)
62      return Promise.reject(new Error("Invalid UI version."));
63    if (after < this.#batch.version) return Promise.resolve(this.snapshot());
64    if (this.#wait) return Promise.reject(new Error("Only one pending poll per UI session."));
65    return new Promise((resolve, reject) => {
66      this.#wait = { resolve, reject };
67      this.#waitTimer = setTimeout(() => {
68        this.#clearWait();
69        resolve(this.snapshot());
70      }, 25_000);
71    });
72  }
73
74  #send(sink: Sink): void {
75    if (sink.busy || !sink.pending || this.#sink !== sink) return;
76    const batch = sink.pending;
77    sink.pending = undefined;
78    sink.busy = true;
79    void Promise.resolve().then(() => sink.receiver.receive(batch)).catch(() => {
80      if (this.#sink === sink) this.#sink = undefined;
81      sink.pending = undefined;
82      sink.receiver.dispose();
83    }).finally(() => {
84      sink.busy = false;
85      this.#send(sink);
86    });
87  }
88
89  #assertOpen(): void {
90    if (this.#closed) throw new Error("Sidebar session closed.");
91  }
92
93  #clearWait(): void {
94    clearTimeout(this.#waitTimer);
95    this.#waitTimer = undefined;
96    this.#wait = undefined;
97  }
98
99  dispose(): void {
100    this.#closed = true;
101    clearTimeout(this.#timer);
102    this.#pending = undefined;
103    this.#sink?.receiver.dispose();
104    this.#sink = undefined;
105    this.#wait?.reject(new Error("Sidebar session closed."));
106    this.#clearWait();
107  }
108}
109