const portPost = MessagePort.prototype.postMessage;
const portClose = MessagePort.prototype.close;
const readerClosed = descriptor(ReadableDefaultReader.prototype, "closed")!
  .get as () => Promise<void>;

function send(port: Port, value: RequestPacket | ReplyPacket): unknown {
  return apply(portPost, port, [value]);
}
function disconnect(port: Port): void {
  apply(portClose, port, []);
}
function prepareWireStream(_state: unknown, source: object): StreamRecord {
  const record: StreamRecord = { source, readable: null, writable: null };
  if (descriptor(source, "_readableStreamController")) {
    record.kind = "readable";
    record.readable = source as ReadableResource;
  } else if (descriptor(source, "_writableStreamController")) {
    record.kind = "writable";
    record.writable = source as WritableResource;
  } else if (descriptor(source, "_transformStreamController")) {
    record.kind = "transform";
    record.readable = ownValue(source, "_readable") as ReadableResource;
    record.writable = ownValue(source, "_writable") as WritableResource;
  } else throw dataCloneError("Invalid transferable stream");
  validateRecord(record);
  if (record.readable) record.readPair = op_port_pair();
  if (record.writable) record.writePair = op_port_pair();
  return record;
}
function serveStream(
  port: Port,
  resource: Reader | Writer,
  readable: boolean,
): void {
  let finished = false, disconnected = false, active = 0;
  function finish() {
    finished = true;
    if (active === 0 && !disconnected) {
      disconnected = true;
      disconnect(port);
    }
  }
  const closed = invoke(readable ? readerClosed : writerClosed, resource);
  then(closed, undefined, (error) => {
    if (!disconnected) {
      send(port, [null, false, cloneError(error)]);
      disconnected = true;
      disconnect(port);
    }
  });
  setPortCloseHandler(port, () => {
    if (!disconnected) {
      disconnected = true;
      const reason = dataCloneError("Transferred stream endpoint closed");
      markPromiseHandled(
        invoke(readable ? cancelReader : abortWriter, resource, [reason]),
      );
      disconnect(port);
    }
  });
  addListener(port, "message", (event) => {
    if (finished || disconnected) return;
    const packet = messageData(event) as RequestPacket;
    const id = packet[0], command = packet[1], value = packet[2];
    active++;
    let operation;
    try {
      if (readable) {
        operation = command === "read"
          ? invoke(read, resource)
          : invoke(cancelReader, resource, [value]);
      } else {
        operation = command === "write"
          ? invoke(write, resource, [value])
          : command === "close"
          ? invoke(closeWriter, resource)
          : invoke(abortWriter, resource, [value]);
      }
    } catch (error) {
      operation = safePromiseReject(error);
    }
    then(safePromiseResolve(operation), (result) => {
      active--;
      if (disconnected) return;
      try {
        send(port, [id, true, result]);
      } catch (error) {
        markPromiseHandled(
          invoke(readable ? cancelReader : abortWriter, resource, [error]),
        );
        send(port, [id, false, cloneError(error)]);
        finish();
        return;
      }
      if (
        finished || command === "cancel" || command === "close" ||
        command === "abort" ||
        (readable && (result as ReadResult | undefined)?.done)
      ) finish();
    }, (error) => {
      active--;
      if (!disconnected) {
        send(port, [id, false, cloneError(error)]);
        finish();
      }
    });
  });
  startPort(port);
}
function commitWireStream(
  _state: unknown,
  _source: object,
  record: StreamRecord,
): void {
  if (record.readable) {
    record.reader = acquireReader(record.readable);
    op_port_move(record.readPair![1]);
    serveStream(adoptPort(record.readPair![0]), record.reader, true);
  }
  if (record.writable) {
    record.writer = acquireWriter(record.writable);
    op_port_move(record.writePair![1]);
    serveStream(adoptPort(record.writePair![0]), record.writer, false);
  }
}
function streamClient(id: number) {
  const port = adoptPort(id),
    pending = new Map<
      number,
      { resolve(value: unknown): void; reject(reason: unknown): void }
    >();
  let next = 0, ended = false, onerror: ((reason: unknown) => void) | undefined;
  function fail(error: unknown): void {
    if (ended) return;
    ended = true;
    mapForEach(pending, (entry) => entry.reject(error));
    mapClear(pending);
    onerror?.(error);
    disconnect(port);
  }
  setPortCloseHandler(
    port,
    () => fail(dataCloneError("Transferred stream endpoint closed")),
  );
  addListener(port, "message", (event) => {
    const packet = messageData(event) as ReplyPacket;
    const id = packet[0], ok = packet[1], value = packet[2];
    if (id === null) {
      fail(value);
      return;
    }
    const entry = mapGet(pending, id);
    if (!entry) return;
    mapDelete(pending, id);
    if (ok) entry.resolve(value);
    else entry.reject(value);
  });
  startPort(port);
  return {
    set onerror(callback: (reason: unknown) => void) {
      onerror = callback;
    },
    request<C extends RequestPacket[1]>(
      command: C,
      value?: unknown,
    ): Promise<C extends "read" ? ReadResult : void> {
      if (ended) {
        return safePromiseReject(
          dataCloneError("Transferred stream endpoint closed"),
        );
      }
      const id = next++;
      return safeNewPromise<C extends "read" ? ReadResult : void>(
        (resolve, reject) => {
          mapSet(pending, id, {
            resolve: resolve as (value: unknown) => void,
            reject,
          });
          try {
            send(port, [id, command, value]);
            const retain = () =>
              then(referencePort(port), () => {
                if (!ended && mapGet(pending, id)) retain();
              });
            retain();
          } catch (error) {
            fail(error);
          }
        },
      );
    },
    close() {
      ended = true;
      disconnect(port);
    },
  };
}
function decodeWireStream(packet: StreamPacket): StreamResource {
  const kind = packet[0],
    readId = packet[1],
    writeId = packet[2],
    writableClosed = packet[3];
  let readable, writable;
  if (readId !== null) {
    const client = streamClient(readId);
    readable = new Readable({
      start(controller) {
        client.onerror = (error) => controller.error(error);
      },
      pull(controller) {
        return then(client.request("read"), (result) => {
          if (result.done) {
            controller.close();
            client.close();
          } else controller.enqueue(result.value);
        });
      },
      cancel(reason) {
        return then(
          client.request("cancel", reason),
          () => client.close(),
          (error) => {
            client.close();
            throw error;
          },
        );
      },
    }, { highWaterMark: 0 });
  }
  if (writeId !== null) {
    const client = streamClient(writeId);
    writable = new Writable(
      writableClosed ? {} : {
        start(controller) {
          client.onerror = (error) => controller.error(error);
        },
        write(chunk) {
          return client.request("write", chunk);
        },
        close() {
          return then(client.request("close"), () => client.close());
        },
        abort(reason) {
          return then(client.request("abort", reason), () => client.close());
        },
      },
    );
    if (writableClosed) {
      markPromiseHandled(invoke(closeStream, writable));
      client.close();
    }
  }
  if (kind === "readable") return readable!;
  if (kind === "writable") return writable!;
  const transform = new Transform();
  define(transform, "_readable", { value: readable });
  define(transform, "_writable", { value: writable });
  return transform;
}
registerWire<undefined, StreamPacket, StreamRecord>({
  name: "Stream",
  accepts: (value) =>
    !!(
      descriptor(value, "_readableStreamController") ||
      descriptor(value, "_writableStreamController") ||
      descriptor(value, "_transformStreamController")
    ),
  encode: (
    _state,
    _source,
    record,
  ) => [
    record.kind!,
    record.readPair?.[1] ?? null,
    record.writePair?.[1] ?? null,
    record.writableState === "closed",
  ],
  decode: decodeWireStream,
  prepare: prepareWireStream,
  validate: (_state, _source, record) => validateRecord(record),
  commit: commitWireStream,
  rollback: (_state, _source, record) => {
    const pairs = [record.readPair, record.writePair];
    for (let i = 0; i < pairs.length; i++) {
      const pair = pairs[i];
      if (pair) {
        op_port_close(pair[0]);
        op_port_close(pair[1]);
      }
    }
  },
});