// Ported from https://github.com/nodejs/undici/pull/907

'use strict';

const assert = require('node:assert');
const { Readable } = require('node:stream');
const {
  RequestAbortedError,
  NotSupportedError,
  InvalidArgumentError,
  AbortError,
} = require('../core/errors');
const util = require('../core/util');
const { ReadableStreamFrom } = require('../core/util');

const kConsume = Symbol('kConsume');
const kReading = Symbol('kReading');
const kBody = Symbol('kBody');
const kAbort = Symbol('kAbort');
const kContentType = Symbol('kContentType');
const kContentLength = Symbol('kContentLength');
const kUsed = Symbol('kUsed');
const kBytesRead = Symbol('kBytesRead');

const noop = () => {};

/**
 * @class
 * @extends {Readable}
 * @see https://fetch.spec.whatwg.org/#body
 */
class BodyReadable extends Readable {
  /**
   * @param {object} opts
   * @param {(this: Readable, size: number) => void} opts.resume
   * @param {() => (void | null)} opts.abort
   * @param {string} [opts.contentType = '']
   * @param {number} [opts.contentLength]
   * @param {number} [opts.highWaterMark = 64 * 1024]
   */
  constructor({
    resume,
    abort,
    contentType = '',
    contentLength,
    highWaterMark = 64 * 1024, // Same as nodejs fs streams.
  }) {
    super({
      autoDestroy: true,
      read: resume,
      highWaterMark,
    });

    this._readableState.dataEmitted = false;

    this[kAbort] = abort;

    /**
     * @type {Consume | null}
     */
    this[kConsume] = null;
    this[kBytesRead] = 0;
    /**
     * @type {ReadableStream|null}
     */
    this[kBody] = null;
    this[kUsed] = false;
    this[kContentType] = contentType;
    this[kContentLength] =
      Number.isFinite(contentLength) ? contentLength : null;

    // Is stream being consumed through Readable API?
    // This is an optimization so that we avoid checking
    // for 'data' and 'readable' listeners in the hot path
    // inside push().
    this[kReading] = false;
  }

  /**
   * @param {Error|null} err
   * @param {(error:(Error|null)) => void} callback
   * @returns {void}
   */
  _destroy(err, callback) {
    if (!err && !this._readableState.endEmitted) {
      err = new RequestAbortedError();
    }

    if (err) {
      this[kAbort]();
    }

    // Workaround for Node "bug". If the stream is destroyed in same
    // tick as it is created, then a user who is waiting for a
    // promise (i.e micro tick) for installing an 'error' listener will
    // never get a chance and will always encounter an unhandled exception.
    if (!this[kUsed]) {
      setImmediate(() => {
        callback(err);
      });
    } else {
      callback(err);
    }
  }

  /**
   * @param {string} event
   * @param {(...args: any[]) => void} listener
   * @returns {this}
   */
  on(event, listener) {
    if (event === 'data' || event === 'readable') {
      this[kReading] = true;
      this[kUsed] = true;
    }
    return super.on(event, listener);
  }

  /**
   * @param {string} event
   * @param {(...args: any[]) => void} listener
   * @returns {this}
   */
  addListener(event, listener) {
    return this.on(event, listener);
  }

  /**
   * @param {string|symbol} event
   * @param {(...args: any[]) => void} listener
   * @returns {this}
   */
  off(event, listener) {
    const ret = super.off(event, listener);
    if (event === 'data' || event === 'readable') {
      this[kReading] =
        this.listenerCount('data') > 0 || this.listenerCount('readable') > 0;
    }
    return ret;
  }

  /**
   * @param {string|symbol} event
   * @param {(...args: any[]) => void} listener
   * @returns {this}
   */
  removeListener(event, listener) {
    return this.off(event, listener);
  }

  /**
   * @param {Buffer|null} chunk
   * @returns {boolean}
   */
  push(chunk) {
    this[kBytesRead] += chunk ? chunk.length : 0;

    if (this[kConsume] && chunk !== null) {
      consumePush(this[kConsume], chunk);
      return this[kReading] ? super.push(chunk) : true;
    }
    return super.push(chunk);
  }

  /**
   * Consumes and returns the body as a string.
   *
   * @see https://fetch.spec.whatwg.org/#dom-body-text
   * @returns {Promise<string>}
   */
  text() {
    return consume(this, 'text');
  }

  /**
   * Consumes and returns the body as a JavaScript Object.
   *
   * @see https://fetch.spec.whatwg.org/#dom-body-json
   * @returns {Promise<unknown>}
   */
  json() {
    return consume(this, 'json');
  }

  /**
   * Consumes and returns the body as a Blob
   *
   * @see https://fetch.spec.whatwg.org/#dom-body-blob
   * @returns {Promise<Blob>}
   */
  blob() {
    return consume(this, 'blob');
  }

  /**
   * Consumes and returns the body as an Uint8Array.
   *
   * @see https://fetch.spec.whatwg.org/#dom-body-bytes
   * @returns {Promise<Uint8Array>}
   */
  bytes() {
    return consume(this, 'bytes');
  }

  /**
   * Consumes and returns the body as an ArrayBuffer.
   *
   * @see https://fetch.spec.whatwg.org/#dom-body-arraybuffer
   * @returns {Promise<ArrayBuffer>}
   */
  arrayBuffer() {
    return consume(this, 'arrayBuffer');
  }

  /**
   * Not implemented
   *
   * @see https://fetch.spec.whatwg.org/#dom-body-formdata
   * @throws {NotSupportedError}
   */
  async formData() {
    // TODO: Implement.
    throw new NotSupportedError();
  }

  /**
   * Returns true if the body is not null and the body has been consumed.
   * Otherwise, returns false.
   *
   * @see https://fetch.spec.whatwg.org/#dom-body-bodyused
   * @readonly
   * @returns {boolean}
   */
  get bodyUsed() {
    return util.isDisturbed(this);
  }

  /**
   * @see https://fetch.spec.whatwg.org/#dom-body-body
   * @readonly
   * @returns {ReadableStream}
   */
  get body() {
    if (!this[kBody]) {
      this[kBody] = ReadableStreamFrom(this);
      if (this[kConsume]) {
        // TODO: Is this the best way to force a lock?
        this[kBody].getReader(); // Ensure stream is locked.
        assert(this[kBody].locked);
      }
    }
    return this[kBody];
  }

  /**
   * Dumps the response body by reading `limit` number of bytes.
   * @param {object} opts
   * @param {number} [opts.limit = 131072] Number of bytes to read.
   * @param {AbortSignal} [opts.signal] An AbortSignal to cancel the dump.
   * @returns {Promise<null>}
   */
  async dump(opts) {
    const signal = opts?.signal;

    if (
      signal != null &&
      (typeof signal !== 'object' || !('aborted' in signal))
    ) {
      throw new InvalidArgumentError('signal must be an AbortSignal');
    }

    const limit =
      opts?.limit && Number.isFinite(opts.limit) ? opts.limit : 128 * 1024;

    signal?.throwIfAborted();

    if (this._readableState.closeEmitted) {
      return null;
    }

    return await new Promise((resolve, reject) => {
      if (
        (this[kContentLength] && this[kContentLength] > limit) ||
        this[kBytesRead] > limit
      ) {
        this.destroy(new AbortError());
      }

      if (signal) {
        const onAbort = () => {
          this.destroy(signal.reason ?? new AbortError());
        };
        signal.addEventListener('abort', onAbort);
        this.on('close', function () {
          signal.removeEventListener('abort', onAbort);
          if (signal.aborted) {
            reject(signal.reason ?? new AbortError());
          } else {
            resolve(null);
          }
        });
      } else {
        this.on('close', resolve);
      }

      this.on('error', noop)
        .on('data', () => {
          if (this[kBytesRead] > limit) {
            this.destroy();
          }
        })
        .resume();
    });
  }

  /**
   * @param {BufferEncoding} encoding
   * @returns {this}
   */
  setEncoding(encoding) {
    if (Buffer.isEncoding(encoding)) {
      this._readableState.encoding = encoding;
    }
    return this;
  }
}

/**
 * @see https://streams.spec.whatwg.org/#readablestream-locked
 * @param {BodyReadable} bodyReadable
 * @returns {boolean}
 */
function isLocked(bodyReadable) {
  // Consume is an implicit lock.
  return (
    bodyReadable[kBody]?.locked === true || bodyReadable[kConsume] !== null
  );
}

/**
 * @see https://fetch.spec.whatwg.org/#body-unusable
 * @param {BodyReadable} bodyReadable
 * @returns {boolean}
 */
function isUnusable(bodyReadable) {
  return util.isDisturbed(bodyReadable) || isLocked(bodyReadable);
}

/**
 * @typedef {object} Consume
 * @property {string} type
 * @property {BodyReadable} stream
 * @property {((value?: any) => void)} resolve
 * @property {((err: Error) => void)} reject
 * @property {number} length
 * @property {Buffer[]} body
 */

/**
 * @param {BodyReadable} stream
 * @param {string} type
 * @returns {Promise<any>}
 */
function consume(stream, type) {
  assert(!stream[kConsume]);

  return new Promise((resolve, reject) => {
    if (isUnusable(stream)) {
      const rState = stream._readableState;
      if (rState.destroyed && rState.closeEmitted === false) {
        stream
          .on('error', (err) => {
            reject(err);
          })
          .on('close', () => {
            reject(new TypeError('unusable'));
          });
      } else {
        reject(rState.errored ?? new TypeError('unusable'));
      }
    } else {
      queueMicrotask(() => {
        stream[kConsume] = {
          type,
          stream,
          resolve,
          reject,
          length: 0,
          body: [],
        };

        stream
          .on('error', function (err) {
            consumeFinish(this[kConsume], err);
          })
          .on('close', function () {
            if (this[kConsume].body !== null) {
              consumeFinish(this[kConsume], new RequestAbortedError());
            }
          });

        consumeStart(stream[kConsume]);
      });
    }
  });
}

/**
 * @param {Consume} consume
 * @returns {void}
 */
function consumeStart(consume) {
  if (consume.body === null) {
    return;
  }

  const { _readableState: state } = consume.stream;

  if (state.bufferIndex) {
    const start = state.bufferIndex;
    const end = state.buffer.length;
    for (let n = start; n < end; n++) {
      consumePush(consume, state.buffer[n]);
    }
  } else {
    for (const chunk of state.buffer) {
      consumePush(consume, chunk);
    }
  }

  if (state.endEmitted) {
    consumeEnd(this[kConsume], this._readableState.encoding);
  } else {
    consume.stream.on('end', function () {
      consumeEnd(this[kConsume], this._readableState.encoding);
    });
  }

  consume.stream.resume();

  while (consume.stream.read() != null) {
    // Loop
  }
}

/**
 * @param {Buffer[]} chunks
 * @param {number} length
 * @param {BufferEncoding} encoding
 * @returns {string}
 */
function chunksDecode(chunks, length, encoding) {
  if (chunks.length === 0 || length === 0) {
    return '';
  }
  const buffer =
    chunks.length === 1 ? chunks[0] : Buffer.concat(chunks, length);
  const bufferLength = buffer.length;

  // Skip BOM.
  const start =
    (
      bufferLength > 2 &&
      buffer[0] === 0xef &&
      buffer[1] === 0xbb &&
      buffer[2] === 0xbf
    ) ?
      3
    : 0;
  if (!encoding || encoding === 'utf8' || encoding === 'utf-8') {
    return buffer.utf8Slice(start, bufferLength);
  } else {
    return buffer.subarray(start, bufferLength).toString(encoding);
  }
}

/**
 * @param {Buffer[]} chunks
 * @param {number} length
 * @returns {Uint8Array}
 */
function chunksConcat(chunks, length) {
  if (chunks.length === 0 || length === 0) {
    return new Uint8Array(0);
  }
  if (chunks.length === 1) {
    // fast-path
    return new Uint8Array(chunks[0]);
  }
  const buffer = new Uint8Array(Buffer.allocUnsafeSlow(length).buffer);

  let offset = 0;
  for (let i = 0; i < chunks.length; ++i) {
    const chunk = chunks[i];
    buffer.set(chunk, offset);
    offset += chunk.length;
  }

  return buffer;
}

/**
 * @param {Consume} consume
 * @param {BufferEncoding} encoding
 * @returns {void}
 */
function consumeEnd(consume, encoding) {
  const { type, body, resolve, stream, length } = consume;

  try {
    if (type === 'text') {
      resolve(chunksDecode(body, length, encoding));
    } else if (type === 'json') {
      resolve(JSON.parse(chunksDecode(body, length, encoding)));
    } else if (type === 'arrayBuffer') {
      resolve(chunksConcat(body, length).buffer);
    } else if (type === 'blob') {
      resolve(new Blob(body, { type: stream[kContentType] }));
    } else if (type === 'bytes') {
      resolve(chunksConcat(body, length));
    }

    consumeFinish(consume);
  } catch (err) {
    stream.destroy(err);
  }
}

/**
 * @param {Consume} consume
 * @param {Buffer} chunk
 * @returns {void}
 */
function consumePush(consume, chunk) {
  consume.length += chunk.length;
  consume.body.push(chunk);
}

/**
 * @param {Consume} consume
 * @param {Error} [err]
 * @returns {void}
 */
function consumeFinish(consume, err) {
  if (consume.body === null) {
    return;
  }

  if (err) {
    consume.reject(err);
  } else {
    consume.resolve();
  }

  // Reset the consume object to allow for garbage collection.
  consume.type = null;
  consume.stream = null;
  consume.resolve = null;
  consume.reject = null;
  consume.length = 0;
  consume.body = null;
}

module.exports = {
  Readable: BodyReadable,
  chunksDecode,
};