import {
  NumberIsFinite,
  ObjectCreate,
  ObjectFreeze,
  RangeError,
  ReflectApply as apply,
  Set,
  SetPrototypeAdd as add,
  SetPrototypeHas as has,
  TypedArrayPrototypeGetTag as tag,
  TypeError,
  WeakMap,
  WeakMapPrototypeGet as get,
  WeakMapPrototypeSet as set,
} from "ext:core/primordials";
import { reportNonfatalException } from "ext:core/event_hooks";
import {
  markPromiseHandled,
  promiseThen as then,
  safePromise,
} from "ext:core/promises";
import {
  op_http_cancel,
  op_http_cli_listen,
  op_http_finished,
  op_http_listen,
  op_http_next,
  op_http_read,
  op_http_respond,
  op_http_respond_file,
  op_http_stop,
  op_http_upgrade,
  op_http_wait,
  op_http_write,
} from "ext:core/ops";
import { dictionary, state, usv } from "ext:web/idl.js";
import {
  abort,
  AbortSignal,
  addListener,
  DOMException,
  removeListener,
  secret,
  signals,
} from "ext:web/events.js";
import {
  enqueue,
  errorStream,
  finish,
  read,
  ReadableStream,
  readerCancel,
  release,
  streamCancel,
  streamReader,
} from "ext:webstreams/streams.js";
import { bodies, messageState, usable } from "ext:http_types/body.js";
import { fromPairs, headerEntries, headers } from "ext:http_types/headers.js";
import { Request } from "ext:http_types/request.js";
import { Response } from "ext:http_types/response.js";
import { adoptWebSocket } from "ext:websocket/websocket.js";
import { takeFile } from "./file.js";

import type {
  HttpAddress,
  HttpListener,
  HttpRequestMetadata,
} from "ext:core/ops";
import type {
  ReadableByteStreamController,
  ReadableStreamDefaultReader,
} from "ext:webstreams/streams.js";
import type { WebSocket } from "ext:websocket/websocket.js";

import type { ServeOptions, UnixServeOptions } from "avril:std";

type StreamAddr = Readonly<HttpAddress>;
type TcpAddr = Extract<StreamAddr, { transport: "tcp" }>;
type UnixAddr = Extract<StreamAddr, { transport: "unix" }>;
interface ServeInfo<A extends StreamAddr = StreamAddr> {
  readonly remoteAddr: A;
}
type ServeHandler<A extends StreamAddr = StreamAddr> = (request: Request, info: ServeInfo<A>) => unknown;
type ErrorHandler = (error: unknown) => unknown;
interface HttpServer<A extends StreamAddr = StreamAddr> {
  readonly addr: A;
  readonly finished: Promise<void>;
  readonly shutdown: () => Promise<void>;
}

const requests = new WeakMap<object, number>();

function address(value: HttpAddress): StreamAddr {
  return ObjectFreeze(
    value.transport === "unix"
      ? { __proto__: null, transport: "unix", path: value.path }
      : {
        __proto__: null,
        transport: "tcp",
        hostname: value.hostname,
        port: value.port,
      },
  );
}

function upgradeWebSocket(
  request: unknown,
  rawOptions: unknown = {},
): { socket: WebSocket; response: Response } {
  const id = get(requests, request);
  if (id === undefined) {
    throw new TypeError("Expected an incoming HTTP request");
  }
  const requestState = messageState(request, "request");
  usable(requestState.body);
  const options = dictionary(rawOptions);
  const protocol = options.protocol === undefined
    ? null
    : usv(options.protocol);
  const bound = op_http_upgrade([id, protocol]);
  const socket = adoptWebSocket(bound[0], requestState.url);
  const response = ObjectCreate(Response.prototype) as Response,
    h = fromPairs(bound[1]);
  state(headers, h).immutable = true;
  set(bodies, response, {
    kind: "response",
    body: { stream: null, bytes: null, used: false, type: null },
    headers: h,
    status: 101,
    statusText: "Switching Protocols",
    type: "default",
    url: "",
    redirected: false,
    upgrade: id,
  });
  return { socket, response };
}

async function dispatch(
  metadata: HttpRequestMetadata,
  handler: ServeHandler,
  onError: ErrorHandler | undefined,
): Promise<void> {
  const id = metadata[0], signal = new AbortSignal(secret);
  let controller: ReadableByteStreamController | undefined,
    reader: ReadableStreamDefaultReader<unknown> | undefined,
    bodyEnded = !metadata[4],
    sent = false;
  const reported = new Set<unknown>();
  function report(error: unknown): void {
    if (state(signals, signal).aborted || has(reported, error)) return;
    add(reported, error);
    reportNonfatalException(error);
  }
  const disconnected = new DOMException(
    "HTTP client disconnected",
    "AbortError",
  );
  then(op_http_wait(id), (canceled) => {
    if (canceled) abort(signal, disconnected);
    if (!bodyEnded && controller) errorStream(controller, disconnected);
    if (reader) markPromiseHandled(readerCancel(reader, disconnected));
  }, () => abort(signal, disconnected));
  const stream = metadata[4]
    ? new ReadableStream<Uint8Array<ArrayBuffer>>({
      __proto__: null,
      type: "bytes",
      start(c) {
        controller = c;
      },
      async pull(c) {
        try {
          const bytes = await safePromise(op_http_read(id));
          if (bytes === null) {
            bodyEnded = true;
            finish(c);
          } else if (bytes.length) enqueue(c, bytes);
        } catch (error) {
          bodyEnded = true;
          errorStream(c, error);
        }
      },
      cancel() {
        bodyEnded = true;
      },
    }, { __proto__: null, highWaterMark: 0 })
    : null;
  const h = fromPairs(metadata[3]);
  state(headers, h).immutable = true;
  const request = ObjectCreate(Request.prototype) as Request;
  set(bodies, request, {
    kind: "request",
    url: metadata[2],
    method: metadata[1],
    headers: h,
    signal,
    redirect: "follow",
    cache: "default",
    credentials: "same-origin",
    mode: "cors",
    integrity: "",
    keepalive: false,
    referrer: "",
    referrerPolicy: "",
    priority: "auto",
    body: { stream, bytes: null, used: false, type: null },
  });
  set(requests, request, id);
  const info = ObjectFreeze({
    __proto__: null,
    remoteAddr: address(metadata[5]),
  });
  try {
    let response: unknown;
    try {
      response = await safePromise(handler(request, info));
      const s = messageState(response, "response");
      usable(s.body);
      if (s.status < 200 && !(s.status === 101 && s.upgrade === id)) {
        throw new TypeError("Invalid HTTP response status");
      }
    } catch (error) {
      if (!onError) report(error);
      try {
        response = onError
          ? await safePromise(onError(error))
          : new Response("Internal Server Error", { status: 500 });
        const s = messageState(response, "response");
        usable(s.body);
        if (s.status < 200 && !(s.status === 101 && s.upgrade === id)) {
          throw new TypeError("Invalid HTTP response status");
        }
      } catch (error) {
        report(error);
        response = new Response("Internal Server Error", { status: 500 });
      }
    }
    const s = messageState(response, "response"), b = s.body;
    if (state(signals, signal).aborted) {
      if (b.stream) markPromiseHandled(streamCancel(b.stream, disconnected));
      return;
    }
    const file = b.stream === null ? null : takeFile(b.stream, signal);
    if (file !== null) {
      b.used = true;
      op_http_respond_file([id, s.status, headerEntries(s.headers), file]);
      sent = true;
      return;
    }
    const streaming = b.stream !== null && b.bytes === null;
    if (streaming) reader = streamReader(b.stream);
    b.used = b.stream !== null;
    const wantsBody = op_http_respond([
      id,
      s.status,
      headerEntries(s.headers),
      b.bytes,
      streaming,
    ]);
    sent = true;
    if (b.bytes !== null) markPromiseHandled(streamCancel(b.stream));
    if (reader) {
      if (wantsBody) {
        while (true) {
          const result = await safePromise(read(reader));
          if (result.done) break;
          if (tag(result.value) !== "Uint8Array") {
            throw new TypeError("Response chunks must be Uint8Array");
          }
          // The native write still checks shared and resizable backing stores.
          if (
            !await safePromise(
              op_http_write([id, result.value as Uint8Array<ArrayBuffer>]),
            )
          ) {
            markPromiseHandled(readerCancel(reader, disconnected));
            return;
          }
        }
        await safePromise(op_http_write([id, null]));
      } else markPromiseHandled(readerCancel(reader));
    }
  } catch (error) {
    report(error);
    if (!sent) {
      try {
        const fallback = new Response("Internal Server Error", { status: 500 });
        const s = messageState(fallback, "response");
        op_http_respond([
          id,
          s.status,
          headerEntries(s.headers),
          s.body.bytes,
          false,
        ]);
      } catch {
        op_http_cancel(id);
      }
    } else op_http_cancel(id);
    if (reader) markPromiseHandled(readerCancel(reader));
  } finally {
    if (reader) release(reader);
    reader = undefined;
  }
}

function serve(handler: ServeHandler<TcpAddr>): HttpServer<TcpAddr>;
function serve(options: ServeOptions, handler: ServeHandler<TcpAddr>): HttpServer<TcpAddr>;
function serve(options: UnixServeOptions, handler: ServeHandler<UnixAddr>): HttpServer<UnixAddr>;
function serve(rawOptions: unknown, handler: unknown = undefined): HttpServer {
  if (typeof rawOptions === "function") {
    handler = rawOptions;
    rawOptions = {};
  }
  const options = dictionary(rawOptions);
  if (typeof handler !== "function") {
    throw new TypeError("serve requires a handler");
  }
  const hostname = options.hostname === undefined
    ? "127.0.0.1"
    : usv(options.hostname);
  const port = options.port === undefined ? 8000 : +(options.port as number);
  if (!NumberIsFinite(port) || port < 0 || port > 65535 || port % 1 !== 0) {
    throw new RangeError("Invalid HTTP listen port");
  }
  const signal = options.signal,
    onError = options.onError,
    onListen = options.onListen;
  if (onError !== undefined && typeof onError !== "function") {
    throw new TypeError("Invalid onError callback");
  }
  if (onListen !== undefined && typeof onListen !== "function") {
    throw new TypeError("Invalid onListen callback");
  }
  const path = options.path === undefined ? undefined : usv(options.path);
  if (
    path !== undefined &&
    (options.hostname !== undefined || options.port !== undefined)
  ) {
    throw new TypeError(
      "Unix socket path cannot be combined with hostname or port",
    );
  }
  if (signal !== undefined && state(signals, signal).aborted) {
    throw state(signals, signal).reason;
  }
  const bound = op_http_listen(
    path === undefined
      ? { __proto__: null, transport: "tcp", hostname, port }
      : { __proto__: null, transport: "unix", path },
  );
  return startServer(
    bound,
    handler as ServeHandler,
    onError as ErrorHandler | undefined,
    signal,
    onListen as ((addr: StreamAddr) => unknown) | undefined,
  );
}

function startServer(
  bound: HttpListener,
  handler: ServeHandler,
  onError: ErrorHandler | undefined,
  signal: unknown,
  onListen: ((addr: StreamAddr) => unknown) | undefined,
): HttpServer {
  const id = bound[0], addr = address(bound[1]);
  const stop = () => op_http_stop([id, true]);
  if (signal !== undefined) addListener(signal, "abort", stop, { once: true });
  const finished = then(op_http_finished(id), () => {
    if (signal !== undefined) removeListener(signal, "abort", stop);
  }, (error) => {
    if (signal !== undefined) removeListener(signal, "abort", stop);
    throw error;
  });
  markPromiseHandled(finished);
  then(
    (async () => {
      while (true) {
        const metadata = await safePromise(op_http_next(id));
        if (metadata === null) break;
        markPromiseHandled(dispatch(metadata, handler, onError));
      }
    })(),
    () => {},
    stop,
  );
  const server = ObjectFreeze({
    __proto__: null,
    addr,
    finished,
    shutdown() {
      op_http_stop([id, false]);
      return finished;
    },
  });
  try {
    if (onListen) onListen(addr);
  } catch (error) {
    stop();
    throw error;
  }
  return server;
}

function startCLI(namespace: Record<string, unknown>): Promise<void> {
  const module = namespace.default;
  if (module === null || typeof module !== "object") {
    throw new TypeError(
      "avril serve requires a default export object with a fetch function",
    );
  }
  const fetch = (module as Record<string, unknown>).fetch;
  if (typeof fetch !== "function") {
    throw new TypeError(
      "avril serve requires a default export object with a fetch function",
    );
  }
  const bound = op_http_cli_listen(null);
  const server = startServer(
    bound,
    (request, info) => apply(fetch, module, [request, info]),
    undefined,
    undefined,
    undefined,
  );
  return server.finished;
}

export type { HttpServer, ServeHandler, ServeInfo, StreamAddr };
export { serve, startCLI, upgradeWebSocket };