1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeFS from "node:fs";
3import * as NodeNet from "node:net";
4import * as NodeReadline from "node:readline";
5import type * as NodeStream from "node:stream";
6
7import * as Effect from "effect/Effect";
8import * as Option from "effect/Option";
9import * as Predicate from "effect/Predicate";
10import * as Result from "effect/Result";
11import * as Schema from "effect/Schema";
12import { decodeJsonResult } from "@t3tools/shared/schemaJson";
13import * as HostProcess from "@t3tools/shared/HostProcess";
14
15export class BootstrapFdStatError extends Schema.TaggedError<BootstrapFdStatError>()(
16 "BootstrapFdStatError",
17 {
18 fd: Schema.Number,
19 cause: Schema.Defect(),
20 },
21) {
22 override get message(): string {
23 return `Failed to stat bootstrap file descriptor ${this.fd}.`;
24 }
25}
26
27export class BootstrapInputStreamOpenError extends Schema.TaggedError<BootstrapInputStreamOpenError>()(
28 "BootstrapInputStreamOpenError",
29 {
30 fd: Schema.Number,
31 platform: Schema.String,
32 fdPath: Schema.optional(Schema.String),
33 cause: Schema.Defect(),
34 },
35) {
36 override get message(): string {
37 const path = this.fdPath === undefined ? "" : ` via '${this.fdPath}'`;
38 return `Failed to open bootstrap input stream for file descriptor ${this.fd}${path} on '${this.platform}'.`;
39 }
40}
41
42export class BootstrapEnvelopeReadError extends Schema.TaggedError<BootstrapEnvelopeReadError>()(
43 "BootstrapEnvelopeReadError",
44 {
45 fd: Schema.Number,
46 cause: Schema.Defect(),
47 },
48) {
49 override get message(): string {
50 return `Failed to read bootstrap envelope from file descriptor ${this.fd}.`;
51 }
52}
53
54export class BootstrapEnvelopeDecodeError extends Schema.TaggedError<BootstrapEnvelopeDecodeError>()(
55 "BootstrapEnvelopeDecodeError",
56 {
57 fd: Schema.Number,
58 cause: Schema.Defect(),
59 },
60) {
61 override get message(): string {
62 return `Failed to decode bootstrap envelope from file descriptor ${this.fd}.`;
63 }
64}
65
66export const BootstrapError = Schema.Union([
67 BootstrapFdStatError,
68 BootstrapInputStreamOpenError,
69 BootstrapEnvelopeReadError,
70 BootstrapEnvelopeDecodeError,
71]);
72export type BootstrapError = typeof BootstrapError.Type;
73
74export const readBootstrapEnvelope = Effect.fn("readBootstrapEnvelope")(function* <A, I>(
75 schema: Schema.Codec<A, I>,
76 fd: number,
77 options?: {
78 timeoutMs?: number;
79 },
80): Effect.fn.Return<Option.Option<A>, BootstrapError> {
81 const fdReady = yield* isFdReady(fd);
82 if (!fdReady) return Option.none();
83
84 const stream = yield* makeBootstrapInputStream(fd);
85
86 const timeoutMs = options?.timeoutMs ?? 1000;
87
88 return yield* Effect.callback<
89 Option.Option<A>,
90 BootstrapEnvelopeReadError | BootstrapEnvelopeDecodeError
91 >((resume) => {
92 const input = NodeReadline.createInterface({
93 input: stream,
94 crlfDelay: Infinity,
95 });
96
97 const cleanup = () => {
98 stream.removeListener("error", handleError);
99 input.removeListener("line", handleLine);
100 input.removeListener("close", handleClose);
101 input.close();
102 stream.destroy();
103 };
104
105 const handleError = (error: Error) => {
106 if (isUnavailableBootstrapFdError(error)) {
107 resume(Effect.succeedNone);
108 return;
109 }
110 resume(
111 Effect.fail(
112 new BootstrapEnvelopeReadError({
113 fd,
114 cause: error,
115 }),
116 ),
117 );
118 };
119
120 const handleLine = (line: string) => {
121 const parsed = decodeJsonResult(schema)(line);
122 if (Result.isSuccess(parsed)) {
123 resume(Effect.succeedSome(parsed.success));
124 } else {
125 resume(
126 Effect.fail(
127 new BootstrapEnvelopeDecodeError({
128 fd,
129 cause: parsed.failure,
130 }),
131 ),
132 );
133 }
134 };
135
136 const handleClose = () => {
137 resume(Effect.succeedNone);
138 };
139
140 stream.once("error", handleError);
141 input.once("line", handleLine);
142 input.once("close", handleClose);
143
144 return Effect.sync(cleanup);
145 }).pipe(Effect.timeoutOption(timeoutMs), Effect.map(Option.flatten));
146});
147
148const isUnavailableBootstrapFdError = Predicate.compose(
149 Predicate.hasProperty("code"),
150 (_) => _.code === "EBADF" || _.code === "ENOENT",
151);
152
153const isFdReady = (fd: number) =>
154 Effect.try({
155 try: () => NodeFS.fstatSync(fd),
156 catch: (error) =>
157 new BootstrapFdStatError({
158 fd,
159 cause: error,
160 }),
161 }).pipe(
162 Effect.as(true),
163 Effect.catchTags({
164 BootstrapFdStatError: (error) =>
165 isUnavailableBootstrapFdError(error.cause) ? Effect.succeed(false) : Effect.fail(error),
166 }),
167 );
168
169const makeBootstrapInputStream = (fd: number) =>
170 Effect.gen(function* () {
171 const platform = yield* HostProcess.Platform;
172 const fdPath = resolveFdPath(fd, platform);
173 return yield* Effect.try<NodeStream.Readable, BootstrapInputStreamOpenError>({
174 try: () => {
175 if (fdPath === undefined) {
176 return makeDirectBootstrapStream(fd);
177 }
178
179 let streamFd: number | undefined;
180 try {
181 streamFd = NodeFS.openSync(fdPath, "r");
182 return NodeFS.createReadStream("", {
183 fd: streamFd,
184 encoding: "utf8",
185 autoClose: true,
186 });
187 } catch (error) {
188 if (isBootstrapFdPathDuplicationError(error)) {
189 if (streamFd !== undefined) {
190 NodeFS.closeSync(streamFd);
191 }
192 return makeDirectBootstrapStream(fd);
193 }
194 throw error;
195 }
196 },
197 catch: (error) =>
198 new BootstrapInputStreamOpenError({
199 fd,
200 platform,
201 ...(fdPath === undefined ? {} : { fdPath }),
202 cause: error,
203 }),
204 });
205 });
206
207const makeDirectBootstrapStream = (fd: number): NodeStream.Readable => {
208 try {
209 return NodeFS.createReadStream("", {
210 fd,
211 encoding: "utf8",
212 autoClose: true,
213 });
214 } catch {
215 const stream = new NodeNet.Socket({
216 fd,
217 readable: true,
218 writable: false,
219 });
220 stream.setEncoding("utf8");
221 return stream;
222 }
223};
224
225// Stdin pipes inherited across the wsl.exe boundary report EACCES when we try
226// to re-open them via /proc/self/fd/0 — fall back to reading the fd directly
227// in that case, the same way we already do for ENXIO/EINVAL/EPERM.
228const isBootstrapFdPathDuplicationError = Predicate.compose(
229 Predicate.hasProperty("code"),
230 (_) => _.code === "ENXIO" || _.code === "EINVAL" || _.code === "EPERM" || _.code === "EACCES",
231);
232
233function resolveFdPath(fd: number, platform: NodeJS.Platform): string | undefined {
234 if (platform === "linux") {
235 return `/proc/self/fd/${fd}`;
236 }
237 if (platform === "win32") {
238 return undefined;
239 }
240 return `/dev/fd/${fd}`;
241}