//go:build integration

package engine

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"net/url"
	"os"
	"strings"
	"testing"
	"time"

	"cloud.google.com/go/bigquery"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
	"github.com/stretchr/testify/require"
	"google.golang.org/api/iterator"
)

// TestGCPTopSQLSourceDiscoverySmoke is an explicitly gated, read-only side-path
// test. It reads BigQuery table metadata and GCS child prefixes through ambient
// ADC, but never runs DDL or queries customer rows. Logs contain hashes and
// bounded counts only.
func TestGCPTopSQLSourceDiscoverySmoke(t *testing.T) {
	project := strings.TrimSpace(os.Getenv("CLINIC_GCP_TOPSQL_SMOKE_PROJECT"))
	datasets := splitNonEmpty(os.Getenv("CLINIC_GCP_TOPSQL_SMOKE_DATASETS"))
	if project == "" || len(datasets) == 0 {
		t.Skip("set CLINIC_GCP_TOPSQL_SMOKE_PROJECT and CLINIC_GCP_TOPSQL_SMOKE_DATASETS")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	queryEngine, err := NewBigQueryEngine(&BigQueryConfig{ProjectID: project})
	require.NoError(t, err)
	t.Cleanup(func() { _ = queryEngine.client.Close() })

	targets := 0
	nonEmptyInstanceTargets := 0
	repairedDuplicateRoots := 0
	for _, dataset := range datasets {
		tables := queryEngine.client.Dataset(dataset).Tables(ctx)
		for {
			table, err := tables.Next()
			if errors.Is(err, iterator.Done) {
				break
			}
			if isBigQueryNotFound(err) {
				break
			}
			require.NoError(t, err)
			kind := topSQLTableKind(table.TableID)
			if kind != "topsql_tidb" && kind != "topsql_tikv" {
				continue
			}
			metadata, err := table.Metadata(ctx)
			require.NoError(t, err)
			if metadata.ExternalDataConfig == nil || len(metadata.ExternalDataConfig.SourceURIs) == 0 {
				continue
			}
			root, ok := topSQLExternalRootFromSourceURI(metadata.ExternalDataConfig.SourceURIs[0])
			if !ok {
				continue
			}
			root, repaired := repairDuplicateTopSQLComponentRoot(root, kind)
			if repaired {
				repairedDuplicateRoots++
			}
			definition, err := queryEngine.discoverTopSQLExternalDefinition(ctx, table.TableID, root)
			if err != nil {
				diagnostics, ok := AsBigQueryTopSQLSetupError(err)
				require.True(t, ok, "source discovery returned an untyped error")
				require.Failf(t, "source discovery failed", "target=%s operation=%s source=%s reason=%s http=%d", smokeTargetHash(root), diagnostics.Operation, diagnostics.ErrorSource, diagnostics.Reason, diagnostics.HTTPStatus)
			}
			targets++
			if definition.instanceCount > 0 {
				nonEmptyInstanceTargets++
			}
			t.Logf("target=%s kind=%s duplicate_component_repaired=%t instances=%d source_uris=%d", smokeTargetHash(root), kind, repaired, definition.instanceCount, len(definition.sourceURIs))
			break
		}
	}

	require.Positive(t, targets, "no existing TopSQL metric external roots were available")
	require.Positive(t, nonEmptyInstanceTargets, "all discovered TopSQL metric roots were empty")
	t.Logf("targets=%d duplicate_component_roots=%d nonempty_instance_targets=%d", targets, repairedDuplicateRoots, nonEmptyInstanceTargets)
}

// TestGCPTopSQLQuerySmoke exercises the final BigQuery query boundary with an
// unqualified physical view name. It reads no customer rows: the date cannot
// match a real partition and LIMIT 0 prevents result materialization. It does
// not run table management or DDL.
func TestGCPTopSQLQuerySmoke(t *testing.T) {
	if os.Getenv("CLINIC_GCP_TOPSQL_LOGICAL_E2E") != "1" {
		t.Skip("set CLINIC_GCP_TOPSQL_LOGICAL_E2E=1 after managed empty views are reconciled")
	}
	project := strings.TrimSpace(os.Getenv("CLINIC_GCP_TOPSQL_SMOKE_PROJECT"))
	datasets := splitNonEmpty(os.Getenv("CLINIC_GCP_TOPSQL_SMOKE_DATASETS"))
	if project == "" || len(datasets) == 0 {
		t.Skip("set CLINIC_GCP_TOPSQL_SMOKE_PROJECT and CLINIC_GCP_TOPSQL_SMOKE_DATASETS")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	for _, dataset := range datasets {
		bootstrap, err := NewBigQueryEngine(&BigQueryConfig{ProjectID: project})
		require.NoError(t, err)
		datasetMetadata, err := bootstrap.client.Dataset(dataset).Metadata(ctx)
		_ = bootstrap.client.Close()
		if isBigQueryNotFound(err) {
			continue
		}
		require.NoError(t, err)
		region := datasetMetadata.Location
		suffix := "_" + strings.ReplaceAll(region, "-", "_")
		if region == "" || !strings.HasSuffix(dataset, suffix) {
			continue
		}
		datasetPrefix := strings.TrimSuffix(dataset, suffix)
		queryEngine, err := NewBigQueryEngine(&BigQueryConfig{ProjectID: project, Dataset: datasetPrefix})
		require.NoError(t, err)

		tables := queryEngine.client.Dataset(dataset).Tables(ctx)
		for {
			table, err := tables.Next()
			if errors.Is(err, iterator.Done) {
				break
			}
			require.NoError(t, err)
			if _, ok := bigQueryTopSQLViewClusterID(table.TableID); !ok {
				continue
			}
			metadata, err := table.Metadata(ctx)
			require.NoError(t, err)
			if metadata.Type != bigquery.ViewTable {
				continue
			}

			sql := fmt.Sprintf("SELECT * FROM `%s` WHERE date = '__clinic_e2e_no_date__' LIMIT 0", table.TableID)
			result, err := queryEngine.Execute(ctx, sql, &dto.DataSourceInfo{Vendor: "gcp", Region: region})
			if err != nil {
				if diagnostics, ok := AsBigQueryQueryError(err); ok {
					require.Failf(t, "read-only TopSQL query failed", "target=%s source=%s reason=%s shape=%s errors=%d", smokeTargetHash(dataset+"/"+table.TableID), diagnostics.ErrorSource, diagnostics.Reason, diagnostics.MessageShape, diagnostics.ErrorCount)
				}
				if diagnostics, ok := AsUnknownFieldError(err); ok {
					require.Failf(t, "read-only TopSQL query used an unknown field", "target=%s fields=%v", smokeTargetHash(dataset+"/"+table.TableID), diagnostics.Fields)
				}
				require.Failf(t, "read-only TopSQL query failed", "target=%s error_type=%T", smokeTargetHash(dataset+"/"+table.TableID), err)
			}
			require.Zero(t, result.Metadata.RowCount)
			t.Logf("target=%s region=%s rows=%d", smokeTargetHash(dataset+"/"+table.TableID), region, result.Metadata.RowCount)
			_ = queryEngine.client.Close()
			return
		}
		_ = queryEngine.client.Close()
	}
	t.Fatal("no existing regional logical TopSQL view was available")
}

// TestGCPTopSQLPhysicalQuerySmoke validates default dataset and location
// routing against an existing nonempty physical table without running DDL or
// returning rows.
func TestGCPTopSQLPhysicalQuerySmoke(t *testing.T) {
	if os.Getenv("CLINIC_GCP_TOPSQL_PHYSICAL_E2E") != "1" {
		t.Skip("set CLINIC_GCP_TOPSQL_PHYSICAL_E2E=1 after physical tables are reconciled")
	}
	project := strings.TrimSpace(os.Getenv("CLINIC_GCP_TOPSQL_SMOKE_PROJECT"))
	datasets := splitNonEmpty(os.Getenv("CLINIC_GCP_TOPSQL_SMOKE_DATASETS"))
	if project == "" || len(datasets) == 0 {
		t.Skip("set CLINIC_GCP_TOPSQL_SMOKE_PROJECT and CLINIC_GCP_TOPSQL_SMOKE_DATASETS")
	}

	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	for _, dataset := range datasets {
		bootstrap, err := NewBigQueryEngine(&BigQueryConfig{ProjectID: project})
		require.NoError(t, err)
		datasetMetadata, err := bootstrap.client.Dataset(dataset).Metadata(ctx)
		_ = bootstrap.client.Close()
		if isBigQueryNotFound(err) {
			continue
		}
		require.NoError(t, err)
		region := datasetMetadata.Location
		suffix := "_" + strings.ReplaceAll(region, "-", "_")
		if region == "" || !strings.HasSuffix(dataset, suffix) {
			continue
		}
		queryEngine, err := NewBigQueryEngine(&BigQueryConfig{ProjectID: project, Dataset: strings.TrimSuffix(dataset, suffix)})
		require.NoError(t, err)

		tables := queryEngine.client.Dataset(dataset).Tables(ctx)
		for {
			table, err := tables.Next()
			if errors.Is(err, iterator.Done) {
				break
			}
			require.NoError(t, err)
			kind := topSQLTableKind(table.TableID)
			if kind != "topsql_tidb" && kind != "topsql_tikv" {
				continue
			}
			metadata, err := table.Metadata(ctx)
			require.NoError(t, err)
			if metadata.Type != bigquery.ExternalTable || metadata.ExternalDataConfig == nil || len(metadata.ExternalDataConfig.SourceURIs) == 0 {
				continue
			}
			root, ok := topSQLExternalRootFromSourceURI(metadata.ExternalDataConfig.SourceURIs[0])
			if !ok {
				continue
			}
			root, _ = repairDuplicateTopSQLComponentRoot(root, kind)
			definition, err := queryEngine.discoverTopSQLExternalDefinition(ctx, table.TableID, root)
			if err != nil || definition.empty || !isExpectedTopSQLExternalTable(metadata, definition) {
				continue
			}

			sql := fmt.Sprintf("SELECT * FROM `%s` WHERE date = '__clinic_e2e_no_date__' LIMIT 0", table.TableID)
			result, err := queryEngine.Execute(ctx, sql, &dto.DataSourceInfo{Vendor: "gcp", Region: region})
			if diagnostics, ok := AsBigQueryQueryError(err); ok {
				require.Failf(t, "read-only physical TopSQL query failed", "target=%s source=%s reason=%s shape=%s errors=%d", smokeTargetHash(dataset+"/"+table.TableID), diagnostics.ErrorSource, diagnostics.Reason, diagnostics.MessageShape, diagnostics.ErrorCount)
			}
			require.NoError(t, err)
			require.Zero(t, result.Metadata.RowCount)
			t.Logf("target=%s region=%s rows=%d", smokeTargetHash(dataset+"/"+table.TableID), region, result.Metadata.RowCount)
			_ = queryEngine.client.Close()
			return
		}
		_ = queryEngine.client.Close()
	}
	t.Fatal("no existing regional nonempty physical TopSQL table was available")
}

func repairDuplicateTopSQLComponentRoot(root, kind string) (string, bool) {
	component := strings.TrimPrefix(kind, "topsql_")
	suffix := "/component=" + component
	withoutLast := strings.TrimSuffix(root, suffix)
	if withoutLast == root || !strings.HasSuffix(withoutLast, suffix) {
		return root, false
	}
	return withoutLast, true
}

func topSQLExternalRootFromSourceURI(raw string) (string, bool) {
	u, err := url.Parse(raw)
	if err != nil || u.Scheme != "gs" || u.Host == "" {
		return "", false
	}
	path := strings.TrimRight(u.Path, "/")
	for _, marker := range []string{"/instance=", "/date="} {
		if index := strings.Index(path, marker); index >= 0 {
			path = path[:index]
			break
		}
	}
	path = strings.TrimSuffix(path, "/*")
	if path == "" {
		return "", false
	}
	return "gs://" + u.Host + path, true
}

func splitNonEmpty(value string) []string {
	parts := strings.Split(value, ",")
	result := make([]string, 0, len(parts))
	for _, part := range parts {
		if part = strings.TrimSpace(part); part != "" {
			result = append(result, part)
		}
	}
	return result
}

func smokeTargetHash(value string) string {
	sum := sha256.Sum256([]byte(value))
	return hex.EncodeToString(sum[:6])
}