package replay

import (
	"testing"

	"github.com/go-logr/logr"
	"github.com/stretchr/testify/require"
	"helm.sh/helm/v3/pkg/action"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/kubernetes/scheme"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
)

func TestBuildInputPaths(t *testing.T) {
	r := &Reconciler{}

	t.Run("aws appends role arn and external id", func(t *testing.T) {
		externalID := "ext-id"
		task := cloudv1.ReplayTask{
			Name: "aws-task",
			SourceAuditLogAddrs: []cloudv1.AuditLogAddr{
				{
					AWS: &cloudv1.AWSAuditLog{
						S3URL:      "s3://bucket/prefix",
						RoleArn:    "arn:aws:iam::123456789012:role/replay",
						ExternalID: &externalID,
					},
				},
			},
		}

		paths, err := r.buildInputPaths(task, logr.Discard())
		require.NoError(t, err)
		require.Equal(t, []string{
			"s3://bucket/prefix?role-arn=arn:aws:iam::123456789012:role/replay&external-id=ext-id",
		}, paths)
	})

	t.Run("aliyun passes s3 compatible url through unchanged", func(t *testing.T) {
		task := cloudv1.ReplayTask{
			Name: "aliyun-task",
			SourceAuditLogAddrs: []cloudv1.AuditLogAddr{
				{
					Aliyun: &cloudv1.AliyunAuditLog{
						BucketURL: "s3://db-auditlog-fang-test/10480461862511845599/tidb-*?endpoint=http://oss-cn-hangzhou.aliyuncs.com&access-key=test-ak&secret-access-key=test-sk",
					},
				},
			},
		}

		paths, err := r.buildInputPaths(task, logr.Discard())
		require.NoError(t, err)
		require.Equal(t, []string{
			"s3://db-auditlog-fang-test/10480461862511845599/tidb-*?endpoint=http://oss-cn-hangzhou.aliyuncs.com&access-key=test-ak&secret-access-key=test-sk",
		}, paths)
	})
}

func TestReconcileDeletionCleansGeneratedKubeConfigSecret(t *testing.T) {
	s := runtime.NewScheme()
	require.NoError(t, scheme.AddToScheme(s))
	require.NoError(t, cloudv1.AddToScheme(s))

	now := metav1.Now()
	replay := &cloudv1.Replay{
		ObjectMeta: metav1.ObjectMeta{
			Name:              "replay-1",
			Namespace:         "replay-ns",
			Finalizers:        []string{"replay.cloud.pingcap.com/finalizer"},
			DeletionTimestamp: &now,
		},
		Spec: cloudv1.ReplaySpec{
			TargetClusterType: cloudv1.ReplayClusterTypePremium,
			Kubernetes: cloudv1.Kubernetes{
				Name:     "dataplane-1",
				Endpoint: "https://example.com:6443",
				Account:  "1234567890",
				Provider: consts.ProviderTypeACK,
				KubeConfigSecretRef: &corev1.SecretReference{
					Name:      "replay-1-kubeconfig",
					Namespace: "replay-ns",
				},
			},
		},
	}
	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "replay-1-kubeconfig",
			Namespace: "replay-ns",
		},
		Data: map[string][]byte{
			"kubeconfig": []byte("fake-kubeconfig"),
		},
	}

	kubeClient := fake.NewClientBuilder().
		WithScheme(s).
		WithStatusSubresource(replay).
		WithObjects(replay, secret).
		Build()
	remoteClient := fake.NewClientBuilder().
		WithScheme(scheme.Scheme).
		Build()
	manager := &fakeReplayDataplaneManager{
		registered: map[string]bool{
			"replay-replay-1-dataplane-1": true,
		},
		clients: map[string]client.Client{
			"replay-replay-1-dataplane-1": remoteClient,
		},
	}
	r := &Reconciler{
		Client:        kubeClient,
		Cache:         kubeClient,
		EventRecorder: record.NewFakeRecorder(8),
		Manager:       manager,
		RootConfig: config.RootConfig{
			ControllerConfig: config.ControllerConfig{
				Provider:    consts.ProviderTypeACK,
				ServicePlan: consts.ServiceTypePremium,
			},
		},
	}

	_, err := r.Reconcile(t.Context(), ctrl.Request{
		NamespacedName: client.ObjectKey{Name: "replay-1", Namespace: "replay-ns"},
	})
	require.NoError(t, err)

	err = kubeClient.Get(t.Context(), client.ObjectKey{Name: "replay-1-kubeconfig", Namespace: "replay-ns"}, &corev1.Secret{})
	require.True(t, apierrors.IsNotFound(err), "expected generated kubeconfig secret to be deleted, got %v", err)
}

func TestHandleDeployStateKeepsStoppedReplayStopped(t *testing.T) {
	r := &Reconciler{}
	replay := &cloudv1.Replay{
		Status: cloudv1.ReplayStatus{
			Phase:   cloudv1.ReplayStopped,
			Message: "paused",
		},
	}

	res, errs, retriableErrs := r.handleDeployState(t.Context(), logr.Discard(), nil, replay)
	require.Equal(t, ctrl.Result{}, res)
	require.NoError(t, errs)
	require.NoError(t, retriableErrs)
	require.Equal(t, cloudv1.ReplayStopped, replay.Status.Phase)
	require.Equal(t, "paused", replay.Status.Message)
}

type fakeReplayDataplaneManager struct {
	registered   map[string]bool
	deregistered []string
	clients      map[string]client.Client
}

func (f *fakeReplayDataplaneManager) IsRegistered(name string) bool {
	return f.registered[name]
}

func (f *fakeReplayDataplaneManager) Register(name string, cfg *rest.Config) error {
	if f.registered == nil {
		f.registered = make(map[string]bool)
	}
	f.registered[name] = true
	_ = cfg
	return nil
}

func (f *fakeReplayDataplaneManager) Deregister(name string) {
	f.deregistered = append(f.deregistered, name)
	delete(f.registered, name)
}

func (f *fakeReplayDataplaneManager) GetClient(name string) client.Client {
	return f.clients[name]
}

func (f *fakeReplayDataplaneManager) GetHelmConfig(string) *action.Configuration {
	return nil
}

func (f *fakeReplayDataplaneManager) GetRestConfig(name string) *rest.Config {
	_ = name
	return nil
}