//go:build unit

package task

import (
	"testing"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
)

func newTestCRRFDaemonConfig(overrides func(*CRRFDaemonConfig)) *CRRFDaemonConfig {
	cfg := &CRRFDaemonConfig{
		backup: &cloudv1.Backup{
			ObjectMeta: metav1.ObjectMeta{Name: "my-log-backup"},
		},
		cluster: &cloudv1.Cluster{
			ObjectMeta: metav1.ObjectMeta{Name: "abc123"},
		},
		upstreamStorage: "s3://bucket/backup",
		taskName:        "my-log-backup",
		pdAddr:          "tidbcluster-pd.tidb-ns.svc:2379",
	}
	if overrides != nil {
		overrides(cfg)
	}
	return cfg
}

func TestCRRFDaemonConfig_GetName(t *testing.T) {
	cases := map[string]struct {
		backupName string
		want       string
	}{
		"normal backup name produces prefixed release name": {
			backupName: "my-log-backup",
			want:       "crrf-daemon-my-log-backup",
		},
	}
	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			cfg := newTestCRRFDaemonConfig(func(c *CRRFDaemonConfig) {
				c.backup.Name = tc.backupName
			})
			if got := cfg.GetName(); got != tc.want {
				t.Errorf("GetName() = %q, want %q", got, tc.want)
			}
		})
	}
}

func TestCRRFDaemonConfig_GetNamespace(t *testing.T) {
	cfg := newTestCRRFDaemonConfig(func(c *CRRFDaemonConfig) {
		c.cluster.Name = "abc123"
	})
	if got := cfg.GetNamespace(); got != "tidbabc123" {
		t.Errorf("GetNamespace() = %q, want %q", got, "tidbabc123")
	}
}

func TestCRRFDaemonConfig_ToHelmValues(t *testing.T) {
	cases := map[string]struct {
		setup          func(*CRRFDaemonConfig)
		checkImage     func(t *testing.T, values map[string]interface{})
		wantDownstream bool
	}{
		"no image set leaves image key absent": {
			setup: func(c *CRRFDaemonConfig) { c.image = "" },
			checkImage: func(t *testing.T, values map[string]interface{}) {
				t.Helper()
				if _, ok := values["image"]; ok {
					t.Error("expected 'image' key to be absent when image is empty")
				}
			},
		},
		"image with tag is split into repository and tag": {
			setup: func(c *CRRFDaemonConfig) { c.image = "pingcap/br:v6.5.0" },
			checkImage: func(t *testing.T, values map[string]interface{}) {
				t.Helper()
				img, ok := values["image"].(map[string]interface{})
				if !ok {
					t.Fatal("expected 'image' to be a map")
				}
				if img["repository"] != "pingcap/br" {
					t.Errorf("repository = %q, want %q", img["repository"], "pingcap/br")
				}
				if img["tag"] != "v6.5.0" {
					t.Errorf("tag = %q, want %q", img["tag"], "v6.5.0")
				}
			},
		},
		"image from registry with port is split at last colon": {
			setup: func(c *CRRFDaemonConfig) {
				c.image = "123456789.dkr.ecr.us-east-1.amazonaws.com/pingcap/br:v6.5.0"
			},
			checkImage: func(t *testing.T, values map[string]interface{}) {
				t.Helper()
				img, ok := values["image"].(map[string]interface{})
				if !ok {
					t.Fatal("expected 'image' to be a map")
				}
				if img["repository"] != "123456789.dkr.ecr.us-east-1.amazonaws.com/pingcap/br" {
					t.Errorf("repository = %q", img["repository"])
				}
				if img["tag"] != "v6.5.0" {
					t.Errorf("tag = %q, want %q", img["tag"], "v6.5.0")
				}
			},
		},
		"image without tag sets empty tag": {
			setup: func(c *CRRFDaemonConfig) { c.image = "pingcap/br" },
			checkImage: func(t *testing.T, values map[string]interface{}) {
				t.Helper()
				img, ok := values["image"].(map[string]interface{})
				if !ok {
					t.Fatal("expected 'image' to be a map")
				}
				if img["repository"] != "pingcap/br" {
					t.Errorf("repository = %q, want %q", img["repository"], "pingcap/br")
				}
				if img["tag"] != "" {
					t.Errorf("tag = %q, want empty", img["tag"])
				}
			},
		},
		"downstream storage included when set": {
			setup:          func(c *CRRFDaemonConfig) { c.downstreamStorage = "s3://dst-bucket/backup" },
			wantDownstream: true,
			checkImage:     func(t *testing.T, _ map[string]interface{}) {},
		},
		"downstream storage absent when empty": {
			setup:          func(c *CRRFDaemonConfig) { c.downstreamStorage = "" },
			wantDownstream: false,
			checkImage:     func(t *testing.T, _ map[string]interface{}) {},
		},
	}

	for name, tc := range cases {
		t.Run(name, func(t *testing.T) {
			cfg := newTestCRRFDaemonConfig(tc.setup)
			values := cfg.ToHelmValues()

			// Validate args sub-map is present
			args, ok := values["args"].(map[string]interface{})
			if !ok {
				t.Fatal("expected 'args' to be a map in helm values")
			}

			// Always-present fields inside args
			for _, key := range []string{"upstreamStorage", "taskName", "pdAddr", "statusAddr"} {
				if _, ok := args[key]; !ok {
					t.Errorf("expected key %q to be present in helm values args", key)
				}
			}

			tc.checkImage(t, values)

			if tc.wantDownstream {
				if _, ok := args["downstreamStorage"]; !ok {
					t.Error("expected 'downstreamStorage' key to be present in args")
				}
			} else {
				if _, ok := args["downstreamStorage"]; ok {
					t.Error("expected 'downstreamStorage' key to be absent from args when empty")
				}
			}
		})
	}
}

// Regression test: before the fix, image was set as the full "repo:tag" string in
// "repository" with an empty "tag", causing Helm to render "repo:tag:" (double colon).
func TestCRRFDaemonConfig_ToHelmValues_ImageNeverProducesDoubleColon(t *testing.T) {
	cfg := newTestCRRFDaemonConfig(func(c *CRRFDaemonConfig) {
		c.image = "pingcap/br:v6.5.0"
	})
	values := cfg.ToHelmValues()
	img := values["image"].(map[string]interface{})
	repo := img["repository"].(string)
	tag := img["tag"].(string)
	full := repo + ":" + tag
	// Should produce "pingcap/br:v6.5.0", not "pingcap/br:v6.5.0:"
	if full != "pingcap/br:v6.5.0" {
		t.Errorf("rendered image = %q, want %q", full, "pingcap/br:v6.5.0")
	}
}

// Ensure Backup.Spec.Storage being nil is handled upstream; test that Storage ptr helper works.
func TestCRRFStorageDefault(t *testing.T) {
	// CloudV1 uses *string for Spec.Storage; ptr.Deref should return "" for nil.
	var s *string
	got := ptrDeref(s, "")
	if got != "" {
		t.Errorf("expected empty string for nil pointer, got %q", got)
	}
	val := "s3://bucket"
	got = ptrDeref(&val, "")
	if got != "s3://bucket" {
		t.Errorf("expected %q, got %q", "s3://bucket", got)
	}
}

// ptrDeref is a local helper equivalent to k8s.io/utils/ptr.Deref for testing.
func ptrDeref(p *string, def string) string {
	if p == nil {
		return def
	}
	return *p
}