/*
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"

	cdcv1 "github.com/tidbcloud/ticdc-service-ng/api/v1"
	"github.com/tidbcloud/ticdc-service-ng/controllers/ticdc/task"
	"github.com/tidbcloud/ticdc-service-ng/pkg/cdccluster"
	"github.com/tidbcloud/ticdc-service-ng/pkg/cert"
	"github.com/tidbcloud/ticdc-service-ng/pkg/cloud"
	"github.com/tidbcloud/ticdc-service-ng/pkg/config"
	"github.com/tidbcloud/ticdc-service-ng/pkg/deploy"
	"github.com/tidbcloud/ticdc-service-ng/pkg/k8s"
	"github.com/tidbcloud/ticdc-service-ng/pkg/regionalservice"
	"github.com/tidbcloud/ticdc-service-ng/pkg/util"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"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
	Cache              client.Reader
	RegionalCli        regionalservice.Client
	CDCClusterProvider cdccluster.Provider // inject for integration test
	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
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)
	}
	sharedPoolID := ticdc.Spec.SharedPoolID

	dataplaneNamespace := genTiDBNamespace(sharedPoolID)
	certCommonName := genCertCommonName(sharedPoolID)
	certIssuerName := cert.TiDBComponentsSelfSignedIssuerName

	sharedPool, err := r.RegionalCli.GetSharedPoolByID(ctx, sharedPoolID)
	if err != nil || sharedPool == nil {
		return ctrl.Result{}, fmt.Errorf("fail to get sharedPool: %v", err)
	}
	appbox, err := r.RegionalCli.GetDataPlaneByID(ctx, sharedPool.DataplaneName)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("fail to get appbox: %v", err)
	}
	var (
		cdcClusterProvider cdccluster.Provider
		cloudProvider      cloud.Provider
	)

	certProvider := cloud.NewSelfSignedCertificateProvider(certIssuerName, certCommonName)

	if r.CDCClusterProvider != nil { // inject for integration test
		cdcClusterProvider = r.CDCClusterProvider
	} else {
		logger.Info("Using tidb-operator v2", "TidbCluster", sharedPoolID)
		cdcClusterProvider = cdccluster.NewV2ClusterProvider(r.Config.TiCDCPreStopImage, r.Config.Zones, appbox.Tags)
	}

	switch p := r.Config.Provider; p {
	case config.ProviderAWS:
		cloudProvider = cloud.NewAWSProvider(appbox, r.Config)
	case config.ProviderAliCloud:
		cloudProvider = cloud.NewAliCloudProvider(appbox, r.Config)
	case config.ProviderAzure:
		cloudProvider = cloud.NewAzureProvider(appbox, r.Config)
	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)
	}
	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,
		Config:          r.Config,
	}
	tasks := []deploy.Task{
		task.NewGuardian(c),
		task.NewVolumeAttributesClassTask(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
		errs          error
		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.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.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
	}
	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)
}

func genTiDBNamespace(sharedPoolID string) string {
	return fmt.Sprintf("tidb%s", sharedPoolID)
}
func genCertCommonName(clusterID string) string {
	ns := genTiDBNamespace(clusterID)
	return util.FnvHash(ns)
}