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

const common = require('../common');
const assert = require('assert');
const {
  dump,
  from,
  share,
  text,
} = require('stream/iter');

const { setTimeout, setImmediate } = require('timers/promises');

// =============================================================================
// Async share()
// =============================================================================

async function testBasicShare() {
  const shared = share(from('hello shared'));

  const consumer = shared.pull();
  const data = await text(consumer);
  assert.strictEqual(data, 'hello shared');
}

async function testShareMultipleConsumers() {
  async function* gen() {
    yield [new TextEncoder().encode('chunk1')];
    yield [new TextEncoder().encode('chunk2')];
    yield [new TextEncoder().encode('chunk3')];
  }

  const shared = share(gen(), { budget: 16384 });

  const c1 = shared.pull();
  const c2 = shared.pull();

  assert.strictEqual(shared.consumerCount, 2);

  const [data1, data2] = await Promise.all([
    text(c1),
    text(c2),
  ]);

  assert.strictEqual(data1, 'chunk1chunk2chunk3');
  assert.strictEqual(data2, 'chunk1chunk2chunk3');
}

async function testShareConsumerCount() {
  const shared = share(from('data'));

  assert.strictEqual(shared.consumerCount, 0);

  const c1 = shared.pull();
  assert.strictEqual(shared.consumerCount, 1);

  const c2 = shared.pull();
  assert.strictEqual(shared.consumerCount, 2);

  // Cancel detaches all consumers
  shared.cancel();
  assert.strictEqual(shared.consumerCount, 0);

  // Both should complete immediately
  const [data1, data2] = await Promise.all([
    text(c1),
    text(c2),
  ]);
  assert.strictEqual(data1, '');
  assert.strictEqual(data2, '');
}

async function testShareCancel() {
  const shared = share(from('data'));
  const consumer = shared.pull();

  shared.cancel();
  assert.strictEqual(shared.consumerCount, 0);

  const batches = [];
  for await (const batch of consumer) {
    batches.push(batch);
  }
  assert.strictEqual(batches.length, 0);
}

async function testShareCancelMidIteration() {
  // Verify that cancel during iteration stops data flow
  let sourceReturnCalled = false;
  const enc = new TextEncoder();
  async function* gen() {
    try {
      yield [enc.encode('a')];
      yield [enc.encode('b')];
      yield [enc.encode('c')];
    } finally {
      sourceReturnCalled = true;
    }
  }
  const shared = share(gen(), { budget: 16384 });
  const consumer = shared.pull();

  const items = [];
  for await (const batch of consumer) {
    for (const chunk of batch) {
      items.push(new TextDecoder().decode(chunk));
    }
    // Cancel after first batch
    shared.cancel();
  }
  assert.strictEqual(items.length, 1);
  assert.strictEqual(items[0], 'a');

  await setImmediate();
  assert.strictEqual(sourceReturnCalled, true);
}

async function testShareCancelWithReason() {
  const shared = share(from('data'));
  const consumer = shared.pull();

  shared.cancel(new Error('share cancelled'));

  await assert.rejects(
    async () => {
      // eslint-disable-next-line no-unused-vars
      for await (const _ of consumer) {
        assert.fail('Should not reach here');
      }
    },
    { message: 'share cancelled' },
  );
}

async function testShareCancelWithFalsyReason() {
  for (const reason of [0, '', false, null]) {
    const shared = share(from('data'));
    const iterator = shared.pull()[Symbol.asyncIterator]();

    shared.cancel(reason);

    await assert.rejects(iterator.next(), (error) => error === reason);
  }
}

async function testShareCancelWhileSourcePullPending() {
  const noReason = { __proto__: null };

  for (const reason of [noReason, 0]) {
    const sourceStarted = Promise.withResolvers();
    const sourceNext = Promise.withResolvers();
    let nextCalls = 0;
    let returnCalls = 0;
    const source = {
      __proto__: null,
      [Symbol.asyncIterator]() {
        return {
          __proto__: null,
          next() {
            nextCalls++;
            sourceStarted.resolve();
            return sourceNext.promise;
          },
          async return() {
            returnCalls++;
            return { __proto__: null, done: true, value: undefined };
          },
        };
      },
    };
    const shared = share(source);
    const iterator = shared.pull()[Symbol.asyncIterator]();
    const read = iterator.next().then(
      (value) => ({ __proto__: null, rejected: false, value }),
      (error) => ({ __proto__: null, rejected: true, error }),
    );

    await sourceStarted.promise;
    if (reason === noReason) {
      shared.cancel();
    } else {
      shared.cancel(reason);
    }

    const timedOut = { __proto__: null };
    const outcome = await Promise.race([
      read,
      setImmediate(timedOut),
    ]);
    assert.notStrictEqual(outcome, timedOut);
    if (reason === noReason) {
      assert.strictEqual(outcome.rejected, false);
      assert.strictEqual(outcome.value.done, true);
    } else {
      assert.strictEqual(outcome.rejected, true);
      assert.strictEqual(outcome.error, reason);
    }
    assert.strictEqual(nextCalls, 1);

    sourceNext.resolve({
      __proto__: null,
      done: false,
      value: [Uint8Array.of(1)],
    });
    await setImmediate();
    assert.strictEqual(returnCalls, 1);
  }
}

async function testShareAbortSignal() {
  const ac = new AbortController();
  const reason = new Error('share aborted');
  const enc = new TextEncoder();
  async function* source() {
    yield [enc.encode('a')];
    yield [enc.encode('b')];
  }
  const shared = share(source(), {
    budget: 16384,
    backpressure: 'unbounded',
    signal: ac.signal,
  });
  const fast = shared.pull()[Symbol.asyncIterator]();
  shared.pull();

  await fast.next();
  const read = fast.next();
  const rejected = assert.rejects(read, (error) => error === reason);
  ac.abort(reason);

  await rejected;
}

async function testShareAbortSignalWhileSourcePullPending() {
  const ac = new AbortController();
  const {
    promise: resumePromise,
    resolve: resume,
  } = Promise.withResolvers();
  const {
    promise: sourceStartedPromise,
    resolve: sourceStarted,
  } = Promise.withResolvers();

  const source = {
    __proto__: null,
    [Symbol.asyncIterator]() {
      return {
        __proto__: null,
        async next() {
          sourceStarted();
          await resumePromise;
          return { __proto__: null, done: true, value: undefined };
        },
      };
    },
  };

  const shared = share(source, { signal: ac.signal });
  const iter1 = shared.pull()[Symbol.asyncIterator]();
  const iter2 = shared.pull()[Symbol.asyncIterator]();
  const read1 = iter1.next();
  const read2 = iter2.next();
  const rejected1 = assert.rejects(read1, { name: 'AbortError' });
  const rejected2 = assert.rejects(read2, { name: 'AbortError' });

  await sourceStartedPromise;
  ac.abort();
  resume();

  await Promise.all([rejected1, rejected2]);
}

async function testSharePullAbortSignalRejectsPendingNext() {
  const ac = new AbortController();
  const reason = new Error('pull aborted');
  const shared = share(
    // eslint-disable-next-line require-yield
    (async function* never() {
      await new Promise(() => {});
    })(),
  );
  const iter = shared.pull({ signal: ac.signal })[Symbol.asyncIterator]();

  const pendingNext = iter.next();
  const rejected = assert.rejects(pendingNext, (error) => error === reason);
  ac.abort(reason);

  await rejected;
  shared.cancel();
}

async function testSharePullPreAbortedSignalDoesNotAddConsumer() {
  const reason = new Error('already aborted');
  const signal = AbortSignal.abort(reason);
  const shared = share(from('data'));
  const iter = shared.pull({ signal })[Symbol.asyncIterator]();

  assert.strictEqual(shared.consumerCount, 0);
  await assert.rejects(iter.next(), (error) => error === reason);
  assert.strictEqual(shared.consumerCount, 0);
}

async function testShareAlreadyAborted() {
  const shared = share(from('data'), { signal: AbortSignal.abort() });
  const consumer = shared.pull();

  await assert.rejects(async () => {
    // eslint-disable-next-line no-unused-vars
    for await (const _ of consumer) {
      assert.fail('Should not reach here');
    }
  }, { name: 'AbortError' });
}

// =============================================================================
// Source error propagation
// =============================================================================

async function testShareSourceError() {
  async function* failingSource() {
    yield [new TextEncoder().encode('a')];
    throw new Error('share source boom');
  }
  const shared = share(failingSource());
  const c1 = shared.pull();
  const c2 = shared.pull();

  await assert.rejects(async () => {
    await dump(c1);
  }, { message: 'share source boom' });
  await assert.rejects(async () => {
    await dump(c2);
  }, { message: 'share source boom' });
}

async function testShareSourceErrorFollowsBufferedData() {
  const reason = new Error('share source boom');
  async function* failingSource() {
    yield [Uint8Array.of(1)];
    throw reason;
  }

  const shared = share(failingSource());
  const fast = shared.pull()[Symbol.asyncIterator]();
  const slow = shared.pull()[Symbol.asyncIterator]();
  const returned = shared.pull()[Symbol.asyncIterator]();
  await returned.return();

  assert.deepStrictEqual((await fast.next()).value, [Uint8Array.of(1)]);
  await assert.rejects(fast.next(), (error) => error === reason);

  assert.deepStrictEqual((await slow.next()).value, [Uint8Array.of(1)]);
  await assert.rejects(slow.next(), (error) => error === reason);
  assert.strictEqual((await returned.next()).done, true);
}

async function testShareLateJoiningConsumer() {
  // A consumer that joins after some data has been consumed should only
  // see data remaining in the buffer (not items already trimmed).
  const enc = new TextEncoder();
  async function* gen() {
    yield [enc.encode('a')];
    yield [enc.encode('b')];
    yield [enc.encode('c')];
  }
  const shared = share(gen(), { budget: 16384 });

  // First consumer reads all data
  const c1 = shared.pull();
  const data1 = await text(c1);
  assert.strictEqual(data1, 'abc');

  // Late-joining consumer: source is exhausted, buffer has been trimmed
  // past all data by c1's reads, so c2 gets nothing.
  const c2 = shared.pull();
  const data2 = await text(c2);
  assert.strictEqual(data2, '');
}

async function testShareRetainsBufferWhenAllConsumersDetach() {
  // Data that a consumer had not read yet must stay available to consumers
  // that attach after every previous consumer has detached.
  const enc = new TextEncoder();
  async function* gen() {
    yield [enc.encode('a')];
    yield [enc.encode('b')];
    yield [enc.encode('c')];
  }
  const shared = share(gen(), { budget: 16384 });
  const c1 = shared.pull()[Symbol.asyncIterator]();
  const c2 = shared.pull()[Symbol.asyncIterator]();
  assert.deepStrictEqual((await c1.next()).value, [enc.encode('a')]);
  // c2 has not read 'a' yet; it is the last consumer to detach.
  await c1.return();
  await c2.return();
  assert.strictEqual(shared.consumerCount, 0);

  assert.strictEqual(await text(shared.pull()), 'abc');
}

async function testShareDropOldestSplitsOversizedBatches() {
  // from() combines the values of this sync generator into a single batch
  // that is much larger than the budget. Evicting that batch as a whole would
  // leave the slower consumer with nothing at all.
  function* source() {
    for (let i = 0; i < 50; i++) {
      const chunk = new Uint8Array(4096);
      chunk[0] = i;
      yield chunk;
    }
  }
  const shared = share(source(), {
    budget: 65536,
    backpressure: 'drop-oldest',
  });
  const fast = shared.pull()[Symbol.asyncIterator]();
  const slow = shared.pull()[Symbol.asyncIterator]();

  const fastSeen = [];
  for (let r = await fast.next(); !r.done; r = await fast.next()) {
    for (const chunk of r.value) fastSeen.push(chunk[0]);
  }
  assert.deepStrictEqual(fastSeen, Array.from({ length: 50 }, (_, i) => i));

  const slowSeen = [];
  for (let r = await slow.next(); !r.done; r = await slow.next()) {
    for (const chunk of r.value) slowSeen.push(chunk[0]);
  }
  // The slow consumer lost the oldest chunks but keeps an in-order suffix
  // that fits the budget.
  assert.ok(slowSeen.length > 0);
  assert.ok(slowSeen.length * 4096 < 65536);
  assert.deepStrictEqual(
    slowSeen,
    Array.from({ length: slowSeen.length }, (_, i) => 50 - slowSeen.length + i));
}

async function testShareConsumerBreak() {
  // Verify that a consumer breaking mid-iteration detaches properly
  const enc = new TextEncoder();
  async function* gen() {
    yield [enc.encode('a')];
    yield [enc.encode('b')];
    yield [enc.encode('c')];
  }
  const shared = share(gen(), { budget: 16384 });
  const c1 = shared.pull();
  const c2 = shared.pull();

  assert.strictEqual(shared.consumerCount, 2);

  // c1 breaks after first batch
  // eslint-disable-next-line no-unused-vars
  for await (const _ of c1) {
    break;
  }
  // c1 should be detached
  assert.strictEqual(shared.consumerCount, 1);

  // c2 should still get all data
  const data2 = await text(c2);
  assert.strictEqual(data2, 'abc');
}

async function testShareMultipleConsumersConcurrentPull() {
  // Multiple consumers pulling concurrently should each receive all items
  // even when only one item is pulled from source at a time.
  async function* slowSource() {
    const enc = new TextEncoder();
    for (let i = 0; i < 5; i++) {
      await setTimeout(1);
      yield [enc.encode(`item-${i}`)];
    }
  }
  const shared = share(slowSource());
  const c1 = shared.pull();
  const c2 = shared.pull();
  const c3 = shared.pull();

  const [t1, t2, t3] = await Promise.all([
    text(c1), text(c2), text(c3),
  ]);

  const expected = 'item-0item-1item-2item-3item-4';
  assert.strictEqual(t1, expected);
  assert.strictEqual(t2, expected);
  assert.strictEqual(t3, expected);
}

async function testShareConsumerConcurrentNextCalls() {
  async function* source() {
    const enc = new TextEncoder();
    yield [enc.encode('first')];
    yield [enc.encode('second')];
  }

  const shared = share(source());
  const it = shared.pull()[Symbol.asyncIterator]();
  const first = it.next();
  const second = it.next();

  const [r1, r2] = await Promise.all([first, second]);
  const dec = new TextDecoder();
  assert.strictEqual(dec.decode(r1.value[0]), 'first');
  assert.strictEqual(dec.decode(r2.value[0]), 'second');
}

// share() accepts string source directly (normalized via from())
async function testShareStringSource() {
  const shared = share('hello-share');
  const result = await text(shared.pull());
  assert.strictEqual(result, 'hello-share');
}

Promise.all([
  testBasicShare(),
  testShareMultipleConsumers(),
  testShareConsumerCount(),
  testShareCancel(),
  testShareCancelMidIteration(),
  testShareCancelWithReason(),
  testShareCancelWithFalsyReason(),
  testShareCancelWhileSourcePullPending(),
  testShareAbortSignal(),
  testShareAbortSignalWhileSourcePullPending(),
  testSharePullAbortSignalRejectsPendingNext(),
  testSharePullPreAbortedSignalDoesNotAddConsumer(),
  testShareAlreadyAborted(),
  testShareSourceError(),
  testShareSourceErrorFollowsBufferedData(),
  testShareLateJoiningConsumer(),
  testShareRetainsBufferWhenAllConsumersDetach(),
  testShareDropOldestSplitsOversizedBatches(),
  testShareConsumerBreak(),
  testShareMultipleConsumersConcurrentPull(),
  testShareConsumerConcurrentNextCalls(),
  testShareStringSource(),
]).then(common.mustCall());