package agent

import (
	"context"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/desktop/internal/agent/drivercontract"
)

func TestForwardOpenCodeTranslatedLinesBoundsOversizedLine(t *testing.T) {
	lines := make(chan string, 1)
	reader, writer := io.Pipe()
	stop := make(chan struct{})
	abort := make(chan struct{})
	oversizedBytes := make(chan int, 1)
	done := make(chan struct{})
	go func() {
		defer close(done)
		forwardOpenCodeTranslatedLines(lines, writer, stop, abort, func(bytes int) string {
			oversizedBytes <- bytes
			return `{"kind":"error","message":"translated event exceeded process line limit"}`
		})
	}()
	lines <- strings.Repeat("x", maxProcessLineBytes)
	close(stop)

	payload, err := io.ReadAll(reader)
	require.NoError(t, err)
	<-done
	require.Equal(t, maxProcessLineBytes+1, <-oversizedBytes)
	require.Equal(t, `{"kind":"error","message":"translated event exceeded process line limit"}`+"\n", string(payload))
	require.Empty(t, lines)
}

func TestOpenCodeConnectSSEReportsCloseBeforeFirstDataFrame(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/event", r.URL.Path)
		w.Header().Set("Content-Type", "text/event-stream")
		w.WriteHeader(http.StatusOK)
		if flusher, ok := w.(http.Flusher); ok {
			flusher.Flush()
		}
	}))
	defer server.Close()

	driver := &OpenCodeDriver{
		serverURL:  server.URL,
		password:   "pw",
		httpClient: server.Client(),
		workDir:    t.TempDir(),
	}
	ready := make(chan openCodeSSEReadyEvent, 1)

	err := driver.connectSSE(context.Background(), drivercontract.LaunchConfig{WorkDir: driver.workDir}, ready)

	require.Error(t, err)
	require.Contains(t, err.Error(), "SSE stream closed before first data frame")
	select {
	case ev := <-ready:
		require.False(t, ev.ready)
		require.Error(t, ev.err)
		require.Contains(t, ev.err.Error(), "SSE stream closed before first data frame")
	default:
		t.Fatal("connectSSE did not signal pre-ready close")
	}
	require.True(t, driver.sseDeliveryBlocked)
}

func TestOpenCodeConnectSSEEmitsRuntimeDebugLifecycle(t *testing.T) {
	firstFrameFlushed := make(chan struct{})
	releaseClose := make(chan struct{})
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		require.Equal(t, "/event", r.URL.Path)
		w.Header().Set("Content-Type", "text/event-stream")
		w.WriteHeader(http.StatusOK)
		if flusher, ok := w.(http.Flusher); ok {
			_, _ = w.Write([]byte(`data: {"type":"server.connected","properties":{}}` + "\n\n"))
			flusher.Flush()
		}
		close(firstFrameFlushed)
		<-releaseClose
	}))
	defer server.Close()

	driver := &OpenCodeDriver{
		serverURL:           server.URL,
		password:            "pw",
		httpClient:          server.Client(),
		workDir:             t.TempDir(),
		translatedLines:     make(chan string, 16),
		runtimeDebugEnabled: true,
	}
	debugLogPath := attachOpenCodeRuntimeDebugLogForTest(t, driver, driver.workDir)
	ready := make(chan openCodeSSEReadyEvent, 1)
	done := make(chan error, 1)
	go func() {
		done <- driver.connectSSE(context.Background(), drivercontract.LaunchConfig{AgentID: "agt_debug", WorkDir: driver.workDir}, ready)
	}()

	select {
	case ev := <-ready:
		require.True(t, ev.ready)
	case <-time.After(time.Second):
		t.Fatal("connectSSE did not signal ready")
	}
	require.False(t, driver.sseDeliveryBlocked)
	<-firstFrameFlushed
	close(releaseClose)
	require.ErrorIs(t, <-done, errOpenCodeSSEReplayGap)
	require.True(t, driver.sseDeliveryBlocked)

	var messages []string
	for _, event := range readOpenCodeRuntimeDebugEntries(t, debugLogPath) {
		messages = append(messages, event.Event)
	}
	require.Contains(t, messages, "opencode sse connect started")
	require.Contains(t, messages, "opencode sse first data received")
	require.Contains(t, messages, "opencode sse stream closed")
	require.Empty(t, driver.translatedLines)
}

func TestOpenCodeSubscribeSSERetiresGenerationAfterReplayGap(t *testing.T) {
	var requests atomic.Int32
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		requests.Add(1)
		require.Equal(t, "/event", r.URL.Path)
		w.Header().Set("Content-Type", "text/event-stream")
		w.WriteHeader(http.StatusOK)
		if flusher, ok := w.(http.Flusher); ok {
			_, _ = w.Write([]byte(`data: {"type":"server.connected","properties":{}}` + "\n\n"))
			flusher.Flush()
		}
	}))
	defer server.Close()

	retired := make(chan struct{})
	summarizeFence := &openCodeSummarizeBoundaryFence{
		sessionID:           "sess-1",
		sessionGeneration:   7,
		summarizeGeneration: 3,
		idleCh:              make(chan struct{}),
		retiredCh:           make(chan struct{}),
	}
	driver := &OpenCodeDriver{
		serverURL:           server.URL,
		password:            "pw",
		workDir:             t.TempDir(),
		sessionID:           "sess-1",
		sessionGeneration:   7,
		summarizeGeneration: 3,
		summarizeInProgress: true,
		summarizeFence:      summarizeFence,
		translatedLines:     make(chan string, 4),
		serverCancel:        sync.OnceFunc(func() { close(retired) }),
	}
	ready := make(chan openCodeSSEReadyEvent, 1)
	driver.subscribeSSE(context.Background(), drivercontract.LaunchConfig{
		AgentID: "agt-gap",
		WorkDir: driver.workDir,
	}, ready)

	select {
	case <-retired:
	default:
		t.Fatal("post-ready SSE loss did not retire the runtime generation")
	}
	require.Equal(t, int32(1), requests.Load(), "replayless SSE loss must not reconnect and reopen delivery")
	require.True(t, driver.sseDeliveryBlocked)
	require.True(t, driver.generationRetiring)
	require.Equal(t, uint64(8), driver.sessionGeneration)
	require.Equal(t, uint64(4), driver.summarizeGeneration)
	select {
	case <-summarizeFence.retiredCh:
	default:
		t.Fatal("SSE retirement did not invalidate in-flight summarize completion")
	}
}