package sourcereading

import (
	"bufio"
	"context"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/http/httptest"
	"net/http/httptrace"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/redis/go-redis/v9"
	"go.etcd.io/raft/v3"
	pb "go.etcd.io/raft/v3/raftpb"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	healthpb "google.golang.org/grpc/health/grpc_health_v1"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
)

func waitSignal(t *testing.T, ch <-chan struct{}) {
	t.Helper()
	select {
	case <-ch:
	case <-time.After(5 * time.Second):
		t.Fatal("signal timeout")
	}
}

func TestHTTP(t *testing.T) {
	contexts := make(chan context.Context, 2)
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		contexts <- r.Context()
		w.WriteHeader(http.StatusTeapot)
		_, _ = io.WriteString(w, "tea")
	}))
	defer srv.Close()
	tr := &http.Transport{MaxIdleConnsPerHost: 1}
	defer tr.CloseIdleConnections()
	client := &http.Client{Transport: tr, Timeout: 5 * time.Second}
	reused := make([]bool, 0, 2)
	for range 2 {
		// Each request has one connection acquisition; channel transfer avoids a trace callback race.
		got := make(chan bool, 1)
		trace := &httptrace.ClientTrace{GotConn: func(i httptrace.GotConnInfo) { got <- i.Reused }}
		req, err := http.NewRequestWithContext(httptrace.WithClientTrace(context.Background(), trace), "GET", srv.URL, nil)
		if err != nil {
			t.Fatal(err)
		}
		resp, err := client.Do(req)
		if err != nil {
			t.Fatal(err)
		}
		body, readErr := io.ReadAll(resp.Body)
		closeErr := resp.Body.Close()
		if readErr != nil || closeErr != nil || resp.StatusCode != 418 || string(body) != "tea" {
			t.Fatalf("status=%d body=%q read=%v close=%v", resp.StatusCode, body, readErr, closeErr)
		}
		reused = append(reused, <-got)
		serverCtx := <-contexts
		waitSignal(t, serverCtx.Done())
		if !errors.Is(serverCtx.Err(), context.Canceled) {
			t.Fatal(serverCtx.Err())
		}
	}
	if !slices.Equal(reused, []bool{false, true}) {
		t.Fatal(reused)
	}
	t.Logf("status=418 transport_error=nil reused=%v handler_context=context canceled", reused)
}

func TestOnce(t *testing.T) {
	var once sync.Once
	var calls atomic.Int32
	var wg sync.WaitGroup
	var value int
	var bad atomic.Int32
	for range 32 {
		wg.Go(func() {
			once.Do(func() { calls.Add(1); value = 42 })
			if value != 42 {
				bad.Add(1)
			}
		})
	}
	wg.Wait()
	if calls.Load() != 1 || bad.Load() != 0 {
		t.Fatal(calls.Load(), bad.Load())
	}
	var panicOnce sync.Once
	panicCalls := 0
	func() {
		defer func() {
			if recover() != "boom" {
				t.Error("missing panic")
			}
		}()
		panicOnce.Do(func() { panicCalls++; panic("boom") })
	}()
	panicOnce.Do(func() { panicCalls++ })
	if panicCalls != 1 {
		t.Fatal(panicCalls)
	}
	t.Logf("initializer_calls=%d readers=32 wrong_values=%d panic_calls=%d", calls.Load(), bad.Load(), panicCalls)
}

// This deliberately wrong implementation records "started", not "finished".
func TestBrokenOnceCounterexample(t *testing.T) {
	var started atomic.Bool
	entered, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{})
	defer close(release)
	brokenDo := func(f func()) {
		if started.CompareAndSwap(false, true) {
			f()
		}
	}
	go func() {
		brokenDo(func() { close(entered); <-release })
		close(finished)
	}()
	waitSignal(t, entered)
	brokenDo(func() { t.Error("unexpected second initializer") })
	select {
	case <-finished:
		t.Fatal("first initializer unexpectedly finished")
	default:
		t.Log("broken_second_call_returned=true first_initializer_finished=false")
	}
	// The release deferred above lets the goroutine exit; join in cleanup.
	t.Cleanup(func() { waitSignal(t, finished) })
}

func TestContext(t *testing.T) {
	parent, cancelParent := context.WithCancelCause(context.Background())
	defer cancelParent(nil)
	child, cancelChild := context.WithCancelCause(parent)
	defer cancelChild(nil)
	sibling, cancelSibling := context.WithCancel(parent)
	defer cancelSibling()
	detached := context.WithoutCancel(parent)
	cancelChild(errors.New("child stopped"))
	parentBefore := parent.Err()
	cancelParent(errors.New("request stopped"))
	waitSignal(t, sibling.Done())
	if parentBefore != nil || context.Cause(child).Error() != "child stopped" || context.Cause(sibling).Error() != "request stopped" || detached.Done() != nil {
		t.Fatal("unexpected cancellation")
	}
	t.Logf("parent_before=%v child_cause=%q sibling_cause=%q detached_done_nil=%v", parentBefore, context.Cause(child), context.Cause(sibling), detached.Done() == nil)
}

func TestGin(t *testing.T) {
	gin.SetMode(gin.TestMode)
	r := gin.New()
	var events []string
	r.Use(func(c *gin.Context) {
		events = append(events, "outer-before")
		c.Next()
		events = append(events, "outer-after")
	})
	r.Use(func(c *gin.Context) {
		events = append(events, "auth")
		if c.Query("deny") == "1" {
			c.AbortWithStatus(401)
			events = append(events, "after-abort")
			return
		}
		c.Next()
	})
	var copied *gin.Context
	r.GET("/users/:id", func(c *gin.Context) {
		events = append(events, "handler")
		data := []int{1}
		c.Set("data", data)
		copied = c.Copy()
		data[0] = 9
		c.String(200, c.Param("id"))
	})
	for _, target := range []string{"/users/7", "/users/7?deny=1"} {
		events = nil
		w := httptest.NewRecorder()
		r.ServeHTTP(w, httptest.NewRequest("GET", target, nil))
		expected := []string{"outer-before", "auth", "handler", "outer-after"}
		wantStatus := 200
		if strings.Contains(target, "deny") {
			expected = []string{"outer-before", "auth", "after-abort", "outer-after"}
			wantStatus = 401
		}
		if w.Code != wantStatus || !slices.Equal(events, expected) {
			t.Fatal(w.Code, events)
		}
		t.Logf("status=%d events=%v", w.Code, events)
	}
	data, _ := copied.Get("data")
	if copied.Param("id") != "7" || data.([]int)[0] != 9 {
		t.Fatal(copied.Params, data)
	}
	t.Logf("copy_param=%s copied_slice=%v", copied.Param("id"), data)
}

type probeHealth struct {
	healthpb.UnimplementedHealthServer
	entered chan struct{}
	stopped chan struct{}
}

func (s *probeHealth) Check(ctx context.Context, r *healthpb.HealthCheckRequest) (*healthpb.HealthCheckResponse, error) {
	switch r.Service {
	case "fail":
		return nil, status.Error(codes.Unavailable, "planned failure")
	case "wait":
		close(s.entered)
		<-ctx.Done()
		close(s.stopped)
		return nil, status.FromContextError(ctx.Err()).Err()
	default:
		return &healthpb.HealthCheckResponse{Status: healthpb.HealthCheckResponse_SERVING}, nil
	}
}
func TestGRPC(t *testing.T) {
	listener := bufconn.Listen(1 << 20)
	var intercepted atomic.Int32
	srv := grpc.NewServer(grpc.UnaryInterceptor(func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
		intercepted.Add(1)
		return handler(ctx, req)
	}))
	svc := &probeHealth{entered: make(chan struct{}), stopped: make(chan struct{})}
	healthpb.RegisterHealthServer(srv, svc)
	serveDone := make(chan struct{})
	go func() { defer close(serveDone); _ = srv.Serve(listener) }()
	defer func() { srv.Stop(); _ = listener.Close(); waitSignal(t, serveDone) }()
	cc, err := grpc.NewClient("passthrough:///memory", grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) { return listener.DialContext(ctx) }))
	if err != nil {
		t.Fatal(err)
	}
	defer cc.Close()
	client := healthpb.NewHealthClient(cc)
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	result, err := client.Check(ctx, &healthpb.HealthCheckRequest{})
	if err != nil || result.Status != healthpb.HealthCheckResponse_SERVING {
		t.Fatal(result, err)
	}
	_, failErr := client.Check(ctx, &healthpb.HealthCheckRequest{Service: "fail"})
	if status.Code(failErr) != codes.Unavailable {
		t.Fatal(failErr)
	}
	callCtx, cancelCall := context.WithCancel(ctx)
	defer cancelCall()
	resultCh := make(chan error, 1)
	go func() { _, e := client.Check(callCtx, &healthpb.HealthCheckRequest{Service: "wait"}); resultCh <- e }()
	waitSignal(t, svc.entered)
	cancelCall()
	var canceledErr error
	select {
	case canceledErr = <-resultCh:
	case <-ctx.Done():
		t.Fatal(ctx.Err())
	}
	waitSignal(t, svc.stopped)
	if status.Code(canceledErr) != codes.Canceled || intercepted.Load() != 3 {
		t.Fatal(canceledErr, intercepted.Load())
	}
	t.Logf("normal=%s failure=%s canceled=%s server_observed_cancel=true interceptions=%d", result.Status, status.Code(failErr), status.Code(canceledErr), intercepted.Load())
}

func TestRaft(t *testing.T) {
	storage := raft.NewMemoryStorage()
	// A fixture snapshot establishes a one-voter group. It is not a recovered disk snapshot.
	snapshot := pb.Snapshot{Metadata: pb.SnapshotMetadata{Index: 1, Term: 1, ConfState: pb.ConfState{Voters: []uint64{1}}}}
	if err := storage.ApplySnapshot(snapshot); err != nil {
		t.Fatal(err)
	}
	if err := storage.SetHardState(pb.HardState{Term: 1, Commit: 1}); err != nil {
		t.Fatal(err)
	}
	rn, err := raft.NewRawNode(&raft.Config{ID: 1, Storage: storage, Applied: 1, ElectionTick: 10, HeartbeatTick: 1, MaxSizePerMsg: 4096, MaxInflightMsgs: 16, Logger: &raft.DefaultLogger{Logger: log.New(io.Discard, "", 0)}})
	if err != nil {
		t.Fatal(err)
	}
	initial := rn.BasicStatus().RaftState
	var applied []string
	pump := func() {
		for n := 0; rn.HasReady(); n++ {
			if n >= 20 {
				t.Fatal("unexpected Ready loop")
			}
			rd := rn.Ready()
			if !raft.IsEmptySnap(rd.Snapshot) {
				t.Fatal("unexpected snapshot")
			}
			if err := storage.Append(rd.Entries); err != nil {
				t.Fatal(err)
			}
			if !raft.IsEmptyHardState(rd.HardState) {
				if err := storage.SetHardState(rd.HardState); err != nil {
					t.Fatal(err)
				}
			}
			if len(rd.Messages) != 0 {
				t.Fatalf("single-node fixture produced outbound messages: %v", rd.Messages)
			}
			for _, e := range rd.CommittedEntries {
				if e.Type != pb.EntryNormal {
					t.Fatal("unexpected configuration entry")
				}
				if len(e.Data) > 0 {
					applied = append(applied, string(e.Data))
				}
			}
			rn.Advance(rd)
		}
	}
	if err := rn.Campaign(); err != nil {
		t.Fatal(err)
	}
	pump()
	if rn.BasicStatus().RaftState != raft.StateLeader {
		t.Fatal(rn.BasicStatus())
	}
	if err := rn.Propose([]byte("set x=1")); err != nil {
		t.Fatal(err)
	}
	before := len(applied)
	pump()
	if before != 0 || !slices.Equal(applied, []string{"set x=1"}) {
		t.Fatal(before, applied)
	}
	t.Logf("state=%s->%s applied_before_pump=%d applied_after_pump=%q", initial, rn.BasicStatus().RaftState, before, applied)
}

// readCommand accepts only RESP arrays of bulk strings emitted by this fixture's client.
func readCommand(r *bufio.Reader) ([]string, error) {
	header, err := r.ReadString('\n')
	if err != nil {
		return nil, err
	}
	if !strings.HasPrefix(header, "*") {
		return nil, fmt.Errorf("not an array: %q", header)
	}
	n, err := strconv.Atoi(strings.TrimSuffix(header[1:], "\r\n"))
	if err != nil || n < 1 || n > 16 {
		return nil, fmt.Errorf("invalid array")
	}
	args := make([]string, n)
	for i := range args {
		line, e := r.ReadString('\n')
		if e != nil {
			return nil, e
		}
		if !strings.HasPrefix(line, "$") {
			return nil, fmt.Errorf("not bulk")
		}
		size, e := strconv.Atoi(strings.TrimSuffix(line[1:], "\r\n"))
		if e != nil || size < 0 || size > 1024 {
			return nil, fmt.Errorf("invalid bulk")
		}
		b := make([]byte, size+2)
		if _, e = io.ReadFull(r, b); e != nil {
			return nil, e
		}
		if string(b[size:]) != "\r\n" {
			return nil, fmt.Errorf("invalid ending")
		}
		args[i] = string(b[:size])
	}
	return args, nil
}
func TestRedis(t *testing.T) {
	var wg sync.WaitGroup
	var getCount atomic.Int32
	client := redis.NewClient(&redis.Options{
		Addr: "memory:0", Protocol: 2, DisableIdentity: true, MaxRetries: -1, PoolSize: 1,
		ContextTimeoutEnabled: true, ReadTimeout: time.Second, WriteTimeout: time.Second,
		Dialer: func(ctx context.Context, _, _ string) (net.Conn, error) {
			if err := ctx.Err(); err != nil {
				return nil, err
			}
			c, s := net.Pipe()
			wg.Go(func() {
				defer s.Close()
				r := bufio.NewReader(s)
				for {
					args, err := readCommand(r)
					if err != nil {
						return
					}
					reply := "-ERR unsupported fixture command\r\n"
					switch strings.ToLower(args[0]) {
					case "hello":
						reply = "-ERR unknown command 'hello'\r\n"
					case "get":
						getCount.Add(1)
						if len(args) == 2 && args[1] == "present" {
							reply = "$2\r\nok\r\n"
						} else {
							reply = "$-1\r\n"
						}
					}
					if _, err := io.WriteString(s, reply); err != nil {
						return
					}
				}
			})
			return c, nil
		},
	})
	defer func() { _ = client.Close(); wg.Wait() }()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	// Warm up connection and verify a normal command traverses encoding and decoding.
	value, err := client.Get(ctx, "present").Result()
	if err != nil || value != "ok" {
		t.Fatal(value, err)
	}
	before := getCount.Load()
	pipe := client.Pipeline()
	present := pipe.Get(ctx, "present")
	missing := pipe.Get(ctx, "missing")
	queued := getCount.Load() - before
	_, execErr := pipe.Exec(ctx)
	if queued != 0 || present.Val() != "ok" || !errors.Is(missing.Err(), redis.Nil) || !errors.Is(execErr, redis.Nil) {
		t.Fatal(queued, present, missing, execErr)
	}
	t.Logf("get=%q queued_wire_gets=%d pipeline_present=%q pipeline_missing_is_nil=%v exec_is_nil=%v", value, queued, present.Val(), errors.Is(missing.Err(), redis.Nil), errors.Is(execErr, redis.Nil))
}