package replay

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"github.com/tidbcloud/eks-provider/controllers/replay/task"
	"github.com/tidbcloud/eks-provider/controllers/replay/task/common"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
)

// Reconciler is the replay reconciler
type Reconciler struct {
	Logger        logr.Logger
	EventRecorder record.EventRecorder
	Client        client.Client
	Cache         client.Reader
	Manager       dataplane.K8sClientManager
	RootConfig    config.RootConfig
}

// Helper functions for common operations

// getClientset creates a kubernetes clientset from envProvider
func (r *Reconciler) getClientset(envProvider dataplane.EnvProvider) (*kubernetes.Clientset, error) {
	restConfig := envProvider.GetRestConfig()
	if restConfig == nil {
		return nil, fmt.Errorf("failed to get REST config from envProvider")
	}
	return kubernetes.NewForConfig(restConfig)
}

// getEffectiveReplicas returns the effective replica count (defaults to 1 if 0)
func getEffectiveReplicas(replicas int32) int32 {
	if replicas == 0 {
		return 1
	}
	return replicas
}

// updateTaskErrorMessages updates the error messages for tasks in replay status
func updateTaskErrorMessages(replay *cloudv1.Replay, taskErrors map[string][]string) {
	for i := range replay.Status.Tasks {
		taskName := replay.Status.Tasks[i].Name
		if errors, hasError := taskErrors[taskName]; hasError {
			replay.Status.Tasks[i].Message = strings.Join(errors, "; ")
		} else {
			replay.Status.Tasks[i].Message = ""
		}
	}
}

// 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 ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	logger.Info("Reconciling Replay")

	replay := &cloudv1.Replay{}
	if err := r.Cache.Get(ctx, req.NamespacedName, replay); err != nil {
		if apierrors.IsNotFound(err) {
			logger.Info("Replay CR not found, skipping", "replay", req.NamespacedName)
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, fmt.Errorf("unable to get replay %s: %v", req.NamespacedName, err)
	}

	// Check if replay is paused
	if replay.Spec.Paused != nil && *replay.Spec.Paused {
		logger.Info("Replay is paused, skipping reconciliation", "replay", req.NamespacedName)
		return ctrl.Result{}, nil
	}

	deleted := !replay.DeletionTimestamp.IsZero()

	// Use NewWithKubenets to create EnvProvider and auto-register
	// Registration key: "replay-{replay-name}" to avoid conflicts with Cluster CRs
	registrationKey := fmt.Sprintf("replay-%s-%s", replay.Name, replay.Spec.Kubernetes.Name)

	var dataplaneKubeData []byte
	if ref := replay.Spec.Kubernetes.KubeConfigSecretRef; ref != nil {
		var err error
		if dataplaneKubeData, err = dataplane.GetDataplaneKubeConfigFromMetaSecret(ctx, r.Client, ref); err != nil {
			r.EventRecorder.Eventf(replay, corev1.EventTypeWarning, "CannotGetKubeConfig", "cannot get kubeconfig, import:%s, err:%v", req.NamespacedName, err)
			return ctrl.Result{}, fmt.Errorf("cannot get kubeconfig: %w", err)
		}
		logger.Info("get dataplane kubeconfig from appbox KubeConfigSecretRef", "kubeconfig_len", len(dataplaneKubeData))
	}

	envProvider, err := dataplane.NewWithKubenets(
		r.Manager,
		registrationKey,
		&replay.Spec.Kubernetes,
		r.RootConfig.Network.Region,
		r.RootConfig.FullManage.AssumeRoleName,
		r.RootConfig.FullManage.AssumeRoleExternalID,
		dataplaneKubeData, // Use Endpoint + CAData from Kubernetes, not kubeconfig secret
	)
	if err != nil {
		logger.Error(err, "Failed to create dataplane provider",
			"registrationKey", registrationKey,
			"clusterName", replay.Spec.Kubernetes.Name,
			"endpoint", replay.Spec.Kubernetes.Endpoint,
			"region", r.RootConfig.Network.Region,
			"assumeRoleName", r.RootConfig.FullManage.AssumeRoleName,
			"account", replay.Spec.Kubernetes.Account)
		return ctrl.Result{}, fmt.Errorf("failed to create dataplane provider: %v", err)
	}

	// Get the registered client
	remote := envProvider.GetClient()

	// Build task context
	c := task.NewContext(r.EventRecorder, r.Logger, r.Client, remote, replay, r.RootConfig)
	c, err = DefaultReplayContext(c)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("failed to initialize context: %v", err)
	}

	var res ctrl.Result
	var errs error
	var retriableErrs error

	if deleted {
		// Handle deletion - destroy infrastructure
		tasks := c.Provider.Tasks()
		deployer := deploy.NewDeployer(tasks, nil, logger)

		results := deployer.Destroy(ctx)
		res, errs, retriableErrs = deploy.MinRequeue(results)
		if errs == nil && retriableErrs == nil {
			// All tasks destroyed successfully, remove finalizer
			controllerutil.RemoveFinalizer(replay, task.Finalizer)

			// Deregister the k8s client to prevent memory leak
			r.Manager.Deregister(registrationKey)
			logger.Info("Deregistered dataplane client", "key", registrationKey)
			if err := r.updateReplayDeletionMetadata(ctx, replay); err != nil {
				r.EventRecorder.Event(replay, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
				return res, err
			}
			if err := r.cleanupGeneratedKubeConfigSecret(ctx, replay); err != nil {
				logger.Error(err, "Failed to cleanup generated kubeconfig secret after replay deletion", "replay", req.NamespacedName)
			}
		} else if err := r.updateReplayDeletionStatus(ctx, replay); err != nil {
			r.EventRecorder.Event(replay, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
			return res, err
		}
	} else {
		// State machine: handle spec.state transitions
		res, errs, retriableErrs = r.handleStateTransition(ctx, logger, c, replay, remote, envProvider)
	}

	r.EventRecorder.Event(replay, corev1.EventTypeNormal, "ReconcileSuccess", "")
	if retriableErrs != nil {
		logger.Info("retriable error", "error", retriableErrs)
	}
	return res, errs
}

// handleStateTransition handles the state machine for Replay CR
// State transitions:
// - Deploy: Create infrastructure (StatefulSet, NodeGroup, etc.)
// - Active: Call HTTP API to start replay tasks
// - Pause: Call HTTP API to cancel replay tasks
// - Delete: Delete infrastructure (handled in Reconcile deletion path)
func (r *Reconciler) handleStateTransition(
	ctx context.Context,
	logger logr.Logger,
	c *task.Context,
	replay *cloudv1.Replay,
	remote client.Client,
	envProvider dataplane.EnvProvider,
) (ctrl.Result, error, error) {
	logger.Info("State machine", "spec.state", replay.Spec.State, "status.phase", replay.Status.Phase)

	var res ctrl.Result
	var errs, retriableErrs error

	switch replay.Spec.State {
	case cloudv1.ReplayStateDeploy:
		res, errs, retriableErrs = r.handleDeployState(ctx, logger, c, replay)
	case cloudv1.ReplayStateActive:
		res, errs, retriableErrs = r.handleActiveState(ctx, logger, c, replay, remote, envProvider)
	case cloudv1.ReplayStatePause:
		res, errs, retriableErrs = r.handlePauseState(ctx, logger, c, replay, remote, envProvider)
	default:
		replay.Status.Message = fmt.Sprintf("Unknown state: %s", replay.Spec.State)
		replay.Status.LastUpdateTime = metav1.Now()
		logger.Info("Unknown state", "state", replay.Spec.State)
	}

	// Update Replay CR (spec and status)
	if err := r.updateReplayCR(ctx, replay); err != nil {
		r.EventRecorder.Event(replay, corev1.EventTypeWarning, "ReconcileFailed", err.Error())
		return res, err, nil
	}

	return res, errs, retriableErrs
}

// handleDeployState handles the Deploy state - creates infrastructure
func (r *Reconciler) handleDeployState(
	ctx context.Context,
	logger logr.Logger,
	c *task.Context,
	replay *cloudv1.Replay,
) (ctrl.Result, error, error) {
	currentPhase := replay.Status.Phase
	if currentPhase != "" && currentPhase != cloudv1.ReplayPending && currentPhase != cloudv1.ReplayCreating {
		logger.Info("Infrastructure already deployed", "phase", currentPhase)
		return ctrl.Result{}, nil, nil
	}

	deployer := deploy.NewDeployer(c.Provider.Tasks(), nil, logger)
	results := deployer.Deploy(ctx)
	res, errs, retriableErrs := deploy.MinRequeue(results)

	if errs == nil && retriableErrs == nil {
		replay.Status.Phase = cloudv1.ReplayRunning
		replay.Status.Message = fmt.Sprintf("Infrastructure deployed, %d StatefulSet(s) ready", len(replay.Spec.Tasks))
		replay.Status.LastUpdateTime = metav1.Now()
	}

	return res, errs, retriableErrs
}

// handleActiveState handles the Active state - ensures infrastructure and starts tasks
func (r *Reconciler) handleActiveState(
	ctx context.Context,
	logger logr.Logger,
	c *task.Context,
	replay *cloudv1.Replay,
	remote client.Client,
	envProvider dataplane.EnvProvider,
) (ctrl.Result, error, error) {
	var res ctrl.Result
	var errs, retriableErrs error

	// Ensure infrastructure is deployed first
	if replay.Status.Phase != cloudv1.ReplayRunning && replay.Status.Phase != cloudv1.ReplayStopped {
		res, errs, retriableErrs = r.ensureInfrastructure(ctx, logger, c, replay)
		if errs != nil || retriableErrs != nil {
			logger.Error(fmt.Errorf("infrastructure deployment failed"), "Cannot activate replay")
			return res, errs, retriableErrs
		}
		replay.Status.Phase = cloudv1.ReplayRunning
		replay.Status.LastUpdateTime = metav1.Now()
	}

	// Only start tasks if phase is Running or Stopped
	if replay.Status.Phase != cloudv1.ReplayRunning && replay.Status.Phase != cloudv1.ReplayStopped {
		return res, errs, retriableErrs
	}

	// Sync, start, and sync again
	r.syncTaskStatusesBestEffort(ctx, logger, c, replay, envProvider, "before starting")

	if err := r.startReplayTasks(ctx, logger, c, replay, remote, envProvider); err != nil {
		logger.Error(err, "Failed to start replay tasks")
		replay.Status.Message = fmt.Sprintf("Failed to start tasks: %v", err)
		replay.Status.LastUpdateTime = metav1.Now()
		retriableErrs = err
	} else {
		// Update phase to Running after successfully starting tasks
		replay.Status.Phase = cloudv1.ReplayRunning
		replay.Status.Message = "Replay tasks started successfully"
		replay.Status.LastUpdateTime = metav1.Now()
	}

	r.syncTaskStatusesBestEffort(ctx, logger, c, replay, envProvider, "after start")

	// Requeue after 5 minutes to sync task status periodically
	// This reduces the reconcile frequency for running replay tasks
	return ctrl.Result{RequeueAfter: 2 * time.Minute}, errs, retriableErrs
}

// handlePauseState handles the Pause state - stops running tasks
func (r *Reconciler) handlePauseState(
	ctx context.Context,
	logger logr.Logger,
	c *task.Context,
	replay *cloudv1.Replay,
	remote client.Client,
	envProvider dataplane.EnvProvider,
) (ctrl.Result, error, error) {
	if replay.Status.Phase == cloudv1.ReplayStopped {
		logger.Info("Replay already in stopped state, skipping reconcile")
		return ctrl.Result{}, nil, nil
	}

	var err error
	r.syncTaskStatusesBestEffort(ctx, logger, c, replay, envProvider, "before stopping")

	if replay.Status.Phase == cloudv1.ReplayRunning {
		if err2 := r.stopReplayTasks(ctx, logger, c, replay, remote, envProvider); err2 != nil {
			logger.Error(err2, "Failed to stop replay tasks")
			replay.Status.Message = fmt.Sprintf("Failed to stop tasks: %v", err2)
			replay.Status.LastUpdateTime = metav1.Now()
			err = err2
		} else {
			replay.Status.Phase = cloudv1.ReplayStopped
			replay.Status.Message = "Replay tasks stopped successfully"
			replay.Status.LastUpdateTime = metav1.Now()
		}
	} else {
		logger.Info("Replay tasks already stopped or not running", "phase", replay.Status.Phase)
		replay.Status.Phase = cloudv1.ReplayStopped
		replay.Status.LastUpdateTime = metav1.Now()
	}

	r.syncTaskStatusesBestEffort(ctx, logger, c, replay, envProvider, "after stop")

	return ctrl.Result{}, err, nil
}

// ensureInfrastructure deploys infrastructure if not already deployed
func (r *Reconciler) ensureInfrastructure(
	ctx context.Context,
	logger logr.Logger,
	c *task.Context,
	replay *cloudv1.Replay,
) (ctrl.Result, error, error) {
	tasks := c.Provider.Tasks()
	deployer := deploy.NewDeployer(tasks, nil, logger)
	results := deployer.Deploy(ctx)
	return deploy.MinRequeue(results)
}

// syncTaskStatusesBestEffort syncs task statuses, logging errors without failing
func (r *Reconciler) syncTaskStatusesBestEffort(
	ctx context.Context,
	logger logr.Logger,
	c *task.Context,
	replay *cloudv1.Replay,
	envProvider dataplane.EnvProvider,
	phase string,
) {
	logger.Info("Syncing task status", "phase", phase)
	if err := r.syncTaskStatuses(ctx, logger, c, replay, envProvider); err != nil {
		logger.Error(err, "Failed to sync task status", "phase", phase)
	}
}

// updateReplayCR updates both spec and status of the Replay CR
func (r *Reconciler) updateReplayCR(ctx context.Context, replay *cloudv1.Replay) error {
	replayStatus := replay.Status.DeepCopy()
	if err := r.Client.Update(ctx, replay); err != nil {
		return err
	}
	replay.Status = *replayStatus
	return r.Client.Status().Update(ctx, replay)
}

func (r *Reconciler) updateReplayDeletionMetadata(ctx context.Context, replay *cloudv1.Replay) error {
	return client.IgnoreNotFound(r.Client.Update(ctx, replay))
}

func (r *Reconciler) updateReplayDeletionStatus(ctx context.Context, replay *cloudv1.Replay) error {
	return client.IgnoreNotFound(r.Client.Status().Update(ctx, replay))
}

func (r *Reconciler) cleanupGeneratedKubeConfigSecret(ctx context.Context, replay *cloudv1.Replay) error {
	if replay == nil || replay.Spec.Kubernetes.KubeConfigSecretRef == nil {
		return nil
	}

	ref := replay.Spec.Kubernetes.KubeConfigSecretRef
	if ref.Namespace != replay.Namespace || ref.Name != fmt.Sprintf("%s-kubeconfig", replay.Name) {
		return nil
	}

	secret := &corev1.Secret{
		ObjectMeta: metav1.ObjectMeta{
			Name:      ref.Name,
			Namespace: ref.Namespace,
		},
	}
	if err := r.Client.Delete(ctx, secret); err != nil && !apierrors.IsNotFound(err) {
		return fmt.Errorf("failed to delete generated kubeconfig secret %s/%s: %w", ref.Namespace, ref.Name, err)
	}
	return nil
}

// startReplayTasks calls HTTP API to start replay tasks for all StatefulSet pods
// Uses Kubernetes proxy to access pods
// Checks task status before starting to avoid duplicate starts
func (r *Reconciler) startReplayTasks(
	ctx context.Context,
	logger logr.Logger,
	c *task.Context,
	replay *cloudv1.Replay,
	remote client.Client,
	envProvider dataplane.EnvProvider,
) error {
	secret, err := r.getPasswordSecret(ctx, replay)
	if err != nil {
		return err
	}

	clientset, err := r.getClientset(envProvider)
	if err != nil {
		return err
	}

	namespace := c.GetReplayNameSpace()
	taskErrors := make(map[string][]string)
	startedCount, skippedCount, failedCount := 0, 0, 0

	// Start replay tasks for each task and each replica
	for _, replayTask := range replay.Spec.Tasks {
		password, ok := secret.Data[replayTask.Username]
		if !ok {
			errMsg := fmt.Sprintf("password not found for username '%s' in secret %s/%s",
				replayTask.Username, replay.Spec.PasswordSecretRef.Namespace, replay.Spec.PasswordSecretRef.Name)
			logger.Error(fmt.Errorf("%s", errMsg), "Skipping task", "task", replayTask.Name)
			taskErrors[replayTask.Name] = append(taskErrors[replayTask.Name], errMsg)
			continue
		}

		inputPaths, err := r.buildInputPaths(replayTask, logger)
		if err != nil {
			taskErrors[replayTask.Name] = append(taskErrors[replayTask.Name], err.Error())
			continue
		}
		logger.Info("Built input paths for task", "task", replayTask.Name, "inputPaths", inputPaths)

		// Start each replica
		replicas := getEffectiveReplicas(replayTask.Replicas)
		for i := int32(0); i < replicas; i++ {
			podName := task.GeneratePodName(replay.Name, replayTask.Name, i)
			replayClient := common.NewReplayClientViaKubeProxy(
				k8s.NewHttpLinkByPodProxy(ctx, clientset, namespace, podName, strconv.Itoa(task.DefaultSvcPort)))

			// Check if already running
			if status, err := replayClient.GetTaskStatus(); err == nil && status != nil && status.Status == common.TaskStatusRunning {
				logger.Info("Task already running, skipping", "pod", podName, "task", replayTask.Name)
				skippedCount++
				continue
			}

			// Build and send create task request
			// Use replica-specific config if available, otherwise use all configs for merging
			replicaConfigs := replayTask.ReplayConfigs
			if int(i) < len(replayTask.ReplayConfigs) {
				// Use config for this specific replica
				replicaConfigs = []cloudv1.ReplayConfig{replayTask.ReplayConfigs[i]}
			}
			req := r.buildCreateTaskRequest(replay, replayTask, inputPaths, password, replicaConfigs)
			logger.Info("Starting replay task via Pod proxy", "pod", podName, "task", replayTask.Name, "namespace", namespace, "replicaIndex", i)
			if err := replayClient.CreateTask(req, logger); err != nil {
				logger.Error(err, "Failed to start replay task", "pod", podName, "task", replayTask.Name)
				failedCount++
				taskErrors[replayTask.Name] = append(taskErrors[replayTask.Name], fmt.Sprintf("Pod %s: %v", podName, err))
				continue
			}
			logger.Info("Successfully started replay task", "pod", podName, "task", replayTask.Name)
			startedCount++
		}
	}

	updateTaskErrorMessages(replay, taskErrors)

	logger.Info("Start replay tasks summary", "started", startedCount, "skipped", skippedCount, "failed", failedCount)

	if failedCount > 0 {
		return fmt.Errorf("failed to start any replay tasks: %d failures", failedCount)
	}
	return nil
}

// getPasswordSecret retrieves the password secret
func (r *Reconciler) getPasswordSecret(ctx context.Context, replay *cloudv1.Replay) (*corev1.Secret, error) {
	secret := &corev1.Secret{}
	err := r.Client.Get(ctx, client.ObjectKey{
		Namespace: replay.Spec.PasswordSecretRef.Namespace,
		Name:      replay.Spec.PasswordSecretRef.Name,
	}, secret)
	if err != nil {
		return nil, fmt.Errorf("failed to get password secret: %w", err)
	}
	return secret, nil
}

// buildInputPaths constructs input paths from audit log addresses.
// For Aliyun, replay runtime expects an S3-compatible OSS URL that is already
// consumable by the downstream storage parser, so we pass BucketURL through as-is
// instead of appending role-based auth parameters.
func (r *Reconciler) buildInputPaths(replayTask cloudv1.ReplayTask, logger logr.Logger) ([]string, error) {
	inputPaths := []string{}
	for _, addr := range replayTask.SourceAuditLogAddrs {
		if addr.AWS != nil {
			s3URL := addr.AWS.S3URL
			if s3URL == "" {
				logger.Error(fmt.Errorf("s3URL is empty"), "Invalid audit log configuration", "task", replayTask.Name)
				continue
			}
			if addr.AWS.RoleArn != "" {
				s3URL = fmt.Sprintf("%s?role-arn=%s", s3URL, addr.AWS.RoleArn)
				if addr.AWS.ExternalID != nil {
					s3URL = fmt.Sprintf("%s&external-id=%s", s3URL, *addr.AWS.ExternalID)
				}
			}
			inputPaths = append(inputPaths, s3URL)
		}
		if addr.Aliyun != nil {
			if addr.Aliyun.BucketURL == "" {
				logger.Error(fmt.Errorf("bucketURL is empty"), "Invalid aliyun audit log configuration", "task", replayTask.Name)
				continue
			}
			inputPaths = append(inputPaths, addr.Aliyun.BucketURL)
		}
	}

	if len(inputPaths) == 0 {
		return nil, fmt.Errorf("task %s: no valid audit log addresses configured in SourceAuditLogAddrs", replayTask.Name)
	}
	return inputPaths, nil
}

// buildCreateTaskRequest constructs a CreateTaskRequest from replay configuration
// replicaConfigs should contain the config(s) for this specific replica
func (r *Reconciler) buildCreateTaskRequest(
	replay *cloudv1.Replay,
	replayTask cloudv1.ReplayTask,
	inputPaths []string,
	password []byte,
	replicaConfigs []cloudv1.ReplayConfig,
) common.CreateTaskRequest {
	mergedConfig := r.mergeReplayConfig(replay.Spec.ReplayConfig, replicaConfigs)

	req := common.CreateTaskRequest{
		Addr:            replayTask.TargetEndpoint,
		Input:           strings.Join(inputPaths, ","),
		Username:        replayTask.Username,
		Password:        string(password),
		Format:          mergedConfig.Format,
		ReadOnly:        mergedConfig.ReadOnly,
		IgnoreErrs:      mergedConfig.IgnoreErrs,
		PsClose:         mergedConfig.PsClose,
		CheckpointPath:  mergedConfig.CheckpointPath,
		BufSize:         int(mergedConfig.BufSize),
		StartTime:       mergedConfig.StartTime,
		CmdStartTime:    mergedConfig.CommandStartTime,
		CmdEndTime:      mergedConfig.CommandEndTime,
		DryRun:          mergedConfig.DryRun,
		DynamicInput:    mergedConfig.DynamicInput,
		ReplayerCount:   mergedConfig.ReplayerCount,
		ReplayerIndex:   mergedConfig.ReplayerIndex,
		OutputPath:      mergedConfig.OutputPath,
		CustomArguments: mergedConfig.CustomArguments,
	}

	// Use default checkpoint path if not specified
	if req.CheckpointPath == "" {
		req.CheckpointPath = "/data/checkpoint_" + req.CmdEndTime + ".json"
	}
	if req.DryRun {
		req.CheckpointPath += ".dryrun"
	}

	// Set optional speed
	if mergedConfig.Speed != "" {
		if speed, err := strconv.ParseFloat(mergedConfig.Speed, 64); err == nil {
			req.Speed = speed
		}
	}

	return req
}

// stopReplayTasks calls HTTP API to stop/cancel replay tasks for all StatefulSet pods
// Checks task status before stopping to avoid unnecessary stop calls
func (r *Reconciler) stopReplayTasks(
	ctx context.Context,
	logger logr.Logger,
	c *task.Context,
	replay *cloudv1.Replay,
	remote client.Client,
	envProvider dataplane.EnvProvider,
) error {
	clientset, err := r.getClientset(envProvider)
	if err != nil {
		return err
	}

	namespace := c.GetReplayNameSpace()
	taskErrors := make(map[string][]string)
	stoppedCount, skippedCount, failedCount := 0, 0, 0

	// Stop replay tasks for each task and each replica
	for _, replayTask := range replay.Spec.Tasks {
		replicas := getEffectiveReplicas(replayTask.Replicas)
		for i := int32(0); i < replicas; i++ {
			podName := task.GeneratePodName(replay.Name, replayTask.Name, i)
			replayClient := common.NewReplayClientViaKubeProxy(
				k8s.NewHttpLinkByPodProxy(ctx, clientset, namespace, podName, strconv.Itoa(task.DefaultSvcPort)))

			// Check if already stopped
			if status, err := replayClient.GetTaskStatus(); err == nil && status != nil {
				if status.Status == common.TaskStatusCanceled || status.Status == common.TaskStatusDone {
					logger.Info("Task already stopped, skipping", "pod", podName, "task", replayTask.Name, "status", status.Status)
					skippedCount++
					continue
				}
			}

			// Stop the task
			req := common.CancelTaskRequest{
				Type:     common.TaskTypeReplay,
				Graceful: true,
			}
			logger.Info("Stopping replay task via Pod proxy", "pod", podName, "task", replayTask.Name, "namespace", namespace)
			_, err := replayClient.CancelTask(req)
			if err != nil {
				logger.Error(err, "Failed to stop replay task", "pod", podName, "task", replayTask.Name)
				failedCount++
				taskErrors[replayTask.Name] = append(taskErrors[replayTask.Name], fmt.Sprintf("Pod %s: %v", podName, err))
				continue
			}
			logger.Info("Successfully stopped replay task", "pod", podName, "task", replayTask.Name)
			stoppedCount++
		}
	}

	updateTaskErrorMessages(replay, taskErrors)

	logger.Info("Stop replay tasks summary", "stopped", stoppedCount, "skipped", skippedCount, "failed", failedCount)

	if failedCount > 0 && stoppedCount == 0 && skippedCount == 0 {
		return fmt.Errorf("failed to stop any replay tasks: %d failures", failedCount)
	}
	return nil
}

// syncTaskStatuses queries the actual status of all replay tasks via HTTP API
// and updates the Replay CR's status.Tasks field accordingly
func (r *Reconciler) syncTaskStatuses(
	ctx context.Context,
	logger logr.Logger,
	c *task.Context,
	replay *cloudv1.Replay,
	envProvider dataplane.EnvProvider,
) error {
	clientset, err := r.getClientset(envProvider)
	if err != nil {
		return err
	}

	namespace := c.GetReplayNameSpace()
	taskStatusMap := make(map[string]*cloudv1.ReplayTaskStatus)

	// Query status for each task
	for _, replayTask := range replay.Spec.Tasks {
		taskStatus := r.syncTaskStatus(ctx, logger, clientset, namespace, replay, replayTask)
		taskStatusMap[replayTask.Name] = taskStatus
	}

	// Update replay.Status.Tasks, merging with existing error messages
	existingTaskStatusMap := make(map[string]cloudv1.ReplayTaskStatus)
	for _, existingTask := range replay.Status.Tasks {
		existingTaskStatusMap[existingTask.Name] = existingTask
	}

	taskStatuses := make([]cloudv1.ReplayTaskStatus, 0, len(replay.Spec.Tasks))
	for _, replayTask := range replay.Spec.Tasks {
		if status, ok := taskStatusMap[replayTask.Name]; ok {
			if existingStatus, exists := existingTaskStatusMap[replayTask.Name]; exists {
				if existingStatus.Message != "" {
					// If task has existing error message, keep it unless task is now running
					if status.Message == "" {
						// Keep existing error message if new sync has no message
						status.Message = existingStatus.Message
					}
				}
			}
			taskStatuses = append(taskStatuses, *status)
		}
	}
	replay.Status.Tasks = taskStatuses

	return nil
}

// syncTaskStatus syncs status for a single task
func (r *Reconciler) syncTaskStatus(
	ctx context.Context,
	logger logr.Logger,
	clientset *kubernetes.Clientset,
	namespace string,
	replay *cloudv1.Replay,
	replayTask cloudv1.ReplayTask,
) *cloudv1.ReplayTaskStatus {
	replicas := getEffectiveReplicas(replayTask.Replicas)

	taskStatus := &cloudv1.ReplayTaskStatus{
		Name: replayTask.Name,
	}

	replicaStatuses := make([]cloudv1.ReplayReplicaStatus, 0, replicas)

	for i := int32(0); i < replicas; i++ {
		podName := task.GeneratePodName(replay.Name, replayTask.Name, i)
		replicaStatus := r.syncReplicaStatus(ctx, logger, clientset, namespace, podName, i, replayTask)

		replicaStatuses = append(replicaStatuses, replicaStatus)

		logger.Info("Pod task status",
			"pod", podName,
			"phase", replicaStatus.Phase,
			"ready", replicaStatus.Ready)
	}
	taskStatus.ReplicaStatuses = replicaStatuses
	return taskStatus
}

// syncReplicaStatus syncs status for a single replica
func (r *Reconciler) syncReplicaStatus(
	ctx context.Context,
	logger logr.Logger,
	clientset *kubernetes.Clientset,
	namespace string,
	podName string,
	index int32,
	replayTask cloudv1.ReplayTask,
) cloudv1.ReplayReplicaStatus {
	replicaStatus := cloudv1.ReplayReplicaStatus{
		PodName: podName,
		Index:   index,
		Phase:   cloudv1.ReplayTaskPhasePending,
		Ready:   false,
	}

	// Query task status via Pod proxy
	httpLink := k8s.NewHttpLinkByPodProxy(ctx, clientset, namespace, podName, strconv.Itoa(task.DefaultSvcPort))
	replayClient := common.NewReplayClientViaKubeProxy(httpLink)

	status, err := replayClient.GetTaskStatus()
	if err != nil {
		logger.Error(err, "Failed to get task status", "pod", podName, "task", replayTask.Name)
		replicaStatus.Message = err.Error()
		return replicaStatus
	}

	// Populate task info
	replicaStatus.TaskInfo = &cloudv1.ReplayTaskInfo{
		Type:                 status.Type,
		Status:               status.Status,
		StartTime:            status.StartTime,
		EndTime:              status.EndTime,
		LastCommandTimestamp: status.LastCmdTs,
		Error:                status.Error,
		SourceAuditLogAddrs:  replayTask.SourceAuditLogAddrs,
	}

	// Update replica phase based on status
	switch status.Status {
	case common.TaskStatusRunning:
		replicaStatus.Phase = cloudv1.ReplayTaskPhaseRunning
		replicaStatus.Ready = true
	case common.TaskStatusCanceled, common.TaskStatusDone:
		replicaStatus.Phase = cloudv1.ReplayTaskPhaseStopped
		replicaStatus.Ready = false
	default:
		logger.Info("Unknown task status", "pod", podName, "status", status.Status)
		replicaStatus.Phase = cloudv1.ReplayTaskPhasePending
	}

	return replicaStatus
}

// mergeReplayConfig merges global ReplayConfig with task-level ReplayConfigs
// Task-level configs have higher priority and will override global config
// If taskConfigs is empty or nil, returns the global config
func (r *Reconciler) mergeReplayConfig(globalConfig cloudv1.ReplayConfig, taskConfigs []cloudv1.ReplayConfig) cloudv1.ReplayConfig {
	// Start with global config as base
	merged := globalConfig

	// If no task-level configs, return global config
	if len(taskConfigs) == 0 {
		return merged
	}

	// Apply task-level configs in order (later configs override earlier ones)
	// We only override if the field is non-empty/non-zero, so we preserve the "omitempty" semantic
	for _, taskConfig := range taskConfigs {
		// Override non-empty string fields
		if taskConfig.Format != "" {
			merged.Format = taskConfig.Format
		}
		if taskConfig.PsClose != "" {
			merged.PsClose = taskConfig.PsClose
		}
		if taskConfig.Image != "" {
			merged.Image = taskConfig.Image
		}
		if taskConfig.LogFile != "" {
			merged.LogFile = taskConfig.LogFile
		}
		if taskConfig.Speed != "" {
			merged.Speed = taskConfig.Speed
		}
		if taskConfig.CommandStartTime != "" {
			merged.CommandStartTime = taskConfig.CommandStartTime
		}
		if taskConfig.CommandEndTime != "" {
			merged.CommandEndTime = taskConfig.CommandEndTime
		}
		if taskConfig.StartTime != "" {
			merged.StartTime = taskConfig.StartTime
		}
		if taskConfig.CheckpointPath != "" {
			merged.CheckpointPath = taskConfig.CheckpointPath
		}
		if taskConfig.OutputPath != "" {
			merged.OutputPath = taskConfig.OutputPath
		}

		// Override non-zero numeric fields
		if taskConfig.BufSize != 0 {
			merged.BufSize = taskConfig.BufSize
		}
		if taskConfig.ReplayerCount != 0 {
			merged.ReplayerCount = taskConfig.ReplayerCount
		}
		if taskConfig.ReplayerIndex != 0 {
			merged.ReplayerIndex = taskConfig.ReplayerIndex
		}

		// If task config has the field set to true, override the merged config
		if taskConfig.IgnoreErrs {
			merged.IgnoreErrs = true
		}
		if taskConfig.ReadOnly {
			merged.ReadOnly = true
		}
		if taskConfig.DryRun {
			merged.DryRun = true
		}
		if taskConfig.DynamicInput {
			merged.DynamicInput = true
		}

		if taskConfig.CustomArguments != nil {
			merged.CustomArguments = taskConfig.CustomArguments
		}
	}

	return merged
}

func (r *Reconciler) SetupWithManager(mgr ctrl.Manager, preds ...predicate.Predicate) error {
	// Add index to Replay object, used for listing Replay by k8s name
	if err := mgr.GetFieldIndexer().IndexField(context.Background(), &cloudv1.Replay{}, "spec.kubernetes.name", func(obj client.Object) []string {
		replay := obj.(*cloudv1.Replay)
		return []string{replay.Spec.Kubernetes.Name}
	}); err != nil {
		return err
	}

	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.Replay{}, builder.WithPredicates(preds...)).
		WithOptions(controller.Options{RateLimiter: k8s.RetryLimiter, MaxConcurrentReconciles: 10}).
		Complete(r)
}