1// Adapted from opencode (MIT). Copyright (c) 2025 opencode. See LICENSE.opencode.
2// The engines run in a worker per open file, so no layout, recalculation or painting holds the window. These are the
3// two ends of that worker's protocol: typed calls by method name, a key that lets a newer call replace a queued one
4// (as the markdown worker's queue does), transferable payloads, and a worker that fails every pending call when it dies.
5
6import * as Schema from "effect/Schema";
7
8/** What one method takes and returns. */
9export type Method<Input, Output> = { readonly input: Input; readonly output: Output };
10
11/** A worker's methods by name. */
12export type Methods = { readonly [name: string]: Method<unknown, unknown> };
13
14type Request = {
15 readonly id: number;
16 readonly method: string;
17 readonly key?: string | undefined;
18 readonly input: unknown;
19};
20
21/** Withdraws a call whose caller aborted it. */
22type Cancel = { readonly cancel: number };
23
24type Response =
25 /** `transferred` lists the objects the reply moved, so a reply nobody waits for can close its bitmaps. */
26 | {
27 readonly id: number;
28 readonly ok: true;
29 readonly value: unknown;
30 readonly transferred: readonly Transferable[];
31 }
32 | {
33 readonly id: number;
34 readonly ok: false;
35 readonly superseded: boolean;
36 readonly message: string;
37 };
38
39/** A value a method returns with the buffers to move to the caller rather than copy. */
40export class Transfer<T> {
41 constructor(
42 readonly value: T,
43 readonly transfer: readonly Transferable[],
44 ) {}
45}
46
47/** A call that a newer call with the same key replaced before it ran. */
48export class SupersededError extends Schema.TaggedError<SupersededError>()("SupersededError", {
49 message: Schema.String,
50}) {}
51
52/** A call made after the worker failed or was closed, or pending when that happened. */
53class WorkerClosedError extends Schema.TaggedError<WorkerClosedError>()("WorkerClosedError", {
54 message: Schema.String,
55}) {}
56
57type CallOptions = {
58 /** A newer call with the same key replaces this one while it waits to run. */
59 readonly key?: string;
60 /** Buffers to move to the worker rather than copy, such as the file's bytes or a transferred canvas. */
61 readonly transfer?: readonly Transferable[];
62 /** Drops the reply and a call still queued; a call that has started sees its handler's signal abort. */
63 readonly signal?: AbortSignal;
64};
65
66/** A call waiting for its reply. */
67type Pending = {
68 readonly settle: (response: Response) => void;
69 readonly reject: (error: Error) => void;
70};
71
72/** The next call id, and why the worker stopped answering once it has. */
73type ClientState = { next: number; failure: Error | undefined };
74
75/** The window's end: starts the worker and calls its methods. */
76export function createWorkerClient<M extends Methods>(worker: Worker) {
77 const pending = new Map<number, Pending>();
78 const state: ClientState = { next: 0, failure: undefined };
79
80 const fail = (error: Error) => {
81 state.failure ??= error;
82 pending.forEach((call) => call.reject(error));
83 pending.clear();
84 worker.terminate();
85 };
86
87 worker.addEventListener("message", (event: MessageEvent<Response>) => {
88 const call = pending.get(event.data.id);
89
90 // A reply that crossed its call's cancel on the way: its bitmaps would hold their memory until collected.
91 if (!call) {
92 if (event.data.ok) closeBitmaps(event.data.transferred);
93
94 return;
95 }
96
97 pending.delete(event.data.id);
98 call.settle(event.data);
99 });
100
101 worker.addEventListener("error", (event) =>
102 fail(new WorkerClosedError({ message: event.message || "The Office engine stopped" })),
103 );
104 worker.addEventListener("messageerror", () =>
105 fail(new WorkerClosedError({ message: "The Office engine sent an unreadable reply" })),
106 );
107
108 return {
109 call<K extends keyof M & string>(
110 method: K,
111 input: M[K]["input"],
112 options: CallOptions = {},
113 ): Promise<M[K]["output"]> {
114 if (state.failure) return Promise.reject(state.failure);
115
116 if (options.signal?.aborted)
117 return Promise.reject(new WorkerClosedError({ message: "The call was aborted" }));
118
119 const id = ++state.next;
120
121 return new Promise<M[K]["output"]>((resolve, reject) => {
122 const timeout = setTimeout(
123 () =>
124 fail(
125 new WorkerClosedError({
126 message: "This document took too long to preview. Open it in another app.",
127 }),
128 ),
129 30_000,
130 );
131 const abort = () => {
132 clearTimeout(timeout);
133 pending.delete(id);
134 reject(new WorkerClosedError({ message: "The call was aborted" }));
135 worker.postMessage({ cancel: id } satisfies Cancel, []);
136 };
137
138 options.signal?.addEventListener("abort", abort, { once: true });
139 pending.set(id, {
140 settle: (response) => {
141 clearTimeout(timeout);
142 options.signal?.removeEventListener("abort", abort);
143
144 if (!response.ok)
145 return reject(
146 response.superseded
147 ? new SupersededError({ message: response.message })
148 : new Error(response.message),
149 );
150
151 // SAFETY: the worker answers each id with the output of the method that id called (see `serveWorker`).
152 resolve(response.value as M[K]["output"]);
153 },
154 reject: (error) => {
155 clearTimeout(timeout);
156 options.signal?.removeEventListener("abort", abort);
157 reject(error);
158 },
159 });
160 worker.postMessage({ id, method, key: options.key, input } satisfies Request, [
161 ...(options.transfer ?? []),
162 ]);
163 });
164 },
165 /** Stops the worker, which frees its engine's memory, and fails every pending call. */
166 close() {
167 fail(new WorkerClosedError({ message: "The Office engine was closed" }));
168 },
169 };
170}
171
172/** Whether a call runs, and which call it is, so its caller can abort it. */
173type ServerState = {
174 running: boolean;
175 current: { readonly id: number; readonly controller: AbortController } | undefined;
176};
177
178/**
179 * The worker's end: runs calls one at a time, in order, letting messages through between them. A handler's signal
180 * aborts when its caller aborts the call.
181 */
182export function serveWorker<M extends Methods>(handlers: {
183 readonly [K in keyof M]: (
184 input: M[K]["input"],
185 signal: AbortSignal,
186 ) =>
187 | M[K]["output"]
188 | Transfer<M[K]["output"]>
189 | Promise<M[K]["output"] | Transfer<M[K]["output"]>>;
190}) {
191 const queue: Request[] = [];
192
193 const state: ServerState = { running: false, current: undefined };
194
195 const reply = (response: Response, transfer: readonly Transferable[] = []) =>
196 self.postMessage(response, { transfer: [...transfer] });
197
198 const run = async () => {
199 if (state.running) return;
200
201 state.running = true;
202
203 // Each call returns to the event loop before it leaves the queue, so calls that arrived meanwhile can replace it.
204 while (queue.length > 0) {
205 await new Promise((resolve) => setTimeout(resolve));
206
207 const request = queue.shift();
208
209 if (!request) break;
210
211 const handler = handlers[request.method];
212 const controller = new AbortController();
213
214 state.current = { id: request.id, controller };
215
216 if (!handler) {
217 reply({
218 id: request.id,
219 ok: false,
220 superseded: false,
221 message: `Unknown method ${request.method}`,
222 });
223 continue;
224 }
225
226 // SAFETY: the client sends each method the input its `Methods` entry declares (see `createWorkerClient`).
227 const input = request.input as never;
228
229 await Promise.resolve()
230 .then(() => handler(input, controller.signal))
231 .then(
232 (result) => {
233 const transfer = result instanceof Transfer ? result.transfer : [];
234
235 // The caller already settled a call it aborted, so nothing would take the reply or its bitmaps.
236 if (controller.signal.aborted) return closeBitmaps(transfer);
237
238 reply(
239 {
240 id: request.id,
241 ok: true,
242 value: result instanceof Transfer ? result.value : result,
243 transferred: transfer,
244 },
245 transfer,
246 );
247 },
248 (cause: unknown) => {
249 if (controller.signal.aborted) return;
250
251 reply({
252 id: request.id,
253 ok: false,
254 superseded: false,
255 message: cause instanceof Error ? cause.message : String(cause),
256 });
257 },
258 );
259 }
260
261 state.running = false;
262 state.current = undefined;
263 };
264
265 self.addEventListener("message", (event: MessageEvent<Request | Cancel>) => {
266 // The caller already settled a cancelled call, so it gets no reply.
267 if ("cancel" in event.data) {
268 const id = event.data.cancel;
269 const queued = queue.findIndex((request) => request.id === id);
270
271 if (queued !== -1) queue.splice(queued, 1);
272
273 if (state.current?.id === id) state.current.controller.abort();
274
275 return;
276 }
277
278 const key = event.data.key;
279 const replaced = key === undefined ? -1 : queue.findIndex((request) => request.key === key);
280
281 if (replaced !== -1) {
282 const [old] = queue.splice(replaced, 1);
283
284 if (old)
285 reply({ id: old.id, ok: false, superseded: true, message: "Replaced by a newer call" });
286 }
287
288 queue.push(event.data);
289 void run();
290 });
291}
292
293/** Frees the bitmaps among objects that were moved for a reply nobody takes. */
294function closeBitmaps(transfer: readonly Transferable[]) {
295 transfer.forEach((item) => {
296 if (item instanceof ImageBitmap) item.close();
297 });
298}