1/**
2 * Deno implementation of the child process spawner service.
3 *
4 * This module uses `Deno.Command` and Web Streams directly. Deno cannot create
5 * detached process groups, so killing a handle terminates only its direct child;
6 * descendants spawned by that child are left running.
7 *
8 * @since 4.0.0
9 */
10import type * as Arr from "effect/Array"
11import * as Deferred from "effect/Deferred"
12import * as Effect from "effect/Effect"
13import * as Exit from "effect/Exit"
14import * as Layer from "effect/Layer"
15import * as Path from "effect/Path"
16import * as PlatformError from "effect/PlatformError"
17import * as Predicate from "effect/Predicate"
18import * as ChildProcess from "effect/process/ChildProcess"
19import type { ChildProcessHandle } from "effect/process/ChildProcessSpawner"
20import {
21 ChildProcessSpawner,
22 ExitCode,
23 make as makeSpawner,
24 makeHandle,
25 ProcessId
26} from "effect/process/ChildProcessSpawner"
27import type * as Scope from "effect/Scope"
28import * as Sink from "effect/Sink"
29import * as Stream from "effect/Stream"
30
31const commandString = (command: ChildProcess.Command): string => {
32 const { commands } = flattenCommand(command)
33 return commands.map((command) => `${command.command} ${command.args.join(" ")}`).join(" | ")
34}
35
36const toPlatformError = (
37 method: string,
38 cause: unknown,
39 command: ChildProcess.Command
40): PlatformError.PlatformError => {
41 const errorName = Predicate.hasProperty(cause, "name") ? cause.name : undefined
42 const tag = errorName === "NotFound" ?
43 "NotFound" :
44 errorName === "PermissionDenied" ?
45 "PermissionDenied" :
46 errorName === "TimedOut"
47 ? "TimedOut"
48 : "Unknown"
49 return PlatformError.systemError({
50 _tag: tag,
51 module: "ChildProcess",
52 method,
53 pathOrDescriptor: commandString(command),
54 syscall: `${method} ${commandString(command).trim()}`,
55 cause
56 })
57}
58
59const unsupported = (option: "additionalFds" | "detached") =>
60 PlatformError.badArgument({
61 module: "ChildProcessSpawner",
62 method: "spawn",
63 description: `The ${option} option is unsupported because Deno has no equivalent`
64 })
65
66const make = Effect.gen(function*() {
67 const path = yield* Path.Path
68
69 const resolveWorkingDirectory = (options: ChildProcess.CommandOptions) =>
70 options.cwd === undefined ? undefined : path.resolve(options.cwd)
71
72 const resolveEnvironment = (options: ChildProcess.CommandOptions) => {
73 if (options.env === undefined) return undefined
74 const env: Record<string, string> = {}
75 for (const [key, value] of Object.entries(options.env)) {
76 if (value !== undefined) env[key] = value
77 }
78 return env
79 }
80
81 const resolveStdinOption = (options: ChildProcess.CommandOptions): ChildProcess.StdinConfig => {
82 const defaultConfig: ChildProcess.StdinConfig = { stream: "pipe", encoding: "utf-8", endOnDone: true }
83 if (options.stdin === undefined) {
84 return defaultConfig
85 }
86 if (typeof options.stdin === "string") {
87 return { ...defaultConfig, stream: options.stdin }
88 }
89 if (Stream.isStream(options.stdin)) {
90 return { ...defaultConfig, stream: options.stdin }
91 }
92 return {
93 stream: options.stdin.stream,
94 encoding: options.stdin.encoding ?? defaultConfig.encoding,
95 endOnDone: options.stdin.endOnDone ?? defaultConfig.endOnDone
96 }
97 }
98
99 const resolveOutputOption = (
100 options: ChildProcess.CommandOptions,
101 streamName: "stdout" | "stderr"
102 ): ChildProcess.StdoutConfig => {
103 const option = options[streamName]
104 if (option === undefined) return { stream: "pipe" }
105 if (typeof option === "string" || Sink.isSink(option)) return { stream: option }
106 return { stream: option.stream }
107 }
108
109 const inputToStdioOption = (
110 input: ChildProcess.CommandInput
111 ): "piped" | "inherit" | "null" =>
112 Stream.isStream(input) || input === "pipe" || input === "overlapped" ?
113 "piped" :
114 input === "ignore"
115 ? "null"
116 : "inherit"
117
118 const outputToStdioOption = (
119 output: ChildProcess.CommandOutput | undefined
120 ): "piped" | "inherit" | "null" =>
121 Sink.isSink(output) || output === "pipe" || output === "overlapped" || output === undefined ?
122 "piped" :
123 output === "ignore"
124 ? "null"
125 : "inherit"
126
127 const setupChildStdin = (
128 command: ChildProcess.StandardCommand,
129 childProcess: Deno.ChildProcess,
130 config: ChildProcess.StdinConfig
131 ) =>
132 Effect.suspend(() => {
133 if (inputToStdioOption(config.stream) !== "piped") return Effect.succeed(Sink.drain)
134 const sink = Sink.fromWritableStream({
135 evaluate: () => childProcess.stdin,
136 onError: (cause) => toPlatformError("fromWritable(stdin)", cause, command),
137 closeOnDone: config.endOnDone
138 })
139 return Stream.isStream(config.stream)
140 ? Effect.as(Effect.forkScoped(Stream.run(config.stream, sink)), sink)
141 : Effect.succeed(sink)
142 })
143
144 const setupChildOutputStreams = (
145 command: ChildProcess.StandardCommand,
146 childProcess: Deno.ChildProcess,
147 stdoutConfig: ChildProcess.StdoutConfig,
148 stderrConfig: ChildProcess.StderrConfig
149 ) => {
150 let stdout: Stream.Stream<Uint8Array, PlatformError.PlatformError> =
151 outputToStdioOption(stdoutConfig.stream) === "piped"
152 ? Stream.fromReadableStream({
153 evaluate: () => childProcess.stdout,
154 onError: (cause) => toPlatformError("fromReadable(stdout)", cause, command)
155 })
156 : Stream.empty
157 let stderr: Stream.Stream<Uint8Array, PlatformError.PlatformError> =
158 outputToStdioOption(stderrConfig.stream) === "piped"
159 ? Stream.fromReadableStream({
160 evaluate: () => childProcess.stderr,
161 onError: (cause) => toPlatformError("fromReadable(stderr)", cause, command)
162 })
163 : Stream.empty
164
165 if (Sink.isSink(stdoutConfig.stream)) stdout = Stream.transduce(stdout, stdoutConfig.stream)
166 if (Sink.isSink(stderrConfig.stream)) stderr = Stream.transduce(stderr, stderrConfig.stream)
167
168 return { stdout, stderr, all: Stream.merge(stdout, stderr) }
169 }
170
171 const spawn = (
172 command: ChildProcess.StandardCommand,
173 executable: string,
174 args: ReadonlyArray<string>,
175 options: Deno.CommandOptions
176 ) =>
177 Effect.try({
178 try: () => {
179 const childProcess = new Deno.Command(executable, { ...options, args: [...args] }).spawn()
180 const exitSignal = Deferred.makeUnsafe<Deno.CommandStatus, PlatformError.PlatformError>()
181 childProcess.status.then(
182 (status) => Deferred.doneUnsafe(exitSignal, Exit.succeed(status)),
183 (cause) => Deferred.doneUnsafe(exitSignal, Exit.fail(toPlatformError("status", cause, command)))
184 )
185 return [childProcess, exitSignal] as const
186 },
187 catch: (cause) => toPlatformError("spawn", cause, command)
188 })
189
190 const killProcess = (
191 command: ChildProcess.StandardCommand,
192 childProcess: Deno.ChildProcess,
193 signal: ChildProcess.Signal
194 ) =>
195 Effect.try({
196 try: () => childProcess.kill(signal as Deno.Signal),
197 catch: (cause) => toPlatformError("kill", cause, command)
198 })
199
200 const withTimeout = (
201 childProcess: Deno.ChildProcess,
202 command: ChildProcess.StandardCommand,
203 options: ChildProcess.KillOptions | undefined
204 ) =>
205 <A, E, R>(
206 kill: (
207 command: ChildProcess.StandardCommand,
208 childProcess: Deno.ChildProcess,
209 signal: ChildProcess.Signal
210 ) => Effect.Effect<A, E, R>
211 ) => {
212 const killSignal = options?.killSignal ?? "SIGTERM"
213 return options?.forceKillAfter === undefined
214 ? kill(command, childProcess, killSignal)
215 : Effect.timeoutOrElse(kill(command, childProcess, killSignal), {
216 duration: options.forceKillAfter,
217 orElse: () => kill(command, childProcess, "SIGKILL")
218 })
219 }
220
221 const getSourceStream = (
222 handle: ChildProcessHandle,
223 from: "stdout" | "stderr" | "all"
224 ): Stream.Stream<Uint8Array, PlatformError.PlatformError> => {
225 switch (from) {
226 case "stdout":
227 return handle.stdout
228 case "stderr":
229 return handle.stderr
230 case "all":
231 return handle.all
232 }
233 }
234
235 const spawnCommand: (
236 command: ChildProcess.Command
237 ) => Effect.Effect<ChildProcessHandle, PlatformError.PlatformError, Scope.Scope> = Effect.fnUntraced(function*(cmd) {
238 switch (cmd._tag) {
239 case "StandardCommand": {
240 if (cmd.options.additionalFds !== undefined) {
241 return yield* Effect.fail(unsupported("additionalFds"))
242 }
243 if (cmd.options.detached !== undefined) {
244 return yield* Effect.fail(unsupported("detached"))
245 }
246
247 const stdinConfig = resolveStdinOption(cmd.options)
248 const stdoutConfig = resolveOutputOption(cmd.options, "stdout")
249 const stderrConfig = resolveOutputOption(cmd.options, "stderr")
250 let isReferenced = true
251
252 let executable = cmd.command
253 let args = cmd.args
254 let windowsRawArguments = false
255 if (cmd.options.shell) {
256 const shellCommand = `${cmd.command} ${cmd.args.join(" ")}`
257 if (Deno.build.os === "windows") {
258 executable = typeof cmd.options.shell === "string" ? cmd.options.shell : "cmd.exe"
259 args = ["/d", "/s", "/c", shellCommand]
260 windowsRawArguments = true
261 } else {
262 executable = typeof cmd.options.shell === "string" ? cmd.options.shell : "/bin/sh"
263 args = ["-c", shellCommand]
264 }
265 }
266
267 const cwd = resolveWorkingDirectory(cmd.options)
268 const env = resolveEnvironment(cmd.options)
269 const [childProcess, exitSignal] = yield* Effect.acquireRelease(
270 spawn(cmd, executable, args, {
271 ...(cwd === undefined ? undefined : { cwd }),
272 ...(env === undefined ? undefined : { env }),
273 clearEnv: cmd.options.env !== undefined && cmd.options.extendEnv !== true,
274 stdin: inputToStdioOption(stdinConfig.stream),
275 stdout: outputToStdioOption(stdoutConfig.stream),
276 stderr: outputToStdioOption(stderrConfig.stream),
277 windowsRawArguments
278 }),
279 Effect.fnUntraced(function*([childProcess, exitSignal]) {
280 const exited = yield* Deferred.isDone(exitSignal)
281 if (exited || !isReferenced) return
282 const killWithTimeout = withTimeout(childProcess, cmd, cmd.options)
283 yield* killWithTimeout((command, childProcess, signal) =>
284 killProcess(command, childProcess, signal).pipe(
285 Effect.andThen(Deferred.await(exitSignal))
286 )
287 ).pipe(Effect.ignore)
288 })
289 )
290
291 const reref = Effect.sync(() => {
292 if (!isReferenced) {
293 childProcess.ref()
294 isReferenced = true
295 }
296 })
297 const unref = Effect.sync(() => {
298 if (isReferenced) {
299 childProcess.unref()
300 isReferenced = false
301 }
302 return reref
303 })
304 const stdin = yield* setupChildStdin(cmd, childProcess, stdinConfig)
305 const { all, stderr, stdout } = setupChildOutputStreams(cmd, childProcess, stdoutConfig, stderrConfig)
306 const isRunning = Effect.map(Deferred.isDone(exitSignal), (done) => !done)
307 const exitCode = Effect.flatMap(Deferred.await(exitSignal), (status) =>
308 Predicate.isNull(status.signal)
309 ? Effect.succeed(ExitCode(status.code))
310 : Effect.fail(toPlatformError(
311 "exitCode",
312 new globalThis.Error(`Process interrupted due to receipt of signal: '${status.signal}'`),
313 cmd
314 )))
315 const kill = (options?: ChildProcess.KillOptions | undefined) => {
316 const killWithTimeout = withTimeout(childProcess, cmd, options)
317 return killWithTimeout((command, childProcess, signal) =>
318 killProcess(command, childProcess, signal).pipe(
319 Effect.andThen(Deferred.await(exitSignal))
320 )
321 ).pipe(Effect.asVoid)
322 }
323
324 return makeHandle({
325 pid: ProcessId(childProcess.pid),
326 exitCode,
327 isRunning,
328 kill,
329 stdin,
330 stdout,
331 stderr,
332 all,
333 getInputFd: () => Sink.drain,
334 getOutputFd: () => Stream.empty,
335 unref
336 })
337 }
338 case "PipedCommand": {
339 const { commands, pipeOptions } = flattenCommand(cmd)
340 const [root, ...pipeline] = commands
341 const handles = [yield* spawnCommand(root)]
342
343 for (let i = 0; i < pipeline.length; i++) {
344 const command = pipeline[i]
345 const options = pipeOptions[i] ?? {}
346 const stdinConfig = resolveStdinOption(command.options)
347 const from = options.from ?? "stdout"
348
349 if (
350 (from !== "stdout" && from !== "stderr" && from !== "all") ||
351 (options.to ?? "stdin") !== "stdin"
352 ) {
353 return yield* Effect.fail(unsupported("additionalFds"))
354 }
355 const sourceStream = getSourceStream(handles[handles.length - 1], from)
356 handles.push(
357 yield* spawnCommand(ChildProcess.make(command.command, command.args, {
358 ...command.options,
359 stdin: { ...stdinConfig, stream: sourceStream }
360 }))
361 )
362 }
363
364 const handle = handles[handles.length - 1]
365 const kill = (options?: ChildProcess.KillOptions | undefined) =>
366 Effect.forEach([...handles].reverse(), (handle) => Effect.ignore(handle.kill(options)), { discard: true })
367 const unref = Effect.gen(function*() {
368 const rerefs: Array<Effect.Effect<void, PlatformError.PlatformError>> = []
369 for (const handle of handles) rerefs.push(yield* handle.unref)
370 return Effect.forEach([...rerefs].reverse(), (reref) => reref, { discard: true })
371 })
372
373 return makeHandle({
374 pid: handle.pid,
375 exitCode: handle.exitCode,
376 isRunning: handle.isRunning,
377 kill,
378 stdin: handle.stdin,
379 stdout: handle.stdout,
380 stderr: handle.stderr,
381 all: handle.all,
382 getInputFd: handle.getInputFd,
383 getOutputFd: handle.getOutputFd,
384 unref
385 })
386 }
387 }
388 })
389
390 return makeSpawner(spawnCommand)
391})
392
393/**
394 * Layer that provides the Deno child process spawner.
395 *
396 * @category layers
397 * @since 4.0.0
398 */
399export const layer: Layer.Layer<ChildProcessSpawner, never, Path.Path> = Layer.effect(ChildProcessSpawner, make)
400
401/**
402 * Result of flattening a pipeline of commands.
403 *
404 * @category models
405 * @since 4.0.0
406 */
407export interface FlattenedPipeline {
408 readonly commands: Arr.NonEmptyReadonlyArray<ChildProcess.StandardCommand>
409 readonly pipeOptions: ReadonlyArray<ChildProcess.PipeOptions>
410}
411
412/**
413 * Flattens a command into standard commands and their pipe options.
414 *
415 * @category transforming
416 * @since 4.0.0
417 */
418export const flattenCommand = (command: ChildProcess.Command): FlattenedPipeline => {
419 const commands: Array<ChildProcess.StandardCommand> = []
420 const pipeOptions: Array<ChildProcess.PipeOptions> = []
421
422 const flatten = (command: ChildProcess.Command): void => {
423 switch (command._tag) {
424 case "StandardCommand":
425 commands.push(command)
426 break
427 case "PipedCommand":
428 flatten(command.left)
429 pipeOptions.push(command.options)
430 flatten(command.right)
431 break
432 }
433 }
434 flatten(command)
435
436 const [first, ...rest] = commands
437 if (first === undefined) throw new Error("flattenCommand produced empty commands array")
438 return { commands: [first, ...rest], pipeOptions }
439}