/* Copyright 2021.

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 cluster

import (
	"context"
	"fmt"
	"net/http"
	"time"

	awskmsv1alpha1 "github.com/crossplane/provider-aws/apis/kms/v1alpha1"
	"github.com/go-logr/logr"
	"github.com/tidbcloud/shared-provider/pkg/types"
	"github.com/tidbcloud/shared-provider/pkg/utils/labels"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	apitypes "k8s.io/apimachinery/pkg/types"
	"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/controller"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/source"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	configsdk "github.com/tidbcloud/config-service/sdk"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/providers/aws"
	"github.com/tidbcloud/shared-provider/pkg/gitops"
	"github.com/tidbcloud/shared-provider/pkg/manager"
	"github.com/tidbcloud/shared-provider/pkg/utils"
	"github.com/tidbcloud/shared-provider/pkg/utils/namespace"
)

var ClusterGVK = cloudv1.GroupVersion.WithKind("Cluster")

// Reconciler reconciles a Network object.
type Reconciler struct {
	Client      client.Client
	InfraClient client.Client

	Cache      client.Reader
	InfraCache client.Reader

	Recorder         record.EventRecorder
	InfraRecorder    record.EventRecorder
	bandwidthHandler *BandwidthHandler

	logr.Logger

	AwsClient               aws.Client
	ConfigService           configsdk.Interface
	BaseProxy               string
	PublicHostedZoneDomain  string
	PublicHostedZoneId      string
	KubeClusterName         string
	LoadBalancerClusterName string
	RemoteWriteUrl          string
	EvictDuration           time.Duration
	ConfigProvider          gitops.ConfigProvider
	QuerierForTidbMonitor   func(key apitypes.NamespacedName) Querier
	Env                     string

	AWSRegion string
}

// reconcileContext is the local context of a single cluster reconciliation, it is NOT concurrency safe.
type reconcileContext struct {
	*Reconciler
	logr.Logger

	ctx      context.Context
	provider *cloudv1.Provider
	network  *cloudv1.Network
	tc       *v1alpha1.TidbCluster
	ti       *v1alpha1.TidbInitializer
	tm       *v1alpha1.TidbMonitor
	cluster  *cloudv1.Cluster

	// tags is tags for cloud resource which are inherited from Cluster labels
	// with prefix tags.tidbcloud.com/
	tags map[string]string

	providerType string

	tcKey, tiKey, tmKey client.ObjectKey

	awsRegion, kmsKeyId string

	querier Querier

	backupRole   string
	backupBucket string
}

func (r *Reconciler) newReconcileContext(ctx context.Context, req ctrl.Request) (*reconcileContext, error) {
	cluster := &cloudv1.Cluster{}
	if err := r.InfraCache.Get(ctx, req.NamespacedName, cluster); err != nil {
		if apierrors.IsNotFound(err) {
			return nil, nil
		}
		return nil, fmt.Errorf("can't get cluster from cache: %w", err)
	}

	if kubeCluster := labels.GetKubeClusterFromLabels(cluster.Labels); kubeCluster != r.KubeClusterName {
		r.Info("unexpected cluster scheduling", "kubecluster", kubeCluster)
		return nil, nil
	}

	network := &cloudv1.Network{}
	netKey := client.ObjectKey{Name: cluster.Spec.NetworkName, Namespace: req.Namespace}
	if err := r.InfraCache.Get(ctx, netKey, network); err != nil {
		return nil, fmt.Errorf("can't get network from cache: %w", err)
	}

	provider := &cloudv1.Provider{}
	providerKey := client.ObjectKey{Name: network.Spec.Provider}
	if err := r.InfraCache.Get(ctx, providerKey, provider); err != nil {
		return nil, fmt.Errorf("can't get provider from cache: %w", err)
	}

	ns := namespace.GetOwnedNamespaceNameFromNetwork(network)
	tcKey := client.ObjectKey{Name: "db" + cluster.Name, Namespace: ns}
	tiKey := client.ObjectKey{Name: "db" + cluster.Name, Namespace: ns}
	tmKey := client.ObjectKey{Name: "db" + cluster.Name, Namespace: ns}

	tc := &v1alpha1.TidbCluster{}
	if err := r.Cache.Get(ctx, tcKey, tc); err != nil {
		if !apierrors.IsNotFound(err) {
			return nil, fmt.Errorf("can't get tidbcluster from cache: %w", err)
		}
		tc = nil
	}

	ti := &v1alpha1.TidbInitializer{}
	if err := r.Cache.Get(ctx, tiKey, ti); err != nil {
		if !apierrors.IsNotFound(err) {
			return nil, fmt.Errorf("can't get tidbinitializer from cache: %w", err)
		}
		ti = nil
	}

	tm := &v1alpha1.TidbMonitor{}
	if err := r.Cache.Get(ctx, tmKey, tm); err != nil {
		if !apierrors.IsNotFound(err) {
			return nil, fmt.Errorf("can't get tidbmonitor from cache: %w", err)
		}
		tm = nil
	}

	var kmsKeyId string
	if cluster.Annotations[kmsPermissionFinishAnno] != "" && cluster.DeletionTimestamp == nil {
		k := &awskmsv1alpha1.Key{}
		if err := r.Cache.Get(ctx, client.ObjectKey{Name: cluster.ClusterScopeName()}, k); err != nil {
			return nil, fmt.Errorf("can't get key from cache: %w", err)
		}
		if k.Status.AtProvider.KeyID == nil {
			return nil, fmt.Errorf("get keyid is nil")
		}
		kmsKeyId = *k.Status.AtProvider.KeyID
	}

	querier := r.QuerierForTidbMonitor(tmKey)

	networkSharedStatus, err := types.SharedNetworkStatusFromProvider(network)
	if err != nil {
		return nil, fmt.Errorf("can't get network shared status: %w", err)
	}

	return &reconcileContext{
		Reconciler: r,
		Logger:     ctrl.LoggerFrom(ctx),

		ctx:          ctx,
		provider:     provider,
		providerType: provider.Name,
		network:      network,
		tc:           tc,
		ti:           ti,
		tm:           tm,
		cluster:      cluster,
		tcKey:        tcKey,
		tiKey:        tiKey,
		tmKey:        tmKey,
		awsRegion:    r.AWSRegion,
		kmsKeyId:     kmsKeyId,
		querier:      querier,
		tags:         cloudv1.TagsFromLabels(cluster.Labels),

		backupRole:   networkSharedStatus.BackupRole,
		backupBucket: networkSharedStatus.BackupBucket,
	}, nil
}

func (r *reconcileContext) haveKms() bool {
	return r.cluster.Annotations[kmsPermissionFinishAnno] != ""
}

func (r *reconcileContext) haveTc() bool {
	return r.tc != nil
}

func (r *reconcileContext) haveTi() bool {
	return r.ti != nil
}

func (r *reconcileContext) haveTm() bool {
	return r.tm != nil
}

func (r *reconcileContext) needPublicEndpoint() bool {
	return r.cluster.Spec.SqlEndpoints.Public.IsEnabled() && r.cluster.Status.SqlEndpoints.PublicEndpoint == nil
}

func (r *reconcileContext) needPrivateEndpoint() bool {
	return r.cluster.Spec.SqlEndpoints.Private != nil &&
		r.cluster.Spec.SqlEndpoints.Private.IsEnabled()
}

func (r *reconcileContext) needPrivateLink() bool {
	_, ok := r.cluster.Annotations[annoKeyPrivateLinkSpec]
	return ok
}

//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusters,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusters/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusters/finalizers,verbs=update
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=providers,verbs=get;list;watch
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=imports;backups;restores,verbs=*
//+kubebuilder:rbac:groups=pingcap.com,resources=tidbclusters,verbs=*
//+kubebuilder:rbac:groups=pingcap.com,resources=tidbmonitors,verbs=*
//+kubebuilder:rbac:groups=route53.aws.crossplane.io,resources=resourcerecordsets,verbs=*
//+kubebuilder:rbac:groups=kms.aws.crossplane.io,resources=keys,verbs=*
//+kubebuilder:rbac:groups=cert-manager.io,resources=certificates,verbs=*
//+kubebuilder:rbac:groups=core,resources=serviceaccounts;services;persistentvolumeclaims;secrets;configmaps;events,verbs=*
//+kubebuilder:rbac:groups=core,resources=pods;nodes,verbs=get;list;watch

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := ctrl.LoggerFrom(ctx)
	log.Info("start reconcile cluster")
	start := time.Now()
	defer func() {
		log.Info("finish reconcile cluster", "time", time.Since(start))
	}()

	rx, err := r.newReconcileContext(ctx, req)
	if err != nil || rx == nil {
		return ctrl.Result{}, err
	}

	if err = rx.ensureFinalizer(); err != nil {
		return ctrl.Result{}, err
	}

	if err = rx.updateDefaultValue(); err != nil {
		return ctrl.Result{}, err
	}

	var phase cloudv1.ClusterPhase
	var reason string
	var needUpdatePhase bool

	if rx.cluster.DeletionTimestamp != nil {
		if rx.cluster.Status.Phase != cloudv1.ClusterPhaseDeleting &&
			rx.cluster.Status.Phase != cloudv1.ClusterPhasePendingDelete {
			// shortcut to PendingDelete phase whenever cluster is get deleted
			phase = cloudv1.ClusterPhasePendingDelete
			reason = "cluster is pending to be deleted"
			needUpdatePhase = true
		}
	} else {
		if rx.cluster.Status.Phase == "" {
			// shortcut to Pending phase when cluster is newly created
			phase = cloudv1.ClusterPhasePending
			reason = "cluster is waiting to be reconciled"
			needUpdatePhase = true
		} else if rx.cluster.Status.Phase != cloudv1.ClusterPhaseCreating && !rx.haveTc() {
			// force to Creating phase when tc is nil
			phase = cloudv1.ClusterPhaseCreating
			reason = "ensuring tidb cluster"
			needUpdatePhase = true
		}
	}

	if needUpdatePhase {
		if err = rx.updateProgress(phase, reason); err != nil {
			return ctrl.Result{}, err
		}
	}

	clusterPhase := rx.cluster.Status.Phase
	switch clusterPhase {
	case cloudv1.ClusterPhaseCreating, cloudv1.ClusterPhasePending:
		err = rx.createCluster()
	case cloudv1.ClusterPhaseRunning, cloudv1.ClusterPhaseUpgrading, cloudv1.ClusterPhaseScaling:
		err = rx.reconcileCluster()
	case cloudv1.ClusterPhasePendingDelete:
		err = rx.pendingDeleteCluster()
	case cloudv1.ClusterPhaseDeleting:
		err = rx.deleteCluster()
	case cloudv1.ClusterPhaseUnavailable, cloudv1.ClusterPhaseHibernated:
		// when cluster is unavailable, skip reconciliation and observe until it recovers for safety
		err = rx.observeCluster()
	default:
		// unknown phase, try to recover the real phase via observation
		err = rx.observeCluster()
	}

	if clusterPhase == cloudv1.ClusterPhaseDeleting && err == nil {
		err = rx.tryRemoveFinalizer()
	}

	return utils.ReconcileErr(log, err)
}

// SetupWithManager sets up the controller with the Manager.
func (r *Reconciler) SetupWithManager(mgr manager.Manager) error {
	r.Client = mgr.GetClient()
	r.InfraClient = mgr.InfraCluster().GetClient()
	r.Cache = mgr.GetCache()
	r.InfraCache = mgr.InfraCluster().GetCache()
	r.Recorder = mgr.GetEventRecorderFor("Cluster")
	r.InfraRecorder = mgr.InfraCluster().GetEventRecorderFor("Cluster")
	r.bandwidthHandler = NewBandwidthHandler(r.Client, r.InfraClient)

	r.QuerierForTidbMonitor = func(tmKey client.ObjectKey) Querier {
		return newHTTPClientQueyer(
			&http.Client{},
			fmt.Sprintf("http://%s.%s:9090", tmKey.Name+"-prometheus", tmKey.Namespace),
		)
	}

	c, err := controller.New("Cluster", mgr, controller.Options{
		Reconciler: r,
	})
	if err != nil {
		return fmt.Errorf("can't new a controller: %w", err)
	}

	r.Logger = c.GetLogger()

	return r.watches(c, mgr)
}

func (r *Reconciler) watches(c controller.Controller, mgr manager.Manager) error {
	ptp := utils.GetPredicateForProviderTypes(types.ProviderTypeGCPSharedTier, types.ProviderTypeAWSSharedTier)
	p, err := predicate.LabelSelectorPredicate(metav1.LabelSelector{
		MatchLabels: map[string]string{
			labels.LabelKeyKubernetesCluster: r.KubeClusterName,
		},
	})
	if err != nil {
		return fmt.Errorf("invalid label selector: %w", err)
	}
	if err := c.Watch(
		source.NewKindWithCache(&cloudv1.Cluster{}, mgr.InfraCluster().GetCache()),
		&handler.EnqueueRequestForObject{},
		ptp,
		p,
	); err != nil {
		return fmt.Errorf("can't watch cluster: %w", err)
	}

	cp, err := predicate.LabelSelectorPredicate(metav1.LabelSelector{
		MatchLabels: map[string]string{
			labels.LabelKeyKubernetesCluster: r.KubeClusterName,
		},
	})
	if err != nil {
		return fmt.Errorf("invalid label selector: %w", err)
	}
	if err := c.Watch(
		source.NewKindWithCache(&v1alpha1.TidbCluster{}, mgr.GetCache()),
		&handler.Funcs{
			UpdateFunc: r.handleUpdateTcEvent,
			DeleteFunc: r.handleDeleteTcEvent,
		},
		cp,
	); err != nil {
		return fmt.Errorf("can't watch tidbcluster: %w", err)
	}

	return nil
}