1/**
2 * helpers for building applications with an HTTP server, with integrations with
3 * `@clo/lib/progress`
4 *
5 * @module
6 */
7
8export interface Server {
9 port: number;
10 url: string;
11 close(): Promise<void>;
12 [Symbol.asyncDispose](): Promise<void>;
13}
14
15export interface ServeOptions {
16 /** respond to a server request */
17 respond: (request: Request, opts: ServeContext) => Promise<Response>;
18 /**
19 * take over the connection of a request to switch protocols, like to a
20 * websocket. without it, such a request is responded to like any other.
21 */
22 upgrade?: (request: Request, socket: stream.Duplex, opts: ServeContext) => void;
23 /**
24 * which requests to switch protocols go to `upgrade`, given without their
25 * bodies. the rest are responded to like any other. defaults to all of them.
26 */
27 shouldUpgrade?: (request: Request) => boolean;
28 /** port 0 hosts on a random port */
29 port: number;
30
31 progress?: progress.Ref;
32}
33
34export interface ServeContext {
35 node: progress.Node;
36 /** the other end of the connection, which is a proxy's for a request through one */
37 remoteAddress: string | undefined;
38}
39
40/**
41 * this server is not optimized performance and likely includes bugs.
42 *
43 * more production ready alternatives: `srvx` and `@hono/node-server`.
44 */
45export async function serve(
46 { respond, upgrade, shouldUpgrade = () => true, port, progress: rootNode = progress.nullNode }: ServeOptions,
47): Promise<Server> {
48 const server = nodeHttp.createServer({
49 shouldUpgradeCallback: (req) => upgrade !== undefined && shouldUpgrade(toRequest(req, false)),
50 }, (req, res) => {
51 const request = toRequest(req);
52 const requestNode = rootNode.start(`${request.method} ${req.url}`);
53 respond(request, { node: requestNode, remoteAddress: req.socket.remoteAddress })
54 .then((response) => {
55 // the client went away while the response was made
56 if (res.destroyed) return void response.body?.cancel().catch(() => {});
57 res.writeHead(
58 response.status,
59 response.statusText,
60 Array.from(response.headers.entries()),
61 );
62 if (response.body) {
63 // unlike `pipe`, this cancels the body when the client goes away,
64 // and a failing body closes the connection instead of the process.
65 stream.pipeline(
66 stream.Readable.fromWeb(response.body as import("node:stream/web").ReadableStream),
67 res,
68 (err) => {
69 if (err && exception.code(err) !== "ERR_STREAM_PREMATURE_CLOSE") console.error(err);
70 },
71 );
72 } else {
73 res.end();
74 }
75 })
76 .catch((err) => {
77 console.error(err);
78 })
79 .finally(() => {
80 requestNode.end();
81 });
82 });
83
84 const upgraded = new Set<stream.Duplex>();
85 if (upgrade) {
86 server.on("upgrade", (req: nodeHttp.IncomingMessage, socket: stream.Duplex, head: Buffer) => {
87 // what the client sent after the request is already in the new protocol
88 if (head.length > 0) socket.unshift(head);
89 const request = toRequest(req);
90 const requestNode = rootNode.start(`${request.method} ${req.url}`);
91 upgraded.add(socket);
92 // a connection that resets errors, which would end the process with no listener
93 socket.on("error", () => {});
94 socket.once("close", () => {
95 upgraded.delete(socket);
96 requestNode.end();
97 });
98 try {
99 upgrade(request, socket, { node: requestNode, remoteAddress: req.socket.remoteAddress });
100 } catch (err) {
101 console.error(err);
102 socket.destroy();
103 }
104 });
105 }
106
107 const actualPort = await new Promise<number>((resolve, reject) => {
108 async function onError(err: unknown) {
109 if (exception.code(err) !== "EADDRINUSE") {
110 reject(err);
111 return;
112 }
113 const info = await subprocess.lookupProcessOnPort(port).catch(() => null);
114 if (!info) return void reject(err);
115
116 reject(
117 new Error(
118 `Port ${port} is in use by ${JSON.stringify(info.command)} (pid ${info.pid})`,
119 ),
120 );
121 }
122 server.on("error", onError);
123 server.listen(port, () => {
124 server.off("error", onError);
125 const addr = server.address();
126 ASSERT(addr && typeof addr === "object");
127 resolve(addr.port);
128 });
129 });
130
131 function close() {
132 return new Promise<void>((resolve, reject) => {
133 server.close((err) => {
134 if (err) reject(err);
135 else resolve();
136 });
137 // an upgraded connection has no response to finish, so it would hold the server open
138 for (const socket of upgraded) socket.destroy();
139 });
140 }
141
142 return {
143 port: actualPort,
144 url: `http://localhost${actualPort === 80 ? "" : ":" + actualPort}/`,
145 close,
146 [Symbol.asyncDispose]: close,
147 };
148}
149
150function toRequest(req: nodeHttp.IncomingMessage, body = true): Request {
151 const headers = new Headers();
152 for (const key in req.headers) {
153 const value = req.headers[key];
154 if (value == null) continue;
155 if (Array.isArray(value)) {
156 for (const item of value) headers.append(key, item);
157 } else {
158 headers.append(key, value);
159 }
160 }
161 const method = UNWRAP(req.method).toUpperCase();
162 const hasNoBody = !body || method === "GET" || method === "HEAD";
163 return new Request(
164 "http://" + (req.headers.host ?? "localhost") + UNWRAP(req.url),
165 {
166 method,
167 headers,
168 body: hasNoBody
169 ? undefined
170 : stream.Readable.toWeb(req) as ReadableStream,
171 // @ts-expect-error
172 duplex: hasNoBody ? undefined : "half",
173 },
174 );
175}
176
177/**
178 * parses the [`Range`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Guides/Range_requests)
179 * header for a resource of `fileSize` bytes. `null` means the header is
180 * malformed and should be ignored. otherwise, the satisfiable ranges are
181 * returned clamped to the resource, so an empty array is a `416` response.
182 */
183export function parseRangeHeader(
184 rangeHeader: string,
185 fileSize: number,
186): Range[] | null {
187 const [unit, ranges] = rangeHeader.split("=");
188 if (!ranges) return null;
189 if (unit !== "bytes") return null;
190
191 const result: Range[] = [];
192 const rangeParts = ranges.split(",");
193
194 for (const range of rangeParts) {
195 const match = /^\s*(\d*)-(\d*)\s*$/.exec(range);
196 if (!match) return null;
197 const [, first = "", last = ""] = match;
198 if (first === "" && last === "") return null;
199 if (first !== "" && last !== "" && +last < +first) return null;
200
201 // `-500` is the last 500 bytes, `500-` is everything after the first 500.
202 const start = first === "" ? Math.max(0, fileSize - +last) : +first;
203 const end = first === "" || last === "" ? fileSize - 1 : Math.min(+last, fileSize - 1);
204 if (start <= end) result.push({ start, end });
205 }
206
207 // validate that ranges do not intersect
208 result.sort((a, b) => a.start - b.start);
209 for (let i = 1; i < result.length; i++) {
210 if (UNWRAP(result[i]).start <= UNWRAP(result[i - 1]).end) {
211 return null;
212 }
213 }
214
215 return result;
216}
217
218export interface Range {
219 /** inclusive */
220 start: number;
221 /** inclusive */
222 end: number;
223}
224
225import * as nodeHttp from "node:http";
226import * as stream from "node:stream";
227import { ASSERT, UNWRAP } from "./assert.ts";
228import * as exception from "./exception.ts";
229import * as node from "./node.ts";
230import * as progress from "./progress.ts";
231import * as subprocess from "./subprocess.ts";