package integration

import (
	"context"
	"reflect"
	"testing"
	"time"

	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"

	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/api/v1alpha1"
	"github.com/tidbcloud/cluster-service-ng/internal/capacity-planner/pkg/controller"
)

// recordingClient is the test-only observation point: it wraps the
// manager's client and records every Get key. The Reconciler's Reconcile
// always Gets its request key, so observing a Get for dp-x/wired proves
// the event for that exact object reached Reconcile (initial-list
// reconciles of pre-existing objects record their own keys and are
// filtered out).
type recordingClient struct {
	client.Client
	gets chan types.NamespacedName
}

func (c *recordingClient) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
	select {
	case c.gets <- key:
	default:
	}
	return c.Client.Get(ctx, key, obj, opts...)
}

// TestManagerWiringEventLoopDisabledNoSideEffect proves the full event chain works
// (API server -> informer -> workqueue -> Reconcile) for a freshly created
// Autoscaler, using the production Reconciler and its production
// SetupWithManager (no test-side wiring copy), and that DISABLED mode leaves
// the object and the cluster untouched.
func TestManagerWiringEventLoopDisabledNoSideEffect(t *testing.T) {
	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()

	// Snapshot pre-existing Autoscalers from sibling tests so this test is
	// self-contained regardless of run order.
	before := &v1alpha1.AutoscalerList{}
	if err := testClient.List(ctx, before); err != nil {
		t.Fatal(err)
	}
	known := map[string]bool{}
	for _, it := range before.Items {
		known[it.Name] = true
	}

	mgr, err := ctrl.NewManager(testCfg, ctrl.Options{
		Scheme:  schemeForTest(),
		Metrics: metricsserver.Options{BindAddress: "0"},
	})
	if err != nil {
		t.Fatalf("new manager: %v", err)
	}
	rec := &recordingClient{Client: mgr.GetClient(), gets: make(chan types.NamespacedName, 16)}
	// Production wiring: the real Reconciler behind the recording client.
	if err := (&controller.Reconciler{Client: rec, Scheme: mgr.GetScheme()}).SetupWithManager(mgr); err != nil {
		t.Fatalf("SetupWithManager: %v", err)
	}

	mgrCtx, mgrCancel := context.WithCancel(ctx)
	done := make(chan struct{})
	go func() {
		defer close(done)
		if err := mgr.Start(mgrCtx); err != nil {
			t.Errorf("manager start: %v", err)
		}
	}()
	defer func() {
		mgrCancel()
		<-done
	}()

	as := validAutoscaler()
	as.Name = "wired"
	as.Spec.Mode = v1alpha1.AutoscalerModeDisabled
	wiredKey := types.NamespacedName{Namespace: as.Namespace, Name: as.Name}
	if err := testClient.Create(ctx, as); err != nil {
		t.Fatalf("create Autoscaler: %v", err)
	}
	defer func() {
		_ = testClient.Delete(t.Context(), as)
	}()

	// Wait until Reconcile ran for dp-x/wired specifically (keys from
	// initial-list reconciles of pre-existing objects are ignored).
	deadline := time.After(15 * time.Second)
	sawWired := false
	for !sawWired {
		select {
		case key := <-rec.gets:
			if key == wiredKey {
				sawWired = true
			}
		case <-deadline:
			t.Fatalf("Reconcile for %s was not observed within 15s", wiredKey)
		}
	}

	got := &v1alpha1.Autoscaler{}
	if err := testClient.Get(ctx, wiredKey, got); err != nil {
		t.Fatal(err)
	}
	if !reflect.DeepEqual(got.Spec, as.Spec) {
		t.Fatalf("spec mutated by reconciler:\ngot  %+v\nwant %+v", got.Spec, as.Spec)
	}
	if !reflect.DeepEqual(got.Status, v1alpha1.AutoscalerStatus{}) {
		t.Fatalf("status written by disabled reconciler: %+v", got.Status)
	}
	// No new objects: exactly the pre-existing set plus the one we created,
	// and nothing deleted.
	after := &v1alpha1.AutoscalerList{}
	if err := testClient.List(ctx, after); err != nil {
		t.Fatal(err)
	}
	names := map[string]bool{}
	for _, it := range after.Items {
		names[it.Name] = true
	}
	if !names[as.Name] {
		t.Fatal("wired Autoscaler missing after reconcile")
	}
	for name := range known {
		if !names[name] {
			t.Fatalf("pre-existing Autoscaler %q disappeared", name)
		}
	}
	if len(names) != len(known)+1 {
		t.Fatalf("reconciler must not create objects: before %v after %v", known, names)
	}
}