package persistence

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/query"
	domainrls "github.com/nexuscrm/kernel/pkg/rls"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/rs/zerolog/log"
)

// RunAnalytics executes an aggregation query
func (r *QueryRepository) RunAnalytics(ctx context.Context, tableSchema *models.ObjectMetadata, q models.AnalyticsQuery, rlsResult *domainrls.Result) (interface{}, error) {
	return r.runAnalytics(ctx, tableSchema, q, rlsResult, nil)
}

// RunAnalyticsWithCorrelatedMemberships executes an aggregation query with
// service-validated correlated membership predicates.
func (r *QueryRepository) RunAnalyticsWithCorrelatedMemberships(ctx context.Context, tableSchema *models.ObjectMetadata, q models.AnalyticsQuery, rlsResult *domainrls.Result, memberships []ports.QueryCorrelatedMembershipPlan) (interface{}, error) {
	return r.runAnalytics(ctx, tableSchema, q, rlsResult, memberships)
}

func (r *QueryRepository) runAnalytics(ctx context.Context, tableSchema *models.ObjectMetadata, q models.AnalyticsQuery, rlsResult *domainrls.Result, memberships []ports.QueryCorrelatedMembershipPlan) (interface{}, error) {
	// Use WithMetadata to enable alias resolution where necessary
	readTable, err := r.resolveReadTable(ctx, tableSchema)
	if err != nil {
		return nil, err
	}
	builder := query.From(readTable).WithMetadata(tableSchema)

	if err := r.applyAnalyticsFilters(ctx, builder, tableSchema, q, rlsResult, readTable, memberships); err != nil {
		return nil, err
	}

	if err := r.applyAnalyticsOperation(builder, tableSchema, q, readTable); err != nil {
		return nil, err
	}

	return r.executeAndFormatAnalytics(ctx, builder, q)
}

func (r *QueryRepository) applyAnalyticsFilters(ctx context.Context, builder *query.Builder, tableSchema *models.ObjectMetadata, q models.AnalyticsQuery, rlsResult *domainrls.Result, readTable string, memberships []ports.QueryCorrelatedMembershipPlan) error {
	if err := validateParentFilterReadTarget(tableSchema, readTable, q.ParentFilterPaths); err != nil {
		return err
	}
	if err := applyParentFilterJoins(builder, tableSchema, readTable, q.ParentFilterPaths, q.ScanAll); err != nil {
		return err
	}

	if !q.ScanAll && schemaHasIsDeleted(tableSchema) {
		builder.ExcludeDeleted()
	}
	builder.SkipID()

	if guard, err := analyticsParentGroupByMatchedParentGuard(tableSchema, readTable, q); err != nil {
		return err
	} else if guard != "" {
		builder.WhereRaw(guard, nil)
	}

	if q.FilterExpr != "" {
		mapper := newFieldMapperForTable(tableSchema, readTable)
		allowMemberPaths := len(q.ParentFilterPaths) > 0
		var unsupportedParentPath string
		if allowMemberPaths {
			mapper = parentFilterFieldMapperForTable(tableSchema, readTable, q.ParentFilterPaths, func(identifier string) {
				if unsupportedParentPath == "" {
					unsupportedParentPath = identifier
				}
			})
		}
		sqlWhere, args, err := formula.ToSQLWithOptions(q.FilterExpr, formula.SQLOptions{
			AllowMemberPaths: allowMemberPaths,
			Mapper:           mapper,
			FunctionLowerer:  parentFilterNullBlankFunctionLowerer(q.ParentFilterPaths),
		})
		if err != nil {
			return apperrors.NewValidationError("filterExpr", fmt.Sprintf("invalid filter expression: %v", err))
		}
		if unsupportedParentPath != "" {
			return apperrors.NewValidationError("filterExpr", fmt.Sprintf("unsupported relationship field filter path %q", unsupportedParentPath))
		}
		if err := validateParentFilterNullBlankSQL(sqlWhere, q.ParentFilterPaths); err != nil {
			return err
		}
		if err := validateParentFilterListSQL(sqlWhere, q.ParentFilterPaths); err != nil {
			return err
		}
		sqlWhere = requireMatchedParentForNullSQL(sqlWhere, q.ParentFilterPaths)
		builder.WhereRaw(sqlWhere, args)
	}
	readRLSResult, err := r.prepareRLSForRead(ctx, rlsResult, q.ScanAll)
	if err != nil {
		return err
	}
	readRLSResult, err = prepareRLSForReadTarget(readRLSResult, tableSchema, readTable)
	if err != nil {
		return err
	}
	if rlsWhere := domainrls.ToSQLFilter(readRLSResult, readTable); rlsWhere != "" {
		rlsWhere = qualifyParentFilterRootRLSFilter(rlsWhere, tableSchema, readTable, q.ParentFilterPaths)
		builder.WhereRaw(rlsWhere, nil)
	}

	if q.Having != "" {
		resolvedHaving, err := analyticsHavingExpression(tableSchema, readTable, q)
		if err != nil {
			return err
		}
		builder.HavingRaw(resolvedHaving, nil)
	}

	if len(q.Criteria) > 0 {
		for _, c := range q.Criteria {
			if isValidFieldName(c.Field) && isValidFilterOp(c.Op) {
				opUpper := strings.ToUpper(strings.TrimSpace(c.Op))
				if parentExpression, ok, err := parentFilterStructuredClauseExpression(tableSchema, readTable, q.ParentFilterPaths, c.Field, "Criteria"); err != nil {
					return err
				} else if ok {
					if err := applyParentFilterStructuredCriterion(builder, parentExpression, opUpper, c.Val, c.Field, "Criteria"); err != nil {
						return err
					}
					continue
				}
				builder.WhereOp(c.Field, c.Op, c.Val)
			}
		}
	}

	if err := r.applyCorrelatedMemberships(ctx, builder, tableSchema, readTable, memberships); err != nil {
		return err
	}

	return nil
}

func (r *QueryRepository) applyAnalyticsOperation(builder *query.Builder, tableSchema *models.ObjectMetadata, q models.AnalyticsQuery, readTable string) error {
	if len(q.Aggregates) > 0 {
		return r.applyMultiAggregateAnalyticsOperation(builder, tableSchema, q, readTable)
	}

	switch q.Operation {
	case OpCount:
		builder.AddSelectRaw(fmt.Sprintf("%s as val", countSelectExpression(tableSchema, readTable)))

	case OpGroupBy:
		agg := countSelectExpression(tableSchema, readTable)
		if q.Field != nil {
			if !isValidFieldName(*q.Field) {
				return apperrors.NewValidationError("field", fmt.Sprintf("invalid aggregation field: %s", *q.Field))
			}
			aggFunc := "SUM"
			if q.AggregateType != "" {
				validAggs := map[string]bool{"SUM": true, "AVG": true, "MIN": true, "MAX": true, "COUNT": true}
				if validAggs[strings.ToUpper(q.AggregateType)] {
					aggFunc = strings.ToUpper(q.AggregateType)
				}
			}
			aggCol := analyticsRootColumnExpression(tableSchema, readTable, *q.Field, len(q.ParentFilterPaths) > 0)
			if aggFunc == "COUNT" {
				agg = fmt.Sprintf("COUNT(%s)", aggCol)
			} else {
				agg = fmt.Sprintf("%s(%s)", aggFunc, aggCol)
			}
		}

		groupByExpr, err := analyticsGroupByExpression(tableSchema, q, readTable)
		if err != nil {
			return err
		}

		builder.AddSelectRaw(fmt.Sprintf("%s as name", groupByExpr))
		builder.AddSelectRaw(fmt.Sprintf("%s as val", agg))
		builder.GroupByRaw(groupByExpr)
		if err := r.applyAnalyticsGroupOrderAndLimit(builder, tableSchema, q, readTable, "val"); err != nil {
			return err
		}

	default: // sum, avg, min, max
		validOps := map[string]bool{
			"SUM": true, "AVG": true, "MIN": true, "MAX": true,
		}
		op := strings.ToUpper(q.Operation)
		if !validOps[op] {
			return apperrors.NewValidationError("operation", fmt.Sprintf("unsupported analytics operation: %s", q.Operation))
		}
		if q.Field == nil || !isValidFieldName(*q.Field) {
			return apperrors.NewValidationError("field", fmt.Sprintf("invalid analytics field: %v", q.Field))
		}
		aggCol := analyticsRootColumnExpression(tableSchema, readTable, *q.Field, len(q.ParentFilterPaths) > 0)
		builder.AddSelectRaw(fmt.Sprintf("%s(%s) as val", op, aggCol))
	}
	return nil
}

func (r *QueryRepository) applyMultiAggregateAnalyticsOperation(builder *query.Builder, tableSchema *models.ObjectMetadata, q models.AnalyticsQuery, readTable string) error {
	if q.Operation == OpGroupBy {
		groupByExpr, err := analyticsGroupByExpression(tableSchema, q, readTable)
		if err != nil {
			return err
		}
		builder.AddSelectRaw(fmt.Sprintf("%s as name", groupByExpr))
		for i, aggregate := range q.Aggregates {
			expr, err := analyticsAggregateExpression(tableSchema, aggregate, readTable, len(q.ParentFilterPaths) > 0)
			if err != nil {
				return err
			}
			builder.AddSelectRaw(fmt.Sprintf("%s as agg%d", expr, i))
		}
		builder.GroupByRaw(groupByExpr)
		return r.applyAnalyticsGroupOrderAndLimit(builder, tableSchema, q, readTable, "agg0")
	}

	for i, aggregate := range q.Aggregates {
		expr, err := analyticsAggregateExpression(tableSchema, aggregate, readTable, len(q.ParentFilterPaths) > 0)
		if err != nil {
			return err
		}
		builder.AddSelectRaw(fmt.Sprintf("%s as agg%d", expr, i))
	}
	return nil
}

func analyticsAggregateExpression(tableSchema *models.ObjectMetadata, aggregate models.AnalyticsAggregate, readTable string, qualifyRoot bool) (string, error) {
	aggFunc := strings.ToUpper(strings.TrimSpace(aggregate.Type))
	validAggs := map[string]bool{"SUM": true, "AVG": true, "MIN": true, "MAX": true, "COUNT": true}
	if !validAggs[aggFunc] {
		return "", apperrors.NewValidationError("aggregate", fmt.Sprintf("unsupported aggregate type: %s", aggregate.Type))
	}
	field := strings.TrimSpace(aggregate.Field)
	if field == "" {
		if aggFunc != "COUNT" {
			return "", apperrors.NewValidationError("field", fmt.Sprintf("%s aggregate requires a field", aggFunc))
		}
		return countSelectExpression(tableSchema, readTable), nil
	}
	if !isValidFieldName(field) {
		return "", apperrors.NewValidationError("field", fmt.Sprintf("invalid aggregation field: %s", field))
	}
	aggCol := analyticsRootColumnExpression(tableSchema, readTable, field, qualifyRoot)
	if aggFunc == "COUNT" {
		return fmt.Sprintf("COUNT(%s)", aggCol), nil
	}
	return fmt.Sprintf("%s(%s)", aggFunc, aggCol), nil
}

func analyticsGroupByExpression(tableSchema *models.ObjectMetadata, q models.AnalyticsQuery, readTable string) (string, error) {
	if q.GroupBy == nil {
		return "", apperrors.NewValidationError("groupBy", fmt.Sprintf("invalid group by field: %v", q.GroupBy))
	}
	if parentExpression, ok, err := parentFilterStructuredClauseExpression(tableSchema, readTable, q.ParentFilterPaths, *q.GroupBy, "groupBy"); err != nil {
		return "", err
	} else if ok {
		return parentExpression, nil
	}
	if !isValidFieldName(*q.GroupBy) {
		return "", apperrors.NewValidationError("groupBy", fmt.Sprintf("invalid group by field: %s", *q.GroupBy))
	}
	return analyticsRootColumnExpression(tableSchema, readTable, *q.GroupBy, len(q.ParentFilterPaths) > 0), nil
}

func analyticsRootColumnExpression(schema *models.ObjectMetadata, readTable string, field string, qualify bool) string {
	col := readTargetColumnName(schema, readTable, field)
	if !qualify {
		return fmt.Sprintf("`%s`", col)
	}
	return fmt.Sprintf("`%s`.`%s`", readTable, col)
}

func analyticsParentGroupByMatchedParentGuard(tableSchema *models.ObjectMetadata, readTable string, q models.AnalyticsQuery) (string, error) {
	if q.GroupBy == nil {
		return "", nil
	}
	index, ok := parentFilterPathIndex(q.ParentFilterPaths, *q.GroupBy)
	if !ok {
		if _, _, err := parentFilterStructuredClauseExpression(tableSchema, readTable, q.ParentFilterPaths, *q.GroupBy, "GroupBy"); err != nil {
			return "", err
		}
		return "", nil
	}
	return parentFilterMatchedParentGuard(tableSchema, readTable, index, q.ParentFilterPaths[index]), nil
}

func analyticsHavingExpression(tableSchema *models.ObjectMetadata, readTable string, q models.AnalyticsQuery) (string, error) {
	resolvedHaving := q.Having
	if len(q.ParentFilterPaths) > 0 {
		var err error
		resolvedHaving, err = replaceParentFilterPathIdentifiers(resolvedHaving, tableSchema, readTable, q.ParentFilterPaths, "Having")
		if err != nil {
			return "", err
		}
	}
	replacements := buildFieldReplacements(tableSchema, readTable)
	if len(q.ParentFilterPaths) > 0 {
		replacements = buildQualifiedFieldReplacements(tableSchema, readTable)
	}
	if len(replacements) > 0 {
		resolvedHaving = replaceRollupFilterIdentifiers(resolvedHaving, replacements)
	}
	return resolvedHaving, nil
}

func (r *QueryRepository) applyAnalyticsGroupOrderAndLimit(builder *query.Builder, tableSchema *models.ObjectMetadata, q models.AnalyticsQuery, readTable string, aggregateAlias string) error {
	if q.SortField != "" {
		if !isValidFieldName(q.SortField) {
			builder.OrderBy(fmt.Sprintf("`%s`", aggregateAlias), "DESC")
		} else {
			field := q.SortField
			direction := q.SortDirection
			if direction != "ASC" && direction != "DESC" {
				direction = "DESC"
			}
			if field == "name" || field == "val" || field == "value" {
				sortCol := field
				if field == "val" || field == "value" {
					sortCol = aggregateAlias
				}
				builder.OrderBy(fmt.Sprintf("`%s`", sortCol), direction)
			} else {
				sortExpr, err := parentFilterStructuredSortExpression(tableSchema, readTable, q.ParentFilterPaths, field, "SortField")
				if err != nil {
					return err
				}
				if sortExpr != "" {
					builder.OrderByRaw(fmt.Sprintf("%s %s", sortExpr, direction), nil)
				}
			}
		}
	} else {
		builder.OrderBy(fmt.Sprintf("`%s`", aggregateAlias), "DESC")
	}

	limit := constants.DefaultSOSLLimit
	if q.Limit > 0 {
		limit = q.Limit
	}
	builder.Limit(limit)
	if q.Offset > 0 {
		builder.Offset(q.Offset)
	}
	return nil
}

func (r *QueryRepository) executeAndFormatAnalytics(ctx context.Context, builder *query.Builder, q models.AnalyticsQuery) (interface{}, error) {
	queryP := builder.Build()
	log.Debug().Str("sql", queryP.SQL).Any("params", queryP.Params).Msg("DEBUG_ANALYTICS_SQL")

	exec := r.GetExecutor(ctx)
	rows, err := exec.QueryContext(ctx, queryP.SQL, queryP.Params...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	results, err := query.ScanRowsToSObjects(rows)
	if err != nil {
		return nil, err
	}

	// Post-process to ensure numeric types for aggregate result columns.
	for _, row := range results {
		for key, v := range row {
			if !isAnalyticsAggregateResultColumn(key) {
				continue
			}
			if s, isStr := v.(string); isStr {
				if f, err := strconv.ParseFloat(s, 64); err == nil {
					row[key] = f
				}
			}
		}
	}

	if q.Operation == OpGroupBy {
		return results, nil
	}

	if len(q.Aggregates) > 0 {
		if len(results) > 0 {
			return results[0], nil
		}
		return models.SObject{}, nil
	}

	if len(results) > 0 {
		return results[0]["val"], nil
	}

	return 0, nil
}

func isAnalyticsAggregateResultColumn(key string) bool {
	if key == "val" {
		return true
	}
	if !strings.HasPrefix(key, "agg") || len(key) == len("agg") {
		return false
	}
	for _, c := range key[len("agg"):] {
		if c < '0' || c > '9' {
			return false
		}
	}
	return true
}