diff --git a/docs/docs/network/index.mdx b/docs/docs/network/index.mdx index 58b41e1..2213d9e 100644 --- a/docs/docs/network/index.mdx +++ b/docs/docs/network/index.mdx @@ -6,7 +6,7 @@ description: "Overview of the engine networking docs" # Network Overview -This page describes the engine's networking libraries: `network-server` and `network-client`. Together they provide a small, pragmatic networking layer used by the example `pong-network` game. +This page describes the engine's networking libraries: `network-server` and `network-client`. Together they provide a small, pragmatic networking layer used by the example `pong-network` game. The same apps can also run without a game server, hosted by one of the players. ## Goals @@ -21,3 +21,4 @@ This page describes the engine's networking libraries: `network-server` and `net - [Packet Channels](./packet-framing) - [Network Server API](./network-server-api) - [Network Client API](./network-client-api) +- [Peer-to-peer Networking](./network-peer) diff --git a/docs/docs/network/network-peer.mdx b/docs/docs/network/network-peer.mdx new file mode 100644 index 0000000..9d30b82 --- /dev/null +++ b/docs/docs/network/network-peer.mdx @@ -0,0 +1,134 @@ +--- +title: "Peer-to-peer Networking" +icon: "book" +description: "Run the same client and server apps without a game server: one player hosts" +--- + +# Peer-to-peer Networking + +A game made of a client app and a server app can run **without a game server**. +The first player to open the game hosts it: their browser tab runs the server +app next to their client app. The other players join that tab over WebRTC, on +the same four [channels](./packet-framing) and typed packets. + +Nothing changes in the game code: the same systems, components and +`ctx.network` APIs run in both modes. Only one line of configuration picks the +mode. + +## Turning it on + +In the workspace `nanoforge.config.ts`: + +```ts +import { defineConfig } from "@nanoforge-dev/config"; + +export default defineConfig({ + type: "workspace", + packages: ["apps/*"], + network: "p2p", // "dedicated" (default) runs the server app in Bun +}); +``` + +The environment is the same as with a dedicated server: + +```dotenv +NANOFORGE_CLIENT_SERVER_ADDRESS=127.0.0.1 +NANOFORGE_CLIENT_SERVER_TCP_PORT=4445 +NANOFORGE_SERVER_LISTENING_TCP_PORT=4445 + +# STUN for every player; add a TURN server for players behind strict NATs +NANOFORGE_ICE_SERVERS=stun:stun.cloudflare.com:3478 +``` + +## How it works + +``` + server process (Bun): signaling only + ╱ ╲ + host tab ───────── player tab one WebRTC connection per player, + server app ╲ carrying the four channels + + client app╲───── player tab +``` + +- `nf build` builds the server app a second time, for the browser, into the + client's output (`server/`). The engine picks browser builds of + `@nanoforge-dev/ecs/server` and `@nanoforge-dev/network/server` on its own. +- `nf start` sets `NETWORK_MODE=p2p` for both apps. The Bun server process then + only pairs players for the WebRTC handshake, on `LISTENING_TCP_PORT`. Game + traffic never goes through it, and one process serves any number of games. + The server app still starts there, with no players: its sends do nothing. +- A page opened **without** `?room=` hosts: its client app starts the server + app in the same tab and joins it through an in-memory loopback. The server + app gets the client's environment. +- A page opened **with** `?room=CODE` joins that game's host over WebRTC. +- The host and its client app keep separate worlds, exactly as on two machines. + +## Inviting players + +The host logs its invite link to the console. The game can also show it: + +| `ctx.network` member | Host | Player who joined | Dedicated mode | +| -------------------- | --------------------- | --------------------- | -------------- | +| `roomCode` | `"ABC234"` | `"ABC234"` | `undefined` | +| `inviteLink` | page URL `?room=CODE` | page URL `?room=CODE` | `undefined` | + +```ts +export const showInvite = (registry: Registry, ctx: Context) => { + const link = ctx.network.inviteLink; + if (link) inviteText.text(`Invite your friends: ${link}`); +}; +``` + +## Developing, testing and deploying + +- **Editor:** play runs the game in dedicated mode. The game code is the same in + both modes. +- **Locally:** `nf build && nf start`, open the game, then open the logged + invite link in a second tab. +- **Hosted:** build and enable a host as usual. Each player who opens the game + without a link hosts their own game, so one host serves many games at once. + +## Players coming and going + +- **A player joins.** `server.clients.onConnect` fires on the host once the + player's four channels are open, so the listener can already send to it. +- **A player leaves.** `server.clients.onDisconnect` fires as soon as the + signaling server reports it gone, a data channel closes, or its connection + fails. +- **The host leaves.** The game ends: the signaling server closes the room, and + each player's `isConnected()` turns `false` about a second later. + +## Limits + +- **Message size.** Keep each packet at or below 64 KiB: WebRTC data channels + cap message size. +- **`ReliableUnordered` is really unordered**, unlike over a dedicated server's + WebSocket. +- **NAT.** A direct connection fails for some pairs of players, typically on + strict corporate or mobile networks. A TURN server in `ICE_SERVERS` relays + them; the default list only has STUN servers. +- **Trust.** The host runs the server app, so it can cheat. Check what players + send with `definePacket`'s `validate`. Anyone with the room code can join. +- **Background tabs.** Browsers slow down timers in hidden tabs, so the server + app in the host's tab times its ticks with a Web Worker: the game keeps its + pace for every player when the host switches tabs. Only the host's own view + slows down while hidden. +- **Server environment.** The server app gets the client's environment: + `NANOFORGE_SERVER_*` variables do not reach it in this mode. + +## Signaling protocol + +JSON text frames, at most 64 KiB each: + +| Who | Sends | Server forwards | +| ------ | ------------------------------------ | ------------------------------------- | +| Host | connects without a `room` | `{type:"room", room}` to the host | +| Player | connects with `?room=CODE` | HTTP 404 when the room does not exist | +| Player | any frame (its WebRTC offer and ICE) | `{type:"signal", from, data}` to host | +| Host | `{type:"signal", to, data}` | `data` to that player of its room | +| Player | closes | `{type:"leave", from}` to the host | +| Host | closes | close code `4001` to its players | + +The host welcomes a player (`{type:"welcome", id, token}`) once its data +channels are open, so players speak the same protocol as with a dedicated UDP +server. diff --git a/modules/ecs/package.json b/modules/ecs/package.json index 6724cdd..1ad0a1e 100644 --- a/modules/ecs/package.json +++ b/modules/ecs/package.json @@ -46,6 +46,10 @@ } }, "./server": { + "browser": { + "types": "./dist/client/index.d.ts", + "default": "./dist/client/index.js" + }, "require": { "types": "./dist/server/index.d.cts", "default": "./dist/server/index.cjs" diff --git a/modules/network/README.md b/modules/network/README.md index 358bb16..7c666b0 100644 --- a/modules/network/README.md +++ b/modules/network/README.md @@ -15,7 +15,7 @@ ## About -`@nanoforge-dev/network` is NanoForge's built-in networking library. It ships two subpath entry points — `@nanoforge-dev/network/client` and `@nanoforge-dev/network/server` — each sending packets on four channels: `Channel.ReliableOrdered` and `Channel.ReliableUnordered` over TCP (WebSocket), `Channel.UnreliableOrdered` and `Channel.UnreliableUnordered` over UDP (WebRTC data channels). The root entry, `@nanoforge-dev/network`, exposes the types shared by both, such as `NetworkConfig`. Splitting client and server into separate entry points keeps the server's runtime-only dependencies (`Bun.serve`, `node-datachannel`) out of a client's bundle graph entirely. +`@nanoforge-dev/network` is NanoForge's built-in networking library. It ships two subpath entry points — `@nanoforge-dev/network/client` and `@nanoforge-dev/network/server` — each sending packets on four channels: `Channel.ReliableOrdered` and `Channel.ReliableUnordered` over TCP (WebSocket), `Channel.UnreliableOrdered` and `Channel.UnreliableUnordered` over UDP (WebRTC data channels). The root entry, `@nanoforge-dev/network`, exposes the types shared by both, such as `NetworkConfig`. Splitting client and server into separate entry points keeps the server's runtime-only dependencies (`Bun.serve`, `node-datachannel`) out of a client's bundle graph entirely. The same two entry points also run a game without a game server, hosted by one of the players (see [Peer-to-peer](#peer-to-peer)). > **The server entry point requires the [Bun](https://bun.sh) runtime.** `@nanoforge-dev/network/server` is built on Bun's native WebSocket server (`Bun.serve`) and does not run on Node.js. The client entry point is runtime-agnostic and targets the browser. @@ -241,6 +241,37 @@ starts a new session with a new id. Using a channel whose port wasn't configured on that side throws. +### Peer-to-peer + +Set `network: "p2p"` in the workspace `nanoforge.config.ts` and the same client +and server apps run without a game server: the first player to open the game +hosts it, running the server app in their own tab, and the others join that tab +over WebRTC, on the same four channels and typed packets. No game code changes. + +```ts +// nanoforge.config.ts (workspace) +export default defineConfig({ type: "workspace", packages: ["apps/*"], network: "p2p" }); +``` + +- `nf build` also builds the server app for the browser, into the client's + output. `nf start` sets `NETWORK_MODE=p2p` for both apps. +- The Bun server process only pairs players (WebRTC signaling), on + `LISTENING_TCP_PORT`. Players reach it with the usual `SERVER_ADDRESS` and + `SERVER_TCP_PORT`, so the env stays the same as with a dedicated server. +- A page opened without `?room=` hosts a game; `?room=CODE` joins one. The host + logs its invite link, and the game can show it: + +```ts +ctx.network.roomCode; // "ABC234", on the host and on players who joined +ctx.network.inviteLink; // "https://example.com/?room=ABC234", to send to other players +``` + +The game ends when the host leaves: each player's `isConnected()` turns `false`. +Keep packets at or below 64 KiB and add a TURN server to `ICE_SERVERS` for +players behind strict NATs. The host's server keeps its pace when its tab is +hidden. See the +[peer-to-peer docs](https://github.com/NanoForge-dev/Engine/blob/main/docs/docs/network/network-peer.mdx). + ## Links - [GitHub][source] diff --git a/modules/network/package.json b/modules/network/package.json index fc88026..873b71d 100644 --- a/modules/network/package.json +++ b/modules/network/package.json @@ -46,6 +46,10 @@ } }, "./server": { + "browser": { + "types": "./dist/server-browser/browser.d.ts", + "default": "./dist/server-browser/browser.js" + }, "require": { "types": "./dist/server/index.d.cts", "default": "./dist/server/index.cjs" diff --git a/modules/network/src/client/client-session.ts b/modules/network/src/client/client-session.ts index 2ccc3fd..6daaa78 100644 --- a/modules/network/src/client/client-session.ts +++ b/modules/network/src/client/client-session.ts @@ -32,6 +32,7 @@ export type WelcomeWait = { * @param ip - Server hostname or IP. * @param port - Server port. * @param session - Current client session. + * @param params - Extra query parameters, such as a peer-to-peer `room`. * @returns The URL to connect to. */ export const buildServerUrl = ( @@ -39,10 +40,16 @@ export const buildServerUrl = ( ip: string, port: number, session: ClientSession, + params: Record = {}, ): string => { const url = `ws${wss ? "s" : ""}://${ip}:${port}`; - if (session.token === undefined) return url; - return `${url}/?${SESSION_TOKEN_PARAM}=${encodeURIComponent(session.token)}`; + const query = Object.entries({ + ...params, + ...(session.token !== undefined && { [SESSION_TOKEN_PARAM]: session.token }), + }) + .map(([key, value]) => `${key}=${encodeURIComponent(value)}`) + .join("&"); + return query === "" ? url : `${url}/?${query}`; }; /** diff --git a/modules/network/src/client/client.network.library.ts b/modules/network/src/client/client.network.library.ts index 71aea66..a5ac7d0 100644 --- a/modules/network/src/client/client.network.library.ts +++ b/modules/network/src/client/client.network.library.ts @@ -2,40 +2,24 @@ import { type Context, type InitContext, Library, defineLibraryKey } from "@nano import type { EditorAwareLibrary, EditorInitContext } from "@nanoforge-dev/editor-lib"; import { registerEnv } from "@nanoforge-dev/env"; +import type { PeerHost } from "../server/peer-host"; import { CHANNELS, CONNECT_TIMEOUT_MS, - type Channel, type ChannelOptions, - DEFAULT_CHANNEL, + RELIABLE_CHANNELS, + UNRELIABLE_CHANNELS, isReliableChannel, } from "../shared/channels"; -import { - type NetworkData, - type NetworkPayload, - encodeNetworkPayload, -} from "../shared/network-data"; -import { - PacketDispatcher, - type TypedPacket, - encodePacket, - encodeTypedPacket, - sortPackets, -} from "../shared/packet"; +import type { NetworkData, NetworkPayload } from "../shared/network-data"; +import { PEER_HOST_KEY, ROOM_PARAM } from "../shared/signaling"; import { type Trace, createTrace, traceClient } from "../shared/trace"; -import type { ClientSession } from "./client-session"; import { ClientConfigNetwork } from "./config.client.network"; -import type { - ClientPacketInfo, - ClientPacketsApi, - NetworkClientContextApi, -} from "./network-client-context.type"; +import { NetworkClient } from "./network-client"; +import type { ClientPacketsApi, NetworkClientContextApi } from "./network-client-context.type"; import { TCPClient } from "./tcp.client.network"; import { UDPClient } from "./udp.client.network"; -/** Delay between two checks of the channels while `__init` waits for them. */ -const CONNECT_POLL_MS = 50; - /** * Built-in network library for client-side applications. * @@ -52,6 +36,10 @@ const CONNECT_POLL_MS = 50; * - `SERVER_UDP_PORT` — signaling port for UDP/WebRTC (optional). * - `WSS` — set to `"true"` to use `wss://` / `https://` (default: `false`). * - `ICE_SERVERS` — STUN/TURN servers for the UDP transport, comma-separated or a JSON array (default: `[]`). + * - `NETWORK_MODE` — `p2p` (set by `nf` from `network: "p2p"`): without `?room=` in + * the page URL, this player hosts the game, running the server app in this + * tab; with it, the player joins that room's host over WebRTC. The signaling + * server is reached at `SERVER_ADDRESS` and `SERVER_TCP_PORT`. * * @example * ```ts @@ -66,22 +54,25 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary private _tcp?: TCPClient; private _udp?: UDPClient; - private readonly _session: ClientSession = {}; - private readonly _dispatcher = new PacketDispatcher(); - /** Typed packets drained from the transports, dispatched by the next `__run`. */ - private _typed: TypedPacket[] = []; - /** `sendData` packets drained from the transports, returned by `getReceivedPackets`. */ - private readonly _raw = new Map(); - private readonly _packets: ClientPacketsApi = { - on: (packet, listener) => this._dispatcher.on(packet, listener), - read: (packet) => this._dispatcher.read(packet), - send: (packet, ...[data, options]) => - this.sendBytes(options?.channel ?? packet.channel, encodeTypedPacket(packet, data)), - }; + private readonly _client = new NetworkClient( + "NetworkClientLibrary", + (channel) => + `Channel ${channel} needs ${isReliableChannel(channel) ? "SERVER_TCP_PORT" : "SERVER_UDP_PORT"} to be set`, + ); /** Client id assigned by the server, `undefined` until transport is welcomed. */ public get clientId(): string | undefined { - return this._session.id; + return this._client.clientId; + } + + /** See `NetworkClientContextApi.roomCode`. */ + public get roomCode(): string | undefined { + return this._client.roomCode; + } + + /** See `NetworkClientContextApi.inviteLink`. */ + public get inviteLink(): string | undefined { + return this._client.inviteLink; } private _trace: Trace | undefined; @@ -93,7 +84,7 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary /** Typed packets: per-type listeners, `read` and `send`. */ public get packets(): ClientPacketsApi { - return this._packets; + return this._client.packets; } public override async __init(ctx: InitContext): Promise { @@ -103,12 +94,14 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary throw new Error("NetworkClientLibrary: no server port specified to connect to."); } + if (config.NETWORK_MODE === "p2p") return this.connectPeer(ctx, config); + if (config.SERVER_TCP_PORT !== undefined) { this._tcp = new TCPClient( +config.SERVER_TCP_PORT, config.SERVER_ADDRESS, config.WSS, - this._session, + this._client.session, ); } @@ -118,11 +111,21 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary config.SERVER_ADDRESS, config.WSS, config.ICE_SERVERS, - this._session, + this._client.session, ); } - await this.openChannels(); + // TCP carries the reliable channels, UDP the unreliable ones. + this._client.transports = { + ...(this._tcp && Object.fromEntries(RELIABLE_CHANNELS.map((c) => [c, this._tcp]))), + ...(this._udp && Object.fromEntries(UNRELIABLE_CHANNELS.map((c) => [c, this._udp]))), + }; + + // TCP first, so UDP joins its session with the token from the TCP welcome. + const transports: [string, TCPClient | UDPClient][] = []; + if (this._tcp) transports.push(["TCP", this._tcp]); + if (this._udp) transports.push(["UDP", this._udp]); + await this._client.openChannels(transports); } /** @@ -132,8 +135,8 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary * @param options - `channel` to send on, `Channel.ReliableOrdered` by default. * @throws When the channel's port was not configured */ - public sendData(data: NetworkPayload, { channel = DEFAULT_CHANNEL }: ChannelOptions = {}): void { - this.sendBytes(channel, encodePacket(null, encodeNetworkPayload(data))); + public sendData(data: NetworkPayload, options?: ChannelOptions): void { + this._client.sendData(data, options); } /** @@ -148,14 +151,8 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary * @returns Array of packets, each read with `bytes()`, `text()`, `json()`… * @throws When the channel's port was not configured. */ - public getReceivedPackets({ channel }: ChannelOptions = {}): NetworkData[] { - if (channel === undefined) { - return this.configuredChannels().flatMap((c) => this.getReceivedPackets({ channel: c })); - } - this.drain(channel); - const packets = this._raw.get(channel) ?? []; - this._raw.delete(channel); - return packets; + public getReceivedPackets(options?: ChannelOptions): NetworkData[] { + return this._client.getReceivedPackets(options); } /** @@ -164,13 +161,8 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary * @param options - `channel` to check, every configured channel by default. * @throws When the channel's port was not configured. */ - public isConnected({ channel }: ChannelOptions = {}): boolean { - if (channel === undefined) { - const channels = this.configuredChannels(); - return channels.length > 0 && channels.every((c) => this.isConnected({ channel: c })); - } - if (isReliableChannel(channel)) return this.requireTcp(channel).isConnected(); - return this.requireUdp(channel).isConnected(channel); + public isConnected(options?: ChannelOptions): boolean { + return this._client.isConnected(options); } /** @@ -192,100 +184,79 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary /** Calls the listeners of the typed packets received since the last tick. */ public override async __run(ctx: Context): Promise { - this._raw.clear(); - for (const channel of this.configuredChannels()) this.drain(channel); - const typed = this._typed; - this._typed = []; - this._dispatcher.dispatch(typed, ctx); + this._client.run(ctx); } public override expose(): NetworkClientContextApi { - // eslint-disable-next-line @typescript-eslint/no-this-alias - const library = this; - return { - sendData: (data, options) => library.sendData(data, options), - getReceivedPackets: (options) => library.getReceivedPackets(options), - isConnected: (options) => library.isConnected(options), - get packets() { - return library.packets; - }, - get clientId() { - return library.clientId; - }, - }; + return this._client.expose(); } - /** - * Connect every configured transport and wait until its channels are open. - * - * @remarks - * Channels cannot be opened once the game runs, and connection errors are - * best caught at start, so this runs during `__init`. - * - * @throws When a transport fails to connect, or the channels are not open - * within `CONNECT_TIMEOUT_MS`. - */ - private async openChannels(): Promise { - const deadline = Date.now() + CONNECT_TIMEOUT_MS; - const channels = this.configuredChannels(); - - // TCP first, so UDP joins its session with the token from the TCP welcome. - const transports: [string, TCPClient | UDPClient][] = []; - if (this._tcp) transports.push(["TCP", this._tcp]); - if (this._udp) transports.push(["UDP", this._udp]); + private async connectPeer(ctx: InitContext, config: ClientConfigNetwork): Promise { + if (config.SERVER_TCP_PORT === undefined) { + throw new Error( + "NetworkClientLibrary: p2p needs SERVER_TCP_PORT to reach the signaling server.", + ); + } - const failures: unknown[] = []; - for (const [name, transport] of transports) { - try { - await transport.connect(); - } catch (error) { - failures.push(new Error(`${name} connection failed`, { cause: error })); - } + const room = new URLSearchParams(location.search).get(ROOM_PARAM); + if (room !== null) { + const udp = (this._udp = new UDPClient( + +config.SERVER_TCP_PORT, + config.SERVER_ADDRESS, + config.WSS, + config.ICE_SERVERS, + this._client.session, + CHANNELS, + { [ROOM_PARAM]: room }, + )); + this._client.transports = Object.fromEntries(CHANNELS.map((c) => [c, udp])); + this._client.room = { room }; + await this._client.openChannels([ + ["Peer", { connect: () => udp.connect(CONNECT_TIMEOUT_MS) }], + ]); + return; } - if (failures.length > 0) { - throw new Error("NetworkClientLibrary: could not connect to the server", { - cause: failures.length === 1 ? failures[0] : failures, - }); + + const host = await this.startServer(ctx); + const local = host.attachLocal(); + this._client.session.id = local.id; + this._client.transports = Object.fromEntries(CHANNELS.map((c) => [c, local.transport])); + this._client.room = host; + console.info(`Invite: ${this.inviteLink}`); + } + + private async startServer(ctx: InitContext): Promise { + const main = ctx.files.get("/server/main.js"); + if (main === undefined) { + throw new Error( + 'NetworkClientLibrary: p2p needs the server app built for the browser: run `nf build` with network: "p2p".', + ); } - while (!channels.every((channel) => this.isConnected({ channel }))) { + const files = new Map( + [...ctx.files] + .filter(([path]) => path.startsWith("/server/")) + .map(([path, url]) => [path.slice("/server".length), url]), + ); + const server = await import(main); + let failure: { error: unknown } | undefined; + server.main({ files, env: ctx.env }).catch((error: unknown) => (failure = { error })); + + const deadline = Date.now() + CONNECT_TIMEOUT_MS; + for (;;) { + const host = Reflect.get(globalThis, PEER_HOST_KEY) as PeerHost | undefined; + if (host) return host; + if (failure) { + throw new Error("NetworkClientLibrary: the server app failed to start", { + cause: failure.error, + }); + } if (Date.now() >= deadline) { - const pending = channels.filter((channel) => !this.isConnected({ channel })); throw new Error( - `NetworkClientLibrary: channels ${pending.join(", ")} not open after ${CONNECT_TIMEOUT_MS}ms`, + `NetworkClientLibrary: the server app did not start within ${CONNECT_TIMEOUT_MS}ms`, ); } - await new Promise((resolve) => setTimeout(resolve, CONNECT_POLL_MS)); + await new Promise((resolve) => setTimeout(resolve, 50)); } } - - private configuredChannels(): Channel[] { - return CHANNELS.filter((channel) => - isReliableChannel(channel) ? this._tcp !== undefined : this._udp !== undefined, - ); - } - - private sendBytes(channel: Channel, bytes: Uint8Array): void { - if (isReliableChannel(channel)) this.requireTcp(channel).sendData(channel, bytes); - else this.requireUdp(channel).sendData(channel, bytes); - } - - private drain(channel: Channel): void { - const packets = isReliableChannel(channel) - ? this.requireTcp(channel).getReceivedPackets(channel) - : this.requireUdp(channel).getReceivedPackets(channel); - const raw = this._raw.get(channel) ?? []; - sortPackets(packets, { channel }, raw, this._typed); - this._raw.set(channel, raw); - } - - private requireTcp(channel: Channel): TCPClient { - if (!this._tcp) throw new Error(`Channel ${channel} needs SERVER_TCP_PORT to be set`); - return this._tcp; - } - - private requireUdp(channel: Channel): UDPClient { - if (!this._udp) throw new Error(`Channel ${channel} needs SERVER_UDP_PORT to be set`); - return this._udp; - } } diff --git a/modules/network/src/client/network-client-context.type.ts b/modules/network/src/client/network-client-context.type.ts index 68b660e..b067374 100644 --- a/modules/network/src/client/network-client-context.type.ts +++ b/modules/network/src/client/network-client-context.type.ts @@ -59,6 +59,10 @@ export interface NetworkClientContextApi { readonly packets: ClientPacketsApi; /** Client id assigned by the server, shared by TCP and UDP. `undefined` until welcomed. */ readonly clientId: string | undefined; + /** Code of the peer-to-peer game this player hosts or joined, `undefined` otherwise. */ + readonly roomCode: string | undefined; + /** Link to send to other players so they join this peer-to-peer game, `undefined` otherwise. */ + readonly inviteLink: string | undefined; /** * Send bytes, a string or a JSON value to the server, on * `Channel.ReliableOrdered` unless another `channel` is given. diff --git a/modules/network/src/client/network-client.ts b/modules/network/src/client/network-client.ts new file mode 100644 index 0000000..b3f8219 --- /dev/null +++ b/modules/network/src/client/network-client.ts @@ -0,0 +1,207 @@ +import type { Context } from "@nanoforge-dev/common"; + +import { + CHANNELS, + CONNECT_TIMEOUT_MS, + type Channel, + type ChannelOptions, + DEFAULT_CHANNEL, +} from "../shared/channels"; +import { + type NetworkData, + type NetworkPayload, + encodeNetworkPayload, +} from "../shared/network-data"; +import { + PacketDispatcher, + type TypedPacket, + encodePacket, + encodeTypedPacket, + sortPackets, +} from "../shared/packet"; +import { ROOM_PARAM } from "../shared/signaling"; +import type { ClientTransport } from "../shared/transport"; +import type { ClientSession } from "./client-session"; +import type { + ClientPacketInfo, + ClientPacketsApi, + NetworkClientContextApi, +} from "./network-client-context.type"; + +/** Delay between two checks of the channels while `openChannels` waits for them. */ +const CONNECT_POLL_MS = 50; + +/** + * Transport-independent half of a client: session, typed and raw packet + * queues, and routing of each channel to the transport carrying it. + * + * @remarks + * Shared by `NetworkClientLibrary` (TCP and UDP to a dedicated server) and + * the `NETWORK_MODE=p2p` mode (WebRTC to a host, or loopback on the host). + */ +export class NetworkClient { + /** Transport carrying each channel; a missing channel throws when used. */ + public transports: Partial>> = {}; + /** Session shared with the transports, filled by the server's welcome. */ + public readonly session: ClientSession = {}; + /** Where the room code of a peer-to-peer game is read from: the host, or the joined room. */ + public room: { readonly room: string | undefined } | undefined; + + private readonly _dispatcher = new PacketDispatcher(); + /** Typed packets drained from the transports, dispatched by the next `run`. */ + private _typed: TypedPacket[] = []; + /** `sendData` packets drained from the transports, returned by `getReceivedPackets`. */ + private readonly _raw = new Map(); + private readonly _packets: ClientPacketsApi = { + on: (packet, listener) => this._dispatcher.on(packet, listener), + read: (packet) => this._dispatcher.read(packet), + send: (packet, ...[data, options]) => + this.sendBytes(options?.channel ?? packet.channel, encodeTypedPacket(packet, data)), + }; + + /** + * @param _name - Library name, prefixed to connection errors. + * @param _requirement - Error message for a channel with no transport. + */ + constructor( + private readonly _name: string, + private readonly _requirement: (channel: Channel) => string, + ) {} + + /** Client id assigned by the server, `undefined` until welcomed. */ + public get clientId(): string | undefined { + return this.session.id; + } + + /** See `NetworkClientContextApi.roomCode`. */ + public get roomCode(): string | undefined { + return this.room?.room; + } + + /** See `NetworkClientContextApi.inviteLink`. */ + public get inviteLink(): string | undefined { + const code = this.roomCode; + if (code === undefined) return undefined; + return `${location.origin}${location.pathname}?${ROOM_PARAM}=${encodeURIComponent(code)}`; + } + + /** Typed packets: per-type listeners, `read` and `send`. */ + public get packets(): ClientPacketsApi { + return this._packets; + } + + /** See `NetworkClientContextApi.sendData`. */ + public sendData(data: NetworkPayload, { channel = DEFAULT_CHANNEL }: ChannelOptions = {}): void { + this.sendBytes(channel, encodePacket(null, encodeNetworkPayload(data))); + } + + /** See `NetworkClientContextApi.getReceivedPackets`. */ + public getReceivedPackets({ channel }: ChannelOptions = {}): NetworkData[] { + if (channel === undefined) { + return this.configuredChannels().flatMap((c) => this.getReceivedPackets({ channel: c })); + } + this.drain(channel); + const packets = this._raw.get(channel) ?? []; + this._raw.delete(channel); + return packets; + } + + /** See `NetworkClientContextApi.isConnected`. */ + public isConnected({ channel }: ChannelOptions = {}): boolean { + if (channel === undefined) { + const channels = this.configuredChannels(); + return channels.length > 0 && channels.every((c) => this.isConnected({ channel: c })); + } + return this.require(channel).isConnected(channel); + } + + /** Calls the listeners of the typed packets received since the last tick. */ + public run(ctx: Context): void { + this._raw.clear(); + for (const channel of this.configuredChannels()) this.drain(channel); + const typed = this._typed; + this._typed = []; + this._dispatcher.dispatch(typed, ctx); + } + + /** + * Connect each transport, in order, then wait until every routed channel is + * open. + * + * @param transports - Transports to connect, named for error messages. The + * first one welcomed gives its session token to the next. + * @throws When a transport fails to connect, or the channels are not open + * within `CONNECT_TIMEOUT_MS`. + */ + public async openChannels(transports: [string, { connect(): Promise }][]): Promise { + const deadline = Date.now() + CONNECT_TIMEOUT_MS; + const channels = this.configuredChannels(); + + const failures: unknown[] = []; + for (const [name, transport] of transports) { + try { + await transport.connect(); + } catch (error) { + failures.push(new Error(`${name} connection failed`, { cause: error })); + } + } + if (failures.length > 0) { + throw new Error(`${this._name}: could not connect to the server`, { + cause: failures.length === 1 ? failures[0] : failures, + }); + } + + while (!channels.every((channel) => this.isConnected({ channel }))) { + if (Date.now() >= deadline) { + const pending = channels.filter((channel) => !this.isConnected({ channel })); + throw new Error( + `${this._name}: channels ${pending.join(", ")} not open after ${CONNECT_TIMEOUT_MS}ms`, + ); + } + await new Promise((resolve) => setTimeout(resolve, CONNECT_POLL_MS)); + } + } + + public expose(): NetworkClientContextApi { + // eslint-disable-next-line @typescript-eslint/no-this-alias + const client = this; + return { + sendData: (data, options) => client.sendData(data, options), + getReceivedPackets: (options) => client.getReceivedPackets(options), + isConnected: (options) => client.isConnected(options), + get packets() { + return client.packets; + }, + get clientId() { + return client.clientId; + }, + get roomCode() { + return client.roomCode; + }, + get inviteLink() { + return client.inviteLink; + }, + }; + } + + private configuredChannels(): Channel[] { + return CHANNELS.filter((channel) => this.transports[channel] !== undefined); + } + + private sendBytes(channel: Channel, bytes: Uint8Array): void { + this.require(channel).sendData(channel, bytes); + } + + private drain(channel: Channel): void { + const packets = this.require(channel).getReceivedPackets(channel); + const raw = this._raw.get(channel) ?? []; + sortPackets(packets, { channel }, raw, this._typed); + this._raw.set(channel, raw); + } + + private require(channel: Channel): ClientTransport { + const transport = this.transports[channel]; + if (!transport) throw new Error(this._requirement(channel)); + return transport; + } +} diff --git a/modules/network/src/client/udp.client.network.ts b/modules/network/src/client/udp.client.network.ts index 5f7305e..531a233 100644 --- a/modules/network/src/client/udp.client.network.ts +++ b/modules/network/src/client/udp.client.network.ts @@ -1,12 +1,13 @@ import { Channel, + DATA_CHANNEL_OPTIONS, PacketSequencer, UNRELIABLE_CHANNELS, - UNRELIABLE_CHANNEL_OPTIONS, - type UnreliableChannel, } from "../shared/channels"; +import { HOST_LEFT_CODE } from "../shared/signaling"; import { type ClientSession, + WELCOME_TIMEOUT_MS, type WelcomeWait, applyWelcome, buildServerUrl, @@ -24,13 +25,14 @@ type DataChannelState = { /** * WebRTC connection to a NanoForge UDP server, carrying the unreliable - * channels. + * channels — or to a peer-to-peer host, carrying all four. * * @remarks * Uses a WebSocket signaling channel to perform the SDP/ICE handshake, then - * opens one RTCDataChannel per unreliable channel, labelled with its - * {@link Channel} value. Both use `ordered: false` and `maxRetransmits: 0` - * for minimal latency. Each data channel message is one packet; + * opens one RTCDataChannel per channel it carries, labelled with its + * {@link Channel} value and set up with `DATA_CHANNEL_OPTIONS`: the + * unreliable ones use `ordered: false` and `maxRetransmits: 0` for minimal + * latency. Each data channel message is one packet; * `UnreliableOrdered` packets carry a sequence number and late ones are * dropped. * @@ -42,15 +44,25 @@ type DataChannelState = { * `__init`. */ export class UDPClient { - private readonly _channels = new Map(); + private readonly _channels = new Map(); private _welcome: WelcomeWait | null = null; + private _webSocket: WebSocket | null = null; + private _peerConnection: RTCPeerConnection | null = null; + /** + * @param _carried - Channels to open, the unreliable ones for a dedicated + * server. + * @param _params - Extra query parameters of the signaling URL, such as a + * peer-to-peer `room`. + */ constructor( private _port: number, private _ip: string, private _wss: boolean, private _iceServers: RTCIceServer[] = [], private readonly _session: ClientSession = {}, + private readonly _carried: readonly Channel[] = UNRELIABLE_CHANNELS, + private readonly _params: Record = {}, ) {} /** @@ -62,14 +74,16 @@ export class UDPClient { * server's welcome is received. The data channels may become open shortly * after — check `isConnected`. * + * @param welcomeTimeoutMs - How long to wait for the welcome. A peer-to-peer + * host only welcomes a joiner once its channels are open, so it waits longer. * @throws When the signaling socket fails or closes before the welcome, or - * when no welcome arrives within `WELCOME_TIMEOUT_MS`. + * when no welcome arrives within `welcomeTimeoutMs`. */ - public async connect(): Promise { - const welcome = (this._welcome = waitForWelcome("UDP")); + public async connect(welcomeTimeoutMs = WELCOME_TIMEOUT_MS): Promise { + const welcome = (this._welcome = waitForWelcome("UDP", welcomeTimeoutMs)); try { - const webSocket: WebSocket = this.connectToServerWebSocket(); - const rtcPeerConnection: RTCPeerConnection = this.getRtcChannelFromIceServer(); + const webSocket = (this._webSocket = this.connectToServerWebSocket()); + const rtcPeerConnection = (this._peerConnection = this.getRtcChannelFromIceServer()); this.setupIceConnection(rtcPeerConnection, webSocket); await this.sendIceOffer(rtcPeerConnection, webSocket); } catch (error) { @@ -85,22 +99,33 @@ export class UDPClient { return this._session.id; } + /** + * Close the signaling socket and the peer connection. + */ + public close(): void { + this._webSocket?.close(); + this._peerConnection?.close(); + this._webSocket = null; + this._peerConnection = null; + this._channels.clear(); + } + /** * Return `true` when the data channel of `channel` is open. * - * @param channel - Unreliable channel to check. + * @param channel - Channel to check. */ - public isConnected(channel: UnreliableChannel): boolean { + public isConnected(channel: Channel): boolean { return this._channels.get(channel)?.channel.readyState === "open"; } /** - * Send a payload to the server on an unreliable channel. + * Send a payload to the server on a channel. * - * @param channel - Unreliable channel to send on. + * @param channel - Channel to send on. * @param data - Raw payload bytes. */ - public sendData(channel: UnreliableChannel, data: Uint8Array): void { + public sendData(channel: Channel, data: Uint8Array): void { const state = this._channels.get(channel); if (!state) { console.error("UDP not connected"); @@ -110,12 +135,12 @@ export class UDPClient { } /** - * Return the packets received on an unreliable channel since the last call. + * Return the packets received on a channel since the last call. * - * @param channel - Unreliable channel to read. + * @param channel - Channel to read. * @returns Array of packet buffers, in arrival order. */ - public getReceivedPackets(channel: UnreliableChannel): Uint8Array[] { + public getReceivedPackets(channel: Channel): Uint8Array[] { const state = this._channels.get(channel); if (!state) return []; const packets = state.packets; @@ -124,7 +149,7 @@ export class UDPClient { } private connectToServerWebSocket(): WebSocket { - const serverUrl = buildServerUrl(this._wss, this._ip, this._port, this._session); + const serverUrl = buildServerUrl(this._wss, this._ip, this._port, this._session, this._params); console.log("Try to connect for UDP to " + serverUrl.split("?")[0]); const webSocket = new WebSocket(serverUrl); @@ -132,22 +157,28 @@ export class UDPClient { console.error("UDP connection error : WebSocket Error", { cause: e }); this._welcome?.settle(new Error("UDP connection error", { cause: e })); }; - webSocket.onclose = (): void => { + webSocket.onclose = (event: CloseEvent): void => { this._welcome?.settle(new Error("UDP signaling closed before the server welcome")); + if (event.code === HOST_LEFT_CODE) this.close(); }; return webSocket; } private getRtcChannelFromIceServer(): RTCPeerConnection { const rtcPeerConnection = new RTCPeerConnection({ iceServers: this._iceServers }); - for (const channel of UNRELIABLE_CHANNELS) { + // Chrome keeps the data channels open after the connection failed. + rtcPeerConnection.onconnectionstatechange = () => { + const state = rtcPeerConnection.connectionState; + if (state === "failed" || state === "closed") this.close(); + }; + for (const channel of this._carried) { this.openDataChannel(rtcPeerConnection, channel); } return rtcPeerConnection; } - private openDataChannel(rtcPeerConnection: RTCPeerConnection, channel: UnreliableChannel): void { - const dataChannel = rtcPeerConnection.createDataChannel(channel, UNRELIABLE_CHANNEL_OPTIONS); + private openDataChannel(rtcPeerConnection: RTCPeerConnection, channel: Channel): void { + const dataChannel = rtcPeerConnection.createDataChannel(channel, DATA_CHANNEL_OPTIONS[channel]); dataChannel.binaryType = "arraybuffer"; const state: DataChannelState = { channel: dataChannel, diff --git a/modules/network/src/server/browser.ts b/modules/network/src/server/browser.ts new file mode 100644 index 0000000..86e514a --- /dev/null +++ b/modules/network/src/server/browser.ts @@ -0,0 +1,28 @@ +import "./context-augmentation"; + +export type { NetworkConfig } from "../shared/config.network"; +export { Channel, type ChannelOptions } from "../shared/channels"; +export { NetworkData, type NetworkPayload } from "../shared/network-data"; +export { + type PacketArgs, + type PacketDefinition, + type PacketListener, + type PacketOptions, + definePacket, +} from "../shared/packet"; +export type { + ClientId, + ClientInfo, + ClientListener, + ClientsApi, + ConnectionInfo, + Transport, +} from "./client-registry"; +export { ServerConfigNetwork } from "./config.server.network"; +export type { + NetworkServerContextApi, + ServerPacketInfo, + ServerPacketsApi, +} from "./network-server-context.type"; +export { NetworkServerLibrary } from "./server.network.browser"; +export type { UDPServerIceConfig } from "./udp.server.network"; diff --git a/modules/network/src/server/client-registry.ts b/modules/network/src/server/client-registry.ts index e5e1437..3b5077d 100644 --- a/modules/network/src/server/client-registry.ts +++ b/modules/network/src/server/client-registry.ts @@ -1,12 +1,13 @@ -import type { SocketAddress } from "bun"; - import { SESSION_TOKEN_PARAM } from "../shared/session"; /** Session-wide client identifier, shared by the TCP and UDP transports. */ export type ClientId = string; -/** Transport a connection was made over. */ -export type Transport = "tcp" | "udp"; +/** + * Transport a connection was made over: `"local"` is the host's own player and + * `"peer"` a WebRTC link to a peer host (`NETWORK_MODE=p2p`). + */ +export type Transport = "tcp" | "udp" | "local" | "peer"; /** What the server knows about one transport connection of a client. */ export type ConnectionInfo = { @@ -100,7 +101,8 @@ export const getSessionToken = (request: Request): string | null => export const buildConnectionInfo = ( transport: Transport, request: Request, - server: { requestIP(request: Request): SocketAddress | null }, + // Structural `SocketAddress`, so this file builds without Bun's types. + server: { requestIP(request: Request): { address: string; port: number; family: string } | null }, ): ConnectionInfo => { const remote = server.requestIP(request); const params = [...new URL(request.url).searchParams].filter( @@ -159,11 +161,24 @@ export class ClientRegistry implements ClientsApi { * * @param token - Session token presented by the client, or `null` for a new session. * @param connection - Connection to attach. + * @param register - Called with a new session before the connect listeners, so + * the transport can already be sent to from inside them. * @returns The session id and token, or `null` when {@link canAttach} refuses it. */ + public attach( + token: null, + connection: ConnectionInfo, + register?: (session: { id: ClientId; token: string }) => void, + ): { id: ClientId; token: string }; + public attach( + token: string | null, + connection: ConnectionInfo, + register?: (session: { id: ClientId; token: string }) => void, + ): { id: ClientId; token: string } | null; public attach( token: string | null, connection: ConnectionInfo, + register?: (session: { id: ClientId; token: string }) => void, ): { id: ClientId; token: string } | null { if (!this.canAttach(token, connection.transport)) return null; @@ -184,8 +199,10 @@ export class ClientRegistry implements ClientsApi { }; this._sessions.set(session.info.id, session); this._tokens.set(session.token, session.info.id); + const created = { id: session.info.id, token: session.token }; + register?.(created); this.notify(this._connectListeners, session.info); - return { id: session.info.id, token: session.token }; + return created; } /** diff --git a/modules/network/src/server/network-server.ts b/modules/network/src/server/network-server.ts new file mode 100644 index 0000000..c72156a --- /dev/null +++ b/modules/network/src/server/network-server.ts @@ -0,0 +1,179 @@ +import type { Context } from "@nanoforge-dev/common"; + +import { CHANNELS, type Channel, type ChannelOptions, DEFAULT_CHANNEL } from "../shared/channels"; +import { + type NetworkData, + type NetworkPayload, + encodeNetworkPayload, +} from "../shared/network-data"; +import { + PacketDispatcher, + type TypedPacket, + encodePacket, + encodeTypedPacket, + sortPackets, +} from "../shared/packet"; +import type { ServerTransport } from "../shared/transport"; +import type { ClientId, ClientInfo, ClientRegistry, ClientsApi } from "./client-registry"; +import type { + NetworkServerContextApi, + ServerPacketInfo, + ServerPacketsApi, +} from "./network-server-context.type"; + +/** + * Transport-independent half of a server: client sessions, typed and raw + * packet queues, and routing of each channel to the transport carrying it. + * + * @remarks + * Shared by `NetworkServerLibrary` (TCP and UDP transports, on Bun) and the + * browser `NetworkServerLibrary` (WebRTC and loopback, in the host's tab), so it + * imports nothing runtime-specific. + */ +export class NetworkServer { + /** Transport carrying each channel; a missing channel throws when used. */ + public transports: Partial>> = {}; + + private readonly _dispatcher = new PacketDispatcher(); + private _typed: TypedPacket[] = []; + private readonly _raw = new Map>(); + private readonly _packets: ServerPacketsApi = { + on: (packet, listener) => this._dispatcher.on(packet, listener), + read: (packet) => this._dispatcher.read(packet), + sendToAll: (packet, ...[data, options]) => + this.sendBytesToAll(options?.channel ?? packet.channel, encodeTypedPacket(packet, data)), + sendToClient: (clientId, packet, ...[data, options]) => + this.sendBytesToClient( + clientId, + options?.channel ?? packet.channel, + encodeTypedPacket(packet, data), + ), + }; + + /** + * @param registry - Client sessions, shared with the transports. + * @param requirement - Error message for a channel with no transport. + */ + constructor( + private readonly _registry: ClientRegistry, + private readonly _requirement: (channel: Channel) => string, + ) {} + + /** Client sessions shared by the transports. */ + public get clients(): ClientsApi { + return this._registry; + } + + /** Typed packets: per-type listeners, `read` and sending. */ + public get packets(): ServerPacketsApi { + return this._packets; + } + + /** See `NetworkServerContextApi.sendToAll`. */ + public sendToAll(data: NetworkPayload, { channel = DEFAULT_CHANNEL }: ChannelOptions = {}): void { + this.sendBytesToAll(channel, encodePacket(null, encodeNetworkPayload(data))); + } + + /** See `NetworkServerContextApi.sendToClient`. */ + public sendToClient( + clientId: ClientId, + data: NetworkPayload, + { channel = DEFAULT_CHANNEL }: ChannelOptions = {}, + ): void { + this.sendBytesToClient(clientId, channel, encodePacket(null, encodeNetworkPayload(data))); + } + + /** See `NetworkServerContextApi.getReceivedPackets`. */ + public getReceivedPackets({ channel }: ChannelOptions = {}): Map { + if (channel !== undefined) { + const clients = this.drain(channel); + const packets = this._raw.get(channel) ?? new Map(); + this._raw.delete(channel); + for (const clientId of clients) if (!packets.has(clientId)) packets.set(clientId, []); + return packets; + } + + const merged = new Map(); + for (const c of this.configuredChannels()) { + this.getReceivedPackets({ channel: c }).forEach((packets, clientId) => { + merged.set(clientId, [...(merged.get(clientId) ?? []), ...packets]); + }); + } + return merged; + } + + /** See `NetworkServerContextApi.getConnectedClients`. */ + public getConnectedClients({ channel }: ChannelOptions = {}): ClientId[] { + if (channel !== undefined) return this.require(channel).getConnectedClients(channel); + return [ + ...new Set(this.configuredChannels().flatMap((c) => this.require(c).getConnectedClients(c))), + ]; + } + + /** See `NetworkServerContextApi.getClientInfo`. */ + public getClientInfo(clientId: ClientId): ClientInfo | undefined { + return this._registry.get(clientId); + } + + /** Calls the listeners of the typed packets received since the last tick. */ + public run(ctx: Context): void { + this._raw.clear(); + for (const channel of this.configuredChannels()) this.drain(channel); + const typed = this._typed; + this._typed = []; + this._dispatcher.dispatch(typed, ctx); + } + + public expose(): NetworkServerContextApi { + // eslint-disable-next-line @typescript-eslint/no-this-alias + const server = this; + return { + sendToAll: (data, options) => server.sendToAll(data, options), + sendToClient: (clientId, data, options) => server.sendToClient(clientId, data, options), + getReceivedPackets: (options) => server.getReceivedPackets(options), + getConnectedClients: (options) => server.getConnectedClients(options), + getClientInfo: (clientId) => server.getClientInfo(clientId), + get clients() { + return server.clients; + }, + get packets() { + return server.packets; + }, + }; + } + + private configuredChannels(): Channel[] { + return CHANNELS.filter((channel) => this.transports[channel] !== undefined); + } + + private sendBytesToAll(channel: Channel, bytes: Uint8Array): void { + this.require(channel).sendToEverybody(channel, bytes); + } + + private sendBytesToClient(clientId: ClientId, channel: Channel, bytes: Uint8Array): void { + this.require(channel).sendToClient(channel, clientId, bytes); + } + + /** + * Move what each client sent on `channel` to the typed and raw queues. + * + * @returns The clients the transport reported, packets or not. + */ + private drain(channel: Channel): ClientId[] { + const received = this.require(channel).getReceivedPackets(channel); + const raw = this._raw.get(channel) ?? new Map(); + received.forEach((packets, clientId) => { + const clientRaw = raw.get(clientId) ?? []; + sortPackets(packets, { clientId, channel }, clientRaw, this._typed); + if (clientRaw.length > 0) raw.set(clientId, clientRaw); + }); + this._raw.set(channel, raw); + return [...received.keys()]; + } + + private require(channel: Channel): ServerTransport { + const transport = this.transports[channel]; + if (!transport) throw new Error(this._requirement(channel)); + return transport; + } +} diff --git a/modules/network/src/server/peer-host.ts b/modules/network/src/server/peer-host.ts new file mode 100644 index 0000000..2228682 --- /dev/null +++ b/modules/network/src/server/peer-host.ts @@ -0,0 +1,348 @@ +import { waitForWelcome } from "../client/client-session"; +import { CHANNELS, Channel, PacketSequencer, channelFromLabel } from "../shared/channels"; +import type { WelcomeMessage } from "../shared/session"; +import { type HostSignalMessage, MAX_SIGNAL_BYTES, parseSignal } from "../shared/signaling"; +import type { ClientTransport, ServerTransport } from "../shared/transport"; +import type { ClientId, ClientRegistry, ConnectionInfo } from "./client-registry"; + +/** Host side of one WebRTC data channel. */ +type DataChannelState = { + channel: RTCDataChannel; + packets: Uint8Array[]; + /** Only set on `UnreliableOrdered`. */ + sequencer?: PacketSequencer; +}; + +/** Host side of one joiner's link. */ +type Joiner = { + /** Id the signaling server knows the joiner by. */ + signalId: string; + peerConnection: RTCPeerConnection; + /** ICE candidates received before the offer was applied. */ + pendingCandidates: RTCIceCandidateInit[]; + channels: Map; + /** Whether its offer was received: later ones are ignored. */ + offered: boolean; + /** Session id, once every channel is open. */ + id?: ClientId; +}; + +/** In-memory link between the host's server and its own player. */ +type LocalLink = { + id: ClientId; + /** Packets the server sent to the local player, per channel. */ + inbox: Map; + /** Packets the local player sent to the server, per channel. */ + outbox: Map; +}; + +// shortcut: the browser does not expose a joiner's address before ICE ends; read getStats() if games need it. +const connectionInfo = (transport: "local" | "peer"): ConnectionInfo => ({ + transport, + address: "unknown", + port: 0, + family: "unknown", + params: {}, + connectedAt: Date.now(), +}); + +const take = (queues: Map, channel: Channel): Uint8Array[] => { + const packets = queues.get(channel) ?? []; + queues.set(channel, []); + return packets; +}; + +const push = (queues: Map, channel: Channel, data: Uint8Array): void => { + const queue = queues.get(channel); + if (queue) queue.push(data); + else queues.set(channel, [data]); +}; + +/** + * Server transport of a peer-to-peer host: answers the WebRTC offers of the + * joiners its signaling socket forwards, and carries the four channels to + * each of them over data channels. + * + * @remarks + * Runs in the host's browser, on the native `RTCPeerConnection`. The answer + * side is the one of `UDPServer`, without its libdatachannel-only options. A + * joiner joins the host's sessions, and is welcomed, once all four of its data + * channels are open, so `clients.onConnect` means it can be sent to. It + * leaves when its signaling socket closes, a data channel closes, or its + * connection fails. + * + * The host's own player is a `"local"` session reached through in-memory + * queues (see {@link attachLocal}). + */ +export class PeerHost implements ServerTransport { + private _socket: WebSocket | null = null; + /** Joiners by signaling id, from their first frame on. */ + private readonly _joiners = new Map(); + /** Welcomed joiners by session id. */ + private readonly _clients = new Map(); + private _local: LocalLink | undefined; + private _room: string | undefined; + + constructor( + private readonly _registry: ClientRegistry, + private readonly _iceServers: RTCIceServer[] = [], + ) {} + + /** Code joiners use to reach this host; `undefined` once signaling closed. */ + public get room(): string | undefined { + return this._room; + } + + /** + * Open the signaling socket and get a room for this host. + * + * @param url - Signaling server URL, without a room. + * @returns The room code joiners connect with. + * @throws When the socket fails or closes before the server gives a room. + */ + public async open(url: string): Promise { + const wait = waitForWelcome("Signaling"); + let room = ""; + + const socket = (this._socket = new WebSocket(url)); + socket.onerror = (event) => wait.settle(new Error("Signaling error", { cause: event })); + socket.onclose = () => { + wait.settle(new Error("Signaling closed before giving a room")); + // The server closed the room: its joiners are told the host left. + this._room = undefined; + }; + socket.onmessage = (event: MessageEvent) => { + const message = typeof event.data === "string" ? parseSignal(event.data) : null; + if (message?.type === "room" && typeof message.room === "string" && room === "") { + room = this._room = message.room; + wait.settle(); + } + if (typeof message?.from !== "string") return; + if (message.type === "signal") this.receiveSignal(message.from, message.data); + // Sooner than WebRTC: Chrome keeps a vanished peer's channels open until ICE gives up. + if (message.type === "leave") this.removeJoiner(message.from); + }; + + await wait.promise; + return room; + } + + /** + * Register the host's own player as a `"local"` session. + * + * @returns Its session id, and the client transport its `NetworkClient` uses: + * what it sends reaches this server under that id at the next drain, and + * what this server sends it is read at its own next drain. + */ + public attachLocal(): { id: ClientId; transport: ClientTransport } { + const { id } = this._registry.attach(null, connectionInfo("local")); + const local: LocalLink = (this._local = { id, inbox: new Map(), outbox: new Map() }); + return { + id, + transport: { + sendData: (channel, data) => push(local.outbox, channel, data), + getReceivedPackets: (channel) => take(local.inbox, channel), + isConnected: () => true, + }, + }; + } + + /** Close the signaling socket and every joiner's connection, without notifying. */ + public close(): void { + this._socket?.close(); + this._socket = null; + this._room = undefined; + const joiners = [...this._joiners.values()]; + this._joiners.clear(); + this._clients.clear(); + this._local = undefined; + for (const joiner of joiners) joiner.peerConnection.close(); + } + + public sendToEverybody(channel: Channel, data: Uint8Array): void { + if (this._local) push(this._local.inbox, channel, data); + for (const joiner of this._clients.values()) { + const state = joiner.channels.get(channel); + if (state) this.send(state, data); + } + } + + public sendToClient(channel: Channel, clientId: ClientId, data: Uint8Array): void { + if (clientId === this._local?.id) { + push(this._local.inbox, channel, data); + return; + } + const state = this._clients.get(clientId)?.channels.get(channel); + if (!state) { + console.error(`Unknown client on ${channel}: ${clientId}`); + return; + } + this.send(state, data); + } + + public getReceivedPackets(channel: Channel): Map { + const packets = new Map(); + if (this._local) packets.set(this._local.id, take(this._local.outbox, channel)); + this._clients.forEach((joiner, id) => { + const state = joiner.channels.get(channel); + if (!state) return; + packets.set(id, state.packets); + state.packets = []; + }); + return packets; + } + + public getConnectedClients(channel: Channel): ClientId[] { + const joiners = [...this._clients] + .filter(([, joiner]) => joiner.channels.has(channel)) + .map(([id]) => id); + return this._local ? [this._local.id, ...joiners] : joiners; + } + + private send(state: DataChannelState, data: Uint8Array): void { + if (state.channel.readyState !== "open") return; + state.channel.send(state.sequencer ? state.sequencer.wrap(data) : new Uint8Array(data)); + } + + private signal(signalId: string, data: unknown): void { + if (this._socket?.readyState !== WebSocket.OPEN) return; + const frame = JSON.stringify({ + type: "signal", + to: signalId, + data, + } satisfies HostSignalMessage); + // A crafted offer can inflate the answer past the server's cap, which would drop the host's socket. + if (new TextEncoder().encode(frame).length > MAX_SIGNAL_BYTES) { + console.error(`Peer signaling frame too large, dropping joiner: ${signalId}`); + this.removeJoiner(signalId); + return; + } + this._socket.send(frame); + } + + /** Handle a frame a joiner sent through the signaling server. */ + private receiveSignal(signalId: string, data: unknown): void { + if (typeof data !== "object" || data === null) return; + const message = data as Record; + let joiner = this._joiners.get(signalId); + if (!joiner) { + // Only an offer costs a peer connection; a joiner sends it before any ICE candidate. + if (message.type !== "offer") return; + // Created before any `await`, so ICE candidates sent along with the offer are queued, not lost. + joiner = this.addJoiner(signalId); + } + + if (message.type === "offer" && !joiner.offered) { + joiner.offered = true; + this.answer(joiner, message.offer as RTCSessionDescriptionInit).catch((error: unknown) => { + console.error(`Peer offer failed for joiner: ${signalId}`, { cause: error }); + this.removeJoiner(signalId); + }); + } + + if (message.type === "ice" && typeof message.candidate === "object" && message.candidate) { + const candidate = message.candidate as RTCIceCandidateInit; + if (!joiner.peerConnection.remoteDescription) { + joiner.pendingCandidates.push(candidate); + return; + } + joiner.peerConnection.addIceCandidate(candidate).catch((error: unknown) => { + console.error(`Peer ICE candidate rejected for joiner: ${signalId}`, { cause: error }); + }); + } + } + + private addJoiner(signalId: string): Joiner { + const peerConnection = new RTCPeerConnection({ iceServers: this._iceServers }); + const joiner: Joiner = { + signalId, + peerConnection, + pendingCandidates: [], + channels: new Map(), + offered: false, + }; + this._joiners.set(signalId, joiner); + + peerConnection.onicecandidate = (event) => { + if (event.candidate) this.signal(signalId, { type: "ice", candidate: event.candidate }); + }; + peerConnection.onconnectionstatechange = () => { + const state = peerConnection.connectionState; + if (state === "failed" || state === "closed") this.removeJoiner(signalId); + }; + peerConnection.ondatachannel = (event) => this.receiveDataChannel(joiner, event.channel); + return joiner; + } + + private async answer(joiner: Joiner, offer: RTCSessionDescriptionInit): Promise { + const { peerConnection } = joiner; + await peerConnection.setRemoteDescription(offer); + + for (const candidate of joiner.pendingCandidates) { + await peerConnection.addIceCandidate(candidate); + } + joiner.pendingCandidates.length = 0; + + await peerConnection.setLocalDescription(await peerConnection.createAnswer()); + const answer = peerConnection.localDescription; + this.signal(joiner.signalId, { + type: "answer", + answer: answer ? { type: answer.type, sdp: answer.sdp } : null, + }); + } + + private receiveDataChannel(joiner: Joiner, channel: RTCDataChannel): void { + const kind = channelFromLabel(channel.label); + if (kind === undefined || joiner.channels.has(kind)) { + console.error(`Peer data channel "${channel.label}" refused for joiner: ${joiner.signalId}`); + channel.close(); + return; + } + + channel.binaryType = "arraybuffer"; + const state: DataChannelState = { + channel, + packets: [], + ...(kind === Channel.UnreliableOrdered ? { sequencer: new PacketSequencer() } : {}), + }; + joiner.channels.set(kind, state); + + channel.onopen = () => this.welcomeWhenOpen(joiner); + channel.onmessage = (event: MessageEvent) => { + if (!(event.data instanceof ArrayBuffer)) return; + const packet = new Uint8Array(event.data); + const payload = state.sequencer ? state.sequencer.unwrap(packet) : packet; + if (payload) state.packets.push(payload); + }; + channel.onclose = () => this.removeJoiner(joiner.signalId); + channel.onerror = (event) => { + console.error(`Peer ${kind} error for joiner: ${joiner.signalId}`, { cause: event }); + }; + // A channel announced by `ondatachannel` may already be open. + this.welcomeWhenOpen(joiner); + } + + /** Join the joiner to the sessions once all four channels are open. */ + private welcomeWhenOpen(joiner: Joiner): void { + if (joiner.id !== undefined || this._joiners.get(joiner.signalId) !== joiner) return; + const open = CHANNELS.every((c) => joiner.channels.get(c)?.channel.readyState === "open"); + if (!open) return; + + // Reachable before the connect listeners run, so they can send to it. + const { id, token } = this._registry.attach(null, connectionInfo("peer"), ({ id }) => { + joiner.id = id; + this._clients.set(id, joiner); + }); + this.signal(joiner.signalId, { type: "welcome", id, token } satisfies WelcomeMessage); + } + + private removeJoiner(signalId: string): void { + const joiner = this._joiners.get(signalId); + if (!joiner) return; + this._joiners.delete(signalId); + joiner.peerConnection.close(); + if (joiner.id === undefined) return; + this._clients.delete(joiner.id); + this._registry.detach(joiner.id, "peer"); + } +} diff --git a/modules/network/src/server/server.network.base.ts b/modules/network/src/server/server.network.base.ts new file mode 100644 index 0000000..18d1e65 --- /dev/null +++ b/modules/network/src/server/server.network.base.ts @@ -0,0 +1,122 @@ +import { type Context, Library, defineLibraryKey } from "@nanoforge-dev/common"; +import type { EditorAwareLibrary, EditorInitContext } from "@nanoforge-dev/editor-lib"; + +import type { Channel, ChannelOptions } from "../shared/channels"; +import type { NetworkData, NetworkPayload } from "../shared/network-data"; +import type { Trace } from "../shared/trace"; +import { type ClientId, type ClientInfo, ClientRegistry, type ClientsApi } from "./client-registry"; +import { NetworkServer } from "./network-server"; +import type { NetworkServerContextApi, ServerPacketsApi } from "./network-server-context.type"; + +/** + * Public API of `NetworkServerLibrary`, shared by its Bun build (TCP and UDP + * servers) and its browser build (peer-to-peer host). + */ +export abstract class NetworkServerLibraryBase extends Library implements EditorAwareLibrary { + readonly key = defineLibraryKey("network"); + + protected readonly _registry = new ClientRegistry(); + protected readonly _server: NetworkServer; + protected _trace: Trace | undefined; + + /** @param requirement - Error message for a channel with no transport. */ + constructor(requirement: (channel: Channel) => string) { + super({ runAfter: ["ecs"] }); + this._server = new NetworkServer(this._registry, requirement); + } + + /** Client sessions shared by the transports. */ + public get clients(): ClientsApi { + return this._server.clients; + } + + /** Typed packets: per-type listeners, `read` and sending. */ + public get packets(): ServerPacketsApi { + return this._server.packets; + } + + /** + * Send a payload to every client reachable on a channel. + * + * @param data - Bytes, a string, or any value `JSON.stringify` can encode. + * @param options - `channel` to send on, `Channel.ReliableOrdered` by default. + * @throws When the channel's port was not configured, or `data` cannot be + * encoded as JSON. + */ + public sendToAll(data: NetworkPayload, options?: ChannelOptions): void { + this._server.sendToAll(data, options); + } + + /** + * Send a payload to one client. + * + * @param clientId - Client identifier. + * @param data - Bytes, a string, or any value `JSON.stringify` can encode. + * @param options - `channel` to send on, `Channel.ReliableOrdered` by default. + * @throws When the channel's port was not configured, or `data` cannot be + * encoded as JSON. + */ + public sendToClient(clientId: ClientId, data: NetworkPayload, options?: ChannelOptions): void { + this._server.sendToClient(clientId, data, options); + } + + /** + * Return the packets each client sent since the last call. + * + * @remarks + * Call this method once per frame: packets left unread for a whole tick are + * dropped, and typed packets never show up here. Without a `channel`, the + * packets of every configured channel are merged per client, reliable channels first. + * + * @param options - `channel` to read from, every channel by default. + * @returns Map of client id to packets, each read with `bytes()`, `text()`, + * `json()`… + * @throws When the channel's port was not configured. + */ + public getReceivedPackets(options?: ChannelOptions): Map { + return this._server.getReceivedPackets(options); + } + + /** + * Return the clients reachable on a channel. + * + * @param options - `channel` to check, any configured channel by default. + * @returns Snapshot array of client ids. + * @throws When the channel's port was not configured. + */ + public getConnectedClients(options?: ChannelOptions): ClientId[] { + return this._server.getConnectedClients(options); + } + + /** + * Return what is known about a client session: its address, user agent, + * query parameters and attached transports. + * + * @param clientId - Client identifier. + * @returns The live session info, or `undefined` when the client is gone. + */ + public getClientInfo(clientId: ClientId): ClientInfo | undefined { + return this._server.getClientInfo(clientId); + } + + /** + * Traces packets for the editor (`network-trace`, `network-stats`) while + * its `welcome` asks for it. + */ + public abstract __editorInit(ctx: EditorInitContext): Promise; + + /** Sends the totals of the packets traced for the editor (also while paused). */ + // eslint-disable-next-line @typescript-eslint/no-unused-vars + public override async __events(_ctx: Context): Promise { + this._trace?.tick(); + } + + /** Calls the listeners of the typed packets received since the last tick. */ + public override async __run(ctx: Context): Promise { + this._server.run(ctx); + } + + public override expose(): NetworkServerContextApi { + return this._server.expose(); + } +} diff --git a/modules/network/src/server/server.network.browser.ts b/modules/network/src/server/server.network.browser.ts new file mode 100644 index 0000000..1cf2f0e --- /dev/null +++ b/modules/network/src/server/server.network.browser.ts @@ -0,0 +1,58 @@ +import type { Context, InitContext } from "@nanoforge-dev/common"; +import type { EditorInitContext } from "@nanoforge-dev/editor-lib"; +import { registerEnv } from "@nanoforge-dev/env"; + +import { buildServerUrl } from "../client/client-session"; +import { ClientConfigNetwork } from "../client/config.client.network"; +import { CHANNELS } from "../shared/channels"; +import { PEER_HOST_KEY } from "../shared/signaling"; +import { createTrace, traceServer } from "../shared/trace"; +import { PeerHost } from "./peer-host"; +import { NetworkServerLibraryBase } from "./server.network.base"; + +/** + * `NetworkServerLibrary` as bundled for the browser, when the game sets + * `network: "p2p"`: it runs in the host player's tab, accepts joiners over + * WebRTC through the signaling server, and lets the page's client join it + * through an in-memory loopback. + * + * @remarks + * Reads the client's environment: `SERVER_ADDRESS`, `SERVER_TCP_PORT` and + * `WSS` locate the signaling server, `ICE_SERVERS` the STUN/TURN servers. + */ +export class NetworkServerLibrary extends NetworkServerLibraryBase { + private _host?: PeerHost; + + constructor() { + super((channel) => `Channel ${channel} is not open`); + } + + public override async __init(ctx: InitContext): Promise { + const config = await registerEnv(ClientConfigNetwork, ctx.env); + if (config.SERVER_TCP_PORT === undefined) { + throw new Error( + "NetworkServerLibrary: p2p needs SERVER_TCP_PORT to reach the signaling server.", + ); + } + + const host = (this._host = new PeerHost(this._registry, config.ICE_SERVERS)); + await host.open(buildServerUrl(config.WSS, config.SERVER_ADDRESS, +config.SERVER_TCP_PORT, {})); + this._server.transports = Object.fromEntries(CHANNELS.map((channel) => [channel, host])); + Reflect.set(globalThis, PEER_HOST_KEY, host); + } + + // eslint-disable-next-line @typescript-eslint/no-unused-vars + public override async __clear(_ctx: Context): Promise { + this._host?.close(); + this._registry.clear(); + delete this._host; + this._server.transports = {}; + Reflect.deleteProperty(globalThis, PEER_HOST_KEY); + } + + public override async __editorInit(ctx: EditorInitContext): Promise { + const trace = createTrace(ctx.editor); + if (this._host) traceServer(this._host, "udp", trace); + this._trace = trace; + } +} diff --git a/modules/network/src/server/server.network.library.ts b/modules/network/src/server/server.network.library.ts index 6a1783e..2563d0f 100644 --- a/modules/network/src/server/server.network.library.ts +++ b/modules/network/src/server/server.network.library.ts @@ -1,37 +1,29 @@ -import { type Context, type InitContext, Library, defineLibraryKey } from "@nanoforge-dev/common"; -import type { EditorAwareLibrary, EditorInitContext } from "@nanoforge-dev/editor-lib"; +import type { Context, InitContext } from "@nanoforge-dev/common"; +import type { EditorInitContext } from "@nanoforge-dev/editor-lib"; import { registerEnv } from "@nanoforge-dev/env"; import { CHANNELS, type Channel, - type ChannelOptions, - DEFAULT_CHANNEL, + RELIABLE_CHANNELS, + UNRELIABLE_CHANNELS, isReliableChannel, } from "../shared/channels"; -import { - type NetworkData, - type NetworkPayload, - encodeNetworkPayload, -} from "../shared/network-data"; -import { - PacketDispatcher, - type TypedPacket, - encodePacket, - encodeTypedPacket, - sortPackets, -} from "../shared/packet"; -import { type Trace, createTrace, traceServer } from "../shared/trace"; -import { type ClientId, type ClientInfo, ClientRegistry, type ClientsApi } from "./client-registry"; +import { createTrace, traceServer } from "../shared/trace"; +import type { ServerTransport } from "../shared/transport"; import { ServerConfigNetwork } from "./config.server.network"; -import type { - NetworkServerContextApi, - ServerPacketInfo, - ServerPacketsApi, -} from "./network-server-context.type"; +import { NetworkServerLibraryBase } from "./server.network.base"; +import { SignalingServer } from "./signaling.server.network"; import { TCPServer } from "./tcp.server.network"; import { UDPServer } from "./udp.server.network"; +const NO_CLIENTS: ServerTransport = { + sendToEverybody: () => {}, + sendToClient: () => {}, + getReceivedPackets: () => new Map(), + getConnectedClients: () => [], +}; + /** * Built-in network library for server-side applications. * @@ -48,6 +40,9 @@ import { UDPServer } from "./udp.server.network"; * - `ICE_SERVERS` — STUN/TURN servers for the UDP transport, comma-separated or a JSON array (default: `[]`). * - `ICE_PORT` — fixed, multiplexed UDP port for the UDP transport (optional). * - `ADVERTISE_IP` — public address written into host ICE candidates (optional). + * - `NETWORK_MODE` — `p2p` (set by `nf` from `network: "p2p"`): the game runs in + * the host player's browser, and this server only pairs players, on + * `LISTENING_TCP_PORT`: the server app still runs here, with no players. * * @example * ```ts @@ -59,43 +54,16 @@ import { UDPServer } from "./udp.server.network"; * }); * ``` */ -export class NetworkServerLibrary extends Library implements EditorAwareLibrary { - readonly key = defineLibraryKey("network"); - +export class NetworkServerLibrary extends NetworkServerLibraryBase { private _tcp?: TCPServer; private _udp?: UDPServer; - - private readonly _registry = new ClientRegistry(); - private readonly _dispatcher = new PacketDispatcher(); - private _typed: TypedPacket[] = []; - private readonly _raw = new Map>(); - private readonly _packets: ServerPacketsApi = { - on: (packet, listener) => this._dispatcher.on(packet, listener), - read: (packet) => this._dispatcher.read(packet), - sendToAll: (packet, ...[data, options]) => - this.sendBytesToAll(options?.channel ?? packet.channel, encodeTypedPacket(packet, data)), - sendToClient: (clientId, packet, ...[data, options]) => - this.sendBytesToClient( - clientId, - options?.channel ?? packet.channel, - encodeTypedPacket(packet, data), - ), - }; - - /** Client sessions shared by the TCP and UDP servers. */ - public get clients(): ClientsApi { - return this._registry; - } - - private _trace: Trace | undefined; + private _signaling?: SignalingServer; constructor() { - super({ runAfter: ["ecs"] }); - } - - /** Typed packets: per-type listeners, `read` and sending. */ - public get packets(): ServerPacketsApi { - return this._packets; + super( + (channel) => + `Channel ${channel} needs ${isReliableChannel(channel) ? "LISTENING_TCP_PORT" : "LISTENING_UDP_PORT"} to be set`, + ); } public override async __init(ctx: InitContext): Promise { @@ -119,6 +87,21 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary throw new Error("NetworkServerLibrary: both WSS_CERT and WSS_KEY must be provided together."); } + if (config.NETWORK_MODE === "p2p") { + if (config.LISTENING_TCP_PORT === undefined) { + throw new Error("NetworkServerLibrary: p2p needs LISTENING_TCP_PORT for signaling."); + } + this._signaling = new SignalingServer( + +config.LISTENING_TCP_PORT, + config.LISTENING_INTERFACE, + config.WSS_CERT, + config.WSS_KEY, + ); + this._signaling.listen(); + this._server.transports = Object.fromEntries(CHANNELS.map((c) => [c, NO_CLIENTS])); + return; + } + if (config.LISTENING_TCP_PORT !== undefined) { this._tcp = new TCPServer( +config.LISTENING_TCP_PORT, @@ -145,194 +128,38 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary ); this._udp.listen(); } + + this.routeChannels(); } // eslint-disable-next-line @typescript-eslint/no-unused-vars public override async __clear(_ctx: Context): Promise { this._tcp?.close(); this._udp?.close(); + this._signaling?.stop(); this._registry.clear(); delete this._tcp; delete this._udp; - } - - /** - * Send a payload to every client reachable on a channel. - * - * @param data - Bytes, a string, or any value `JSON.stringify` can encode. - * @param options - `channel` to send on, `Channel.ReliableOrdered` by default. - * @throws When the channel's port was not configured, or `data` cannot be - * encoded as JSON. - */ - public sendToAll(data: NetworkPayload, { channel = DEFAULT_CHANNEL }: ChannelOptions = {}): void { - this.sendBytesToAll(channel, encodePacket(null, encodeNetworkPayload(data))); - } - - /** - * Send a payload to one client. - * - * @param clientId - Client identifier. - * @param data - Bytes, a string, or any value `JSON.stringify` can encode. - * @param options - `channel` to send on, `Channel.ReliableOrdered` by default. - * @throws When the channel's port was not configured, or `data` cannot be - * encoded as JSON. - */ - public sendToClient( - clientId: ClientId, - data: NetworkPayload, - { channel = DEFAULT_CHANNEL }: ChannelOptions = {}, - ): void { - this.sendBytesToClient(clientId, channel, encodePacket(null, encodeNetworkPayload(data))); - } - - /** - * Return the packets each client sent since the last call. - * - * @remarks - * Call this method once per frame: packets left unread for a whole tick are - * dropped, and typed packets never show up here. Without a `channel`, the - * packets of every configured channel are merged per client, reliable channels first. - * - * @param options - `channel` to read from, every channel by default. - * @returns Map of client id to packets, each read with `bytes()`, `text()`, - * `json()`… - * @throws When the channel's port was not configured. - */ - public getReceivedPackets({ channel }: ChannelOptions = {}): Map { - if (channel !== undefined) { - const clients = this.drain(channel); - const packets = this._raw.get(channel) ?? new Map(); - this._raw.delete(channel); - for (const clientId of clients) if (!packets.has(clientId)) packets.set(clientId, []); - return packets; - } - - const merged = new Map(); - for (const c of this.configuredChannels()) { - this.getReceivedPackets({ channel: c }).forEach((packets, clientId) => { - merged.set(clientId, [...(merged.get(clientId) ?? []), ...packets]); - }); - } - return merged; - } - - /** - * Return the clients reachable on a channel. - * - * @param options - `channel` to check, any configured channel by default. - * @returns Snapshot array of client ids. - * @throws When the channel's port was not configured. - */ - public getConnectedClients({ channel }: ChannelOptions = {}): ClientId[] { - if (channel !== undefined) { - if (isReliableChannel(channel)) return this.requireTcp(channel).getConnectedClients(); - return this.requireUdp(channel).getConnectedClients(channel); - } - return [ - ...new Set([ - ...(this._tcp?.getConnectedClients() ?? []), - ...(this._udp?.getConnectedClients() ?? []), - ]), - ]; - } - - /** - * Return what is known about a client session: its address, user agent, - * query parameters and attached transports. - * - * @param clientId - Client identifier. - * @returns The live session info, or `undefined` when the client is gone. - */ - public getClientInfo(clientId: ClientId): ClientInfo | undefined { - return this._registry.get(clientId); + delete this._signaling; + this.routeChannels(); } /** * Traces packets for the editor (`network-trace`, `network-stats`) while * its `welcome` asks for it. */ - public async __editorInit(ctx: EditorInitContext): Promise { + public override async __editorInit(ctx: EditorInitContext): Promise { const trace = createTrace(ctx.editor); if (this._tcp) traceServer(this._tcp, "tcp", trace); if (this._udp) traceServer(this._udp, "udp", trace); this._trace = trace; } - /** Sends the totals of the packets traced for the editor (also while paused). */ - // eslint-disable-next-line @typescript-eslint/no-unused-vars - public override async __events(_ctx: Context): Promise { - this._trace?.tick(); - } - - /** Calls the listeners of the typed packets received since the last tick. */ - public override async __run(ctx: Context): Promise { - this._raw.clear(); - for (const channel of this.configuredChannels()) this.drain(channel); - const typed = this._typed; - this._typed = []; - this._dispatcher.dispatch(typed, ctx); - } - - public override expose(): NetworkServerContextApi { - // eslint-disable-next-line @typescript-eslint/no-this-alias - const library = this; - return { - sendToAll: (data, options) => library.sendToAll(data, options), - sendToClient: (clientId, data, options) => library.sendToClient(clientId, data, options), - getReceivedPackets: (options) => library.getReceivedPackets(options), - getConnectedClients: (options) => library.getConnectedClients(options), - getClientInfo: (clientId) => library.getClientInfo(clientId), - get clients() { - return library.clients; - }, - get packets() { - return library.packets; - }, + /** TCP carries the reliable channels, UDP the unreliable ones. */ + private routeChannels(): void { + this._server.transports = { + ...(this._tcp && Object.fromEntries(RELIABLE_CHANNELS.map((c) => [c, this._tcp]))), + ...(this._udp && Object.fromEntries(UNRELIABLE_CHANNELS.map((c) => [c, this._udp]))), }; } - - private configuredChannels(): Channel[] { - return CHANNELS.filter((channel) => - isReliableChannel(channel) ? this._tcp !== undefined : this._udp !== undefined, - ); - } - - private sendBytesToAll(channel: Channel, bytes: Uint8Array): void { - if (isReliableChannel(channel)) this.requireTcp(channel).sendToEverybody(channel, bytes); - else this.requireUdp(channel).sendToEverybody(channel, bytes); - } - - private sendBytesToClient(clientId: ClientId, channel: Channel, bytes: Uint8Array): void { - if (isReliableChannel(channel)) this.requireTcp(channel).sendToClient(channel, clientId, bytes); - else this.requireUdp(channel).sendToClient(channel, clientId, bytes); - } - - /** - * Move what each client sent on `channel` to the typed and raw queues. - * - * @returns The clients the transport reported, packets or not. - */ - private drain(channel: Channel): ClientId[] { - const received = isReliableChannel(channel) - ? this.requireTcp(channel).getReceivedPackets(channel) - : this.requireUdp(channel).getReceivedPackets(channel); - const raw = this._raw.get(channel) ?? new Map(); - received.forEach((packets, clientId) => { - const clientRaw = raw.get(clientId) ?? []; - sortPackets(packets, { clientId, channel }, clientRaw, this._typed); - if (clientRaw.length > 0) raw.set(clientId, clientRaw); - }); - this._raw.set(channel, raw); - return [...received.keys()]; - } - - private requireTcp(channel: Channel): TCPServer { - if (!this._tcp) throw new Error(`Channel ${channel} needs LISTENING_TCP_PORT to be set`); - return this._tcp; - } - - private requireUdp(channel: Channel): UDPServer { - if (!this._udp) throw new Error(`Channel ${channel} needs LISTENING_UDP_PORT to be set`); - return this._udp; - } } diff --git a/modules/network/src/server/signaling.server.network.ts b/modules/network/src/server/signaling.server.network.ts new file mode 100644 index 0000000..12999d3 --- /dev/null +++ b/modules/network/src/server/signaling.server.network.ts @@ -0,0 +1,161 @@ +import type { Server, ServerWebSocket } from "bun"; + +import { + HOST_LEFT_CODE, + type JoinerSignalMessage, + type LeaveMessage, + MAX_SIGNAL_BYTES, + ROOM_PARAM, + type RoomMessage, + parseSignal, +} from "../shared/signaling"; + +/** Room codes avoid 0/O and 1/I, so they can be read aloud or typed. */ +const ROOM_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"; +const ROOM_LENGTH = 6; + +type SignalingSocketData = + { role: "host"; room?: string } | { role: "joiner"; room: string; id: string }; + +type SignalingSocket = ServerWebSocket; + +type Room = { host: SignalingSocket; joiners: Map }; + +/** 6 characters from a 32-letter alphabet: each random byte maps without bias. */ +const generateRoomCode = (): string => + Array.from( + crypto.getRandomValues(new Uint8Array(ROOM_LENGTH)), + (byte) => ROOM_ALPHABET[byte % ROOM_ALPHABET.length], + ).join(""); + +/** + * WebSocket server pairing peer-to-peer hosts with their joiners for the + * WebRTC handshake (see `shared/signaling.ts` for the protocol). + * + * @remarks + * Game traffic never goes through it: once a joiner's data channels are open, + * its signaling socket is idle. Frames are written by players, so they are + * parsed defensively, capped at `MAX_SIGNAL_BYTES`, and a joiner can only ever + * reach its own room's host. + * + * Runs on `Bun.serve`; started by `NetworkServerLibrary` when + * `NETWORK_MODE` is `p2p`, on `LISTENING_TCP_PORT`. + */ +export class SignalingServer { + private readonly _rooms = new Map(); + private _server: Server | undefined; + + constructor( + private _port: number, + private _host: string, + private _cert?: string, + private _key?: string, + ) {} + + /** Start accepting hosts and joiners. */ + public listen(): void { + const cert = this._cert; + const key = this._key; + const secure = cert !== undefined && key !== undefined; + + this._server = Bun.serve({ + port: this._port, + hostname: this._host, + ...(secure ? { tls: { cert: Bun.file(cert), key: Bun.file(key) } } : {}), + fetch: (request, server) => { + const room = new URL(request.url).searchParams.get(ROOM_PARAM); + if (room !== null && !this._rooms.has(room)) { + return new Response("Unknown room", { status: 404 }); + } + + const data: SignalingSocketData = + room === null ? { role: "host" } : { role: "joiner", room, id: crypto.randomUUID() }; + if (server.upgrade(request, { data })) return undefined; + return new Response("Expected a WebSocket connection", { status: 426 }); + }, + websocket: { + maxPayloadLength: MAX_SIGNAL_BYTES, + open: (webSocket) => this.open(webSocket), + message: (webSocket, message) => + this.message( + webSocket, + typeof message === "string" ? message : new TextDecoder().decode(message), + ), + close: (webSocket) => this.close(webSocket), + }, + }); + + console.log( + `${secure ? "Secure " : ""}Signaling server listening on ` + + `${secure ? "wss" : "ws"}://${this._host}:${this._port}`, + ); + } + + /** Stop the server and forget every room. */ + public stop(): void { + this._server?.stop(true); + this._server = undefined; + this._rooms.clear(); + } + + private open(webSocket: SignalingSocket): void { + const data = webSocket.data; + if (data.role === "host") { + let room = generateRoomCode(); + while (this._rooms.has(room)) room = generateRoomCode(); + data.room = room; + this._rooms.set(room, { host: webSocket, joiners: new Map() }); + webSocket.send(JSON.stringify({ type: "room", room } satisfies RoomMessage)); + return; + } + + // The host may have left between the upgrade request and now. + const room = this._rooms.get(data.room); + if (!room) { + webSocket.close(1001, "Unknown room"); + return; + } + room.joiners.set(data.id, webSocket); + } + + private message(webSocket: SignalingSocket, frame: string): void { + const data = webSocket.data; + const room = data.room === undefined ? undefined : this._rooms.get(data.room); + const message = parseSignal(frame); + if (!room || !message) return; + + if (data.role === "joiner") { + if (room.joiners.get(data.id) !== webSocket) return; + room.host.send( + JSON.stringify({ + type: "signal", + from: data.id, + data: message, + } satisfies JoinerSignalMessage), + ); + return; + } + + if (message.type !== "signal" || typeof message.to !== "string") return; + if (message.data === undefined) return; + room.joiners.get(message.to)?.send(JSON.stringify(message.data)); + } + + private close(webSocket: SignalingSocket): void { + const data = webSocket.data; + const code = data.room; + const room = code === undefined ? undefined : this._rooms.get(code); + if (!room || code === undefined) return; + + if (data.role === "joiner") { + if (room.joiners.get(data.id) !== webSocket) return; + room.joiners.delete(data.id); + room.host.send(JSON.stringify({ type: "leave", from: data.id } satisfies LeaveMessage)); + return; + } + + if (room.host !== webSocket) return; + this._rooms.delete(code); + for (const joiner of room.joiners.values()) joiner.close(HOST_LEFT_CODE, "Host left"); + } +} diff --git a/modules/network/src/shared/channels.ts b/modules/network/src/shared/channels.ts index df91d08..cfc0315 100644 --- a/modules/network/src/shared/channels.ts +++ b/modules/network/src/shared/channels.ts @@ -117,6 +117,28 @@ export const UNRELIABLE_CHANNEL_OPTIONS: RTCDataChannelInit = { maxRetransmits: 0, }; +/** + * WebRTC data channel options of every channel, for links where all four run + * over WebRTC (peer-to-peer). The unreliable ones match + * `UNRELIABLE_CHANNEL_OPTIONS`; `ReliableUnordered` really is unordered here, + * unlike over the TCP WebSocket. + */ +export const DATA_CHANNEL_OPTIONS: Record = { + [Channel.ReliableOrdered]: {}, + [Channel.ReliableUnordered]: { ordered: false }, + [Channel.UnreliableOrdered]: UNRELIABLE_CHANNEL_OPTIONS, + [Channel.UnreliableUnordered]: UNRELIABLE_CHANNEL_OPTIONS, +}; + +/** + * Find the channel a WebRTC data channel stands for, on a peer-to-peer link. + * + * @param label - Label of the data channel, which is the channel value. + * @returns The channel, or `undefined` for an unknown label. + */ +export const channelFromLabel = (label: string): Channel | undefined => + CHANNELS.find((channel) => channel === label); + /** * Find the unreliable channel a WebRTC data channel stands for. * diff --git a/modules/network/src/shared/config.network.ts b/modules/network/src/shared/config.network.ts index 600fb5d..e1c3bdb 100644 --- a/modules/network/src/shared/config.network.ts +++ b/modules/network/src/shared/config.network.ts @@ -1,4 +1,4 @@ -import { Default, Expose, IsArray, Transform } from "@nanoforge-dev/env"; +import { Default, Expose, IsArray, IsIn, Transform } from "@nanoforge-dev/env"; import { DEFAULT_NANOFORGE_ICE_SERVERS } from "../default"; @@ -65,4 +65,15 @@ export class NetworkConfig { @Default(DEFAULT_NANOFORGE_ICE_SERVERS) @IsArray() ICE_SERVERS!: RTCIceServer[]; + + /** + * `p2p` when the game's `nanoforge.config.ts` sets `network: "p2p"`: the + * server app runs in the host player's browser. Set by the `nf` CLI. + * + * @default "dedicated" + */ + @Expose() + @Default("dedicated") + @IsIn(["dedicated", "p2p"]) + NETWORK_MODE!: "dedicated" | "p2p"; } diff --git a/modules/network/src/shared/signaling.ts b/modules/network/src/shared/signaling.ts new file mode 100644 index 0000000..7288e11 --- /dev/null +++ b/modules/network/src/shared/signaling.ts @@ -0,0 +1,55 @@ +/** + * Messages of the peer-to-peer signaling server (`NETWORK_MODE=p2p`), + * which pairs a host with its joiners for the WebRTC handshake only. + * + * @remarks + * A host connects without a `room` query parameter and is given a room code; + * a joiner connects with `?room=`. Every frame a joiner sends reaches + * the host wrapped in a `signal` with the joiner's id; the host answers with a + * `signal` addressed `to` that id, and the server forwards its `data` alone. + * Joiners therefore speak the same protocol as with a dedicated UDP server. + */ + +/** Query parameter a joiner names its room with. */ +export const ROOM_PARAM = "room"; + +/** + * Close code of a joiner's signaling socket when its host left: the game is + * over, whatever its data channels report (Chrome kept those of a closed host + * tab open for minutes in testing). + */ +export const HOST_LEFT_CODE = 4001; + +/** Global key under which the browser `NetworkServerLibrary` publishes its `PeerHost` for the page's client. */ +export const PEER_HOST_KEY = Symbol.for("nanoforge.network.host"); + +/** Largest signaling frame accepted, far above any SDP offer or ICE candidate. */ +export const MAX_SIGNAL_BYTES = 64 * 1024; + +/** Server → host: the code joiners use to reach this host. */ +export type RoomMessage = { type: "room"; room: string }; + +/** Server → host: a frame a joiner sent. */ +export type JoinerSignalMessage = { type: "signal"; from: string; data: unknown }; + +/** Host → server: a frame for one joiner of its room. */ +export type HostSignalMessage = { type: "signal"; to: string; data: unknown }; + +/** Server → host: a joiner's signaling socket closed. */ +export type LeaveMessage = { type: "leave"; from: string }; + +/** + * Parse a signaling frame written by a peer. + * + * @returns The parsed object, or `null` for anything else than a JSON object. + */ +export const parseSignal = (frame: string): Record | null => { + try { + const message: unknown = JSON.parse(frame); + return typeof message === "object" && message !== null + ? (message as Record) + : null; + } catch { + return null; + } +}; diff --git a/modules/network/src/shared/trace.ts b/modules/network/src/shared/trace.ts index 7d55fe1..948d7fb 100644 --- a/modules/network/src/shared/trace.ts +++ b/modules/network/src/shared/trace.ts @@ -4,6 +4,8 @@ import type { EditorNetworkTrace, } from "@nanoforge-dev/editor-lib"; +import type { ClientTransport, ServerTransport } from "./transport"; + type Transport = EditorNetworkTrace["transport"]; export interface Trace { @@ -91,17 +93,6 @@ export const createTrace = (editor: EditorContextApi): Trace => { return trace; }; -interface ClientTransport { - sendData(channel: C, data: Uint8Array): void; - getReceivedPackets(channel: C): Uint8Array[]; -} - -interface ServerTransport { - sendToEverybody(channel: C, data: Uint8Array): void; - sendToClient(channel: C, clientId: string, data: Uint8Array): void; - getReceivedPackets(channel: C): Map; -} - /** Traces the packets a client transport sends and receives. */ export const traceClient = ( transport: ClientTransport, diff --git a/modules/network/src/shared/transport.ts b/modules/network/src/shared/transport.ts new file mode 100644 index 0000000..0ca3f84 --- /dev/null +++ b/modules/network/src/shared/transport.ts @@ -0,0 +1,20 @@ +/** + * Client side of a transport: what `NetworkClient` needs from the TCP, UDP, + * peer or loopback links carrying its channels. + */ +export interface ClientTransport { + sendData(channel: C, data: Uint8Array): void; + getReceivedPackets(channel: C): Uint8Array[]; + isConnected(channel: C): boolean; +} + +/** + * Server side of a transport: what `NetworkServer` needs from the TCP, UDP or + * peer-host links carrying its channels. + */ +export interface ServerTransport { + sendToEverybody(channel: C, data: Uint8Array): void; + sendToClient(channel: C, clientId: string, data: Uint8Array): void; + getReceivedPackets(channel: C): Map; + getConnectedClients(channel: C): string[]; +} diff --git a/modules/network/test/client/client.network.p2p.spec.ts b/modules/network/test/client/client.network.p2p.spec.ts new file mode 100644 index 0000000..73fffdb --- /dev/null +++ b/modules/network/test/client/client.network.p2p.spec.ts @@ -0,0 +1,163 @@ +import type { InitContext } from "@nanoforge-dev/common"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { NetworkClientLibrary, definePacket } from "../../src/client"; +import { PEER_HOST_KEY } from "../../src/shared/signaling"; + +const env = { + NETWORK_MODE: "p2p", + SERVER_ADDRESS: "127.0.0.1", + SERVER_TCP_PORT: "4445", +}; + +const makeInitContext = (files: [string, string][] = []): InitContext => ({ + vars: { get: () => undefined, set: () => {} }, + env, + files: new Map(files), +}); + +const moduleUrl = (code: string) => `data:text/javascript,${encodeURIComponent(code)}`; + +const Play = definePacket("play"); + +describe("NetworkClientLibrary in p2p mode", () => { + let sockets: any[]; + let dataChannels: any[]; + + beforeEach(() => { + sockets = []; + dataChannels = []; + vi.stubGlobal("location", { search: "", origin: "https://game.test", pathname: "/play" }); + vi.stubGlobal( + "WebSocket", + Object.assign( + vi.fn(function (this: any, url: string) { + Object.assign(this, { url, readyState: 1, send: vi.fn(), close: vi.fn() }); + sockets.push(this); + }), + { OPEN: 1 }, + ), + ); + vi.stubGlobal( + "RTCPeerConnection", + vi.fn(function (this: any) { + Object.assign(this, { + close: vi.fn(), + createDataChannel: vi.fn((label: string) => { + const channel = { label, readyState: "open", send: vi.fn() }; + dataChannels.push(channel); + return channel; + }), + createOffer: vi.fn().mockResolvedValue({ type: "offer", sdp: "v=0" }), + setLocalDescription: vi.fn().mockResolvedValue(undefined), + }); + }), + ); + vi.spyOn(console, "log").mockImplementation(() => {}); + }); + + afterEach(() => { + Reflect.deleteProperty(globalThis, PEER_HOST_KEY); + Reflect.deleteProperty(globalThis, "__serverArgs"); + vi.useRealTimers(); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + it("should join the room named in the page URL over WebRTC", async () => { + vi.stubGlobal("location", { + search: "?room=ABC234", + origin: "https://game.test", + pathname: "/play", + }); + const lib = new NetworkClientLibrary(); + const init = lib.__init(makeInitContext()); + await new Promise((resolve) => setTimeout(resolve)); + await sockets[0].onmessage({ + data: JSON.stringify({ type: "welcome", id: "client-7", token: "t0k3n" }), + }); + await init; + + expect(sockets[0].url).toBe("ws://127.0.0.1:4445/?room=ABC234"); + expect(dataChannels.map((channel) => channel.label)).toStrictEqual([ + "reliable-ordered", + "reliable-unordered", + "unreliable-ordered", + "unreliable-unordered", + ]); + expect(lib.clientId).toBe("client-7"); + expect(lib.isConnected()).toBe(true); + expect(lib.roomCode).toBe("ABC234"); + expect(lib.inviteLink).toBe("https://game.test/play?room=ABC234"); + }); + + it("should host: run the server app in this page and join it through the loopback", async () => { + const info = vi.spyOn(console, "info").mockImplementation(() => {}); + const sendData = vi.fn(); + const main = moduleUrl(`export const main = async (args) => { + globalThis.__serverArgs = args; + globalThis[Symbol.for("nanoforge.network.host")] = { + room: "XYZ789", + attachLocal: () => ({ + id: "local-0", + transport: globalThis.__localTransport, + }), + }; + await new Promise(() => {}); + };`); + Reflect.set(globalThis, "__localTransport", { + sendData, + getReceivedPackets: () => [], + isConnected: () => true, + }); + const lib = new NetworkClientLibrary(); + await lib.__init( + makeInitContext([ + ["/server/main.js", main], + ["/server/libecs.wasm", "blob:wasm"], + ["/libecs.wasm", "blob:client-wasm"], + ]), + ); + + const args = Reflect.get(globalThis, "__serverArgs"); + expect(args.files).toStrictEqual( + new Map([ + ["/main.js", main], + ["/libecs.wasm", "blob:wasm"], + ]), + ); + expect(args.env).toBe(env); + expect(lib.clientId).toBe("local-0"); + expect(lib.isConnected()).toBe(true); + expect(lib.roomCode).toBe("XYZ789"); + expect(lib.inviteLink).toBe("https://game.test/play?room=XYZ789"); + expect(info).toHaveBeenCalledWith("Invite: https://game.test/play?room=XYZ789"); + + lib.packets.send(Play); + expect(sendData).toHaveBeenCalledWith("reliable-ordered", expect.any(Uint8Array)); + Reflect.deleteProperty(globalThis, "__localTransport"); + }); + + it("should fail to host without the server app built for the browser", async () => { + await expect(new NetworkClientLibrary().__init(makeInitContext())).rejects.toThrow( + "p2p needs the server app built for the browser", + ); + }); + + it("should fail to host when the server app fails to start", async () => { + const main = moduleUrl(`export const main = async () => { throw new Error("no wasm"); };`); + await expect( + new NetworkClientLibrary().__init(makeInitContext([["/server/main.js", main]])), + ).rejects.toThrow("the server app failed to start"); + }); + + it("should need SERVER_TCP_PORT to reach the signaling server", async () => { + const ctx = { + ...makeInitContext(), + env: { ...env, SERVER_TCP_PORT: undefined, SERVER_UDP_PORT: "4444" }, + }; + await expect(new NetworkClientLibrary().__init(ctx)).rejects.toThrow( + "p2p needs SERVER_TCP_PORT", + ); + }); +}); diff --git a/modules/network/test/client/udp.client.spec.ts b/modules/network/test/client/udp.client.spec.ts index 90d29b0..bea21cb 100644 --- a/modules/network/test/client/udp.client.spec.ts +++ b/modules/network/test/client/udp.client.spec.ts @@ -1,7 +1,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { UDPClient } from "../../src/client/udp.client.network"; -import { Channel, PacketSequencer } from "../../src/shared/channels"; +import { CHANNELS, Channel, PacketSequencer } from "../../src/shared/channels"; const welcome = { type: "welcome", id: "client-0", token: "t0k3n" }; @@ -108,6 +108,65 @@ describe("UDPClient", () => { ]); }); + it("should open the four channels of a peer-to-peer link in a room", async () => { + const client = new UDPClient(9000, "127.0.0.1", false, [], {}, CHANNELS, { room: "ABC234" }); + await connectWelcomed(client); + + expect(vi.mocked(WebSocket)).toHaveBeenCalledWith("ws://127.0.0.1:9000/?room=ABC234"); + expect(pc.createDataChannel.mock.calls).toStrictEqual([ + ["reliable-ordered", {}], + ["reliable-unordered", { ordered: false }], + ["unreliable-ordered", { ordered: false, maxRetransmits: 0 }], + ["unreliable-unordered", { ordered: false, maxRetransmits: 0 }], + ]); + + channels.get("reliable-ordered")!.readyState = "open"; + client.sendData(Channel.ReliableOrdered, new Uint8Array([1])); + expect(client.isConnected(Channel.ReliableOrdered)).toBe(true); + expect(channels.get("reliable-ordered")!.send).toHaveBeenCalledWith(new Uint8Array([1])); + }); + + it("should close when the signaling server says the host left", async () => { + const client = new UDPClient(9000, "127.0.0.1", false, [], {}, CHANNELS, { room: "ABC234" }); + await connectWelcomed(client); + channels.get("reliable-ordered")!.readyState = "open"; + pc.close = vi.fn(); + ws.close = vi.fn(); + + ws.onclose({ code: 1006 }); + expect(client.isConnected(Channel.ReliableOrdered)).toBe(true); + ws.onclose({ code: 4001 }); + expect(client.isConnected(Channel.ReliableOrdered)).toBe(false); + expect(pc.close).toHaveBeenCalled(); + }); + + it("should close when the peer connection fails, though its channels say open", async () => { + const client = new UDPClient(8081, "127.0.0.1", false); + await connectWelcomed(client); + channels.get("unreliable-ordered")!.readyState = "open"; + pc.close = vi.fn(); + ws.close = vi.fn(); + + pc.connectionState = "disconnected"; + pc.onconnectionstatechange(); + expect(client.isConnected(Channel.UnreliableOrdered)).toBe(true); + pc.connectionState = "failed"; + pc.onconnectionstatechange(); + expect(client.isConnected(Channel.UnreliableOrdered)).toBe(false); + }); + + it("should close the signaling socket and the peer connection", async () => { + const client = new UDPClient(8081, "127.0.0.1", false); + await connectWelcomed(client); + ws.close = vi.fn(); + pc.close = vi.fn(); + + client.close(); + expect(ws.close).toHaveBeenCalled(); + expect(pc.close).toHaveBeenCalled(); + expect(client.isConnected(Channel.UnreliableOrdered)).toBe(false); + }); + it("should report each channel connected once its data channel opens", async () => { const client = new UDPClient(8081, "127.0.0.1", false); await connectWelcomed(client); @@ -257,7 +316,7 @@ describe("UDPClient", () => { it("should reject when the signaling socket closes before the welcome", async () => { const client = new UDPClient(8081, "127.0.0.1", false); const connecting = client.connect(); - ws.onclose(); + ws.onclose({ code: 1006 }); await expect(connecting).rejects.toThrow("UDP signaling closed before the server welcome"); }); diff --git a/modules/network/test/server/client-registry.spec.ts b/modules/network/test/server/client-registry.spec.ts index 096f6d6..b8f693c 100644 --- a/modules/network/test/server/client-registry.spec.ts +++ b/modules/network/test/server/client-registry.spec.ts @@ -95,6 +95,17 @@ describe("ClientRegistry", () => { expect(onConnect).toHaveBeenCalledWith(registry.get(tcp.id)); }); + it("should register a new session before notifying its connection", () => { + const registry = new ClientRegistry(); + const order: string[] = []; + registry.onConnect(() => order.push("listener")); + + const session = registry.attach(null, connection("peer"), ({ id }) => + order.push(`register ${id}`), + ); + expect(order).toStrictEqual([`register ${session.id}`, "listener"]); + }); + it("should notify a disconnection only once the last transport closes", () => { const registry = new ClientRegistry(); const onDisconnect = vi.fn(); diff --git a/modules/network/test/server/peer-host.spec.ts b/modules/network/test/server/peer-host.spec.ts new file mode 100644 index 0000000..b371e79 --- /dev/null +++ b/modules/network/test/server/peer-host.spec.ts @@ -0,0 +1,360 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { ClientRegistry } from "../../src/server/client-registry"; +import { PeerHost } from "../../src/server/peer-host"; +import { CHANNELS, Channel, PacketSequencer } from "../../src/shared/channels"; + +/** Stand-in for a data channel a joiner opened, labelled `label`. */ +const makeDataChannel = (label: string) => ({ + label, + readyState: "connecting", + binaryType: "", + send: vi.fn(), + close: vi.fn(), + onopen: null as null | (() => void), + onmessage: null as null | ((event: { data: unknown }) => void), + onclose: null as null | (() => void), + onerror: null as null | ((event: unknown) => void), +}); + +describe("PeerHost", () => { + let socket: any; + let peers: any[]; + /** Resolves the pending `setRemoteDescription`, to hold the offer handler mid-await. */ + let applyRemote: () => void; + let registry: ClientRegistry; + + beforeEach(() => { + let nextId = 0; + vi.spyOn(crypto, "randomUUID").mockImplementation( + () => `client-${nextId++}` as ReturnType, + ); + peers = []; + vi.stubGlobal( + "WebSocket", + Object.assign( + vi.fn(function () { + return (socket = { + readyState: 1, + send: vi.fn(), + close: vi.fn(), + onerror: null, + onclose: null, + onmessage: null, + }); + }), + { OPEN: 1 }, + ), + ); + vi.stubGlobal( + "RTCPeerConnection", + vi.fn(function () { + const pc: any = { + remoteDescription: null, + localDescription: null, + connectionState: "new", + onicecandidate: null, + onconnectionstatechange: null, + ondatachannel: null, + setRemoteDescription: vi.fn( + (description: unknown) => + new Promise((resolve) => { + applyRemote = () => { + pc.remoteDescription = description; + resolve(); + }; + }), + ), + addIceCandidate: vi.fn().mockResolvedValue(undefined), + createAnswer: vi.fn().mockResolvedValue({ type: "answer", sdp: "v=0 answer" }), + setLocalDescription: vi.fn(async (description: unknown) => { + pc.localDescription = description; + }), + close: vi.fn(), + }; + peers.push(pc); + return pc; + }), + ); + registry = new ClientRegistry(); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + const fromServer = (message: unknown) => socket.onmessage({ data: JSON.stringify(message) }); + const fromJoiner = (from: string, data: unknown) => fromServer({ type: "signal", from, data }); + const sent = () => socket.send.mock.calls.map(([frame]: [string]) => JSON.parse(frame)); + const flush = () => new Promise((resolve) => setTimeout(resolve)); + + const openHost = async () => { + const host = new PeerHost(registry, [{ urls: "stun:stun.example.com" }]); + const opening = host.open("ws://127.0.0.1:9000"); + fromServer({ type: "room", room: "ABC234" }); + expect(await opening).toBe("ABC234"); + return host; + }; + + /** Run a joiner's offer through to its four open data channels. */ + const connectJoiner = async (host: PeerHost, from = "joiner-a") => { + fromJoiner(from, { type: "offer", offer: { type: "offer", sdp: "v=0" } }); + const pc = peers.at(-1); + applyRemote(); + await flush(); + const channels = CHANNELS.map((label) => makeDataChannel(label)); + for (const channel of channels) pc.ondatachannel({ channel }); + for (const channel of channels) { + channel.readyState = "open"; + channel.onopen?.(); + } + const welcome = sent().find((frame: any) => frame.data.type === "welcome"); + return { pc, channels, id: welcome?.data.id as string, host }; + }; + + describe("signaling", () => { + it("should resolve the room the signaling server gives", async () => { + await openHost(); + expect(WebSocket).toHaveBeenCalledWith("ws://127.0.0.1:9000"); + }); + + it("should reject when the signaling socket closes before a room", async () => { + const opening = new PeerHost(registry).open("ws://127.0.0.1:9000"); + socket.onclose(); + await expect(opening).rejects.toThrow("Signaling closed before giving a room"); + }); + + it("should answer a joiner's offer through the signaling server", async () => { + await openHost(); + fromJoiner("joiner-a", { type: "offer", offer: { type: "offer", sdp: "v=0" } }); + applyRemote(); + await flush(); + + expect(RTCPeerConnection).toHaveBeenCalledWith({ + iceServers: [{ urls: "stun:stun.example.com" }], + }); + expect(sent()).toStrictEqual([ + { + type: "signal", + to: "joiner-a", + data: { type: "answer", answer: { type: "answer", sdp: "v=0 answer" } }, + }, + ]); + }); + + it("should queue ICE candidates that arrive while the offer is being applied", async () => { + await openHost(); + fromJoiner("joiner-a", { type: "offer", offer: { type: "offer", sdp: "v=0" } }); + fromJoiner("joiner-a", { type: "ice", candidate: { candidate: "c1" } }); + const pc = peers[0]; + expect(peers).toHaveLength(1); + expect(pc.addIceCandidate).not.toHaveBeenCalled(); + + applyRemote(); + await flush(); + fromJoiner("joiner-a", { type: "ice", candidate: { candidate: "c2" } }); + expect(pc.addIceCandidate.mock.calls).toStrictEqual([ + [{ candidate: "c1" }], + [{ candidate: "c2" }], + ]); + }); + + it("should trickle the host's ICE candidates to the joiner", async () => { + await openHost(); + fromJoiner("joiner-a", { type: "offer", offer: { type: "offer", sdp: "v=0" } }); + peers[0].onicecandidate({ candidate: { candidate: "h1" } }); + + expect(sent()).toStrictEqual([ + { type: "signal", to: "joiner-a", data: { type: "ice", candidate: { candidate: "h1" } } }, + ]); + }); + + it("should not open a peer connection before a joiner's offer", async () => { + await openHost(); + fromJoiner("joiner-a", {}); + fromJoiner("joiner-a", { type: "ice", candidate: { candidate: "c1" } }); + expect(peers).toHaveLength(0); + }); + + it("should drop a joiner rather than send a frame the server would refuse", async () => { + await openHost(); + const error = vi.spyOn(console, "error").mockImplementation(() => {}); + fromJoiner("joiner-a", { type: "offer", offer: { type: "offer", sdp: "v=0" } }); + peers[0].createAnswer.mockResolvedValue({ type: "answer", sdp: "a".repeat(64 * 1024) }); + applyRemote(); + await flush(); + + expect(socket.send).not.toHaveBeenCalled(); + expect(peers[0].close).toHaveBeenCalled(); + expect(error).toHaveBeenCalledWith(expect.stringContaining("too large")); + }); + + it("should forget its room when the signaling socket closes", async () => { + const host = await openHost(); + expect(host.room).toBe("ABC234"); + socket.onclose(); + expect(host.room).toBeUndefined(); + }); + + it("should ignore frames that are not objects", async () => { + await openHost(); + fromJoiner("joiner-a", "offer"); + socket.onmessage({ data: "{not json" }); + expect(peers).toHaveLength(0); + }); + }); + + describe("joiners", () => { + it("should welcome a joiner, and connect it, once its four channels are open", async () => { + const host = await openHost(); + const connected = vi.fn(); + registry.onConnect(connected); + fromJoiner("joiner-a", { type: "offer", offer: { type: "offer", sdp: "v=0" } }); + applyRemote(); + await flush(); + const channels = CHANNELS.map((label) => makeDataChannel(label)); + for (const channel of channels) peers[0].ondatachannel({ channel }); + for (const channel of channels.slice(0, 3)) { + channel.readyState = "open"; + channel.onopen?.(); + } + expect(connected).not.toHaveBeenCalled(); + expect(host.getConnectedClients(Channel.ReliableOrdered)).toStrictEqual([]); + + channels[3]!.readyState = "open"; + channels[3]!.onopen?.(); + expect(sent().at(-1)).toStrictEqual({ + type: "signal", + to: "joiner-a", + data: { type: "welcome", id: "client-0", token: expect.any(String) }, + }); + expect(connected).toHaveBeenCalledWith( + expect.objectContaining({ id: "client-0", transports: { peer: expect.anything() } }), + ); + expect(host.getConnectedClients(Channel.ReliableOrdered)).toStrictEqual(["client-0"]); + expect(channels.map((channel) => channel.binaryType)).toStrictEqual( + CHANNELS.map(() => "arraybuffer"), + ); + }); + + it("should let the connect listeners send to the new joiner", async () => { + const host = await openHost(); + registry.onConnect(({ id }) => + host.sendToClient(Channel.ReliableOrdered, id, new Uint8Array([9])), + ); + const { channels } = await connectJoiner(host); + + expect(channels[0]!.send).toHaveBeenCalledWith(new Uint8Array([9])); + }); + + it("should close a data channel with an unknown label", async () => { + await openHost(); + fromJoiner("joiner-a", { type: "offer", offer: { type: "offer", sdp: "v=0" } }); + vi.spyOn(console, "error").mockImplementation(() => {}); + const chat = makeDataChannel("chat"); + peers[0].ondatachannel({ channel: chat }); + expect(chat.close).toHaveBeenCalled(); + }); + + it("should send and receive on each channel, sequencing UnreliableOrdered", async () => { + const host = await openHost(); + const { channels, id } = await connectJoiner(host); + const [reliable, , ordered] = channels; + + host.sendToClient(Channel.ReliableOrdered, id, new Uint8Array([1])); + host.sendToEverybody(Channel.UnreliableOrdered, new Uint8Array([2])); + expect(reliable!.send).toHaveBeenCalledWith(new Uint8Array([1])); + expect(ordered!.send).toHaveBeenCalledWith(new PacketSequencer().wrap(new Uint8Array([2]))); + + const sequencer = new PacketSequencer(); + const first = sequencer.wrap(new Uint8Array([7])); + const second = sequencer.wrap(new Uint8Array([8])); + reliable!.onmessage?.({ data: new Uint8Array([5]).buffer }); + ordered!.onmessage?.({ data: second.buffer }); + ordered!.onmessage?.({ data: first.buffer }); + ordered!.onmessage?.({ data: "text" }); + + expect(host.getReceivedPackets(Channel.ReliableOrdered).get(id)).toStrictEqual([ + new Uint8Array([5]), + ]); + expect(host.getReceivedPackets(Channel.UnreliableOrdered).get(id)).toStrictEqual([ + new Uint8Array([8]), + ]); + expect(host.getReceivedPackets(Channel.UnreliableOrdered).get(id)).toStrictEqual([]); + }); + + it("should disconnect a joiner when one of its channels closes", async () => { + const host = await openHost(); + const disconnected = vi.fn(); + registry.onDisconnect(disconnected); + const { pc, channels, id } = await connectJoiner(host); + + channels[1]!.onclose?.(); + expect(pc.close).toHaveBeenCalled(); + expect(disconnected).toHaveBeenCalledWith(expect.objectContaining({ id })); + expect(host.getConnectedClients(Channel.ReliableOrdered)).toStrictEqual([]); + }); + + it("should disconnect a joiner whose connection failed", async () => { + const host = await openHost(); + const { pc, id } = await connectJoiner(host); + + pc.connectionState = "failed"; + pc.onconnectionstatechange(); + expect(registry.get(id)).toBeUndefined(); + }); + + it("should disconnect a joiner whose signaling socket closed, welcomed or not", async () => { + const host = await openHost(); + const disconnected = vi.fn(); + registry.onDisconnect(disconnected); + const { id } = await connectJoiner(host, "joiner-a"); + fromJoiner("joiner-b", { type: "offer", offer: { type: "offer", sdp: "v=0" } }); + + fromServer({ type: "leave", from: "joiner-a" }); + fromServer({ type: "leave", from: "joiner-b" }); + expect(peers[0].close).toHaveBeenCalled(); + expect(peers[1].close).toHaveBeenCalled(); + expect(disconnected.mock.calls).toStrictEqual([[expect.objectContaining({ id })]]); + expect(host.getConnectedClients(Channel.ReliableOrdered)).toStrictEqual([]); + }); + + it("should close every connection without notifying", async () => { + const host = await openHost(); + const disconnected = vi.fn(); + registry.onDisconnect(disconnected); + const { pc } = await connectJoiner(host); + + host.close(); + expect(socket.close).toHaveBeenCalled(); + expect(pc.close).toHaveBeenCalled(); + expect(disconnected).not.toHaveBeenCalled(); + }); + }); + + describe("local player", () => { + it("should loop the host's own player back through in-memory queues", async () => { + const host = await openHost(); + const { id, transport } = host.attachLocal(); + const joiner = await connectJoiner(host); + + expect(registry.get(id)?.transports).toStrictEqual({ local: expect.anything() }); + expect(transport.isConnected(Channel.UnreliableOrdered)).toBe(true); + expect(host.getConnectedClients(Channel.ReliableOrdered)).toStrictEqual([id, joiner.id]); + + host.sendToEverybody(Channel.ReliableOrdered, new Uint8Array([1])); + host.sendToClient(Channel.ReliableOrdered, id, new Uint8Array([2])); + expect(transport.getReceivedPackets(Channel.ReliableOrdered)).toStrictEqual([ + new Uint8Array([1]), + new Uint8Array([2]), + ]); + expect(transport.getReceivedPackets(Channel.ReliableOrdered)).toStrictEqual([]); + + transport.sendData(Channel.UnreliableUnordered, new Uint8Array([3])); + expect(host.getReceivedPackets(Channel.UnreliableUnordered).get(id)).toStrictEqual([ + new Uint8Array([3]), + ]); + }); + }); +}); diff --git a/modules/network/test/server/server.network.browser.spec.ts b/modules/network/test/server/server.network.browser.spec.ts new file mode 100644 index 0000000..5d8b13b --- /dev/null +++ b/modules/network/test/server/server.network.browser.spec.ts @@ -0,0 +1,82 @@ +import type { Context, InitContext } from "@nanoforge-dev/common"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { NetworkServerLibrary } from "../../src/server/browser"; +import { PeerHost } from "../../src/server/peer-host"; +import { Channel } from "../../src/shared/channels"; +import { PEER_HOST_KEY } from "../../src/shared/signaling"; + +const makeInitContext = (env: Record): InitContext => ({ + vars: { get: () => undefined, set: () => {} }, + env, + files: new Map(), +}); + +const env = { SERVER_ADDRESS: "127.0.0.1", SERVER_TCP_PORT: "4445", NETWORK_MODE: "p2p" }; + +describe("NetworkServerLibrary (browser)", () => { + let socket: any; + + beforeEach(() => { + socket = undefined; + vi.stubGlobal( + "WebSocket", + Object.assign( + vi.fn(function (this: any, url: string) { + socket = Object.assign(this, { url, readyState: 1, send: vi.fn(), close: vi.fn() }); + }), + { OPEN: 1 }, + ), + ); + }); + + afterEach(() => { + Reflect.deleteProperty(globalThis, PEER_HOST_KEY); + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + const init = async () => { + const lib = new NetworkServerLibrary(); + const initializing = lib.__init(makeInitContext(env)); + await vi.waitFor(() => expect(socket).toBeDefined()); + socket.onmessage({ data: JSON.stringify({ type: "room", room: "ABC234" }) }); + await initializing; + return lib; + }; + + it("should open a room and publish its host for the page's client", async () => { + const lib = await init(); + const host = Reflect.get(globalThis, PEER_HOST_KEY); + + expect(lib.key).toBe("network"); + expect(socket.url).toBe("ws://127.0.0.1:4445"); + expect(host).toBeInstanceOf(PeerHost); + expect(host.room).toBe("ABC234"); + }); + + it("should serve its own player through the loopback on every channel", async () => { + const lib = await init(); + const { id, transport } = Reflect.get(globalThis, PEER_HOST_KEY).attachLocal(); + + lib.sendToClient(id, "hi", { channel: Channel.UnreliableOrdered }); + expect(transport.getReceivedPackets(Channel.UnreliableOrdered)).toHaveLength(1); + expect(lib.getConnectedClients()).toStrictEqual([id]); + }); + + it("should close the room on __clear", async () => { + const lib = await init(); + await lib.__clear({} as Context); + + expect(socket.close).toHaveBeenCalled(); + expect(Reflect.get(globalThis, PEER_HOST_KEY)).toBeUndefined(); + expect(() => lib.sendToAll("hi")).toThrow("is not open"); + }); + + it("should need SERVER_TCP_PORT to reach the signaling server", async () => { + const ctx = makeInitContext({ SERVER_ADDRESS: "127.0.0.1", SERVER_UDP_PORT: "4444" }); + await expect(new NetworkServerLibrary().__init(ctx)).rejects.toThrow( + "p2p needs SERVER_TCP_PORT", + ); + }); +}); diff --git a/modules/network/test/server/server.network.library.spec.ts b/modules/network/test/server/server.network.library.spec.ts index 3d5e5ef..e474a41 100644 --- a/modules/network/test/server/server.network.library.spec.ts +++ b/modules/network/test/server/server.network.library.spec.ts @@ -136,6 +136,29 @@ describe("NetworkServerLibrary", () => { ).not.toThrow(); }); + it("should only start a signaling server on LISTENING_TCP_PORT in p2p mode", async () => { + const stop = vi.fn(); + serve.mockImplementation(() => ({ stop })); + const lib = new NetworkServerLibrary(); + await lib.__init(makeInitContext({ LISTENING_TCP_PORT: "4445", NETWORK_MODE: "p2p" })); + + expect(serve).toHaveBeenCalledTimes(1); + expect(serve.mock.calls[0]?.[0].port).toBe(4445); + expect(serve.mock.calls[0]?.[0].websocket.maxPayloadLength).toBe(64 * 1024); + expect(() => lib.sendToAll("hi", { channel: Channel.UnreliableUnordered })).not.toThrow(); + expect(lib.getConnectedClients()).toStrictEqual([]); + expect(lib.getReceivedPackets()).toStrictEqual(new Map()); + await lib.__clear({} as never); + expect(stop).toHaveBeenCalledWith(true); + }); + + it("should need LISTENING_TCP_PORT in p2p mode", async () => { + const ctx = makeInitContext({ LISTENING_UDP_PORT: "4444", NETWORK_MODE: "p2p" }); + await expect(new NetworkServerLibrary().__init(ctx)).rejects.toThrow( + "p2p needs LISTENING_TCP_PORT", + ); + }); + it("should default LISTENING_INTERFACE to 0.0.0.0 when not provided", async () => { const ctx = makeInitContext({ LISTENING_TCP_PORT: "9000" }); const lib = new NetworkServerLibrary(); diff --git a/modules/network/test/server/signaling.server.spec.ts b/modules/network/test/server/signaling.server.spec.ts new file mode 100644 index 0000000..7ed3c3a --- /dev/null +++ b/modules/network/test/server/signaling.server.spec.ts @@ -0,0 +1,139 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { SignalingServer } from "../../src/server/signaling.server.network"; + +const serve = vi.fn(); +const file = vi.fn((path: string) => ({ path })); + +beforeEach(() => { + let nextId = 0; + vi.spyOn(crypto, "randomUUID").mockImplementation( + () => `joiner-${nextId++}` as ReturnType, + ); + serve.mockImplementation(() => ({ stop: vi.fn() })); + vi.stubGlobal("Bun", { serve, file }); +}); + +afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + vi.clearAllMocks(); +}); + +const listen = () => { + const server = new SignalingServer(9000, "0.0.0.0"); + server.listen(); + return { server, options: serve.mock.calls[0]?.[0] }; +}; + +/** Upgrade and open a socket the way Bun does; `response` is set when the upgrade is refused. */ +const connect = (options: any, url = "http://localhost:9000/") => { + let data: any; + const response = options.fetch(new Request(url), { + upgrade: (_request: unknown, opts: { data: unknown }) => { + data = opts.data; + return true; + }, + }); + const webSocket = { data, send: vi.fn(), close: vi.fn() }; + if (data) options.websocket.open(webSocket); + return { webSocket, response }; +}; + +const sent = (webSocket: { send: ReturnType }) => + webSocket.send.mock.calls.map(([frame]) => JSON.parse(frame)); + +/** A host with its room code, and a joiner of that room. */ +const pair = () => { + const { options } = listen(); + const host = connect(options).webSocket; + const [{ room }] = sent(host); + host.send.mockClear(); + const joiner = connect(options, `http://localhost:9000/?room=${room}`).webSocket; + return { options, host, joiner, room }; +}; + +describe("SignalingServer", () => { + it("should give a host a 6-character room code", () => { + const { options } = listen(); + const { webSocket } = connect(options); + + expect(sent(webSocket)).toStrictEqual([{ type: "room", room: expect.any(String) }]); + expect(sent(webSocket)[0].room).toMatch(/^[A-HJ-NP-Z2-9]{6}$/); + }); + + it("should cap signaling frames", () => { + const { options } = listen(); + expect(options.websocket.maxPayloadLength).toBe(64 * 1024); + }); + + it("should refuse a joiner of an unknown room", async () => { + const { options } = listen(); + const { response } = connect(options, "http://localhost:9000/?room=NOPE42"); + + expect(response.status).toBe(404); + }); + + it("should forward a joiner's frames to its host, with the joiner's id", () => { + const { options, host, joiner } = pair(); + + options.websocket.message(joiner, JSON.stringify({ type: "offer", offer: { sdp: "v=0" } })); + expect(sent(host)).toStrictEqual([ + { type: "signal", from: "joiner-0", data: { type: "offer", offer: { sdp: "v=0" } } }, + ]); + }); + + it("should forward the host's frames to the joiner they are addressed to", () => { + const { options, host, joiner } = pair(); + const welcome = { type: "welcome", id: "p1", token: "t" }; + + options.websocket.message( + host, + JSON.stringify({ type: "signal", to: "joiner-0", data: welcome }), + ); + expect(sent(joiner)).toStrictEqual([welcome]); + }); + + it("should not let a host reach a joiner of another room", () => { + const { options, joiner } = pair(); + const otherHost = connect(options).webSocket; + + options.websocket.message( + otherHost, + JSON.stringify({ type: "signal", to: "joiner-0", data: { type: "answer" } }), + ); + expect(joiner.send).not.toHaveBeenCalled(); + }); + + it("should drop frames that are not JSON objects", () => { + const { options, host, joiner } = pair(); + + options.websocket.message(joiner, "{not json"); + options.websocket.message(joiner, "42"); + options.websocket.message(host, "{not json"); + options.websocket.message(host, JSON.stringify({ type: "signal", to: "joiner-0" })); + expect(host.send).not.toHaveBeenCalled(); + expect(joiner.send).not.toHaveBeenCalled(); + }); + + it("should tell the host when a joiner leaves", () => { + const { options, host, joiner } = pair(); + + options.websocket.close(joiner); + expect(sent(host)).toStrictEqual([{ type: "leave", from: "joiner-0" }]); + }); + + it("should close the room and its joiners when the host leaves", () => { + const { options, host, joiner, room } = pair(); + + options.websocket.close(host); + expect(joiner.close).toHaveBeenCalledWith(4001, "Host left"); + expect(connect(options, `http://localhost:9000/?room=${room}`).response.status).toBe(404); + }); + + it("should stop Bun's server", () => { + const { server } = listen(); + server.stop(); + expect(serve.mock.results[0]?.value.stop).toHaveBeenCalledWith(true); + }); +}); diff --git a/modules/network/test/shared/channels.spec.ts b/modules/network/test/shared/channels.spec.ts index 4353df4..912ae0e 100644 --- a/modules/network/test/shared/channels.spec.ts +++ b/modules/network/test/shared/channels.spec.ts @@ -1,8 +1,12 @@ import { describe, expect, it } from "vitest"; import { + CHANNELS, Channel, + DATA_CHANNEL_OPTIONS, PacketSequencer, + UNRELIABLE_CHANNEL_OPTIONS, + channelFromLabel, decodeReliableFrame, encodeReliableFrame, unreliableChannelFromLabel, @@ -56,6 +60,28 @@ describe("unreliableChannelFromLabel", () => { }); }); +describe("channelFromLabel", () => { + it("should map the label of every channel", () => { + for (const channel of CHANNELS) expect(channelFromLabel(channel)).toBe(channel); + }); + + it("should reject unknown labels", () => { + expect(channelFromLabel("chat")).toBeUndefined(); + }); +}); + +describe("DATA_CHANNEL_OPTIONS", () => { + it("should keep the unreliable options of a dedicated server", () => { + expect(DATA_CHANNEL_OPTIONS[Channel.UnreliableOrdered]).toBe(UNRELIABLE_CHANNEL_OPTIONS); + expect(DATA_CHANNEL_OPTIONS[Channel.UnreliableUnordered]).toBe(UNRELIABLE_CHANNEL_OPTIONS); + }); + + it("should retransmit the reliable channels", () => { + expect(DATA_CHANNEL_OPTIONS[Channel.ReliableOrdered]).toStrictEqual({}); + expect(DATA_CHANNEL_OPTIONS[Channel.ReliableUnordered]).toStrictEqual({ ordered: false }); + }); +}); + describe("PacketSequencer", () => { it("should number outgoing packets from zero", () => { const sequencer = new PacketSequencer(); diff --git a/modules/network/tsdown.config.ts b/modules/network/tsdown.config.ts index ce475f9..24ba71f 100644 --- a/modules/network/tsdown.config.ts +++ b/modules/network/tsdown.config.ts @@ -12,4 +12,9 @@ export default [ outDir: "dist/server", tsconfig: "tsconfig.server.json", }), + createTsdownConfig({ + entry: "src/server/browser.ts", + outDir: "dist/server-browser", + tsconfig: "tsconfig.server.json", + }), ]; diff --git a/packages/core/src/application/nanoforge-application.ts b/packages/core/src/application/nanoforge-application.ts index f66fa69..a6f7b6c 100644 --- a/packages/core/src/application/nanoforge-application.ts +++ b/packages/core/src/application/nanoforge-application.ts @@ -117,11 +117,21 @@ export abstract class NanoforgeApplication { const tickMs = performance.now() - tickTimer; for (const observer of observers) observer.onTick?.(tickMs, Date.now()); - setTimeout(loop, tickLengthMs + tickStart - Date.now()); + this.schedule(loop, tickLengthMs + tickStart - Date.now()); }; this.appState.setIsRunning(true); - setTimeout(loop); + this.schedule(loop, 0); + } + + /** + * Call `tick` once `delayMs` elapsed: the next tick of the game loop. + * + * @param tick - Next tick of the game loop. + * @param delayMs - Delay before it, in milliseconds. + */ + protected schedule(tick: () => void, delayMs: number): void { + setTimeout(tick, delayMs); } protected async initialize(options: RunOptions | ClientRunOptions): Promise { diff --git a/packages/core/src/application/nanoforge-server.ts b/packages/core/src/application/nanoforge-server.ts index a7c996f..fe378be 100644 --- a/packages/core/src/application/nanoforge-server.ts +++ b/packages/core/src/application/nanoforge-server.ts @@ -19,6 +19,8 @@ import { NanoforgeApplication } from "./nanoforge-application"; * ``` */ export class NanoforgeServer extends NanoforgeApplication { + private _clock?: Worker; + /** * Initialise all registered libraries in dependency order and prepare the * engine for the game loop. @@ -33,4 +35,22 @@ export class NanoforgeServer extends NanoforgeApplication { public async init(options: ServerRunOptions): Promise { await this.initialize(options); } + + /** + * In a browser (a peer-to-peer host), ticks are timed by a worker: the + * timers of a hidden tab are throttled, which would slow the game down for + * every player. + */ + protected override schedule(tick: () => void, delayMs: number): void { + if (typeof window === "undefined" || typeof Worker === "undefined") { + super.schedule(tick, delayMs); + return; + } + const source = "onmessage = (event) => setTimeout(() => postMessage(0), event.data);"; + this._clock ??= new Worker( + URL.createObjectURL(new Blob([source], { type: "text/javascript" })), + ); + this._clock.onmessage = () => tick(); + this._clock.postMessage(Math.max(0, delayMs)); + } } diff --git a/packages/core/test/nanoforge-server.spec.ts b/packages/core/test/nanoforge-server.spec.ts new file mode 100644 index 0000000..260150a --- /dev/null +++ b/packages/core/test/nanoforge-server.spec.ts @@ -0,0 +1,48 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { NanoforgeServer } from "../src/application/nanoforge-server"; + +const options = { files: new Map(), env: {} }; + +describe("NanoforgeServer tick clock", () => { + afterEach(() => { + vi.unstubAllGlobals(); + vi.restoreAllMocks(); + }); + + it("should time its ticks with a worker in a browser, where hidden tabs throttle timers", async () => { + const workers: { onmessage: (() => void) | null; postMessage: ReturnType }[] = []; + vi.stubGlobal("window", {}); + vi.stubGlobal( + "Worker", + vi.fn(function (this: (typeof workers)[number]) { + this.onmessage = null; + this.postMessage = vi.fn(); + workers.push(this); + }), + ); + vi.stubGlobal("URL", { createObjectURL: vi.fn(() => "blob:clock") }); + const server = new NanoforgeServer({ tickRate: 50 }); + await server.init(options); + + await server.run(); + expect(workers).toHaveLength(1); + expect(workers[0]!.postMessage).toHaveBeenCalledWith(0); + + workers[0]!.onmessage?.(); + await vi.waitFor(() => expect(workers[0]!.postMessage).toHaveBeenCalledTimes(2)); + expect(workers[0]!.postMessage.mock.calls[1]![0]).toBeGreaterThan(0); + expect(workers).toHaveLength(1); + }); + + it("should use plain timers outside a browser", async () => { + const timeout = vi.spyOn(globalThis, "setTimeout"); + vi.stubGlobal("Worker", vi.fn()); + const server = new NanoforgeServer({ tickRate: 50 }); + await server.init(options); + + await server.run(); + expect(timeout).toHaveBeenCalled(); + expect(Worker).not.toHaveBeenCalled(); + }); +});