1/**
2 * Node.js runtime support for workers that serve Effect worker requests.
3 *
4 * `NodeWorkerRunner` supplies the Node implementation of the Effect worker
5 * runner platform. The exported `layer` runs inside a `node:worker_threads`
6 * worker through `parentPort`, or inside a child process through
7 * `process.send`. It listens for parent messages, runs handlers registered with
8 * `WorkerRunner`, sends replies over the same channel, and closes when the
9 * parent sends the close message.
10 *
11 * @since 4.0.0
12 */
13import * as Cause from "effect/Cause"
14import * as Deferred from "effect/Deferred"
15import * as Effect from "effect/Effect"
16import * as Exit from "effect/Exit"
17import * as Fiber from "effect/Fiber"
18import * as Layer from "effect/Layer"
19import * as Scope from "effect/Scope"
20import { WorkerError, WorkerReceiveError, WorkerSpawnError } from "effect/workers/WorkerError"
21import * as WorkerRunner from "effect/workers/WorkerRunner"
22import * as WorkerThreads from "node:worker_threads"
23
24/**
25 * Provides the `WorkerRunnerPlatform` for code running inside a Node worker
26 * thread or child process, routing parent messages to the registered handler
27 * and sending responses back through the parent channel.
28 *
29 * @category layers
30 * @since 4.0.0
31 */
32export const layer: Layer.Layer<WorkerRunner.WorkerRunnerPlatform> = Layer.succeed(WorkerRunner.WorkerRunnerPlatform)({
33 start<O = unknown, I = unknown>() {
34 return Effect.gen(function*() {
35 if (!WorkerThreads.parentPort && !process.send) {
36 return yield* new WorkerError({
37 reason: new WorkerSpawnError({ message: "not in a worker" })
38 })
39 }
40
41 const sendRaw = WorkerThreads.parentPort
42 ? (_portId: number, message: any, transfers?: any) => WorkerThreads.parentPort!.postMessage(message, transfers)
43 : (_portId: number, message: any, _transfers?: any) => process.send!(message)
44 const sendUnsafe = (_portId: number, message: O, transfers?: ReadonlyArray<unknown>) =>
45 sendRaw(_portId, [1, message], transfers)
46 const send = (_portId: number, message: O, transfers?: ReadonlyArray<unknown>) =>
47 Effect.sync(() => sendUnsafe(_portId, message, transfers))
48
49 const run = <A, E, R>(
50 handler: (portId: number, message: I) => Effect.Effect<A, E, R> | void
51 ): Effect.Effect<void, WorkerError, R> =>
52 Effect.scopedWith(Effect.fnUntraced(function*(scope) {
53 const closeLatch = Deferred.makeUnsafe<void, WorkerError>()
54 const trackFiber = Fiber.runIn(scope)
55 const services = yield* Effect.context<R>()
56 const runFork = Effect.runForkWith(services)
57 const onExit = (exit: Exit.Exit<any, E>) => {
58 if (exit._tag === "Failure" && !Cause.hasInterruptsOnly(exit.cause)) {
59 runFork(Effect.logError("unhandled error in worker", exit.cause))
60 }
61 }
62 const port = WorkerThreads.parentPort ?? process
63 function onMessage(message: WorkerRunner.PlatformMessage<I>) {
64 if (message[0] === 0) {
65 const result = handler(0, message[1])
66 if (Effect.isEffect(result)) {
67 const fiber = runFork(result)
68 fiber.addObserver(onExit)
69 trackFiber(fiber)
70 }
71 } else {
72 if (WorkerThreads.parentPort) {
73 WorkerThreads.parentPort.close()
74 } else {
75 process.channel?.unref()
76 }
77 Deferred.doneUnsafe(closeLatch, Exit.void)
78 }
79 }
80 port.on("message", onMessage)
81
82 function onMessageError(cause: unknown) {
83 Deferred.doneUnsafe(
84 closeLatch,
85 new WorkerError({
86 reason: new WorkerReceiveError({
87 message: "received messageerror event",
88 cause
89 })
90 })
91 )
92 }
93 function onError(cause: unknown) {
94 Deferred.doneUnsafe(
95 closeLatch,
96 new WorkerError({
97 reason: new WorkerReceiveError({
98 message: "received error event",
99 cause
100 })
101 })
102 )
103 }
104 if (WorkerThreads.parentPort) {
105 WorkerThreads.parentPort.on("messageerror", onMessageError)
106 WorkerThreads.parentPort.on("error", onError)
107 }
108
109 yield* Scope.addFinalizer(
110 scope,
111 Effect.sync(() => {
112 port.off("message", onMessage)
113 if (WorkerThreads.parentPort) {
114 WorkerThreads.parentPort.off("messageerror", onMessageError)
115 WorkerThreads.parentPort.off("error", onError)
116 }
117 })
118 )
119
120 sendRaw(0, [0])
121
122 return yield* Deferred.await(closeLatch)
123 }))
124
125 return { run, send, sendUnsafe }
126 })
127 }
128})