package runtimeprocess

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

	"github.com/pingcap-inc/nexuscrm-api-host/processhttp"
)

func TestNoWorkProcessLifecycleAndOperationalSurface(t *testing.T) {
	process, err := Assemble(Config{
		Role: "runtime-worker", Version: "test-sha", AdminAddress: "127.0.0.1:0",
		Capabilities: []string{},
	})
	if err != nil {
		t.Fatal(err)
	}
	if process.State() != StateAssembled || process.Live() || process.Ready() {
		t.Fatalf("assembled state = %v live=%v ready=%v", process.State(), process.Live(), process.Ready())
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	if process.State() != StateReady || !process.Live() || !process.Ready() {
		t.Fatalf("started state = %v live=%v ready=%v", process.State(), process.Live(), process.Ready())
	}
	assertProbe(t, "http://"+process.AdminAddress()+processhttp.LivePath, http.StatusOK)
	assertProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusOK)
	metrics := readMetrics(t, "http://"+process.AdminAddress()+MetricsPath)
	for _, fact := range []string{
		`nexuscrm_runtime_process_info{role="runtime-worker",version="test-sha"} 1`,
		`nexuscrm_runtime_process_live{role="runtime-worker"} 1`,
		`nexuscrm_runtime_process_ready{role="runtime-worker"} 1`,
		`nexuscrm_runtime_process_state{role="runtime-worker"} 2`,
	} {
		if !strings.Contains(metrics, fact) {
			t.Fatalf("metrics missing %q:\n%s", fact, metrics)
		}
	}
	if strings.Contains(metrics, "nexuscrm_runtime_admission_limit") {
		t.Fatal("no-work worker exported API admission metrics")
	}

	process.BeginDrain()
	if process.State() != StateDraining || process.Ready() {
		t.Fatalf("draining state = %v ready=%v", process.State(), process.Ready())
	}
	assertProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusServiceUnavailable)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := process.Stop(ctx); err != nil {
		t.Fatal(err)
	}
	if process.State() != StateStopped || process.Live() || process.Ready() {
		t.Fatalf("stopped state = %v live=%v ready=%v", process.State(), process.Live(), process.Ready())
	}
	if err := process.Wait(); err != nil {
		t.Fatalf("Wait = %v", err)
	}
}

func TestAdmissionMetricsRequireOneOwnerAndReportCurrentValues(t *testing.T) {
	process, err := Assemble(Config{
		Role: "runtime-api", Version: "test", AdminAddress: "127.0.0.1:0",
		Admission: func() AdmissionSnapshot {
			return AdmissionSnapshot{Limit: 2, InFlight: 1, Accepted: 7, RejectedFull: 3, RejectedDraining: 2}
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		if err := process.Stop(ctx); err != nil {
			t.Fatal(err)
		}
	}()
	metrics := readMetrics(t, "http://"+process.AdminAddress()+MetricsPath)
	for _, fact := range []string{
		`nexuscrm_runtime_admission_limit{role="runtime-api"} 2`,
		`nexuscrm_runtime_admission_in_flight{role="runtime-api"} 1`,
		`nexuscrm_runtime_admission_accepted_total{role="runtime-api"} 7`,
		`nexuscrm_runtime_admission_rejected_total{reason="capacity",role="runtime-api"} 3`,
		`nexuscrm_runtime_admission_rejected_total{reason="draining",role="runtime-api"} 2`,
	} {
		if !strings.Contains(metrics, fact) {
			t.Fatalf("metrics missing %q:\n%s", fact, metrics)
		}
	}
}

func TestMetricsReportsCachedReadinessWithoutRunningActiveCheck(t *testing.T) {
	var checks atomic.Int64
	process, err := Assemble(Config{
		Role: "runtime-worker", Version: "test", AdminAddress: "127.0.0.1:0",
		Ready: func() bool {
			checks.Add(1)
			return true
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	if process.Ready() {
		t.Fatal("assembled process reported ready")
	}
	if got := checks.Load(); got != 0 {
		t.Fatalf("assembled readiness invoked role callback %d times, want 0", got)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		if err := process.Stop(ctx); err != nil {
			t.Fatal(err)
		}
	}()

	metrics := readMetrics(t, "http://"+process.AdminAddress()+MetricsPath)
	if got := checks.Load(); got != 0 {
		t.Fatalf("readiness checks after metrics scrape = %d, want 0", got)
	}
	for _, fact := range []string{
		`nexuscrm_runtime_process_ready{role="runtime-worker"} 0`,
		`nexuscrm_runtime_process_readiness_check_known{role="runtime-worker"} 0`,
		`nexuscrm_runtime_process_readiness_last_check_timestamp_seconds{role="runtime-worker"} 0`,
	} {
		if !strings.Contains(metrics, fact) {
			t.Fatalf("metrics missing %q:\n%s", fact, metrics)
		}
	}

	assertProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusOK)
	if got := checks.Load(); got != 1 {
		t.Fatalf("readiness checks after active probe = %d, want 1", got)
	}
	metrics = readMetrics(t, "http://"+process.AdminAddress()+MetricsPath)
	if got := checks.Load(); got != 1 {
		t.Fatalf("readiness checks after second metrics scrape = %d, want 1", got)
	}
	for _, fact := range []string{
		`nexuscrm_runtime_process_ready{role="runtime-worker"} 1`,
		`nexuscrm_runtime_process_readiness_check_known{role="runtime-worker"} 1`,
	} {
		if !strings.Contains(metrics, fact) {
			t.Fatalf("metrics missing %q:\n%s", fact, metrics)
		}
	}
	if strings.Contains(metrics, `nexuscrm_runtime_process_readiness_last_check_timestamp_seconds{role="runtime-worker"} 0`) {
		t.Fatalf("active readiness timestamp was not recorded:\n%s", metrics)
	}
}

func TestReadinessObservationKeepsNewestOverlappingProbe(t *testing.T) {
	var checks atomic.Int64
	firstStarted := make(chan struct{})
	releaseFirst := make(chan struct{})
	process, err := Assemble(Config{
		Role: "runtime-worker", Version: "test", AdminAddress: "127.0.0.1:0",
		Ready: func() bool {
			if checks.Add(1) == 1 {
				close(firstStarted)
				<-releaseFirst
				return false
			}
			return true
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		if err := process.Stop(ctx); err != nil {
			t.Fatal(err)
		}
	}()

	firstDone := make(chan bool, 1)
	go func() { firstDone <- process.Ready() }()
	select {
	case <-firstStarted:
	case <-time.After(time.Second):
		t.Fatal("first readiness probe did not start")
	}
	if !process.Ready() {
		t.Fatal("newer readiness probe returned false")
	}
	close(releaseFirst)
	if <-firstDone {
		t.Fatal("older readiness probe changed its own result")
	}
	if snapshot := process.readinessSnapshot(); !snapshot.Known || !snapshot.Ready {
		t.Fatalf("readiness snapshot = %+v, want newer successful result", snapshot)
	}
}

func TestProcessFailsClosedOnPartialListenFailure(t *testing.T) {
	process, err := Assemble(Config{
		Role: "runtime-api", Version: "test", AdminAddress: "127.0.0.1:0",
		GRPCAddress: "invalid-address", GRPCServer: nil,
	})
	if err == nil || process != nil {
		t.Fatalf("Assemble = %#v, %v; want paired gRPC config rejection", process, err)
	}
}

func TestGRPCRoleRequiresAdmissionDrainHook(t *testing.T) {
	server := &rpcServerStub{}
	config := Config{Role: "runtime-api", Version: "test", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0", GRPCServer: server}
	if process, err := Assemble(config); err == nil || process != nil {
		t.Fatalf("Assemble(%+v) = %#v, %v; want drain lifecycle rejection", config, process, err)
	}
}

func TestGRPCShutdownForcesStopAtOwnerDeadline(t *testing.T) {
	server := newBlockingRPCServer()
	process, err := Assemble(Config{
		Role: "runtime-api", Version: "test", AdminAddress: "127.0.0.1:0",
		GRPCAddress: "127.0.0.1:0", GRPCServer: server, BeginDrain: func() {},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	select {
	case <-server.started:
	case <-time.After(time.Second):
		t.Fatal("RPC server did not start")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
	defer cancel()
	if err := process.Stop(ctx); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("Stop error = %v, want deadline exceeded", err)
	}
	select {
	case <-server.stopCalled:
	default:
		t.Fatal("owner deadline did not force transport Stop")
	}
	if process.State() != StateStopped {
		t.Fatalf("state = %v, want stopped", process.State())
	}
}

func TestProcessOwnsControllerStartCancellationAndJoin(t *testing.T) {
	started := make(chan struct{})
	stopped := make(chan struct{})
	process, err := Assemble(Config{
		Role: "runtime-api", Version: "test", AdminAddress: "127.0.0.1:0",
		Controller: func(ctx context.Context) error {
			close(started)
			<-ctx.Done()
			close(stopped)
			return nil
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	select {
	case <-started:
		t.Fatal("controller started before the process owned its listeners")
	default:
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	select {
	case <-started:
	case <-time.After(time.Second):
		t.Fatal("controller did not start with the process")
	}
	if !process.Ready() {
		t.Fatal("live controller did not contribute to readiness")
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := process.Stop(ctx); err != nil {
		t.Fatal(err)
	}
	select {
	case <-stopped:
	default:
		t.Fatal("Stop returned before the controller was canceled and joined")
	}
}

func TestDiagnosticsAndContentionProfilingAreIndependentAndLoopbackOnly(t *testing.T) {
	disabled, err := Assemble(Config{Role: "runtime-api", Version: "test", AdminAddress: "127.0.0.1:0"})
	if err != nil {
		t.Fatal(err)
	}
	request := httptest.NewRequest(http.MethodGet, "/debug/pprof/goroutine", nil)
	response := httptest.NewRecorder()
	disabled.adminHandler.ServeHTTP(response, request)
	if response.Code != http.StatusNotFound {
		t.Fatalf("disabled profiling status = %d, want 404", response.Code)
	}

	diagnostics, err := Assemble(Config{
		Role: "runtime-api", Version: "test", AdminAddress: "127.0.0.1:0", ExposeDiagnostics: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	response = httptest.NewRecorder()
	diagnostics.adminHandler.ServeHTTP(response, request)
	if response.Code != http.StatusOK || response.Body.Len() == 0 {
		t.Fatalf("diagnostics response = %d/%d bytes", response.Code, response.Body.Len())
	}

	enabled, err := Assemble(Config{
		Role: "runtime-api", Version: "test", AdminAddress: "127.0.0.1:0", EnableProfiling: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	response = httptest.NewRecorder()
	enabled.adminHandler.ServeHTTP(response, request)
	if response.Code != http.StatusOK || response.Body.Len() == 0 {
		t.Fatalf("enabled profiling response = %d/%d bytes", response.Code, response.Body.Len())
	}
	if _, err := Assemble(Config{
		Role: "runtime-api", Version: "test", AdminAddress: "0.0.0.0:9091", EnableProfiling: true,
	}); err == nil {
		t.Fatal("profiling was allowed on a non-loopback listener")
	}
	if _, err := Assemble(Config{
		Role: "runtime-api", Version: "test", AdminAddress: "0.0.0.0:9091", ExposeDiagnostics: true,
	}); err == nil {
		t.Fatal("diagnostics were allowed on a non-loopback listener")
	}
}

func TestLoopbackListenAddressClassification(t *testing.T) {
	for _, test := range []struct {
		address  string
		loopback bool
	}{
		{address: "127.0.0.1:9091", loopback: true},
		{address: "localhost:9091", loopback: true},
		{address: "[::1]:9091", loopback: true},
		{address: ":9091", loopback: false},
		{address: "0.0.0.0:9091", loopback: false},
		{address: "[::]:9091", loopback: false},
		{address: "not-an-address", loopback: false},
	} {
		if got := IsLoopbackListenAddress(test.address); got != test.loopback {
			t.Errorf("IsLoopbackListenAddress(%q) = %t, want %t", test.address, got, test.loopback)
		}
	}
}

func TestUnexpectedControllerExitRemovesReadinessAndWakesWait(t *testing.T) {
	want := errors.New("controller failed")
	release := make(chan struct{})
	process, err := Assemble(Config{
		Role: "runtime-api", Version: "test", AdminAddress: "127.0.0.1:0",
		Controller: func(context.Context) error {
			<-release
			return want
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	close(release)
	if err := process.Wait(); !errors.Is(err, want) {
		t.Fatalf("Wait = %v, want controller failure", err)
	}
	if process.Ready() {
		t.Fatal("process remained ready after its owned controller exited")
	}
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err := process.Stop(ctx); err != nil {
		t.Fatal(err)
	}
}

type rpcServerStub struct{}

func (*rpcServerStub) Serve(net.Listener) error { return nil }
func (*rpcServerStub) GracefulStop()            {}
func (*rpcServerStub) Stop()                    {}

type blockingRPCServer struct {
	done       chan struct{}
	started    chan struct{}
	stopCalled chan struct{}
	stopOnce   sync.Once
	mu         sync.Mutex
	listener   net.Listener
}

func newBlockingRPCServer() *blockingRPCServer {
	return &blockingRPCServer{done: make(chan struct{}), started: make(chan struct{}), stopCalled: make(chan struct{})}
}

func (s *blockingRPCServer) Serve(listener net.Listener) error {
	s.mu.Lock()
	s.listener = listener
	s.mu.Unlock()
	close(s.started)
	<-s.done
	return nil
}
func (s *blockingRPCServer) GracefulStop() { <-s.done }
func (s *blockingRPCServer) Stop() {
	s.stopOnce.Do(func() {
		s.mu.Lock()
		if s.listener != nil {
			_ = s.listener.Close()
		}
		s.mu.Unlock()
		close(s.stopCalled)
		close(s.done)
	})
}

func assertProbe(t *testing.T, url string, want int) {
	t.Helper()
	client := http.Client{Timeout: time.Second}
	response, err := client.Get(url)
	if err != nil {
		t.Fatal(err)
	}
	defer response.Body.Close()
	if response.StatusCode != want {
		t.Fatalf("GET %s status = %d, want %d", url, response.StatusCode, want)
	}
}

func readMetrics(t *testing.T, url string) string {
	t.Helper()
	client := http.Client{Timeout: time.Second}
	response, err := client.Get(url)
	if err != nil {
		t.Fatal(err)
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		t.Fatalf("GET %s status = %d", url, response.StatusCode)
	}
	body, err := io.ReadAll(response.Body)
	if err != nil {
		t.Fatal(err)
	}
	return string(body)
}