package impl

import (
	"context"
	"testing"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	bucket_mock "github.com/tidbcloud/dataflow-service/app/models/bucket/mock"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service/app/repo/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

// newTestRTCModifier constructs a synchronousRTCModifier with the given mocks.
// Pass nil for any mock that the test does not exercise.
func newTestRTCModifier(
	backupSettingRepo *repo_mock.BackupSettingRepo,
	bucketRepo *repo_mock.BucketRepo,
	awsBucketManagerModel *bucket_mock.AWSBucketManagerModel,
	awsBucketConfigRepo *repo_mock.AWSBucketConfigurationRepo,
	logBackupRepo *repo_mock.LogBackupRepo,
) synchronousRTCModifier {
	return synchronousRTCModifier{
		dbProvider:                 &fakeTransProvider{},
		backupSettingRepo:          backupSettingRepo,
		bucketRepo:                 bucketRepo,
		awsBucketManagerModel:      awsBucketManagerModel,
		awsBucketConfigurationRepo: awsBucketConfigRepo,
		logBackupRepo:              logBackupRepo,
	}
}

// makeBackupSettingWithMasterBucket returns a BackupSetting that passes loadAndValidate.
func makeBackupSettingWithMasterBucket(clusterID, bucketID uint64) *entity.BackupSetting {
	prefix := "test-prefix"
	return &entity.BackupSetting{
		Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
		ClusterID: entity.ID(clusterID),
		Bucket: &entity.BucketSetting{
			Prefix: &prefix,
			Buckets: []entity.BackupSettingBucket{
				{Master: true, BucketId: bucketID, Regions: []string{"us-east-1"}},
			},
		},
	}
}

// makeValidBucket returns a Bucket entity with the minimum fields needed by applyRTCChange.
func makeValidBucket(bucketID uint64) *entity.Bucket {
	return &entity.Bucket{
		Updatable:      entity.Updatable{Base: entity.Base{ID: entity.ID(bucketID)}},
		Name:           "test-bucket",
		ProjectID:      "0",
		CloudAccountID: pointer.To("123456789012"),
		Regions:        entity.List{"us-east-1"},
		Extension:      entity.JSONMap{},
	}
}

// ─── execute: validation error paths ───────────────────────────────────────

func TestSynchronousRTCModifier_Execute_BackupSettingLoadFails(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	modifier := newTestRTCModifier(backupSettingRepo, nil, nil, nil, nil)

	loadErr := bizerr.New(bizerr.ErrCodeDB, "db error")
	backupSettingRepo.On("Get", mock.Anything, mock.Anything).Return(nil, loadErr).Once()

	err := modifier.execute(context.Background(), &domain.ReplicationRTCModificationReq{ClusterID: 1})
	require.Error(t, err)
	assert.Equal(t, loadErr, err)
}

func TestSynchronousRTCModifier_Execute_NilBucket(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	modifier := newTestRTCModifier(backupSettingRepo, nil, nil, nil, nil)

	backupSettingRepo.On("Get", mock.Anything, mock.Anything).Return(
		&entity.BackupSetting{ClusterID: entity.ID(1)}, nil,
	).Once()

	err := modifier.execute(context.Background(), &domain.ReplicationRTCModificationReq{ClusterID: 1})
	require.Error(t, err)
}

func TestSynchronousRTCModifier_Execute_NilBucketPrefix(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	modifier := newTestRTCModifier(backupSettingRepo, nil, nil, nil, nil)

	backupSettingRepo.On("Get", mock.Anything, mock.Anything).Return(
		&entity.BackupSetting{
			ClusterID: entity.ID(1),
			Bucket:    &entity.BucketSetting{Prefix: nil},
		}, nil,
	).Once()

	err := modifier.execute(context.Background(), &domain.ReplicationRTCModificationReq{ClusterID: 1})
	require.Error(t, err)
}

func TestSynchronousRTCModifier_Execute_AWSBucketConfigLoadFails(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	bucketRepo := repo_mock.NewBucketRepo(t)
	awsBucketConfigRepo := repo_mock.NewAWSBucketConfigurationRepo(t)
	modifier := newTestRTCModifier(backupSettingRepo, bucketRepo, nil, awsBucketConfigRepo, nil)

	backupSettingRepo.On("Get", mock.Anything, mock.Anything).Return(
		makeBackupSettingWithMasterBucket(1, 42), nil,
	).Once()
	bucketRepo.On("Load", mock.Anything, uint64(42)).Return(makeValidBucket(42), nil).Once()
	configErr := bizerr.New(bizerr.ErrCodeDB, "aws config not found")
	awsBucketConfigRepo.On("Get", mock.Anything, mock.Anything).Return(nil, configErr).Once()

	err := modifier.execute(context.Background(), &domain.ReplicationRTCModificationReq{ClusterID: 1, BucketID: 42})
	require.Error(t, err)
	assert.Equal(t, configErr, err)
}

// ─── execute: dry-run ────────────────────────────────────────────────────────

func TestSynchronousRTCModifier_Execute_DryRun_ReturnsNilWithoutCallingCloudAPI(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	bucketRepo := repo_mock.NewBucketRepo(t)
	awsBucketConfigRepo := repo_mock.NewAWSBucketConfigurationRepo(t)
	awsBucketManagerModel := bucket_mock.NewAWSBucketManagerModel(t)
	modifier := newTestRTCModifier(backupSettingRepo, bucketRepo, awsBucketManagerModel, awsBucketConfigRepo, nil)

	backupSettingRepo.On("Get", mock.Anything, mock.Anything).Return(
		makeBackupSettingWithMasterBucket(1, 42), nil,
	).Once()
	bucketRepo.On("Load", mock.Anything, uint64(42)).Return(makeValidBucket(42), nil).Once()
	awsBucketConfigRepo.On("Get", mock.Anything, mock.Anything).Return(
		&entity.AWSBucketConfiguration{}, nil,
	).Once()

	err := modifier.execute(context.Background(), &domain.ReplicationRTCModificationReq{
		ClusterID: 1, BucketID: 42, Enable: true, DryRun: true,
	})
	require.NoError(t, err)
	awsBucketManagerModel.AssertNotCalled(t, "ModifyReplicationRTC", mock.Anything, mock.Anything)
}

// ─── applyRTCChange ──────────────────────────────────────────────────────────

func TestSynchronousRTCModifier_ApplyRTCChange_ModifyReplicationRTCFails(t *testing.T) {
	awsBucketManagerModel := bucket_mock.NewAWSBucketManagerModel(t)
	modifier := newTestRTCModifier(nil, nil, awsBucketManagerModel, nil, nil)

	modErr := bizerr.New(bizerr.ErrCodeGRPC, "cloud api error")
	awsBucketManagerModel.On("ModifyReplicationRTC", mock.Anything, mock.Anything).Return(modErr).Once()

	backupSetting := makeBackupSettingWithMasterBucket(1, 42)

	err := modifier.applyRTCChange(context.Background(), zap.NewNop(), backupSetting, makeValidBucket(42), &entity.AWSBucketConfiguration{}, true)
	require.Error(t, err)
	assert.Equal(t, modErr, err)
}

func TestSynchronousRTCModifier_ApplyRTCChange_AWSBucketConfigSaveFails(t *testing.T) {
	awsBucketManagerModel := bucket_mock.NewAWSBucketManagerModel(t)
	awsBucketConfigRepo := repo_mock.NewAWSBucketConfigurationRepo(t)
	logBackupRepo := repo_mock.NewLogBackupRepo(t)
	modifier := newTestRTCModifier(nil, nil, awsBucketManagerModel, awsBucketConfigRepo, logBackupRepo)

	awsBucketManagerModel.On("ModifyReplicationRTC", mock.Anything, mock.Anything).Return(nil).Once()
	awsBucketConfigRepo.On("Save", mock.Anything, mock.Anything).Return(
		bizerr.New(bizerr.ErrCodeDB, "save config error"),
	).Once()

	backupSetting := makeBackupSettingWithMasterBucket(1, 42)

	err := modifier.applyRTCChange(context.Background(), zap.NewNop(), backupSetting, makeValidBucket(42), &entity.AWSBucketConfiguration{}, true)
	require.Error(t, err)
}

func TestSynchronousRTCModifier_ApplyRTCChange_BucketSaveFails(t *testing.T) {
	bucketRepo := repo_mock.NewBucketRepo(t)
	awsBucketManagerModel := bucket_mock.NewAWSBucketManagerModel(t)
	awsBucketConfigRepo := repo_mock.NewAWSBucketConfigurationRepo(t)
	logBackupRepo := repo_mock.NewLogBackupRepo(t)
	modifier := newTestRTCModifier(nil, bucketRepo, awsBucketManagerModel, awsBucketConfigRepo, logBackupRepo)

	awsBucketManagerModel.On("ModifyReplicationRTC", mock.Anything, mock.Anything).Return(nil).Once()
	awsBucketConfigRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	bucketRepo.On("Save", mock.Anything, mock.Anything).Return(
		bizerr.New(bizerr.ErrCodeDB, "save bucket error"),
	).Once()

	backupSetting := makeBackupSettingWithMasterBucket(1, 42)

	err := modifier.applyRTCChange(context.Background(), zap.NewNop(), backupSetting, makeValidBucket(42), &entity.AWSBucketConfiguration{}, true)
	require.Error(t, err)
}

func TestSynchronousRTCModifier_ApplyRTCChange_EnableRTC_UpdatesConfigAndBucket(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	bucketRepo := repo_mock.NewBucketRepo(t)
	awsBucketManagerModel := bucket_mock.NewAWSBucketManagerModel(t)
	awsBucketConfigRepo := repo_mock.NewAWSBucketConfigurationRepo(t)
	logBackupRepo := repo_mock.NewLogBackupRepo(t)
	modifier := newTestRTCModifier(backupSettingRepo, bucketRepo, awsBucketManagerModel, awsBucketConfigRepo, logBackupRepo)

	loadedBucket := makeValidBucket(42)
	awsBucketManagerModel.On("ModifyReplicationRTC", mock.Anything, mock.MatchedBy(
		func(req *domain.ModifyReplicationRTCReq) bool {
			return req.EnableRTC && req.Name == "test-bucket" && req.RuleID == "test-prefix"
		},
	)).Return(nil).Once()
	awsBucketConfigRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	bucketRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	backupSettingRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	logBackupRepo.On("List", mock.Anything, mock.Anything).Return([]entity.LogBackup{}, nil).Once()

	backupSetting := makeBackupSettingWithMasterBucket(1, 42)
	awsBucketConfig := &entity.AWSBucketConfiguration{}

	err := modifier.applyRTCChange(context.Background(), zap.NewNop(), backupSetting, loadedBucket, awsBucketConfig, true)
	require.NoError(t, err)
	assert.Equal(t, pointer.To(true), awsBucketConfig.RTCEnabled)
	assert.True(t, loadedBucket.GetRTCEnabled())
	assert.Equal(t, pointer.To(true), backupSetting.Bucket.RTCEnabled)
}

func TestSynchronousRTCModifier_ApplyRTCChange_DisableRTC_UpdatesConfigAndBucket(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	bucketRepo := repo_mock.NewBucketRepo(t)
	awsBucketManagerModel := bucket_mock.NewAWSBucketManagerModel(t)
	awsBucketConfigRepo := repo_mock.NewAWSBucketConfigurationRepo(t)
	logBackupRepo := repo_mock.NewLogBackupRepo(t)
	modifier := newTestRTCModifier(backupSettingRepo, bucketRepo, awsBucketManagerModel, awsBucketConfigRepo, logBackupRepo)

	loadedBucket := makeValidBucket(42)
	loadedBucket.SetRTCEnabled(true)
	awsBucketManagerModel.On("ModifyReplicationRTC", mock.Anything, mock.MatchedBy(
		func(req *domain.ModifyReplicationRTCReq) bool { return !req.EnableRTC },
	)).Return(nil).Once()
	awsBucketConfigRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	bucketRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	backupSettingRepo.On("Save", mock.Anything, mock.Anything).Return(nil).Once()
	logBackupRepo.On("List", mock.Anything, mock.Anything).Return([]entity.LogBackup{}, nil).Once()

	backupSetting := makeBackupSettingWithMasterBucket(1, 42)
	awsBucketConfig := &entity.AWSBucketConfiguration{RTCEnabled: pointer.To(true)}

	err := modifier.applyRTCChange(context.Background(), zap.NewNop(), backupSetting, loadedBucket, awsBucketConfig, false)
	require.NoError(t, err)
	assert.Equal(t, pointer.To(false), awsBucketConfig.RTCEnabled)
	assert.False(t, loadedBucket.GetRTCEnabled())
	assert.Equal(t, pointer.To(false), backupSetting.Bucket.RTCEnabled)
}

// ─── updateRunningLogBackupSnapshots ────────────────────────────────────────

func TestSynchronousRTCModifier_UpdateLogBackupSnapshots_ListFails(t *testing.T) {
	logBackupRepo := repo_mock.NewLogBackupRepo(t)
	modifier := newTestRTCModifier(nil, nil, nil, nil, logBackupRepo)

	listErr := bizerr.New(bizerr.ErrCodeDB, "list error")
	logBackupRepo.On("List", mock.Anything, mock.Anything).Return(nil, listErr).Once()

	err := modifier.updateRunningLogBackupSnapshots(context.Background(), zap.NewNop(), 1, true)
	require.Error(t, err)
}

func TestSynchronousRTCModifier_UpdateLogBackupSnapshots_SkipsLogBackupWithNilSnapshot(t *testing.T) {
	logBackupRepo := repo_mock.NewLogBackupRepo(t)
	modifier := newTestRTCModifier(nil, nil, nil, nil, logBackupRepo)

	logBackupRepo.On("List", mock.Anything, mock.Anything).Return(
		[]entity.LogBackup{{BucketSettingSnapshot: nil}}, nil,
	).Once()

	err := modifier.updateRunningLogBackupSnapshots(context.Background(), zap.NewNop(), 1, true)
	require.NoError(t, err)
	logBackupRepo.AssertNotCalled(t, "Save", mock.Anything, mock.Anything)
}

func TestSynchronousRTCModifier_UpdateLogBackupSnapshots_SaveFails(t *testing.T) {
	logBackupRepo := repo_mock.NewLogBackupRepo(t)
	modifier := newTestRTCModifier(nil, nil, nil, nil, logBackupRepo)

	logBackupRepo.On("List", mock.Anything, mock.Anything).Return(
		[]entity.LogBackup{
			{
				Updatable:             entity.Updatable{Base: entity.Base{ID: entity.ID(10)}},
				BucketSettingSnapshot: &entity.BucketSetting{},
			},
		}, nil,
	).Once()
	logBackupRepo.On("Save", mock.Anything, mock.Anything).Return(
		bizerr.New(bizerr.ErrCodeDB, "save error"),
	).Once()

	err := modifier.updateRunningLogBackupSnapshots(context.Background(), zap.NewNop(), 10, true)
	require.Error(t, err)
}

func TestSynchronousRTCModifier_UpdateLogBackupSnapshots_UpdatesRTCEnabledOnAllActiveBackups(t *testing.T) {
	logBackupRepo := repo_mock.NewLogBackupRepo(t)
	modifier := newTestRTCModifier(nil, nil, nil, nil, logBackupRepo)

	lb1 := entity.LogBackup{
		Updatable:             entity.Updatable{Base: entity.Base{ID: entity.ID(11)}},
		BucketSettingSnapshot: &entity.BucketSetting{},
	}
	lb2 := entity.LogBackup{
		Updatable:             entity.Updatable{Base: entity.Base{ID: entity.ID(12)}},
		BucketSettingSnapshot: &entity.BucketSetting{},
	}
	logBackupRepo.On("List", mock.Anything, mock.Anything).Return(
		[]entity.LogBackup{lb1, lb2}, nil,
	).Once()
	logBackupRepo.On("Save", mock.Anything, mock.MatchedBy(func(lb *entity.LogBackup) bool {
		return lb.BucketSettingSnapshot != nil &&
			lb.BucketSettingSnapshot.RTCEnabled != nil &&
			*lb.BucketSettingSnapshot.RTCEnabled
	})).Return(nil).Times(2)

	err := modifier.updateRunningLogBackupSnapshots(context.Background(), zap.NewNop(), 11, true)
	require.NoError(t, err)
}

// ─── execute: BucketID path ───────────────────────────────────────────────────

// TestSynchronousRTCModifier_Execute_BucketID_LoadFails verifies that when BucketID is provided
// but the bucket repo returns an error, execute returns that error.
func TestSynchronousRTCModifier_Execute_BucketID_LoadFails(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	bucketRepo := repo_mock.NewBucketRepo(t)
	modifier := newTestRTCModifier(backupSettingRepo, bucketRepo, nil, nil, nil)

	backupSettingRepo.On("Get", mock.Anything, mock.Anything).Return(
		makeBackupSettingWithMasterBucket(1, 42), nil,
	).Once()
	loadErr := bizerr.New(bizerr.ErrCodeDB, "bucket not found")
	bucketRepo.On("Load", mock.Anything, uint64(55)).Return(nil, loadErr).Once()

	err := modifier.execute(context.Background(), &domain.ReplicationRTCModificationReq{ClusterID: 1, BucketID: 55})
	require.Error(t, err)
	assert.Equal(t, loadErr, err)
}

// TestSynchronousRTCModifier_Execute_BucketID_ProjectMismatch verifies that when BucketID refers
// to a bucket belonging to a different project than the backup setting, execute returns an error.
func TestSynchronousRTCModifier_Execute_BucketID_ProjectMismatch(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	bucketRepo := repo_mock.NewBucketRepo(t)
	modifier := newTestRTCModifier(backupSettingRepo, bucketRepo, nil, nil, nil)

	bs := makeBackupSettingWithMasterBucket(1, 42)
	bs.ProjectID = entity.ID(100)
	backupSettingRepo.On("Get", mock.Anything, mock.Anything).Return(bs, nil).Once()

	wrongProjectBucket := makeValidBucket(55)
	wrongProjectBucket.ProjectID = "999" // different project
	bucketRepo.On("Load", mock.Anything, uint64(55)).Return(wrongProjectBucket, nil).Once()

	err := modifier.execute(context.Background(), &domain.ReplicationRTCModificationReq{ClusterID: 1, BucketID: 55})
	require.Error(t, err)
	assert.Contains(t, err.Error(), "does not belong to project")
}

// TestSynchronousRTCModifier_Execute_DryRun_WithBucketID verifies that when BucketID is provided
// and DryRun is true, execute succeeds without calling the cloud API.
func TestSynchronousRTCModifier_Execute_DryRun_WithBucketID(t *testing.T) {
	backupSettingRepo := repo_mock.NewBackupSettingRepo(t)
	bucketRepo := repo_mock.NewBucketRepo(t)
	awsBucketManagerModel := bucket_mock.NewAWSBucketManagerModel(t)
	awsBucketConfigRepo := repo_mock.NewAWSBucketConfigurationRepo(t)
	modifier := newTestRTCModifier(backupSettingRepo, bucketRepo, awsBucketManagerModel, awsBucketConfigRepo, nil)

	bs := makeBackupSettingWithMasterBucket(1, 42)
	bs.ProjectID = entity.ID(100)
	backupSettingRepo.On("Get", mock.Anything, mock.Anything).Return(bs, nil).Once()

	sameProjBucket := makeValidBucket(55)
	sameProjBucket.ProjectID = "100"
	bucketRepo.On("Load", mock.Anything, uint64(55)).Return(sameProjBucket, nil).Once()
	awsBucketConfigRepo.On("Get", mock.Anything, mock.Anything).Return(&entity.AWSBucketConfiguration{}, nil).Once()

	err := modifier.execute(context.Background(), &domain.ReplicationRTCModificationReq{
		ClusterID: 1, BucketID: 55, Enable: true, DryRun: true,
	})
	require.NoError(t, err)
	awsBucketManagerModel.AssertNotCalled(t, "ModifyReplicationRTC", mock.Anything, mock.Anything)
}