1/**
2 * Client-side worker primitives shared by browser, Node, and Bun adapters.
3 *
4 * This module defines the platform-neutral `Worker` client, the `WorkerPlatform`
5 * service that creates workers by numeric id, and the `Spawner` service used to
6 * find platform-specific worker instances. `makePlatform` wraps platform setup
7 * and listen hooks into a `WorkerPlatform`, buffers outgoing messages until the
8 * worker is ready, runs incoming messages with Effect handlers, and ties worker
9 * cleanup to scope lifetime.
10 *
11 * @stability unstable
12 * @since 4.0.0
13 */
14import * as Context from "../Context.ts"
15import type * as Deferred from "../Deferred.ts"
16import * as Effect from "../Effect.ts"
17import * as FiberSet from "../FiberSet.ts"
18import * as Latch from "../Latch.ts"
19import * as Layer from "../Layer.ts"
20import * as Scope from "../Scope.ts"
21import { WorkerError, WorkerSendError } from "./WorkerError.ts"
22
23/**
24 * Service that spawns effect `Worker` instances for numeric worker ids using
25 * the configured `Spawner`.
26 *
27 * @stability unstable
28 * @category services
29 * @since 4.0.0
30 */
31export class WorkerPlatform extends Context.Service<WorkerPlatform, {
32 readonly spawn: <O = unknown, I = unknown>(
33 id: number
34 ) => Effect.Effect<Worker<O, I>, WorkerError, Spawner>
35}>()("effect/workers/Worker/WorkerPlatform") {}
36
37/**
38 * Effect-based worker abstraction that can send input messages and run a
39 * long-lived handler for output messages, failing with `WorkerError` or handler
40 * errors.
41 *
42 * @stability unstable
43 * @category models
44 * @since 4.0.0
45 */
46export interface Worker<O = unknown, I = unknown> {
47 readonly send: (
48 message: I,
49 transfers?: ReadonlyArray<unknown>
50 ) => Effect.Effect<void, WorkerError>
51 readonly run: <A, E, R>(
52 handler: (message: O) => Effect.Effect<A, E, R>,
53 options?:
54 | {
55 readonly onSpawn?: Effect.Effect<void> | undefined
56 }
57 | undefined
58 ) => Effect.Effect<never, E | WorkerError, R>
59}
60
61/**
62 * Wraps platform-specific send and run functions into a `Worker`, translating
63 * platform ready/data messages and running the optional `onSpawn` effect when
64 * the worker reports readiness.
65 *
66 * @stability unstable
67 * @category constructors
68 * @since 4.0.0
69 */
70export const makeUnsafe = (options: {
71 readonly send: (
72 message: unknown,
73 transfers?: ReadonlyArray<unknown>
74 ) => Effect.Effect<void, WorkerError>
75 readonly run: <A, E, R>(
76 handler: (message: PlatformMessage) => Effect.Effect<A, E, R>
77 ) => Effect.Effect<never, E | WorkerError, R>
78}): Worker<any, any> => ({
79 send: options.send,
80 run(handler, options_) {
81 const onSpawn = options_?.onSpawn ?? Effect.void
82 return options.run((msg) => {
83 if (msg[0] === 0) return onSpawn
84 return handler(msg[1])
85 })
86 }
87})
88
89/**
90 * Internal worker platform protocol message: `[0]` signals readiness and
91 * `[1, payload]` carries data.
92 *
93 * @stability unstable
94 * @category models
95 * @since 4.0.0
96 */
97export type PlatformMessage = readonly [ready: 0] | readonly [data: 1, unknown]
98
99/**
100 * Phantom identifier for the service that maps worker ids to platform-specific
101 * worker instances.
102 *
103 * @stability unstable
104 * @category models
105 * @since 4.0.0
106 */
107export interface Spawner {
108 readonly _: unique symbol
109}
110
111/**
112 * Service tag for the worker `SpawnerFn`.
113 *
114 * @stability unstable
115 * @category services
116 * @since 4.0.0
117 */
118export const Spawner: Context.Service<
119 Spawner,
120 SpawnerFn<unknown>
121> = Context.Service("effect/workers/Worker/Spawner")
122
123/**
124 * Function that creates or locates a platform-specific worker instance for a
125 * numeric worker id.
126 *
127 * @stability unstable
128 * @category models
129 * @since 4.0.0
130 */
131export interface SpawnerFn<W = unknown> {
132 (id: number): W
133}
134
135/**
136 * Creates a layer that provides a worker `Spawner` service from a `SpawnerFn`.
137 *
138 * @stability unstable
139 * @category layers
140 * @since 4.0.0
141 */
142export const layerSpawner: <W = unknown>(
143 spawner: SpawnerFn<W>
144) => Layer.Layer<Spawner> = Layer.succeed(Spawner)
145
146/**
147 * Creates a `WorkerPlatform` from platform-specific setup and listen hooks,
148 * buffering sent messages until the worker is ready and scoping port cleanup to
149 * the worker run.
150 *
151 * @stability unstable
152 * @category constructors
153 * @since 4.0.0
154 */
155export const makePlatform = <W>() =>
156<
157 P extends {
158 readonly postMessage: (message: any, transfers?: any | undefined) => void
159 }
160>(options: {
161 readonly setup: (options: {
162 readonly worker: W
163 readonly scope: Scope.Scope
164 }) => Effect.Effect<P, WorkerError>
165 readonly listen: (options: {
166 readonly port: P
167 readonly emit: (data: any) => void
168 readonly deferred: Deferred.Deferred<never, WorkerError>
169 readonly scope: Scope.Scope
170 }) => Effect.Effect<void>
171}): WorkerPlatform["Service"] =>
172 WorkerPlatform.of({
173 spawn<O, I>(id: number) {
174 return Effect.gen(function*() {
175 const spawn = (yield* Spawner) as SpawnerFn<W>
176 let currentPort: P | undefined
177 const buffer: Array<[unknown, ReadonlyArray<unknown> | undefined]> = []
178 const sendToPort = (port: P, message: unknown, transfers?: ReadonlyArray<unknown>) =>
179 Effect.try({
180 try: () => port.postMessage([0, message], transfers as any),
181 catch: (cause) =>
182 new WorkerError({
183 reason: new WorkerSendError({
184 message: "Failed to send message to worker",
185 cause
186 })
187 })
188 })
189
190 const run = <A, E, R>(
191 handler: (_: O) => Effect.Effect<A, E, R>,
192 opts?: { readonly onSpawn?: Effect.Effect<void> | undefined }
193 ) =>
194 Effect.uninterruptibleMask((restore) =>
195 Effect.scopedWith(
196 Effect.fnUntraced(function*(scope) {
197 const port = yield* options.setup({
198 worker: spawn(id),
199 scope
200 })
201 yield* Scope.addFinalizer(
202 scope,
203 Effect.sync(() => {
204 currentPort = undefined
205 })
206 )
207 const fiberSet = yield* FiberSet.make<
208 any,
209 WorkerError | E
210 >().pipe(Scope.provide(scope))
211 const run = yield* FiberSet.runtime(fiberSet)<R>()
212 const ready = Latch.makeUnsafe()
213 yield* options.listen({
214 port,
215 scope,
216 emit(data: PlatformMessage) {
217 if (data[0] === 0) {
218 if (opts?.onSpawn) {
219 run(Effect.ensuring(opts.onSpawn, ready.open))
220 } else {
221 ready.openUnsafe()
222 }
223 return
224 }
225 run(handler(data[1] as O))
226 },
227 deferred: fiberSet.deferred as any
228 })
229 // Fail fast if the worker dies before signalling readiness,
230 // and allow interruption while waiting for the handshake.
231 yield* restore(Effect.raceFirst(ready.await, FiberSet.join(fiberSet)))
232 currentPort = port
233 if (buffer.length > 0) {
234 for (const [message, transfers] of buffer) {
235 yield* sendToPort(port, message, transfers)
236 }
237 buffer.length = 0
238 }
239 return (yield* restore(FiberSet.join(fiberSet))) as never
240 })
241 )
242 )
243
244 const send = (message: I, transfers?: ReadonlyArray<unknown>) =>
245 Effect.suspend(() => {
246 if (currentPort === undefined) {
247 buffer.push([message, transfers])
248 return Effect.void
249 }
250 return sendToPort(currentPort, message, transfers)
251 })
252
253 return { run, send }
254 })
255 }
256 })