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

import {
  SSE_CONTINUE,
  createFetchSseConnection,
} from "../../../src/adapters/realtime/sse/fetch-sse-connection.ts";
import {
  realtimeTransportRecoveryCommitted,
  type RealtimeRecoveryCheckpoint,
} from "../../../src/application/ports/realtime/event-authority.ts";
import type { RealtimeResult } from "../../../src/application/ports/realtime/shared.ts";
import type {
  StreamRegistrationId,
} from "../../../src/contracts/realtime-streams.ts";
import {
  realtimeFailure,
  realtimeSuccess,
} from "../../../src/adapters/realtime/result.ts";

const encoder = new TextEncoder();
const continueEvent = () => realtimeSuccess(SSE_CONTINUE);
const RECOVERY_CHECKPOINT = Object.freeze({
  recoveryMode: "CURSOR" as const,
  streamEpoch: "stream-epoch.0001",
  lastAppliedSequence: "1",
  resumeCursor: "cursor-1",
}) as RealtimeRecoveryCheckpoint;
const RECOVERY_OUTCOME = realtimeTransportRecoveryCommitted(
  "REFERENCE_STREAM" as StreamRegistrationId,
  RECOVERY_CHECKPOINT,
);

function eventStream(
  chunks: readonly string[],
  options: Readonly<{
    status?: number;
    contentType?: string;
  }> = {},
): Response {
  return new Response(
    new ReadableStream<Uint8Array>({
      start(controller) {
        for (const chunk of chunks) {
          controller.enqueue(encoder.encode(chunk));
        }
        controller.close();
      },
    }),
    {
      status: options.status ?? 200,
      headers: {
        "Content-Type":
          options.contentType ?? "text/event-stream; charset=utf-8",
      },
    },
  );
}

function connection(
  fetcher: typeof fetch,
  recoveryMode: "CURSOR" | "SESSION_REBUILD" | "SNAPSHOT_ONLY" =
    "CURSOR",
) {
  return createFetchSseConnection({
    endpoint: "https://app.example.test/events",
    applicationOrigin: "https://app.example.test",
    recoveryMode,
    fetcher,
  });
}

describe("fetch SSE connection", () => {
  it("uses the fixed request profile and streams events sequentially", async () => {
    const response = eventStream([
      ": heartbeat\r\n",
      "retry: 2500\n",
      "id: cursor-1\ndata: first\n",
      "data: second\n\n",
    ]);
    const fetcher = vi.fn(
      async (_input: RequestInfo | URL, _init?: RequestInit) =>
        response,
    );
    const received: string[] = [];
    const comments = vi.fn();
    const hints = vi.fn();
    const runtime = connection(fetcher as typeof fetch);

    await expect(
      runtime.read({
        resumeCursor: "cursor-0",
        onEvent: async (event) => {
          received.push(event.data);
          return realtimeSuccess(SSE_CONTINUE);
        },
        onComment: comments,
        onRetryHint: hints,
      }),
    ).resolves.toEqual({
      ok: true,
      value: {
        kind: "EOF",
        incompleteEventDiscarded: false,
        retryHintMs: 2_500,
      },
    });

    expect(received).toEqual(["first\nsecond"]);
    expect(comments).toHaveBeenCalledOnce();
    expect(hints).toHaveBeenCalledWith(2_500);
    const [target, init] = fetcher.mock.calls[0] ?? [];
    expect(target).toBe("https://app.example.test/events");
    expect(init).toMatchObject({
      method: "GET",
      credentials: "same-origin",
      redirect: "error",
      cache: "no-store",
      referrerPolicy: "no-referrer",
    });
    expect(new Headers(init?.headers)).toEqual(
      new Headers({
        Accept: "text/event-stream",
        "Last-Event-ID": "cursor-0",
      }),
    );
  });

  it("holds event consumption behind the validated open barrier gate", async () => {
    const runtime = connection(
      (async () =>
        eventStream([
          "id: cursor-1\ndata: after-barrier\n\n",
        ])) as typeof fetch,
    );
    let release:
      | ((result: RealtimeResult<void>) => void)
      | undefined;
    const gate = new Promise<RealtimeResult<void>>((resolve) => {
      release = resolve;
    });
    const onOpen = vi.fn(() => gate);
    const onEvent = vi.fn(continueEvent);

    const reading = runtime.read({
      resumeCursor: null,
      onOpen,
      onEvent,
    });
    await vi.waitFor(() => expect(onOpen).toHaveBeenCalledOnce());
    expect(onEvent).not.toHaveBeenCalled();

    release?.(realtimeSuccess(undefined));
    await expect(reading).resolves.toMatchObject({
      ok: true,
      value: { kind: "EOF" },
    });
    expect(onEvent).toHaveBeenCalledOnce();
  });

  it("propagates an exact open-barrier failure before reading events", async () => {
    const runtime = connection(
      (async () =>
        eventStream([
          "id: cursor-1\ndata: unreachable\n\n",
        ])) as typeof fetch,
    );
    const barrierFailure = realtimeFailure(
      "SCOPE_FENCED",
      "RECOVER",
      false,
    );
    const onEvent = vi.fn(continueEvent);

    await expect(
      runtime.read({
        resumeCursor: null,
        onOpen: () => barrierFailure,
        onEvent,
      }),
    ).resolves.toBe(barrierFailure);
    expect(onEvent).not.toHaveBeenCalled();
  });

  it("treats 204 as terminal and rejects incorrect media types", async () => {
    const terminal = connection(
      (async () => new Response(null, { status: 204 })) as typeof fetch,
    );
    await expect(
      terminal.read({
        resumeCursor: null,
        onEvent: continueEvent,
      }),
    ).resolves.toEqual({
      ok: true,
      value: { kind: "NO_RECONNECT" },
    });

    const wrongType = connection(
      (async () =>
        eventStream(["data: value\n\n"], {
          contentType: "application/json",
        })) as typeof fetch,
    );
    await expect(
      wrongType.read({
        resumeCursor: null,
        onEvent: continueEvent,
      }),
    ).resolves.toEqual({
      ok: false,
      error: {
        kind: "PROTOCOL_MISMATCH",
        operation: "CONNECT",
        retryable: false,
      },
    });
  });

  it.each([
    [401, "AUTH_REQUIRED", false],
    [403, "FORBIDDEN", false],
    [409, "CURSOR_EXPIRED", false],
    [410, "CURSOR_EXPIRED", false],
    [429, "RATE_LIMITED", true],
    [502, "PROVIDER_UNAVAILABLE", true],
    [503, "PROVIDER_UNAVAILABLE", true],
    [504, "PROVIDER_UNAVAILABLE", true],
    [500, "PROTOCOL_MISMATCH", false],
  ] as const)(
    "maps HTTP %i to %s without exposing a response body",
    async (status, kind, retryable) => {
      const runtime = connection(
        (async () =>
          new Response("private backend text", {
            status,
            headers:
              status === 429 || status === 503
                ? { "Retry-After": "10" }
                : undefined,
          })) as typeof fetch,
      );
      const result = await runtime.read({
        resumeCursor: null,
        onEvent: continueEvent,
      });
      expect(result).toEqual({
        ok: false,
        error: {
          kind,
          operation: "CONNECT",
          retryable,
        },
      });
      expect(JSON.stringify(result)).not.toContain("private backend");
    },
  );

  it.each([
    [429, "RATE_LIMITED"],
    [503, "PROVIDER_UNAVAILABLE"],
  ] as const)(
    "does not retry HTTP %i without a bounded server hint",
    async (status, kind) => {
      const runtime = connection(
        (async () => new Response(null, { status })) as typeof fetch,
      );
      await expect(
        runtime.read({
          resumeCursor: null,
          onEvent: continueEvent,
        }),
      ).resolves.toEqual({
        ok: false,
        error: {
          kind,
          operation: "CONNECT",
          retryable: false,
        },
      });
    },
  );

  it("does not clamp an excessive Retry-After into an early retry", async () => {
    const runtime = connection(
      (async () =>
        new Response(null, {
          status: 429,
          headers: { "Retry-After": "120" },
        })) as typeof fetch,
    );
    await expect(
      runtime.read({
        resumeCursor: null,
        onEvent: continueEvent,
      }),
    ).resolves.toEqual({
      ok: false,
      error: {
        kind: "RATE_LIMITED",
        operation: "CONNECT",
        retryable: false,
      },
    });
  });

  it("enforces direct event IDs only for CURSOR recovery", async () => {
    const missingCursorId = connection(
      (async () =>
        eventStream(["data: value\n\n"])) as typeof fetch,
    );
    await expect(
      missingCursorId.read({
        resumeCursor: null,
        onEvent: continueEvent,
      }),
    ).resolves.toEqual({
      ok: false,
      error: {
        kind: "PROTOCOL_MISMATCH",
        operation: "DECODE",
        retryable: false,
      },
    });

    const unexpectedCursorId = connection(
      (async () =>
        eventStream([
          "id: cursor-1\ndata: value\n\n",
        ])) as typeof fetch,
      "SNAPSHOT_ONLY",
    );
    await expect(
      unexpectedCursorId.read({
        resumeCursor: null,
        onEvent: continueEvent,
      }),
    ).resolves.toEqual({
      ok: false,
      error: {
        kind: "PROTOCOL_MISMATCH",
        operation: "DECODE",
        retryable: false,
      },
    });
  });

  it("stops the old stream after authority recovery commits", async () => {
    const runtime = connection(
      (async () =>
        eventStream([
          "id: cursor-1\ndata: first\n\n",
          "id: cursor-2\ndata: stale-generation\n\n",
        ])) as typeof fetch,
    );
    const onEvent = vi.fn(() =>
      realtimeSuccess(RECOVERY_OUTCOME),
    );

    await expect(
      runtime.read({
        resumeCursor: null,
        onEvent,
      }),
    ).resolves.toEqual({
      ok: true,
      value: RECOVERY_OUTCOME,
    });
    expect(onEvent).toHaveBeenCalledOnce();
  });

  it("propagates a canonical consumer failure and aborts its event generation", async () => {
    const runtime = connection(
      (async () =>
        eventStream([
          "id: cursor-1\ndata: forbidden\n\n",
        ])) as typeof fetch,
    );
    let handlerSignal: AbortSignal | undefined;

    await expect(
      runtime.read({
        resumeCursor: null,
        onEvent: (_event, signal) => {
          handlerSignal = signal;
          return realtimeFailure("FORBIDDEN", "RECEIVE");
        },
      }),
    ).resolves.toEqual(
      realtimeFailure("FORBIDDEN", "RECEIVE"),
    );
    expect(handlerSignal?.aborted).toBe(true);
  });

  it("fences recovery immediately and waits for bounded reader cancellation", async () => {
    let releaseCancellation: (() => void) | undefined;
    let cancellationStarted = false;
    const stream = new ReadableStream<Uint8Array>({
      start(controller) {
        controller.enqueue(
          encoder.encode("id: cursor-1\ndata: first\n\n"),
        );
      },
      cancel() {
        cancellationStarted = true;
        return new Promise<void>((resolve) => {
          releaseCancellation = resolve;
        });
      },
    });
    const runtime = connection(
      (async () =>
        new Response(stream, {
          headers: { "Content-Type": "text/event-stream" },
        })) as typeof fetch,
    );
    const pending = runtime.read({
      resumeCursor: null,
      onEvent: () =>
        realtimeSuccess(RECOVERY_OUTCOME),
    });
    let settled = false;
    void pending.then(() => {
      settled = true;
    });
    for (let turn = 0; turn < 20; turn += 1) {
      await Promise.resolve();
    }

    expect(cancellationStarted).toBe(true);
    expect(settled).toBe(false);
    releaseCancellation?.();
    await expect(pending).resolves.toEqual({
      ok: true,
      value: RECOVERY_OUTCOME,
    });
  });

  it("discards incomplete EOF and closes an active reader idempotently", async () => {
    const incomplete = connection(
      (async () =>
        eventStream([
          "id: cursor-1\ndata: incomplete\n",
        ])) as typeof fetch,
    );
    await expect(
      incomplete.read({
        resumeCursor: null,
        onEvent() {
          throw new Error("must not run");
        },
      }),
    ).resolves.toEqual({
      ok: true,
      value: {
        kind: "EOF",
        incompleteEventDiscarded: true,
        retryHintMs: null,
      },
    });

    const pendingStream = new ReadableStream<Uint8Array>({
      pull: () => new Promise<void>(() => {}),
    });
    const active = connection(
      (async () =>
        new Response(pendingStream, {
          headers: { "Content-Type": "text/event-stream" },
        })) as typeof fetch,
    );
    const pending = active.read({
      resumeCursor: null,
      onEvent: continueEvent,
    });
    await Promise.resolve();
    await Promise.resolve();
    active.close();
    active.close();
    await expect(pending).resolves.toMatchObject({
      ok: false,
      error: {
        kind: "ABORTED",
        operation: "CONNECT",
        retryable: false,
      },
    });
    await expect(
      active.read({
        resumeCursor: null,
        onEvent: continueEvent,
      }),
    ).resolves.toEqual({
      ok: false,
      error: {
        kind: "CLOSED",
        operation: "CONNECT",
        retryable: false,
      },
    });
  });

  it("dispatches a CR-terminated blank block completed at EOF", async () => {
    const runtime = connection(
      (async () =>
        eventStream([
          "id: cursor-1\rdata: final\r\r",
        ])) as typeof fetch,
    );
    const received = vi.fn(() =>
      realtimeSuccess(SSE_CONTINUE),
    );
    await expect(
      runtime.read({
        resumeCursor: null,
        onEvent: received,
      }),
    ).resolves.toMatchObject({
      ok: true,
      value: {
        kind: "EOF",
        incompleteEventDiscarded: false,
      },
    });
    expect(received).toHaveBeenCalledWith(
      expect.objectContaining({ data: "final" }),
      expect.any(AbortSignal),
    );
  });

  it("rejects arbitrary endpoints and mode-incoherent cursors before fetch", async () => {
    expect(() =>
      createFetchSseConnection({
        endpoint: "https://other.example.test/events",
        applicationOrigin: "https://app.example.test",
        recoveryMode: "CURSOR",
      }),
    ).toThrow(TypeError);
    expect(() =>
      createFetchSseConnection({
        endpoint:
          "https://app.example.test/events?token=not-allowed",
        applicationOrigin: "https://app.example.test",
        recoveryMode: "CURSOR",
      }),
    ).toThrow(TypeError);

    const fetcher = vi.fn(
      async () => eventStream(["data: unreachable\n\n"]),
    );
    const snapshotOnly = connection(
      fetcher as unknown as typeof fetch,
      "SNAPSHOT_ONLY",
    );
    await expect(
      snapshotOnly.read({
        resumeCursor: "cursor-not-allowed",
        onEvent: continueEvent,
      }),
    ).resolves.toEqual({
      ok: false,
      error: {
        kind: "PROTOCOL_MISMATCH",
        operation: "CONNECT",
        retryable: false,
      },
    });
    expect(fetcher).not.toHaveBeenCalled();
  });
});