import {
  ObjectCreate,
  TypedArrayPrototypeGetTag as tag,
  TypeError,
  Uint8Array,
  uncurryThis,
  WeakMapPrototypeSet as set,
} from "ext:core/primordials";
import {
  markPromiseHandled,
  promiseThen as then,
  safePromise,
} from "ext:core/promises";
import { state } from "ext:web/idl.js";
import { userAgent } from "ext:web/navigator.js";
import {
  op_fetch_cancel,
  op_fetch_create,
  op_fetch_read,
  op_fetch_read_all,
  op_fetch_send,
  op_fetch_write,
} from "ext:core/ops";
import {
  enqueue,
  errorStream,
  finish,
  read,
  ReadableStream,
  readerCancel,
  release,
  streamCancel,
  streamReader,
} from "ext:webstreams/streams.js";
import { addListener, removeListener, signals } from "ext:web/events.js";
import { Request } from "ext:http_types/request.js";
import { Response } from "ext:http_types/response.js";
import {
  fromPairs,
  getHeader,
  headerEntries,
  headers,
  setHeader,
} from "ext:http_types/headers.js";
import { bodies, bodyCollectors, usable } from "ext:http_types/body.js";

import type { RequestState } from "ext:http_types/body.js";
import type {
  ReadableByteStreamController,
  ReadableStreamDefaultReader,
} from "ext:webstreams/streams.js";
import type { FetchMetadata } from "ext:core/ops";

interface FetchResource {
  id: number;
  cleanup(): void;
}

const WeakRef = globalThis.WeakRef,
  FinalizationRegistry = globalThis.FinalizationRegistry;
const deref = uncurryThis(
    (WeakRef.prototype as WeakRef<ReadableByteStreamController>).deref,
  ),
  registerFinalizer = uncurryThis(
    (FinalizationRegistry.prototype as FinalizationRegistry<FetchResource>)
      .register,
  );
function failure(error: unknown): TypeError {
  // Guest message getters and stringification can abort the request.
  return new TypeError((error as { message?: string }).message);
}
const finalizer = new FinalizationRegistry<FetchResource>(({ id, cleanup }) => {
  cleanup();
  op_fetch_cancel(id);
});
async function fetch(
  input: unknown,
  init: unknown = undefined,
): Promise<Response> {
  if (arguments.length === 0) throw new TypeError("fetch requires input");
  const request = new Request(input, init),
    s = state(bodies, request) as RequestState,
    b = s.body;
  if (state(signals, s.signal).aborted) throw state(signals, s.signal).reason;
  usable(b);
  const id = op_fetch_create(null);
  let controller: WeakRef<ReadableByteStreamController> | undefined,
    uploadError: TypeError | undefined,
    uploadReader: ReadableStreamDefaultReader<unknown> | undefined;
  const aborted = () => {
    if (uploadReader) {
      markPromiseHandled(
        readerCancel(uploadReader, state(signals, s.signal).reason),
      );
    }
    op_fetch_cancel(id);
    const c = controller && deref(controller);
    if (c) errorStream(c, state(signals, s.signal).reason);
  };
  addListener(s.signal, "abort", aborted, { once: true });
  const cleanup = () => removeListener(s.signal, "abort", aborted);
  let pending: Promise<FetchMetadata>;
  try {
    if (getHeader(s.headers, "user-agent") === null) {
      setHeader(s.headers, "user-agent", userAgent);
    }
    const entries = headerEntries(s.headers);
    b.used = b.stream !== null;
    pending = op_fetch_send([
      id,
      s.url,
      s.method,
      entries,
      b.bytes,
      b.stream !== null && b.bytes === null,
      s.redirect,
    ]);
    if (b.bytes !== null) markPromiseHandled(streamCancel(b.stream));
    else if (b.stream) {
      const reader = streamReader(b.stream);
      uploadReader = reader;
      then(
        (async () => {
          try {
            while (true) {
              const result = await safePromise(read(reader));
              if (result.done) break;
              if (tag(result.value) !== "Uint8Array") {
                throw new TypeError("Upload chunks must be Uint8Array");
              }
              // The native write still checks shared and resizable backing stores.
              if (
                !await safePromise(
                  op_fetch_write([id, result.value as Uint8Array<ArrayBuffer>]),
                )
              ) {
                try {
                  await safePromise(readerCancel(reader));
                } catch {
                  // The body is no longer wanted; failing cleanup must not fail the response.
                }
                return;
              }
            }
            await safePromise(op_fetch_write([id, null]));
          } finally {
            release(reader);
            uploadReader = undefined;
          }
        })(),
        () => {},
        (error) => {
          uploadError = failure(error);
          op_fetch_cancel(id);
          const c = controller && deref(controller);
          if (c) errorStream(c, uploadError);
        },
      );
    }
    const reply = await safePromise(pending);
    const status = reply[0],
      statusText = reply[1],
      list = reply[2],
      url = reply[3],
      redirected = reply[4],
      abandoned = reply[5];
    // The upload loop only notices on its next write, which may never come in time.
    if (abandoned && uploadReader) {
      markPromiseHandled(readerCancel(uploadReader));
    }
    if (state(signals, s.signal).aborted) {
      throw state(signals, s.signal).reason;
    }
    const h = fromPairs(list);
    state(headers, h).immutable = true;
    let stream: ReadableStream<Uint8Array<ArrayBuffer>> | null = null;
    if (
      s.method === "HEAD" || status === 204 || status === 205 ||
      status === 304
    ) {
      cleanup();
      op_fetch_cancel(id);
    } else {
      let collecting = false;
      stream = new ReadableStream<Uint8Array<ArrayBuffer>>({
        type: "bytes",
        start(c) {
          controller = new WeakRef(c);
        },
        async pull(c) {
          try {
            const bytes = await safePromise(
              collecting ? op_fetch_read_all(id) : op_fetch_read(id),
            );
            if (bytes !== null && bytes.length) enqueue(c, bytes);
            if (bytes === null || collecting) {
              cleanup();
              finish(c);
            }
          } catch (error) {
            cleanup();
            op_fetch_cancel(id);
            errorStream(
              c,
              state(signals, s.signal).aborted
                ? state(signals, s.signal).reason
                : failure(error),
            );
          }
        },
        cancel() {
          cleanup();
          op_fetch_cancel(id);
        },
      }, { highWaterMark: 0 });
      // Read through the stream so it still owns disturbance, locking, and errors.
      set(bodyCollectors, stream, async (reader, convert) => {
        collecting = true;
        const result = await safePromise(read(reader));
        if (!result.done && tag(result.value) !== "Uint8Array") {
          throw new TypeError("Body stream must contain Uint8Array chunks");
        }
        // Guest promise hooks can retain stream chunks; Blob storage must not alias them.
        return {
          __proto__: null,
          value: convert(
            result.done
              ? new Uint8Array()
              : new Uint8Array(result.value as Uint8Array),
          ),
        };
      });
      registerFinalizer(finalizer, stream, { id, cleanup });
    }
    const response = ObjectCreate(Response.prototype) as Response;
    set(bodies, response, {
      kind: "response",
      body: { stream, bytes: null, used: false, type: null },
      headers: h,
      status,
      statusText,
      url,
      redirected,
      type: "basic",
    });
    return response;
  } catch (error) {
    cleanup();
    op_fetch_cancel(id);
    throw state(signals, s.signal).aborted
      ? state(signals, s.signal).reason
      : uploadError ?? failure(error);
  }
}

export { fetch };