1/**
2 * Node socket adapters for Effect sockets.
3 *
4 * This module opens `node:net` or `node:tls` connections, or wraps existing
5 * Node `Duplex` streams, and presents them as `Socket.Socket` values, socket
6 * channels, or layers. It also exposes the `NetSocket` service tag for the
7 * underlying Node socket and re-exports the `ws` package namespace.
8 *
9 * @since 4.0.0
10 */
11import * as Arr from "effect/Array"
12import * as Channel from "effect/Channel"
13import * as Context from "effect/Context"
14import type * as Duration from "effect/Duration"
15import * as Effect from "effect/Effect"
16import * as Function from "effect/Function"
17import { identity } from "effect/Function"
18import * as Latch from "effect/Latch"
19import * as Layer from "effect/Layer"
20import * as Redacted from "effect/Redacted"
21import * as Scope from "effect/Scope"
22import * as Socket from "effect/socket/Socket"
23import { Buffer } from "node:buffer"
24import * as Net from "node:net"
25import type { Duplex } from "node:stream"
26import * as Tls from "node:tls"
27
28const isDeno = "Deno" in globalThis
29
30/**
31 * @stability unstable
32 * @category re-exports
33 * @since 4.0.0
34 */
35export * as NodeWS from "ws"
36
37/**
38 * Service tag for the underlying Node `net.Socket` associated with the current
39 * socket connection.
40 *
41 * @category services
42 * @since 4.0.0
43 */
44export class NetSocket extends Context.Service<NetSocket, Net.Socket>()(
45 "@effect/platform-node/NodeSocket/NetSocket"
46) {}
47
48const readAvailable = (
49 conn: Duplex
50): Arr.NonEmptyReadonlyArray<Uint8Array | string> | undefined => {
51 const first = conn.read() as Uint8Array | string | null
52 if (first === null) return undefined
53 const second = conn.read() as Uint8Array | string | null
54 if (second === null) return [first]
55 const out: [Uint8Array | string, ...Array<Uint8Array | string>] = [first, second]
56 let chunk: Uint8Array | string | null
57 while ((chunk = conn.read() as Uint8Array | string | null) !== null) {
58 out.push(chunk)
59 }
60 return out
61}
62
63const toBuffers = (input: string | Uint8Array | ReadonlyArray<string | Uint8Array>): Array<Buffer> =>
64 Arr.map(Arr.ensure(input), (value) => Buffer.from(value))
65
66const closeSocket = (conn: Net.Socket, isOpen: boolean) => {
67 if (conn.closed !== false) return
68 if (!isOpen || !("destroySoon" in conn)) {
69 conn.destroy()
70 } else {
71 conn.destroySoon()
72 }
73}
74
75/**
76 * Opens a Node TCP connection as an Effect socket.
77 *
78 * **When to use**
79 *
80 * Use to create a `Socket.Socket` whose reader acquisition dials
81 * `net.createConnection`.
82 *
83 * **Details**
84 *
85 * Supports `openTimeout` and closes or destroys the underlying socket when the
86 * reader scope is finalized.
87 *
88 * @category constructors
89 * @since 4.0.0
90 */
91export const makeNet = (
92 options: Net.NetConnectOpts & {
93 readonly openTimeout?: Duration.Input | undefined
94 }
95): Effect.Effect<Socket.Socket> =>
96 fromDuplex(
97 Effect.contextWith((context: Context.Context<Scope.Scope>) => {
98 let conn: Net.Socket | undefined
99 let isOpen = false
100 return Effect.flatMap(
101 Scope.addFinalizer(
102 Context.get(context, Scope.Scope),
103 Effect.sync(() => {
104 if (!conn) return
105 closeSocket(conn, isOpen)
106 })
107 ),
108 () =>
109 Effect.callback<Net.Socket, Socket.SocketError, never>((resume) => {
110 conn = Net.createConnection(options)
111 conn.once("connect", () => {
112 isOpen = true
113 resume(Effect.succeed(conn!))
114 })
115 conn.on("error", (cause: Error) => {
116 resume(Effect.fail(
117 new Socket.SocketError({
118 reason: new Socket.SocketOpenError({ kind: "Unknown", cause })
119 })
120 ))
121 })
122 })
123 )
124 }),
125 options
126 )
127
128/**
129 * Adapts a Node `Duplex` into a `Socket.Socket`.
130 *
131 * **Details**
132 *
133 * Reader acquisition opens the duplex and keeps it paused: each pull drains
134 * Node's internal buffer with `stream.read()`. On Node 26+ that yields each
135 * buffered chunk as a separate batch element with no copy. Earlier Node
136 * concatenates the buffer into one chunk. The stream is never resumed, so once
137 * Node's buffer reaches its `highWaterMark` the kernel receive window closes
138 * and the peer blocks: backpressure is end-to-end with no buffering above
139 * Node's own.
140 *
141 * Writes use `write()` return-value backpressure, awaiting one `drain` when
142 * the internal buffer is full. `writeAll` corks the stream around the batch.
143 * Releasing the writer scope half-closes the stream (`end()`).
144 *
145 * @category constructors
146 * @since 4.0.0
147 */
148export const fromDuplex = <RO>(
149 open: Effect.Effect<Duplex, Socket.SocketError, RO>,
150 options?: {
151 readonly openTimeout?: Duration.Input | undefined
152 readonly tlsServer?: boolean | undefined
153 }
154): Effect.Effect<Socket.Socket, never, Exclude<RO, Scope.Scope>> =>
155 Effect.withFiber<Socket.Socket, never, Exclude<RO, Scope.Scope>>((fiber) => {
156 let currentSocket: Duplex | undefined
157 const latch = Latch.makeUnsafe(false)
158 const openServices = fiber.context as Context.Context<RO>
159 const isServer = options?.tlsServer === true
160 const secureEvent = isServer ? "secure" : "secureConnect"
161
162 const reader: Socket.Socket["reader"] = Effect.gen(function*() {
163 const scope = yield* Effect.scope
164 let conn = yield* Scope.provide(open, scope).pipe(
165 options?.openTimeout !== undefined ?
166 Effect.timeoutOrElse({
167 duration: options.openTimeout,
168 orElse: () =>
169 Effect.fail(
170 new Socket.SocketError({
171 reason: new Socket.SocketOpenError({ kind: "Timeout", cause: new Error("Connection timed out") })
172 })
173 )
174 }) :
175 identity
176 )
177
178 type ReadResume = (
179 effect: Effect.Effect<Arr.NonEmptyReadonlyArray<Uint8Array | string>, Socket.SocketError>
180 ) => void
181
182 let error: Socket.SocketError | undefined
183 let waiter: ReadResume | undefined
184 // Bytes consumed by read() while the parked pull was interrupted.
185 let pending: Arr.NonEmptyReadonlyArray<Uint8Array | string> | undefined
186 let reading = false
187 let closed = false
188 let upgradeAvailable = true
189
190 // The only place a waiter is resumed, so delivery order has one owner:
191 // retained bytes first, then freshly read bytes, then the error.
192 function drain() {
193 if (waiter === undefined || reading) return
194 let chunk = pending
195 pending = undefined
196 if (chunk === undefined && !closed) {
197 // A data listener can interrupt and replace the pull synchronously.
198 // The replacement must wait for this read to preserve byte order.
199 reading = true
200 try {
201 chunk = readAvailable(conn)
202 } finally {
203 reading = false
204 }
205 }
206 // A close during read() discards the bytes it consumed.
207 if (closed) chunk = undefined
208 if (waiter === undefined) {
209 pending = chunk
210 return
211 }
212 const result = chunk !== undefined
213 ? Effect.succeed(chunk)
214 : error !== undefined
215 ? Effect.fail(error)
216 : undefined
217 if (result === undefined) return
218 const resume = waiter
219 waiter = undefined
220 resume(result)
221 }
222 // Normal EOF: already consumed bytes are still delivered before the error.
223 function end(err: Socket.SocketError) {
224 error ??= err
225 drain()
226 }
227 // Teardown or failure: consumed bytes are discarded.
228 function close(err: Socket.SocketError) {
229 error ??= err
230 closed = true
231 pending = undefined
232 drain()
233 }
234 function onEnd() {
235 end(new Socket.SocketError({ reason: new Socket.SocketCloseError({ code: 1000 }) }))
236 }
237 function onError(cause: Error) {
238 close(
239 new Socket.SocketError({
240 reason: new Socket.SocketReadError({ cause })
241 })
242 )
243 }
244 function onClose(hadError: boolean) {
245 const err = new Socket.SocketError({
246 reason: new Socket.SocketCloseError({ code: hadError ? 1006 : 1000 })
247 })
248 if (hadError) close(err)
249 else end(err)
250 }
251
252 function attachReadListeners(conn: Duplex) {
253 conn.on("readable", drain)
254 conn.on("end", onEnd)
255 conn.on("error", onError)
256 conn.on("close", onClose)
257 }
258
259 function detachReadListeners(conn: Duplex) {
260 conn.off("readable", drain)
261 conn.off("end", onEnd)
262 conn.off("error", onError)
263 conn.off("close", onClose)
264 }
265
266 // Deno's node:net compatibility layer stops emitting `readable` after
267 // an explicit pause. The stream is already non-flowing without one.
268 if (!isDeno) conn.pause()
269 attachReadListeners(conn)
270 yield* Scope.addFinalizer(
271 scope,
272 Effect.sync(() => {
273 // resume a pull blocked in another fiber before detaching
274 close(
275 new Socket.SocketError({
276 reason: new Socket.SocketCloseError({ code: 1006 })
277 })
278 )
279 detachReadListeners(conn)
280 latch.closeUnsafe()
281 currentSocket = undefined
282 upgradeAvailable = false
283 })
284 )
285
286 currentSocket = conn
287 latch.openUnsafe()
288
289 const pull = Effect.callback<Arr.NonEmptyReadonlyArray<Uint8Array | string>, Socket.SocketError>((resume) => {
290 waiter = resume
291 drain()
292 // Resumed synchronously: nothing to cancel.
293 if (waiter !== resume) return
294 return Effect.sync(() => {
295 if (waiter === resume) waiter = undefined
296 })
297 })
298
299 const upgrade: Socket.Reader["upgrade"] = (upgradeOptions = {}) =>
300 Effect.suspend(() => {
301 if (!upgradeAvailable) {
302 return Effect.fail(
303 new Socket.SocketError({
304 reason: new Socket.SocketUpgradeError({
305 cause: new Error("socket is already upgraded or closed")
306 })
307 })
308 )
309 }
310 const hasKey = upgradeOptions.key !== undefined
311 const hasCert = upgradeOptions.cert !== undefined
312 if ((isServer && (!hasKey || !hasCert)) || hasKey !== hasCert) {
313 return Effect.fail(
314 new Socket.SocketError({
315 reason: new Socket.SocketUpgradeError({
316 cause: new Error(
317 isServer
318 ? "server TLS upgrade requires both key and cert"
319 : "TLS upgrade credentials must include both key and cert"
320 )
321 })
322 })
323 )
324 }
325 return Effect.callback<void, Socket.SocketError>((resume) => {
326 const raw = conn
327 detachReadListeners(raw)
328
329 let tls: Tls.TLSSocket
330 try {
331 const secureContext = Tls.createSecureContext({
332 key: upgradeOptions.key === undefined
333 ? undefined
334 : Arr.map(
335 Arr.ensure(upgradeOptions.key),
336 (value) => Buffer.from(Redacted.value(value))
337 ),
338 cert: upgradeOptions.cert === undefined ? undefined : toBuffers(upgradeOptions.cert),
339 ca: upgradeOptions.ca === undefined ? undefined : toBuffers(upgradeOptions.ca),
340 passphrase: upgradeOptions.passphrase === undefined
341 ? undefined
342 : Redacted.value(upgradeOptions.passphrase)
343 })
344 const tlsOptions = {
345 secureContext,
346 ALPNProtocols: upgradeOptions.alpnProtocols === undefined
347 ? undefined
348 : [...upgradeOptions.alpnProtocols],
349 requestCert: upgradeOptions.requestCert,
350 rejectUnauthorized: upgradeOptions.rejectUnauthorized
351 }
352 tls = isServer
353 ? new Tls.TLSSocket(raw as Net.Socket, { ...tlsOptions, isServer: true })
354 : Tls.connect({ ...tlsOptions, socket: raw as Net.Socket })
355 } catch (cause) {
356 attachReadListeners(raw)
357 resume(Effect.fail(
358 new Socket.SocketError({
359 reason: new Socket.SocketUpgradeError({ cause })
360 })
361 ))
362 return
363 }
364
365 conn = tls
366 currentSocket = tls
367 tls.pause()
368
369 function cleanup() {
370 tls.off(secureEvent, succeed)
371 tls.off("error", failUpgrade)
372 tls.off("close", onUpgradeClose)
373 }
374 function succeed() {
375 cleanup()
376 upgradeAvailable = false
377 attachReadListeners(tls)
378 resume(Effect.void)
379 }
380 function failUpgrade(cause: unknown) {
381 cleanup()
382 const upgradeError = new Socket.SocketError({
383 reason: new Socket.SocketUpgradeError({ cause })
384 })
385 close(upgradeError)
386 resume(Effect.fail(upgradeError))
387 }
388 function onUpgradeClose() {
389 failUpgrade(new Error("socket closed during TLS upgrade"))
390 }
391
392 tls.once(secureEvent, succeed)
393 tls.once("error", failUpgrade)
394 tls.once("close", onUpgradeClose)
395
396 return Effect.sync(() => {
397 cleanup()
398 close(
399 new Socket.SocketError({
400 reason: new Socket.SocketCloseError({ code: 1006 })
401 })
402 )
403 tls.destroy()
404 })
405 })
406 })
407
408 return { pull, upgrade }
409 }).pipe(
410 Effect.updateContext((input: Context.Context<Scope.Scope>) => Context.merge(openServices, input))
411 ) as Socket.Socket["reader"]
412
413 const awaitDrain = (conn: Duplex) =>
414 Effect.callback<void, Socket.SocketError>((resume) => {
415 function cleanup() {
416 conn.off("drain", onDrain)
417 conn.off("error", onError)
418 conn.off("close", onClose)
419 }
420 function onDrain() {
421 cleanup()
422 resume(Effect.void)
423 }
424 function onError(cause: Error) {
425 cleanup()
426 resume(Effect.fail(
427 new Socket.SocketError({
428 reason: new Socket.SocketWriteError({ cause })
429 })
430 ))
431 }
432 function onClose() {
433 cleanup()
434 resume(Effect.fail(
435 new Socket.SocketError({
436 reason: new Socket.SocketWriteError({ cause: new Error("socket closed") })
437 })
438 ))
439 }
440 conn.on("drain", onDrain)
441 conn.on("error", onError)
442 conn.on("close", onClose)
443 return Effect.sync(cleanup)
444 })
445
446 const write = (
447 chunk: Uint8Array | string | Socket.CloseEvent
448 ): Effect.Effect<void, Socket.SocketError> =>
449 Effect.suspend(() => {
450 const conn = currentSocket
451 if (conn === undefined) return latch.whenOpen(write(chunk))
452 if (Socket.isCloseEvent(chunk)) {
453 conn.destroy(chunk.code > 1000 ? new Error(`closed with code ${chunk.code}`) : undefined)
454 return Effect.void
455 }
456 try {
457 return conn.write(chunk) ? Effect.void : awaitDrain(conn)
458 } catch (cause) {
459 return Effect.fail(
460 new Socket.SocketError({
461 reason: new Socket.SocketWriteError({ cause })
462 })
463 )
464 }
465 })
466
467 const writeAll = (
468 chunks: Arr.NonEmptyReadonlyArray<Uint8Array | string>
469 ): Effect.Effect<void, Socket.SocketError> =>
470 Effect.suspend(() => {
471 const conn = currentSocket
472 if (conn === undefined) return latch.whenOpen(writeAll(chunks))
473 let needsDrain = false
474 try {
475 if (chunks.length === 1) {
476 needsDrain = !conn.write(chunks[0])
477 } else {
478 conn.cork()
479 try {
480 for (let i = 0; i < chunks.length; i++) {
481 needsDrain = !conn.write(chunks[i]) || needsDrain
482 }
483 } finally {
484 conn.uncork()
485 }
486 }
487 } catch (cause) {
488 return Effect.fail(
489 new Socket.SocketError({
490 reason: new Socket.SocketWriteError({ cause })
491 })
492 )
493 }
494 return needsDrain ? awaitDrain(conn) : Effect.void
495 })
496
497 const writer: Socket.Socket["writer"] = Effect.acquireRelease(
498 Effect.succeed({ write, writeAll }),
499 () =>
500 Effect.sync(() => {
501 if (!currentSocket || currentSocket.writableEnded) return
502 currentSocket.end()
503 })
504 )
505
506 return Effect.succeed(Socket.make({ reader, writer }))
507 })
508
509/**
510 * Creates a `Channel` over a TCP socket, reading arrays of `Uint8Array`
511 * chunks and writing arrays of bytes, strings, or socket close events.
512 *
513 * @category constructors
514 * @since 4.0.0
515 */
516export const makeNetChannel = <IE = never>(
517 options: Net.NetConnectOpts
518): Channel.Channel<
519 Arr.NonEmptyReadonlyArray<Uint8Array>,
520 Socket.SocketError | IE,
521 void,
522 Arr.NonEmptyReadonlyArray<Uint8Array | string | Socket.CloseEvent>,
523 IE
524> =>
525 Channel.unwrap(
526 Effect.map(makeNet(options), Socket.toChannelWith<IE>())
527 )
528
529/**
530 * Provides a `Socket.Socket` by opening a TCP connection with the supplied
531 * Node `net` connection options.
532 *
533 * @category layers
534 * @since 4.0.0
535 */
536export const layerNet: (options: Net.NetConnectOpts) => Layer.Layer<
537 Socket.Socket,
538 Socket.SocketError
539> = Function.flow(makeNet, Layer.effect(Socket.Socket))
540
541/**
542 * Opens a Node TLS connection as an Effect socket.
543 *
544 * **When to use**
545 *
546 * Use to create a `Socket.Socket` whose reader acquisition dials
547 * `tls.connect` and completes once the TLS handshake has finished.
548 *
549 * **Details**
550 *
551 * Accepts the same options as `tls.connect`, so trust anchors (`ca`), client
552 * certificates (`cert` / `key`), ALPN protocols, and `servername` are set
553 * there. A failed handshake, including an untrusted or expired peer
554 * certificate, fails with a `SocketOpenError`. Supports `openTimeout` and
555 * destroys the underlying socket when the reader scope is finalized.
556 *
557 * @category constructors
558 * @since 4.0.0
559 */
560export const makeTls = (
561 options: Tls.ConnectionOptions & {
562 readonly openTimeout?: Duration.Input | undefined
563 }
564): Effect.Effect<Socket.Socket> =>
565 fromDuplex(
566 Effect.contextWith((context: Context.Context<Scope.Scope>) => {
567 let conn: Tls.TLSSocket | undefined
568 let isOpen = false
569 return Effect.flatMap(
570 Scope.addFinalizer(
571 Context.get(context, Scope.Scope),
572 Effect.sync(() => {
573 if (!conn) return
574 closeSocket(conn, isOpen)
575 })
576 ),
577 () =>
578 Effect.callback<Tls.TLSSocket, Socket.SocketError, never>((resume) => {
579 conn = Tls.connect(options)
580 conn.once("secureConnect", () => {
581 isOpen = true
582 resume(Effect.succeed(conn!))
583 })
584 conn.on("error", (cause: Error) => {
585 resume(Effect.fail(
586 new Socket.SocketError({
587 reason: new Socket.SocketOpenError({ kind: "Unknown", cause })
588 })
589 ))
590 })
591 })
592 )
593 }),
594 options
595 )
596
597/**
598 * Creates a `Channel` over a TLS socket, reading arrays of `Uint8Array`
599 * chunks and writing arrays of bytes, strings, or socket close events.
600 *
601 * @category constructors
602 * @since 4.0.0
603 */
604export const makeTlsChannel = <IE = never>(
605 options: Tls.ConnectionOptions
606): Channel.Channel<
607 Arr.NonEmptyReadonlyArray<Uint8Array>,
608 Socket.SocketError | IE,
609 void,
610 Arr.NonEmptyReadonlyArray<Uint8Array | string | Socket.CloseEvent>,
611 IE
612> =>
613 Channel.unwrap(
614 Effect.map(makeTls(options), Socket.toChannelWith<IE>())
615 )
616
617/**
618 * Provides a `Socket.Socket` by opening a TLS connection with the supplied
619 * Node `tls` connection options.
620 *
621 * @category layers
622 * @since 4.0.0
623 */
624export const layerTls: (options: Tls.ConnectionOptions) => Layer.Layer<
625 Socket.Socket,
626 Socket.SocketError
627> = Function.flow(makeTls, Layer.effect(Socket.Socket))