/*
Copyright 2025.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package controller

import (
	"context"
	"fmt"

	compute "cloud.google.com/go/compute/apiv1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	cdcv1 "github.com/tidbcloud/ticdc-service/api/v1"
	"github.com/tidbcloud/ticdc-service/controllers/ticdc/task"
	"github.com/tidbcloud/ticdc-service/pkg/cdccluster"
	"github.com/tidbcloud/ticdc-service/pkg/cloud"
	"github.com/tidbcloud/ticdc-service/pkg/config"
	"github.com/tidbcloud/ticdc-service/pkg/regionalservice"
	"github.com/tidbcloud/ticdc-service/pkg/tidbcluster"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
)

// TiCDCReconciler reconciles a TiCDC object
type TiCDCReconciler struct {
	EventRecorder record.EventRecorder
	client.Client
	Scheme             *runtime.Scheme
	Cache              client.Reader
	RegionalCli        regionalservice.Client
	CDCClusterProvider cdccluster.Provider // inject for integration test
	ClusterNS          string              // inject for integration test
	Scaler             cloud.Scaler        // inject for integration test

	ForwardingRuleClient *compute.ForwardingRulesClient
	Config               *config.Config
}

// +kubebuilder:rbac:groups=cdc.tidbcloud.com,resources=ticdcs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=cdc.tidbcloud.com,resources=ticdcs/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=cdc.tidbcloud.com,resources=ticdcs/finalizers,verbs=update

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// TODO(user): Modify the Reconcile function to compare the state specified by
// the TiCDC object against the actual cluster state, and then
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.19.1/pkg/reconcile
//
//nolint:gocyclo
func (r *TiCDCReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	logger.Info("Reconciling TiCDC")

	ticdc := &cdcv1.TiCDC{}
	if err := r.Cache.Get(ctx, req.NamespacedName, ticdc); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, fmt.Errorf("unable to get ticdc %s: %v", req.NamespacedName, err)
	}

	tenantID := ticdc.Labels[cloudv1.TenantTagKey]
	clusterID := ticdc.Spec.ClusterID
	projectID := ticdc.Labels[cloudv1.ProjectTagKey]

	// ticdc cluster is deployed in the same namespace as tidb cluster
	// In theory, we should get the namespace from cluster-service
	// but we use the naming convention as a temporary solution
	// due to cluster-service does not provide such API yet.
	var clusterNS string
	if r.ClusterNS != "" {
		clusterNS = r.ClusterNS
	} else {
		clusterNS = fmt.Sprintf("%s-%s", r.Config.ClusterNSPrefix, projectID)
	}

	tidbclusterProvider, err := tidbcluster.NewClusterCRProvider(ctx, r.Client, clusterNS, clusterID, r.Config.Region)
	if err != nil {
		return ctrl.Result{}, err
	}
	dataplaneNamespace, err := tidbclusterProvider.GetNamespace(ctx, clusterID)
	if err != nil {
		return ctrl.Result{}, err
	}
	certCommonName, err := tidbclusterProvider.GetCertCommonName(ctx, clusterID)
	if err != nil {
		return ctrl.Result{}, err
	}
	certIssuerName, err := tidbclusterProvider.GetCertIssuerName(ctx, clusterID)
	if err != nil {
		return ctrl.Result{}, err
	}
	clusterRole, err := tidbclusterProvider.GetClusterRole(ctx, clusterID)
	if err != nil {
		return ctrl.Result{}, err
	}

	appbox, err := r.RegionalCli.GetAPPBoxByTenantIDAndClusterID(ctx, tenantID, clusterID)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("fail to get appbox: %v", err)
	}
	var (
		cdcClusterProvider cdccluster.Provider
		cloudProvider      cloud.Provider
		certProvider       cloud.CertificateProvider
	)

	selfSigned, err := tidbclusterProvider.FeatureEnabled(
		ctx,
		clusterID,
		consts.ClusterFeatureUseSelfSignedCertForTiDBComponents,
	)
	if err != nil {
		return ctrl.Result{}, err
	}
	if selfSigned {
		certProvider = cloud.NewSelfSignedCertificateProvider(certIssuerName, certCommonName)
	} else {
		certProvider = cloud.NewPCACertificateProvider(certIssuerName, certCommonName)
	}

	if r.CDCClusterProvider != nil { // inject for integration test
		cdcClusterProvider = r.CDCClusterProvider
	} else {
		v2Enabled, err := tidbclusterProvider.FeatureEnabled(
			ctx,
			clusterID,
			consts.ClusterFeatureUseTiDBOperatorV2,
		)
		if err != nil {
			return ctrl.Result{}, err
		}
		if v2Enabled {
			logger.Info("Using tidb-operator v2", "TidbCluster", clusterID)
			cdcClusterProvider = cdccluster.NewV2ClusterProvider(r.Config.TiCDCPreStopImage)
		} else {
			logger.Info("Using tidb-operator v1", "TidbCluster", clusterID)
			cdcClusterProvider = cdccluster.NewV1ClusterProvider()
		}
	}

	switch p := r.Config.Provider; p {
	case config.ProviderAWS:
		cloudProvider = cloud.NewAWSProvider(appbox, r.Config)
	case config.ProviderGCP:
		cloudProvider, err = cloud.NewGcpProvider(appbox, r.Config)
		if err != nil {
			return ctrl.Result{}, fmt.Errorf("fail to build gcp Provider: %v", err)
		}
	case config.ProviderAzure:
		tid := r.Config.Azure.CrossplaneProviderTenantID
		cid := r.Config.Azure.CrossplaneProviderClientID
		cloudProvider, err = cloud.NewAzureProvider(ctx, tid, cid, r.RegionalCli, r.Client, appbox)
		if err != nil {
			return ctrl.Result{}, fmt.Errorf("fail to build azure Provider: %v", err)
		}
	default:
		return ctrl.Result{}, fmt.Errorf("unsupported provider: %s", p)
	}
	remote, err := cloudProvider.GetKubeClient(ctx)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to get remote kube client: %v", err)
	}
	var scaler cloud.Scaler
	if r.Scaler != nil {
		scaler = r.Scaler // inject for integration test
	} else {
		scaler = cloud.NewScaler(remote, dataplaneNamespace, ticdc.Name)
	}
	c := &task.Context{
		Recorder:        r.EventRecorder,
		Logger:          logger,
		Env:             r.Config.Env,
		Local:           r.Client,
		Cdc:             ticdc,
		Remote:          remote,
		Namespace:       dataplaneNamespace,
		CloudProvider:   cloudProvider,
		ClusterProvider: cdcClusterProvider,
		CertProvider:    certProvider,
		Scaler:          scaler,
		Config:          r.Config,
		ClusterRole:     clusterRole.CloudIamIdentity,
	}
	tasks := []deploy.Task{
		task.NewGuardian(c),
		task.NewProviderConfigTask(c),
		task.NewNodePoolTask(c),
		task.NewStorageClassTask(c),
		task.NewCertificateTask(c),
		task.NewClusterTask(c),
	}
	deployer := deploy.NewDeployer(tasks, nil, logger)
	deleted := !ticdc.DeletionTimestamp.IsZero()
	// finalizer is added to cdcv1.TiCDC in Guardian task
	// but Guardian task can't detect other task destroy results
	// so we remove finalizer in the main controller
	var res ctrl.Result
	var errs error
	var retriableErrs error
	if deleted {
		results := deployer.Destroy(ctx)
		res, errs, retriableErrs = deploy.MinRequeue(results)
		if errs == nil && retriableErrs == nil {
			// all tasks destroy finished without error, so remove finalizer
			controllerutil.RemoveFinalizer(ticdc, task.Finalizer)
		}
		if err := r.Client.Update(ctx, ticdc); err != nil {
			r.EventRecorder.Event(ticdc, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
			return res, err
		}
	} else {
		results := deployer.Deploy(ctx)
		res, errs, retriableErrs = deploy.MinRequeue(results)
		if errs == nil && retriableErrs == nil {
			// all tasks deploy finish without error
			logger.Info("All tasks completed")
		}
		cdcStatus := ticdc.Status.DeepCopy()
		if err := r.Client.Update(ctx, ticdc); err != nil {
			r.EventRecorder.Event(ticdc, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
			return res, err
		}
		ticdc.Status = *cdcStatus
		if err := r.Client.Status().Update(ctx, ticdc); err != nil {
			r.EventRecorder.Event(ticdc, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
			return res, err
		}
	}
	if errs != nil {
		r.EventRecorder.Event(ticdc, corev1.EventTypeWarning, "ReconcileFailed", errs.Error())
		return res, errs
	} else if retriableErrs != nil {
		r.EventRecorder.Event(ticdc, corev1.EventTypeWarning, "ReconcileFailed", retriableErrs.Error())
		return res, retriableErrs
	} else {
		r.EventRecorder.Event(ticdc, corev1.EventTypeNormal, "ReconcileSuccess", "")
		return res, nil
	}
}

// SetupWithManager sets up the controller with the Manager.
func (r *TiCDCReconciler) SetupWithManager(mgr ctrl.Manager, preds ...predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&cdcv1.TiCDC{}, builder.WithPredicates(preds...)).
		WithOptions(controller.Options{RateLimiter: k8s.RetryLimiter, MaxConcurrentReconciles: 5}).
		Named("ticdc").
		Complete(r)
}