import { createMachineActor, type HandlerContext, type MessageActor } from '../../../core/actors/create-machine-actor';
import { effect } from '../../../core/signals/effect';
import { peek } from '../../../core/signals/primitives';
import { SerialRunner, Task } from '../../../core/tasks/task';
import {
  type BackBufferConfig,
  calculateBackBufferFlushPoint,
  DEFAULT_BACK_BUFFER_CONFIG,
} from '../../../media/buffer/back-buffer';
import {
  calculateForwardFlushPoint,
  DEFAULT_FORWARD_BUFFER_CONFIG,
  type ForwardBufferConfig,
  getSegmentsToLoad,
} from '../../../media/buffer/forward-buffer';
import {
  type AddressableObject,
  type AudioTrack,
  SEGMENT_TIME_EPSILON,
  type Segment,
  type VideoTrack,
} from '../../../media/types';
import type { AppendInitMessage, AppendSegmentMessage, RemoveMessage, SourceBufferActor } from './source-buffer';

// ============================================================================
// BUFFER STATE TYPES
// ============================================================================

/**
 * Buffer state for a single SourceBuffer.
 * Tracks which init segment and media segments are loaded.
 */
export interface SourceBufferState {
  /** Track ID of the loaded init segment */
  initTrackId?: string;
  /** Loaded media segments (unordered - selectors derive ordering) */
  segments: Array<{ id: string; trackId: string }>;
}

/**
 * Buffer state for all SourceBuffers.
 */
export interface BufferState {
  video?: SourceBufferState;
  audio?: SourceBufferState;
}

// ============================================================================
// MESSAGE PROTOCOL
// ============================================================================

/** Track types that have SourceBuffers (video and audio only). */
export type SegmentLoaderTrack = VideoTrack | AudioTrack;

/**
 * Message sent to a SegmentLoaderActor.
 *
 * `range` is optional to distinguish loading modes:
 * - No range: load init segment only (metadata preload mode)
 * - With range: load init + all segments overlapping [start, end]
 *
 * `start` and `end` are raw time values — no segment snapping.
 * The actor maps them onto segment boundaries internally.
 */
export type SegmentLoaderMessage = {
  type: 'load';
  track: SegmentLoaderTrack;
  range?: { start: number; end: number };
};

// ============================================================================
// LOAD TASK
// ============================================================================

/**
 * A LoadTask is the intent to perform one unit of SegmentLoader work.
 * Unlike SourceBufferMessage, fetch-based tasks carry a URL rather than
 * pre-fetched data — the runner fetches and appends them in sequence.
 *
 * Derived from SourceBufferMessage types by removing `data` and adding
 * a fetch URL via AddressableObject.
 *
 * @todo Rename — "LoadTask" risks confusion with the `Task` class used for
 * SourceBufferActor scheduling. These are closer to operation descriptors or
 * messages than tasks in that sense.
 */
export type LoadTask =
  | (Omit<AppendInitMessage, 'data'> & AddressableObject)
  | (Omit<AppendSegmentMessage, 'data'> & AddressableObject)
  | RemoveMessage;

// ============================================================================
// ACTOR INTERFACE
// ============================================================================

/** Finite states of the actor. */
export type SegmentLoaderActorState = 'idle' | 'loading' | 'destroyed';

/** Non-finite (extended) data managed by the actor. */
export interface SegmentLoaderActorContext {
  /** Track ID of the init segment currently being fetched/appended, or null. */
  inFlightInitTrackId: string | null;
  /** Segment ID currently being fetched/appended, or null. */
  inFlightSegmentId: string | null;
}

export type SegmentLoaderActor = MessageActor<SegmentLoaderActorState, SegmentLoaderActorContext, SegmentLoaderMessage>;

/**
 * Configuration for `createSegmentLoaderActor`. Each sub-config is
 * spread over the corresponding `DEFAULT_*_CONFIG` so callers can
 * override individual fields.
 */
export interface SegmentLoaderActorConfig {
  forwardBuffer?: Partial<ForwardBufferConfig>;
  backBuffer?: Partial<BackBufferConfig>;
}

// ============================================================================
// HELPERS
// ============================================================================

type FetchBytes = (
  addressable: AddressableObject,
  options?: RequestInit & { minChunkSize?: number }
) => Promise<AsyncIterable<Uint8Array>>;

/**
 * Resolves when the SourceBufferActor snapshot reaches 'idle'.
 * Rejects if the signal is aborted or the actor is destroyed.
 *
 * Used to sequence SourceBufferActor operations without awaiting send()
 * directly — send() is fire-and-forget; callers observe completion via
 * state transition.
 */
function waitForIdle(snapshot: SourceBufferActor['snapshot'], signal: AbortSignal): Promise<void> {
  return new Promise((resolve, reject) => {
    if (snapshot.get().value === 'idle') {
      resolve();
      return;
    }
    if (snapshot.get().value === 'destroyed') {
      reject(new DOMException('Aborted', 'AbortError'));
      return;
    }
    if (signal.aborted) {
      reject(signal.reason);
      return;
    }

    let stop: (() => void) | undefined;

    const cleanup = (fn: () => void) => {
      stop?.();
      signal.removeEventListener('abort', onAbort);
      fn();
    };

    const onAbort = () => cleanup(() => reject(signal.reason));

    stop = effect(() => {
      const value = snapshot.get().value;
      if (value === 'idle') cleanup(resolve);
      else if (value === 'destroyed') cleanup(() => reject(new DOMException('Aborted', 'AbortError')));
    });

    signal.addEventListener('abort', onAbort, { once: true });
  });
}

// ============================================================================
// LOAD TASK FACTORY
// ============================================================================

interface LoadTaskOptions {
  getContext: () => SegmentLoaderActorContext;
  setContext: (ctx: SegmentLoaderActorContext) => void;
  fetchBytes: FetchBytes;
  sourceBufferActor: SourceBufferActor;
}

/**
 * Wraps a LoadTask descriptor into a Task that fetches (if needed) and
 * forwards to SourceBufferActor. Updates in-flight context around async
 * operations so the loading handler can make accurate continue/preempt
 * decisions at any point.
 */
function makeLoadTask(
  op: LoadTask,
  { getContext, setContext, fetchBytes, sourceBufferActor }: LoadTaskOptions
): Task<void> {
  return new Task(async (taskSignal) => {
    if (taskSignal.aborted) return;

    if (op.type === 'remove') {
      sourceBufferActor.send(op);
      await waitForIdle(sourceBufferActor.snapshot, taskSignal);
      return;
    }

    if (op.type === 'append-init') {
      setContext({ ...getContext(), inFlightInitTrackId: op.meta.trackId });
      try {
        // Init segments are small and need the full body before appending.
        // minChunkSize: Infinity accumulates all chunks into one before yielding.
        const data = await fetchBytes(op, { signal: taskSignal, minChunkSize: Infinity });
        if (!taskSignal.aborted) {
          sourceBufferActor.send({ type: 'append-init', data, meta: op.meta });
          await waitForIdle(sourceBufferActor.snapshot, taskSignal);
        }
      } finally {
        setContext({ ...getContext(), inFlightInitTrackId: null });
      }
      return;
    }

    // append-segment: await headers eagerly (starts the HTTP connection and
    // records the fetch in observers like tests), then pass the body stream
    // directly to the actor so chunks are appended as they arrive.
    setContext({ ...getContext(), inFlightSegmentId: op.meta.id });
    try {
      const stream = await fetchBytes(op, { signal: taskSignal });
      if (!taskSignal.aborted) {
        sourceBufferActor.send({ type: 'append-segment', data: stream, meta: op.meta });
        await waitForIdle(sourceBufferActor.snapshot, taskSignal);
      }
    } finally {
      setContext({ ...getContext(), inFlightSegmentId: null });
    }
  });
}

// ============================================================================
// IMPLEMENTATION
// ============================================================================

/**
 * Creates a SegmentLoaderActor for one track type (video or audio).
 *
 * Receives load assignments via `send()` and owns all execution: planning,
 * removes, fetches, and appends. Coordinates with the SourceBufferActor for
 * all physical SourceBuffer operations.
 *
 * Planning (Cases 1–3) happens in the `load` handler on every incoming
 * message, producing an ordered LoadTask list. The runner drains that list
 * sequentially via SerialRunner. When a new message arrives mid-run, the
 * handler replans and either continues the in-flight operation (abortPending
 * + schedule new remainder) or preempts it (abortAll + cancel SourceBuffer
 * if needed + schedule new plan).
 *
 * @param sourceBufferActor - Shared SourceBufferActor reference (not owned)
 * @param fetchBytes - Tracked fetch closure (owns throughput sampling for segments).
 *   Accepts an optional `minChunkSize` in options; init segments pass `Infinity`
 *   so the entire body accumulates as one chunk before appending.
 */
export function createSegmentLoaderActor(
  sourceBufferActor: SourceBufferActor,
  fetchBytes: FetchBytes,
  config: SegmentLoaderActorConfig = {}
): SegmentLoaderActor {
  type UserState = Exclude<SegmentLoaderActorState, 'destroyed'>;
  type Ctx = HandlerContext<UserState, SegmentLoaderActorContext, () => SerialRunner>;

  const forwardBufferConfig: ForwardBufferConfig = { ...DEFAULT_FORWARD_BUFFER_CONFIG, ...config.forwardBuffer };
  const backBufferConfig: BackBufferConfig = { ...DEFAULT_BACK_BUFFER_CONFIG, ...config.backBuffer };

  const getBufferedSegments = (allSegments: readonly Segment[]): Segment[] => {
    // Exclude partial segments — they are still being streamed and must not be
    // treated as fully buffered for load planning or buffer window calculations.
    //
    // `peek` defensively: `load` handlers run synchronously inside `send()`,
    // which is called from inside the dispatcher reactor's `effects:` body.
    // A tracked `.snapshot.get()` here would leak the source-buffer-actor's
    // snapshot into the dispatcher's dep set, causing the dispatcher to re-
    // fire on every SourceBufferActor state change. Mirrors the fix applied
    // to the text-track loader in `b3f44efe`.
    const bufferedIds = new Set(
      peek(sourceBufferActor.snapshot)
        .context.segments.filter((s) => !s.partial)
        .map((s) => s.id)
    );
    return allSegments.filter((s) => bufferedIds.has(s.id));
  };

  /**
   * Translate a load message into an ordered LoadTask list based on committed
   * actor state. In-flight awareness is handled separately in the load handler.
   *
   * @todo Rename alongside LoadTask (e.g. planOps).
   *
   * Case 1 — Removes: forward and back buffer flush points, segment-aligned.
   *   ABR-style track switches (same content, different bitrate) do not flush:
   *   appending new content overwrites existing buffer ranges, and the actor's
   *   time-aligned deduplication keeps the segment model accurate as new
   *   segments arrive.
   *
   *   Cross-rendition track switches (audio language change, text language
   *   change) do flush: the buffered content is semantically incompatible with
   *   the newly-selected track, so overwrite-on-append would leave stale
   *   content playing until each replacement segment lands. Today's predicate:
   *   `actorCtx.initTrackLanguage !== track.language` — fires for language
   *   changes, no-ops for video / same-language audio bitrate switches.
   *   Future stage: pluggable predicate / strategy at actor construction time
   *   for codec-change (5.1 surround) and other cross-rendition shapes.
   *
   * Case 2 — Init: schedule if not yet committed for this track.
   *
   * Case 3 — Segments: all segments in the load window not yet committed.
   */
  const planTasks = (message: SegmentLoaderMessage): LoadTask[] => {
    const { track, range } = message;
    // `peek` for the same reason as `getBufferedSegments` above — avoid
    // leaking the SourceBufferActor snapshot into the calling dispatcher's
    // tracking scope.
    const actorCtx = peek(sourceBufferActor.snapshot).context;
    const bufferedSegments = getBufferedSegments(track.segments);
    const currentTime = range?.start ?? 0;
    const tasks: LoadTask[] = [];

    // Cross-rendition switch check (mid-stream language change). Fires when
    // (a) an init segment has already been committed for some track,
    // (b) the newly-selected track is a different track, and
    // (c) the languages differ. The buffered range from the current segment
    // boundary forward is treated as stale (new track's same-timestamp
    // segments will overwrite it via MSE append-at-same-timestamp); no
    // explicit `remove` task is emitted for the cross-rendition range.
    // Computed against the currently-buffered segments (stable reference;
    // new track's playlist may not be resolved yet at first load).
    const isCrossRenditionSwitch =
      actorCtx.initTrackId !== undefined &&
      actorCtx.initTrackId !== track.id &&
      actorCtx.initTrackLanguage !== track.language;

    // Two categories of "buffered content that should not gate planning":
    //
    // - `removes` — content that needs an explicit `remove` task (out-of-window
    //   forward content, back-buffer content beyond the keep window). Emitted
    //   as `{ type: 'remove' }` tasks.
    // - `staleRanges` — content that the new appends will overwrite at the
    //   same timestamps (cross-rendition switch). No explicit `remove` —
    //   MSE's overwrite-on-append handles it, which gives a much smaller
    //   perceived audio gap than `remove`-then-fetch-then-append.
    //
    // Both categories affect `effectiveBuffered` so `getSegmentsToLoad`
    // re-plans new-track segments inside them.
    const removes: Array<{ start: number; end: number }> = [];
    const staleRanges: Array<{ start: number; end: number }> = [];
    if (range) {
      if (isCrossRenditionSwitch) {
        // Mark current-segment-start onward as stale. Falls back to the
        // first buffered segment after currentTime when the playhead sits
        // in a buffer gap.
        const currentSeg = actorCtx.segments.find(
          (s) => s.startTime <= currentTime && s.startTime + s.duration > currentTime
        );
        const staleStart = currentSeg?.startTime ?? actorCtx.segments.find((s) => s.startTime > currentTime)?.startTime;
        if (staleStart !== undefined) {
          staleRanges.push({ start: staleStart, end: Infinity });
        }
      }
      const forwardFlushStart = calculateForwardFlushPoint(bufferedSegments, currentTime, forwardBufferConfig);
      if (forwardFlushStart < Infinity) {
        removes.push({ start: forwardFlushStart, end: Infinity });
      }
      const backFlushEnd = calculateBackBufferFlushPoint(bufferedSegments, currentTime, backBufferConfig);
      if (backFlushEnd > 0) {
        removes.push({ start: 0, end: backFlushEnd });
      }
      for (const r of removes) tasks.push({ type: 'remove', start: r.start, end: r.end });
    }

    // Treat any segment overlapping a planned remove OR a stale range as
    // not-buffered. Without this, sibling renditions that share segment IDs
    // and startTimes (audio language variants) would see the new track's
    // same-startTime segments marked "buffered" by the pre-flush snapshot,
    // skip them in `getSegmentsToLoad`, and leave a permanent gap from
    // `currentSegmentStart` to the end of the old buffer window — stalling
    // playback.
    const overlapsStale = (seg: { startTime: number; duration: number }): boolean => {
      const segEnd = seg.startTime + seg.duration;
      return (
        removes.some((r) => seg.startTime < r.end && segEnd > r.start) ||
        staleRanges.some((r) => seg.startTime < r.end && segEnd > r.start)
      );
    };
    const effectiveBuffered =
      removes.length + staleRanges.length > 0 ? bufferedSegments.filter((s) => !overlapsStale(s)) : bufferedSegments;

    // Case 2: Init
    if (actorCtx.initTrackId !== track.id) {
      tasks.push({
        type: 'append-init',
        meta: { trackId: track.id, language: track.language },
        url: track.initialization.url,
        ...(track.initialization.byteRange !== undefined && { byteRange: track.initialization.byteRange }),
      });
    }

    // Case 3: Segments
    if (range) {
      const segmentsToLoad = getSegmentsToLoad(
        track.segments,
        effectiveBuffered,
        currentTime,
        forwardBufferConfig
      ).filter((seg) => {
        // Quality-aware filter: skip segments already covered by equal-or-higher-quality
        // content in the actor context. Preserves buffered high-quality content during
        // ABR downgrades; loads during upgrades and for uncovered positions.
        // Actor entries that overlap a planned remove OR a stale range are treated
        // as nonexistent here — they're about to be flushed or overwritten, so the
        // new-track segment must load regardless of the existing entry's bandwidth.
        const existing = actorCtx.segments.find(
          (s) => !overlapsStale(s) && Math.abs(s.startTime - seg.startTime) < SEGMENT_TIME_EPSILON
        );
        // Partial segments are still streaming — treat as not buffered so they
        // are always re-planned (avoids relying on incomplete data).
        if (existing?.partial) return true;
        if (!existing?.trackBandwidth || !track.bandwidth) return true;
        return track.bandwidth > existing.trackBandwidth;
      });
      for (const segment of segmentsToLoad) {
        tasks.push({
          type: 'append-segment',
          meta: {
            id: segment.id,
            startTime: segment.startTime,
            duration: segment.duration,
            trackId: track.id,
            trackBandwidth: track.bandwidth,
          },
          url: segment.url,
          ...(segment.byteRange !== undefined && { byteRange: segment.byteRange }),
        });
      }
    }

    return tasks;
  };

  const scheduleAll = (tasks: LoadTask[], { getContext, setContext, runner }: Ctx): void => {
    tasks.forEach((op) => {
      runner
        .schedule(makeLoadTask(op, { getContext, setContext, fetchBytes, sourceBufferActor }))
        .then(undefined, (e: unknown) => {
          if (e instanceof Error && e.name === 'AbortError') return;
          // On unexpected fetch/append errors, abort remaining tasks so a failed
          // init doesn't cause segment fetches to proceed with no init segment.
          console.error('Unexpected error in segment loader:', e);
          runner.abortPending();
        });
    });
  };

  return createMachineActor<UserState, SegmentLoaderActorContext, SegmentLoaderMessage, () => SerialRunner>({
    runner: () => new SerialRunner(),
    initial: 'idle',
    context: { inFlightInitTrackId: null, inFlightSegmentId: null },
    states: {
      idle: {
        on: {
          load: (msg, ctx) => {
            const allTasks = planTasks(msg);
            if (allTasks.length === 0) return;
            ctx.transition('loading');
            scheduleAll(allTasks, ctx);
          },
        },
      },
      loading: {
        onSettled: 'idle',
        on: {
          load: (msg, ctx) => {
            const { context, runner } = ctx;
            const allTasks = planTasks(msg);

            // Determine whether the in-flight operation is still needed.
            const inFlightStillNeeded =
              (context.inFlightSegmentId !== null &&
                allTasks.some((t) => t.type === 'append-segment' && t.meta.id === context.inFlightSegmentId)) ||
              (context.inFlightInitTrackId !== null &&
                allTasks.some((t) => t.type === 'append-init' && t.meta.trackId === context.inFlightInitTrackId));

            if (inFlightStillNeeded) {
              // Continue: abort only the pending queue, let the in-flight task finish.
              // Schedule everything except the in-flight item — it covers that slot.
              runner.abortPending();
              scheduleAll(
                allTasks.filter(
                  (t) =>
                    !(t.type === 'append-segment' && t.meta.id === context.inFlightSegmentId) &&
                    !(t.type === 'append-init' && t.meta.trackId === context.inFlightInitTrackId)
                ),
                ctx
              );
            } else {
              // Preempt: abort everything and replan.
              runner.abortAll();
              // Cancel SourceBufferActor tasks when a segment is in-flight (always
              // discard) or when a track switch is happening (new track's init follows).
              // For a same-track seek with an in-flight init, skip cancel — the task's
              // signal is not aborted (abortAll was called on the runner, but the init
              // task already completed or will complete via its own signal path).
              const cancelSourceBuffer =
                context.inFlightSegmentId !== null ||
                (context.inFlightInitTrackId !== null &&
                  allTasks.some((t) => t.type === 'append-init' && t.meta.trackId !== context.inFlightInitTrackId));
              if (cancelSourceBuffer) {
                sourceBufferActor.send({ type: 'cancel' });
              }
              scheduleAll(allTasks, ctx);
            }
          },
        },
      },
    },
  });
}