package bucket

import (
	"context"
	"fmt"
	"time"

	"github.com/go-logr/logr"
	"go.opentelemetry.io/otel"
	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"

	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/cloud/azure"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/bucket/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/k8s"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
)

var tracer = otel.Tracer("BucketReconciler")

type Reconciler struct {
	client.Client
	Log           logr.Logger
	EventRecorder record.EventRecorder
	BucketRW      repository.BucketReaderWriter
	DataPlaneRW   repository.DataPlaneReaderWriter
	APPBoxRW      repository.APPBoxReaderWriter
	Config        rsconfig.RootConfig
	AzureCli      azure.Client
}

func NewReconciler(
	cli client.Client,
	log logr.Logger,
	recorder record.EventRecorder,
	rsconfig rsconfig.RootConfig,
	azureCli azure.Client,
) *Reconciler {
	return &Reconciler{
		Client:        cli,
		Log:           log,
		EventRecorder: recorder,
		BucketRW:      repository.NewBucketReaderWriter(cli),
		DataPlaneRW:   repository.NewDataPlaneReaderWriter(cli),
		APPBoxRW:      repository.NewAPPBoxReaderWriter(cli),
		Config:        rsconfig,
		AzureCli:      azureCli,
	}
}

func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := r.Log.WithValues("bucket", req.NamespacedName)
	ctx = log.IntoContext(ctx, logger)

	start := time.Now()
	defer func() {
		logger.Info("Finish RegionalServer Bucket reconcile", "take", time.Since(start))
	}()
	bucket := new(v1alpha1.Bucket)
	if err := r.Get(ctx, req.NamespacedName, bucket); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Info("Bucket not found, skipping reconcile")
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, fmt.Errorf("unable to fetch Bucket: %w", err)
	}
	if utils.IsPaused(bucket.GetObjectMeta()) {
		logger.Info("reconciliation is paused either through the pause annotation",
			"annotation", v1alpha1.AnnotationKeyReconciliationPaused)
		r.EventRecorder.Event(bucket, corev1.EventTypeWarning, "ReconciliationPaused", "reconciliation is paused through the annotation "+v1alpha1.AnnotationKeyReconciliationPaused)
	}
	controllers.EnsureDataPlaneIDLabel(bucket, bucket.Spec.DataplaneId)

	c, err := (&task.Context{
		Local:      r.Client,
		Logger:     logger,
		Config:     r.Config,
		Bucket:     bucket,
		DataPlaneR: r.DataPlaneRW,
		AppBoxR:    r.APPBoxRW,
		AzureCli:   r.AzureCli,
	}).Default()
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("unable to create task context: %w", err)
	}

	tasks := GetBucketTasks(c)

	if bucket.DeletionTimestamp.IsZero() && isNotReadyBucket(bucket) {
		tasks = r.setHookOnTasks(tasks, c)
	}
	d := deploy.NewDeployer(tasks, nil, logger)

	var res ctrl.Result
	if bucket.DeletionTimestamp.IsZero() {
		res, err = r.updateCreate(ctx, logger, c, d.Deploy(ctx))
	} else {
		if controllers.DeletionConfirmed(bucket) {
			bucket.Status.Phase = v1alpha1.BucketPhaseDeleting
			if err := r.BucketRW.UpdateStatus(ctx, bucket); err != nil {
				return ctrl.Result{}, fmt.Errorf("failed to update bucket status: %w", err)
			}
			res, err = r.updateDelete(ctx, logger, c, d.Destroy(ctx))
		} else {
			logger.Info("waiting for bucket deletion to be confirmed")
			return ctrl.Result{}, fmt.Errorf("bucket deletion is not confirmed")
		}
	}
	if err == nil {
		r.EventRecorder.Event(bucket, corev1.EventTypeNormal, "ReconcileSuccess", "")
	} else {
		r.EventRecorder.Event(bucket, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
	}
	return res, err
}

func (r *Reconciler) setHookOnTasks(tasks []deploy.Task, c *task.Context) []deploy.Task {
	hook := func(ctc context.Context, task deploy.Task) {
		name := task.Name()
		passTimeKey := utils.GenTaskFirstPastAnnotationKey(name)
		_, ok := c.Bucket.Annotations[passTimeKey]
		if ok {
			return
		}
		if c.Bucket.Annotations == nil {
			c.Bucket.Annotations = map[string]string{}
		}
		c.Bucket.Annotations[passTimeKey] = time.Now().Format(time.RFC3339)
		log.FromContext(ctc).Info("pass task when creating bucket", "task", name)
	}
	for idx, t := range tasks {
		tasks[idx] = deploy.NewTaskWraper(t, hook, tracer)
	}
	return tasks
}

func isNotReadyBucket(bucket *v1alpha1.Bucket) bool {
	return bucket.Status.Phase != v1alpha1.BucketPhaseReady
}

func (r *Reconciler) updateCreate(ctx context.Context, logger logr.Logger, t *task.Context, results []deploy.Result) (ctrl.Result, error) {
	res, err, retriableErrs := deploy.MinRequeue(results)
	if err != nil {
		logger.Info("deploy create encounters error", "unretriable", err)
	} else if retriableErrs != nil {
		logger.Info("deploy create encounters error", "retriable", retriableErrs)
	}

	if e := r.updateBucket(ctx, logger, t.Bucket); e != nil {
		return res, fmt.Errorf("failed to update Bucket: %w", e)
	}
	return res, err
}

func (r *Reconciler) updateBucket(ctx context.Context, logger logr.Logger, bucket *v1alpha1.Bucket) error {
	bucketStatus := bucket.Status.DeepCopy()
	if err := r.BucketRW.Update(ctx, bucket); err != nil {
		logger.Error(err, "failed to update Bucket")
		return err
	}
	bucket.Status = *bucketStatus
	if err := r.BucketRW.UpdateStatus(ctx, bucket); err != nil && !apierrors.IsNotFound(err) {
		logger.Error(err, "failed to update Bucket status")
		return err
	}
	return nil
}

func (r *Reconciler) updateDelete(ctx context.Context, logger logr.Logger, t *task.Context, results []deploy.Result) (ctrl.Result, error) {
	res, err, retriableErrs := deploy.MinRequeue(results)
	if err != nil {
		logger.Info("deploy delete encounters error", "unretriable", err)
	} else if retriableErrs != nil {
		logger.Info("deploy delete encounters error", "retriable", retriableErrs)
	} else {
		logger.Info("deploy delete successfully, remove finalizer")
		controllerutil.RemoveFinalizer(t.Bucket, v1alpha1.Finalizer)
	}

	if e := r.updateBucket(ctx, logger, t.Bucket); e != nil {
		return res, fmt.Errorf("failed to update Bucket: %w", e)
	}
	return res, err
}

func (r *Reconciler) SetupWithManager(mgr ctrl.Manager, pds ...predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&v1alpha1.Bucket{}, builder.WithPredicates(pds...)).
		WithOptions(controller.Options{
			RateLimiter: k8s.RetryLimiter,
		}).
		Complete(r)
}