Add Deno socket server support (#6724)

Author: tim-smart

.changeset/add-deno-socket-server.md

@@ -0,0 +1,5 @@
+---
+"@effect/platform-deno": patch
+---
+
+Add native Deno TCP, Unix, and TLS socket server adapters.

packages/platform-deno/src/DenoSocketServer.ts

@@ -0,0 +1,120 @@
+/**
+ * Native Deno TCP and Unix socket servers for Effect's unstable socket server
+ * API.
+ *
+ * The plain `make` and `layer` names intentionally differ from
+ * `DenoSocket.makeTcp` and `DenoSocket.layerTcp` because this module has only
+ * one server kind. Deno has no standalone WebSocket server, so WebSocket
+ * constructors are omitted, and there is no `IncomingMessage` analogue.
+ * Connections made before `run` starts remain in the kernel backlog instead of
+ * being accepted and buffered as they are by the Node implementation. A second
+ * concurrent `run` waits for the first because listener access is guarded by a
+ * semaphore.
+ *
+ * Deno-native TLS server constructors are included even though the Node socket
+ * server has no corresponding dedicated constructors. TLS is deliberately not
+ * tested here because the repository has no certificate fixture.
+ *
+ * @since 4.0.0
+ */
+import * as Effect from "effect/Effect"
+import * as Function from "effect/Function"
+import * as Layer from "effect/Layer"
+import type * as Scope from "effect/Scope"
+import * as SocketServer from "effect/unstable/socket/SocketServer"
+import { closeListener, fromListener } from "./internal/denoSocketServer.ts"
+
+/**
+ * Native Deno options for listening on a TCP or Unix socket.
+ *
+ * @category models
+ * @since 4.0.0
+ */
+export type ListenOptions =
+  | (Deno.TcpListenOptions & { transport?: "tcp" })
+  | (Deno.UnixListenOptions & { transport: "unix" })
+
+/**
+ * Native Deno options and certified key material for listening with TLS.
+ *
+ * @category models
+ * @since 4.0.0
+ */
+export type TlsListenOptions = Deno.ListenTlsOptions & Deno.TlsCertifiedKeyPem
+
+/**
+ * Creates a scoped socket server using a native Deno TCP or Unix listener.
+ *
+ * @category constructors
+ * @since 4.0.0
+ */
+export const make: (
+  options: ListenOptions
+) => Effect.Effect<
+  SocketServer.SocketServer["Service"],
+  SocketServer.SocketServerError,
+  Scope.Scope
+> = Effect.fnUntraced(function*(options) {
+  const listener = yield* Effect.acquireRelease(
+    Effect.try({
+      try: () => options.transport === "unix" ? Deno.listen(options) : Deno.listen(options),
+      catch: openError
+    }),
+    closeListener
+  )
+  return fromListener(listener)
+})
+
+/**
+ * Provides a socket server using a scoped native Deno TCP or Unix listener.
+ *
+ * @category layers
+ * @since 4.0.0
+ */
+export const layer: (
+  options: ListenOptions
+) => Layer.Layer<SocketServer.SocketServer, SocketServer.SocketServerError> = Function.flow(
+  make,
+  Layer.effect(SocketServer.SocketServer)
+)
+
+/**
+ * Creates a scoped TLS socket server using a native Deno TLS listener.
+ *
+ * @category constructors
+ * @since 4.0.0
+ */
+export const makeTls: (
+  options: TlsListenOptions
+) => Effect.Effect<
+  SocketServer.SocketServer["Service"],
+  SocketServer.SocketServerError,
+  Scope.Scope
+> = Effect.fnUntraced(function*(options) {
+  const listener = yield* Effect.acquireRelease(
+    Effect.try({
+      try: () => Deno.listenTls(options),
+      catch: openError
+    }),
+    closeListener
+  )
+  return fromListener(listener)
+})
+
+/**
+ * Provides a TLS socket server using a scoped native Deno TLS listener.
+ *
+ * @category layers
+ * @since 4.0.0
+ */
+export const layerTls: (
+  options: TlsListenOptions
+) => Layer.Layer<SocketServer.SocketServer, SocketServer.SocketServerError> = Function.flow(
+  makeTls,
+  Layer.effect(SocketServer.SocketServer)
+)
+
+const openError = (cause: unknown) =>
+  new SocketServer.SocketServerError({
+    reason: new SocketServer.SocketServerOpenError({ cause })
+  })

packages/platform-deno/src/index.ts

@@ -59,6 +59,11 @@ export * as DenoServices from "./DenoServices.ts"
  */
 export * as DenoSocket from "./DenoSocket.ts"
 
+/**
+ * @since 4.0.0
+ */
+export * as DenoSocketServer from "./DenoSocketServer.ts"
+
 /**
  * @since 4.0.0
  */

packages/platform-deno/src/internal/denoSocketServer.ts

@@ -0,0 +1,89 @@
+import * as Cause from "effect/Cause"
+import * as Context from "effect/Context"
+import * as Effect from "effect/Effect"
+import * as Exit from "effect/Exit"
+import * as Fiber from "effect/Fiber"
+import { pipe } from "effect/Function"
+import * as References from "effect/References"
+import * as Scope from "effect/Scope"
+import * as Semaphore from "effect/Semaphore"
+import type * as Socket from "effect/unstable/socket/Socket"
+import * as SocketServer from "effect/unstable/socket/SocketServer"
+import * as DenoSocket from "../DenoSocket.ts"
+
+const initialBackoffMillis = 10
+const maximumBackoffMillis = 1_000
+
+type Listener = Deno.Listener<Deno.Conn, Deno.NetAddr | Deno.UnixAddr>
+
+export const fromListener = (listener: Listener): SocketServer.SocketServer["Service"] => {
+  const semaphore = Semaphore.makeUnsafe(1)
+
+  const run = <R, E, _>(handler: (socket: Socket.Socket) => Effect.Effect<_, E, R>) =>
+    semaphore.withPermit(Effect.gen(function*() {
+      const scope = yield* Scope.make()
+      const services = Context.omit(Scope.Scope)(yield* Effect.context<R>()) as Context.Context<R>
+      const trackFiber = Fiber.runIn(scope)
+      let failures = 0
+
+      const loop: Effect.Effect<never> = Effect.suspend(() =>
+        Effect.tryPromise(() => listener.accept()).pipe(
+          Effect.matchEffect({
+            onFailure: (error) => {
+              const cause = error.cause
+              if (isTeardownError(cause)) return Effect.never
+              const backoff = Math.min(initialBackoffMillis * 2 ** failures++, maximumBackoffMillis)
+              return reportUnhandledError(Cause.fail(cause)).pipe(
+                Effect.andThen(Effect.sleep(backoff)),
+                Effect.andThen(loop)
+              )
+            },
+            onSuccess: (conn) => {
+              failures = 0
+              pipe(
+                DenoSocket.fromConn(Effect.acquireRelease(Effect.succeed(conn), closeConn)),
+                Effect.flatMap(handler),
+                Effect.ensuring(closeConn(conn)),
+                Effect.catchCause(reportUnhandledError),
+                Effect.runForkWith(Context.add(services, DenoSocket.Conn, conn)),
+                trackFiber
+              )
+              return loop
+            }
+          })
+        )
+      )
+
+      return yield* loop.pipe(Effect.ensuring(Scope.close(scope, Exit.void)))
+    }))
+
+  const address = listener.addr
+  return SocketServer.SocketServer.of({
+    address: "path" in address
+      ? { _tag: "UnixAddress", path: address.path }
+      : { _tag: "TcpAddress", hostname: address.hostname, port: address.port },
+    run
+  })
+}
+
+export const closeListener = (listener: Listener): Effect.Effect<void> =>
+  Effect.try(() => listener.close()).pipe(
+    Effect.catch(({ cause }) => isTeardownError(cause) ? Effect.void : Effect.die(cause))
+  )
+
+const closeConn = (conn: Deno.Conn): Effect.Effect<void> =>
+  Effect.try(() => conn.close()).pipe(
+    Effect.catch(({ cause }) => isTeardownError(cause) ? Effect.void : Effect.die(cause))
+  )
+
+const isTeardownError = (cause: unknown): boolean =>
+  cause instanceof Deno.errors.BadResource || cause instanceof Deno.errors.Interrupted
+
+const reportUnhandledError = <E>(cause: Cause.Cause<E>) =>
+  Effect.withFiber<void>((fiber) => {
+    const unhandledLogLevel = fiber.getRef(References.UnhandledLogLevel)
+    if (unhandledLogLevel) {
+      return Effect.logWithLevel(unhandledLogLevel)(cause, "Unhandled error in SocketServer")
+    }
+    return Effect.void
+  })