package service

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"math"
	"sort"
	"strconv"
	"strings"
	"unicode"
	"unicode/utf8"
)

const (
	maxBYOCStatementsBinaryPlanBytes     = 2 << 20
	maxBYOCStatementsBinaryPlanTextBytes = 4 << 20
	maxBYOCStatementsBinaryPlanCellBytes = 256 << 10
	maxBYOCStatementsBinaryPlanNodes     = 4096
	maxBYOCStatementsBinaryPlanDepth     = 256
	byocStatementsPlanDiscardedText      = "(plan discarded because too long)"
)

type byocStatementsBinaryPlan struct {
	DiscardedDueToTooLong bool                      `json:"discardedDueToTooLong"`
	Main                  *byocStatementsPlanNode   `json:"main"`
	CTEs                  []*byocStatementsPlanNode `json:"ctes"`
	WithRuntimeStats      bool                      `json:"withRuntimeStats"`
}

type byocStatementsPlanNode struct {
	Name              string                    `json:"name"`
	Labels            []string                  `json:"labels"`
	EstRows           json.RawMessage           `json:"estRows"`
	Cost              json.RawMessage           `json:"cost"`
	ActRows           json.RawMessage           `json:"actRows"`
	TaskType          string                    `json:"taskType"`
	StoreType         string                    `json:"storeType"`
	AccessObjects     []json.RawMessage         `json:"accessObjects"`
	OperatorInfo      string                    `json:"operatorInfo"`
	RootBasicExecInfo json.RawMessage           `json:"rootBasicExecInfo"`
	RootGroupExecInfo json.RawMessage           `json:"rootGroupExecInfo"`
	CopExecInfo       json.RawMessage           `json:"copExecInfo"`
	MemoryBytes       json.RawMessage           `json:"memoryBytes"`
	DiskBytes         json.RawMessage           `json:"diskBytes"`
	Children          []*byocStatementsPlanNode `json:"children"`
}

// addBYOCStatementsBinaryPlanText adds a human-readable execution-plan table
// to an O11y Statements plan-detail response. O11y owns the binary-plan JSON
// schema, so this adapter is deliberately best effort: existing text wins and
// every decoding, size, or formatting failure leaves the original response
// byte-for-byte unchanged.
func addBYOCStatementsBinaryPlanText(body []byte) []byte {
	var response map[string]json.RawMessage
	if err := decodeBYOCStatementsJSON(body, &response); err != nil || response == nil {
		return body
	}

	if raw, ok := response["binary_plan_text"]; ok {
		var existing string
		if json.Unmarshal(raw, &existing) == nil && existing != "" {
			return body
		}
	}

	var binaryPlan string
	if raw, ok := response["binary_plan"]; !ok || json.Unmarshal(raw, &binaryPlan) != nil || binaryPlan == "" {
		return body
	}
	text, err := decodeBYOCStatementsBinaryPlan(binaryPlan)
	if err != nil || text == "" || len(text) > maxBYOCStatementsBinaryPlanTextBytes {
		return body
	}
	encodedText, err := json.Marshal(text)
	if err != nil {
		return body
	}
	response["binary_plan_text"] = encodedText
	result, err := json.Marshal(response)
	if err != nil {
		return body
	}
	return result
}

func decodeBYOCStatementsBinaryPlan(raw string) (string, error) {
	if len(raw) == 0 || len(raw) > maxBYOCStatementsBinaryPlanBytes {
		return "", fmt.Errorf("binary plan exceeds the allowed size")
	}
	if err := validateBYOCStatementsJSONDepth([]byte(raw), maxBYOCStatementsBinaryPlanDepth); err != nil {
		return "", err
	}

	var plan byocStatementsBinaryPlan
	if err := decodeBYOCStatementsJSON([]byte(raw), &plan); err != nil {
		return "", fmt.Errorf("decode binary plan JSON: %w", err)
	}
	if plan.DiscardedDueToTooLong {
		return byocStatementsPlanDiscardedText, nil
	}
	if plan.Main == nil {
		return "", fmt.Errorf("binary plan main tree is required")
	}

	headings := []string{"id", "estRows", "estCost", "task", "access object", "operator info"}
	if plan.WithRuntimeStats {
		headings = []string{"id", "estRows", "estCost", "actRows", "task", "access object", "execution info", "operator info", "memory", "disk"}
	}
	rows := make([][]string, 0, 64)
	nodeCount := 0
	if err := appendBYOCStatementsPlanRows(plan.Main, "", true, plan.WithRuntimeStats, 0, &nodeCount, &rows); err != nil {
		return "", err
	}
	for _, cte := range plan.CTEs {
		if err := appendBYOCStatementsPlanRows(cte, "", true, plan.WithRuntimeStats, 0, &nodeCount, &rows); err != nil {
			return "", err
		}
	}
	if len(rows) == 0 {
		return "", fmt.Errorf("binary plan contains no operators")
	}
	return formatBYOCStatementsPlanTable(headings, rows)
}

func appendBYOCStatementsPlanRows(
	node *byocStatementsPlanNode,
	indent string,
	isLast bool,
	withRuntimeStats bool,
	depth int,
	nodeCount *int,
	rows *[][]string,
) error {
	if node == nil || strings.TrimSpace(node.Name) == "" {
		return fmt.Errorf("binary plan operator name is required")
	}
	if depth > maxBYOCStatementsBinaryPlanDepth {
		return fmt.Errorf("binary plan tree is too deep")
	}
	(*nodeCount)++
	if *nodeCount > maxBYOCStatementsBinaryPlanNodes {
		return fmt.Errorf("binary plan contains too many operators")
	}

	identifier := prettyBYOCStatementsPlanIdentifier(
		sanitizeBYOCStatementsPlanCell(node.Name)+formatBYOCStatementsPlanLabels(node.Labels),
		indent,
		isLast,
	)
	estRows := formatBYOCStatementsPlanFloat(node.EstRows)
	estCost := formatBYOCStatementsPlanFloat(node.Cost)
	task := formatBYOCStatementsPlanTask(node.TaskType, node.StoreType)
	accessObjects, err := formatBYOCStatementsPlanAccessObjects(node.AccessObjects)
	if err != nil {
		return err
	}
	operatorInfo := sanitizeBYOCStatementsPlanCell(node.OperatorInfo)

	row := []string{identifier, estRows, estCost}
	if withRuntimeStats {
		executionInfo, err := formatBYOCStatementsPlanExecutionInfo(node)
		if err != nil {
			return err
		}
		row = append(row,
			formatBYOCStatementsPlanScalar(node.ActRows, "0"),
			task,
			accessObjects,
			executionInfo,
			operatorInfo,
			formatBYOCStatementsPlanBytes(node.MemoryBytes),
			formatBYOCStatementsPlanBytes(node.DiskBytes),
		)
	} else {
		row = append(row, task, accessObjects, operatorInfo)
	}
	for _, cell := range row {
		if len(cell) > maxBYOCStatementsBinaryPlanCellBytes {
			return fmt.Errorf("binary plan field exceeds the allowed size")
		}
	}
	*rows = append(*rows, row)

	childIndent := byocStatementsPlanChildIndent(indent, isLast)
	for i, child := range node.Children {
		if err := appendBYOCStatementsPlanRows(
			child,
			childIndent,
			i == len(node.Children)-1,
			withRuntimeStats,
			depth+1,
			nodeCount,
			rows,
		); err != nil {
			return err
		}
	}
	return nil
}

func formatBYOCStatementsPlanExecutionInfo(node *byocStatementsPlanNode) (string, error) {
	values := []json.RawMessage{node.RootBasicExecInfo, node.RootGroupExecInfo, node.CopExecInfo}
	parts := make([]string, 0, len(values))
	for _, raw := range values {
		part, err := formatBYOCStatementsPlanJSONValue(raw, true)
		if err != nil {
			return "", err
		}
		if part != "" {
			parts = append(parts, part)
		}
	}
	return strings.Join(parts, ", "), nil
}

func formatBYOCStatementsPlanAccessObjects(objects []json.RawMessage) (string, error) {
	parts := make([]string, 0, len(objects))
	for _, raw := range objects {
		var object map[string]json.RawMessage
		if err := decodeBYOCStatementsJSON(raw, &object); err != nil || object == nil {
			return "", fmt.Errorf("decode binary plan access object")
		}
		if scanRaw, ok := object["scanObject"]; ok {
			formatted, err := formatBYOCStatementsScanObject(scanRaw)
			if err != nil {
				return "", err
			}
			if formatted != "" {
				parts = append(parts, formatted)
				continue
			}
		}
		formatted, err := formatBYOCStatementsPlanJSONValue(raw, true)
		if err != nil {
			return "", err
		}
		if formatted != "" {
			parts = append(parts, formatted)
		}
	}
	return strings.Join(parts, ", "), nil
}

func formatBYOCStatementsScanObject(raw json.RawMessage) (string, error) {
	var scan struct {
		Database   string   `json:"database"`
		Table      string   `json:"table"`
		Partitions []string `json:"partitions"`
		Indexes    []struct {
			Name             string   `json:"name"`
			Cols             []string `json:"cols"`
			IsClusteredIndex bool     `json:"isClusteredIndex"`
		} `json:"indexes"`
	}
	if err := decodeBYOCStatementsJSON(raw, &scan); err != nil {
		return "", fmt.Errorf("decode binary plan scan object: %w", err)
	}
	parts := make([]string, 0, 2+len(scan.Indexes))
	table := sanitizeBYOCStatementsPlanCell(scan.Table)
	database := sanitizeBYOCStatementsPlanCell(scan.Database)
	if table != "" {
		if database != "" {
			table = database + "." + table
		}
		parts = append(parts, "table:"+table)
	}
	if len(scan.Partitions) != 0 {
		partitions := make([]string, 0, len(scan.Partitions))
		for _, partition := range scan.Partitions {
			partitions = append(partitions, sanitizeBYOCStatementsPlanCell(partition))
		}
		parts = append(parts, "partition:"+strings.Join(partitions, ","))
	}
	for _, index := range scan.Indexes {
		kind := "index:"
		if index.IsClusteredIndex {
			kind = "clustered index:"
		}
		columns := make([]string, 0, len(index.Cols))
		for _, column := range index.Cols {
			columns = append(columns, sanitizeBYOCStatementsPlanCell(column))
		}
		value := kind + sanitizeBYOCStatementsPlanCell(index.Name)
		if len(columns) != 0 {
			value += "(" + strings.Join(columns, ", ") + ")"
		}
		parts = append(parts, value)
	}
	return strings.Join(parts, ", "), nil
}

func formatBYOCStatementsPlanJSONValue(raw json.RawMessage, topLevel bool) (string, error) {
	if len(bytes.TrimSpace(raw)) == 0 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
		return "", nil
	}
	var value any
	if err := decodeBYOCStatementsJSON(raw, &value); err != nil {
		return "", fmt.Errorf("decode binary plan field: %w", err)
	}
	return renderBYOCStatementsPlanJSONValue(value, topLevel), nil
}

func renderBYOCStatementsPlanJSONValue(value any, topLevel bool) string {
	switch typed := value.(type) {
	case nil:
		return ""
	case string:
		return sanitizeBYOCStatementsPlanCell(typed)
	case json.Number:
		return typed.String()
	case bool:
		return strconv.FormatBool(typed)
	case []any:
		parts := make([]string, 0, len(typed))
		for _, item := range typed {
			if part := renderBYOCStatementsPlanJSONValue(item, false); part != "" {
				parts = append(parts, part)
			}
		}
		if topLevel {
			return strings.Join(parts, ", ")
		}
		return "[" + strings.Join(parts, ", ") + "]"
	case map[string]any:
		keys := make([]string, 0, len(typed))
		for key := range typed {
			keys = append(keys, key)
		}
		sort.Strings(keys)
		parts := make([]string, 0, len(keys))
		for _, key := range keys {
			parts = append(parts, sanitizeBYOCStatementsPlanCell(key)+":"+renderBYOCStatementsPlanJSONValue(typed[key], false))
		}
		if topLevel {
			return strings.Join(parts, ", ")
		}
		return "{" + strings.Join(parts, ", ") + "}"
	default:
		return sanitizeBYOCStatementsPlanCell(fmt.Sprint(typed))
	}
}

func formatBYOCStatementsPlanFloat(raw json.RawMessage) string {
	value := formatBYOCStatementsPlanScalar(raw, "0")
	parsed, err := strconv.ParseFloat(value, 64)
	if err != nil || math.IsNaN(parsed) || math.IsInf(parsed, 0) {
		return "0.00"
	}
	return strconv.FormatFloat(parsed, 'f', 2, 64)
}

func formatBYOCStatementsPlanScalar(raw json.RawMessage, fallback string) string {
	if len(bytes.TrimSpace(raw)) == 0 || bytes.Equal(bytes.TrimSpace(raw), []byte("null")) {
		return fallback
	}
	var value any
	if err := decodeBYOCStatementsJSON(raw, &value); err != nil {
		return fallback
	}
	switch typed := value.(type) {
	case string:
		return sanitizeBYOCStatementsPlanCell(typed)
	case json.Number:
		return typed.String()
	default:
		return fallback
	}
}

func formatBYOCStatementsPlanBytes(raw json.RawMessage) string {
	value := formatBYOCStatementsPlanScalar(raw, "N/A")
	if value == "" || value == "-1" || strings.EqualFold(value, "N/A") {
		return "N/A"
	}
	bytesValue, err := strconv.ParseInt(value, 10, 64)
	if err != nil || bytesValue < 0 {
		return value
	}
	return formatBYOCStatementsByteCount(bytesValue)
}

func formatBYOCStatementsByteCount(value int64) string {
	const (
		kib = int64(1 << 10)
		mib = int64(1 << 20)
		gib = int64(1 << 30)
	)
	if value <= kib {
		return fmt.Sprintf("%d Bytes", value)
	}
	unit, unitName := kib, "KB"
	if value > gib {
		unit, unitName = gib, "GB"
	} else if value > mib {
		unit, unitName = mib, "MB"
	}
	decimal := 1
	result := float64(value) / float64(unit)
	if value%unit == 0 {
		decimal = 0
	} else if result < 10 {
		decimal = 2
	}
	return strconv.FormatFloat(result, 'f', decimal, 64) + " " + unitName
}

func formatBYOCStatementsPlanTask(taskType, storeType string) string {
	taskType = sanitizeBYOCStatementsPlanCell(taskType)
	storeType = sanitizeBYOCStatementsPlanCell(storeType)
	if taskType == "" {
		taskType = "unknown"
	}
	if strings.EqualFold(taskType, "root") || strings.EqualFold(taskType, "unknown") || storeType == "" {
		return taskType
	}
	return taskType + "[" + storeType + "]"
}

func formatBYOCStatementsPlanLabels(labels []string) string {
	var result strings.Builder
	for _, label := range labels {
		switch label {
		case "empty":
		case "buildSide":
			result.WriteString("(Build)")
		case "probeSide":
			result.WriteString("(Probe)")
		case "seedPart":
			result.WriteString("(Seed Part)")
		case "recursivePart":
			result.WriteString("(Recursive Part)")
		}
	}
	return result.String()
}

func formatBYOCStatementsPlanTable(headings []string, rows [][]string) (string, error) {
	allRows := make([][]string, 0, len(rows)+1)
	allRows = append(allRows, headings)
	allRows = append(allRows, rows...)
	maxRunes := make([]int, len(headings))
	for _, row := range allRows {
		if len(row) != len(headings) {
			return "", fmt.Errorf("binary plan row has an invalid column count")
		}
		for column, value := range row {
			if width := utf8.RuneCountInString(value); width > maxRunes[column] {
				maxRunes[column] = width
			}
		}
	}

	totalBytes := 1
	for _, row := range allRows {
		rowBytes := 3
		for column, value := range row {
			rowBytes += 3 + len(value) + maxRunes[column] - utf8.RuneCountInString(value)
		}
		if rowBytes > maxBYOCStatementsBinaryPlanTextBytes-totalBytes {
			return "", fmt.Errorf("binary plan text exceeds the allowed size")
		}
		totalBytes += rowBytes
	}

	var result strings.Builder
	result.Grow(totalBytes)
	result.WriteByte('\n')
	for _, row := range allRows {
		for column, value := range row {
			result.WriteString("| ")
			result.WriteString(value)
			result.WriteString(strings.Repeat(" ", maxRunes[column]-utf8.RuneCountInString(value)))
			result.WriteByte(' ')
		}
		result.WriteString("|\n")
	}
	return result.String(), nil
}

func prettyBYOCStatementsPlanIdentifier(identifier, indent string, isLast bool) string {
	if indent == "" {
		return identifier
	}
	characters := []rune(indent)
	for i := len(characters) - 1; i >= 0; i-- {
		if characters[i] == '│' {
			if isLast {
				characters[i] = '└'
			} else {
				characters[i] = '├'
			}
			break
		}
	}
	characters[len(characters)-1] = '─'
	return string(characters) + identifier
}

func byocStatementsPlanChildIndent(indent string, isLast bool) string {
	characters := []rune(indent)
	if isLast {
		for i := len(characters) - 1; i >= 0; i-- {
			if characters[i] == '│' {
				characters[i] = ' '
				break
			}
		}
	}
	return string(append(characters, '│', ' '))
}

func sanitizeBYOCStatementsPlanCell(value string) string {
	return strings.Map(func(character rune) rune {
		if unicode.IsControl(character) {
			return ' '
		}
		return character
	}, value)
}

func validateBYOCStatementsJSONDepth(raw []byte, maxDepth int) error {
	depth := 0
	inString := false
	escaped := false
	for _, character := range raw {
		if inString {
			switch {
			case escaped:
				escaped = false
			case character == '\\':
				escaped = true
			case character == '"':
				inString = false
			}
			continue
		}
		switch character {
		case '"':
			inString = true
		case '{', '[':
			depth++
			if depth > maxDepth {
				return fmt.Errorf("binary plan JSON is too deeply nested")
			}
		case '}', ']':
			depth--
		}
	}
	return nil
}

func decodeBYOCStatementsJSON(raw []byte, target any) error {
	decoder := json.NewDecoder(bytes.NewReader(raw))
	decoder.UseNumber()
	if err := decoder.Decode(target); err != nil {
		return err
	}
	var trailing any
	if err := decoder.Decode(&trailing); err != io.EOF {
		return fmt.Errorf("trailing JSON data")
	}
	return nil
}