package analytics

import (
	"context"
	"database/sql"
	"fmt"
	"math"
	"strings"
	"sync"
	"time"

	pkgErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/perftiming"
	"github.com/nexuscrm/kernel/pkg/ports"
	domainrls "github.com/nexuscrm/kernel/pkg/rls"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/kernel/pkg/utils"
	"github.com/nexuscrm/shared/pkg/constants"
	neutralcontract "github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	"github.com/nexuscrm/shared/pkg/searchmetadata"
	zlog "github.com/rs/zerolog/log"
	"golang.org/x/sync/errgroup"
)

// QueryService handles all query operations
type QueryService struct {
	repo        ports.QueryRepository
	metadata    MetadataSource
	permissions PermissionChecker
	validator   *SecurityValidator
	logger      *logging.Logger

	historyDB      *sql.DB
	historyViews   HistoryViewResolver
	historyFactory HistoryQueryFactory
}

// SetHistoryBackend wires the field-history execution backend used by the
// neutral Query.QueryHistory contract. The history query implementation is
// supplied by an injected HistoryQueryFactory (an audit-backed adapter wired at
// the composition root); QueryService owns the public query contract surface.
func (qs *QueryService) SetHistoryBackend(db *sql.DB, views HistoryViewResolver) {
	if qs == nil {
		return
	}
	qs.historyDB = db
	qs.historyViews = views
}

// SetHistoryFactory injects the audit-backed factory used to build the history
// query executor. Wired at the composition root, inverting the analytics ->
// features/audit edge.
func (qs *QueryService) SetHistoryFactory(factory HistoryQueryFactory) {
	if qs == nil {
		return
	}
	qs.historyFactory = factory
}

// NewQueryService creates a new QueryService
func NewQueryService(
	repo ports.QueryRepository,
	metadata MetadataSource,
	permissions PermissionChecker,
	logger *logging.Logger,
) *QueryService {
	return &QueryService{
		repo:        repo,
		metadata:    metadata,
		permissions: permissions,
		validator:   NewSecurityValidator(permissions, metadata),
		logger:      logger,
	}
}

func scopedReadBypassesUserAccess(ctx context.Context, objectName string) bool {
	return constants.GetSecurityContext(ctx) >= constants.SystemContext ||
		constants.IsScopedInternalAccess(ctx, objectName, constants.PermRead)
}

func scopedReadBypassesFieldVisibility(ctx context.Context, objectName string) bool {
	return constants.GetSecurityContext(ctx) >= constants.SystemContext ||
		constants.ScopedInternalAccessBypassesFLS(ctx, objectName, constants.PermRead)
}

func scopedReadEnforcesFieldVisibility(ctx context.Context, objectName string) bool {
	return constants.GetSecurityContext(ctx) < constants.SystemContext &&
		constants.ScopedInternalAccessEnforcesFLS(ctx, objectName, constants.PermRead)
}

func (qs *QueryService) checkObjectReadAccess(ctx context.Context, objectName string, currentUser *models.UserSession) error {
	if scopedReadBypassesUserAccess(ctx, objectName) {
		if qs.permissions.IsBlockedObject(ctx, objectName, constants.PermRead) {
			return pkgErrors.NewPermissionError("read", objectName)
		}
		return nil
	}
	if !qs.permissions.CheckObjectPermissionWithUser(ctx, objectName, constants.PermRead, currentUser) {
		return pkgErrors.NewPermissionError("read", objectName)
	}
	return nil
}

func (qs *QueryService) readRLSResult(ctx context.Context, schema *models.ObjectMetadata, objectName string, currentUser *models.UserSession) *domainrls.Result {
	if scopedReadBypassesUserAccess(ctx, objectName) {
		return nil
	}
	return qs.permissions.BuildRLSResult(ctx, schema, objectName, currentUser)
}

func (qs *QueryService) visibleFieldsForQuery(ctx context.Context, objectName string, schema *models.ObjectMetadata, currentUser *models.UserSession) ([]string, error) {
	visibleFields, err := qs.systemFieldsForQuery(ctx, objectName, schema, currentUser)
	if err != nil {
		return nil, err
	}
	persistable := models.PersistableFields(schema)
	fieldSet := make(map[string]bool, len(persistable))
	for _, field := range persistable {
		fieldSet[field.APIName] = true
	}

	// Add custom fields that are visible. SystemContext and legacy scoped
	// internal read see all fields except security-sensitive values; scoped
	// reads that opt into EnforceFLS preserve user field visibility.
	for _, field := range persistable {
		if field.IsSystem || field.IsNameField {
			continue
		}
		readable, err := qs.fieldReadableForQueryWithSchema(ctx, objectName, schema, field, currentUser)
		if err != nil {
			return nil, err
		}
		if readable {
			visibleFields = append(visibleFields, field.APIName)
			if field.IsPolymorphic {
				typeCol := schemautil.GetPolymorphicTypeColumnName(field.APIName)
				if fieldSet[typeCol] {
					visibleFields = append(visibleFields, typeCol)
				}
			}
		}
	}
	return visibleFields, nil
}

func (qs *QueryService) systemFieldsForQuery(ctx context.Context, objectName string, schema *models.ObjectMetadata, currentUser *models.UserSession) ([]string, error) {
	systemFields := make([]string, 0, len(schema.Fields))
	for i := range schema.Fields {
		field := &schema.Fields[i]
		if field.IsVirtual {
			continue
		}
		if field.IsSystem || field.IsNameField {
			systemFields = append(systemFields, field.APIName)
		}
	}
	filteredSystemFields := make([]string, 0, len(systemFields))
	for _, apiName := range systemFields {
		field := models.FindField(schema, apiName)
		if field != nil && qs.fieldReadBlockedWithSchema(ctx, objectName, schema, *field) {
			continue
		}
		filteredSystemFields = append(filteredSystemFields, apiName)
	}
	if !scopedReadEnforcesFieldVisibility(ctx, objectName) {
		return filteredSystemFields, nil
	}

	visibleFields := make([]string, 0, len(filteredSystemFields))
	for _, apiName := range filteredSystemFields {
		field := models.FindField(schema, apiName)
		if field == nil || field.IsSystem || !field.IsNameField {
			visibleFields = append(visibleFields, apiName)
			continue
		}
		visible := false
		if qs.permissions != nil {
			var err error
			visible, err = qs.checkFieldVisibilityWithSchema(ctx, objectName, schema, field, currentUser)
			if err != nil {
				return nil, err
			}
		}
		if visible {
			visibleFields = append(visibleFields, field.APIName)
		}
	}
	return visibleFields, nil
}

func (qs *QueryService) fieldReadableForQueryWithSchema(ctx context.Context, objectName string, schema *models.ObjectMetadata, field models.FieldMetadata, currentUser *models.UserSession) (bool, error) {
	if qs.fieldReadBlockedWithSchema(ctx, objectName, schema, field) {
		return false, nil
	}
	if field.IsSystem {
		return true, nil
	}
	if field.IsNameField && !scopedReadEnforcesFieldVisibility(ctx, objectName) {
		return true, nil
	}
	if scopedReadBypassesFieldVisibility(ctx, objectName) {
		return true, nil
	}
	if qs.permissions == nil {
		return false, nil
	}
	return qs.checkFieldVisibilityWithSchema(ctx, objectName, schema, &field, currentUser)
}

func (qs *QueryService) checkFieldVisibilityWithSchema(ctx context.Context, objectName string, schema *models.ObjectMetadata, field *models.FieldMetadata, currentUser *models.UserSession) (bool, error) {
	if checker, ok := qs.permissions.(SchemaBoundFieldPermissionChecker); ok {
		return checker.CheckFieldVisibilityWithSchema(ctx, schema, field, currentUser)
	}
	return qs.permissions.CheckFieldVisibilityWithUser(ctx, objectName, field.APIName, currentUser)
}

func (qs *QueryService) fieldReadBlockedWithSchema(ctx context.Context, objectName string, schema *models.ObjectMetadata, field models.FieldMetadata) bool {
	if _, ok := qs.permissions.(SchemaBoundFieldPermissionChecker); ok {
		return FieldReadBlockedByMetadata(&field)
	}
	if qs.permissions != nil {
		if qs.permissions.IsFieldReadBlocked(ctx, objectName, field.APIName) {
			return true
		}
	}
	if qs.metadata != nil {
		return FieldReadBlockedByMetadata(models.FindField(schema, field.APIName))
	}
	return false
}

// Legacy single-field helpers retain the existing local package contract for
// callers that have not yet loaded a schema. Bulk paths use the schema-bound
// forms above and never materialize the object again per field.
func (qs *QueryService) fieldReadableForQuery(ctx context.Context, objectName string, field models.FieldMetadata, currentUser *models.UserSession) (bool, error) {
	if qs.metadata == nil {
		return false, nil
	}
	schema := qs.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return false, nil
	}
	return qs.fieldReadableForQueryWithSchema(ctx, objectName, schema, field, currentUser)
}

func (qs *QueryService) projectSystemPermissionVirtualFields(ctx context.Context, records []models.SObject, schema *models.ObjectMetadata, selectFields []string) error {
	if qs.permissions == nil || schema == nil || len(records) == 0 {
		return nil
	}
	systemPermissionVirtuals := models.VirtualFieldsByKind(schema, constants.VirtualFieldSystemPermission)
	if len(systemPermissionVirtuals) == 0 {
		return nil
	}
	projectFields := systemPermissionVirtuals
	if len(selectFields) > 0 {
		projectFields = make([]models.FieldMetadata, 0)
		added := make(map[string]bool)
		for _, requested := range selectFields {
			f := models.FindField(schema, requested)
			if f != nil && f.IsVirtual && f.VirtualKind == constants.VirtualFieldSystemPermission {
				key := strings.ToLower(f.APIName)
				if !added[key] {
					projectFields = append(projectFields, *f)
					added[key] = true
				}
			}
		}
	}
	if len(projectFields) == 0 {
		return nil
	}
	return qs.projectPermissionSetGrantFields(ctx, records, projectFields)
}

// ProjectPreviewSyntheticRecords applies the same default read projection that
// QueryWithFilter uses when no explicit select list is supplied. It exists for
// request-local no-DML preview rows that cannot be queried from durable storage
// but still must pass through object access, record access, and FLS projection
// before Flow can use them as recordLookup results.
func (qs *QueryService) ProjectPreviewSyntheticRecords(ctx context.Context, objectName string, records []models.SObject, currentUser *models.UserSession) ([]models.SObject, error) {
	if len(records) == 0 {
		return []models.SObject{}, nil
	}
	if err := qs.checkObjectReadAccess(ctx, objectName, currentUser); err != nil {
		return nil, err
	}
	schema := qs.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return nil, pkgErrors.NewNotFoundError("Object", objectName)
	}

	visibleFields, err := qs.visibleFieldsForQuery(ctx, objectName, schema, currentUser)
	if err != nil {
		return nil, err
	}
	visible := make(map[string]struct{}, len(visibleFields))
	for _, field := range visibleFields {
		visible[field] = struct{}{}
	}

	scopedRead := scopedReadBypassesUserAccess(ctx, objectName)
	if !scopedRead && qs.permissions != nil {
		qs.permissions.PrimeReadAccessInputCache(ctx, schema.APIName, records, currentUser)
	}
	projected := make([]models.SObject, 0, len(records))
	for _, record := range records {
		if record == nil {
			continue
		}
		if !scopedRead && qs.permissions != nil && !qs.permissions.CheckRecordAccess(ctx, schema, record, constants.PermRead, currentUser) {
			continue
		}
		row := make(models.SObject, len(visible))
		for field := range visible {
			if value, ok := record[field]; ok {
				row[field] = value
			}
		}
		projected = append(projected, row)
	}
	if len(projected) == 0 {
		return projected, nil
	}
	projected, err = qs.hydrateVirtualFields(ctx, projected, schema, visibleFields, currentUser, nil)
	if err != nil {
		return nil, err
	}
	projected = projectVirtualCompoundFields(projected, schema, visibleFields)
	if err := qs.projectSystemPermissionVirtualFields(ctx, projected, schema, nil); err != nil {
		return nil, err
	}
	return projected, nil
}

// ProjectRecordsForUserRead applies the normal user-context field projection to
// records that were loaded through a trusted internal read path.
func (qs *QueryService) ProjectRecordsForUserRead(ctx context.Context, objectName string, records []models.SObject, selectFields []string, currentUser *models.UserSession) ([]models.SObject, error) {
	if len(records) == 0 {
		return []models.SObject{}, nil
	}
	schema := qs.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return nil, pkgErrors.NewNotFoundError("Object", objectName)
	}

	visibleFields, err := qs.visibleFieldsForQuery(ctx, objectName, schema, currentUser)
	if err != nil {
		return nil, err
	}
	expandedSelectFields := expandRelationshipSelectFields(schema, selectFields)
	canonicalizedFields := canonicalizeFieldNames(schema, expandedSelectFields)
	queryFields := buildQueryFields(schema, models.QueryRequest{
		ObjectAPIName: objectName,
		SelectFields:  canonicalizedFields,
	}, visibleFields)
	allowed := make(map[string]struct{}, len(queryFields))
	for _, field := range queryFields {
		allowed[field] = struct{}{}
	}

	projected := make([]models.SObject, 0, len(records))
	for _, record := range records {
		if record == nil {
			continue
		}
		row := make(models.SObject, len(allowed))
		for field := range allowed {
			if value, ok := record[field]; ok {
				row[field] = value
			}
		}
		projected = append(projected, row)
	}
	if len(projected) == 0 {
		return projected, nil
	}
	projected, err = qs.hydrateVirtualFields(ctx, projected, schema, queryFields, currentUser, parseDotNotationChildren(schema, canonicalizedFields))
	if err != nil {
		return nil, err
	}
	projectionFields := canonicalizedFields
	if len(projectionFields) == 0 {
		projectionFields = visibleFields
	}
	projected = projectVirtualCompoundFields(projected, schema, projectionFields)
	if err := qs.projectSystemPermissionVirtualFields(ctx, projected, schema, canonicalizedFields); err != nil {
		return nil, err
	}
	if err := qs.projectRelationshipSelectFields(ctx, schema, projected, selectFields, currentUser); err != nil {
		return nil, err
	}
	return projected, nil
}

func (qs *QueryService) ValidateFieldsForUserRead(ctx context.Context, objectName string, fieldNames []string, currentUser *models.UserSession) error {
	if len(fieldNames) == 0 {
		return nil
	}
	schema := qs.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return pkgErrors.NewNotFoundError("Object", objectName)
	}
	visibleFields, err := qs.visibleFieldsForQuery(ctx, objectName, schema, currentUser)
	if err != nil {
		return err
	}
	visible := make(map[string]struct{}, len(visibleFields))
	for _, field := range visibleFields {
		visible[strings.ToLower(field)] = struct{}{}
	}
	for _, field := range canonicalizeFieldNames(schema, expandRelationshipSelectFields(schema, fieldNames)) {
		if strings.TrimSpace(field) == "" {
			continue
		}
		if _, ok := visible[strings.ToLower(field)]; !ok {
			return pkgErrors.NewPermissionError("read", fmt.Sprintf("%s.%s", objectName, field))
		}
	}
	return nil
}

func queryCountTotalSize(result interface{}) (int, bool) {
	value, ok := utils.ToFloat64(result)
	if !ok || value < 0 || math.Trunc(value) != value {
		return 0, false
	}
	maxInt := int(^uint(0) >> 1)
	if value > float64(maxInt) {
		return 0, false
	}
	return int(value), true
}

// CountWithSpec executes one exact filtered count without first entering the
// generic row-query pipeline. runAnalytics remains the single authority owner
// for object access, filter FLS and RLS; this method only removes the duplicate
// row projection/visible-field preparation that QueryWithSpec would perform
// before recognizing COUNT() as an aggregate.
func (qs *QueryService) CountWithSpec(ctx context.Context, req queryspec.CountSpec, currentUser *models.UserSession) (int, error) {
	filterPredicate, correlatedMemberships, err := splitCorrelatedMemberships(req.Filter)
	if err != nil {
		return 0, err
	}
	if len(correlatedMemberships) != 0 {
		return 0, pkgErrors.NewValidationError("Filter", "correlated membership is not supported by exact count")
	}
	filter, err := lowerQueryFilterPredicate(filterPredicate)
	if err != nil {
		return 0, err
	}
	query := models.AnalyticsQuery{
		ObjectAPIName: req.ObjectAPIName,
		Operation:     constants.AnalyticsOperationCount,
		FilterExpr:    filter.expression,
		Criteria:      modelQueryCriteria(filter.criteria),
		ScanAll:       req.ScanAll,
	}
	result, err := qs.runAnalytics(ctx, query, currentUser, nil)
	if err != nil {
		return 0, err
	}
	count, ok := queryCountTotalSize(result)
	if !ok {
		return 0, pkgErrors.NewInternalError("unexpected COUNT() result format", nil)
	}
	return count, nil
}

// QueryWithSpec executes a query through the neutral analytics contract.
func (qs *QueryService) QueryWithSpec(
	ctx context.Context,
	req queryspec.QuerySpec,
	currentUser *models.UserSession,
) (result *models.QueryResult, err error) {
	if !req.ReferenceHydration.Valid() {
		return nil, pkgErrors.NewValidationError("ReferenceHydration", "unsupported reference hydration mode")
	}
	start := time.Now()
	defer func() {
		if !perftiming.Enabled(ctx) {
			return
		}
		rowCount := 0
		if result != nil {
			rowCount = len(result.Records)
		}
		perftiming.Event(ctx, "query_timing", start, err).
			Str("operation", "QueryWithSpec").
			Str("object", req.ObjectAPIName).
			Int("limit", req.Limit).
			Int("rows", rowCount).
			Bool("has_error", err != nil).
			Msg("Query timing")
	}()

	// Check permissions. SystemContext and metadata-owned standard-action read
	// scopes bypass user read/RLS while still enforcing blocked-object policy.
	stageStart := time.Now()
	if err := qs.checkObjectReadAccess(ctx, req.ObjectAPIName, currentUser); err != nil {
		logQueryStageTiming(ctx, "QueryWithSpec", "checkObjectReadAccess", req.ObjectAPIName, stageStart, err, 0, 0)
		return nil, err
	}
	logQueryStageTiming(ctx, "QueryWithSpec", "checkObjectReadAccess", req.ObjectAPIName, stageStart, nil, 0, 0)

	stageStart = time.Now()
	schema := qs.metadata.GetSchema(ctx, req.ObjectAPIName)
	if schema == nil {
		logQueryStageTiming(ctx, "QueryWithSpec", "metadata.GetSchema", req.ObjectAPIName, stageStart, pkgErrors.NewNotFoundError("Object", req.ObjectAPIName), 0, 0)
		return nil, pkgErrors.NewNotFoundError("Object", req.ObjectAPIName)
	}
	logQueryStageTiming(ctx, "QueryWithSpec", "metadata.GetSchema", req.ObjectAPIName, stageStart, nil, len(schema.Fields), 0)

	stageStart = time.Now()
	rlsResult := qs.readRLSResult(ctx, schema, req.ObjectAPIName, currentUser)
	logQueryStageTiming(ctx, "QueryWithSpec", "readRLSResult", req.ObjectAPIName, stageStart, nil, 0, 0)

	// Build visible field list. This is the set of columns the repository
	// will SELECT from the physical storage table.
	//
	// PersistableFields filters out schema-level virtuals (IsVirtual=true),
	// which is what keeps SystemPermission projections (Permissions<X>)
	// out of raw SELECT statements — the refactor's main concern. It does
	// NOT filter formula / rollup fields — those use their own type
	// gating downstream (formula_view_service.go routes them through TiDB
	// views; getStorageColumns in bulk_insert.go adds an explicit
	// FieldTypeFormula / RollupSummary skip). The composition is:
	//   PersistableFields  — excludes decorator-emitted virtuals
	//   + formula/rollup type checks where needed — excludes computed types
	// For the query SELECT list, virtuals are materialized post-fetch by
	// projectPermissionSetGrantFields / hydrateVirtualFields.
	stageStart = time.Now()
	visibleFields, err := qs.visibleFieldsForQuery(ctx, req.ObjectAPIName, schema, currentUser)
	logQueryStageTiming(ctx, "QueryWithSpec", "visibleFieldsForQuery", req.ObjectAPIName, stageStart, err, len(visibleFields), 0)
	if err != nil {
		return nil, err
	}

	// Canonicalize SelectFields to match schema PascalCase (SOQL is case-insensitive)
	canonicalizedFields := canonicalizeFieldNames(schema, req.SelectFields)

	// Convert API DTO to Domain Model for Repository
	// Detect xMruAccessedAt sentinel sort field — resolve to MRU JOIN ordering
	sortField := req.SortField
	sortDirection := req.SortDirection
	var mruUserID string
	if sortField == "xMruAccessedAt" && currentUser != nil {
		mruUserID = currentUser.ID
		sortField = ""
		sortDirection = ""
	}

	filterPredicate, correlatedMemberships, err := splitCorrelatedMemberships(req.Filter)
	if err != nil {
		return nil, err
	}

	filter, err := lowerQueryFilterPredicate(filterPredicate)
	if err != nil {
		return nil, err
	}
	domainReq := models.QueryRequest{
		ObjectAPIName: req.ObjectAPIName,
		FilterExpr:    filter.expression,
		SortField:     sortField,
		SortDirection: sortDirection,
		Limit:         req.Limit,
		Offset:        req.Offset,
		SkipTotalSize: req.SkipTotalSize,
		ScanAll:       req.ScanAll,
		SelectFields:  canonicalizedFields,
		MRUUserID:     mruUserID,
	}
	// Map Criteria
	if len(filter.criteria) > 0 {
		domainReq.Criteria = modelQueryCriteria(filter.criteria)
	}
	// Map OrderBy
	if req.OrderBy != nil {
		domainReq.OrderBy = make([]models.SortCriterion, len(req.OrderBy))
		for i, o := range req.OrderBy {
			domainReq.OrderBy[i] = models.SortCriterion{
				Field:     o.Field,
				Direction: o.Direction,
			}
		}
	}

	// CHECK: Is this an Analytics/Aggregate Query?
	stageStart = time.Now()
	analyticsReq := req
	analyticsReq.Filter = filterPredicate
	isAgg, analyticsPlan, err := qs.buildAnalyticsQuery(analyticsReq)
	logQueryStageTiming(ctx, "QueryWithSpec", "buildAnalyticsQuery", req.ObjectAPIName, stageStart, err, len(req.SelectFields), 0)
	if err != nil {
		return nil, err
	}
	if isAgg && len(req.SubQueries) > 0 {
		return nil, pkgErrors.NewValidationError("SubQueries", "aggregate queries cannot include child relationship subqueries")
	}
	if isAgg {
		if limit := AggregateFetchLimitFromContext(ctx); limit > 0 {
			analyticsPlan.query.Limit = limit
		}
		stageStart = time.Now()
		correlatedMembershipPlans, err := qs.buildCorrelatedMembershipPlans(ctx, req.ObjectAPIName, schema, correlatedMemberships, currentUser)
		logQueryStageTiming(ctx, "QueryWithSpec", "buildCorrelatedMembershipPlans", req.ObjectAPIName, stageStart, err, len(correlatedMemberships), 0)
		if err != nil {
			return nil, err
		}
		// Execute Analytics Query
		stageStart = time.Now()
		result, err := qs.runAnalytics(ctx, analyticsPlan.query, currentUser, correlatedMembershipPlans)
		logQueryStageTiming(ctx, "QueryWithSpec", "RunAnalytics", req.ObjectAPIName, stageStart, err, len(req.SelectFields), 0)
		if err != nil {
			return nil, err
		}
		if analyticsPlan.resultShape == analyticsResultShapeCountTotalSize {
			totalSize, ok := queryCountTotalSize(result)
			if !ok {
				return nil, pkgErrors.NewInternalError("unexpected COUNT() result format", nil)
			}
			return &models.QueryResult{
				TotalSize: totalSize,
				Done:      true,
				Records:   []models.SObject{},
			}, nil
		}

		// Transform Result to Standard QueryResult
		aggRecords := make([]models.SObject, 0)

		if rows, ok := result.([]models.SObject); ok {
			// GROUP BY Results
			aggRecords = make([]models.SObject, len(rows))

			groupByField := ""
			if analyticsPlan.query.GroupBy != nil {
				groupByField = *analyticsPlan.query.GroupBy
			}

			for i, row := range rows {
				newRow := models.SObject{}

				for _, output := range analyticsPlan.aggregateOutputs {
					if v, ok := row[output.resultKey]; ok {
						newRow[output.outputKey] = v
					}
				}

				// Map "name" → GroupBy field name
				// RunAnalytics hydrates lookup IDs to names automatically.
				if v, ok := row["name"]; ok {
					if groupByField != "" {
						newRow[groupByField] = v
					}
				}

				aggRecords[i] = newRow
			}
		} else if row, ok := result.(models.SObject); ok {
			newRow := models.SObject{}
			for _, output := range analyticsPlan.aggregateOutputs {
				if v, ok := row[output.resultKey]; ok {
					newRow[output.outputKey] = v
				}
			}
			aggRecords = append(aggRecords, newRow)
		} else {
			// Single Scalar Result (e.g. COUNT())
			// Wrap in a single record with the SOQL aggregate output key.
			aggRecords = append(aggRecords, models.SObject{
				analyticsPlan.outputKey: result,
			})
		}

		return &models.QueryResult{
			TotalSize:   len(aggRecords),
			Done:        true,
			Records:     aggRecords,
			IsAggregate: true,
		}, nil
	}

	stageStart = time.Now()
	parentFilterPaths, err := qs.validateQueryReadClauses(ctx, req.ObjectAPIName, schema, domainReq, currentUser)
	if err != nil {
		logQueryStageTiming(ctx, "QueryWithSpec", "validateQueryReadClauses", req.ObjectAPIName, stageStart, err, len(domainReq.SelectFields), 0)
		return nil, err
	}
	domainReq.ParentFilterPaths = parentFilterPaths
	logQueryStageTiming(ctx, "QueryWithSpec", "validateQueryReadClauses", req.ObjectAPIName, stageStart, nil, len(domainReq.SelectFields), 0)

	stageStart = time.Now()
	correlatedMembershipPlans, err := qs.buildCorrelatedMembershipPlans(ctx, req.ObjectAPIName, schema, correlatedMemberships, currentUser)
	logQueryStageTiming(ctx, "QueryWithSpec", "buildCorrelatedMembershipPlans", req.ObjectAPIName, stageStart, err, len(correlatedMemberships), 0)
	if err != nil {
		return nil, err
	}

	stageStart = time.Now()
	if err := validateQuerySelections(schema, domainReq.SelectFields, func(targetAPIName string) *models.ObjectMetadata {
		if qs.metadata == nil {
			return nil
		}
		return qs.metadata.GetSchema(ctx, targetAPIName)
	}); err != nil {
		logQueryStageTiming(ctx, "QueryWithSpec", "validateQuerySelections", req.ObjectAPIName, stageStart, err, len(domainReq.SelectFields), 0)
		return nil, err
	}
	logQueryStageTiming(ctx, "QueryWithSpec", "validateQuerySelections", req.ObjectAPIName, stageStart, nil, len(domainReq.SelectFields), 0)

	stageStart = time.Now()
	queryFields := buildQueryFields(schema, domainReq, visibleFields)
	logQueryStageTiming(ctx, "QueryWithSpec", "buildQueryFields", req.ObjectAPIName, stageStart, nil, len(queryFields), 0)

	// Normal Query: Delegate to Repository
	repoReq := domainReq
	repoReq.SelectFields = queryFields
	stageStart = time.Now()
	var queryResult *models.QueryResult
	if len(correlatedMembershipPlans) > 0 {
		correlatedRepo, ok := qs.repo.(ports.CorrelatedMembershipQueryRepository)
		if !ok {
			err = pkgErrors.NewInternalError("correlated membership query executor is not configured", nil)
		} else {
			queryResult, err = correlatedRepo.FindWithCorrelatedMemberships(ctx, schema, repoReq, rlsResult, queryFields, correlatedMembershipPlans)
		}
	} else {
		queryResult, err = qs.repo.Find(ctx, schema, repoReq, rlsResult, queryFields)
	}
	repoRows := 0
	if queryResult != nil {
		repoRows = len(queryResult.Records)
	}
	logQueryStageTiming(ctx, "QueryWithSpec", "repo.Find", req.ObjectAPIName, stageStart, err, len(queryFields), repoRows)
	if err != nil {
		return nil, err
	}

	// Hydrate virtual fields (formulas, booleans)
	stageStart = time.Now()
	queryResult.Records, err = qs.hydrateVirtualFieldsWithReferenceMode(
		ctx, queryResult.Records, schema, queryFields, currentUser,
		parseDotNotationChildren(schema, domainReq.SelectFields),
		req.ReferenceHydration, domainReq.SelectFields,
	)
	logQueryStageTiming(ctx, "QueryWithSpec", "hydrateVirtualFields", req.ObjectAPIName, stageStart, err, len(queryFields), len(queryResult.Records))
	if err != nil {
		return nil, err
	}
	projectionFields := domainReq.SelectFields
	if len(projectionFields) == 0 {
		projectionFields = visibleFields
	}
	stageStart = time.Now()
	queryResult.Records = projectVirtualCompoundFields(queryResult.Records, schema, projectionFields)
	logQueryStageTiming(ctx, "QueryWithSpec", "projectVirtualCompoundFields", req.ObjectAPIName, stageStart, nil, len(projectionFields), len(queryResult.Records))
	stageStart = time.Now()
	queryResult.Records = projectExplicitRelationshipSelections(schema, queryResult.Records, domainReq.SelectFields)
	logQueryStageTiming(ctx, "QueryWithSpec", "projectExplicitRelationshipSelections", req.ObjectAPIName, stageStart, nil, len(domainReq.SelectFields), len(queryResult.Records))

	// Virtual-field projection from backing stores into returned records.
	// Driven by the decorated schema's virtual-field declarations, not a
	// PermissionSet-specific branch.
	stageStart = time.Now()
	if err := qs.projectSystemPermissionVirtualFields(ctx, queryResult.Records, schema, req.SelectFields); err != nil {
		logQueryStageTiming(ctx, "QueryWithSpec", "projectSystemPermissionVirtualFields", req.ObjectAPIName, stageStart, err, len(req.SelectFields), len(queryResult.Records))
		return nil, err
	}
	logQueryStageTiming(ctx, "QueryWithSpec", "projectSystemPermissionVirtualFields", req.ObjectAPIName, stageStart, nil, len(req.SelectFields), len(queryResult.Records))

	// Execute subqueries for child relationships
	if len(req.SubQueries) > 0 {
		domainSubQueries := make([]models.SubQueryRequest, len(req.SubQueries))
		for i, sq := range req.SubQueries {
			filter, err := lowerQueryFilterPredicate(sq.Filter)
			if err != nil {
				return nil, err
			}
			domainSubQueries[i] = models.SubQueryRequest{
				RelationshipName: sq.RelationshipName,
				SelectFields:     sq.SelectFields,
				FilterExpr:       filter.expression,
				Criteria:         modelQueryCriteria(filter.criteria),
				Limit:            sq.Limit,
			}
			if len(sq.OrderBy) > 0 {
				domainSubQueries[i].OrderBy = make([]models.SortCriterion, len(sq.OrderBy))
				for j, o := range sq.OrderBy {
					domainSubQueries[i].OrderBy[j] = models.SortCriterion{Field: o.Field, Direction: o.Direction}
				}
			}
		}
		stageStart = time.Now()
		queryResult.Records, err = qs.executeSubQueries(ctx, queryResult.Records, schema, domainSubQueries, currentUser, req.ScanAll)
		logQueryStageTiming(ctx, "QueryWithSpec", "executeSubQueries", req.ObjectAPIName, stageStart, err, len(req.SubQueries), len(queryResult.Records))
		if err != nil {
			return nil, err
		}
	}

	return queryResult, nil
}

func (qs *QueryService) Query(ctx context.Context, req queryspec.QuerySpec, currentUser *models.UserSession) (*neutralcontract.QueryResult, error) {
	result, err := qs.QueryWithSpec(ctx, req, currentUser)
	if err != nil {
		return nil, err
	}
	neutral := neutralQueryResultFromModel(result)
	if neutral == nil {
		// contract.Query invariant: a nil error guarantees a non-nil result.
		// A no-rows query is represented as an empty, Done result here — never a
		// nil pointer — so neutral consumers never need per-call-site nil guards.
		neutral = &neutralcontract.QueryResult{Done: true}
	}
	return neutral, nil
}

func (qs *QueryService) QueryHistory(ctx context.Context, req neutralcontract.HistoryQueryRequest, user *models.UserSession) (*neutralcontract.HistoryQueryResult, error) {
	if qs == nil || qs.historyFactory == nil {
		return nil, &neutralcontract.HistoryQueryError{
			Kind:    neutralcontract.HistoryQueryErrorUnavailable,
			Message: "history query backend is not configured",
		}
	}
	return qs.historyFactory.
		Build(qs.historyDB, qs.metadata, qs.permissions, qs.historyViews).
		QueryHistory(ctx, req, user)
}

func neutralQueryResultFromModel(result *models.QueryResult) *neutralcontract.QueryResult {
	if result == nil {
		return nil
	}
	return &neutralcontract.QueryResult{
		TotalSize:        result.TotalSize,
		TotalSizeUnknown: result.TotalSizeUnknown,
		Done:             result.Done,
		Records:          neutralRecordsFromSObjects(result.Records),
		IsAggregate:      result.IsAggregate,
	}
}

func neutralRecordsFromSObjects(records []models.SObject) []neutralcontract.Record {
	out := make([]neutralcontract.Record, 0, len(records))
	for _, record := range records {
		out = append(out, neutralRecordFromSObject(record, true))
	}
	return out
}

func neutralRecordFromSObject(record models.SObject, includeObjectType bool) neutralcontract.Record {
	copied := make(neutralcontract.Record, len(record))
	for key, value := range record {
		if models.IsRecordMetadataKey(key) {
			continue
		}
		copied[key] = neutralRecordValue(value)
	}
	if includeObjectType {
		if objectType := models.RecordObjectType(record); objectType != "" {
			copied[models.ObjectTypeMetadataKey] = objectType
		}
	}
	return copied
}

func neutralRecordValue(value interface{}) interface{} {
	switch v := value.(type) {
	case models.SObject:
		return neutralRecordFromSObject(v, true)
	case neutralcontract.Record:
		return neutralContractRecordFromMap(map[string]interface{}(v))
	case map[string]interface{}:
		if models.IsNeutralRecordMap(v) {
			return neutralContractRecordFromMap(v)
		}
		return neutralPlainMap(v)
	case []models.SObject:
		records := make([]neutralcontract.Record, 0, len(v))
		for _, record := range v {
			records = append(records, neutralRecordFromSObject(record, true))
		}
		return records
	case []neutralcontract.Record:
		records := make([]neutralcontract.Record, 0, len(v))
		for _, record := range v {
			records = append(records, neutralContractRecordFromMap(map[string]interface{}(record)))
		}
		return records
	case []map[string]interface{}:
		values := make([]interface{}, 0, len(v))
		for _, item := range v {
			values = append(values, neutralRecordValue(item))
		}
		return values
	case []interface{}:
		values := make([]interface{}, 0, len(v))
		for _, item := range v {
			values = append(values, neutralRecordValue(item))
		}
		return values
	default:
		return value
	}
}

func neutralPlainMap(source map[string]interface{}) map[string]interface{} {
	copied := make(map[string]interface{}, len(source))
	for key, value := range source {
		copied[key] = neutralRecordValue(value)
	}
	return copied
}

func neutralContractRecordFromMap(source map[string]interface{}) neutralcontract.Record {
	objectType, _ := source[models.ObjectTypeMetadataKey].(string)
	copied := make(neutralcontract.Record, len(source))
	for key, value := range source {
		if models.IsRecordMetadataKey(key) {
			continue
		}
		copied[key] = neutralRecordValue(value)
	}
	if objectType != "" {
		copied[models.ObjectTypeMetadataKey] = objectType
	}
	return copied
}

// executeSubQueries resolves child relationship subqueries and attaches results to parent records.
// For each subquery, it finds the child object and FK field by scanning all schemas for reference
// fields that point to the parent object and whose RelationshipName matches.
func (qs *QueryService) executeSubQueries(
	ctx context.Context,
	parentRecords []models.SObject,
	parentSchema *models.ObjectMetadata,
	subQueries []models.SubQueryRequest,
	currentUser *models.UserSession,
	scanAll bool,
) ([]models.SObject, error) {
	if len(subQueries) == 0 {
		return parentRecords, nil
	}

	// Collect parent IDs
	parentIDs := make([]string, 0, len(parentRecords))
	parentIDSet := make(map[string]struct{}, len(parentRecords))
	for _, rec := range parentRecords {
		id, ok := rec[constants.FieldID].(string)
		if !ok || strings.TrimSpace(id) == "" {
			return nil, pkgErrors.NewInternalError("subquery parent row is missing its authoritative record identifier", nil)
		}
		parentIDs = append(parentIDs, id)
		parentIDSet[id] = struct{}{}
	}

	allSchemas := qs.metadata.GetSchemas(ctx)

	for _, sq := range subQueries {
		// Authoritative SoT: _System_Field.ChildRelationshipName via the
		// shared resolver — same path used by sobject_describe.go and
		// crt_synthesizer.findChildRelationship (issue #153). Non-matched
		// cases (A/C/D in the F1 truth table) fail closed here. A requested
		// relationship is either executed and attached (including an explicit
		// empty slice) or the complete query fails; omission is never used as an
		// execution-error carrier.
		resolution := ResolveChildRelationshipAuthoritative(allSchemas, parentSchema.APIName, sq.RelationshipName)
		switch resolution.Status {
		case ChildRelStatusMatched:
		case ChildRelStatusNoCandidates, ChildRelStatusNoMatch:
			return nil, pkgErrors.NewValidationError(
				"SubQuery.RelationshipName",
				fmt.Sprintf("child relationship %q is not declared on object %q", sq.RelationshipName, parentSchema.APIName),
			)
		case ChildRelStatusPackReapplyRequired:
			return nil, pkgErrors.NewInternalError("child relationship metadata is incomplete", nil)
		default:
			return nil, pkgErrors.NewInternalError("child relationship resolver returned an unknown status", nil)
		}
		childObj := resolution.ChildObjectAPIName
		fkField := resolution.FKFieldAPIName

		// Check permissions on child object. Scoped internal read uses the same
		// generic helper as the top-level query path.
		if err := qs.checkObjectReadAccess(ctx, childObj, currentUser); err != nil {
			return nil, err
		}

		childSchema := qs.metadata.GetSchema(ctx, childObj)
		if childSchema == nil {
			return nil, pkgErrors.NewInternalError("resolved child relationship object is absent from the bound metadata view", nil)
		}

		parentIDValues := make([]interface{}, 0, len(parentIDs))
		for _, id := range parentIDs {
			parentIDValues = append(parentIDValues, id)
		}
		criteria := []models.QueryCriterion{{
			Field: fkField,
			Op:    "IN",
			Val:   parentIDValues,
		}}
		if sq.FilterExpr != "" {
			if err := qs.validateFormulaReadClause(ctx, childObj, childSchema, sq.FilterExpr, "SubQuery.FilterExpr", currentUser); err != nil {
				return parentRecords, err
			}
		}
		for _, criterion := range sq.Criteria {
			if err := qs.validateClauseFieldReadable(ctx, childObj, childSchema, criterion.Field, "SubQuery.Criteria", currentUser); err != nil {
				return parentRecords, err
			}
			criteria = append(criteria, criterion)
		}
		for _, sort := range sq.OrderBy {
			if err := qs.validateClauseFieldReadable(ctx, childObj, childSchema, sort.Field, "SubQuery.OrderBy", currentUser); err != nil {
				return parentRecords, err
			}
		}

		childReq := models.QueryRequest{
			ObjectAPIName: childObj,
			FilterExpr:    sq.FilterExpr,
			Criteria:      criteria,
			SelectFields:  sq.SelectFields,
			OrderBy:       sq.OrderBy,
			ScanAll:       scanAll,
		}
		requestedChildFields := append([]string(nil), sq.SelectFields...)
		// Propagate limit to bound the child record fetch.
		// Apply an absolute ceiling of 10 000 to prevent runaway queries
		// when the parent set is large (e.g. 2 000 parents × 200 = 400 000).
		const maxChildLimit = 10000
		if sq.Limit > 0 {
			childReq.Limit = sq.Limit * len(parentIDs)
		} else {
			// Default safety cap: 200 child records per parent
			childReq.Limit = 200 * len(parentIDs)
		}
		if childReq.Limit > maxChildLimit {
			childReq.Limit = maxChildLimit
		}
		// Ensure the FK field and Id are selected so we can group results
		hasFk := false
		hasId := false
		for _, f := range childReq.SelectFields {
			if strings.EqualFold(f, fkField) {
				hasFk = true
			}
			if strings.EqualFold(f, constants.FieldID) {
				hasId = true
			}
		}
		if !hasFk {
			childReq.SelectFields = append(childReq.SelectFields, fkField)
		}
		if !hasId {
			childReq.SelectFields = append([]string{constants.FieldID}, childReq.SelectFields...)
		}

		// Build visible fields for the child query.
		childVisibleFields, err := qs.visibleFieldsForQuery(ctx, childObj, childSchema, currentUser)
		if err != nil {
			return parentRecords, err
		}

		childRLSResult := qs.readRLSResult(ctx, childSchema, childObj, currentUser)

		if err := validateQuerySelections(childSchema, requestedChildFields, func(targetAPIName string) *models.ObjectMetadata {
			if qs.metadata == nil {
				return nil
			}
			return qs.metadata.GetSchema(ctx, targetAPIName)
		}); err != nil {
			return parentRecords, err
		}

		queryFields := buildQueryFields(childSchema, childReq, childVisibleFields)
		idField := models.FindField(childSchema, constants.FieldID)
		foreignKeyField := models.FindField(childSchema, fkField)
		if idField == nil || foreignKeyField == nil ||
			qs.fieldReadBlockedWithSchema(ctx, childObj, childSchema, *idField) ||
			qs.fieldReadBlockedWithSchema(ctx, childObj, childSchema, *foreignKeyField) {
			return nil, pkgErrors.NewPermissionError("read", childObj)
		}
		queryFields, err = appendSubqueryExecutionField(queryFields, childSchema, constants.FieldID)
		if err != nil {
			return nil, err
		}
		queryFields, err = appendSubqueryExecutionField(queryFields, childSchema, fkField)
		if err != nil {
			return nil, err
		}
		// An empty root page still has to validate every requested relationship,
		// permission, clause, and projection dependency. Once those checks pass,
		// there is no child work to execute and no relationship value to attach.
		if len(parentIDs) == 0 {
			continue
		}
		childRepoReq := childReq
		childRepoReq.SelectFields = queryFields

		childResult, err := qs.repo.Find(ctx, childSchema, childRepoReq, childRLSResult, queryFields)
		if err != nil {
			return nil, fmt.Errorf("execute child relationship %q: %w", sq.RelationshipName, err)
		}
		if childResult == nil {
			return nil, pkgErrors.NewInternalError("child relationship repository returned no result", nil)
		}
		childResult.Records, err = qs.hydrateVirtualFields(ctx, childResult.Records, childSchema, queryFields, currentUser, parseDotNotationChildren(childSchema, requestedChildFields))
		if err != nil {
			return parentRecords, err
		}
		childProjectionFields := requestedChildFields
		if len(childProjectionFields) == 0 {
			childProjectionFields = childVisibleFields
		}
		childResult.Records = projectVirtualCompoundFields(childResult.Records, childSchema, childProjectionFields)

		// Group child records by parent ID
		grouped := make(map[string][]models.SObject)
		for _, childRec := range childResult.Records {
			// A child row becomes a neutral nested-record carrier at this point.
			// Preserve the object identity already resolved from authoritative
			// relationship metadata so every downstream neutral projection can
			// apply the child's field types without re-resolving the relationship.
			models.StampObjectType(childRec, childObj)
			if models.RecordObjectType(childRec) != childObj {
				return parentRecords, fmt.Errorf("subquery %q returned record type %q, want %q", sq.RelationshipName, models.RecordObjectType(childRec), childObj)
			}
			parentID, ok := childRec[fkField].(string)
			if !ok || strings.TrimSpace(parentID) == "" {
				return nil, pkgErrors.NewInternalError("child relationship row is missing its grouping identifier", nil)
			}
			if _, requested := parentIDSet[parentID]; !requested {
				return nil, pkgErrors.NewInternalError("child relationship repository returned a row outside the requested parent set", nil)
			}
			grouped[parentID] = append(grouped[parentID], childRec)
		}
		childResult.Records = projectExplicitRelationshipSelections(childSchema, childResult.Records, requestedChildFields)
		childResult.Records = projectSubqueryGroupingFields(childResult.Records, fkField, requestedChildFields, childVisibleFields)

		// Apply per-parent limit if specified
		if sq.Limit > 0 {
			for pid, recs := range grouped {
				if len(recs) > sq.Limit {
					grouped[pid] = recs[:sq.Limit]
				}
			}
		}

		// Attach to parent records
		for i := range parentRecords {
			parentID := parentIDs[i]
			childRecs := grouped[parentID]
			if childRecs == nil {
				childRecs = []models.SObject{}
			}
			// App-blind L0 attaches the NEUTRAL child-records slice under the
			// relationship name and does NOT mint the SF child-subquery envelope
			// {totalSize, done, records}. That SF wire shape is L1-owned: the
			// dialect (dialect-sf-crm query_projection.go wrapChildSubqueryEnvelope)
			// re-wraps this []models.SObject slice into the envelope, computing
			// totalSize = len and done = true (subqueries were always done: true,
			// so no information is lost). Empty subqueries emit an empty slice.
			parentRecords[i][sq.RelationshipName] = childRecs
		}
	}

	return parentRecords, nil
}

func appendSubqueryExecutionField(fields []string, schema *models.ObjectMetadata, fieldAPIName string) ([]string, error) {
	field := models.FindField(schema, fieldAPIName)
	if field == nil || field.IsVirtual {
		return nil, pkgErrors.NewInternalError("child relationship execution dependency is absent from metadata", nil)
	}
	return appendLookupDisplayField(fields, field.APIName), nil
}

func querySObjectString(value interface{}) string {
	switch v := value.(type) {
	case nil:
		return ""
	case string:
		return strings.TrimSpace(v)
	case []byte:
		return strings.TrimSpace(string(v))
	default:
		return strings.TrimSpace(fmt.Sprint(v))
	}
}

// QueryWithFilter executes a query with a formula expression filter
func (qs *QueryService) QueryWithFilter(
	ctx context.Context,
	objectName string,
	filterExpr string,
	currentUser *models.UserSession,
	orderBy string,
	orderDirection string,
	limit int,
) (*models.QueryResult, error) {
	req := queryspec.QuerySpec{
		ObjectAPIName: objectName,
		Filter:        queryspec.FormulaPredicate(filterExpr),
		SortField:     orderBy,
		SortDirection: orderDirection,
		Limit:         limit,
	}
	return qs.QueryWithSpec(ctx, req, currentUser)
}

// SearchWithSpec executes a single-object search through the neutral analytics
// contract. The legacy SearchSingleObject helper remains for internal call
// sites while they migrate away from raw filter expressions.
func (qs *QueryService) SearchWithSpec(ctx context.Context, req queryspec.SearchSpec, currentUser *models.UserSession) ([]neutralcontract.Record, error) {
	filterExpr, err := lowerQueryExpressionPredicate(req.Filter)
	if err != nil {
		return nil, err
	}
	results, err := qs.searchObjectWithOptions(
		ctx,
		req.ObjectAPIName,
		req.Term,
		append([]string(nil), req.SearchFields...),
		append([]string(nil), req.ReturnFields...),
		req.Limit,
		req.Offset,
		req.SortField,
		req.SortDirection,
		filterExpr,
		currentUser,
		ports.SearchOptions{Prefix: req.Prefix, ForceBaseTable: req.ForceBaseTable},
	)
	if err != nil {
		return nil, err
	}
	return neutralRecordsFromSObjects(results), nil
}

// SearchSingleObject searches a specific object with filtering and pagination.
// It uses the default full-search match (case-folded substring, formula-view
// read target).
func (qs *QueryService) SearchSingleObject(
	ctx context.Context,
	objectName string,
	term string,
	searchFields []string,
	returnFields []string,
	limit int,
	offset int,
	orderBy string,
	orderDir string,
	filterExpr string,
	currentUser *models.UserSession,
) ([]models.SObject, error) {
	return qs.searchObjectWithOptions(ctx, objectName, term, searchFields, returnFields, limit, offset, orderBy, orderDir, filterExpr, currentUser, ports.SearchOptions{})
}

// SearchSuggestions runs the name-prefix autocomplete search behind
// GET /search/suggestions. It reuses the same FLS/readable-field pipeline as
// SearchSingleObject but matches an anchored prefix against index-eligible
// physical name columns and reads the object's base table (never the formula
// view), since no formula field participates in a name-only suggestion match.
//
// It overfetches one extra row (limit+1) so the handler can report
// hasMoreResults reliably: the repository applies an exact LIMIT with no
// overfetch of its own, so "exactly limit matches, no more" would otherwise be
// indistinguishable from "truncated". The caller is responsible for trimming
// the returned slice back to its display limit.
func (qs *QueryService) SearchSuggestions(
	ctx context.Context,
	objectName string,
	term string,
	searchFields []string,
	returnFields []string,
	limit int,
	currentUser *models.UserSession,
) ([]models.SObject, error) {
	overfetch := limit
	if overfetch > 0 {
		overfetch = limit + 1
	}
	return qs.searchObjectWithOptions(ctx, objectName, term, searchFields, returnFields, overfetch, 0, "", "", "", currentUser, ports.SearchOptions{Prefix: true, ForceBaseTable: true})
}

func (qs *QueryService) searchObjectWithOptions(
	ctx context.Context,
	objectName string,
	term string,
	searchFields []string,
	returnFields []string,
	limit int,
	offset int,
	orderBy string,
	orderDir string,
	filterExpr string,
	currentUser *models.UserSession,
	opts ports.SearchOptions,
) ([]models.SObject, error) {
	funcStart := time.Now()
	defer func() {
		zlog.Debug().
			Dur("func_latency_ms", time.Since(funcStart)).
			Str("object_name", objectName).
			Int("term_length", len(term)).
			Int("limit", limit).
			Bool("prefix", opts.Prefix).
			Msg("[SEARCH_METRIC] searchObjectWithOptions function latency")
	}()

	if err := qs.checkObjectReadAccess(ctx, objectName, currentUser); err != nil {
		return []models.SObject{}, nil
	}

	schema := qs.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return nil, pkgErrors.NewNotFoundError("Object", objectName)
	}

	// 1. Determine Search Fields (Fields to check against 'term')
	// 1. Strict Search Fields: We rely on the caller (Handler) to provide fields (expanded from scope).
	// If none provided, we search nothing.
	readableFields, err := qs.readableSearchFields(ctx, objectName, schema, currentUser)
	if err != nil {
		return nil, err
	}
	finalSearchFields := filterReadableFieldList(schema, searchFields, readableFields)

	if len(finalSearchFields) == 0 {
		return []models.SObject{}, nil
	}

	// 2. Determine Select Fields (Fields to return)
	requestedReturnFields := returnFields
	if len(requestedReturnFields) == 0 {
		requestedReturnFields = []string{constants.FieldID}

		var nameField string
		for _, f := range schema.Fields {
			if f.IsNameField {
				nameField = f.APIName
				break
			}
		}

		if nameField != "" {
			requestedReturnFields = append(requestedReturnFields, nameField)
		}
	} else {
		// Ensure Id is always selected
		hasId := false
		for _, f := range requestedReturnFields {
			if strings.EqualFold(f, constants.FieldID) {
				hasId = true
				break
			}
		}
		if !hasId {
			requestedReturnFields = append([]string{constants.FieldID}, requestedReturnFields...)
		}
	}
	requestedReturnFields = filterReadableFieldList(schema, requestedReturnFields, readableFields)
	fieldsToSelect := expandCompoundReturnFields(schema, requestedReturnFields)

	if limit <= 0 {
		limit = 500
	}

	rlsResult := qs.readRLSResult(ctx, schema, objectName, currentUser)

	// Delegate to Repository
	results, err := qs.repo.Search(ctx, schema, term, finalSearchFields, fieldsToSelect, limit, offset, orderBy, orderDir, filterExpr, rlsResult, opts)
	if err != nil {
		return []models.SObject{}, err
	}

	results = projectVirtualCompoundFields(results, schema, requestedReturnFields)
	return results, nil
}

func (qs *QueryService) readableSearchFields(ctx context.Context, objectName string, schema *models.ObjectMetadata, currentUser *models.UserSession) (map[string]string, error) {
	readable := make(map[string]string)
	add := func(apiName string) {
		field := models.FindField(schema, apiName)
		if field == nil {
			return
		}
		readable[strings.ToLower(field.APIName)] = field.APIName
	}

	systemFields, err := qs.systemFieldsForQuery(ctx, objectName, schema, currentUser)
	if err != nil {
		return nil, err
	}
	for _, apiName := range systemFields {
		add(apiName)
	}

	for _, field := range schema.Fields {
		if field.IsVirtual && field.VirtualKind != constants.VirtualFieldCompound {
			continue
		}
		if qs.fieldReadBlockedWithSchema(ctx, objectName, schema, field) {
			continue
		}
		readable, err := qs.fieldReadableForQueryWithSchema(ctx, objectName, schema, field, currentUser)
		if err != nil {
			return nil, err
		}
		if readable {
			add(field.APIName)
		}
	}

	return readable, nil
}

func filterReadableFieldList(schema *models.ObjectMetadata, fields []string, readable map[string]string) []string {
	if schema == nil || len(fields) == 0 || len(readable) == 0 {
		return nil
	}

	filtered := make([]string, 0, len(fields))
	seen := make(map[string]struct{}, len(fields))
	for _, apiName := range fields {
		field := models.FindField(schema, apiName)
		if field == nil {
			continue
		}
		canonical, ok := readable[strings.ToLower(field.APIName)]
		if !ok {
			continue
		}
		key := strings.ToLower(canonical)
		if _, exists := seen[key]; exists {
			continue
		}
		seen[key] = struct{}{}
		filtered = append(filtered, canonical)
	}
	return filtered
}

func expandCompoundReturnFields(schema *models.ObjectMetadata, fields []string) []string {
	if schema == nil || len(fields) == 0 {
		return fields
	}

	expanded := make([]string, 0, len(fields))
	seen := make(map[string]struct{}, len(fields))
	add := func(apiName string) {
		if apiName == "" {
			return
		}
		if _, ok := seen[apiName]; ok {
			return
		}
		seen[apiName] = struct{}{}
		expanded = append(expanded, apiName)
	}

	for _, field := range fields {
		if isVirtualCompoundField(schema, field) {
			for _, member := range schemautil.CompoundMemberNames(schema, field) {
				add(member)
			}
			continue
		}
		add(field)
		for _, member := range schemautil.CompoundMemberNames(schema, field) {
			add(member)
		}
	}

	return expanded
}

// GlobalSearch searches across all objects (Parallel Execution)
func (qs *QueryService) GlobalSearch(ctx context.Context, term string, currentUser *models.UserSession) ([]models.SearchResult, error) {
	globalStart := time.Now()
	defer func() {
		zlog.Debug().
			Dur("total_latency_ms", time.Since(globalStart)).
			Int("term_length", len(term)).
			Msg("[SEARCH_METRIC] GlobalSearch total latency")
	}()

	schemas := qs.metadata.GetSchemas(ctx)
	results := make([]models.SearchResult, 0)
	var mu sync.Mutex

	// Filter searchable schemas first
	var searchableSchemas []*models.ObjectMetadata
	for _, schema := range schemas {
		if searchmetadata.IsUserSearchableObject(schema) {
			searchableSchemas = append(searchableSchemas, schema)
		}
	}

	zlog.Debug().
		Int("searchable_object_count", len(searchableSchemas)).
		Int("term_length", len(term)).
		Msg("[SEARCH_METRIC] GlobalSearch will search N objects")

	// Use errgroup for bounded concurrency
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(10)

	for _, schema := range searchableSchemas {
		s := schema // capture loop variable
		g.Go(func() error {
			objectStart := time.Now()
			defer func() {
				zlog.Debug().
					Dur("object_latency_ms", time.Since(objectStart)).
					Str("object_name", s.APIName).
					Int("term_length", len(term)).
					Msg("[SEARCH_METRIC] SearchSingleObject per-object latency")
			}()

			// Check read permission, including scoped standard-action access.
			if err := qs.checkObjectReadAccess(ctx, s.APIName, currentUser); err != nil {
				return nil
			}

			// Push down limit: Global search only shows top 5 per object
			matches, err := qs.SearchSingleObject(ctx, s.APIName, term, nil, nil, 5, 0, "", "", "", currentUser)
			if err != nil {
				qs.logger.Warn(fmt.Sprintf("Search failed for %s: %v", s.APIName, err))
				return nil
			}

			if len(matches) > 0 {
				mu.Lock()
				results = append(results, models.SearchResult{
					ObjectAPIName: s.APIName,
					ObjectLabel:   s.PluralLabel,
					Icon:          s.Icon,
					Matches:       matches,
				})
				mu.Unlock()
			}
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}

	zlog.Info().
		Int("results_object_count", len(results)).
		Int("term_length", len(term)).
		Msg("[SEARCH_METRIC] GlobalSearch completed")

	return results, nil
}

func (qs *QueryService) GlobalSearchWithSpec(ctx context.Context, req queryspec.GlobalSearchSpec, currentUser *models.UserSession) ([]neutralcontract.SearchResult, error) {
	results, err := qs.GlobalSearch(ctx, req.Term, currentUser)
	if err != nil {
		return nil, err
	}
	out := make([]neutralcontract.SearchResult, 0, len(results))
	for _, result := range results {
		out = append(out, neutralcontract.SearchResult{
			ObjectLabel:   result.ObjectLabel,
			ObjectAPIName: result.ObjectAPIName,
			Icon:          result.Icon,
			Matches:       neutralRecordsFromSObjects(result.Matches),
		})
	}
	return out, nil
}

// RunAnalytics executes an analytics query
func (qs *QueryService) RunAnalytics(ctx context.Context, analyticsQuery models.AnalyticsQuery, currentUser *models.UserSession) (interface{}, error) {
	return qs.runAnalytics(ctx, analyticsQuery, currentUser, nil)
}

func (qs *QueryService) runAnalytics(ctx context.Context, analyticsQuery models.AnalyticsQuery, currentUser *models.UserSession, memberships []ports.QueryCorrelatedMembershipPlan) (interface{}, error) {
	objectName := analyticsQuery.ObjectAPIName
	bypassFieldVisibility := scopedReadBypassesFieldVisibility(ctx, objectName)

	if err := qs.checkObjectReadAccess(ctx, objectName, currentUser); err != nil {
		return nil, err
	}

	schema := qs.metadata.GetSchema(ctx, objectName)
	if schema == nil {
		return nil, pkgErrors.NewNotFoundError("Object", objectName)
	}

	// Aggregate operands are explicit selections too. Validate their schema
	// identity before FLS fallback or storage can treat an unknown name as a
	// permitted column. COUNT() contributes no operand.
	aggregateFields := analyticsAggregateReadFields(analyticsQuery)
	if err := validateQuerySelections(schema, aggregateFields, func(targetAPIName string) *models.ObjectMetadata {
		return qs.metadata.GetSchema(ctx, targetAPIName)
	}); err != nil {
		return nil, err
	}

	// FLS Check: Ensure user has visibility to aggregated fields.
	for _, field := range aggregateFields {
		fieldMetadata := models.FindField(schema, field)
		if bypassFieldVisibility {
			if fieldMetadata != nil && qs.fieldReadBlockedWithSchema(ctx, objectName, schema, *fieldMetadata) {
				return nil, pkgErrors.NewPermissionError("read field "+field+" on", objectName)
			}
		} else {
			var visible bool
			var err error
			if fieldMetadata != nil {
				visible, err = qs.checkFieldVisibilityWithSchema(ctx, objectName, schema, fieldMetadata, currentUser)
			} else {
				visible, err = qs.permissions.CheckFieldVisibilityWithUser(ctx, objectName, field, currentUser)
			}
			if err != nil {
				return nil, err
			}
			if !visible {
				return nil, pkgErrors.NewPermissionError("read field "+field+" on", objectName)
			}
		}
	}
	parentFilterPaths, err := qs.validateAnalyticsReadClauses(ctx, objectName, schema, analyticsQuery, currentUser)
	if err != nil {
		return nil, err
	}
	analyticsQuery.ParentFilterPaths = parentFilterPaths

	rlsResult := qs.readRLSResult(ctx, schema, objectName, currentUser)

	// Delegate to Repository (Pass Schema for Alias Resolution)
	var val interface{}
	if len(memberships) > 0 {
		correlatedRepo, ok := qs.repo.(ports.CorrelatedMembershipAnalyticsRepository)
		if !ok {
			return nil, pkgErrors.NewInternalError("correlated membership aggregate executor is not configured", nil)
		}
		val, err = correlatedRepo.RunAnalyticsWithCorrelatedMemberships(ctx, schema, analyticsQuery, rlsResult, memberships)
	} else {
		val, err = qs.repo.RunAnalytics(ctx, schema, analyticsQuery, rlsResult)
	}
	if err != nil {
		return nil, err
	}

	if analyticsQuery.Operation == constants.AnalyticsOperationGroupBy {
		results, ok := val.([]models.SObject)
		if !ok {
			return val, nil
		}

		// Check if group_by field is a Lookup - if so, hydrate names
		if analyticsQuery.GroupBy != nil {
			groupByField := *analyticsQuery.GroupBy
			results, err = qs.hydrateAnalyticsGroupNames(ctx, results, schema, groupByField, currentUser)
			if err != nil {
				return nil, err
			}
		}

		return results, nil
	}

	// val is scalar (float64, int64, etc)
	return val, nil
}

func analyticsAggregateReadFields(q models.AnalyticsQuery) []string {
	if len(q.Aggregates) == 0 {
		if q.Field == nil || strings.TrimSpace(*q.Field) == "" {
			return nil
		}
		return []string{*q.Field}
	}
	fields := make([]string, 0, len(q.Aggregates))
	seen := map[string]struct{}{}
	for _, aggregate := range q.Aggregates {
		field := strings.TrimSpace(aggregate.Field)
		if field == "" {
			continue
		}
		key := strings.ToLower(field)
		if _, ok := seen[key]; ok {
			continue
		}
		fields = append(fields, field)
		seen[key] = struct{}{}
	}
	return fields
}

// hydrateAnalyticsGroupNames resolves lookup field IDs to human-readable names for analytics group_by results
func (qs *QueryService) hydrateAnalyticsGroupNames(ctx context.Context, results []models.SObject, schema *models.ObjectMetadata, groupByField string, currentUser *models.UserSession) ([]models.SObject, error) {
	// Find the field definition
	var fieldDef *models.FieldMetadata
	for i := range schema.Fields {
		if strings.EqualFold(schema.Fields[i].APIName, groupByField) {
			fieldDef = &schema.Fields[i]
			break
		}
	}

	// Check if it's a lookup field - MUST check nil first
	if fieldDef == nil {
		return results, nil
	}

	isLookup := strings.EqualFold(string(fieldDef.Type), string(constants.FieldTypeReference))
	if !isLookup {
		return results, nil
	}

	// Get the referenced object from metadata
	refObject := lookupRefObject(*fieldDef)
	if refObject == "" {
		return results, nil
	}

	// Collect unique IDs from results
	ids := make([]string, 0, len(results))
	for _, r := range results {
		if name, ok := r["name"]; ok && name != nil {
			idStr := fmt.Sprintf("%v", name)
			if idStr != "" {
				ids = append(ids, idStr)
			}
		}
	}

	if len(ids) == 0 {
		return results, nil
	}

	// Get the schema for the referenced object to find name and ID fields
	refSchema := qs.metadata.GetSchema(ctx, refObject)
	if refSchema == nil {
		return results, nil
	}

	// Determine ID field from metadata (not hardcoded)
	idField := getLookupIdField(refSchema)

	// Find name field from metadata
	nameField := constants.FieldName
	for _, f := range refSchema.Fields {
		if f.IsNameField {
			nameField = f.APIName
			break
		}
	}
	displayFields, err := qs.lookupDisplayFieldsForRead(ctx, refSchema.APIName, refSchema, currentUser)
	if err != nil {
		return nil, err
	}
	if !utils.ContainsString(displayFields, nameField) {
		return results, nil
	}

	// Build the read-path with the same field set as normal lookup hydration.
	fetchFields := displayFields
	if len(fetchFields) == 0 {
		return results, nil
	}
	bypassUserAccess := scopedReadBypassesUserAccess(ctx, refSchema.APIName)
	requiresRecordAccessCheck := !bypassUserAccess && currentUser != nil && qs.permissions != nil
	if requiresRecordAccessCheck {
		fetchFields = appendRecordAccessInputFields(fetchFields, refSchema)
	}

	// Fetch candidate records by ID and apply record-level access to display-name projection.
	// This keeps aggregate row counts unchanged while suppressing leaked names for unreadable targets.
	lookupResults, err := qs.repo.GetRecordsByIDs(ctx, refSchema, ids, fetchFields)
	if err != nil {
		return results, nil // Fallback to original on error
	}
	if requiresRecordAccessCheck {
		qs.permissions.PrimeReadAccessInputCache(ctx, refSchema.APIName, lookupResults, currentUser)
	}

	// Build ID -> Name map using the metadata-driven ID field
	nameMap := make(map[string]string)
	for _, rec := range lookupResults {
		if requiresRecordAccessCheck && !qs.permissions.CheckRecordAccess(ctx, refSchema, rec, constants.PermRead, currentUser) {
			continue
		}
		id := fmt.Sprintf("%v", rec[idField])
		name := extractLookupDisplayName(refSchema, rec)
		if name == "" {
			name = fmt.Sprintf("%v", rec[nameField])
		}
		if id != "" && id != "<nil>" && name != "" && name != "<nil>" {
			nameMap[id] = name
		}
	}

	// Replace IDs with names
	for i := range results {
		if name, ok := results[i]["name"]; ok && name != nil {
			idStr := fmt.Sprintf("%v", name)
			if resolvedName, found := nameMap[idStr]; found {
				results[i]["name"] = resolvedName
			}
		}
	}

	return results, nil
}

// canonicalizeFieldNames maps SOQL field names (case-insensitive) to their
// canonical PascalCase form from the object schema. Unrecognized names pass through.
func canonicalizeFieldNames(schema *models.ObjectMetadata, fields []string) []string {
	if len(fields) == 0 || schema == nil {
		return fields
	}
	lookup := make(map[string]string, len(schema.Fields))
	for _, f := range schema.Fields {
		lookup[strings.ToLower(f.APIName)] = f.APIName
	}
	out := make([]string, len(fields))
	for i, f := range fields {
		if canonical, ok := lookup[strings.ToLower(f)]; ok {
			out[i] = canonical
		} else {
			out[i] = f
		}
	}
	return out
}

// ExecuteRawSQL executes a raw SQL query with parameters (Validated and Secured)
func (qs *QueryService) ExecuteRawSQL(ctx context.Context, sql string, params []interface{}, user *models.UserSession) ([]models.SObject, error) {
	// Validate and Rewrite SQL (RLS/FLS)
	// This ensures users can only see what they are allowed to see.
	// Note: Raw SQL from QueryService does not use context variables, so pass nil.
	safeSQL, safeParams, err := qs.validator.ValidateAndRewrite(ctx, sql, params, user, nil)
	if err != nil {
		return nil, pkgErrors.NewInternalError("security validation failed", err)
	}

	// Delegate to Repository
	return qs.repo.ExecuteRawSQL(ctx, safeSQL, safeParams)
}

// projectPermissionSetGrantFields hydrates selected SystemPermission virtual
// fields on PermissionSet rows from the _System_PermissionSetPermission
// junction. Uses a single batched grants-lookup — O(N rows + 1 query).
func (qs *QueryService) projectPermissionSetGrantFields(ctx context.Context, records []models.SObject, fields []models.FieldMetadata) error {
	if len(records) == 0 {
		return nil
	}
	if len(fields) == 0 {
		return nil
	}
	ids := make([]string, 0, len(records))
	for _, r := range records {
		if id, ok := r[constants.FieldID].(string); ok && id != "" {
			ids = append(ids, id)
		}
	}
	grantsByID, err := qs.permissions.GetPermissionSetGrantsMany(ctx, ids)
	if err != nil {
		return err
	}
	for _, r := range records {
		id, _ := r[constants.FieldID].(string)
		ProjectPermissionGrantFieldsOntoRecord(r, fields, grantsByID[id])
	}
	return nil
}