fix: continue with setup-server, remote

Author: kettanaito

src/browser/setup-worker.ts

@@ -72,7 +72,7 @@ export function setupWorker(
            * Service Workers do not intercept WebSocket connections.
            */
           new InterceptorSource({
-            interceptors: [new WebSocketInterceptor()],
+            interceptors: [new WebSocketInterceptor() as any],
           }),
         ],
         handlers,

src/core/new/sources/remote-process-source.ts

@@ -0,0 +1,18 @@
+import type {
+  SerializedRequest,
+  SerializedResponse,
+} from './packets/http-packet'
+
+export type StreamEventMap = {
+  'stream:chunk': (chunk: Uint8Array | undefined) => void
+  'stream:error': (reason?: unknown) => void
+  'stream:end': () => void
+}
+
+export type NetworkSessionEventMap = StreamEventMap & {
+  request: (request: SerializedRequest) => void
+}
+
+export type RpcServerEventMap = StreamEventMap & {
+  response: (response: SerializedResponse | undefined) => void
+}

src/core/rpc/events.ts

@@ -0,0 +1,84 @@
+import {
+  RequestHandler,
+  type RequestHandlerDefaultInfo,
+} from '../../handlers/RequestHandler'
+import type { ResponseResolutionContext } from '../../utils/executeHandlers'
+import { NetworkHttpTransport } from '../transports/http-transport'
+
+interface RemoteRequestHandlerParsedResult {
+  response: Response | undefined
+}
+
+type RemoteRequestHandlerResolverExtras = {
+  response: Response | undefined
+}
+
+export class RemoteRequestHandler extends RequestHandler<
+  RequestHandlerDefaultInfo,
+  RemoteRequestHandlerParsedResult,
+  RemoteRequestHandlerResolverExtras
+> {
+  #transport: NetworkHttpTransport
+
+  constructor(args: { port: number }) {
+    super({
+      info: {
+        header: 'RemoteRequestHandler',
+      },
+      resolver({ response }: RemoteRequestHandlerResolverExtras) {
+        return response
+      },
+    })
+
+    this.#transport = new NetworkHttpTransport({
+      port: args.port,
+    })
+  }
+
+  async parse(args: {
+    request: Request
+    resolutionContext?: ResponseResolutionContext
+  }): Promise<any> {
+    const response = await this.#transport
+      .handleRequest({ request: args.request })
+      .catch(() => undefined)
+
+    const parsedResult = await super.parse(args)
+
+    if (response != null) {
+      parsedResult.response = response
+    }
+
+    return parsedResult
+  }
+
+  predicate(args: {
+    request: Request
+    parsedResult: RemoteRequestHandlerParsedResult
+    resolutionContext?: ResponseResolutionContext
+  }): boolean | Promise<boolean> {
+    // This handler is considered matching if the remote process
+    // decided to handle the intercepted request.
+    return args.parsedResult.response != null
+  }
+
+  protected extendResolverArgs(args: {
+    request: Request
+    parsedResult: RemoteRequestHandlerParsedResult
+  }): RemoteRequestHandlerResolverExtras {
+    const resolverArgs = super.extendResolverArgs(args)
+    resolverArgs.response = args.parsedResult.response
+    return resolverArgs
+  }
+
+  log(_args: {
+    request: Request
+    response: Response
+    parsedResult: RemoteRequestHandlerParsedResult
+  }): void {
+    /**
+     * @note Skip logging. This is an internal request handler.
+     */
+    return
+  }
+}

src/core/rpc/handlers/remote-request-handler.ts

@@ -0,0 +1,45 @@
+import { until } from 'until-async'
+import {
+  WebSocketHandler,
+  type WebSocketHandlerConnection,
+  type WebSocketResolutionContext,
+} from '../../handlers/WebSocketHandler'
+import { NetworkSession } from '../session'
+import { NetworkWebSocketTransport } from '../transports/websocket-transport'
+
+export class RemoteWebSocketHandler extends WebSocketHandler {
+  #transport: NetworkWebSocketTransport
+
+  constructor(args: { session: NetworkSession }) {
+    super(/.*/)
+
+    this.#transport = new NetworkWebSocketTransport({
+      session: args.session,
+    })
+  }
+
+  public async run(
+    connection: Omit<WebSocketHandlerConnection, 'params'>,
+    resolutionContext?: WebSocketResolutionContext,
+  ): Promise<boolean> {
+    const [error, remoteConnection] = await until(() => {
+      return this.#transport.send({
+        clientUrl: connection.client.url,
+        resolutionContext,
+      })
+    })
+
+    /**
+     * @todo Check if rejecting with no reason still falls through this check.
+     */
+    if (error) {
+      return false
+    }
+
+    /**
+     * @todo How will this work if THIS process has a WS client emitting events
+     * and THAT process has event listeners? Should we serialize those events now?
+     */
+    return this.connect(remoteConnection)
+  }
+}

src/core/rpc/handlers/remote-websocket-handler.ts

@@ -0,0 +1,97 @@
+import type { Socket } from 'socket.io-client'
+import { DeferredPromise } from '@open-draft/deferred-promise'
+import type { NetworkPacket } from '.'
+import type { SessionSocket } from '../session'
+import { emitReadableStream, WebSocketReadableStreamSource } from '../utils'
+import type { StreamEventMap } from '../events'
+
+export interface SerializedRequest {
+  method: string
+  url: string
+  headers: Array<[string, string]>
+  hasBodyStream: boolean
+}
+
+export interface SerializedResponse {
+  status: number
+  statusText: string
+  headers: Array<[string, string]>
+  hasBodyStream: boolean
+}
+
+export class HttpPacket implements NetworkPacket {
+  constructor(private readonly request: Request) {}
+
+  async send(socket: SessionSocket): Promise<Response | undefined> {
+    const intentionPromise = new DeferredPromise<Response | undefined>()
+    const serializedRequest = serializeHttpRequest(this.request)
+
+    socket.emit('request', serializedRequest)
+
+    if (this.request.body != null) {
+      emitReadableStream(this.request.body, socket)
+    }
+
+    socket.once('response', (serializedResponse) => {
+      const response =
+        serializedResponse != null
+          ? deserializeHttpResponse(serializedResponse, socket)
+          : undefined
+
+      intentionPromise.resolve(response)
+    })
+
+    return intentionPromise
+  }
+}
+
+export function serializeHttpRequest(request: Request): SerializedRequest {
+  return {
+    method: request.method,
+    url: request.url,
+    headers: Array.from(request.headers),
+    hasBodyStream: request.body != null,
+  }
+}
+
+export function deserializeHttpRequest(
+  serializedRequest: SerializedRequest,
+  socket: Socket<StreamEventMap, any>,
+): Request {
+  const { method, url, headers, hasBodyStream } = serializedRequest
+
+  return new Request(url, {
+    method,
+    headers,
+    body: hasBodyStream
+      ? new ReadableStream(new WebSocketReadableStreamSource(socket))
+      : null,
+  })
+}
+
+export function serializeHttpResponse(response: Response): SerializedResponse {
+  return {
+    status: response.status,
+    statusText: response.statusText,
+    headers: Array.from(response.headers),
+    hasBodyStream: response.body != null,
+  }
+}
+
+export function deserializeHttpResponse(
+  serializedResponse: SerializedResponse,
+  socket: SessionSocket,
+): Response {
+  const { status, statusText, headers, hasBodyStream } = serializedResponse
+
+  return new Response(
+    hasBodyStream
+      ? new ReadableStream(new WebSocketReadableStreamSource(socket))
+      : null,
+    {
+      status,
+      statusText,
+      headers,
+    },
+  )
+}

src/core/rpc/packets/http-packet.ts

@@ -0,0 +1,5 @@
+import type { SessionSocket } from '../session'
+
+export interface NetworkPacket {
+  send(socket: SessionSocket): Promise<any>
+}

src/core/rpc/packets/index.ts

@@ -0,0 +1,19 @@
+import type { NetworkPacket } from '.'
+import type { WebSocketResolutionContext } from '../../handlers/WebSocketHandler'
+import type { SessionSocket } from '../session'
+
+export class WebSocketPacket implements NetworkPacket {
+  constructor(
+    private readonly args: {
+      url: string
+      resolutionContext?: WebSocketResolutionContext
+    },
+  ) {}
+
+  async send(socket: SessionSocket): Promise<any> {
+    // 1. Create a frame that describe this WS connection.
+    // 2. Send it over the `ws`.
+    // 3. (?) Return the response?
+    socket.send('...TODO...')
+  }
+}

src/core/rpc/packets/websocket-packet.ts

@@ -0,0 +1,83 @@
+import * as http from 'node:http'
+import { DeferredPromise } from '@open-draft/deferred-promise'
+import { Server } from 'socket.io'
+import type { Socket } from 'socket.io-client'
+import { Emitter, TypedEvent } from 'rettime'
+import type { NetworkSessionEventMap, StreamEventMap } from './events'
+import {
+  deserializeHttpRequest,
+  serializeHttpResponse,
+} from './packets/http-packet'
+import { emitReadableStream } from './utils'
+
+type RpcServerPublicEventMap = {
+  request: TypedEvent<{ request: Request }, Response>
+}
+
+export class RpcServer extends Emitter<RpcServerPublicEventMap> {
+  #server: Server<NetworkSessionEventMap, RpcServerPublicEventMap>
+
+  constructor() {
+    super()
+
+    const httpServer = http.createServer()
+
+    this.#server = new Server()
+    this.#server.attach(httpServer)
+
+    this.#server.on('connection', (client) => {
+      client.on('request', async (serializedRequest) => {
+        const request = deserializeHttpRequest(
+          serializedRequest,
+          client as unknown as Socket<any, StreamEventMap>,
+        )
+
+        /** @todo Notify the consumer there's been a request! */
+        const results = await this.emitAsPromise(
+          new TypedEvent('request', {
+            data: {
+              request,
+            },
+          }),
+        )
+
+        const response = new Response('hello world')
+
+        // client.emit('response', serializeHttpResponse(response))
+
+        if (response.body != null) {
+          emitReadableStream(
+            response.body,
+            client as unknown as Socket<any, StreamEventMap>,
+          )
+        }
+      })
+    })
+  }
+
+  public async listen(port: number): Promise<void> {
+    const listenPromise = new DeferredPromise<void>()
+    const { httpServer } = this.#server
+
+    httpServer
+      .listen(port, () => listenPromise.resolve())
+      .once('error', (error) => listenPromise.reject(error))
+
+    return listenPromise
+  }
+
+  public async close(): Promise<void> {
+    const closePromise = new DeferredPromise<void>()
+
+    this.#server.disconnectSockets()
+    this.#server.close((error) => {
+      if (error) {
+        closePromise.reject(error)
+      } else {
+        closePromise.resolve()
+      }
+    })
+
+    return closePromise
+  }
+}