/*
Copyright 2022.

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 clusterconfig

import (
	"context"
	"fmt"
	"net/http"
	"reflect"
	"strconv"
	"strings"
	"sync"

	"github.com/go-logr/logr"
	"github.com/google/go-cmp/cmp"
	gh "github.com/google/go-github/v56/github"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/prometheus/client_golang/prometheus"
	metric "github.com/tidbcloud/config-manager/pkg/gitops-v2/metric"
	"github.com/tidbcloud/config-manager/pkg/metrics"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	sdk "github.com/tidbcloud/infra-provider/pkg/gitops-sdk"
	"github.com/tidbcloud/infra-provider/pkg/gitops-sdk/consts"
	"github.com/tidbcloud/infra-provider/pkg/gitops-sdk/revision"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	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/handler"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
)

const (
	ConfigRevisionRefField = ".spec.configRevisionRef.name"

	reasonImageUpdated  = "image updated"
	reasonConfigUpdated = "configuration updated"

	ghOwner = "tidbcloud"
	ghRepo  = "tidb-configuration"
)

// we use interface just for unittest.
type CommentsAppender interface {
	ListPullRequestsWithCommit(ctx context.Context, owner, repo, sha string, opts *gh.ListOptions) ([]*gh.PullRequest, *gh.Response, error)
	CreateComment(ctx context.Context, owner string, repo string, number int, comment *gh.IssueComment) (*gh.IssueComment, *gh.Response, error)
}

type GithubClientWrapper struct {
	*gh.Client
}

func (gca *GithubClientWrapper) ListPullRequestsWithCommit(ctx context.Context, owner, repo, sha string, opts *gh.ListOptions) ([]*gh.PullRequest, *gh.Response, error) {
	return gca.PullRequests.ListPullRequestsWithCommit(ctx, owner, repo, sha, opts)
}

func (gca *GithubClientWrapper) CreateComment(ctx context.Context, owner string, repo string, number int, comment *gh.IssueComment) (*gh.IssueComment, *gh.Response, error) {
	return gca.Issues.CreateComment(ctx, owner, repo, number, comment)
}

// ClusterConfigReconciler reconciles a ClusterConfig object
type ClusterConfigReconciler struct {
	client.Client
	revisioner revision.Interface
	g          sdk.CCInterface
	Log        logr.Logger

	enableNotification bool
	ghClient           CommentsAppender
	jumpEndpoint       string // with schema
	commentsCache      sync.Map
}

// Reconcile with cluster object, for every enqueued cluster,
// 1. find related raw ClusterConfigs in infra-api
// 4. change RevisionRef on the Cluster.
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusterconfigs,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusterconfigs/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusterconfigs/finalizers,verbs=update
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusters,verbs=get;list;watch;update;patch
// +kubebuilder:rbac:groups=cloud.pingcap.com,resources=configrevisions,verbs=get;list;watch;create;update;patch;delete
func (r *ClusterConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := r.Log.WithValues("cluster", req.NamespacedName)

	c := &cloudv1.Cluster{}
	if err := r.Get(ctx, req.NamespacedName, c); err != nil {
		// we'll ignore not-found errors, since they can't be fixed by an immediate
		// requeue (we'll need to wait for a new notification), and we can get them
		// on deleted requests.
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	switch c.Labels[cloudv1.ProviderTypeLabel] {
	case consts.ProviderTypeAWSFreeTier, consts.ProviderTypeAWSSharedTier, consts.ProviderTypeGCPSharedTier:
		log.Info("skip for serverless cluster")
		return ctrl.Result{}, nil
	case consts.ProviderTypeAWS, consts.ProviderTypeGCP:
		log.Info("skip for gardener-arch cluster")
		return ctrl.Result{}, nil
	}

	// Do nothing if the cluster is deleting.
	// ClusterConfig will be GCed by background workers.
	if !c.GetDeletionTimestamp().IsZero() {
		return ctrl.Result{}, nil
	}

	// calculate overlayed ClusterConfig, we treat this as VirtualNode
	cc, err := r.g.DescribeClusterConfiguration(c)
	if err != nil {
		return ctrl.Result{}, err
	}

	oldCr, err := GetActiveClusterRevision(r.revisioner, c)
	if err != nil {
		return ctrl.Result{}, err
	}
	cr, err := sdk.GetClusterRevision(r.revisioner, cc, false)
	if err != nil {
		return ctrl.Result{}, err
	}
	lastCR, err := GetLastCreatedConfigRevision(r.revisioner, c.Name)
	if err != nil {
		return ctrl.Result{}, err
	}
	if revision.EqualRevision(lastCR, cr) {
		log.Info("configRevision.spec is not changed between last generated one and new one, skip.", "lastCR", lastCR.Name, "newCR", cr.Name)
		return ctrl.Result{}, nil
	}

	analysisPlanAndRecordMetrics(c, oldCr, cr, cc.IsDryrun(), c.Labels[consts.ClusterTagKey], c.Labels[cloudv1.ProviderTypeLabel])
	logPlan(log, oldCr.Spec, cr.Spec, c, cc.IsDryrun(), c.Labels[cloudv1.ProviderTypeLabel])

	if cc.IsDryrun() {
		return ctrl.Result{}, nil
	}
	// may create a new revision.
	cr, err = sdk.GetClusterRevision(r.revisioner, cc, true)
	if err != nil {
		log.Error(err, "get cluster config revision")
		return ctrl.Result{}, err
	}
	log.Info("debug configrevision", "name", cr.Name, "cc_labels", cc.GetLabels(), "cr_labels", cr.GetLabels())
	// need add some mark on annotation as the config revision is processed?
	var isTestInfraGitCommit bool

	// skip cc create by test-infra on staging.
	// if the cluter config is created by test-infra. the gitCommitID field is always `hacked-config-by-test-infra`.
	if cc.Spec.ConfigMeta.GitCommitID == "hacked-config-by-test-infra" {
		isTestInfraGitCommit = true
	}
	mergedPr, err := r.getMergedPRByCommit(ctx, c, cc.Spec.ConfigMeta.GitCommitID)
	if err != nil {
		log.Error(err, "get pr by commit")
		return ctrl.Result{}, err
	}
	if r.enableNotification && !isTestInfraGitCommit {
		if err := r.addPRMessageToGithub(ctx, c, cr, mergedPr); err != nil {
			// no need to log
			return ctrl.Result{}, err
		}
		return ctrl.Result{}, nil
	}

	//update, err := r.updateClusterRevisionRef(ctx, c, cr, cc.IsDryrun())
	//if err != nil {
	//	log.Error(err, "update cluster failed")
	//	return ctrl.Result{}, err
	//}
	//
	//if update {
	//	// Debug event
	//	err = SendEvent(c, oldCr, cr)
	//	if err != nil {
	//		log.Error(err, "send update cluster config event failed")
	//	}
	//}
	return ctrl.Result{}, nil
}

//func (r *ClusterConfigReconciler) updateClusterRevisionRef(ctx context.Context, c *cloudv1.Cluster, rev *cloudv1.ConfigRevision, dryrun bool) (bool, error) {
//	if c.Spec.ConfigRevisionRef != nil && c.Spec.ConfigRevisionRef.Name == rev.Name {
//		return false, nil
//	}
//	c.Spec.ConfigRevisionRef = &cloudv1.Reference{
//		Name: rev.Name,
//	}
//	return true, r.Update(ctx, c)
//}

// TODO: some fix left
func (r *ClusterConfigReconciler) addPRMessageToGithub(ctx context.Context, c *cloudv1.Cluster, rev *cloudv1.ConfigRevision, pr *gh.PullRequest) error {
	log := r.Log.WithValues("cluster", c.Name, "revision", rev.Name).WithValues("prNum", pr.GetNumber())

	// if configRevisionRef is equal to rev.Name, no need to add pr comment.
	if c.Spec.ConfigRevisionRef != nil && c.Spec.ConfigRevisionRef.Name == rev.Name {
		log.Info("no configRevision change, skip")
		return nil
	}
	// TODO: set base branch to filter prs
	if val, loaded := r.commentsCache.Load(c.Name); loaded {
		previous, ok := val.(string)
		if ok && previous == rev.Name {
			// no need to send GitHub PR comments
			log.Info("already sent, skip")
			return nil
		}
	}
	log.Info("try to send PR comment")

	// TODO: use template
	clusterID := c.Name
	projectID := c.Labels[consts.ProjectTagKey]
	prNumber := pr.GetNumber()
	url := fmt.Sprintf("%s?clusterid=%v&projectid=%v&pr=%v&starttime=%v&configrevisionid=%v",
		r.jumpEndpoint,
		clusterID,
		projectID,
		prNumber,
		rev.CreationTimestamp.Unix(),
		rev.Name)
	comment, resp, err := r.ghClient.CreateComment(ctx, ghOwner, ghRepo, prNumber, &gh.IssueComment{
		Body: &url,
	})
	if err != nil {
		log.Error(err, "create pull request comment failed")
		return err
	}
	if resp.StatusCode >= 300 {
		log.WithValues("status", resp.Status).Error(err, "create pull request comment failed")
		return fmt.Errorf("resp with status %v", resp.Status)
	}
	log.Info("comment sent", "commentID", comment.GetID())
	r.commentsCache.Store(clusterID, rev.Name)
	return nil
}

func (r *ClusterConfigReconciler) getMergedPRByCommit(ctx context.Context, c *cloudv1.Cluster, commitHash string) (*gh.PullRequest, error) {
	log := r.Log.WithValues("commit", commitHash)

	prs, resp, err := r.ghClient.ListPullRequestsWithCommit(ctx, ghOwner, ghRepo, commitHash, &gh.ListOptions{})
	if err != nil {
		log.Error(err, "list pull request with commit failed")
		return nil, err
	}
	if resp.StatusCode != http.StatusOK {
		log.WithValues("status", resp.Status).Error(err, "list pull request with commit failed")
		return nil, fmt.Errorf("resp with status %v", resp.Status)
	}

	var mergedPRs []*gh.PullRequest
	for _, pr := range prs {
		// the field 'Merged' is NOT accurate and can't be used. Follow official doc I found the field is not existing
		// but provide the 'MergedAt'.
		if ptr.Deref(pr.State, "") == "closed" &&
			pr.MergedAt != nil {
			mergedPRs = append(mergedPRs, pr)
		}
	}
	metrics.PRNumByCommit.WithLabelValues(c.Name, commitHash).Set(float64(len(mergedPRs)))
	// in theory, there should be one and only one merged pr that is associated this commit.
	if len(mergedPRs) != 1 {
		err := fmt.Errorf("the count of PRs assocated with this commit is not 1")
		log.WithValues("count", len(mergedPRs)).Error(err, "mergedPrs", "mergedPRs", mergedPRs)
		return nil, err
	}

	log.Info("found PR", "prNum", mergedPRs[0].Number)
	return mergedPRs[0], nil
}

// GetActiveClusterRevision returns the used ConfigRevision for the cluster.
func GetActiveClusterRevision(rv revision.Interface, c *cloudv1.Cluster) (*cloudv1.ConfigRevision, error) {

	if c.Spec.ConfigRevisionRef == nil || c.Spec.ConfigRevisionRef.Name == "" {
		return &cloudv1.ConfigRevision{
			ObjectMeta: metav1.ObjectMeta{},
			Spec:       cloudv1.ConfigRevisionSpec{},
		}, nil
	}

	cr, err := rv.GetConfigRevisions(client.ObjectKey{Name: c.Spec.ConfigRevisionRef.Name})
	return cr, err
}

// GetLastCreatedConfigRevision returns the last generated ConfigRevision or empty ConfigRevision for the cluster.
func GetLastCreatedConfigRevision(rv revision.Interface, clusterName string) (*cloudv1.ConfigRevision, error) {
	crs, err := rv.ListConfigRevisions(cloudv1.SelectAllConfigRevisions4Cluster(clusterName))
	if err != nil {
		return nil, fmt.Errorf("list configRevisions: %w", err)
	}
	latestCR := cloudv1.LatestCreatedConfigRevision(crs)
	return &latestCR, nil
}

func NewClusterConfigReconciler(mgr ctrl.Manager, log logr.Logger, opts ...Option) *ClusterConfigReconciler {
	r := &ClusterConfigReconciler{
		Log:        log,
		Client:     mgr.GetClient(),
		revisioner: revision.NewRevisioner(mgr.GetClient()),
		g:          &sdk.Gitops{Cli: mgr.GetClient()},
	}
	for _, opt := range opts {
		opt(r)
	}
	return r
}

// SetupWithManager sets up the controller with the Manager.
func (r *ClusterConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {
	if err := mgr.GetFieldIndexer().IndexField(context.Background(), &cloudv1.Cluster{}, ConfigRevisionRefField, func(o client.Object) []string {
		c := o.(*cloudv1.Cluster)
		if c.Spec.ConfigRevisionRef == nil {
			return nil
		}
		return []string{c.Spec.ConfigRevisionRef.Name}
	}); err != nil {
		return err
	}

	providerTypePredicate, err := predicate.LabelSelectorPredicate(
		metav1.LabelSelector{
			MatchExpressions: []metav1.LabelSelectorRequirement{
				{
					Key:      cloudv1.ProviderTypeLabel,
					Operator: metav1.LabelSelectorOpIn,
					Values:   []string{consts.ProviderTypeEKS, consts.ProviderTypeGKE},
				},
			},
		})
	if err != nil {
		return err
	}

	// enqueue all clusters and clusters that are related to ClusterConfig.
	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.Cluster{}, builder.WithPredicates(instanceTypeChangedPredicate{}, providerTypePredicate)).
		Watches(
			&cloudv1.ClusterConfig{},
			handler.EnqueueRequestsFromMapFunc(r.findClusterFromClusterConfig),
			builder.WithPredicates(predicate.And(predicate.GenerationChangedPredicate{}, omitConfigMetaChangePredicate{})),
		).
		Complete(r)
}

func (r *ClusterConfigReconciler) findClusterFromClusterConfig(ctx context.Context, clusterConfig client.Object) []reconcile.Request {
	cc, ok := clusterConfig.(*cloudv1.ClusterConfig)
	if !ok {
		r.Log.Error(nil, "object is not ClusterConfig", "object", clusterConfig)
		return []reconcile.Request{}
	}
	relatedClusters := &cloudv1.ClusterList{}
	if _, exist := cc.Spec.Selector.MatchLabels[cloudv1.TenantTagKey]; exist {
		// find matched clusters via tenant/cluster selector
		selector, err := metav1.LabelSelectorAsSelector(cc.Spec.Selector)
		if err != nil {
			r.Log.Error(err, "invalid label selector", "labelselector", cc.Spec.Selector)
			return []reconcile.Request{}
		}
		err = r.List(context.TODO(), relatedClusters, &client.ListOptions{
			LabelSelector: selector,
		})
		if err != nil {
			r.Log.Error(err, "failed to list clusters for clusterconfig %s", cc.Name)
			return []reconcile.Request{}
		}
	} else {
		// for base and instanceType clusterconfig, find all the clusters via version selector.
		selector := labels.SelectorFromSet(labels.Set(map[string]string{
			cloudv1.VersionLabel: cc.Labels[cloudv1.VersionLabel],
		}))

		crs, err := r.revisioner.ListConfigRevisions(selector)
		if err != nil {
			r.Log.Error(err, "failed to list config revisions for clusterconfig %s", cc.Name)
			return []reconcile.Request{}
		}

		for _, cr := range crs {
			cs := &cloudv1.ClusterList{}
			if err = r.Client.List(context.TODO(), cs, &client.ListOptions{
				FieldSelector: fields.OneTermEqualSelector(ConfigRevisionRefField, cr.Name),
			}); err != nil {
				continue
			}
			relatedClusters.Items = append(relatedClusters.Items, cs.Items...)
		}
	}

	requests := make([]reconcile.Request, len(relatedClusters.Items))
	for i, item := range relatedClusters.Items {
		requests[i] = reconcile.Request{
			NamespacedName: types.NamespacedName{
				Name:      item.GetName(),
				Namespace: item.GetNamespace(),
			},
		}
	}
	return requests
}

func logPlan(log logr.Logger, before, after interface{}, c *cloudv1.Cluster, dryrun bool, providerType string) {

	// filter v1/serveless cluster
	if providerType != consts.ProviderTypeEKS && providerType != consts.ProviderTypeGKE {
		return
	}

	if !dryrun && (c.Spec.ConfigRevisionRef == nil || c.Spec.ConfigRevisionRef.Name == "") {
		log.Error(fmt.Errorf("ClusterRevisionRef is Nil"), fmt.Sprintf("cluster %s, please check it immediately", c.Name))
	}

	mode := "execute"
	if dryrun {
		mode = "dryrun"
	}
	d := cmp.Diff(before, after)
	if d != "" {
		log.Info(fmt.Sprintf("UpdateClusterRevision: cluster %s, mode [%s]: topology: %+v\t diff: %s", c.Name, mode, c.Spec.Cluster, strings.ReplaceAll(d, "\n", "\r")))
	}
}

// analysisPlanAndRecordMetrics will record the affected components for dryrun clusters via metric.
// only consider image and config changes here,
// other changes that might result in component restart shall not be introduced by gitops.
func analysisPlanAndRecordMetrics(c *cloudv1.Cluster, old *cloudv1.ConfigRevision, new *cloudv1.ConfigRevision, dryrun bool, clusterId, providerType string) {
	// filter test cluster
	if _, err := strconv.ParseUint(clusterId, 10, 64); err != nil {
		return
	}

	// filter v1/serveless cluster
	if providerType != consts.ProviderTypeEKS && providerType != consts.ProviderTypeGKE {
		// clear all the metrics related to this clusterId
		metric.ConfigDryrunAffected.DeletePartialMatch(prometheus.Labels{"cluster_id": clusterId})
		return
	}

	if !dryrun {
		// clear all the metrics related to this clusterId
		metric.ConfigDryrunAffected.DeletePartialMatch(prometheus.Labels{"cluster_id": clusterId})
		metric.ConfigRevisionRefIsNil.DeletePartialMatch(prometheus.Labels{"cluster_id": clusterId})
		return
	}

	if c.Spec.ConfigRevisionRef == nil || c.Spec.ConfigRevisionRef.Name == "" {
		metric.ConfigRevisionRefIsNil.WithLabelValues(clusterId).Set(1)
	}

	record := func(clusterId string, comp string, reason string, val float64) {
		// avoid multiple reason present at the same time.
		metric.ConfigDryrunAffected.DeletePartialMatch(prometheus.Labels{"cluster_id": clusterId, "component": comp})
		metric.ConfigDryrunAffected.WithLabelValues(clusterId, comp, reason).Set(val)
	}

	if old.Spec.PD != nil && (old.Spec.PD.Image() != new.Spec.PD.Image()) {
		record(clusterId, v1alpha1.PDMemberType.String(), reasonImageUpdated, 1)
	} else if old.Spec.PD != nil && new.Spec.PD != nil && !reflect.DeepEqual(old.Spec.PD.Config, new.Spec.PD.Config) {
		record(clusterId, v1alpha1.PDMemberType.String(), reasonConfigUpdated, 1)
	} else {
		record(clusterId, v1alpha1.PDMemberType.String(), "", 0)
	}

	if old.Spec.TiDB != nil && (old.Spec.TiDB.Image() != new.Spec.TiDB.Image()) {
		record(clusterId, v1alpha1.TiDBMemberType.String(), reasonImageUpdated, 1)
	} else if old.Spec.TiDB != nil && new.Spec.TiDB != nil && !reflect.DeepEqual(old.Spec.TiDB.Config, new.Spec.TiDB.Config) {
		record(clusterId, v1alpha1.TiDBMemberType.String(), reasonConfigUpdated, 1)
	} else {
		record(clusterId, v1alpha1.TiDBMemberType.String(), "", 0)
	}

	if old.Spec.TiKV != nil && (old.Spec.TiKV.Image() != new.Spec.TiKV.Image()) {
		record(clusterId, v1alpha1.TiKVMemberType.String(), reasonImageUpdated, 1)
	} else if old.Spec.TiKV != nil && new.Spec.TiKV != nil && !reflect.DeepEqual(old.Spec.TiKV.Config, new.Spec.TiKV.Config) {
		record(clusterId, v1alpha1.TiKVMemberType.String(), reasonConfigUpdated, 1)
	} else {
		record(clusterId, v1alpha1.TiKVMemberType.String(), "", 0)
	}

	if old.Spec.TiProxy != nil && (old.Spec.TiProxy.Image() != new.Spec.TiProxy.Image()) {
		record(clusterId, v1alpha1.TiProxyMemberType.String(), reasonImageUpdated, 1)
	} else if old.Spec.TiProxy != nil && new.Spec.TiProxy != nil && !reflect.DeepEqual(old.Spec.TiProxy.Config, new.Spec.TiProxy.Config) {
		record(clusterId, v1alpha1.TiProxyMemberType.String(), reasonConfigUpdated, 1)
	} else {
		record(clusterId, v1alpha1.TiProxyMemberType.String(), "", 0)
	}

	if old.Spec.TiFlash != nil && (old.Spec.TiFlash.Image() != new.Spec.TiFlash.Image()) {
		record(clusterId, v1alpha1.TiFlashMemberType.String(), reasonImageUpdated, 1)
	} else if old.Spec.TiFlash != nil && new.Spec.TiFlash != nil && !reflect.DeepEqual(old.Spec.TiFlash.Config, new.Spec.TiFlash.Config) {
		record(clusterId, v1alpha1.TiFlashMemberType.String(), reasonConfigUpdated, 1)
	} else {
		record(clusterId, v1alpha1.TiFlashMemberType.String(), "", 0)
	}
}