char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somgenerate std typedefs from typescript impl38d7f6c

main
branchesmainall refs →
12.6 KiB432 linesraw
1import {
2  NumberIsFinite,
3  ObjectCreate,
4  ObjectFreeze,
5  RangeError,
6  ReflectApply as apply,
7  Set,
8  SetPrototypeAdd as add,
9  SetPrototypeHas as has,
10  TypedArrayPrototypeGetTag as tag,
11  TypeError,
12  WeakMap,
13  WeakMapPrototypeGet as get,
14  WeakMapPrototypeSet as set,
15} from "ext:core/primordials";
16import { reportNonfatalException } from "ext:core/event_hooks";
17import {
18  markPromiseHandled,
19  promiseThen as then,
20  safePromise,
21} from "ext:core/promises";
22import {
23  op_http_cancel,
24  op_http_cli_listen,
25  op_http_finished,
26  op_http_listen,
27  op_http_next,
28  op_http_read,
29  op_http_respond,
30  op_http_respond_file,
31  op_http_stop,
32  op_http_upgrade,
33  op_http_wait,
34  op_http_write,
35} from "ext:core/ops";
36import { dictionary, state, usv } from "ext:web/idl.js";
37import {
38  abort,
39  AbortSignal,
40  addListener,
41  DOMException,
42  removeListener,
43  secret,
44  signals,
45} from "ext:web/events.js";
46import {
47  enqueue,
48  errorStream,
49  finish,
50  read,
51  ReadableStream,
52  readerCancel,
53  release,
54  streamCancel,
55  streamReader,
56} from "ext:webstreams/streams.js";
57import { bodies, messageState, usable } from "ext:http_types/body.js";
58import { fromPairs, headerEntries, headers } from "ext:http_types/headers.js";
59import { Request } from "ext:http_types/request.js";
60import { Response } from "ext:http_types/response.js";
61import { adoptWebSocket } from "ext:websocket/websocket.js";
62import { takeFile } from "./file.js";
63
64import type {
65  HttpAddress,
66  HttpListener,
67  HttpRequestMetadata,
68} from "ext:core/ops";
69import type {
70  ReadableByteStreamController,
71  ReadableStreamDefaultReader,
72} from "ext:webstreams/streams.js";
73import type { WebSocket } from "ext:websocket/websocket.js";
74
75import type { ServeOptions, UnixServeOptions } from "avril:std";
76
77type StreamAddr = Readonly<HttpAddress>;
78type TcpAddr = Extract<StreamAddr, { transport: "tcp" }>;
79type UnixAddr = Extract<StreamAddr, { transport: "unix" }>;
80interface ServeInfo<A extends StreamAddr = StreamAddr> {
81  readonly remoteAddr: A;
82}
83type ServeHandler<A extends StreamAddr = StreamAddr> = (request: Request, info: ServeInfo<A>) => unknown;
84type ErrorHandler = (error: unknown) => unknown;
85interface HttpServer<A extends StreamAddr = StreamAddr> {
86  readonly addr: A;
87  readonly finished: Promise<void>;
88  readonly shutdown: () => Promise<void>;
89}
90
91const requests = new WeakMap<object, number>();
92
93function address(value: HttpAddress): StreamAddr {
94  return ObjectFreeze(
95    value.transport === "unix"
96      ? { __proto__: null, transport: "unix", path: value.path }
97      : {
98        __proto__: null,
99        transport: "tcp",
100        hostname: value.hostname,
101        port: value.port,
102      },
103  );
104}
105
106function upgradeWebSocket(
107  request: unknown,
108  rawOptions: unknown = {},
109): { socket: WebSocket; response: Response } {
110  const id = get(requests, request);
111  if (id === undefined) {
112    throw new TypeError("Expected an incoming HTTP request");
113  }
114  const requestState = messageState(request, "request");
115  usable(requestState.body);
116  const options = dictionary(rawOptions);
117  const protocol = options.protocol === undefined
118    ? null
119    : usv(options.protocol);
120  const bound = op_http_upgrade([id, protocol]);
121  const socket = adoptWebSocket(bound[0], requestState.url);
122  const response = ObjectCreate(Response.prototype) as Response,
123    h = fromPairs(bound[1]);
124  state(headers, h).immutable = true;
125  set(bodies, response, {
126    kind: "response",
127    body: { stream: null, bytes: null, used: false, type: null },
128    headers: h,
129    status: 101,
130    statusText: "Switching Protocols",
131    type: "default",
132    url: "",
133    redirected: false,
134    upgrade: id,
135  });
136  return { socket, response };
137}
138
139async function dispatch(
140  metadata: HttpRequestMetadata,
141  handler: ServeHandler,
142  onError: ErrorHandler | undefined,
143): Promise<void> {
144  const id = metadata[0], signal = new AbortSignal(secret);
145  let controller: ReadableByteStreamController | undefined,
146    reader: ReadableStreamDefaultReader<unknown> | undefined,
147    bodyEnded = !metadata[4],
148    sent = false;
149  const reported = new Set<unknown>();
150  function report(error: unknown): void {
151    if (state(signals, signal).aborted || has(reported, error)) return;
152    add(reported, error);
153    reportNonfatalException(error);
154  }
155  const disconnected = new DOMException(
156    "HTTP client disconnected",
157    "AbortError",
158  );
159  then(op_http_wait(id), (canceled) => {
160    if (canceled) abort(signal, disconnected);
161    if (!bodyEnded && controller) errorStream(controller, disconnected);
162    if (reader) markPromiseHandled(readerCancel(reader, disconnected));
163  }, () => abort(signal, disconnected));
164  const stream = metadata[4]
165    ? new ReadableStream<Uint8Array<ArrayBuffer>>({
166      __proto__: null,
167      type: "bytes",
168      start(c) {
169        controller = c;
170      },
171      async pull(c) {
172        try {
173          const bytes = await safePromise(op_http_read(id));
174          if (bytes === null) {
175            bodyEnded = true;
176            finish(c);
177          } else if (bytes.length) enqueue(c, bytes);
178        } catch (error) {
179          bodyEnded = true;
180          errorStream(c, error);
181        }
182      },
183      cancel() {
184        bodyEnded = true;
185      },
186    }, { __proto__: null, highWaterMark: 0 })
187    : null;
188  const h = fromPairs(metadata[3]);
189  state(headers, h).immutable = true;
190  const request = ObjectCreate(Request.prototype) as Request;
191  set(bodies, request, {
192    kind: "request",
193    url: metadata[2],
194    method: metadata[1],
195    headers: h,
196    signal,
197    redirect: "follow",
198    cache: "default",
199    credentials: "same-origin",
200    mode: "cors",
201    integrity: "",
202    keepalive: false,
203    referrer: "",
204    referrerPolicy: "",
205    priority: "auto",
206    body: { stream, bytes: null, used: false, type: null },
207  });
208  set(requests, request, id);
209  const info = ObjectFreeze({
210    __proto__: null,
211    remoteAddr: address(metadata[5]),
212  });
213  try {
214    let response: unknown;
215    try {
216      response = await safePromise(handler(request, info));
217      const s = messageState(response, "response");
218      usable(s.body);
219      if (s.status < 200 && !(s.status === 101 && s.upgrade === id)) {
220        throw new TypeError("Invalid HTTP response status");
221      }
222    } catch (error) {
223      if (!onError) report(error);
224      try {
225        response = onError
226          ? await safePromise(onError(error))
227          : new Response("Internal Server Error", { status: 500 });
228        const s = messageState(response, "response");
229        usable(s.body);
230        if (s.status < 200 && !(s.status === 101 && s.upgrade === id)) {
231          throw new TypeError("Invalid HTTP response status");
232        }
233      } catch (error) {
234        report(error);
235        response = new Response("Internal Server Error", { status: 500 });
236      }
237    }
238    const s = messageState(response, "response"), b = s.body;
239    if (state(signals, signal).aborted) {
240      if (b.stream) markPromiseHandled(streamCancel(b.stream, disconnected));
241      return;
242    }
243    const file = b.stream === null ? null : takeFile(b.stream, signal);
244    if (file !== null) {
245      b.used = true;
246      op_http_respond_file([id, s.status, headerEntries(s.headers), file]);
247      sent = true;
248      return;
249    }
250    const streaming = b.stream !== null && b.bytes === null;
251    if (streaming) reader = streamReader(b.stream);
252    b.used = b.stream !== null;
253    const wantsBody = op_http_respond([
254      id,
255      s.status,
256      headerEntries(s.headers),
257      b.bytes,
258      streaming,
259    ]);
260    sent = true;
261    if (b.bytes !== null) markPromiseHandled(streamCancel(b.stream));
262    if (reader) {
263      if (wantsBody) {
264        while (true) {
265          const result = await safePromise(read(reader));
266          if (result.done) break;
267          if (tag(result.value) !== "Uint8Array") {
268            throw new TypeError("Response chunks must be Uint8Array");
269          }
270          // The native write still checks shared and resizable backing stores.
271          if (
272            !await safePromise(
273              op_http_write([id, result.value as Uint8Array<ArrayBuffer>]),
274            )
275          ) {
276            markPromiseHandled(readerCancel(reader, disconnected));
277            return;
278          }
279        }
280        await safePromise(op_http_write([id, null]));
281      } else markPromiseHandled(readerCancel(reader));
282    }
283  } catch (error) {
284    report(error);
285    if (!sent) {
286      try {
287        const fallback = new Response("Internal Server Error", { status: 500 });
288        const s = messageState(fallback, "response");
289        op_http_respond([
290          id,
291          s.status,
292          headerEntries(s.headers),
293          s.body.bytes,
294          false,
295        ]);
296      } catch {
297        op_http_cancel(id);
298      }
299    } else op_http_cancel(id);
300    if (reader) markPromiseHandled(readerCancel(reader));
301  } finally {
302    if (reader) release(reader);
303    reader = undefined;
304  }
305}
306
307function serve(handler: ServeHandler<TcpAddr>): HttpServer<TcpAddr>;
308function serve(options: ServeOptions, handler: ServeHandler<TcpAddr>): HttpServer<TcpAddr>;
309function serve(options: UnixServeOptions, handler: ServeHandler<UnixAddr>): HttpServer<UnixAddr>;
310function serve(rawOptions: unknown, handler: unknown = undefined): HttpServer {
311  if (typeof rawOptions === "function") {
312    handler = rawOptions;
313    rawOptions = {};
314  }
315  const options = dictionary(rawOptions);
316  if (typeof handler !== "function") {
317    throw new TypeError("serve requires a handler");
318  }
319  const hostname = options.hostname === undefined
320    ? "127.0.0.1"
321    : usv(options.hostname);
322  const port = options.port === undefined ? 8000 : +(options.port as number);
323  if (!NumberIsFinite(port) || port < 0 || port > 65535 || port % 1 !== 0) {
324    throw new RangeError("Invalid HTTP listen port");
325  }
326  const signal = options.signal,
327    onError = options.onError,
328    onListen = options.onListen;
329  if (onError !== undefined && typeof onError !== "function") {
330    throw new TypeError("Invalid onError callback");
331  }
332  if (onListen !== undefined && typeof onListen !== "function") {
333    throw new TypeError("Invalid onListen callback");
334  }
335  const path = options.path === undefined ? undefined : usv(options.path);
336  if (
337    path !== undefined &&
338    (options.hostname !== undefined || options.port !== undefined)
339  ) {
340    throw new TypeError(
341      "Unix socket path cannot be combined with hostname or port",
342    );
343  }
344  if (signal !== undefined && state(signals, signal).aborted) {
345    throw state(signals, signal).reason;
346  }
347  const bound = op_http_listen(
348    path === undefined
349      ? { __proto__: null, transport: "tcp", hostname, port }
350      : { __proto__: null, transport: "unix", path },
351  );
352  return startServer(
353    bound,
354    handler as ServeHandler,
355    onError as ErrorHandler | undefined,
356    signal,
357    onListen as ((addr: StreamAddr) => unknown) | undefined,
358  );
359}
360
361function startServer(
362  bound: HttpListener,
363  handler: ServeHandler,
364  onError: ErrorHandler | undefined,
365  signal: unknown,
366  onListen: ((addr: StreamAddr) => unknown) | undefined,
367): HttpServer {
368  const id = bound[0], addr = address(bound[1]);
369  const stop = () => op_http_stop([id, true]);
370  if (signal !== undefined) addListener(signal, "abort", stop, { once: true });
371  const finished = then(op_http_finished(id), () => {
372    if (signal !== undefined) removeListener(signal, "abort", stop);
373  }, (error) => {
374    if (signal !== undefined) removeListener(signal, "abort", stop);
375    throw error;
376  });
377  markPromiseHandled(finished);
378  then(
379    (async () => {
380      while (true) {
381        const metadata = await safePromise(op_http_next(id));
382        if (metadata === null) break;
383        markPromiseHandled(dispatch(metadata, handler, onError));
384      }
385    })(),
386    () => {},
387    stop,
388  );
389  const server = ObjectFreeze({
390    __proto__: null,
391    addr,
392    finished,
393    shutdown() {
394      op_http_stop([id, false]);
395      return finished;
396    },
397  });
398  try {
399    if (onListen) onListen(addr);
400  } catch (error) {
401    stop();
402    throw error;
403  }
404  return server;
405}
406
407function startCLI(namespace: Record<string, unknown>): Promise<void> {
408  const module = namespace.default;
409  if (module === null || typeof module !== "object") {
410    throw new TypeError(
411      "avril serve requires a default export object with a fetch function",
412    );
413  }
414  const fetch = (module as Record<string, unknown>).fetch;
415  if (typeof fetch !== "function") {
416    throw new TypeError(
417      "avril serve requires a default export object with a fetch function",
418    );
419  }
420  const bound = op_http_cli_listen(null);
421  const server = startServer(
422    bound,
423    (request, info) => apply(fetch, module, [request, info]),
424    undefined,
425    undefined,
426    undefined,
427  );
428  return server.finished;
429}
430
431export type { HttpServer, ServeHandler, ServeInfo, StreamAddr };
432export { serve, startCLI, upgradeWebSocket };