package revision

import (
	"context"
	"fmt"
	"hash/fnv"
	"reflect"
	"strconv"
	"strings"

	"time"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/pingcap/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/util"
)

type Interface interface {
	ListConfigRevisions(selector labels.Selector) ([]cloudv1.ConfigRevision, error)
	// CreateConfigRevision attempts to create the revision. If name collision occurs,
	// a unique identifier is added to the hash of the revision and renamed.
	// If the returned error is not nil, creation failed.
	// If the returned error is nil, the returned ConfigRevision has been created.
	CreateConfigRevision(revision *cloudv1.ConfigRevision) (*cloudv1.ConfigRevision, error)

	RenewConfigRevision(revision *cloudv1.ConfigRevision) error

	DeleteConfigRevision(revision *cloudv1.ConfigRevision) error
}

func NewRevisioner(cli client.Client) Interface {
	return &revisioner{
		client: cli,
	}
}

type revisioner struct {
	client client.Client
}

func (r *revisioner) ListConfigRevisions(selector labels.Selector) ([]cloudv1.ConfigRevision, error) {
	crs := &cloudv1.ConfigRevisionList{}
	err := r.client.List(context.TODO(), crs, &client.ListOptions{
		LabelSelector: selector,
	})
	if err != nil {
		return nil, errors.Annotatef(err, "list configRevisions with selector %s", selector.String())
	}
	return crs.Items, nil
}

func (r *revisioner) CreateConfigRevision(revision *cloudv1.ConfigRevision) (*cloudv1.ConfigRevision, error) {
	clone := revision.DeepCopy()
	// Continue to attempt to create the revision updating the name with a new hash on each iteration
	probe := int64(0)
	for {
		var hash string
		if probe > 0 {
			hash = HashConfigRevision(revision, &probe)
		} else {
			hash = HashConfigRevision(revision, nil)
		}
		index := strings.LastIndex(revision.Name, "-")
		clone.Name = revision.Name[0:index+1] + hash
		clone.Labels[cloudv1.ConfigRevisionHashLabel] = hash
		err := r.client.Create(context.TODO(), clone)
		if apierrors.IsAlreadyExists(err) {
			exists := &cloudv1.ConfigRevision{}
			err := r.client.Get(context.TODO(), client.ObjectKey{Name: clone.Name}, exists)
			if err != nil {
				return nil, err
			}
			if reflect.DeepEqual(exists.Spec, clone.Spec) {
				return exists, nil
			}
			probe++
			continue
		}
		return clone, err
	}
}

func (r *revisioner) RenewConfigRevision(revision *cloudv1.ConfigRevision) error {
	patch := client.MergeFrom(revision.DeepCopy())
	if revision.Annotations == nil {
		revision.Annotations = make(map[string]string)
	}
	revision.Annotations[cloudv1.ConfigRevisionActiveAnnotation] = time.Now().Format(time.RFC3339)
	return r.client.Patch(context.TODO(), revision, patch)
}

func (r *revisioner) DeleteConfigRevision(revision *cloudv1.ConfigRevision) error {
	return r.client.Delete(context.TODO(), revision)
}

// NewConfigRevision will take a snapshot of ClusterConfig as ConfigRevision
// and set its name to the concatenation of layers tuple and hash.
func NewConfigRevision(cc *cloudv1.ClusterConfig) *cloudv1.ConfigRevision {
	clone := cc.DeepCopy()
	cr := &cloudv1.ConfigRevision{
		ObjectMeta: metav1.ObjectMeta{
			Labels: clone.Labels,
		},
		Spec: cloudv1.ConfigRevisionSpec{
			Version:        clone.Spec.Version,
			TiDB:           clone.Spec.TiDB,
			PD:             clone.Spec.PD,
			TiKV:           clone.Spec.TiKV,
			TiFlash:        clone.Spec.TiFlash,
			TiFlashLearner: clone.Spec.TiFlashLearner,
			Backup:         clone.Spec.Backup,
			Restore:        clone.Spec.Restore,
			Lightning:      clone.Spec.Lightning,
			TiCDC:          clone.Spec.TiCDC,
		},
	}
	if cr.GetLabels() == nil {
		cr.ObjectMeta.Labels = make(map[string]string)
	}
	hash := HashConfigRevision(cr, nil)
	prefix := "base" + cc.Labels[cloudv1.VersionLabel]
	if t, ok := cc.Labels[cloudv1.TenantLabel]; ok {
		prefix = fmt.Sprintf("%s-t%s", prefix, t)
	}
	if c, ok := cc.Labels[cloudv1.ClusterLabel]; ok {
		prefix = fmt.Sprintf("%s-c%s", prefix, c)
	}
	cr.Name = fmt.Sprintf("%s-%s", prefix, hash)
	cr.Labels[cloudv1.ConfigRevisionHashLabel] = hash
	if cr.Annotations == nil {
		cr.Annotations = make(map[string]string)
	}
	cr.Annotations[cloudv1.ConfigRevisionActiveAnnotation] = time.Now().Format(time.RFC3339)
	return cr
}

func HashConfigRevision(revision *cloudv1.ConfigRevision, salt *int64) string {
	hf := fnv.New32()
	util.DeepHashObject(hf, revision.Spec)

	if salt != nil {
		hf.Write([]byte(strconv.FormatInt(*salt, 10)))
	}
	return fmt.Sprintf("%x", hf.Sum32())
}

func EqualRevision(lhs *cloudv1.ConfigRevision, rhs *cloudv1.ConfigRevision) bool {
	var lhsHash, rhsHash *uint32
	if lhs == nil || rhs == nil {
		return lhs == rhs
	}
	if hs, found := lhs.Labels[cloudv1.ConfigRevisionHashLabel]; found {
		hash, err := strconv.ParseInt(hs, 10, 32)
		if err == nil {
			lhsHash = new(uint32)
			*lhsHash = uint32(hash)
		}
	}
	if hs, found := rhs.Labels[cloudv1.ConfigRevisionHashLabel]; found {
		hash, err := strconv.ParseInt(hs, 10, 32)
		if err == nil {
			rhsHash = new(uint32)
			*rhsHash = uint32(hash)
		}
	}
	if lhsHash != nil && rhsHash != nil && *lhsHash != *rhsHash {
		return false
	}
	return reflect.DeepEqual(lhs.Spec, rhs.Spec)
}