myxce/src/frontend/apiserver/store/query.go

package store

import (
	"apiserver/utils"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"sort"
	"strings"
)

const PER_PAGE_LIMIT = 100

// ExecuteQuery runs the full filter → sort → paginate → select pipeline.
func ExecuteQuery(rows []utils.OrderMap, opts *utils.QueryOptions, fields []string) ([]utils.OrderMap, string) {
	if opts == nil {
		return SelectFields(rows, fields), ""
	}

	// 1. Filter — both the flat AND-conditions AND the compound (per-field OR,
	// array-aware) conditions. bug-1490-adjacent: CompConditions used to be
	// silently dropped here, so GET /xce/v1/jobs/:jobids returned the whole
	// job table (the mcp scheduler_get_job tool never saw the requested job).
	if len(opts.Conditions) > 0 {
		rows = FilterRows(rows, opts.Conditions)
	}
	if len(opts.CompConditions) > 0 {
		rows = FilterCompConditions(rows, opts.CompConditions)
	}

	// 2. Sort
	if len(opts.Orders) > 0 {
		SortRows(rows, opts.Orders)
	}

	// 3. Paginate
	limit := opts.Limit
	if limit <= 0 {
		limit = PER_PAGE_LIMIT
	}
	rows, nextToken := PaginateRows(rows, limit, opts.Bookmark, opts.Orders)

	// 4. Select fields
	rows = SelectFields(rows, fields)

	return rows, nextToken
}

// MatchesCondition evaluates a single filter condition against a row.
func MatchesCondition(row utils.OrderMap, cond utils.Condition) bool {
	fieldVal := row.Data[cond.Column]

	switch cond.Operator {
	case utils.OpEQ:
		return compareValues(fieldVal, cond.Value.Value) == 0
	case utils.OpNE:
		return compareValues(fieldVal, cond.Value.Value) != 0
	case utils.OpLT:
		return compareValues(fieldVal, cond.Value.Value) < 0
	case utils.OpLTE:
		return compareValues(fieldVal, cond.Value.Value) <= 0
	case utils.OpGT:
		return compareValues(fieldVal, cond.Value.Value) > 0
	case utils.OpGTE:
		return compareValues(fieldVal, cond.Value.Value) >= 0
	case utils.OpIN:
		return valueInList(fieldVal, cond.Value.Value)
	case utils.OpNOTIN:
		return !valueInList(fieldVal, cond.Value.Value)
	case utils.OpLIKE:
		return matchLike(fieldVal, cond.Value.Value)
	case utils.OpNOTLIKE:
		return !matchLike(fieldVal, cond.Value.Value)
	case utils.OpBETWEEN:
		return valueBetween(fieldVal, cond.Value.Value)
	case utils.OpISNULL:
		return fieldVal == nil
	case utils.OpNOTNULL:
		return fieldVal != nil
	}
	return false
}

// MatchesBitmaskCondition handles bitmask fields (job_status, host_status).
// For bitmask EQ: (val & mask) != 0 means the status is set.
// For bitmask NE: (val & mask) == 0 means the status is NOT set.
// null/notnull are handled on the raw field value (bug-1504: job_status
// notnull previously fell through to `return false`, so `jobs list -a`
// filtered out every job).
func MatchesBitmaskCondition(row utils.OrderMap, field string, bitmask int64, op utils.OperatorType) bool {
	fieldVal := row.Data[field]
	switch op {
	case utils.OpISNULL:
		return fieldVal == nil
	case utils.OpNOTNULL:
		return fieldVal != nil
	}
	val := toInt64(fieldVal)
	result := val & bitmask
	switch op {
	case utils.OpNE: // bitmask "eq" is stored as OpNE (val & mask != 0)
		return result != 0
	case utils.OpEQ: // bitmask "ne" is stored as OpEQ (val & mask == 0)
		return result == 0
	}
	return false
}

// IsBitmaskField returns true for fields that use bitmask-based filtering.
func IsBitmaskField(field string) bool {
	return field == "job_status" || field == "host_status"
}

// rowMatchesCondition evaluates a single condition against a row, dispatching
// bitmask fields (job_status/host_status) to MatchesBitmaskCondition and all
// others to MatchesCondition.
func rowMatchesCondition(row utils.OrderMap, cond utils.Condition) bool {
	if IsBitmaskField(cond.Column) {
		bitmask, ok := cond.Value.Value.(int64)
		if !ok {
			return false
		}
		return MatchesBitmaskCondition(row, cond.Column, bitmask, cond.Operator)
	}
	return MatchesCondition(row, cond)
}

// FilterRows applies all conditions (AND logic) to rows.
func FilterRows(rows []utils.OrderMap, conditions []utils.Condition) []utils.OrderMap {
	result := make([]utils.OrderMap, 0, len(rows))
	for _, row := range rows {
		match := true
		for _, cond := range conditions {
			if !rowMatchesCondition(row, cond) {
				match = false
				break
			}
		}
		if match {
			result = append(result, row)
		}
	}
	return result
}

// FilterCompConditions applies compound conditions — the per-field OR structure
// built by ConvertToNewFilterArray for array-aware queries (e.g. job_id with
// array_idx sub-ranges in GET /xce/v1/jobs/:jobids, or job_name filters in
// GET /xce/v1/jobs). A row must match EVERY field key present (AND across
// fields); within a field, the row matches if ANY of its ConditionArrays
// matches. A ConditionArray matches when its MainCondition holds AND either it
// has no SubORConditions or any sub-condition holds.
func FilterCompConditions(rows []utils.OrderMap, comp utils.CompCondition) []utils.OrderMap {
	if len(comp) == 0 {
		return rows
	}
	result := make([]utils.OrderMap, 0, len(rows))
	for _, row := range rows {
		match := true
		for _, arrays := range comp {
			fieldMatch := false
			for _, arr := range arrays {
				if !rowMatchesCondition(row, arr.MainCondition) {
					continue
				}
				if len(arr.SubORConditions) == 0 {
					fieldMatch = true
					break
				}
				for _, sub := range arr.SubORConditions {
					if rowMatchesCondition(row, sub) {
						fieldMatch = true
						break
					}
				}
				if fieldMatch {
					break
				}
			}
			if !fieldMatch {
				match = false
				break
			}
		}
		if match {
			result = append(result, row)
		}
	}
	return result
}

// SortRows sorts by multiple fields with asc/desc.
func SortRows(rows []utils.OrderMap, orders []utils.SortOrder) {
	sort.SliceStable(rows, func(i, j int) bool {
		for _, ord := range orders {
			cmp := compareValues(rows[i].Data[ord.Column], rows[j].Data[ord.Column])
			if cmp == 0 {
				continue
			}
			if ord.Ascending {
				return cmp < 0
			}
			return cmp > 0
		}
		return false
	})
}

// PaginateRows applies limit + page_token (bookmark-based pagination).
func PaginateRows(rows []utils.OrderMap, limit int, pageToken string, orders []utils.SortOrder) ([]utils.OrderMap, string) {
	startIdx := 0
	if pageToken != "" {
		startIdx = findPageTokenIndex(rows, pageToken, orders)
	}
	if limit <= 0 {
		limit = PER_PAGE_LIMIT
	}
	endIdx := startIdx + limit
	if endIdx > len(rows) {
		endIdx = len(rows)
	}
	if startIdx >= len(rows) {
		return []utils.OrderMap{}, ""
	}

	result := rows[startIdx:endIdx]
	nextToken := ""
	if endIdx < len(rows) {
		nextToken = encodePageToken(result[len(result)-1], orders)
	}
	return result, nextToken
}

// SelectFields projects only requested fields from rows.
func SelectFields(rows []utils.OrderMap, fields []string) []utils.OrderMap {
	if len(fields) == 0 {
		return rows
	}
	result := make([]utils.OrderMap, len(rows))
	for i, row := range rows {
		newData := make(map[string]interface{})
		for _, f := range fields {
			if v, ok := row.Data[f]; ok {
				newData[f] = v
			}
		}
		result[i] = utils.OrderMap{Fields: fields, Data: newData}
	}
	return result
}

// --- Helper functions ---

// compareValues does type-aware comparison. Returns -1, 0, or 1.
func compareValues(a, b interface{}) int {
	if a == nil && b == nil {
		return 0
	}
	if a == nil {
		return -1
	}
	if b == nil {
		return 1
	}

	// Try int64 comparison
	aInt, aIsInt := toInt64Safe(a)
	bInt, bIsInt := toInt64Safe(b)
	if aIsInt && bIsInt {
		if aInt < bInt {
			return -1
		}
		if aInt > bInt {
			return 1
		}
		return 0
	}

	// Try float64 comparison
	aFloat, aIsFloat := toFloat64Safe(a)
	bFloat, bIsFloat := toFloat64Safe(b)
	if aIsFloat && bIsFloat {
		if aFloat < bFloat {
			return -1
		}
		if aFloat > bFloat {
			return 1
		}
		return 0
	}

	// String comparison
	aStr := fmt.Sprintf("%v", a)
	bStr := fmt.Sprintf("%v", b)
	if aStr < bStr {
		return -1
	}
	if aStr > bStr {
		return 1
	}
	return 0
}

func toInt64(v interface{}) int64 {
	val, _ := toInt64Safe(v)
	return val
}

func toInt64Safe(v interface{}) (int64, bool) {
	switch val := v.(type) {
	case int64:
		return val, true
	case int:
		return int64(val), true
	case int32:
		return int64(val), true
	case uint64:
		return int64(val), true
	case float64:
		return int64(val), true
	}
	return 0, false
}

func toFloat64Safe(v interface{}) (float64, bool) {
	switch val := v.(type) {
	case float64:
		return val, true
	case float32:
		return float64(val), true
	case int64:
		return float64(val), true
	case int:
		return float64(val), true
	}
	return 0, false
}

// valueInList checks if val is in a list of Values.
func valueInList(val interface{}, list interface{}) bool {
	switch l := list.(type) {
	case []utils.Value:
		for _, item := range l {
			if compareValues(val, item.Value) == 0 {
				return true
			}
		}
	case []interface{}:
		for _, item := range l {
			if compareValues(val, item) == 0 {
				return true
			}
		}
	}
	return false
}

// matchLike implements SQL LIKE with Go string matching.
func matchLike(fieldVal, pattern interface{}) bool {
	fieldStr := fmt.Sprintf("%v", fieldVal)
	patternStr := fmt.Sprintf("%v", pattern)

	// If pattern contains SQL wildcards, translate them
	if strings.Contains(patternStr, "%") {
		// Simple wildcard matching
		if strings.HasPrefix(patternStr, "%") && strings.HasSuffix(patternStr, "%") {
			return strings.Contains(strings.ToLower(fieldStr), strings.ToLower(strings.Trim(patternStr, "%")))
		}
		if strings.HasPrefix(patternStr, "%") {
			return strings.HasSuffix(strings.ToLower(fieldStr), strings.ToLower(strings.TrimPrefix(patternStr, "%")))
		}
		if strings.HasSuffix(patternStr, "%") {
			return strings.HasPrefix(strings.ToLower(fieldStr), strings.ToLower(strings.TrimSuffix(patternStr, "%")))
		}
	}

	// Default: substring match (contains)
	return strings.Contains(strings.ToLower(fieldStr), strings.ToLower(patternStr))
}

// valueBetween checks if val is between the first two elements of a Value array.
func valueBetween(val interface{}, rangeVal interface{}) bool {
	switch r := rangeVal.(type) {
	case []utils.Value:
		if len(r) >= 2 {
			return compareValues(val, r[0].Value) >= 0 && compareValues(val, r[1].Value) <= 0
		}
	}
	return false
}

// Page token encoding/decoding using base64 JSON.
func encodePageToken(lastRow utils.OrderMap, orders []utils.SortOrder) string {
	tokenData := make(map[string]interface{})
	for _, ord := range orders {
		if v, ok := lastRow.Data[ord.Column]; ok {
			tokenData[ord.Column] = v
		}
	}
	jsonBytes, err := json.Marshal(tokenData)
	if err != nil {
		return ""
	}
	return base64.URLEncoding.EncodeToString(jsonBytes)
}

func findPageTokenIndex(rows []utils.OrderMap, pageToken string, orders []utils.SortOrder) int {
	jsonBytes, err := base64.URLEncoding.DecodeString(pageToken)
	if err != nil {
		return 0
	}
	var tokenData map[string]interface{}
	if err := json.Unmarshal(jsonBytes, &tokenData); err != nil {
		return 0
	}

	for i, row := range rows {
		match := true
		for _, ord := range orders {
			tokenVal, ok := tokenData[ord.Column]
			if !ok {
				continue
			}
			if compareValues(row.Data[ord.Column], tokenVal) != 0 {
				match = false
				break
			}
		}
		if match {
			return i + 1 // Start after the matched row
		}
	}
	return 0
}