package gardener

import (
	"context"
	"fmt"

	"github.com/pingcap/errors"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/intstr"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"

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

const (
	dataflowAgentName     = "dataflow-agent"
	dataflowAgentSvcName  = "dataflow-agent"
	dataflowAgentGRPCPort = 443

	dataflowAgentGcpSecretName = "dataflow-agent-gcp-credentials"
	gcpCredentialsKey          = "privateKey"
	gcpCredentialsBase         = "/root/gcp"
	gcpCredentialsPath         = "key.json"
)

func deployDataflowAgent(
	ctx context.Context,
	infraCli client.Client,
	shootCli client.Client,
	image string,
	network *cloudv1.Network,
	loadBalancerSourceRanges []string,
) (controllerutil.OperationResult, error) {
	var (
		op  controllerutil.OperationResult
		err error
	)
	labels := map[string]string{
		"app": dataflowAgentName,
	}
	cr := &rbacv1.ClusterRole{
		ObjectMeta: metav1.ObjectMeta{
			Name:   dataflowAgentName,
			Labels: labels,
		},
	}
	_, err = controllerutil.CreateOrUpdate(ctx, shootCli, cr, func() error {
		cr.Rules = []rbacv1.PolicyRule{
			{
				APIGroups: []string{""},
				Resources: []string{"secrets", "services", "endpoints", "serviceaccounts"},
				Verbs:     []string{"get", "watch", "list", "create", "delete", "update"},
			},
			{
				APIGroups: []string{"batch"},
				Resources: []string{"jobs"},
				Verbs:     []string{"get", "list", "watch", "create", "update", "delete"},
			},
			// Allow proxy to bind subjects to the shoot-operator role
			{
				APIGroups: []string{"rbac.authorization.k8s.io"},
				Resources: []string{"clusterrolebindings"},
				Verbs:     []string{"create", "get", "update"},
			},
			{
				APIGroups:     []string{"rbac.authorization.k8s.io"},
				Resources:     []string{"clusterroles"},
				Verbs:         []string{"bind"},
				ResourceNames: []string{shootClusterOperatorRole},
			},
		}
		return nil
	})
	if err != nil {
		return op, errors.AddStack(err)
	}

	sa := &corev1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Name:      dataflowAgentName,
			Namespace: consts.LocalOperatorNamespace,
			Labels:    labels,
		},
	}
	_, err = k8s.CreateIgnoreExists(ctx, shootCli, sa)
	if err != nil {
		return op, err
	}

	crb := &rbacv1.ClusterRoleBinding{
		ObjectMeta: metav1.ObjectMeta{
			Name:   dataflowAgentName,
			Labels: labels,
		},
		Subjects: []rbacv1.Subject{{
			Kind:      rbacv1.ServiceAccountKind,
			Name:      dataflowAgentName,
			Namespace: consts.LocalOperatorNamespace,
		}},
		RoleRef: rbacv1.RoleRef{
			Kind:     "ClusterRole",
			Name:     dataflowAgentName,
			APIGroup: rbacv1.GroupName,
		},
	}
	_, err = k8s.CreateIgnoreExists(ctx, shootCli, crb)
	if err != nil {
		return op, err
	}

	// create GCPCredentials secret
	if network.Spec.Provider == consts.ProviderTypeGCP {
		saName := GetGCPAccessServiceAccountName(network.Spec.Region, network.GetProjectId())
		keyName := GetGCPAccessServiceAccountKeyName(saName)
		secretName := types.NamespacedName{Namespace: network.Namespace, Name: keyName}

		var keySecret corev1.Secret
		if err := infraCli.Get(ctx, secretName, &keySecret); err != nil {
			return op, err
		}
		creds := keySecret.Data[gcpCredentialsKey]
		if len(creds) == 0 {
			return op, fmt.Errorf("gcp service account key is not ready yet")
		}
		secret := &corev1.Secret{
			ObjectMeta: metav1.ObjectMeta{
				Name:      dataflowAgentGcpSecretName,
				Namespace: consts.LocalOperatorNamespace,
			},
		}
		data := map[string][]byte{gcpCredentialsKey: creds}
		_, err = controllerutil.CreateOrUpdate(ctx, shootCli, secret, func() error {
			secret.Data = data
			return nil
		})
		if err != nil {
			return op, err
		}
	}

	deploy := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:      dataflowAgentName,
			Namespace: consts.LocalOperatorNamespace,
		},
		Spec: appsv1.DeploymentSpec{
			Selector: &metav1.LabelSelector{
				MatchLabels: labels,
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: labels,
				},
			},
		},
	}
	_, err = controllerutil.CreateOrUpdate(ctx, shootCli, deploy, func() error {
		deploy.Spec.Replicas = pointer.Int32Ptr(1)
		deploy.Spec.Template.Spec = corev1.PodSpec{
			ServiceAccountName: dataflowAgentName,
			Volumes: []corev1.Volume{
				{
					Name: "tls-certs",
					VolumeSource: corev1.VolumeSource{
						Secret: &corev1.SecretVolumeSource{
							SecretName:  "proxy-certificate-secret",
							DefaultMode: pointer.Int32(0422),
						},
					},
				},
			},
			Containers: []corev1.Container{{
				Name:    "agent",
				Image:   image,
				Command: []string{"./dataflow-agent"},
				Resources: corev1.ResourceRequirements{
					Limits: corev1.ResourceList{
						corev1.ResourceCPU:    resource.MustParse("100m"),
						corev1.ResourceMemory: resource.MustParse("256Mi"),
					},
					Requests: corev1.ResourceList{
						corev1.ResourceCPU:    resource.MustParse("20m"),
						corev1.ResourceMemory: resource.MustParse("80Mi"),
					},
				},
				VolumeMounts: []corev1.VolumeMount{
					{
						Name:      "tls-certs",
						MountPath: "/proxy-tls",
					},
				},
			}},
		}
		// add GCPCredentials
		if network.Spec.Provider == consts.ProviderTypeGCP {
			deploy.Spec.Template.Spec.Containers[0].VolumeMounts = append(
				deploy.Spec.Template.Spec.Containers[0].VolumeMounts,
				corev1.VolumeMount{Name: "gcp-credentials", MountPath: gcpCredentialsBase})
			deploy.Spec.Template.Spec.Volumes = append(
				deploy.Spec.Template.Spec.Volumes,
				corev1.Volume{
					Name: "gcp-credentials",
					VolumeSource: corev1.VolumeSource{
						Secret: &corev1.SecretVolumeSource{
							SecretName: dataflowAgentGcpSecretName,
							Items:      []corev1.KeyToPath{{Key: gcpCredentialsKey, Path: gcpCredentialsPath}}}}})
		}
		return nil
	})
	if err != nil {
		return op, err
	}

	svc := &corev1.Service{
		ObjectMeta: metav1.ObjectMeta{
			Name:      dataflowAgentSvcName,
			Namespace: consts.LocalOperatorNamespace,
		},
	}
	return controllerutil.CreateOrUpdate(ctx, shootCli, svc, func() error {
		svc.Spec.Ports = []corev1.ServicePort{{
			Port:       dataflowAgentGRPCPort,
			TargetPort: intstr.FromInt(dataflowAgentGRPCPort),
			Protocol:   corev1.ProtocolTCP,
		}}
		svc.Spec.Selector = labels
		svc.Spec.Type = corev1.ServiceTypeLoadBalancer
		svc.Spec.LoadBalancerSourceRanges = loadBalancerSourceRanges
		return nil
	})
}