package task

import (
	"context"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"text/template"

	"github.com/stretchr/testify/assert"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	rbacv1 "k8s.io/api/rbac/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"

	agentv2 "github.com/tidbcloud/observability/agent-control-server/api/v2"
	"github.com/tidbcloud/observability/agent-control-server/pkg/agentbuild"
	"github.com/tidbcloud/observability/agent-control-server/pkg/config"
	"github.com/tidbcloud/observability/agent-control-server/pkg/consts"
	"github.com/tidbcloud/observability/agent-control-server/pkg/reconciler"
	"github.com/tidbcloud/observability/agent-control-server/pkg/testkits"
)

func makeVectorTask(agent *agentv2.SharedPoolAgent, opts testkits.TaskOptions) *VectorTask {
	objs := []client.Object{agent}
	if len(opts.LocalObjects) > 0 {
		objs = append(objs, opts.LocalObjects...)
	}
	localCli := testkits.BuildClient(objs...)

	ctx := &reconciler.Context[*agentv2.SharedPoolAgent]{
		Agent:     agent,
		Local:     localCli,
		Remote:    testkits.NewClientGetterWithOpts(opts),
		Config:    &opts.Config,
		Templates: opts.Templates,
	}
	return NewVectorTask(ctx)
}

func baseVectorConfig() config.RootConfig {
	return config.RootConfig{
		O11yConfig: config.ObservabilityInfra{
			Vendor: "aws",
			CustomerConfigsV2: config.CustomerConfigsV2{
				NextGen: config.CustomerNextGen{
					SharedPool: config.CustomerNextGenSharedPool{
						Default: config.CustomerAgents{
							Vector: &config.VectorConfig{
								Chart: &config.CustomerHelmChart{
									Name:    "vector",
									Version: "1.0.0",
								},
								Image: "vectorio/vector:latest",
							},
						},
						Specs: config.CustomerAgentSpecs{
							{
								Name:   "default",
								Vector: &corev1.ResourceRequirements{},
							},
						},
					},
				},
			},
		},
	}
}

func assertRenderTemplateValues(t *testing.T, templateFile string, values map[string]string, expected string) {
	templatePath := filepath.Join("..", "..", "..", "examples", "templates", templateFile)
	rawTemplate, err := os.ReadFile(templatePath)
	assert.NoError(t, err)

	tmpl, err := template.New(templateFile).Parse(string(rawTemplate))
	assert.NoError(t, err)

	renderedValues, err := agentbuild.BuildVectorValues(tmpl, values)
	assert.NoError(t, err)

	assert.Equal(t, strings.TrimSpace(expected), strings.TrimSpace(renderedValues))
}

func TestVectorTask(t *testing.T) {
	tests := []struct {
		name   string
		verify func(t *testing.T)
	}{
		{
			name: "Name",
			verify: func(t *testing.T) {
				task := &VectorTask{}
				assert.Equal(t, "Vector", task.Name())
			},
		},
		{
			name: "ClusterMeta",
			verify: func(t *testing.T) {
				agent := testkits.NewSharedPoolAgentBuilder().Build()
				task := makeVectorTask(agent, testkits.TaskOptions{})
				meta := task.ClusterMeta()

				assert.Equal(t, consts.AgentSharedPool, meta.Agent)
				assert.Equal(t, "test-cluster", meta.K8sClusterName)
				assert.Equal(t, "tenant-1", meta.TenantID)
				assert.Equal(t, "app-1", meta.ApplicationID)
				assert.Equal(t, "tidbcloud", meta.BizType)
			},
		},
		{
			name: "buildName",
			verify: func(t *testing.T) {
				agent := testkits.NewSharedPoolAgentBuilder().Build()
				task := makeVectorTask(agent, testkits.TaskOptions{})
				assert.Equal(t, "o11y-vector-instance123", task.buildName("instance123"))
			},
		},
		{
			name: "buildNamespace normal mode",
			verify: func(t *testing.T) {
				agent := testkits.NewSharedPoolAgentBuilder().Build()
				task := makeVectorTask(agent, testkits.TaskOptions{Config: baseVectorConfig()})
				assert.Equal(t, "tidbpool-456", task.buildNamespace())
			},
		},
		{
			name: "buildNamespace serverless mode",
			verify: func(t *testing.T) {
				cfg := baseVectorConfig()
				cfg.Mode = config.ModeServerless
				agent := testkits.NewSharedPoolAgentBuilder().Build()
				task := makeVectorTask(agent, testkits.TaskOptions{Config: cfg})
				assert.Equal(t, consts.ServerlessObservabilityNamespace, task.buildNamespace())
			},
		},
		{
			name: "buildRole",
			verify: func(t *testing.T) {
				agent := testkits.NewSharedPoolAgentBuilder().Build()
				task := makeVectorTask(agent, testkits.TaskOptions{})

				role, err := task.buildRole("o11y-vector-instance123", "tidbpool-456")()
				assert.NoError(t, err)
				assert.Equal(t, "o11y-vector-instance123", role.Name)
				assert.Equal(t, "tidbpool-456", role.Namespace)
				assert.Len(t, role.Rules, 1)
				assert.Equal(t, []string{""}, role.Rules[0].APIGroups)
				assert.Equal(t, []string{"pods"}, role.Rules[0].Resources)
				assert.Equal(t, []string{"get", "list", "watch"}, role.Rules[0].Verbs)
			},
		},
		{
			name: "buildRoleBinding",
			verify: func(t *testing.T) {
				agent := testkits.NewSharedPoolAgentBuilder().Build()
				task := makeVectorTask(agent, testkits.TaskOptions{})

				binding, err := task.buildRoleBinding("o11y-vector-instance123", "tidbpool-456")()
				assert.NoError(t, err)
				assert.Len(t, binding.Subjects, 1)
				assert.Equal(t, "ServiceAccount", binding.Subjects[0].Kind)
				assert.Equal(t, "o11y-vector-instance123", binding.Subjects[0].Name)
				assert.Equal(t, "Role", binding.RoleRef.Kind)
				assert.Equal(t, "o11y-vector-instance123", binding.RoleRef.Name)
				assert.Equal(t, "rbac.authorization.k8s.io", binding.RoleRef.APIGroup)
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tt.verify(t)
		})
	}
}

func TestVectorTask_PostCreate(t *testing.T) {
	tests := []struct {
		name    string
		agent   *agentv2.SharedPoolAgent
		opts    testkits.TaskOptions
		wantErr bool
		verify  func(t *testing.T, opts testkits.TaskOptions)
	}{
		{
			name:    "post create fails with remote client error",
			agent:   testkits.NewSharedPoolAgentBuilder().Build(),
			opts:    testkits.TaskOptions{ClientError: true},
			wantErr: true,
		},
		{
			name:    "post create fails when statefulset not ready",
			agent:   testkits.NewSharedPoolAgentBuilder().Build(),
			opts:    testkits.TaskOptions{Client: testkits.BuildClient()},
			wantErr: true,
		},
		{
			name:  "post create succeeds when statefulset ready",
			agent: testkits.NewSharedPoolAgentBuilder().Build(),
			opts: testkits.TaskOptions{
				Client: testkits.BuildClient(
					&appsv1.StatefulSet{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "o11y-vector-instance123",
							Namespace: "tidbpool-456",
						},
						Spec:   appsv1.StatefulSetSpec{Replicas: ptr.To(int32(1))},
						Status: appsv1.StatefulSetStatus{AvailableReplicas: 1},
					},
					&corev1.PersistentVolumeClaim{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "data-o11y-vector-instance123-0",
							Namespace: "tidbpool-456",
						},
					},
				),
			},
			wantErr: false,
			verify: func(t *testing.T, opts testkits.TaskOptions) {
				pvc := &corev1.PersistentVolumeClaim{}
				err := opts.Client.Get(context.Background(), types.NamespacedName{
					Name:      "data-o11y-vector-instance123-0",
					Namespace: "tidbpool-456",
				}, pvc)
				assert.NoError(t, err)
				assert.Equal(t, consts.TagValueComponentO11y, pvc.Labels[consts.TagKeyComponent])
				assert.Equal(t, "tenant-1", pvc.Labels[consts.TagKeyTenant])
				assert.Equal(t, "pool-456", pvc.Labels[consts.TagKeyResourcePool])
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			task := makeVectorTask(tt.agent, tt.opts)
			task.targets = []vectorTarget{{
				InstanceID:  "instance123",
				ReleaseName: "o11y-vector-instance123",
			}}
			result := task.PostCreate(context.Background())
			if tt.wantErr {
				assert.Error(t, result.Error())
			} else {
				assert.NoError(t, result.Error())
			}
			if tt.verify != nil {
				tt.verify(t, tt.opts)
			}
		})
	}
}

func TestVectorTask_PostDelete(t *testing.T) {
	tests := []struct {
		name    string
		agent   *agentv2.SharedPoolAgent
		opts    testkits.TaskOptions
		wantErr bool
		verify  func(t *testing.T, opts testkits.TaskOptions)
	}{
		{
			name:    "post delete fails with remote client error",
			agent:   testkits.NewSharedPoolAgentBuilder().Build(),
			opts:    testkits.TaskOptions{ClientError: true},
			wantErr: true,
		},
		{
			name:    "post delete succeeds when statefulset deleted",
			agent:   testkits.NewSharedPoolAgentBuilder().Build(),
			opts:    testkits.TaskOptions{Client: testkits.BuildClient()},
			wantErr: false,
		},
		{
			name:  "post delete fails when statefulset still exists",
			agent: testkits.NewSharedPoolAgentBuilder().Build(),
			opts: testkits.TaskOptions{
				Client: testkits.BuildClient(
					&appsv1.StatefulSet{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "o11y-vector-instance123",
							Namespace: "tidbpool-456",
						},
					},
					&corev1.Pod{
						ObjectMeta: metav1.ObjectMeta{
							Name:      "db-instance123-tikv-0",
							Namespace: "tidbpool-456",
							Labels: map[string]string{
								"pingcap.com/component": "tikv",
							},
						},
					},
				),
			},
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			task := makeVectorTask(tt.agent, tt.opts)
			result := task.PostDelete(context.Background())
			if tt.wantErr {
				assert.Error(t, result.Error())
			} else {
				assert.NoError(t, result.Error())
			}
			if tt.verify != nil {
				tt.verify(t, tt.opts)
			}
		})
	}
}

func TestVectorTask_Create(t *testing.T) {
	tmpl, err := template.New("test").Parse(`
image: {{.ImageRepo}}:{{.ImageTag}}
`)
	assert.NoError(t, err)

	tests := []struct {
		name    string
		agent   *agentv2.SharedPoolAgent
		opts    testkits.TaskOptions
		wantErr bool
		verify  func(t *testing.T, opts testkits.TaskOptions)
	}{
		{
			name:  "create fails with remote client error",
			agent: testkits.NewSharedPoolAgentBuilder().Build(),
			opts: testkits.TaskOptions{
				ClientError: true,
				Config:      baseVectorConfig(),
			},
			wantErr: true,
		},
		{
			name:  "create fails with missing template",
			agent: testkits.NewSharedPoolAgentBuilder().Build(),
			opts: testkits.TaskOptions{
				Client: testkits.BuildClient(),
				Config: baseVectorConfig(),
			},
			wantErr: true,
		},
		{
			name:  "create fails with alicloud vendor when secret not found",
			agent: testkits.NewSharedPoolAgentBuilder().Build(),
			opts: func() testkits.TaskOptions {
				cfg := baseVectorConfig()
				cfg.O11yConfig.Vendor = "alicloud"
				return testkits.TaskOptions{
					Client: testkits.BuildClient(),
					Config: cfg,
				}
			}(),
			wantErr: true,
		},
		{
			name: "create fails when spec not found",
			agent: testkits.NewSharedPoolAgentBuilder().
				WithAnnotations(map[string]string{consts.AnnotationKeyO11yDesiredSpec: "nonexistent"}).
				Build(),
			opts: func() testkits.TaskOptions {
				cfg := baseVectorConfig()
				cfg.O11yConfig.CustomerConfigsV2.NextGen.SharedPool.Specs = nil
				return testkits.TaskOptions{
					Client: testkits.BuildClient(),
					Config: cfg,
				}
			}(),
			wantErr: true,
		},
		{
			name:  "create succeeds when no TiKV pods",
			agent: testkits.NewSharedPoolAgentBuilder().Build(),
			opts: testkits.TaskOptions{
				Client:    testkits.BuildClient(),
				Config:    baseVectorConfig(),
				Templates: config.Templates{"sharedpool/vector-values-aws.yaml": tmpl},
			},
			wantErr: false,
		},
		{
			name:  "serverless create without TiKV pods",
			agent: testkits.NewSharedPoolAgentBuilder().Build(),
			opts: func() testkits.TaskOptions {
				cfg := baseVectorConfig()
				cfg.Mode = config.ModeServerless
				return testkits.TaskOptions{
					Client: testkits.BuildClient(
						makeServerlessTLSSecret(t, serverlessVectorDefaultSecret, consts.ServerlessObservabilityNamespace, false),
					),
					HelmError: true,
					Config:    cfg,
					Templates: config.Templates{"sharedpool/vector-values-aws.yaml": tmpl},
				}
			}(),
			wantErr: true,
			verify: func(t *testing.T, opts testkits.TaskOptions) {
				role := &rbacv1.Role{}
				err := opts.Client.Get(context.Background(), types.NamespacedName{
					Name:      serverlessVectorName,
					Namespace: consts.ServerlessObservabilityNamespace,
				}, role)
				assert.NoError(t, err)

				roleBinding := &rbacv1.RoleBinding{}
				err = opts.Client.Get(context.Background(), types.NamespacedName{
					Name:      serverlessVectorName,
					Namespace: consts.ServerlessObservabilityNamespace,
				}, roleBinding)
				assert.NoError(t, err)

				tlsSecret := &corev1.Secret{}
				err = opts.Client.Get(context.Background(), types.NamespacedName{
					Name:      serverlessTLSSecretName,
					Namespace: consts.ServerlessObservabilityNamespace,
				}, tlsSecret)
				assert.NoError(t, err)
				assert.True(t, strings.HasPrefix(string(tlsSecret.Data[corev1.TLSPrivateKeyKey]), "-----BEGIN PRIVATE KEY-----"))
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			task := makeVectorTask(tt.agent, tt.opts)
			result := task.Create(context.Background())
			if tt.wantErr {
				assert.Error(t, result.Error())
			} else {
				assert.NoError(t, result.Error())
			}
			if tt.verify != nil {
				tt.verify(t, tt.opts)
			}
		})
	}
}

func TestVectorTask_Delete(t *testing.T) {
	tests := []struct {
		name    string
		agent   *agentv2.SharedPoolAgent
		opts    testkits.TaskOptions
		wantErr bool
	}{
		{
			name:  "delete fails with remote client error",
			agent: testkits.NewSharedPoolAgentBuilder().Build(),
			opts: testkits.TaskOptions{
				ClientError: true,
				Config:      baseVectorConfig(),
			},
			wantErr: true,
		},
		{
			name:  "delete succeeds with valid remote client",
			agent: testkits.NewSharedPoolAgentBuilder().Build(),
			opts: testkits.TaskOptions{
				Client: testkits.BuildClient(),
				Config: baseVectorConfig(),
			},
			wantErr: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			task := makeVectorTask(tt.agent, tt.opts)
			result := task.Delete(context.Background())
			if tt.wantErr {
				assert.Error(t, result.Error())
			} else {
				assert.True(t, result.IsDoneState() || result.Error() != nil)
			}
		})
	}
}

func TestVectorTask_buildInstanceIDs(t *testing.T) {
	tests := []struct {
		name             string
		remoteClient     client.Client
		wantInstanceIDs  []string
		wantInstanceSize int
	}{
		{
			name: "extracts instanceIDs from TiKV pods",
			remoteClient: testkits.BuildClient(
				&corev1.Pod{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "db-instance123-tikv-0",
						Namespace: "tidbpool-456",
						Labels:    map[string]string{"pingcap.com/component": "tikv"},
					},
				},
				&corev1.Pod{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "db-instance456-tikv-0",
						Namespace: "tidbpool-456",
						Labels:    map[string]string{"pingcap.com/component": "tikv"},
					},
				},
				&corev1.Pod{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "pd-0",
						Namespace: "tidbpool-456",
						Labels:    map[string]string{"pingcap.com/component": "pd"},
					},
				},
			),
			wantInstanceIDs:  []string{"instance123", "instance456"},
			wantInstanceSize: 2,
		},
		{
			name:             "returns empty when no TiKV pods",
			remoteClient:     testkits.BuildClient(),
			wantInstanceIDs:  nil,
			wantInstanceSize: 0,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			agent := testkits.NewSharedPoolAgentBuilder().Build()
			task := makeVectorTask(agent, testkits.TaskOptions{Client: tt.remoteClient})
			instanceIDs, err := task.buildInstanceIDs(context.Background(), tt.remoteClient)
			assert.NoError(t, err)
			assert.Len(t, instanceIDs, tt.wantInstanceSize)
			for _, instanceID := range tt.wantInstanceIDs {
				assert.Contains(t, instanceIDs, instanceID)
			}
		})
	}
}

func TestVectorTask_buildTargets(t *testing.T) {
	tests := []struct {
		name    string
		config  config.RootConfig
		fakeCli client.Client
		verify  func(t *testing.T, targets []vectorTarget)
	}{
		{
			name:   "premium mode builds targets from TiKV pods",
			config: baseVectorConfig(),
			fakeCli: testkits.BuildClient(
				&corev1.Pod{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "db-instance123-tikv-0",
						Namespace: "tidbpool-456",
						Labels:    map[string]string{"pingcap.com/component": "tikv"},
					},
				},
				&corev1.Pod{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "db-instance456-tikv-0",
						Namespace: "tidbpool-456",
						Labels:    map[string]string{"pingcap.com/component": "tikv"},
					},
				},
			),
			verify: func(t *testing.T, targets []vectorTarget) {
				assert.Len(t, targets, 2)
				assert.Equal(t, "instance123", targets[0].InstanceID)
				assert.Equal(t, "o11y-vector-instance123", targets[0].ReleaseName)
				assert.Equal(t, "instance456", targets[1].InstanceID)
				assert.Equal(t, "o11y-vector-instance456", targets[1].ReleaseName)
			},
		},
		{
			name:   "premium mode returns nil when no TiKV pods",
			config: baseVectorConfig(),
			fakeCli: testkits.BuildClient(
				&corev1.Pod{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "pd-0",
						Namespace: "tidbpool-456",
						Labels:    map[string]string{"pingcap.com/component": "pd"},
					},
				},
			),
			verify: func(t *testing.T, targets []vectorTarget) {
				assert.Nil(t, targets)
			},
		},
		{
			name: "serverless mode ignores TiKV pods",
			config: func() config.RootConfig {
				cfg := baseVectorConfig()
				cfg.Mode = config.ModeServerless
				return cfg
			}(),
			fakeCli: testkits.BuildClient(
				&corev1.Pod{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "db-instance456-tikv-0",
						Namespace: consts.ServerlessObservabilityNamespace,
						Labels:    map[string]string{"pingcap.com/component": "tikv"},
					},
				},
				&corev1.Pod{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "db-instance123-tikv-0",
						Namespace: consts.ServerlessObservabilityNamespace,
						Labels:    map[string]string{"pingcap.com/component": "tikv"},
					},
				},
			),
			verify: func(t *testing.T, targets []vectorTarget) {
				assert.Len(t, targets, 1)
				assert.Empty(t, targets[0].InstanceID)
				assert.Equal(t, serverlessVectorName, targets[0].ReleaseName)
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			agent := testkits.NewSharedPoolAgentBuilder().Build()
			task := makeVectorTask(agent, testkits.TaskOptions{
				Config: tt.config,
				Client: tt.fakeCli,
			})
			targets, err := task.buildTargets(context.Background(), tt.fakeCli)
			assert.NoError(t, err)
			tt.verify(t, targets)
		})
	}
}

func TestVectorTask_listVectorPVCNames(t *testing.T) {
	tests := []struct {
		name        string
		config      config.RootConfig
		fakeCli     client.Client
		releaseName string
		namespace   string
		wantNames   []string
	}{
		{
			name: "serverless mode",
			config: func() config.RootConfig {
				cfg := baseVectorConfig()
				cfg.Mode = config.ModeServerless
				return cfg
			}(),
			fakeCli: testkits.BuildClient(
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-sharedpool-2",
						Namespace: consts.ServerlessObservabilityNamespace,
					},
				},
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-sharedpool-0",
						Namespace: consts.ServerlessObservabilityNamespace,
					},
				},
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-sharedpool-1",
						Namespace: consts.ServerlessObservabilityNamespace,
					},
				},
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-other-release-0",
						Namespace: consts.ServerlessObservabilityNamespace,
					},
				},
			),
			releaseName: serverlessVectorName,
			namespace:   consts.ServerlessObservabilityNamespace,
			wantNames: []string{
				"data-o11y-vector-sharedpool-0",
				"data-o11y-vector-sharedpool-1",
				"data-o11y-vector-sharedpool-2",
			},
		},
		{
			name:   "premium mode",
			config: baseVectorConfig(),
			fakeCli: testkits.BuildClient(
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-instance123-1",
						Namespace: "tidbpool-456",
					},
				},
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-instance123-0",
						Namespace: "tidbpool-456",
					},
				},
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-instance456-0", // different release, should be ignored
						Namespace: "tidbpool-456",
					},
				},
			),
			releaseName: "o11y-vector-instance123",
			namespace:   "tidbpool-456",
			wantNames: []string{
				"data-o11y-vector-instance123-0",
				"data-o11y-vector-instance123-1",
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			task := makeVectorTask(testkits.NewSharedPoolAgentBuilder().Build(), testkits.TaskOptions{Config: tt.config})
			names, err := task.listVectorPVCNames(context.Background(), tt.fakeCli, tt.releaseName, tt.namespace)
			assert.NoError(t, err)
			assert.Equal(t, tt.wantNames, names)
		})
	}
}

func TestVectorTask_buildKeyspaceToVmTenants(t *testing.T) {
	tests := []struct {
		name         string
		localObjects []client.Object
		want         string
		wantErr      bool
	}{
		{
			name:         "returns empty when no TiDBGroupAgents",
			localObjects: nil,
			want:         "",
		},
		{
			name: "skips TiDBGroupAgents with different SharedPoolID",
			localObjects: []client.Object{
				testkits.NewTiDBGroupAgentBuilder().
					WithSharedPoolID("other-pool").
					WithClusterID("cluster-1").
					WithAnnotations(map[string]string{
						consts.AnnotationKeyO11yVMAccountID: "acct-1",
						consts.AnnotationKeyO11yVMProjectID: "proj-1",
					}).
					Build(),
			},
			want: "",
		},
		{
			name: "skips TiDBGroupAgents missing required annotations",
			localObjects: []client.Object{
				testkits.NewTiDBGroupAgentBuilder().
					WithClusterID("cluster-1").
					Build(),
			},
			want: "",
		},
		{
			name: "builds sorted keyspace string from matching agents",
			localObjects: []client.Object{
				testkits.NewTiDBGroupAgentBuilder().
					WithName("agent-b").
					WithClusterID("cluster-2").
					WithAnnotations(map[string]string{
						consts.AnnotationKeyO11yVMAccountID: "acct-2",
						consts.AnnotationKeyO11yVMProjectID: "proj-2",
					}).
					Build(),
				testkits.NewTiDBGroupAgentBuilder().
					WithName("agent-a").
					WithClusterID("cluster-1").
					WithAnnotations(map[string]string{
						consts.AnnotationKeyO11yVMAccountID: "acct-1",
						consts.AnnotationKeyO11yVMProjectID: "proj-1",
					}).
					Build(),
			},
			want: "cluster-1:acct-1:proj-1,cluster-2:acct-2:proj-2",
		},
		{
			name: "mix of matching and non-matching agents",
			localObjects: []client.Object{
				testkits.NewTiDBGroupAgentBuilder().
					WithName("agent-a").
					WithClusterID("cluster-1").
					WithAnnotations(map[string]string{
						consts.AnnotationKeyO11yVMAccountID: "acct-1",
						consts.AnnotationKeyO11yVMProjectID: "proj-1",
					}).
					Build(),
				testkits.NewTiDBGroupAgentBuilder().
					WithName("agent-b").
					WithSharedPoolID("other-pool").
					WithClusterID("cluster-2").
					WithAnnotations(map[string]string{
						consts.AnnotationKeyO11yVMAccountID: "acct-2",
						consts.AnnotationKeyO11yVMProjectID: "proj-2",
					}).
					Build(),
				testkits.NewTiDBGroupAgentBuilder().
					WithName("agent-c").
					WithClusterID("cluster-3").
					Build(),
			},
			want: "cluster-1:acct-1:proj-1",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			agent := testkits.NewSharedPoolAgentBuilder().Build()
			task := makeVectorTask(agent, testkits.TaskOptions{
				Config:       baseVectorConfig(),
				LocalObjects: tt.localObjects,
			})
			got, err := task.buildKeyspaceToVmTenants(context.Background())
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tt.want, got)
			}
		})
	}
}

func TestVectorTask_deleteVectorPVCs(t *testing.T) {
	tests := []struct {
		name        string
		config      config.RootConfig
		fakeCli     client.Client
		releaseName string
		wantErr     bool
		verify      func(t *testing.T, cli client.Client, config config.RootConfig)
	}{
		{
			name:   "serverless mode deletes PVCs for release",
			config: func() config.RootConfig { cfg := baseVectorConfig(); cfg.Mode = config.ModeServerless; return cfg }(),
			fakeCli: testkits.BuildClient(
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-sharedpool-0",
						Namespace: consts.ServerlessObservabilityNamespace,
					},
				},
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-sharedpool-1",
						Namespace: consts.ServerlessObservabilityNamespace,
					},
				},
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-other-release-0",
						Namespace: consts.ServerlessObservabilityNamespace,
					},
				},
			),
			releaseName: serverlessVectorName,
			verify: func(t *testing.T, cli client.Client, cfg config.RootConfig) {
				pvcList := &corev1.PersistentVolumeClaimList{}
				err := cli.List(context.Background(), pvcList, client.InNamespace(consts.ServerlessObservabilityNamespace))
				assert.NoError(t, err)
				assert.Len(t, pvcList.Items, 1)
				assert.Equal(t, "data-other-release-0", pvcList.Items[0].Name)
			},
		},
		{
			name:   "premium mode deletes PVCs for release",
			config: baseVectorConfig(),
			fakeCli: testkits.BuildClient(
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-instance123-0",
						Namespace: "tidbpool-456",
					},
				},
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-instance123-1",
						Namespace: "tidbpool-456",
					},
				},
				&corev1.PersistentVolumeClaim{
					ObjectMeta: metav1.ObjectMeta{
						Name:      "data-o11y-vector-instance456-0",
						Namespace: "tidbpool-456",
					},
				},
			),
			releaseName: "o11y-vector-instance123",
			verify: func(t *testing.T, cli client.Client, cfg config.RootConfig) {
				pvcList := &corev1.PersistentVolumeClaimList{}
				err := cli.List(context.Background(), pvcList, client.InNamespace("tidbpool-456"))
				assert.NoError(t, err)
				assert.Len(t, pvcList.Items, 1)
				assert.Equal(t, "data-o11y-vector-instance456-0", pvcList.Items[0].Name)
			},
		},
		{
			name:        "no matching PVCs is not an error",
			config:      baseVectorConfig(),
			fakeCli:     testkits.BuildClient(),
			releaseName: "o11y-vector-nonexistent",
			wantErr:     false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			task := makeVectorTask(testkits.NewSharedPoolAgentBuilder().Build(), testkits.TaskOptions{Config: tt.config})
			err := task.deleteVectorPVCs(context.Background(), tt.fakeCli, tt.releaseName)
			if tt.wantErr {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
			}
			if tt.verify != nil {
				tt.verify(t, tt.fakeCli, tt.config)
			}
		})
	}
}

func TestBuildTargetsFromInstanceIDs(t *testing.T) {
	tests := []struct {
		name      string
		instances map[string]struct{}
		wantLen   int
		wantIDs   []string
	}{
		{
			name:      "empty map returns nil",
			instances: map[string]struct{}{},
			wantLen:   0,
			wantIDs:   nil,
		},
		{
			name:      "single entry",
			instances: map[string]struct{}{"abc": {}},
			wantLen:   1,
			wantIDs:   []string{"abc"},
		},
		{
			name:      "multiple entries sorted by instanceID",
			instances: map[string]struct{}{"zzz": {}, "aaa": {}, "mmm": {}},
			wantLen:   3,
			wantIDs:   []string{"aaa", "mmm", "zzz"},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			agent := testkits.NewSharedPoolAgentBuilder().Build()
			task := makeVectorTask(agent, testkits.TaskOptions{})
			got := task.buildTargetsFromInstanceIDs(tt.instances)
			if tt.wantIDs == nil {
				assert.Nil(t, got)
				return
			}
			assert.Len(t, got, tt.wantLen)
			for i, id := range tt.wantIDs {
				assert.Equal(t, id, got[i].InstanceID)
				assert.Equal(t, "o11y-vector-"+id, got[i].ReleaseName)
			}
		})
	}
}