package query

import (
	"bytes"
	"encoding/json"
	"fmt"
	"strings"
)

// ScopeRange filters a partition (or comparable) column with open/closed bounds.
// At least one bound must be set. Bounds are combined with AND.
type ScopeRange struct {
	Gte string `json:"gte,omitempty"`
	Gt  string `json:"gt,omitempty"`
	Lte string `json:"lte,omitempty"`
	Lt  string `json:"lt,omitempty"`
}

// ScopeValue is one scope entry: equality string, IN-list, or range object.
//
// JSON examples:
//
//	"dedicated"
//	["tidb", "tikv"]
//	{"gte":"2026-07-17","lt":"2026-07-20"}
type ScopeValue struct {
	eq  *string
	in  []string
	rng *ScopeRange
}

// Scope binds partition dive-path filters for a dataset.
// Keys are logical field names (e.g. deploy_type, components, date); the
// server maps them onto the physical partition columns of the chosen dataset.
type Scope map[string]ScopeValue

// ScopeEq builds an equality scope value.
func ScopeEq(v string) ScopeValue {
	s := v
	return ScopeValue{eq: &s}
}

// ScopeIn builds an IN-list scope value.
func ScopeIn(vals ...string) ScopeValue {
	out := make([]string, len(vals))
	copy(out, vals)
	return ScopeValue{in: out}
}

// ScopeRng builds a range scope value.
func ScopeRng(r ScopeRange) ScopeValue {
	cp := r
	return ScopeValue{rng: &cp}
}

func (v *ScopeValue) UnmarshalJSON(data []byte) error {
	data = bytes.TrimSpace(data)
	if len(data) == 0 || bytes.Equal(data, []byte("null")) {
		return fmt.Errorf("scope value must not be null")
	}
	switch data[0] {
	case '"':
		var s string
		if err := json.Unmarshal(data, &s); err != nil {
			return err
		}
		v.eq = &s
		v.in = nil
		v.rng = nil
		return nil
	case '[':
		var arr []string
		if err := json.Unmarshal(data, &arr); err != nil {
			return fmt.Errorf("scope array values must be strings: %w", err)
		}
		v.eq = nil
		v.in = arr
		v.rng = nil
		return nil
	case '{':
		var r ScopeRange
		if err := json.Unmarshal(data, &r); err != nil {
			return fmt.Errorf("scope range object: %w", err)
		}
		if r.Gte == "" && r.Gt == "" && r.Lte == "" && r.Lt == "" {
			return fmt.Errorf("scope range requires at least one of gte, gt, lte, lt")
		}
		v.eq = nil
		v.in = nil
		v.rng = &r
		return nil
	default:
		return fmt.Errorf("scope value must be a string, string array, or range object")
	}
}

func (v ScopeValue) MarshalJSON() ([]byte, error) {
	switch {
	case v.eq != nil:
		return json.Marshal(*v.eq)
	case v.in != nil:
		return json.Marshal(v.in)
	case v.rng != nil:
		return json.Marshal(v.rng)
	default:
		return []byte("null"), nil
	}
}

func (v ScopeValue) isZero() bool {
	return v.eq == nil && len(v.in) == 0 && v.rng == nil
}

// isBlank reports whether the value is absent or an empty equality / empty IN-list.
func (v ScopeValue) isBlank() bool {
	if v.isZero() {
		return true
	}
	if v.eq != nil && strings.TrimSpace(*v.eq) == "" {
		return true
	}
	if v.in != nil {
		for _, s := range v.in {
			if strings.TrimSpace(s) != "" && strings.TrimSpace(s) != "*" {
				return false
			}
		}
		return true
	}
	return false
}

// GetString returns the equality string for key, if present.
func (s Scope) GetString(key string) (string, bool) {
	if s == nil {
		return "", false
	}
	v, ok := s[key]
	if !ok || v.eq == nil {
		return "", false
	}
	return *v.eq, true
}

func (s Scope) lookup(keys ...string) (ScopeValue, string, bool) {
	if s == nil {
		return ScopeValue{}, "", false
	}
	for _, k := range keys {
		if v, ok := s[k]; ok && !v.isBlank() {
			return v, k, true
		}
	}
	return ScopeValue{}, "", false
}

// scopeFieldSpec maps a logical scope key onto a physical SQL column.
type scopeFieldSpec struct {
	APIKeys   []string // accepted JSON keys, first is canonical
	SQLColumn string
	Required  bool
	Normalize func(string) (string, error)
	// AllowLegacyDatePair accepts begin_date+end_date when date is absent.
	AllowLegacyDatePair bool
}

func (f scopeFieldSpec) resolve(scope Scope) (ScopeValue, error) {
	if v, _, ok := scope.lookup(f.APIKeys...); ok {
		return v, nil
	}
	if f.AllowLegacyDatePair {
		begin, hasB := scope.GetString("begin_date")
		end, hasE := scope.GetString("end_date")
		if hasB || hasE {
			if !hasB || !hasE {
				return ScopeValue{}, fmt.Errorf("scope.begin_date and scope.end_date must be set together")
			}
			return ScopeRng(ScopeRange{Gte: begin, Lte: end}), nil
		}
	}
	if f.Required {
		return ScopeValue{}, fmt.Errorf("scope.%s is required", f.APIKeys[0])
	}
	return ScopeValue{}, nil
}

func scopeValueToPredicates(col string, v ScopeValue, normalize func(string) (string, error)) ([]string, error) {
	if v.isZero() {
		return nil, nil
	}
	norm := func(s string) (string, error) {
		if normalize == nil {
			return s, nil
		}
		return normalize(s)
	}
	switch {
	case v.eq != nil:
		nv, err := norm(*v.eq)
		if err != nil {
			return nil, err
		}
		if strings.TrimSpace(nv) == "" {
			return nil, fmt.Errorf("%s must not be empty", col)
		}
		return []string{fmt.Sprintf("%s = %s", quoteIdent(col), quoteLiteral(nv))}, nil
	case v.in != nil:
		parts := make([]string, 0, len(v.in))
		for _, raw := range v.in {
			raw = strings.TrimSpace(raw)
			if raw == "" || raw == "*" {
				continue
			}
			nv, err := norm(raw)
			if err != nil {
				return nil, err
			}
			parts = append(parts, quoteLiteral(nv))
		}
		if len(parts) == 0 {
			return nil, fmt.Errorf("%s must list explicit values (\"*\" / empty is not supported)", col)
		}
		return []string{fmt.Sprintf("%s IN (%s)", quoteIdent(col), strings.Join(parts, ", "))}, nil
	case v.rng != nil:
		r := v.rng
		var preds []string
		add := func(op, raw string) error {
			if raw == "" {
				return nil
			}
			nv, err := norm(raw)
			if err != nil {
				return err
			}
			preds = append(preds, fmt.Sprintf("%s %s %s", quoteIdent(col), op, quoteLiteral(nv)))
			return nil
		}
		if err := add(">=", r.Gte); err != nil {
			return nil, err
		}
		if err := add(">", r.Gt); err != nil {
			return nil, err
		}
		if err := add("<=", r.Lte); err != nil {
			return nil, err
		}
		if err := add("<", r.Lt); err != nil {
			return nil, err
		}
		if len(preds) == 0 {
			return nil, fmt.Errorf("%s range requires at least one bound", col)
		}
		return preds, nil
	default:
		return nil, fmt.Errorf("invalid scope value for %s", col)
	}
}