package client

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"reflect"
	"strconv"
	"sync"
	"time"

	"github.com/go-logr/logr"
	prometheusv1 "github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring/v1"
	v1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/pkgv2/gormutil"
	"gorm.io/gorm"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/meta"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/selection"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
)

const (
	DefaultEventTTL = 10 * time.Minute
	// Component name for event source tracking
	ComponentName = "mysql-client"
)

var (
	// globalDB is a singleton gorm.DB instance for MySQL.
	globalDB *gorm.DB
	mu       = sync.Mutex{} // protect globalDB initialization
)

// paginationCursor represents the state for pagination
type paginationCursor struct {
	LastClusterID string `json:"last_cluster_id"`
}

type mysqlClient struct {
	db  *gorm.DB
	log logr.Logger

	componentName string
	eventTTL      time.Duration
}

var _ Client = (*mysqlClient)(nil)

// NewMySQLClient creates a new DBClient instance with the provided gorm.DB connection.
func NewMySQLClient(connCfg *gormutil.Config, logger logr.Logger, componentName string, eventTTL time.Duration) (Client, error) {
	db, err := NewGormDB(connCfg)
	if err != nil {
		return nil, fmt.Errorf("failed to create gorm DB: %w", err)
	}

	if eventTTL == 0 {
		eventTTL = DefaultEventTTL
	}

	return &mysqlClient{
		db:            db,
		log:           logger,
		componentName: componentName,
		eventTTL:      eventTTL,
	}, nil
}

// NewGormDB initializes a global gorm.DB instance using the provided configuration.
func NewGormDB(cfg *gormutil.Config) (*gorm.DB, error) {
	mu.Lock()
	defer mu.Unlock()
	if globalDB != nil {
		return globalDB, nil
	}
	db, err := gormutil.Connect(*cfg)
	if err != nil {
		return nil, fmt.Errorf("failed to connect to MySQL: %w", err)
	}
	sqlDB, err := db.DB()
	if err != nil {
		return nil, fmt.Errorf("failed to get sql.DB from gorm.DB: %w", err)
	}
	if err := sqlDB.Ping(); err != nil {
		return nil, fmt.Errorf("failed to ping database: %w", err)
	}

	globalDB = db
	return db, nil
}

func (c *mysqlClient) RawClient() *gorm.DB {
	return c.db
}

// Get retrieves a cluster from MySQL and converts it to a K8s object
func (c *mysqlClient) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) (err error) {
	start := time.Now()
	defer func() {
		recordOperationDuration("get", start, err)
	}()

	if err = checkObject(obj); err != nil {
		return err
	}
	clusterID := key.Name
	if clusterID == "" {
		return ErrObjkeyInvalid
	}
	log := c.log.WithName("[Get]").WithValues("clusterID", clusterID)

	return c.withTx(ctx, func(tx *gorm.DB) error {
		cluster := &Cluster{ClusterID: clusterID}
		if err := tx.Preload("Spec").Preload("Status").First(&cluster).Error; err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return NewNotFound(clusterID)
			}
			return NewInternalError(fmt.Errorf("failed to get cluster: %w", err))
		}
		log.Info("get cluster successfully")
		// Convert database cluster to K8s object
		return c.convertClusterToK8s(cluster, obj)
	})
}

// List retrieves a list of clusters from MySQL and converts them to K8s objects
func (c *mysqlClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) (err error) {
	start := time.Now()
	defer func() {
		recordOperationDuration("list", start, err)
	}()

	var clusters []Cluster
	query := c.db.WithContext(ctx).Model(&Cluster{}).Preload("Spec").Preload("Status")

	// Apply list options
	listOpts := client.ListOptions{}
	for _, opt := range opts {
		opt.ApplyToList(&listOpts)
	}

	// Apply namespace filter
	if listOpts.Namespace != "" {
		query = query.Where("namespace = ?", listOpts.Namespace)
	}

	// Apply label selector
	if listOpts.LabelSelector != nil {
		query = c.applyLabelSelector(query, listOpts.LabelSelector)
	}

	// Apply continue token for pagination
	if listOpts.Continue != "" {
		cursor, err := c.decodeContinueToken(listOpts.Continue)
		if err != nil {
			return NewBadRequest(fmt.Sprintf("invalid continue token: %v", err))
		}
		// Use cursor to filter results after the last seen record
		// Order by cluster_id for consistent pagination
		query = query.Where("clusters.cluster_id > ?", cursor.LastClusterID)
	}

	// Always order by cluster_id for consistent pagination
	query = query.Order("clusters.cluster_id ASC")

	// Apply limit (fetch one extra to determine if there are more results)
	if listOpts.Limit > 0 {
		query = query.Limit(int(listOpts.Limit + 1))
	}

	if err = query.Find(&clusters).Error; err != nil {
		return NewInternalError(fmt.Errorf("failed to list clusters: %w", err))
	}

	// Check if there are more results and generate continue token
	continueToken, err := c.generateContinueToken(clusters, listOpts.Limit)
	if err != nil {
		return NewInternalError(fmt.Errorf("failed to generate continue token: %w", err))
	}

	// Trim results to the requested limit
	if listOpts.Limit > 0 && int64(len(clusters)) > listOpts.Limit {
		clusters = clusters[:listOpts.Limit]
	}

	// Convert database clusters to K8s objects
	if err = c.convertClusterListToK8s(clusters, list); err != nil {
		return err
	}

	// Set continue token in the list metadata if there are more results
	if continueToken != "" {
		if k8sList, ok := list.(*v1.ClusterList); ok {
			k8sList.Continue = continueToken
		}
	}

	return nil
}

// Create creates a new cluster in MySQL and creates a corresponding event
func (c *mysqlClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) (err error) {
	start := time.Now()
	defer func() {
		recordOperationDuration("create", start, err)
	}()

	if err = checkObject(obj); err != nil {
		return err
	}
	// Convert K8s object to cluster records
	cluster, spec, status, err := c.convertK8sToCluster(obj, true)
	if err != nil {
		return fmt.Errorf("failed to convert K8s object to cluster: %w", err)
	}

	// Set initial resource version
	status.ResourceVersion = 1
	cluster.Spec = *spec
	cluster.Status = *status

	log := c.logger(ctx).WithName("[Create]").WithValues("clusterID", cluster.ClusterID)

	err = c.withTx(ctx, func(tx *gorm.DB) error {
		// Check if cluster already exists
		var existingCount int64
		if err := tx.Model(&Cluster{}).Where("cluster_id = ?", cluster.ClusterID).Count(&existingCount).Error; err != nil {
			return NewInternalError(fmt.Errorf("failed to check existing cluster: %w", err))
		}
		if existingCount > 0 {
			return NewAlreadyExists(cluster.ClusterID)
		}

		// Insert cluster record
		if err := tx.Create(cluster).Error; err != nil {
			if gormutil.IsDuplicatedError(err) {
				return NewAlreadyExists(cluster.ClusterID)
			}
			return NewInternalError(fmt.Errorf("failed to create cluster with spec and status: %w", err))
		}

		// Create cluster event for watch functionality
		event := &ClusterEvent{
			Namespace:       cluster.Namespace,
			Name:            cluster.ClusterID,
			EventType:       EventTypeClusterCreated,
			Status:          EventStatusPending,
			SourceComponent: c.componentName,
			ResourceVersion: cluster.Status.ResourceVersion,
		}

		if err := tx.Create(event).Error; err != nil {
			return NewInternalError(fmt.Errorf("failed to create cluster event: %w", err))
		}

		log.Info("created cluster and related records successfully",
			"clusterID", cluster.ClusterID,
			"eventID", event.EventID)

		return nil
	})

	if err == nil {
		// Update the object with the new resource version
		obj.SetResourceVersion(strconv.FormatUint(status.ResourceVersion, 10))
	}
	return err
}

// Delete deletes a cluster from MySQL and creates a corresponding event
func (c *mysqlClient) Delete(ctx context.Context, obj client.Object, opts ...client.DeleteOption) (err error) {
	start := time.Now()
	defer func() {
		recordOperationDuration("delete", start, err)
	}()

	if err = checkObject(obj); err != nil {
		return err
	}

	// Apply delete options
	deleteOpts := client.DeleteOptions{}
	for _, opt := range opts {
		opt.ApplyToDelete(&deleteOpts)
	}

	clusterID, namespace := obj.GetName(), obj.GetNamespace()
	if clusterID == "" {
		return ErrObjkeyInvalid
	}
	log := c.logger(ctx).WithName("[Delete]").WithValues("clusterID", clusterID, "Options", deleteOpts)

	err = c.withTx(ctx, func(tx *gorm.DB) error {
		// Get the cluster to check if it exists
		var cluster Cluster
		if err := tx.Where("cluster_id = ? ", clusterID).First(&cluster).Error; err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return NewNotFound(clusterID)
			}
			return NewInternalError(fmt.Errorf("failed to get cluster: %w", err))
		}

		// Get status for resource version
		var status ClusterStatus
		if err := tx.Where("cluster_id = ?", clusterID).First(&status).Error; err != nil {
			return NewInternalError(fmt.Errorf("failed to get cluster status: %w", err))
		}

		// Increment resource version for deletion
		oldVersion := status.ResourceVersion
		newVersion := oldVersion + 1
		if isForceDelete(&deleteOpts) {
			if cluster.DeletionTimestamp == nil {
				return ErrDeleteCond
			}

			log.Info("force delete cluster records")

			// With ON DELETE CASCADE, we only need to delete the main cluster record.
			if err := tx.Delete(&Cluster{ClusterID: clusterID}).Error; err != nil {
				return NewInternalError(fmt.Errorf("failed to delete cluster: %w", err))
			}

			// For force delete, the event is terminal. Set `Status` to `StatusDone` directly and add a TTL for cleanup.
			ttlExpiration := time.Now().Add(c.eventTTL)
			event := &ClusterEvent{
				Namespace:       namespace,
				Name:            clusterID,
				EventType:       EventTypeClusterDeleted,
				Status:          EventStatusDone,
				SourceComponent: c.componentName,
				ResourceVersion: newVersion,
				TTLExpiration:   &ttlExpiration,
			}
			if err := tx.Create(event).Error; err != nil {
				return NewInternalError(fmt.Errorf("failed to create cluster event: %w", err))
			}

			log.Info("successfully deleted cluster records")
			return nil
		}

		if cluster.DeletionTimestamp != nil {
			log.Info("found DeletionTimestamp, skip delete", "deletion_timestamp", cluster.DeletionTimestamp.Format(time.RFC3339))
			return nil
		}

		// Soft delete cluster (set deletion timestamp)
		now := time.Now()
		if err := tx.Model(&Cluster{}).Where("cluster_id = ?", clusterID).Update("deletion_timestamp", now).Error; err != nil {
			return NewInternalError(fmt.Errorf("failed to delete cluster: %w", err))
		}

		result := tx.Model(&ClusterStatus{}).Where("cluster_id = ? AND resource_version = ?", clusterID, oldVersion).Update("resource_version", newVersion)
		if result.Error != nil {
			return NewInternalError(fmt.Errorf("failed to update resource version: %w", result.Error))
		}
		if result.RowsAffected == 0 {
			return NewConflict(clusterID, fmt.Errorf("resource version mismatch"))
		}

		// `EventType` is EventTypeClusterUpdated, not EventTypeClusterDeleted, because this is a soft delete.
		event := &ClusterEvent{
			Namespace:       namespace,
			Name:            clusterID,
			EventType:       EventTypeClusterDeleted,
			Status:          EventStatusPending,
			SourceComponent: c.componentName,
			ResourceVersion: newVersion,
		}
		if err := tx.Create(event).Error; err != nil {
			return NewInternalError(fmt.Errorf("failed to create cluster event: %w", err))
		}

		log.Info("successfully updated cluster records", "eventID", event.EventID)
		return nil
	})
	return err
}

// Update updates an existing cluster in MySQL and creates a corresponding event
func (c *mysqlClient) Update(ctx context.Context, obj client.Object, opts ...client.UpdateOption) (err error) {
	start := time.Now()
	defer func() {
		recordOperationDuration("update", start, err)
	}()

	clusterID := obj.GetName()
	if clusterID == "" {
		return ErrObjkeyInvalid
	}
	namespace := obj.GetNamespace()
	log := c.logger(ctx).WithName("[Update]").WithValues("clusterID", clusterID)

	// Convert K8s object to cluster records
	cluster, spec, status, err := c.convertK8sToCluster(obj, false)
	if err != nil {
		return fmt.Errorf("failed to convert K8s object to cluster: %w", err)
	}

	// Increment resource version
	oldVersion := status.ResourceVersion
	status.ResourceVersion = oldVersion + 1

	err = c.withTx(ctx, func(tx *gorm.DB) error {
		var currentStatus ClusterStatus
		if err := tx.Where("cluster_id = ?", clusterID).First(&currentStatus).Error; err != nil {
			if errors.Is(err, gorm.ErrRecordNotFound) {
				return NewNotFound(clusterID)
			}
			return NewInternalError(fmt.Errorf("failed to get current cluster status: %w", err))
		}
		if currentStatus.ResourceVersion != oldVersion {
			return NewConflict(clusterID, fmt.Errorf("resource version mismatch, got %d, expected %d", oldVersion, currentStatus.ResourceVersion))
		}

		// Update cluster record
		if err := tx.Model(&Cluster{}).Where("cluster_id = ?", clusterID).
			Omit("creation_timestamp", "update_timestamp").
			Select("*").Updates(cluster).Error; err != nil {
			return NewInternalError(fmt.Errorf("failed to update cluster: %w", err))
		}

		// Update spec record
		if err := tx.Model(&ClusterSpec{}).Where("cluster_id = ?", clusterID).
			Omit("creation_timestamp", "update_timestamp").
			Select("*").Updates(spec).Error; err != nil {
			return NewInternalError(fmt.Errorf("failed to update cluster spec: %w", err))
		}

		// Update status record with version check
		result := tx.Model(&ClusterStatus{}).Where("cluster_id = ? AND resource_version = ?", clusterID, oldVersion).
			Omit("creation_timestamp", "update_timestamp").
			Select("*").Updates(status)

		if result.Error != nil {
			return NewInternalError(fmt.Errorf("failed to update cluster status: %w", result.Error))
		}
		if result.RowsAffected == 0 {
			return NewConflict(clusterID, fmt.Errorf("resource version mismatch"))
		}

		// Create cluster event
		event := &ClusterEvent{
			Namespace:       namespace,
			Name:            clusterID,
			EventType:       EventTypeClusterUpdated,
			Status:          EventStatusPending,
			SourceComponent: c.componentName,
			ResourceVersion: status.ResourceVersion,
		}

		if err := tx.Create(event).Error; err != nil {
			return NewInternalError(fmt.Errorf("failed to create cluster event: %w", err))
		}

		log.Info("cluster updated successfully",
			"eventID", event.EventID,
			"resourceVersion", status.ResourceVersion)
		return nil
	})

	if err == nil {
		// Update the object with the new resource version
		obj.SetResourceVersion(strconv.FormatUint(status.ResourceVersion, 10))
	}
	return err
}

// Patch patches an existing object in MySQL, be cautious to use it since it's not well tested.
func (c *mysqlClient) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.PatchOption) (err error) {
	start := time.Now()
	defer func() {
		recordOperationDuration("patch", start, err)
	}()

	// Get the current object
	key := client.ObjectKey{Name: obj.GetName(), Namespace: obj.GetNamespace()}
	currentObj := obj.DeepCopyObject().(client.Object)

	if err = c.Get(ctx, key, currentObj); err != nil {
		return err
	}

	// Get patch data
	patchData, err := patch.Data(currentObj)
	if err != nil {
		return NewBadRequest(fmt.Sprintf("failed to get patch data: %v", err))
	}

	// Apply patch based on patch type
	switch patch.Type() {
	case types.MergePatchType:
		// For merge patch, merge the patch data with the current object
		// Marshal current object to JSON
		currentJSON, err := json.Marshal(currentObj)
		if err != nil {
			return NewInternalError(fmt.Errorf("failed to marshal current object: %w", err))
		}

		// Merge patch data
		merged, err := jsonMergePatch(currentJSON, patchData)
		if err != nil {
			return NewBadRequest(fmt.Sprintf("failed to merge patch: %v", err))
		}

		// Unmarshal back to object
		if err = json.Unmarshal(merged, currentObj); err != nil {
			return NewBadRequest(fmt.Sprintf("failed to unmarshal merged object: %v", err))
		}
	case types.JSONPatchType, types.StrategicMergePatchType:
		// For other patch types, use a simplified approach
		if err = json.Unmarshal(patchData, currentObj); err != nil {
			return NewBadRequest(fmt.Sprintf("failed to apply patch: %v", err))
		}
	default:
		return NewBadRequest(fmt.Sprintf("unsupported patch type: %s", patch.Type()))
	}

	// Update the patched object
	return c.Update(ctx, currentObj)
}

// Apply is not supported by the MySQL-backed client because it does not speak
// server-side apply semantics.
func (c *mysqlClient) Apply(ctx context.Context, obj runtime.ApplyConfiguration, opts ...client.ApplyOption) error {
	return fmt.Errorf("apply is not implemented for mysql client")
}

// jsonMergePatch performs a simple JSON merge patch
func jsonMergePatch(original, patch []byte) ([]byte, error) {
	var originalMap, patchMap map[string]interface{}

	if err := json.Unmarshal(original, &originalMap); err != nil {
		return nil, err
	}
	if err := json.Unmarshal(patch, &patchMap); err != nil {
		return nil, err
	}

	// Merge patch into original
	for k, v := range patchMap {
		if v == nil {
			delete(originalMap, k)
		} else {
			originalMap[k] = v
		}
	}

	return json.Marshal(originalMap)
}

// Note: DeleteAllOf is not implemented
func (c *mysqlClient) DeleteAllOf(ctx context.Context, obj client.Object, opts ...client.DeleteAllOfOption) error {
	panic("not implemented")
}

// Status returns a status sub-resource writer
func (c *mysqlClient) Status() client.SubResourceWriter {
	return &mysqlStatusWriter{client: c}
}

// Scheme returns the scheme
func (c *mysqlClient) Scheme() *runtime.Scheme {
	panic("not implemented")
}

// RESTMapper returns the REST mapper
func (c *mysqlClient) RESTMapper() meta.RESTMapper {
	panic("not implemented")
}

// GroupVersionKindFor returns the GroupVersionKind for the given object
func (c *mysqlClient) GroupVersionKindFor(obj runtime.Object) (schema.GroupVersionKind, error) {
	if obj == nil {
		return schema.GroupVersionKind{}, fmt.Errorf("object is nil")
	}

	// Get the GVK from the object
	gvk := obj.GetObjectKind().GroupVersionKind()
	if gvk.Empty() {
		return schema.GroupVersionKind{}, fmt.Errorf("unable to determine GroupVersionKind")
	}

	return gvk, nil
}

// IsObjectNamespaced returns true if the object is namespaced
func (c *mysqlClient) IsObjectNamespaced(obj runtime.Object) (bool, error) {
	if obj == nil {
		return false, fmt.Errorf("object is nil")
	}

	// For now, assume all objects are namespaced
	return true, nil
}

// SubResource returns a sub-resource client
func (c *mysqlClient) SubResource(subResource string) client.SubResourceClient {
	// For now, only support status sub-resource
	if subResource == "status" {
		return &mysqlSubResourceClient{writer: c.Status()}
	}
	return nil
}

// convertK8sToCluster converts a Kubernetes Cluster object to MySQL database records
func (c *mysqlClient) convertK8sToCluster(obj client.Object, creation bool) (*Cluster, *ClusterSpec, *ClusterStatus, error) {
	o11yCluster, ok := obj.(*v1.Cluster)
	if !ok {
		return nil, nil, nil, ErrTypeInvalid
	}

	// Extract labels for region, tenant, projectID, applicationID, bizType
	region := o11yCluster.GetLabel(constants.LabelRegion)
	tenant := o11yCluster.GetLabel(constants.LabelTenantID)
	projectID := o11yCluster.GetLabel(constants.LabelProjectID)
	applicationID := o11yCluster.GetLabel(constants.LabelApplicationID)
	bizType := o11yCluster.GetLabel(constants.LabelBizType)
	ruleTemplateID := o11yCluster.GetLabel(constants.LabelTemplateID)

	// Create Cluster record
	cluster := &Cluster{
		Namespace:      o11yCluster.Namespace,
		ClusterID:      o11yCluster.Name,
		Region:         region,
		Tenant:         tenant,
		ProjectID:      projectID,
		ApplicationID:  applicationID,
		BizType:        bizType,
		RuleTemplateID: ruleTemplateID,
	}

	// Handle deletion timestamp
	if o11yCluster.DeletionTimestamp != nil {
		deletionTime := o11yCluster.DeletionTimestamp.Time
		cluster.DeletionTimestamp = &deletionTime
	}

	// Create ClusterSpec record
	spec := &ClusterSpec{
		ClusterID:              o11yCluster.Name,
		Labels:                 o11yCluster.Labels,
		Annotations:            o11yCluster.Annotations,
		EndpointServiceName:    o11yCluster.Spec.Endpoint.Name,
		EndpointServiceUserARN: o11yCluster.Spec.Endpoint.UserARN,
		EndpointServiceShared:  o11yCluster.Spec.Endpoint.Shared,
		MuteAlert:              o11yCluster.Spec.MuteAlert,
		VMAccountID:            o11yCluster.Spec.VictoriaAccount.AccountID,
		VMProjectID:            o11yCluster.Spec.VictoriaAccount.ProjectID,
		VMSupportServerless:    o11yCluster.Spec.VictoriaAccount.SupportServerless,
		VMClusterName:          o11yCluster.Spec.VictoriaAccount.VMClusterName,
		VMAvailabilityZone:     o11yCluster.Spec.VictoriaAccount.AvailabilityZone,
	}

	// Handle optional fields
	if o11yCluster.Spec.VictoriaAccount.MaxQueueSize != nil {
		spec.VMAlertMaxQueueSize = *o11yCluster.Spec.VictoriaAccount.MaxQueueSize
	}
	if o11yCluster.Spec.VictoriaAccount.RuleSelector != nil {
		spec.VMRuleSelector = *o11yCluster.Spec.VictoriaAccount.RuleSelector
	}
	// Flatten AlertResources into individual fields
	if o11yCluster.Spec.VictoriaAccount.AlertResources.Requests != nil {
		if cpu := o11yCluster.Spec.VictoriaAccount.AlertResources.Requests.Cpu(); cpu != nil {
			spec.VMAlertCPURequest = cpu.String()
		}
		if memory := o11yCluster.Spec.VictoriaAccount.AlertResources.Requests.Memory(); memory != nil {
			spec.VMAlertMemoryRequest = memory.String()
		}
	}
	if o11yCluster.Spec.VictoriaAccount.AlertResources.Limits != nil {
		if cpu := o11yCluster.Spec.VictoriaAccount.AlertResources.Limits.Cpu(); cpu != nil {
			spec.VMAlertCPULimit = cpu.String()
		}
		if memory := o11yCluster.Spec.VictoriaAccount.AlertResources.Limits.Memory(); memory != nil {
			spec.VMAlertMemoryLimit = memory.String()
		}
	}
	if o11yCluster.Spec.VictoriaAccount.Alerting != nil && o11yCluster.Spec.VictoriaAccount.Alerting.Alertmanagers != nil {
		spec.VMAlertManagers = o11yCluster.Spec.VictoriaAccount.Alerting.Alertmanagers
	}

	// Create ClusterStatus record
	status := &ClusterStatus{
		ClusterID: o11yCluster.Name,
	}
	// for creation, return empty status
	if !creation {
		status.Phase = string(o11yCluster.Status.Phase)
		status.DNSName = o11yCluster.Status.DnsName
		status.ServiceName = o11yCluster.Status.ServiceName
		status.UpdateAt = o11yCluster.Status.UpdateAt
	}

	// Parse resource version if present
	if o11yCluster.ResourceVersion != "" {
		rv, err := strconv.ParseUint(o11yCluster.ResourceVersion, 10, 64)
		if err == nil {
			status.ResourceVersion = rv
		}
	}

	return cluster, spec, status, nil
}

// convertClusterToK8s converts MySQL database records to a Kubernetes Cluster object
func (c *mysqlClient) convertClusterToK8s(cluster *Cluster, obj client.Object) error {
	o11yCluster, ok := obj.(*v1.Cluster)
	if !ok {
		return fmt.Errorf("object is not a Cluster type")
	}

	// Set TypeMeta
	o11yCluster.TypeMeta = metav1.TypeMeta{
		APIVersion: "observability.pingcap.com/v1",
		Kind:       "Cluster",
	}

	// Set ObjectMeta
	o11yCluster.ObjectMeta = metav1.ObjectMeta{
		Name:        cluster.ClusterID,
		Namespace:   cluster.Namespace,
		Labels:      cluster.Spec.Labels,
		Annotations: cluster.Spec.Annotations,
		CreationTimestamp: metav1.Time{
			Time: cluster.CreationTimestamp,
		},
		ResourceVersion: strconv.FormatUint(cluster.Status.ResourceVersion, 10),
	}

	// Set deletion timestamp if present
	if cluster.DeletionTimestamp != nil {
		o11yCluster.DeletionTimestamp = &metav1.Time{
			Time: *cluster.DeletionTimestamp,
		}
	}

	// Set Spec
	o11yCluster.Spec = v1.ClusterSpec{
		Endpoint: v1.Endpoint{
			Name:    cluster.Spec.EndpointServiceName,
			UserARN: cluster.Spec.EndpointServiceUserARN,
			Shared:  cluster.Spec.EndpointServiceShared,
		},
		VictoriaAccount: v1.VictoriaAccount{
			AccountID:         cluster.Spec.VMAccountID,
			ProjectID:         cluster.Spec.VMProjectID,
			SupportServerless: cluster.Spec.VMSupportServerless,
			VMClusterName:     cluster.Spec.VMClusterName,
			AvailabilityZone:  cluster.Spec.VMAvailabilityZone,
			RuleSelector:      &cluster.Spec.VMRuleSelector,
			AlertResources:    buildResourceRequirements(cluster.Spec.VMAlertCPURequest, cluster.Spec.VMAlertCPULimit, cluster.Spec.VMAlertMemoryRequest, cluster.Spec.VMAlertMemoryLimit),
		},
		MuteAlert: cluster.Spec.MuteAlert,
	}

	// Set optional fields
	if cluster.Spec.VMAlertMaxQueueSize > 0 {
		maxQueueSize := cluster.Spec.VMAlertMaxQueueSize
		o11yCluster.Spec.VictoriaAccount.MaxQueueSize = &maxQueueSize
	}
	if len(cluster.Spec.VMAlertManagers) > 0 {
		o11yCluster.Spec.VictoriaAccount.Alerting = &prometheusv1.AlertingSpec{
			Alertmanagers: cluster.Spec.VMAlertManagers,
		}
	}

	// Set Status
	o11yCluster.Status = v1.ClusterStatus{
		Phase:       constants.ResourceStatus(cluster.Status.Phase),
		DnsName:     cluster.Status.DNSName,
		ServiceName: cluster.Status.ServiceName,
		UpdateAt:    cluster.Status.UpdateAt,
	}

	return nil
}

// convertClusterListToK8s converts a list of MySQL database records to a Kubernetes ClusterList object
func (c *mysqlClient) convertClusterListToK8s(clusters []Cluster, list client.ObjectList) error {
	k8sClusterList, ok := list.(*v1.ClusterList)
	if !ok {
		return fmt.Errorf("object is not a ClusterList type")
	}

	// Set TypeMeta
	k8sClusterList.TypeMeta = metav1.TypeMeta{
		APIVersion: "observability.pingcap.com/v1",
		Kind:       "ClusterList",
	}

	// Convert each cluster
	k8sClusterList.Items = make([]v1.Cluster, 0, len(clusters))
	for i := range clusters {
		var k8sCluster v1.Cluster
		if err := c.convertClusterToK8s(&clusters[i], &k8sCluster); err != nil {
			return fmt.Errorf("failed to convert cluster %s: %w", clusters[i].ClusterID, err)
		}
		k8sClusterList.Items = append(k8sClusterList.Items, k8sCluster)
	}

	return nil
}

// buildResourceRequirements constructs a corev1.ResourceRequirements from flattened string fields
func buildResourceRequirements(cpuRequest, cpuLimit, memoryRequest, memoryLimit string) corev1.ResourceRequirements {
	resources := corev1.ResourceRequirements{
		Requests: corev1.ResourceList{},
		Limits:   corev1.ResourceList{},
	}

	if cpuRequest != "" {
		if quantity, err := resource.ParseQuantity(cpuRequest); err == nil {
			resources.Requests[corev1.ResourceCPU] = quantity
		}
	}
	if memoryRequest != "" {
		if quantity, err := resource.ParseQuantity(memoryRequest); err == nil {
			resources.Requests[corev1.ResourceMemory] = quantity
		}
	}
	if cpuLimit != "" {
		if quantity, err := resource.ParseQuantity(cpuLimit); err == nil {
			resources.Limits[corev1.ResourceCPU] = quantity
		}
	}
	if memoryLimit != "" {
		if quantity, err := resource.ParseQuantity(memoryLimit); err == nil {
			resources.Limits[corev1.ResourceMemory] = quantity
		}
	}

	return resources
}

// encodeContinueToken encodes a pagination cursor into a base64 string token
func (c *mysqlClient) encodeContinueToken(cursor *paginationCursor) (string, error) {
	if cursor == nil {
		return "", nil
	}
	data, err := json.Marshal(cursor)
	if err != nil {
		return "", fmt.Errorf("failed to marshal cursor: %w", err)
	}
	return base64.StdEncoding.EncodeToString(data), nil
}

// decodeContinueToken decodes a base64 string token into a pagination cursor
func (c *mysqlClient) decodeContinueToken(token string) (*paginationCursor, error) {
	if token == "" {
		return nil, fmt.Errorf("empty token")
	}
	data, err := base64.StdEncoding.DecodeString(token)
	if err != nil {
		return nil, fmt.Errorf("failed to decode token: %w", err)
	}
	var cursor paginationCursor
	if err := json.Unmarshal(data, &cursor); err != nil {
		return nil, fmt.Errorf("failed to unmarshal cursor: %w", err)
	}
	if cursor.LastClusterID == "" {
		return nil, fmt.Errorf("invalid cursor: missing last_cluster_id")
	}
	return &cursor, nil
}

// applyLabelSelector applies label selector filters to the query
func (c *mysqlClient) applyLabelSelector(query *gorm.DB, selector labels.Selector) *gorm.DB {
	requirements, selectable := selector.Requirements()
	if !selectable {
		return query
	}

	needsSpecJoin := false
	for _, req := range requirements {
		key := req.Key()

		// Optimize queries for indexed fields by using table columns instead of JSON_EXTRACT
		if indexedField := c.getIndexedFieldForLabel(key); indexedField != "" {
			query = c.applyIndexedFieldRequirement(query, indexedField, req)
		} else {
			// Fall back to JSON_EXTRACT for other labels
			needsSpecJoin = true
			query = c.applyJSONFieldRequirement(query, key, req)
		}
	}

	// Only join cluster_specs if needed for JSON label queries
	if needsSpecJoin {
		query = query.Joins("LEFT JOIN cluster_specs ON clusters.cluster_id = cluster_specs.cluster_id")
	}

	return query
}

// applyIndexedFieldRequirement applies a label requirement using indexed table columns
func (c *mysqlClient) applyIndexedFieldRequirement(query *gorm.DB, field string, req labels.Requirement) *gorm.DB {
	switch req.Operator() {
	case selection.In, selection.Equals:
		values := req.Values()
		if len(values) > 0 {
			query = query.Where(field+" IN ?", values.List())
		}
	case selection.NotIn, selection.NotEquals:
		values := req.Values()
		if len(values) > 0 {
			query = query.Where(field+" NOT IN ?", values.List())
		}
	case selection.Exists:
		query = query.Where(field + " IS NOT NULL AND " + field + " != ''")
	case selection.DoesNotExist:
		query = query.Where(field + " IS NULL OR " + field + " = ''")
	default:
		panic("not supported Selector")
	}
	return query
}

// applyJSONFieldRequirement applies a label requirement using JSON_EXTRACT
func (c *mysqlClient) applyJSONFieldRequirement(query *gorm.DB, key string, req labels.Requirement) *gorm.DB {
	// Quote the key in the JSON path to handle special characters like '.', '/', '-'
	jsonPath := fmt.Sprintf(`$."%s"`, key)
	switch req.Operator() {
	case selection.In, selection.Equals:
		values := req.Values()
		if len(values) > 0 {
			query = query.Where("JSON_EXTRACT(cluster_specs.labels, ?) IN ?", jsonPath, values.List())
		}
	case selection.NotIn, selection.NotEquals:
		values := req.Values()
		if len(values) > 0 {
			query = query.Where("JSON_EXTRACT(cluster_specs.labels, ?) NOT IN ?", jsonPath, values.List())
		}
	case selection.Exists:
		query = query.Where("JSON_EXTRACT(cluster_specs.labels, ?) IS NOT NULL", jsonPath)
	case selection.DoesNotExist:
		query = query.Where("JSON_EXTRACT(cluster_specs.labels, ?) IS NULL", jsonPath)
	default:
		panic("not supported Selector")
	}
	return query
}

// generateContinueToken generates a continue token if there are more results
func (c *mysqlClient) generateContinueToken(clusters []Cluster, limit int64) (string, error) {
	if limit == 0 || int64(len(clusters)) <= limit {
		return "", nil
	}

	// There are more results, generate continue token from the last record we'll return
	lastCluster := clusters[limit-1]
	return c.encodeContinueToken(&paginationCursor{
		LastClusterID: lastCluster.ClusterID,
	})
}

// getIndexedFieldForLabel returns the table column name for optimized label queries.
// Returns empty string if the label should use JSON_EXTRACT instead.
func (c *mysqlClient) getIndexedFieldForLabel(labelKey string) string {
	switch labelKey {
	case constants.LabelClusterID:
		return "clusters.cluster_id"
	case constants.LabelTenantID:
		return "clusters.tenant"
	case constants.LabelProjectID:
		return "clusters.project_id"
	case constants.LabelApplicationID:
		return "clusters.application_id"
	case constants.LabelBizType:
		return "clusters.biz_type"
	case constants.LabelTemplateID:
		return "clusters.rule_template_id"
	default:
		return ""
	}
}

func (m *mysqlClient) withTx(ctx context.Context, fn func(*gorm.DB) error) (err error) {
	tx := m.db.WithContext(ctx).Begin()
	if tx.Error != nil {
		return fmt.Errorf("failed to begin transaction: %w", tx.Error)
	}

	defer func() {
		if r := recover(); r != nil {
			tx.Rollback()
			err = fmt.Errorf("panic in transaction: %v", r)
		}
	}()

	if err = fn(tx); err != nil {
		tx.Rollback()
		return err
	}

	return tx.Commit().Error
}

func (c *mysqlClient) logger(ctx context.Context) logr.Logger {
	reconcileID := controller.ReconcileIDFromContext(ctx)
	if reconcileID != "" {
		return c.log.WithValues("reconcileID", reconcileID)
	}
	requestID, ok := ctx.Value(string(constants.OpenAPIHeaderRequestId)).(string)
	if !ok {
		return c.log
	}
	return c.log.WithValues("request_id", requestID)
}

func checkObject(obj client.Object) error {
	objVal := reflect.ValueOf(obj)
	if objVal.Kind() != reflect.Pointer {
		return ErrTypeInvalid
	}
	if !objVal.Elem().IsValid() {
		return ErrValueInvalid
	}

	return nil
}