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 = Effect.fn("bootstrap.makeBootstrapInputStream")(function* (
170 fd: number,
171) {
172 const platform = yield* HostProcess.Platform;
173 const fdPath = resolveFdPath(fd, platform);
174 return yield* Effect.try<NodeStream.Readable, BootstrapInputStreamOpenError>({
175 try: () => {
176 if (fdPath === undefined) {
177 return makeDirectBootstrapStream(fd);
178 }
179
180 let streamFd: number | undefined;
181 try {
182 streamFd = NodeFS.openSync(fdPath, "r");
183 return NodeFS.createReadStream("", {
184 fd: streamFd,
185 encoding: "utf8",
186 autoClose: true,
187 });
188 } catch (error) {
189 if (isBootstrapFdPathDuplicationError(error)) {
190 if (streamFd !== undefined) {
191 NodeFS.closeSync(streamFd);
192 }
193 return makeDirectBootstrapStream(fd);
194 }
195 throw error;
196 }
197 },
198 catch: (error) =>
199 new BootstrapInputStreamOpenError({
200 fd,
201 platform,
202 ...(fdPath === undefined ? {} : { fdPath }),
203 cause: error,
204 }),
205 });
206});
207
208const makeDirectBootstrapStream = (fd: number): NodeStream.Readable => {
209 try {
210 return NodeFS.createReadStream("", {
211 fd,
212 encoding: "utf8",
213 autoClose: true,
214 });
215 } catch {
216 const stream = new NodeNet.Socket({
217 fd,
218 readable: true,
219 writable: false,
220 });
221 stream.setEncoding("utf8");
222 return stream;
223 }
224};
225
226// WSL-inherited stdin pipes cannot be reopened via /proc/self/fd/0 (EACCES),
227// so they must be read directly.
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}