1/**
2 * Worker-entrypoint support for Bun worker runners.
3 *
4 * This module exports a `layer` that provides `WorkerRunnerPlatform` for code
5 * already running inside a Bun `Worker`. The platform receives request messages
6 * from the parent-side `BunWorker` platform, runs the registered handler, sends
7 * responses through the worker `postMessage` channel, and closes when the
8 * parent sends the close message.
9 *
10 * @since 4.0.0
11 */
12import * as Cause from "effect/Cause"
13import * as Deferred from "effect/Deferred"
14import * as Effect from "effect/Effect"
15import * as Exit from "effect/Exit"
16import * as Fiber from "effect/Fiber"
17import { identity } from "effect/Function"
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"
22
23declare const self: MessagePort
24
25/**
26 * Provides the `WorkerRunnerPlatform` for code running inside a Bun worker,
27 * routing parent messages to the registered handler and sending responses back
28 * through the worker port.
29 *
30 * @category layers
31 * @since 4.0.0
32 */
33export const layer: Layer.Layer<WorkerRunner.WorkerRunnerPlatform> = Layer.succeed(WorkerRunner.WorkerRunnerPlatform)({
34 start: Effect.fnUntraced(function*<O = unknown, I = unknown>() {
35 if (!("postMessage" in self)) {
36 return yield* new WorkerError({
37 reason: new WorkerSpawnError({ message: "not in a Worker context" })
38 })
39 }
40 const port = self
41 const run = <A, E, R>(
42 handler: (portId: number, message: I) => Effect.Effect<A, E, R> | void
43 ) =>
44 Effect.scopedWith(Effect.fnUntraced(function*(scope) {
45 const closeLatch = Deferred.makeUnsafe<void, WorkerError>()
46 const trackFiber = Fiber.runIn(scope)
47 const services = yield* Effect.context<R>()
48 const runFork = Effect.runForkWith(services)
49 const onExit = (exit: Exit.Exit<any, E>) => {
50 if (exit._tag === "Failure" && !Cause.hasInterruptsOnly(exit.cause)) {
51 runFork(Effect.logError("unhandled error in worker", exit.cause))
52 }
53 }
54
55 function onMessage(event: MessageEvent) {
56 const message = (event as MessageEvent).data as WorkerRunner.PlatformMessage<I>
57 if (message[0] === 0) {
58 const result = handler(0, message[1])
59 if (Effect.isEffect(result)) {
60 const fiber = runFork(result)
61 fiber.addObserver(onExit)
62 trackFiber(fiber)
63 }
64 } else {
65 port.close()
66 Deferred.doneUnsafe(closeLatch, Exit.void)
67 }
68 }
69 function onMessageError(error: MessageEvent) {
70 Deferred.doneUnsafe(
71 closeLatch,
72 new WorkerError({
73 reason: new WorkerReceiveError({
74 message: "received messageerror event",
75 cause: error.data
76 })
77 })
78 )
79 }
80 yield* Scope.addFinalizer(
81 scope,
82 Effect.sync(() => {
83 port.removeEventListener("message", onMessage)
84 port.removeEventListener("messageerror", onMessageError)
85 })
86 )
87 port.addEventListener("message", onMessage)
88 port.addEventListener("messageerror", onMessageError)
89 port.postMessage([0])
90
91 return yield* Deferred.await(closeLatch)
92 }))
93
94 const sendUnsafe = (_portId: number, message: O, transfer?: ReadonlyArray<unknown>) =>
95 port.postMessage([1, message], {
96 transfer: transfer as any
97 })
98 const send = (_portId: number, message: O, transfer?: ReadonlyArray<unknown>) =>
99 Effect.sync(() => sendUnsafe(0, message, transfer))
100 return identity<WorkerRunner.WorkerRunner<any, any>>({ run, send, sendUnsafe })
101 })
102})