package engine

import (
	"errors"
	"sort"
	"strings"
)

const unknownFieldErrorMessage = "query references fields that cannot be resolved in the target cluster schema"

// UnknownFieldError is the vendor-neutral form of a query-engine failure that
// can be attributed to one or more unresolved SQL identifiers.
type UnknownFieldError struct {
	Engine  string
	QueryID string
	Fields  []string
}

func (e *UnknownFieldError) Error() string {
	return unknownFieldErrorMessage
}

// AsUnknownFieldError extracts a safely normalized unknown-field error.
func AsUnknownFieldError(err error) (*UnknownFieldError, bool) {
	var unknown *UnknownFieldError
	if !errors.As(err, &unknown) || unknown == nil {
		return nil, false
	}
	fields := normalizeUnknownFields(unknown.Fields)
	if len(fields) == 0 {
		return nil, false
	}
	return &UnknownFieldError{
		Engine:  unknown.Engine,
		QueryID: unknown.QueryID,
		Fields:  fields,
	}, true
}

// AsBigQueryQueryError extracts safe diagnostics for an unclassified BigQuery
// failure. It never exposes the upstream error message.
func AsBigQueryQueryError(err error) (*BigQueryQueryError, bool) {
	var queryErr *BigQueryQueryError
	if !errors.As(err, &queryErr) || queryErr == nil {
		return nil, false
	}
	queryID := queryErr.QueryID
	if !bigQueryQueryIDPattern.MatchString(queryID) {
		queryID = ""
	}
	reason := queryErr.Reason
	if !bigQueryReasonPattern.MatchString(reason) {
		reason = "unknown"
	}
	errorCount := queryErr.ErrorCount
	if errorCount < 0 {
		errorCount = 0
	}
	return &BigQueryQueryError{
		QueryID:       queryID,
		ErrorSource:   normalizeBigQueryErrorSource(queryErr.ErrorSource),
		Reason:        reason,
		ErrorCount:    errorCount,
		MessageShape:  normalizeBigQueryMessageShape(queryErr.MessageShape),
		HasSuggestion: queryErr.HasSuggestion,
	}, true
}

// AsBigQueryTopSQLSetupError extracts bounded setup diagnostics. It never
// exposes the query, source URI, or the upstream message.
func AsBigQueryTopSQLSetupError(err error) (*BigQueryTopSQLSetupError, bool) {
	var setupErr *BigQueryTopSQLSetupError
	if !errors.As(err, &setupErr) || setupErr == nil {
		return nil, false
	}
	result := *setupErr
	result.Operation = normalizeBigQueryTopSQLSetupOperation(result.Operation)
	result.ObjectKind = normalizeBigQueryTopSQLObjectKind(result.ObjectKind)
	result.ObjectState = normalizeBigQueryTopSQLObjectState(result.ObjectState)
	if !bigQueryQueryIDPattern.MatchString(result.QueryID) {
		result.QueryID = ""
	}
	if !bigQueryReasonPattern.MatchString(result.Reason) {
		result.Reason = "unknown"
	}
	result.ErrorSource = normalizeBigQueryErrorSource(result.ErrorSource)
	if result.HTTPStatus < 0 || result.HTTPStatus > 599 {
		result.HTTPStatus = 0
	}
	if result.InstanceCount < 0 {
		result.InstanceCount = 0
	}
	if result.SourceURICount < 0 {
		result.SourceURICount = 0
	}
	return &result, true
}

func normalizeBigQueryTopSQLSetupOperation(operation string) string {
	switch operation {
	case topSQLSetupOperationDiscover, topSQLSetupOperationMetadata, topSQLSetupOperationDDL, topSQLSetupOperationViewDDL, topSQLSetupOperationEmptyDDL:
		return operation
	default:
		return "unknown"
	}
}

func normalizeBigQueryTopSQLObjectKind(kind string) string {
	switch kind {
	case "topsql", "topsql_tidb", "topsql_tikv", "topsql_sql_meta", "topsql_plan_meta":
		return kind
	default:
		return "unknown"
	}
}

func normalizeBigQueryTopSQLObjectState(state string) string {
	switch state {
	case topSQLObjectStateCurrentParquet,
		topSQLObjectStateLegacyDelta,
		topSQLObjectStateLegacyNested,
		topSQLObjectStateLegacyBroadParquet,
		topSQLObjectStateManagedEmptyView,
		topSQLObjectStateObjectConflict:
		return state
	default:
		return "unknown"
	}
}

func newUnknownFieldError(engineName, queryID string, fields []string) error {
	fields = normalizeUnknownFields(fields)
	if len(fields) == 0 {
		return nil
	}
	return &UnknownFieldError{Engine: engineName, QueryID: queryID, Fields: fields}
}

func normalizeUnknownFields(fields []string) []string {
	unique := make(map[string]struct{}, len(fields))
	for _, field := range fields {
		field = strings.TrimSpace(field)
		if !isSafeSQLIdentifier(field) {
			continue
		}
		unique[field] = struct{}{}
	}
	result := make([]string, 0, len(unique))
	for field := range unique {
		result = append(result, field)
	}
	sort.Strings(result)
	return result
}

func isSafeSQLIdentifier(field string) bool {
	if field == "" || len(field) > 128 {
		return false
	}
	for index, r := range field {
		if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || r == '_' || (index > 0 && r >= '0' && r <= '9') {
			continue
		}
		return false
	}
	return true
}