import DIE from "phpdie";
import { type Ord, minBy, maxBy } from "./utils";

// PromiseWithResolvers is ES2024 — define locally for broader consumer compat
type PromiseWithResolvers<T> = {
  promise: Promise<T>;
  resolve: (value: T | PromiseLike<T>) => void;
  reject: (reason?: unknown) => void;
};
import { toStream } from "./froms";
import { type FlowSource, sflow } from "./index";
import { streamAsyncIterator } from "./streamAsyncIterator";

interface MergeBy {
  <T>(ordFn: (input: T) => Ord, srcs: FlowSource<FlowSource<T>>): sflow<T>;
  <T>(ordFn: (input: T) => Ord): (srcs: FlowSource<FlowSource<T>>) => sflow<T>;
}

/**
 * merge multiple stream by ascend order, assume all input stream is sorted by ascend
 * output stream will be sorted by ascend too.
 *
 * if one of input stream is not sorted by ascend, it will throw an error.
 *
 * @param ordFn a function to get the order of input
 * @param srcs a list of input stream
 * @returns a new stream that merge all input stream by ascend order
 * @deprecated use {@link mergeStreamsByAscend}, this will be removed in next major version
 */
export const mergeAscends: MergeBy = <T>(
  ordFn: (input: T) => Ord,
  _srcs?: FlowSource<FlowSource<T>>,
) => {
  if (!_srcs) return ((srcs: any) => mergeAscends(ordFn, srcs)) as any;
  return sflow(
    new ReadableStream<T>(
      {
        pull: async (ctrl) => {
          const srcs = await sflow(_srcs).toArray();
          const slots = srcs.map(() => undefined as { value: T } | undefined);
          const pendingSlotRemoval = srcs.map(
            () => undefined as PromiseWithResolvers<void> | undefined,
          );
          const drains = srcs.map(() => false);
          let lastMinValue: T | undefined;
          await Promise.all(
            srcs.map(async (src, i) => {
              for await (const value of sflow(src)) {
                while (slots[i] !== undefined) {
                  if (shiftMinValueIfFull()) continue;
                  pendingSlotRemoval[i] = Promise.withResolvers<void>();
                  await pendingSlotRemoval[i]?.promise; // wait for this slot empty;
                }
                slots[i] = { value: value as T };
                shiftMinValueIfFull();
              }
              // done
              drains[i] = true;
              pendingSlotRemoval.map((e) => e?.resolve());
              await Promise.all(pendingSlotRemoval.map((e) => e?.promise));
              const allDrain = drains.every(Boolean);

              if (allDrain) {
                while (slots.some((e) => e !== undefined))
                  shiftMinValueIfFull();
                ctrl.close();
              }

              function shiftMinValueIfFull() {
                const isFull = slots.every(
                  (slot, i) => slot !== undefined || drains[i],
                );
                if (!isFull) return false;
                const fullSlots = slots
                  .flatMap((e) => (e !== undefined ? [e] : []))
                  .map((e) => e.value);
                const minValue = minBy(ordFn, fullSlots);
                const minIndex = slots.findIndex((e) => e?.value === minValue);
                if (lastMinValue !== undefined) {
                  if (ordFn(lastMinValue) > ordFn(minValue))
                    DIE(`
MergeAscendError: one of source stream is not ascending ordered.

stream index: ${minIndex}

prev: ${ordFn(lastMinValue)}
prev: ${JSON.stringify(lastMinValue)}

curr: ${ordFn(minValue)}
curr: ${JSON.stringify(minValue)}
`);
                }
                lastMinValue = minValue;
                ctrl.enqueue(minValue!);
                slots[minIndex] = undefined;
                pendingSlotRemoval[minIndex]?.resolve();
                pendingSlotRemoval[minIndex] = undefined;
                return true;
              }
            }),
          );
        },
      },
      { highWaterMark: 0 },
    ),
  );
};

/**
 * merge multiple stream by ascend order, assume all input stream is sorted by ascend
 * output stream will be sorted by ascend too.
 *
 * if one of input stream is not sorted by ascend, it will throw an error.
 *
 * @param ordFn a function to get the order of input
 * @param srcs a list of input stream
 * @returns a new stream that merge all input stream by ascend order
 * @deprecated use {@link mergeStreamsByAscend}
 */
export const mergeDescends: MergeBy = <T>(
  ordFn: (input: T) => Ord,
  _srcs?: FlowSource<FlowSource<T>>,
) => {
  if (!_srcs) return ((srcs: any) => mergeDescends(ordFn, srcs)) as any;
  return toStream(
    new ReadableStream<T>(
      {
        pull: async (ctrl) => {
          const srcs = await sflow(_srcs).toArray();
          const slots = srcs.map(
            () => undefined as { value: Awaited<T> } | undefined,
          );
          const pendingSlotRemoval = srcs.map(
            () => undefined as PromiseWithResolvers<void> | undefined,
          );
          const drains = srcs.map(() => false);
          let lastMaxValue: T | undefined;
          await Promise.all(
            srcs.map(async (src, i) => {
              const stream = toStream(src);
              const streamIterator = Object.assign(stream, {
                [Symbol.asyncIterator]: streamAsyncIterator,
              });
              for await (const value of streamIterator) {
                while (slots[i] !== undefined) {
                  if (shiftMaxValueIfFull()) continue;
                  pendingSlotRemoval[i] = Promise.withResolvers<void>();
                  await pendingSlotRemoval[i]?.promise; // wait for this slot empty;
                }

                slots[i] = { value: value as T extends Awaited<T> ? T : never };
                shiftMaxValueIfFull();
              }
              // done
              drains[i] = true;
              pendingSlotRemoval.map((e) => e?.resolve());
              await Promise.all(pendingSlotRemoval.map((e) => e?.promise));
              const allDrain = drains.every(Boolean);

              if (allDrain) {
                while (slots.some((e) => e !== undefined))
                  shiftMaxValueIfFull();
                ctrl.close();
              }

              function shiftMaxValueIfFull() {
                const isFull = slots.every(
                  (slot, i) => slot !== undefined || drains[i],
                );
                if (!isFull) return false;
                const fullSlots = slots
                  .flatMap((e) => (e !== undefined ? [e] : []))
                  .map((e) => e.value);
                const maxValue = maxBy(ordFn, fullSlots);
                const maxIndex = slots.findIndex((e) => e?.value === maxValue);
                if (lastMaxValue !== undefined) {
                  if (ordFn(lastMaxValue) < ordFn(maxValue))
                    DIE(`
MergeDescendError: one of source stream is not descending ordered.

stream index: ${maxIndex}

prev: ${ordFn(lastMaxValue)}
prev: ${JSON.stringify(lastMaxValue)}

curr: ${ordFn(maxValue)}
curr: ${JSON.stringify(maxValue)}
`);
                }
                lastMaxValue = maxValue;
                ctrl.enqueue(maxValue!);
                slots[maxIndex] = undefined;
                pendingSlotRemoval[maxIndex]?.resolve();
                pendingSlotRemoval[maxIndex] = undefined;
                return true;
              }
            }),
          );
        },
      },
      { highWaterMark: 0 },
    ),
  );
};