1/**
2 * Node socket server adapters for Effect's unstable socket server API.
3 *
4 * This module turns `node:net` TCP or Unix-domain servers, `node:tls` servers,
5 * and `ws` WebSocket servers into scoped `SocketServer.SocketServer` services.
6 * Use the TCP and TLS constructors when handlers should receive a
7 * `Socket.Socket` backed by a Node `net.Socket`; use the WebSocket
8 * constructors when handlers should receive a `Socket.Socket` backed by `ws`
9 * and have access to the per-connection WebSocket and `IncomingMessage`
10 * services.
11 *
12 * @since 4.0.0
13 */
14import type { Cause } from "effect/Cause"
15import * as Context from "effect/Context"
16import * as Deferred from "effect/Deferred"
17import * as Effect from "effect/Effect"
18import * as Exit from "effect/Exit"
19import * as Fiber from "effect/Fiber"
20import { pipe } from "effect/Function"
21import * as Function from "effect/Function"
22import * as Layer from "effect/Layer"
23import * as NetAddress from "effect/net/NetAddress"
24import * as References from "effect/References"
25import * as Scope from "effect/Scope"
26import * as Socket from "effect/socket/Socket"
27import * as SocketServer from "effect/socket/SocketServer"
28import type * as Http from "node:http"
29import * as Net from "node:net"
30import * as Tls from "node:tls"
31import * as NodeSocket from "./NodeSocket.ts"
32import { NodeWS } from "./NodeSocket.ts"
33
34const isDeno = "Deno" in globalThis
35
36const socketAddressFromNode = (
37 address: string | Net.AddressInfo
38): Effect.Effect<NetAddress.SocketAddress, SocketServer.SocketServerError> =>
39 typeof address === "string"
40 ? Effect.succeed(NetAddress.unixPathAddress(address))
41 : Effect.fromResult(NetAddress.inetAddressFromIpString(address.address, address.port)).pipe(
42 Effect.mapError((cause) =>
43 new SocketServer.SocketServerError({
44 reason: new SocketServer.SocketServerOpenError({ cause })
45 })
46 )
47 )
48
49/**
50 * Service tag for the Node `IncomingMessage` associated with the current
51 * WebSocket server connection.
52 *
53 * @category services
54 * @since 4.0.0
55 */
56export class IncomingMessage extends Context.Service<
57 IncomingMessage,
58 Http.IncomingMessage
59>()("@effect/platform-node-shared/NodeSocketServer/IncomingMessage") {}
60
61/**
62 * Creates a scoped TCP `SocketServer` from a Node `net.Server`, starts
63 * listening with the supplied options, queues pending connections until `run`
64 * is called, and closes the server when the scope ends.
65 *
66 * @category constructors
67 * @since 4.0.0
68 */
69export const make = (
70 options: Net.ServerOpts & Net.ListenOptions
71): Effect.Effect<
72 SocketServer.SocketServer["Service"],
73 SocketServer.SocketServerError,
74 Scope.Scope
75> =>
76 makeNetServer({
77 listenOptions: options,
78 connectionEvent: "connection",
79 createServer: () => Net.createServer(options)
80 })
81
82/**
83 * Provides a TCP `SocketServer` by creating and managing a scoped Node
84 * `net.Server` with the supplied server and listen options.
85 *
86 * @category layers
87 * @since 4.0.0
88 */
89export const layer: (
90 options: Net.ServerOpts & Net.ListenOptions
91) => Layer.Layer<
92 SocketServer.SocketServer,
93 SocketServer.SocketServerError
94> = Function.flow(make, Layer.effect(SocketServer.SocketServer))
95
96/**
97 * Creates a scoped TLS `SocketServer` from a Node `tls.Server`, starts
98 * listening with the supplied options, queues pending connections until `run`
99 * is called, and closes the server when the scope ends.
100 *
101 * **Details**
102 *
103 * Connections reach `run` only once their handshake has completed, and the
104 * handler receives a `Socket.Socket` backed by the `tls.TLSSocket`, which is
105 * also provided as `NodeSocket.NetSocket`. Connections that fail the handshake
106 * are destroyed and the server keeps listening.
107 *
108 * @category constructors
109 * @since 4.0.0
110 */
111export const makeTls = (
112 options: Tls.TlsOptions & Net.ListenOptions
113): Effect.Effect<
114 SocketServer.SocketServer["Service"],
115 SocketServer.SocketServerError,
116 Scope.Scope
117> =>
118 makeNetServer({
119 listenOptions: options,
120 connectionEvent: "secureConnection",
121 createServer: () => {
122 const server = Tls.createServer(options)
123 server.on("tlsClientError", (_error, socket) => {
124 socket.destroy()
125 })
126 return server
127 }
128 })
129
130/**
131 * Provides a TLS `SocketServer` by creating and managing a scoped Node
132 * `tls.Server` with the supplied TLS and listen options.
133 *
134 * @category layers
135 * @since 4.0.0
136 */
137export const layerTls: (
138 options: Tls.TlsOptions & Net.ListenOptions
139) => Layer.Layer<
140 SocketServer.SocketServer,
141 SocketServer.SocketServerError
142> = Function.flow(makeTls, Layer.effect(SocketServer.SocketServer))
143
144/**
145 * Creates a scoped WebSocket `SocketServer` backed by the `ws` package,
146 * providing the WebSocket and its Node `IncomingMessage` to connection
147 * handlers and closing the server when the scope ends.
148 *
149 * @stability unstable
150 * @category constructors
151 * @since 4.0.0
152 */
153export const makeWebSocket: (
154 options: NodeWS.ServerOptions<typeof NodeWS.WebSocket, typeof Http.IncomingMessage>
155) => Effect.Effect<
156 SocketServer.SocketServer["Service"],
157 SocketServer.SocketServerError,
158 Scope.Scope
159> = Effect.fnUntraced(function*(
160 options: NodeWS.ServerOptions
161) {
162 const pendingConnections = new Map<
163 NodeWS.WebSocket,
164 readonly [request: Http.IncomingMessage, remove: () => void]
165 >()
166 const server = yield* Effect.acquireRelease(
167 Effect.sync(() => new NodeWS.WebSocketServer(options)),
168 (server) =>
169 Effect.callback<void>((resume) => {
170 pendingConnections.forEach(([, remove], conn) => {
171 remove()
172 conn.terminate()
173 })
174 server.close(() => resume(Effect.void))
175 })
176 )
177 function defaultHandler(conn: NodeWS.WebSocket, req: Http.IncomingMessage) {
178 const remove = () => {
179 pendingConnections.delete(conn)
180 conn.removeEventListener("close", remove)
181 }
182 pendingConnections.set(conn, [req, remove])
183 conn.addEventListener("close", remove)
184 }
185 let onConnection = defaultHandler
186 server.on("connection", (conn, req) => {
187 // pause immediately so nothing is dropped before a handler acquires the
188 // socket's reader
189 conn.pause()
190 onConnection(conn, req as Http.IncomingMessage)
191 })
192
193 yield* Effect.callback<void, SocketServer.SocketServerError>((resume) => {
194 server.once("error", (error) => {
195 resume(Effect.fail(
196 new SocketServer.SocketServerError({
197 reason: new SocketServer.SocketServerOpenError({
198 cause: error
199 })
200 })
201 ))
202 })
203 server.once("listening", () => {
204 resume(Effect.void)
205 })
206 })
207
208 const run = Effect.fnUntraced(function*<R, E, _>(handler: (socket: Socket.Socket) => Effect.Effect<_, E, R>) {
209 const scope = yield* Scope.make()
210 const services = Context.omit(Scope.Scope)(yield* Effect.context<R>()) as Context.Context<R>
211 const trackFiber = Fiber.runIn(scope)
212 const prevOnConnection = onConnection
213 onConnection = function(conn: NodeWS.WebSocket, req: Http.IncomingMessage) {
214 let context = services
215 context = Context.add(context, IncomingMessage, req)
216 context = Context.add(context, Socket.WebSocket, conn)
217 pipe(
218 Socket.fromWebSocket(
219 Effect.acquireRelease(
220 Effect.succeed(conn),
221 (conn) =>
222 Effect.sync(() => {
223 conn.close()
224 })
225 )
226 ),
227 Effect.flatMap(handler),
228 Effect.catchCause(reportUnhandledError),
229 Effect.runForkWith(context),
230 trackFiber
231 )
232 }
233 pendingConnections.forEach(([req, remove], conn) => {
234 remove()
235 onConnection(conn, req)
236 })
237 return yield* Effect.callback<never>((_resume) => {
238 return Effect.sync(() => {
239 onConnection = prevOnConnection
240 })
241 }).pipe(
242 Effect.ensuring(Scope.close(scope, Exit.void))
243 )
244 })
245
246 const boundAddress = yield* socketAddressFromNode(server.address()!)
247 return SocketServer.SocketServer.of({
248 address: boundAddress,
249 run
250 })
251})
252
253/**
254 * Provides a WebSocket `SocketServer` backed by the `ws` package and managed
255 * with the supplied server options.
256 *
257 * @stability unstable
258 * @category layers
259 * @since 4.0.0
260 */
261export const layerWebSocket: (
262 options: NodeSocket.NodeWS.ServerOptions<typeof NodeSocket.NodeWS.WebSocket, typeof Http.IncomingMessage>
263) => Layer.Layer<
264 SocketServer.SocketServer,
265 SocketServer.SocketServerError
266> = Function.flow(makeWebSocket, Layer.effect(SocketServer.SocketServer))
267
268const reportUnhandledError = <E>(cause: Cause<E>) =>
269 Effect.withFiber<void>((fiber) => {
270 const unhandledLogLevel = fiber.getRef(References.UnhandledLogLevel)
271 if (unhandledLogLevel) {
272 return Effect.logWithLevel(unhandledLogLevel)(cause, "Unhandled error in SocketServer")
273 }
274 return Effect.void
275 })
276
277const makeNetServer = Effect.fnUntraced(function*(options: {
278 readonly createServer: () => Net.Server
279 readonly connectionEvent: "connection" | "secureConnection"
280 readonly listenOptions: Net.ListenOptions
281}) {
282 const errorDeferred = Deferred.makeUnsafe<never, Error>()
283 const pending = new Map<Net.Socket, () => void>()
284 function defaultOnConnection(conn: Net.Socket) {
285 const remove = () => {
286 pending.delete(conn)
287 conn.off("close", remove)
288 conn.off("error", remove)
289 }
290 pending.set(conn, remove)
291 conn.on("close", remove)
292 conn.on("error", remove)
293 }
294 let onConnection = defaultOnConnection
295 // oxlint-disable-next-line prefer-const
296 let server: Net.Server | undefined
297 yield* Effect.addFinalizer(() =>
298 Effect.callback<void>((resume) => {
299 pending.forEach((remove, conn) => {
300 remove()
301 conn.destroy()
302 })
303 server?.close(() => resume(Effect.void))
304 })
305 )
306 server = options.createServer()
307 server.on(options.connectionEvent, (conn: Net.Socket) => {
308 // pause immediately so nothing is dropped before a handler acquires the
309 // socket's reader. Deno's node:net compatibility layer leaves the socket
310 // non-flowing but breaks `readable` events after an explicit pause.
311 if (!isDeno) conn.pause()
312 onConnection(conn)
313 })
314 server.on("error", (err) => Deferred.doneUnsafe(errorDeferred, Exit.fail(err)))
315
316 yield* Effect.callback<void, SocketServer.SocketServerError>((resume) => {
317 server.listen(options.listenOptions, () => resume(Effect.void))
318 }).pipe(
319 Effect.raceFirst(Effect.mapError(Deferred.await(errorDeferred), (err) =>
320 new SocketServer.SocketServerError({
321 reason: new SocketServer.SocketServerOpenError({
322 cause: err
323 })
324 })))
325 )
326
327 const run = Effect.fnUntraced(function*<R, E, _>(handler: (socket: Socket.Socket) => Effect.Effect<_, E, R>) {
328 const scope = yield* Scope.make()
329 const services = Context.omit(Scope.Scope)(yield* Effect.context<R>()) as Context.Context<R>
330 const trackFiber = Fiber.runIn(scope)
331 const prevOnConnection = onConnection
332 onConnection = function(conn: Net.Socket) {
333 let error: Error | undefined
334 conn.on("error", (err) => {
335 error = err
336 })
337 pipe(
338 NodeSocket.fromDuplex(
339 Effect.acquireRelease(
340 Effect.suspend((): Effect.Effect<Net.Socket, Socket.SocketError> => {
341 if (error) {
342 return Effect.fail(
343 new Socket.SocketError({
344 reason: new Socket.SocketOpenError({
345 kind: "Unknown",
346 cause: error
347 })
348 })
349 )
350 } else if (conn.closed) {
351 return Effect.fail(
352 new Socket.SocketError({
353 reason: new Socket.SocketCloseError({ code: 1000 })
354 })
355 )
356 }
357 return Effect.succeed(conn)
358 }),
359 (conn) =>
360 Effect.sync(() => {
361 if (conn.closed === false) {
362 conn.destroySoon()
363 }
364 })
365 ),
366 { tlsServer: true }
367 ),
368 Effect.flatMap(handler),
369 Effect.catchCause(reportUnhandledError),
370 Effect.runForkWith(Context.add(services, NodeSocket.NetSocket, conn)),
371 trackFiber
372 )
373 }
374 pending.forEach((remove, conn) => {
375 remove()
376 onConnection(conn)
377 })
378 return yield* Effect.callback<never>((_resume) => {
379 return Effect.suspend(() => {
380 onConnection = prevOnConnection
381 return Scope.close(scope, Exit.void)
382 })
383 })
384 })
385
386 const boundAddress = yield* socketAddressFromNode(server.address()!)
387 return SocketServer.SocketServer.of({
388 address: boundAddress,
389 run
390 })
391})