import { renderHook, waitFor } from '@testing-library/react'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { useSseStream } from './useSseStream'

function sseResponse(chunks: string[], ok = true): Response {
  const stream = new ReadableStream<Uint8Array>({
    start(controller) {
      for (const chunk of chunks) controller.enqueue(new TextEncoder().encode(chunk))
      controller.close()
    },
  })
  return { ok, body: stream } as unknown as Response
}

function reconnectingFetch(chunks: string[]) {
  return vi
    .fn()
    .mockResolvedValueOnce(sseResponse(chunks))
    .mockResolvedValue(new Promise(() => {}))
}

describe('useSseStream', () => {
  afterEach(() => {
    vi.restoreAllMocks()
  })

  it('stays idle and never fetches when url is null', () => {
    const fetchImpl = vi.fn()
    const { result } = renderHook(() =>
      useSseStream({ url: null, onMessage: vi.fn(), fetchImpl: fetchImpl as unknown as typeof fetch }),
    )
    expect(result.current).toBe('idle')
    expect(fetchImpl).not.toHaveBeenCalled()
  })

  it('parses split multiline events and ignores heartbeat comments', async () => {
    const onMessage = vi.fn()
    const onEvent = vi.fn()
    const fetchImpl = reconnectingFetch([
      ': heartbeat\r\n\r\n',
      'id: opaque/',
      'cursor\r\nevent: item\r\ndata: {"type":"item",\r\ndata: "item":{"id":"a"}}\r\n\r\n',
    ])
    renderHook(() =>
      useSseStream({
        url: 'https://collector.test/events',
        headers: { authorization: 'Bearer tok' },
        onMessage,
        onEvent,
        fetchImpl: fetchImpl as unknown as typeof fetch,
      }),
    )

    await waitFor(() => expect(onMessage).toHaveBeenCalledWith({ type: 'item', item: { id: 'a' } }))
    expect(onEvent).toHaveBeenCalledWith({
      id: 'opaque/cursor',
      event: 'item',
      data: { type: 'item', item: { id: 'a' } },
    })
    expect(fetchImpl).toHaveBeenCalledWith(
      'https://collector.test/events',
      expect.objectContaining({ headers: { authorization: 'Bearer tok' } }),
    )
  })

  it('parses multiple frames delivered in one chunk', async () => {
    const onMessage = vi.fn()
    const fetchImpl = reconnectingFetch(['data: {"type":"item-resolved","id":"a"}\n\ndata: {"type":"item-resolved","id":"b"}\n\n'])
    renderHook(() =>
      useSseStream({ url: 'https://collector.test/events', onMessage, fetchImpl: fetchImpl as unknown as typeof fetch }),
    )

    await waitFor(() => expect(onMessage).toHaveBeenCalledTimes(2))
    expect(onMessage).toHaveBeenNthCalledWith(1, { type: 'item-resolved', id: 'a' })
    expect(onMessage).toHaveBeenNthCalledWith(2, { type: 'item-resolved', id: 'b' })
  })

  it('terminates malformed data frames before later events or reconnects', async () => {
    const onMessage = vi.fn()
    const onStatus = vi.fn()
    const fetchImpl = vi.fn().mockResolvedValue(sseResponse(['id: malformed\ndata: not-json\n\nid: accepted\ndata: {"type":"panel"}\n\n']))
    const { result } = renderHook(() =>
      useSseStream({
        url: 'https://collector.test/events',
        onMessage,
        onStatus,
        retryDelayMs: 1,
        fetchImpl: fetchImpl as unknown as typeof fetch,
      }),
    )

    await waitFor(() => expect(result.current).toBe('closed'))
    expect(onMessage).not.toHaveBeenCalled()
    await new Promise((resolve) => setTimeout(resolve, 20))
    expect(fetchImpl).toHaveBeenCalledTimes(1)
    expect(onStatus).toHaveBeenCalledWith('closed')
  })

  it('resumes from the last delivered opaque ID and suppresses replayed events', async () => {
    const onMessage = vi.fn()
    const fetchImpl = reconnectingFetch([
      'id: cursor/1?x=y\ndata: {"id":"first"}\n\n',
    ])
    fetchImpl.mockResolvedValueOnce(sseResponse(['id: cursor/1?x=y\ndata: {"id":"duplicate"}\n\nid: cursor-2\ndata: {"id":"next"}\n\n']))

    const { unmount } = renderHook(() =>
      useSseStream({
        url: 'https://collector.test/events',
        headers: { authorization: 'Bearer tok' },
        onMessage,
        retryDelayMs: 1,
        fetchImpl: fetchImpl as unknown as typeof fetch,
      }),
    )

    await waitFor(() => expect(onMessage).toHaveBeenCalledTimes(2))
    expect(onMessage).toHaveBeenNthCalledWith(1, { id: 'first' })
    expect(onMessage).toHaveBeenNthCalledWith(2, { id: 'next' })
    expect(fetchImpl.mock.calls[1]?.[1]).toEqual(expect.objectContaining({
      headers: { authorization: 'Bearer tok', 'Last-Event-ID': 'cursor/1?x=y' },
    }))
    unmount()
  })

  it('reconnects after the exact fixture frame closes and resumes its opaque cursor', async () => {
    const onMessage = vi.fn()
    const fetchImpl = vi
      .fn()
      .mockResolvedValueOnce(sseResponse([
        'id: opaque:0003\nevent: task\ndata: {"id":"opaque:0003","source":"fixture","kind":"transcript.chunk","ts":"2026-07-22T00:00:00.000Z","taskId":"t4","attemptId":"attempt-2","payload":{"text":"stream one"}}\n\n',
      ]))
      .mockResolvedValue(new Promise(() => {}))

    const { unmount } = renderHook(() =>
      useSseStream({
        url: 'https://collector.test/events',
        headers: { authorization: 'Bearer tok' },
        onMessage,
        retryDelayMs: 1,
        fetchImpl: fetchImpl as unknown as typeof fetch,
      }),
    )

    await waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(2))
    expect(onMessage).toHaveBeenCalledWith(expect.objectContaining({ id: 'opaque:0003', kind: 'transcript.chunk' }))
    expect(fetchImpl.mock.calls[1]?.[1]).toEqual(expect.objectContaining({
      headers: { authorization: 'Bearer tok', 'Last-Event-ID': 'opaque:0003' },
    }))
    unmount()
  })

  it('reconnects after the stream ends, until unmounted', async () => {
    const onStatus = vi.fn()
    const fetchImpl = vi.fn().mockResolvedValue(sseResponse([]))
    const { unmount } = renderHook(() =>
      useSseStream({
        url: 'https://collector.test/events',
        onMessage: vi.fn(),
        onStatus,
        retryDelayMs: 1,
        fetchImpl: fetchImpl as unknown as typeof fetch,
      }),
    )

    await waitFor(() => expect(fetchImpl.mock.calls.length).toBeGreaterThanOrEqual(2))
    unmount()
    const callsAtUnmount = fetchImpl.mock.calls.length
    await new Promise((resolve) => setTimeout(resolve, 20))
    expect(fetchImpl.mock.calls.length).toBe(callsAtUnmount)
    expect(onStatus).toHaveBeenCalledWith('connecting')
    expect(onStatus).toHaveBeenCalledWith('open')
    expect(onStatus).toHaveBeenCalledWith('closed')
  })

  it('does not reconnect after consumer closes the stream with url null', async () => {
    let controller: ReadableStreamDefaultController<Uint8Array> | undefined
    const fetchImpl = vi.fn().mockResolvedValue({
      ok: true,
      body: new ReadableStream<Uint8Array>({
        start(nextController) {
          controller = nextController
        },
      }),
    } as unknown as Response)
    const { rerender } = renderHook(
      ({ url }) => useSseStream({
        url,
        onMessage: vi.fn(),
        retryDelayMs: 1,
        fetchImpl: fetchImpl as unknown as typeof fetch,
      }),
      { initialProps: { url: 'https://collector.test/events' as string | null } },
    )

    await waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1))
    rerender({ url: null })
    controller?.close()
    await new Promise((resolve) => setTimeout(resolve, 20))
    expect(fetchImpl).toHaveBeenCalledTimes(1)
  })

  it('closes an oversized delimiter-free frame without reconnecting', async () => {
    const onStatus = vi.fn()
    const fetchImpl = vi.fn().mockResolvedValue(sseResponse(['x'.repeat(65_537)]))
    const { result } = renderHook(() => useSseStream({
      url: 'https://collector.test/events', onMessage: vi.fn(), onStatus, retryDelayMs: 1,
      fetchImpl: fetchImpl as unknown as typeof fetch,
    }))

    await waitFor(() => expect(result.current).toBe('closed'))
    await new Promise((resolve) => setTimeout(resolve, 20))
    expect(fetchImpl).toHaveBeenCalledTimes(1)
    expect(onStatus).toHaveBeenCalledWith('closed')
  })

  it('accepts a frame exactly at the byte bound', async () => {
    const onMessage = vi.fn()
    const payload = JSON.stringify({ text: 'x'.repeat(65_519) })
    const fetchImpl = reconnectingFetch([`data: ${payload}\n\n`])
    renderHook(() => useSseStream({ url: 'https://collector.test/events', onMessage, fetchImpl: fetchImpl as unknown as typeof fetch }))

    await waitFor(() => expect(onMessage).toHaveBeenCalledWith({ text: 'x'.repeat(65_519) }))
  })

  it('closes an oversized delimited frame before parsing or delivery', async () => {
    const onMessage = vi.fn()
    const fetchImpl = vi.fn().mockResolvedValue(sseResponse([`data: ${'x'.repeat(65_537)}\n\n`]))
    const { result } = renderHook(() => useSseStream({
      url: 'https://collector.test/events', onMessage, retryDelayMs: 1,
      fetchImpl: fetchImpl as unknown as typeof fetch,
    }))

    await waitFor(() => expect(result.current).toBe('closed'))
    expect(onMessage).not.toHaveBeenCalled()
    await new Promise((resolve) => setTimeout(resolve, 20))
    expect(fetchImpl).toHaveBeenCalledTimes(1)
  })

  it('aborts the in-flight request on unmount', () => {
    let capturedSignal: AbortSignal | undefined
    const fetchImpl = vi.fn().mockImplementation((_url: string, init?: RequestInit) => {
      capturedSignal = init?.signal ?? undefined
      return new Promise(() => {})
    })
    const { unmount } = renderHook(() =>
      useSseStream({ url: 'https://collector.test/events', onMessage: vi.fn(), fetchImpl: fetchImpl as unknown as typeof fetch }),
    )
    expect(capturedSignal?.aborted).toBe(false)
    unmount()
    expect(capturedSignal?.aborted).toBe(true)
  })
})