package automation

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
)

const flowAtomicParentPrelockPageSize = 200

func withFlowAtomicParentPrelocks(ctx context.Context, discovered []recordlocks.ParentKey) context.Context {
	if len(discovered) == 0 {
		return ctx
	}
	all := append(recordlocks.ForcedParentPrelocksFromContext(ctx), discovered...)
	return recordlocks.WithForcedParentPrelocks(ctx, all)
}

func validateFlowAtomicParentPrelocks(meta *models.FlowMetadata) error {
	if meta == nil {
		return nil
	}
	if err := models.ValidateFlowAtomicParentPrelocks(*meta); err != nil {
		return apperrors.NewValidationError("atomicParentPrelocks", err.Error())
	}
	return nil
}

func (e *FlowExecutor) discoverFlowAtomicParentPrelocks(
	ctx context.Context,
	interview *models.FlowInterview,
	meta *models.FlowMetadata,
	user *models.UserSession,
) ([]recordlocks.ParentKey, error) {
	if meta == nil || len(meta.AtomicParentPrelocks) == 0 {
		return nil, nil
	}
	if e.query == nil {
		return nil, apperrors.NewInternalError("atomic parent prelock discovery requires a query service", nil)
	}

	var keys []recordlocks.ParentKey
	discoveredRows := 0
	consumeRows := func(rows int) error {
		limit, bounded := preparedFlowDirectRecordOperationLimit(ctx)
		if !bounded {
			return nil
		}
		if rows < 0 || rows > limit-discoveredRows {
			return fmt.Errorf(
				"%w: atomic parent prelock discovery exceeds %d rows",
				ErrFlowExecutionBudgetExceeded,
				limit,
			)
		}
		discoveredRows += rows
		return nil
	}
	for _, prelock := range meta.AtomicParentPrelocks {
		for i, rawID := range prelock.RecordIDs {
			resolved, resolveErr := e.resolveFlowAtomicParentPrelockValue(ctx, interview, user, rawID)
			if resolveErr != nil {
				return nil, apperrors.NewValidationError("atomicParentPrelocks", fmt.Sprintf("parent prelock %q recordIds[%d]: %v", prelock.Name, i, resolveErr))
			}
			ids, err := flowAtomicParentPrelockIDs(resolved)
			if err != nil {
				return nil, apperrors.NewValidationError("atomicParentPrelocks", fmt.Sprintf("parent prelock %q: %v", prelock.Name, err))
			}
			if err := consumeRows(len(ids)); err != nil {
				return nil, err
			}
			for _, id := range ids {
				keys = append(keys, recordlocks.ParentKey{ObjectName: prelock.ParentObject, RecordID: id})
			}
		}

		if strings.TrimSpace(prelock.SourceObject) == "" {
			continue
		}
		resolvedFilters := make([]models.FlowRecordFilter, len(prelock.Filters))
		copy(resolvedFilters, prelock.Filters)
		for i, condition := range prelock.Filters {
			resolved, resolveErr := e.resolveFlowAtomicParentPrelockValue(ctx, interview, user, condition.Value)
			if resolveErr != nil {
				return nil, apperrors.NewValidationError("atomicParentPrelocks", fmt.Sprintf("parent prelock %q filter %d: %v", prelock.Name, i+1, resolveErr))
			}
			resolvedFilters[i].Value = resolved
			operator := strings.ToUpper(strings.TrimSpace(normalizeFlowOperator(condition.Operator)))
			if operator != "BETWEEN" && operator != "NOTBETWEEN" {
				continue
			}
			if len(parseBetweenValue(resolved)) != 2 {
				return nil, apperrors.NewValidationError("atomicParentPrelocks", fmt.Sprintf("parent prelock %q filter %s requires exactly two resolved bounds", prelock.Name, condition.Field))
			}
		}
		filter := e.buildFilterFromConditionsImpl(resolvedFilters, prelock.FilterLogic, false, func(value interface{}) interface{} {
			return value
		})
		if strings.TrimSpace(filter) == "" || strings.EqualFold(strings.TrimSpace(filter), "1=1") {
			return nil, apperrors.NewValidationError("atomicParentPrelocks", fmt.Sprintf("parent prelock %q resolved to an unbounded source query", prelock.Name))
		}
		for offset := 0; ; {
			pageSize := flowAtomicParentPrelockPageSize
			if limit, bounded := preparedFlowDirectRecordOperationLimit(ctx); bounded {
				remaining := limit - discoveredRows
				pageSize = remaining + 1
				if pageSize > flowAtomicParentPrelockPageSize {
					pageSize = flowAtomicParentPrelockPageSize
				}
			}
			result, err := e.query.QueryWithSpec(ctx, queryspec.QuerySpec{
				ObjectAPIName: prelock.SourceObject,
				SelectFields:  []string{prelock.ParentReferenceField},
				Filter:        queryspec.ExpressionPredicate(queryspec.RawOperand(filter)),
				SortField:     constants.FieldID,
				SortDirection: "ASC",
				Limit:         pageSize,
				Offset:        offset,
				SkipTotalSize: true,
			}, user)
			if err != nil {
				return nil, apperrors.NewInternalError(fmt.Sprintf("atomic parent prelock %q discovery failed", prelock.Name), err)
			}
			if result == nil {
				break
			}
			if err := consumeRows(len(result.Records)); err != nil {
				return nil, err
			}
			for _, row := range result.Records {
				ids, valueErr := flowAtomicParentPrelockIDs(row[prelock.ParentReferenceField])
				if valueErr != nil {
					return nil, apperrors.NewValidationError("atomicParentPrelocks", fmt.Sprintf("parent prelock %q field %s: %v", prelock.Name, prelock.ParentReferenceField, valueErr))
				}
				for _, id := range ids {
					keys = append(keys, recordlocks.ParentKey{ObjectName: prelock.ParentObject, RecordID: id})
				}
			}
			if len(result.Records) < pageSize {
				break
			}
			offset += pageSize
		}
	}
	return recordlocks.OrderedParentKeys(keys), nil
}

func (e *FlowExecutor) resolveFlowAtomicParentPrelockValue(
	ctx context.Context,
	interview *models.FlowInterview,
	user *models.UserSession,
	value interface{},
) (interface{}, error) {
	switch typed := value.(type) {
	case string:
		return nil, fmt.Errorf("bare string value %q is ambiguous; use stringValue for a literal or elementReference for an input reference", typed)
	case []string:
		if len(typed) > 0 {
			return nil, fmt.Errorf("bare string collection is ambiguous; use typed value wrappers")
		}
		return []interface{}{}, nil
	case map[string]interface{}:
		if rawRef, isReference := typed["elementReference"]; isReference {
			if len(typed) != 1 {
				return nil, fmt.Errorf("elementReference wrapper must not contain additional values")
			}
			resolved, present, err := e.ResolveFlowResource(ctx, interview, user, value)
			if err != nil {
				return nil, err
			}
			if !present {
				return nil, fmt.Errorf("elementReference %q did not resolve before the first Flow element", rawRef)
			}
			resolved = normalizeFlowAtomicParentPrelockResolvedValue(resolved)
			if flowAtomicParentPrelockResolvedValueContainsMap(resolved) {
				return nil, fmt.Errorf("elementReference %q resolved to an unsupported map value", rawRef)
			}
			return resolved, nil
		}
		if len(typed) != 1 {
			return nil, fmt.Errorf("typed literal wrapper must not contain additional values")
		}
		if key, literal, isLiteral := flowAtomicParentPrelockTypedLiteral(typed); isLiteral {
			if !flowAtomicParentPrelockLiteralMatchesType(key, literal) {
				return nil, fmt.Errorf("%s must contain a scalar value of the declared type", key)
			}
			return literal, nil
		}
		return nil, fmt.Errorf("map value must be a supported typed literal wrapper or elementReference")
	case []interface{}:
		resolved := make([]interface{}, len(typed))
		for i, item := range typed {
			value, err := e.resolveFlowAtomicParentPrelockValue(ctx, interview, user, item)
			if err != nil {
				return nil, err
			}
			resolved[i] = value
		}
		return resolved, nil
	default:
		resolved, _, err := e.ResolveFlowResource(ctx, interview, user, value)
		return resolved, err
	}
}

func flowAtomicParentPrelockResolvedValueContainsMap(value interface{}) bool {
	switch typed := value.(type) {
	case map[string]interface{}:
		return true
	case models.Record:
		return true
	case []interface{}:
		for _, item := range typed {
			if flowAtomicParentPrelockResolvedValueContainsMap(item) {
				return true
			}
		}
	}
	return false
}

func normalizeFlowAtomicParentPrelockResolvedValue(value interface{}) interface{} {
	switch typed := value.(type) {
	case map[string]interface{}:
		if key, literal, isLiteral := flowAtomicParentPrelockTypedLiteral(typed); isLiteral && flowAtomicParentPrelockLiteralMatchesType(key, literal) {
			return literal
		}
		resolved := make(map[string]interface{}, len(typed))
		for key, item := range typed {
			resolved[key] = normalizeFlowAtomicParentPrelockResolvedValue(item)
		}
		return resolved
	case []interface{}:
		resolved := make([]interface{}, len(typed))
		for i, item := range typed {
			resolved[i] = normalizeFlowAtomicParentPrelockResolvedValue(item)
		}
		return resolved
	default:
		return value
	}
}

func flowAtomicParentPrelockTypedLiteral(value map[string]interface{}) (string, interface{}, bool) {
	if len(value) != 1 {
		return "", nil, false
	}
	for key, literal := range value {
		switch key {
		case "stringValue", "numberValue", "booleanValue", "dateValue", "dateTimeValue":
			return key, literal, true
		}
	}
	return "", nil, false
}

func flowAtomicParentPrelockLiteralMatchesType(key string, value interface{}) bool {
	switch key {
	case "stringValue", "dateValue", "dateTimeValue":
		_, ok := value.(string)
		return ok
	case "booleanValue":
		_, ok := value.(bool)
		return ok
	case "numberValue":
		switch value.(type) {
		case int, int8, int16, int32, int64,
			uint, uint8, uint16, uint32, uint64,
			float32, float64, json.Number:
			return true
		}
	}
	return false
}

func flowAtomicParentPrelockIDs(value interface{}) ([]string, error) {
	switch typed := value.(type) {
	case nil:
		return nil, nil
	case string:
		if id := strings.TrimSpace(typed); id != "" {
			return []string{id}, nil
		}
		return nil, nil
	case []string:
		var ids []string
		for _, item := range typed {
			resolved, err := flowAtomicParentPrelockIDs(item)
			if err != nil {
				return nil, err
			}
			ids = append(ids, resolved...)
		}
		return ids, nil
	case []interface{}:
		var ids []string
		for _, item := range typed {
			resolved, err := flowAtomicParentPrelockIDs(item)
			if err != nil {
				return nil, err
			}
			ids = append(ids, resolved...)
		}
		return ids, nil
	default:
		return nil, fmt.Errorf("record id must resolve to a string or string collection, got %T", value)
	}
}