1/**
2 * Native Deno socket adapters for Effect sockets.
3 *
4 * This module uses Deno-specific names: `makeTcp`, `makeTcpChannel`,
5 * `layerTcp`, and `fromConn` correspond to the Node platform's net and duplex
6 * APIs. Deno does not support `keepAliveInitialDelay`, and the `noDelay` and
7 * `keepAlive` options have no effect on Unix connections. A `CloseEvent` always
8 * closes gracefully because Deno has no equivalent of Node's reset-on-close.
9 *
10 * @since 4.0.0
11 */
12import * as Arr from "effect/Array"
13import * as Channel from "effect/Channel"
14import * as Context from "effect/Context"
15import type * as Duration from "effect/Duration"
16import * as Effect from "effect/Effect"
17import * as Function from "effect/Function"
18import { constVoid } from "effect/Function"
19import * as Latch from "effect/Latch"
20import * as Layer from "effect/Layer"
21import * as Scope from "effect/Scope"
22import * as Socket from "effect/socket/Socket"
23
24/**
25 * Options for opening a TCP or Unix connection.
26 *
27 * @category options
28 * @since 4.0.0
29 */
30export type ConnectOptions = (Deno.ConnectOptions | Deno.UnixConnectOptions) & {
31 readonly noDelay?: boolean | undefined
32 readonly keepAlive?: boolean | undefined
33}
34
35/**
36 * Options for opening a TCP or Unix connection.
37 *
38 * @category options
39 * @since 4.0.0
40 */
41export type TcpOptions = ConnectOptions & {
42 readonly openTimeout?: Duration.Input | undefined
43}
44
45/**
46 * Service tag for the underlying Deno connection.
47 *
48 * @category services
49 * @since 4.0.0
50 */
51export class Conn extends Context.Service<Conn, Deno.Conn>()(
52 "@effect/platform-deno/DenoSocket/Conn"
53) {}
54
55/**
56 * Adapts a Deno connection into an Effect socket.
57 *
58 * **Details**
59 *
60 * Readers over TCP connections can upgrade in place with `Deno.startTls`.
61 * Deno exposes only the client side of this transition; Unix connections fail
62 * the upgrade with a `SocketUpgradeError`.
63 *
64 * `Deno.startTls` does not support client certificates, so `key`, `cert`,
65 * `passphrase`, and `requestCert` do not affect Deno upgrades. Setting
66 * `rejectUnauthorized` to `false` disables hostname verification only;
67 * certificate-chain validation still applies.
68 *
69 * @category constructors
70 * @since 4.0.0
71 */
72export const fromConn = <RO>(
73 open: Effect.Effect<Deno.Conn, Socket.SocketError, RO>
74): Effect.Effect<Socket.Socket, never, Exclude<RO, Scope.Scope>> =>
75 Effect.withFiber<Socket.Socket, never, Exclude<RO, Scope.Scope>>((fiber) => {
76 let current: {
77 readonly conn: Deno.Conn
78 readonly writer: WritableStreamDefaultWriter<Uint8Array>
79 readonly fail: (error: Socket.SocketError) => void
80 } | undefined
81 let tearingDown = false
82 let writeClosed = false
83 const latch = Latch.makeUnsafe(false)
84 const openServices = fiber.context as Context.Context<RO>
85
86 const reader: Socket.Socket["reader"] = Effect.gen(function*() {
87 const scope = yield* Effect.scope
88 let conn: Deno.Conn | undefined
89 let upgradeAvailable = true
90 yield* Scope.addFinalizer(
91 scope,
92 Effect.suspend(() => {
93 tearingDown = true
94 latch.closeUnsafe()
95 current = undefined
96 upgradeAvailable = false
97 return conn === undefined ? Effect.void : close(conn)
98 })
99 )
100 conn = yield* Scope.provide(open, scope)
101 let readerHandle = conn.readable.getReader()
102 let writerHandle = conn.writable.getWriter()
103
104 let error: Socket.SocketError | undefined
105 const fail = (err: Socket.SocketError) => {
106 if (error === undefined) error = err
107 tearingDown = true
108 readerHandle.cancel().catch(constVoid)
109 }
110 current = {
111 conn,
112 writer: writerHandle,
113 fail
114 }
115 tearingDown = false
116 if (writeClosed) {
117 writeClosed = false
118 writerHandle.releaseLock()
119 yield* closeWrite(conn)
120 }
121 latch.openUnsafe()
122
123 const read = Effect.tryPromise({
124 try: () => readerHandle.read(),
125 catch: (cause) =>
126 tearingDown && isTeardownError(cause)
127 ? error ?? new Socket.SocketError({
128 reason: new Socket.SocketCloseError({ code: 1000 })
129 })
130 : new Socket.SocketError({
131 reason: new Socket.SocketReadError({ cause })
132 })
133 })
134 const upgrade: Socket.Reader["upgrade"] = (options = {}) =>
135 Effect.suspend(() => {
136 if (!upgradeAvailable) {
137 return Effect.fail(
138 new Socket.SocketError({
139 reason: new Socket.SocketUpgradeError({
140 cause: new Error("socket is already upgraded or closed")
141 })
142 })
143 )
144 }
145 if (conn!.remoteAddr.transport !== "tcp") {
146 return Effect.fail(
147 new Socket.SocketError({
148 reason: new Socket.SocketUpgradeError({
149 cause: new Error("TLS upgrade requires a TCP connection")
150 })
151 })
152 )
153 }
154 upgradeAvailable = false
155 const raw = conn as Deno.TcpConn
156 readerHandle.releaseLock()
157 writerHandle.releaseLock()
158 return Effect.tryPromise({
159 try: async () => {
160 const tls = await Deno.startTls(raw, {
161 hostname: raw.remoteAddr.hostname,
162 caCerts: options.ca === undefined ? [] : toStrings(options.ca),
163 unsafelyDisableHostnameVerification: options.rejectUnauthorized === false,
164 ...(options.alpnProtocols === undefined ? {} : { alpnProtocols: [...options.alpnProtocols] })
165 })
166 await tls.handshake()
167 return tls
168 },
169 catch: (cause) =>
170 new Socket.SocketError({
171 reason: new Socket.SocketUpgradeError({ cause })
172 })
173 }).pipe(
174 Effect.tap((tls) =>
175 Effect.sync(() => {
176 conn = tls
177 readerHandle = tls.readable.getReader()
178 writerHandle = tls.writable.getWriter()
179 current = { conn: tls, writer: writerHandle, fail }
180 })
181 ),
182 Effect.tapError((upgradeError) => Effect.sync(() => fail(upgradeError))),
183 Effect.asVoid
184 )
185 })
186 return {
187 pull: Effect.suspend(() => {
188 if (error !== undefined) return Effect.fail(error)
189 return Effect.flatMap(read, ({ done, value }) =>
190 done
191 ? Effect.fail(
192 error ?? new Socket.SocketError({
193 reason: new Socket.SocketCloseError({ code: 1000 })
194 })
195 )
196 : Effect.succeed([value] as const))
197 }),
198 upgrade
199 }
200 }).pipe(
201 Effect.updateContext((input: Context.Context<Scope.Scope>) => Context.merge(openServices, input))
202 ) as Socket.Socket["reader"]
203
204 const write = (chunk: Uint8Array | string | Socket.CloseEvent) =>
205 latch.whenOpen(Effect.suspend(() => {
206 const { conn, fail, writer } = current!
207 if (Socket.isCloseEvent(chunk)) {
208 fail(
209 new Socket.SocketError({
210 reason: new Socket.SocketCloseError({ code: chunk.code, closeReason: chunk.reason })
211 })
212 )
213 return close(conn)
214 }
215 const bytes = typeof chunk === "string" ? encoder.encode(chunk) : chunk
216 return Effect.tryPromise({
217 try: () => writer.ready.then(() => writer.write(bytes)),
218 catch: (cause) =>
219 new Socket.SocketError({
220 reason: new Socket.SocketWriteError({ cause })
221 })
222 })
223 }))
224
225 const writeAll = (chunks: Arr.NonEmptyReadonlyArray<Uint8Array | string>) =>
226 latch.whenOpen(Effect.tryPromise({
227 try: async () => {
228 const { writer } = current!
229 for (let i = 0; i < chunks.length; i++) {
230 const chunk = chunks[i]
231 const bytes = typeof chunk === "string" ? encoder.encode(chunk) : chunk
232 await writer.ready
233 await writer.write(bytes)
234 }
235 },
236 catch: (cause) =>
237 new Socket.SocketError({
238 reason: new Socket.SocketWriteError({ cause })
239 })
240 }))
241
242 const writer: Socket.Socket["writer"] = Effect.acquireRelease(
243 Effect.sync(() => {
244 writeClosed = false
245 return { write, writeAll }
246 }),
247 () =>
248 Effect.suspend(() => {
249 if (current === undefined) {
250 writeClosed = true
251 return Effect.void
252 }
253 const { conn, writer } = current
254 writer.releaseLock()
255 return closeWrite(conn)
256 })
257 )
258
259 return Effect.succeed(Socket.make({ reader, writer }))
260 })
261
262/**
263 * Opens a native Deno TCP or Unix connection as an Effect socket.
264 *
265 * **Details**
266 *
267 * An `openTimeout` interrupts acquisition but cannot cancel the in-flight
268 * `Deno.connect` promise. The scope finalizer closes a connection that arrives
269 * after the timeout.
270 *
271 * @category constructors
272 * @since 4.0.0
273 */
274export const makeTcp = (options: TcpOptions): Effect.Effect<Socket.Socket> => {
275 const { keepAlive, noDelay, openTimeout, ...connectOptions } = options
276 const acquire = Effect.contextWith((context: Context.Context<Scope.Scope>) => {
277 let conn: Deno.Conn | undefined
278 let finalized = false
279 return Scope.addFinalizer(
280 Context.get(context, Scope.Scope),
281 Effect.suspend(() => {
282 finalized = true
283 return conn === undefined ? Effect.void : close(conn)
284 })
285 ).pipe(
286 Effect.andThen(Effect.tryPromise({
287 try: () => {
288 const connecting = connectOptions.transport === "unix"
289 ? Deno.connect(connectOptions)
290 : Deno.connect(connectOptions)
291 return connecting.then((connection) => {
292 conn = connection
293 if (finalized) connection.close()
294 return connection
295 })
296 },
297 catch: (cause) =>
298 new Socket.SocketError({
299 reason: new Socket.SocketOpenError({ kind: "Unknown", cause })
300 })
301 })),
302 Effect.tap((connection) =>
303 Effect.sync(() => {
304 if ("setNoDelay" in connection && noDelay !== undefined) {
305 connection.setNoDelay(noDelay)
306 }
307 if ("setKeepAlive" in connection && keepAlive !== undefined) {
308 connection.setKeepAlive(keepAlive)
309 }
310 })
311 )
312 )
313 }).pipe(
314 openTimeout === undefined
315 ? Function.identity
316 : Effect.timeoutOrElse({
317 duration: openTimeout,
318 orElse: () =>
319 Effect.fail(
320 new Socket.SocketError({
321 reason: new Socket.SocketOpenError({ kind: "Timeout", cause: new Error("Connection timed out") })
322 })
323 )
324 })
325 )
326 return fromConn(acquire)
327}
328
329/**
330 * Creates a channel over a native Deno TCP or Unix connection.
331 *
332 * @category constructors
333 * @since 4.0.0
334 */
335export const makeTcpChannel = <IE = never>(
336 options: ConnectOptions
337): Channel.Channel<
338 Arr.NonEmptyReadonlyArray<Uint8Array>,
339 Socket.SocketError | IE,
340 void,
341 Arr.NonEmptyReadonlyArray<Uint8Array | string | Socket.CloseEvent>,
342 IE
343> => Channel.unwrap(Effect.map(makeTcp(options), Socket.toChannelWith<IE>()))
344
345/**
346 * Provides a socket by opening a native Deno TCP or Unix connection.
347 *
348 * @category layers
349 * @since 4.0.0
350 */
351export const layerTcp: (options: ConnectOptions) => Layer.Layer<
352 Socket.Socket,
353 Socket.SocketError
354> = Function.flow(makeTcp, Layer.effect(Socket.Socket))
355
356/**
357 * Creates a socket layer connected to a URL with Deno's global WebSocket.
358 *
359 * @category layers
360 * @since 4.0.0
361 */
362export const layerWebSocket = (url: string, options?: {
363 readonly openTimeout?: Duration.Input | undefined
364 readonly protocols?: string | globalThis.Array<string> | undefined
365 readonly highWaterMark?: number | undefined
366}): Layer.Layer<Socket.Socket> =>
367 Layer.effect(Socket.Socket, Socket.makeWebSocket(url, options)).pipe(
368 Layer.provide(layerWebSocketConstructor)
369 )
370
371/**
372 * Provides the WebSocket constructor backed by `globalThis.WebSocket`.
373 *
374 * @category layers
375 * @since 4.0.0
376 */
377export const layerWebSocketConstructor: Layer.Layer<Socket.WebSocketConstructor> =
378 Socket.layerWebSocketConstructorGlobal
379
380const encoder = new TextEncoder()
381const decoder = new TextDecoder()
382
383const toStrings = (values: string | Uint8Array | ReadonlyArray<string | Uint8Array>): Array<string> =>
384 Arr.map(Arr.ensure(values), (value) => typeof value === "string" ? value : decoder.decode(value))
385
386const isBadResource = (cause: unknown): cause is Deno.errors.BadResource => cause instanceof Deno.errors.BadResource
387
388const isTeardownError = (cause: unknown): boolean => cause instanceof Deno.errors.Interrupted || isBadResource(cause)
389
390const close = (conn: Deno.Conn): Effect.Effect<void> =>
391 Effect.try(() => conn.close()).pipe(
392 Effect.catch(({ cause }) => isBadResource(cause) ? Effect.void : Effect.die(cause))
393 )
394
395const closeWrite = (conn: Deno.Conn): Effect.Effect<void> =>
396 Effect.tryPromise(() => conn.closeWrite()).pipe(
397 Effect.catch((error) => isBadResource(error) ? Effect.void : Effect.die(error))
398 )