// What berry and an environment's worker say to each other: one json value per line, with bytes and the
// environment's errors spelled out so that they arrive as what they were.

import { Buffer } from "node:buffer";
import { ExecutionError, FileError } from "@earendil-works/pi-durable/env";

/** Bumped whenever either side's messages change shape. */
export const PROTOCOL = 3;

/** What the worker says first: the machine it found itself on. */
export type Hello = {
  readonly protocol: number;
  readonly platform: string;
  readonly home: string;
  readonly variables: Record<string, string>;
};

export type FetchOptions = {
  method: string;
  headers: [string, string][];
  body?: Uint8Array;
  redirect: RequestRedirect;
  credentials: RequestCredentials;
  cache: RequestCache;
  referrer: string;
  referrerPolicy: ReferrerPolicy;
  integrity: string;
};

export type FetchResponse = {
  status: number;
  statusText: string;
  headers: [string, string][];
  url: string;
  redirected: boolean;
  handle?: number;
};

export type Request =
  /** Call `m` with `a`, resolving paths against `cwd`. */
  | {
    readonly i: number;
    readonly m: string;
    readonly a: readonly unknown[];
    readonly cwd?: string;
  }
  /** Abort the call `c`. */
  | { readonly c: number };

export type Response =
  | { readonly hello: Hello }
  /** The call `i` settled with `r`. */
  | { readonly i: number; readonly r: unknown }
  /** The call `i` threw, which the environment's methods never mean to. */
  | { readonly i: number; readonly e: string }
  /** Something call `i` streams, while it runs or for as long as what it opened stays open. */
  | { readonly i: number; readonly o: unknown };

export function encode(message: Request | Response): string {
  return JSON.stringify(message, function (this: Record<string, unknown>, key, converted) {
    // what toJSON() made of it, for buffers an array of numbers
    const value = this[key];
    if (value instanceof Uint8Array) {
      return {
        $bytes: Buffer.from(value.buffer, value.byteOffset, value.byteLength).toString("base64"),
      };
    }
    if (value instanceof FileError) {
      return { $file: { code: value.code, message: value.message, path: value.path } };
    }
    if (value instanceof ExecutionError) {
      return { $exec: { code: value.code, message: value.message, spillPath: value.spillPath } };
    }
    return converted;
  });
}

export function decode(line: string): Request | Response {
  return JSON.parse(line, (_, value) => {
    if (typeof value !== "object" || value === null) return value;
    if (typeof value.$bytes === "string") {
      return new Uint8Array(Buffer.from(value.$bytes, "base64"));
    }
    if (value.$file) return new FileError(value.$file.code, value.$file.message, value.$file.path);
    if (value.$exec) {
      const error = new ExecutionError(value.$exec.code, value.$exec.message);
      if (value.$exec.spillPath !== undefined) error.spillPath = value.$exec.spillPath;
      return error;
    }
    return value;
  });
}

/** Lines of a byte stream, as they complete. */
export async function* lines(
  reader: ReadableStreamDefaultReader<Uint8Array>,
  buffered: Uint8Array = new Uint8Array(),
): AsyncGenerator<string> {
  const decoder = new TextDecoder();
  // chunks of the line in progress, which can be a whole file's worth
  let parts: string[] = [];
  let text = decoder.decode(buffered, { stream: true });
  while (true) {
    let start = 0;
    for (let end; (end = text.indexOf("\n", start)) >= 0; start = end + 1) {
      parts.push(text.slice(start, end));
      yield parts.join("");
      parts = [];
    }
    parts.push(text.slice(start));
    const { value, done } = await reader.read();
    if (done) return;
    text = decoder.decode(value, { stream: true });
  }
}