/* Copyright 2021.

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 cluster

import (
	"context"
	"errors"
	"fmt"
	"os"
	"strconv"
	"time"

	alicred "github.com/aliyun/credentials-go/credentials"
	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/sqs"
	"github.com/go-logr/logr"
	"github.com/tidbcloud/aws-shared-provider/cmd/manager/app/config"
	"github.com/tidbcloud/aws-shared-provider/pkg/auditlog/entity"
	"github.com/tidbcloud/aws-shared-provider/pkg/auditlog/repo"
	auditlogutil "github.com/tidbcloud/aws-shared-provider/pkg/auditlog/util"
	"github.com/tidbcloud/aws-shared-provider/pkg/manager"
	"github.com/tidbcloud/aws-shared-provider/pkg/tidbmanager"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/pd"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tidb"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/tidbworker"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	infrao11y "github.com/tidbcloud/observability/agent-control-server/pkg/o11y/api"
	infra "github.com/tidbcloud/tidb-management-service/api/serverless_infra"
	infraclient "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client"
	"github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/redis"
	unmarshaldynamo "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/unmarshal/dynamo"
	unmarshalots "github.com/tidbcloud/tidb-management-service/api/serverless_infra/consumer/unmarshal/tablestore"
	"github.com/tidbcloud/tidb-management-service/api/serverless_infra/meta"
	sourcemysql "github.com/tidbcloud/tidb-management-service/api/serverless_infra/source/mysql"
	sourcermq "github.com/tidbcloud/tidb-management-service/api/serverless_infra/source/rocketmq"
	sourcesqs "github.com/tidbcloud/tidb-management-service/api/serverless_infra/source/sqs"
	infraworkqueue "github.com/tidbcloud/tidb-management-service/api/serverless_infra/workqueue"
	clientv3 "go.etcd.io/etcd/client/v3"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	"k8s.io/klog/v2"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	"sigs.k8s.io/controller-runtime/pkg/source"
)

const (
	MaxConcurrentClusters = 20
)

// Reconciler reconciles a Cluster object.
type Reconciler struct {
	logr.Logger

	InfraClusterCli infraclient.Client
	InfraClient     client.Client

	InfraDoubleWriteCli infraclient.Client

	Client   client.Client
	Recorder record.EventRecorder

	PdClient      pd.HttpPdClient
	TidbWorkerApi tidbworker.Api
	TidbManagers  map[string]tidbmanager.TidbManager

	// Cluster metric agent dependencies introduced by new o11y infra.
	O11yAgentEnabled    bool
	O11yConfig          *cloudv1.ObservabilityInfraConfig
	O11yApi             infrao11y.Api
	O11yAgentAccessArn  string
	O11yEnv             string
	O11yGlobalClusterID string

	AwsCfg                *aws.Config
	AlicloudCred          alicred.Credential
	EtcdKV                clientv3.KV
	EtcdLease             clientv3.Lease
	S3BucketName          string
	S3BackupRoleArn       string
	BaseProxy             string
	KubeClusterName       string
	ImportJobsNamespace   string
	KeyspacePoolNs        string
	EncryptKeyspacePoolNs string
	EtcdTokenPrefixMap    map[string]string
	TidbVersionMap        map[string]string
	TidbPools             []string
	ProviderType          string

	SkipClusterDeleteEtcdAndPd bool

	EnableDoubleWrite      bool
	EnableInfraMysqlClient bool
	AuditLogScheduleRepo   repo.AuditLogScheduleRepo
	TidbController         tidb.Controller
}

// reconcileContext is the local context of a single cluster reconciliation, it is NOT concurrency safe.
type reconcileContext struct {
	*Reconciler
	logr.Logger

	ctx context.Context
	// the cluster will be continuously updated during reconciliation.
	cluster *infra.Cluster

	keyspaceNs    string
	keyspaceToken string
	keyspaceID    uint32

	specTidbPool, tidbVersion string

	specResourceGroups   map[string]*infra.ResourceGroup
	auditLogScheduleRepo repo.AuditLogScheduleRepo
}

func needGetWithConsistent(tidbPool string, cluster *infra.Cluster) bool {
	if cluster.Status.KeyspaceToken == "" || cluster.Annotations[types.AnnoKeyTidbPoolChanging] != "" {
		return true
	}
	if tidbPool != cluster.Status.TidbPool {
		return true
	}
	for rgName, rgSpec := range cluster.Spec.TiDB.ResourceGroups {
		pool := rgSpec.Pool
		if pool == "" {
			pool = tidbPool
		}
		rgStatus := cluster.Status.ResourceGroups[rgName]
		if pool != rgStatus.Pool {
			return true
		}
	}
	return false
}

func (r *Reconciler) newReconcileContext(ctx context.Context, req ctrl.Request) (*reconcileContext, error) {
	log := ctrl.LoggerFrom(ctx)

	var opts []infraclient.GetOptionFunc
	if r.EnableInfraMysqlClient {
		opts = append(opts, infraclient.WithConsistent())
	}
	cluster := &infra.Cluster{ObjectMeta: meta.ObjectMeta{Namespace: req.Namespace, Name: req.Name}}
	if err := r.InfraClusterCli.Get(ctx, cluster, opts...); err != nil {
		if errors.Is(err, infraclient.ErrNotFound) {
			return nil, nil
		}
		return nil, fmt.Errorf("can't get cluster: %w", err)
	}

	if cluster.TagsKubernetes != r.KubeClusterName {
		log.Info("unexpected cluster scheduling", "tagsKubernetes", cluster.TagsKubernetes)
		return nil, nil
	}

	tidbPool := r.getTidbPool(cluster)

	// The first MySQL read is already consistent. Other clients may return cached data,
	// so verify the resource version against the underlay in sensitive states.
	if !r.EnableInfraMysqlClient && needGetWithConsistent(tidbPool, cluster) {
		latestCluster := &infra.Cluster{ObjectMeta: meta.ObjectMeta{Namespace: req.Namespace, Name: req.Name}}
		if err := r.InfraClusterCli.Get(ctx, latestCluster, infraclient.WithConsistent()); err != nil {
			return nil, fmt.Errorf("can't get cluster with consistent: %w", err)
		}
		if latestCluster.ResourceVersion != cluster.ResourceVersion {
			return nil, fmt.Errorf("cluster resourceVersion(%v) is not the latest(%v), reconcile again",
				cluster.ResourceVersion, latestCluster.ResourceVersion)
		}
	}

	log = log.WithValues("rrv", cluster.ResourceVersion)
	if cluster.Status.KeyspaceToken != "" {
		log = log.WithValues("keyspaceToken", cluster.Status.KeyspaceToken)
	}

	keyspaceNs := r.KeyspacePoolNs
	if cluster.Spec.EncryptKeyspace {
		keyspaceNs = r.EncryptKeyspacePoolNs
	}

	tidbMinReplicas := types.GetDefaultTiDBMinReplicas(cluster)
	tidbMaxReplicas, err := r.getTidbMaxReplicas(ctx, tidbPool)
	if err != nil {
		return nil, fmt.Errorf("getTidbMaxReplicas err: %w", err)
	}
	if cluster.Spec.TiDB.MaxReplicas != nil {
		tidbMaxReplicas = *cluster.Spec.TiDB.MaxReplicas
		log.Info("use infra spec.tidb.maxReplicas", "tidbMaxReplicas", tidbMaxReplicas)
	}
	if tidbMaxReplicas < 1 {
		log.Info("tidbMaxReplicas < 1, reset tidbMaxReplicas to 1", "tidbMaxReplicas", tidbMaxReplicas)
		tidbMaxReplicas = 1
	}
	if cluster.Spec.TiDB.MinReplicas != nil {
		tidbMinReplicas = *cluster.Spec.TiDB.MinReplicas
		log.Info("use infra spec.tidb.minReplicas", "tidbMinReplicas", tidbMinReplicas)
	}
	if tidbMinReplicas < 1 {
		log.Info("tidbMinReplicas < 1, reset tidbMinReplicas to defaultMinReplicas", "tidbMinReplicas", tidbMinReplicas)
		tidbMinReplicas = types.GetDefaultTiDBMinReplicas(cluster)
	}
	if tidbMinReplicas > tidbMaxReplicas {
		log.Info("tidbMinReplicas > tidbMaxReplicas, reset tidbMinReplicas to defaultMinReplicas",
			"tidbMinReplicas", tidbMinReplicas, "tidbMaxReplicas", tidbMaxReplicas)
		tidbMinReplicas = types.GetDefaultTiDBMinReplicas(cluster)
	}
	if cluster.Spec.TiDB.Replicas > tidbMaxReplicas {
		log.Info("tidb replicas is greater than tidbMaxReplicas, force reset tidb replicas", "tidbPool", tidbPool,
			"tidb replicas", cluster.Spec.TiDB.Replicas, "tidbMaxReplicas", tidbMaxReplicas,
		)
		if err := r.InfraClusterCli.Patch(ctx, map[string]any{"'spec.tidb'.replicas": tidbMaxReplicas}, cluster); err != nil {
			return nil, fmt.Errorf("patch cluster for tidb replicas err: %w", err)
		}
	} else if cluster.Spec.TiDB.Replicas < tidbMinReplicas {
		log.Info("tidb replicas is less than tidbMinReplicas, force reset tidb replicas", "tidbPool", tidbPool,
			"tidb replicas", cluster.Spec.TiDB.Replicas, "tidbMinReplicas", tidbMinReplicas,
		)
		if err := r.InfraClusterCli.Patch(ctx, map[string]any{"'spec.tidb'.replicas": tidbMinReplicas}, cluster); err != nil {
			return nil, fmt.Errorf("patch cluster for tidb replicas err: %w", err)
		}
	}

	return &reconcileContext{
		Reconciler: r,
		Logger:     log,

		ctx:                  ctx,
		cluster:              cluster,
		keyspaceNs:           keyspaceNs,
		keyspaceToken:        cluster.Status.KeyspaceToken,
		keyspaceID:           cluster.Status.KeyspaceId,
		specTidbPool:         tidbPool,
		specResourceGroups:   cluster.Spec.TiDB.ResourceGroups,
		tidbVersion:          r.TidbVersionMap[tidbPool],
		auditLogScheduleRepo: r.AuditLogScheduleRepo,
	}, nil
}

func (r *Reconciler) getTidbMaxReplicas(ctx context.Context, tidbPoolNamespace string) (int, error) {
	serverlessPoolConfigCm := &corev1.ConfigMap{}
	if err := r.Client.Get(ctx,
		client.ObjectKey{
			Name:      types.ServerlessPoolConfig,
			Namespace: tidbPoolNamespace,
		}, serverlessPoolConfigCm); err != nil {
		return 0, err
	}
	if serverlessPoolConfigCm.Data[types.TidbMaxReplicasKey] == "" {
		return 0, fmt.Errorf("key %s is empty", types.TidbMaxReplicasKey)
	}
	return strconv.Atoi(serverlessPoolConfigCm.Data[types.TidbMaxReplicasKey])
}

func (r *Reconciler) getTidbPool(cluster *infra.Cluster) string {
	tidbPool := r.TidbPools[0] // The first element is default TidbPool
	for _, v := range r.TidbPools {
		if v == cluster.Spec.TiDB.PoolSelector {
			return v
		}
	}

	return tidbPool
}

// buildAuditLogSchedule builds the audit log schedule based on the cluster spec and status.
// It returns true if the schedule needs to be updated, false otherwise.
func (r *reconcileContext) buildAuditLogSchedule(schedule *entity.AuditLogSchedule) bool {
	spec := r.cluster.Spec.AuditLogConfig
	status := r.cluster.Status.AuditLogConfig

	// rotation
	rotationUpdated := false
	if spec.RotationIntervalSeconds != status.RotationIntervalSeconds {
		rotationUpdated = true
		schedule.RotationIntervalSeconds = spec.RotationIntervalSeconds
		r.Info("tidb audit log rotation interval have changed",
			"pre", status.RotationIntervalSeconds,
			"now", spec.RotationIntervalSeconds)
	}
	if spec.RotationSizeBytes != status.RotationSizeBytes {
		rotationUpdated = true
		schedule.RotationSizeBytes = spec.RotationSizeBytes
		r.Info("tidb audit log rotation size have changed",
			"pre", status.RotationIntervalSeconds,
			"now", spec.RotationIntervalSeconds)
	}
	// unredacted
	unredactedUpdated := false
	if spec.Unredacted != status.Unredacted {
		unredactedUpdated = true
		schedule.Unredacted = spec.Unredacted
		r.Info("tidb audit log unredacted have changed",
			"pre", status.Unredacted,
			"now", spec.Unredacted)
	}
	// enable & disable
	enableUpdated := false
	if spec.Enabled != status.Enabled {
		enableUpdated = true
		// enable the audit log
		if spec.Enabled {
			schedule.State = entity.AuditLogScheduleStateEnabled
			schedule.DisableTime = nil
			schedule.ResetRotationTime()
		} else {
			// disable the audit log
			currentTime := time.Now()
			schedule.DisableTime = &currentTime
		}
		r.Info("tidb audit log enabled have changed",
			"pre", status.Enabled,
			"now", spec.Enabled)
	}

	// cloud storage
	cloudStorageUpdated := false
	if spec.CloudStorage != nil {
		// empty cloud storage of status means using tidb cloud storage
		switch spec.CloudStorage.CloudStorageType {
		case infra.CloudStorageTypeS3:
			if status.CloudStorage == nil || status.CloudStorage.S3 == nil ||
				status.CloudStorage.CloudStorageType != infra.CloudStorageTypeS3 ||
				*spec.CloudStorage.S3 != *status.CloudStorage.S3 {
				cloudStorageUpdated = true
			}
		case infra.CloudStorageTypeOSS:
			if status.CloudStorage == nil || status.CloudStorage.OSS == nil ||
				status.CloudStorage.CloudStorageType != infra.CloudStorageTypeOSS ||
				*spec.CloudStorage.OSS != *status.CloudStorage.OSS {
				cloudStorageUpdated = true
			}
		case infra.CloudStorageTypeGCS:
			if status.CloudStorage == nil || status.CloudStorage.GCS == nil ||
				status.CloudStorage.CloudStorageType != infra.CloudStorageTypeGCS ||
				spec.CloudStorage.GCS != status.CloudStorage.GCS {
				cloudStorageUpdated = true
			}
		case infra.CloudStorageTypeAzureBlob:
			if status.CloudStorage == nil || status.CloudStorage.AzureBlob == nil ||
				status.CloudStorage.CloudStorageType != infra.CloudStorageTypeAzureBlob ||
				spec.CloudStorage.AzureBlob != status.CloudStorage.AzureBlob {
				cloudStorageUpdated = true
			}
		case infra.CloudStorageTypeTiDBCloud:
			if status.CloudStorage != nil && status.CloudStorage.CloudStorageType != infra.CloudStorageTypeTiDBCloud {
				cloudStorageUpdated = true
			}
		}
	}
	if cloudStorageUpdated {
		r.refreshAuditLogScheduleCloudStorage(schedule)
		r.Info("tidb audit log cloud storage have changed",
			"pre", status.CloudStorage, "now", spec.CloudStorage)
	}

	return rotationUpdated || enableUpdated || cloudStorageUpdated || unredactedUpdated
}

func (r *reconcileContext) needResetTidbCluster() bool {
	empty := infra.TiDBStatus{}
	if r.cluster.Status.TiDB == empty {
		return true
	}
	if r.specTidbPool != r.cluster.Status.TidbPool {
		r.Info("tidb pool have changed", "old", r.cluster.Status.TidbPool, "now", r.specTidbPool)
		return true
	}
	if r.cluster.Spec.TiDB.Replicas != r.cluster.Status.TiDB.UpdatedReplicas {
		r.Info("tidb replicas have changed",
			"old", r.cluster.Status.TiDB.UpdatedReplicas, "now", r.cluster.Spec.TiDB.Replicas)
		return true
	}
	if len(r.specResourceGroups) != len(r.cluster.Status.ResourceGroups) {
		r.Info("tidb resource groups count have changed", "old", len(r.cluster.Status.ResourceGroups), "now", len(r.specResourceGroups))
		return true
	}
	for rgName, rgSpec := range r.specResourceGroups {
		rgSpecCopy := rgSpec.DeepCopy()
		if rgSpecCopy.Pool == "" {
			rgSpecCopy.Pool = r.specTidbPool
		}
		rgStatus := r.cluster.Status.ResourceGroups[rgName]
		if *rgSpecCopy != rgStatus {
			r.Info("tidb resource group have changed", "rgName", rgName, "old", rgStatus, "now", rgSpecCopy)
			return true
		}
	}

	return false
}

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
func (r *Reconciler) Reconcile(ctx context.Context, req any) (ctrl.Result, error) {
	log := ctrl.LoggerFrom(ctx)
	log.Info("start reconcile cluster")
	start := time.Now()
	defer func() {
		log.Info("finish reconcile cluster", "time", time.Since(start))
	}()

	request := req.(ctrl.Request)
	rx, err := r.newReconcileContext(ctx, request)

	if err != nil || rx == nil {
		return utils.ReconcileErr(log, "Cluster", request.Name, err)
	}

	var phase infra.ClusterPhase
	var reason string
	var needUpdatePhase bool

	if rx.cluster.DeletionTimestamp != nil {
		if rx.cluster.Status.Phase != infra.ClusterPhaseDeleting &&
			rx.cluster.Status.Phase != infra.ClusterPhasePendingDelete {
			// shortcut to PendingDelete phase whenever cluster is get deleted
			phase = infra.ClusterPhasePendingDelete
			reason = "cluster is pending to be deleted"
			needUpdatePhase = true
		}
	} else {
		if rx.cluster.Status.Phase == "" {
			// shortcut to Pending phase when cluster is newly created
			phase = infra.ClusterPhasePending
			reason = "cluster is waiting to be reconciled"
			needUpdatePhase = true
		}
	}

	if needUpdatePhase {
		if err = rx.updateProgress(phase, reason); err != nil {
			return utils.ReconcileErr(log, "Cluster", rx.cluster.Name, err)
		}
	}

	clusterPhase := rx.cluster.Status.Phase
	switch clusterPhase {
	case infra.ClusterPhaseCreating, infra.ClusterPhasePending:
		err = rx.createCluster()
	case infra.ClusterPhaseRunning, infra.ClusterPhaseUpgrading, infra.ClusterPhaseScaling:
		err = rx.reconcileCluster()
	case infra.ClusterPhasePendingDelete:
		err = rx.pendingDeleteCluster()
	case infra.ClusterPhaseDeleting:
		err = rx.deleteCluster()
	case infra.ClusterPhaseUnavailable:
		// when cluster is unavailable, skip reconciliation and observe until it recovers for safety
		err = rx.observeCluster()
	default:
		// unknown phase, try to recover the real phase via observation
		err = rx.observeCluster()
	}

	deleted := false
	if clusterPhase == infra.ClusterPhaseDeleting && err == nil {
		err = rx.InfraClusterCli.Delete(rx.ctx, rx.cluster, infraclient.WithForce())
		deleted = true
	}

	if rx.EnableDoubleWrite && err == nil {
		result := rx.InfraDoubleWriteCli.ForceUpdate(ctx, rx.cluster, deleted)
		log.Info("doubleWriteClient force updates cluster", "result", result, "deleted", deleted)
	}

	return utils.ReconcileErr(log, "Cluster", rx.cluster.Name, err)
}

// SetupWithManager sets up the controller with the Manager.
func (r *Reconciler) SetupWithManager(
	mgr manager.Manager,
	cfg *config.Config,
	redisConsumer *redis.Consumer,
) error {
	r.Client = mgr.GetClient()
	r.InfraClient = mgr.GetClient()
	r.Recorder = mgr.GetEventRecorderFor("Cluster")
	log := mgr.GetLogger().WithValues(
		"controller", "Cluster",
	)
	logConstructor := func(in *any) logr.Logger {
		if in != nil {
			if req, ok := (*in).(reconcile.Request); ok {
				return log.WithValues(
					"object", klog.KRef(req.Namespace, req.Name),
					"namespace", req.Namespace, "name", req.Name,
				)
			}
		}
		return log
	}

	maxConcurrent := MaxConcurrentClusters
	if v := os.Getenv("MAX_CONCURRENT_CLUSTERS"); v != "" {
		if n, err := strconv.Atoi(v); err == nil && n > 0 {
			maxConcurrent = n
		}
	}

	if cfg.EnableInfraMysqlClient {
		c, err := controller.NewTyped("Cluster", mgr, controller.TypedOptions[any]{
			NewQueue: func(
				controllerName string,
				rateLimiter workqueue.TypedRateLimiter[any],
			) workqueue.TypedRateLimitingInterface[any] {
				return infraworkqueue.NewMysqlRateLimitingQueueWithConfig(
					rateLimiter,
					workqueue.TypedRateLimitingQueueConfig[any]{Name: controllerName},
					infra.DefaultClusterEventTTL,
				)
			},
			RateLimiter:             infraworkqueue.ControllerRateLimiter(),
			MaxConcurrentReconciles: maxConcurrent,
			Reconciler:              r,
			LogConstructor:          logConstructor,
		})
		if err != nil {
			return fmt.Errorf("can't new a cluster controller: %w", err)
		}

		r.Logger = c.GetLogger()

		rdsConfig, err := auditlogutil.GetRdsConfig(cfg.CloudProviderType, cfg.InfraRdsSecretArn, cfg.Region)
		if err != nil {
			return fmt.Errorf("get RDS config err: %w", err)
		}
		consumerSource := &sourcemysql.Consumer{
			Logger:  c.GetLogger(),
			ConnCfg: rdsConfig.ToGormutilConfig(),

			Kubernetes: cfg.KubeClusterName,
		}
		return r.watches(c, consumerSource)
	}
	switch {
	case r.AwsCfg != nil:
		c, err := controller.NewTyped("Cluster", mgr, controller.TypedOptions[any]{
			NewQueue: func(
				controllerName string,
				rateLimiter workqueue.TypedRateLimiter[any],
			) workqueue.TypedRateLimitingInterface[any] {
				return infraworkqueue.NewSqsRateLimitingQueueWithConfig(
					rateLimiter,
					workqueue.TypedRateLimitingQueueConfig[any]{Name: controllerName},
				)
			},
			RateLimiter:             infraworkqueue.ControllerRateLimiter(),
			MaxConcurrentReconciles: maxConcurrent,
			Reconciler:              r,
			LogConstructor:          logConstructor,
		})
		if err != nil {
			return fmt.Errorf("can't new aws cluster controller: %w", err)
		}

		r.Logger = c.GetLogger()

		consumerSource := &sourcesqs.Consumer{
			Logger: c.GetLogger(),

			QueueUrl:      cfg.ClusterQueueURL,
			Client:        sqs.NewFromConfig(*r.AwsCfg),
			RedisConsumer: redisConsumer,
			Unmarshal: func(data any) (infra.Object, error) {
				return unmarshaldynamo.Unmarshal(data, func() infra.Object { return &infra.Cluster{} })
			},
		}
		return r.watches(c, consumerSource)
	case r.AlicloudCred != nil:
		c, err := controller.NewTyped("Cluster", mgr, controller.TypedOptions[any]{
			NewQueue: func(
				controllerName string,
				rateLimiter workqueue.TypedRateLimiter[any],
			) workqueue.TypedRateLimitingInterface[any] {
				return infraworkqueue.NewRmqRateLimitingQueueWithConfig(
					rateLimiter,
					workqueue.TypedRateLimitingQueueConfig[any]{Name: controllerName},
				)
			},
			RateLimiter:             infraworkqueue.ControllerRateLimiter(),
			MaxConcurrentReconciles: maxConcurrent,
			Reconciler:              r,
			LogConstructor:          logConstructor,
		})
		if err != nil {
			return fmt.Errorf("can't new alicloud cluster controller: %w", err)
		}

		r.Logger = c.GetLogger()

		consumerSource := &sourcermq.Consumer{
			Logger: c.GetLogger(),

			Topic:             cfg.InfraClusterTable,
			DeadTopic:         cfg.InfraClusterTable + "_dead",
			ConsumerGroup:     cfg.RocketmqConsumerGroup,
			DeadConsumerGroup: cfg.RocketmqConsumerGroup + "_dead",
			Endpoint:          cfg.RocketmqEndpoint,
			Namespace:         cfg.RocketmqInstanceID,
			AccessKey:         cfg.RocketmqUsername,
			AccessSecret:      cfg.RocketmqPassword,
			FilterExpression:  cfg.KubeClusterName,
			RedisConsumer:     redisConsumer,
			Unmarshal: func(data any) (infra.Object, error) {
				return unmarshalots.Unmarshal(data, func() infra.Object { return &infra.Cluster{} })
			},
		}
		return r.watches(c, consumerSource)
	default:
		return fmt.Errorf("unsupported underlay infra cluster")
	}
}

func (r *Reconciler) watches(c controller.TypedController[any], source source.TypedSource[any]) error {
	if err := c.Watch(source); err != nil {
		return fmt.Errorf("can't watch cluster: %w", err)
	}

	return nil
}