package main

import (
	"bytes"
	"flag"
	"fmt"
	"os"
	"os/exec"
	"path/filepath"
	"reflect"
	"sort"
	"strings"

	"gopkg.in/yaml.v2"
)

var sectionOrder = []string{
	"kafka", "mysql", "pulsar", "s3", "gcs", "azureBlobStorage", "backupExport", "essentialExport", "schemaRegistry",
}

const (
	s3AccessKeyPlaceholder  = "S3_ACCESS_KEY_PLACEHOLDER"
	s3SecretKeyPlaceholder  = "S3_SECRET_KEY_PLACEHOLDER"
	azureSASPlaceholder     = "AZURE_SAS_TOKEN_PLACEHOLDER"
	ossAccessKeyPlaceholder = "OSS_ACCESS_KEY_PLACEHOLDER"
	ossSecretKeyPlaceholder = "OSS_SECRET_KEY_PLACEHOLDER"
)

func main() {
	infra := flag.String("infra", "../infrastructure/date-integration-test-env", "multi-cloud Terraform directory")
	verifyGit := flag.Bool("verify-unrelated-against-git", false, "verify tracked non-server fields against HEAD")
	flag.Parse()
	if *verifyGit {
		verifyUnrelatedAgainstGit()
		return
	}

	combined := map[string]interface{}{}
	for _, module := range []string{"clouds/aws", "clouds/gcp", "clouds/azure", "clouds/alicloud"} {
		fragment := terraformFragment(filepath.Join(*infra, module))
		mergeFragment(combined, fragment)
	}
	replaceStaticCloudCredentialsWithPlaceholders(combined)
	normalizeKafkaDependencies(combined)
	common := commonSections(combined)
	environment := environmentSections(combined)

	tracked := []string{"config.dev.yaml", "config.staging.yaml", "config.prod.yaml"}
	locals, err := filepath.Glob("config.local*.yaml")
	check(err)
	sort.Strings(locals)

	writeSections("config.common.yaml", common, 0o644)
	mergeImportSampleDataOSS(combined, "config.common.yaml")
	mergeImportSampleDataAzureBlob(combined, "config.common.yaml")
	for _, path := range tracked {
		writeSections(path, environment, 0o644)
	}
	for _, path := range locals {
		writeSections(path, environment, 0o600)
	}
	fmt.Printf("Updated %d tracked and %d ignored local E2E config files.\n", len(tracked)+1, len(locals))
}

func replaceStaticCloudCredentialsWithPlaceholders(combined map[string]interface{}) {
	replaceListCredentialFields(combined["s3"], map[string]string{
		"accessKey": s3AccessKeyPlaceholder,
		"secretKey": s3SecretKeyPlaceholder,
	})
	replaceListCredentialFields(combined["azureBlobStorage"], map[string]string{
		"sasToken": azureSASPlaceholder,
	})
}

// normalizeKafkaDependencies strips fields the E2E config should not carry and
// enforces cloud-specific shapes on the generated Kafka dependencies:
//   - projectID is dropped: project selection is done via the config's
//     cloud/region, not a per-Kafka dependency field.
//   - GCP Kafka dependencies keep a single "all zones" DNS record key because
//     PSC endpoint selection is decided by the consumer, not a specific zone.
func normalizeKafkaDependencies(combined map[string]interface{}) {
	entries, ok := combined["kafka"].([]interface{})
	if !ok {
		return
	}
	for _, entry := range entries {
		m, ok := entry.(map[interface{}]interface{})
		if !ok {
			continue
		}
		normalizeKafkaEntry(m)
	}
}

func normalizeKafkaEntry(m map[interface{}]interface{}) {
	delete(m, "projectID")
	if m["cloudProvider"] != "gcp" {
		return
	}
	records, ok := m["dnsRecords"].(map[interface{}]interface{})
	if !ok {
		return
	}
	for _, record := range records {
		m["dnsRecords"] = map[interface{}]interface{}{"all zones": record}
		break
	}
}

// mergeImportSampleDataOSS rewrites only the Alibaba Cloud OSS block of the
// manually-maintained importSampleData section in config.common.yaml. External
// AWS/GCP/Azure reference sample-data sources are preserved; the OSS
// sourceURL/endpoint and fixture AK/SK come from the alicloud
// e2e_dependency_yaml fragment, and the AK/SK are redacted to placeholders so
// the tracked file never commits live credentials.
func mergeImportSampleDataOSS(combined map[string]interface{}, configPath string) {
	mergeImportSampleDataSource(combined, configPath, "oss",
		[]string{"sourceURL", "endpoint", "importedDBTable", "customCSVFormat"},
		map[string]string{"accessKeyID": ossAccessKeyPlaceholder, "accessKeySecret": ossSecretKeyPlaceholder})
}

// mergeImportSampleDataAzureBlob rewrites only the Azure Blob block of the
// importSampleData section, mirroring the OSS source above. The dedicated
// import sourceURL and SAS come from the azure e2e_dependency_yaml fragment,
// and the SAS is redacted to a placeholder for the tracked file.
func mergeImportSampleDataAzureBlob(combined map[string]interface{}, configPath string) {
	mergeImportSampleDataSource(combined, configPath, "azureBlob",
		[]string{"sourceURL"},
		map[string]string{"sasToken": azureSASPlaceholder})
}

// mergeImportSampleDataSource patches a single provider sub-block of the
// manual importSampleData section in config.common.yaml from the matching
// e2e_dependency_yaml fragment, preserving external reference sources, and
// rewrites only the importSampleData block in place.
func mergeImportSampleDataSource(combined map[string]interface{}, configPath, subKey string, fields []string, redactions map[string]string) {
	fragment := nestedMapValue(combined["importSampleData"], subKey)
	if fragment == nil {
		return
	}
	data, err := os.ReadFile(configPath)
	check(err)
	parsed := map[interface{}]interface{}{}
	check(yaml.Unmarshal(data, &parsed))
	block := asMap(parsed["importSampleData"])
	if block == nil {
		block = map[interface{}]interface{}{}
		parsed["importSampleData"] = block
	}
	sub := asMap(block[subKey])
	if sub == nil {
		sub = map[interface{}]interface{}{}
		block[subKey] = sub
	}
	for _, field := range fields {
		if value, ok := fragment[field]; ok && value != nil {
			if field == "endpoint" && subKey == "oss" {
				value = normalizeOSSEndpoint(value)
			}
			// Terraform emits customCSVFormat and importedDBTable as maps. Replace
			// those blocks atomically when present; an absent block is left alone
			// for compatibility with legacy Terraform output.
			if generated := asMap(value); generated != nil {
				sub[field] = generated
			} else {
				sub[field] = value
			}
		}
	}
	for field, placeholder := range redactions {
		sub[field] = placeholder
	}
	writeSectionsWithOrder(configPath, map[string]interface{}{"importSampleData": block}, []string{"importSampleData"}, 0o644)
}

func normalizeOSSEndpoint(value interface{}) interface{} {
	endpoint, ok := value.(string)
	if !ok || endpoint == "" || strings.Contains(endpoint, "://") {
		return value
	}
	return "https://" + endpoint
}

func nestedMapValue(parent interface{}, key string) map[interface{}]interface{} {
	block := asMap(parent)
	if block == nil {
		return nil
	}
	return asMap(block[key])
}

func asMap(value interface{}) map[interface{}]interface{} {
	if m, ok := value.(map[interface{}]interface{}); ok {
		return m
	}
	if m, ok := value.(map[string]interface{}); ok {
		converted := make(map[interface{}]interface{}, len(m))
		for key, child := range m {
			converted[key] = child
		}
		return converted
	}
	return nil
}

func replaceListCredentialFields(value interface{}, replacements map[string]string) {
	entries, ok := value.([]interface{})
	if !ok {
		return
	}
	for _, entry := range entries {
		switch typed := entry.(type) {
		case map[interface{}]interface{}:
			for field, placeholder := range replacements {
				if _, exists := typed[field]; exists {
					typed[field] = placeholder
				}
			}
		case map[string]interface{}:
			for field, placeholder := range replacements {
				if _, exists := typed[field]; exists {
					typed[field] = placeholder
				}
			}
		}
	}
}

func verifyUnrelatedAgainstGit() {
	for _, path := range []string{"config.common.yaml", "config.dev.yaml", "config.staging.yaml", "config.prod.yaml"} {
		currentData, err := os.ReadFile(path)
		check(err)
		headPath := filepath.ToSlash(filepath.Join("test/e2e", path))
		headData, err := exec.Command("git", "show", "HEAD:"+headPath).Output()
		if err != nil {
			panic(fmt.Sprintf("read %s from git HEAD: %v", path, err))
		}
		current := map[string]interface{}{}
		head := map[string]interface{}{}
		check(yaml.Unmarshal(currentData, &current))
		check(yaml.Unmarshal(headData, &head))
		if path == "config.common.yaml" {
			verifyStaticCloudCredentialPlaceholders(path, current)
		}
		verifySharedDependencyPlacement(path, current, path == "config.common.yaml")
		for _, key := range sectionOrder {
			if containsValue(current[key], "SET_IN_IGNORED_CONFIG_LOCAL") {
				panic(fmt.Sprintf("tracked %s section %s still contains a credential marker", path, key))
			}
			delete(current, key)
			delete(head, key)
		}
		if path == "config.common.yaml" {
			removeManagedImportSampleDataFields(current)
			removeManagedImportSampleDataFields(head)
			// importData is a manually-maintained test-server fixture section
			// (like the sectionOrder sections above); it is excluded from the
			// git comparison because the Terraform sync does not manage it.
			delete(current, "importData")
			delete(head, "importData")
		}
		if !reflect.DeepEqual(current, head) {
			panic(fmt.Sprintf("non-test-server configuration changed in %s", path))
		}
	}
	locals, err := filepath.Glob("config.local*.yaml")
	check(err)
	for _, path := range locals {
		data, err := os.ReadFile(path)
		check(err)
		parsed := map[string]interface{}{}
		if err := yaml.Unmarshal(data, &parsed); err != nil {
			panic(fmt.Sprintf("parse ignored local config %s: %v", path, err))
		}
		verifySharedDependencyPlacement(path, parsed, false)
		if bytes.Contains(data, []byte("SET_IN_IGNORED_CONFIG_LOCAL")) {
			panic(fmt.Sprintf("ignored local config still contains a redaction marker: %s", path))
		}
		info, err := os.Stat(path)
		check(err)
		if info.Mode().Perm() != 0o600 {
			panic(fmt.Sprintf("ignored local config is not mode 0600: %s", path))
		}
	}
	fmt.Println("Tracked non-test-server configuration matches git HEAD; shared dependencies are valid and static cloud credentials use placeholders.")
}

// removeManagedImportSampleDataFields removes only fields generated by the
// synchronization tool. Other providers, unknown fields, and unmanaged
// importSampleData content remain part of the git comparison.
func removeManagedImportSampleDataFields(parsed map[string]interface{}) {
	managed := map[string][]string{
		"oss":       {"sourceURL", "endpoint", "accessKeyID", "accessKeySecret"},
		"azureBlob": {"sourceURL", "sasToken"},
	}
	nestedManaged := map[string]map[string][]string{
		"oss": {
			"importedDBTable": {"database", "table"},
			"customCSVFormat": {"delimiter", "escapeChar", "lineTerminator", "nullValue", "separator", "skipRows", "splitFile"},
		},
	}
	for provider, fields := range managed {
		sub := importSampleDataSubMap(parsed["importSampleData"], provider)
		if sub == nil {
			continue
		}
		for _, field := range fields {
			deleteMapField(sub, field)
		}
		for block, nestedFields := range nestedManaged[provider] {
			nested := mapField(sub, block)
			for _, field := range nestedFields {
				deleteMapField(nested, field)
			}
			if empty, isMap := isEmptyMap(nested); isMap && empty {
				deleteMapField(sub, block)
			}
		}
		if empty, isMap := isEmptyMap(sub); isMap && empty {
			deleteMapField(parsed["importSampleData"], provider)
		}
	}
	if empty, isMap := isEmptyMap(parsed["importSampleData"]); isMap && empty {
		delete(parsed, "importSampleData")
	}
}

func mapField(value interface{}, field string) interface{} {
	switch typed := value.(type) {
	case map[interface{}]interface{}:
		return typed[field]
	case map[string]interface{}:
		return typed[field]
	default:
		return nil
	}
}

func isEmptyMap(value interface{}) (empty, isMap bool) {
	switch typed := value.(type) {
	case map[interface{}]interface{}:
		return len(typed) == 0, true
	case map[string]interface{}:
		return len(typed) == 0, true
	default:
		return false, false
	}
}

func deleteMapField(value interface{}, field string) {
	switch typed := value.(type) {
	case map[interface{}]interface{}:
		delete(typed, field)
	case map[string]interface{}:
		delete(typed, field)
	}
}

func verifyStaticCloudCredentialPlaceholders(path string, parsed map[string]interface{}) {
	verifyOptionalListCredentialPlaceholder(path, parsed["s3"], "accessKey", s3AccessKeyPlaceholder)
	verifyOptionalListCredentialPlaceholder(path, parsed["s3"], "secretKey", s3SecretKeyPlaceholder)
	verifyListCredentialPlaceholder(path, parsed["azureBlobStorage"], "sasToken", azureSASPlaceholder)
	verifyMapCredentialPlaceholder(path, importSampleDataSubMap(parsed["importSampleData"], "oss"), "accessKeyID", ossAccessKeyPlaceholder)
	verifyMapCredentialPlaceholder(path, importSampleDataSubMap(parsed["importSampleData"], "oss"), "accessKeySecret", ossSecretKeyPlaceholder)
	verifyMapCredentialPlaceholder(path, importSampleDataSubMap(parsed["importSampleData"], "azureBlob"), "sasToken", azureSASPlaceholder)
}

// importSampleDataSubMap returns the raw sub-map for the given subKey of the
// importSampleData section, or nil when the section is absent. It tolerates
// both yaml.v2 map representations so verification does not depend on the
// parser's nested type.
func importSampleDataSubMap(value interface{}, subKey string) interface{} {
	var block map[interface{}]interface{}
	switch typed := value.(type) {
	case map[interface{}]interface{}:
		block = typed
	case map[string]interface{}:
		block = map[interface{}]interface{}{}
		for k, v := range typed {
			block[k] = v
		}
	default:
		return nil
	}
	return block[subKey]
}

func verifyOptionalListCredentialPlaceholder(path string, value interface{}, field, expected string) {
	entries, ok := value.([]interface{})
	if !ok || len(entries) == 0 {
		panic(fmt.Sprintf("tracked %s has no entries while checking optional credential field %s", path, field))
	}
	for _, entry := range entries {
		var raw interface{}
		var exists bool
		switch typed := entry.(type) {
		case map[interface{}]interface{}:
			raw, exists = typed[field]
		case map[string]interface{}:
			raw, exists = typed[field]
		}
		if !exists {
			continue
		}
		actual, valid := raw.(string)
		if !valid || actual != expected {
			panic(fmt.Sprintf("tracked %s credential field %s is not the required placeholder", path, field))
		}
	}
}

func verifyListCredentialPlaceholder(path string, value interface{}, field, expected string) {
	entries, ok := value.([]interface{})
	if !ok || len(entries) == 0 {
		panic(fmt.Sprintf("tracked %s has no entries for credential field %s", path, field))
	}
	for _, entry := range entries {
		verifyMapCredentialPlaceholder(path, entry, field, expected)
	}
}

func verifyMapCredentialPlaceholder(path string, value interface{}, field, expected string) {
	actual, ok := mapStringField(value, field)
	if !ok || actual != expected {
		panic(fmt.Sprintf("tracked %s credential field %s is not the required placeholder", path, field))
	}
}

func mapStringField(value interface{}, field string) (string, bool) {
	var raw interface{}
	switch typed := value.(type) {
	case map[interface{}]interface{}:
		raw = typed[field]
	case map[string]interface{}:
		raw = typed[field]
	default:
		return "", false
	}
	actual, ok := raw.(string)
	return actual, ok
}

func verifySharedDependencyPlacement(path string, parsed map[string]interface{}, common bool) {
	for _, key := range sectionOrder {
		_, present := parsed[key]
		if common && !present {
			panic(fmt.Sprintf("shared dependency %s is missing from %s", key, path))
		}
		if !common && present {
			panic(fmt.Sprintf("shared dependency %s must be inherited from config.common.yaml, but is present in %s", key, path))
		}
	}
}

func containsValue(value interface{}, expected string) bool {
	switch typed := value.(type) {
	case map[string]interface{}:
		for _, child := range typed {
			if containsValue(child, expected) {
				return true
			}
		}
	case map[interface{}]interface{}:
		for _, child := range typed {
			if containsValue(child, expected) {
				return true
			}
		}
	case []interface{}:
		for _, child := range typed {
			if containsValue(child, expected) {
				return true
			}
		}
	case string:
		return typed == expected
	}
	return false
}

func commonSections(combined map[string]interface{}) map[string]interface{} {
	common := map[string]interface{}{}
	for _, key := range sectionOrder {
		if value, ok := combined[key]; ok {
			common[key] = value
		}
	}
	return common
}

func environmentSections(_ map[string]interface{}) map[string]interface{} {
	return map[string]interface{}{}
}

func terraformFragment(dir string) map[string]interface{} {
	cmd := exec.Command("terraform", "-chdir="+dir, "output", "-raw", "e2e_dependency_yaml")
	data, err := cmd.Output()
	if err != nil {
		panic(fmt.Sprintf("read Terraform output from %s: %v", dir, err))
	}
	return decodeTerraformFragment(dir, data)
}

func decodeTerraformFragment(dir string, data []byte) map[string]interface{} {
	if len(bytes.TrimSpace(data)) == 0 {
		panic(fmt.Sprintf("Terraform output e2e_dependency_yaml is missing or empty in %s; apply that cloud root before synchronizing E2E config", dir))
	}
	fragment := map[string]interface{}{}
	check(yaml.Unmarshal(data, &fragment))
	return fragment
}

func mergeFragment(dst, src map[string]interface{}) {
	for key, value := range src {
		switch current := dst[key].(type) {
		case []interface{}:
			dst[key] = append(current, value.([]interface{})...)
		case map[interface{}]interface{}:
			for nestedKey, nestedValue := range value.(map[interface{}]interface{}) {
				current[nestedKey] = nestedValue
			}
		default:
			dst[key] = value
		}
	}
}

func writeSections(path string, sections map[string]interface{}, mode os.FileMode) {
	writeSectionsWithOrder(path, sections, sectionOrder, mode)
}

func writeSectionsWithOrder(path string, sections map[string]interface{}, order []string, mode os.FileMode) {
	data, err := os.ReadFile(path)
	check(err)
	lines := strings.Split(string(data), "\n")
	starts := topLevelStarts(lines)
	remove := map[int]bool{}
	insertAt := len(lines)
	for _, key := range order {
		start, ok := starts[key]
		if !ok {
			continue
		}
		if start < insertAt {
			insertAt = start
		}
		end := nextTopLevelStart(lines, start+1)
		for index := start; index < end; index++ {
			remove[index] = true
		}
	}

	rendered := renderSectionsWithOrder(sections, order)
	var output bytes.Buffer
	inserted := false
	for index, line := range lines {
		if index == insertAt {
			output.WriteString(rendered)
			inserted = true
		}
		if !remove[index] {
			output.WriteString(line)
			if index < len(lines)-1 {
				output.WriteByte('\n')
			}
		}
	}
	if !inserted {
		if output.Len() > 0 && !bytes.HasSuffix(output.Bytes(), []byte("\n")) {
			output.WriteByte('\n')
		}
		output.WriteString(rendered)
	}
	final := append(bytes.TrimRight(output.Bytes(), "\n"), '\n')
	check(os.WriteFile(path, final, mode))
	check(os.Chmod(path, mode))
}

func renderSections(sections map[string]interface{}) string {
	return renderSectionsWithOrder(sections, sectionOrder)
}

func renderSectionsWithOrder(sections map[string]interface{}, order []string) string {
	var output bytes.Buffer
	for _, key := range order {
		value, ok := sections[key]
		if !ok {
			continue
		}
		data, err := yaml.Marshal(map[string]interface{}{key: value})
		check(err)
		output.Write(data)
		output.WriteByte('\n')
	}
	return output.String()
}

func topLevelStarts(lines []string) map[string]int {
	result := map[string]int{}
	for index, line := range lines {
		if !isTopLevelKey(line) {
			continue
		}
		colon := strings.IndexByte(line, ':')
		result[line[:colon]] = index
	}
	return result
}

func nextTopLevelStart(lines []string, from int) int {
	for index := from; index < len(lines); index++ {
		if isTopLevelKey(lines[index]) {
			return index
		}
	}
	return len(lines)
}

func isTopLevelKey(line string) bool {
	if line == "" || !((line[0] >= 'A' && line[0] <= 'Z') || (line[0] >= 'a' && line[0] <= 'z')) {
		return false
	}
	return strings.IndexByte(line, ':') > 0
}

func check(err error) {
	if err != nil {
		panic(err)
	}
}