/*


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 controllers

import (
	"context"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/tidbcloud/config-service/pkg/clinic_event"
	corev1 "k8s.io/api/core/v1"

	"github.com/go-logr/logr"
	"github.com/hashicorp/go-multierror"
	"github.com/pingcap/log"
	pingcapv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pkg/errors"
	"github.com/tidbcloud/config-service/api/v1alpha1"
	"github.com/tidbcloud/config-service/pkg/cluster"
	"github.com/tidbcloud/config-service/pkg/configeval"
	"github.com/tidbcloud/config-service/pkg/tenant"
	"github.com/tidbcloud/config-service/pkg/utils"
	"github.com/tidbcloud/config-service/pkg/worker"
	"github.com/tidbcloud/config-service/sdk/api"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/manager"
)

// RecommendedEvictLeaderTimeout is set according to
// https://docs.pingcap.com/tidb-in-kubernetes/dev/configure-a-tidb-cluster#configure-graceful-upgrade-for-tikv-cluster
const RecommendedEvictLeaderTimeout = "10000m"

// ClusterConfigController reconciles the dbaas cluster configs
// NOTE: since the target of this controller is dbaas cluster, which is not a Kubernetes resource, so this
// controller will fetch clusters from an interface to reconcile and it is not a typical kube-builder controller.
type ClusterConfigController struct {
	infraApiCli client.Client
	client.Client
	Clusters                cluster.Lister
	Tenants                 tenant.Lister
	Log                     logr.Logger
	Scheme                  *runtime.Scheme
	skipEvalNotFoundCluster bool

	workers sync.Map
	lck     sync.Mutex

	activityHub bool
}

func NewClusterConfigController(mgr manager.Manager, infraAPiCli client.Client, logger logr.Logger, cl cluster.Lister, tl tenant.Lister, skipEval, activityHub bool) *ClusterConfigController {
	return &ClusterConfigController{
		infraApiCli:             infraAPiCli,
		Client:                  mgr.GetClient(),
		Clusters:                cl,
		Tenants:                 tl,
		Log:                     logger,
		Scheme:                  mgr.GetScheme(),
		skipEvalNotFoundCluster: skipEval,
		activityHub:             activityHub,
	}
}

func (c *ClusterConfigController) Start(ctx context.Context) error {
	w := c.getAliveClusterWatcher()
	w.Start()
	<-ctx.Done()
	w.Stop()
	return nil
}

func (c *ClusterConfigController) NeedLeaderElection() bool {
	return true
}

func (c *ClusterConfigController) getTenantWatcher() *worker.PeriodWorker {
	return worker.NewPeriodWorker(c.Log, time.Minute*20, func() error {
		tenants, err := c.Tenants.List()
		if err != nil {
			return err
		}

		newTenants := make([]string, 0)
		obsoleteTenants := make(map[string]bool)

		c.workers.Range(func(k, v interface{}) bool {
			if kstr, ok := k.(string); ok {
				// assume all tenants are obsolete first
				obsoleteTenants[kstr] = true
			} else {
				// impossible path, but we'd better be defensive
				c.Log.Error(errors.New("encounter non string key in worker map"), "", "key", k)
			}
			return true
		})
		for _, t := range tenants {
			// only reconcile active tenants
			if strings.Contains(t.Status, "active") {
				tenantId := strconv.FormatUint(t.ID, 10)
				// the tenants still exist in central is not obsolete
				delete(obsoleteTenants, tenantId)
				_, ok := c.workers.Load(tenantId)
				if !ok {
					newTenants = append(newTenants, tenantId)
				}
			}
		}
		for idx, id := range newTenants {
			c.startWorkerForTenant(id, time.Duration(idx%600)*time.Second)
		}
		for id := range obsoleteTenants {
			err := c.stopWorkerForTenant(id)
			if err != nil {
				c.Log.Error(err, "err stop worker")
			}
		}
		return nil
	})
}

// getAliveClusterWatcher query core-server to get alive non-devtier dedicated cluster
// and do reconcile on these alive cluster directly.
func (c *ClusterConfigController) getAliveClusterWatcher() *worker.PeriodWorker {
	return worker.NewPeriodWorker(c.Log, time.Minute*2, func() error {
		// for now we just use one goroutine to process reconcileCluster since there at most 50+ active cluster need to processed by config-server
		// may fan out to a group of goroutine if we cannot process workload in 2 minutes.
		var cnt = 0
		defer func(startTs time.Time) {
			// log duration for debug
			log.Debug("process alive clusters done", zap.Int("count", cnt), zap.Duration("duration", time.Now().Sub(startTs)))
		}(time.Now())
		clusters, err := c.Clusters.ListAliveClusters()
		if err != nil {
			return err
		}
		var errs *multierror.Error
		cnt = len(clusters)
		for i := range clusters {
			cs := clusters[i]
			if rErr := c.reconcileCluster(cs); rErr != nil {
				errs = multierror.Append(errs, rErr)
				continue
			}
		}
		return errs.ErrorOrNil()
	})
}

func (c *ClusterConfigController) reconcile(tenantID string) error {
	log.Debug("reconcile", zap.String("tenant id", tenantID))

	clusters, err := c.Clusters.List(tenantID)
	if err != nil {
		return err
	}
	var errs *multierror.Error
	if len(clusters) < 1 {
		// if the tenant does not have any cluster yet, don't sync it till next tenant sync
		if err := c.stopWorkerForTenant(tenantID); err != nil {
			return err
		}
		return nil
	}
	for i := range clusters {
		cs := clusters[i]
		// maybe parallelize?
		if err := c.reconcileCluster(cs); err != nil {
			errs = multierror.Append(errs, err)
			continue
		}
	}
	return errs.ErrorOrNil()
}

func (c *ClusterConfigController) reconcileCluster(cs cluster.Cluster) error {
	cMeta := cs.GetMeta()
	log.Debug("cluster reconcile",
		zap.String("tenant id", cMeta.TenantID),
		zap.String("cluster id", cMeta.ClusterID))

	if !strings.Contains(cMeta.Status, "NORMAL") {
		log.Debug("skip reconciling cluster",
			zap.String("tenant id", cMeta.TenantID),
			zap.String("cluster id", cMeta.ClusterID),
			zap.String("cluster status", string(cMeta.Status)))
		return nil
	}
	tc, err := cs.GetResource()
	if err != nil {
		return errors.Wrapf(err, "get resource, tenant id %s cluster id %s", cMeta.TenantID, cMeta.ClusterID)
	}
	// TODO: check whether we can perform update (aylei)
	clusterConfig, err := c.findClusterConfig(cMeta.TenantID, cMeta.ClusterID)
	if err != nil {
		return errors.Wrapf(err, "find cluster config, tenant id %s, cluster id %s", cMeta.TenantID, cMeta.ClusterID)
	}

	cls, err := configeval.FindCluster(c.infraApiCli, cMeta.TenantID, cMeta.ClusterID)
	if err != nil {
		if c.skipEvalNotFoundCluster {
			err = configeval.RemoveExpClusterConifg(clusterConfig, cls)
			if err != nil {
				return errors.Wrapf(err, "remove expression from clusterconfig, tenant id %s, cluster id %s", cMeta.TenantID, cMeta.ClusterID)
			}
		} else {
			return errors.Wrapf(err, "find cluster, tenant id %s, cluster id %s", cMeta.TenantID, cMeta.ClusterID)
		}
	} else {
		err = configeval.EvalClusterConfig(c.infraApiCli, clusterConfig, cls)
		if err != nil {
			return errors.Wrapf(err, "eval clusterconfig, tenant id %s, cluster id %s", cMeta.TenantID, cMeta.ClusterID)
		}
	}

	pcc := clusterConfig.GetPDClusterConfig(cMeta.MachineTypeOf(api.ComponentPD))
	pdUpdate, err := cs.UpdatePDClusterConfig(pcc, clusterConfig.IsDryrun())
	if err != nil {
		return errors.Wrapf(err, "update pd config, tenantId %s, clusterId %s", cMeta.TenantID, cMeta.ClusterID)
	}
	if pdUpdate && c.activityHub {
		// Debug event
		err = cs.SendEvent(tc, tc, cluster.PDConfigType)
		if err != nil {
			c.Log.Error(err, "send update pd config event failed")
		}
	}

	tc, err = cs.GetResource()
	if err != nil {
		return errors.Wrapf(err, "get resource, tenant id %s cluster id %s", cMeta.TenantID, cMeta.ClusterID)
	}

	oldTc := tc.DeepCopy()
	_, update, err := cs.UpdateResourceConfig(tc, clusterConfig, func() error {
		return patchTiDBCluster(tc, clusterConfig, cMeta)
	})
	if err != nil {
		return errors.Wrapf(err, "update resource config, tenantId %s, clusterId %s", cMeta.TenantID, cMeta.ClusterID)
	}

	if update && c.activityHub {
		err = cs.SendEvent(oldTc, tc, cluster.ClusterConfigType)
		if err != nil {
			c.Log.Error(err, "send update cluster config event failed")
		}
	}

	return nil
}

func patchTiDBCluster(tc *pingcapv1alpha1.TidbCluster, cc *v1alpha1.ClusterConfig, cMeta *cluster.ClusterMeta) error {
	tc.Spec.Version = cc.Spec.Version
	if tc.Spec.TiDB != nil {
		desired := cc.GetTiDBSpec(cMeta.MachineTypeOf(api.ComponentTiDB))
		tc.Spec.TiDB.Config = computeTiDBConfig(tc.Spec.TiDB.Config, desired.Config)
		tc.Spec.TiDB.Version = desired.Tag
		tc.Spec.TiDB.BaseImage = desired.BaseImage
		// append GOMEMLIMIT to tidb ENV only if
		// 1. cc has annotation clinic.pingcap.com/enable-gomemlimit: "true"
		// 2. resource mem limit is already set on tc.
		if cc.Annotations[clinic_event.ClinicEnableGOMEMLIMIT] == "true" && tc.Spec.TiDB.ResourceRequirements.Limits.Memory().Value() > 0 {
			limit := strconv.FormatInt(int64(0.8*float64(tc.Spec.TiDB.ResourceRequirements.Limits.Memory().Value())), 10)
			needAppend := true
			for i := range tc.Spec.TiDB.Env {
				if tc.Spec.TiDB.Env[i].Name == "GOMEMLIMIT" {
					tc.Spec.TiDB.Env[i].Value = limit
					needAppend = false
					break
				}
			}
			if needAppend {
				tc.Spec.TiDB.Env = append(tc.Spec.TiDB.Env, corev1.EnvVar{
					Name:  "GOMEMLIMIT",
					Value: limit,
				})
			}
		}
	}
	if tc.Spec.TiKV != nil {
		desired := cc.GetTiKVSpec(cMeta.MachineTypeOf(api.ComponentTiKV))
		tc.Spec.TiKV.Config = computeTiKVConfig(tc.Spec.TiKV.Config, desired.Config)
		tc.Spec.TiKV.Version = desired.Tag
		tc.Spec.TiKV.BaseImage = desired.BaseImage

		// Turn up the `evictLeaderTimeout`.
		// The default is 10 minutes, which may cause an error.
		// See more details in https://jira.tidbcloud.com/browse/DM-1051
		// Note: this modification will not make TiKV restart.
		evictLeaderTimeout := RecommendedEvictLeaderTimeout
		tc.Spec.TiKV.EvictLeaderTimeout = &evictLeaderTimeout
	}
	if tc.Spec.PD != nil {
		desired := cc.GetPDSpec(cMeta.MachineTypeOf(api.ComponentPD))
		tc.Spec.PD.Config = computePDConfig(tc.Spec.PD.Config,
			cc.GetPDInstanceConfig(cMeta.MachineTypeOf(api.ComponentPD)))
		tc.Spec.PD.BaseImage = desired.BaseImage
		tc.Spec.PD.Version = desired.Tag
	}
	if tc.Spec.TiFlash != nil {
		desired := cc.GetTiFlashSpec(cMeta.MachineTypeOf(api.ComponentTiFlash))
		tc.Spec.TiFlash.Config.Common = computeTiFlashConfig(tc.Spec.TiFlash.Config.Common, desired.Config)
		tc.Spec.TiFlash.Version = desired.Tag
		tc.Spec.TiFlash.BaseImage = desired.BaseImage
		if cc.Spec.TiFlashLearner != nil {
			desiredLearner := cc.GetTiFlashLearnerSpec(cMeta.MachineTypeOf(api.ComponentTiFlashLearner))
			tc.Spec.TiFlash.Config.Proxy = computeTiFlashLearnerConfig(tc.Spec.TiFlash.Config.Proxy, desiredLearner.Config)
		}
	}
	return nil
}

func (c *ClusterConfigController) startWorkerForTenant(tenantID string, delayDur time.Duration) {
	c.lck.Lock()
	defer c.lck.Unlock()

	c.Log.Info("starting tenant worker", "id", tenantID)
	w := worker.NewPeriodWorkerWithDelay(c.Log, delayDur, time.Minute*2, func() error {
		return c.reconcile(tenantID)
	})
	c.workers.Store(tenantID, w)
	w.Start()
}

func (c *ClusterConfigController) stopWorkerForTenant(tenantID string) error {
	c.lck.Lock()
	defer c.lck.Unlock()

	c.Log.Info("stopping tenant worker", "tenantID", tenantID)
	w, ok := c.workers.Load(tenantID)
	if ok {
		c.workers.Delete(tenantID)
		go func() {
			w.(worker.Worker).Stop()
			c.Log.Info("background tenant is stopped", "tenantID", tenantID)
		}()
		return nil
	}
	return errors.Errorf("tenant %s worker is not found", tenantID)
}

// findClusterConfig find cluster config cr by tenantID and clusterID.
func (c *ClusterConfigController) findClusterConfig(tenantID, clusterID string) (*v1alpha1.ClusterConfig, error) {
	return utils.FindAndValidateClusterConfig(c.Client, tenantID, clusterID)
}