package integrationservice

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

	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	commandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

type fakeSupervisor struct {
	starts   atomic.Int32
	stops    atomic.Int32
	startErr error
	start    func(context.Context) error
}

func (s *fakeSupervisor) StartContext(ctx context.Context) error {
	s.starts.Add(1)
	if s.start != nil {
		return s.start(ctx)
	}
	return s.startErr
}
func (s *fakeSupervisor) Stop() error { s.stops.Add(1); return nil }

type fakeRouter struct {
	started chan struct{}
	result  chan error
}

type immediateExitRouter struct{ err error }

func (r immediateExitRouter) Run(context.Context, func()) error { return r.err }

type productionReceiptLaneStub struct {
	probeErr error
	running  chan struct{}
	stopped  chan struct{}
}

func (r *productionReceiptLaneStub) ProcessBatch(context.Context) (int, error) {
	return 0, r.probeErr
}

func (r *productionReceiptLaneStub) Run(ctx context.Context) error {
	close(r.running)
	<-ctx.Done()
	close(r.stopped)
	return ctx.Err()
}

type productionOutboundLaneStub struct {
	running chan struct{}
	stopped chan struct{}
}

func (r *productionOutboundLaneStub) Run(ctx context.Context, started func()) error {
	close(r.running)
	started()
	<-ctx.Done()
	close(r.stopped)
	return ctx.Err()
}

func TestProductionReceiptRouterStartsAndCancelsBothExistingIntegrationLanes(t *testing.T) {
	receipts := &productionReceiptLaneStub{
		running: make(chan struct{}), stopped: make(chan struct{}),
	}
	outbound := &productionOutboundLaneStub{
		running: make(chan struct{}), stopped: make(chan struct{}),
	}
	router := &productionReceiptRouter{router: receipts, outbound: outbound}
	ctx, cancel := context.WithCancel(t.Context())
	started := make(chan struct{})
	done := make(chan error, 1)
	go func() { done <- router.Run(ctx, func() { close(started) }) }()
	select {
	case <-started:
	case <-time.After(time.Second):
		t.Fatal("composite Integration lifecycle did not start both lanes")
	}
	select {
	case <-receipts.running:
	default:
		t.Fatal("existing receipt router was not running")
	}
	select {
	case <-outbound.running:
	default:
		t.Fatal("outbound intent lifecycle was not running")
	}
	cancel()
	if err := <-done; !errors.Is(err, context.Canceled) {
		t.Fatalf("composite shutdown error = %v", err)
	}
	for name, stopped := range map[string]<-chan struct{}{
		"receipt router": receipts.stopped, "outbound owner": outbound.stopped,
	} {
		select {
		case <-stopped:
		case <-time.After(time.Second):
			t.Fatalf("%s did not observe cancellation", name)
		}
	}
}

func TestProductionReceiptRouterFailsStartupWhenEitherLaneCannotProbe(t *testing.T) {
	receipts := &productionReceiptLaneStub{
		probeErr: errors.New("receipt schema unavailable"),
		running:  make(chan struct{}), stopped: make(chan struct{}),
	}
	outbound := &productionOutboundLaneStub{
		running: make(chan struct{}), stopped: make(chan struct{}),
	}
	started := false
	err := (&productionReceiptRouter{router: receipts, outbound: outbound}).Run(
		t.Context(), func() { started = true },
	)
	if err == nil || !strings.Contains(err.Error(), "receipt schema unavailable") || started {
		t.Fatalf("startup result = %v started=%v", err, started)
	}
	select {
	case <-outbound.stopped:
	case <-time.After(time.Second):
		t.Fatal("peer outbound lane was not canceled after receipt startup failure")
	}
}

func newFakeRouter() *fakeRouter {
	return &fakeRouter{started: make(chan struct{}), result: make(chan error, 1)}
}

func (r *fakeRouter) Run(ctx context.Context, started func()) error {
	started()
	close(r.started)
	select {
	case err := <-r.result:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

type fakePlatform struct {
	ready   atomic.Bool
	closes  atomic.Int32
	view    catalogv1.IntegrationBindingView
	viewErr error
}

func (p *fakePlatform) Ready() bool  { return p.ready.Load() }
func (p *fakePlatform) Close() error { p.closes.Add(1); return nil }
func (p *fakePlatform) GetIntegrationBindingView(context.Context, catalogv1.GetRequest) (catalogv1.IntegrationBindingView, error) {
	return p.view, p.viewErr
}

func TestReadPlatformAuthority(t *testing.T) {
	config := Config{OrgID: "org-test", AppInstallationID: "01900000-0000-7000-8000-000000000001", AuthorizedWorkloadID: "integration-service"}
	client := &fakePlatform{view: validIntegrationBindingView(config)}
	view, err := readPlatformAuthority(t.Context(), client, config)
	if err != nil || view.AuthorizedWorkloadID != config.AuthorizedWorkloadID {
		t.Fatalf("readPlatformAuthority() = (%#v, %v)", view, err)
	}
	client.view.AuthorizedWorkloadID = "other-workload"
	client.view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(client.view)
	if _, err := readPlatformAuthority(t.Context(), client, config); err == nil {
		t.Fatal("readPlatformAuthority() accepted a different workload scope")
	}
}

func TestPinnedPlatformClientKeepsOneImmutableCatalogRevision(t *testing.T) {
	config := Config{
		OrgID: "org-test", AppInstallationID: "01900000-0000-7000-8000-000000000001",
		AuthorizedWorkloadID: "integration-service",
	}
	view := validIntegrationBindingView(config)
	view.Bindings = []catalogv1.IntegrationBinding{
		integrationServingTestBinding(
			catalogv1.BindingConnectorInboundRoute,
			"connector-a",
			"install-a",
			"connector-a/inbound",
			"a",
		),
	}
	setIntegrationServingRouteCommand(view.Bindings, "BoundedInbound")
	view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
	underlying := &fakePlatform{view: view}
	client := newPinnedPlatformClient(underlying, view)

	// Neither a later underlying revision nor mutation of one returned copy may
	// change the process-lifetime authority used by a subsequent handoff.
	underlying.view.Lineage.CatalogRevision++
	first, err := client.GetIntegrationBindingView(t.Context(), catalogv1.GetRequest{})
	if err != nil {
		t.Fatal(err)
	}
	first.Bindings[0].CredentialRefs = append(first.Bindings[0].CredentialRefs, "mutated")
	first.Bindings[0].RuntimeCommand.DefinitionVersion = "mutated"
	second, err := client.GetIntegrationBindingView(t.Context(), catalogv1.GetRequest{})
	if err != nil {
		t.Fatal(err)
	}
	if second.Lineage.CatalogRevision != view.Lineage.CatalogRevision ||
		second.Bindings[0].RuntimeCommand.DefinitionVersion != "flow-v1" ||
		len(second.Bindings[0].CredentialRefs) != 0 {
		t.Fatalf("pinned Integration view drifted: %#v", second)
	}
	minimum := view.Lineage.CatalogRevision + 1
	if _, err := client.GetIntegrationBindingView(t.Context(), catalogv1.GetRequest{MinCatalogRevision: &minimum}); err == nil {
		t.Fatal("pinned Integration view accepted a newer minimum Catalog revision")
	}
}

func (p *fakePlatform) AcceptCommand(context.Context, commandv1.Request) (commandv1.Response, error) {
	return commandv1.Response{}, nil
}

func (p *fakePlatform) ListPendingOutboundIntents(
	context.Context,
	outboundintentv1.ListRequest,
) (outboundintentv1.ListResponse, error) {
	return outboundintentv1.ListResponse{Receipts: p.outboundReceipts()}, nil
}

func (p *fakePlatform) AcknowledgeOutboundIntent(
	_ context.Context,
	request outboundintentv1.AcknowledgeRequest,
) (outboundintentv1.AcknowledgeResponse, error) {
	return outboundintentv1.AcknowledgeResponse{
		IntentID: request.IntentID, Outcome: request.Outcome,
		Receipts: p.outboundReceipts(),
	}, nil
}

func (p *fakePlatform) CommitOutboundResult(
	_ context.Context,
	request outboundintentv1.CommitResultRequest,
) (outboundintentv1.CommitResultResponse, error) {
	return outboundintentv1.CommitResultResponse{
		IntentID: request.IntentID, Receipts: p.outboundReceipts(),
	}, nil
}

func (p *fakePlatform) outboundReceipts() outboundintentv1.Receipts {
	return outboundintentv1.Receipts{
		ProtocolVersion: "1.0", CapabilityVersion: outboundintentv1.CapabilityName,
		CatalogRevision:  p.view.Lineage.CatalogRevision,
		SchemaGeneration: p.view.Lineage.SchemaGeneration,
		EngineGeneration: p.view.Lineage.EngineGeneration,
		SecurityEpoch:    p.view.Receipts.SecurityEpoch,
	}
}

type fakeDatabase struct{ closes atomic.Int32 }

func (d *fakeDatabase) Close() error { d.closes.Add(1); return nil }

type immediateExitHTTPProcess struct {
	err      error
	shutdown atomic.Int32
}

func (*immediateExitHTTPProcess) Start() error                     { return nil }
func (*immediateExitHTTPProcess) Ready() bool                      { return false }
func (*immediateExitHTTPProcess) Address() string                  { return "127.0.0.1:1" }
func (p *immediateExitHTTPProcess) Wait() error                    { return p.err }
func (p *immediateExitHTTPProcess) Shutdown(context.Context) error { p.shutdown.Add(1); return nil }

func TestProcessLifecycleAndDynamicRuntimeReadiness(t *testing.T) {
	supervisor := &fakeSupervisor{}
	router := newFakeRouter()
	runtime := &fakePlatform{}
	runtime.ready.Store(true)
	database := &fakeDatabase{}
	process, err := newProcess("test-version", processTestConfig(), http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		w.WriteHeader(http.StatusNoContent)
	}), supervisor, router, runtime, database)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(t.Context()); err != nil {
		t.Fatal(err)
	}
	<-router.started
	assertHTTPStatus(t, "http://"+process.Address()+"/readyz", http.StatusOK)
	assertHTTPStatus(t, "http://"+process.Address()+"/webhooks/test", http.StatusNoContent)
	metrics, err := http.Get("http://" + process.Address() + "/metrics")
	if err != nil {
		t.Fatal(err)
	}
	metricsBody, readErr := io.ReadAll(metrics.Body)
	metrics.Body.Close()
	if readErr != nil || metrics.StatusCode != http.StatusOK || !strings.Contains(string(metricsBody), "go_goroutines") {
		t.Fatalf("Integration metrics response status=%d read_err=%v contains_runtime_metric=%t", metrics.StatusCode, readErr, strings.Contains(string(metricsBody), "go_goroutines"))
	}
	runtime.ready.Store(false)
	assertHTTPStatus(t, "http://"+process.Address()+"/readyz", http.StatusServiceUnavailable)
	runtime.ready.Store(true)

	waitDone := make(chan error, 1)
	go func() { waitDone <- process.Wait() }()
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := process.Shutdown(shutdownCtx); err != nil {
		t.Fatalf("Shutdown() error = %v", err)
	}
	if err := <-waitDone; err != nil {
		t.Fatalf("Wait() error = %v", err)
	}
	if supervisor.starts.Load() != 1 || supervisor.stops.Load() != 1 || runtime.closes.Load() != 1 || database.closes.Load() != 1 {
		t.Fatalf("unexpected lifecycle counts: starts=%d stops=%d runtime_closes=%d db_closes=%d",
			supervisor.starts.Load(), supervisor.stops.Load(), runtime.closes.Load(), database.closes.Load())
	}
	if err := process.Shutdown(shutdownCtx); err != nil {
		t.Fatalf("second Shutdown() error = %v", err)
	}
	if supervisor.stops.Load() != 1 || runtime.closes.Load() != 1 || database.closes.Load() != 1 {
		t.Fatal("Shutdown() is not idempotent")
	}
}

func TestProcessRejectsUnavailableRuntimeBeforeStartingWorkers(t *testing.T) {
	supervisor := &fakeSupervisor{}
	runtime := &fakePlatform{}
	process, err := newProcess("test-version", processTestConfig(), http.NotFoundHandler(), supervisor, newFakeRouter(), runtime, &fakeDatabase{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(t.Context()); err == nil || supervisor.starts.Load() != 0 {
		t.Fatalf("Start() error = %v, starts = %d", err, supervisor.starts.Load())
	}
}

func TestProcessSurfacesUnexpectedReceiptRouterExit(t *testing.T) {
	router := newFakeRouter()
	runtime := &fakePlatform{}
	runtime.ready.Store(true)
	process, err := newProcess("test-version", processTestConfig(), http.NotFoundHandler(), &fakeSupervisor{}, router, runtime, &fakeDatabase{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(t.Context()); err != nil {
		t.Fatal(err)
	}
	<-router.started
	want := errors.New("router failed")
	router.result <- want
	if err := process.Wait(); !errors.Is(err, want) {
		t.Fatalf("Wait() error = %v, want %v", err, want)
	}
	assertHTTPStatus(t, "http://"+process.Address()+"/readyz", http.StatusServiceUnavailable)
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_ = process.Shutdown(shutdownCtx)
}

func TestProcessNeverAdvertisesReadyAfterImmediateReceiptRouterExit(t *testing.T) {
	runtime := &fakePlatform{}
	runtime.ready.Store(true)
	supervisor := &fakeSupervisor{}
	want := errors.New("router failed during startup")
	process, err := newProcess(
		"test-version", processTestConfig(), http.NotFoundHandler(), supervisor,
		immediateExitRouter{err: want}, runtime, &fakeDatabase{},
	)
	if err != nil {
		t.Fatal(err)
	}
	startErr := process.Start(t.Context())
	if !errors.Is(startErr, want) {
		t.Fatalf("Start() error = %v, want %v", startErr, want)
	}
	if process.isReady() || process.ready.Load() {
		t.Fatal("immediate receipt router exit advertised readiness")
	}
	if process.Address() != "" {
		t.Fatalf("immediate receipt router exit opened ingress at %q", process.Address())
	}
	if supervisor.starts.Load() != 1 || supervisor.stops.Load() != 1 {
		t.Fatalf("unexpected supervisor lifecycle: starts=%d stops=%d", supervisor.starts.Load(), supervisor.stops.Load())
	}
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_ = process.Shutdown(shutdownCtx)
}

func TestProcessNeverAdvertisesReadyAfterImmediateHTTPExit(t *testing.T) {
	runtime := &fakePlatform{}
	runtime.ready.Store(true)
	supervisor := &fakeSupervisor{}
	router := newFakeRouter()
	process, err := newProcess(
		"test-version", processTestConfig(), http.NotFoundHandler(), supervisor,
		router, runtime, &fakeDatabase{},
	)
	if err != nil {
		t.Fatal(err)
	}
	want := errors.New("HTTP failed during startup")
	httpProcess := &immediateExitHTTPProcess{err: want}
	process.http = httpProcess
	startErr := process.Start(t.Context())
	if !errors.Is(startErr, want) {
		t.Fatalf("Start() error = %v, want %v", startErr, want)
	}
	if process.isReady() || process.ready.Load() {
		t.Fatal("immediate HTTP exit advertised readiness")
	}
	if supervisor.starts.Load() != 1 || supervisor.stops.Load() != 1 || httpProcess.shutdown.Load() != 1 {
		t.Fatalf("unexpected lifecycle: supervisor starts=%d stops=%d HTTP shutdowns=%d",
			supervisor.starts.Load(), supervisor.stops.Load(), httpProcess.shutdown.Load())
	}
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	_ = process.Shutdown(shutdownCtx)
}

func TestProcessCleansUpWhenHTTPListenerCannotStart(t *testing.T) {
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	defer listener.Close()
	config := processTestConfig()
	config.HTTPAddress = listener.Addr().String()
	supervisor := &fakeSupervisor{}
	runtime := &fakePlatform{}
	runtime.ready.Store(true)
	process, err := newProcess("test-version", config, http.NotFoundHandler(), supervisor, newFakeRouter(), runtime, &fakeDatabase{})
	if err != nil {
		t.Fatal(err)
	}
	startDone := make(chan error, 1)
	go func() { startDone <- process.Start(t.Context()) }()
	select {
	case err := <-startDone:
		if err == nil {
			t.Fatal("Start() succeeded with an occupied address")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("Start() deadlocked while cleaning up the receipt router")
	}
	if supervisor.starts.Load() != 1 || supervisor.stops.Load() != 1 {
		t.Fatalf("unexpected supervisor lifecycle: starts=%d stops=%d", supervisor.starts.Load(), supervisor.stops.Load())
	}
}

func TestProcessStartupCancellationReachesSupervisor(t *testing.T) {
	entered := make(chan struct{})
	supervisor := &fakeSupervisor{start: func(ctx context.Context) error {
		close(entered)
		<-ctx.Done()
		return ctx.Err()
	}}
	runtime := &fakePlatform{}
	runtime.ready.Store(true)
	process, err := newProcess(
		"test-version", processTestConfig(), http.NotFoundHandler(), supervisor,
		newFakeRouter(), runtime, &fakeDatabase{},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithCancel(context.Background())
	startDone := make(chan error, 1)
	go func() { startDone <- process.Start(ctx) }()
	select {
	case <-entered:
	case <-time.After(time.Second):
		t.Fatal("process startup did not reach the supervisor")
	}
	cancel()
	select {
	case err := <-startDone:
		if !errors.Is(err, context.Canceled) {
			t.Fatalf("Start() error = %v, want context cancellation", err)
		}
	case <-time.After(time.Second):
		t.Fatal("Start() did not return after cancellation")
	}
	if process.ready.Load() {
		t.Fatal("canceled startup advertised readiness")
	}
}

func TestProcessDoesNotOpenIngressWhenStartupIsCanceledAtSupervisorReturn(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	supervisor := &fakeSupervisor{start: func(context.Context) error {
		cancel()
		return nil
	}}
	runtime := &fakePlatform{}
	runtime.ready.Store(true)
	process, err := newProcess(
		"test-version", processTestConfig(), http.NotFoundHandler(), supervisor,
		newFakeRouter(), runtime, &fakeDatabase{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(ctx); !errors.Is(err, context.Canceled) {
		t.Fatalf("Start() error = %v, want context cancellation", err)
	}
	if process.Address() != "" || process.ready.Load() {
		t.Fatalf("canceled startup opened ingress address=%q ready=%t", process.Address(), process.ready.Load())
	}
	if supervisor.stops.Load() != 1 {
		t.Fatalf("supervisor stops = %d, want 1", supervisor.stops.Load())
	}
}

func processTestConfig() Config {
	return Config{
		HTTPAddress: "127.0.0.1:0", ReadHeaderTimeout: time.Second,
		RequestTimeout: time.Second, IdleTimeout: time.Second,
	}
}

func validIntegrationBindingView(config Config) catalogv1.IntegrationBindingView {
	view := catalogv1.IntegrationBindingView{
		OrgID: config.OrgID, AppInstallationID: config.AppInstallationID,
		AuthorizedWorkloadID: config.AuthorizedWorkloadID,
		Lineage: catalogv1.ViewLineage{
			CatalogRevision: 41, SchemaGeneration: 7, EngineGeneration: 8,
			SourceSnapshotDigest: "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
		},
		Bindings: []catalogv1.IntegrationBinding{},
		Receipts: catalogv1.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: catalogv1.IntegrationBindingCapabilityName,
			CatalogRevision: 41, SchemaGeneration: 7, EngineGeneration: 8,
			SecurityEpoch: 1,
		},
	}
	view.Lineage.ViewDigest = catalogv1.ComputeIntegrationBindingViewDigest(view)
	return view
}

func assertHTTPStatus(t *testing.T, url string, want int) {
	t.Helper()
	response, err := http.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)
	}
}