package packs

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log"
	"regexp"
	"sort"
	"strings"
	"time"

	domainschema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
)

var polymorphicBackfillIdentifierRE = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)

var errPolymorphicBackfillCandidateInvariant = errors.New("polymorphic backfill candidate invariant violation")

const (
	polymorphicBackfillStatementTimeout     = 10 * time.Second
	polymorphicBackfillStatementMaxAttempts = 3
	polymorphicBackfillBatchSize            = 500
	polymorphicBackfillCandidateTargetLimit = 16
	polymorphicBackfillWriteBatchSize       = 100
	polymorphicBackfillObjectPrefixSize     = 3
)

type polymorphicBackfillRow struct {
	id          string
	fkID        sql.NullString
	currentType sql.NullString
}

type polymorphicBackfillResolutionState struct {
	unavailableTargetErrors map[string]error
	objectOwnerByKeyPrefix  map[string]string
	objectOwnersLoaded      bool
}

// BackfillPolymorphicDiscriminators repairs server-managed <field>_type values
// for every managed polymorphic reference field. It is idempotent, clears stale
// discriminators for empty FKs, and leaves non-empty FKs unchanged when they do
// not resolve within the current source's declared target set. This preserves
// historical and extension-owned references across source-backed pack reapply.
// When a declared target table is unavailable, an unresolved FK is preserved
// only when active key-prefix metadata proves that unavailable target cannot own
// it. Unknown ownership, ambiguous target identities, and concurrent source
// changes still fail closed.
func BackfillPolymorphicDiscriminators(ctx context.Context, db *sql.DB, defs []domainschema.TableDefinition) error {
	if db == nil {
		return fmt.Errorf("polymorphic discriminator backfill: nil DB")
	}
	defs = domainschema.ExpandPolymorphicDiscriminatorColumnsForDefinitions(defs)
	resolutionState := &polymorphicBackfillResolutionState{
		unavailableTargetErrors: make(map[string]error),
	}
	totalUpdated := 0
	for _, def := range defs {
		tableName := strings.TrimSpace(def.TableName)
		if tableName == "" {
			continue
		}
		for _, col := range def.Columns {
			if !domainschema.IsPolymorphicReferenceColumn(col) {
				continue
			}
			updated, err := backfillPolymorphicDiscriminatorColumn(ctx, db, tableName, col, resolutionState)
			if err != nil {
				return err
			}
			totalUpdated += updated
		}
	}
	if totalUpdated > 0 {
		log.Printf("   BackfillPolymorphicDiscriminators: updated %d discriminator row(s)", totalUpdated)
	}
	return nil
}

func backfillPolymorphicDiscriminatorColumn(
	ctx context.Context,
	db *sql.DB,
	tableName string,
	col domainschema.ColumnDefinition,
	resolutionState *polymorphicBackfillResolutionState,
) (int, error) {
	tableIdent, err := quotePolymorphicBackfillIdentifier(tableName)
	if err != nil {
		return 0, err
	}
	fkIdent, err := quotePolymorphicBackfillIdentifier(col.Name)
	if err != nil {
		return 0, err
	}
	typeColName := domainschema.PolymorphicDiscriminatorColumnName(col.Name)
	typeIdent, err := quotePolymorphicBackfillIdentifier(typeColName)
	if err != nil {
		return 0, err
	}
	idIdent, err := quotePolymorphicBackfillIdentifier(constants.ColID)
	if err != nil {
		return 0, err
	}
	targets, err := normalizePolymorphicBackfillTargets(col.ReferenceTo)
	if err != nil {
		return 0, err
	}
	if len(targets) == 0 {
		return 0, fmt.Errorf("polymorphic discriminator backfill %s.%s has no declared targets", tableName, col.Name)
	}

	updated := 0
	processed := 0
	unresolved := 0
	batches := 0
	afterID := ""
	combinedClassificationEnabled := len(targets) <= polymorphicBackfillCandidateTargetLimit
	candidateClassificationEnabled := combinedClassificationEnabled
	for {
		var (
			rows            []polymorphicBackfillRow
			resolvedTargets map[string][]string
		)
		if candidateClassificationEnabled {
			rows, err = loadPolymorphicBackfillCandidateRows(
				ctx, db, tableName, col.Name, tableIdent, idIdent, fkIdent, typeIdent, targets, afterID,
			)
			if err == nil && len(rows) >= polymorphicBackfillBatchSize {
				// Independent probes do not share an ordering cursor. When the
				// sparse fast path fills a page, discard that discovery result and
				// fall back from the beginning to the canonical database-ordered
				// source pager. This keeps one pagination authority and cannot skip
				// IDs whose database collation differs from Go byte ordering.
				candidateClassificationEnabled = false
				rows, resolvedTargets, err = loadAndResolvePolymorphicBackfillRows(
					ctx, db, tableName, col.Name, tableIdent, idIdent, fkIdent, typeIdent, targets, afterID,
				)
			} else if err == nil {
				resolvedTargets, err = resolvePolymorphicDiscriminatorTargets(
					ctx, db, targets, rows, resolutionState,
				)
			}
			if err != nil {
				if ctxErr := ctx.Err(); ctxErr != nil {
					return updated, ctxErr
				}
				if errors.Is(err, errPolymorphicBackfillCandidateInvariant) {
					return updated, err
				}
				combinedClassificationEnabled = false
				candidateClassificationEnabled = false
			}
		} else if combinedClassificationEnabled {
			rows, resolvedTargets, err = loadAndResolvePolymorphicBackfillRows(
				ctx, db, tableName, col.Name, tableIdent, idIdent, fkIdent, typeIdent, targets, afterID,
			)
			if err != nil {
				if ctxErr := ctx.Err(); ctxErr != nil {
					return updated, ctxErr
				}
				combinedClassificationEnabled = false
				candidateClassificationEnabled = false
			}
		}
		if !combinedClassificationEnabled {
			rows, err = loadPolymorphicBackfillRows(
				ctx, db, tableName, col.Name, tableIdent, idIdent, fkIdent, typeIdent, afterID,
			)
			if err != nil {
				return updated, err
			}
		}
		if len(rows) == 0 {
			break
		}

		lastID := rows[len(rows)-1].id
		if lastID == "" || lastID == afterID {
			return updated, fmt.Errorf("polymorphic discriminator backfill %s.%s did not advance after Id=%q",
				tableName, col.Name, afterID)
		}

		if !combinedClassificationEnabled {
			resolvedTargets, err = resolvePolymorphicDiscriminatorTargets(
				ctx, db, targets, rows, resolutionState,
			)
			if err != nil {
				return updated, fmt.Errorf("polymorphic discriminator backfill resolve %s.%s targets=%v: %w",
					tableName, col.Name, targets, err)
			}
		}
		batchUpdated, batchUnresolved, err := writePolymorphicDiscriminatorBatch(
			ctx, db, tableName, col.Name, tableIdent, idIdent, fkIdent, typeIdent, targets, rows, resolvedTargets,
		)
		if err != nil {
			return updated, err
		}

		updated += batchUpdated
		unresolved += batchUnresolved
		processed += len(rows)
		batches++
		if len(rows) < polymorphicBackfillBatchSize {
			break
		}
		if candidateClassificationEnabled {
			candidateClassificationEnabled = false
		}
		afterID = lastID
	}
	if processed > 0 {
		log.Printf("polymorphic discriminator backfill: table=%s field=%s scanned=%d targets=%d batches=%d updated=%d unresolved=%d unavailable_targets=%d",
			tableName, col.Name, processed, len(targets), batches, updated, unresolved,
			countUnavailablePolymorphicBackfillTargets(targets, resolutionState.unavailableTargetErrors))
	}
	return updated, nil
}

// loadPolymorphicBackfillCandidateRows finds only rows which can require a
// discriminator write.  The former candidate query joined every declared
// target behind one OR-heavy predicate.  TiDB had to classify the entire source
// table before LIMIT could help; on a 235k-row Task table that query took more
// than the bounded statement deadline even when there were zero repair rows.
//
// Keep the same semantics with independent, optimizer-friendly probes:
//   - one indexed source-only probe which preserves fail-closed validation for
//     malformed source IDs;
//   - one source-only probe for stale discriminators on an empty FK; and
//   - one inner-join probe per declared target for rows whose discriminator does
//     not match the target which actually owns the FK.
//
// The returned source rows are resolved again through the canonical batched
// target resolver.  That second step is intentional: it preserves ambiguous-FK,
// unavailable-target, collation, and CAS-write behavior instead of trusting a
// discovery probe as authority.
func loadPolymorphicBackfillCandidateRows(
	ctx context.Context,
	db *sql.DB,
	tableName, fieldName, tableIdent, idIdent, fkIdent, typeIdent string,
	targets []string,
	afterID string,
) ([]polymorphicBackfillRow, error) {
	rowsByID := make(map[string]polymorphicBackfillRow, polymorphicBackfillBatchSize)
	collect := func(operation, query string, args []any) error {
		return runPolymorphicBackfillStatement(ctx, operation, func(statementCtx context.Context) error {
			queryRows, queryErr := db.QueryContext(statementCtx, query, args...)
			if queryErr != nil {
				return queryErr
			}
			defer queryRows.Close()

			seenInProbe := make(map[string]struct{}, polymorphicBackfillBatchSize)
			for queryRows.Next() {
				var row polymorphicBackfillRow
				if scanErr := queryRows.Scan(&row.id, &row.fkID, &row.currentType); scanErr != nil {
					return fmt.Errorf("polymorphic discriminator backfill scan %s.%s: %w", tableName, fieldName, scanErr)
				}
				if row.id == "" {
					return fmt.Errorf("%w: polymorphic discriminator backfill %s.%s encountered empty row Id", errPolymorphicBackfillCandidateInvariant, tableName, fieldName)
				}
				if _, duplicate := seenInProbe[row.id]; duplicate {
					return fmt.Errorf("%w: polymorphic discriminator backfill %s.%s encountered duplicate row Id=%q", errPolymorphicBackfillCandidateInvariant, tableName, fieldName, row.id)
				}
				seenInProbe[row.id] = struct{}{}
				if existing, duplicate := rowsByID[row.id]; duplicate {
					if existing.fkID != row.fkID || existing.currentType != row.currentType {
						return fmt.Errorf("%w: polymorphic discriminator backfill %s.%s source row changed across candidate probes Id=%q", errPolymorphicBackfillCandidateInvariant, tableName, fieldName, row.id)
					}
					continue
				}
				rowsByID[row.id] = row
			}
			if rowsErr := queryRows.Err(); rowsErr != nil {
				return fmt.Errorf("polymorphic discriminator backfill rows %s.%s: %w", tableName, fieldName, rowsErr)
			}
			return nil
		})
	}

	invalidIDQuery := fmt.Sprintf(
		"SELECT /* polymorphic-backfill-invalid-source-id */ source.%s, source.%s, source.%s FROM %s AS source WHERE source.%s IS NULL OR source.%s = '' LIMIT 1",
		idIdent, fkIdent, typeIdent, tableIdent, idIdent, idIdent,
	)
	if err := collect("find invalid source Id rows "+tableName+"."+fieldName, invalidIDQuery, nil); err != nil {
		return nil, err
	}

	staleWhere := fmt.Sprintf(
		"(source.%s IS NULL OR BINARY source.%s = BINARY '') AND source.%s IS NOT NULL AND NOT (BINARY source.%s <=> BINARY '')",
		fkIdent, fkIdent, typeIdent, typeIdent,
	)
	staleArgs := make([]any, 0, 1)
	if afterID != "" {
		staleWhere = fmt.Sprintf("source.%s > ? AND (%s)", idIdent, staleWhere)
		staleArgs = append(staleArgs, afterID)
	}
	staleQuery := fmt.Sprintf(
		"SELECT /* polymorphic-backfill-stale-discriminator */ source.%s, source.%s, source.%s FROM %s AS source WHERE %s ORDER BY source.%s LIMIT %d",
		idIdent, fkIdent, typeIdent, tableIdent, staleWhere, idIdent, polymorphicBackfillBatchSize,
	)
	if err := collect("find stale discriminator rows "+tableName+"."+fieldName, staleQuery, staleArgs); err != nil {
		return nil, err
	}

	for _, target := range targets {
		targetIdent, err := quotePolymorphicBackfillIdentifier(target)
		if err != nil {
			return nil, err
		}
		where := fmt.Sprintf("NOT (BINARY source.%s <=> BINARY ?)", typeIdent)
		args := make([]any, 0, 2)
		if afterID != "" {
			where = fmt.Sprintf("source.%s > ? AND (%s)", idIdent, where)
			args = append(args, afterID)
		}
		args = append(args, target)
		query := fmt.Sprintf(
			"SELECT /* polymorphic-backfill-target-candidate */ source.%s, source.%s, source.%s FROM %s AS source INNER JOIN %s AS matched ON matched.%s = source.%s WHERE %s ORDER BY source.%s LIMIT %d",
			idIdent, fkIdent, typeIdent, tableIdent, targetIdent, idIdent, fkIdent, where, idIdent, polymorphicBackfillBatchSize,
		)
		if err := collect("find target candidate rows "+tableName+"."+fieldName+" target="+target, query, args); err != nil {
			return nil, err
		}
	}

	rows := make([]polymorphicBackfillRow, 0, len(rowsByID))
	for _, row := range rowsByID {
		rows = append(rows, row)
	}
	// Stable byte ordering keeps resolution and write batches deterministic.
	// Sparse candidate pages terminate without a cursor; a full page is
	// discarded before canonical database-ordered paging begins, so this order
	// is never used to advance SQL pagination.
	sort.Slice(rows, func(i, j int) bool { return rows[i].id < rows[j].id })
	return rows, nil
}

func normalizePolymorphicBackfillTargets(rawTargets []string) ([]string, error) {
	targets := make([]string, 0, len(rawTargets))
	seen := make(map[string]struct{}, len(rawTargets))
	for _, rawTarget := range rawTargets {
		target := strings.TrimSpace(rawTarget)
		if target == "" {
			continue
		}
		if _, err := quotePolymorphicBackfillIdentifier(target); err != nil {
			return nil, err
		}
		if _, ok := seen[target]; ok {
			continue
		}
		seen[target] = struct{}{}
		targets = append(targets, target)
	}
	return targets, nil
}

func loadAndResolvePolymorphicBackfillRows(
	ctx context.Context,
	db *sql.DB,
	tableName, fieldName, tableIdent, idIdent, fkIdent, typeIdent string,
	targets []string,
	afterID string,
) ([]polymorphicBackfillRow, map[string][]string, error) {
	selectColumns := []string{"source." + idIdent, "source." + fkIdent, "source." + typeIdent}
	joins := make([]string, len(targets))
	for i, target := range targets {
		targetIdent, err := quotePolymorphicBackfillIdentifier(target)
		if err != nil {
			return nil, nil, err
		}
		alias := fmt.Sprintf("target_%d", i)
		selectColumns = append(selectColumns, alias+"."+idIdent)
		joins[i] = fmt.Sprintf("LEFT JOIN %s AS %s ON %s.%s = source.%s", targetIdent, alias, alias, idIdent, fkIdent)
	}
	args := make([]any, 0, 1)
	sourceQuery := fmt.Sprintf(
		"SELECT /* polymorphic-backfill-classification */ %s, %s, %s FROM %s",
		idIdent, fkIdent, typeIdent, tableIdent,
	)
	if afterID != "" {
		sourceQuery += fmt.Sprintf(" WHERE %s > ?", idIdent)
		args = append(args, afterID)
	}
	sourceQuery += fmt.Sprintf(" ORDER BY %s LIMIT %d", idIdent, polymorphicBackfillBatchSize)
	query := fmt.Sprintf(
		"SELECT %s FROM (%s) AS source %s ORDER BY source.%s",
		strings.Join(selectColumns, ", "), sourceQuery, strings.Join(joins, " "), idIdent,
	)

	var (
		batch    []polymorphicBackfillRow
		resolved map[string][]string
	)
	err := runPolymorphicBackfillStatement(ctx, "classify rows "+tableName+"."+fieldName, func(statementCtx context.Context) error {
		queryRows, queryErr := db.QueryContext(statementCtx, query, args...)
		if queryErr != nil {
			return queryErr
		}
		defer queryRows.Close()

		attemptBatch := make([]polymorphicBackfillRow, 0, polymorphicBackfillBatchSize)
		seenIDs := make(map[string]struct{}, polymorphicBackfillBatchSize)
		matchedTargetByFK := make(map[string]map[int]struct{}, polymorphicBackfillBatchSize)
		attemptResolved := make(map[string][]string, polymorphicBackfillBatchSize)
		for queryRows.Next() {
			var row polymorphicBackfillRow
			targetMatches := make([]sql.NullString, len(targets))
			scanArgs := make([]any, 0, 3+len(targets))
			scanArgs = append(scanArgs, &row.id, &row.fkID, &row.currentType)
			for i := range targetMatches {
				scanArgs = append(scanArgs, &targetMatches[i])
			}
			if scanErr := queryRows.Scan(scanArgs...); scanErr != nil {
				return fmt.Errorf("polymorphic discriminator backfill scan %s.%s: %w", tableName, fieldName, scanErr)
			}
			if row.id == "" {
				return fmt.Errorf("polymorphic discriminator backfill %s.%s encountered empty row Id", tableName, fieldName)
			}
			if _, duplicate := seenIDs[row.id]; duplicate {
				return fmt.Errorf("polymorphic discriminator backfill %s.%s encountered duplicate row Id=%q", tableName, fieldName, row.id)
			}
			seenIDs[row.id] = struct{}{}
			attemptBatch = append(attemptBatch, row)

			if row.fkID.Valid && row.fkID.String != "" {
				fkTargets := matchedTargetByFK[row.fkID.String]
				if fkTargets == nil {
					fkTargets = make(map[int]struct{}, len(targets))
					matchedTargetByFK[row.fkID.String] = fkTargets
				}
				for i, match := range targetMatches {
					if !match.Valid {
						continue
					}
					if _, alreadyMatched := fkTargets[i]; !alreadyMatched {
						fkTargets[i] = struct{}{}
						attemptResolved[row.fkID.String] = append(attemptResolved[row.fkID.String], targets[i])
					}
				}
			}
		}
		if rowsErr := queryRows.Err(); rowsErr != nil {
			return fmt.Errorf("polymorphic discriminator backfill rows %s.%s: %w", tableName, fieldName, rowsErr)
		}
		batch = attemptBatch
		resolved = attemptResolved
		return nil
	})
	return batch, resolved, err
}

func loadPolymorphicBackfillRows(
	ctx context.Context,
	db *sql.DB,
	tableName, fieldName, tableIdent, idIdent, fkIdent, typeIdent, afterID string,
) ([]polymorphicBackfillRow, error) {
	// Page by the primary key before testing the two discriminator inputs. The
	// previous OR predicate made TiDB scan the full table to find every page,
	// even with ORDER BY Id and LIMIT. A bounded primary-key page guarantees
	// progress under the statement deadline; writePolymorphicDiscriminatorBatch
	// still owns the empty-FK, stale-type, and unresolved-target semantics.
	query := fmt.Sprintf("SELECT %s, %s, %s FROM %s", idIdent, fkIdent, typeIdent, tableIdent)
	args := make([]any, 0, 1)
	if afterID != "" {
		query += fmt.Sprintf(" WHERE %s > ?", idIdent)
		args = append(args, afterID)
	}
	query += fmt.Sprintf(" ORDER BY %s LIMIT %d", idIdent, polymorphicBackfillBatchSize)

	var batch []polymorphicBackfillRow
	err := runPolymorphicBackfillStatement(ctx, "load rows "+tableName+"."+fieldName, func(statementCtx context.Context) error {
		rows, queryErr := db.QueryContext(statementCtx, query, args...)
		if queryErr != nil {
			return fmt.Errorf("polymorphic discriminator backfill query %s.%s: %w", tableName, fieldName, queryErr)
		}
		defer rows.Close()

		attemptRows := make([]polymorphicBackfillRow, 0, polymorphicBackfillBatchSize)
		seenIDs := make(map[string]struct{}, polymorphicBackfillBatchSize)
		for rows.Next() {
			var row polymorphicBackfillRow
			if scanErr := rows.Scan(&row.id, &row.fkID, &row.currentType); scanErr != nil {
				return fmt.Errorf("polymorphic discriminator backfill scan %s.%s: %w", tableName, fieldName, scanErr)
			}
			if row.id == "" {
				return fmt.Errorf("polymorphic discriminator backfill %s.%s encountered empty row Id", tableName, fieldName)
			}
			if _, ok := seenIDs[row.id]; ok {
				return fmt.Errorf("polymorphic discriminator backfill %s.%s encountered duplicate row Id=%q", tableName, fieldName, row.id)
			}
			seenIDs[row.id] = struct{}{}
			attemptRows = append(attemptRows, row)
		}
		if rowsErr := rows.Err(); rowsErr != nil {
			return fmt.Errorf("polymorphic discriminator backfill rows %s.%s: %w", tableName, fieldName, rowsErr)
		}
		batch = attemptRows
		return nil
	})
	return batch, err
}

func resolvePolymorphicDiscriminatorTargets(
	ctx context.Context,
	db *sql.DB,
	targets []string,
	rows []polymorphicBackfillRow,
	state *polymorphicBackfillResolutionState,
) (map[string][]string, error) {
	fkIDs := make([]string, 0, len(rows))
	seenFKs := make(map[string]struct{}, len(rows))
	for _, row := range rows {
		if !row.fkID.Valid || row.fkID.String == "" {
			continue
		}
		if _, ok := seenFKs[row.fkID.String]; ok {
			continue
		}
		seenFKs[row.fkID.String] = struct{}{}
		fkIDs = append(fkIDs, row.fkID.String)
	}
	if len(fkIDs) == 0 {
		return map[string][]string{}, nil
	}

	resolved := make(map[string][]string, len(fkIDs))
	idIdent, err := quotePolymorphicBackfillIdentifier(constants.ColID)
	if err != nil {
		return nil, err
	}
	requestedRows := make([]string, len(fkIDs))
	args := make([]any, len(fkIDs))
	for i, fkID := range fkIDs {
		if i == 0 {
			requestedRows[i] = "SELECT ? AS `RequestedId`"
		} else {
			requestedRows[i] = "SELECT ?"
		}
		args[i] = fkID
	}
	for _, target := range targets {
		if _, unavailable := state.unavailableTargetErrors[target]; unavailable {
			continue
		}
		targetIdent, err := quotePolymorphicBackfillIdentifier(target)
		if err != nil {
			return nil, err
		}
		query := fmt.Sprintf(
			"SELECT requested.`RequestedId` FROM (%s) AS requested INNER JOIN %s AS matched ON matched.%s = requested.`RequestedId`",
			strings.Join(requestedRows, " UNION ALL "), targetIdent, idIdent,
		)
		var matches []string
		err = runPolymorphicBackfillStatement(ctx, "resolve target "+target, func(statementCtx context.Context) error {
			queryRows, queryErr := db.QueryContext(statementCtx, query, args...)
			if queryErr != nil {
				return queryErr
			}
			defer queryRows.Close()

			attemptMatches := make([]string, 0)
			for queryRows.Next() {
				var id string
				if scanErr := queryRows.Scan(&id); scanErr != nil {
					return scanErr
				}
				// The query returns the original requested value rather than the
				// stored target Id. The database join owns collation equality while
				// this map retains the byte-exact source FK used by repair rows.
				if _, ok := seenFKs[id]; !ok {
					return fmt.Errorf("target %s returned unexpected id=%q", target, id)
				}
				attemptMatches = append(attemptMatches, id)
			}
			if rowsErr := queryRows.Err(); rowsErr != nil {
				return rowsErr
			}
			matches = attemptMatches
			return nil
		})
		if err != nil {
			// ReferenceTo is a capability universe and may include targets that are
			// not installed in this environment. Cache only a missing-table result;
			// an unknown Id column remains physical schema drift and fails closed.
			if persistence.IsTableNotExistsError(err) {
				state.unavailableTargetErrors[target] = err
				continue
			}
			return nil, fmt.Errorf("check target %s: %w", target, err)
		}
		for _, id := range matches {
			resolved[id] = append(resolved[id], target)
		}
	}
	if err := validateUnresolvedPolymorphicBackfillOwners(ctx, db, targets, fkIDs, resolved, state); err != nil {
		return nil, err
	}
	return resolved, nil
}

func validateUnresolvedPolymorphicBackfillOwners(
	ctx context.Context,
	db *sql.DB,
	targets, fkIDs []string,
	resolved map[string][]string,
	state *polymorphicBackfillResolutionState,
) error {
	unavailableTargets := make([]string, 0)
	for _, target := range targets {
		if _, unavailable := state.unavailableTargetErrors[target]; unavailable {
			unavailableTargets = append(unavailableTargets, target)
		}
	}
	if len(unavailableTargets) == 0 {
		return nil
	}

	unresolvedFKs := make([]string, 0)
	for _, fkID := range fkIDs {
		if len(resolved[fkID]) == 0 {
			unresolvedFKs = append(unresolvedFKs, fkID)
		}
	}
	if len(unresolvedFKs) == 0 {
		return nil
	}

	if !state.objectOwnersLoaded {
		owners, err := loadPolymorphicBackfillObjectKeyPrefixOwners(ctx, db)
		if err != nil {
			return fmt.Errorf("load object ownership while declared target schema is unavailable targets=%v: %w", unavailableTargets, err)
		}
		state.objectOwnerByKeyPrefix = owners
		state.objectOwnersLoaded = true
	}

	for _, fkID := range unresolvedFKs {
		if len(fkID) < polymorphicBackfillObjectPrefixSize {
			return fmt.Errorf("cannot classify unresolved FK %q while declared target schema is unavailable targets=%v: record Id is shorter than the %d-byte object key prefix",
				fkID, unavailableTargets, polymorphicBackfillObjectPrefixSize)
		}
		prefix := fkID[:polymorphicBackfillObjectPrefixSize]
		owner, ok := state.objectOwnerByKeyPrefix[prefix]
		if !ok {
			return fmt.Errorf("cannot classify unresolved FK %q prefix=%q while declared target schema is unavailable targets=%v: active object ownership is unknown",
				fkID, prefix, unavailableTargets)
		}
		if unavailableErr, unavailable := polymorphicBackfillUnavailableTargetError(owner, state.unavailableTargetErrors); unavailable {
			return fmt.Errorf("cannot classify unresolved FK %q prefix=%q owner=%q while declared target schema is unavailable targets=%v: %w",
				fkID, prefix, owner, unavailableTargets, unavailableErr)
		}
	}
	return nil
}

func loadPolymorphicBackfillObjectKeyPrefixOwners(ctx context.Context, db *sql.DB) (map[string]string, error) {
	const query = "SELECT `KeyPrefix`, `ApiName` FROM `_System_Object` WHERE `IsDeleted` = 0 AND `KeyPrefix` IS NOT NULL AND `KeyPrefix` <> '' ORDER BY `KeyPrefix`, `ApiName`"
	var owners map[string]string
	err := runPolymorphicBackfillStatement(ctx, "load object key-prefix ownership", func(statementCtx context.Context) error {
		rows, queryErr := db.QueryContext(statementCtx, query)
		if queryErr != nil {
			return queryErr
		}
		defer rows.Close()

		attemptOwners := make(map[string]string)
		for rows.Next() {
			var prefix, apiName string
			if scanErr := rows.Scan(&prefix, &apiName); scanErr != nil {
				return scanErr
			}
			apiName = strings.TrimSpace(apiName)
			if len(prefix) != polymorphicBackfillObjectPrefixSize || apiName == "" {
				return fmt.Errorf("invalid active object key-prefix ownership prefix=%q owner=%q", prefix, apiName)
			}
			if existing, duplicate := attemptOwners[prefix]; duplicate {
				return fmt.Errorf("ambiguous active object key-prefix ownership prefix=%q owners=%q,%q", prefix, existing, apiName)
			}
			attemptOwners[prefix] = apiName
		}
		if rowsErr := rows.Err(); rowsErr != nil {
			return rowsErr
		}
		owners = attemptOwners
		return nil
	})
	if err != nil {
		return nil, err
	}
	return owners, nil
}

func polymorphicBackfillUnavailableTargetError(owner string, unavailableTargetErrors map[string]error) (error, bool) {
	for target, targetErr := range unavailableTargetErrors {
		if strings.EqualFold(owner, target) {
			return targetErr, true
		}
	}
	return nil, false
}

func countUnavailablePolymorphicBackfillTargets(targets []string, unavailableTargetErrors map[string]error) int {
	count := 0
	for _, target := range targets {
		if _, unavailable := unavailableTargetErrors[target]; unavailable {
			count++
		}
	}
	return count
}

func writePolymorphicDiscriminatorBatch(
	ctx context.Context,
	db *sql.DB,
	tableName, fieldName, tableIdent, idIdent, fkIdent, typeIdent string,
	targets []string,
	rows []polymorphicBackfillRow,
	resolvedTargets map[string][]string,
) (int, int, error) {
	clearRows := make([]polymorphicBackfillRow, 0)
	updateRowsByType := make(map[string][]polymorphicBackfillRow, len(targets))
	unresolved := 0
	for _, row := range rows {
		if !row.fkID.Valid || row.fkID.String == "" {
			if row.currentType.Valid && row.currentType.String != "" {
				clearRows = append(clearRows, row)
			}
			continue
		}

		matches := resolvedTargets[row.fkID.String]
		if len(matches) == 0 {
			// A source-backed pack can encounter extension-owned references outside
			// its declared target set, while historical imports can retain references
			// after their target disappears. No safe discriminator can be inferred
			// from this source alone, so preserve the row and keep repairing the
			// resolvable portion of the batch.
			unresolved++
			continue
		}
		if len(matches) > 1 {
			return 0, unresolved, fmt.Errorf("polymorphic discriminator backfill ambiguous %s.%s row Id=%q fk=%q matched targets=%v",
				tableName, fieldName, row.id, row.fkID.String, matches)
		}
		resolvedType := matches[0]
		if row.currentType.Valid && row.currentType.String == resolvedType {
			continue
		}
		updateRowsByType[resolvedType] = append(updateRowsByType[resolvedType], row)
	}

	updated := 0
	if len(clearRows) > 0 {
		if err := executePolymorphicDiscriminatorCASUpdate(
			ctx, db, tableIdent, idIdent, fkIdent, typeIdent, nil, clearRows,
		); err != nil {
			return updated, unresolved, fmt.Errorf("polymorphic discriminator backfill clear %s.%s rows=%d: %w",
				tableName, fieldName, len(clearRows), err)
		}
		updated += len(clearRows)
	}
	for _, target := range targets {
		targetRows := updateRowsByType[target]
		if len(targetRows) == 0 {
			continue
		}
		if err := executePolymorphicDiscriminatorCASUpdate(
			ctx, db, tableIdent, idIdent, fkIdent, typeIdent, target, targetRows,
		); err != nil {
			return updated, unresolved, fmt.Errorf("polymorphic discriminator backfill update %s.%s type=%q rows=%d: %w",
				tableName, fieldName, target, len(targetRows), err)
		}
		updated += len(targetRows)
	}
	return updated, unresolved, nil
}

// executePolymorphicDiscriminatorCASUpdate binds every repair to the FK and
// discriminator snapshot used during target resolution. A concurrent record
// retarget or clear therefore makes the statement affect fewer rows and fails
// ApplyAll without writing a stale discriminator onto the raced row.
func executePolymorphicDiscriminatorCASUpdate(
	ctx context.Context,
	db *sql.DB,
	tableIdent, idIdent, fkIdent, typeIdent string,
	value any,
	rows []polymorphicBackfillRow,
) error {
	if len(rows) == 0 {
		return nil
	}
	sort.Slice(rows, func(i, j int) bool { return rows[i].id < rows[j].id })
	for start := 0; start < len(rows); start += polymorphicBackfillWriteBatchSize {
		end := start + polymorphicBackfillWriteBatchSize
		if end > len(rows) {
			end = len(rows)
		}
		if err := executePolymorphicDiscriminatorCASWrite(
			ctx, db, tableIdent, idIdent, fkIdent, typeIdent, value, rows[start:end],
		); err != nil {
			return err
		}
	}
	return nil
}

func executePolymorphicDiscriminatorCASWrite(
	ctx context.Context,
	db *sql.DB,
	tableIdent, idIdent, fkIdent, typeIdent string,
	value any,
	rows []polymorphicBackfillRow,
) error {
	predicate := fmt.Sprintf("(%s = ? AND %s <=> ? AND %s <=> ?)", idIdent, fkIdent, typeIdent)
	predicates := make([]string, len(rows))
	args := make([]any, 0, 1+3*len(rows))
	args = append(args, value)
	for i, row := range rows {
		predicates[i] = predicate
		args = append(args, row.id, nullablePolymorphicBackfillValue(row.fkID), nullablePolymorphicBackfillValue(row.currentType))
	}
	query := fmt.Sprintf("UPDATE %s SET %s = ? WHERE %s", tableIdent, typeIdent, strings.Join(predicates, " OR "))
	return runPolymorphicBackfillWrite(ctx, func(statementCtx context.Context) error {
		result, err := db.ExecContext(statementCtx, query, args...)
		if err != nil {
			return err
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return fmt.Errorf("read affected rows: %w", err)
		}
		if affected != int64(len(rows)) {
			return fmt.Errorf("concurrent source row change: affected rows=%d, want %d", affected, len(rows))
		}
		return nil
	})
}

func nullablePolymorphicBackfillValue(value sql.NullString) any {
	if !value.Valid {
		return nil
	}
	return value.String
}

func runPolymorphicBackfillStatement(ctx context.Context, operation string, fn func(context.Context) error) error {
	var lastErr error
	for attempt := 1; attempt <= polymorphicBackfillStatementMaxAttempts; attempt++ {
		statementCtx, cancel := context.WithTimeout(ctx, polymorphicBackfillStatementTimeout)
		lastErr = fn(statementCtx)
		cancel()
		if lastErr == nil || errors.Is(lastErr, sql.ErrNoRows) {
			return lastErr
		}
		if ctxErr := ctx.Err(); ctxErr != nil {
			return ctxErr
		}
		if !isRetryablePolymorphicBackfillStatementError(lastErr) || attempt == polymorphicBackfillStatementMaxAttempts {
			return lastErr
		}
		log.Printf("polymorphic discriminator backfill: retrying transient database operation=%q attempt=%d/%d: %v",
			operation, attempt, polymorphicBackfillStatementMaxAttempts, lastErr)
	}
	return lastErr
}

func runPolymorphicBackfillWrite(ctx context.Context, fn func(context.Context) error) error {
	statementCtx, cancel := context.WithTimeout(ctx, polymorphicBackfillStatementTimeout)
	defer cancel()
	return fn(statementCtx)
}

func isRetryablePolymorphicBackfillStatementError(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, context.DeadlineExceeded) || isTransientDDLConnectionError(err) {
		return true
	}
	return strings.Contains(strings.ToLower(err.Error()), "canceling query due to user request")
}

func quotePolymorphicBackfillIdentifier(name string) (string, error) {
	name = strings.TrimSpace(name)
	if !polymorphicBackfillIdentifierRE.MatchString(name) {
		return "", fmt.Errorf("polymorphic discriminator backfill invalid identifier %q", name)
	}
	return "`" + name + "`", nil
}