// Flags: --experimental-stream-iter
'use strict';

// Coverage tests for transform.js: abort cleanup, strategy/params options,
// early consumer exit triggering finally block.

const common = require('../common');
const assert = require('assert');
const { setTimeout } = require('timers/promises');
const {
  from,
  pull,
  bytes,
  text,
} = require('stream/iter');
const {
  compressGzip,
  decompressGzip,
  compressDeflate,
  decompressDeflate,
  compressBrotli,
  decompressBrotli,
  compressZstd,
  decompressZstd,
} = require('zlib/iter');
const { constants } = require('zlib');

async function roundTrip(input, compress, decompress) {
  return text(pull(from(input), compress, decompress));
}

// Abort mid-compression triggers finally cleanup
async function testAbortMidCompression() {
  const ac = new AbortController();
  const largeInput = 'x'.repeat(100_000);
  const compressed = pull(from(largeInput), compressGzip(),
                          { signal: ac.signal });
  const iter = compressed[Symbol.asyncIterator]();

  // Read one batch then abort
  const first = await iter.next();
  assert.strictEqual(first.done, false);
  ac.abort();
  await assert.rejects(iter.next(), { name: 'AbortError' });
}

// Early consumer exit (break from for-await) triggers finally
async function testEarlyConsumerExit() {
  const largeInput = 'y'.repeat(100_000);
  const compressed = pull(from(largeInput), compressGzip());

  // eslint-disable-next-line no-unused-vars
  for await (const batch of compressed) {
    break; // Early exit — should trigger finally block cleanup
  }
  // If we get here without hanging or crashing, cleanup worked
}

async function testPrefetchedRejectionIsHandled() {
  const reason = new Error('prefetched read failed');
  let reads = 0;
  const source = {
    __proto__: null,
    [Symbol.asyncIterator]() { return this; },
    next() {
      if (reads++ === 0) {
        return Promise.resolve({
          __proto__: null,
          done: false,
          value: [Buffer.alloc(1024 * 1024, 0x61)],
        });
      }
      return Promise.reject(reason);
    },
    return() { return { __proto__: null, done: true }; },
  };
  const controller = new AbortController();
  const transformed = compressGzip().transform(source, {
    signal: controller.signal,
  });

  await assert.rejects(bytes(transformed), (error) => error === reason);
}

async function testReturnDoesNotAwaitPrefetchedRead() {
  const never = new Promise(() => {});
  async function* source() {
    yield [Buffer.alloc(1024 * 1024, 0x61)];
    await never;
  }
  const controller = new AbortController();
  const iterator = compressGzip().transform(source(), {
    signal: controller.signal,
  })[Symbol.asyncIterator]();

  const first = await iterator.next();
  assert.strictEqual(first.done, false);
  const returned = await Promise.race([
    iterator.return().then(() => true),
    setTimeout(100, false),
  ]);
  assert.strictEqual(returned, true);
}

// Gzip with explicit strategy option
async function testGzipWithStrategy() {
  const input = 'strategy test data '.repeat(100);
  const c = compressGzip({ strategy: constants.Z_DEFAULT_STRATEGY });
  const result = await roundTrip(input, c, decompressGzip());
  assert.strictEqual(result, input);
}

// Deflate with Z_FIXED strategy
async function testDeflateWithFixedStrategy() {
  const input = 'fixed strategy '.repeat(100);
  const c = compressDeflate({ strategy: constants.Z_FIXED });
  const result = await roundTrip(input, c, decompressDeflate());
  assert.strictEqual(result, input);
}

// Brotli with custom quality param
async function testBrotliWithParams() {
  const input = 'brotli params test '.repeat(100);
  const params = { [constants.BROTLI_PARAM_QUALITY]: 5 };
  const result = await roundTrip(input, compressBrotli({ params }),
                                 decompressBrotli());
  assert.strictEqual(result, input);
}

// Zstd with custom compression level param
async function testZstdWithParams() {
  const input = 'zstd params test '.repeat(100);
  const params = { [constants.ZSTD_c_compressionLevel]: 10 };
  const result = await roundTrip(input, compressZstd({ params }),
                                 decompressZstd());
  assert.strictEqual(result, input);
}

// Gzip with custom chunkSize
async function testGzipWithChunkSize() {
  const input = 'chunk size test';
  const c = compressGzip({ chunkSize: 256 });
  const d = decompressGzip({ chunkSize: 256 });
  const result = await roundTrip(input, c, d);
  assert.strictEqual(result, input);
}

// Invalid chunkSize throws when transform is invoked
async function testInvalidChunkSize() {
  const tx = compressGzip({ chunkSize: 8 });
  await assert.rejects(
    async () => await bytes(pull(from('data'), tx)),
    { code: 'ERR_OUT_OF_RANGE' },
  );
}

Promise.all([
  testAbortMidCompression(),
  testEarlyConsumerExit(),
  testPrefetchedRejectionIsHandled(),
  testReturnDoesNotAwaitPrefetchedRead(),
  testGzipWithStrategy(),
  testDeflateWithFixedStrategy(),
  testBrotliWithParams(),
  testZstdWithParams(),
  testGzipWithChunkSize(),
  testInvalidChunkSize(),
]).then(common.mustCall());