package impl

import (
	"context"

	"github.com/AlekSi/pointer"
	"github.com/tidbcloud/dataflow-service/app/models/bucket"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo"
	backupRepo "github.com/tidbcloud/dataflow-service/app/repo/backup"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	"github.com/tidbcloud/dataflow-service/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db"
	"go.uber.org/zap"
)

// synchronousRTCModifier modifies S3 RTC synchronously: calls the cloud API directly
// then commits all DB changes in a single transaction, without checking BackupSetting status.
type synchronousRTCModifier struct {
	dbProvider                 db.TransProvider
	backupSettingRepo          backupRepo.BackupSettingRepo
	bucketRepo                 repo.BucketRepo
	awsBucketManagerModel      bucket.AWSBucketManagerModel
	awsBucketConfigurationRepo repo.AWSBucketConfigurationRepo
	logBackupRepo              backupRepo.LogBackupRepo
}

func (s *synchronousRTCModifier) execute(ctx context.Context, req *domain.ReplicationRTCModificationReq) bizerr.BizError {
	logger := log.GetLogger(ctx)
	backupSetting, masterBucket, awsBucketConfig, err := s.loadAndValidate(ctx, logger, req.ClusterID, req.BucketID)
	if err != nil {
		return err
	}

	if req.DryRun {
		logger.Info("dry run for modifying replication RTC",
			zap.Uint64("clusterID", req.ClusterID), zap.Bool("enable", req.Enable))
		return nil
	}

	return s.applyRTCChange(ctx, logger, backupSetting, masterBucket, awsBucketConfig, req.Enable)
}

func (s *synchronousRTCModifier) loadAndValidate(
	ctx context.Context, logger *zap.Logger, clusterID uint64, bucketID uint64,
) (*entity.BackupSetting, *entity.Bucket, *entity.AWSBucketConfiguration, bizerr.BizError) {
	backupSetting, err := s.backupSettingRepo.Get(ctx,
		func(qs entity.BackupSettingQuerySet) entity.BackupSettingQuerySet {
			return qs.ClusterIDEq(entity.ID(clusterID))
		})
	if err != nil {
		logger.Warn("load backup setting error", zap.Error(err))
		return nil, nil, nil, err
	}
	if backupSetting.Bucket == nil || backupSetting.Bucket.Prefix == nil {
		logger.Error("no bucket config or prefix, nothing to do", zap.Uint64("clusterID", clusterID))
		return nil, nil, nil, bizerr_wrapper.NewInternalErr("no bucket prefix in backup setting for cluster %d", clusterID)
	}

	bucket, bizErr := s.bucketRepo.Load(ctx, bucketID)
	if bizErr != nil {
		logger.Warn("load bucket for project validation error",
			zap.Uint64("bucketID", bucketID), zap.Error(bizErr))
		return nil, nil, nil, bizErr
	}
	if bucket.Uint64ProjectID() != uint64(backupSetting.ProjectID) {
		logger.Error("bucket does not belong to cluster project",
			zap.Uint64("bucketID", bucketID),
			zap.Uint64("projectID", uint64(backupSetting.ProjectID)))
		return nil, nil, nil, bizerr_wrapper.NewInvalidInputError(
			"bucket %d does not belong to project of cluster %d", bucketID, clusterID)
	}

	awsBucketConfig, err := s.awsBucketConfigurationRepo.Get(ctx,
		func(qs entity.AWSBucketConfigurationQuerySet) entity.AWSBucketConfigurationQuerySet {
			return qs.BucketIDEq(entity.ID(bucketID))
		})
	if err != nil {
		logger.Warn("load AWS bucket configuration error", zap.Error(err))
		return nil, nil, nil, err
	}
	return backupSetting, bucket, awsBucketConfig, nil
}

func (s *synchronousRTCModifier) applyRTCChange(
	ctx context.Context, logger *zap.Logger,
	backupSetting *entity.BackupSetting, masterBucket *entity.Bucket, awsBucketConfig *entity.AWSBucketConfiguration,
	enable bool,
) bizerr.BizError {
	logger.Info("modifying RTC on replication rule",
		zap.Uint64("backupSettingId", uint64(backupSetting.ID)),
		zap.String("ruleID", *backupSetting.Bucket.Prefix),
		zap.Bool("enable", enable))

	// Call the cloud API first (outside transaction — network call cannot be transactional).
	if modErr := s.awsBucketManagerModel.ModifyReplicationRTC(ctx, &domain.ModifyReplicationRTCReq{
		AWSBucketReq: domain.AWSBucketReq{
			Name:         masterBucket.Name,
			AWSAccountID: *masterBucket.CloudAccountID,
			Region:       masterBucket.Regions[0],
		},
		RuleID:    *backupSetting.Bucket.Prefix,
		EnableRTC: enable,
	}); modErr != nil {
		logger.Warn("failed to modify RTC on replication rule", zap.Error(modErr))
		return modErr
	}

	// Commit all DB updates atomically.
	err := s.dbProvider.Transaction(ctx, func(txCtx context.Context) error {
		awsBucketConfig.RTCEnabled = pointer.To(enable)
		if saveErr := s.awsBucketConfigurationRepo.Save(txCtx, awsBucketConfig); saveErr != nil {
			logger.Warn("failed to save AWSBucketConfiguration", zap.Error(saveErr))
			return saveErr
		}
		masterBucket.SetRTCEnabled(enable)
		if saveErr := s.bucketRepo.Save(txCtx, masterBucket); saveErr != nil {
			logger.Warn("failed to save bucket RTCEnabled", zap.Error(saveErr))
			return saveErr
		}
		backupSetting.Bucket.RTCEnabled = pointer.To(enable)
		if saveErr := s.backupSettingRepo.Save(txCtx, backupSetting); saveErr != nil {
			logger.Warn("failed to save BackupSetting Bucket.RTCEnabled", zap.Error(saveErr))
			return saveErr
		}
		return s.updateRunningLogBackupSnapshots(txCtx, logger, uint64(backupSetting.ClusterID), enable)
	})
	if err != nil {
		logger.Warn("failed to apply replication RTC change", zap.Error(err))
		return bizerr_wrapper.WrapGormError(err, "failed to apply replication RTC change")
	}
	return nil
}

func (s *synchronousRTCModifier) updateRunningLogBackupSnapshots(
	ctx context.Context, logger *zap.Logger, clusterID uint64, enable bool,
) error {
	logBackups, err := s.logBackupRepo.List(ctx,
		func(qs entity.LogBackupQuerySet) entity.LogBackupQuerySet {
			return qs.ClusterIDEq(entity.ID(clusterID)).
				DeletedAtIsNull().
				StatusNotIn(entity.LogBackupStatusStopped, entity.LogBackupStatusFailed)
		},
	)
	if err != nil {
		logger.Warn("list active log backups failed", zap.Uint64("clusterID", clusterID), zap.Error(err))
		return err
	}
	for i := range logBackups {
		lb := &logBackups[i]
		if lb.BucketSettingSnapshot == nil {
			continue
		}
		lb.BucketSettingSnapshot.RTCEnabled = &enable
		if saveErr := s.logBackupRepo.Save(ctx, lb); saveErr != nil {
			logger.Warn("save log backup snapshot failed",
				zap.Uint64("logBackupID", uint64(lb.ID)),
				zap.Bool("enable", enable), zap.Error(saveErr))
			return saveErr
		}
		logger.Info("updated log backup RTC snapshot",
			zap.Uint64("logBackupID", uint64(lb.ID)), zap.Bool("enable", enable))
	}
	return nil
}

func (m *backupSettingModel) ModifyReplicationRTC(ctx context.Context, req *domain.ReplicationRTCModificationReq) bizerr.BizError {
	logger := log.GetLogger(ctx)
	logger.Info("modifying replication RTC synchronously",
		zap.Uint64("clusterID", req.ClusterID), zap.Bool("enable", req.Enable))
	return m.synchronousRTCModifier.execute(ctx, req)
}