package automation

import (
	"context"
	"fmt"
	"strings"
	"time"

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

// executeRecordLookup queries records
func (e *FlowExecutor) executeRecordLookup(ctx context.Context, interview *models.FlowInterview, lookup *models.FlowRecordLookup, user *models.UserSession) (nextElem string, execErr error) {
	ctx, auditFault := withAuditFaultBox(ctx)
	start := time.Now()
	profile := flowRecordLookupProfileEvent{
		Started: start,
		Status:  "success",
		Source:  "query",
	}
	var profilePlan flowRecordLookupProfilePlan
	var auditInputJSON string
	var result *models.QueryResult
	if lookup != nil {
		profile.ElementName = lookup.Name
		profile.ObjectName = lookup.Object
		profile.GetFirstRecordOnly = lookup.GetFirstRecordOnly
		profilePlan = buildFlowRecordLookupProfilePlan(interview, lookup)
		profile.FilterShape = profilePlan.shape
		profile.Batchable = profilePlan.batchable
		profile.BatchableReason = profilePlan.reason
		profile.BatchKeyField = profilePlan.keyField
		profile.CrossObject = profilePlan.crossObject
	}
	if current, ok := flowRuntimeContextFromCtx(ctx).CurrentFlow(); ok {
		profile.FlowName = current.APIName
		profile.FlowVersionID = current.VersionID
	}
	defer func() {
		status := "success"
		errPayload := ""
		var profileErr error
		if execErr != nil {
			status = "error"
			errPayload = execErr.Error()
			profileErr = execErr
		} else if be := auditFault.Err(); be != nil {
			status = "error"
			errPayload = be.Error()
			profileErr = be
		}
		emitFlowAudit(ctx, Row{
			FlowExecutionId:  interview.Id,
			FlowDefinitionId: auditFlowDefinitionIdFromCtx(ctx),
			FlowVersionId:    interview.FlowVersionId,
			ElementId:        lookup.Name,
			ElementType:      "RecordLookup",
			Status:           status,
			DurationMs:       int(time.Since(start).Milliseconds()),
			InputJSON:        auditInputJSON,
			OutputJSON:       recordLookupAuditOutputJSON(result, profile.Source),
			ErrorPayload:     errPayload,
		}, auditSampleRateFromCtx(ctx))
		profile.Status = status
		profile.Err = profileErr
		logFlowRecordLookupProfile(ctx, profile)
	}()
	// Build filter from conditions. Use the ctx-aware resolver so filter
	// right-hand-side references that include `__r` (e.g. "$Record.Order__r.Id")
	// resolve correctly before being embedded in the WHERE clause.
	filter, err := e.buildFilterFromConditionsCtxE(ctx, interview, user, lookup.Filters, lookup.FilterLogic)
	if err != nil {
		if lookup.FaultConnector != nil {
			return e.routeFault(ctx, interview, lookup.FaultConnector.TargetReference, err)
		}
		return "", err
	}
	profile.FilterHash = shortTimingHash(filter)

	limit := 200
	if lookup.GetFirstRecordOnly {
		limit = 1
	} else if boundedLimit, bounded := preparedFlowDirectRecordOperationLimit(ctx); bounded {
		// Bounded prepared execution must distinguish an exact in-bound
		// collection from a truncated one before any downstream mutation. Fetch
		// one sentinel row; the governor check below rejects the complete lookup
		// when that row exists.
		limit = boundedLimit + 1
	}
	auditInputJSON = recordLookupAuditInputJSON(filter, limit)

	// In BeforeSave context the outer trigger transaction is open. Cross-object
	// GetRecords that run within that transaction can return 0 rows because TiDB
	// REPEATABLE READ takes a read snapshot at the transaction's first statement —
	// rows committed by other sessions after that point are invisible to
	// subsequent reads in the same tx. Stripping the transaction (DetachTx) gives
	// the query a fresh connection with a clean read view while preserving all
	// other context values (SystemContext, BeforeSaveFlow marker). (#1781)
	//
	// DetachTx: BeforeSave outer transaction's read snapshot is too old to see
	// records committed by other sessions. Run cross-object GetRecords on a fresh
	// connection to get an up-to-date snapshot.
	//
	// SEMANTIC TRADE-OFF: For same-object lookups, we DO NOT DetachTx — the outer
	// transaction needs to see its own uncommitted $Record mutations. This guard
	// ensures same-object lookups preserve BeforeSave in-memory semantics.
	queryCtx := ctx
	if isBeforeSaveFlow(ctx) && isCrossObjectLookup(interview, lookup.Object) {
		queryCtx = e.persistence.DetachTx(ctx)
	}

	queryLimit := limit
	if queryLimit > 0 {
		queryLimit += len(atomicPostUpsertLookupRecordIDs(ctx, interview, lookup))
	}
	profile.QueryLimit = queryLimit
	cacheKey, cacheable, cacheSkipReason := flowRecordLookupCacheKey(ctx, interview, lookup, user, filter, queryLimit)
	if !cacheable {
		logFlowRecordLookupCache(ctx, lookup, "skip", cacheSkipReason, queryLimit)
	}
	coalesceKey := cacheKey
	coalesceable := cacheable
	if !coalesceable {
		var coalesceSkipReason string
		coalesceKey, coalesceable, coalesceSkipReason = flowRecordLookupInFlightCoalesceKey(ctx, interview, lookup, user, filter, queryLimit)
		if !coalesceable {
			logFlowRecordLookupCache(ctx, lookup, "singleflight_skip", coalesceSkipReason, queryLimit)
		}
	}
	if coalesceable && e.persistence != nil && e.persistence.IsTransactionActive(queryCtx) {
		canDetach, skipReason := flowRecordLookupActiveTxDetachAllowed(ctx, profilePlan)
		if !canDetach {
			coalesceable = false
			logFlowRecordLookupCache(ctx, lookup, "singleflight_skip", skipReason, queryLimit)
		} else {
			queryCtx = e.persistence.DetachTx(queryCtx)
			if e.persistence.IsTransactionActive(queryCtx) {
				coalesceable = false
				logFlowRecordLookupCache(ctx, lookup, "singleflight_skip", "detach_tx_still_active", queryLimit)
			}
		}
	}
	if cacheable {
		if cached, ok := flowRuntimeContextFromCtx(ctx).CachedRecordLookupResult(cacheKey); ok {
			result = cached
			profile.Source = "request_cache"
			logFlowRecordLookupCache(ctx, lookup, "hit", "", queryLimit)
		}
	}
	if result == nil {
		if cacheable {
			logFlowRecordLookupCache(ctx, lookup, "miss", "", queryLimit)
		}
		lookupStarted := time.Now()
		var err error
		var batched bool
		var batchAttempt flowRecordLookupBatchAttempt
		result, batched, batchAttempt, err = e.tryBatchedRecordLookup(ctx, queryCtx, interview, lookup, user, profilePlan, queryLimit)
		profile.BatchSkipReason = batchAttempt.SkipReason
		profile.BatchDetachedTx = batchAttempt.DetachedTx
		profile.BatchKeyFieldFound = batchAttempt.KeyFieldFound
		profile.BatchKeyFieldUnique = batchAttempt.KeyFieldUnique
		if batched {
			profile.Source = "query_batch"
		} else {
			var shared bool
			result, shared, err = e.queryRecordLookupWithSingleflight(coalesceKey, coalesceable, func() (*models.QueryResult, error) {
				return e.queryFlowRecordLookup(queryCtx, lookup, filter, user, queryLimit)
			})
			if shared && !cacheable && coalesceable {
				profile.Source = "query_singleflight"
			}
			stagetiming.Add(ctx, stagetiming.StageRecordLookup, time.Since(lookupStarted))
		}
		if err != nil {
			if lookup.FaultConnector != nil {
				return e.routeFault(ctx, interview, lookup.FaultConnector.TargetReference, err)
			}
			return "", err
		}
		e.seedRecordLookupOwnerUserProof(ctx, queryCtx, lookup.Object, result)
		result, err = e.mergeAtomicPostUpsertLookupRecords(ctx, interview, lookup, user, result, filter, limit)
		if err != nil {
			if lookup.FaultConnector != nil {
				return e.routeFault(ctx, interview, lookup.FaultConnector.TargetReference, err)
			}
			return "", err
		}
		result, err = e.mergePreviewSyntheticLookupRecords(ctx, interview, lookup, user, result, limit)
		if err != nil {
			if lookup.FaultConnector != nil {
				return e.routeFault(ctx, interview, lookup.FaultConnector.TargetReference, err)
			}
			return "", err
		}
		if err := preparedFlowDirectRecordOperationLimitError(
			ctx, "RecordLookup", lookup.Object, len(result.Records),
		); err != nil {
			if lookup.FaultConnector != nil {
				return e.routeFault(ctx, interview, lookup.FaultConnector.TargetReference, err)
			}
			return "", err
		}
		if cacheable {
			flowRuntimeContextFromCtx(ctx).SetCachedRecordLookupResult(cacheKey, result)
			logFlowRecordLookupCache(ctx, lookup, "store", "", queryLimit)
		}
	}
	result = projectFlowRecordLookupResult(result, lookup, e.lookupObjectSchema(ctx, lookup.Object))
	if result != nil {
		profile.Rows = len(result.Records)
	}

	// Stamp neutral object-type metadata on every returned record so a subsequent
	// `LookupName.Parent__r.Name` reference can resolve the target object via
	// metadata. `QueryWithFilter` returns raw rows; without this, downstream
	// `__r` traversal would fall back to the trigger object's type and miss.
	for _, record := range result.Records {
		models.StampObjectType(record, lookup.Object)
	}
	flowRuntimeContextFromCtx(ctx).RememberRecordLookupBaselines(lookup.Object, result.Records)

	// Store results. When the element declares an explicit OutputReference
	// (Salesforce Flow's "storeOutputs" binding), also key the result under
	// that variable name so downstream elements can reference it by the
	// author's chosen name (e.g. "$lead") rather than by the element name.
	// The element-name key is preserved for backwards compatibility with
	// flows that didn't set OutputReference.
	storeKeys := []string{lookup.Name}
	if lookup.OutputReference != "" && lookup.OutputReference != lookup.Name {
		storeKeys = append(storeKeys, lookup.OutputReference)
	}

	if lookup.GetFirstRecordOnly {
		if len(result.Records) > 0 {
			for _, key := range storeKeys {
				interview.ContextData[key] = result.Records[0]
			}

			// Output assignments
			for _, output := range lookup.OutputAssignments {
				if val, ok := result.Records[0][output.Field]; ok {
					interview.ContextData[output.AssignToReference] = val
				}
			}
		} else {
			for _, key := range storeKeys {
				interview.ContextData[key] = nil
			}
		}
	} else {
		for _, key := range storeKeys {
			interview.ContextData[key] = result.Records
		}
	}

	if lookup.Connector != nil {
		return lookup.Connector.TargetReference, nil
	}
	return "", nil
}

func (e *FlowExecutor) queryFlowRecordLookup(ctx context.Context, lookup *models.FlowRecordLookup, filter string, user *models.UserSession, limit int) (*models.QueryResult, error) {
	selectFields := flowRecordLookupExecutionSelectFields(lookup, lookup.SortField)
	if len(selectFields) == 0 {
		return e.query.QueryWithFilter(ctx, lookup.Object, filter, user, lookup.SortField, lookup.SortOrder, limit)
	}
	return e.query.QueryWithSpec(ctx, queryspec.QuerySpec{
		ObjectAPIName:      lookup.Object,
		SelectFields:       selectFields,
		Filter:             queryspec.FormulaPredicate(filter),
		SortField:          lookup.SortField,
		SortDirection:      lookup.SortOrder,
		Limit:              limit,
		ReferenceHydration: flowRecordLookupReferenceHydration(lookup),
	}, user)
}

func flowRecordLookupReferenceHydration(lookup *models.FlowRecordLookup) queryspec.ReferenceHydrationMode {
	if lookup != nil && lookup.QueriedFields != nil {
		return queryspec.ReferenceHydrationExplicitOnly
	}
	return queryspec.ReferenceHydrationDefault
}

// flowRecordLookupSelectFields returns the metadata-authored field closure for
// a RecordLookup. A nil result deliberately preserves the compatibility path
// for metadata that predates queriedFields and expects every readable field.
func flowRecordLookupSelectFields(lookup *models.FlowRecordLookup) []string {
	if lookup == nil || lookup.QueriedFields == nil {
		return nil
	}
	fields := make([]string, 0, len(lookup.QueriedFields))
	seen := make(map[string]struct{}, len(lookup.QueriedFields))
	for _, authored := range lookup.QueriedFields {
		field := strings.TrimSpace(authored)
		if field == "" {
			continue
		}
		key := strings.ToLower(field)
		if _, exists := seen[key]; exists {
			continue
		}
		seen[key] = struct{}{}
		fields = append(fields, field)
	}
	// An explicitly authored empty/blank closure is still a closure. Keep the
	// mandatory record identity instead of silently falling back to every
	// readable field, which would recreate the unsafe collection-update shape.
	if len(fields) == 0 {
		return []string{constants.FieldID}
	}
	return fields
}

func flowRecordLookupProjectionKey(lookup *models.FlowRecordLookup) string {
	fields := flowRecordLookupSelectFields(lookup)
	for i := range fields {
		fields[i] = strings.ToLower(fields[i])
	}
	return strings.Join(fields, "\x1f")
}

func flowRecordLookupExecutionSelectFields(lookup *models.FlowRecordLookup, required ...string) []string {
	fields := flowRecordLookupSelectFields(lookup)
	if len(fields) == 0 {
		return nil
	}
	seen := make(map[string]struct{}, len(fields)+len(required))
	for _, field := range fields {
		seen[strings.ToLower(field)] = struct{}{}
	}
	for _, candidate := range required {
		field := strings.TrimSpace(candidate)
		if field == "" {
			continue
		}
		key := strings.ToLower(field)
		if _, exists := seen[key]; exists {
			continue
		}
		seen[key] = struct{}{}
		fields = append(fields, field)
	}
	return fields
}

func projectFlowRecordLookupResult(result *models.QueryResult, lookup *models.FlowRecordLookup, schema *models.ObjectMetadata) *models.QueryResult {
	selected := flowRecordLookupSelectFields(lookup)
	if result == nil || len(selected) == 0 {
		return result
	}
	allowed := make(map[string]struct{}, len(selected)+1)
	allowed[strings.ToLower(constants.FieldID)] = struct{}{}
	for _, field := range selected {
		allowed[strings.ToLower(field)] = struct{}{}
		if parent, _, ok := strings.Cut(field, "."); ok {
			parent = strings.TrimSpace(parent)
			allowed[strings.ToLower(parent)] = struct{}{}
			if refField := flowRecordLookupReferenceField(schema, parent); refField != nil {
				allowed[strings.ToLower(flowRecordLookupRelationshipName(*refField))] = struct{}{}
			}
		}
	}
	projected := *result
	projected.Records = make([]models.SObject, 0, len(result.Records))
	for _, record := range result.Records {
		row := make(models.SObject, len(allowed))
		for field, value := range record {
			if _, ok := allowed[strings.ToLower(field)]; ok {
				row[field] = value
			}
		}
		projected.Records = append(projected.Records, row)
	}
	return &projected
}

func flowRecordLookupReferenceField(schema *models.ObjectMetadata, parent string) *models.FieldMetadata {
	if schema == nil {
		return nil
	}
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if strings.EqualFold(field.APIName, parent) ||
			(field.RelationshipName != nil && strings.EqualFold(strings.TrimSpace(*field.RelationshipName), parent)) {
			return field
		}
	}
	return nil
}

func flowRecordLookupRelationshipName(field models.FieldMetadata) string {
	if field.RelationshipName != nil && strings.TrimSpace(*field.RelationshipName) != "" {
		return strings.TrimSpace(*field.RelationshipName)
	}
	if strings.HasSuffix(field.APIName, "__c") {
		return strings.TrimSuffix(field.APIName, "__c") + "__r"
	}
	return strings.TrimSuffix(field.APIName, "Id")
}

func flowRecordLookupActiveTxDetachAllowed(ctx context.Context, plan flowRecordLookupProfilePlan) (bool, string) {
	if !isWithinAtomicFlowBoundary(ctx) || isBeforeSaveFlow(ctx) {
		return false, "active_tx"
	}
	if !plan.crossObject {
		return false, "active_tx_same_object"
	}
	if flowRuntimeContextFromCtx(ctx).HasRecordMutation() {
		return false, "active_tx_after_mutation"
	}
	return true, ""
}

func (e *FlowExecutor) seedRecordLookupOwnerUserProof(ctx context.Context, queryCtx context.Context, objectName string, result *models.QueryResult) {
	if result == nil || !strings.EqualFold(strings.TrimSpace(objectName), constants.TableUser) || !e.recordLookupOwnerProofSeedAllowed(ctx, queryCtx) {
		return
	}
	for _, record := range result.Records {
		if record == nil {
			continue
		}
		id, _ := record[constants.FieldID].(string)
		if isTruthyFlowGuardField(record[constants.FieldIsDeleted]) {
			continue
		}
		records.SeedOwnerUserLookupProof(ctx, id)
	}
}

func (e *FlowExecutor) recordLookupOwnerProofSeedAllowed(ctx context.Context, queryCtx context.Context) bool {
	if flowRuntimeContextFromCtx(ctx).HasRecordMutation() {
		return false
	}
	if e == nil || e.persistence == nil || !e.persistence.IsTransactionActive(queryCtx) {
		return true
	}
	return isWithinAtomicFlowBoundary(ctx)
}

func isTruthyFlowGuardField(value interface{}) bool {
	switch v := value.(type) {
	case nil:
		return false
	case bool:
		return v
	case int:
		return v != 0
	case int32:
		return v != 0
	case int64:
		return v != 0
	case uint8:
		return v != 0
	case float64:
		return v != 0
	case []byte:
		return len(v) == 1 && v[0] != 0
	case string:
		return v == "1" || strings.EqualFold(v, "true")
	default:
		return false
	}
}

type recordLookupSingleflightResult struct {
	result *models.QueryResult
}

func (e *FlowExecutor) queryRecordLookupWithSingleflight(cacheKey string, cacheable bool, query func() (*models.QueryResult, error)) (*models.QueryResult, bool, error) {
	// Tests and narrowly scoped internal helpers may construct FlowExecutor
	// literals. A missing group must not be initialized lazily here: concurrent
	// callers would race over ownership. The production constructor always
	// supplies the group; an incomplete literal safely loses only coalescing.
	if e == nil || e.recordLookupQueryGroup == nil || !cacheable || strings.TrimSpace(cacheKey) == "" {
		result, err := query()
		return result, false, err
	}
	value, err, shared := e.recordLookupQueryGroup.Do(cacheKey, func() (interface{}, error) {
		result, err := query()
		if err != nil {
			return nil, err
		}
		return recordLookupSingleflightResult{result: cloneRuntimeQueryResult(result)}, nil
	})
	if err != nil {
		return nil, shared, err
	}
	out, ok := value.(recordLookupSingleflightResult)
	if !ok {
		return nil, shared, fmt.Errorf("record lookup singleflight returned %T", value)
	}
	return cloneRuntimeQueryResult(out.result), shared, nil
}

// mergeAtomicPostUpsertLookupRecords keeps detached reads as the default
// cross-object before-save path, then overlays rows this execution already
// wrote by re-reading only those registry-owned IDs on the ambient atomic tx.
// This closes same-execution created/updated-row visibility gaps without
// hardcoding app-specific literals or switching the whole lookup to ambient tx.
func (e *FlowExecutor) mergeAtomicPostUpsertLookupRecords(
	ctx context.Context,
	interview *models.FlowInterview,
	lookup *models.FlowRecordLookup,
	user *models.UserSession,
	result *models.QueryResult,
	filter string,
	limit int,
) (*models.QueryResult, error) {
	ownedIDs := atomicPostUpsertLookupRecordIDs(ctx, interview, lookup)
	if len(ownedIDs) == 0 {
		return result, nil
	}
	if result == nil {
		result = &models.QueryResult{Done: true}
	}

	rows := make([]models.SObject, 0, len(result.Records)+len(ownedIDs))
	seenIDs := map[string]int{}
	for _, record := range result.Records {
		rows = append(rows, clonePreviewLookupRow(record))
		if id, _ := record[constants.FieldID].(string); strings.TrimSpace(id) != "" {
			seenIDs[strings.TrimSpace(id)] = len(rows) - 1
		}
	}

	appended := 0
	replaced := 0
	removed := 0
	overlayCtx := atomicPostUpsertLookupReadContext(ctx)
	for _, id := range ownedIDs {
		overlay, err := e.queryFlowRecordLookup(overlayCtx, lookup, lookupOwnedRecordFilter(filter, id), user, 1)
		if err != nil {
			return nil, err
		}
		matchedOwned := false
		for _, row := range overlay.Records {
			rowID, _ := row[constants.FieldID].(string)
			rowID = strings.TrimSpace(rowID)
			if rowID == "" || rowID != id {
				continue
			}
			matchedOwned = true
			replacement := clonePreviewLookupRow(row)
			if existing, exists := seenIDs[rowID]; exists {
				rows[existing] = replacement
				replaced++
				continue
			}
			seenIDs[rowID] = len(rows)
			rows = append(rows, replacement)
			appended++
		}
		if !matchedOwned {
			before := len(rows)
			rows = removePreviewLookupRowByID(rows, seenIDs, id)
			if len(rows) < before {
				removed++
			}
		}
	}
	needsLimitTrim := limit > 0 && len(rows) > limit
	if appended == 0 && replaced == 0 && removed == 0 && !needsLimitTrim {
		return result, nil
	}

	if strings.TrimSpace(lookup.SortField) != "" {
		sortPreviewLookupRows(rows, lookup.SortField, lookup.SortOrder)
	}
	if limit > 0 && len(rows) > limit {
		rows = rows[:limit]
	}

	merged := *result
	merged.Records = rows
	merged.TotalSize = result.TotalSize + appended - removed
	if merged.TotalSize < 0 {
		merged.TotalSize = 0
	}
	if merged.TotalSize < len(rows) {
		merged.TotalSize = len(rows)
	}
	if merged.TotalSize == 0 {
		merged.Done = true
	}
	return &merged, nil
}

func atomicPostUpsertLookupRecordIDs(ctx context.Context, interview *models.FlowInterview, lookup *models.FlowRecordLookup) []string {
	if lookup == nil || interview == nil || !isBeforeSaveFlow(ctx) || !isWithinAtomicFlowBoundary(ctx) || !isCrossObjectLookup(interview, lookup.Object) {
		return nil
	}
	postUpsert := flowRuntimeContextFromCtx(ctx).PostUpsertRecords(lookup.Object)
	ownedIDs := make([]string, 0, len(postUpsert))
	ownedSet := map[string]struct{}{}
	for _, record := range postUpsert {
		if record == nil {
			continue
		}
		id, _ := record[constants.FieldID].(string)
		id = strings.TrimSpace(id)
		if id == "" {
			continue
		}
		if _, exists := ownedSet[id]; exists {
			continue
		}
		ownedSet[id] = struct{}{}
		ownedIDs = append(ownedIDs, id)
	}
	return ownedIDs
}

func atomicPostUpsertLookupReadContext(ctx context.Context) context.Context {
	return constants.WithSecurityContext(ctx, constants.UserContext, "Flow RecordLookup owned-row overlay uses invoking user read authority")
}

func lookupOwnedRecordFilter(filter string, id string) string {
	id = strings.ReplaceAll(strings.TrimSpace(id), "'", "''")
	if id == "" {
		return strings.TrimSpace(filter)
	}
	if strings.TrimSpace(filter) == "" {
		return fmt.Sprintf("%s = '%s'", constants.FieldID, id)
	}
	return fmt.Sprintf("(%s) AND %s = '%s'", filter, constants.FieldID, id)
}

func removePreviewLookupRowByID(rows []models.SObject, seenIDs map[string]int, id string) []models.SObject {
	id = strings.TrimSpace(id)
	idx, exists := seenIDs[id]
	if id == "" || !exists {
		return rows
	}
	rows = append(rows[:idx], rows[idx+1:]...)
	delete(seenIDs, id)
	for i := idx; i < len(rows); i++ {
		rowID, _ := rows[i][constants.FieldID].(string)
		rowID = strings.TrimSpace(rowID)
		if rowID != "" {
			seenIDs[rowID] = i
		}
	}
	return rows
}