package resourcemanager

import (
	"context"
	"strconv"
	"sync"

	apierrors "k8s.io/apimachinery/pkg/api/errors"

	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/errors"

	"github.com/tidbcloud/observability/pkg/common/utils"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	"k8s.io/client-go/kubernetes"
)

// dataAccountReaderWriterFactory manager different namespace generator
type dataAccountReaderWriterFactory struct {
	generators map[string]*dataAccountReaderWriter
	mutex      sync.RWMutex
	kubeClient kubernetes.Interface
	config     utils.Config
}

func NewdataAccountReaderWriterFactory(kubeClient kubernetes.Interface, config utils.Config) *dataAccountReaderWriterFactory {
	return &dataAccountReaderWriterFactory{
		generators: make(map[string]*dataAccountReaderWriter),
		kubeClient: kubeClient,
		config:     config,
	}
}

// GetReaderWriter get dataAccountReaderWriter for the given namespace.
func (f *dataAccountReaderWriterFactory) GetReaderWriter(namespace string) DataAccountReaderWriter {
	f.mutex.RLock()
	rw, exists := f.generators[namespace]
	f.mutex.RUnlock()

	if exists {
		return rw
	}

	f.mutex.Lock()
	defer f.mutex.Unlock()

	// double check
	if rw, exists := f.generators[namespace]; exists {
		return rw
	}

	// create a new dataAccountReaderWriter
	segmentMgr := &configMapSegmentMgr{
		name:       constants.DataAccountSegmentConfigMap,
		keyStartId: constants.DataAccountKeyStartID,
		retries:    minRetries,
		kubeClient: f.kubeClient,
		namespace:  namespace,
	}

	generator := utils.NewIdGenerator(f.config, segmentMgr)
	rw = &dataAccountReaderWriter{
		generator:  generator,
		kubeClient: f.kubeClient,
	}

	f.generators[namespace] = rw
	return rw
}

type dataAccountReaderWriter struct {
	generator  *utils.SegmentIdGenerator
	kubeClient kubernetes.Interface
}

func (d dataAccountReaderWriter) UniqueInt32ForCluster(ctx context.Context) (int32, error) {
	return d.generator.Next(ctx)
}

type configMapSegmentMgr struct {
	name       string
	keyStartId string
	retries    int
	kubeClient kubernetes.Interface
	namespace  string
}

const minSegmentStart = "1"
const minRetries = 10

func (p configMapSegmentMgr) GetSegment(ctx context.Context, step int32) (*utils.Segment, error) {
	// Optimistic Lock by k8s resource version
	for i := 0; i < min(minRetries, p.retries); i++ {
		config, err := p.kubeClient.CoreV1().
			ConfigMaps(p.namespace).
			Get(ctx, p.name, metav1.GetOptions{})

		if err != nil {
			// if the ConfigMap does not exist, create it
			if apierrors.IsNotFound(err) {
				createErr := p.createConfigMapIfNotExists(ctx)
				if createErr != nil {
					return nil, WrapClientError(createErr)
				}
				// retry to get the ConfigMap after creation
				continue
			}
			return nil, WrapClientError(err)
		}

		current := config.Data[p.keyStartId]
		if len(current) == 0 {
			current = minSegmentStart
		}

		start, err := strconv.ParseInt(current, 10, 32)
		if err != nil {
			return nil, errors.Wrap(errors.UnrecognizedError, "parse segment failed", err)
		}

		config.Data[p.keyStartId] = strconv.Itoa(int(int32(start) + step))

		conflict, err := p.updateConfigMapIgnoreVersionConflict(ctx, config)
		if err != nil {
			return nil, WrapClientError(err)
		}

		if conflict {
			continue
		} else {
			return utils.NewSegment(int32(start), step), nil
		}
	}

	return nil, errors.NewErrorf(errors.K8SClientError, "get segment timeout")
}

var isConflicted = func(err error) bool {
	return err != nil && apierrors.IsConflict(err)
}

func (p configMapSegmentMgr) updateConfigMapIgnoreVersionConflict(ctx context.Context, config *corev1.ConfigMap) (bool, error) {
	_, err := p.kubeClient.CoreV1().
		ConfigMaps(p.namespace).
		Update(ctx, config, metav1.UpdateOptions{})

	if isConflicted(err) {
		return true, nil
	}

	return false, err
}

// createConfigMapIfNotExists create ConfigMap if it does not exist.
func (p configMapSegmentMgr) createConfigMapIfNotExists(ctx context.Context) error {
	// this config in the reserved namespace, we should not create it
	// for observability, we create the config map in deploy regional service
	if p.namespace == constants.ObservabilityNamespace {
		return errors.NewErrorf(errors.K8SClientError, "namespace %s is reserved, skip create!", p.namespace)
	}
	configMap := &corev1.ConfigMap{
		ObjectMeta: metav1.ObjectMeta{
			Name:      p.name,
			Namespace: p.namespace,
		},
		Data: map[string]string{
			p.keyStartId: minSegmentStart,
		},
	}

	_, err := p.kubeClient.CoreV1().
		ConfigMaps(p.namespace).
		Create(ctx, configMap, metav1.CreateOptions{})

	// if the ConfigMap already exists, we ignore the error
	if apierrors.IsAlreadyExists(err) {
		return nil
	}

	return err
}