package api

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

	"github.com/stretchr/testify/require"
)

// TestStreamBodyReaderHappyPath verifies the watchdog is transparent
// when the source produces data within the idle window. All bytes
// must surface to the caller; the source terminal error (typically
// io.EOF) propagates verbatim.
func TestStreamBodyReaderHappyPath(t *testing.T) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = io.Copy(w, strings.NewReader("hello world"))
	}))
	defer server.Close()

	resp, err := http.DefaultClient.Get(server.URL)
	require.NoError(t, err)
	require.NotNil(t, resp)
	defer resp.Body.Close()

	r := newStreamBodyReaderWithContext(context.Background(), resp.Body, streamBodyConfig{
		IdleTimeout: 5 * time.Second,
	})

	var assembled strings.Builder
	buf := make([]byte, 32)
	for {
		n, err := r.Read(buf)
		if n > 0 {
			assembled.Write(buf[:n])
		}
		if err != nil {
			require.ErrorIs(t, err, io.EOF)
			break
		}
	}
	require.Equal(t, "hello world", assembled.String())
}

// TestStreamBodyReaderIdleTimeoutCancelsUpstreamRequestContext is
// the realistic end-to-end proof: a stalled upstream is interrupted
// by the watchdog cancelling the upstream request context, which
// the standard *http.Transport honors by tearing down the conn and
// unblocking the in-flight body Read. This test must NOT hang.
func TestStreamBodyReaderIdleTimeoutCancelsUpstreamRequestContext(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Write one byte + flush so headers reach the client and Do
		// returns; then stall until the request context is canceled
		// by our watchdog Cancel hook.
		_, _ = w.Write([]byte("x"))
		if f, ok := w.(http.Flusher); ok {
			f.Flush()
		}
		<-r.Context().Done()
	}))
	defer server.Close()

	req, _ := http.NewRequestWithContext(ctx, "GET", server.URL, nil)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)

	r := newStreamBodyReaderWithContext(ctx, resp.Body, streamBodyConfig{
		IdleTimeout: 40 * time.Millisecond,
		Cancel:      cancel,
	})

	buf := make([]byte, 8)
	n, err := r.Read(buf)
	require.NoError(t, err)
	require.Equal(t, 1, n)

	start := time.Now()
	_, err = r.Read(buf)
	elapsed := time.Since(start)
	require.Error(t, err, "blocked Read must return after watchdog cancels")
	require.Less(t, elapsed, 2*time.Second, "transport must tear down conn promptly after ctx cancel")
}

// TestStreamBodyReaderRollingDeadline verifies that each successful
// Read resets the idle window. Multiple slow reads (each below the
// idle window) must all succeed even though their cumulative time
// exceeds one idle window.
func TestStreamBodyReaderRollingDeadline(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		for _, chunk := range []string{"chunk-1", "chunk-2", "chunk-3"} {
			_, _ = w.Write([]byte(chunk))
			if f, ok := w.(http.Flusher); ok {
				f.Flush()
			}
			time.Sleep(30 * time.Millisecond)
		}
	}))
	defer server.Close()

	req, _ := http.NewRequestWithContext(ctx, "GET", server.URL, nil)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)

	r := newStreamBodyReaderWithContext(ctx, resp.Body, streamBodyConfig{
		IdleTimeout: 120 * time.Millisecond,
		Cancel:      cancel,
	})

	var assembled strings.Builder
	buf := make([]byte, 32)
	for {
		n, err := r.Read(buf)
		if n > 0 {
			assembled.Write(buf[:n])
		}
		if err != nil {
			require.ErrorIs(t, err, io.EOF, "must drain to EOF, not idle-timeout")
			break
		}
	}
	require.Equal(t, "chunk-1chunk-2chunk-3", assembled.String())
}

// TestStreamBodyReaderContextCancelUnblocksRead verifies that
// external context cancel (e.g., client disconnect) propagates to
// the watchdog and unblocks a pending Read via Cancel.
func TestStreamBodyReaderContextCancelUnblocksRead(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		if f, ok := w.(http.Flusher); ok {
			f.Flush()
		}
		<-r.Context().Done()
	}))
	defer server.Close()

	req, _ := http.NewRequestWithContext(ctx, "GET", server.URL, nil)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)

	r := newStreamBodyReaderWithContext(ctx, resp.Body, streamBodyConfig{
		IdleTimeout: time.Minute,
		Cancel:      cancel,
	})

	go func() {
		time.Sleep(60 * time.Millisecond)
		cancel()
	}()

	buf := make([]byte, 8)
	start := time.Now()
	_, err = r.Read(buf)
	elapsed := time.Since(start)
	require.Error(t, err)
	require.Less(t, elapsed, 2*time.Second, "Read must return shortly after ctx cancel")
}

// TestStreamBodyReaderCloseViaCancelUnblocksRead verifies that
// Close (not just idle timeout) cancels the upstream request context,
// which the Transport translates into Read interruption. The wrapper
// never writes to the caller buffer outside of Read.
func TestStreamBodyReaderCloseViaCancelUnblocksRead(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		if f, ok := w.(http.Flusher); ok {
			f.Flush()
		}
		<-r.Context().Done()
	}))
	defer server.Close()

	req, _ := http.NewRequestWithContext(ctx, "GET", server.URL, nil)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)

	r := newStreamBodyReaderWithContext(ctx, resp.Body, streamBodyConfig{
		IdleTimeout: time.Minute,
		Cancel:      cancel,
	})

	go func() {
		time.Sleep(60 * time.Millisecond)
		_ = r.Close()
	}()

	buf := make([]byte, 8)
	start := time.Now()
	_, err = r.Read(buf)
	elapsed := time.Since(start)
	require.Error(t, err)
	require.Less(t, elapsed, 2*time.Second, "Close must unblock pending Read via Cancel")
}

// TestStreamBodyReaderOnTimeoutFiresOnce verifies the optional
// OnTimeout callback fires exactly once when the idle window
// elapses, and does NOT fire on plain Close.
func TestStreamBodyReaderOnTimeoutFiresOnce(t *testing.T) {
	var calls atomic.Int32
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte("x"))
		if f, ok := w.(http.Flusher); ok {
			f.Flush()
		}
		<-r.Context().Done()
	}))
	defer server.Close()

	req, _ := http.NewRequestWithContext(ctx, "GET", server.URL, nil)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)

	r := newStreamBodyReaderWithContext(ctx, resp.Body, streamBodyConfig{
		IdleTimeout: 30 * time.Millisecond,
		Cancel:      cancel,
		OnTimeout:   func() { calls.Add(1) },
	})

	buf := make([]byte, 8)
	_, _ = r.Read(buf) // first byte arrives, no timeout
	_, _ = r.Read(buf) // stalls, watchdog fires

	require.Equal(t, int32(1), calls.Load(), "OnTimeout must fire exactly once on idle timeout")
}

// TestStreamBodyReaderCloseDoesNotFireOnTimeout verifies OnTimeout
// is NOT invoked when Close is called externally without an idle
// timeout having fired.
func TestStreamBodyReaderCloseDoesNotFireOnTimeout(t *testing.T) {
	var calls atomic.Int32
	r := newStreamBodyReaderWithContext(context.Background(), &noOpReadCloser{}, streamBodyConfig{
		IdleTimeout: time.Minute,
		OnTimeout:   func() { calls.Add(1) },
	})
	_ = r.Close()
	require.Zero(t, calls.Load(), "Close must NOT fire OnTimeout")
}

// TestStreamBodyReaderNoBufferRaceUnderConcurrentCloseAndRead is
// the regression test for the bbc5fef72 Important #3 defect. Under
// `-race`, this test fails if any detached goroutine writes to the
// caller's buffer outside of Read. The synchronous Read design has
// no such goroutine: only the watchdog exists, and it never touches
// the caller's `p`.
//
// Run with: go test -race -run TestStreamBodyReaderNoBufferRace
func TestStreamBodyReaderNoBufferRaceUnderConcurrentCloseAndRead(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		_, _ = w.Write([]byte("first"))
		if f, ok := w.(http.Flusher); ok {
			f.Flush()
		}
		<-r.Context().Done()
	}))
	defer server.Close()

	req, _ := http.NewRequestWithContext(ctx, "GET", server.URL, nil)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)

	r := newStreamBodyReaderWithContext(ctx, resp.Body, streamBodyConfig{
		IdleTimeout: 30 * time.Millisecond,
		Cancel:      cancel,
	})

	buf := make([]byte, 8)
	n, err := r.Read(buf)
	require.NoError(t, err)
	require.Equal(t, "first", string(buf[:n]))

	// Second Read: source stalls, watchdog cancels ctx, transport
	// unblocks Read. Under -race, any late detached goroutine
	// writing to `buf` would be detected.
	_, err = r.Read(buf)
	require.Error(t, err)
	require.Equal(t, "first", string(buf[:n]), "buf must not be modified by a detached goroutine")
}

// TestStreamBodyReaderSourceErrorPropagates verifies that non-EOF
// errors from the source surface to the caller unchanged. The
// watchdog must not mask upstream errors as timeouts.
func TestStreamBodyReaderSourceErrorPropagates(t *testing.T) {
	sentinel := errors.New("upstream connection reset")
	source := &errorReadCloser{err: sentinel}
	r := newStreamBodyReader(source, streamBodyConfig{
		IdleTimeout: time.Minute,
	})

	buf := make([]byte, 8)
	_, err := r.Read(buf)
	require.ErrorIs(t, err, sentinel, "source error must propagate verbatim")
}

// TestStreamBodyReaderIdleTimeoutDefault verifies that a zero
// IdleTimeout falls back to the package default rather than
// disabling the watchdog (which would reintroduce the hang bug).
func TestStreamBodyReaderIdleTimeoutDefault(t *testing.T) {
	r := newStreamBodyReader(&noOpReadCloser{}, streamBodyConfig{})
	require.Positive(t, r.idleTimeout, "zero IdleTimeout must fall back to a positive default")
	_ = r.Close()
}

// TestStreamBodyReaderAlreadyClosedReturnsClosedSentinel verifies
// that calling Read on an already-closed reader returns the closed
// sentinel without touching the source.
func TestStreamBodyReaderAlreadyClosedReturnsClosedSentinel(t *testing.T) {
	source := &closeCountingReadCloser{}
	r := newStreamBodyReader(source, streamBodyConfig{
		IdleTimeout: time.Minute,
	})
	require.NoError(t, r.Close())
	buf := make([]byte, 8)
	_, err := r.Read(buf)
	require.ErrorIs(t, err, errStreamBodyClosed)
}

// TestStreamBodyReaderConcurrentCloseSafety ensures Close is safe
// to call concurrently from multiple goroutines (gateway teardown
// races against watchdog timer).
func TestStreamBodyReaderConcurrentCloseSafety(t *testing.T) {
	source := &closeCountingReadCloser{}
	r := newStreamBodyReader(source, streamBodyConfig{
		IdleTimeout: time.Minute,
	})

	const goroutines = 8
	var wg sync.WaitGroup
	for range goroutines {
		wg.Add(1)
		go func() {
			defer wg.Done()
			_ = r.Close()
		}()
	}
	wg.Wait()
	require.Equal(t, int32(1), source.closeCount.Load(), "source Close must be called exactly once")
}

// TestStreamBodyReaderPreservesTrailingErrorWithData is the regression
// for the c4dfc8267 Important defect: when source.Read returns
// (n>0, sentinel) — a terminal error with data in flight — the wrapper
// must surface BOTH the bytes AND the sentinel. The prior implementation
// returned (n, nil) and dropped the sentinel, so a half-failed payload
// looked indistinguishable from a clean EOF.
//
// io.Copy / io.ReadAll honor the io.Reader contract: they process n>0
// bytes before considering err, so returning (n, err) together is the
// correct shape.
func TestStreamBodyReaderPreservesTrailingErrorWithData(t *testing.T) {
	sentinel := errors.New("upstream connection reset mid-payload")
	source := &scriptedReadCloser{chunks: []readResult{
		{data: "hello", err: nil},
		{data: "-sent!", err: sentinel},
	}}
	r := newStreamBodyReader(source, streamBodyConfig{
		IdleTimeout: time.Minute,
	})

	var assembled strings.Builder
	buf := make([]byte, 32)
	var seenErr error
	for range 5 {
		n, err := r.Read(buf)
		if n > 0 {
			assembled.Write(buf[:n])
		}
		if err != nil {
			seenErr = err
			break
		}
	}
	require.Equal(t, "hello-sent!", assembled.String(), "must deliver both data chunks")
	require.ErrorIs(t, seenErr, sentinel, "trailing sentinel error must propagate")
}

// TestStreamBodyReaderDoesNotExtendIdleOnZeroRead verifies the rolling
// deadline discipline: only n>0 resets it. A source that returns
// (0, nil) — net/http can do this between chunks of a chunked body —
// must NOT extend the idle window, otherwise a wedged source that
// spins nil reads could starve the watchdog indefinitely.
func TestStreamBodyReaderDoesNotExtendIdleOnZeroRead(t *testing.T) {
	source := &zeroNilReadCloser{}
	timer := newFakeStreamBodyTimer()
	timedOut := make(chan struct{})
	r := newStreamBodyReader(source, streamBodyConfig{
		IdleTimeout: time.Minute,
		OnTimeout:   func() { close(timedOut) },
		NewTimer: func(d time.Duration) streamBodyTimer {
			require.Equal(t, time.Minute, d)
			return timer
		},
	})
	defer r.Close()

	buf := make([]byte, 8)
	_, err := r.Read(buf)
	require.NoError(t, err)
	_, err = r.Read(buf)
	require.NoError(t, err)
	require.Equal(t, int32(0), timer.resetCount.Load(), "zero-byte reads must not reset the idle timer")

	timer.fire()
	select {
	case <-timedOut:
	case <-t.Context().Done():
		require.FailNow(t, "watchdog did not observe the timer signal")
	}
}

// --- test helpers ---

type noOpReadCloser struct{}

func (noOpReadCloser) Read([]byte) (int, error) { return 0, io.EOF }
func (noOpReadCloser) Close() error             { return nil }

type closeCountingReadCloser struct {
	closeCount atomic.Int32
}

func (r *closeCountingReadCloser) Read([]byte) (int, error) { return 0, io.EOF }
func (r *closeCountingReadCloser) Close() error {
	r.closeCount.Add(1)
	return nil
}

type errorReadCloser struct {
	err error
}

func (r *errorReadCloser) Read([]byte) (int, error) { return 0, r.err }
func (r *errorReadCloser) Close() error             { return nil }

// readResult is a single scripted Read return value.
type readResult struct {
	data string
	err  error
}

// scriptedReadCloser replays a fixed sequence of Read results, then
// returns io.EOF forever. Used to drive the wrapper through exact
// shapes (data, trailing-error-with-data, etc.) the real net/http
// body would only emit stochastically.
type scriptedReadCloser struct {
	chunks []readResult
	calls  int
	closed atomic.Bool
}

func (r *scriptedReadCloser) Read(p []byte) (int, error) {
	if r.closed.Load() {
		return 0, errStreamBodyClosed
	}
	if r.calls >= len(r.chunks) {
		return 0, io.EOF
	}
	chunk := r.chunks[r.calls]
	r.calls++
	n := copy(p, chunk.data)
	return n, chunk.err
}

func (r *scriptedReadCloser) Close() error {
	r.closed.Store(true)
	return nil
}

// zeroNilReadCloser returns (0, nil) on every Read — the pathological
// case net/http emits between chunks of a chunked body. Used to prove
// the rolling deadline is NOT extended on byte-less reads.
type zeroNilReadCloser struct {
	closed atomic.Bool
}

type fakeStreamBodyTimer struct {
	c          chan time.Time
	resetCount atomic.Int32
}

func newFakeStreamBodyTimer() *fakeStreamBodyTimer {
	return &fakeStreamBodyTimer{c: make(chan time.Time, 1)}
}

func (t *fakeStreamBodyTimer) C() <-chan time.Time { return t.c }
func (t *fakeStreamBodyTimer) Stop() bool          { return true }
func (t *fakeStreamBodyTimer) Reset(time.Duration) bool {
	t.resetCount.Add(1)
	return true
}
func (t *fakeStreamBodyTimer) fire() { t.c <- time.Now() }

func (r *zeroNilReadCloser) Read([]byte) (int, error) {
	if r.closed.Load() {
		return 0, errStreamBodyClosed
	}
	return 0, nil
}

func (r *zeroNilReadCloser) Close() error {
	r.closed.Store(true)
	return nil
}