package config

import (
	"os"
	"path/filepath"
	"strings"
	"testing"
)

func TestTableTemplateIsDeltaTable(t *testing.T) {
	tests := []struct {
		name string
		ddl  string
		want bool
	}{
		{
			name: "delta table property",
			ddl:  "CREATE EXTERNAL TABLE db.tbl LOCATION 's3://bucket/path' TBLPROPERTIES ('table_type' = 'DELTA')",
			want: true,
		},
		{
			name: "parquet external table",
			ddl:  "CREATE EXTERNAL TABLE db.tbl (id BIGINT) STORED AS PARQUET LOCATION 's3://bucket/path'",
			want: false,
		},
		{
			name: "view",
			ddl:  "CREATE OR REPLACE VIEW db.v AS SELECT * FROM db.tbl",
			want: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if got := (TableTemplate{DDL: tt.ddl}).IsDeltaTable(); got != tt.want {
				t.Fatalf("IsDeltaTable() = %v, want %v", got, tt.want)
			}
		})
	}
}

func TestLoadClinicAPIBaseIsIndependentFromWarehouseEngines(t *testing.T) {
	t.Setenv("TEST_CLINIC_API_BASE", "http://clinic-internal-api:80")
	configPath := filepath.Join(t.TempDir(), "config-dataproxy.yaml")
	err := os.WriteFile(configPath, []byte(`
clinic_api_base: "${TEST_CLINIC_API_BASE}"
`), 0o600)
	if err != nil {
		t.Fatalf("WriteFile() error = %v", err)
	}

	baseURL, err := LoadClinicAPIBase(configPath)
	if err != nil {
		t.Fatalf("LoadClinicAPIBase() error = %v", err)
	}
	if baseURL != "http://clinic-internal-api:80" {
		t.Fatalf("LoadClinicAPIBase() = %q", baseURL)
	}
}

func TestLoadClinicAPIBaseRejectsUnsafeURL(t *testing.T) {
	for _, value := range []string{"", "clinic-internal-api", "ftp://clinic", "http://user@clinic", "http://clinic?target=other"} {
		t.Run(value, func(t *testing.T) {
			configPath := filepath.Join(t.TempDir(), "config-dataproxy.yaml")
			if err := os.WriteFile(configPath, []byte("clinic_api_base: \""+value+"\"\n"), 0o600); err != nil {
				t.Fatalf("WriteFile() error = %v", err)
			}
			_, err := LoadClinicAPIBase(configPath)
			if err == nil || !strings.Contains(err.Error(), "clinic_api_base") {
				t.Fatalf("LoadClinicAPIBase(%q) error = %v", value, err)
			}
		})
	}
}

func TestLoadTidbMgmtHTTPBase(t *testing.T) {
	configPath := filepath.Join(t.TempDir(), "config-dataproxy.yaml")
	requireContent := "tidb_mgmt_http_base: \"https://us-west-2.prod.aws.tidbcloud.com\"\n"
	if err := os.WriteFile(configPath, []byte(requireContent), 0o600); err != nil {
		t.Fatalf("WriteFile() error = %v", err)
	}
	baseURL, err := LoadTidbMgmtHTTPBase(configPath)
	if err != nil {
		t.Fatalf("LoadTidbMgmtHTTPBase() error = %v", err)
	}
	if baseURL != "https://us-west-2.prod.aws.tidbcloud.com" {
		t.Fatalf("LoadTidbMgmtHTTPBase() = %q", baseURL)
	}
}

func TestLoadTidbMgmtHTTPBaseRejectsUnsafeURL(t *testing.T) {
	for _, value := range []string{"tidb-mgmt", "ftp://tidb-mgmt", "https://user@tidb-mgmt", "https://tidb-mgmt?target=other"} {
		t.Run(value, func(t *testing.T) {
			configPath := filepath.Join(t.TempDir(), "config-dataproxy.yaml")
			if err := os.WriteFile(configPath, []byte("tidb_mgmt_http_base: \""+value+"\"\n"), 0o600); err != nil {
				t.Fatalf("WriteFile() error = %v", err)
			}
			_, err := LoadTidbMgmtHTTPBase(configPath)
			if err == nil || !strings.Contains(err.Error(), "tidb_mgmt_http_base") {
				t.Fatalf("LoadTidbMgmtHTTPBase(%q) error = %v", value, err)
			}
		})
	}
}

func TestLoadTiCDCConfig(t *testing.T) {
	configPath := filepath.Join(t.TempDir(), "config-dataproxy.yaml")
	err := os.WriteFile(configPath, []byte(`
ticdc:
  dataflow_addr: "dataflow.internal:10103"
  dataflow_ng_addr: "dataflow.internal:10104"
  serverless_addr: "dataflow.internal:10101"
`), 0o600)
	if err != nil {
		t.Fatalf("WriteFile() error = %v", err)
	}

	cfg, err := LoadTiCDCConfig(configPath)
	if err != nil {
		t.Fatalf("LoadTiCDCConfig() error = %v", err)
	}
	if cfg.DataflowAddr != "dataflow.internal:10103" || cfg.DataflowNGAddr != "dataflow.internal:10104" || cfg.ServerlessAddr != "dataflow.internal:10101" {
		t.Fatalf("LoadTiCDCConfig() = %#v", cfg)
	}
}

func TestLoadTiCDCConfigRequiresBothAddresses(t *testing.T) {
	for _, content := range []string{
		"ticdc:\n  dataflow_ng_addr: dataflow.internal:10104\n",
		"ticdc:\n  dataflow_addr: dataflow.internal:10103\n",
		"ticdc:\n  dataflow_addr: https://dataflow.internal\n  dataflow_ng_addr: dataflow.internal:10104\n  serverless_addr: dataflow.internal:10101\n",
	} {
		configPath := filepath.Join(t.TempDir(), "config-dataproxy.yaml")
		if err := os.WriteFile(configPath, []byte(content), 0o600); err != nil {
			t.Fatalf("WriteFile() error = %v", err)
		}
		if _, err := LoadTiCDCConfig(configPath); err == nil {
			t.Fatalf("LoadTiCDCConfig(%q) unexpectedly succeeded", content)
		}
	}
}

func TestLoadDataProxyConfigLoadsNestedInstanceProjectionRefresh(t *testing.T) {
	configPath := filepath.Join(t.TempDir(), "config-dataproxy.yaml")
	err := os.WriteFile(configPath, []byte(`
aws:
  enabled: true
  table_manager:
    instance_projection_refresh_seconds: 600
`), 0o600)
	if err != nil {
		t.Fatalf("WriteFile() error = %v", err)
	}

	cfg, err := loadDataProxyConfig(configPath)
	if err != nil {
		t.Fatalf("loadDataProxyConfig() error = %v", err)
	}
	if cfg.AWS.TableManager.InstanceProjectionRefreshSeconds != 600 {
		t.Fatalf("InstanceProjectionRefreshSeconds = %d", cfg.AWS.TableManager.InstanceProjectionRefreshSeconds)
	}
}

func TestLoadDataProxyConfigRejectsMisplacedInstanceProjectionRefresh(t *testing.T) {
	configPath := filepath.Join(t.TempDir(), "config-dataproxy.yaml")
	err := os.WriteFile(configPath, []byte(`
aws:
  enabled: true
  instance_projection_refresh_seconds: 600
`), 0o600)
	if err != nil {
		t.Fatalf("WriteFile() error = %v", err)
	}

	_, err = loadDataProxyConfig(configPath)
	if err == nil || !strings.Contains(err.Error(), "aws.table_manager.instance_projection_refresh_seconds") {
		t.Fatalf("loadDataProxyConfig() error = %v", err)
	}
}

func TestLoadDataProxyConfigMaxComputeSlowQueryParquetPredicatePushdown(t *testing.T) {
	tests := []struct {
		name        string
		setting     string
		wantEnabled bool
		wantRegions []string
	}{
		{name: "omitted"},
		{
			name: "enabled for selected regions",
			setting: "    enable_slow_query_parquet_predicate_pushdown: true\n" +
				"    slow_query_parquet_predicate_pushdown_regions: [ap-southeast-1, ap-northeast-1]\n",
			wantEnabled: true,
			wantRegions: []string{"ap-southeast-1", "ap-northeast-1"},
		},
		{name: "explicitly disabled", setting: "    enable_slow_query_parquet_predicate_pushdown: false\n"},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			configPath := filepath.Join(t.TempDir(), "config-dataproxy.yaml")
			content := "alicloud:\n  maxcompute:\n" + tt.setting
			if err := os.WriteFile(configPath, []byte(content), 0o600); err != nil {
				t.Fatalf("WriteFile() error = %v", err)
			}

			cfg, err := loadDataProxyConfig(configPath)
			if err != nil {
				t.Fatalf("loadDataProxyConfig() error = %v", err)
			}
			mc := cfg.AliCloud.MaxCompute
			if mc.EnableSlowQueryParquetPredicatePushdown != tt.wantEnabled {
				t.Fatalf("EnableSlowQueryParquetPredicatePushdown = %v, want %v", mc.EnableSlowQueryParquetPredicatePushdown, tt.wantEnabled)
			}
			if len(mc.SlowQueryParquetPredicatePushdownRegions) != len(tt.wantRegions) {
				t.Fatalf("SlowQueryParquetPredicatePushdownRegions = %#v, want %#v", mc.SlowQueryParquetPredicatePushdownRegions, tt.wantRegions)
			}
			for i := range tt.wantRegions {
				if mc.SlowQueryParquetPredicatePushdownRegions[i] != tt.wantRegions[i] {
					t.Fatalf("SlowQueryParquetPredicatePushdownRegions = %#v, want %#v", mc.SlowQueryParquetPredicatePushdownRegions, tt.wantRegions)
				}
			}
		})
	}
}

func TestDataProxyConfigValidateRequiresPredicatePushdownRegionAllowlist(t *testing.T) {
	cfg := DefaultConfig()
	cfg.AliCloud.Enabled = true
	cfg.AliCloud.MaxCompute.ProjectPrefix = "clinic_proxy_test"
	cfg.AliCloud.MaxCompute.AccessKey = "test-access-key"
	cfg.AliCloud.MaxCompute.SecretKey = "test-secret-key"
	cfg.AliCloud.MaxCompute.EnableSlowQueryParquetPredicatePushdown = true

	err := cfg.Validate()
	if err == nil || !strings.Contains(err.Error(), "slow_query_parquet_predicate_pushdown_regions is required") {
		t.Fatalf("Validate() error = %v, want required region allowlist", err)
	}

	cfg.AliCloud.MaxCompute.SlowQueryParquetPredicatePushdownRegions = []string{"ap-southeast-1"}
	if err := cfg.Validate(); err != nil {
		t.Fatalf("Validate() with explicit region allowlist error = %v", err)
	}
}