package automation

import (
	"context"
	"fmt"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/kernel/pkg/perftiming"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	"github.com/nexuscrm/shared/pkg/stagetiming"
)

const (
	recordLookupBatchWaitEnv       = "NEXUSCRM_FLOW_RECORD_LOOKUP_BATCH_WAIT_MS"
	recordLookupBatchMaxEntriesEnv = "NEXUSCRM_FLOW_RECORD_LOOKUP_BATCH_MAX_ENTRIES"
	defaultRecordLookupBatchMax    = 64
)

type flowRecordLookupBatcher struct {
	wait time.Duration
	max  int

	mu      sync.Mutex
	batches map[string]*flowRecordLookupBatch
}

type flowRecordLookupBatch struct {
	key   string
	req   flowRecordLookupBatchRequest
	run   func(context.Context, flowRecordLookupBatchRequest, []*flowRecordLookupBatchCall) (map[string]*models.QueryResult, error)
	calls []*flowRecordLookupBatchCall
}

type flowRecordLookupBatchCall struct {
	ctx      context.Context
	valueKey string
	value    interface{}
	resultCh chan flowRecordLookupBatchResult
}

type flowRecordLookupBatchResult struct {
	result *models.QueryResult
	err    error
}

type flowRecordLookupBatchRequest struct {
	groupKey           string
	valueKey           string
	value              interface{}
	object             string
	keyField           string
	selectFields       []string
	referenceHydration queryspec.ReferenceHydrationMode
	predicate          queryspec.Predicate
	limit              int
	collection         bool
	user               *models.UserSession
}

type flowRecordLookupBatchAttempt struct {
	SkipReason     string
	DetachedTx     bool
	KeyFieldFound  bool
	KeyFieldUnique bool
}

func newFlowRecordLookupBatcherFromEnv() *flowRecordLookupBatcher {
	waitMillis := parsePositiveIntEnv(recordLookupBatchWaitEnv)
	if waitMillis <= 0 {
		return nil
	}
	maxEntries := parsePositiveIntEnv(recordLookupBatchMaxEntriesEnv)
	if maxEntries <= 0 {
		maxEntries = defaultRecordLookupBatchMax
	}
	return newFlowRecordLookupBatcher(time.Duration(waitMillis)*time.Millisecond, maxEntries)
}

func newFlowRecordLookupBatcher(wait time.Duration, max int) *flowRecordLookupBatcher {
	if wait <= 0 || max <= 0 {
		return nil
	}
	return &flowRecordLookupBatcher{
		wait:    wait,
		max:     max,
		batches: map[string]*flowRecordLookupBatch{},
	}
}

func parsePositiveIntEnv(name string) int {
	raw := strings.TrimSpace(os.Getenv(name))
	if raw == "" {
		return 0
	}
	value, err := strconv.Atoi(raw)
	if err != nil || value <= 0 {
		return 0
	}
	return value
}

func (b *flowRecordLookupBatcher) Do(
	ctx context.Context,
	req flowRecordLookupBatchRequest,
	run func(context.Context, flowRecordLookupBatchRequest, []*flowRecordLookupBatchCall) (map[string]*models.QueryResult, error),
) (*models.QueryResult, bool, error) {
	if b == nil || strings.TrimSpace(req.groupKey) == "" || strings.TrimSpace(req.valueKey) == "" || run == nil {
		return nil, false, nil
	}
	if ctx == nil {
		ctx = context.Background()
	}
	call := &flowRecordLookupBatchCall{
		ctx:      ctx,
		valueKey: req.valueKey,
		value:    req.value,
		resultCh: make(chan flowRecordLookupBatchResult, 1),
	}

	b.mu.Lock()
	batch := b.batches[req.groupKey]
	if batch == nil {
		batch = &flowRecordLookupBatch{key: req.groupKey, req: req, run: run}
		b.batches[req.groupKey] = batch
		time.AfterFunc(b.wait, func() {
			b.flush(req.groupKey)
		})
	}
	batch.calls = append(batch.calls, call)
	shouldFlush := len(batch.calls) >= b.max
	b.mu.Unlock()

	if shouldFlush {
		go b.flush(req.groupKey)
	}

	select {
	case out := <-call.resultCh:
		return out.result, true, out.err
	case <-ctx.Done():
		return nil, true, ctx.Err()
	}
}

func (b *flowRecordLookupBatcher) flush(key string) {
	b.mu.Lock()
	batch := b.batches[key]
	if batch == nil {
		b.mu.Unlock()
		return
	}
	delete(b.batches, key)
	calls := append([]*flowRecordLookupBatchCall(nil), batch.calls...)
	run := batch.run
	b.mu.Unlock()

	if len(calls) == 0 || run == nil {
		return
	}
	activeCtx, activeCancel, activeCalls := activeRecordLookupBatchCalls(calls)
	if len(activeCalls) == 0 {
		for _, call := range calls {
			notifyRecordLookupBatchCall(call, nil, callContextErr(call))
		}
		return
	}
	defer activeCancel()
	results, err := run(activeCtx, batch.req, activeCalls)
	for _, call := range calls {
		callErr := err
		if ctxErr := callContextErr(call); ctxErr != nil {
			callErr = ctxErr
		}
		notifyRecordLookupBatchCall(call, results, callErr)
	}
}

func activeRecordLookupBatchCalls(calls []*flowRecordLookupBatchCall) (context.Context, context.CancelFunc, []*flowRecordLookupBatchCall) {
	active := make([]*flowRecordLookupBatchCall, 0, len(calls))
	var activeCtx context.Context
	for _, call := range calls {
		if call == nil || callContextErr(call) != nil {
			continue
		}
		active = append(active, call)
		if call.ctx != nil {
			if activeCtx == nil {
				activeCtx = context.WithoutCancel(call.ctx)
			}
		}
	}
	if activeCtx == nil {
		activeCtx = context.Background()
	}
	// Keep request-scoped values from an active caller, but make cancellation
	// batch-owned so one caller's deadline cannot abort work still needed by
	// other active callers.
	activeCtx, cancel := context.WithCancel(activeCtx)
	cancelWhenAllRecordLookupBatchCallersDone(activeCtx, cancel, active)
	return activeCtx, cancel, active
}

func cancelWhenAllRecordLookupBatchCallersDone(batchCtx context.Context, cancel context.CancelFunc, calls []*flowRecordLookupBatchCall) {
	var wg sync.WaitGroup
	waiters := 0
	for _, call := range calls {
		if call == nil {
			continue
		}
		if call.ctx == nil || call.ctx.Done() == nil {
			// A non-cancellable active caller may still be waiting, so a
			// cancellable subset cannot prove that the batch is abandoned.
			return
		}
		wg.Add(1)
		waiters++
		go func(done <-chan struct{}) {
			defer wg.Done()
			select {
			case <-done:
			case <-batchCtx.Done():
			}
		}(call.ctx.Done())
	}
	if waiters == 0 {
		return
	}
	go func() {
		wg.Wait()
		cancel()
	}()
}

func callContextErr(call *flowRecordLookupBatchCall) error {
	if call == nil || call.ctx == nil {
		return nil
	}
	return call.ctx.Err()
}

func notifyRecordLookupBatchCall(call *flowRecordLookupBatchCall, results map[string]*models.QueryResult, err error) {
	if call == nil {
		return
	}
	out := flowRecordLookupBatchResult{err: err}
	if err == nil && results != nil {
		out.result = results[call.valueKey]
		if out.result == nil {
			out.result = &models.QueryResult{Done: true}
		}
	}
	select {
	case call.resultCh <- out:
	default:
	}
}

func (e *FlowExecutor) tryBatchedRecordLookup(
	ctx context.Context,
	queryCtx context.Context,
	interview *models.FlowInterview,
	lookup *models.FlowRecordLookup,
	user *models.UserSession,
	plan flowRecordLookupProfilePlan,
	queryLimit int,
) (*models.QueryResult, bool, flowRecordLookupBatchAttempt, error) {
	attempt := flowRecordLookupBatchAttempt{}
	skip := func(reason string) (*models.QueryResult, bool, flowRecordLookupBatchAttempt, error) {
		attempt.SkipReason = reason
		return nil, false, attempt, nil
	}
	if e == nil || e.recordLookupBatcher == nil || e.query == nil || e.persistence == nil || lookup == nil {
		return skip("disabled")
	}
	// Collection lookups can batch when the dynamic equality key is unique. The
	// uniqueness gate below preserves each caller's result cardinality.
	if !plan.batchable || queryLimit < 1 {
		return skip("shape")
	}
	if isPreviewMode(ctx) || constants.HasStandardActionInternalAccessScopes(ctx) {
		return skip("context_scope")
	}
	if e.persistence.IsTransactionActive(queryCtx) {
		canDetach, skipReason := flowRecordLookupActiveTxDetachAllowed(ctx, plan)
		if !canDetach {
			return skip(skipReason)
		}
		queryCtx = e.persistence.DetachTx(queryCtx)
		attempt.DetachedTx = true
		if e.persistence.IsTransactionActive(queryCtx) {
			return skip("detach_tx_still_active")
		}
	}
	if len(atomicPostUpsertLookupRecordIDs(ctx, interview, lookup)) != 0 {
		return skip("post_upsert_overlay")
	}
	keyFound, keyUnique := e.recordLookupBatchKeyUniqueness(ctx, lookup.Object, plan.keyField)
	attempt.KeyFieldFound = keyFound
	attempt.KeyFieldUnique = keyUnique
	if !keyUnique {
		if !keyFound {
			return skip("key_not_found")
		}
		return skip("key_not_unique")
	}

	req, ok, err := e.buildRecordLookupBatchRequest(ctx, interview, lookup, user, plan, queryLimit)
	if err != nil {
		return nil, false, attempt, err
	}
	if !ok {
		return skip("request_build")
	}
	start := time.Now()
	result, batched, err := e.recordLookupBatcher.Do(queryCtx, req, e.executeRecordLookupBatch)
	if batched {
		stagetiming.Add(ctx, stagetiming.StageRecordLookup, time.Since(start))
	}
	return result, batched, attempt, err
}

func (e *FlowExecutor) recordLookupBatchKeyUniqueness(ctx context.Context, objectName, fieldName string) (bool, bool) {
	fieldName = strings.TrimSpace(fieldName)
	if strings.EqualFold(fieldName, constants.FieldID) {
		return true, true
	}
	if schema := e.lookupObjectSchema(ctx, objectName); schema != nil {
		if field, ok := schema.FieldsMap[strings.ToLower(fieldName)]; ok && field != nil {
			return true, field.IsUnique
		}
		for i := range schema.Fields {
			field := &schema.Fields[i]
			if strings.EqualFold(field.APIName, fieldName) {
				return true, field.IsUnique
			}
		}
	}
	field := e.lookupFieldMeta(ctx, objectName, fieldName)
	if field == nil {
		return false, false
	}
	return true, field.IsUnique
}

func (e *FlowExecutor) buildRecordLookupBatchRequest(
	ctx context.Context,
	interview *models.FlowInterview,
	lookup *models.FlowRecordLookup,
	user *models.UserSession,
	plan flowRecordLookupProfilePlan,
	queryLimit int,
) (flowRecordLookupBatchRequest, bool, error) {
	if lookup == nil || strings.TrimSpace(plan.keyField) == "" {
		return flowRecordLookupBatchRequest{}, false, nil
	}

	predicates := make([]queryspec.Predicate, 0, len(lookup.Filters))
	predicateKeyParts := make([]string, 0, len(lookup.Filters))
	var keyValue interface{}
	keyFound := false
	for _, filter := range lookup.Filters {
		if err := validateIdentifier(filter.Field); err != nil {
			return flowRecordLookupBatchRequest{}, false, nil
		}
		operator := strings.ToUpper(strings.TrimSpace(normalizeFlowOperator(filter.Operator)))
		op, ok := flowRecordLookupBatchOperator(operator)
		if !ok {
			return flowRecordLookupBatchRequest{}, false, nil
		}
		value, _, err := e.ResolveFlowResource(ctx, interview, user, filter.Value)
		if err != nil {
			return flowRecordLookupBatchRequest{}, false, err
		}
		value, _ = extractTypedValue(value)
		if strings.EqualFold(strings.TrimSpace(filter.Field), plan.keyField) && flowRecordLookupValueIsDynamicReference(filter.Value) {
			if op != "=" {
				return flowRecordLookupBatchRequest{}, false, nil
			}
			keyValue = value
			keyFound = true
			continue
		}
		if value == nil {
			return flowRecordLookupBatchRequest{}, false, nil
		}
		predicateValueKey, ok := flowRecordLookupBatchValueKey(value)
		if !ok {
			return flowRecordLookupBatchRequest{}, false, nil
		}
		predicateKeyParts = append(predicateKeyParts,
			strings.ToLower(strings.TrimSpace(filter.Field))+"\x00"+op+"\x00"+predicateValueKey)
		predicates = append(predicates, queryspec.FieldPredicate(filter.Field, op, value))
	}
	if !keyFound {
		return flowRecordLookupBatchRequest{}, false, nil
	}
	valueKey, ok := flowRecordLookupBatchValueKey(keyValue)
	if !ok {
		return flowRecordLookupBatchRequest{}, false, nil
	}

	groupKey := flowRecordLookupBatchGroupKey(ctx, lookup, user, plan, queryLimit, strings.Join(predicateKeyParts, "\x1f"))
	return flowRecordLookupBatchRequest{
		groupKey:           groupKey,
		valueKey:           valueKey,
		value:              keyValue,
		object:             lookup.Object,
		keyField:           plan.keyField,
		selectFields:       flowRecordLookupExecutionSelectFields(lookup, plan.keyField),
		referenceHydration: flowRecordLookupReferenceHydration(lookup),
		predicate:          queryspec.And(predicates...),
		limit:              queryLimit,
		collection:         !lookup.GetFirstRecordOnly,
		user:               user,
	}, true, nil
}

func flowRecordLookupBatchOperator(operator string) (string, bool) {
	switch operator {
	case "EQUALTO", "=", "==":
		return "=", true
	case "NOTEQUALTO", "!=", "<>":
		return "!=", true
	default:
		return "", false
	}
}

func flowRecordLookupBatchValueKey(value interface{}) (string, bool) {
	switch v := value.(type) {
	case nil:
		return "", false
	case string:
		if strings.TrimSpace(v) == "" {
			return "", false
		}
		return "s:" + v, true
	case []byte:
		return flowRecordLookupBatchValueKey(string(v))
	case fmt.Stringer:
		return flowRecordLookupBatchValueKey(v.String())
	case bool:
		return "b:" + strconv.FormatBool(v), true
	case int:
		return "i:" + strconv.Itoa(v), true
	case int8, int16, int32, int64:
		return "i:" + fmt.Sprint(v), true
	case uint, uint8, uint16, uint32, uint64:
		return "u:" + fmt.Sprint(v), true
	case float32, float64:
		return "f:" + fmt.Sprint(v), true
	default:
		s := fmt.Sprint(v)
		if strings.TrimSpace(s) == "" || s == "<nil>" {
			return "", false
		}
		return "x:" + s, true
	}
}

func flowRecordLookupBatchStringValue(value interface{}) (string, bool) {
	switch v := value.(type) {
	case nil:
		return "", false
	case string:
		if strings.TrimSpace(v) == "" {
			return "", false
		}
		return v, true
	case []byte:
		return flowRecordLookupBatchStringValue(string(v))
	case fmt.Stringer:
		return flowRecordLookupBatchStringValue(v.String())
	default:
		return "", false
	}
}

func flowRecordLookupBatchValuesSQLMatch(rowValue, callValue interface{}) bool {
	rowKey, rowKeyOK := flowRecordLookupBatchValueKey(rowValue)
	callKey, callKeyOK := flowRecordLookupBatchValueKey(callValue)
	if rowKeyOK && callKeyOK && rowKey == callKey {
		return true
	}

	rowString, rowStringOK := flowRecordLookupBatchStringValue(rowValue)
	callString, callStringOK := flowRecordLookupBatchStringValue(callValue)
	if rowStringOK && callStringOK {
		return strings.EqualFold(rowString, callString)
	}
	return false
}

func flowRecordLookupBatchHasStringRemapAmbiguity(calls []*flowRecordLookupBatchCall) bool {
	seen := make([]string, 0, len(calls))
	for _, call := range calls {
		if call == nil {
			continue
		}
		value, ok := flowRecordLookupBatchStringValue(call.value)
		if !ok {
			continue
		}
		if value != strings.TrimSpace(value) {
			return true
		}
		for _, existing := range seen {
			if existing != value && strings.EqualFold(strings.TrimSpace(existing), strings.TrimSpace(value)) {
				return true
			}
		}
		seen = append(seen, value)
	}
	return false
}

func flowRecordLookupBatchGroupKey(ctx context.Context, lookup *models.FlowRecordLookup, user *models.UserSession, plan flowRecordLookupProfilePlan, queryLimit int, predicateKey string) string {
	var b strings.Builder
	appendLookupCacheKeyPart(&b, "batch-v1")
	appendLookupCacheKeyPart(&b, strings.ToLower(strings.TrimSpace(lookup.Object)))
	appendLookupCacheKeyPart(&b, strings.ToLower(strings.TrimSpace(plan.keyField)))
	appendLookupCacheKeyPart(&b, plan.shape)
	appendLookupCacheKeyPart(&b, predicateKey)
	appendLookupCacheKeyPart(&b, flowRecordLookupProjectionKey(lookup))
	appendLookupCacheKeyPart(&b, strconv.Itoa(queryLimit))
	appendLookupCacheKeyPart(&b, fmt.Sprintf("%v", constants.GetSecurityContext(ctx)))
	appendLookupCacheKeyPart(&b, flowRecordLookupCacheUserKey(user))
	return b.String()
}

func (e *FlowExecutor) executeRecordLookupBatch(
	ctx context.Context,
	req flowRecordLookupBatchRequest,
	calls []*flowRecordLookupBatchCall,
) (results map[string]*models.QueryResult, err error) {
	start := time.Now()
	uniqueValues := make([]interface{}, 0, len(calls))
	seen := map[string]struct{}{}
	for _, call := range calls {
		if call == nil {
			continue
		}
		if _, ok := seen[call.valueKey]; ok {
			continue
		}
		seen[call.valueKey] = struct{}{}
		uniqueValues = append(uniqueValues, call.value)
	}
	defer func() {
		logFlowRecordLookupBatch(ctx, req.object, req.keyField, len(calls), len(uniqueValues), results, start, err)
	}()
	if len(uniqueValues) == 0 {
		return map[string]*models.QueryResult{}, nil
	}
	if req.collection {
		// Collection lookups were newly admitted to this path by this PR. Until
		// remapping has authoritative SQL type/collation semantics, let SQL own
		// each value instead of interpreting a shared IN result in Go.
		return e.executeRecordLookupBatchByValue(ctx, req, calls)
	}
	if flowRecordLookupBatchHasStringRemapAmbiguity(calls) {
		// Without field collation metadata, one shared IN result cannot safely
		// remap case/whitespace-equivalent caller values. Let SQL own each equality.
		return e.executeRecordLookupBatchByValue(ctx, req, calls)
	}

	predicate := queryspec.And(req.predicate, queryspec.FieldPredicate(req.keyField, "IN", uniqueValues))
	result, err := e.query.QueryWithSpec(ctx, queryspec.QuerySpec{
		ObjectAPIName:      req.object,
		SelectFields:       append([]string(nil), req.selectFields...),
		Filter:             predicate,
		Limit:              len(uniqueValues),
		ReferenceHydration: req.referenceHydration,
	}, req.user)
	if err != nil {
		return nil, err
	}

	byKey := make(map[string]*models.QueryResult, len(uniqueValues))
	for _, row := range result.Records {
		rowValue := row[req.keyField]
		for _, call := range calls {
			if call == nil {
				continue
			}
			if _, exists := byKey[call.valueKey]; exists {
				continue
			}
			// The batch query uses ordinary SQL equality/IN predicates. NexusCRM
			// tables default to utf8mb4_unicode_ci, so string result remapping must
			// follow case-insensitive predicate semantics instead of exact Go keys.
			if !flowRecordLookupBatchValuesSQLMatch(rowValue, call.value) {
				continue
			}
			byKey[call.valueKey] = &models.QueryResult{
				TotalSize: 1,
				Done:      true,
				Records:   []models.SObject{cloneRuntimeSObject(row)},
			}
		}
	}
	for _, call := range calls {
		if call == nil {
			continue
		}
		if _, exists := byKey[call.valueKey]; !exists {
			byKey[call.valueKey] = &models.QueryResult{Done: true}
		}
	}
	return byKey, nil
}

func (e *FlowExecutor) executeRecordLookupBatchByValue(
	ctx context.Context,
	req flowRecordLookupBatchRequest,
	calls []*flowRecordLookupBatchCall,
) (map[string]*models.QueryResult, error) {
	byKey := make(map[string]*models.QueryResult, len(calls))
	for _, call := range calls {
		if call == nil {
			continue
		}
		if _, exists := byKey[call.valueKey]; exists {
			continue
		}
		predicate := queryspec.And(req.predicate, queryspec.FieldPredicate(req.keyField, "=", call.value))
		result, err := e.query.QueryWithSpec(ctx, queryspec.QuerySpec{
			ObjectAPIName:      req.object,
			SelectFields:       append([]string(nil), req.selectFields...),
			Filter:             predicate,
			Limit:              req.limit,
			ReferenceHydration: req.referenceHydration,
		}, req.user)
		if err != nil {
			return nil, err
		}
		if result == nil {
			result = &models.QueryResult{Done: true}
		}
		byKey[call.valueKey] = cloneRuntimeQueryResult(result)
	}
	for _, call := range calls {
		if call == nil {
			continue
		}
		if _, exists := byKey[call.valueKey]; !exists {
			byKey[call.valueKey] = &models.QueryResult{Done: true}
		}
	}
	return byKey, nil
}

func logFlowRecordLookupBatch(ctx context.Context, objectName, keyField string, calls, uniqueValues int, results map[string]*models.QueryResult, started time.Time, err error) {
	if !perftiming.Enabled(ctx) {
		return
	}
	rows := 0
	for _, result := range results {
		if result != nil {
			rows += len(result.Records)
		}
	}
	perftiming.Event(ctx, "flow_record_lookup_batch", started, err).
		Str("object", objectName).
		Str("key_field", keyField).
		Int("calls", calls).
		Int("unique_values", uniqueValues).
		Int("rows", rows).
		Msg("Flow RecordLookup batch")
}