package impl

import (
	"strings"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/app/models/domain"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	"github.com/tidbcloud/pkgv2/bizerr"
)

type replayTestEncryptor struct{}

func (replayTestEncryptor) EncryptBase64(text string) (string, bizerr.BizError) {
	return "enc:" + text, nil
}
func (replayTestEncryptor) DecryptBase64(text string) (string, bizerr.BizError) {
	return strings.TrimPrefix(text, "enc:"), nil
}
func (replayTestEncryptor) EncryptBase64Inplace(text *string) bizerr.BizError {
	if text != nil && *text != "" {
		*text = "enc:" + *text
	}
	return nil
}
func (replayTestEncryptor) DecryptBase64Inplace(text *string) bizerr.BizError { return nil }

func encryptedAliyunBucketURLForTest(bucketURL string) string {
	return encryptedAliyunBucketURLPrefix + "enc:" + bucketURL
}

func TestMergeReplayConfigUsesReplaceSemantics(t *testing.T) {
	existingReplayStartTime := "2026-04-08T10:00:00Z"
	existingRunStartTime := "2026-04-08T10:10:00Z"
	existingReplayIndex := int32(7)
	existingFormat := "audit_log_plugin"
	existingIgnoreErrs := true
	existingPsClose := "always"
	existingBufSize := int32(1000000)
	existingSpeed := "1"
	existingDynamicInput := true
	existingReadOnly := true
	newRunStartTime := "2026-04-11T08:00:00Z"
	newReplayIndex := int32(0)
	newFormat := ""
	newIgnoreErrs := false
	newPsClose := ""
	newBufSize := int32(0)
	newSpeed := ""
	newDynamicInput := false
	newReadOnly := false

	replayEntity := &entity.Replay{
		ReplayConfig: &entity.ReplayCommonConfig{
			Username: "root",
			Password: "encrypted-password",
			ReplayConfig: entity.ReplayConfig{
				ReplayStartTime: &existingReplayStartTime,
				RunStartTime:    &existingRunStartTime,
				ReplayIndex:     &existingReplayIndex,
				Format:          &existingFormat,
				IgnoreErrs:      &existingIgnoreErrs,
				PsClose:         &existingPsClose,
				BufSize:         &existingBufSize,
				Speed:           &existingSpeed,
				DynamicInput:    &existingDynamicInput,
				ReadOnly:        &existingReadOnly,
				CustomArguments: map[string]string{"wait-on-eof": "true"},
			},
		},
	}

	newConfig := &entity.ReplayCommonConfig{
		ReplayConfig: entity.ReplayConfig{
			RunStartTime: &newRunStartTime,
			ReplayIndex:  &newReplayIndex,
			Format:       &newFormat,
			IgnoreErrs:   &newIgnoreErrs,
			PsClose:      &newPsClose,
			BufSize:      &newBufSize,
			Speed:        &newSpeed,
			DynamicInput: &newDynamicInput,
			ReadOnly:     &newReadOnly,
		},
	}

	model := &replayModelImpl{}
	err := model.mergeReplayConfig(replayEntity, newConfig)
	require.NoError(t, err)

	require.Equal(t, "root", replayEntity.ReplayConfig.Username)
	require.Equal(t, "encrypted-password", replayEntity.ReplayConfig.Password)
	require.Nil(t, replayEntity.ReplayConfig.ReplayConfig.ReplayStartTime)
	require.Equal(t, newRunStartTime, *replayEntity.ReplayConfig.ReplayConfig.RunStartTime)
	require.Equal(t, int32(0), *replayEntity.ReplayConfig.ReplayConfig.ReplayIndex)
	require.Equal(t, "", *replayEntity.ReplayConfig.ReplayConfig.Format)
	require.False(t, *replayEntity.ReplayConfig.ReplayConfig.IgnoreErrs)
	require.Equal(t, "", *replayEntity.ReplayConfig.ReplayConfig.PsClose)
	require.Equal(t, int32(0), *replayEntity.ReplayConfig.ReplayConfig.BufSize)
	require.Equal(t, "", *replayEntity.ReplayConfig.ReplayConfig.Speed)
	require.False(t, *replayEntity.ReplayConfig.ReplayConfig.DynamicInput)
	require.False(t, *replayEntity.ReplayConfig.ReplayConfig.ReadOnly)
	require.Nil(t, replayEntity.ReplayConfig.ReplayConfig.CustomArguments)
}

func TestMergeReplayConfigOmittedFieldsAreClearedUnderReplaceSemantics(t *testing.T) {
	existingFormat := "audit_log_plugin"
	existingIgnoreErrs := true

	replayEntity := &entity.Replay{
		ReplayConfig: &entity.ReplayCommonConfig{
			ReplayConfig: entity.ReplayConfig{
				Format:     &existingFormat,
				IgnoreErrs: &existingIgnoreErrs,
			},
		},
	}

	model := &replayModelImpl{}
	err := model.mergeReplayConfig(replayEntity, &entity.ReplayCommonConfig{})
	require.NoError(t, err)
	require.Nil(t, replayEntity.ReplayConfig.ReplayConfig.Format)
	require.Nil(t, replayEntity.ReplayConfig.ReplayConfig.IgnoreErrs)
}

func TestApplyReplayUpdatesReplacesSourceAuditCredentialAndEncryptsAliyunSecrets(t *testing.T) {
	replayEntity := &entity.Replay{
		SourceAuditCredential: &entity.AuditLogAddr{
			AWS: &entity.AWSAuditLog{
				S3URL: "s3://bucket/path",
			},
		},
	}
	model := &replayModelImpl{encryptor: replayTestEncryptor{}}

	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       "oss://bucket/new-path",
				AccessKeyID:     "new-ak",
				SecretAccessKey: "new-sk",
			},
		},
	})
	require.NoError(t, err)
	require.NotNil(t, replayEntity.SourceAuditCredential)
	require.Nil(t, replayEntity.SourceAuditCredential.AWS)
	require.Nil(t, replayEntity.SourceAuditCredential.GCP)
	require.NotNil(t, replayEntity.SourceAuditCredential.Aliyun)
	require.Equal(
		t,
		encryptedAliyunBucketURLForTest("oss://bucket/new-path"),
		replayEntity.SourceAuditCredential.Aliyun.BucketURL,
	)
	require.Equal(t, "enc:new-ak", replayEntity.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "enc:new-sk", replayEntity.SourceAuditCredential.Aliyun.SecretAccessKey)
}

func TestApplyReplayUpdatesRejectsSourceAuditCredentialPartialAliyunKeyPair(t *testing.T) {
	replayEntity := &entity.Replay{}
	model := &replayModelImpl{encryptor: replayTestEncryptor{}}

	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:   "oss://bucket/new-path",
				AccessKeyID: "new-ak",
			},
		},
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "access_key_id and secret_access_key must be provided together")
}

func TestApplyReplayUpdatesOverwritesSourceAuditCredentialAliyunKeysWithEmptyInput(t *testing.T) {
	replayEntity := &entity.Replay{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       "oss://bucket/old-path",
				AccessKeyID:     "enc:old-ak",
				SecretAccessKey: "enc:old-sk",
			},
		},
	}
	model := &replayModelImpl{encryptor: replayTestEncryptor{}}

	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL: "oss://bucket/new-path",
			},
		},
	})
	require.NoError(t, err)
	require.NotNil(t, replayEntity.SourceAuditCredential)
	require.NotNil(t, replayEntity.SourceAuditCredential.Aliyun)
	require.Equal(
		t,
		encryptedAliyunBucketURLForTest("oss://bucket/new-path"),
		replayEntity.SourceAuditCredential.Aliyun.BucketURL,
	)
	require.Empty(t, replayEntity.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Empty(t, replayEntity.SourceAuditCredential.Aliyun.SecretAccessKey)
}

func TestApplyReplayUpdatesTreatsSourceAuditCredentialAliyunPlaceholdersAsInput(t *testing.T) {
	replayEntity := &entity.Replay{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       encryptedAliyunBucketURLForTest("oss://bucket/old-path"),
				AccessKeyID:     "enc:old-ak",
				SecretAccessKey: "enc:old-sk",
			},
		},
	}
	model := &replayModelImpl{encryptor: replayTestEncryptor{}}

	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       "oss://bucket/new-path",
				AccessKeyID:     "{ACCESS_KEY_ID}",
				SecretAccessKey: "{SECRET_ACCESS_KEY}",
			},
		},
	})
	require.NoError(t, err)
	require.NotNil(t, replayEntity.SourceAuditCredential)
	require.NotNil(t, replayEntity.SourceAuditCredential.Aliyun)
	require.Equal(
		t,
		encryptedAliyunBucketURLForTest("oss://bucket/new-path"),
		replayEntity.SourceAuditCredential.Aliyun.BucketURL,
	)
	require.Equal(t, "enc:{ACCESS_KEY_ID}", replayEntity.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "enc:{SECRET_ACCESS_KEY}", replayEntity.SourceAuditCredential.Aliyun.SecretAccessKey)

	prepared := cloneReplayForExecution(replayEntity)
	err = decryptReplayAuditLogSecretFields(replayTestEncryptor{}, prepared, auditLogSecretScopeExecution)
	require.NoError(t, err)
	require.Equal(t, "oss://bucket/new-path", prepared.SourceAuditCredential.Aliyun.BucketURL)
	require.Equal(t, "{ACCESS_KEY_ID}", prepared.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "{SECRET_ACCESS_KEY}", prepared.SourceAuditCredential.Aliyun.SecretAccessKey)
}

func TestApplyReplayUpdatesTreatsSourceAuditCredentialAliyunPlaceholderMixAsInput(t *testing.T) {
	replayEntity := &entity.Replay{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       encryptedAliyunBucketURLForTest("oss://bucket/old-path"),
				AccessKeyID:     "enc:old-ak",
				SecretAccessKey: "enc:old-sk",
			},
		},
	}
	model := &replayModelImpl{encryptor: replayTestEncryptor{}}

	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       "oss://bucket/new-path",
				AccessKeyID:     "{ACCESS_KEY_ID}",
				SecretAccessKey: "new-sk",
			},
		},
	})
	require.NoError(t, err)
	require.Equal(t, "enc:{ACCESS_KEY_ID}", replayEntity.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "enc:new-sk", replayEntity.SourceAuditCredential.Aliyun.SecretAccessKey)
}

func TestEncryptReplayAuditLogSecretFieldsEncryptsTaskAliyunSecrets(t *testing.T) {
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{
			{
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/path",
								AccessKeyID:     "task-ak",
								SecretAccessKey: "task-sk",
							},
						},
					},
				},
			},
		},
	}

	err := encryptReplayAuditLogSecretFields(replayTestEncryptor{}, replayEntity)
	require.NoError(t, err)
	addr := replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun
	require.NotNil(t, addr)
	require.Equal(t, encryptedAliyunBucketURLForTest("oss://bucket/path"), addr.BucketURL)
	require.Equal(t, "enc:task-ak", addr.AccessKeyID)
	require.Equal(t, "enc:task-sk", addr.SecretAccessKey)
}

func TestEncryptReplayAuditLogSecretFieldsEncryptsForgedAliyunBucketURLPrefix(t *testing.T) {
	forgedBucketURL := encryptedAliyunBucketURLPrefix +
		"s3://bucket/path/tidb-?endpoint=http://oss-cn-hangzhou.aliyuncs.com&access-key=ak&secret-access-key=sk"
	replayEntity := &entity.Replay{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       forgedBucketURL,
				AccessKeyID:     "ak",
				SecretAccessKey: "sk",
			},
		},
		ReplayTasks: entity.ReplayTaskList{
			{
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       forgedBucketURL,
								AccessKeyID:     "task-ak",
								SecretAccessKey: "task-sk",
							},
						},
					},
				},
			},
		},
	}

	err := encryptReplayAuditLogSecretFields(replayTestEncryptor{}, replayEntity)
	require.NoError(t, err)

	require.Equal(t, encryptedAliyunBucketURLForTest(forgedBucketURL), replayEntity.SourceAuditCredential.Aliyun.BucketURL)
	require.Equal(t, "enc:ak", replayEntity.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "enc:sk", replayEntity.SourceAuditCredential.Aliyun.SecretAccessKey)
	taskAliyun := replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun
	require.Equal(t, encryptedAliyunBucketURLForTest(forgedBucketURL), taskAliyun.BucketURL)
	require.Equal(t, "enc:task-ak", taskAliyun.AccessKeyID)
	require.Equal(t, "enc:task-sk", taskAliyun.SecretAccessKey)

	prepared := cloneReplayForExecution(replayEntity)
	err = decryptReplayAuditLogSecretFields(replayTestEncryptor{}, prepared, auditLogSecretScopeExecution)
	require.NoError(t, err)
	require.Equal(t, forgedBucketURL, prepared.SourceAuditCredential.Aliyun.BucketURL)
	require.Equal(t, "ak", prepared.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "sk", prepared.SourceAuditCredential.Aliyun.SecretAccessKey)
	require.Equal(t, forgedBucketURL, prepared.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.BucketURL)
	require.Equal(t, "task-ak", prepared.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.AccessKeyID)
	require.Equal(t, "task-sk", prepared.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.SecretAccessKey)
}

func TestDecryptReplaySecretFieldsForExecutionDecryptsAliyunBucketURL(t *testing.T) {
	replayEntity := &entity.Replay{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       encryptedAliyunBucketURLForTest("oss://bucket/path"),
				AccessKeyID:     "enc:ak",
				SecretAccessKey: "enc:sk",
			},
		},
		ReplayTasks: entity.ReplayTaskList{
			{
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL: encryptedAliyunBucketURLForTest("s3://bucket/task-path"),
							},
						},
					},
				},
			},
		},
	}

	err := decryptReplayAuditLogSecretFields(replayTestEncryptor{}, replayEntity, auditLogSecretScopeExecution)
	require.NoError(t, err)
	require.Equal(t, "oss://bucket/path", replayEntity.SourceAuditCredential.Aliyun.BucketURL)
	require.Equal(t, "ak", replayEntity.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "sk", replayEntity.SourceAuditCredential.Aliyun.SecretAccessKey)
	require.Equal(
		t,
		"s3://bucket/task-path",
		replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.BucketURL,
	)
}

func TestDecryptReplayAuditLogBucketURLsForTransportKeepsAliyunKeysEncrypted(t *testing.T) {
	replayEntity := &entity.Replay{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       encryptedAliyunBucketURLForTest("s3://bucket/path?access-key=ak&secret-access-key=sk"),
				AccessKeyID:     "enc:ak",
				SecretAccessKey: "enc:sk",
			},
		},
	}

	err := decryptReplayAuditLogSecretFields(replayTestEncryptor{}, replayEntity, auditLogSecretScopeTransport)
	require.NoError(t, err)
	require.Equal(t, "s3://bucket/path?access-key=ak&secret-access-key=sk", replayEntity.SourceAuditCredential.Aliyun.BucketURL)
	require.Equal(t, "enc:ak", replayEntity.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "enc:sk", replayEntity.SourceAuditCredential.Aliyun.SecretAccessKey)
}

func TestEntityToReplayForReadDecryptsAliyunBucketURLOnClone(t *testing.T) {
	encryptedBucketURL := encryptedAliyunBucketURLForTest("s3://bucket/path?access-key=ak&secret-access-key=sk")
	replayEntity := &entity.Replay{
		SourceAuditCredential: &entity.AuditLogAddr{
			Aliyun: &entity.AliyunAuditLog{
				BucketURL:       encryptedBucketURL,
				AccessKeyID:     "enc:ak",
				SecretAccessKey: "enc:sk",
			},
		},
	}
	model := &replayModelImpl{encryptor: replayTestEncryptor{}}

	replayDomain, err := model.entityToReplayForRead(replayEntity)
	require.NoError(t, err)
	require.NotNil(t, replayDomain.SourceAuditCredential)
	require.NotNil(t, replayDomain.SourceAuditCredential.Aliyun)
	require.Equal(t, "s3://bucket/path?access-key=ak&secret-access-key=sk", replayDomain.SourceAuditCredential.Aliyun.BucketURL)
	require.Equal(t, "enc:ak", replayDomain.SourceAuditCredential.Aliyun.AccessKeyID)
	require.Equal(t, "enc:sk", replayDomain.SourceAuditCredential.Aliyun.SecretAccessKey)
	require.Equal(t, encryptedBucketURL, replayEntity.SourceAuditCredential.Aliyun.BucketURL)
}

func TestApplyReplayUpdatesReplacesSourceAuditCredentialWithEmptyStructLikeMaster(t *testing.T) {
	replayEntity := &entity.Replay{
		SourceAuditCredential: &entity.AuditLogAddr{
			AWS: &entity.AWSAuditLog{
				S3URL: "s3://bucket/path",
			},
		},
	}
	model := &replayModelImpl{}

	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		SourceAuditCredential: &entity.AuditLogAddr{},
	})
	require.NoError(t, err)
	require.NotNil(t, replayEntity.SourceAuditCredential)
	require.Nil(t, replayEntity.SourceAuditCredential.AWS)
	require.Nil(t, replayEntity.SourceAuditCredential.GCP)
	require.Nil(t, replayEntity.SourceAuditCredential.Aliyun)
}

func TestApplyReplayUpdatesClearsExplicitEmptyReplayTasks(t *testing.T) {
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{{Name: "task-1"}},
	}
	model := &replayModelImpl{}

	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		ReplayTasks: entity.ReplayTaskList{},
	})
	require.NoError(t, err)
	require.NotNil(t, replayEntity.ReplayTasks)
	require.Empty(t, replayEntity.ReplayTasks)
}

func TestApplyReplayUpdatesUsesReplaceSemanticsForExistingTaskMaterial(t *testing.T) {
	username := "root"
	password := "enc:existing-password"
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/prefix*",
								AccessKeyID:     "task-ak",
								SecretAccessKey: "task-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{
					Endpoint: "tidb.premium.example.com:4000",
					Username: &username,
					Password: &password,
				},
				Replicas:        2,
				ReplayStartTime: "2026-04-01T00:00:00Z",
				RunStartTime:    "2026-04-01T01:00:00Z",
				Resources: &entity.ReplayTaskResources{
					InstanceType: "ecs.g6.large",
					StorageSize:  "500Gi",
				},
				PerReplayConfigs: []entity.ReplayConfig{
					{CustomArguments: map[string]string{"worker-count": "4"}},
				},
				Image: "pingcap/replayer:v1",
			},
		},
	}

	model := &replayModelImpl{encryptor: replayTestEncryptor{}}
	newUsername := "admin"
	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataTarget: entity.ReplayDataTarget{
					Username: &newUsername,
				},
			},
		},
	})
	require.NoError(t, err)
	require.Len(t, replayEntity.ReplayTasks, 1)

	task := replayEntity.ReplayTasks[0]
	require.Empty(t, task.DataSource.SourceAuditLogAddrs)
	require.Empty(t, task.DataTarget.Endpoint)
	require.NotNil(t, task.DataTarget.Username)
	require.Equal(t, "admin", *task.DataTarget.Username)
	require.NotNil(t, task.DataTarget.Password)
	require.Equal(t, "enc:existing-password", *task.DataTarget.Password)
	require.Equal(t, int32(0), task.Replicas)
	require.Empty(t, task.ReplayStartTime)
	require.Empty(t, task.RunStartTime)
	require.Nil(t, task.Resources)
	require.Empty(t, task.PerReplayConfigs)
	require.Empty(t, task.Image)
}

func TestApplyReplayUpdatesUsesTaskAliyunURLOnlyWhenBucketURLChanges(t *testing.T) {
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/prefix/tidb-*",
								AccessKeyID:     "task-ak",
								SecretAccessKey: "task-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{Endpoint: "tidb.premium.example.com:4000"},
			},
		},
	}

	model := &replayModelImpl{encryptor: replayTestEncryptor{}}
	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL: "s3://bucket/other-prefix/tidb-*?endpoint=http://oss-ap-southeast-1.aliyuncs.com",
							},
						},
					},
				},
			},
		},
	})
	require.NoError(t, err)
	require.Len(t, replayEntity.ReplayTasks, 1)
	addr := replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun
	require.NotNil(t, addr)
	require.Equal(
		t,
		encryptedAliyunBucketURLForTest("s3://bucket/other-prefix/tidb-*?endpoint=http://oss-ap-southeast-1.aliyuncs.com"),
		addr.BucketURL,
	)
	require.Empty(t, addr.AccessKeyID)
	require.Empty(t, addr.SecretAccessKey)
}

func TestApplyReplayUpdatesEncryptsTaskAliyunKeysWithoutNormalizingBucket(t *testing.T) {
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/prefix/tidb-*",
								AccessKeyID:     "task-ak",
								SecretAccessKey: "task-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{Endpoint: "tidb.premium.example.com:4000"},
			},
		},
	}

	model := &replayModelImpl{encryptor: replayTestEncryptor{}}
	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "s3://bucket/prefix/tidb-*",
								AccessKeyID:     "new-ak",
								SecretAccessKey: "new-sk",
							},
						},
					},
				},
			},
		},
	})
	require.NoError(t, err)
	addr := replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun
	require.NotNil(t, addr)
	require.Equal(t, encryptedAliyunBucketURLForTest("s3://bucket/prefix/tidb-*"), addr.BucketURL)
	require.Equal(t, "enc:new-ak", addr.AccessKeyID)
	require.Equal(t, "enc:new-sk", addr.SecretAccessKey)
}

func TestApplyReplayUpdatesTreatsTaskAliyunPlaceholdersAsInput(t *testing.T) {
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       encryptedAliyunBucketURLForTest("oss://bucket/old-prefix/tidb-*"),
								AccessKeyID:     "enc:task-ak",
								SecretAccessKey: "enc:task-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{Endpoint: "tidb.premium.example.com:4000"},
			},
		},
	}

	model := &replayModelImpl{encryptor: replayTestEncryptor{}}
	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/new-prefix/tidb-*",
								AccessKeyID:     "{ACCESS_KEY_ID}",
								SecretAccessKey: "{SECRET_ACCESS_KEY}",
							},
						},
					},
				},
			},
		},
	})
	require.NoError(t, err)
	addr := replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun
	require.NotNil(t, addr)
	require.Equal(t, encryptedAliyunBucketURLForTest("oss://bucket/new-prefix/tidb-*"), addr.BucketURL)
	require.Equal(t, "enc:{ACCESS_KEY_ID}", addr.AccessKeyID)
	require.Equal(t, "enc:{SECRET_ACCESS_KEY}", addr.SecretAccessKey)

	prepared := cloneReplayForExecution(replayEntity)
	err = decryptReplayAuditLogSecretFields(replayTestEncryptor{}, prepared, auditLogSecretScopeExecution)
	require.NoError(t, err)
	preparedAliyun := prepared.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun
	require.Equal(t, "oss://bucket/new-prefix/tidb-*", preparedAliyun.BucketURL)
	require.Equal(t, "{ACCESS_KEY_ID}", preparedAliyun.AccessKeyID)
	require.Equal(t, "{SECRET_ACCESS_KEY}", preparedAliyun.SecretAccessKey)
}

func TestApplyReplayUpdatesRejectsTaskAliyunPartialKeyPair(t *testing.T) {
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/prefix/tidb-*",
								AccessKeyID:     "enc:task-ak",
								SecretAccessKey: "enc:task-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{Endpoint: "tidb.premium.example.com:4000"},
			},
		},
	}

	model := &replayModelImpl{encryptor: replayTestEncryptor{}}
	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:   "oss://bucket/prefix/tidb-*",
								AccessKeyID: "new-ak",
							},
						},
					},
				},
			},
		},
	})
	require.Error(t, err)
	require.Contains(t, err.Error(), "access_key_id and secret_access_key must be provided together")
}

func TestApplyReplayUpdatesTreatsTaskAliyunPlaceholderMixAsInput(t *testing.T) {
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       encryptedAliyunBucketURLForTest("oss://bucket/prefix/tidb-*"),
								AccessKeyID:     "enc:task-ak",
								SecretAccessKey: "enc:task-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{Endpoint: "tidb.premium.example.com:4000"},
			},
		},
	}

	model := &replayModelImpl{encryptor: replayTestEncryptor{}}
	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/prefix/tidb-*",
								AccessKeyID:     "new-ak",
								SecretAccessKey: "{SECRET_ACCESS_KEY}",
							},
						},
					},
				},
			},
		},
	})
	require.NoError(t, err)
	addr := replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun
	require.Equal(t, "enc:new-ak", addr.AccessKeyID)
	require.Equal(t, "enc:{SECRET_ACCESS_KEY}", addr.SecretAccessKey)
}

func TestApplyReplayUpdatesUsesTaskAliyunUpdateInputWithoutMergingExistingCredential(t *testing.T) {
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/prefix/tidb-*",
								AccessKeyID:     "enc:task-ak",
								SecretAccessKey: "enc:task-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{Endpoint: "tidb.premium.example.com:4000"},
			},
		},
	}
	incomingAliyun := &entity.AliyunAuditLog{}
	updateReq := &domain.UpdateReplayReq{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{Aliyun: incomingAliyun},
					},
				},
			},
		},
	}

	model := &replayModelImpl{encryptor: replayTestEncryptor{}}
	err := model.applyReplayUpdates(replayEntity, updateReq)
	require.NoError(t, err)
	require.Empty(t, incomingAliyun.BucketURL)
	require.Empty(t, incomingAliyun.AccessKeyID)
	require.Empty(t, incomingAliyun.SecretAccessKey)
	require.Empty(t, replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.BucketURL)
	require.Empty(t, replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.AccessKeyID)
	require.Empty(t, replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs[0].Aliyun.SecretAccessKey)
}

func TestApplyReplayUpdatesAppliesUsernameOnlyPatchWithoutExistingPassword(t *testing.T) {
	replayEntity := &entity.Replay{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataSource: entity.ReplayDataSource{
					SourceAuditLogAddrs: []entity.AuditLogAddr{
						{
							Aliyun: &entity.AliyunAuditLog{
								BucketURL:       "oss://bucket/prefix*",
								AccessKeyID:     "task-ak",
								SecretAccessKey: "task-sk",
							},
						},
					},
				},
				DataTarget: entity.ReplayDataTarget{
					Endpoint: "tidb.premium.example.com:4000",
				},
			},
		},
	}

	model := &replayModelImpl{encryptor: replayTestEncryptor{}}
	newUsername := "admin"
	err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
		ReplayTasks: entity.ReplayTaskList{
			{
				Name: "group-0",
				DataTarget: entity.ReplayDataTarget{
					Username: &newUsername,
				},
			},
		},
	})
	require.NoError(t, err)
	require.Len(t, replayEntity.ReplayTasks, 1)
	require.NotNil(t, replayEntity.ReplayTasks[0].DataTarget.Username)
	require.Equal(t, "admin", *replayEntity.ReplayTasks[0].DataTarget.Username)
	require.Nil(t, replayEntity.ReplayTasks[0].DataTarget.Password)
}

func TestApplyReplayUpdatesAppliesIncompleteNewReplayTaskLikeMaster(t *testing.T) {
	model := &replayModelImpl{encryptor: replayTestEncryptor{}}

	t.Run("missing source audit log addrs", func(t *testing.T) {
		replayEntity := &entity.Replay{
			ReplayTasks: entity.ReplayTaskList{
				{
					Name: "group-0",
					DataSource: entity.ReplayDataSource{
						SourceAuditLogAddrs: []entity.AuditLogAddr{
							{Aliyun: &entity.AliyunAuditLog{BucketURL: "oss://bucket/prefix", AccessKeyID: "ak", SecretAccessKey: "sk"}},
						},
					},
					DataTarget: entity.ReplayDataTarget{Endpoint: "tidb.premium.example.com:4000"},
				},
			},
		}

		err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
			ReplayTasks: entity.ReplayTaskList{
				{
					Name:       "group-1",
					DataTarget: entity.ReplayDataTarget{Endpoint: "tidb.premium.example.com:4001"},
				},
			},
		})
		require.NoError(t, err)
		require.Len(t, replayEntity.ReplayTasks, 1)
		require.Equal(t, "group-1", replayEntity.ReplayTasks[0].Name)
		require.Empty(t, replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs)
		require.Equal(t, "tidb.premium.example.com:4001", replayEntity.ReplayTasks[0].DataTarget.Endpoint)
	})

	t.Run("missing target endpoint", func(t *testing.T) {
		replayEntity := &entity.Replay{}

		err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
			ReplayTasks: entity.ReplayTaskList{
				{
					Name: "group-1",
					DataSource: entity.ReplayDataSource{
						SourceAuditLogAddrs: []entity.AuditLogAddr{
							{Aliyun: &entity.AliyunAuditLog{BucketURL: "oss://bucket/prefix", AccessKeyID: "ak", SecretAccessKey: "sk"}},
						},
					},
				},
			},
		})
		require.NoError(t, err)
		require.Len(t, replayEntity.ReplayTasks, 1)
		require.Equal(t, "group-1", replayEntity.ReplayTasks[0].Name)
		require.Len(t, replayEntity.ReplayTasks[0].DataSource.SourceAuditLogAddrs, 1)
		require.Empty(t, replayEntity.ReplayTasks[0].DataTarget.Endpoint)
	})

	t.Run("username without password", func(t *testing.T) {
		replayEntity := &entity.Replay{}
		username := "new-user"

		err := model.applyReplayUpdates(replayEntity, &domain.UpdateReplayReq{
			ReplayTasks: entity.ReplayTaskList{
				{
					Name: "group-1",
					DataSource: entity.ReplayDataSource{
						SourceAuditLogAddrs: []entity.AuditLogAddr{
							{Aliyun: &entity.AliyunAuditLog{BucketURL: "oss://bucket/prefix", AccessKeyID: "ak", SecretAccessKey: "sk"}},
						},
					},
					DataTarget: entity.ReplayDataTarget{
						Endpoint: "tidb.premium.example.com:4001",
						Username: &username,
					},
				},
			},
		})
		require.NoError(t, err)
		require.Len(t, replayEntity.ReplayTasks, 1)
		require.Equal(t, "group-1", replayEntity.ReplayTasks[0].Name)
		require.NotNil(t, replayEntity.ReplayTasks[0].DataTarget.Username)
		require.Equal(t, "new-user", *replayEntity.ReplayTasks[0].DataTarget.Username)
		require.Nil(t, replayEntity.ReplayTasks[0].DataTarget.Password)
	})
}