package catalogapp

import (
	"fmt"
	"sort"
	"strings"
	"unicode"
)

type catalogCheckTokenKind byte

const (
	catalogCheckTokenWord catalogCheckTokenKind = iota + 1
	catalogCheckTokenString
	catalogCheckTokenOpen
	catalogCheckTokenClose
	catalogCheckTokenEqual
)

type catalogCheckToken struct {
	kind  catalogCheckTokenKind
	value string
}

type catalogCheckNode struct {
	op       string
	value    string
	children []catalogCheckNode
}

type catalogCheckParser struct {
	tokens []catalogCheckToken
	index  int
}

// canonicalCatalogCheckClause parses the deliberately small SQL grammar used
// by the two admission constraints. Comparing a canonical boolean tree avoids
// trusting a same-name permissive CHECK while tolerating harmless TiDB/MySQL
// whitespace, quoting, charset-introducer, parenthesis, and term-order output.
func canonicalCatalogCheckClause(clause string) (string, error) {
	tokens, err := tokenizeCatalogCheckClause(clause)
	if err != nil {
		return "", err
	}
	parser := catalogCheckParser{tokens: tokens}
	node, err := parser.parseOr()
	if err != nil {
		return "", err
	}
	if parser.index != len(tokens) {
		return "", fmt.Errorf("unexpected token %q", tokens[parser.index].value)
	}
	return canonicalCatalogCheckNode(node), nil
}

func tokenizeCatalogCheckClause(clause string) ([]catalogCheckToken, error) {
	tokens := make([]catalogCheckToken, 0, 32)
	for index := 0; index < len(clause); {
		character := clause[index]
		if unicode.IsSpace(rune(character)) {
			index++
			continue
		}
		switch character {
		case '(':
			tokens = append(tokens, catalogCheckToken{kind: catalogCheckTokenOpen, value: "("})
			index++
		case ')':
			tokens = append(tokens, catalogCheckToken{kind: catalogCheckTokenClose, value: ")"})
			index++
		case '=':
			tokens = append(tokens, catalogCheckToken{kind: catalogCheckTokenEqual, value: "="})
			index++
		case '`':
			start := index + 1
			index = start
			for index < len(clause) && clause[index] != '`' {
				index++
			}
			if index == len(clause) || index == start {
				return nil, fmt.Errorf("invalid quoted identifier")
			}
			tokens = append(tokens, catalogCheckToken{kind: catalogCheckTokenWord, value: strings.ToLower(clause[start:index])})
			index++
		case '\'':
			value, next, err := readCatalogCheckString(clause, index)
			if err != nil {
				return nil, err
			}
			tokens = append(tokens, catalogCheckToken{kind: catalogCheckTokenString, value: value})
			index = next
		default:
			if !isCatalogCheckWordByte(character) {
				return nil, fmt.Errorf("unsupported CHECK character %q", character)
			}
			start := index
			for index < len(clause) && isCatalogCheckWordByte(clause[index]) {
				index++
			}
			tokens = append(tokens, catalogCheckToken{kind: catalogCheckTokenWord, value: strings.ToLower(clause[start:index])})
		}
	}
	if len(tokens) == 0 {
		return nil, fmt.Errorf("empty CHECK clause")
	}
	return tokens, nil
}

func readCatalogCheckString(clause string, start int) (string, int, error) {
	var value strings.Builder
	for index := start + 1; index < len(clause); index++ {
		if clause[index] != '\'' {
			value.WriteByte(clause[index])
			continue
		}
		if index+1 < len(clause) && clause[index+1] == '\'' {
			value.WriteByte('\'')
			index++
			continue
		}
		return value.String(), index + 1, nil
	}
	return "", 0, fmt.Errorf("unterminated CHECK string")
}

func isCatalogCheckWordByte(value byte) bool {
	return value >= 'a' && value <= 'z' ||
		value >= 'A' && value <= 'Z' ||
		value >= '0' && value <= '9' || value == '_'
}

func (parser *catalogCheckParser) parseOr() (catalogCheckNode, error) {
	left, err := parser.parseAnd()
	if err != nil {
		return catalogCheckNode{}, err
	}
	children := []catalogCheckNode{left}
	for parser.consumeWord("or") {
		right, err := parser.parseAnd()
		if err != nil {
			return catalogCheckNode{}, err
		}
		children = append(children, right)
	}
	if len(children) == 1 {
		return left, nil
	}
	return catalogCheckNode{op: "or", children: children}, nil
}

func (parser *catalogCheckParser) parseAnd() (catalogCheckNode, error) {
	left, err := parser.parsePrimary()
	if err != nil {
		return catalogCheckNode{}, err
	}
	children := []catalogCheckNode{left}
	for parser.consumeWord("and") {
		right, err := parser.parsePrimary()
		if err != nil {
			return catalogCheckNode{}, err
		}
		children = append(children, right)
	}
	if len(children) == 1 {
		return left, nil
	}
	return catalogCheckNode{op: "and", children: children}, nil
}

func (parser *catalogCheckParser) parsePrimary() (catalogCheckNode, error) {
	if parser.consume(catalogCheckTokenOpen) {
		node, err := parser.parseOr()
		if err != nil {
			return catalogCheckNode{}, err
		}
		if !parser.consume(catalogCheckTokenClose) {
			return catalogCheckNode{}, fmt.Errorf("missing closing parenthesis")
		}
		return node, nil
	}
	identifier, err := parser.take(catalogCheckTokenWord)
	if err != nil {
		return catalogCheckNode{}, fmt.Errorf("CHECK predicate identifier: %w", err)
	}
	if parser.consume(catalogCheckTokenEqual) {
		// MySQL may render a string literal with a charset introducer.
		if parser.index < len(parser.tokens) && parser.tokens[parser.index].kind == catalogCheckTokenWord &&
			strings.HasPrefix(parser.tokens[parser.index].value, "_") {
			parser.index++
		}
		literal, err := parser.take(catalogCheckTokenString)
		if err != nil {
			return catalogCheckNode{}, fmt.Errorf("CHECK equality literal: %w", err)
		}
		return catalogCheckNode{op: "eq", value: identifier + "=" + literal}, nil
	}
	if !parser.consumeWord("is") {
		return catalogCheckNode{}, fmt.Errorf("CHECK predicate for %s must use equality or IS NULL", identifier)
	}
	not := parser.consumeWord("not")
	if !parser.consumeWord("null") {
		return catalogCheckNode{}, fmt.Errorf("CHECK predicate for %s has invalid IS expression", identifier)
	}
	op := "null"
	if not {
		op = "notnull"
	}
	return catalogCheckNode{op: op, value: identifier}, nil
}

func (parser *catalogCheckParser) consume(kind catalogCheckTokenKind) bool {
	if parser.index >= len(parser.tokens) || parser.tokens[parser.index].kind != kind {
		return false
	}
	parser.index++
	return true
}

func (parser *catalogCheckParser) consumeWord(word string) bool {
	if parser.index >= len(parser.tokens) || parser.tokens[parser.index].kind != catalogCheckTokenWord ||
		parser.tokens[parser.index].value != word {
		return false
	}
	parser.index++
	return true
}

func (parser *catalogCheckParser) take(kind catalogCheckTokenKind) (string, error) {
	if parser.index >= len(parser.tokens) || parser.tokens[parser.index].kind != kind {
		return "", fmt.Errorf("unexpected or absent token")
	}
	value := parser.tokens[parser.index].value
	parser.index++
	return value, nil
}

func canonicalCatalogCheckNode(node catalogCheckNode) string {
	if node.op != "and" && node.op != "or" {
		return node.op + "(" + node.value + ")"
	}
	children := make([]string, 0, len(node.children))
	for _, child := range node.children {
		if child.op == node.op {
			for _, grandchild := range child.children {
				children = append(children, canonicalCatalogCheckNode(grandchild))
			}
			continue
		}
		children = append(children, canonicalCatalogCheckNode(child))
	}
	sort.Strings(children)
	return node.op + "(" + strings.Join(children, ",") + ")"
}