import { EventEmitter } from "node:events";
import { afterEach, describe, expect, it, vi } from "vitest";
import { InMemoryEventStore } from "@agent-studio/harness";
import * as sseModule from "../src/sse/sse-connection.js";

class FakeResponse extends EventEmitter {
  chunks: string[] = [];
  ended = false;
  shouldBackpressure = false;

  write(chunk: string): boolean {
    this.chunks.push(chunk);
    return !this.shouldBackpressure;
  }

  end(): void {
    this.ended = true;
    this.emit("close");
  }
}

afterEach(() => vi.useRealTimers());

describe("SseConnection", () => {
  it("replays after Last-Event-ID and gives it precedence over sequence", () => {
    const store = new InMemoryEventStore();
    const first = store.append("ses_1", { type: "session.started", payload: { cwd: "." } });
    const second = store.append("ses_1", { type: "run.started", runId: "run_1", payload: { modelId: "demo" } });
    const third = store.append("ses_1", { type: "assistant.delta", runId: "run_1", payload: { messageId: "msg_1", delta: "hi" } });
    const Response = Reflect.get(sseModule, "SseConnection") as unknown as new (options: Record<string, unknown>) => { start(): void; close(): void };
    expect(Response, "SseConnection must be exported").toBeDefined();
    const response = new FakeResponse();
    const connection = new Response({ sessionId: "ses_1", eventStore: store, response, cursor: { eventId: second.eventId, sequence: first.sequence } });

    connection.start();

    const output = response.chunks.join("");
    expect(output).toContain(`id: ${third.eventId}`);
    expect(output).not.toContain(`id: ${first.eventId}`);
    connection.close();
  });

  it("emits reset for a retained-history gap and streams live events", () => {
    const store = new InMemoryEventStore(1);
    store.append("ses_1", { type: "session.started", payload: { cwd: "." } });
    store.append("ses_1", { type: "run.started", runId: "run_1", payload: { modelId: "demo" } });
    const response = new FakeResponse();
    const Response = Reflect.get(sseModule, "SseConnection") as unknown as new (options: Record<string, unknown>) => { start(): void; close(): void };
    const connection = new Response({ sessionId: "ses_1", eventStore: store, response, cursor: { sequence: 0 } });
    connection.start();
    store.append("ses_1", { type: "assistant.delta", runId: "run_1", payload: { messageId: "msg_1", delta: "live" } });

    const output = response.chunks.join("");
    expect(output).toContain("event: stream.reset");
    expect(output).toContain('"delta":"live"');
    connection.close();
  });

  it("clears its subscriber and heartbeat when closed", () => {
    vi.useFakeTimers();
    const store = new InMemoryEventStore() as InMemoryEventStore & { listenerCount(sessionId: string): number };
    const response = new FakeResponse();
    const Response = Reflect.get(sseModule, "SseConnection") as unknown as new (options: Record<string, unknown>) => { start(): void; close(): void };
    const connection = new Response({ sessionId: "ses_1", eventStore: store, response, heartbeatMs: 15_000 });
    connection.start();
    expect(store.listenerCount("ses_1")).toBe(1);

    connection.close();

    expect(store.listenerCount("ses_1")).toBe(0);
    expect(vi.getTimerCount()).toBe(0);
  });

  it("closes a persistently backpressured client at the configured bound", () => {
    const store = new InMemoryEventStore();
    const response = new FakeResponse();
    response.shouldBackpressure = true;
    const Response = Reflect.get(sseModule, "SseConnection") as unknown as new (options: Record<string, unknown>) => { start(): void };
    const connection = new Response({ sessionId: "ses_1", eventStore: store, response, maxQueuedBytes: 20 });
    connection.start();
    store.append("ses_1", { type: "assistant.delta", runId: "run_1", payload: { messageId: "msg_1", delta: "a long live chunk" } });
    expect(response.ended).toBe(true);
  });
});