smol

import { Readable } from "node:stream";
import http, { ServerResponse } from "node:http";

import { Result$Error } from "./gleam.mjs";

import {
  adapt as adaptFetch,
  adaptNode,
  nodeToRequest,
  responseToNode,
} from "./adapt.ffi.mjs";
import { upgrade } from "./smol.ffi.mjs";
import { started } from "./smol.mjs";

import {
  Builder$Builder$handler,
  Builder$Builder$port,
  Builder$Builder$interface,
  Builder$Builder$on_error,
  Builder$Builder$after_start,
} from "./smol/internal/builder.mjs";

async function importWebSocket() {
  const mod = await import(new URL(`./vendor/${"ws.cjs"}`, import.meta.url).href);
  return mod.default ?? mod;
}

export async function start(builder) {
  try {
    const listener = adaptNode(Builder$Builder$handler(builder));
    const options = {};
    const server = http.createServer(options, listener);
    server.addListener("upgrade", adaptUpgrade(Builder$Builder$handler(builder)));

    let disposed = false;
    const stop = () => {
      if (disposed) {
        return;
      }
      disposed = true;

      process.off("SIGINT", onExit);
      process.off("SIGTERM", onExit);
      process.off("exit", onExit);

      return new Promise((resolve, reject) => {
        server.close((err) => (err ? reject(err) : resolve()));
      });
    };
    const onExit = async () => {
      try {
        await stop();
      } finally {
        process.exit();
      }
    };

    await new Promise((resolve, reject) => {
      server.once("error", reject);
      server.listen(
        Builder$Builder$port(builder),
        Builder$Builder$interface(builder),
        511,
        resolve,
      );
    });

    process.on("SIGINT", onExit);
    process.on("SIGTERM", onExit);
    process.on("exit", onExit);

    server.on("error", (err) => {
      Builder$Builder$on_error(builder)(err.stack, err);
      // FIXME: can we not just die here? maybe restart the server?
      process.exit(1);
    });

    const addressInfo = server.address();
    const port = addressInfo.port;
    const address = addressInfo.address;
    Builder$Builder$after_start(builder)(port, address);

    return started(port, address, stop);
  } catch (e) {
    return Result$Error(undefined);
  }
}

function adaptUpgrade(handler) {
  const fetchHandler = adaptFetch(handler);
  let wssPromise;

  const upgradeHandler = async (req, socket, head) => {
    const request = nodeToRequest({ toWebStream: Readable.toWeb }, req);
    const response = await fetchHandler(request);

    if (response.body[upgrade]) {
      const runtime = response.body[upgrade];
      const wss = await (wssPromise ??= importWebSocket().then((WebSocket) => {
        return new WebSocket.Server({ noServer: true, perMessageDeflate: true });
      }));

      wss.handleUpgrade(req, socket, head, (ws) => {
        runtime.send = (message) => ws.send(message);
        runtime.close = () => ws.close();
        runtime.onOpen();

        ws.on("message", (data, isBinary) =>
          runtime.onMessage(isBinary ? data : data.toString()),
        );
        ws.on("close", runtime.onClose);
      });
    } else {
      const res = new ServerResponse(req);
      res.assignSocket(socket);
      responseToNode(response, res);
    }
  };

  return upgradeHandler;
}