import { describe, expect, it, vi } from "vitest";

import type { ClockPort } from "../../../src/application/ports/clock-port.ts";
import { definePollLeasePolicy } from "../../../src/application/policies/bounded-polling.ts";
import {
  createBoundedPollCoordinator,
  type BoundedPollAttemptResult,
  type BoundedPollEnvironment,
} from "../../../src/adapters/realtime/polling/bounded-poll-coordinator.ts";

type Sleeper = {
  dueAt: number;
  resolve(): void;
  reject(): void;
  signal?: AbortSignal;
  onAbort?: () => void;
};

class ManualClock implements ClockPort {
  current = 0;
  readonly sleepers: Sleeper[] = [];

  now(): number {
    return this.current;
  }

  sleep(milliseconds: number, signal?: AbortSignal): Promise<void> {
    return new Promise((resolve, reject) => {
      if (signal?.aborted) {
        reject(new DOMException("Aborted", "AbortError"));
        return;
      }
      const sleeper: Sleeper = {
        dueAt: this.current + milliseconds,
        resolve: () => {
          signal?.removeEventListener("abort", sleeper.onAbort!);
          resolve();
        },
        reject: () => {
          signal?.removeEventListener("abort", sleeper.onAbort!);
          reject(new DOMException("Aborted", "AbortError"));
        },
        signal,
      };
      sleeper.onAbort = () => {
        this.remove(sleeper);
        sleeper.reject();
      };
      signal?.addEventListener("abort", sleeper.onAbort, {
        once: true,
      });
      this.sleepers.push(sleeper);
    });
  }

  advance(milliseconds: number): void {
    this.current += milliseconds;
    const ready = this.sleepers
      .filter((sleeper) => sleeper.dueAt <= this.current)
      .sort((left, right) => left.dueAt - right.dueAt);
    for (const sleeper of ready) {
      this.remove(sleeper);
      sleeper.resolve();
    }
  }

  private remove(target: Sleeper): void {
    const index = this.sleepers.indexOf(target);
    if (index >= 0) this.sleepers.splice(index, 1);
  }
}

class MutableEnvironment implements BoundedPollEnvironment {
  visible = true;
  connected = true;
  readonly visibilityListeners = new Set<
    (visibility: "HIDDEN" | "VISIBLE") => void
  >();
  readonly onlineListeners = new Set<(online: boolean) => void>();

  visibility(): "HIDDEN" | "VISIBLE" {
    return this.visible ? "VISIBLE" : "HIDDEN";
  }

  online(): boolean {
    return this.connected;
  }

  subscribeVisibility(
    listener: (visibility: "HIDDEN" | "VISIBLE") => void,
  ): () => void {
    this.visibilityListeners.add(listener);
    return () => this.visibilityListeners.delete(listener);
  }

  subscribeOnline(listener: (online: boolean) => void): () => void {
    this.onlineListeners.add(listener);
    return () => this.onlineListeners.delete(listener);
  }

  hide(): void {
    this.visible = false;
    for (const listener of this.visibilityListeners) listener("HIDDEN");
  }
}

const policy = definePollLeasePolicy({
  operationId: "GET_JOB_STATUS",
  owner: "reference-job",
  minimumIntervalMs: 5_000,
  successIntervalMs: 5_000,
  maxIntervalMs: 60_000,
  maxAttempts: 3,
  maxElapsedMs: 60_000,
  maxResponseBytes: 1_024,
  visibility: "VISIBLE_ONLY",
  fallbackReason: "CONVERGENCE",
  terminalStates: ["COMPLETED", "FAILED"],
});
const operation = {
  operationId: "GET_JOB_STATUS",
  contractVersion: 2,
  protocol: "REST",
  semantics: "QUERY",
  method: "GET",
  replayPolicy: "SAFE",
  retry: "never",
  maxResponseBytes: 1_024,
  transportMaxAttempts: 1,
  authRecoveryCount: 0,
  maxCumulativeSleepMs: 0,
  serverStream: false,
} as const;

async function flush(): Promise<void> {
  for (let turn = 0; turn < 12; turn += 1) {
    await Promise.resolve();
  }
}

describe("bounded poll coordinator", () => {
  it("chains completed attempts without overlap and stops at a terminal state", async () => {
    const clock = new ManualClock();
    const environment = new MutableEnvironment();
    let active = 0;
    let highWatermark = 0;
    const execute = vi
      .fn<
        (
          input: Readonly<{
            operationId: string;
            attempt: number;
            maxResponseBytes: number;
            signal: AbortSignal;
          }>,
        ) => Promise<BoundedPollAttemptResult<string>>
      >()
      .mockImplementation(async ({ attempt }) => {
        active += 1;
        highWatermark = Math.max(highWatermark, active);
        await Promise.resolve();
        active -= 1;
        return {
          ok: true,
          value: {
            kind: "VALUE",
            value: attempt === 1 ? "working" : "done",
            responseBytes: 32,
            state: attempt === 1 ? "RUNNING" : "COMPLETED",
          },
        };
      });
    const onValue = vi.fn();
    const coordinator = createBoundedPollCoordinator({
      policy,
      operation,
      execute,
      environment,
      clock,
      random: () => 0.5,
    });

    const result = coordinator.run({ onValue });
    expect(execute).not.toHaveBeenCalled();
    clock.advance(5_000);
    await flush();
    expect(execute).toHaveBeenCalledTimes(1);
    expect(clock.sleepers).toHaveLength(1);
    clock.advance(5_000);
    await flush();

    await expect(result).resolves.toEqual({
      ok: true,
      value: {
        kind: "TERMINAL",
        attempts: 2,
        state: "COMPLETED",
        value: "done",
      },
    });
    expect(onValue.mock.calls.map(([value]) => value)).toEqual([
      "working",
      "done",
    ]);
    expect(highWatermark).toBe(1);
    expect(coordinator.getState()).toBe("IDLE");
    expect(environment.visibilityListeners.size).toBe(0);
    expect(environment.onlineListeners.size).toBe(0);
  });

  it("passes the strictest response ceiling to the executor before decode", async () => {
    const clock = new ManualClock();
    const execute = vi.fn(
      async ({
        maxResponseBytes,
      }: Readonly<{ maxResponseBytes: number }>) => ({
        ok: true as const,
        value: {
          kind: "VALUE" as const,
          value: "done",
          responseBytes: maxResponseBytes,
          state: "COMPLETED",
        },
      }),
    );
    const coordinator = createBoundedPollCoordinator<string>({
      policy,
      operation: {
        ...operation,
        maxResponseBytes: 4_096,
      },
      execute,
      environment: new MutableEnvironment(),
      clock,
    });

    const pending = coordinator.run();
    clock.advance(5_000);

    await expect(pending).resolves.toMatchObject({
      ok: true,
      value: { state: "COMPLETED" },
    });
    expect(execute).toHaveBeenCalledWith(
      expect.objectContaining({ maxResponseBytes: 1_024 }),
    );
  });

  it("does not start another attempt while a request is unresolved", async () => {
    const clock = new ManualClock();
    const environment = new MutableEnvironment();
    let resolveFirst:
      | ((result: BoundedPollAttemptResult<string>) => void)
      | undefined;
    const execute = vi
      .fn()
      .mockImplementationOnce(
        () =>
          new Promise<BoundedPollAttemptResult<string>>((resolve) => {
            resolveFirst = resolve;
          }),
      )
      .mockResolvedValue({
        ok: true,
        value: {
          kind: "VALUE",
          value: "done",
          responseBytes: 10,
          state: "COMPLETED",
        },
      });
    const coordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      execute,
      environment,
      clock,
      random: () => 0.5,
    });

    const pending = coordinator.run();
    clock.advance(5_000);
    await flush();
    expect(execute).toHaveBeenCalledTimes(1);
    clock.advance(30_000);
    await flush();
    expect(execute).toHaveBeenCalledTimes(1);

    resolveFirst?.({
      ok: true,
      value: { kind: "UNCHANGED", responseBytes: 0 },
    });
    await flush();
    clock.advance(5_000);
    await flush();
    expect(execute).toHaveBeenCalledTimes(2);
    await expect(pending).resolves.toMatchObject({
      ok: true,
      value: { attempts: 2, state: "COMPLETED" },
    });
  });

  it("ends the lease when an in-flight executor ignores abort and exceeds max elapsed", async () => {
    const clock = new ManualClock();
    let attemptSignal: AbortSignal | undefined;
    let settleAttempt:
      | ((result: BoundedPollAttemptResult<string>) => void)
      | undefined;
    const execute = vi.fn(
      ({ signal }: Readonly<{ signal: AbortSignal }>) => {
        attemptSignal = signal;
        return new Promise<BoundedPollAttemptResult<string>>(
          (resolve) => {
            settleAttempt = resolve;
          },
        );
      },
    );
    const coordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      execute,
      environment: new MutableEnvironment(),
      clock,
    });

    const pending = coordinator.run();
    clock.advance(5_000);
    await flush();
    expect(execute).toHaveBeenCalledTimes(1);
    expect(attemptSignal?.aborted).toBe(false);

    clock.advance(54_999);
    await flush();
    expect(coordinator.getState()).toBe("RUNNING");
    clock.advance(1);
    await flush();

    await expect(pending).resolves.toEqual({
      ok: false,
      error: {
        kind: "POLL_BUDGET_EXHAUSTED",
        operation: "POLL",
        retryable: false,
      },
    });
    expect(attemptSignal?.aborted).toBe(true);
    expect(coordinator.getState()).toBe("DRAINING");
    await expect(coordinator.run()).resolves.toMatchObject({
      ok: false,
      error: { kind: "PROTOCOL_MISMATCH" },
    });
    settleAttempt?.({
      ok: false,
      error: {
        kind: "ABORTED",
        operation: "POLL",
        retryable: false,
      },
    });
    await flush();
    expect(coordinator.getState()).toBe("IDLE");
  });

  it("returns promptly when a caller aborts an executor that ignores its signal", async () => {
    const clock = new ManualClock();
    const caller = new AbortController();
    let settleAttempt:
      | ((result: BoundedPollAttemptResult<string>) => void)
      | undefined;
    const execute = vi.fn(
      () =>
        new Promise<BoundedPollAttemptResult<string>>(
          (resolve) => {
            settleAttempt = resolve;
          },
        ),
    );
    const coordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      execute,
      environment: new MutableEnvironment(),
      clock,
    });

    const pending = coordinator.run({ signal: caller.signal });
    clock.advance(5_000);
    await flush();
    expect(execute).toHaveBeenCalledTimes(1);
    caller.abort();

    await expect(pending).resolves.toEqual({
      ok: false,
      error: {
        kind: "ABORTED",
        operation: "POLL",
        retryable: false,
      },
    });
    expect(coordinator.getState()).toBe("DRAINING");
    settleAttempt?.({
      ok: false,
      error: {
        kind: "ABORTED",
        operation: "POLL",
        retryable: false,
      },
    });
    await flush();
    expect(coordinator.getState()).toBe("IDLE");
  });

  it("fails closed and aborts the attempt when the lease deadline clock is unavailable", async () => {
    let current = 0;
    let sleeps = 0;
    const clock: ClockPort = {
      now: () => current,
      sleep: async (milliseconds) => {
        sleeps += 1;
        if (sleeps > 1) throw new Error("deadline unavailable");
        current += milliseconds;
      },
    };
    let attemptSignal: AbortSignal | undefined;
    const execute = vi.fn(
      async ({ signal }: Readonly<{ signal: AbortSignal }>) => {
        attemptSignal = signal;
        return {
          ok: true as const,
          value: {
            kind: "UNCHANGED" as const,
            responseBytes: 0 as const,
          },
        };
      },
    );
    const coordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      execute,
      environment: new MutableEnvironment(),
      clock,
    });

    await expect(coordinator.run()).resolves.toEqual({
      ok: false,
      error: {
        kind: "PROVIDER_UNAVAILABLE",
        operation: "POLL",
        retryable: false,
      },
    });
    expect(execute).toHaveBeenCalledOnce();
    expect(attemptSignal?.aborted).toBe(true);
  });

  it("fences a non-cooperative apply callback at the lease deadline", async () => {
    const clock = new ManualClock();
    let applyContext:
      | Readonly<{
          signal: AbortSignal;
          isCurrent(): boolean;
      }>
      | undefined;
    let releaseApply: (() => void) | undefined;
    const coordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      execute: async () => ({
        ok: true,
        value: {
          kind: "VALUE",
          value: "working",
          responseBytes: 10,
          state: "RUNNING",
        },
      }),
      environment: new MutableEnvironment(),
      clock,
    });

    const pending = coordinator.run({
      onValue: (_value, context) => {
        applyContext = context;
        return new Promise<void>((resolve) => {
          releaseApply = resolve;
        });
      },
    });
    clock.advance(5_000);
    await flush();
    expect(applyContext?.isCurrent()).toBe(true);

    clock.advance(55_000);
    await flush();

    await expect(pending).resolves.toEqual({
      ok: false,
      error: {
        kind: "POLL_BUDGET_EXHAUSTED",
        operation: "POLL",
        retryable: false,
      },
    });
    expect(applyContext?.signal.aborted).toBe(true);
    expect(applyContext?.isCurrent()).toBe(false);
    expect(coordinator.getState()).toBe("DRAINING");
    releaseApply?.();
    await flush();
    expect(coordinator.getState()).toBe("IDLE");
  });

  it("honors Retry-After as a floor and exhausts finite attempts", async () => {
    const clock = new ManualClock();
    const execute = vi
      .fn()
      .mockResolvedValueOnce({
        ok: false,
        error: {
          kind: "RATE_LIMITED",
          operation: "POLL",
          retryable: true,
          retryAfterMs: 10_000,
        },
      })
      .mockResolvedValue({
        ok: true,
        value: { kind: "UNCHANGED", responseBytes: 0 },
      });
    const coordinator = createBoundedPollCoordinator<string>({
      policy: definePollLeasePolicy({
        ...policy,
        maxAttempts: 2,
      }),
      operation,
      execute,
      environment: new MutableEnvironment(),
      clock,
      random: () => 0,
    });

    const pending = coordinator.run();
    clock.advance(5_000);
    await flush();
    expect(execute).toHaveBeenCalledTimes(1);
    clock.advance(9_999);
    await flush();
    expect(execute).toHaveBeenCalledTimes(1);
    clock.advance(1);
    await flush();

    await expect(pending).resolves.toEqual({
      ok: false,
      error: {
        kind: "POLL_BUDGET_EXHAUSTED",
        operation: "POLL",
        retryable: false,
      },
    });
    expect(execute).toHaveBeenCalledTimes(2);
  });

  it.each(["RATE_LIMITED", "PROVIDER_UNAVAILABLE"] as const)(
    "does not retry %s without a bounded server hint",
    async (kind) => {
      const clock = new ManualClock();
      const execute = vi.fn().mockResolvedValue({
        ok: false,
        error: {
          kind,
          operation: "POLL",
          retryable: true,
        },
      });
      const coordinator = createBoundedPollCoordinator<string>({
        policy,
        operation,
        execute,
        environment: new MutableEnvironment(),
        clock,
        random: () => 0,
      });

      const pending = coordinator.run();
      clock.advance(5_000);
      await flush();
      await expect(pending).resolves.toEqual({
        ok: false,
        error: {
          kind,
          operation: "POLL",
          retryable: false,
        },
      });
      expect(execute).toHaveBeenCalledTimes(1);
    },
  );

  it("ignores Retry-After for retryable failure kinds that cannot carry the hint", async () => {
    const clock = new ManualClock();
    const execute = vi
      .fn()
      .mockResolvedValueOnce({
        ok: false,
        error: {
          kind: "CONNECT_TIMEOUT",
          operation: "POLL",
          retryable: true,
          retryAfterMs: 10_000,
        },
      })
      .mockResolvedValue({
        ok: true,
        value: {
          kind: "VALUE",
          value: "done",
          responseBytes: 10,
          state: "COMPLETED",
        },
      });
    const coordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      execute,
      environment: new MutableEnvironment(),
      clock,
      random: () => 0,
    });

    const pending = coordinator.run();
    clock.advance(5_000);
    await flush();
    expect(execute).toHaveBeenCalledTimes(1);
    clock.advance(4_999);
    await flush();
    expect(execute).toHaveBeenCalledTimes(1);
    clock.advance(1);

    await expect(pending).resolves.toMatchObject({
      ok: true,
      value: { attempts: 2, state: "COMPLETED" },
    });
    expect(execute).toHaveBeenCalledTimes(2);
  });

  it.each(["AUTH_REQUIRED", "FORBIDDEN"] as const)(
    "never retries terminal %s failures even when the executor marks them retryable",
    async (kind) => {
      const clock = new ManualClock();
      const execute = vi.fn().mockResolvedValue({
        ok: false,
        error: {
          kind,
          operation: "POLL",
          retryable: true,
          retryAfterMs: 10_000,
        },
      });
      const coordinator = createBoundedPollCoordinator<string>({
        policy,
        operation,
        execute,
        environment: new MutableEnvironment(),
        clock,
      });

      const pending = coordinator.run();
      clock.advance(5_000);

      await expect(pending).resolves.toEqual({
        ok: false,
        error: {
          kind,
          operation: "POLL",
          retryable: false,
        },
      });
      expect(execute).toHaveBeenCalledOnce();
    },
  );

  it("aborts in-flight work on hidden lifecycle and rejects late scope results", async () => {
    const hiddenClock = new ManualClock();
    const hiddenEnvironment = new MutableEnvironment();
    const hiddenCoordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      environment: hiddenEnvironment,
      clock: hiddenClock,
      execute: ({ signal }) =>
        new Promise((resolve) => {
          signal.addEventListener(
            "abort",
            () =>
              resolve({
                ok: false,
                error: {
                  kind: "ABORTED",
                  operation: "POLL",
                  retryable: false,
                },
              }),
            { once: true },
          );
        }),
    });
    const hidden = hiddenCoordinator.run();
    hiddenClock.advance(5_000);
    await flush();
    hiddenEnvironment.hide();
    await expect(hidden).resolves.toEqual({
      ok: false,
      error: {
        kind: "ABORTED",
        operation: "POLL",
        retryable: false,
      },
    });

    const scopeClock = new ManualClock();
    let current = true;
    let resolveAttempt:
      | ((result: BoundedPollAttemptResult<string>) => void)
      | undefined;
    const onValue = vi.fn();
    const scopeCoordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      environment: new MutableEnvironment(),
      clock: scopeClock,
      isCurrent: () => current,
      execute: () =>
        new Promise((resolve) => {
          resolveAttempt = resolve;
        }),
    });
    const fenced = scopeCoordinator.run({ onValue });
    scopeClock.advance(5_000);
    await flush();
    current = false;
    resolveAttempt?.({
      ok: true,
      value: {
        kind: "VALUE",
        value: "late",
        responseBytes: 10,
        state: "COMPLETED",
      },
    });
    await expect(fenced).resolves.toEqual({
      ok: false,
      error: {
        kind: "SCOPE_FENCED",
        operation: "POLL",
        retryable: false,
      },
    });
    expect(onValue).not.toHaveBeenCalled();
  });

  it("rejects accessor and extra-key attempt results without re-reading them", async () => {
    const accessorClock = new ManualClock();
    const readOk = vi.fn(() => true);
    const accessorResult = Object.defineProperties(
      {},
      {
        ok: {
          enumerable: true,
          get: readOk,
        },
        value: {
          enumerable: true,
          value: {
            kind: "VALUE",
            value: "unsafe",
            responseBytes: 1,
            state: "COMPLETED",
          },
        },
      },
    ) as BoundedPollAttemptResult<string>;
    const accessorCoordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      environment: new MutableEnvironment(),
      clock: accessorClock,
      execute: async () => accessorResult,
    });
    const accessorRun = accessorCoordinator.run();
    accessorClock.advance(5_000);
    await expect(accessorRun).resolves.toMatchObject({
      ok: false,
      error: { kind: "PROTOCOL_MISMATCH" },
    });
    expect(readOk).not.toHaveBeenCalled();

    const extraClock = new ManualClock();
    const extraCoordinator = createBoundedPollCoordinator<string>({
      policy,
      operation,
      environment: new MutableEnvironment(),
      clock: extraClock,
      execute: async () =>
        ({
          ok: true,
          value: {
            kind: "VALUE",
            value: "unsafe",
            responseBytes: 1,
            state: "COMPLETED",
          },
          extra: true,
        }) as BoundedPollAttemptResult<string>,
    });
    const extraRun = extraCoordinator.run();
    extraClock.advance(5_000);
    await expect(extraRun).resolves.toMatchObject({
      ok: false,
      error: { kind: "PROTOCOL_MISMATCH" },
    });
  });

  it("fails closed on response ceilings, concurrent runs and close", async () => {
    const clock = new ManualClock();
    const coordinator = createBoundedPollCoordinator<string>({
      policy,
      operation: {
        ...operation,
        maxResponseBytes: 512,
      },
      environment: new MutableEnvironment(),
      clock,
      execute: async () => ({
        ok: true,
        value: {
          kind: "VALUE",
          value: "oversized",
          responseBytes: 513,
          state: "RUNNING",
        },
      }),
    });
    const first = coordinator.run();
    await expect(coordinator.run()).resolves.toEqual({
      ok: false,
      error: {
        kind: "PROTOCOL_MISMATCH",
        operation: "POLL",
        retryable: false,
      },
    });
    clock.advance(5_000);
    await expect(first).resolves.toEqual({
      ok: false,
      error: {
        kind: "PROTOCOL_MISMATCH",
        operation: "POLL",
        retryable: false,
      },
    });

    const pending = coordinator.run();
    coordinator.close();
    coordinator.close();
    await expect(pending).resolves.toEqual({
      ok: false,
      error: {
        kind: "CLOSED",
        operation: "POLL",
        retryable: false,
      },
    });
    expect(coordinator.getState()).toBe("CLOSED");
  });
});