const Readable = globalThis.ReadableStream;
const Writable = globalThis.WritableStream;
const Transform = globalThis.TransformStream;
const ReadableDefaultReader = globalThis.ReadableStreamDefaultReader;
const WritableDefaultWriter = globalThis.WritableStreamDefaultWriter;
const getReader = Readable.prototype.getReader;
const read = ReadableDefaultReader.prototype.read;
const cancelReader = ReadableDefaultReader.prototype.cancel;
const getWriter = Writable.prototype.getWriter;
const write = WritableDefaultWriter.prototype.write;
const closeWriter = WritableDefaultWriter.prototype.close;
const closeStream = Writable.prototype.close;
const abortWriter = WritableDefaultWriter.prototype.abort;
const writerClosed = Object.getOwnPropertyDescriptor(
  WritableDefaultWriter.prototype,
  "closed",
)!
  .get as () => Promise<void>;
const isExtensible = Object.isExtensible;
const transferStates = new WeakMap<object, StreamRecord>();
function invoke<F extends (...args: never[]) => unknown>(
  fn: F,
  receiver: unknown,
  args: unknown[] = [],
): ReturnType<F> {
  return apply(fn, receiver, args) as ReturnType<F>;
}
function cloneRejection<T>(promise: Promise<T>): Promise<T> {
  return then(promise, undefined, (error) => {
    throw cloneError(error);
  });
}
function cloneError(error: unknown): unknown {
  try {
    return cloneValue(error);
  } catch (cloneError) {
    return cloneError;
  }
}
function cancelAfterFailure(
  record: StreamRecord,
  error: unknown,
): Promise<never> {
  let cancellation;
  try {
    cancellation = invoke(cancelReader, record.reader, [error]);
  } catch {
    throw error;
  }
  return then(cancellation, () => {
    throw error;
  }, () => {
    throw error;
  });
}
function makeReadable(record: StreamRecord): ReadableResource {
  const source = {
    pull(controller: ReadController) {
      const reading = then(invoke(read, record.reader), (result) => {
        if (result.done) controller.close();
        else controller.enqueue(cloneValue(result.value));
      });
      return then(
        reading,
        undefined,
        (error) => cancelAfterFailure(record, cloneError(error)),
      );
    },
    cancel(reason: unknown) {
      let copied;
      try {
        copied = cloneValue(reason);
      } catch (error) {
        return cancelAfterFailure(record, error);
      }
      return cloneRejection(invoke(cancelReader, record.reader, [copied]));
    },
  };
  return new Readable(source, { highWaterMark: 0 });
}
function abortAfterFailure(
  record: StreamRecord,
  error: unknown,
): Promise<never> {
  let abort;
  try {
    abort = invoke(abortWriter, record.writer, [error]);
  } catch {
    throw error;
  }
  return then(abort, () => {
    throw error;
  }, () => {
    throw error;
  });
}
function makeWritable(record: StreamRecord): WritableResource {
  return new Writable({
    start(controller) {
      record.controller = controller;
    },
    write(chunk) {
      let copied;
      try {
        copied = cloneValue(chunk);
      } catch (error) {
        return abortAfterFailure(record, error);
      }
      return cloneRejection(invoke(write, record.writer, [copied]));
    },
    close() {
      if (record.writableState === "closed") {
        return invoke(writerClosed, record.writer);
      }
      return cloneRejection(invoke(closeWriter, record.writer));
    },
    abort(reason) {
      let copied;
      try {
        copied = cloneValue(reason);
      } catch (error) {
        return abortAfterFailure(record, error);
      }
      return cloneRejection(invoke(abortWriter, record.writer, [copied]));
    },
  }, { highWaterMark: 1 });
}
function prepareMarked(_state: unknown, source: object): StreamResource {
  const record: StreamRecord = {
    source,
    readable: null,
    writable: null,
    reader: null,
    writer: null,
  };
  if (descriptor(source, "_readableStreamController")) {
    record.readable = source as ReadableResource;
    record.kind = "readable";
  } else if (descriptor(source, "_writableStreamController")) {
    record.writable = source as WritableResource;
    record.kind = "writable";
  } else if (descriptor(source, "_transformStreamController")) {
    record.readable = ownValue(source, "_readable") as ReadableResource;
    record.writable = ownValue(source, "_writable") as WritableResource;
    record.kind = "transform";
  } else {
    throw dataCloneError("This object is not a transferable stream");
  }
  validateRecord(record);
  if (record.readable) record.readableClone = makeReadable(record);
  if (record.writable) record.writableClone = makeWritable(record);
  let destination: StreamResource | undefined = record.readableClone;
  if (record.kind === "transform") {
    destination = new Transform();
    define(destination, "_readable", { value: record.readableClone });
    define(destination, "_writable", { value: record.writableClone });
  } else if (record.kind === "writable") destination = record.writableClone;
  record.destination = destination;
  set(transferStates, destination!, record);
  return destination!;
}
function validateMarked(
  _state: unknown,
  _source: object,
  destination: StreamResource,
): void {
  validateRecord(get(transferStates, destination)!);
}
function validateRecord(record: StreamRecord): void {
  if (record.readable) {
    if (!descriptor(record.readable, "_readableStreamController")) {
      throw dataCloneError("Invalid readable stream state");
    }
    validateSlot(record.readable, "_reader");
  }
  if (record.writable) {
    if (!descriptor(record.writable, "_writableStreamController")) {
      throw dataCloneError("Invalid writable stream state");
    }
    validateSlot(record.writable, "_writer");
  }
  if (record.writable) {
    record.writableState = ownValue(record.writable, "_state");
  }
  if (
    (record.readable && getPrototype(record.readable) !== Readable.prototype &&
      !isExtensible(record.readable)) ||
    (record.writable && getPrototype(record.writable) !== Writable.prototype &&
      !isExtensible(record.writable))
  ) {
    throw dataCloneError(
      "not yet implemented: transferring this non-extensible stream",
    );
  }
}
function commitMarked(
  _state: unknown,
  _source: object,
  destination: StreamResource,
): void {
  const record = get(transferStates, destination)!;
  if (record.readable) record.reader = acquireReader(record.readable);
  if (record.writable) record.writer = acquireWriter(record.writable);
  if (record.writable) {
    if (record.writableState === "closed") {
      const closing = invoke(closeStream, record.writableClone);
      markPromiseHandled(closing);
      then(closing, undefined, (error) => {
        try {
          record.controller!.error(cloneError(error));
        } catch (cloneError) {
          record.controller!.error(cloneError);
        }
      });
    }
    then(invoke(writerClosed, record.writer), undefined, (error) => {
      try {
        record.controller!.error(cloneValue(error));
      } catch (cloneError) {
        record.controller!.error(cloneError);
      }
    });
  }
}
function validateSlot(stream: object, key: PropertyKey): void {
  const property = descriptor(stream, key);
  if (property?.value !== undefined) {
    throw dataCloneError("Cannot transfer a locked stream");
  }
  if (!property || !("value" in property) || !property.writable) {
    throw dataCloneError(
      "not yet implemented: transferring a stream with immutable internal state",
    );
  }
}
function ownValue(object: object, key: PropertyKey): unknown {
  const property = descriptor(object, key);
  return property && "value" in property ? property.value : undefined;
}
function acquireReader(stream: ReadableResource): Reader {
  const prototype = getPrototype(stream);
  if (prototype === Readable.prototype) return invoke(getReader, stream);
  setPrototype(stream, Readable.prototype);
  try {
    return invoke(getReader, stream);
  } finally {
    setPrototype(stream, prototype);
  }
}
function acquireWriter(stream: WritableResource): Writer {
  const prototype = getPrototype(stream);
  if (prototype === Writable.prototype) return invoke(getWriter, stream);
  setPrototype(stream, Writable.prototype);
  try {
    return invoke(getWriter, stream);
  } finally {
    setPrototype(stream, prototype);
  }
}
registerMarkedTransferable(prepareMarked, validateMarked, commitMarked);