guest.ts

import {
  encodeBase64,
  decodeBase64,
} from "https://deno.land/std/encoding/base64.ts";
import process from "node:process";
import SuperJSON from "npm:[email protected]";

SuperJSON.registerCustom(
  {
    isApplicable: (v: any): v is Function => typeof v === "function",
    serialize: (_v: Function) => "",
    deserialize: () => {
      // This will cause typeOutput to detect a function
      // and trigger the right behavior in ValEditor.
      return Symbol.for("function") as unknown as Function;
    },
  },
  "function"
);

/**
 * Given an object of environment variables, create a stub
 * that simulates the same interface as Deno.env
 */
export function createDenoEnvStub(
  input: Record<string, string>
): typeof Deno.env {
  return {
    get(key: string) {
      return input[key];
    },
    has(key: string) {
      return input[key] !== undefined;
    },
    toObject() {
      return { ...input };
    },
    set(_key: string, _value: string) {
      // Stub
    },
    delete(_key: string) {
      // Stub
    },
  };
}

// For now Deno.env is backed by this global object so that we can inject values
// into it at runtime. This will need to change if we ever handle multiple
// requests at once.
const envInput: Record<string, string> = {};

async function readBody(resp: Response) {
  try {
    return encodeBase64(await resp.arrayBuffer());
  } catch (_e) {
    return null;
  }
}

/**
 * Given a Response object, serialize it.
 * Note: if you try this twice on the same Response, it'll
 * crash! Streams, like resp.arrayBuffer(), can only
 * be consumed once.
 */
export async function serializeResponse(resp: Response) {
  return {
    headers: [...resp.headers.entries()],
    body: await readBody(resp),
    status: resp.status,
    statusText: resp.statusText,
  };
}

/**
 * Express API mocks ----------------------------------------------------------
 */

export class RequestLike {
  #options: SerializedExpressRequest;

  constructor(options: SerializedExpressRequest) {
    this.#options = options;
  }

  // https://expressjs.com/en/api.html#req.method
  get method() {
    return this.#options.method;
  }

  // https://expressjs.com/en/api.html#req.query
  get query() {
    return this.#options.query;
  }

  get body() {
    return this.#options.body;
  }

  /**
   * Stubs
   */

  // https://expressjs.com/en/api.html#req.baseUrl
  get baseUrl() {
    return "/";
  }

  // https://expressjs.com/en/api.html#req.params
  get params() {
    return {};
  }

  // https://expressjs.com/en/api.html#req.secure
  get secure() {
    return true;
  }

  // https://expressjs.com/en/api.html#req.subdomains
  get subdomains() {
    return [];
  }

  // https://expressjs.com/en/api.html#req.fresh
  get fresh() {
    return true;
  }

  // https://expressjs.com/en/api.html#req.protocol
  get protocol() {
    return "https";
  }

  // https://expressjs.com/en/api.html#req.path
  get path() {
    return this.#options.path;
  }

  // https://expressjs.com/en/4x/api.html#req.originalUrl
  get originalUrl() {
    return this.#options.originalUrl;
  }

  acceptsCharsets() {
    return true;
  }

  acceptsEncodings() {
    return true;
  }

  acceptsLanguages() {
    return true;
  }

  accepts() {
    return true;
  }

  get(name: string) {
    return this.#options.headers[String(name).toLowerCase()];
  }
}

interface Timing {
  executionStart: number;
  importComplete: number;
  executionComplete: number;
}

interface DoneMessage {
  type: "done";
  wallTime: number;
}
interface ErrorMessage {
  type: "error";
  value: { name: string; message: string; stack?: string };
}
interface ReadyMessage {
  type: "ready";
}
interface ReturnMessage {
  type: "return";
  value: any;
}
interface ExportsMessage {
  type: "exports";
  value: any;
}
interface ExpressResponseMessage {
  type: "expressresponse";
  name: string;
  args: unknown[];
}
export type Message =
  | DoneMessage
  | ErrorMessage
  | ReadyMessage
  | ReturnMessage
  | ExportsMessage
  | ExpressResponseMessage;

export class ResponseLike {
  #messages: Message[];
  constructor(messages: Message[]) {
    this.#messages = messages;
  }
  #stub(name: string, args: unknown[]) {
    this.#messages.push({
      type: "expressresponse",
      name,
      args,
    });
    return this;
  }
  json(...args: unknown[]) {
    return this.#stub("json", args);
  }
  jsonp(...args: unknown[]) {
    return this.#stub("jsonp", args);
  }
  status(...args: unknown[]) {
    return this.#stub("status", args);
  }
  send(...args: unknown[]) {
    return this.#stub("send", args);
  }
  type(...args: unknown[]) {
    return this.#stub("type", args);
  }
  get(...args: unknown[]) {
    return this.#stub("get", args);
  }
  redirect(...args: unknown[]) {
    return this.#stub("redirect", args);
  }
  end(...args: unknown[]) {
    return this.#stub("end", args);
  }
  set(...args: unknown[]) {
    return this.#stub("set", args);
  }
}

/**
 * Deserialization ------------------------------------------------------------
 */
const TAG = "@@valtown-type";
function getDeserializationTag(arg: unknown): string | undefined {
  if (
    arg &&
    typeof arg === "object" &&
    TAG in arg &&
    typeof arg[TAG] === "string"
  ) {
    return arg[TAG];
  }
}

type SerializedRequest = {
  [TAG]: "request";
  url: string;
  method: string;
  headers: [string, string][];
  body?: string;
};

type SerializedExpressRequest = {
  [TAG]: "express-request";
  method: string;
  protocol: string;
  hostname: string;
  xhr: boolean;
  body: string;
  query: Record<string, string>;
  headers: Record<string, string>;
  path: string;
  originalUrl: string;
};

type SerializedExpressResponse = {
  [TAG]: "express-response";
};

function deserializeRequest(arg: SerializedRequest) {
  return new Request(arg.url, {
    method: arg.method,
    headers: arg.headers,
    ...(arg.body ? { body: decodeBase64(arg.body) } : {}),
  });
}

function deserializeExpressRequest(arg: SerializedExpressRequest) {
  return new RequestLike(arg);
}

function deserializeExpressResponse(
  _arg: SerializedExpressResponse,
  messages: Message[]
) {
  return new ResponseLike(messages);
}

export function deserializeCustom(messages: Message[]): (arg: unknown) => any {
  return (arg: unknown) => {
    // Sniff for custom types that can't make it through
    // structuredClone and deserialize them into custom
    // objects.
    switch (getDeserializationTag(arg)) {
      case "request": {
        return deserializeRequest(arg as SerializedRequest);
      }
      case "express-request": {
        return deserializeExpressRequest(arg as SerializedExpressRequest);
      }
      case "express-response": {
        return deserializeExpressResponse(
          arg as SerializedExpressResponse,
          messages
        );
      }
      default: {
        return arg;
      }
    }
  };
}

const WALL_TIME_START = Date.now();

function getMainExport(
  mod: any
): { ok: true; value: any } | { ok: false; error: Error } {
  if ("default" in mod) {
    return { ok: true, value: mod.default };
  }

  // If the val has exactly one named export, we run that.

  const exports = Object.keys(mod);

  if (exports.length > 1) {
    const error = new Error(
      `Vals require a default export, or exactly one named export. This val exports: ${exports.join(
        ", "
      )}`
    );
    error.name = "ImportValError";
    return { ok: false, error };
  } else if (exports.length === 0) {
    const error = new Error(
      "Vals require a default export, or exactly one named export. This val has none."
    );
    error.name = "ImportValError";
    return { ok: false, error };
  }

  return { ok: true, value: mod[exports[0]] };
}

const handleInternalRequest = async (
  args: any,
  mod: any,
  req: Request
): Promise<Message[]> => {
  try {
    if (Array.isArray(args)) {
      const exp = getMainExport(mod);
      if (!exp.ok) {
        return [{ type: "error", value: serializeError(exp.error) }];
      }
      if (typeof exp.value === "function") {
        const expressMessages: Message[] = [];
        let value = await exp.value(
          ...args.map(deserializeCustom(expressMessages))
        );
        if (value instanceof Response) {
          value = await serializeResponse(value);
        }
        return [{ type: "return", value }, ...expressMessages];
      } else {
        return [{ type: "return", value: await exp.value }];
      }
    } else {
      // Await every exported value
      const awaitedMod = Object.fromEntries(
        await Promise.all(
          Object.entries(mod).map(async ([key, v]) => {
            const resolved = await v;
            if (resolved instanceof Response) {
              return [key, await serializeResponse(resolved)];
            }
            return [key, resolved];
          })
        )
      );
      const exp = getMainExport(awaitedMod);
      const out: Message[] = [
        { type: "return", value: exp.ok ? exp.value : undefined },
        { type: "exports", value: awaitedMod },
      ];
      return out;
    }
  } catch (e) {
    return [{ type: "error", value: serializeError(e) }];
  }
};

const serializeError = (e: Error) => {
  return { name: e.name, message: e.message, stack: cleanStack(e.stack) };
};

const serializeExportsAndReturn = (msgs: Message[]): Message[] => {
  return msgs.map((msg) => {
    switch (msg.type) {
      case "exports":
      case "return":
        return { type: msg.type, value: SuperJSON.serialize(msg.value) };
      default:
        return msg;
    }
  });
};

const handleRequest = async (mod: any, req: Request) => {
  try {
    const traceparent = req.headers.get("traceparent");
    if (traceparent) {
      envInput["vt-traceparent"] = traceparent;
    }
    if (req.headers.has("X-VT-Internal-Request")) {
      try {
        // If this is not an http request, the body contains the request
        // arguments. Process them.
        const { args }: { args: any } = SuperJSON.deserialize(await req.json());
        let resp = await handleInternalRequest(args, mod, req);
        resp = serializeExportsAndReturn(resp);
        if (JSON.stringify(resp).length > 10_000_000) {
          Response.json([
            {
              type: "error",
              value: {
                name: "WS_PAYLOAD_TOO_LARGE",
                message: `The response is too large to process`,
              },
            },
          ]);
        }
        return Response.json(resp);
      } catch (e) {
        return Response.json([{ type: "error", value: serializeError(e) }]);
      }
    }
  } catch (e) {
    return Response.json({ error: serializeError(e) }, { status: 500 });
  }
  // This is an http request.
  try {
    // If this is an http request then we expect that the request is an external
    // request and we can process it with the handler directly.
    const exp = getMainExport(mod);
    if (!exp.ok) {
      return Response.json(
        { error: serializeError(exp.error) },
        { status: 500 }
      );
    }
    // TODO: these are all http requests, so what to do if the response is not a
    // Response object?
    const argResp = await (async () => {
      if (typeof exp.value === "function") {
        const value = await exp.value(req);
        if (value instanceof Response) {
          return value;
        }
        return value;
      } else {
        await exp.value;
      }
    })();
    return argResp;
  } catch (e) {
    // Do not remove this console.log, we want this error in the logs.
    console.log(cleanStack(e.stack));
    return Response.json(
      { error: serializeError(e) },
      {
        status: 500,
        headers: { "X-VT-Error": "true" },
      }
    );
  }
};

function cleanStack(str?: string) {
  if (!str) return undefined;
  return str
    .split("\n")
    .filter(
      (line) =>
        !line.includes(import.meta.url) &&
        !line.includes("deno-http-worker/deno-bootstrap/index.ts")
    )
    .join("\n");
}

let initialized = false;
let initializing = false;
let mod: any;

const pendingRequests: {
  req: Request;
  resolve: (value: Response | Promise<Response>) => void;
  reject: (reason?: unknown) => void;
}[] = [];

// We expect a single init request and then a follow-on request from the user
// and then we send values back.
export default {
  onError(e: any) {
    return new Response(e.message, { status: 400 });
  },
  async fetch(req: Request): Promise<Response> {
    try {
      if (initializing) {
        return new Promise<Response>((resolve, reject) => {
          pendingRequests.push({ req, resolve, reject });
        });
      }
      if (!initialized) {
        initializing = true;
        let { env, entrypoint } = await req.json();
        if (!entrypoint) {
          // This request will error and future requests will hang.
          return new Response("No source or import value found", {
            status: 500,
          });
        }
        process.env = env ? { ...env } : {};
        Object.assign(envInput, env);
        process.cwd = () => "/app/";
        Object.defineProperty(Deno, "env", {
          value: createDenoEnvStub(envInput),
        });
        mod = await import(entrypoint);
        initialized = true;
        initializing = false;
        for (const { req, resolve } of pendingRequests) {
          resolve(handleRequest(mod, req));
        }
        return Response.json({});
      }
      // Process requests that are not pending
      return handleRequest(mod, req);
    } catch (e) {
      console.error(e);
      return Response.json({ error: serializeError(e) }, { status: 500 });
    }
  },
};