package main

import (
	"bytes"
	"fmt"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"testing"

	"gopkg.in/yaml.v2"
)

func TestWriteSectionsPreservesUnrelatedConfig(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	original := `orgID: "org-before"
projectID: "project-before"
kafka:
  - bootstrapEndpoints: old.example:39092
    password: old
cases2Run:
  - ChangefeedBasic
mysql:
  - endpoint: old.example:3306
    password: old
mustCleanup: true
`
	if err := os.WriteFile(path, []byte(original), 0o600); err != nil {
		t.Fatal(err)
	}

	sections := map[string]interface{}{
		"kafka": []interface{}{map[interface{}]interface{}{
			"bootstrapEndpoints": "new.example:40001",
			"password":           "new-secret",
		}},
		"mysql": []interface{}{map[interface{}]interface{}{
			"endpoint": "new.example:40002",
			"password": "new-secret",
		}},
		"schemaRegistry": nil,
	}
	writeSections(path, sections, 0o600)

	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	result := string(data)
	for _, preserved := range []string{
		`orgID: "org-before"`,
		`projectID: "project-before"`,
		"cases2Run:\n  - ChangefeedBasic",
		"mustCleanup: true",
	} {
		if !strings.Contains(result, preserved) {
			t.Fatalf("unrelated content was not preserved: %q\n%s", preserved, result)
		}
	}
	if strings.Contains(result, "old.example") || strings.Count(result, "new.example") != 2 {
		t.Fatalf("test-server blocks were not replaced exactly once:\n%s", result)
	}
	if strings.Count(result, "new-secret") != 2 {
		t.Fatalf("test-server credentials were not written exactly once per dependency:\n%s", result)
	}
	if strings.HasSuffix(result, "\n\n") {
		t.Fatalf("generated config has an extra blank line at EOF")
	}

	writeSections(path, sections, 0o600)
	second, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	if string(second) != result {
		t.Fatalf("test-server rewrite is not idempotent:\nfirst:\n%s\nsecond:\n%s", result, second)
	}
}

func TestTrackedSectionsRetainCredentials(t *testing.T) {
	t.Parallel()
	environment := map[string]interface{}{
		"mysql": []interface{}{map[interface{}]interface{}{
			"user":     "e2e_admin",
			"password": "generated-password",
		}},
		"pulsar": []interface{}{map[interface{}]interface{}{
			"token": "generated-token",
		}},
	}

	trackedYAML := renderSections(environment)
	if !strings.Contains(trackedYAML, "generated-password") || !strings.Contains(trackedYAML, "generated-token") {
		t.Fatalf("tracked rendering is missing generated test credentials")
	}
	if strings.Contains(trackedYAML, "SET_IN_IGNORED_CONFIG_LOCAL") {
		t.Fatalf("tracked rendering still contains a credential marker")
	}
}

func TestReplaceStaticCloudCredentialsWithPlaceholders(t *testing.T) {
	t.Parallel()
	combined := map[string]interface{}{
		"s3": []interface{}{map[interface{}]interface{}{
			"accessKey": "generated-access-key",
			"secretKey": "generated-secret-key",
			"bucketURI": "s3://example/cdc/",
		}},
		"azureBlobStorage": []interface{}{map[interface{}]interface{}{
			"sasToken":     "generated-sas-token",
			"containerURI": "https://example.blob.core.windows.net/cdc/",
		}},
	}

	replaceStaticCloudCredentialsWithPlaceholders(combined)

	s3 := combined["s3"].([]interface{})[0].(map[interface{}]interface{})
	if s3["accessKey"] != s3AccessKeyPlaceholder || s3["secretKey"] != s3SecretKeyPlaceholder {
		t.Fatalf("S3 credentials were not replaced with placeholders: %#v", s3)
	}
	if s3["bucketURI"] != "s3://example/cdc/" {
		t.Fatalf("S3 non-credential fields changed: %#v", s3)
	}

	azure := combined["azureBlobStorage"].([]interface{})[0].(map[interface{}]interface{})
	if azure["sasToken"] != azureSASPlaceholder {
		t.Fatalf("Azure SAS was not replaced with a placeholder: %#v", azure)
	}
	if azure["containerURI"] != "https://example.blob.core.windows.net/cdc/" {
		t.Fatalf("Azure non-credential fields changed: %#v", azure)
	}
}

func TestCommonOwnsAllGeneratedDependencies(t *testing.T) {
	t.Parallel()
	combined := map[string]interface{}{
		"kafka": []interface{}{
			map[interface{}]interface{}{"networkType": "NetworkType_Public", "authType": "DISABLE"},
			map[interface{}]interface{}{"networkType": "NetworkType_Public", "authType": "SASL_PLAIN"},
			map[interface{}]interface{}{"networkType": "NetworkType_Private_Link", "authType": "DISABLE"},
		},
		"mysql":            []interface{}{map[interface{}]interface{}{"endpoint": "example:40000"}},
		"pulsar":           []interface{}{map[interface{}]interface{}{"endpoint": "example:40001"}},
		"s3":               []interface{}{map[interface{}]interface{}{"bucketURI": "s3://example/"}},
		"gcs":              []interface{}{map[interface{}]interface{}{"bucketURI": "gs://example/"}},
		"schemaRegistry":   map[interface{}]interface{}{"password": "schema-secret"},
		"azureBlobStorage": []interface{}{},
		"backupExport":     map[interface{}]interface{}{},
		"essentialExport":  map[interface{}]interface{}{},
	}

	common := commonSections(combined)
	environment := environmentSections(combined)
	if got := len(common["kafka"].([]interface{})); got != 3 {
		t.Fatalf("common Kafka count = %d, want 3", got)
	}
	for _, key := range sectionOrder {
		if _, ok := common[key]; !ok {
			t.Fatalf("shared dependency %s must be written to config.common.yaml", key)
		}
		if _, ok := environment[key]; ok {
			t.Fatalf("environment config must inherit %s from config.common.yaml", key)
		}
	}
}

func TestMergeImportSampleDataOSSPreservesExternalSources(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	original := `mustCleanup: true
importSampleData:
  s3:
    sourceURL: "s3://tidbcloud-sample-data/data-ingestion/"
    roleARN: "arn:aws:iam::801626783489:role/import-sample-access"
  gcs:
    sourceURL: "gs://tidbcloud-samples-us-west1/"
  azureBlob:
    sourceURL: "https://tcidmsampledata.blob.core.windows.net/sql/"
  oss:
    sourceURL: "oss://old-bucket/test-with-schema/"
    accessKeyID: "old-ak"
    accessKeySecret: "old-sk"
    importedDBTable:
      database: "test"
      table: "a"
`
	if err := os.WriteFile(path, []byte(original), 0o644); err != nil {
		t.Fatal(err)
	}

	combined := map[string]interface{}{
		"importSampleData": map[interface{}]interface{}{
			"oss": map[interface{}]interface{}{
				"sourceURL":       "oss://dit-abc123/import/",
				"endpoint":        "oss-ap-southeast-1.aliyuncs.com",
				"accessKeyID":     "generated-ak",
				"accessKeySecret": "generated-sk",
			},
		},
	}
	mergeImportSampleDataOSS(combined, path)

	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	result := string(data)
	for _, preserved := range []string{
		"s3://tidbcloud-sample-data/data-ingestion/",
		"arn:aws:iam::801626783489:role/import-sample-access",
		"gs://tidbcloud-samples-us-west1/",
		"https://tcidmsampledata.blob.core.windows.net/sql/",
		"database: test",
	} {
		if !strings.Contains(result, preserved) {
			t.Fatalf("external import sample source was not preserved: %q\n%s", preserved, result)
		}
	}
	for _, want := range []string{
		"oss://dit-abc123/import/",
		"https://oss-ap-southeast-1.aliyuncs.com",
		ossAccessKeyPlaceholder,
		ossSecretKeyPlaceholder,
	} {
		if !strings.Contains(result, want) {
			t.Fatalf("generated OSS import config missing %q\n%s", want, result)
		}
	}
	for _, forbidden := range []string{"old-ak", "old-sk", "generated-ak", "generated-sk"} {
		if strings.Contains(result, forbidden) {
			t.Fatalf("OSS AK/SK credential leaked into output: %q", forbidden)
		}
	}
}

func TestMergeImportSampleDataOSSReplacesCompleteGeneratedBlocks(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	original := `importSampleData:
  oss:
    customCSVFormat:
      delimiter: old
      unmanaged: preserve-only-when-not-generated
    importedDBTable:
      database: old
      stale: remove-when-generated
`
	if err := os.WriteFile(path, []byte(original), 0o644); err != nil {
		t.Fatal(err)
	}
	combined := map[string]interface{}{
		"importSampleData": map[interface{}]interface{}{
			"oss": map[interface{}]interface{}{
				"endpoint": "oss-ap-southeast-1.aliyuncs.com",
				"customCSVFormat": map[interface{}]interface{}{
					"delimiter":      `"`,
					"escapeChar":     `\`,
					"lineTerminator": "\n",
					"nullValue":      `\N`,
					"separator":      ";",
					"skipRows":       1,
					"splitFile":      false,
				},
				"importedDBTable": map[interface{}]interface{}{
					"database": "test",
					"table":    "a",
				},
			},
		},
	}
	mergeImportSampleDataOSS(combined, path)
	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	result := string(data)
	if strings.Contains(result, "stale: remove-when-generated") || strings.Contains(result, "unmanaged: preserve-only-when-not-generated") {
		t.Fatalf("generated nested blocks were not replaced atomically:\n%s", result)
	}
	if !strings.Contains(result, "https://oss-ap-southeast-1.aliyuncs.com") {
		t.Fatalf("generated OSS fields are missing or endpoint was not normalized:\n%s", result)
	}
	parsed := map[string]interface{}{}
	if err := yaml.Unmarshal(data, &parsed); err != nil {
		t.Fatal(err)
	}
	oss := importSampleDataSubMap(parsed["importSampleData"], "oss")
	format := mapField(oss, "customCSVFormat")
	if lineTerminator, ok := mapStringField(format, "lineTerminator"); !ok || lineTerminator != "\n" {
		t.Fatalf("generated CSV line terminator = %q, want an actual LF", lineTerminator)
	}
	first := append([]byte(nil), data...)
	mergeImportSampleDataOSS(combined, path)
	second, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	if !bytes.Equal(first, second) {
		t.Fatalf("full OSS import merge is not byte-idempotent:\nfirst:\n%s\nsecond:\n%s", first, second)
	}
}

func TestRemoveManagedImportSampleDataFieldsPreservesUnknownFields(t *testing.T) {
	t.Parallel()
	current := map[string]interface{}{}
	head := map[string]interface{}{}
	for _, parsed := range []map[string]interface{}{current, head} {
		parsed["importSampleData"] = map[interface{}]interface{}{
			"oss": map[interface{}]interface{}{
				"sourceURL": "generated",
				"customCSVFormat": map[interface{}]interface{}{
					"delimiter": `"`,
					"unmanaged": "same",
				},
			},
		}
	}
	removeManagedImportSampleDataFields(current)
	removeManagedImportSampleDataFields(head)
	if !reflect.DeepEqual(current, head) {
		t.Fatalf("managed field removal should produce equal maps: %#v != %#v", current, head)
	}
	if strings.Contains(fmt.Sprintf("%#v", current), "delimiter") {
		t.Fatalf("managed nested field was not removed: %#v", current)
	}
	current["importSampleData"].(map[interface{}]interface{})["oss"].(map[interface{}]interface{})["customCSVFormat"].(map[interface{}]interface{})["unmanaged"] = "changed"
	if reflect.DeepEqual(current, head) {
		t.Fatal("unknown nested field changes must remain detectable")
	}
}

func TestVerifyUnrelatedAgainstGitOwnership(t *testing.T) {
	// This test changes cwd and PATH for a short-lived fake git command, so it
	// must remain serial with other package tests.
	legalCurrent, head := verifierFixtureYAML()
	tests := map[string]struct {
		mutateCurrent    func(string) string
		mutateHead       func(string) string
		mutateEnvCurrent func(string) string
		mutateEnvHead    func(string) string
		wantPass         bool
	}{
		"managed import fields may change": {
			mutateCurrent: func(value string) string {
				value = strings.Replace(value, "oss://new-bucket/import/csv/", "oss://current-bucket/import/csv/", 1)
				value = strings.Replace(value, "https://oss-ap-southeast-1.aliyuncs.com", "https://current-oss.example", 1)
				value = strings.Replace(value, "database: test", "database: current", 1)
				value = strings.Replace(value, "delimiter: '\"'", "delimiter: ','", 1)
				return strings.Replace(value, "https://external-azure-samples/", "https://current-external-azure-samples/", 1)
			},
			mutateHead: func(value string) string {
				value = strings.Replace(value, "oss://new-bucket/import/csv/", "oss://old-bucket/import/csv/", 1)
				value = strings.Replace(value, "https://oss-ap-southeast-1.aliyuncs.com", "https://old-oss.example", 1)
				value = strings.Replace(value, "database: test", "database: old", 1)
				value = strings.Replace(value, "delimiter: '\"'", "delimiter: ';'", 1)
				return strings.Replace(value, "https://external-azure-samples/", "https://old-external-azure-samples/", 1)
			},
			wantPass: true,
		},
		"external S3 import field is rejected": {
			mutateCurrent: func(value string) string {
				return strings.Replace(value, "s3://external-samples/", "s3://changed-external-samples/", 1)
			},
			mutateHead: func(value string) string { return value },
		},
		"external GCS import field is rejected": {
			mutateCurrent: func(value string) string {
				return strings.Replace(value, "gs://external-samples/", "gs://changed-external-samples/", 1)
			},
			mutateHead: func(value string) string { return value },
		},
		"unknown OSS nested field is rejected": {
			mutateCurrent: func(value string) string {
				return strings.Replace(value, "unmanaged: same", "unmanaged: changed", 1)
			},
			mutateHead: func(value string) string { return value },
		},
		"malformed OSS nested block is rejected": {
			mutateCurrent: func(value string) string {
				start := strings.Index(value, "    customCSVFormat:\n")
				end := strings.Index(value[start:], "\ns3:")
				return value[:start] + "    customCSVFormat: malformed" + value[start+end:]
			},
			mutateHead: func(value string) string { return value },
		},
		"environment change is rejected": {
			mutateCurrent: func(value string) string { return value },
			mutateHead: func(value string) string {
				return value
			},
			mutateEnvCurrent: func(value string) string { return strings.Replace(value, "org-same", "org-current", 1) },
			mutateEnvHead:    func(value string) string { return strings.Replace(value, "org-same", "org-head", 1) },
			wantPass:         false,
		},
		"real OSS credential is rejected": {
			mutateCurrent: func(value string) string {
				return strings.Replace(value, ossAccessKeyPlaceholder, "real-access-key", 1)
			},
			mutateHead: func(value string) string { return value },
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			current := tc.mutateCurrent(legalCurrent)
			expectedHead := tc.mutateHead(head)
			envCurrent, envHead := "orgID: org-same\n", "orgID: org-same\n"
			if tc.mutateEnvCurrent != nil {
				envCurrent = tc.mutateEnvCurrent(envCurrent)
			}
			if tc.mutateEnvHead != nil {
				envHead = tc.mutateEnvHead(envHead)
			}
			if got := runVerifierFixture(t, current, expectedHead, envCurrent, envHead); got != tc.wantPass {
				t.Fatalf("verifyUnrelatedAgainstGit pass=%v, want %v", got, tc.wantPass)
			}
		})
	}
}

func verifierFixtureYAML() (current, head string) {
	fixture := `runAsCI: true
mustCleanup: true
orgID: org-same
importSampleData:
  s3:
    sourceURL: s3://external-samples/
  gcs:
    sourceURL: gs://external-samples/
  azureBlob:
    sourceURL: https://external-azure-samples/
    sasToken: AZURE_SAS_TOKEN_PLACEHOLDER
  oss:
    sourceURL: oss://new-bucket/import/csv/
    endpoint: https://oss-ap-southeast-1.aliyuncs.com
    accessKeyID: OSS_ACCESS_KEY_PLACEHOLDER
    accessKeySecret: OSS_SECRET_KEY_PLACEHOLDER
    importedDBTable:
      database: test
      table: a
    customCSVFormat:
      delimiter: '"'
      escapeChar: \
      lineTerminator: "\n"
      nullValue: \\N
      separator: ;
      skipRows: 1
      splitFile: false
      unmanaged: same
s3:
  - accessKey: S3_ACCESS_KEY_PLACEHOLDER
    secretKey: S3_SECRET_KEY_PLACEHOLDER
azureBlobStorage:
  - sasToken: AZURE_SAS_TOKEN_PLACEHOLDER
kafka: []
mysql: []
pulsar: []
gcs: []
backupExport: {}
essentialExport: {}
schemaRegistry: {}
`
	return fixture, fixture
}

func runVerifierFixture(t *testing.T, current, head, envCurrent, envHead string) (passed bool) {
	t.Helper()
	dir := t.TempDir()
	if err := os.WriteFile(filepath.Join(dir, "config.common.yaml"), []byte(current), 0o644); err != nil {
		t.Fatal(err)
	}
	for _, path := range []string{"config.dev.yaml", "config.staging.yaml", "config.prod.yaml"} {
		if err := os.WriteFile(filepath.Join(dir, path), []byte(envCurrent), 0o644); err != nil {
			t.Fatal(err)
		}
	}
	headPath := filepath.Join(dir, "head-common.yaml")
	if err := os.WriteFile(headPath, []byte(head), 0o644); err != nil {
		t.Fatal(err)
	}
	headEnvPath := filepath.Join(dir, "head-env.yaml")
	if err := os.WriteFile(headEnvPath, []byte(envHead), 0o644); err != nil {
		t.Fatal(err)
	}
	gitPath := filepath.Join(dir, "git")
	gitScript := fmt.Sprintf("#!/bin/sh\ncase \"$2\" in\n  *config.common.yaml) /bin/cat %q ;;\n  *) /bin/cat %q ;;\nesac\n", headPath, headEnvPath)
	if err := os.WriteFile(gitPath, []byte(gitScript), 0o755); err != nil {
		t.Fatal(err)
	}
	oldWD, err := os.Getwd()
	if err != nil {
		t.Fatal(err)
	}
	oldPath := os.Getenv("PATH")
	if err := os.Chdir(dir); err != nil {
		t.Fatal(err)
	}
	if err := os.Setenv("PATH", dir+string(os.PathListSeparator)+oldPath); err != nil {
		t.Fatal(err)
	}
	defer func() {
		_ = os.Setenv("PATH", oldPath)
		_ = os.Chdir(oldWD)
	}()
	passed = true
	defer func() {
		if recover() != nil {
			passed = false
		}
	}()
	verifyUnrelatedAgainstGit()
	return passed
}

func TestImportSampleDataSubMap(t *testing.T) {
	t.Parallel()
	block := map[interface{}]interface{}{"oss": map[interface{}]interface{}{"accessKeyID": "x"}}
	if importSampleDataSubMap(block, "oss") == nil {
		t.Fatal("nested OSS value was not found for map[interface{}]interface{}")
	}
	stringMap := map[string]interface{}{"importSampleData": map[string]interface{}{"oss": map[string]interface{}{"accessKeyID": "x"}}}
	if importSampleDataSubMap(stringMap["importSampleData"], "oss") == nil {
		t.Fatal("nested OSS value was not found for map[string]interface{}")
	}
	if importSampleDataSubMap(nil, "oss") != nil {
		t.Fatal("absent importSampleData section must return nil")
	}
}

func TestMergeImportSampleDataAzureBlobPreservesExternalSources(t *testing.T) {
	t.Parallel()
	dir := t.TempDir()
	path := filepath.Join(dir, "config.yaml")
	original := `mustCleanup: true
importSampleData:
  s3:
    sourceURL: "s3://tidbcloud-sample-data/data-ingestion/"
  azureBlob:
    sourceURL: "https://old-samples.blob.core.windows.net/sql/"
    sasToken: "sv=old-token"
`
	if err := os.WriteFile(path, []byte(original), 0o644); err != nil {
		t.Fatal(err)
	}

	combined := map[string]interface{}{
		"importSampleData": map[interface{}]interface{}{
			"azureBlob": map[interface{}]interface{}{
				"sourceURL": "https://dit0af0ecda743c.blob.core.windows.net/import/",
				"sasToken":  "sv=generated-sas",
			},
		},
	}
	mergeImportSampleDataAzureBlob(combined, path)

	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	result := string(data)
	for _, preserved := range []string{
		"s3://tidbcloud-sample-data/data-ingestion/",
		"https://dit0af0ecda743c.blob.core.windows.net/import/",
	} {
		if !strings.Contains(result, preserved) {
			t.Fatalf("import sample source not present/updated: %q\n%s", preserved, result)
		}
	}
	if !strings.Contains(result, azureSASPlaceholder) {
		t.Fatalf("Azure SAS token was not redacted to placeholder\n%s", result)
	}
	for _, forbidden := range []string{"sv=old-token", "sv=generated-sas"} {
		if strings.Contains(result, forbidden) {
			t.Fatalf("Azure SAS token leaked into output: %q", forbidden)
		}
	}
}

func TestNormalizeKafkaDependenciesStripsProjectIDAndGCPZoneKey(t *testing.T) {
	t.Parallel()
	combined := map[string]interface{}{
		"kafka": []interface{}{
			map[interface{}]interface{}{
				"cloudProvider": "aws",
				"projectID":     "1,2",
				"dnsRecords":    map[interface{}]interface{}{"usw2-az1": "az1.example"},
			},
			map[interface{}]interface{}{
				"cloudProvider": "gcp",
				"projectID":     "1,2",
				"dnsRecords":    map[interface{}]interface{}{"us-west1-a": "us-west1.example"},
			},
		},
	}

	normalizeKafkaDependencies(combined)

	kafka := combined["kafka"].([]interface{})
	aws := kafka[0].(map[interface{}]interface{})
	if _, ok := aws["projectID"]; ok {
		t.Fatalf("AWS Kafka dependency kept projectID: %#v", aws)
	}
	awsRecords := aws["dnsRecords"].(map[interface{}]interface{})
	if v := awsRecords["usw2-az1"]; v != "az1.example" {
		t.Fatalf("AWS Kafka DNS records changed unexpectedly: %#v", awsRecords)
	}
	gcp := kafka[1].(map[interface{}]interface{})
	if _, ok := gcp["projectID"]; ok {
		t.Fatalf("GCP Kafka dependency kept projectID: %#v", gcp)
	}
	gcpRecords := gcp["dnsRecords"].(map[interface{}]interface{})
	if len(gcpRecords) != 1 || gcpRecords["all zones"] != "us-west1.example" {
		t.Fatalf("GCP Kafka DNS records were not normalized to 'all zones': %#v", gcpRecords)
	}
}

func TestDecodeTerraformFragmentRejectsMissingOutput(t *testing.T) {
	t.Parallel()

	defer func() {
		recovered := recover()
		if recovered == nil {
			t.Fatal("empty Terraform output did not fail")
		}
		message := recovered.(string)
		if !strings.Contains(message, "apply that cloud root") || !strings.Contains(message, "clouds/alicloud") {
			t.Fatalf("unexpected panic: %s", message)
		}
	}()

	decodeTerraformFragment("clouds/alicloud", nil)
}

func TestDecodeTerraformFragment(t *testing.T) {
	t.Parallel()

	fragment := decodeTerraformFragment("clouds/alicloud", []byte(`mysql:
- cloudProvider: alicloud
  password: test-password
`))
	mysql := fragment["mysql"].([]interface{})
	if len(mysql) != 1 {
		t.Fatalf("MySQL dependency count = %d, want 1", len(mysql))
	}
}