package persistence

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

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/formulastorage"
)

type packOwnedFieldState struct {
	id           string
	ownerKind    string
	ownerPack    string
	isCustomized bool
	isDeleted    bool
	trackHistory bool
}

// EnsurePackOwnedField creates or verifies a disabled-history pack-owned field
// without routing through the admin field writer. It exists for app-pack
// installation paths that must preserve OwnerKind=pack / OwnerPack=<pack> and
// keep IsCustomized=false on fresh rows. TrackHistory creation or convergence
// is deliberately not owned here: callers must use the active-transaction
// BatchSaveFieldMetadataWithCustomizationPolicy exact-owner path.
func (r *SchemaRepository) EnsurePackOwnedField(ctx context.Context, objectAPIName string, field schema.FieldWithContext) error {
	objectAPIName = strings.TrimSpace(objectAPIName)
	if objectAPIName == "" {
		return apperrors.NewValidationError("objectAPIName", "object API name is required")
	}
	field.TableName = objectAPIName

	col := field.ColumnDefinition
	col.Name = strings.TrimSpace(col.Name)
	if col.Name == "" {
		return apperrors.NewValidationError("fieldName", fmt.Sprintf("field name is required for %s", objectAPIName))
	}
	field.ColumnDefinition = col
	if err := r.validateCatalogFieldContextWriteProtectionProjection(field); err != nil {
		return err
	}

	ownerPack := strings.TrimSpace(field.OwnerPack)
	if ownerPack == "" && field.Field != nil && field.Field.OwnerPack != nil {
		ownerPack = strings.TrimSpace(*field.Field.OwnerPack)
	}
	if strings.TrimSpace(field.OwnerKind) != constants.FieldOwnerKindPack || ownerPack == "" {
		return apperrors.NewValidationError("OwnerKind", fmt.Sprintf("field %s.%s requires pack ownership", objectAPIName, col.Name))
	}
	field.OwnerKind = constants.FieldOwnerKindPack
	field.OwnerPack = ownerPack

	objectID, err := r.loadObjectIDByAPIName(ctx, objectAPIName)
	if err != nil {
		return err
	}
	field.ObjectID = objectID
	if strings.TrimSpace(field.FieldID) == "" {
		field.FieldID = schema.GenerateFieldID(objectAPIName, col.Name)
	}

	state, found, err := r.loadPackOwnedFieldState(ctx, objectAPIName, col.Name)
	if err != nil {
		return err
	}
	desiredTrackHistory := field.Field != nil && field.Field.TrackHistory
	if !found && desiredTrackHistory {
		return apperrors.NewValidationError(constants.FieldSysField_TrackHistory, "new fields must be created with trackHistory disabled; enable it in a separate committed transition")
	}
	if found {
		if err := validateExistingPackOwnedField(objectAPIName, col.Name, ownerPack, state); err != nil {
			return err
		}
		// EnsurePackOwnedField is intentionally non-transactional. Any existing
		// TrackHistory divergence must fail before DDL or metadata mutation and
		// converge only through the active-transaction exact-owner reapply path.
		if state.trackHistory != desiredTrackHistory {
			return apperrors.NewValidationError(constants.FieldSysField_TrackHistory, fmt.Sprintf(
				"existing pack-owned field %s.%s requires active-transaction exact-owner history tracking convergence",
				objectAPIName,
				col.Name,
			))
		}
		plan, err := r.ResolvePackOwnedFieldConvergence(ctx, objectAPIName, field)
		if err != nil {
			return err
		}
		if plan.Action == schema.PackOwnedFieldConvergenceFailBeforeMutation {
			return apperrors.NewValidationError("physicalColumnShape", plan.Detail)
		}
		if packOwnedConvergenceRequiresPhysicalDDL(plan) {
			if err := r.ensurePackOwnedPhysicalColumn(ctx, objectAPIName, col); err != nil {
				return err
			}
			if err := r.verifyPackOwnedPhysicalConvergence(objectAPIName, col); err != nil {
				return err
			}
		}
		field = r.preparePackOwnedFieldMetadata(field, col, ownerPack)
		if err := r.batchSaveFieldMetadata(ctx, []schema.FieldWithContext{field}, nil, false, false); err != nil {
			return err
		}
		return nil
	}

	plan, err := r.ResolvePackOwnedFieldConvergence(ctx, objectAPIName, field)
	if err != nil {
		return err
	}
	if plan.Action == schema.PackOwnedFieldConvergenceFailBeforeMutation {
		return apperrors.NewValidationError("physicalColumnShape", plan.Detail)
	}
	var storage formulastorage.Resolution
	if field.Field != nil {
		storage, err = formulastorage.Resolve(formulastorage.SpecFromField(field.Field))
	} else {
		storage, err = formulastorage.Resolve(formulastorage.SpecFromColumn(field.ColumnDefinition))
	}
	if err != nil {
		return err
	}
	if storage.RequiresPhysicalColumn() && packOwnedConvergenceRequiresPhysicalDDL(plan) {
		if err := r.ensurePackOwnedPhysicalColumn(ctx, objectAPIName, col); err != nil {
			return err
		}
		if err := r.verifyPackOwnedPhysicalConvergence(objectAPIName, col); err != nil {
			return err
		}
	}

	field = r.preparePackOwnedFieldMetadata(field, col, ownerPack)

	if err := r.batchSaveFieldMetadata(ctx, []schema.FieldWithContext{field}, nil, false, false); err != nil {
		return err
	}

	state, found, err = r.loadPackOwnedFieldState(ctx, objectAPIName, col.Name)
	if err != nil {
		return err
	}
	if !found {
		return apperrors.NewInternalError(fmt.Sprintf("pack-owned field %s.%s metadata was not written", objectAPIName, col.Name), nil)
	}
	if err := validateExistingPackOwnedField(objectAPIName, col.Name, ownerPack, state); err != nil {
		return err
	}
	if storage.RequiresPhysicalColumn() {
		exists, err := r.checkColumnExists(objectAPIName, col.Name)
		if err != nil {
			return apperrors.NewInternalError(fmt.Sprintf("failed to verify column existence for %s.%s", objectAPIName, col.Name), err)
		}
		if !exists {
			return apperrors.NewInternalError(fmt.Sprintf("pack-owned field %s.%s metadata written but physical column is missing", objectAPIName, col.Name), nil)
		}
	}
	return nil
}

func (r *SchemaRepository) preparePackOwnedFieldMetadata(field schema.FieldWithContext, col schema.ColumnDefinition, ownerPack string) schema.FieldWithContext {
	if field.Field == nil {
		field.Field = r.buildFieldMetadataFromColumn(col)
	}
	fieldCopy := *field.Field
	field.Field = &fieldCopy
	field.Field.APIName = col.Name
	if field.Field.Label == "" {
		field.Field.Label = col.Label
	}
	field.Field.OwnerKind = constants.FieldOwnerKindPack
	field.Field.OwnerPack = &ownerPack
	field.Field.IsUnique = col.Unique
	return field
}

func (r *SchemaRepository) ResolvePackOwnedFieldConvergence(ctx context.Context, objectAPIName string, field schema.FieldWithContext) (schema.PackOwnedFieldConvergencePlan, error) {
	objectAPIName = strings.TrimSpace(objectAPIName)
	col := field.ColumnDefinition
	col.Name = strings.TrimSpace(col.Name)
	plan := schema.PackOwnedFieldConvergencePlan{
		ObjectAPIName: objectAPIName,
		FieldAPIName:  col.Name,
	}
	if objectAPIName == "" {
		return plan, apperrors.NewValidationError("objectAPIName", "object API name is required")
	}
	if col.Name == "" {
		return plan, apperrors.NewValidationError("fieldName", fmt.Sprintf("field name is required for %s", objectAPIName))
	}
	ownerPack := strings.TrimSpace(field.OwnerPack)
	if ownerPack == "" && field.Field != nil && field.Field.OwnerPack != nil {
		ownerPack = strings.TrimSpace(*field.Field.OwnerPack)
	}
	if strings.TrimSpace(field.OwnerKind) != constants.FieldOwnerKindPack || ownerPack == "" {
		return plan, apperrors.NewValidationError("OwnerKind", fmt.Sprintf("field %s.%s requires pack ownership", objectAPIName, col.Name))
	}
	if _, err := r.loadObjectIDByAPIName(ctx, objectAPIName); err != nil {
		return plan, err
	}

	state, found, err := r.loadPackOwnedFieldState(ctx, objectAPIName, col.Name)
	if err != nil {
		return plan, err
	}
	var storage formulastorage.Resolution
	if field.Field != nil {
		storage, err = formulastorage.Resolve(formulastorage.SpecFromField(field.Field))
	} else {
		storage, err = formulastorage.Resolve(formulastorage.SpecFromColumn(field.ColumnDefinition))
	}
	if err != nil {
		return plan, err
	}
	if !found {
		if storage.FormulaBacked && !storage.RequiresPhysicalColumn() {
			return r.resolvePackOwnedMetadataOnlyConvergence(objectAPIName, col, plan, schema.PackOwnedFieldConvergenceMetadataOnlyCreate)
		}
		return r.resolvePackOwnedPhysicalConvergence(objectAPIName, col, plan, schema.PackOwnedFieldConvergenceCreate)
	}
	if err := validateExistingPackOwnedField(objectAPIName, col.Name, ownerPack, state); err != nil {
		plan.Action = schema.PackOwnedFieldConvergenceFailBeforeMutation
		plan.Reason = "metadata-owner-conflict"
		plan.Detail = err.Error()
		return plan, nil
	}
	if storage.FormulaBacked && !storage.RequiresPhysicalColumn() {
		return r.resolvePackOwnedMetadataOnlyConvergence(objectAPIName, col, plan, schema.PackOwnedFieldConvergenceVerifyOnly)
	}
	return r.resolvePackOwnedPhysicalConvergence(objectAPIName, col, plan, schema.PackOwnedFieldConvergenceVerifyOnly)
}

func (r *SchemaRepository) resolvePackOwnedMetadataOnlyConvergence(objectAPIName string, col schema.ColumnDefinition, plan schema.PackOwnedFieldConvergencePlan, matchAction schema.PackOwnedFieldConvergenceAction) (schema.PackOwnedFieldConvergencePlan, error) {
	exists, err := r.checkColumnExists(objectAPIName, col.Name)
	if err != nil {
		return plan, apperrors.NewInternalError(fmt.Sprintf("failed to check column existence for %s.%s", objectAPIName, col.Name), err)
	}
	if exists {
		plan.Action = schema.PackOwnedFieldConvergenceFailBeforeMutation
		plan.Reason = "metadata-only-physical-column"
		plan.Detail = fmt.Sprintf("metadata-only field %s.%s has an unexpected physical column", objectAPIName, col.Name)
		return plan, nil
	}
	plan.Action = matchAction
	if matchAction == schema.PackOwnedFieldConvergenceMetadataOnlyCreate {
		plan.Reason = "metadata-missing"
	} else {
		plan.Reason = "metadata-only-active"
	}
	return plan, nil
}

func (r *SchemaRepository) resolvePackOwnedPhysicalConvergence(objectAPIName string, col schema.ColumnDefinition, plan schema.PackOwnedFieldConvergencePlan, matchAction schema.PackOwnedFieldConvergenceAction) (schema.PackOwnedFieldConvergencePlan, error) {
	exists, err := r.checkColumnExists(objectAPIName, col.Name)
	if err != nil {
		return plan, apperrors.NewInternalError(fmt.Sprintf("failed to check column existence for %s.%s", objectAPIName, col.Name), err)
	}
	if !exists {
		if matchAction == schema.PackOwnedFieldConvergenceVerifyOnly {
			plan.Action = schema.PackOwnedFieldConvergenceRepairMissingPhysical
			plan.Reason = "physical-missing"
			return plan, nil
		}
		plan.Action = matchAction
		plan.Reason = "metadata-missing"
		return plan, nil
	}
	spec, err := r.getPhysicalColumnSpec(objectAPIName, col.Name)
	if err != nil {
		return plan, apperrors.NewInternalError(fmt.Sprintf("failed to read physical column spec for %s.%s", objectAPIName, col.Name), err)
	}
	if !r.packOwnedPhysicalColumnMatches(spec, col) {
		plan.Action = schema.PackOwnedFieldConvergenceFailBeforeMutation
		plan.Reason = "physical-shape-drift"
		plan.Detail = fmt.Sprintf("field %s.%s physical column shape = %s nullable=%v default=%s, want %s nullable=%v default=%s",
			objectAPIName,
			col.Name,
			spec.ColumnType,
			spec.IsNullable,
			formatPackOwnedPhysicalDefault(spec.ColumnDefault),
			r.ResolveColumnSqlType(col),
			col.Nullable,
			formatPackOwnedDesiredDefault(col.Default),
		)
		return plan, nil
	}
	constraintDrift, err := r.packOwnedPhysicalConstraintDrift(objectAPIName, col)
	if err != nil {
		return plan, err
	}
	if constraintDrift != "" {
		plan.Action = schema.PackOwnedFieldConvergenceFailBeforeMutation
		plan.Reason = "physical-constraint-drift"
		plan.Detail = constraintDrift
		return plan, nil
	}
	plan.Action = matchAction
	if matchAction == schema.PackOwnedFieldConvergenceCreate {
		plan.Reason = "metadata-missing-physical-matches"
	} else {
		plan.Reason = "physical-matches"
	}
	return plan, nil
}

func (r *SchemaRepository) packOwnedPhysicalConstraintDrift(tableName string, col schema.ColumnDefinition) (string, error) {
	if col.Unique {
		exists, err := r.packOwnedUniqueConstraintExists(tableName, col.Name)
		if err != nil {
			return "", apperrors.NewInternalError(fmt.Sprintf("failed to read unique constraint for %s.%s", tableName, col.Name), err)
		}
		if !exists {
			return fmt.Sprintf("field %s.%s is missing required unique constraint", tableName, col.Name), nil
		}
	}

	if len(col.ReferenceTo) == 1 && !col.IsPolymorphic && !strings.EqualFold(col.ReferenceTo[0], "parent") {
		exists, err := r.packOwnedForeignKeyConstraintExists(tableName, col.Name, col.ReferenceTo[0], col.OnDelete)
		if err != nil {
			return "", apperrors.NewInternalError(fmt.Sprintf("failed to read foreign key constraint for %s.%s", tableName, col.Name), err)
		}
		if !exists {
			return fmt.Sprintf("field %s.%s is missing required foreign key constraint to %s", tableName, col.Name, col.ReferenceTo[0]), nil
		}
	}
	return "", nil
}

func (r *SchemaRepository) packOwnedUniqueConstraintExists(tableName string, columnName string) (bool, error) {
	indexName := schema.GenerateUniqueIndexName(tableName, columnName)
	const q = `
		SELECT COUNT(*)
		  FROM INFORMATION_SCHEMA.STATISTICS s
		 WHERE s.TABLE_SCHEMA = DATABASE()
		   AND s.TABLE_NAME = ?
		   AND s.INDEX_NAME = ?
		   AND s.COLUMN_NAME = ?
		   AND s.NON_UNIQUE = 0
		   AND NOT EXISTS (
		       SELECT 1
		         FROM INFORMATION_SCHEMA.STATISTICS extra
		        WHERE extra.TABLE_SCHEMA = s.TABLE_SCHEMA
		          AND extra.TABLE_NAME = s.TABLE_NAME
		          AND extra.INDEX_NAME = s.INDEX_NAME
		          AND extra.COLUMN_NAME <> s.COLUMN_NAME
		   )
	`
	var count int
	if err := r.db.QueryRow(q, tableName, indexName, columnName).Scan(&count); err != nil {
		return false, err
	}
	return count > 0, nil
}

func (r *SchemaRepository) packOwnedForeignKeyConstraintExists(tableName string, columnName string, referencedTableName string, onDelete string) (bool, error) {
	deleteRules := packOwnedAllowedDeleteRules(onDelete)
	const q = `
		SELECT COUNT(*)
		  FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE k
		  JOIN INFORMATION_SCHEMA.REFERENTIAL_CONSTRAINTS rc
		    ON rc.CONSTRAINT_SCHEMA = k.CONSTRAINT_SCHEMA
		   AND rc.CONSTRAINT_NAME = k.CONSTRAINT_NAME
		 WHERE k.TABLE_SCHEMA = DATABASE()
		   AND k.TABLE_NAME = ?
		   AND k.COLUMN_NAME = ?
		   AND k.REFERENCED_TABLE_NAME = ?
		   AND k.REFERENCED_COLUMN_NAME = ?
		   AND UPPER(COALESCE(rc.DELETE_RULE, '')) IN (?, ?)
	`
	var count int
	if err := r.db.QueryRow(q, tableName, columnName, referencedTableName, constants.ColID, deleteRules[0], deleteRules[1]).Scan(&count); err != nil {
		return false, err
	}
	return count > 0, nil
}

func packOwnedAllowedDeleteRules(onDelete string) [2]string {
	rule := strings.ToUpper(strings.TrimSpace(onDelete))
	if rule == "" {
		return [2]string{"RESTRICT", "NO ACTION"}
	}
	return [2]string{rule, rule}
}

func packOwnedConvergenceRequiresPhysicalDDL(plan schema.PackOwnedFieldConvergencePlan) bool {
	return plan.Action == schema.PackOwnedFieldConvergenceRepairMissingPhysical ||
		(plan.Action == schema.PackOwnedFieldConvergenceCreate && plan.Reason == "metadata-missing")
}

func (r *SchemaRepository) verifyPackOwnedPhysicalConvergence(objectAPIName string, col schema.ColumnDefinition) error {
	plan := schema.PackOwnedFieldConvergencePlan{ObjectAPIName: objectAPIName, FieldAPIName: col.Name}
	plan, err := r.resolvePackOwnedPhysicalConvergence(objectAPIName, col, plan, schema.PackOwnedFieldConvergenceVerifyOnly)
	if err != nil {
		return err
	}
	if plan.Action == schema.PackOwnedFieldConvergenceFailBeforeMutation {
		return apperrors.NewValidationError("physicalColumnShape", plan.Detail)
	}
	if plan.Action != schema.PackOwnedFieldConvergenceVerifyOnly {
		return apperrors.NewValidationError("physicalColumnShape", fmt.Sprintf("field %s.%s physical column did not converge after DDL: action=%s reason=%s", objectAPIName, col.Name, plan.Action, plan.Reason))
	}
	return nil
}

func (r *SchemaRepository) packOwnedPhysicalColumnMatches(spec *physicalColumnSpec, col schema.ColumnDefinition) bool {
	if spec == nil {
		return false
	}
	desiredType := r.ResolveColumnSqlType(col)
	if packOwnedNormalizeColumnType(spec.ColumnType) != packOwnedNormalizeColumnType(desiredType) {
		return false
	}
	if spec.IsNullable != col.Nullable {
		return false
	}
	if !packOwnedPhysicalDefaultMatches(spec.ColumnDefault, col.Default, desiredType) {
		return false
	}
	if strings.TrimSpace(col.CollationName) != "" {
		if spec.CollationName == nil || !strings.EqualFold(strings.TrimSpace(*spec.CollationName), strings.TrimSpace(col.CollationName)) {
			return false
		}
	}
	return true
}

func packOwnedPhysicalDefaultMatches(actual *string, desired string, desiredType string) bool {
	if !SqlTypeAllowsDefaultClause(desiredType) {
		return true
	}
	desired = strings.TrimSpace(desired)
	if desired == "" {
		return actual == nil
	}
	if actual == nil {
		return false
	}
	actualDefault := normalizePackOwnedColumnDefault(*actual)
	desiredDefault := normalizePackOwnedColumnDefault(desired)
	if packOwnedNormalizeColumnType(desiredType) == "tinyint(1)" {
		actualBool, actualOK := normalizePackOwnedBooleanDefault(actualDefault)
		desiredBool, desiredOK := normalizePackOwnedBooleanDefault(desiredDefault)
		if actualOK && desiredOK {
			return actualBool == desiredBool
		}
	}
	if packOwnedIsTemporalColumnType(desiredType) {
		actualTimestamp, actualOK := normalizePackOwnedTimestampDefault(actualDefault)
		desiredTimestamp, desiredOK := normalizePackOwnedTimestampDefault(desiredDefault)
		if actualOK && desiredOK {
			return actualTimestamp == desiredTimestamp
		}
	}
	return strings.EqualFold(actualDefault, desiredDefault)
}

func normalizePackOwnedColumnDefault(v string) string {
	v = strings.TrimSpace(v)
	if len(v) >= 2 {
		first := v[0]
		last := v[len(v)-1]
		if (first == '\'' && last == '\'') || (first == '"' && last == '"') {
			v = strings.TrimSpace(v[1 : len(v)-1])
		}
	}
	return v
}

func packOwnedNormalizeColumnType(v string) string {
	lower := strings.ToLower(strings.TrimSpace(v))
	switch lower {
	case "boolean", "bool":
		return "tinyint(1)"
	case "int", "integer":
		return "int(11)"
	case "bigint":
		return "bigint(20)"
	case "smallint":
		return "smallint(6)"
	case "tinyint":
		return "tinyint(4)"
	default:
		return lower
	}
}

func normalizePackOwnedBooleanDefault(v string) (bool, bool) {
	normalized := strings.ToLower(strings.TrimSpace(v))
	for len(normalized) >= 2 && normalized[0] == '(' && normalized[len(normalized)-1] == ')' {
		normalized = strings.TrimSpace(normalized[1 : len(normalized)-1])
	}
	switch normalized {
	case "1", "true":
		return true, true
	case "0", "false":
		return false, true
	default:
		return false, false
	}
}

func packOwnedIsTemporalColumnType(v string) bool {
	normalized := packOwnedNormalizeColumnType(v)
	return strings.HasPrefix(normalized, "datetime") || strings.HasPrefix(normalized, "timestamp")
}

func normalizePackOwnedTimestampDefault(v string) (string, bool) {
	normalized := strings.ToLower(strings.TrimSpace(v))
	for len(normalized) >= 2 && normalized[0] == '(' && normalized[len(normalized)-1] == ')' {
		normalized = strings.TrimSpace(normalized[1 : len(normalized)-1])
	}
	switch normalized {
	case "current_timestamp", "current_timestamp()":
		return "current_timestamp", true
	default:
		return "", false
	}
}

func formatPackOwnedPhysicalDefault(v *string) string {
	if v == nil {
		return "<NULL>"
	}
	return *v
}

func formatPackOwnedDesiredDefault(v string) string {
	if strings.TrimSpace(v) == "" {
		return "<NULL>"
	}
	return v
}

func (r *SchemaRepository) loadObjectIDByAPIName(ctx context.Context, objectAPIName string) (string, error) {
	var objectID string
	err := r.db.QueryRowContext(ctx,
		"SELECT Id FROM _System_Object WHERE ApiName = ? AND COALESCE(IsDeleted, 0) = 0 LIMIT 1",
		objectAPIName,
	).Scan(&objectID)
	if err == sql.ErrNoRows {
		return "", apperrors.NewNotFoundError("object", objectAPIName)
	}
	if err != nil {
		return "", apperrors.NewInternalError(fmt.Sprintf("failed to resolve object ID for %s", objectAPIName), err)
	}
	return objectID, nil
}

func (r *SchemaRepository) loadPackOwnedFieldState(ctx context.Context, objectAPIName string, fieldAPIName string) (packOwnedFieldState, bool, error) {
	var state packOwnedFieldState
	var isCustomized int
	var isDeleted int
	var trackHistory int
	err := r.db.QueryRowContext(ctx,
		"SELECT f.Id, COALESCE(f.OwnerKind, ''), COALESCE(f.OwnerPack, ''), COALESCE(f.IsCustomized, 0), COALESCE(f.IsDeleted, 0), COALESCE(f.TrackHistory, 0) FROM _System_Field f JOIN _System_Object o ON o.Id = f.ObjectId WHERE o.ApiName = ? AND f.ApiName = ? LIMIT 1",
		objectAPIName,
		fieldAPIName,
	).Scan(&state.id, &state.ownerKind, &state.ownerPack, &isCustomized, &isDeleted, &trackHistory)
	if err == sql.ErrNoRows {
		return packOwnedFieldState{}, false, nil
	}
	if err != nil {
		return packOwnedFieldState{}, false, apperrors.NewInternalError(fmt.Sprintf("failed to load field ownership for %s.%s", objectAPIName, fieldAPIName), err)
	}
	state.ownerKind = strings.TrimSpace(state.ownerKind)
	state.ownerPack = strings.TrimSpace(state.ownerPack)
	state.isCustomized = isCustomized != 0
	state.isDeleted = isDeleted != 0
	state.trackHistory = trackHistory != 0
	return state, true, nil
}

func validateExistingPackOwnedField(objectAPIName string, fieldAPIName string, ownerPack string, state packOwnedFieldState) error {
	if state.ownerKind != constants.FieldOwnerKindPack {
		return apperrors.NewValidationError("OwnerKind", fmt.Sprintf("field %s.%s OwnerKind = %q, want %q", objectAPIName, fieldAPIName, state.ownerKind, constants.FieldOwnerKindPack))
	}
	if state.ownerPack != ownerPack {
		return apperrors.NewValidationError("OwnerPack", fmt.Sprintf("field %s.%s OwnerPack = %q, want %q", objectAPIName, fieldAPIName, state.ownerPack, ownerPack))
	}
	if state.isCustomized {
		return apperrors.NewValidationError("IsCustomized", fmt.Sprintf("field %s.%s is customized and cannot be overwritten by pack %q", objectAPIName, fieldAPIName, ownerPack))
	}
	if state.isDeleted {
		return apperrors.NewValidationError("IsDeleted", fmt.Sprintf("field %s.%s is deleted and cannot be adopted by pack %q", objectAPIName, fieldAPIName, ownerPack))
	}
	return nil
}

func (r *SchemaRepository) ensurePackOwnedPhysicalColumn(ctx context.Context, tableName string, col schema.ColumnDefinition) error {
	if err := validateCatalogColumnWriteProtectionProjection(tableName, col); err != nil {
		return err
	}
	if err := validateDropColumnNames(tableName, col.Name); err != nil {
		return err
	}
	if err := r.ValidateFieldDefinition(col); err != nil {
		return err
	}

	exists, err := r.checkColumnExists(tableName, col.Name)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to check column existence for %s.%s", tableName, col.Name), err)
	}
	if !exists {
		desiredType := r.ResolveColumnSqlType(col)
		stmts := PlanAddColumnStatements(tableName, col.Name, desiredType, col.Nullable, col.Default, col.CollationName)
		for _, stmt := range stmts {
			if _, err := r.db.ExecContext(ctx, stmt); err != nil {
				return apperrors.NewInternalError(fmt.Sprintf("failed to add column to table %s", tableName), err)
			}
		}
	}

	if col.Unique {
		idxName := schema.GenerateUniqueIndexName(tableName, col.Name)
		idxDDL := fmt.Sprintf("ALTER TABLE `%s` ADD UNIQUE KEY `%s` (`%s`)", tableName, idxName, col.Name)
		if _, err := r.db.ExecContext(ctx, idxDDL); err != nil && !isDuplicateDDLError(err) {
			return apperrors.NewInternalError("failed to add Unique Index", err)
		}
	}

	if len(col.ReferenceTo) == 1 && !col.IsPolymorphic && !strings.EqualFold(col.ReferenceTo[0], "parent") {
		safeColName := schema.GenerateSafeIdentifier(col.Name, "col")
		rawFKName := fmt.Sprintf("fk_%s_%s", tableName, safeColName)
		fkName := schema.GenerateSafeIdentifier(rawFKName, "fk")
		fkDDL := fmt.Sprintf("ALTER TABLE `%s` ADD CONSTRAINT `%s` FOREIGN KEY (`%s`) REFERENCES `%s` (`%s`)",
			tableName, fkName, safeColName, col.ReferenceTo[0], constants.ColID)
		if col.OnDelete != "" {
			fkDDL += fmt.Sprintf(" ON DELETE %s", col.OnDelete)
		}
		if _, err := r.db.ExecContext(ctx, fkDDL); err != nil && !isDuplicateDDLError(err) {
			return apperrors.NewInternalError("failed to add foreign key constraint", err)
		}
	}

	return nil
}

func isDuplicateDDLError(err error) bool {
	if err == nil {
		return false
	}
	msg := err.Error()
	return strings.Contains(msg, "Duplicate") || strings.Contains(msg, "already exists")
}