package engine

import (
	"context"
	"crypto/sha256"
	"errors"
	"fmt"
	"net"
	"net/url"
	"regexp"
	"slices"
	"sort"
	"strings"
	"time"

	cloudauth "cloud.google.com/go/auth"
	"cloud.google.com/go/bigquery"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/dto"
	"github.com/pingcap-inc/clinic/clinic_data_service/pkg/log"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"google.golang.org/api/googleapi"
	"google.golang.org/api/iterator"
)

const (
	topSQLMaxSourceURIs  = 10000
	topSQLMaxDDLBytes    = 900 * 1024
	topSQLEmptyInstance  = "__clinic_no_instance__"
	topSQLEmptyDate      = "__clinic_no_date__"
	topSQLSetupJobWindow = 5 * time.Minute
	topSQLDiscoveryLimit = 8
)

const (
	topSQLObjectStateCurrentParquet     = "current_parquet"
	topSQLObjectStateLegacyDelta        = "legacy_delta"
	topSQLObjectStateLegacyNested       = "legacy_nested_component"
	topSQLObjectStateLegacyBroadParquet = "legacy_broad_parquet"
	topSQLObjectStateManagedEmptyView   = "managed_empty_view"
	topSQLObjectStateObjectConflict     = "object_conflict"
)

const (
	topSQLSetupOperationDiscover = "discover_source_uris"
	topSQLSetupOperationMetadata = "get_metadata"
	topSQLSetupOperationDDL      = "create_or_replace_external_table"
	topSQLSetupOperationViewDDL  = "create_or_replace_view"
	topSQLSetupOperationEmptyDDL = "create_empty_view"
)

var (
	topSQLInstanceNamePattern      = regexp.MustCompile("^[A-Za-z0-9_.:-]{1,255}$")
	errTopSQLInvalidDateLayout     = errors.New("invalid GCS date layout")
	errTopSQLInvalidInstanceLayout = errors.New("invalid GCS instance layout")
	errTopSQLInvalidInstanceName   = errors.New("invalid GCS instance name")
)

const (
	topSQLReasonGCSListFailed         = "gcs_list_failed"
	topSQLReasonInvalidDateLayout     = "invalid_date_layout"
	topSQLReasonInvalidInstanceLayout = "invalid_instance_layout"
	topSQLReasonInvalidInstanceName   = "invalid_instance_name"
	topSQLReasonCredentialError       = "credential_error"
	topSQLReasonNetworkError          = "network_error"
)

// BigQueryTopSQLSetupError retains diagnostics that are useful to operators
// without retaining SQL, GCS paths, object names, or upstream messages.
type BigQueryTopSQLSetupError struct {
	Operation      string
	ObjectKind     string
	ObjectState    string
	Reason         string
	ErrorSource    string
	HTTPStatus     int
	QueryID        string
	InstanceCount  int
	SourceURICount int
}

func (e *BigQueryTopSQLSetupError) Error() string {
	return "BigQuery TopSQL setup failed"
}

type topSQLExternalDefinition struct {
	root               string
	kind               string
	month              string
	sourceURIs         []string
	partitionColumns   []string
	partitionURIPrefix string
	schema             bigquery.Schema
	instanceCount      int
	empty              bool
}

type topSQLSetupJob interface {
	ID() string
	Wait(context.Context) (*bigquery.JobStatus, error)
}

type topSQLSetupJobRunner interface {
	Run(context.Context, string, string, string) (topSQLSetupJob, error)
	Lookup(context.Context, string, string) (topSQLSetupJob, error)
}

type bigQuerySetupJobRunner struct {
	client *bigquery.Client
}

func (r *bigQuerySetupJobRunner) Run(ctx context.Context, query, jobID, location string) (topSQLSetupJob, error) {
	q := r.client.Query(query)
	q.UseStandardSQL = true
	q.JobID = jobID
	q.Location = location
	return q.Run(ctx)
}

func (r *bigQuerySetupJobRunner) Lookup(ctx context.Context, jobID, location string) (topSQLSetupJob, error) {
	return r.client.JobFromIDLocation(ctx, jobID, location)
}

// topSQLExternalTableDefinition is a deterministic helper for unit tests and
// static DDL validation. Runtime setup must use discovery instead.
func topSQLExternalTableDefinition(table, gcsURI string) (*topSQLExternalDefinition, error) {
	kind := topSQLTableKind(table)
	root := strings.TrimRight(gcsURI, "/")
	if kind == "" || !strings.HasPrefix(root, "gs://") {
		return nil, errors.New("invalid TopSQL external table definition")
	}
	definition := &topSQLExternalDefinition{root: root, kind: kind, month: topSQLTableMonth(table), schema: topSQLBigQuerySchema(kind)}
	if kind == "topsql_tidb" || kind == "topsql_tikv" {
		definition.partitionColumns = []string{"instance", "date"}
		definition.sourceURIs = topSQLInstanceSourceURIs(root, nil, definition.month)
	} else {
		definition.partitionColumns = []string{"date"}
		definition.sourceURIs = []string{root + "/date=" + topSQLDatePattern(definition.month)}
	}
	definition.partitionURIPrefix = root
	return definition, validateTopSQLSourceURIs(definition)
}

func (e *BigQueryEngine) EnsureTopSQLExternalTable(ctx context.Context, dataset, table, gcsURI, region string) error {
	key := strings.Join([]string{dataset, table, strings.TrimRight(gcsURI, "/"), region}, "|")
	_, err, _ := e.topSQLSourceGroup.Do(key, func() (interface{}, error) {
		definition, err := e.discoverTopSQLExternalDefinition(ctx, table, gcsURI)
		if err != nil {
			return nil, err
		}
		fingerprint := topSQLSourceFingerprint(definition.sourceURIs)
		if previous, ok := e.topSQLSources.Load(key); ok && previous == fingerprint {
			return nil, nil
		}
		if err := e.reconcileTopSQLExternalTable(ctx, dataset, table, definition, region); err != nil {
			return nil, err
		}
		e.topSQLSources.Store(key, fingerprint)
		return nil, nil
	})
	return err
}

func (e *BigQueryEngine) discoverTopSQLExternalDefinition(ctx context.Context, table, gcsURI string) (*topSQLExternalDefinition, error) {
	kind := topSQLTableKind(table)
	if kind == "" {
		return nil, newBigQueryTopSQLSetupError(topSQLSetupOperationDiscover, "", topSQLObjectStateObjectConflict, "invalid_table", 0, 0, nil)
	}
	root := strings.TrimRight(gcsURI, "/")
	if !strings.HasPrefix(root, "gs://") {
		return nil, newBigQueryTopSQLSetupError(topSQLSetupOperationDiscover, kind, topSQLObjectStateObjectConflict, "invalid_source", 0, 0, nil)
	}
	definition := &topSQLExternalDefinition{root: root, kind: kind, month: topSQLTableMonth(table), schema: topSQLBigQuerySchema(kind)}
	if kind == "topsql_tidb" || kind == "topsql_tikv" {
		definition.partitionColumns = []string{"instance", "date"}
		definition.partitionURIPrefix = root
		instances, err := e.discoverTopSQLInstances(ctx, root, definition.month)
		if err != nil {
			return nil, newBigQueryTopSQLSetupError(topSQLSetupOperationDiscover, kind, topSQLObjectStateCurrentParquet, topSQLDiscoveryReason(err), 0, 0, err)
		}
		definition.sourceURIs = topSQLInstanceSourceURIs(root, instances, definition.month)
		definition.instanceCount = len(instances)
		definition.empty = len(instances) == 0
	} else {
		definition.partitionColumns = []string{"date"}
		definition.partitionURIPrefix = root
		hasDatePartitions, err := e.topSQLHasDatePartitions(ctx, root, definition.month)
		if err != nil {
			return nil, newBigQueryTopSQLSetupError(topSQLSetupOperationDiscover, kind, topSQLObjectStateCurrentParquet, topSQLDiscoveryReason(err), 0, 0, err)
		}
		if hasDatePartitions {
			definition.sourceURIs = []string{root + "/date=" + topSQLDatePattern(definition.month)}
		} else {
			definition.sourceURIs = []string{root + "/date=" + topSQLEmptyDate + "/*"}
			definition.empty = true
		}
	}
	if err := validateTopSQLSourceURIs(definition); err != nil {
		return nil, newBigQueryTopSQLSetupError(topSQLSetupOperationDiscover, kind, topSQLObjectStateCurrentParquet, "source_uri_limit", 0, len(definition.sourceURIs), err)
	}
	return definition, nil
}

func (e *BigQueryEngine) topSQLHasDatePartitions(ctx context.Context, root, month string) (bool, error) {
	source, err := dto.ParseDataSource(root)
	if err != nil || source == nil || source.Bucket == "" {
		return false, errors.New("invalid GCS source")
	}
	lister, err := e.getGCSObjectLister(ctx)
	if err != nil {
		return false, err
	}
	prefix := strings.Trim(source.Path, "/")
	if prefix != "" {
		prefix += "/"
	}
	listPrefix := prefix
	if month != "" {
		listPrefix += "date=" + topSQLMonthPathPrefix(month)
	}
	objectIterator := lister.ListPrefixes(ctx, source.Bucket, listPrefix)
	for {
		name, err := objectIterator.Next()
		if err == iterator.Done {
			return false, nil
		}
		if err != nil {
			return false, err
		}
		if !strings.HasPrefix(name, prefix) || !strings.HasSuffix(name, "/") {
			return false, errTopSQLInvalidDateLayout
		}
		entry := strings.TrimSuffix(strings.TrimPrefix(name, prefix), "/")
		if entry == "_delta_log" {
			continue
		}
		if strings.HasPrefix(entry, "date=") && !strings.Contains(entry, "/") && validTopSQLPartitionDate(strings.TrimPrefix(entry, "date="), month) {
			return true, nil
		}
		return false, errTopSQLInvalidDateLayout
	}
}

func (e *BigQueryEngine) discoverTopSQLInstances(ctx context.Context, root, month string) ([]string, error) {
	source, err := dto.ParseDataSource(root)
	if err != nil || source == nil || source.Bucket == "" {
		return nil, errors.New("invalid GCS source")
	}
	lister, err := e.getGCSObjectLister(ctx)
	if err != nil {
		return nil, err
	}
	prefix := strings.Trim(source.Path, "/")
	if prefix != "" {
		prefix += "/"
	}
	objectIterator := lister.ListPrefixes(ctx, source.Bucket, prefix)
	instances := make(map[string]struct{})
	for {
		name, err := objectIterator.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return nil, err
		}
		if !strings.HasPrefix(name, prefix) || !strings.HasSuffix(name, "/") {
			return nil, errTopSQLInvalidInstanceLayout
		}
		entry := strings.TrimSuffix(strings.TrimPrefix(name, prefix), "/")
		if entry == "_delta_log" {
			continue
		}
		if !strings.HasPrefix(entry, "instance=") || strings.Contains(entry, "/") {
			return nil, errTopSQLInvalidInstanceLayout
		}
		instance := strings.TrimPrefix(entry, "instance=")
		if instance == topSQLEmptyInstance || !topSQLInstanceNamePattern.MatchString(instance) {
			return nil, errTopSQLInvalidInstanceName
		}
		instances[instance] = struct{}{}
	}
	candidates := make([]string, 0, len(instances))
	for instance := range instances {
		candidates = append(candidates, instance)
	}
	sort.Strings(candidates)
	hasDatePartitions := make([]bool, len(candidates))
	group, groupCtx := errgroup.WithContext(ctx)
	group.SetLimit(topSQLDiscoveryLimit)
	for index, instance := range candidates {
		index, instance := index, instance
		group.Go(func() error {
			available, err := e.topSQLHasDatePartitions(groupCtx, root+"/instance="+instance, month)
			if err == nil {
				hasDatePartitions[index] = available
			}
			return err
		})
	}
	if err := group.Wait(); err != nil {
		return nil, err
	}
	result := make([]string, 0, len(candidates))
	for index, instance := range candidates {
		if hasDatePartitions[index] {
			result = append(result, instance)
		}
	}
	return result, nil
}

func topSQLDiscoveryReason(err error) string {
	switch {
	case errors.Is(err, errTopSQLInvalidDateLayout):
		return topSQLReasonInvalidDateLayout
	case errors.Is(err, errTopSQLInvalidInstanceLayout):
		return topSQLReasonInvalidInstanceLayout
	case errors.Is(err, errTopSQLInvalidInstanceName):
		return topSQLReasonInvalidInstanceName
	default:
		return topSQLReasonGCSListFailed
	}
}

func topSQLInstanceSourceURIs(root string, instances []string, month string) []string {
	datePattern := topSQLDatePattern(month)
	if len(instances) == 0 {
		return []string{root + "/instance=" + topSQLEmptyInstance + "/date=" + datePattern}
	}
	uris := make([]string, 0, len(instances))
	for _, instance := range instances {
		uris = append(uris, root+"/instance="+instance+"/date="+datePattern)
	}
	return uris
}

func topSQLDatePattern(month string) string {
	if month == "" {
		return "*"
	}
	return topSQLMonthPathPrefix(month) + "*"
}

func topSQLMonthPathPrefix(month string) string {
	return month[:4] + "-" + month[4:] + "-"
}

func validTopSQLPartitionDate(value, month string) bool {
	date, err := time.Parse("2006-01-02", value)
	if err != nil {
		return false
	}
	return month == "" || date.Format("200601") == month
}

func validateTopSQLSourceURIs(definition *topSQLExternalDefinition) error {
	if len(definition.sourceURIs) == 0 || len(definition.sourceURIs) > topSQLMaxSourceURIs {
		return errors.New("TopSQL source URI count exceeds limit")
	}
	for _, uri := range definition.sourceURIs {
		if strings.Count(uri, "*") != 1 || strings.Contains(uri, "/_delta_log/") {
			return errors.New("invalid TopSQL source URI")
		}
	}
	if len(topSQLExternalTableDDL("project", "dataset", "table", definition)) > topSQLMaxDDLBytes {
		return errors.New("TopSQL DDL exceeds limit")
	}
	return nil
}

func (e *BigQueryEngine) reconcileTopSQLExternalTable(ctx context.Context, dataset, table string, definition *topSQLExternalDefinition, region string) error {
	tbl := e.client.Dataset(dataset).Table(table)
	reconciliationVersion := "missing"
	existingState := "missing"
	metadata, err := tbl.Metadata(ctx)
	if err == nil {
		reconciliationVersion = topSQLReconciliationVersion(metadata)
		state := classifyTopSQLExternalObject(metadata, definition)
		existingState = state
		if definition.empty && state == topSQLObjectStateManagedEmptyView {
			return nil
		}
		if !definition.empty && state == topSQLObjectStateCurrentParquet && isExpectedTopSQLExternalTable(metadata, definition) {
			return nil
		}
		if state == topSQLObjectStateObjectConflict {
			return newBigQueryTopSQLSetupError(topSQLSetupOperationMetadata, definition.kind, state, "object_conflict", definition.instanceCount, len(definition.sourceURIs), nil)
		}
	} else if !isBigQueryNotFound(err) {
		return newBigQueryTopSQLSetupError(topSQLSetupOperationMetadata, definition.kind, "", "metadata_failed", definition.instanceCount, len(definition.sourceURIs), err)
	}
	setupOperation := topSQLSetupOperationDDL
	desiredState := topSQLObjectStateCurrentParquet
	setupQuery := topSQLExternalTableDDL(e.projectID, dataset, table, definition)
	if definition.empty {
		setupOperation = topSQLSetupOperationEmptyDDL
		desiredState = topSQLObjectStateManagedEmptyView
		setupQuery = topSQLEmptyViewDDL(e.projectID, dataset, table, definition)
		if existingState != "missing" {
			setupQuery = "DROP TABLE IF EXISTS " + bigQueryObjectName(e.projectID, dataset, table) + "; " + setupQuery
		}
	} else if existingState == topSQLObjectStateManagedEmptyView {
		setupQuery = "DROP VIEW IF EXISTS " + bigQueryObjectName(e.projectID, dataset, table) + "; " + setupQuery
	}
	queryID, err := e.runTopSQLSetupQuery(ctx, setupQuery, region, reconciliationVersion)
	if err != nil {
		return newBigQueryTopSQLSetupError(setupOperation, definition.kind, desiredState, "ddl_failed", definition.instanceCount, len(definition.sourceURIs), err).withQueryID(queryID)
	}
	log.Info(ctx, "TOPSQL_BIGQUERY_SOURCE_REFRESHED",
		zap.String("topsql_object_kind", definition.kind),
		zap.Int("topsql_instance_count", definition.instanceCount),
		zap.Int("topsql_source_uri_count", len(definition.sourceURIs)),
		zap.String("bigquery_query_id", queryID),
	)
	return nil
}

func (e *BigQueryEngine) EnsureTopSQLView(ctx context.Context, dataset, table, tidbRoot, region string) error {
	scope, ok := parseBigQueryTopSQLViewScope(table)
	if !ok {
		return newBigQueryTopSQLSetupError(topSQLSetupOperationViewDDL, "", topSQLObjectStateObjectConflict, "invalid_table", 0, 0, nil)
	}
	tbl := e.client.Dataset(dataset).Table(table)
	reconciliationVersion := "missing"
	replaceLegacyTable := false
	if metadata, err := tbl.Metadata(ctx); err == nil {
		reconciliationVersion = topSQLReconciliationVersion(metadata)
		if metadata.Type == bigquery.ViewTable {
			return nil
		}
		if !isReplaceableTopSQLViewObject(metadata, tidbRoot) {
			return newBigQueryTopSQLSetupError(topSQLSetupOperationMetadata, "topsql", topSQLObjectStateObjectConflict, "object_conflict", 0, 0, nil)
		}
		replaceLegacyTable = true
	} else if !isBigQueryNotFound(err) {
		return newBigQueryTopSQLSetupError(topSQLSetupOperationMetadata, "topsql", "", "metadata_failed", 0, 0, err)
	}
	setupQuery := topSQLViewSetupQuery(e.projectID, dataset, table, scope, replaceLegacyTable)
	queryID, err := e.runTopSQLSetupQuery(ctx, setupQuery, region, reconciliationVersion)
	if err != nil {
		return newBigQueryTopSQLSetupError(topSQLSetupOperationViewDDL, "topsql", topSQLObjectStateCurrentParquet, "ddl_failed", 0, 0, err).withQueryID(queryID)
	}
	return nil
}

func isExpectedTopSQLExternalTable(metadata *bigquery.TableMetadata, definition *topSQLExternalDefinition) bool {
	if metadata == nil || metadata.Type != bigquery.ExternalTable || metadata.ExternalDataConfig == nil {
		return false
	}
	config := metadata.ExternalDataConfig
	if config.SourceFormat != bigquery.Parquet || !slices.Equal(config.SourceURIs, definition.sourceURIs) || config.HivePartitioningOptions == nil {
		return false
	}
	hive := config.HivePartitioningOptions
	return hive.Mode == bigquery.CustomHivePartitioningMode && hive.SourceURIPrefix == definition.partitionURIPrefix && hive.RequirePartitionFilter
}

func classifyTopSQLExternalObject(metadata *bigquery.TableMetadata, definition *topSQLExternalDefinition) string {
	if isExpectedTopSQLEmptyView(metadata, definition) {
		return topSQLObjectStateManagedEmptyView
	}
	if metadata == nil || metadata.Type != bigquery.ExternalTable || metadata.ExternalDataConfig == nil {
		return topSQLObjectStateObjectConflict
	}
	config := metadata.ExternalDataConfig
	if isExpectedTopSQLExternalTable(metadata, definition) || topSQLManagedSourceURIs(config.SourceURIs, definition) {
		return topSQLObjectStateCurrentParquet
	}
	if len(config.SourceURIs) != 1 {
		return topSQLObjectStateObjectConflict
	}
	uri := strings.TrimSuffix(strings.TrimRight(config.SourceURIs[0], "/"), "/*")
	if config.SourceFormat == DeltaLakeFormat && uri == definition.root {
		return topSQLObjectStateLegacyDelta
	}
	if config.SourceFormat != bigquery.Parquet {
		return topSQLObjectStateObjectConflict
	}
	if uri == definition.root {
		return topSQLObjectStateLegacyBroadParquet
	}
	if strings.HasSuffix(uri, "/component="+strings.TrimPrefix(definition.kind, "topsql_")) {
		return topSQLObjectStateLegacyNested
	}
	return topSQLObjectStateObjectConflict
}

func isExpectedTopSQLEmptyView(metadata *bigquery.TableMetadata, definition *topSQLExternalDefinition) bool {
	return metadata != nil && metadata.Type == bigquery.ViewTable &&
		strings.TrimSpace(metadata.ViewQuery) == topSQLEmptyViewQuery(definition)
}

func topSQLManagedSourceURIs(uris []string, definition *topSQLExternalDefinition) bool {
	if len(uris) == 0 {
		return false
	}
	if definition.kind == "topsql_tidb" || definition.kind == "topsql_tikv" {
		for _, uri := range uris {
			prefix := definition.root + "/instance="
			dateSuffix := "/date=" + topSQLDatePattern(definition.month)
			if !strings.HasPrefix(uri, prefix) || !strings.HasSuffix(uri, dateSuffix) {
				return false
			}
			instance := strings.TrimSuffix(strings.TrimPrefix(uri, prefix), dateSuffix)
			if instance != topSQLEmptyInstance && !topSQLInstanceNamePattern.MatchString(instance) {
				return false
			}
		}
		return true
	}
	return len(uris) == 1 && (uris[0] == definition.root+"/date="+topSQLDatePattern(definition.month) || uris[0] == definition.root+"/date="+topSQLEmptyDate+"/*")
}

func isReplaceableTopSQLViewObject(metadata *bigquery.TableMetadata, tidbRoot string) bool {
	if metadata == nil || metadata.Type != bigquery.ExternalTable || metadata.ExternalDataConfig == nil || len(metadata.ExternalDataConfig.SourceURIs) != 1 {
		return false
	}
	uri := strings.TrimSuffix(strings.TrimRight(metadata.ExternalDataConfig.SourceURIs[0], "/"), "/*")
	return (metadata.ExternalDataConfig.SourceFormat == DeltaLakeFormat || metadata.ExternalDataConfig.SourceFormat == bigquery.Parquet) && uri == strings.TrimRight(tidbRoot, "/")
}

func topSQLExternalTableDDL(projectID, dataset, table string, definition *topSQLExternalDefinition) string {
	fields := make([]string, 0, len(definition.schema))
	for _, field := range definition.schema {
		typ := string(field.Type)
		if field.Type == bigquery.IntegerFieldType {
			typ = "INT64"
		}
		fields = append(fields, field.Name+" "+typ)
	}
	partitions := make([]string, 0, len(definition.partitionColumns))
	for _, column := range definition.partitionColumns {
		partitions = append(partitions, column+" STRING")
	}
	uris := make([]string, 0, len(definition.sourceURIs))
	for _, uri := range definition.sourceURIs {
		uris = append(uris, topSQLStringLiteral(uri))
	}
	return fmt.Sprintf("CREATE OR REPLACE EXTERNAL TABLE %s (%s) WITH PARTITION COLUMNS (%s) OPTIONS (format = 'PARQUET', uris = [%s], hive_partition_uri_prefix = %s, require_hive_partition_filter = TRUE)", bigQueryObjectName(projectID, dataset, table), strings.Join(fields, ", "), strings.Join(partitions, ", "), strings.Join(uris, ", "), topSQLStringLiteral(definition.partitionURIPrefix))
}

func topSQLEmptyViewDDL(projectID, dataset, table string, definition *topSQLExternalDefinition) string {
	return "CREATE VIEW " + bigQueryObjectName(projectID, dataset, table) + " AS " + topSQLEmptyViewQuery(definition)
}

func topSQLEmptyViewQuery(definition *topSQLExternalDefinition) string {
	fields := make([]string, 0, len(definition.schema)+len(definition.partitionColumns))
	for _, field := range definition.schema {
		typ := string(field.Type)
		if field.Type == bigquery.IntegerFieldType {
			typ = "INT64"
		}
		fields = append(fields, fmt.Sprintf("CAST(NULL AS %s) AS %s", typ, bigQueryColumnName(field.Name)))
	}
	for _, column := range definition.partitionColumns {
		fields = append(fields, "CAST(NULL AS STRING) AS "+bigQueryColumnName(column))
	}
	return "SELECT " + strings.Join(fields, ", ") + " WHERE FALSE"
}

func bigQueryColumnName(column string) string {
	return fmt.Sprintf("%c%s%c", 96, column, 96)
}

func topSQLStringLiteral(value string) string {
	return "'" + strings.ReplaceAll(value, "'", "\\'") + "'"
}

func topSQLBigQuerySchema(kind string) bigquery.Schema {
	fields := []struct {
		name string
		typ  bigquery.FieldType
	}{
		{"db", bigquery.StringFieldType}, {"keyspace", bigquery.StringFieldType},
		{"plan_digest", bigquery.StringFieldType}, {"region_id", bigquery.StringFieldType},
		{"sql_digest", bigquery.StringFieldType}, {"table", bigquery.StringFieldType},
		{"table_id", bigquery.IntegerFieldType}, {"tag_label", bigquery.StringFieldType},
		{"timestamps", bigquery.IntegerFieldType}, {"topsql_cpu_time_ms", bigquery.IntegerFieldType},
		{"topsql_logical_read_bytes", bigquery.IntegerFieldType}, {"topsql_logical_write_bytes", bigquery.IntegerFieldType},
		{"topsql_network_in_bytes", bigquery.IntegerFieldType}, {"topsql_network_out_bytes", bigquery.IntegerFieldType},
		{"topsql_read_keys", bigquery.IntegerFieldType}, {"topsql_stmt_duration_count", bigquery.IntegerFieldType},
		{"topsql_stmt_duration_sum_ns", bigquery.IntegerFieldType}, {"topsql_stmt_exec_count", bigquery.IntegerFieldType},
		{"topsql_write_keys", bigquery.IntegerFieldType},
	}
	if kind == "topsql_sql_meta" {
		fields = []struct {
			name string
			typ  bigquery.FieldType
		}{{"normalized_sql", bigquery.StringFieldType}, {"sql_digest", bigquery.StringFieldType}}
	}
	if kind == "topsql_plan_meta" {
		fields = []struct {
			name string
			typ  bigquery.FieldType
		}{{"encoded_normalized_plan", bigquery.StringFieldType}, {"normalized_plan", bigquery.StringFieldType}, {"plan_digest", bigquery.StringFieldType}}
	}
	schema := make(bigquery.Schema, 0, len(fields))
	for _, field := range fields {
		schema = append(schema, &bigquery.FieldSchema{Name: field.name, Type: field.typ})
	}
	return schema
}

func topSQLViewDDL(projectID, dataset, table string, scope bigQueryTopSQLViewScope) string {
	base := fmt.Sprintf("%c%s.%s", 96, projectID, dataset)
	fields := []string{"t.db", "t.keyspace", "t.plan_digest", "t.region_id", "t.sql_digest", "t.\u0060table\u0060", "t.table_id", "t.tag_label", "t.timestamps", "t.topsql_cpu_time_ms", "t.topsql_logical_read_bytes", "t.topsql_logical_write_bytes", "t.topsql_network_in_bytes", "t.topsql_network_out_bytes", "t.topsql_read_keys", "t.topsql_stmt_duration_count", "t.topsql_stmt_duration_sum_ns", "t.topsql_stmt_exec_count", "t.topsql_write_keys", "t.instance", "t.date"}
	selectPart := func(source, component, sqlMeta, planMeta string) string {
		return fmt.Sprintf("SELECT %s, '%s' AS component, sm.normalized_sql, pm.normalized_plan, pm.encoded_normalized_plan FROM %s t LEFT JOIN %s sm ON t.sql_digest = sm.sql_digest AND t.date = sm.date LEFT JOIN %s pm ON t.plan_digest = pm.plan_digest AND t.date = pm.date", strings.Join(fields, ", "), component, source, sqlMeta, planMeta)
	}
	months := scope.months
	if len(months) == 0 {
		months = []string{""}
	}
	selects := make([]string, 0, len(months)*2)
	for _, month := range months {
		suffix := scope.clusterID
		if month != "" {
			suffix += "_m" + month
		}
		tidb := base + ".topsql_tidb_" + suffix + "\u0060"
		tikv := base + ".topsql_tikv_" + suffix + "\u0060"
		sqlMeta := base + ".topsql_sql_meta_" + suffix + "\u0060"
		planMeta := base + ".topsql_plan_meta_" + suffix + "\u0060"
		selects = append(selects,
			selectPart(tidb, "tidb", sqlMeta, planMeta),
			selectPart(tikv, "tikv", sqlMeta, planMeta),
		)
	}
	return "CREATE OR REPLACE VIEW " + bigQueryObjectName(projectID, dataset, table) + " AS " + strings.Join(selects, " UNION ALL ")
}

func topSQLViewSetupQuery(projectID, dataset, table string, scope bigQueryTopSQLViewScope, replaceLegacyTable bool) string {
	query := topSQLViewDDL(projectID, dataset, table, scope)
	if replaceLegacyTable {
		return "DROP TABLE IF EXISTS " + bigQueryObjectName(projectID, dataset, table) + "; " + query
	}
	return query
}

func bigQueryObjectName(projectID, dataset, table string) string {
	parts := []string{projectID, dataset}
	if table != "" {
		parts = append(parts, table)
	}
	return fmt.Sprintf("%c%s%c", 96, strings.Join(parts, "."), 96)
}

func topSQLReconciliationVersion(metadata *bigquery.TableMetadata) string {
	if metadata != nil && metadata.ETag != "" {
		return metadata.ETag
	}
	return "existing"
}

func topSQLSetupJobID(query, reconciliationVersion string, now time.Time) string {
	bucket := now.UTC().Truncate(topSQLSetupJobWindow).Unix()
	sum := sha256.Sum256([]byte(fmt.Sprintf("%s\n%s\n%d", query, reconciliationVersion, bucket)))
	return fmt.Sprintf("clinic_topsql_%d_%x", bucket, sum[:12])
}

func (e *BigQueryEngine) runTopSQLSetupQuery(ctx context.Context, query, region, reconciliationVersion string) (string, error) {
	runner := e.topSQLSetupJobs
	if runner == nil {
		if e.client == nil {
			return "", errors.New("BigQuery setup job runner is unavailable")
		}
		runner = &bigQuerySetupJobRunner{client: e.client}
	}
	now := time.Now
	if e.topSQLNow != nil {
		now = e.topSQLNow
	}
	jobID := topSQLSetupJobID(query, reconciliationVersion, now())
	job, err := runner.Run(ctx, query, jobID, region)
	if err != nil {
		if !isBigQueryAlreadyExists(err) {
			return "", err
		}
		job, err = runner.Lookup(ctx, jobID, region)
		if err != nil {
			return jobID, err
		}
	}
	queryID := job.ID()
	status, err := job.Wait(ctx)
	if err != nil {
		return queryID, err
	}
	if status == nil {
		return queryID, errors.New("BigQuery setup job returned no status")
	}
	if err := status.Err(); err != nil {
		return queryID, err
	}
	return queryID, nil
}

func isBigQueryAlreadyExists(err error) bool {
	var apiErr *googleapi.Error
	return errors.As(err, &apiErr) && apiErr != nil && apiErr.Code == 409
}

func newBigQueryTopSQLSetupError(operation, kind, state, reason string, instances, sourceURIs int, cause error) *BigQueryTopSQLSetupError {
	result := &BigQueryTopSQLSetupError{Operation: operation, ObjectKind: kind, ObjectState: state, Reason: reason, ErrorSource: bigQueryErrorSourceUnknown, InstanceCount: instances, SourceURICount: sourceURIs}
	if cause == nil {
		return result
	}
	result.ErrorSource, result.Reason = bigQueryTopSQLFailureDetails(cause, reason)
	var apiErr *googleapi.Error
	if errors.As(cause, &apiErr) && apiErr != nil {
		result.HTTPStatus = apiErr.Code
		return result
	}
	var authErr *cloudauth.Error
	if errors.As(cause, &authErr) && authErr != nil && authErr.Response != nil {
		result.HTTPStatus = authErr.Response.StatusCode
	}
	return result
}

func (e *BigQueryTopSQLSetupError) withQueryID(queryID string) *BigQueryTopSQLSetupError {
	if bigQueryQueryIDPattern.MatchString(queryID) {
		e.QueryID = queryID
	}
	return e
}

func bigQueryTopSQLFailureDetails(err error, fallback string) (string, string) {
	if reason := topSQLDiscoveryReason(err); reason != topSQLReasonGCSListFailed {
		return bigQueryErrorSourceValidation, reason
	}
	var queryErr *bigquery.Error
	if errors.As(err, &queryErr) && queryErr != nil {
		if bigQueryReasonPattern.MatchString(queryErr.Reason) {
			return bigQueryErrorSourceBigQuery, queryErr.Reason
		}
		return bigQueryErrorSourceBigQuery, fallback
	}
	var apiErr *googleapi.Error
	if errors.As(err, &apiErr) && apiErr != nil {
		if len(apiErr.Errors) > 0 && bigQueryReasonPattern.MatchString(apiErr.Errors[0].Reason) {
			return bigQueryErrorSourceGoogleAPI, apiErr.Errors[0].Reason
		}
		return bigQueryErrorSourceGoogleAPI, fallback
	}
	var authErr *cloudauth.Error
	if errors.As(err, &authErr) && authErr != nil {
		return bigQueryErrorSourceAuth, topSQLReasonCredentialError
	}
	var urlErr *url.Error
	if errors.As(err, &urlErr) && urlErr != nil {
		return bigQueryErrorSourceNetwork, topSQLReasonNetworkError
	}
	var netErr net.Error
	if errors.As(err, &netErr) && netErr != nil {
		return bigQueryErrorSourceNetwork, topSQLReasonNetworkError
	}
	return bigQueryErrorSourceGCS, fallback
}

func isBigQueryNotFound(err error) bool {
	if err == nil {
		return false
	}
	var apiErr *googleapi.Error
	if errors.As(err, &apiErr) && apiErr != nil && apiErr.Code == 404 {
		return true
	}
	return strings.Contains(err.Error(), "notFound") || strings.Contains(err.Error(), "Not found")
}

func topSQLSourceFingerprint(uris []string) string {
	copyURIs := append([]string(nil), uris...)
	sort.Strings(copyURIs)
	sum := sha256.Sum256([]byte(strings.Join(copyURIs, "\n")))
	return fmt.Sprintf("%x", sum[:])
}