package test

import (
	"context"
	"database/sql"
	"net"
	"net/url"
	"os"
	"strconv"
	"testing"
	"time"

	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
	"github.com/Azure/azure-sdk-for-go/sdk/azidentity"
	"github.com/microsoft/go-mssqldb"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/config"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/service"
)

const defaultAzureSynapseQuery = `
SELECT TOP 2
    date,
    Query_time,
    Query,
    digest
FROM OPENROWSET(
    BULK 'https://o11yprodjapaneast.dfs.core.windows.net/o11y-prod-shared-japaneast/deltalake/1372813089192471291/019db46c-ce89-7052-8407-05d202ff4095/slowlogs/',
    FORMAT = 'DELTA'
) AS rows
WHERE date = '20260422'
`

// const defaultAzureSynapseQuery = `
// SELECT TOP 3
//     date,
//     Query_time,
//     Query,
//     digest
// FROM OPENROWSET(
//     BULK 'https://o11yprodsoutheastasia.dfs.core.windows.net/o11y-prod-shared-southeastasia/deltalake/1372813089189381281/019c974a-66c4-7a02-abcb-f2d536b55747/slowlogs/',
//     FORMAT = 'DELTA'
// ) AS rows
// WHERE date = '20260226'
// `

func TestAzure(t *testing.T) {
	server := os.Getenv("AZURE_SYNAPSE_SERVER")
	if server == "" {
		t.Skip("AZURE_SYNAPSE_SERVER is not set")
	}
	port := 1433
	database := envOrDefault("AZURE_SYNAPSE_DATABASE", "master")

	// =========================
	// 1. Azure Service Principal
	// =========================
	tenantID := os.Getenv("AZURE_TENANT_ID")
	clientID := os.Getenv("AZURE_CLIENT_ID")
	clientSecret := os.Getenv("AZURE_CLIENT_SECRET")
	if tenantID == "" || clientID == "" || clientSecret == "" {
		t.Skip("AZURE_TENANT_ID, AZURE_CLIENT_ID, and AZURE_CLIENT_SECRET are required")
	}

	cred, err := azidentity.NewClientSecretCredential(
		tenantID,
		clientID,
		clientSecret,
		nil,
	)
	if err != nil {
		t.Fatalf("create credential failed: %v", err)
	}

	token, err := cred.GetToken(context.Background(), policy.TokenRequestOptions{
		Scopes: []string{"https://database.windows.net/.default"},
	})
	if err != nil {
		t.Fatalf("get token failed: %v", err)
	}

	// =========================
	// 2. 使用 Access Token 连接 Synapse
	// =========================
	connector, err := mssql.NewAccessTokenConnector(
		buildSQLServerConnString(server, port, database),
		func() (string, error) {
			return token.Token, nil
		},
	)
	if err != nil {
		t.Fatalf("create connector failed: %v", err)
	}

	db := sql.OpenDB(connector)
	defer db.Close()

	db.SetMaxOpenConns(10)
	db.SetMaxIdleConns(5)
	db.SetConnMaxLifetime(30 * time.Minute)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	err = db.PingContext(ctx)
	if err != nil {
		t.Fatalf("ping failed: %v", err)
	}

	t.Log("connect synapse success")

	// =========================
	// 3. Query Delta table
	// =========================
	query := envOrDefault("AZURE_SYNAPSE_QUERY", defaultAzureSynapseQuery)
	rows, err := db.QueryContext(ctx, query)
	if err != nil {
		t.Fatalf("query failed: %v", err)
	}
	defer rows.Close()

	for rows.Next() {
		var date string
		var queryTime float64
		var queryText string
		var digest string

		err := rows.Scan(
			&date,
			&queryTime,
			&queryText,
			&digest,
		)
		if err != nil {
			t.Fatalf("scan failed: %v", err)
		}

		t.Log("===================================")
		t.Log("date:", date)
		t.Log("query_time:", queryTime)
		t.Log("digest:", digest)
		t.Log("query:", queryText)
	}

	if err := rows.Err(); err != nil {
		t.Fatalf("rows error: %v", err)
	}
}

func TestAzureDataProxyServiceQuery(t *testing.T) {
	server := os.Getenv("AZURE_SYNAPSE_SERVER")
	tenantID := os.Getenv("AZURE_TENANT_ID")
	clientID := os.Getenv("AZURE_CLIENT_ID")
	clientSecret := os.Getenv("AZURE_CLIENT_SECRET")
	if server == "" || tenantID == "" || clientID == "" || clientSecret == "" {
		t.Skip("Azure Synapse env vars are not set")
	}

	dataSource := envOrDefault("AZURE_DELTA_DATASOURCE", "https://o11yprodsoutheastasia.dfs.core.windows.net/o11y-prod-shared-southeastasia/deltalake/1372813089189381281/019c974a-66c4-7a02-abcb-f2d536b55747/slowlogs/")
	region := envOrDefault("AZURE_REGION", "southeastasia")
	clusterID := envOrDefault("AZURE_CLUSTER_ID", "1372813089189381281")
	dateValue := envOrDefault("AZURE_DELTA_DATE", "20260226")

	cfg := config.DefaultConfig()
	cfg.Azure.Enabled = true
	cfg.Azure.Synapse.Server = server
	cfg.Azure.Synapse.Database = envOrDefault("AZURE_SYNAPSE_DATABASE", "master")
	cfg.Azure.Synapse.AuthType = "aad_service_principal"
	cfg.Azure.Synapse.TenantID = tenantID
	cfg.Azure.Synapse.ClientID = clientID
	cfg.Azure.Synapse.ClientSecret = clientSecret

	svc, err := service.NewDataProxyService(cfg)
	if err != nil {
		t.Fatalf("create data proxy service failed: %v", err)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()

	result, err := svc.Query(ctx, &dto.DataProxyQueryRequest{
		SQL:       "SELECT date, Query_time, Query, digest FROM clinic_data_proxy.slow_query_logs WHERE date IN ('" + dateValue + "') LIMIT 3",
		ClusterId: clusterID,
		Timeout:   60 * time.Second,
		Options: dto.DataProxyQueryOptions{
			Vendor:     "azure",
			Region:     region,
			DataSource: dataSource,
		},
	})
	if err != nil {
		t.Fatalf("data proxy query failed: %v", err)
	}
	if result.Metadata.Engine != "synapse" || result.Metadata.Vendor != "azure" {
		t.Fatalf("unexpected metadata: %+v", result.Metadata)
	}
	t.Logf("data proxy query success: rows=%d queryID=%s", len(result.Rows), result.Metadata.QueryID)
}

func buildSQLServerConnString(server string, port int, database string) string {
	values := url.Values{}
	values.Set("database", database)
	values.Set("encrypt", "true")

	connURL := url.URL{
		Scheme:   "sqlserver",
		Host:     net.JoinHostPort(server, strconv.Itoa(port)),
		RawQuery: values.Encode(),
	}

	return connURL.String()
}

func envOrDefault(key, fallback string) string {
	if value := os.Getenv(key); value != "" {
		return value
	}
	return fallback
}