package persistence

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

	pkgerrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/query"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/formulastorage"
	"github.com/nexuscrm/shared/pkg/models"
)

// RollupRepository handles the dynamic SQL execution required for calculating and applying rollups.
// It bypasses the standard PersistenceService to avoid circular dependencies and overhead for system calculations.
type RollupRepository struct {
	db             *sql.DB
	schemaProvider SchemaProvider
	viewResolver   ports.ViewResolver
}

const rollupCurrentReadLockAlias = "__rollup_lock"
const rollupParentIDPageSize = 100000
const maxRollupFilterBatchSize = 256

// NewRollupRepository creates a new RollupRepository
func NewRollupRepository(db *sql.DB) *RollupRepository {
	return &RollupRepository{
		db: db,
	}
}

// SetSchemaProvider wires a SchemaProvider (typically MetadataService) so
// rollup aggregation can resolve the authoritative child-object schema from the
// shared metadata cache instead of issuing direct DB lookups.
func (r *RollupRepository) SetSchemaProvider(sp SchemaProvider) {
	r.schemaProvider = sp
}

// SetViewResolver injects the ViewResolver so rollup aggregation can read from
// the authoritative view when the child object exposes active-computed fields.
func (r *RollupRepository) SetViewResolver(vr ports.ViewResolver) {
	r.viewResolver = vr
}

func (r *RollupRepository) getSchema(ctx context.Context, objectAPIName string) (*models.ObjectMetadata, error) {
	if r.schemaProvider != nil {
		schema := r.schemaProvider.GetSchema(ctx, objectAPIName)
		if schema == nil {
			return nil, pkgerrors.NewValidationError("summaryObject", fmt.Sprintf("schema not found for rollup object: %s", objectAPIName))
		}
		return schema, nil
	}
	return nil, pkgerrors.NewInternalError("rollup repository schema provider not wired", nil)
}

func (r *RollupRepository) resolveReadTable(ctx context.Context, objectAPIName string, schema *models.ObjectMetadata) string {
	if schema == nil {
		return objectAPIName
	}
	if formulastorage.HasActiveComputedFields(schema) {
		// Rollup hot paths may run while holding a parent-row lock inside a
		// pessimistic transaction. Routing by schema shape keeps that critical
		// section on the active transaction connection instead of triggering a
		// FormulaViewService.ViewExists information_schema probe on a second pooled
		// connection, which can deadlock against sibling waiters under load.
		return "v_" + strings.ToLower(schema.APIName)
	}
	if r.viewResolver == nil {
		return objectAPIName
	}
	return r.viewResolver.ResolveTableName(ctx, schema)
}

// rollupRequiresView reports whether a single rollup actually needs the
// authoritative v_<object> view. A rollup only needs it when the aggregated
// field is an active-computed field, or when its filter references one.
// Object-level routing (any active-computed field on the object) forces every
// rollup through the wide view even when the aggregate and filter touch only
// physical columns, which makes a locking current-read materialize the whole
// view and row-lock the base table for no reason.
func (r *RollupRepository) rollupRequiresView(schema *models.ObjectMetadata, summaryField string, filter *string) bool {
	if schema == nil {
		return false
	}
	if summaryField != "" && summaryField != "*" && formulastorage.SchemaFieldIsActiveComputed(schema, summaryField) {
		return true
	}
	if filter != nil && *filter != "" {
		return rollupFilterReferencesActiveComputed(schema, *filter)
	}
	return false
}

// rollupFilterReferencesActiveComputed conservatively reports whether the raw
// filter mentions any active-computed field (by API or physical name). It scans
// both backtick-quoted and bare identifiers and skips string literals.
func rollupFilterReferencesActiveComputed(schema *models.ObjectMetadata, filter string) bool {
	computed := make(map[string]struct{}, len(schema.Fields))
	for i := range schema.Fields {
		if !formulastorage.IsActiveComputedField(&schema.Fields[i]) {
			continue
		}
		if api := strings.ToLower(strings.TrimSpace(schema.Fields[i].APIName)); api != "" {
			computed[api] = struct{}{}
		}
		phys := actualColumnName(schema.Fields[i].APIName, schema.Fields[i].PhysicalName)
		if phys = strings.ToLower(strings.TrimSpace(phys)); phys != "" {
			computed[phys] = struct{}{}
		}
	}
	if len(computed) == 0 {
		return false
	}
	for _, ident := range filterIdentifierTokens(filter) {
		if _, ok := computed[ident]; ok {
			return true
		}
	}
	return false
}

// filterIdentifierTokens returns the lowercased identifiers found in a raw SQL
// filter fragment, covering backtick-quoted and bare word identifiers while
// skipping single/double-quoted string literals.
func filterIdentifierTokens(sqlText string) []string {
	isPart := func(c byte) bool {
		return c == '_' || c >= '0' && c <= '9' || c >= 'A' && c <= 'Z' || c >= 'a' && c <= 'z'
	}
	var tokens []string
	for i := 0; i < len(sqlText); {
		ch := sqlText[i]
		if ch == '\'' || ch == '"' {
			quote := ch
			for i++; i < len(sqlText); i++ {
				if sqlText[i] == quote {
					i++
					break
				}
			}
			continue
		}
		if ch == '`' {
			start := i + 1
			for i++; i < len(sqlText); i++ {
				if sqlText[i] == '`' {
					break
				}
			}
			if start <= i && i <= len(sqlText) {
				tokens = append(tokens, strings.ToLower(sqlText[start:min(i, len(sqlText))]))
			}
			if i < len(sqlText) {
				i++
			}
			continue
		}
		if ch == '_' || ch >= 'A' && ch <= 'Z' || ch >= 'a' && ch <= 'z' {
			j := i + 1
			for j < len(sqlText) && isPart(sqlText[j]) {
				j++
			}
			tokens = append(tokens, strings.ToLower(sqlText[i:j]))
			i = j
			continue
		}
		i++
	}
	return tokens
}

// GetExecutor returns the transactional executor if present in context, or the DB connection
func (r *RollupRepository) GetExecutor(ctx context.Context) Executor {
	if tx, ok := ctx.Value(txContextKey{}).(*sql.Tx); ok && tx != nil {
		return tx
	}
	return r.db
}

// CalculateRollup executes the aggregation query and returns the raw result.
// It supports COUNT, SUM, MIN, MAX, AVG.
// The query is constructed dynamically based on metadata configuration.
func (r *RollupRepository) CalculateRollup(ctx context.Context, calcType, summaryObject, summaryField, relationField, parentID string, filter *string) (interface{}, error) {
	return r.calculateRollup(ctx, calcType, summaryObject, summaryField, relationField, parentID, filter, false)
}

// CalculateRollupCurrentRead executes the aggregate as a locking current-read.
// Formula-view-routed child reads aggregate from the authoritative view while
// joining the base child table as a lock source.
func (r *RollupRepository) CalculateRollupCurrentRead(ctx context.Context, calcType, summaryObject, summaryField, relationField, parentID string, filter *string) (interface{}, error) {
	return r.calculateRollup(ctx, calcType, summaryObject, summaryField, relationField, parentID, filter, true)
}

func (r *RollupRepository) RecordMatchesRollupFilter(ctx context.Context, summaryObject, recordID string, filter *string) (bool, error) {
	if !isValidFieldName(summaryObject) {
		return false, pkgerrors.NewValidationError("summaryObject", fmt.Sprintf("invalid summary object: %s", summaryObject))
	}

	schema, err := r.getSchema(ctx, summaryObject)
	if err != nil {
		return false, pkgerrors.NewInternalError("failed to load rollup summary schema", err)
	}
	// Only consult the view when the match filter references an active-computed
	// field; a physical-column filter is correct and cheaper on the base table.
	readTable := summaryObject
	if r.rollupRequiresView(schema, "", filter) {
		readTable = r.resolveReadTable(ctx, summaryObject, schema)
	}

	builder := query.From(summaryObject).
		WithMetadata(schema).
		SkipID().
		AddSelectRaw("1").
		Where(constants.FieldID, recordID).
		WhereRaw(fmt.Sprintf("`%s` = ?", constants.ColIsDeleted), []interface{}{false}).
		Limit(1)
	if filter != nil && *filter != "" {
		filterSQL := *filter
		if readTable != summaryObject {
			filterSQL = replaceRollupFilterIdentifiers(filterSQL, rollupViewFilterReplacements(schema, summaryObject))
		}
		builder = builder.WhereRaw(filterSQL, nil)
	}

	qr := builder.Build()
	qr.SQL = rewriteRollupReadSourceAlias(qr.SQL, summaryObject, readTable)

	exec := r.GetExecutor(ctx)
	row := exec.QueryRowContext(ctx, qr.SQL, qr.Params...)
	var matched int
	if err := row.Scan(&matched); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return false, nil
		}
		return false, err
	}
	return true, nil
}

// RecordMatchesRollupFilters evaluates several filters against the same record
// image with at most one SQL statement per authoritative read source (physical
// table or active-formula view). The caller supplies an immutable pre- or
// post-write image boundary; this method only removes duplicate round trips and
// does not cache results across a write.
func (r *RollupRepository) RecordMatchesRollupFilters(
	ctx context.Context,
	summaryObject string,
	recordID string,
	filters []*string,
) ([]bool, error) {
	if len(filters) == 0 {
		return []bool{}, nil
	}
	if len(filters) > maxRollupFilterBatchSize {
		return nil, pkgerrors.NewValidationError(
			"filters",
			fmt.Sprintf("too many rollup filters: %d (max %d)", len(filters), maxRollupFilterBatchSize),
		)
	}
	if !isValidFieldName(summaryObject) {
		return nil, pkgerrors.NewValidationError("summaryObject", fmt.Sprintf("invalid summary object: %s", summaryObject))
	}
	schema, err := r.getSchema(ctx, summaryObject)
	if err != nil {
		return nil, pkgerrors.NewInternalError("failed to load rollup summary schema", err)
	}

	type readGroup struct {
		readTable string
		indexes   []int
	}
	groupsByTable := make(map[string]*readGroup, 2)
	groupOrder := make([]string, 0, 2)
	for index, filter := range filters {
		readTable := summaryObject
		if r.rollupRequiresView(schema, "", filter) {
			readTable = r.resolveReadTable(ctx, summaryObject, schema)
		}
		group := groupsByTable[readTable]
		if group == nil {
			group = &readGroup{readTable: readTable}
			groupsByTable[readTable] = group
			groupOrder = append(groupOrder, readTable)
		}
		group.indexes = append(group.indexes, index)
	}

	results := make([]bool, len(filters))
	for _, readTable := range groupOrder {
		group := groupsByTable[readTable]
		groupResults, queryErr := r.recordMatchesRollupFilterGroup(
			ctx, summaryObject, recordID, schema, readTable, filters, group.indexes,
		)
		if queryErr != nil {
			return nil, queryErr
		}
		for offset, index := range group.indexes {
			results[index] = groupResults[offset]
		}
	}
	return results, nil
}

func (r *RollupRepository) recordMatchesRollupFilterGroup(
	ctx context.Context,
	summaryObject string,
	recordID string,
	schema *models.ObjectMetadata,
	readTable string,
	filters []*string,
	indexes []int,
) (results []bool, retErr error) {
	builder := query.From(summaryObject).
		WithMetadata(schema).
		SkipID()
	for _, index := range indexes {
		predicate := "1 = 1"
		if filter := filters[index]; filter != nil && *filter != "" {
			predicate = *filter
			if readTable != summaryObject {
				predicate = replaceRollupFilterIdentifiers(
					predicate,
					rollupViewFilterReplacements(schema, summaryObject),
				)
			}
		}
		builder = builder.AddSelectRaw(
			fmt.Sprintf("MAX(CASE WHEN (%s) THEN 1 ELSE 0 END)", predicate),
		)
	}
	builder = builder.
		Where(constants.FieldID, recordID).
		WhereRaw(fmt.Sprintf("`%s` = ?", constants.ColIsDeleted), []interface{}{false})

	qr := builder.Build()
	qr.SQL = rewriteRollupReadSourceAlias(qr.SQL, summaryObject, readTable)
	exec := r.GetExecutor(ctx)
	sqlStarted := time.Now()
	defer func() {
		logRepositorySQLTiming(
			ctx,
			"RollupRepository.RecordMatchesRollupFilters",
			summaryObject,
			readTable,
			qr.SQL,
			len(qr.Params),
			len(indexes),
			1,
			sqlStarted,
			retErr,
		)
	}()

	values := make([]sql.NullInt64, len(indexes))
	destinations := make([]interface{}, len(values))
	for index := range values {
		destinations[index] = &values[index]
	}
	if err := exec.QueryRowContext(ctx, qr.SQL, qr.Params...).Scan(destinations...); err != nil {
		return nil, err
	}
	results = make([]bool, len(values))
	for index := range values {
		results[index] = values[index].Valid && values[index].Int64 != 0
	}
	return results, nil
}

func (r *RollupRepository) calculateRollup(ctx context.Context, calcType, summaryObject, summaryField, relationField, parentID string, filter *string, forUpdate bool) (result interface{}, retErr error) {
	// 1. Build Aggregation Query
	// SELECT FUNC(field) FROM child WHERE rel_field = ? AND IsDeleted = false

	normalizedCalcType := strings.ToUpper(strings.TrimSpace(calcType))
	normalizedSummaryField := strings.TrimSpace(summaryField)
	if normalizedCalcType == RollupTypeCount && normalizedSummaryField == "" {
		normalizedSummaryField = "*"
	}

	if !isValidFieldName(summaryObject) {
		return nil, pkgerrors.NewValidationError("summaryObject", fmt.Sprintf("invalid summary object: %s", summaryObject))
	}
	if !isValidFieldName(normalizedSummaryField) && normalizedSummaryField != "*" {
		return nil, pkgerrors.NewValidationError("summaryField", fmt.Sprintf("invalid summary field: %s", summaryField))
	}
	if !isValidFieldName(relationField) {
		return nil, pkgerrors.NewValidationError("relationField", fmt.Sprintf("invalid relation field: %s", relationField))
	}
	schema, err := r.getSchema(ctx, summaryObject)
	if err != nil {
		return nil, pkgerrors.NewInternalError("failed to load rollup summary schema", err)
	}
	summaryFieldMeta := rollupSummaryFieldMetadata(schema, normalizedSummaryField)
	// Route to the authoritative view only when this specific rollup actually
	// reads an active-computed field (aggregated column or a filter reference).
	// A COUNT/SUM over physical columns with a physical-column filter does not
	// need the view; forcing it through v_<object> turns a locking current-read
	// into a full view materialization plus base-table row locks (the 25s
	// Campaign.NumberOfContacts convertLead timeout).
	readTable := summaryObject
	if r.rollupRequiresView(schema, normalizedSummaryField, filter) {
		readTable = r.resolveReadTable(ctx, summaryObject, schema)
	}
	hasRawFilter := filter != nil && *filter != ""
	// Under READ COMMITTED the pinned pessimistic connection already reads the
	// latest committed siblings with a plain SELECT, and the parent row is
	// locked before this recompute runs (see WithReadCommittedIsolation). So the
	// child FOR UPDATE — whose only jobs are latest-committed visibility under
	// REPEATABLE READ and per-row locking — is redundant and is dropped. Only
	// the FOR UPDATE suffix is affected: viewRoutedCurrentRead stays keyed on
	// the read source so view-backed reads keep their lock-source JOIN and
	// filter-identifier rewrite intact (RC just omits the trailing FOR UPDATE),
	// keeping the query byte-identical to the REPEATABLE READ path minus locking.
	lockRows := forUpdate && !readCommittedIsolationFromContext(ctx)
	viewRoutedCurrentRead := forUpdate && !strings.EqualFold(readTable, summaryObject)

	funcMap := map[string]string{
		RollupTypeCount: RollupTypeCount,
		RollupTypeSum:   RollupTypeSum,
		RollupTypeMin:   RollupTypeMin,
		RollupTypeMax:   RollupTypeMax,
		RollupTypeAvg:   RollupTypeAvg,
	}

	aggFunc, ok := funcMap[normalizedCalcType]
	if !ok {
		return nil, pkgerrors.NewValidationError("calcType", fmt.Sprintf("unsupported rollup type: %s", calcType))
	}

	aggExpression := "*"
	if normalizedCalcType != RollupTypeCount {
		aggExpression = fmt.Sprintf("`%s`", normalizedSummaryField)
		if viewRoutedCurrentRead {
			aggExpression = fmt.Sprintf("`%s`.`%s`", summaryObject, normalizedSummaryField)
		}
	}

	// Build against the logical object name first so generated column
	// qualifiers remain stable when we later retarget the top-level FROM
	// clause to an alternate authoritative read source such as v_<object>.
	builder := query.From(summaryObject).
		WithMetadata(schema).
		SkipID().
		AddSelectRaw(fmt.Sprintf("%s(%s)", aggFunc, aggExpression))
	if viewRoutedCurrentRead {
		builder = builder.
			WhereRaw(fmt.Sprintf("`%s`.`%s` = ?", rollupCurrentReadLockAlias, resolvePhysicalName(schema, constants.FieldIsDeleted)), []interface{}{false}).
			WhereRaw(fmt.Sprintf("`%s`.`%s` = ?", rollupCurrentReadLockAlias, resolvePhysicalName(schema, relationField)), []interface{}{parentID}).
			// Repeat the parent predicate on the view-aliased side. The INNER JOIN
			// on Id already makes this logically redundant, but without it TiDB does
			// not push the parent filter into the v_<object> scan and materializes
			// the entire view before joining (the 692k-row Campaign.NumberOfResponses
			// FOR UPDATE that hit the 25s timeout). With it, the predicate pushes
			// into the base scan and the view reads only this parent's children.
			// The view exposes columns by API name (see rollupViewFilterReplacements),
			// so reference relationField directly rather than its physical name.
			WhereRaw(fmt.Sprintf("`%s`.`%s` = ?", summaryObject, relationField), []interface{}{parentID})
	} else {
		builder = builder.
			Where(relationField, parentID).
			WhereRaw(fmt.Sprintf("`%s` = ?", constants.ColIsDeleted), []interface{}{false})
	}

	// Add optional filter
	if hasRawFilter {
		if viewRoutedCurrentRead {
			builder.WhereRaw(replaceRollupFilterIdentifiers(*filter, rollupViewFilterReplacements(schema, summaryObject)), nil)
		} else {
			builder.WhereRaw(*filter, nil)
		}
	}

	qr := builder.Build()
	if viewRoutedCurrentRead {
		qr.SQL = rewriteRollupCurrentReadViewSource(qr.SQL, summaryObject, readTable, resolvePhysicalName(schema, constants.FieldID))
	} else {
		qr.SQL = rewriteRollupReadSourceAlias(qr.SQL, summaryObject, readTable)
	}
	if lockRows {
		qr.SQL += " FOR UPDATE"
	}

	// 2. Execute Query
	exec := r.GetExecutor(ctx)
	sqlStarted := time.Now()
	defer func() {
		logRepositorySQLTiming(ctx, "RollupRepository.calculateRollup", summaryObject, readTable, qr.SQL, len(qr.Params), 1, 1, sqlStarted, retErr)
	}()
	row := exec.QueryRowContext(ctx, qr.SQL, qr.Params...)

	var scanDest interface{}
	// For SUM/AVG on empty set, SQL returns NULL. We need to handle that.
	var nullFloat sql.NullFloat64
	var nullString sql.NullString // For MIN/MAX on dates/text

	// Decide scan destination based on expected return type from DB perspective
	switch normalizedCalcType {
	case RollupTypeCount:
		// Count always returns a number (int)
		if err := row.Scan(&scanDest); err != nil {
			return nil, err
		}
		// Convert byte array or other types to int/float if needed, but usually driver handles scalar
		// TiDB/MySQL driver usually returns int64 for COUNT
		return scanDest, nil
	case RollupTypeSum, RollupTypeAvg:
		// Return float/decimal
		if err := row.Scan(&nullFloat); err != nil {
			return nil, err
		}
		if nullFloat.Valid {
			return nullFloat.Float64, nil
		}
		return 0, nil // Default for Sum/Avg is 0
	case RollupTypeMin, RollupTypeMax:
		// Could be number, date, or text
		if rollupFieldIsTemporal(summaryFieldMeta) {
			var nullTime sql.NullTime
			if err := row.Scan(&nullTime); err != nil {
				return nil, err
			}
			if nullTime.Valid {
				if rollupFieldEffectiveType(*summaryFieldMeta) == constants.FieldTypeDate {
					return nullTime.Time.Format("2006-01-02"), nil
				}
				return nullTime.Time, nil
			}
			return nil, nil
		}
		if err := row.Scan(&nullString); err != nil {
			return nil, err
		}
		if nullString.Valid {
			return nullString.String, nil
		}
		return nil, nil // Default for Min/Max is null
	}

	return 0, nil
}

func rollupSummaryFieldMetadata(schema *models.ObjectMetadata, apiName string) *models.FieldMetadata {
	if schema == nil || strings.TrimSpace(apiName) == "" {
		return nil
	}
	for i := range schema.Fields {
		if strings.EqualFold(schema.Fields[i].APIName, apiName) {
			return &schema.Fields[i]
		}
	}
	return nil
}

func rollupFieldIsTemporal(field *models.FieldMetadata) bool {
	if field == nil {
		return false
	}
	switch rollupFieldEffectiveType(*field) {
	case constants.FieldTypeDate, constants.FieldTypeDatetime:
		return true
	default:
		return false
	}
}

func rollupFieldEffectiveType(field models.FieldMetadata) constants.SchemaFieldType {
	base := constants.NormalizeFieldType(string(field.Type))
	if base != constants.FieldTypeFormula && base != constants.FieldTypeRollupSummary {
		return base
	}
	if field.ReturnType != nil {
		if normalized := constants.NormalizeFieldType(string(*field.ReturnType)); normalized != "" {
			return normalized
		}
	}
	return base
}

func rewriteRollupReadSourceAlias(sqlText string, logicalObject string, readTable string) string {
	logicalObject = strings.TrimSpace(logicalObject)
	readTable = strings.TrimSpace(readTable)
	if sqlText == "" || logicalObject == "" || readTable == "" || strings.EqualFold(logicalObject, readTable) {
		return sqlText
	}
	needle := fmt.Sprintf(" FROM `%s`", logicalObject)
	replacement := fmt.Sprintf(" FROM `%s` AS `%s`", readTable, logicalObject)
	return strings.Replace(sqlText, needle, replacement, 1)
}

func rewriteRollupCurrentReadViewSource(sqlText string, logicalObject string, readTable string, physicalID string) string {
	logicalObject = strings.TrimSpace(logicalObject)
	readTable = strings.TrimSpace(readTable)
	physicalID = strings.TrimSpace(physicalID)
	if physicalID == "" {
		physicalID = constants.FieldID
	}
	if sqlText == "" || logicalObject == "" || readTable == "" {
		return sqlText
	}
	needle := fmt.Sprintf(" FROM `%s`", logicalObject)
	replacement := fmt.Sprintf(
		" FROM `%s` AS `%s` INNER JOIN `%s` AS `%s` ON `%s`.`%s` = `%s`.`%s`",
		logicalObject,
		rollupCurrentReadLockAlias,
		readTable,
		logicalObject,
		logicalObject,
		constants.FieldID,
		rollupCurrentReadLockAlias,
		physicalID,
	)
	return strings.Replace(sqlText, needle, replacement, 1)
}

func rollupViewFilterReplacements(schema *models.ObjectMetadata, logicalObject string) map[string]string {
	replacements := make(map[string]string, len(schema.Fields)*2)
	for _, field := range schema.Fields {
		if strings.TrimSpace(field.APIName) == "" {
			continue
		}
		qualified := fmt.Sprintf("`%s`.`%s`", logicalObject, field.APIName)
		replacements[strings.ToLower(field.APIName)] = qualified
		physical := actualColumnName(field.APIName, field.PhysicalName)
		if strings.TrimSpace(physical) != "" {
			replacements[strings.ToLower(physical)] = qualified
		}
	}
	return replacements
}

func replaceRollupFilterIdentifiers(sqlText string, replacements map[string]string) string {
	isPart := func(c byte) bool {
		return c == '_' || c >= '0' && c <= '9' || c >= 'A' && c <= 'Z' || c >= 'a' && c <= 'z'
	}
	isSpace := func(c byte) bool { return c == ' ' || c == '\n' || c == '\r' || c == '\t' }
	var out strings.Builder
	for i, subqueryDepth := 0, 0; i < len(sqlText); {
		ch := sqlText[i]
		if ch == '\'' || ch == '"' {
			quote, start := ch, i
			for i++; i < len(sqlText); i++ {
				if sqlText[i] == quote {
					i++
					break
				}
			}
			out.WriteString(sqlText[start:i])
			continue
		}
		if ch == '`' {
			start := i
			for i++; i < len(sqlText); i++ {
				if sqlText[i] == '`' {
					break
				}
			}
			if i >= len(sqlText) {
				out.WriteString(sqlText[start:])
				break
			}
			identifier := sqlText[start+1 : i]
			end := i + 1
			qualified := (start > 0 && sqlText[start-1] == '.') || (end < len(sqlText) && sqlText[end] == '.')
			if subqueryDepth == 0 && !qualified {
				if v := replacements[strings.ToLower(identifier)]; v != "" {
					out.WriteString(v)
					i = end
					continue
				}
			}
			out.WriteString(sqlText[start:end])
			i = end
			continue
		}
		if ch == '(' {
			j := i + 1
			for j < len(sqlText) && isSpace(sqlText[j]) {
				j++
			}
			if subqueryDepth > 0 || j+6 <= len(sqlText) && strings.EqualFold(sqlText[j:j+6], "select") {
				subqueryDepth++
			}
		} else if ch == ')' && subqueryDepth > 0 {
			subqueryDepth--
		}
		if subqueryDepth == 0 && (ch == '_' || ch >= 'A' && ch <= 'Z' || ch >= 'a' && ch <= 'z') {
			j := i + 1
			for j < len(sqlText) && isPart(sqlText[j]) {
				j++
			}
			qualified := (i > 0 && sqlText[i-1] == '.') || (j < len(sqlText) && sqlText[j] == '.')
			if v := replacements[strings.ToLower(sqlText[i:j])]; v != "" && !qualified {
				out.WriteString(v)
			} else {
				out.WriteString(sqlText[i:j])
			}
			i = j
			continue
		}
		out.WriteByte(ch)
		i++
	}
	return out.String()
}

// GetParentIDsPage retrieves one keyset page of record IDs for a given object.
func (r *RollupRepository) GetParentIDsPage(ctx context.Context, objectName string, afterID string, limit int) ([]string, error) {
	if !isValidFieldName(objectName) {
		return nil, pkgerrors.NewValidationError("objectName", fmt.Sprintf("invalid object name: %s", objectName))
	}
	if limit <= 0 {
		limit = rollupParentIDPageSize
	}

	builder := query.From(objectName).
		AddSelectRaw(constants.ColID).
		WhereRaw(fmt.Sprintf("`%s` = ?", constants.ColIsDeleted), []interface{}{false}).
		OrderBy(constants.ColID, constants.SortASC).
		Limit(limit)
	if afterID != "" {
		builder = builder.WhereRaw(fmt.Sprintf("`%s` > ?", constants.ColID), []interface{}{afterID})
	}

	qr := builder.Build()
	rows, err := r.db.QueryContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return nil, pkgerrors.NewInternalError("failed to query parent IDs", err)
	}
	defer rows.Close()

	var ids []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, pkgerrors.NewInternalError("failed to scan row", err)
		}
		ids = append(ids, id)
	}

	if err := rows.Err(); err != nil {
		return nil, err
	}

	return ids, nil
}