/**
 * The stream bridge: `atom.fromStream` drains an Effect `Stream` into a reactive
 * atom on a scope-bound fiber. A component reads that atom with `yield*` and
 * re-renders per emission; when the runtime is disposed (the component unmounts),
 * the scope closes and the drain fiber is interrupted — no leaked subscription.
 */
import { act } from 'react';
import { createRoot } from 'react-dom/client';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import * as Context from 'effect/Context';
import * as Effect from 'effect/Effect';
import * as Layer from 'effect/Layer';
import * as Stream from 'effect/Stream';
import { atom, defer, mount, rec } from '../../index.client.ts';

Reflect.set(globalThis, 'IS_REACT_ACT_ENVIRONMENT', true);

let container: HTMLDivElement;
beforeEach(() => {
  container = document.createElement('div');
  document.body.appendChild(container);
});
afterEach(() => container.remove());

const settle = (ms: number): Promise<void> =>
  act(async () => {
    await new Promise((r) => setTimeout(r, ms));
  });

describe('atom.fromStream — the stream bridge (#45)', () => {
  it('drains a Stream into an atom and interrupts the fiber on dispose', async () => {
    let finalized = false;
    // Emits 1, 2, 3 then stays open (never completes), with a finalizer that only
    // runs on interruption — so it proves the fiber was interrupted, not completed.
    const feed = Stream.make(1, 2, 3).pipe(
      Stream.concat(Stream.never),
      Stream.ensuring(Effect.sync(() => void (finalized = true))),
    );

    class Ticker extends Context.Service<Ticker>()('test/Ticker', {
      make: Effect.gen(function* () {
        const price = yield* atom.fromStream('price', feed, 0);
        return { price };
      }),
    }) {
      static readonly layer = Layer.effect(Ticker, Ticker.make);
    }

    const Live = rec(function* () {
      const ticker = yield* Ticker;
      // A stream atom is born Volatile — `defer` discharges V at the read (the
      // example app's leaf shows the other discharge, `.hot()`).
      const price = yield* defer(ticker.price);
      return <span data-x="p">{price}</span>;
    });

    const root = createRoot(container);
    await act(async () => root.render(mount(Ticker.layer, Live)));
    await settle(30);
    // The stream drained into the atom; the component reflects the latest emission.
    expect(container.querySelector('[data-x="p"]')?.textContent).toBe('3');
    expect(finalized).toBe(false); // still open — the fiber is alive

    await act(async () => root.unmount());
    await settle(40);
    // Disposing the runtime closes the service scope → the drain fiber is interrupted.
    expect(finalized).toBe(true);
  });
});