diff --git a/packages/http-transport/package.json b/packages/http-transport/package.json index a667507a..dc2e2709 100644 --- a/packages/http-transport/package.json +++ b/packages/http-transport/package.json @@ -54,6 +54,7 @@ "@nmtjs/common": "workspace:*", "@nmtjs/gateway": "workspace:*", "@nmtjs/protocol": "workspace:*", + "@nmtjs/server": "workspace:*", "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.67.0" }, "peerDependencies": { @@ -62,6 +63,7 @@ "@nmtjs/core": "workspace:*", "@nmtjs/gateway": "workspace:*", "@nmtjs/protocol": "workspace:*", + "@nmtjs/server": "workspace:*", "@types/bun": "^1.3.0", "@types/deno": "^2.3.0", "@types/node": "^24.13.2", diff --git a/packages/http-transport/src/adapter.ts b/packages/http-transport/src/adapter.ts new file mode 100644 index 00000000..14f1e807 --- /dev/null +++ b/packages/http-transport/src/adapter.ts @@ -0,0 +1,39 @@ +import type { + ServerHost, + ServerHostOptions, + ServerRuntimeName, +} from '@nmtjs/server' + +import type { HttpAdapterParams, HttpAdapterServer } from './types.ts' + +/** + * Shared-server mode mounts onto the provided host; listen mode owns a + * private one — either way a ServerHost is the only socket owner, so both + * modes exercise the same server code path. + */ +export function createHostAdapter( + createHost: (options: ServerHostOptions) => ServerHost, + params: HttpAdapterParams, +): HttpAdapterServer { + let host: ServerHost + if (params.server) { + host = params.server + } else if (params.listen) { + host = createHost({ + listen: params.listen, + tls: params.tls, + maxRequestBodySize: params.maxRequestBodySize, + runtime: params.runtime, + }) + } else { + throw new Error('Either `server` or `listen` option is required') + } + host.setFetchHandler(params.fetchHandler) + return { + get runtime() { + return host.native + }, + start: () => host.start(), + stop: () => host.stop(), + } +} diff --git a/packages/http-transport/src/constants.ts b/packages/http-transport/src/constants.ts index 774931fd..caef05fc 100644 --- a/packages/http-transport/src/constants.ts +++ b/packages/http-transport/src/constants.ts @@ -2,8 +2,8 @@ import type { MetadataKind } from '@nmtjs/application' import { createMeta } from '@nmtjs/application' import { ErrorCode } from '@nmtjs/protocol' -// Matches Bun's maxRequestBodySize default so all runtimes behave the same -export const DEFAULT_MAX_REQUEST_BODY_SIZE = 128 * 1024 * 1024 +// Single source in @nmtjs/server: the host enforces the same default bound +export { DEFAULT_MAX_REQUEST_BODY_SIZE } from '@nmtjs/server' export enum HttpStatus { Continue = 100, diff --git a/packages/http-transport/src/runtimes/bun.ts b/packages/http-transport/src/runtimes/bun.ts index 0e83fbbe..f0fd38fa 100644 --- a/packages/http-transport/src/runtimes/bun.ts +++ b/packages/http-transport/src/runtimes/bun.ts @@ -1,86 +1,15 @@ import type { ApplicationTransport } from '@nmtjs/application' import type { ConnectionType } from '@nmtjs/protocol' import { ProxyableTransportType } from '@nmtjs/gateway' +import { createServerHost } from '@nmtjs/server/bun' -import type { - HttpAdapterParams, - HttpAdapterServer, - HttpTransportOptions, -} from '../types.ts' +import type { HttpAdapterParams, HttpTransportOptions } from '../types.ts' +import { createHostAdapter } from '../adapter.ts' import * as injectables from '../injectables.ts' import { createHTTPTransportWorker } from '../server.ts' -import { - InternalServerErrorHttpResponse, - NotFoundHttpResponse, - OkResponse, -} from '../utils.ts' -function adapterFactory(params: HttpAdapterParams<'bun'>): HttpAdapterServer { - let server: Bun.Server | null = null - - function createServer() { - const routes = - typeof params.runtime?.routes === 'object' && params.runtime.routes - ? params.runtime.routes - : {} - - return globalThis.Bun.serve({ - ...params.runtime, - // Bun's own default (128MiB) applies when neither option is set - maxRequestBodySize: - params.runtime?.maxRequestBodySize ?? params.maxRequestBodySize, - unix: params.listen.unix as string, - port: params.listen.port ?? 0, - hostname: params.listen.hostname, - reusePort: params.listen.reusePort, - tls: params.tls - ? { - cert: params.tls.cert, - key: params.tls.key, - passphrase: params.tls.passphrase, - } - : undefined, - routes: Object.assign({}, routes, { - '/healthy': { GET: OkResponse }, - }) as any, - async fetch(request) { - const url = new URL(request.url) - try { - if (request.headers.get('upgrade') === 'websocket') - return NotFoundHttpResponse() - const { body, headers, method } = request - return await params.fetchHandler( - { url, method, headers }, - body, - request.signal, - ) - } catch (err) { - // TODO: proper logging - console.error(err) - // params.logger.error({ err }, 'Error in fetch handler') - return InternalServerErrorHttpResponse() - } - }, - } as any) - } - - return { - runtime: { - get bun() { - return server! - }, - }, - start: async () => { - server = createServer() - return server!.url.href - }, - stop: async () => { - if (server) { - await server.stop() - server = null - } - }, - } +function adapterFactory(params: HttpAdapterParams<'bun'>) { + return createHostAdapter(createServerHost, params) } export const HttpTransport: ApplicationTransport< diff --git a/packages/http-transport/src/runtimes/deno.ts b/packages/http-transport/src/runtimes/deno.ts index 233310f8..847ae6d4 100644 --- a/packages/http-transport/src/runtimes/deno.ts +++ b/packages/http-transport/src/runtimes/deno.ts @@ -1,128 +1,15 @@ import type { ApplicationTransport } from '@nmtjs/application' import type { ConnectionType } from '@nmtjs/protocol' import { ProxyableTransportType } from '@nmtjs/gateway' +import { createServerHost } from '@nmtjs/server/deno' -import type { - DenoServer, - HttpAdapterParams, - HttpAdapterServer, - HttpTransportOptions, -} from '../types.ts' +import type { HttpAdapterParams, HttpTransportOptions } from '../types.ts' +import { createHostAdapter } from '../adapter.ts' import * as injectables from '../injectables.ts' import { createHTTPTransportWorker } from '../server.ts' -import { - InternalServerErrorHttpResponse, - NotFoundHttpResponse, - OkResponse, -} from '../utils.ts' -interface DenoNetAddr { - transport: 'tcp' | 'udp' - hostname: string - port: number -} - -interface DenoUnixAddr { - transport: 'unix' | 'unixpacket' - path: string -} - -interface DenoVsockAddr { - transport: 'vsock' - cid: number - port: number -} - -type DenoAddr = DenoNetAddr | DenoUnixAddr | DenoVsockAddr - -function adapterFactory(params: HttpAdapterParams<'deno'>): HttpAdapterServer { - let server: DenoServer | null = null - - function createServer() { - const listenOptions = params.listen.unix - ? { path: params.listen.unix } - : { - port: params.listen.port, - hostname: params.listen.hostname, - reusePort: params.listen.reusePort, - } - const options = { - ...listenOptions, - tls: params.tls - ? { - cert: params.tls.cert, - key: params.tls.key, - passphrase: params.tls.passphrase, - } - : undefined, - } - - return new Promise<{ server: DenoServer; addr: DenoAddr }>((resolve) => { - const server = globalThis.Deno.serve({ - ...params.runtime, - ...options, - handler: async (request: Request) => { - const url = new URL(request.url) - if (url.pathname === '/healthy') { - return OkResponse() - } - try { - if (request.headers.get('upgrade') === 'websocket') { - return NotFoundHttpResponse() - } - const { headers, method, body } = request - return await params.fetchHandler( - { url, method, headers }, - body, - request.signal, - ) - } catch (err) { - // TODO: proper logging - console.error(err) - // params.logger.error({ err }, 'Error in fetch handler') - return InternalServerErrorHttpResponse() - } - }, - onListen(addr: DenoAddr) { - setTimeout(() => { - resolve({ server, addr }) - }, 1) - }, - }) - }) - } - - return { - runtime: { - get deno() { - return server! - }, - }, - start: async () => { - const { server: _server, addr } = await createServer() - server = _server - switch (addr.transport) { - case 'unix': - case 'unixpacket': - return `unix://${addr.path}` - case 'tcp': - case 'udp': { - const proto = params.tls ? 'https' : 'http' - return `${proto}://${addr.hostname}:${addr.port}` - } - case 'vsock': - return `vsock://${addr.cid}:${addr.port}` - default: - throw new Error(`Unsupported address transport`) - } - }, - stop: async () => { - if (server) { - await server.shutdown() - server = null - } - }, - } +function adapterFactory(params: HttpAdapterParams<'deno'>) { + return createHostAdapter(createServerHost, params) } export const HttpTransport: ApplicationTransport< diff --git a/packages/http-transport/src/runtimes/node.ts b/packages/http-transport/src/runtimes/node.ts index 32055cf6..4e616019 100644 --- a/packages/http-transport/src/runtimes/node.ts +++ b/packages/http-transport/src/runtimes/node.ts @@ -1,326 +1,15 @@ -import {} from 'node:dns' - import type { ApplicationTransport } from '@nmtjs/application' import type { ConnectionType } from '@nmtjs/protocol' import { ProxyableTransportType } from '@nmtjs/gateway' -import { App, SSLApp, us_socket_local_port } from 'uWebSockets.js' +import { createServerHost } from '@nmtjs/server/node' -import type { - HttpAdapterParams, - HttpAdapterServer, - HttpTransportOptions, -} from '../types.ts' -import { DEFAULT_MAX_REQUEST_BODY_SIZE } from '../constants.ts' +import type { HttpAdapterParams, HttpTransportOptions } from '../types.ts' +import { createHostAdapter } from '../adapter.ts' import * as injectables from '../injectables.ts' import { createHTTPTransportWorker } from '../server.ts' -import { - InternalServerErrorHttpResponse, - NotFoundHttpResponse, - OkResponse, - PayloadTooLargeError, -} from '../utils.ts' - -const statusResponse = OkResponse() -const statusResponseBuffer = await statusResponse.arrayBuffer() - -type UwsResponse = Parameters< - Parameters['any']>[1] ->[0] & { - aborted?: boolean - wakeWritable?: () => void - cancelBody?: () => void -} - -function adapterFactory(params: HttpAdapterParams<'node'>): HttpAdapterServer { - const maxBodySize = params.maxRequestBodySize ?? DEFAULT_MAX_REQUEST_BODY_SIZE - const server = params.tls - ? SSLApp({ - passphrase: params.tls.passphrase, - key_file_name: params.tls.key, - cert_file_name: params.tls.cert, - }) - : App() - - server - .get('/healthy', (res) => { - res.cork(() => { - res - .writeStatus(`${statusResponse.status} ${statusResponse.statusText}`) - .end(statusResponseBuffer) - }) - }) - .any('/*', async (res, req) => { - const requestController = new AbortController() - const uwsRes = res as UwsResponse - let aborted = false - let bodyController: ReadableStreamDefaultController | undefined - - res.onAborted(() => { - aborted = true - uwsRes.aborted = true - uwsRes.wakeWritable?.() - uwsRes.cancelBody?.() - requestController.abort() - - try { - bodyController?.error(requestController.signal.reason) - } catch {} - }) - - let response = NotFoundHttpResponse() - - const headers = new Headers() - const method = req.getMethod() - req.forEach((k, v) => headers.append(k, v)) - - const host = headers.get('host') || 'localhost' - const forwardedProto = headers.get('x-forwarded-proto') - const proto = forwardedProto - ? forwardedProto === 'https' - ? 'https' - : 'http' - : params.tls - ? 'https' - : 'http' - const url = new URL(req.getUrl(), `${proto}://${host}`) - url.search = req.getQuery() ? `?${req.getQuery()}` : '' - try { - // uWS delivers chunks without backpressure, so cap what gets copied - // into memory before the whole body arrives - let received = 0 - let capped = false - const body = new ReadableStream({ - start(controller) { - bodyController = controller - res.onDataV2((chunk, maxRemainingBodyLength) => { - if (aborted || capped) return - if (chunk) { - received += chunk.byteLength - if (received > maxBodySize) { - capped = true - controller.error(new PayloadTooLargeError()) - return - } - const copy = Buffer.allocUnsafe(chunk.byteLength) - copy.set(new Uint8Array(chunk)) - controller.enqueue(copy) - } - if (maxRemainingBodyLength === 0n) controller.close() - }) - }, - }) - response = await params.fetchHandler( - { url, method, headers }, - body, - requestController.signal, - ) - } catch (err) { - // TODO: proper logging - console.error(err) - // params.logger.error({ err }, 'Error in fetch handler') - response = InternalServerErrorHttpResponse() - } - if (aborted) return undefined - else { - const fixedContentLength = response.body - ? getContentLength(response.headers) - : undefined - res.cork(() => { - if (aborted) return undefined - res.writeStatus( - `${response.status.toString()} ${response.statusText}`, - ) - response.headers.forEach((v, k) => { - if ( - typeof fixedContentLength === 'number' && - k.toLowerCase() === 'content-length' - ) { - return - } - - res.writeHeader(k, v) - }) - }) - if (response.body) { - try { - await handleResponseBody(uwsRes, response, fixedContentLength) - } catch { - if (!aborted) res.cork(() => res.close()) - } - } else { - if (!aborted) res.cork(() => res.end()) - } - } - }) - - return { - runtime: { node: server }, - start: () => - new Promise((resolve, reject) => { - if (params.listen.unix) { - server.listen_unix((socket) => { - if (socket) { - resolve('unix://' + params.listen.unix) - } else { - reject(new Error('Failed to start WebSockets server')) - } - }, params.listen.unix) - } else if (typeof params.listen.port === 'number') { - const proto = params.tls ? 'https' : 'http' - const hostname = params.listen.hostname || '127.0.0.1' - - server.listen(hostname, params.listen.port, (socket) => { - if (socket) { - resolve(`${proto}://${hostname}:${us_socket_local_port(socket)}`) - } else { - reject(new Error('Failed to start WebSockets server')) - } - }) - } else { - reject(new Error('Invalid listen parameters')) - } - }), - stop: () => { - server.close() - }, - } -} - -async function handleResponseBody( - res: UwsResponse, - response: Response, - fixedContentLength?: number, -): Promise { - if (!response.body) return - - if (typeof fixedContentLength === 'number') { - await handleFixedLengthStream(res, response.body, fixedContentLength) - return - } - - await handleChunkedStream(res, response.body) -} - -async function handleFixedLengthStream( - res: UwsResponse, - body: ReadableStream, - totalSize: number, -): Promise { - const reader = body.getReader() - try { - if (totalSize === 0) { - if (!res.aborted) res.cork(() => res.endWithoutBody(0)) - return - } - - let responded = false - while (!res.aborted && !responded) { - const { done, value } = await reader.read() - if (done) break - if (value.byteLength === 0) continue - responded = await handleFixedChunk(res, value, totalSize) - } - - if (!responded && !res.aborted) res.cork(() => res.close()) - } finally { - reader.releaseLock() - } -} - -// exported for tests: the waiter dispatch is timing-sensitive and needs -// deterministic coverage against a controlled response double -export async function handleChunkedStream( - res: UwsResponse, - body: ReadableStream, -): Promise { - const reader = body.getReader() - // abort must also cancel a pending read(): a stalled source would otherwise - // keep the pump and the reader lock alive forever - res.cancelBody = () => { - reader.cancel().catch(() => {}) - } - // uWS honors only the first onWritable registration per response, so a - // single handler dispatches drain (and abort) events to the pending waiter - let writableHandlerRegistered = false - const waitWritable = () => - new Promise((resolve, reject) => { - if (!writableHandlerRegistered) { - writableHandlerRegistered = true - res.onWritable(() => { - res.wakeWritable?.() - return true - }) - } - res.wakeWritable = () => { - res.wakeWritable = undefined - if (res.aborted) reject(new Error('Response aborted')) - else resolve() - } - }) - try { - while (!res.aborted) { - const { done, value } = await reader.read() - if (done) break - if (value.byteLength === 0) continue - - // cork() returns the response object, not write()'s backpressure flag - let ok = true - res.cork(() => { - ok = res.write(value) - }) - if (!ok) await waitWritable() - } - - if (!res.aborted) res.cork(() => res.end()) - } finally { - res.cancelBody = undefined - reader.releaseLock() - } -} - -function handleFixedChunk( - res: UwsResponse, - chunk: Uint8Array, - totalSize: number, -): Promise { - return new Promise((resolve, reject) => { - const chunkOffset = res.getWriteOffset() - - const write = (offset: number) => { - if (res.aborted) { - reject(new Error('Response aborted')) - return false - } - - const relativeOffset = offset - chunkOffset - const remaining = - relativeOffset > 0 ? chunk.subarray(relativeOffset) : chunk - - let ok = false - let done = false - res.cork(() => { - ;[ok, done] = res.tryEnd(remaining, totalSize) - }) - - if (done || ok) { - resolve(done) - return ok - } - - res.onWritable(write) - return ok - } - - write(chunkOffset) - }) -} - -function getContentLength(headers: Headers): number | undefined { - const raw = headers.get('content-length') - if (!raw) return undefined - const value = Number(raw) - return Number.isSafeInteger(value) && value >= 0 ? value : undefined +function adapterFactory(params: HttpAdapterParams<'node'>) { + return createHostAdapter(createServerHost, params) } export const HttpTransport: ApplicationTransport< diff --git a/packages/http-transport/src/types.ts b/packages/http-transport/src/types.ts index ef309a7c..26ec8802 100644 --- a/packages/http-transport/src/types.ts +++ b/packages/http-transport/src/types.ts @@ -1,25 +1,57 @@ import type { MaybePromise, OneOf } from '@nmtjs/common' +import type { + DenoServer, + ServerFetchHandler, + ServerHost, + ServerListenOptions, + ServerNativeHandles, + ServerRequest, + ServerRuntimeName, + ServerRuntimeOptions, + ServerTlsOptions, +} from '@nmtjs/server' -export type HttpTransportServerRequest = { - url: URL - method: string - headers: Headers -} +export type { DenoServer } + +export type HttpTransportServerRequest = ServerRequest + +export type HttpTransportListenOptions = ServerListenOptions + +export type HttpTransportTlsOptions = ServerTlsOptions + +export type HttpTransportRuntimeBun = ServerRuntimeOptions['bun'] +export type HttpTransportRuntimeNode = ServerRuntimeOptions['node'] + +export type HttpTransportRuntimeDeno = ServerRuntimeOptions['deno'] + +export type HttpTransportRuntimes = ServerRuntimeOptions + +/** + * The transport either owns its socket (`listen` mode) or mounts onto a + * shared `ServerHost` (`server` mode), so one HTTP server can carry + * multiple transports (e.g. HTTP and WS on a single listen address). + */ export type HttpTransportOptions< - R extends keyof HttpTransportRuntimes = keyof HttpTransportRuntimes, + R extends ServerRuntimeName = ServerRuntimeName, > = { - listen: HttpTransportListenOptions cors?: HttpTransportCorsOptions - tls?: HttpTransportTlsOptions /** * Maximum request body size in bytes. Requests exceeding it are rejected * with 413 Payload Too Large. Defaults to 128MiB (Bun's own default, kept * consistent across runtimes). */ maxRequestBodySize?: number - runtime?: HttpTransportRuntimes[R] -} +} & OneOf< + [ + { + listen: HttpTransportListenOptions + tls?: HttpTransportTlsOptions + runtime?: HttpTransportRuntimes[R] + }, + { server: ServerHost }, + ] +> export type HttpTransportCorsCustomParams = { /** @@ -49,70 +81,17 @@ export type HttpTransportCorsOptions = request: HttpTransportServerRequest, ) => boolean | HttpTransportCorsCustomParams) -export type HttpTransportListenOptions = OneOf< - [{ port: number; hostname?: string; reusePort?: boolean }, { unix: string }] -> - -export type HttpTransportRuntimeBun = Partial< - Pick< - import('bun').Serve.Options, - 'development' | 'id' | 'maxRequestBodySize' | 'idleTimeout' | 'ipv6Only' - > & - import('bun').Serve.Routes -> - -export type HttpTransportRuntimeNode = {} - -export type HttpTransportRuntimeDeno = {} - -export type HttpTransportRuntimes = { - bun: HttpTransportRuntimeBun - node: HttpTransportRuntimeNode - deno: HttpTransportRuntimeDeno -} - -export type HttpTransportTlsOptions = { - /** - * File path or inlined TLS certificate in PEM format (required). - */ - cert?: string - /** - * File path or inlined TLS private key in PEM format (required). - */ - key?: string - /** - * Passphrase for the private key (optional). - */ - passphrase?: string -} - -export type HttpAdapterParams< - R extends keyof HttpTransportRuntimes = keyof HttpTransportRuntimes, -> = { - listen: HttpTransportListenOptions - fetchHandler: ( - request: HttpTransportServerRequest, - body: ReadableStream | null, - signal: AbortSignal, - ) => MaybePromise - cors?: HttpTransportCorsOptions - tls?: HttpTransportTlsOptions - maxRequestBodySize?: number - runtime?: HttpTransportRuntimes[R] -} - -export type DenoServer = ReturnType +export type HttpAdapterParams = + HttpTransportOptions & { + fetchHandler: ServerFetchHandler + } export interface HttpAdapterServer { - runtime: { - bun?: import('bun').Server - node?: import('uWebSockets.js').TemplatedApp - deno?: DenoServer - } + runtime: ServerNativeHandles stop: () => MaybePromise start: () => MaybePromise } export type HttpAdapterServerFactory< - R extends keyof HttpTransportRuntimes = keyof HttpTransportRuntimes, + R extends ServerRuntimeName = ServerRuntimeName, > = (params: HttpAdapterParams) => HttpAdapterServer diff --git a/packages/http-transport/src/utils.ts b/packages/http-transport/src/utils.ts index e3cd223e..ab2ca874 100644 --- a/packages/http-transport/src/utils.ts +++ b/packages/http-transport/src/utils.ts @@ -1,6 +1,16 @@ import { ErrorCode } from '@nmtjs/protocol' import { ProtocolError } from '@nmtjs/protocol/server' +// Response helpers and PayloadTooLargeError live in @nmtjs/server: the +// runtime hosts throw/build them, and instanceof matching in this package +// requires the same class identity +export { + InternalServerErrorHttpResponse, + NotFoundHttpResponse, + OkResponse, + PayloadTooLargeError, +} from '@nmtjs/server' + export const InternalError = (message = 'Internal Server Error') => new ProtocolError(ErrorCode.InternalServerError, message) @@ -12,23 +22,3 @@ export const ForbiddenError = (message = 'Forbidden') => export const RequestTimeoutError = (message = 'Request Timeout') => new ProtocolError(ErrorCode.RequestTimeout, message) - -export const NotFoundHttpResponse = () => - new Response('Not Found', { - status: 404, - headers: { 'Content-Type': 'text/plain' }, - }) - -export const InternalServerErrorHttpResponse = () => - new Response('Internal Server Error', { - status: 500, - headers: { 'Content-Type': 'text/plain' }, - }) - -export const OkResponse = () => new Response('OK', { status: 200 }) - -export class PayloadTooLargeError extends Error { - constructor(message = 'Payload Too Large') { - super(message) - } -} diff --git a/packages/http-transport/tests/_helpers/test-utils.ts b/packages/http-transport/tests/_helpers/test-utils.ts index c81b9204..3d99840b 100644 --- a/packages/http-transport/tests/_helpers/test-utils.ts +++ b/packages/http-transport/tests/_helpers/test-utils.ts @@ -59,7 +59,7 @@ export async function createTestServer( const server = new HttpTransportServer(stubAdapterFactory, { listen: { port: 0 }, ...options, - }) + } as HttpTransportOptions) await server.start(params) return server } diff --git a/packages/http-transport/tests/node-runtime.spec.ts b/packages/http-transport/tests/node-runtime.spec.ts index a3abb598..41285da8 100644 --- a/packages/http-transport/tests/node-runtime.spec.ts +++ b/packages/http-transport/tests/node-runtime.spec.ts @@ -4,7 +4,7 @@ import { BaseServerFormat, ProtocolFormats } from '@nmtjs/protocol/server' import { describe, expect, it, vi } from 'vitest' import type { HttpTransportServerRequest } from '../src/types.ts' -import { handleChunkedStream, HttpTransport } from '../src/runtimes/node.ts' +import { HttpTransport } from '../src/runtimes/node.ts' class TestJsonFormat extends BaseServerFormat { accept = ['application/json'] @@ -143,146 +143,4 @@ describe('node runtime adapter', () => { } }) }) - - describe('handleChunkedStream writable dispatcher', () => { - // scripted stand-in for uWS HttpResponse: only the first onWritable - // registration takes effect, matching real uWS behavior - function createResDouble(writeResults: boolean[]) { - const results = [...writeResults] - let writes = 0 - let ends = 0 - let registrations = 0 - let handler: ((offset: number) => boolean) | undefined - const res: any = { - aborted: false, - wakeWritable: undefined, - cork(cb: () => void) { - cb() - return res - }, - write() { - writes++ - return results.length > 0 ? (results.shift() as boolean) : true - }, - end() { - ends++ - return res - }, - onWritable(h: (offset: number) => boolean) { - registrations++ - handler ??= h - return res - }, - } - return { - res, - fireWritable: () => handler!(0), - counts: () => ({ writes, ends, registrations }), - // mirrors what the route's onAborted handler does - abort: () => { - res.aborted = true - res.wakeWritable?.() - res.cancelBody?.() - }, - } - } - - const chunks = (n: number) => - new ReadableStream({ - start(controller) { - for (let i = 0; i < n; i++) controller.enqueue(new Uint8Array(8)) - controller.close() - }, - }) - - const tick = () => new Promise((resolve) => setImmediate(resolve)) - - it('resumes exactly one pending waiter per drain across cycles', async () => { - const double = createResDouble([false, true, false, true]) - const pump = handleChunkedStream(double.res, chunks(4)) - - await tick() - expect(double.counts()).toEqual({ writes: 1, ends: 0, registrations: 1 }) - - double.fireWritable() - await tick() - // resumed once: write 2 succeeded, write 3 hit backpressure again and - // parked a second waiter without re-registering the handler - expect(double.counts()).toEqual({ writes: 3, ends: 0, registrations: 1 }) - - double.fireWritable() - await tick() - await pump - expect(double.counts()).toEqual({ writes: 4, ends: 1, registrations: 1 }) - }) - - it('writable callback without a pending waiter is a no-op returning true', async () => { - const double = createResDouble([false]) - const pump = handleChunkedStream(double.res, chunks(1)) - - await tick() - expect(double.fireWritable()).toBe(true) - await tick() - await pump - - // stream is finished; a late drain event finds no waiter - expect(() => double.fireWritable()).not.toThrow() - expect(double.fireWritable()).toBe(true) - }) - - it('abort while a waiter is pending settles the wait and exits the pump', async () => { - const double = createResDouble([false]) - const body = chunks(2) - const pump = handleChunkedStream(double.res, body) - - await tick() - expect(double.counts().writes).toBe(1) - - double.abort() - await expect(pump).rejects.toThrow('Response aborted') - expect(double.counts()).toEqual({ writes: 1, ends: 0, registrations: 1 }) - // reader lock is released on exit - expect(() => body.getReader()).not.toThrow() - }) - - it('handles a drain cycle followed by an abort', async () => { - const double = createResDouble([false, false]) - const pump = handleChunkedStream(double.res, chunks(3)) - - await tick() - expect(double.counts().writes).toBe(1) - - double.fireWritable() - await tick() - // resumed and immediately backpressured again - expect(double.counts().writes).toBe(2) - - double.abort() - await expect(pump).rejects.toThrow('Response aborted') - expect(double.counts()).toEqual({ writes: 2, ends: 0, registrations: 1 }) - }) - - it('abort while read() is stalled cancels the reader and exits the pump', async () => { - const double = createResDouble([]) - let cancelled = false - const body = new ReadableStream({ - // source never produces: the pump parks inside reader.read() - pull: () => new Promise(() => {}), - cancel() { - cancelled = true - }, - }) - const pump = handleChunkedStream(double.res, body) - - await tick() - expect(double.counts().writes).toBe(0) - - double.abort() - // reader.cancel() resolves the pending read as done and the pump exits - // without writing or ending the aborted response - await pump - expect(cancelled).toBe(true) - expect(double.counts()).toEqual({ writes: 0, ends: 0, registrations: 0 }) - }) - }) }) diff --git a/packages/neem/package.json b/packages/neem/package.json index 4bb2cf13..9233f189 100644 --- a/packages/neem/package.json +++ b/packages/neem/package.json @@ -61,9 +61,16 @@ } }, "devDependencies": { + "@nmtjs/gateway": "workspace:*", + "@nmtjs/http-transport": "workspace:*", + "@nmtjs/json-format": "workspace:*", "@nmtjs/metrics": "workspace:*", + "@nmtjs/protocol": "workspace:*", "@nmtjs/proxy": "1.0.0-beta.7", - "@types/node": "^24.13.2" + "@nmtjs/server": "workspace:*", + "@nmtjs/ws-transport": "workspace:*", + "@types/node": "^24.13.2", + "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.67.0" }, "files": [ "bin", diff --git a/packages/neem/tests/e2e/fixtures/cases/shared-server/api.planner.ts b/packages/neem/tests/e2e/fixtures/cases/shared-server/api.planner.ts new file mode 100644 index 00000000..04f40c10 --- /dev/null +++ b/packages/neem/tests/e2e/fixtures/cases/shared-server/api.planner.ts @@ -0,0 +1,5 @@ +import { defineRuntimePlanner } from '@nmtjs/neem' + +export default defineRuntimePlanner(() => ({ + workers: [{ label: 'api' }], +})) diff --git a/packages/neem/tests/e2e/fixtures/cases/shared-server/api.runtime.ts b/packages/neem/tests/e2e/fixtures/cases/shared-server/api.runtime.ts new file mode 100644 index 00000000..5fca3e26 --- /dev/null +++ b/packages/neem/tests/e2e/fixtures/cases/shared-server/api.runtime.ts @@ -0,0 +1,13 @@ +import { defineRuntime } from '@nmtjs/neem' + +export default defineRuntime({ + name: 'api', + planner: './api.planner.ts', + proxy: { routing: { type: 'default' } }, + worker: { + entry: './shared-server.worker.ts', + // uWS loads its .node binary via a runtime-computed require that + // bundling cannot see; keep it external and resolve from node_modules + build: { rolldown: { external: ['uWebSockets.js'] } }, + }, +}) diff --git a/packages/neem/tests/e2e/fixtures/cases/shared-server/neem.config.ts b/packages/neem/tests/e2e/fixtures/cases/shared-server/neem.config.ts new file mode 100644 index 00000000..7002cc99 --- /dev/null +++ b/packages/neem/tests/e2e/fixtures/cases/shared-server/neem.config.ts @@ -0,0 +1,11 @@ +import { defineConfig } from '@nmtjs/neem' + +export default defineConfig({ + logger: '../../shared/support/logger.ts', + proxy: { + hostname: '127.0.0.1', + port: Number.parseInt(process.env.NEEM_PROXY_PORT ?? '0', 10), + healthChecks: { interval: 50 }, + }, + runtimes: ['./api.runtime.ts'], +}) diff --git a/packages/neem/tests/e2e/fixtures/cases/shared-server/shared-server.worker.ts b/packages/neem/tests/e2e/fixtures/cases/shared-server/shared-server.worker.ts new file mode 100644 index 00000000..85fbecf2 --- /dev/null +++ b/packages/neem/tests/e2e/fixtures/cases/shared-server/shared-server.worker.ts @@ -0,0 +1,101 @@ +import type { + GatewayApi, + GatewayResolvedProcedure, + TransportWorker, +} from '@nmtjs/gateway' +import type { NeemRuntimeUpstream, NeemRuntimeWorkerContext } from '@nmtjs/neem' +import type { ConnectionType } from '@nmtjs/protocol' +import { Container, createLogger, Hooks } from '@nmtjs/core' +import { Gateway, ProxyableTransportType } from '@nmtjs/gateway' +import { HttpTransport } from '@nmtjs/http-transport/node' +import { JsonFormat } from '@nmtjs/json-format/server' +import { defineRuntimeWorker } from '@nmtjs/neem' +import { ProtocolFormats } from '@nmtjs/protocol/server' +import { createServerHost } from '@nmtjs/server/node' +import { WsTransport } from '@nmtjs/ws-transport/node' + +import { record } from '../../shared/support/_events.ts' + +// resolved procedures carry an empty meta view: the HTTP transport reads the +// allowed-methods meta from it and falls back to its POST-only default +type ResolvedProcedure = GatewayResolvedProcedure & { + meta: Map +} + +export default defineRuntimeWorker({ + definition: { fixture: 'shared-server' }, + createRuntime(ctx: NeemRuntimeWorkerContext) { + let gateway: Gateway | undefined + + return { + async start(): Promise { + const logger = createLogger( + { pinoOptions: { enabled: false } }, + 'shared-server', + ) + const container = new Container({ logger }) + + // both transports mount onto one socket: the gateway then reports + // the same bound URL under both proxyable types + const host = createServerHost({ + listen: { port: 0, hostname: '127.0.0.1' }, + }) + const http = await HttpTransport.factory({ server: host }) + const ws = await WsTransport.factory({ server: host }) + + const api: GatewayApi = { + resolve: async ({ procedure }) => ({ + name: procedure, + stream: false, + meta: new Map(), + }), + call: async ({ procedure, payload }) => ({ + procedure, + payload: payload ?? null, + runtime: ctx.name, + }), + } + + gateway = new Gateway({ + logger, + container, + hooks: new Hooks(), + formats: new ProtocolFormats([new JsonFormat()]), + transports: { + http: { + // variance-only cast, same as the transports' own e2e suites: + // this api's resolve() satisfies what each worker consumes + transport: http as unknown as TransportWorker< + ConnectionType, + ResolvedProcedure + >, + proxyable: ProxyableTransportType.HTTP, + }, + ws: { + transport: ws as unknown as TransportWorker< + ConnectionType, + ResolvedProcedure + >, + proxyable: ProxyableTransportType.WS, + }, + }, + api, + heartbeat: false, + }) + + const hosts = await gateway.start() + record({ event: 'shared-server-hosts', name: ctx.name, hosts }) + + return hosts.map((entry) => ({ + type: entry.type as NeemRuntimeUpstream['type'], + url: entry.url, + })) + }, + async stop() { + const current = gateway + gateway = undefined + await current?.stop() + }, + } + }, +}) diff --git a/packages/neem/tests/e2e/shared-server.spec.ts b/packages/neem/tests/e2e/shared-server.spec.ts new file mode 100644 index 00000000..014e3052 --- /dev/null +++ b/packages/neem/tests/e2e/shared-server.spec.ts @@ -0,0 +1,104 @@ +import { afterEach, describe, expect, it } from 'vitest' + +import type { SpawnedNeem } from './support/e2e.ts' +import { + createNeemFixture, + getDistinctFreePorts, + readRuntimeEvents, + spawnNeem, + waitFor, +} from './support/e2e.ts' + +const fixtures: Array<{ cleanup: () => Promise }> = [] +const spawned: SpawnedNeem[] = [] + +afterEach(async () => { + await Promise.all(spawned.splice(0).map((neem) => neem.stop())) + await Promise.all(fixtures.splice(0).map((fixture) => fixture.cleanup())) +}) + +describe('Neem proxy with a shared transport server', () => { + it('routes HTTP and WS traffic to one upstream registered under both types', async () => { + const fixture = await createNeemFixture({ config: 'shared-server' }) + fixtures.push(fixture) + const [proxyPort] = await getDistinctFreePorts(1) + + const neem = spawnNeem( + ['dev', '--config', fixture.configFile, '--outDir', fixture.outDir], + { + env: { + NEEM_PROXY_PORT: String(proxyPort), + NEEM_RUNTIME_EVENTS_FILE: fixture.eventsFile, + }, + }, + ) + spawned.push(neem) + + await neem.waitForEvent((event) => event.event === 'runtime:ready', 30_000) + + // the gateway reported one bound URL under both proxyable types + const hostsEvent = await waitFor(async () => { + const events = await readRuntimeEvents(fixture.eventsFile) + return events.find((event) => event.event === 'shared-server-hosts') + }, 30_000) + const hosts = hostsEvent.hosts as Array<{ type: string; url: string }> + expect(hosts).toHaveLength(2) + expect(new Set(hosts.map((host) => host.url)).size).toBe(1) + expect(hosts.map((host) => host.type).sort()).toEqual(['http', 'ws']) + + const details = () => + [neem.stdout(), neem.stderr()].filter(Boolean).join('\n') + + // HTTP RPC round-trips through the real native proxy + const echoed = await waitFor( + async () => { + const response = await fetch(`http://127.0.0.1:${proxyPort}/echo`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ ping: 1 }), + }) + if (response.status !== 200) return false + return (await response.json()) as Record + }, + 30_000, + details, + ) + expect(echoed).toMatchObject({ procedure: 'echo', payload: { ping: 1 } }) + + // the host-owned health route flows through the same proxied address + const healthy = await fetch(`http://127.0.0.1:${proxyPort}/healthy`) + expect(healthy.status).toBe(200) + + // a WS upgrade through the same proxy port reaches the ws transport: + // open fires only after the gateway accepted the connection (format + // negotiation via query params), proving upgrade headers and query + // traversed the proxy onto the shared socket + const socket = await waitFor( + () => + new Promise((resolve) => { + const ws = new WebSocket( + `ws://127.0.0.1:${proxyPort}/?accept=application/json&content-type=application/json`, + ) + // bound per attempt: a socket that neither opens nor errors would + // otherwise pin waitFor past its own deadline + const timeout = setTimeout(() => { + ws.close() + resolve(false) + }, 1_000) + ws.onopen = () => { + clearTimeout(timeout) + resolve(ws) + } + ws.onerror = () => { + clearTimeout(timeout) + resolve(false) + } + }), + 30_000, + details, + ) + socket.close() + + await neem.stop() + }, 60_000) +}) diff --git a/packages/neem/tests/unit/proxy.spec.ts b/packages/neem/tests/unit/proxy.spec.ts index da6b6908..dd12feb0 100644 --- a/packages/neem/tests/unit/proxy.spec.ts +++ b/packages/neem/tests/unit/proxy.spec.ts @@ -73,6 +73,36 @@ describe('Neem proxy helpers', () => { ]) }) + it('keeps http and ws upstreams sharing one URL as distinct entries', () => { + // a shared-server transport setup reports the same bound address under + // both proxyable types; merging them would drop one routing table entry + const url = 'http://127.0.0.1:3000/' + const desired = createDesiredUpstreams([ + { + runtimeName: 'api', + upstreams: [ + { type: 'http', url }, + { type: 'ws', url }, + ], + }, + ]) + + expect([...desired.values()]).toEqual([ + expect.objectContaining({ + runtimeName: 'api', + count: 1, + upstream: { type: 'http', url }, + proxyUpstream: expect.objectContaining({ transport: 'http' }), + }), + expect.objectContaining({ + runtimeName: 'api', + count: 1, + upstream: { type: 'ws', url }, + proxyUpstream: expect.objectContaining({ transport: 'ws' }), + }), + ]) + }) + it('creates native proxy options from active manifest runtimes', () => { const config: NeemProxyConfig = { hostname: '127.0.0.1', diff --git a/packages/server/package.json b/packages/server/package.json new file mode 100644 index 00000000..efb8ba02 --- /dev/null +++ b/packages/server/package.json @@ -0,0 +1,90 @@ +{ + "name": "@nmtjs/server", + "description": "Shared HTTP server host for Neemata transports across runtimes.", + "type": "module", + "exports": { + ".": { + "types": "./src/index.ts", + "import": "./dist/index.js", + "module-sync": "./dist/index.js" + }, + "./bun": { + "types": "./src/runtimes/bun.ts", + "import": "./dist/runtimes/bun.js", + "module-sync": "./dist/runtimes/bun.js" + }, + "./deno": { + "types": "./src/runtimes/deno.ts", + "import": "./dist/runtimes/deno.js", + "module-sync": "./dist/runtimes/deno.js" + }, + "./node": { + "types": "./src/runtimes/node.ts", + "import": "./dist/runtimes/node.js", + "module-sync": "./dist/runtimes/node.js" + } + }, + "publishConfig": { + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js", + "module-sync": "./dist/index.js" + }, + "./bun": { + "types": "./dist/runtimes/bun.d.ts", + "import": "./dist/runtimes/bun.js", + "module-sync": "./dist/runtimes/bun.js" + }, + "./deno": { + "types": "./dist/runtimes/deno.d.ts", + "import": "./dist/runtimes/deno.js", + "module-sync": "./dist/runtimes/deno.js" + }, + "./node": { + "types": "./dist/runtimes/node.d.ts", + "import": "./dist/runtimes/node.js", + "module-sync": "./dist/runtimes/node.js" + } + } + }, + "dependencies": { + "crossws": "0.4.6" + }, + "devDependencies": { + "@nmtjs/common": "workspace:*", + "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.67.0" + }, + "peerDependencies": { + "@nmtjs/common": "workspace:*", + "@types/bun": "^1.3.0", + "@types/deno": "^2.3.0", + "@types/node": "^24.13.2", + "uWebSockets.js": "^20.67.0" + }, + "peerDependenciesMeta": { + "@types/bun": { + "optional": true + }, + "@types/deno": { + "optional": true + }, + "@types/node": { + "optional": true + }, + "uWebSockets.js": { + "optional": true + } + }, + "engines": { + "node": ">=24", + "bun": ">=1.3.0", + "deno": ">=2.3.0" + }, + "files": [ + "dist", + "src", + "LICENSE.md", + "README.md" + ] +} diff --git a/packages/server/src/host.ts b/packages/server/src/host.ts new file mode 100644 index 00000000..efb7d38d --- /dev/null +++ b/packages/server/src/host.ts @@ -0,0 +1,107 @@ +import type { + ServerFetchHandler, + ServerHost, + ServerHostOptions, + ServerNativeHandles, + ServerRuntimeName, + ServerWebSocketRegistration, +} from './types.ts' + +export const DEFAULT_MAX_REQUEST_BODY_SIZE = 1024 * 1024 * 128 // 128MiB + +/** + * Reference-counted host base. Registration happens while the host is idle + * (transports register at construction, before any gateway start), and the + * runtime server is materialized lazily in bind() from the collected + * registrations — required by runtimes like Bun where fetch/websocket + * handlers must all be known at serve() time, and it also makes a + * stop()-then-start() cycle rebind cleanly on every runtime. + */ +export abstract class BaseServerHost< + R extends ServerRuntimeName, +> implements ServerHost { + abstract readonly runtime: R + protected fetchHandler?: ServerFetchHandler + protected webSocket?: ServerWebSocketRegistration + #refs = 0 + #bound: Promise | null = null + #closing: Promise | null = null + + constructor(protected readonly options: ServerHostOptions) {} + + abstract get native(): ServerNativeHandles + + setFetchHandler(handler: ServerFetchHandler): void { + if (this.#bound) { + throw new Error('Cannot register a fetch handler on a started server') + } + if (this.fetchHandler) { + throw new Error('A fetch handler is already registered on this server') + } + this.fetchHandler = handler + } + + setWebSocket(registration: ServerWebSocketRegistration): void { + if (this.#bound) { + throw new Error('Cannot register a WebSocket handler on a started server') + } + if (this.webSocket) { + throw new Error( + 'A WebSocket handler is already registered on this server', + ) + } + this.webSocket = registration + } + + isSendSuccess(_status: number): boolean { + return true + } + + async start(): Promise { + this.#refs++ + try { + // a rebind must not race a close() still tearing the old socket down + if (this.#closing) await this.#closing.catch(() => undefined) + this.#bound ??= this.bind() + return await this.#bound + } catch (error) { + // roll back this claim; the last failed claimant clears the rejected + // bind so a later start() can attempt a fresh one + this.#refs-- + if (this.#refs === 0) this.#bound = null + throw error + } + } + + async stop(): Promise { + if (this.#refs === 0) return + this.#refs-- + // the socket closes only after every registrant has stopped + if (this.#refs > 0) return + const bound = this.#bound + if (!bound) return + // a concurrent in-flight bind must settle before closing; #bound stays + // claimed for the whole wait so a racing start() joins the live socket + // instead of binding a second one + await bound.catch(() => undefined) + if (this.#refs > 0 || this.#bound !== bound) return + this.#bound = null + this.#closing = Promise.resolve(this.close()) + try { + await this.#closing + } finally { + this.#closing = null + } + } + + protected get maxRequestBodySize(): number { + return this.options.maxRequestBodySize ?? DEFAULT_MAX_REQUEST_BODY_SIZE + } + + /** + * Materialize the runtime server from the collected registrations, bind + * the socket and resolve with the bound URL. + */ + protected abstract bind(): Promise + protected abstract close(): Promise +} diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts new file mode 100644 index 00000000..3944f5ef --- /dev/null +++ b/packages/server/src/index.ts @@ -0,0 +1,3 @@ +export * from './host.ts' +export * from './types.ts' +export * from './utils.ts' diff --git a/packages/server/src/runtimes/bun.ts b/packages/server/src/runtimes/bun.ts new file mode 100644 index 00000000..5dd925fe --- /dev/null +++ b/packages/server/src/runtimes/bun.ts @@ -0,0 +1,105 @@ +import createAdapter from 'crossws/adapters/bun' + +import type { + ServerHost, + ServerHostOptions, + ServerNativeHandles, +} from '../types.ts' +import { BaseServerHost } from '../host.ts' +import { + InternalServerErrorHttpResponse, + NotFoundHttpResponse, + OkResponse, +} from '../utils.ts' + +class BunServerHost extends BaseServerHost<'bun'> { + readonly runtime = 'bun' as const + #server: Bun.Server | null = null + + get native(): ServerNativeHandles { + return { bun: this.#server ?? undefined } + } + + // Bun send status: >0 = bytes sent, -1 = backpressure applied (will + // drain), 0 = dropped — only a drop is a failed delivery + override isSendSuccess(status: number): boolean { + return status !== 0 + } + + protected async bind(): Promise { + const { listen, tls, runtime } = this.options + const adapter = this.webSocket + ? createAdapter({ hooks: this.webSocket.hooks }) + : null + const wsOptions = this.webSocket?.options + const fetchHandler = this.fetchHandler + + const routes = + typeof runtime?.routes === 'object' && runtime.routes + ? runtime.routes + : {} + + this.#server = globalThis.Bun.serve({ + ...runtime, + // Bun's own default (128MiB) applies when neither option is set + maxRequestBodySize: + runtime?.maxRequestBodySize ?? this.options.maxRequestBodySize, + unix: listen.unix as string, + port: listen.port ?? 0, + hostname: listen.hostname, + reusePort: listen.reusePort, + tls: tls + ? { + cert: tls.cert, + key: tls.key, + passphrase: tls.passphrase, + } + : undefined, + websocket: adapter ? { ...wsOptions, ...adapter.websocket } : undefined, + // bare handler instead of a method map: health probes may use HEAD, + // keep the route method-agnostic like the other runtime hosts + routes: Object.assign({}, routes, { + '/healthy': OkResponse, + }) as any, + async fetch(request: Request, server: Bun.Server) { + try { + if (request.headers.get('upgrade') === 'websocket') { + if (!adapter) return NotFoundHttpResponse() + return await adapter.handleUpgrade(request, server) + } + if (!fetchHandler) return NotFoundHttpResponse() + const url = new URL(request.url) + const { body, headers, method } = request + return await fetchHandler( + { url, method, headers }, + body, + request.signal, + ) + } catch (err) { + // TODO: proper logging + console.error(err) + return InternalServerErrorHttpResponse() + } + }, + } as any) + + // Bun reports unix sockets as `unix:///path`; keep the cross-runtime + // `proto+unix://` contract instead + if (listen.unix) { + return `${tls ? 'https' : 'http'}+unix://${listen.unix}` + } + return this.#server!.url.href + } + + protected async close(): Promise { + const server = this.#server + this.#server = null + if (server) await server.stop() + } +} + +export function createServerHost( + options: ServerHostOptions<'bun'>, +): ServerHost<'bun'> { + return new BunServerHost(options) +} diff --git a/packages/server/src/runtimes/deno.ts b/packages/server/src/runtimes/deno.ts new file mode 100644 index 00000000..6adecef5 --- /dev/null +++ b/packages/server/src/runtimes/deno.ts @@ -0,0 +1,130 @@ +import createAdapter from 'crossws/adapters/deno' + +import type { + DenoServer, + ServerHost, + ServerHostOptions, + ServerNativeHandles, +} from '../types.ts' +import { BaseServerHost } from '../host.ts' +import { + InternalServerErrorHttpResponse, + NotFoundHttpResponse, + OkResponse, +} from '../utils.ts' + +interface DenoNetAddr { + transport: 'tcp' | 'udp' + hostname: string + port: number +} + +interface DenoUnixAddr { + transport: 'unix' | 'unixpacket' + path: string +} + +interface DenoVsockAddr { + transport: 'vsock' + cid: number + port: number +} + +type DenoAddr = DenoNetAddr | DenoUnixAddr | DenoVsockAddr + +class DenoServerHost extends BaseServerHost<'deno'> { + readonly runtime = 'deno' as const + #server: DenoServer | null = null + + get native(): ServerNativeHandles { + return { deno: this.#server ?? undefined } + } + + protected bind(): Promise { + const { listen, tls } = this.options + const adapter = this.webSocket + ? createAdapter({ hooks: this.webSocket.hooks }) + : null + const fetchHandler = this.fetchHandler + + const listenOptions = listen.unix + ? { path: listen.unix } + : { + port: listen.port, + hostname: listen.hostname, + reusePort: listen.reusePort, + } + const options = { + ...listenOptions, + tls: tls + ? { + cert: tls.cert, + key: tls.key, + passphrase: tls.passphrase, + } + : undefined, + } + + return new Promise((resolve) => { + const server = globalThis.Deno.serve({ + ...this.options.runtime, + ...options, + handler: async (request: Request, info: any) => { + const url = new URL(request.url) + if (url.pathname === '/healthy') return OkResponse() + try { + if (request.headers.get('upgrade') === 'websocket') { + if (!adapter) return NotFoundHttpResponse() + return await adapter.handleUpgrade(request, info as any) + } + if (!fetchHandler) return NotFoundHttpResponse() + const { headers, method, body } = request + return await fetchHandler( + { url, method, headers }, + body, + request.signal, + ) + } catch (err) { + // TODO: proper logging + console.error(err) + return InternalServerErrorHttpResponse() + } + }, + onListen: (addr: DenoAddr) => { + this.#server = server + setTimeout(() => { + resolve(formatDenoUrl(addr, Boolean(tls))) + }, 1) + }, + }) + }) + } + + protected async close(): Promise { + const server = this.#server + this.#server = null + if (server) await server.shutdown() + } +} + +function formatDenoUrl(addr: DenoAddr, secure: boolean): string { + const proto = secure ? 'https' : 'http' + switch (addr.transport) { + case 'unix': + case 'unixpacket': + return `${proto}+unix://${addr.path}` + case 'tcp': + case 'udp': + return `${proto}://${addr.hostname}:${addr.port}` + case 'vsock': + return `vsock://${addr.cid}:${addr.port}` + default: + throw new Error(`Unsupported address transport`) + } +} + +export function createServerHost( + options: ServerHostOptions<'deno'>, +): ServerHost<'deno'> { + return new DenoServerHost(options) +} diff --git a/packages/server/src/runtimes/node.ts b/packages/server/src/runtimes/node.ts new file mode 100644 index 00000000..fee45b85 --- /dev/null +++ b/packages/server/src/runtimes/node.ts @@ -0,0 +1,400 @@ +import { Buffer } from 'node:buffer' + +import type { TemplatedApp } from 'uWebSockets.js' +import createAdapter from 'crossws/adapters/uws' +import { App, SSLApp, us_socket_local_port } from 'uWebSockets.js' + +import type { + ServerHost, + ServerHostOptions, + ServerNativeHandles, + ServerWebSocketRuntimeOptions, +} from '../types.ts' +import { BaseServerHost } from '../host.ts' +import { + InternalServerErrorHttpResponse, + NotFoundHttpResponse, + OkResponse, + PayloadTooLargeError, + PayloadTooLargeHttpResponse, +} from '../utils.ts' + +const statusResponse = OkResponse() +const statusResponseBuffer = await statusResponse.arrayBuffer() + +/** + * uWS defaults to 16KiB and CLOSES the socket on larger frames — the + * gateway's own upload credit grants (64KiB) plus the 5-byte frame header + * already exceed that, killing every blob upload. Inline WS payloads are + * capped here at 1 MiB by design: larger data should ride blob streams, + * which are chunked at credit size. + */ +export const DEFAULT_WS_MAX_PAYLOAD = 1024 * 1024 +/** + * uWS defaults to 64KiB and DROPS frames above it — the same order of + * magnitude as outstanding stream credit; a higher ceiling keeps + * abort-on-drop a safety net, not a common path. + */ +export const DEFAULT_WS_MAX_BACKPRESSURE = 1024 * 1024 + +/** + * ?? per field instead of spread-defaults: an explicitly-undefined user + * value (e.g. from an optional env var) must not erase the framework + * defaults and resurrect uWS's frame-killing 16KiB/64KiB limits. + */ +export function resolveUwsWsOptions( + ws?: ServerWebSocketRuntimeOptions['node'], +) { + return { + ...ws, + maxPayloadLength: ws?.maxPayloadLength ?? DEFAULT_WS_MAX_PAYLOAD, + maxBackpressure: ws?.maxBackpressure ?? DEFAULT_WS_MAX_BACKPRESSURE, + } +} + +type UwsResponse = Parameters< + Parameters['any']>[1] +>[0] & { + aborted?: boolean + wakeWritable?: () => void + cancelBody?: () => void +} + +class NodeServerHost extends BaseServerHost<'node'> { + readonly runtime = 'node' as const + #server: TemplatedApp | null = null + + get native(): ServerNativeHandles { + return { node: this.#server ?? undefined } + } + + // uWS send status: 1 = sent, 0 = buffered (will drain), 2 = dropped + // due to backpressure limit — only a drop is a failed delivery + override isSendSuccess(status: number): boolean { + return status !== 2 + } + + protected bind(): Promise { + const server = this.options.tls + ? SSLApp({ + passphrase: this.options.tls.passphrase, + key_file_name: this.options.tls.key, + cert_file_name: this.options.tls.cert, + }) + : App() + + if (this.webSocket) { + const adapter = createAdapter({ hooks: this.webSocket.hooks }) + server.ws('/*', { + ...resolveUwsWsOptions(this.webSocket.options), + ...adapter.websocket, + }) + } + + server + // any() instead of get(): health probes may use HEAD, keep the route + // method-agnostic like the other runtime hosts + .any('/healthy', (res) => { + res.cork(() => { + res + .writeStatus( + `${statusResponse.status} ${statusResponse.statusText}`, + ) + .end(statusResponseBuffer) + }) + }) + .any('/*', (res, req) => this.handleRequest(res as UwsResponse, req)) + + this.#server = server + + return new Promise((resolve, reject) => { + const { listen, tls } = this.options + const proto = tls ? 'https' : 'http' + if (listen.unix) { + server.listen_unix((socket) => { + if (socket) { + resolve(`${proto}+unix://${listen.unix}`) + } else { + reject(new Error('Failed to start server')) + } + }, listen.unix) + } else if (typeof listen.port === 'number') { + const hostname = listen.hostname || '127.0.0.1' + server.listen(hostname, listen.port, (socket) => { + if (socket) { + resolve(`${proto}://${hostname}:${us_socket_local_port(socket)}`) + } else { + reject(new Error('Failed to start server')) + } + }) + } else { + reject(new Error('Invalid listen parameters')) + } + }) + } + + protected async close(): Promise { + this.#server?.close() + this.#server = null + } + + private async handleRequest( + uwsRes: UwsResponse, + req: Parameters[1]>[1], + ) { + const res = uwsRes + const requestController = new AbortController() + let aborted = false + let bodyController: ReadableStreamDefaultController | undefined + + res.onAborted(() => { + aborted = true + uwsRes.aborted = true + uwsRes.wakeWritable?.() + uwsRes.cancelBody?.() + requestController.abort() + + try { + bodyController?.error(requestController.signal.reason) + } catch {} + }) + + let response = NotFoundHttpResponse() + + const headers = new Headers() + const method = req.getMethod() + req.forEach((k, v) => headers.append(k, v)) + + const host = headers.get('host') || 'localhost' + const forwardedProto = headers.get('x-forwarded-proto') + const proto = forwardedProto + ? forwardedProto === 'https' + ? 'https' + : 'http' + : this.options.tls + ? 'https' + : 'http' + const url = new URL(req.getUrl(), `${proto}://${host}`) + url.search = req.getQuery() ? `?${req.getQuery()}` : '' + // Upgrade requests reach here only when no WebSocket handler is mounted + // (a mounted ws route claims them inside uWS): keep parity with the + // other runtimes and reject instead of treating them as plain HTTP + const isUpgrade = headers.get('upgrade') === 'websocket' + if (this.fetchHandler && !isUpgrade) { + const maxBodySize = this.maxRequestBodySize + try { + // uWS delivers chunks without backpressure, so cap what gets copied + // into memory before the whole body arrives + let received = 0 + let capped = false + const body = new ReadableStream({ + start(controller) { + bodyController = controller + res.onDataV2((chunk, maxRemainingBodyLength) => { + if (aborted || capped) return + if (chunk) { + received += chunk.byteLength + if (received > maxBodySize) { + capped = true + controller.error(new PayloadTooLargeError()) + return + } + const copy = Buffer.allocUnsafe(chunk.byteLength) + copy.set(new Uint8Array(chunk)) + controller.enqueue(copy) + } + if (maxRemainingBodyLength === 0n) controller.close() + }) + }, + }) + response = await this.fetchHandler( + { url, method, headers }, + body, + requestController.signal, + ) + } catch (err) { + // a tenant that consumed the capped body without mapping the error + // itself must still produce a 413, not a generic 500 + if (err instanceof PayloadTooLargeError) { + response = PayloadTooLargeHttpResponse() + } else { + // TODO: proper logging + console.error(err) + response = InternalServerErrorHttpResponse() + } + } + } + if (aborted) return undefined + else { + const fixedContentLength = response.body + ? getContentLength(response.headers) + : undefined + res.cork(() => { + if (aborted) return undefined + res.writeStatus(`${response.status.toString()} ${response.statusText}`) + response.headers.forEach((v, k) => { + if ( + typeof fixedContentLength === 'number' && + k.toLowerCase() === 'content-length' + ) { + return + } + + res.writeHeader(k, v) + }) + }) + if (response.body) { + try { + await handleResponseBody(uwsRes, response, fixedContentLength) + } catch { + if (!aborted) res.cork(() => res.close()) + } + } else { + if (!aborted) res.cork(() => res.end()) + } + } + } +} + +async function handleResponseBody( + res: UwsResponse, + response: Response, + fixedContentLength?: number, +): Promise { + if (!response.body) return + + if (typeof fixedContentLength === 'number') { + await handleFixedLengthStream(res, response.body, fixedContentLength) + return + } + + await handleChunkedStream(res, response.body) +} + +async function handleFixedLengthStream( + res: UwsResponse, + body: ReadableStream, + totalSize: number, +): Promise { + const reader = body.getReader() + try { + if (totalSize === 0) { + if (!res.aborted) res.cork(() => res.endWithoutBody(0)) + return + } + + let responded = false + while (!res.aborted && !responded) { + const { done, value } = await reader.read() + if (done) break + if (value.byteLength === 0) continue + responded = await handleFixedChunk(res, value, totalSize) + } + + if (!responded && !res.aborted) res.cork(() => res.close()) + } finally { + reader.releaseLock() + } +} + +// exported for tests: the waiter dispatch is timing-sensitive and needs +// deterministic coverage against a controlled response double +export async function handleChunkedStream( + res: UwsResponse, + body: ReadableStream, +): Promise { + const reader = body.getReader() + // abort must also cancel a pending read(): a stalled source would otherwise + // keep the pump and the reader lock alive forever + res.cancelBody = () => { + reader.cancel().catch(() => {}) + } + // uWS honors only the first onWritable registration per response, so a + // single handler dispatches drain (and abort) events to the pending waiter + let writableHandlerRegistered = false + const waitWritable = () => + new Promise((resolve, reject) => { + if (!writableHandlerRegistered) { + writableHandlerRegistered = true + res.onWritable(() => { + res.wakeWritable?.() + return true + }) + } + res.wakeWritable = () => { + res.wakeWritable = undefined + if (res.aborted) reject(new Error('Response aborted')) + else resolve() + } + }) + try { + while (!res.aborted) { + const { done, value } = await reader.read() + if (done) break + if (value.byteLength === 0) continue + + // cork() returns the response object, not write()'s backpressure flag + let ok = true + res.cork(() => { + ok = res.write(value) + }) + if (!ok) await waitWritable() + } + + if (!res.aborted) res.cork(() => res.end()) + } finally { + res.cancelBody = undefined + reader.releaseLock() + } +} + +function handleFixedChunk( + res: UwsResponse, + chunk: Uint8Array, + totalSize: number, +): Promise { + return new Promise((resolve, reject) => { + const chunkOffset = res.getWriteOffset() + + const write = (offset: number) => { + if (res.aborted) { + reject(new Error('Response aborted')) + return false + } + + const relativeOffset = offset - chunkOffset + const remaining = + relativeOffset > 0 ? chunk.subarray(relativeOffset) : chunk + + let ok = false + let done = false + res.cork(() => { + ;[ok, done] = res.tryEnd(remaining, totalSize) + }) + + if (done || ok) { + resolve(done) + return ok + } + + res.onWritable(write) + return ok + } + + write(chunkOffset) + }) +} + +function getContentLength(headers: Headers): number | undefined { + const raw = headers.get('content-length') + if (!raw) return undefined + + const value = Number(raw) + return Number.isSafeInteger(value) && value >= 0 ? value : undefined +} + +export function createServerHost( + options: ServerHostOptions<'node'>, +): ServerHost<'node'> { + return new NodeServerHost(options) +} diff --git a/packages/server/src/types.ts b/packages/server/src/types.ts new file mode 100644 index 00000000..38367199 --- /dev/null +++ b/packages/server/src/types.ts @@ -0,0 +1,138 @@ +import type { MaybePromise, OneOf } from '@nmtjs/common' +import type { Hooks } from 'crossws' + +export type ServerRuntimeName = 'node' | 'bun' | 'deno' + +export type ServerListenOptions = OneOf< + [{ port: number; hostname?: string; reusePort?: boolean }, { unix: string }] +> + +export type ServerTlsOptions = { + /** + * File path or inlined TLS certificate in PEM format (required). + */ + cert?: string + /** + * File path or inlined TLS private key in PEM format (required). + */ + key?: string + /** + * Passphrase for the private key (optional). + */ + passphrase?: string +} + +export type ServerRequest = { + url: URL + method: string + headers: Headers +} + +export type ServerFetchHandler = ( + request: ServerRequest, + body: ReadableStream | null, + signal: AbortSignal, +) => MaybePromise + +/** + * WebSocket behavior overrides per runtime. These belong to the WebSocket + * tenant of a host, not the host itself — though Bun applies them + * server-wide, since `Bun.serve` accepts a single `websocket` config. + */ +export type ServerWebSocketRuntimeOptions = { + node: Partial< + Pick< + import('uWebSockets.js').WebSocketBehavior, + | 'maxBackpressure' + | 'maxPayloadLength' + | 'maxLifetime' + | 'closeOnBackpressureLimit' + | 'idleTimeout' + | 'compression' + | 'sendPingsAutomatically' + > + > + bun: Partial< + Pick< + import('bun').WebSocketHandler, + | 'backpressureLimit' + | 'maxPayloadLength' + | 'closeOnBackpressureLimit' + | 'idleTimeout' + | 'perMessageDeflate' + | 'sendPings' + > + > + deno: {} +} + +export type ServerWebSocketRegistration< + R extends ServerRuntimeName = ServerRuntimeName, +> = { + hooks: Partial + options?: ServerWebSocketRuntimeOptions[R] +} + +/** + * Server-level runtime options: settings that configure the socket/server + * itself rather than any single protocol mounted on it. + */ +export type ServerRuntimeOptions = { + node: {} + bun: Partial< + Pick< + import('bun').Serve.Options, + 'development' | 'id' | 'maxRequestBodySize' | 'idleTimeout' | 'ipv6Only' + > & + import('bun').Serve.Routes + > + deno: {} +} + +export type ServerHostOptions = + { + listen: ServerListenOptions + tls?: ServerTlsOptions + /** + * Maximum request body size in bytes accepted by the host before the + * request is rejected/aborted. Defaults to 128MiB (Bun's own default, kept + * consistent across runtimes). Transports may enforce stricter caps of + * their own on top of this bound. + */ + maxRequestBodySize?: number + runtime?: ServerRuntimeOptions[R] + } + +export type DenoServer = ReturnType + +export type ServerNativeHandles = { + node?: import('uWebSockets.js').TemplatedApp + bun?: import('bun').Server + deno?: DenoServer +} + +/** + * A runtime HTTP server that transports mount onto instead of owning a + * socket. One host can carry both an HTTP fetch handler and a WebSocket + * upgrade handler, so multiple transports share a single listen address. + * + * Lifecycle is reference-counted: every registrant calls start()/stop() + * through its own lifecycle, the socket binds on the first start() and + * closes on the last stop(). + */ +export interface ServerHost { + readonly runtime: R + /** + * Raw runtime server handle; populated only while the host is bound. + */ + readonly native: ServerNativeHandles + setFetchHandler(handler: ServerFetchHandler): void + setWebSocket(registration: ServerWebSocketRegistration): void + /** + * Numeric WebSocket peer send() statuses are runtime-specific (uWS vs + * Bun), so only the host knows whether one means a dropped frame. + */ + isSendSuccess(status: number): boolean + start(): Promise + stop(): Promise +} diff --git a/packages/server/src/utils.ts b/packages/server/src/utils.ts new file mode 100644 index 00000000..a2753657 --- /dev/null +++ b/packages/server/src/utils.ts @@ -0,0 +1,30 @@ +export const NotFoundHttpResponse = () => + new Response('Not Found', { + status: 404, + headers: { 'Content-Type': 'text/plain' }, + }) + +export const InternalServerErrorHttpResponse = () => + new Response('Internal Server Error', { + status: 500, + headers: { 'Content-Type': 'text/plain' }, + }) + +export const OkResponse = () => new Response('OK', { status: 200 }) + +export const PayloadTooLargeHttpResponse = () => + new Response('Payload Too Large', { + status: 413, + headers: { 'Content-Type': 'text/plain' }, + }) + +/** + * Shared error class: thrown inside runtime body translation and matched by + * `instanceof` in transports to map onto 413 responses — both sides must see + * the same class identity. + */ +export class PayloadTooLargeError extends Error { + constructor(message = 'Payload Too Large') { + super(message) + } +} diff --git a/packages/server/tests/chunked-stream.spec.ts b/packages/server/tests/chunked-stream.spec.ts new file mode 100644 index 00000000..287b7f84 --- /dev/null +++ b/packages/server/tests/chunked-stream.spec.ts @@ -0,0 +1,147 @@ +import { describe, expect, it } from 'vitest' + +import { handleChunkedStream } from '../src/runtimes/node.ts' + +describe('node host', () => { + describe('handleChunkedStream writable dispatcher', () => { + // scripted stand-in for uWS HttpResponse: only the first onWritable + // registration takes effect, matching real uWS behavior + function createResDouble(writeResults: boolean[]) { + const results = [...writeResults] + let writes = 0 + let ends = 0 + let registrations = 0 + let handler: ((offset: number) => boolean) | undefined + const res: any = { + aborted: false, + wakeWritable: undefined, + cork(cb: () => void) { + cb() + return res + }, + write() { + writes++ + return results.length > 0 ? (results.shift() as boolean) : true + }, + end() { + ends++ + return res + }, + onWritable(h: (offset: number) => boolean) { + registrations++ + handler ??= h + return res + }, + } + return { + res, + fireWritable: () => handler!(0), + counts: () => ({ writes, ends, registrations }), + // mirrors what the route's onAborted handler does + abort: () => { + res.aborted = true + res.wakeWritable?.() + res.cancelBody?.() + }, + } + } + + const chunks = (n: number) => + new ReadableStream({ + start(controller) { + for (let i = 0; i < n; i++) controller.enqueue(new Uint8Array(8)) + controller.close() + }, + }) + + const tick = () => new Promise((resolve) => setImmediate(resolve)) + + it('resumes exactly one pending waiter per drain across cycles', async () => { + const double = createResDouble([false, true, false, true]) + const pump = handleChunkedStream(double.res, chunks(4)) + + await tick() + expect(double.counts()).toEqual({ writes: 1, ends: 0, registrations: 1 }) + + double.fireWritable() + await tick() + // resumed once: write 2 succeeded, write 3 hit backpressure again and + // parked a second waiter without re-registering the handler + expect(double.counts()).toEqual({ writes: 3, ends: 0, registrations: 1 }) + + double.fireWritable() + await tick() + await pump + expect(double.counts()).toEqual({ writes: 4, ends: 1, registrations: 1 }) + }) + + it('writable callback without a pending waiter is a no-op returning true', async () => { + const double = createResDouble([false]) + const pump = handleChunkedStream(double.res, chunks(1)) + + await tick() + expect(double.fireWritable()).toBe(true) + await tick() + await pump + + // stream is finished; a late drain event finds no waiter + expect(() => double.fireWritable()).not.toThrow() + expect(double.fireWritable()).toBe(true) + }) + + it('abort while a waiter is pending settles the wait and exits the pump', async () => { + const double = createResDouble([false]) + const body = chunks(2) + const pump = handleChunkedStream(double.res, body) + + await tick() + expect(double.counts().writes).toBe(1) + + double.abort() + await expect(pump).rejects.toThrow('Response aborted') + expect(double.counts()).toEqual({ writes: 1, ends: 0, registrations: 1 }) + // reader lock is released on exit + expect(() => body.getReader()).not.toThrow() + }) + + it('handles a drain cycle followed by an abort', async () => { + const double = createResDouble([false, false]) + const pump = handleChunkedStream(double.res, chunks(3)) + + await tick() + expect(double.counts().writes).toBe(1) + + double.fireWritable() + await tick() + // resumed and immediately backpressured again + expect(double.counts().writes).toBe(2) + + double.abort() + await expect(pump).rejects.toThrow('Response aborted') + expect(double.counts()).toEqual({ writes: 2, ends: 0, registrations: 1 }) + }) + + it('abort while read() is stalled cancels the reader and exits the pump', async () => { + const double = createResDouble([]) + let cancelled = false + const body = new ReadableStream({ + // source never produces: the pump parks inside reader.read() + pull: () => new Promise(() => {}), + cancel() { + cancelled = true + }, + }) + const pump = handleChunkedStream(double.res, body) + + await tick() + expect(double.counts().writes).toBe(0) + + double.abort() + // reader.cancel() resolves the pending read as done and the pump exits + // without writing or ending the aborted response + await pump + expect(cancelled).toBe(true) + expect(double.counts()).toEqual({ writes: 0, ends: 0, registrations: 0 }) + }) + }) +}) diff --git a/packages/server/tests/shared-host.spec.ts b/packages/server/tests/shared-host.spec.ts new file mode 100644 index 00000000..2b74ac44 --- /dev/null +++ b/packages/server/tests/shared-host.spec.ts @@ -0,0 +1,170 @@ +import { defineHooks } from 'crossws' +import { describe, expect, it } from 'vitest' + +import { createServerHost } from '../src/runtimes/node.ts' + +const openSocket = (url: string) => + new Promise((resolve, reject) => { + const ws = new WebSocket(url) + ws.onopen = () => resolve(ws) + ws.onerror = () => reject(new Error('WebSocket failed to connect')) + }) + +const echoHooks = () => + defineHooks({ + message(peer, message) { + peer.send(`echo:${message.text()}`) + }, + }) + +describe('shared node server host', () => { + it('serves HTTP, WebSocket and /healthy on a single socket', async () => { + const host = createServerHost({ + listen: { port: 0, hostname: '127.0.0.1' }, + }) + host.setFetchHandler( + async (request) => new Response(`hello:${request.url.pathname}`), + ) + host.setWebSocket({ hooks: echoHooks() }) + + const url = await host.start() + try { + const response = await fetch(`${url}/route`) + expect(response.status).toBe(200) + expect(await response.text()).toBe('hello:/route') + + const healthy = await fetch(`${url}/healthy`) + expect(healthy.status).toBe(200) + + const ws = await openSocket(url.replace('http', 'ws')) + const reply = new Promise((resolve) => { + ws.onmessage = (event) => resolve(String(event.data)) + }) + ws.send('ping') + expect(await reply).toBe('echo:ping') + ws.close() + } finally { + await host.stop() + } + }) + + it('closes the socket only after the last registrant stops', async () => { + const host = createServerHost({ + listen: { port: 0, hostname: '127.0.0.1' }, + }) + host.setFetchHandler(async () => new Response('ok')) + host.setWebSocket({ hooks: echoHooks() }) + + // two registrants claim the same bound socket + const first = await host.start() + const second = await host.start() + expect(second).toBe(first) + + await host.stop() + // one registrant remains: still serving + const response = await fetch(`${first}/healthy`) + expect(response.status).toBe(200) + + await host.stop() + await expect(fetch(`${first}/healthy`)).rejects.toThrow() + }) + + it('start racing an in-flight final stop reclaims the live socket', async () => { + const host = createServerHost({ + listen: { port: 0, hostname: '127.0.0.1' }, + }) + host.setFetchHandler(async () => new Response('ok')) + + const first = await host.start() + // final stop() not awaited: the racing start() must join the still-bound + // socket instead of binding a second server the stop then orphans + const stopping = host.stop() + const second = await host.start() + await stopping + + expect(second).toBe(first) + try { + const response = await fetch(`${first}/healthy`) + expect(response.status).toBe(200) + } finally { + await host.stop() + } + await expect(fetch(`${first}/healthy`)).rejects.toThrow() + }) + + it('rebinds after a full stop', async () => { + const host = createServerHost({ + listen: { port: 0, hostname: '127.0.0.1' }, + }) + host.setFetchHandler(async () => new Response('ok')) + + const first = await host.start() + await host.stop() + + const second = await host.start() + try { + const response = await fetch(`${second}/healthy`) + expect(response.status).toBe(200) + } finally { + await host.stop() + } + // port 0 means the rebound socket may live elsewhere; only liveness + // of the second URL matters + expect(second).not.toBe('') + expect(first).not.toBe('') + }) + + it('rejects upgrade requests without a WebSocket tenant', async () => { + const host = createServerHost({ + listen: { port: 0, hostname: '127.0.0.1' }, + }) + host.setFetchHandler(async () => new Response('ok')) + + const url = await host.start() + try { + await expect(openSocket(url.replace('http', 'ws'))).rejects.toThrow() + // plain HTTP is unaffected + const response = await fetch(url) + expect(response.status).toBe(200) + } finally { + await host.stop() + } + }) + + it('serves 404 for plain HTTP without a fetch tenant', async () => { + const host = createServerHost({ + listen: { port: 0, hostname: '127.0.0.1' }, + }) + host.setWebSocket({ hooks: echoHooks() }) + + const url = await host.start() + try { + const response = await fetch(`${url}/anything`) + expect(response.status).toBe(404) + // health stays host-owned + const healthy = await fetch(`${url}/healthy`) + expect(healthy.status).toBe(200) + } finally { + await host.stop() + } + }) + + it('rejects duplicate registrations and registration after start', async () => { + const host = createServerHost({ + listen: { port: 0, hostname: '127.0.0.1' }, + }) + host.setFetchHandler(async () => new Response('ok')) + expect(() => host.setFetchHandler(async () => new Response('no'))).toThrow( + 'already registered', + ) + + await host.start() + try { + expect(() => host.setWebSocket({ hooks: echoHooks() })).toThrow( + 'started server', + ) + } finally { + await host.stop() + } + }) +}) diff --git a/packages/server/tsconfig.build.json b/packages/server/tsconfig.build.json new file mode 100644 index 00000000..894a1b64 --- /dev/null +++ b/packages/server/tsconfig.build.json @@ -0,0 +1,12 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "rootDir": "./src", + "noEmit": false, + "outDir": "./dist", + "declaration": true, + "sourceMap": true, + "isolatedDeclarations": false + }, + "include": ["src/**/*"] +} diff --git a/packages/server/tsconfig.json b/packages/server/tsconfig.json new file mode 100644 index 00000000..20c38438 --- /dev/null +++ b/packages/server/tsconfig.json @@ -0,0 +1,7 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "." + }, + "include": ["src/**/*", "tests/**/*"] +} diff --git a/packages/server/vitest.config.ts b/packages/server/vitest.config.ts new file mode 100644 index 00000000..4114b485 --- /dev/null +++ b/packages/server/vitest.config.ts @@ -0,0 +1,5 @@ +import { defineProject } from 'vitest/config' + +export default defineProject({ + test: { environment: 'node', include: ['tests/**/*.spec.ts'] }, +}) diff --git a/packages/ws-transport/package.json b/packages/ws-transport/package.json index bb62d3ea..d2bf9ebe 100644 --- a/packages/ws-transport/package.json +++ b/packages/ws-transport/package.json @@ -57,8 +57,10 @@ "@nmtjs/common": "workspace:*", "@nmtjs/contract": "workspace:*", "@nmtjs/gateway": "workspace:*", + "@nmtjs/http-transport": "workspace:*", "@nmtjs/json-format": "workspace:*", "@nmtjs/protocol": "workspace:*", + "@nmtjs/server": "workspace:*", "@nmtjs/type": "workspace:*", "@nmtjs/ws-client": "workspace:*", "uWebSockets.js": "github:uNetworking/uWebSockets.js#v20.67.0" @@ -69,6 +71,7 @@ "@nmtjs/core": "workspace:*", "@nmtjs/gateway": "workspace:*", "@nmtjs/protocol": "workspace:*", + "@nmtjs/server": "workspace:*", "@types/bun": "^1.3.0", "@types/deno": "^2.3.0", "@types/node": "^24.13.2", diff --git a/packages/ws-transport/src/adapter.ts b/packages/ws-transport/src/adapter.ts new file mode 100644 index 00000000..3e20b794 --- /dev/null +++ b/packages/ws-transport/src/adapter.ts @@ -0,0 +1,36 @@ +import type { + ServerHost, + ServerHostOptions, + ServerRuntimeName, +} from '@nmtjs/server' + +import type { WsAdapterParams, WsAdapterServer } from './types.ts' + +/** + * Shared-server mode mounts onto the provided host; listen mode owns a + * private one — either way a ServerHost is the only socket owner, so both + * modes exercise the same server code path. + */ +export function createHostAdapter( + createHost: (options: ServerHostOptions) => ServerHost, + params: WsAdapterParams, +): WsAdapterServer { + let host: ServerHost + if (params.server) { + host = params.server + } else if (params.listen) { + host = createHost({ + listen: params.listen, + tls: params.tls, + runtime: params.runtime, + }) + } else { + throw new Error('Either `server` or `listen` option is required') + } + host.setWebSocket({ hooks: params.wsHooks, options: params.ws }) + return { + start: () => host.start(), + stop: () => host.stop(), + isSendSuccess: (status) => host.isSendSuccess(status), + } +} diff --git a/packages/ws-transport/src/runtimes/bun.ts b/packages/ws-transport/src/runtimes/bun.ts index ea9817cd..64df1091 100644 --- a/packages/ws-transport/src/runtimes/bun.ts +++ b/packages/ws-transport/src/runtimes/bun.ts @@ -1,74 +1,15 @@ import type { ApplicationTransport } from '@nmtjs/application' import type { ConnectionType } from '@nmtjs/protocol' import { ProxyableTransportType } from '@nmtjs/gateway' -import createAdapter from 'crossws/adapters/bun' +import { createServerHost } from '@nmtjs/server/bun' -import type { - WsAdapterParams, - WsAdapterServer, - WsTransportOptions, -} from '../types.ts' +import type { WsAdapterParams, WsTransportOptions } from '../types.ts' +import { createHostAdapter } from '../adapter.ts' import * as injectables from '../injectables.ts' import { createWSTransportWorker } from '../server.ts' -import { - InternalServerErrorHttpResponse, - NotFoundHttpResponse, - StatusResponse, -} from '../utils.ts' -function adapterFactory(params: WsAdapterParams<'bun'>): WsAdapterServer { - const adapter = createAdapter({ hooks: params.wsHooks }) - - let server: Bun.Server | null = null - - function createServer() { - return globalThis.Bun.serve( - // @ts-expect-error ts bs - { - ...params.runtime?.server, - unix: params.listen.unix, - port: params.listen.port, - hostname: params.listen.hostname, - reusePort: params.listen.reusePort, - tls: params.tls - ? { - cert: params.tls.cert, - key: params.tls.key, - passphrase: params.tls.passphrase, - } - : undefined, - websocket: { ...params.runtime?.ws, ...adapter.websocket }, - routes: { '/healthy': StatusResponse() }, - async fetch(request, server) { - try { - if (request.headers.get('upgrade') === 'websocket') { - return await adapter.handleUpgrade(request, server) - } - } catch (err) { - console.error('Error in WebSocket fetch handler', err) - return InternalServerErrorHttpResponse() - } - return NotFoundHttpResponse() - }, - }, - ) - } - - return { - start: async () => { - server = createServer() - return server!.url.href - }, - stop: async () => { - if (server) { - await server.stop() - server = null - } - }, - // Bun send status: >0 = bytes sent, -1 = backpressure applied (will - // drain), 0 = dropped — only a drop is a failed delivery - isSendSuccess: (status) => status !== 0, - } +function adapterFactory(params: WsAdapterParams<'bun'>) { + return createHostAdapter(createServerHost, params) } export const WsTransport: ApplicationTransport< diff --git a/packages/ws-transport/src/runtimes/deno.ts b/packages/ws-transport/src/runtimes/deno.ts index 98f71222..73cc960d 100644 --- a/packages/ws-transport/src/runtimes/deno.ts +++ b/packages/ws-transport/src/runtimes/deno.ts @@ -1,114 +1,15 @@ import type { ApplicationTransport } from '@nmtjs/application' import type { ConnectionType } from '@nmtjs/protocol' import { ProxyableTransportType } from '@nmtjs/gateway' -import createAdapter from 'crossws/adapters/deno' +import { createServerHost } from '@nmtjs/server/deno' -import type { - WsAdapterParams, - WsAdapterServer, - WsTransportOptions, -} from '../types.ts' +import type { WsAdapterParams, WsTransportOptions } from '../types.ts' +import { createHostAdapter } from '../adapter.ts' import * as injectables from '../injectables.ts' import { createWSTransportWorker } from '../server.ts' -import { - InternalServerErrorHttpResponse, - NotFoundHttpResponse, - StatusResponse, -} from '../utils.ts' -type DenoServer = ReturnType -interface DenoNetAddr { - transport: 'tcp' | 'udp' - hostname: string - port: number -} - -interface DenoUnixAddr { - transport: 'unix' | 'unixpacket' - path: string -} - -interface DenoVsockAddr { - transport: 'vsock' - cid: number - port: number -} - -type DenoAddr = DenoNetAddr | DenoUnixAddr | DenoVsockAddr - -function adapterFactory(params: WsAdapterParams<'deno'>): WsAdapterServer { - const adapter = createAdapter({ hooks: params.wsHooks }) - - let server: DenoServer | null = null - - function createServer() { - const listenOptions = params.listen.unix - ? { path: params.listen.unix } - : { - port: params.listen.port, - hostname: params.listen.hostname, - reusePort: params.listen.reusePort, - } - const options = { - ...listenOptions, - tls: params.tls - ? { - cert: params.tls.cert, - key: params.tls.key, - passphrase: params.tls.passphrase, - } - : undefined, - } - - return new Promise<{ server: DenoServer; addr: DenoAddr }>((resolve) => { - const server = globalThis.Deno.serve({ - ...params.runtime?.server, - ...options, - handler: async (request: Request, info: any) => { - const url = new URL(request.url) - if (url.pathname === '/healthy') return StatusResponse() - try { - if (request.headers.get('upgrade') === 'websocket') { - return await adapter.handleUpgrade(request, info as any) - } - } catch (err) { - console.error('Error in WebSocket fetch handler', err) - return InternalServerErrorHttpResponse() - } - return NotFoundHttpResponse() - }, - onListen(addr: DenoAddr) { - setTimeout(() => { - resolve({ server, addr }) - }, 1) - }, - }) - }) - } - - return { - start: async () => { - const { server: _server, addr } = await createServer() - server = _server - const proto = params.tls ? 'https' : 'http' - - switch (addr.transport) { - case 'unix': - return `${proto}+unix://${addr.path}` - case 'tcp': { - return `${proto}://${addr.hostname}:${addr.port}` - } - default: - throw new Error(`Unsupported address transport`) - } - }, - stop: async () => { - if (server) { - await server.shutdown() - server = null - } - }, - } +function adapterFactory(params: WsAdapterParams<'deno'>) { + return createHostAdapter(createServerHost, params) } export const WsTransport: ApplicationTransport< diff --git a/packages/ws-transport/src/runtimes/node.ts b/packages/ws-transport/src/runtimes/node.ts index 25fcd10b..2c5cd467 100644 --- a/packages/ws-transport/src/runtimes/node.ts +++ b/packages/ws-transport/src/runtimes/node.ts @@ -1,106 +1,23 @@ import type { ApplicationTransport } from '@nmtjs/application' import type { ConnectionType } from '@nmtjs/protocol' import { ProxyableTransportType } from '@nmtjs/gateway' -import createAdapter from 'crossws/adapters/uws' -import { App, SSLApp, us_socket_local_port } from 'uWebSockets.js' +import { createServerHost } from '@nmtjs/server/node' -import type { - WsAdapterParams, - WsAdapterServer, - WsTransportOptions, - WsTransportRuntimeNode, -} from '../types.ts' +import type { WsAdapterParams, WsTransportOptions } from '../types.ts' +import { createHostAdapter } from '../adapter.ts' import * as injectables from '../injectables.ts' import { createWSTransportWorker } from '../server.ts' -import { StatusResponse } from '../utils.ts' -const statusResponse = StatusResponse() -const statusResponseBuffer = await statusResponse.arrayBuffer() +// re-exported for compat: the uWS behavior defaults are applied by the +// shared server host, but they remain part of this transport's public API +export { + DEFAULT_WS_MAX_BACKPRESSURE, + DEFAULT_WS_MAX_PAYLOAD, + resolveUwsWsOptions, +} from '@nmtjs/server/node' -/** - * uWS defaults to 16KiB and CLOSES the socket on larger frames — the - * gateway's own upload credit grants (64KiB) plus the 5-byte frame header - * already exceed that, killing every blob upload. Inline WS payloads are - * capped here at 1 MiB by design: larger data should ride blob streams, - * which are chunked at credit size. - */ -export const DEFAULT_WS_MAX_PAYLOAD = 1024 * 1024 -/** - * uWS defaults to 64KiB and DROPS frames above it — the same order of - * magnitude as outstanding stream credit; a higher ceiling keeps - * abort-on-drop a safety net, not a common path. - */ -export const DEFAULT_WS_MAX_BACKPRESSURE = 1024 * 1024 - -/** - * ?? per field instead of spread-defaults: an explicitly-undefined user - * value (e.g. from an optional env var) must not erase the framework - * defaults and resurrect uWS's frame-killing 16KiB/64KiB limits. - */ -export function resolveUwsWsOptions(ws?: WsTransportRuntimeNode['ws']) { - return { - ...ws, - maxPayloadLength: ws?.maxPayloadLength ?? DEFAULT_WS_MAX_PAYLOAD, - maxBackpressure: ws?.maxBackpressure ?? DEFAULT_WS_MAX_BACKPRESSURE, - } -} - -function adapterFactory(params: WsAdapterParams<'node'>): WsAdapterServer { - const adapter = createAdapter({ hooks: params.wsHooks }) - - const server = params.tls - ? SSLApp({ - passphrase: params.tls.passphrase, - key_file_name: params.tls.key, - cert_file_name: params.tls.cert, - }) - : App() - - server - .ws('/*', { - ...resolveUwsWsOptions(params.runtime?.ws), - ...adapter.websocket, - }) - .get('/healthy', (res) => { - res.cork(() => { - res - .writeStatus(`${statusResponse.status} ${statusResponse.statusText}`) - .end(statusResponseBuffer) - }) - }) - - return { - start: () => - new Promise((resolve, reject) => { - const proto = params.tls ? 'https' : 'http' - if (params.listen.unix) { - server.listen_unix((socket) => { - if (socket) { - resolve(`${proto}+unix://` + params.listen.unix) - } else { - reject(new Error('Failed to start WebSockets server')) - } - }, params.listen.unix) - } else if (typeof params.listen.port === 'number') { - const hostname = params.listen.hostname || '127.0.0.1' - server.listen(hostname, params.listen.port, (socket) => { - if (socket) { - resolve(`${proto}://${hostname}:${us_socket_local_port(socket)}`) - } else { - reject(new Error('Failed to start WebSockets server')) - } - }) - } else { - reject(new Error('Invalid listen parameters')) - } - }), - stop: () => { - server.close() - }, - // uWS send status: 1 = sent, 0 = buffered (will drain), 2 = dropped - // due to backpressure limit — only a drop is a failed delivery - isSendSuccess: (status) => status !== 2, - } +function adapterFactory(params: WsAdapterParams<'node'>) { + return createHostAdapter(createServerHost, params) } export const WsTransport: ApplicationTransport< diff --git a/packages/ws-transport/src/types.ts b/packages/ws-transport/src/types.ts index e83562bd..69a67146 100644 --- a/packages/ws-transport/src/types.ts +++ b/packages/ws-transport/src/types.ts @@ -1,11 +1,16 @@ import type { MaybePromise, OneOf } from '@nmtjs/common' +import type { + ServerHost, + ServerListenOptions, + ServerRequest, + ServerRuntimeName, + ServerRuntimeOptions, + ServerTlsOptions, + ServerWebSocketRuntimeOptions, +} from '@nmtjs/server' import type { Hooks } from 'crossws' -export type WsTransportServerRequest = { - url: URL - method: string - headers: Headers -} +export type WsTransportServerRequest = ServerRequest export type WsTransportPeerContext = { connectionId: string } @@ -13,114 +18,46 @@ declare module 'crossws' { interface PeerContext extends WsTransportPeerContext {} } -export type WsTransportOptions< - R extends keyof WsTransportRuntimes = keyof WsTransportRuntimes, -> = { - listen: WsTransportListenOptions - cors?: WsTransportCorsOptions - tls?: WsTransportTlsOptions - runtime?: WsTransportRuntimes[R] -} - -export type WsTransportCorsCustomParams = { - allowMethods?: string[] - allowHeaders?: string[] - allowCredentials?: string - maxAge?: string - exposeHeaders?: string[] - requestHeaders?: string[] - requestMethod?: string -} - -export type WsTransportCorsOptions = - | true - | string[] - | WsTransportCorsCustomParams - | (( - origin: string, - request: WsTransportServerRequest, - ) => boolean | WsTransportCorsCustomParams) - -export type WsTransportListenOptions = OneOf< - [{ port: number; hostname?: string; reusePort?: boolean }, { unix: string }] -> - -export type WsTransportRuntimeBun = { - ws?: Partial< - Pick< - import('bun').WebSocketHandler, - | 'backpressureLimit' - | 'maxPayloadLength' - | 'closeOnBackpressureLimit' - | 'idleTimeout' - | 'perMessageDeflate' - | 'sendPings' - > - > - server?: Partial< - Pick< - import('bun').Serve.Options, - 'development' | 'id' | 'maxRequestBodySize' | 'idleTimeout' | 'ipv6Only' - > - > -} - -export type WsTransportRuntimeNode = { - /** - * Raw uWS websocket behavior overrides. Unless set, the transport applies - * its own defaults for `maxPayloadLength` and `maxBackpressure` (1 MiB - * each): inline WS payloads are capped at 1 MiB — larger data should ride - * blob streams, which are chunked at credit size — and the value is - * deliberately above the largest upload frame (64KiB credit grant plus the - * frame header), since uWS closes the socket on oversized frames and drops - * frames over the backpressure limit. - */ - ws?: Partial< - Pick< - import('uWebSockets.js').WebSocketBehavior, - | 'maxBackpressure' - | 'maxPayloadLength' - | 'maxLifetime' - | 'closeOnBackpressureLimit' - | 'idleTimeout' - | 'compression' - | 'sendPingsAutomatically' - > - > -} +export type WsTransportListenOptions = ServerListenOptions -export type WsTransportRuntimeDeno = { server?: {} } +export type WsTransportTlsOptions = ServerTlsOptions -export type WsTransportRuntimes = { - bun: WsTransportRuntimeBun - node: WsTransportRuntimeNode - deno: WsTransportRuntimeDeno -} +export type WsTransportRuntimes = ServerRuntimeOptions -export type WsTransportTlsOptions = { - /** - * File path or inlined TLS certificate in PEM format (required). - */ - cert?: string - /** - * File path or inlined TLS private key in PEM format (required). - */ - key?: string +/** + * The transport either owns its socket (`listen` mode) or mounts onto a + * shared `ServerHost` (`server` mode), so one HTTP server can carry + * multiple transports (e.g. HTTP and WS on a single listen address). + */ +export type WsTransportOptions< + R extends ServerRuntimeName = ServerRuntimeName, +> = { /** - * Passphrase for the private key (optional). + * Raw websocket behavior overrides for the runtime. Unless set, the + * transport applies its own defaults — on Node (uWS), + * `maxPayloadLength` and `maxBackpressure` default to 1 MiB each: inline + * WS payloads are capped at 1 MiB — larger data should ride blob + * streams, which are chunked at credit size — and the value is + * deliberately above the largest upload frame (64KiB credit grant plus + * the frame header), since uWS closes the socket on oversized frames and + * drops frames over the backpressure limit. */ - passphrase?: string -} + ws?: ServerWebSocketRuntimeOptions[R] +} & OneOf< + [ + { + listen: WsTransportListenOptions + tls?: WsTransportTlsOptions + runtime?: WsTransportRuntimes[R] + }, + { server: ServerHost }, + ] +> -export type WsAdapterParams< - R extends keyof WsTransportRuntimes = keyof WsTransportRuntimes, -> = { - listen: WsTransportListenOptions - wsHooks: Hooks - cors?: WsTransportCorsOptions - tls?: WsTransportTlsOptions - runtime?: WsTransportRuntimes[R] -} +export type WsAdapterParams = + WsTransportOptions & { + wsHooks: Hooks + } export interface WsAdapterServer { stop: () => MaybePromise @@ -131,5 +68,5 @@ export interface WsAdapterServer { } export type WsAdapterServerFactory< - R extends keyof WsTransportRuntimes = keyof WsTransportRuntimes, + R extends ServerRuntimeName = ServerRuntimeName, > = (params: WsAdapterParams) => WsAdapterServer diff --git a/packages/ws-transport/src/utils.ts b/packages/ws-transport/src/utils.ts index 7d9ae10c..4d91acee 100644 --- a/packages/ws-transport/src/utils.ts +++ b/packages/ws-transport/src/utils.ts @@ -1,6 +1,13 @@ import { ErrorCode } from '@nmtjs/protocol' import { ProtocolError } from '@nmtjs/protocol/server' +// Response helpers live in @nmtjs/server, shared by every runtime host +export { + InternalServerErrorHttpResponse, + NotFoundHttpResponse, + OkResponse as StatusResponse, +} from '@nmtjs/server' + export const InternalError = (message = 'Internal Server Error') => new ProtocolError(ErrorCode.InternalServerError, message) @@ -12,17 +19,3 @@ export const ForbiddenError = (message = 'Forbidden') => export const RequestTimeoutError = (message = 'Request Timeout') => new ProtocolError(ErrorCode.RequestTimeout, message) - -export const NotFoundHttpResponse = () => - new Response('Not Found', { - status: 404, - headers: { 'Content-Type': 'text/plain' }, - }) - -export const InternalServerErrorHttpResponse = () => - new Response('Internal Server Error', { - status: 500, - headers: { 'Content-Type': 'text/plain' }, - }) - -export const StatusResponse = () => new Response('OK', { status: 200 }) diff --git a/packages/ws-transport/tests/shared-server.spec.ts b/packages/ws-transport/tests/shared-server.spec.ts new file mode 100644 index 00000000..caabb7cb --- /dev/null +++ b/packages/ws-transport/tests/shared-server.spec.ts @@ -0,0 +1,111 @@ +import { HttpTransport } from '@nmtjs/http-transport/node' +import { createServerHost } from '@nmtjs/server/node' +import { describe, expect, it, vi } from 'vitest' + +import { WsTransport } from '../src/runtimes/node.ts' + +const textEncoder = new TextEncoder() +const textDecoder = new TextDecoder() + +function createHttpParams() { + const connection = { + id: 'http-connection', + encoder: { + contentType: 'application/json', + encode: (data: unknown) => + textEncoder.encode(JSON.stringify(data ?? null)), + }, + decoder: { + decode: (buffer: Uint8Array) => JSON.parse(textDecoder.decode(buffer)), + }, + [Symbol.asyncDispose]: async () => {}, + } + return { + formats: { + supportsDecoder: (contentType: string) => + contentType.startsWith('application/json'), + supportsEncoder: () => true, + }, + onConnect: vi.fn(async () => connection), + onDisconnect: vi.fn(async () => {}), + onMessage: vi.fn(async () => {}), + resolve: vi.fn(async () => ({ meta: new Map() })), + onRpc: vi.fn(async () => ({ ok: true })), + } as any +} + +function createWsParams() { + const connection = { + id: 'ws-connection', + [Symbol.asyncDispose]: async () => {}, + } + return { + formats: {}, + onConnect: vi.fn(async () => connection), + onDisconnect: vi.fn(async () => {}), + onMessage: vi.fn(async () => {}), + resolve: vi.fn(), + onRpc: vi.fn(), + } as any +} + +const openSocket = (url: string) => + new Promise((resolve, reject) => { + const ws = new WebSocket(url) + ws.onopen = () => resolve(ws) + ws.onerror = () => reject(new Error('WebSocket failed to connect')) + }) + +describe('http + ws transports on a shared server', () => { + it('serves both transports from one listen address', async () => { + const host = createServerHost({ + listen: { port: 0, hostname: '127.0.0.1' }, + }) + + const httpParams = createHttpParams() + const wsParams = createWsParams() + + const httpWorker = await HttpTransport.factory({ server: host }) + const wsWorker = await WsTransport.factory({ server: host }) + + const httpUrl = await httpWorker.start(httpParams) + const wsUrl = await wsWorker.start(wsParams) + // both workers report the same bound address — exactly what the + // gateway hands to the proxy for both upstream types + expect(wsUrl).toBe(httpUrl) + + let httpStopped = false + try { + const response = await fetch(`${httpUrl}/procedure`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: '{"n":1}', + }) + expect(response.status).toBe(200) + expect(await response.json()).toEqual({ ok: true }) + expect(httpParams.onRpc).toHaveBeenCalledOnce() + + const ws = await openSocket(httpUrl.replace('http', 'ws')) + expect(wsParams.onConnect).toHaveBeenCalledOnce() + ws.send(new Uint8Array([1, 2, 3])) + await vi.waitFor(() => expect(wsParams.onMessage).toHaveBeenCalled()) + const { data } = wsParams.onMessage.mock.calls[0][0] + expect(new Uint8Array(data)).toEqual(new Uint8Array([1, 2, 3])) + ws.close() + await vi.waitFor(() => expect(wsParams.onDisconnect).toHaveBeenCalled()) + + // one worker stopping must not kill the other's socket + await httpWorker.stop(httpParams) + httpStopped = true + const wsStillUp = await openSocket(httpUrl.replace('http', 'ws')) + wsStillUp.close() + } finally { + // teardown must run even when the liveness probe above fails, or the + // bound host leaks into the rest of the suite + if (!httpStopped) await httpWorker.stop(httpParams) + await wsWorker.stop(wsParams) + } + + await expect(fetch(`${httpUrl}/healthy`)).rejects.toThrow() + }) +}) diff --git a/packages/ws-transport/tests/stream-e2e.spec.ts b/packages/ws-transport/tests/stream-e2e.spec.ts index 2600ae57..68b8f27a 100644 --- a/packages/ws-transport/tests/stream-e2e.spec.ts +++ b/packages/ws-transport/tests/stream-e2e.spec.ts @@ -9,6 +9,7 @@ import type { TransportWorker, } from '@nmtjs/gateway' import type { ConnectionType } from '@nmtjs/protocol' +import type { ServerWebSocketRuntimeOptions } from '@nmtjs/server' import { RuntimeClient } from '@nmtjs/client' import { blobType, c } from '@nmtjs/contract' import { Container, createLogger, Hooks } from '@nmtjs/core' @@ -29,7 +30,6 @@ import { t } from '@nmtjs/type' import { WsTransportFactory } from '@nmtjs/ws-client' import { afterEach, describe, expect, it } from 'vitest' -import type { WsTransportRuntimeNode } from '../src/types.ts' import { WsTransport } from '../src/runtimes/node.ts' /** @@ -163,7 +163,7 @@ async function createHarness(options: { handlers: Handlers streamIdleTimeout?: number rpcBackpressureWindow?: number - runtimeWs?: WsTransportRuntimeNode['ws'] + runtimeWs?: ServerWebSocketRuntimeOptions['node'] plugins?: ClientPlugin[] }) { const logger = createLogger({ pinoOptions: { enabled: false } }, 'stream-e2e') @@ -181,7 +181,7 @@ async function createHarness(options: { const transport = WsTransport.factory({ listen: { port: 0, hostname: '127.0.0.1' }, - runtime: options.runtimeWs ? { ws: options.runtimeWs } : undefined, + ws: options.runtimeWs, }) const gateway = new Gateway({ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 5b67a28d..61f27e70 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -204,6 +204,9 @@ importers: '@nmtjs/protocol': specifier: workspace:* version: link:../protocol + '@nmtjs/server': + specifier: workspace:* + version: link:../server uWebSockets.js: specifier: github:uNetworking/uWebSockets.js#v20.67.0 version: https://codeload.github.com/uNetworking/uWebSockets.js/tar.gz/e0f56ebb4b349017f006e14d1cd29052f2a7121d @@ -277,15 +280,36 @@ importers: specifier: 'catalog:' version: 4.4.3 devDependencies: + '@nmtjs/gateway': + specifier: workspace:* + version: link:../gateway + '@nmtjs/http-transport': + specifier: workspace:* + version: link:../http-transport + '@nmtjs/json-format': + specifier: workspace:* + version: link:../json-format '@nmtjs/metrics': specifier: workspace:* version: link:../metrics + '@nmtjs/protocol': + specifier: workspace:* + version: link:../protocol '@nmtjs/proxy': specifier: 1.0.0-beta.7 version: 1.0.0-beta.7 + '@nmtjs/server': + specifier: workspace:* + version: link:../server + '@nmtjs/ws-transport': + specifier: workspace:* + version: link:../ws-transport '@types/node': specifier: ^24.13.2 version: 24.13.2 + uWebSockets.js: + specifier: github:uNetworking/uWebSockets.js#v20.67.0 + version: https://codeload.github.com/uNetworking/uWebSockets.js/tar.gz/e0f56ebb4b349017f006e14d1cd29052f2a7121d packages/nmtjs: dependencies: @@ -377,6 +401,28 @@ importers: specifier: ^0.3.3 version: 0.3.3 + packages/server: + dependencies: + '@types/bun': + specifier: ^1.3.0 + version: 1.3.2(@types/react@19.1.9) + '@types/deno': + specifier: ^2.3.0 + version: 2.3.0 + '@types/node': + specifier: ^24.13.2 + version: 24.13.2 + crossws: + specifier: 0.4.6 + version: 0.4.6(srvx@0.12.4) + devDependencies: + '@nmtjs/common': + specifier: workspace:* + version: link:../common + uWebSockets.js: + specifier: github:uNetworking/uWebSockets.js#v20.67.0 + version: https://codeload.github.com/uNetworking/uWebSockets.js/tar.gz/e0f56ebb4b349017f006e14d1cd29052f2a7121d + packages/type: dependencies: '@standard-schema/spec': @@ -507,12 +553,18 @@ importers: '@nmtjs/gateway': specifier: workspace:* version: link:../gateway + '@nmtjs/http-transport': + specifier: workspace:* + version: link:../http-transport '@nmtjs/json-format': specifier: workspace:* version: link:../json-format '@nmtjs/protocol': specifier: workspace:* version: link:../protocol + '@nmtjs/server': + specifier: workspace:* + version: link:../server '@nmtjs/type': specifier: workspace:* version: link:../type diff --git a/tsconfig.build.json b/tsconfig.build.json index b11d0cdc..67c9f745 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -48,6 +48,9 @@ { "path": "./packages/pubsub/tsconfig.build.json" }, + { + "path": "./packages/server/tsconfig.build.json" + }, { "path": "./packages/type/tsconfig.build.json" }, diff --git a/tsconfig.json b/tsconfig.json index 2f12b36e..6d03f83c 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -48,6 +48,9 @@ { "path": "./packages/pubsub/tsconfig.json" }, + { + "path": "./packages/server/tsconfig.json" + }, { "path": "./packages/type/tsconfig.json" },