package cluster

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/samber/lo"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"

	globaldedicatedv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/dedicated/v1"
	"github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/entity"
	configbackuprepo "github.com/tidbcloud/tidb-management-service/dedicated-global-service/domain/repository/configbackup"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
)

const (
	defaultClusterConfigBackupIntervalSec = int32(86400)    // 1 day
	minClusterConfigBackupIntervalSec     = int32(600)      // 10 minutes
	maxClusterConfigBackupIntervalSec     = int32(31536000) // 365 days

	enableConcurrentConfigBackup = true
	maxConcurrentConfigBackup    = 10
	singleConfigBackupTimeout    = 5 * time.Minute
)

func (s *serviceImpl) EnableClusterConfigBackup(
	ctx context.Context,
	req *globaldedicatedv1.EnableClusterConfigBackupRequest,
) (*globaldedicatedv1.ClusterConfigBackupPolicy, bizerr.BizError) {
	if req.GetClusterId() == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	intervalSec := req.GetIntervalSec()
	if intervalSec == 0 {
		intervalSec = defaultClusterConfigBackupIntervalSec
	}
	if intervalSec < minClusterConfigBackupIntervalSec || intervalSec > maxClusterConfigBackupIntervalSec {
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"interval_sec must be between %d and %d",
			minClusterConfigBackupIntervalSec,
			maxClusterConfigBackupIntervalSec,
		)
	}

	cluster, err := s.clusterRepo.Get(ctx, req.GetClusterId(), false)
	if err != nil {
		return nil, bizerr.WrapError(err, "get cluster failed")
	}
	if !isClusterAvailableForConfigBackup(cluster.State) {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster %d is not available", req.GetClusterId())
	}

	now := time.Now()

	if req.GetValidateOnly() {
		// For validate only, we still fetch the existing policy to return what it would look like
		policy, getErr := s.configBackupPolicyRepo.GetByClusterID(ctx, req.GetClusterId())
		if getErr != nil && getErr.ErrCode() != bizerr.ErrCodeRecordNotFound {
			return nil, bizerr.WrapError(getErr, "get cluster config backup policy failed")
		}
		if policy == nil {
			policy = &entity.ClusterConfigBackupPolicy{
				ClusterID:   req.GetClusterId(),
				Enabled:     true,
				IntervalSec: intervalSec,
				NextRunAt:   lo.ToPtr(now),
			}
		} else {
			wasEnabled := policy.Enabled
			policy.Enabled = true
			policy.IntervalSec = intervalSec
			policy.DeleteBackupPending = false
			policy.DeletePolicyPending = false
			if !wasEnabled {
				policy.NextRunAt = lo.ToPtr(now)
			} else {
				policy.NextRunAt = lo.ToPtr(now.Add(time.Duration(intervalSec) * time.Second))
			}
		}
		return toClusterConfigBackupPolicyProto(policy), nil
	}

	if err := s.configBackupPolicyRepo.Enable(ctx, req.GetClusterId(), intervalSec, now); err != nil {
		return nil, bizerr.WrapError(err, "enable cluster config backup policy failed")
	}

	// Fetch the latest policy to return
	policy, err := s.configBackupPolicyRepo.GetByClusterID(ctx, req.GetClusterId())
	if err != nil {
		return nil, bizerr.WrapError(err, "get cluster config backup policy failed after enable")
	}

	return toClusterConfigBackupPolicyProto(policy), nil
}

func (s *serviceImpl) DisableClusterConfigBackup(
	ctx context.Context,
	req *globaldedicatedv1.DisableClusterConfigBackupRequest,
) (*globaldedicatedv1.ClusterConfigBackupPolicy, bizerr.BizError) {
	if req.GetClusterId() == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster_id is required")
	}

	policy, err := s.configBackupPolicyRepo.GetByClusterID(ctx, req.GetClusterId())
	if err != nil && err.ErrCode() != bizerr.ErrCodeRecordNotFound {
		return nil, bizerr.WrapError(err, "get cluster config backup policy failed")
	}

	// If the policy does not exist, it is already disabled, so we return directly without creating a dirty record
	if policy == nil {
		return &globaldedicatedv1.ClusterConfigBackupPolicy{
			ClusterId: req.GetClusterId(),
			Enabled:   false,
		}, nil
	}

	policy.Enabled = false
	policy.NextRunAt = nil

	if req.GetValidateOnly() {
		return toClusterConfigBackupPolicyProto(policy), nil
	}

	disableErr := s.configBackupPolicyRepo.Disable(ctx, req.GetClusterId(), &configbackuprepo.DisableOption{
		DeleteBackupPending: !req.GetRetainBackup(),
		DeletePolicyPending: false,
	})
	if disableErr != nil {
		return nil, bizerr.WrapError(disableErr, "disable cluster config backup policy failed")
	}

	return toClusterConfigBackupPolicyProto(policy), nil
}

func (s *serviceImpl) RunDueClusterConfigBackups(ctx context.Context, limit int) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = bizerr.New(bizerr.ErrCodePanic, "panic in RunDueClusterConfigBackups: %v", r)
			tilog.Error(ctx, "panic recovered in RunDueClusterConfigBackups", zap.Any("panic", r))
		}
	}()

	startTime := time.Now()
	tilog.Info(ctx, "RunDueClusterConfigBackups started", zap.Int("limit", limit))

	defer func() {
		tilog.Info(ctx, "RunDueClusterConfigBackups finished",
			zap.Duration("cost", time.Since(startTime)),
			zap.Error(err))
	}()

	if err := s.cleanupPendingClusterConfigBackups(ctx, limit); err != nil {
		// Log the error but do not block the core backup process
		tilog.Error(ctx, "cleanup pending cluster config backups failed", zap.Error(err))
	}

	now := time.Now()
	policies, err := s.configBackupPolicyRepo.ClaimDue(ctx, now, limit)
	if err != nil {
		return err
	}

	if len(policies) == 0 {
		return nil
	}

	if enableConcurrentConfigBackup {
		s.runClusterConfigBackupsConcurrently(ctx, policies, now)
	} else {
		s.runClusterConfigBackupsSequentially(ctx, policies, now)
	}

	return nil
}

func (s *serviceImpl) runClusterConfigBackupsSequentially(ctx context.Context, policies []*entity.ClusterConfigBackupPolicy, now time.Time) {
	for _, policy := range policies {
		s.runSingleClusterConfigBackup(ctx, policy.ClusterID, now)
	}
}

func (s *serviceImpl) runClusterConfigBackupsConcurrently(ctx context.Context, policies []*entity.ClusterConfigBackupPolicy, now time.Time) {
	sem := make(chan struct{}, maxConcurrentConfigBackup)
	var wg sync.WaitGroup

	tilog.Info(ctx, "start concurrent config backup",
		zap.Int("total_policies", len(policies)),
		zap.Int("concurrency_limit", maxConcurrentConfigBackup))

	for _, policy := range policies {
		p := policy // Explicitly assign to a local variable to avoid loop variable capture issues
		wg.Add(1)
		sem <- struct{}{} // Acquire token, block if channel is full

		go func() {
			defer wg.Done()
			defer func() { <-sem }() // Release token

			s.runSingleClusterConfigBackup(ctx, p.ClusterID, now)
		}()
	}

	tilog.Info(ctx, "all concurrent config backup tasks dispatched, waiting for completion")

	wg.Wait()

	tilog.Info(ctx, "all concurrent config backup tasks completed")
}

func (s *serviceImpl) cleanupPendingClusterConfigBackups(ctx context.Context, limit int) error {
	policies, err := s.configBackupPolicyRepo.ListPendingCleanup(ctx, limit)
	if err != nil {
		return err
	}

	for _, policy := range policies {
		if policy.DeleteBackupPending {
			if err := s.configBackupRepo.DeleteByClusterID(ctx, policy.ClusterID); err != nil &&
				err.ErrCode() != bizerr.ErrCodeRecordNotFound {
				tilog.Error(ctx, "delete pending config backup failed", zap.Uint64("cluster_id", policy.ClusterID), zap.Error(err))
				continue
			}
		}

		if policy.DeletePolicyPending {
			if err := s.configBackupPolicyRepo.DeleteByClusterID(ctx, policy.ClusterID); err != nil &&
				err.ErrCode() != bizerr.ErrCodeRecordNotFound {
				tilog.Error(ctx, "delete pending config backup policy failed", zap.Uint64("cluster_id", policy.ClusterID), zap.Error(err))
				continue
			}
			continue
		}

		if policy.DeleteBackupPending {
			if err := s.configBackupPolicyRepo.ClearDeleteBackupPending(ctx, policy.ClusterID); err != nil {
				tilog.Error(ctx, "update config backup policy pending status failed", zap.Uint64("cluster_id", policy.ClusterID), zap.Error(err))
				continue
			}
		}
	}

	return nil
}

func (s *serviceImpl) runSingleClusterConfigBackup(ctx context.Context, clusterID uint64, now time.Time) {
	defer func() {
		if r := recover(); r != nil {
			tilog.Error(ctx, "panic recovered in single cluster config backup",
				zap.Uint64("cluster_id", clusterID),
				zap.Any("panic", r))
			_ = s.configBackupPolicyRepo.RecordFailure(ctx, clusterID, now, fmt.Sprintf("panic: %v", r))
		}
	}()

	cluster, err := s.clusterRepo.Get(ctx, clusterID, false)
	if err != nil {
		tilog.Error(ctx, "get cluster failed during config backup", zap.Uint64("cluster_id", clusterID), zap.Error(err))
		if err.ErrCode() != bizerr.ErrCodeRecordNotFound {
			_ = s.configBackupPolicyRepo.RecordFailure(ctx, clusterID, now, err.Error())
		}
		return
	}

	if !isClusterAvailableForConfigBackup(cluster.State) {
		errMsg := bizerr.New(bizerr.ErrCodeParams, "cluster %d is not available for config backup", clusterID).Error()
		tilog.Info(ctx, "skip config backup due to cluster state", zap.Uint64("cluster_id", clusterID), zap.String("state", cluster.State.String()))
		_ = s.configBackupPolicyRepo.RecordFailure(ctx, clusterID, now, errMsg)
		return
	}

	// Create a timeout context for the core I/O operations to prevent goroutine leak and worker blocking
	opCtx, cancel := context.WithTimeout(ctx, singleConfigBackupTimeout)
	defer cancel()

	backup, err := s.captureClusterConfigBackup(opCtx, cluster)
	if err != nil {
		tilog.Error(ctx, "capture cluster config backup failed", zap.Uint64("cluster_id", clusterID), zap.Error(err))
		_ = s.configBackupPolicyRepo.RecordFailure(ctx, clusterID, now, err.Error())
		return
	}

	if err := s.configBackupRepo.Upsert(ctx, backup); err != nil {
		tilog.Error(ctx, "upsert cluster config backup failed", zap.Uint64("cluster_id", clusterID), zap.Error(err))
		_ = s.configBackupPolicyRepo.RecordFailure(ctx, clusterID, now, err.Error())
		return
	}

	if err := s.configBackupPolicyRepo.RecordSuccess(ctx, clusterID, now); err != nil {
		tilog.Error(ctx, "record cluster config backup success failed", zap.Uint64("cluster_id", clusterID), zap.Error(err))
		return
	}

	tilog.Info(ctx, "cluster config backup captured",
		zap.Uint64("cluster_id", clusterID),
		zap.Time("captured_at", backup.CapturedAt),
	)
}

func (s *serviceImpl) captureClusterConfigBackup(
	ctx context.Context,
	cluster *entity.Cluster,
) (*entity.ClusterConfigBackup, bizerr.BizError) {
	regionEn, errBiz := s.regionRepo.SelectRegionById(ctx, cluster.RegionId, cluster.OrganizationId)
	if errBiz != nil {
		return nil, bizerr.WrapError(errBiz, "select cluster region failed")
	}

	systemVariables, err := s.getClusterSystemVariables(ctx, cluster, "backup-source")
	if err != nil {
		return nil, bizerr.WrapError(err, "capture system variables failed")
	}
	pdConfig, err := s.getPdConfig(ctx, regionEn, cluster, "")
	if err != nil {
		return nil, bizerr.WrapError(err, "capture pd config failed")
	}
	pdServerConfig, err := s.getPdConfig(ctx, regionEn, cluster, "server")
	if err != nil {
		return nil, bizerr.WrapError(err, "capture pd server config failed")
	}
	placementRules, err := s.getNonTiflashPlacementRules(ctx, regionEn, cluster)
	if err != nil {
		return nil, bizerr.WrapError(err, "capture placement rules failed")
	}
	storeLimits, err := s.getStoreLimits(ctx, regionEn, cluster)
	if err != nil {
		return nil, bizerr.WrapError(err, "capture store limits failed")
	}
	newCollationEnabled := s.getNewCollationEnabledForBackup(ctx, cluster)

	return &entity.ClusterConfigBackup{
		ClusterID:           cluster.ClusterId,
		SystemVariables:     systemVariables,
		PdConfig:            pdConfig,
		PdServerConfig:      pdServerConfig,
		PlacementRules:      placementRules,
		StoreLimits:         storeLimits,
		NewCollationEnabled: newCollationEnabled,
		CapturedAt:          time.Now(),
	}, nil
}

func (s *serviceImpl) getNewCollationEnabledForBackup(ctx context.Context, cluster *entity.Cluster) *bool {
	newCollationEnabled, err := s.isNewCollationEnabled(ctx, cluster)
	if err != nil {
		tilog.Warn(ctx, "failed to capture new_collation_enabled for config backup",
			zap.Uint64("cluster_id", cluster.ClusterId),
			zap.Error(err),
		)
		return nil
	}

	return lo.ToPtr(newCollationEnabled)
}

func toClusterConfigBackupPolicyProto(policy *entity.ClusterConfigBackupPolicy) *globaldedicatedv1.ClusterConfigBackupPolicy {
	if policy == nil {
		return nil
	}
	out := &globaldedicatedv1.ClusterConfigBackupPolicy{
		Id:          policy.ID,
		ClusterId:   policy.ClusterID,
		Enabled:     policy.Enabled,
		IntervalSec: policy.IntervalSec,
		LastError:   policy.LastError,
	}
	if policy.NextRunAt != nil {
		out.NextRunAt = timestamppb.New(*policy.NextRunAt)
	}
	if policy.LastBackupAt != nil {
		out.LastBackupAt = timestamppb.New(*policy.LastBackupAt)
	}
	if policy.LastSuccessAt != nil {
		out.LastSuccessAt = timestamppb.New(*policy.LastSuccessAt)
	}
	return out
}

func isClusterAvailableForConfigBackup(state globaldedicatedv1.Cluster_State) bool {
	return state == globaldedicatedv1.Cluster_AVAILABLE ||
		state == globaldedicatedv1.Cluster_MODIFYING ||
		state == globaldedicatedv1.Cluster_SCALING
}