package pkg

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

	"go.keploy.io/server/v3/pkg/models"
	"go.keploy.io/server/v3/pkg/provenswitch"
	"go.uber.org/zap"
)

// slowApp is an app whose answer comes only once the test releases it: a
// loaded machine, or an app that waits on a slow dependency.
func slowApp(t *testing.T) (url string, entered <-chan struct{}, release func()) {
	t.Helper()
	in, out := make(chan struct{}, 1), make(chan struct{})
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		in <- struct{}{}
		<-out
		_, _ = w.Write([]byte(`{"ok":true}`))
	}))
	var released bool
	rel := func() {
		if !released {
			released = true
			close(out)
		}
	}
	t.Cleanup(func() { rel(); srv.Close() })
	return srv.URL, in, rel
}

// Without a bound of the user's (Unbounded, under the proven switch: row W1)
// a test waits for the app's answer, however long it takes: an answer that
// comes after what used to be keploy's default is the app's answer. With
// one, the bound holds, as before.
func TestAnUnboundedTestWaitsForTheAppsAnswer(t *testing.T) {
	for _, unbounded := range []bool{true, false} {
		url, entered, release := slowApp(t)
		tc := &models.TestCase{Name: "t", Kind: models.HTTP, HTTPReq: models.HTTPReq{Method: "GET", URL: url + "/x"}}
		done := make(chan error, 1)
		go func() {
			_, err := SimulateHTTP(context.Background(), tc, "set", zap.NewNop(), SimulationConfig{APITimeout: 1, Unbounded: unbounded})
			done <- err
		}()
		<-entered
		if unbounded {
			// The app answers past the 1 s APITimeout (induced load).
			time.Sleep(1500 * time.Millisecond)
			release()
			if err := <-done; err != nil {
				t.Fatalf("an unbounded test gave up on the app's late answer: %v", err)
			}
			continue
		}
		if err := <-done; err == nil {
			t.Fatal("a test bounded by APITimeout waited past it")
		}
		release()
	}
}

// The same for a gRPC call: unbounded, the response drain ends on the app's
// end of the stream, however late it comes.
func TestAnUnboundedGRPCTestWaitsForTheAppsEndOfStream(t *testing.T) {
	hold := make(chan struct{})
	addr := startStreamingServer(t, &streamingServer{respond: [][]byte{pbMsg("first")}, holdOpen: hold})
	tc := grpcTestCase(addr, []models.GrpcLengthPrefixedMessage{{}})
	done := make(chan error, 1)
	go func() {
		_, err := SimulateGRPC(context.Background(), tc, "set", zap.NewNop(), SimulationConfig{APITimeout: 1, Unbounded: true})
		done <- err
	}()
	// The app ends its stream past the 1 s APITimeout (induced load).
	time.Sleep(1500 * time.Millisecond)
	close(hold)
	if err := <-done; err != nil {
		t.Fatalf("an unbounded gRPC test gave up on the app's late end of stream: %v", err)
	}
}

// errReadStopped is what a write to a stream the comparison stopped reading
// returns.
var errReadStopped = errors.New("the comparison stopped reading")

// heldOpen is what CompareHTTPStream made of a stream the app holds open past
// its recording, and how many keepalives beyond the recording it read.
type heldOpen struct {
	ok   bool
	info *StreamMismatchInfo
	err  error
	read int
}

// readHeldOpen streams body to CompareHTTPStream and then keepalive, one write
// at a time, as an app that keeps the stream open past its recording does,
// until the comparison stops reading. A write returns once the comparison has
// read it, so read counts the keepalives it took: no clock. One that takes
// every one of limit and still reads is the hang (row W1): the app's stream
// is then ended for it, and the test fails.
func readHeldOpen(t *testing.T, expected models.HTTPResp, cfg HTTPStreamConfig, body, keepalive string, limit int) heldOpen {
	t.Helper()
	pr, pw := io.Pipe()
	done := make(chan heldOpen, 1)
	go func() {
		ok, _, info, err := CompareHTTPStream(expected, pr, cfg, map[string]struct{}{}, zap.NewNop())
		_ = pr.CloseWithError(errReadStopped)
		done <- heldOpen{ok: ok, info: info, err: err}
	}()
	if _, err := pw.Write([]byte(body)); err != nil {
		t.Fatalf("the comparison stopped reading inside the recorded frames: %v", err)
	}
	n := 0
	for ; n < limit; n++ {
		if _, err := pw.Write([]byte(keepalive)); err != nil {
			break
		}
	}
	if n == limit {
		_ = pw.Close()
		<-done
		t.Fatalf("the comparison read %d keepalives beyond the recording and still waited: with no clock it waits as long as the app keeps the stream open", limit)
	}
	got := <-done
	got.read = n
	return got
}

// Under the proven switch, with no clock on the app's answer (row W1), a
// stream the app keeps open past its recording ends on its first keepalive
// beyond the recorded ones: the read does not wait for an end that never
// comes, and the test has no verdict (ErrStreamHeldOpen), never a pass. The
// recorded keepalives are the recording's: an SSE keepalive it holds is one
// of its frames. Bytes beyond a recorded binary stream fail it at once. A
// stream that ends with its recording passes, as before.
func TestAStreamHeldOpenPastItsRecordingEndsOnItsKeepalives(t *testing.T) {
	sse := HTTPStreamConfig{Mode: HTTPStreamModeSSE}
	ndjson := HTTPStreamConfig{Mode: HTTPStreamModeNDJSON}
	sseWithPing := models.HTTPResp{StreamBody: append(frames("data", "a", "b"),
		models.HTTPStreamChunk{Data: []models.HTTPStreamDataField{{Key: "comment", Value: " ping"}}})}
	for _, tc := range []struct {
		name      string
		cfg       HTTPStreamConfig
		expected  models.HTTPResp
		body      string
		keepalive string
	}{
		{"SSE", sse, models.HTTPResp{StreamBody: frames("data", "a", "b")}, "data: a\n\ndata: b\n\n", ": ping\n\n"},
		{"SSE with a recorded keepalive", sse, sseWithPing, "data: a\n\ndata: b\n\n: ping\n\n", ": ping\n\n"},
		{"NDJSON", ndjson, models.HTTPResp{StreamBody: frames("raw", `{"id":1}`, `{"id":2}`)}, "{\"id\":1}\n{\"id\":2}\n", "\n"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			t.Setenv(provenswitch.Env, "1")
			got := readHeldOpen(t, tc.expected, tc.cfg, tc.body, tc.keepalive, 50)
			if !errors.Is(got.err, ErrStreamHeldOpen) {
				t.Fatalf("a stream held open past its recording gave %v (ok %v, %+v), want ErrStreamHeldOpen", got.err, got.ok, got.info)
			}
			if got.read != 1 {
				t.Fatalf("the read ended on keepalive %d beyond the recording, want the first", got.read)
			}
			if !got.ok || got.info != nil {
				t.Fatalf("the recorded frames that came did not match: %+v", got.info)
			}

			// A stream that ends with its recording passes.
			ok, _, info, err := CompareHTTPStream(tc.expected, strings.NewReader(tc.body), tc.cfg, map[string]struct{}{}, zap.NewNop())
			if !ok || err != nil {
				t.Fatalf("a stream that ended with its recording gave %v, %v (%+v)", ok, err, info)
			}

			// Without the switch the comparison stops at the last recorded
			// frame and reads no keepalive, as before.
			t.Setenv(provenswitch.Env, "")
			if got := readHeldOpen(t, tc.expected, tc.cfg, tc.body, tc.keepalive, 50); !got.ok || got.err != nil || got.read != 0 {
				t.Fatalf("without the switch: %+v, want a match that read no keepalive", got)
			}
		})
	}

	// Short of its recording: two keepalives in a row with no line between
	// them, and the app stalled (ErrStreamStalled, an ErrStreamHeldOpen):
	// no verdict, the lines that came having matched. One keepalive between
	// two lines is none.
	t.Run("NDJSON short of its recording", func(t *testing.T) {
		t.Setenv(provenswitch.Env, "1")
		expected := models.HTTPResp{StreamBody: frames("raw", `{"id":1}`, `{"id":2}`)}
		got := readHeldOpen(t, expected, ndjson, "{\"id\":1}\n", "\n", 50)
		if !errors.Is(got.err, ErrStreamStalled) || !errors.Is(got.err, ErrStreamHeldOpen) || !got.ok {
			t.Fatalf("a stream stalled short of its recording gave %+v, want ErrStreamStalled", got)
		}
		if got.read != 2 {
			t.Fatalf("the read ended on keepalive %d, want the second", got.read)
		}
		ok, _, info, err := CompareHTTPStream(expected, strings.NewReader("{\"id\":1}\n\n{\"id\":2}\n"), ndjson, map[string]struct{}{}, zap.NewNop())
		if !ok || err != nil {
			t.Fatalf("one keepalive between two lines gave %v, %v (%+v)", ok, err, info)
		}
	})

	t.Run("binary", func(t *testing.T) {
		t.Setenv(provenswitch.Env, "1")
		got := readHeldOpen(t, models.HTTPResp{StreamBody: frames("raw", "abcd")}, HTTPStreamConfig{Mode: HTTPStreamModeBinary}, "abcd", "ef", 50)
		if got.ok || got.err != nil || got.info == nil {
			t.Fatalf("bytes beyond a recorded binary stream gave %+v, want a size mismatch", got)
		}
		if got.read != 1 {
			t.Fatalf("the read ended %d writes beyond the recording, want the first", got.read)
		}
	})
}