package analytics

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

	pkgErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/schemautil"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
)

// ReportService manages report execution
type ReportService struct {
	queryService ports.QueryService
	metadata     MetadataSource
	permissions  PermissionChecker // may be nil in tests
	logger       *logging.Logger
	// virtualSources is an immutable-after-boot registry of app-owned source
	// bindings. It owns dispatch only; metadata and permissions remain the shape
	// and authorization sources of truth.
	virtualSourcesMu sync.RWMutex
	virtualSources   map[string]contract.VirtualReportSourceRegistration
	// reportDialect owns product-specific report definition syntax. The
	// default neutral dialect does not translate Salesforce report keys.
	reportDialect contract.ReportDefinitionDialect
	// crtResolver is the CRT dispatch hook installed by the service wiring
	// layer. Nil in unit tests that do not exercise the CRT path — the
	// executeReport dispatcher short-circuits when it is nil, preserving
	// pre-CRT behavior.
	crtResolver CRTResolver
	// standardObjs auto-routes known object names before the legacy
	// resolveSourceObject string-trim heuristic; may be nil in tests.
	standardObjs *StandardObjectResolver
	// rawDB is a narrow *sql.DB handle used for reading dialect-requested Report
	// columns that are not registered in the metadata registry. Nil in tests —
	// supplemental fields degrade silently.
	rawDB ReportRawDB
}

// ReportServiceOption is the functional-option type for NewReportService.
// Adding new dependencies to ReportService goes through this surface so
// every existing caller — especially tests — keeps compiling without
// update.
type ReportServiceOption func(*ReportService)

// WithCRTResolver installs the CRT dispatch hook. See design §4.1.
func WithCRTResolver(r CRTResolver) ReportServiceOption {
	return func(s *ReportService) { s.crtResolver = r }
}

// WithRawDB installs a raw sql.DB handle used for reading dialect-requested
// Report columns that are not registered in the metadata registry. Tests that
// don't need supplemental report fields can leave this unset.
func WithRawDB(db ReportRawDB) ReportServiceOption {
	return func(s *ReportService) { s.rawDB = db }
}

// ReportRawDB is the narrow *sql.DB surface the report service uses for
// metadata-bypassing reads. Both *sql.DB and *sql.Tx satisfy it.
type ReportRawDB interface {
	QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row
}

// WithStandardObjectResolver installs the auto-scanned standard-object
// resolver consulted by (*ReportService).resolveSourceObject. See §4.9.
func WithStandardObjectResolver(std *StandardObjectResolver) ReportServiceOption {
	return func(s *ReportService) { s.standardObjs = std }
}

const defaultReportExecutionLimit = 2000

// NewReportService creates a new ReportService. permSvc may be nil in tests; when
// nil, the "myTeam" scope degrades to "mine" (no role-hierarchy expansion).
// The variadic ReportServiceOption surface is how new dependencies (CRT
// resolver, standard-object resolver) attach without churning the ~30 test
// call sites that construct the service with the historical 4-argument
// form.
func NewReportService(queryService ports.QueryService, metadataSvc MetadataSource, permSvc PermissionChecker, logger *logging.Logger, opts ...ReportServiceOption) *ReportService {
	s := &ReportService{
		queryService:  queryService,
		metadata:      metadataSvc,
		permissions:   permSvc,
		logger:        logger,
		reportDialect: NeutralReportDefinitionDialect{},
	}
	for _, opt := range opts {
		opt(s)
	}
	// MEDIUM-1: surface the nil-resolver configuration so operators can
	// detect an incomplete wiring quickly (stored CRTs will 404 until a
	// resolver is installed via WithCRTResolver).
	if s.crtResolver == nil && s.logger != nil {
		s.logger.Warn("CRT resolver not wired; CRT reports will 404 for stored types")
	}
	return s
}

// RunReport executes a report by ID with optional filters
func (s *ReportService) RunReport(ctx context.Context, reportID string, user *models.UserSession, runtimeFilters map[string]interface{}) (*models.ReportResult, error) {
	def, err := s.loadReportDefinition(ctx, reportID, user)
	if err != nil {
		return nil, err
	}

	return s.RunReportDefinition(ctx, def, user, runtimeFilters, 0)
}

// ExecuteReportDefinition executes a report definition directly without loading it from persistence.
func (s *ReportService) ExecuteReportDefinition(ctx context.Context, def *models.ReportDefinition, user *models.UserSession, runtimeFilters map[string]interface{}) (*models.ReportResult, error) {
	if def == nil {
		return nil, pkgErrors.NewValidationError("reportMetadata", "report definition is required")
	}
	return s.RunReportDefinition(ctx, def, user, runtimeFilters, 0)
}

// GetReportMetadata returns only the report definition without execution
func (s *ReportService) GetReportMetadata(ctx context.Context, reportID string, user *models.UserSession) (*models.ReportDefinition, error) {
	return s.loadReportDefinition(ctx, reportID, user)
}

// RunReportDefinition executes an in-memory report definition without requiring
// a saved Report record.
func (s *ReportService) RunReportDefinition(ctx context.Context, def *models.ReportDefinition, user *models.UserSession, runtimeFilters map[string]interface{}, limit int) (*models.ReportResult, error) {
	return s.RunReportDefinitionPage(ctx, def, user, runtimeFilters, limit, 0)
}

// RunReportDefinitionPage executes an in-memory report definition with an
// explicit offset. Every supported source must either consume both page bounds
// or fail before execution; callers never receive a silently repeated page.
func (s *ReportService) RunReportDefinitionPage(ctx context.Context, def *models.ReportDefinition, user *models.UserSession, runtimeFilters map[string]interface{}, limit int, offset int) (*models.ReportResult, error) {
	if def == nil {
		return nil, pkgErrors.NewInternalError("report definition is required", nil)
	}
	if offset < 0 {
		return nil, pkgErrors.NewValidationError("offset", "report offset cannot be negative")
	}
	normalized := *def
	normalized.NormalizeBoundaryAliases()
	return s.executeReport(ctx, &normalized, user, runtimeFilters, normalizeReportExecutionLimit(limit), offset)
}

func (s *ReportService) loadReportDefinition(ctx context.Context, reportID string, user *models.UserSession) (*models.ReportDefinition, error) {
	result, err := s.queryService.QueryWithSpec(ctx, queryspec.QuerySpec{
		ObjectAPIName: constants.TableReport,
		Filter:        queryspec.FieldPredicate("Id", "=", reportID),
		Limit:         1,
	}, user)
	if err != nil {
		return nil, pkgErrors.NewInternalError("failed to load report definition", err)
	}
	if len(result.Records) == 0 {
		return nil, pkgErrors.NewNotFoundError("Report", reportID)
	}

	reportRow := result.Records[0]

	for field, value := range s.fetchReportSupplementalFields(ctx, reportID) {
		reportRow[field] = value
	}
	return s.parseReportDefinition(reportRow)
}

func (s *ReportService) fetchReportSupplementalFields(ctx context.Context, reportID string) map[string]interface{} {
	if s.rawDB == nil {
		return nil
	}
	fields := sanitizeReportSupplementalFields(s.reportDefinitionDialect().SupplementalStoredDefinitionFields())
	if len(fields) == 0 {
		return nil
	}
	selects := make([]string, len(fields))
	values := make([]sql.NullString, len(fields))
	dest := make([]interface{}, len(fields))
	for i, field := range fields {
		selects[i] = fmt.Sprintf("`%s`", field)
		dest[i] = &values[i]
	}
	query := fmt.Sprintf("SELECT %s FROM `Report` WHERE `Id` = ? AND `IsDeleted` = 0 LIMIT 1", strings.Join(selects, ", "))
	if err := s.rawDB.QueryRowContext(ctx, query, reportID).Scan(dest...); err != nil {
		return nil
	}
	out := make(map[string]interface{}, len(fields))
	for i, field := range fields {
		if values[i].Valid && values[i].String != "" {
			out[field] = values[i].String
		}
	}
	return out
}

func sanitizeReportSupplementalFields(fields []string) []string {
	out := make([]string, 0, len(fields))
	seen := map[string]struct{}{}
	for _, field := range fields {
		field = strings.TrimSpace(field)
		if field == "" || !isReportSupplementalFieldName(field) {
			continue
		}
		if _, ok := seen[field]; ok {
			continue
		}
		seen[field] = struct{}{}
		out = append(out, field)
	}
	return out
}

func isReportSupplementalFieldName(field string) bool {
	for i := 0; i < len(field); i++ {
		c := field[i]
		if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' {
			continue
		}
		return false
	}
	return true
}

func normalizeReportExecutionLimit(limit int) int {
	if limit <= 0 || limit > defaultReportExecutionLimit {
		return defaultReportExecutionLimit
	}
	return limit
}

func (s *ReportService) parseReportDefinition(row models.SObject) (*models.ReportDefinition, error) {
	return s.reportDefinitionDialect().ParseStoredDefinition(contract.Record(row))
}

func (s *ReportService) reportDefinitionDialect() contract.ReportDefinitionDialect {
	if s == nil {
		return NeutralReportDefinitionDialect{}
	}
	return effectiveReportDefinitionDialect(s.reportDialect)
}

func (s *ReportService) resolveSourceObject(sourceObj string) string {
	var lookup contract.ReportSourceObjectLookup
	if s != nil && s.standardObjs != nil {
		lookup = s.standardObjs.Lookup
	}
	return s.reportDefinitionDialect().ResolveSourceObject(sourceObj, lookup)
}

func (s *ReportService) ResolveReportSourceObject(sourceObj string) string {
	if s == nil {
		return ""
	}
	return s.resolveSourceObject(sourceObj)
}

func (s *ReportService) ResolveReportSourceBaseObject(ctx context.Context, sourceToken string, user *models.UserSession) (string, error) {
	if s == nil {
		return "", nil
	}
	sourceToken = strings.TrimSpace(sourceToken)
	if sourceToken == "" {
		return "", nil
	}
	if tmpl := LookupJoinTemplate(sourceToken); tmpl != nil {
		return strings.TrimSpace(tmpl.PrimaryTable), nil
	}
	registration, ok, err := s.registeredVirtualReportSourceForExecution(ctx, sourceToken, user)
	if err != nil {
		return "", err
	}
	if ok {
		return registration.ParentObjectAPIName, nil
	}
	if s.crtResolver != nil {
		crtTmpl, _, err := s.crtResolver.Resolve(ctx, sourceToken, user)
		if err != nil {
			return "", err
		}
		if crtTmpl != nil {
			return strings.TrimSpace(crtTmpl.PrimaryTable), nil
		}
	}
	return s.resolveSourceObject(sourceToken), nil
}

func (s *ReportService) normalizeReportColumn(column string) string {
	return s.reportDefinitionDialect().NormalizeColumn(column)
}

func (s *ReportService) executeReport(ctx context.Context, def *models.ReportDefinition, user *models.UserSession, runtimeFilters map[string]interface{}, limit int, offset int) (*models.ReportResult, error) {
	sourceToken := reportExecutionSourceToken(def)

	// Check for cross-object JOIN template — dispatch to JOIN path if found.
	if tmpl := LookupJoinTemplate(sourceToken); tmpl != nil {
		if offset > 0 {
			return nil, pkgErrors.NewValidationError("offset", "report source does not support offset pagination")
		}
		return s.executeJoinReport(ctx, def, tmpl, user, runtimeFilters, limit)
	}
	// Exact boot-time registrations own execution only when their token does not
	// collide with a stored custom report type.
	registration, ok, err := s.registeredVirtualReportSourceForExecution(ctx, sourceToken, user)
	if err != nil {
		return nil, err
	}
	if ok {
		return s.executeVirtualReport(ctx, def, registration, user, runtimeFilters, limit, offset)
	}

	// CRT dispatch hook (design §4.1 step 2). When the SourceObject refers
	// to a stored Custom Report Type or one of the synthetic "CustomEntity$…"
	// shapes, the resolver hands back a synthesized *JoinTemplate plus any
	// degradation warnings. Nil template → the resolver declined and we
	// continue down the single-table path.
	if s.crtResolver != nil {
		crtTmpl, warnings, err := s.crtResolver.Resolve(ctx, sourceToken, user)
		if err != nil {
			return nil, err
		}
		if crtTmpl != nil {
			if offset > 0 {
				return nil, pkgErrors.NewValidationError("offset", "report source does not support offset pagination")
			}
			filterCols := make([]string, 0, len(def.Filters))
			for _, f := range def.Filters {
				filterCols = append(filterCols, f.Column)
			}
			decorated := decorateTemplateForCRT(ctx, crtTmpl, def.Columns, filterCols, newCRTMetadataAdapter(s.metadata))
			result, execErr := s.executeJoinReport(ctx, def, decorated, user, runtimeFilters, limit)
			if execErr != nil {
				return nil, execErr
			}
			if len(warnings) > 0 {
				result.Warnings = append(result.Warnings, warnings...)
			}
			return result, nil
		}
	}

	// Normalise SourceObject: SF synced data may store report type strings
	// rather than plain object names. Resolve before querying.
	objectAPIName := s.resolveSourceObject(sourceToken)

	// Resolve product column references to DB field names upfront through the
	// installed report dialect. Filter out cross-object fields that don't exist
	// on the target table.
	var schema *models.ObjectMetadata
	if s.metadata != nil {
		schema = s.metadata.GetSchema(ctx, objectAPIName)
	}
	knownFields := make(map[string]bool)
	if schema != nil {
		for _, f := range schema.Fields {
			knownFields[f.APIName] = true
			if f.PhysicalName != "" {
				knownFields[f.PhysicalName] = true
			}
		}
	}
	resolvedGrouping := make([]string, 0, len(def.Grouping))
	for _, g := range def.Grouping {
		resolved := s.normalizeReportColumn(g)
		if schema == nil || knownFields[resolved] {
			resolvedGrouping = append(resolvedGrouping, resolved)
		}
		// Skip cross-object fields (e.g. ALIAS → Alias on User, not on Opportunity)
	}
	resolvedAggregates := make([]models.ReportAggregate, len(def.Aggregates))
	for i, a := range def.Aggregates {
		resolvedAggregates[i] = models.ReportAggregate{
			Field: s.normalizeReportColumn(a.Field),
			Type:  a.Type,
		}
	}

	// 3. Execution strategy based on the normalized report shape.
	if shouldExecuteGroupedReport(def, resolvedGrouping) {
		// Only use the pure analytics (GROUP BY) path when there are no detail columns.
		// If the report has detail columns, fall through to the multi-group path
		// which fetches actual rows for the table display.
		if len(resolvedGrouping) == 1 && len(def.Columns) == 0 {
			criteria := s.buildCriteria(ctx, def.Filters, objectAPIName, user)
			criteria = s.appendRuntimeFilters(ctx, criteria, objectAPIName, runtimeFilters)
			criteria = s.applyShowMeScope(ctx, criteria, def.Scope, objectAPIName, user)

			analyticQuery := models.AnalyticsQuery{
				ObjectAPIName: objectAPIName,
				Operation:     constants.AnalyticsOperationGroupBy,
				GroupBy:       &resolvedGrouping[0],
				Criteria:      criteria,
				Limit:         limit,
				Offset:        offset,
			}
			if len(resolvedAggregates) > 0 {
				if resolvedAggregates[0].Field != "" {
					analyticQuery.Field = &resolvedAggregates[0].Field
				}
				analyticQuery.AggregateType = resolvedAggregates[0].Type
			}
			res, err := s.queryService.RunAnalytics(ctx, analyticQuery, user)
			if err != nil {
				return nil, err
			}

			rows, ok := res.([]models.SObject)
			if !ok {
				return nil, pkgErrors.NewInternalError("unexpected analytics result format", nil)
			}

			resultDef := *def
			resultDef.DisplayMode = reportDisplayModeForExecution(resultDef.EffectiveDisplayMode(), resolvedGrouping)
			resultDef.NormalizeBoundaryAliases()
			resultDef.Grouping = resolvedGrouping
			resultDef.Aggregates = resolvedAggregates
			return &models.ReportResult{
				ReportMetadata: resultDef,
				Rows:           rows,
			}, nil
		}
	}

	// Default: Tabular (Rows)
	criteria := s.buildCriteria(ctx, def.Filters, objectAPIName, user)
	criteria = s.appendRuntimeFilters(ctx, criteria, objectAPIName, runtimeFilters)
	criteria = s.applyShowMeScope(ctx, criteria, def.Scope, objectAPIName, user)

	// Apply the neutral date filter. Product-specific presets are normalized by
	// the installed dialect before the executor sees concrete bounds.
	if df := def.EffectiveDateFilter(); df != nil && df.Column != "" {
		dateField := s.normalizeReportColumn(df.Column)
		startDate, endDate := s.normalizeDateBoundsForField(ctx, objectAPIName, dateField, df.StartDate, df.EndDate)
		if startDate != "" {
			criteria = append(criteria, models.QueryCriterion{Field: dateField, Op: ">=", Val: startDate})
		}
		if endDate != "" {
			criteria = append(criteria, models.QueryCriterion{Field: dateField, Op: "<=", Val: endDate})
		}
	}

	// Resolve product column references to DB field names for SelectFields.
	// Lookup fields (e.g. AccountId) get the _Name display column in resolvedColumns
	// but the base field in selectFields so the query engine hydrates the name.
	var selectFields []string
	var resolvedColumns []string // Column names for display (matched to row data keys)
	seen := make(map[string]bool)
	if len(def.Columns) > 0 {
		for _, col := range def.Columns {
			resolved := s.normalizeReportColumn(col)
			if seen[resolved] {
				continue
			}
			seen[resolved] = true
			selectFields = append(selectFields, resolved)
			// For lookup/reference fields ending in "Id", display the _Name column
			if strings.HasSuffix(resolved, "Id") && resolved != "Id" {
				resolvedColumns = append(resolvedColumns, resolved+"_Name")
			} else {
				resolvedColumns = append(resolvedColumns, resolved)
			}
		}
	}
	// Also include grouping fields so chart data can be computed from rows
	for _, g := range resolvedGrouping {
		if !seen[g] {
			selectFields = append(selectFields, g)
			seen[g] = true
		}
	}

	// Map to neutral query spec — filters handle volume, 2000 matches SF behavior
	queryReq := queryspec.QuerySpec{
		ObjectAPIName: objectAPIName,
		Limit:         limit,
		Offset:        offset,
		SelectFields:  selectFields, // Only fetch report columns + grouping fields
	}
	var queryCriteria []queryspec.QueryCriterion

	// Give the installed report dialect a chance to combine per-filter
	// fragments into a neutral formula expression. Fall back to the plain
	// Criteria path when the dialect declines.
	// Pre-resolve lookup-field contains/notContain filters into concrete
	// Id IN lists. Without this the formula path produces LIKE against a
	// column that stores IDs, silently matching nothing.
	lookupFrags, lookupWarnings := s.buildLookupFilterFragments(ctx, def.Filters, objectAPIName, user)
	if expr := s.compileReportFilterExpression(ctx, def, criteria, s.newPicklistResolverForService(), objectAPIName, lookupFrags); expr != "" {
		queryReq.Filter = queryspec.FormulaPredicate(expr)
	} else if criteria != nil {
		queryCriteria = make([]queryspec.QueryCriterion, len(criteria))
		for i, c := range criteria {
			queryCriteria[i] = queryspec.QueryCriterion{
				Field: c.Field,
				Op:    c.Op,
				Val:   c.Val,
			}
		}
		queryReq.Filter = queryspec.CriteriaPredicate(queryCriteria)
	}

	var retryWarnings []models.CRTWarning
	retryWarnings = append(retryWarnings, lookupWarnings...)
	queryRes, err := s.queryService.QueryWithSpec(ctx, queryReq, user)
	if err != nil && !def.DegradeOnError {
		return nil, err
	}
	if err != nil {
		// Retry 1: drop SelectFields (unknown column in SELECT)
		if len(selectFields) > 0 {
			droppedFields := make([]string, len(selectFields))
			copy(droppedFields, selectFields)
			queryReq.SelectFields = nil
			resolvedColumns = nil
			queryRes, err = s.queryService.QueryWithSpec(ctx, queryReq, user)
			if err == nil {
				s.logger.Warn(fmt.Sprintf("report retry: dropped SelectFields %v", droppedFields))
				retryWarnings = append(retryWarnings, models.CRTWarning{
					Code:         WarnReportSelectFieldsDropped,
					Message:      fmt.Sprintf("SelectFields dropped on retry: %s", strings.Join(droppedFields, ", ")),
					SourceObject: def.SourceObject,
				})
			}
		}
		// Retry 2: progressively remove criteria that reference unknown columns
		for err != nil && len(queryCriteria) > 0 {
			var colErr *pkgErrors.ColumnNotFoundError
			if errors.As(err, &colErr) && colErr.ColumnName != "" {
				// Typed error: remove the criterion whose field matches the column name
				removed := false
				for i, c := range queryCriteria {
					if c.Field == colErr.ColumnName || strings.EqualFold(c.Field, colErr.ColumnName) {
						s.logger.Warn(fmt.Sprintf("report retry: dropped criterion %q (column not found)", c.Field))
						retryWarnings = append(retryWarnings, models.CRTWarning{
							Code:         WarnReportCriterionDropped,
							Message:      fmt.Sprintf("Criterion on field %q dropped: column not found", c.Field),
							SourceObject: c.Field,
						})
						queryCriteria = append(queryCriteria[:i], queryCriteria[i+1:]...)
						queryReq.Filter = queryspec.CriteriaPredicate(queryCriteria)
						removed = true
						break
					}
				}
				if !removed {
					// Column name from error didn't match any criterion — drop all
					s.logger.Warn(fmt.Sprintf("report retry: dropped all %d criteria (column %q not in criteria list)", len(queryCriteria), colErr.ColumnName))
					retryWarnings = append(retryWarnings, models.CRTWarning{
						Code:         WarnReportAllCriteriaDropped,
						Message:      fmt.Sprintf("All %d criteria dropped: column %q not matched to any criterion", len(queryCriteria), colErr.ColumnName),
						SourceObject: def.SourceObject,
					})
					queryCriteria = nil
					queryReq.Filter = queryspec.Predicate{}
				}
			} else {
				// Untyped error — can't identify which criterion failed, drop all
				s.logger.Warn(fmt.Sprintf("report retry: dropped all %d criteria (untyped error)", len(queryCriteria)))
				retryWarnings = append(retryWarnings, models.CRTWarning{
					Code:         WarnReportAllCriteriaDropped,
					Message:      fmt.Sprintf("All %d criteria dropped on retry: unrecognized query error", len(queryCriteria)),
					SourceObject: def.SourceObject,
				})
				queryCriteria = nil
				queryReq.Filter = queryspec.Predicate{}
			}
			queryRes, err = s.queryService.QueryWithSpec(ctx, queryReq, user)
		}
		if err != nil {
			return nil, err
		}
	}

	rows := queryRes.Records

	// Replace def fields with resolved names so the frontend can match row keys
	resultDef := *def
	resultDef.DisplayMode = reportDisplayModeForExecution(resultDef.EffectiveDisplayMode(), resolvedGrouping)
	resultDef.NormalizeBoundaryAliases()
	if len(resolvedColumns) > 0 {
		resultDef.Columns = resolvedColumns
	}
	resultDef.Grouping = resolvedGrouping
	resultDef.Aggregates = resolvedAggregates

	// Propagate the unpaginated total so "Total Records" matches SF even
	// when the result set exceeds the execution limit. QueryRepository
	// already ran a COUNT(*) when fetching >limit rows; reading TotalSize
	// here avoids a second round-trip. Falls back to len(rows) when the
	// count wasn't issued (result fit within the page).
	totalCount := queryRes.TotalSize
	if totalCount == 0 {
		totalCount = len(rows)
	}

	return &models.ReportResult{
		ReportMetadata: resultDef,
		Rows:           rows,
		TotalCount:     totalCount,
		Warnings:       retryWarnings,
	}, nil
}

func (s *ReportService) compileReportFilterExpression(ctx context.Context, def *models.ReportDefinition, resolvedCriteria []models.QueryCriterion, resolver *picklistResolver, objectAPIName string, lookupFrags map[int]string) string {
	_ = resolvedCriteria // reserved for future use; scope filter merging goes here
	fragments := s.reportFormulaFragments(ctx, def.Filters, resolver, objectAPIName, lookupFrags)
	expr, ok := s.reportDefinitionDialect().BuildFilterExpression(contract.ReportFilterExpressionRequest{
		Language:      contract.ReportFilterExpressionLanguageFormula,
		FilterLogic:   def.EffectiveFilterLogic(),
		BooleanFilter: def.EffectiveFilterLogic(),
		Filters:       def.Filters,
		Fragments:     fragments,
	})
	if !ok {
		return ""
	}
	return expr.Expression
}

func shouldExecuteGroupedReport(_ *models.ReportDefinition, resolvedGrouping []string) bool {
	return len(resolvedGrouping) > 0
}

func reportDisplayModeForExecution(displayMode string, resolvedGrouping []string) string {
	if len(resolvedGrouping) > 0 {
		return "SUMMARY"
	}
	return displayMode
}

func reportExecutionSourceToken(def *models.ReportDefinition) string {
	if def == nil {
		return ""
	}
	if sourceRef := strings.TrimSpace(def.SourceRef); sourceRef != "" {
		return sourceRef
	}
	return strings.TrimSpace(def.SourceObject)
}

func (s *ReportService) reportFormulaFragments(ctx context.Context, filters []models.ReportFilter, resolver *picklistResolver, objectAPIName string, lookupFrags map[int]string) []contract.ReportFilterExpressionFragment {
	fragments := make([]contract.ReportFilterExpressionFragment, 0, len(filters))
	for i, filter := range filters {
		expr := ""
		if lookupFrags != nil {
			expr = lookupFrags[i]
		}
		if expr == "" {
			expr = reportFilterToFormula(ctx, filter, resolver, objectAPIName, s.normalizeReportColumn)
		}
		if expr == "" {
			continue
		}
		fragments = append(fragments, contract.ReportFilterExpressionFragment{
			Index:      i + 1,
			Expression: expr,
		})
	}
	return fragments
}

// reportFilterToFormula emits a single-predicate formula fragment. Field
// names are resolved through the installed report dialect so product aliases
// line up with the physical column and inherit the same mapper the repository
// applies to its other expressions.
//
// Some report products treat comma-separated equals/notEqual values as an IN/
// NOT IN membership check. The dialect decides when this formula path is active.
// Emit an expr-lang `in` expression so the compiled SQL uses IN(...) and
// matches SF row counts.
func reportFilterToFormula(ctx context.Context, f models.ReportFilter, resolver *picklistResolver, objectAPIName string, normalizeColumn func(string) string) string {
	col := normalizeColumn(f.Column)
	if col == "" {
		return ""
	}
	// Apply picklist label → API value translation so filter values align
	// with what the DB stores. Safe no-op for non-picklist fields and when
	// the resolver or object is not wired.
	resolveOne := func(v string) string {
		if resolver == nil || objectAPIName == "" {
			return v
		}
		return resolver.resolve(ctx, objectAPIName, col, v)
	}
	op := strings.ToLower(f.Operator)
	if (op == "equals" || op == "eq" || op == "=" ||
		op == "notequal" || op == "neq" || op == "!=" || op == "<>") && strings.Contains(f.Value, ",") {
		parts := splitAndTrimMulti(f.Value)
		if len(parts) > 1 {
			quoted := make([]string, len(parts))
			for i, p := range parts {
				quoted[i] = `"` + escapeFormulaString(resolveOne(p)) + `"`
			}
			list := "[" + strings.Join(quoted, ",") + "]"
			if op == "equals" || op == "eq" || op == "=" {
				return fmt.Sprintf(`%s in %s`, col, list)
			}
			return fmt.Sprintf(`!(%s in %s)`, col, list)
		}
	}
	// Multi-value contains: SF treats comma-separated literals as an OR
	// chain of CONTAINS clauses. notContain is the De Morgan dual (AND of
	// NOT CONTAINS).
	if (op == "contains" || op == "notcontain" || op == "doesnotcontain") && strings.Contains(f.Value, ",") {
		parts := splitAndTrimMulti(f.Value)
		if len(parts) > 1 {
			fragments := make([]string, len(parts))
			if op == "contains" {
				for i, p := range parts {
					fragments[i] = fmt.Sprintf(`CONTAINS(%s, "%s")`, col, escapeFormulaString(resolveOne(p)))
				}
				return "(" + strings.Join(fragments, " OR ") + ")"
			}
			for i, p := range parts {
				fragments[i] = fmt.Sprintf(`NOT(CONTAINS(%s, "%s"))`, col, escapeFormulaString(resolveOne(p)))
			}
			return "(" + strings.Join(fragments, " AND ") + ")"
		}
	}
	val := escapeFormulaString(resolveOne(f.Value))
	switch op {
	case "equals", "eq", "=":
		return fmt.Sprintf(`%s = "%s"`, col, val)
	case "notequal", "neq", "!=", "<>":
		return fmt.Sprintf(`%s != "%s"`, col, val)
	case "greaterthan", "gt":
		return fmt.Sprintf(`%s > "%s"`, col, val)
	case "greaterorequal", "gte", ">=":
		return fmt.Sprintf(`%s >= "%s"`, col, val)
	case "lessthan", "lt":
		return fmt.Sprintf(`%s < "%s"`, col, val)
	case "lessorequal", "lte", "<=":
		return fmt.Sprintf(`%s <= "%s"`, col, val)
	case "contains":
		return fmt.Sprintf(`CONTAINS(%s, "%s")`, col, val)
	case "notcontain", "doesnotcontain":
		return fmt.Sprintf(`NOT(CONTAINS(%s, "%s"))`, col, val)
	case "startswith":
		return fmt.Sprintf(`BEGINS(%s, "%s")`, col, val)
	}
	// Unknown operator — fall through to equality so a malformed filter
	// still participates but doesn't nuke the whole expression.
	return fmt.Sprintf(`%s = "%s"`, col, val)
}

func splitAndTrimMulti(s string) []string {
	raw := strings.Split(s, ",")
	out := make([]string, 0, len(raw))
	for _, p := range raw {
		p = strings.TrimSpace(p)
		if p != "" {
			out = append(out, p)
		}
	}
	return out
}

func escapeFormulaString(s string) string {
	// Formula string literals use double quotes; escape embedded quotes and
	// backslashes so the parser doesn't trip on user-supplied filter values.
	s = strings.ReplaceAll(s, `\`, `\\`)
	s = strings.ReplaceAll(s, `"`, `\"`)
	return s
}

// buildLookupFilterFragments pre-resolves each filter that targets a lookup
// field with a string operator, producing a formula-compatible
// `<fk> in [...]` (or `!(<fk> in [...])`) fragment. The map is keyed by
// the filter index (0-based) so callers can substitute the fragment in
// place of the default compiled one.
//
// Filters that aren't lookup-qualified or whose target resolution fails
// are omitted — the caller falls back to reportFilterToFormula.
func (s *ReportService) buildLookupFilterFragments(ctx context.Context, filters []models.ReportFilter, objectAPIName string, user *models.UserSession) (map[int]string, []models.CRTWarning) {
	if s == nil || s.metadata == nil || len(filters) == 0 || objectAPIName == "" {
		return nil, nil
	}
	out := make(map[int]string)
	var warnings []models.CRTWarning
	for i, f := range filters {
		if !isLookupStringFilter(f.Operator) {
			continue
		}
		ids, ok := s.resolveLookupFilterIDs(ctx, objectAPIName, f, user)
		if !ok {
			if s.logger != nil {
				s.logger.Warn(fmt.Sprintf("report lookup filter dropped: column=%s operator=%s object=%s", f.Column, f.Operator, objectAPIName))
			}
			warnings = append(warnings, models.CRTWarning{
				Code:         WarnReportLookupFilterDropped,
				Message:      fmt.Sprintf("Filter on %q (%s) was not applied: lookup field resolution failed", f.Column, f.Operator),
				SourceObject: objectAPIName,
			})
			continue
		}
		fk := s.normalizeReportColumn(f.Column)
		if fk == "" {
			continue
		}
		if len(ids) == 0 {
			// Zero matches on the referenced object. Inject an always-
			// false fragment so the boolean combinator propagates the
			// emptiness correctly (`col = "__no_match__"`).
			out[i] = fmt.Sprintf(`%s = "__no_match__"`, fk)
			continue
		}
		quoted := make([]string, len(ids))
		for j, id := range ids {
			quoted[j] = `"` + escapeFormulaString(id) + `"`
		}
		list := "[" + strings.Join(quoted, ",") + "]"
		op := strings.ToLower(f.Operator)
		if op == "notcontain" || op == "doesnotcontain" {
			out[i] = fmt.Sprintf(`!(%s in %s)`, fk, list)
		} else {
			out[i] = fmt.Sprintf(`%s in %s`, fk, list)
		}
	}
	return out, warnings
}

// isLookupStringFilter reports whether the filter operator is a string
// comparison that, when applied to a reference (lookup) field, should
// actually hit the referenced object's Name column rather than the raw
// ID. Equality operators are excluded because SF stores the ID directly
// and an equals filter on a lookup is already correct.
func isLookupStringFilter(op string) bool {
	switch strings.ToLower(op) {
	case "contains", "notcontain", "doesnotcontain", "startswith":
		return true
	}
	return false
}

// resolveLookupFilterIDs pre-queries the target object of a lookup field
// for Ids whose Name matches the filter pattern(s). Returns (ids, true)
// when the filter column actually is a lookup on objectAPIName; otherwise
// returns (nil, false) and the caller falls through to the normal path.
//
// Multi-value comma-separated values are treated as an OR chain across
// contains clauses — SF's filter UI multi-selects this way.
func (s *ReportService) resolveLookupFilterIDs(ctx context.Context, objectAPIName string, f models.ReportFilter, user *models.UserSession) ([]string, bool) {
	if s.metadata == nil {
		return nil, false
	}
	fk := s.normalizeReportColumn(f.Column)
	if fk == "" {
		return nil, false
	}
	schema := s.metadata.GetSchema(ctx, objectAPIName)
	if schema == nil {
		return nil, false
	}
	var fkField *models.FieldMetadata
	for i := range schema.Fields {
		fld := &schema.Fields[i]
		if strings.EqualFold(fld.APIName, fk) || strings.EqualFold(fld.PhysicalName, fk) {
			fkField = fld
			break
		}
	}
	if fkField == nil {
		return nil, false
	}
	if strings.ToLower(string(fkField.Type)) != "reference" {
		return nil, false
	}
	if len(fkField.ReferenceTo) == 0 {
		return nil, false
	}
	target := fkField.ReferenceTo[0]

	// Build IN-list by querying the target for names that match any of
	// the comma-separated patterns. For `contains` and `startsWith`,
	// wrap each pattern appropriately.
	vals := splitAndTrimMulti(f.Value)
	if len(vals) == 0 {
		vals = []string{f.Value}
	}
	patterns := make([]interface{}, 0, len(vals))
	for _, v := range vals {
		switch strings.ToLower(f.Operator) {
		case "contains", "notcontain", "doesnotcontain":
			patterns = append(patterns, "%"+v+"%")
		case "startswith":
			patterns = append(patterns, v+"%")
		default:
			patterns = append(patterns, "%"+v+"%")
		}
	}
	// Each pattern is ORed. Build a criterion list the repo can consume.
	// The query service doesn't support OR in Criteria, so assemble a
	// FilterExpr using expr-lang's `||`.
	//
	// For notContain/doesNotContain we still use `contains` to find
	// matching Ids, then the caller inverts with NOT IN.
	clauses := make([]string, 0, len(patterns))
	for _, p := range patterns {
		clauses = append(clauses, fmt.Sprintf(`CONTAINS(Name, %q)`, strings.Trim(p.(string), "%")))
	}
	filterExpr := strings.Join(clauses, " || ")

	req := queryspec.QuerySpec{
		ObjectAPIName: target,
		SelectFields:  []string{"Id"},
		Filter:        queryspec.FormulaPredicate(filterExpr),
		Limit:         5000,
	}
	res, err := s.queryService.QueryWithSpec(ctx, req, user)
	if err != nil || res == nil {
		return nil, false
	}
	ids := make([]string, 0, len(res.Records))
	for _, row := range res.Records {
		if id := row.GetString("Id"); id != "" {
			ids = append(ids, id)
		}
	}
	return ids, true
}

// applyShowMeScope appends OwnerId criteria based on the report's "Show me" scope.
//   - "" or "all"       → no filter
//   - "mine"            → OwnerId = current user
//   - "myTeam"          → OwnerId IN (current user + role subordinates)
//
// If the source object's schema does not expose an OwnerId field, scope is
// silently skipped so the query still runs (consistent with SF behavior for
// objects without ownership). If permSvc is nil or role expansion fails,
// "myTeam" gracefully degrades to "mine" so reports never silently leak data.
func (s *ReportService) applyShowMeScope(ctx context.Context, criteria []models.QueryCriterion, scope string, objectAPIName string, user *models.UserSession) []models.QueryCriterion {
	if user == nil {
		return criteria
	}
	normalized := strings.ToLower(strings.TrimSpace(scope))
	if normalized == "" || normalized == "all" {
		return criteria
	}

	// Skip scope filtering when the target object has no OwnerId column.
	if s.metadata != nil {
		if schema := s.metadata.GetSchema(ctx, objectAPIName); schema != nil {
			hasOwner := false
			for _, f := range schema.Fields {
				if f.APIName == "OwnerId" || f.PhysicalName == "OwnerId" {
					hasOwner = true
					break
				}
			}
			if !hasOwner {
				return criteria
			}
		}
	}

	switch normalized {
	case "mine":
		return append(criteria, models.QueryCriterion{Field: "OwnerId", Op: "=", Val: user.ID})
	case "myteam":
		ownerIDs := []string{user.ID}
		if s.permissions != nil && user.RoleID != nil && *user.RoleID != "" {
			subs, err := s.permissions.GetSubordinateUserIDs(ctx, *user.RoleID)
			if err == nil && len(subs) > 0 {
				ownerIDs = append(ownerIDs, subs...)
			}
		}
		if len(ownerIDs) == 1 {
			return append(criteria, models.QueryCriterion{Field: "OwnerId", Op: "=", Val: ownerIDs[0]})
		}
		return append(criteria, models.QueryCriterion{Field: "OwnerId", Op: "IN", Val: ownerIDs})
	}
	return criteria
}

func (s *ReportService) normalizeDateBoundsForField(ctx context.Context, objectAPIName, fieldAPIName, start, end string) (string, string) {
	if !s.isDateTimeField(ctx, objectAPIName, fieldAPIName) {
		return start, end
	}
	if isDateOnlyFilterBound(start) {
		start = start + " 00:00:00"
	}
	if isDateOnlyFilterBound(end) {
		end = end + " 23:59:59"
	}
	return start, end
}

func (s *ReportService) isDateTimeField(ctx context.Context, objectAPIName, fieldAPIName string) bool {
	switch strings.ToLower(strings.TrimSpace(fieldAPIName)) {
	case "createddate", "lastmodifieddate", "systemmodstamp":
		return true
	}

	if s == nil || s.metadata == nil || objectAPIName == "" || fieldAPIName == "" {
		return false
	}

	schema := s.metadata.GetSchema(ctx, objectAPIName)
	if schema == nil {
		return false
	}

	field := schemautil.FindField(schema, fieldAPIName)
	if field == nil {
		return false
	}

	if strings.EqualFold(string(field.Type), string(constants.FieldTypeDatetime)) {
		return true
	}
	return field.LogicalType != nil && strings.EqualFold(string(*field.LogicalType), string(constants.FieldTypeDatetime))
}

func isDateOnlyFilterBound(value string) bool {
	if len(value) != len("2006-01-02") {
		return false
	}
	_, err := time.Parse("2006-01-02", value)
	return err == nil
}

// newPicklistResolverForService builds a picklistResolver backed by the
// ReportService's MetadataService. Returns a resolver that is a no-op when
// s.metadata is nil (matches the existing tolerance of nil metadata across
// report execution — unit tests construct ReportService without metadata).
//
// See issue #476 for why this exists: report filter values arrive as
// user-facing labels ("Free Trial") but the DB stores API values
// ("free_trial"). A single resolver applied at every filter entry point
// fixes every report/dashboard/runtime filter path at once.
func (s *ReportService) newPicklistResolverForService() *picklistResolver {
	if s == nil || s.metadata == nil {
		return newPicklistResolver(nil)
	}
	return newPicklistResolver(func(ctx context.Context, objectAPIName string) *models.ObjectMetadata {
		return s.metadata.GetSchema(ctx, objectAPIName)
	})
}

func (s *ReportService) buildCriteria(ctx context.Context, filters []models.ReportFilter, objectAPIName string, user *models.UserSession) []models.QueryCriterion {
	if len(filters) == 0 {
		return nil
	}

	resolver := s.newPicklistResolverForService()

	criteria := make([]models.QueryCriterion, 0, len(filters))
	for _, f := range filters {
		// Cross-object lookup filter: when the column is a reference field and
		// the operator is a string op like `contains`, report products commonly
		// resolve by joining to the target and filtering on its Name column. Our
		// single-table path has no JOIN surface, so we pre-query the target
		// object for matching Ids and rewrite the filter to
		// `<fk> IN (ids)`. Handles multi-value comma-separated values
		// as an OR over `contains` clauses (SF `contains` semantics).
		if s.metadata != nil && isLookupStringFilter(f.Operator) {
			ids, ok := s.resolveLookupFilterIDs(ctx, objectAPIName, f, user)
			if ok {
				fk := s.normalizeReportColumn(f.Column)
				if len(ids) == 0 {
					// Filter matched zero rows on the referenced object.
					// Inject an unsatisfiable criterion so the report
					// returns nothing (matching SF semantics).
					criteria = append(criteria, models.QueryCriterion{Field: fk, Op: "=", Val: "__no_match__"})
					continue
				}
				vals := make([]interface{}, len(ids))
				for i, id := range ids {
					vals[i] = id
				}
				inOp := "IN"
				if strings.EqualFold(f.Operator, "notContain") || strings.EqualFold(f.Operator, "doesNotContain") || strings.EqualFold(f.Operator, "notEqual") || strings.EqualFold(f.Operator, "neq") {
					inOp = "NOT IN"
				}
				criteria = append(criteria, models.QueryCriterion{Field: fk, Op: inOp, Val: vals})
				continue
			}
		}

		field := s.normalizeReportColumn(f.Column)

		op := "="
		switch f.Operator {
		case "equals", "eq":
			op = "="
		case "notEqual", "neq":
			op = "!="
		case "contains":
			op = "LIKE"
		case "notContain", "doesNotContain":
			op = "NOT LIKE"
		case "greaterThan", "gt":
			op = ">"
		case "greaterOrEqual", "gte":
			op = ">="
		case "lessThan", "lt":
			op = "<"
		case "lessOrEqual", "lte":
			op = "<="
		}

		val := f.Value

		// Convert SF boolean strings to DB integer values
		if strings.EqualFold(val, "true") {
			val = "1"
		} else if strings.EqualFold(val, "false") {
			val = "0"
		}

		// Handle comma-separated multi-value (e.g. "Open,Assigned,Working")
		if (op == "=" || op == "!=") && strings.Contains(val, ",") {
			parts := strings.Split(val, ",")
			vals := make([]interface{}, len(parts))
			for j, p := range parts {
				vals[j] = resolver.resolve(ctx, objectAPIName, field, strings.TrimSpace(p))
			}
			inOp := "IN"
			if op == "!=" {
				inOp = "NOT IN"
			}
			criteria = append(criteria, models.QueryCriterion{Field: field, Op: inOp, Val: vals})
			continue
		}

		// Resolve single-value picklist labels → API values. Non-picklist
		// fields and unmatched values pass through unchanged.
		val = resolver.resolve(ctx, objectAPIName, field, val)

		if f.Operator == "contains" || f.Operator == "notContain" || f.Operator == "doesNotContain" {
			val = fmt.Sprintf("%%%s%%", val)
		}

		criteria = append(criteria, models.QueryCriterion{Field: field, Op: op, Val: val})
	}

	return criteria
}

func (s *ReportService) appendRuntimeFilters(ctx context.Context, criteria []models.QueryCriterion, objectAPIName string, filters map[string]interface{}) []models.QueryCriterion {
	if filters == nil {
		return criteria
	}

	resolver := s.newPicklistResolverForService()

	for k, v := range filters {
		switch k {
		case "__date_range":
			continue
		case "startDate":
			continue
		default:
			valStr := fmt.Sprintf("%v", v)
			// Resolve picklist labels → API values (same rationale as buildCriteria).
			// Runtime filters use the same user-facing labels as definition filters,
			// so they need the same translation. See issue #476.
			valStr = resolver.resolveMulti(ctx, objectAPIName, k, valStr)
			criteria = append(criteria, models.QueryCriterion{
				Field: k,
				Op:    "=",
				Val:   valStr,
			})
		}
	}

	// Handle Date Range
	if dr, ok := filters["__date_range"].(map[string]interface{}); ok {
		field, _ := dr["field"].(string)
		start := dr["start"]
		end := dr["end"]

		if field != "" {
			if start != nil {
				startStr := fmt.Sprintf("%v", start)
				// Handle time.Time formatting if needed, but %v might work for simple ISO or similar if stringer?
				// Usually DB expects 'YYYY-MM-DD...'
				if t, ok := start.(time.Time); ok {
					startStr = t.Format("2006-01-02 15:04:05")
				}
				criteria = append(criteria, models.QueryCriterion{
					Field: field,
					Op:    ">=",
					Val:   startStr,
				})
			}
			if end != nil {
				endStr := fmt.Sprintf("%v", end)
				if t, ok := end.(time.Time); ok {
					endStr = t.Format("2006-01-02 15:04:05")
				}
				criteria = append(criteria, models.QueryCriterion{
					Field: field,
					Op:    "<=",
					Val:   endStr,
				})
			}
		}
	}

	return criteria
}