'use strict';

// New Streams API - Push Stream Implementation
//
// Creates a bonded pair of writer and async iterable for push-based streaming
// with built-in backpressure.

const {
  ArrayPrototypePush,
  ObjectSetPrototypeOf,
  PromisePrototypeThen,
  PromiseReject,
  PromiseResolve,
  PromiseWithResolvers,
  SafeWeakSet,
  SymbolAsyncDispose,
  SymbolAsyncIterator,
  SymbolDispose,
} = primordials;

const {
  codes: {
    ERR_INVALID_STATE,
  },
} = require('internal/errors');

const {
  drainableProtocol,
} = require('internal/streams/iter/types');

const {
  IterResult,
  PendingRequest,
  PendingWrite,
  kNullOnceOption,
  kPushDefaultBudget,
  kResolvedPromise,
  createBatchEntry,
  onSignalAbort,
  toWriterUint8Array,
  convertChunks,
  getWriterSignal,
  parsePullArgs,
  validateBatchEntry,
  validateBudget,
} = require('internal/streams/iter/utils');
const {
  converters,
} = require('internal/streams/iter/webidl');

const {
  pullWithConsumerCleanup,
} = require('internal/streams/iter/pull');

const {
  RingBuffer,
} = require('internal/streams/iter/ringbuffer');

const consumerReturnErrors = new SafeWeakSet();

function isConsumerReturnError(error) {
  return consumerReturnErrors.has(error);
}

function raceEndWithSignal(promise, signal) {
  if (!signal) return promise;
  if (signal.aborted) return PromiseReject(signal.reason);

  const {
    promise: signaledPromise,
    resolve,
    reject,
  } = PromiseWithResolvers();
  const onAbort = () => reject(signal.reason);

  signal.addEventListener('abort', onAbort, kNullOnceOption);
  PromisePrototypeThen(
    promise,
    (value) => {
      signal.removeEventListener('abort', onAbort);
      resolve(value);
    },
    (reason) => {
      signal.removeEventListener('abort', onAbort);
      reject(reason);
    },
  );
  return signaledPromise;
}

// =============================================================================
// PushQueue - Internal Queue with Chunk-Based Backpressure
// =============================================================================

class PushQueue {
  /** Buffered chunks (each slot is from one write/writev call) */
  #slots = new RingBuffer();
  /** Pending writes waiting for buffer space */
  #pendingWrites = new RingBuffer();
  /** Pending reads waiting for data */
  #pendingReads = new RingBuffer();
  /** Pending drains waiting for backpressure to clear */
  #pendingDrains = [];
  /** Writer state: 'open' | 'closing' | 'closed' | 'errored' */
  #writerState = 'open';
  /** Consumer state: 'active' | 'returned' | 'thrown' */
  #consumerState = 'active';
  /** Error that closed the writer */
  #writerError;
  /** Error supplied by the consumer */
  #consumerError;
  /** Total bytes written */
  #bytesWritten = 0;
  /** Pending end promise (resolves when consumer drains past end sentinel) */
  #pendingEnd = null;

  /** Configuration */
  #budget;
  #backpressure;
  #signal;
  #abortHandler;
  /** Cumulative byte size of buffered batches */
  #bufferedBytes = 0;

  constructor(options = { __proto__: null }) {
    options = converters.PushStreamOptions(options, {
      __proto__: null,
      context: 'options',
    });
    const {
      budget = kPushDefaultBudget,
      backpressure = 'strict',
      signal,
    } = options;
    validateBudget(budget);
    this.#budget = budget;
    this.#backpressure = backpressure;
    this.#signal = signal;
    this.#abortHandler = undefined;

    if (this.#signal) {
      this.#abortHandler = () => {
        this.fail(this.#signal.reason);
      };
      onSignalAbort(this.#signal, this.#abortHandler);
    }
  }

  // ===========================================================================
  // Writer Methods
  // ===========================================================================

  /**
   * Check whether the slots buffer has capacity.
   * Returns null if writer is closed/errored or consumer has terminated.
   * @returns {boolean | null}
   */
  get canWrite() {
    if (this.#writerState !== 'open' || this.#consumerState !== 'active') {
      return null;
    }
    if (this.#bufferedBytes >= this.#budget) {
      return false;
    }
    return true;
  }

  get signal() {
    return this.#signal;
  }

  /**
   * Check if a sync write would be accepted.
   * @returns {boolean}
   */
  canWriteSync() {
    if (this.#writerState !== 'open') return false;
    if (this.#consumerState !== 'active') return false;
    if ((this.#backpressure === 'strict' ||
          this.#backpressure === 'unbounded') &&
        this.#bufferedBytes >= this.#budget) {
      return false;
    }
    return true;
  }

  /**
   * Write chunks synchronously if possible.
   * Returns true if write completed, false if buffer is full.
   * @returns {boolean}
   */
  writeSync(chunks) {
    if (this.#writerState !== 'open') return false;
    if (this.#consumerState !== 'active') return false;

    return this.#writeEntry(createBatchEntry(chunks));
  }

  #writeEntry(entry) {
    const batchSize = entry.byteLength;

    // Skip empty chunks -- zero-byte writes would accumulate infinitely
    // without ever triggering backpressure under a byte-budget model.
    if (batchSize === 0) return true;

    if (this.#bufferedBytes >= this.#budget) {
      switch (this.#backpressure) {
        case 'strict':
          return false;
        case 'unbounded':
          return false;
        case 'drop-oldest':
          while (this.#bufferedBytes >= this.#budget &&
                 this.#slots.length > 0) {
            const evicted = this.#slots.shift();
            this.#bufferedBytes -= evicted.byteLength;
          }
          break;
        case 'drop-newest':
          // Discard this write, but return true
          this.#bytesWritten += batchSize;
          return true;
      }
    }

    this.#slots.push(entry);
    this.#bufferedBytes += batchSize;
    this.#bytesWritten += batchSize;

    this.#resolvePendingReads();
    // After drop-oldest, evicting a large chunk may bring us under budget.
    // Resolve pending drains so writers waiting on backpressure can proceed.
    if (this.#bufferedBytes < this.#budget) {
      this.#resolvePendingDrains(true);
    }
    return true;
  }

  /**
   * Write chunks asynchronously.
   * If signal is provided, a write blocked on backpressure will reject
   * immediately when the signal fires. The cancelled write is removed from
   * pendingWrites so it does not occupy a slot. The queue itself is NOT put
   * into an error state - this is per-operation cancellation, not terminal
   * failure.
   * @returns {Promise<void>}
   */
  async writeAsync(chunks, signal) {
    // Check writer state before signal (spec order: state, then signal)
    if (this.#writerState === 'closed') {
      throw new ERR_INVALID_STATE.TypeError('Writer is closed');
    }
    if (this.#writerState === 'closing') {
      throw new ERR_INVALID_STATE.TypeError('Writer is closing');
    }
    if (this.#writerState === 'errored') {
      throw this.#writerError;
    }
    if (this.#consumerState !== 'active') {
      if (this.#consumerState === 'thrown') throw this.#consumerError;
      throw new ERR_INVALID_STATE.TypeError('Stream closed by consumer');
    }

    // Check for pre-aborted signal (after state checks per spec)
    signal?.throwIfAborted();

    const entry = createBatchEntry(chunks);
    if (this.#writeEntry(entry)) {
      return;
    }

    // Buffer is full
    switch (this.#backpressure) {
      case 'strict':
        if (this.#pendingWrites.length >= 1) {
          throw new ERR_INVALID_STATE.RangeError(
            'Backpressure violation: too many pending writes. ' +
            'Await each write() call to respect backpressure.');
        }
        return this.#createPendingWrite(entry, signal);
      case 'unbounded':
        return this.#createPendingWrite(entry, signal);
      default:
        throw new ERR_INVALID_STATE(
          'Unexpected: writeSync should have handled non-strict policy');
    }
  }

  /**
   * Create a pending write promise, optionally racing against a signal.
   * If the signal fires, the entry is removed from pendingWrites and the
   * promise rejects. Signal listeners are cleaned up on normal resolution.
   * @returns {Promise<void>}
   */
  #createPendingWrite(batch, signal) {
    const { promise, resolve, reject } = PromiseWithResolvers();
    const entry = new PendingWrite(batch, resolve, reject);
    this.#pendingWrites.push(entry);

    if (signal) {
      const onAbort = () => {
        // Remove from queue so it doesn't occupy a slot
        const idx = this.#pendingWrites.indexOf(entry);
        if (idx !== -1) {
          this.#pendingWrites.removeAt(idx);
          this.#resolvePendingReads();
        }
        reject(signal.reason);
      };

      // Wrap resolve/reject to clean up signal listener
      entry.resolve = function() {
        signal.removeEventListener('abort', onAbort);
        resolve();
      };
      entry.reject = function(reason) {
        signal.removeEventListener('abort', onAbort);
        reject(reason);
      };

      signal.addEventListener('abort', onAbort, kNullOnceOption);
    }

    return promise;
  }

  /**
   * Signal end of stream. Returns total bytes written.
   * @returns {number}
   */
  end() {
    if (this.#writerState === 'errored') {
      return -2; // Signal to reject with stored error
    }
    if (this.#writerState === 'closing') {
      return -3; // Signal to PushWriter: wait for drain to complete
    }
    if (this.#writerState === 'closed') {
      return this.#bytesWritten; // Idempotent
    }

    this.#resolvePendingDrains(false);

    // If there is no accepted work to drain, close immediately.
    if (this.#slots.length === 0 && this.#pendingWrites.length === 0) {
      this.#writerState = 'closed';
      this.#cleanup();
      this.#resolvePendingReads();
      return this.#bytesWritten;
    }

    // Accepted work remains: close after it drains and the consumer pulls EOF.
    this.#writerState = 'closing';
    return -3; // Signal to PushWriter: create deferred end promise
  }

  /**
   * Called when the consumer pulls past all accepted data while the writer is
   * closing. Transitions to 'closed' and resolves the pending end promise.
   */
  endDrained() {
    if (this.#writerState !== 'closing' ||
        this.#slots.length > 0 ||
        this.#pendingWrites.length > 0) {
      return;
    }
    this.#writerState = 'closed';
    this.#cleanup();
    if (this.#pendingEnd) {
      this.#pendingEnd.resolve(this.#bytesWritten);
      this.#pendingEnd = null;
    }
  }

  /**
   * Put queue into terminal error state.
   * No-op if errored or closed (fully drained).
   * If closing (draining), short-circuits the drain.
   */
  fail(reason) {
    if (this.#writerState === 'errored' || this.#writerState === 'closed') {
      return;
    }

    const wasClosing = this.#writerState === 'closing';
    this.#writerState = 'errored';
    this.#writerError = reason;
    this.#cleanup();
    this.#rejectPendingReads(this.#writerError);
    this.#rejectPendingDrains(this.#writerError);
    this.#rejectPendingWrites(this.#writerError);

    if (wasClosing) {
      // Short-circuit the graceful drain: reject the pending end promise
      if (this.#pendingEnd) {
        this.#pendingEnd.reject(this.#writerError);
        this.#pendingEnd = null;
      }
    }
  }

  get totalBytesWritten() {
    return this.#bytesWritten;
  }

  get error() {
    return this.#writerError;
  }

  get backpressurePolicy() {
    return this.#backpressure;
  }

  get writerState() {
    return this.#writerState;
  }

  getPendingEndPromise() {
    if (this.#pendingEnd === null) {
      const { promise, resolve, reject } = PromiseWithResolvers();
      this.#pendingEnd = { __proto__: null, promise, resolve, reject };
    }
    return this.#pendingEnd.promise;
  }

  /**
   * Wait for backpressure to clear (canWrite becomes true).
   * @returns {Promise<void>}
   */
  waitForDrain() {
    const { promise, resolve, reject } = PromiseWithResolvers();
    ArrayPrototypePush(this.#pendingDrains, new PendingRequest(resolve, reject));
    return promise;
  }

  // ===========================================================================
  // Consumer Methods
  // ===========================================================================

  async read() {
    if (this.#consumerState === 'returned') {
      return new IterResult(true, undefined);
    }
    if (this.#consumerState === 'thrown') {
      throw this.#consumerError;
    }

    // If there's data in the buffer, return it immediately
    if (this.#slots.length > 0) {
      const result = this.#drain();
      this.#resolvePendingWrites();
      return new IterResult(false, result);
    }

    // Buffer empty and writer closing = drain complete
    if (this.#writerState === 'closing') {
      this.endDrained();
      return new IterResult(true, undefined);
    }

    if (this.#writerState === 'closed') {
      return new IterResult(true, undefined);
    }

    if (this.#writerState === 'errored') {
      throw this.#writerError;
    }

    const { promise, resolve, reject } = PromiseWithResolvers();
    this.#pendingReads.push(new PendingRequest(resolve, reject));
    return promise;
  }

  consumerReturn() {
    if (this.#consumerState !== 'active') return;
    this.#consumerState = 'returned';
    const error = new ERR_INVALID_STATE.TypeError('Stream closed by consumer');
    consumerReturnErrors.add(error);
    this.#terminateWriterFromConsumer(error);
    this.#resolvePendingReads();
    // Resolve pending drains with false - no more data will be consumed
    this.#resolvePendingDrains(false);
  }

  consumerThrow(error) {
    if (this.#consumerState !== 'active') return;
    this.#consumerState = 'thrown';
    this.#consumerError = error;
    this.#terminateWriterFromConsumer(error);
    this.#rejectPendingReads(error);
    // Reject pending drains - the consumer errored
    this.#rejectPendingDrains(error);
  }

  // ===========================================================================
  // Private Methods
  // ===========================================================================

  #drain() {
    try {
      if (this.#slots.length === 1) {
        const result = validateBatchEntry(this.#slots.shift());
        this.#bufferedBytes = 0;
        return result;
      }

      const result = [];
      for (let i = 0; i < this.#slots.length; i++) {
        const batch = validateBatchEntry(this.#slots.get(i));
        for (let j = 0; j < batch.length; j++) {
          ArrayPrototypePush(result, batch[j]);
        }
      }
      this.#slots.clear();
      this.#bufferedBytes = 0;
      return result;
    } catch (error) {
      this.#slots.clear();
      this.#bufferedBytes = 0;
      this.fail(error);
      throw error;
    }
  }

  #terminateWriterFromConsumer(error) {
    this.#slots.clear();
    this.#bufferedBytes = 0;
    if (this.#writerState === 'open' || this.#writerState === 'closing') {
      this.#writerState = 'errored';
      this.#writerError = error;
    }
    this.#cleanup();
    this.#rejectPendingWrites(error);
    if (this.#pendingEnd) {
      this.#pendingEnd.reject(error);
      this.#pendingEnd = null;
    }
  }

  #resolvePendingReads() {
    while (this.#pendingReads.length > 0) {
      if (this.#consumerState === 'returned') {
        const pending = this.#pendingReads.shift();
        pending.resolve(new IterResult(true, undefined));
      } else if (this.#consumerState === 'thrown') {
        const pending = this.#pendingReads.shift();
        pending.reject(this.#consumerError);
      } else if (this.#slots.length > 0) {
        const pending = this.#pendingReads.shift();
        try {
          const result = this.#drain();
          this.#resolvePendingWrites();
          pending.resolve(new IterResult(false, result));
        } catch (error) {
          pending.reject(error);
        }
      } else if (this.#writerState === 'closing' &&
                 this.#slots.length === 0 &&
                 this.#pendingWrites.length === 0) {
        this.endDrained();
        const pending = this.#pendingReads.shift();
        pending.resolve(new IterResult(true, undefined));
      } else if (this.#writerState === 'closed') {
        const pending = this.#pendingReads.shift();
        pending.resolve(new IterResult(true, undefined));
      } else if (this.#writerState === 'errored') {
        const pending = this.#pendingReads.shift();
        pending.reject(this.#writerError);
      } else {
        break;
      }
    }
  }

  #resolvePendingWrites() {
    while (this.#pendingWrites.length > 0 &&
          this.#bufferedBytes < this.#budget) {
      const pending = this.#pendingWrites.shift();
      try {
        validateBatchEntry(pending.batch);
        this.#slots.push(pending.batch);
        this.#bufferedBytes += pending.batch.byteLength;
        this.#bytesWritten += pending.batch.byteLength;
        pending.resolve();
      } catch (error) {
        pending.reject(error);
      }
    }

    if (this.#bufferedBytes < this.#budget) {
      this.#resolvePendingDrains(true);
    }
  }

  #resolvePendingDrains(canWrite) {
    const drains = this.#pendingDrains;
    this.#pendingDrains = [];
    for (let i = 0; i < drains.length; i++) {
      drains[i].resolve(canWrite);
    }
  }

  #rejectPendingDrains(error) {
    const drains = this.#pendingDrains;
    this.#pendingDrains = [];
    for (let i = 0; i < drains.length; i++) {
      drains[i].reject(error);
    }
  }

  #rejectPendingReads(error) {
    while (this.#pendingReads.length > 0) {
      this.#pendingReads.shift().reject(error);
    }
  }

  #rejectPendingWrites(error) {
    while (this.#pendingWrites.length > 0) {
      this.#pendingWrites.shift().reject(error);
    }
  }

  #cleanup() {
    if (this.#signal && this.#abortHandler) {
      this.#signal.removeEventListener('abort', this.#abortHandler);
      this.#abortHandler = undefined;
    }
  }
}

// =============================================================================
// PushWriter Implementation
// =============================================================================

class PushWriter {
  #queue;

  constructor(queue) {
    this.#queue = queue;
  }

  [drainableProtocol]() {
    const canWrite = this.canWrite;
    if (canWrite === null) return null;
    if (canWrite) return PromiseResolve(true);
    return this.#queue.waitForDrain();
  }

  get canWrite() {
    return this.#queue.canWrite;
  }

  write(chunk, options) {
    const bytes = toWriterUint8Array(chunk);
    const signal = getWriterSignal(options);
    if (!signal && this.#queue.canWriteSync()) {
      this.#queue.writeSync([bytes]);
      return kResolvedPromise;
    }
    return this.#queue.writeAsync([bytes], signal);
  }

  writev(chunks, options) {
    const bytes = convertChunks(chunks);
    const signal = getWriterSignal(options);
    if (!signal && this.#queue.writeSync(bytes)) {
      return kResolvedPromise;
    }
    return this.#queue.writeAsync(bytes, signal);
  }

  writeSync(chunk) {
    const bytes = toWriterUint8Array(chunk);
    return this.#queue.writeSync([bytes]);
  }

  writevSync(chunks) {
    const bytes = convertChunks(chunks);
    return this.#queue.writeSync(bytes);
  }

  end(options) {
    const signal = getWriterSignal(options);
    const state = this.#queue.writerState;
    if (state === 'errored') return PromiseReject(this.#queue.error);
    if (state === 'closed') {
      return PromiseResolve(this.#queue.totalBytesWritten);
    }
    if (signal?.aborted) return PromiseReject(signal.reason);

    const result = this.#queue.end();
    if (result === -2) {
      // Errored: reject with stored error
      return PromiseReject(this.#queue.error);
    }
    if (result === -3) {
      return raceEndWithSignal(this.#queue.getPendingEndPromise(), signal);
    }
    // >= 0: byte count (immediate close or idempotent)
    return PromiseResolve(result);
  }

  endSync() {
    const result = this.#queue.end();
    if (result === -2) return -1; // Errored
    if (result === -3) return -1; // Buffer not empty, can't wait
    return result;
  }

  fail(reason) {
    this.#queue.fail(reason);
  }

  [SymbolAsyncDispose]() {
    const state = this.#queue.writerState;
    if (state === 'closing') {
      return this.#queue.getPendingEndPromise();
    }
    if (state === 'open') {
      this.fail();
    }
    return PromiseResolve();
  }

  [SymbolDispose]() {
    this.fail();
  }
}

// =============================================================================
// Readable Implementation
// =============================================================================

function createReadable(queue) {
  return {
    __proto__: null,
    [SymbolAsyncIterator]() {
      return ObjectSetPrototypeOf({
        async next() {
          return queue.read();
        },
        async return() {
          queue.consumerReturn();
          return new IterResult(true, undefined);
        },
        async throw(error) {
          queue.consumerThrow(error);
          throw error;
        },
      }, null);
    },
  };
}

// =============================================================================
// Stream.push() Factory
// =============================================================================

function parseArgs(args) {
  const result = parsePullArgs(args);
  // PushQueue constructor requires a non-undefined options object.
  if (result.options === undefined) {
    result.options = { __proto__: null };
  }
  return result;
}

/**
 * Create a push stream with optional transforms.
 * @param {...(Function|object)} args - Transforms, then options (optional)
 * @returns {{ writer: Writer, readable: AsyncIterable<Uint8Array[]> }}
 */
function push(...args) {
  const { transforms, options } = parseArgs(args);

  const queue = new PushQueue(options);
  const writer = new PushWriter(queue);
  const rawReadable = createReadable(queue);

  // Apply transforms lazily if provided
  let readable;
  if (transforms.length > 0) {
    readable = pullWithConsumerCleanup(
      rawReadable, transforms, queue.signal);
  } else {
    readable = rawReadable;
  }

  return { __proto__: null, writer, readable };
}

module.exports = {
  isConsumerReturnError,
  push,
};