package packs

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"reflect"
	"sort"
	"strings"
	"time"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	domainschema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/textutil"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/formulastorage"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/assertion"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
	"github.com/pingcap-inc/nexuscrm-controlplane/packspec"
)

// Pack spec JSON is owned by the packspec package (cycle-free).

// ---- Pack Spec Types (logical metadata, no physical SQL) ----

type ObjectSpec struct {
	APIName                string                         `json:"apiName"`
	KeyPrefix              string                         `json:"keyPrefix,omitempty"`
	Label                  string                         `json:"label"`
	PluralLabel            string                         `json:"pluralLabel,omitempty"`
	TableType              string                         `json:"tableType"`
	Category               string                         `json:"category"`
	CustomSetting          bool                           `json:"customSetting,omitempty"`
	SharingModel           string                         `json:"sharingModel"`
	EnableHierarchySharing bool                           `json:"enableHierarchySharing"`
	Description            string                         `json:"description"`
	AccessPolicy           *models.ObjectAccessPolicy     `json:"accessPolicy,omitempty"`
	ReadSource             *models.ObjectReadSourceConfig `json:"readSource,omitempty"`
	PathField              string                         `json:"pathField,omitempty"`
	DisplayNameField       string                         `json:"displayNameField,omitempty"`
	LockRule               string                         `json:"lockRule,omitempty"`
	ListFields             []string                       `json:"listFields"`
	SearchAllDefaultFields []string                       `json:"searchAllDefaultFields,omitempty"`
	DefaultChildRows       []models.DefaultChildRow       `json:"defaultChildRows,omitempty"`
	ChildCloneRules        []models.ChildCloneRule        `json:"childCloneRules,omitempty"`
	ExclusiveSiblingFields []models.ExclusiveSiblingField `json:"exclusiveSiblingFields,omitempty"`
	Fields                 []FieldSpec                    `json:"fields"`
	Indices                []domainschema.IndexDefinition `json:"indices,omitempty"`
	ManagedRelationships   []relationshipSpec             `json:"managedRelationships,omitempty"`
}

func (s ObjectSpec) MarshalJSON() ([]byte, error) {
	type alias ObjectSpec
	raw, err := json.Marshal(alias(s))
	if err != nil {
		return nil, err
	}
	var doc map[string]json.RawMessage
	if err := json.Unmarshal(raw, &doc); err != nil {
		return nil, err
	}
	if s.ListFields == nil {
		delete(doc, "listFields")
		return json.Marshal(doc)
	}
	encoded, err := json.Marshal(s.ListFields)
	if err != nil {
		return nil, err
	}
	doc["listFields"] = encoded
	return json.Marshal(doc)
}

type listFieldsObjectSpecOccurrence struct {
	specIndex int
}

func ValidateListFieldsForSpecs(specs []ObjectSpec) error {
	return ValidateListFieldsForSpecsWithSnapshot(specs, specs)
}

func ValidateListFieldsForSpecsWithSnapshot(specs []ObjectSpec, snapshotSpecs []ObjectSpec) error {
	if len(specs) == 0 {
		return nil
	}
	if len(snapshotSpecs) == 0 {
		snapshotSpecs = specs
	}
	if err := validateListFieldsSingleObjectSpecOwnership(specs, snapshotSpecs); err != nil {
		return fmt.Errorf("ListFields validation failed:\n  - %s", err)
	}
	return nil
}

func validateListFieldsSingleObjectSpecOwnership(specs []ObjectSpec, snapshotSpecs []ObjectSpec) error {
	occurrences := make(map[string][]listFieldsObjectSpecOccurrence)
	for specIndex, spec := range snapshotSpecs {
		objectAPIName := strings.TrimSpace(spec.APIName)
		if objectAPIName == "" {
			continue
		}
		key := strings.ToLower(objectAPIName)
		occurrences[key] = append(occurrences[key], listFieldsObjectSpecOccurrence{
			specIndex: specIndex,
		})
	}

	var errs []string
	reported := map[string]bool{}
	for _, spec := range specs {
		objectAPIName := strings.TrimSpace(spec.APIName)
		if objectAPIName == "" || spec.ListFields == nil {
			continue
		}
		key := strings.ToLower(objectAPIName)
		if reported[key] {
			continue
		}
		matches := occurrences[key]
		if len(matches) <= 1 {
			continue
		}
		specIndexes := make([]string, 0, len(matches))
		for _, occurrence := range matches {
			specIndexes = append(specIndexes, fmt.Sprintf("spec[%d]", occurrence.specIndex))
		}
		errs = append(errs, fmt.Sprintf(
			"object %s declares listFields but is authored by %d object_spec specs (%s); ListFields is object-level metadata and must be declared only when the object has a single object_spec authoring surface",
			objectAPIName,
			len(matches),
			strings.Join(specIndexes, ", "),
		))
		reported[key] = true
	}
	if len(errs) == 0 {
		return nil
	}
	return fmt.Errorf("%s", strings.Join(errs, "\n  - "))
}

type listFieldsSourceObjectSpecOccurrence struct {
	businessKey string
	specIndex   int
}

func validateInstallableSourceBackedListFields(source *ResolvedPackSource) []PackDiagnostic {
	if source == nil || source.Manifest == nil {
		return nil
	}
	parsedObjectSpecs := make(map[string][]ObjectSpec)
	for _, component := range source.Manifest.Components {
		if component.Kind != ComponentKindObjectSpec {
			continue
		}
		payload, err := sourceBackedComponentPayload(source, component)
		p := componentPayloadLabel(component)
		if err != nil {
			return []PackDiagnostic{diag("component.payload.unavailable", p, err.Error())}
		}
		specs, err := parseObjectSpecsJSON(payload, componentPayloadLabel(component))
		if err != nil {
			return []PackDiagnostic{diag("component.object_spec.invalid_json", p, err.Error())}
		}
		parsedObjectSpecs[strings.TrimSpace(component.BusinessKey)] = specs
	}
	if len(parsedObjectSpecs) == 0 {
		return nil
	}
	if err := validateSourceBackedListFieldsSingleObjectSpecOwnership(parsedObjectSpecs); err != nil {
		return []PackDiagnostic{diag("component.object_spec.list_fields_split_object", "object_spec", err.Error())}
	}
	return nil
}

func validateInstallableSourceBackedChildRelationshipNames(source *ResolvedPackSource) []PackDiagnostic {
	if source == nil || source.Manifest == nil {
		return nil
	}
	var objectSpecs []ObjectSpec
	for _, component := range source.Manifest.Components {
		if component.Kind != ComponentKindObjectSpec {
			continue
		}
		payload, err := sourceBackedComponentPayload(source, component)
		p := componentPayloadLabel(component)
		if err != nil {
			return []PackDiagnostic{diag("component.payload.unavailable", p, err.Error())}
		}
		specs, err := parseObjectSpecsJSON(payload, componentPayloadLabel(component))
		if err != nil {
			return []PackDiagnostic{diag("component.object_spec.invalid_json", p, err.Error())}
		}
		objectSpecs = append(objectSpecs, specs...)
	}
	if len(objectSpecs) == 0 {
		return nil
	}
	decls := ChildRelFKDeclsFromObjectSpecs(objectSpecs)
	if err := ValidateChildRelationshipNameSyntax(decls); err != nil {
		return []PackDiagnostic{diag("component.object_spec.child_relationship_name_invalid", "object_spec", err.Error())}
	}
	if err := ValidateChildRelationshipNameParentFieldCollisions(decls, objectSpecs); err != nil {
		return []PackDiagnostic{diag("component.object_spec.child_relationship_name_invalid", "object_spec", err.Error())}
	}
	if err := ValidateChildRelationshipUniqueness(decls); err != nil {
		return []PackDiagnostic{diag("component.object_spec.child_relationship_name_duplicate", "object_spec", err.Error())}
	}
	return nil
}

func validateSourceBackedListFieldsSingleObjectSpecOwnership(specsByBusinessKey map[string][]ObjectSpec) error {
	occurrences := map[string][]listFieldsSourceObjectSpecOccurrence{}
	keys := make([]string, 0, len(specsByBusinessKey))
	for key := range specsByBusinessKey {
		keys = append(keys, key)
	}
	sort.Strings(keys)

	for _, businessKey := range keys {
		specs := specsByBusinessKey[businessKey]
		for specIndex, spec := range specs {
			objectAPIName := strings.TrimSpace(spec.APIName)
			if objectAPIName == "" {
				continue
			}
			key := strings.ToLower(objectAPIName)
			occurrences[key] = append(occurrences[key], listFieldsSourceObjectSpecOccurrence{
				businessKey: businessKey,
				specIndex:   specIndex,
			})
		}
	}

	var errs []string
	reported := map[string]bool{}
	for _, businessKey := range keys {
		specs := specsByBusinessKey[businessKey]
		for _, spec := range specs {
			objectAPIName := strings.TrimSpace(spec.APIName)
			if objectAPIName == "" || spec.ListFields == nil {
				continue
			}
			key := strings.ToLower(objectAPIName)
			if reported[key] {
				continue
			}
			matches := occurrences[key]
			if len(matches) <= 1 {
				continue
			}
			labels := make([]string, 0, len(matches))
			for _, occurrence := range matches {
				labels = append(labels, fmt.Sprintf("%q spec[%d]", occurrence.businessKey, occurrence.specIndex))
			}
			errs = append(errs, fmt.Sprintf(
				"object %s declares listFields in object_spec %q but is authored by %d object_spec components/specs (%s); ListFields is object-level metadata and must use one object_spec authoring surface",
				objectAPIName,
				businessKey,
				len(matches),
				strings.Join(labels, ", "),
			))
			reported[key] = true
		}
	}
	if len(errs) == 0 {
		return nil
	}
	return fmt.Errorf("%s", strings.Join(errs, "\n  - "))
}

type FieldSpec struct {
	APIName               string                        `json:"apiName"`
	Label                 string                        `json:"label"`
	Type                  string                        `json:"type"` // logical: string, currency, reference, phone, ...
	Length                int                           `json:"length,omitempty"`
	Required              bool                          `json:"required,omitempty"`
	Unique                bool                          `json:"unique,omitempty"`
	TrackHistory          bool                          `json:"trackHistory,omitempty"`
	IsNameField           bool                          `json:"isNameField,omitempty"`
	Createable            *bool                         `json:"createable,omitempty"`
	Updateable            *bool                         `json:"updateable,omitempty"`
	Default               string                        `json:"default,omitempty"`
	HelpText              string                        `json:"inlineHelpText,omitempty"`
	SiblingMaxParentField string                        `json:"siblingMaxParentField,omitempty"`
	Formula               string                        `json:"formula,omitempty"`
	ReturnType            string                        `json:"returnType,omitempty"`
	FormulaMode           string                        `json:"formulaMode,omitempty"`
	FormulaStorageMode    string                        `json:"formulaStorageMode,omitempty"`
	ProjectionConfig      *models.FieldProjectionConfig `json:"projection,omitempty"`
	Validator             string                        `json:"validator,omitempty"`
	ValidatorConfig       string                        `json:"validatorConfig,omitempty"`
	ControllingField      string                        `json:"controllingField,omitempty"`
	ReferenceTo           []string                      `json:"referenceTo,omitempty"`
	WritableReferenceTo   []string                      `json:"writableReferenceTo,omitempty"`
	DeleteConstraint      string                        `json:"deleteConstraint,omitempty"`
	RelationshipName      string                        `json:"relationshipName,omitempty"`
	ChildRelationshipName string                        `json:"childRelationshipName,omitempty"`
	IsPolymorphic         bool                          `json:"isPolymorphic,omitempty"`
	WriteProtectionPolicy string                        `json:"writeProtectionPolicy,omitempty"`
	ReadProtectionPolicy  string                        `json:"readProtectionPolicy,omitempty"`
	Options               models.PicklistOptions        `json:"options,omitempty"`
	RollupConfig          *models.RollupConfig          `json:"rollupConfig,omitempty"`
	// IsCompound / ComposedFrom — Salesforce-style compound parent declaration.
	// When true, the field is a read-only describe projection composed from the
	// listed member fields; not createable/updateable (enforced by
	// ComputeIntrinsic Layer 1e). Issue #1582 RFC v1.5 PR B.
	IsCompound   bool     `json:"isCompound,omitempty"`
	ComposedFrom []string `json:"composedFrom,omitempty"`
}

// isPolymorphicReference delegates to ResolveFieldRelationshipSemantics so
// pack-authored multi-target lookups and references resolve through the same
// SSoT as tenant CRUD and persisted-read paths. Do not re-derive polymorphic
// ownership from (Type, len(ReferenceTo)) inline.
func (f FieldSpec) isPolymorphicReference() bool {
	sem := models.ResolveFieldRelationshipSemantics(&models.FieldMetadata{
		Type:          constants.SchemaFieldType(f.Type),
		ReferenceTo:   f.ReferenceTo,
		IsPolymorphic: f.IsPolymorphic,
	}, models.LifecycleAuthoredWrite)
	return sem.IsPolymorphic
}

func (f FieldSpec) normalizedDeleteConstraint() constants.DeleteRule {
	return constants.NormalizeDeleteRule(constants.DeleteRule(f.DeleteConstraint))
}

type relationshipSpec struct {
	FieldAPIName    string `json:"fieldApiName"`
	ParentObject    string `json:"parentObject"`
	IsSharingParent *bool  `json:"isSharingParent,omitempty"`
	CascadeDelete   *bool  `json:"cascadeDelete,omitempty"`
}

// ---- Pack ----

// PlatformCorePack creates platform-structural standard objects from a logical
// metadata specification. Product/application objects are owned by app packs.
//
// This is the metadata-first authoring model: the pack spec declares logical metadata,
// and the applier derives physical schema from it.
type PlatformCorePack struct{}

func (p *PlatformCorePack) Name() string       { return "platform-core" }
func (p *PlatformCorePack) Category() Category { return PlatformCore }

func (p *PlatformCorePack) Apply(ctx context.Context, deps ApplyDeps) error {
	specs, err := parseStandardObjectsSpec()
	if err != nil {
		return fmt.Errorf("parse standard objects spec: %w", err)
	}
	return applyObjectSpecs(ctx, deps, "platform-core", filterObjectSpecsByCategory(specs, "system"), true, true)
}

func applyObjectSpecs(ctx context.Context, deps ApplyDeps, packName string, specs []ObjectSpec, preserveCustomized bool, skipTenantOwnedFields bool) error {
	var err error
	ctx, err = preflightMetadataTransactionExecutor(ctx, deps, packName)
	if err != nil {
		return err
	}
	return applyObjectSpecsWithResolver(ctx, deps, packName, specs, preserveCustomized, skipTenantOwnedFields, newObjectSpecFieldResolver(specs))
}

func applyObjectSpecsWithResolver(ctx context.Context, deps ApplyDeps, packName string, specs []ObjectSpec, preserveCustomized bool, skipTenantOwnedFields bool, resolver objectSpecFieldResolver) error {
	return applyObjectSpecsWithResolverAndValidation(ctx, deps, packName, specs, preserveCustomized, skipTenantOwnedFields, resolver, packFieldAuthoringCompat{}, nil)
}

func applyObjectSpecsWithResolverAndValidation(ctx context.Context, deps ApplyDeps, packName string, specs []ObjectSpec, preserveCustomized bool, skipTenantOwnedFields bool, resolver objectSpecFieldResolver, fieldCompat packFieldAuthoringCompat, formulaFunctionBundlesForValidation []formula.FunctionBundle, childCloneRuleValidationSpecs ...[]ObjectSpec) error {
	var err error
	ctx, err = preflightMetadataTransactionExecutor(ctx, deps, packName)
	if err != nil {
		return err
	}
	log.Printf("📦 Applying %s pack (metadata-first objects)...", packName)
	start := time.Now()

	if len(specs) == 0 {
		log.Printf("✅ %s pack applied: 0 objects (%.1fs)", packName, time.Since(start).Seconds())
		return nil
	}
	if err := ValidateRollupCyclesWithResolver(specs, resolver); err != nil {
		return fmt.Errorf("validate rollup cycles: %w", err)
	}
	if err := ValidateDefaultChildRowsForSpecs(ctx, specs, nil, nil, true); err != nil {
		return fmt.Errorf("validate default child rows: %w", err)
	}
	childCloneSpecs := specs
	if len(childCloneRuleValidationSpecs) > 0 && len(childCloneRuleValidationSpecs[0]) > 0 {
		childCloneSpecs = childCloneRuleValidationSpecs[0]
	}
	if err := ValidateChildCloneRulesForSpecsWithSnapshot(ctx, specs, childCloneSpecs, deps.Metadata, deps.Metadata == nil); err != nil {
		return fmt.Errorf("validate child clone rules: %w", err)
	}
	if err := ValidateExclusiveSiblingFieldsForSpecsWithSnapshot(ctx, specs, childCloneSpecs, deps.Metadata, deps.Metadata == nil); err != nil {
		return fmt.Errorf("validate exclusive sibling fields: %w", err)
	}
	if err := ValidateListFieldsForSpecsWithSnapshot(specs, childCloneSpecs); err != nil {
		return fmt.Errorf("validate list fields: %w", err)
	}

	// 1. Validate LockRule expressions before any DB mutations.
	// A non-boolean or syntactically-broken LockRule in an external pack would
	// pass apply/verify silently and only fail at runtime (denying all edits/deletes
	// when the formula errors). Validating here (before BatchCreatePhysicalTables
	// and BatchRegisterTables) ensures a bad expression aborts cleanly without
	// leaving a partially-applied object that has a physical table but no matching
	// object/field metadata. Uses the runtime-probe gate from ValidateLockRuleExpression.
	if err := validateSpecLockRulesWithFunctionBundles(specs, resolver, formulaFunctionBundlesForValidation); err != nil {
		return fmt.Errorf("validate locked conditions: %w", err)
	}
	if err := validateSpecReadSourcesForApply(packName, specs); err != nil {
		return fmt.Errorf("validate read sources: %w", err)
	}
	// Issue #1110 / PR #1133: object.displayNameField must resolve to a real,
	// label-renderable field within spec.Fields. Without this gate a
	// source-backed pack could install with a blank, padded, unresolved, or
	// wrong-type displayNameField, persist it to _System_Object via
	// buildObjectMetadataWithResolver, and break frontend lookups/search/
	// recycle-bin when getListRecordName tries to read _Display/_Name
	// siblings that don't exist.
	if err := validateSpecDisplayNameFieldsWithResolver(specs, resolver); err != nil {
		return fmt.Errorf("validate display name fields: %w", err)
	}

	// 2. Convert logical spec → TableDefinition (derive physical types)
	defs, err := convertSpecToTableDefinitionsWithResolver(specs, deps.Schema, resolver)
	if err != nil {
		return fmt.Errorf("convert spec to table definitions: %w", err)
	}
	authoredFields := newObjectSpecAuthoredFieldSet(specs)
	uniqueConstraintDefs, err := objectSpecUniquePhysicalConstraintDefinitionsForApply(deps.DB, packName, defs, preserveCustomized, authoredFields)
	if err != nil {
		return fmt.Errorf("resolve unique physical constraints: %w", err)
	}
	formulaStorageTransitions, err := planObjectSpecFormulaStorageTransitions(ctx, deps.DB, uniqueConstraintDefs)
	if err != nil {
		return fmt.Errorf("plan formula storage transitions: %w", err)
	}
	for _, transition := range formulaStorageTransitions {
		if !offlinecandidateauthority.FormulaStorageTransitionAllowed(
			ctx, packName, transition.objectAPIName, transition.fieldAPIName,
		) {
			return fmt.Errorf(
				"formula storage transition for %s.%s requires stopped-Runtime offline candidate execution with an exact Catalog-owned plan",
				transition.objectAPIName,
				transition.fieldAPIName,
			)
		}
	}
	if err := rebuildObjectSpecFormulaStorageTransitionSources(ctx, deps, formulaStorageTransitions); err != nil {
		return fmt.Errorf("prepare authoritative formula storage transition source: %w", err)
	}
	physicalDefs := make([]domainschema.TableDefinition, 0, len(defs))
	for _, def := range uniqueConstraintDefs {
		if def.ReadSource != nil {
			continue
		}
		physicalCols := make([]domainschema.ColumnDefinition, 0, len(def.Columns))
		for _, col := range def.Columns {
			resolved, err := formulastorage.Resolve(formulastorage.SpecFromColumn(col))
			if !resolved.RequiresPhysicalColumn() {
				continue
			}
			physicalCols = append(physicalCols, col)
			_ = err
		}
		def.Columns = physicalCols
		physicalDefs = append(physicalDefs, def)
	}

	// 3. Preflight or explicitly repair existing duplicate data before
	// unique-index DDL. Older installs could accumulate duplicates while
	// object_spec.unique was not enforced, so the repair gate must run before
	// any schema or metadata mutation can make uniqueness visible.
	if err := preflightOrRepairObjectSpecUniqueLegacyDuplicates(deps.DB, packName, uniqueConstraintDefs, deps.Options.RepairLegacyUniqueDuplicates, newObjectSpecRequiredFieldIndex(specs)); err != nil {
		return fmt.Errorf("prepare unique legacy duplicates: %w", err)
	}

	// 4. Create physical tables (DDL first, same as custom object path)
	if err := deps.Schema.BatchCreatePhysicalTables(ctx, physicalDefs); err != nil {
		return fmt.Errorf("create physical tables: %w", err)
	}

	// 5. Register tables in _System_Table
	if err := deps.Schema.BatchRegisterTables(physicalDefs, nil); err != nil {
		log.Printf("   ⚠️  BatchRegisterTables: %v (may already exist)", err)
	}

	// 6. Reconcile physical schema before publishing field metadata. Runtime
	// code treats _System_Field.IsUnique as the SSoT for unique-key behavior,
	// so the backing unique index must already exist before IsUnique=true can
	// become visible in live metadata.
	if err := schemaProvisioner.ReconcileTableColumns(deps.DB, physicalDefs); err != nil {
		return fmt.Errorf("reconcile table columns: %w", err)
	}
	if err := dropOwnedObjectSpecUniqueDowngrades(deps.DB, packName, uniqueConstraintDefs); err != nil {
		return fmt.Errorf("drop stale owned unique indexes: %w", err)
	}
	if err := applyObjectSpecFormulaStorageTransitions(ctx, deps.DB, formulaStorageTransitions); err != nil {
		return fmt.Errorf("backfill formula storage transitions: %w", err)
	}
	if err := reconcilePackTableIndices(deps.DB, physicalDefs); err != nil {
		return fmt.Errorf("reconcile table indices: %w", err)
	}
	deps.Schema.EnsureRelationshipFKIndexes(physicalDefs)
	if err := assertObjectSpecUniquePhysicalConstraints(deps.DB, packName, uniqueConstraintDefs); err != nil {
		return fmt.Errorf("verify unique physical constraints: %w", err)
	}

	// Build field metadata before persisting readSource object metadata so
	// cutover-blocking live-field checks can fail closed before ReadSource goes live.
	fields, err := buildFieldMetadataWithResolver(packName, specs, deps.Schema, skipTenantOwnedFields, resolver, fieldCompat)
	if err != nil {
		return fmt.Errorf("build field metadata: %w", err)
	}
	if err := validateReadSourceLegacySurfaceCompatibility(ctx, deps.DB, packName, defs, fields); err != nil {
		return fmt.Errorf("validate readSource legacy surfaces: %w", err)
	}

	// Register object metadata in _System_Object
	objects := buildObjectMetadataWithResolver(specs, resolver)
	if err := preservePackOwnedAccessPolicySidecars(ctx, deps.DB, objects); err != nil {
		return fmt.Errorf("preserve pack-owned AccessPolicy sidecars: %w", err)
	}
	if err := applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(ctx, deps, packName, defs, objects, fields, preserveCustomized); err != nil {
		return err
	}
	if hasReadSourceDefinitions(defs) {
		if err := dropReadSourceLegacyPhysicalTables(ctx, deps.DB, defs); err != nil {
			return fmt.Errorf("drop readSource legacy physical tables: %w", err)
		}
	}

	if deps.Options.SkipPolymorphicDiscriminatorBackfill {
		log.Printf("controlplane: operator override enabled; skipping polymorphic discriminator backfill for pack %s", packName)
	} else if err := BackfillPolymorphicDiscriminators(ctx, deps.DB, physicalDefs); err != nil {
		return fmt.Errorf("backfill polymorphic discriminators: %w", err)
	}

	// 8. Reconcile managed access policies (SharingModel, EnableHierarchySharing)
	// Must happen HERE because pack object defs are not in substrate table defs.
	if err := ReconcilePackAccessPolicies(deps.DB, defs); err != nil {
		return fmt.Errorf("reconcile access policies: %w", err)
	}

	// 9. Reconcile managed relationship semantics.
	if err := reconcileManagedRelationships(ctx, deps.DB, specs, defs); err != nil {
		return fmt.Errorf("reconcile managed relationships: %w", err)
	}

	log.Printf("✅ %s pack applied: %d objects (%.1fs)", packName, len(specs), time.Since(start).Seconds())
	return nil
}

type metadataTransactionExecutorPreflightContextKey struct{}

// preflightMetadataTransactionExecutor proves that the transaction manager can
// open an attempt and expose a concrete persistence executor before object-spec
// resolution, ownership work, or schema mutation begins. The actual metadata
// transaction validates its executor again as defense in depth.
func preflightMetadataTransactionExecutor(ctx context.Context, deps ApplyDeps, packName string) (context.Context, error) {
	if preflighted, _ := ctx.Value(metadataTransactionExecutorPreflightContextKey{}).(bool); preflighted {
		return ctx, nil
	}
	if isNilInterface(deps.TxManager) {
		return ctx, fmt.Errorf("metadata apply for %s requires a transaction manager", packName)
	}
	if err := deps.TxManager.WithTransaction(ctx, func(txCtx context.Context) error {
		tx := deps.TxManager.ExtractTx(txCtx)
		if isNilInterface(tx) {
			return fmt.Errorf("extract transaction executor for metadata apply %s", packName)
		}
		if _, ok := tx.(persistence.Executor); !ok {
			return fmt.Errorf("extract transaction executor for metadata apply %s", packName)
		}
		return nil
	}); err != nil {
		return ctx, err
	}
	return context.WithValue(ctx, metadataTransactionExecutorPreflightContextKey{}, true), nil
}

// applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt publishes object
// and field metadata through one validated pessimistic transaction. This lets
// injected persistence adapters take blocking semantic-projection locks without
// opening a nested transaction. Fresh-field history activation remains a
// separate validated transaction. Validation happens before either persistence
// writer can fall back to its root DB executor.
func applyBatchSaveFieldMetadataWithCustomizationPolicyAttempt(ctx context.Context, deps ApplyDeps, packName string, defs []domainschema.TableDefinition, objects []*models.ObjectMetadata, fields []metadata.FieldWithContext, preserveCustomized bool) error {
	if isNilInterface(deps.TxManager) {
		return fmt.Errorf("metadata apply for %s requires a transaction manager", packName)
	}

	var expectedFieldsByObject map[string]map[string]struct{}
	if hasReadSourceDefinitions(defs) {
		expectedFieldsByObject = expectedReadSourceFieldsByObject(fields)
	}
	var pendingFreshHistoryFieldIDs []string
	if err := deps.TxManager.WithPessimisticTransaction(ctx, func(txCtx context.Context) error {
		tx := deps.TxManager.ExtractTx(txCtx)
		if isNilInterface(tx) {
			return fmt.Errorf("extract transaction executor for metadata apply %s", packName)
		}
		txExec, ok := tx.(persistence.Executor)
		if !ok {
			return fmt.Errorf("extract transaction executor for metadata apply %s", packName)
		}
		phaseOneFields, pending, err := prepareFreshHistoryTrackingFields(txCtx, txExec, fields)
		if err != nil {
			return err
		}

		if err := deps.Schema.BatchSaveObjectMetadataWithCustomizationPolicy(txCtx, objects, tx, preserveCustomized); err != nil {
			return fmt.Errorf("save object metadata: %w", err)
		}
		if hasReadSourceDefinitions(defs) {
			if err := reconcileReadSourceLegacyMetadataSurfaces(txCtx, txExec, packName, defs, expectedFieldsByObject); err != nil {
				return fmt.Errorf("reconcile readSource legacy metadata surfaces: %w", err)
			}
		}
		if err := deps.Schema.BatchSaveFieldMetadataWithCustomizationPolicy(txCtx, phaseOneFields, tx, preserveCustomized); err != nil {
			return fmt.Errorf("save field metadata: %w", err)
		}
		pendingFreshHistoryFieldIDs = pending
		return nil
	}); err != nil {
		return err
	}
	if len(pendingFreshHistoryFieldIDs) == 0 {
		return nil
	}

	return deps.TxManager.WithTransaction(ctx, func(txCtx context.Context) error {
		tx := deps.TxManager.ExtractTx(txCtx)
		if isNilInterface(tx) {
			return fmt.Errorf("extract transaction executor for history tracking activation %s", packName)
		}
		if _, ok := tx.(persistence.Executor); !ok {
			return fmt.Errorf("extract transaction executor for history tracking activation %s", packName)
		}
		for _, fieldID := range pendingFreshHistoryFieldIDs {
			if _, err := deps.Schema.ApplyHistoryTrackingTransition(txCtx, fieldID, true, tx, preserveCustomized, false); err != nil {
				return fmt.Errorf("enable history tracking for fresh field %s: %w", fieldID, err)
			}
		}
		return nil
	})
}

// prepareFreshHistoryTrackingFields keeps an authored true value for existing
// fields, but creates a fresh field in the only valid initial state: disabled.
// The caller enables the returned IDs in one separate committed transaction.
func prepareFreshHistoryTrackingFields(ctx context.Context, exec persistence.Executor, fields []metadata.FieldWithContext) ([]metadata.FieldWithContext, []string, error) {
	prepared := append([]metadata.FieldWithContext(nil), fields...)
	pending := make([]string, 0)
	for i, field := range fields {
		if field.Field == nil || !field.Field.TrackHistory {
			continue
		}
		fieldID := field.FieldID
		if fieldID == "" {
			fieldID = domainschema.GenerateFieldID(field.TableName, field.ColumnDefinition.Name)
		}
		var marker int
		err := exec.QueryRowContext(ctx, "SELECT 1 FROM _System_Field WHERE Id = ?", fieldID).Scan(&marker)
		switch {
		case err == nil:
			continue
		case errors.Is(err, sql.ErrNoRows):
			fieldCopy := field
			metadataCopy := *field.Field
			metadataCopy.TrackHistory = false
			fieldCopy.Field = &metadataCopy
			prepared[i] = fieldCopy
			pending = append(pending, fieldID)
		default:
			return nil, nil, fmt.Errorf("classify fresh history field %s: %w", fieldID, err)
		}
	}
	sort.Strings(pending)
	return prepared, pending, nil
}

func isNilInterface(value interface{}) bool {
	if value == nil {
		return true
	}
	reflected := reflect.ValueOf(value)
	switch reflected.Kind() {
	case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice:
		return reflected.IsNil()
	default:
		return false
	}
}

func (p *PlatformCorePack) Verify(_ context.Context, deps VerifyDeps) error {
	if err := validatePlatformCompatibilityDeclarations(); err != nil {
		return fmt.Errorf("verify platform compatibility declarations: %w", err)
	}
	result := &assertion.AssertionResult{}
	schemaProvisioner.AssertManagedAccessPolicyConsistency(deps.DB, result, packspec.StandardObjectPoliciesByCategory("system"))
	rels := append(
		packspec.StandardRelationshipSemanticsByCategory("system"),
		domainschema.ManagedStandardRelationshipSemanticsList()...,
	)
	schemaProvisioner.AssertManagedRelationshipConsistency(deps.DB, result, rels, managedTableDefinitions())
	return assertionResultError(p.Name(), result)
}

func assertionResultError(packName string, result *assertion.AssertionResult) error {
	var errs []string
	for _, violation := range result.Violations {
		if violation.Severity == constants.SeverityError {
			errs = append(errs, fmt.Sprintf("%s: %s", violation.Object, violation.Description))
		}
	}
	if len(errs) == 0 {
		return nil
	}
	return fmt.Errorf("%s managed metadata verify failed:\n  - %s", packName, strings.Join(errs, "\n  - "))
}

// ---- Spec Parsing ----

func parseStandardObjectsSpec() ([]ObjectSpec, error) {
	return parseObjectSpecsJSON(packspec.StandardObjectsSpecJSON(), "standard_objects_spec.json")
}

func parseObjectSpecsJSON(raw []byte, source string) ([]ObjectSpec, error) {
	var specs []ObjectSpec
	if err := json.Unmarshal(raw, &specs); err != nil {
		return nil, apperrors.NewInternalError("failed to parse "+source, err)
	}
	for _, spec := range specs {
		for _, field := range spec.Fields {
			if strings.TrimSpace(field.DeleteConstraint) != "" && field.normalizedDeleteConstraint() == "" {
				return nil, fmt.Errorf("parse %s: %s.%s deleteConstraint %q is invalid", source, spec.APIName, field.APIName, field.DeleteConstraint)
			}
		}
	}
	return specs, nil
}

func filterObjectSpecsByCategory(specs []ObjectSpec, categories ...string) []ObjectSpec {
	allowed := make(map[string]struct{}, len(categories))
	for _, category := range categories {
		allowed[strings.ToLower(category)] = struct{}{}
	}
	filtered := make([]ObjectSpec, 0, len(specs))
	for _, spec := range specs {
		if _, ok := allowed[strings.ToLower(spec.Category)]; ok {
			filtered = append(filtered, spec)
		}
	}
	return filtered
}

// ---- Logical → Physical Conversion ----

func convertSpecToTableDefinitions(specs []ObjectSpec, sm *metadata.SchemaManager) ([]domainschema.TableDefinition, error) {
	return convertSpecToTableDefinitionsWithResolver(specs, sm, newObjectSpecFieldResolver(specs))
}

func convertSpecToTableDefinitionsWithResolver(specs []ObjectSpec, sm *metadata.SchemaManager, resolver objectSpecFieldResolver) ([]domainschema.TableDefinition, error) {
	defs := make([]domainschema.TableDefinition, 0, len(specs))

	for _, spec := range specs {
		// Mirror Salesforce Metadata API: <sharingModel> is mandatory on
		// CustomObject deploy. NexusCRM L0 install-pack rejects pack object
		// specs that omit sharingModel rather than infer one — SharingModel
		// is L1 authoritative product-default with L2 admin override, never
		// an L0-inferred fact (CLAUDE.md "L0 is blind to L2 instances",
		// "one fact, one SSoT, one read path"). describeSObject does not
		// expose org-wide-default; pack tooling must source it via Tooling
		// API EntityDefinition.InternalSharingModel or declare a defensive
		// default explicitly.
		if spec.SharingModel == "" {
			return nil, fmt.Errorf(
				"object %s: pack must declare sharingModel (L0 does not infer L1 authorization policy; "+
					"valid values: PublicReadWrite, PublicRead, Private, ControlledByParent, FolderSharing, FolderContentSharing)",
				spec.APIName,
			)
		}
		def := domainschema.TableDefinition{
			TableName: spec.APIName,
			// Source-backed verify rebuilds TableDefinition from ObjectSpec and
			// later compares live _System_Object.KeyPrefix against def.KeyPrefix.
			KeyPrefix:              strings.TrimSpace(spec.KeyPrefix),
			TableType:              spec.TableType,
			Category:               spec.Category,
			Label:                  spec.Label,
			Description:            spec.Description,
			IsManaged:              true,
			SharingModel:           spec.SharingModel,
			EnableHierarchySharing: spec.EnableHierarchySharing,
			AccessPolicy:           spec.AccessPolicy,
			LockRule:               spec.LockRule,
			DisplayNameField:       spec.DisplayNameField,
			ReadSource:             cloneObjectReadSourceConfig(spec.ReadSource),
			Indices:                append([]domainschema.IndexDefinition(nil), spec.Indices...),
		}

		// System columns first (Id, OwnerId, CreatedDate, ...)
		injectedSystemCols := managedObjectSystemColumnsForSpec(spec)
		for _, sc := range injectedSystemCols {
			def.Columns = append(def.Columns, sc)
		}

		// Business fields: derive physical type from logical type
		for _, field := range spec.Fields {
			col, err := buildPackFieldSpecColumnDefinitionWithResolver(sm, spec.APIName, field, "", resolver)
			if err != nil {
				return nil, fmt.Errorf("object %s field %s: %w", spec.APIName, field.APIName, err)
			}
			def.Columns = append(def.Columns, col)
		}

		def = domainschema.ExpandPolymorphicDiscriminatorColumns(def)
		defs = append(defs, def)
	}

	return defs, nil
}

func reconcileReadSourceLegacySurfaces(ctx context.Context, db *sql.DB, packName string, defs []domainschema.TableDefinition, fields []metadata.FieldWithContext) error {
	if db == nil {
		return fmt.Errorf("db is unavailable for readSource legacy-surface reconcile")
	}

	expectedFieldsByObject := expectedReadSourceFieldsByObject(fields)
	if err := validateReadSourceLegacySurfaceCompatibilityWithExpectedFields(ctx, db, packName, defs, expectedFieldsByObject, requiredFieldsByObjectFromFieldContexts(fields)); err != nil {
		return err
	}
	if err := reconcileReadSourceLegacyMetadataSurfaces(ctx, db, packName, defs, expectedFieldsByObject); err != nil {
		return err
	}
	return dropReadSourceLegacyPhysicalTables(ctx, db, defs)
}

func reconcileReadSourceLegacyMetadataSurfaces(ctx context.Context, exec persistence.Executor, packName string, defs []domainschema.TableDefinition, expectedFieldsByObject map[string]map[string]struct{}) error {
	objectRegistry, err := loadPackObjectRegistryWithExecutor(exec)
	if err != nil {
		return fmt.Errorf("load object registry for readSource legacy-surface reconcile: %w", err)
	}

	for _, def := range defs {
		if def.ReadSource == nil {
			continue
		}
		wantReadSource, err := canonicalReadSourceJSON(def.ReadSource)
		if err != nil {
			return fmt.Errorf("%s object %s ReadSource canonicalization failed: %w", packName, def.TableName, err)
		}
		objectRow, ok := objectRegistry[strings.ToLower(strings.TrimSpace(def.TableName))]
		if !ok || objectRow.deleted {
			return fmt.Errorf("%s _System_Object row for %s missing live ReadSource cutover before dropping legacy surface", packName, def.TableName)
		}
		liveReadSource := ""
		if objectRow.readSource.Valid {
			liveReadSource, err = canonicalStoredReadSourceJSON(objectRow.readSource.String)
			if err != nil {
				return fmt.Errorf("%s _System_Object row for %s ReadSource is invalid: %w", packName, def.TableName, err)
			}
		}
		if liveReadSource != wantReadSource {
			return fmt.Errorf("%s _System_Object row for %s ReadSource = %s, want %q before dropping legacy surface", packName, def.TableName, formatPackColumnDefault(objectRow.readSource.Valid, objectRow.readSource.String), wantReadSource)
		}

		expectedFields := expectedFieldsByObject[strings.ToLower(strings.TrimSpace(def.TableName))]
		liveFields, err := loadReadSourceLiveFields(ctx, exec, def.TableName)
		if err != nil {
			return fmt.Errorf("load live field registry for %s: %w", def.TableName, err)
		}
		stalePackFields, _, _ := classifyReadSourceLiveFields(liveFields, expectedFields, packName)

		if _, err := exec.ExecContext(
			ctx,
			fmt.Sprintf("UPDATE `%s` SET `%s` = 1, `%s` = CURRENT_TIMESTAMP WHERE `%s` = ? AND COALESCE(`%s`, 0) = 0",
				constants.TableTable,
				constants.FieldIsDeleted,
				constants.FieldLastModifiedDate,
				constants.FieldSysTable_TableName,
				constants.FieldIsDeleted,
			),
			def.TableName,
		); err != nil {
			return fmt.Errorf("soft-delete stale table registry %s: %w", def.TableName, err)
		}

		for _, row := range stalePackFields {
			if _, err := exec.ExecContext(
				ctx,
				`UPDATE _System_Field f
JOIN _System_Object o ON o.Id = f.ObjectId
SET f.IsDeleted = 1, f.LastModifiedDate = CURRENT_TIMESTAMP
WHERE o.ApiName = ? AND f.ApiName = ? AND COALESCE(f.IsDeleted, 0) = 0
  AND ((COALESCE(f.OwnerKind, '') = ? AND COALESCE(f.OwnerPack, '') = ?)
    OR (COALESCE(f.OwnerKind, '') = ? AND COALESCE(f.OwnerPack, '') = ?))`,
				def.TableName,
				row.fieldAPIName,
				constants.FieldOwnerKindPack,
				packName,
				constants.FieldOwnerKindPackRemoving,
				packName,
			); err != nil {
				return fmt.Errorf("soft-delete stale field %s.%s: %w", def.TableName, row.fieldAPIName, err)
			}
		}
	}

	return nil
}

func dropReadSourceLegacyPhysicalTables(ctx context.Context, exec persistence.Executor, defs []domainschema.TableDefinition) error {
	for _, def := range defs {
		if def.ReadSource == nil {
			continue
		}
		if _, err := exec.ExecContext(ctx, fmt.Sprintf("DROP TABLE IF EXISTS `%s`", def.TableName)); err != nil {
			return fmt.Errorf("drop stale physical table %s: %w", def.TableName, err)
		}
	}
	return nil
}

func hasReadSourceDefinitions(defs []domainschema.TableDefinition) bool {
	for _, def := range defs {
		if def.ReadSource != nil {
			return true
		}
	}
	return false
}

func expectedReadSourceFieldsByObject(fields []metadata.FieldWithContext) map[string]map[string]struct{} {
	expectedFieldsByObject := make(map[string]map[string]struct{})
	for _, field := range fields {
		objectKey := strings.ToLower(strings.TrimSpace(field.TableName))
		if objectKey == "" {
			continue
		}
		fieldAPIName := ""
		if field.Field != nil {
			fieldAPIName = field.Field.APIName
		}
		if fieldAPIName == "" {
			fieldAPIName = field.ColumnDefinition.Name
		}
		fieldKey := strings.ToLower(strings.TrimSpace(fieldAPIName))
		if fieldKey == "" {
			continue
		}
		if expectedFieldsByObject[objectKey] == nil {
			expectedFieldsByObject[objectKey] = make(map[string]struct{})
		}
		expectedFieldsByObject[objectKey][fieldKey] = struct{}{}
	}
	return expectedFieldsByObject
}

func requiredFieldsByObjectFromFieldContexts(fields []metadata.FieldWithContext) objectSpecRequiredFieldIndex {
	index := objectSpecRequiredFieldIndex{}
	for _, field := range fields {
		if field.Field == nil {
			continue
		}
		objectKey := strings.ToLower(strings.TrimSpace(field.TableName))
		if objectKey == "" {
			continue
		}
		fieldAPIName := strings.TrimSpace(field.Field.APIName)
		if fieldAPIName == "" {
			fieldAPIName = strings.TrimSpace(field.ColumnDefinition.Name)
		}
		fieldKey := strings.ToLower(fieldAPIName)
		if fieldKey == "" {
			continue
		}
		if index[objectKey] == nil {
			index[objectKey] = map[string]objectSpecFieldConstraint{}
		}
		constraint := objectSpecFieldConstraint{required: field.Field.Required}
		if field.Field.MaxLength != nil {
			constraint.maxLength = *field.Field.MaxLength
		}
		index[objectKey][fieldKey] = constraint
	}
	return index
}

func validateReadSourceLegacySurfaceCompatibility(ctx context.Context, db *sql.DB, packName string, defs []domainschema.TableDefinition, fields []metadata.FieldWithContext) error {
	if db == nil {
		return fmt.Errorf("db is unavailable for readSource legacy-surface validation")
	}
	return validateReadSourceLegacySurfaceCompatibilityWithExpectedFields(ctx, db, packName, defs, expectedReadSourceFieldsByObject(fields), requiredFieldsByObjectFromFieldContexts(fields))
}

func validateReadSourceLegacySurfaceCompatibilityWithExpectedFields(ctx context.Context, db *sql.DB, packName string, defs []domainschema.TableDefinition, expectedFieldsByObject map[string]map[string]struct{}, requiredFields objectSpecRequiredFieldIndex) error {
	physicalTables, err := loadPhysicalTableSet(db)
	if err != nil {
		return fmt.Errorf("load physical tables for readSource validation: %w", err)
	}
	physicalColumns, err := schemaProvisioner.LoadPhysicalColumnShapes(db)
	if err != nil {
		return fmt.Errorf("load physical columns for readSource validation: %w", err)
	}
	for _, def := range defs {
		if def.ReadSource == nil {
			continue
		}
		if _, _, contractDrifts := collectReadSourcePhysicalContractDrifts(packName, def, physicalTables, physicalColumns, requiredFields); len(contractDrifts) > 0 {
			sort.Strings(contractDrifts)
			return &packObjectPhysicalSchemaDriftError{drifts: contractDrifts}
		}
		for _, col := range def.Columns {
			resolved, err := formulastorage.Resolve(formulastorage.SpecFromColumn(col))
			_ = err
			if !resolved.RequiresFormulaView() {
				continue
			}
			return fmt.Errorf(
				"readSource object %s declares active computed field %s; readSource + formula routing is unsupported",
				def.TableName,
				col.Name,
			)
		}

		expectedFields := expectedFieldsByObject[strings.ToLower(strings.TrimSpace(def.TableName))]
		liveFields, err := loadReadSourceLiveFields(ctx, db, def.TableName)
		if err != nil {
			return fmt.Errorf("load live field registry for %s: %w", def.TableName, err)
		}
		_, unexpectedLiveFields, activeExpectedComputedFields := classifyReadSourceLiveFields(liveFields, expectedFields, packName)
		if len(unexpectedLiveFields) > 0 || len(activeExpectedComputedFields) > 0 {
			problems := make([]string, 0, len(unexpectedLiveFields)+len(activeExpectedComputedFields))
			for _, row := range unexpectedLiveFields {
				problems = append(problems, fmt.Sprintf("%s owner=%s", row.fieldAPIName, formatReadSourceFieldOwner(row)))
			}
			for _, row := range activeExpectedComputedFields {
				problems = append(problems, fmt.Sprintf("%s owner=%s has active computed metadata", row.fieldAPIName, formatReadSourceFieldOwner(row)))
			}
			return fmt.Errorf(
				"readSource cutover for %s is blocked by unsupported live fields: %s",
				def.TableName,
				strings.Join(problems, ", "),
			)
		}
	}
	return nil
}

// ---- Metadata Builders ----

// validateSpecLockRules validates the LockRule expression for each spec that
// declares one using the schema-aware strategy from formula.ValidateLockRuleExpression.
//
// Each spec is converted to a minimal models.ObjectMetadata so the validator
// can prove field existence and boolean return type. Unsupported constructs are
// rejected at pack apply / verify time with clear authoring-time errors:
//   - Syntax errors → INVALID
//   - Unknown function names (VLOOKUP, REGEX, …) → INVALID
//   - $Permission.* → INVALID (Permission context is never injected in lock gate)
//   - Relationship traversal (__r paths) → INVALID (lock gate has no Fetcher)
//   - Unknown field references → INVALID (field not in object schema)
//   - Non-boolean expression → INVALID
//   - Supported session globals ($User, $Profile, $UserRole) and record fields → VALID
func validateSpecLockRules(specs []ObjectSpec, resolver objectSpecFieldResolver) error {
	return validateSpecLockRulesWithFunctionBundles(specs, resolver, nil)
}

func validateSpecLockRulesWithFunctionBundles(specs []ObjectSpec, resolver objectSpecFieldResolver, formulaFunctionBundles []formula.FunctionBundle) error {
	eng := formula.NewEngine(formula.WithoutRuntimeFunctionBundles())
	if len(formulaFunctionBundles) > 0 {
		if err := eng.RegisterFunctionBundles(formula.CloneFunctionBundles(formulaFunctionBundles)); err != nil {
			return fmt.Errorf("load formula function bundles: %w", err)
		}
	}
	// Build full schemas (preserves ReturnType, FormulaMode, etc.) then enrich
	// with the same system fields that specToMinimalObjectMetadata adds, so
	// ValidateLockRuleExpression sees the same field surface as the runtime.
	fullSchemas := buildObjectMetadataWithResolver(specs, resolver)
	schemaByName := make(map[string]*models.ObjectMetadata, len(fullSchemas))
	for _, s := range fullSchemas {
		schemaByName[s.APIName] = s
	}
	for _, spec := range specs {
		expr := strings.TrimSpace(spec.LockRule)
		if expr == "" {
			continue
		}
		specSchema := schemaByName[spec.APIName]
		if specSchema == nil {
			specSchema = specToMinimalObjectMetadata(spec)
		} else {
			// Prepend table-type-specific system fields so the lock-rule validator
			// sees the same field surface as the runtime DDL paths (RecordTypeId is
			// present for custom objects, absent for standard/system objects).
			sysFields := managedObjectSystemFieldsForSpec(spec)
			combined := make([]models.FieldMetadata, 0, len(sysFields)+len(specSchema.Fields))
			combined = append(combined, sysFields...)
			combined = append(combined, specSchema.Fields...)
			clone := *specSchema
			clone.Fields = combined
			specSchema = &clone
		}
		if err := formula.ValidateLockRuleExpression(expr, eng, specSchema); err != nil {
			return fmt.Errorf("object %s: lockRule %w", spec.APIName, err)
		}
	}
	return nil
}

func validateSpecReadSourcesForApply(packName string, specs []ObjectSpec) error {
	if strings.EqualFold(strings.TrimSpace(packName), "platform-core") {
		return validatePlatformCoreReadSources(specs)
	}
	return validateSpecReadSources(specs)
}

func validateSpecReadSources(specs []ObjectSpec) error {
	for _, spec := range specs {
		if spec.ReadSource == nil {
			continue
		}
		if _, err := models.NormalizeObjectReadSourceConfig(spec.ReadSource); err != nil {
			return fmt.Errorf("object %s: %w", spec.APIName, err)
		}
		return fmt.Errorf("object %s: readSource authoring is not yet supported in the current pack contract", spec.APIName)
	}
	return nil
}

func validatePlatformCoreReadSources(specs []ObjectSpec) error {
	for _, spec := range specs {
		if spec.ReadSource == nil {
			continue
		}
		normalizedReadSource, err := models.NormalizeObjectReadSourceConfig(spec.ReadSource)
		if err != nil {
			return fmt.Errorf("object %s: %w", spec.APIName, err)
		}
		if !supportsPackReadSourceSpec(spec) {
			return fmt.Errorf("object %s: platform-core readSource is supported only for the system standard object RecordType", spec.APIName)
		}
		if normalizedReadSource.PhysicalTable != constants.TableRecordType {
			return fmt.Errorf("object %s: platform-core readSource physical table must be %s", spec.APIName, constants.TableRecordType)
		}
	}
	return nil
}

func supportsPackReadSourceSpec(spec ObjectSpec) bool {
	return strings.EqualFold(strings.TrimSpace(spec.APIName), constants.ObjectRecordType) &&
		constants.TableType(spec.TableType) == constants.TableTypeStandardObject &&
		strings.EqualFold(strings.TrimSpace(spec.Category), "system")
}

func managedObjectSystemColumnsForSpec(spec ObjectSpec) []domainschema.ColumnDefinition {
	if supportsPackReadSourceSpec(spec) {
		return []domainschema.ColumnDefinition{
			{
				Name:       constants.FieldID,
				Type:       "VARCHAR(255)",
				PrimaryKey: true,
				Nullable:   false,
			},
			{
				Name:     constants.FieldCreatedDate,
				Type:     "DATETIME",
				Default:  "CURRENT_TIMESTAMP",
				Nullable: false,
			},
			{
				Name:     constants.FieldLastModifiedDate,
				Type:     "DATETIME",
				Default:  "CURRENT_TIMESTAMP",
				Nullable: false,
			},
			{
				Name:     constants.FieldIsDeleted,
				Type:     "TINYINT(1)",
				Default:  "0",
				Nullable: false,
			},
		}
	}
	return persistence.ManagedObjectSystemColumns(spec.APIName, constants.TableType(spec.TableType))
}

func managedObjectSystemFieldsForSpec(spec ObjectSpec) []models.FieldMetadata {
	fields := persistence.ManagedObjectSystemFieldMetadata(spec.APIName, constants.TableType(spec.TableType))
	if !supportsPackReadSourceSpec(spec) {
		return fields
	}
	supported := map[string]struct{}{
		strings.ToLower(constants.FieldID):               {},
		strings.ToLower(constants.FieldCreatedDate):      {},
		strings.ToLower(constants.FieldLastModifiedDate): {},
		strings.ToLower(constants.FieldIsDeleted):        {},
	}
	out := make([]models.FieldMetadata, 0, len(supported))
	for _, field := range fields {
		if _, ok := supported[strings.ToLower(strings.TrimSpace(field.APIName))]; ok {
			out = append(out, field)
		}
	}
	return out
}

// specToMinimalObjectMetadata converts an ObjectSpec to a minimal
// *models.ObjectMetadata containing only the fields required for LockRule
// expression validation (APIName, Type, ReferenceTo).
//
// Standard system fields (Id, OwnerId, CreatedDate, etc.) are always present
// on every object at runtime. They are added here so that LockRule expressions
// referencing them — e.g. `OwnerId == $User.Id` — pass pack apply / verify
// validation without requiring the spec author to re-declare them in Fields.
func specToMinimalObjectMetadata(spec ObjectSpec) *models.ObjectMetadata {
	// Standard system fields present on every object at runtime.
	// Types match Salesforce v63 semantics and the frozen-formula schema walker's
	// type inference (see schema_validation.go schemaTypeFromFieldType).
	injectedSystemFields := managedObjectSystemFieldsForSpec(spec)
	sysFields := make([]models.FieldMetadata, 0, len(injectedSystemFields))
	for _, field := range injectedSystemFields {
		sysFields = append(sysFields, models.FieldMetadata{
			APIName:          field.APIName,
			Type:             field.Type,
			ReferenceTo:      append([]string(nil), field.ReferenceTo...),
			RelationshipName: field.RelationshipName,
		})
	}

	fields := make([]models.FieldMetadata, 0, len(sysFields)+len(spec.Fields))
	fields = append(fields, sysFields...)
	for _, fs := range spec.Fields {
		relName := fs.RelationshipName
		var relNamePtr *string
		if relName != "" {
			relNamePtr = &relName
		}
		fields = append(fields, models.FieldMetadata{
			APIName:          fs.APIName,
			Type:             constants.SchemaFieldType(fs.Type),
			ReferenceTo:      append([]string(nil), fs.ReferenceTo...),
			RelationshipName: relNamePtr,
		})
	}
	return &models.ObjectMetadata{
		APIName: spec.APIName,
		Label:   spec.Label,
		Fields:  fields,
	}
}

func preservePackOwnedAccessPolicySidecars(ctx context.Context, db *sql.DB, objects []*models.ObjectMetadata) error {
	if db == nil {
		return nil
	}
	for _, obj := range objects {
		if obj == nil || strings.TrimSpace(obj.APIName) == "" {
			continue
		}
		if err := preserveObjectPackOwnedAccessPolicySidecars(ctx, db, obj); err != nil {
			return err
		}
	}
	return nil
}

func preserveObjectPackOwnedAccessPolicySidecars(ctx context.Context, db *sql.DB, obj *models.ObjectMetadata) error {
	var liveRaw, recordMutationPackName, recordMutationPackValue, lifecyclePackName, lifecyclePackValue sql.NullString
	err := db.QueryRowContext(ctx,
		fmt.Sprintf(
			"SELECT `%s`, `%s`, `%s`, `%s`, `%s` FROM `%s` WHERE LOWER(`%s`) = ? AND COALESCE(`%s`, 0) = 0 LIMIT 1",
			constants.FieldSysObject_AccessPolicy,
			constants.FieldSysObject_AccessPolicyRecordMutationPackName,
			constants.FieldSysObject_AccessPolicyRecordMutationPackValue,
			constants.FieldSysObject_AccessPolicyLifecyclePackName,
			constants.FieldSysObject_AccessPolicyLifecyclePackValue,
			constants.TableObject,
			constants.FieldSysObject_APIName,
			constants.FieldIsDeleted,
		),
		strings.ToLower(strings.TrimSpace(obj.APIName)),
	).Scan(&liveRaw, &recordMutationPackName, &recordMutationPackValue, &lifecyclePackName, &lifecyclePackValue)
	if err == sql.ErrNoRows {
		return nil
	}
	if err != nil {
		return fmt.Errorf("query %s AccessPolicy sidecars: %w", obj.APIName, err)
	}

	policy, err := domainschema.PreservePackOwnedAccessPolicySidecars(
		obj.AccessPolicy,
		domainschema.AccessPolicySidecarSnapshot{
			LiveRaw:                 liveRaw,
			RecordMutationPackName:  recordMutationPackName,
			RecordMutationPackValue: recordMutationPackValue,
			LifecyclePackName:       lifecyclePackName,
			LifecyclePackValue:      lifecyclePackValue,
		},
		domainschema.AccessPolicySidecarPreserveOptions{
			ObjectAPIName:          obj.APIName,
			RequireOwnedProjection: true,
		},
	)
	if err != nil {
		return err
	}
	obj.AccessPolicy = policy
	return nil
}

func buildObjectMetadata(specs []ObjectSpec) []*models.ObjectMetadata {
	return buildObjectMetadataWithResolver(specs, newObjectSpecFieldResolver(specs))
}

func buildObjectMetadataWithResolver(specs []ObjectSpec, resolver objectSpecFieldResolver) []*models.ObjectMetadata {
	objects := make([]*models.ObjectMetadata, 0, len(specs))
	for _, spec := range specs {
		label := spec.Label
		pluralLabel := spec.PluralLabel
		if pluralLabel == "" {
			// Route through the canonical textutil.Pluralize so
			// source-backed pack reapply emits the same plural the
			// managed-metadata path would emit for a Label
			// with the same shape. Naive `label + "s"` here was the
			// root cause of NX#492 (Opportunity → "Opportunitys",
			// Price Book Entry → "Price Book Entrys" on every reapply).
			pluralLabel = textutil.Pluralize(label)
		}
		desc := spec.Description
		isCustom := constants.TableType(spec.TableType) == constants.TableTypeCustomObject

		obj := &models.ObjectMetadata{
			ID:                     domainschema.GenerateObjectID(spec.APIName),
			APIName:                spec.APIName,
			Label:                  label,
			PluralLabel:            pluralLabel,
			Description:            &desc,
			TableType:              constants.TableType(spec.TableType),
			IsCustom:               isCustom,
			CustomSetting:          spec.CustomSetting,
			Searchable:             true,
			SharingModel:           models.SharingModel(spec.SharingModel),
			EnableHierarchySharing: spec.EnableHierarchySharing,
			AccessPolicy:           spec.AccessPolicy,
			ReadSource:             cloneObjectReadSourceConfig(spec.ReadSource),
			DefaultChildRows:       append([]models.DefaultChildRow(nil), spec.DefaultChildRows...),
			ChildCloneRules:        append([]models.ChildCloneRule(nil), spec.ChildCloneRules...),
			ExclusiveSiblingFields: append([]models.ExclusiveSiblingField(nil), spec.ExclusiveSiblingFields...),
		}
		if spec.ListFields != nil {
			obj.ListFields = append([]string{}, spec.ListFields...)
		}
		if spec.LockRule != "" {
			lc := spec.LockRule
			obj.LockRule = &lc
		}
		if spec.KeyPrefix != "" {
			keyPrefix := spec.KeyPrefix
			obj.KeyPrefix = &keyPrefix
		}
		if spec.PathField != "" {
			pathField := spec.PathField
			obj.PathField = &pathField
		}
		if spec.DisplayNameField != "" {
			displayNameField := spec.DisplayNameField
			obj.DisplayNameField = &displayNameField
		}
		if spec.LockRule != "" {
			lockRule := spec.LockRule
			obj.LockRule = &lockRule
		}
		if len(spec.Fields) > 0 {
			obj.Fields = make([]models.FieldMetadata, 0, len(spec.Fields))
			for _, fSpec := range spec.Fields {
				createable := true
				if fSpec.Createable != nil {
					createable = *fSpec.Createable
				}
				updateable := true
				if fSpec.Updateable != nil {
					updateable = *fSpec.Updateable
				}
				field := models.FieldMetadata{
					APIName:               fSpec.APIName,
					Label:                 fSpec.Label,
					Type:                  constants.SchemaFieldType(fSpec.Type),
					Required:              packFieldRequired(fSpec),
					IsMasterDetail:        packFieldIsMasterDetail(fSpec),
					IsNameField:           fSpec.IsNameField,
					Createable:            createable,
					Updateable:            updateable,
					ReferenceTo:           append([]string(nil), fSpec.ReferenceTo...),
					WritableReferenceTo:   append([]string(nil), fSpec.WritableReferenceTo...),
					IsPolymorphic:         fSpec.isPolymorphicReference(),
					Options:               fSpec.Options,
					RollupConfig:          cloneNormalizedRollupConfigWithResolver(fSpec.RollupConfig, resolver),
					IsCompound:            fSpec.IsCompound,
					ComposedFrom:          append([]string(nil), fSpec.ComposedFrom...),
					ProjectionConfig:      cloneProjectionConfig(fSpec.ProjectionConfig),
					WriteProtectionPolicy: fSpec.WriteProtectionPolicy,
					ReadProtectionPolicy:  fSpec.ReadProtectionPolicy,
				}
				if rule := fSpec.normalizedDeleteConstraint(); rule != "" {
					field.DeleteRule = &rule
				}
				if fSpec.Default != "" {
					def := fSpec.Default
					field.DefaultValue = &def
				}
				if fSpec.HelpText != "" {
					helpText := fSpec.HelpText
					field.HelpText = &helpText
				}
				if strings.TrimSpace(fSpec.SiblingMaxParentField) != "" {
					parentField := strings.TrimSpace(fSpec.SiblingMaxParentField)
					field.SiblingMaxParentField = &parentField
				}
				if fSpec.Formula != "" {
					formula := fSpec.Formula
					field.Formula = &formula
				}
				if fSpec.ReturnType != "" {
					rt := constants.SchemaFieldType(fSpec.ReturnType)
					field.ReturnType = &rt
				}
				if fSpec.FormulaMode != "" {
					mode := fSpec.FormulaMode
					field.FormulaMode = &mode
				}
				if fSpec.FormulaStorageMode != "" {
					mode := fSpec.FormulaStorageMode
					field.FormulaStorageMode = &mode
				}
				if fSpec.RelationshipName != "" {
					rel := fSpec.RelationshipName
					field.RelationshipName = &rel
				}
				if fSpec.ChildRelationshipName != "" {
					childRel := fSpec.ChildRelationshipName
					field.ChildRelationshipName = &childRel
				}
				obj.Fields = append(obj.Fields, field)
			}
		}
		objects = append(objects, obj)
	}
	return objects
}

func cloneObjectReadSourceConfig(in *models.ObjectReadSourceConfig) *models.ObjectReadSourceConfig {
	if in == nil {
		return nil
	}
	out := *in
	return &out
}

// tenantOwnedFields returns the set of (object, field) pairs registered by a
// tenant package. Core uses it only as ownership metadata; standard Salesforce
// fields remain part of the platform baseline even when a tenant overlays
// labels/options later.
func tenantOwnedFields() map[string]map[string]bool {
	manifest := packspec.TenantFieldManifestJSON()
	if len(manifest) == 0 {
		return nil
	}

	var specs []ObjectSpec
	if err := json.Unmarshal(manifest, &specs); err != nil {
		return nil
	}
	ownership := make(map[string]map[string]bool, len(specs))
	for _, spec := range specs {
		if len(spec.Fields) == 0 {
			continue
		}
		fieldSet := make(map[string]bool, len(spec.Fields))
		for _, field := range spec.Fields {
			fieldSet[field.APIName] = true
		}
		ownership[spec.APIName] = fieldSet
	}
	return ownership
}

func buildFieldMetadata(packName string, specs []ObjectSpec, sm *metadata.SchemaManager, skipTenantOwnedFields bool) ([]metadata.FieldWithContext, error) {
	return buildFieldMetadataWithResolver(packName, specs, sm, skipTenantOwnedFields, newObjectSpecFieldResolver(specs), packFieldAuthoringCompat{})
}

func buildFieldMetadataWithResolver(packName string, specs []ObjectSpec, sm *metadata.SchemaManager, skipTenantOwnedFields bool, resolver objectSpecFieldResolver, compat packFieldAuthoringCompat) ([]metadata.FieldWithContext, error) {
	var fields []metadata.FieldWithContext
	var tenantOwned map[string]map[string]bool
	if skipTenantOwnedFields {
		// Transitional embedded platform-core installs still honor the legacy
		// tenant ownership manifest so L0 does not seize tenant-owned __c facts.
		// Source-backed object_spec installs must NOT apply that filter: the
		// pack's declared fields are the source of truth for live _System_Field
		// convergence, including pack-owned custom fields.
		tenantOwned = tenantOwnedFields()
	}

	// System field metadata is table-type-specific: custom objects get
	// RecordTypeId, while standard/system objects keep the generic baseline.
	for _, spec := range specs {
		specStart := len(fields)
		objectID := domainschema.GenerateObjectID(spec.APIName)
		injectedStdFields := managedObjectSystemFieldsForSpec(spec)

		// System fields
		for _, sf := range injectedStdFields {
			fieldID := domainschema.GenerateFieldID(spec.APIName, sf.APIName)
			sfCopy := sf
			// Provenance is derived at read time by fieldprovenance from
			// the pack manifest; no stored Source tag needed.
			fields = append(fields, metadata.FieldWithContext{
				Field:     &sfCopy,
				TableName: spec.APIName,
				ObjectID:  objectID,
				FieldID:   fieldID,
				OwnerKind: constants.FieldOwnerKindPack,
				OwnerPack: packName,
			})
		}

		// Pack-declared fields. Tenant/custom fields are written by tenant packs.
		declaredFieldNames := make(map[string]bool, len(spec.Fields))
		discriminatorParents := make(map[string]string)
		for _, fSpec := range spec.Fields {
			declaredFieldNames[strings.ToLower(fSpec.APIName)] = true
			if fSpec.isPolymorphicReference() {
				discriminatorParents[strings.ToLower(domainschema.PolymorphicDiscriminatorColumnName(fSpec.APIName))] = fSpec.APIName
			}
		}
		for _, fSpec := range spec.Fields {
			if shouldSkipPackField(spec, fSpec, tenantOwned) {
				continue
			}
			discriminatorParent := discriminatorParents[strings.ToLower(fSpec.APIName)]
			fieldWithContext, err := buildPackFieldSpecWriteWithResolver(packName, spec.APIName, objectID, fSpec, sm, discriminatorParent, resolver, compat)
			if err != nil {
				return nil, fmt.Errorf("object %s field %s: %w", spec.APIName, fSpec.APIName, err)
			}
			fields = append(fields, fieldWithContext)

			if fSpec.isPolymorphicReference() {
				typeColDef := domainschema.PolymorphicDiscriminatorColumn(fieldWithContext.ColumnDefinition)
				if declaredFieldNames[strings.ToLower(typeColDef.Name)] {
					continue
				}
				typeField := &models.FieldMetadata{
					APIName:                          typeColDef.Name,
					Label:                            typeColDef.Label,
					Type:                             constants.FieldTypeString,
					Createable:                       false,
					Updateable:                       false,
					GeneratedForPolymorphicReference: fSpec.APIName,
				}
				fields = append(fields, metadata.FieldWithContext{
					Field:                typeField,
					TableName:            spec.APIName,
					ObjectID:             objectID,
					FieldID:              domainschema.GenerateFieldID(spec.APIName, typeColDef.Name),
					ColumnDefinition:     typeColDef,
					CapabilitiesDeclared: true,
					OwnerKind:            constants.FieldOwnerKindPack,
					OwnerPack:            packName,
				})
			}
		}

		// When no payload field declares IsNameField and exactly one field is an
		// autonumber, stamp that field as IsNameField. This covers standard objects
		// (e.g. Order.OrderNumber, Quote.QuoteNumber) whose display field is an
		// autonumber not explicitly marked in the pack payload JSON.
		stampSoleAutonumberAsNameField(fields[specStart:])
	}

	return fields, nil
}

// stampSoleAutonumberAsNameField stamps IsNameField=true on the sole autonumber
// field in a per-object FieldWithContext slice when no field already declares
// IsNameField. Must be called with only the fields for one object.
//
// Activation condition: no field has IsNameField=true AND exactly one field has
// type autonumber. Multiple autonumbers → ambiguous, no stamp.
func stampSoleAutonumberAsNameField(fields []metadata.FieldWithContext) {
	for _, f := range fields {
		if f.Field != nil && f.Field.IsNameField {
			return
		}
	}
	soleIdx := -1
	for i, f := range fields {
		if f.Field == nil {
			continue
		}
		if strings.EqualFold(string(f.Field.Type), string(constants.FieldTypeAutonumber)) {
			if soleIdx >= 0 {
				return // more than one autonumber — ambiguous, no stamp
			}
			soleIdx = i
		}
	}
	if soleIdx >= 0 {
		fields[soleIdx].Field.IsNameField = true
	}
}

func shouldSkipPackField(spec ObjectSpec, field FieldSpec, tenantOwned map[string]map[string]bool) bool {
	fields := tenantOwned[spec.APIName]
	if fields == nil {
		return false
	}
	return fields[field.APIName] && isCustomFieldAPIName(field.APIName)
}

func isCustomFieldAPIName(apiName string) bool {
	return strings.HasSuffix(apiName, "__c")
}

func cloneProjectionConfig(in *models.FieldProjectionConfig) *models.FieldProjectionConfig {
	if in == nil {
		return nil
	}
	out := *in
	if len(in.Matches) > 0 {
		out.Matches = append([]models.FieldProjectionMatch(nil), in.Matches...)
	}
	return &out
}

func reconcileManagedRelationships(ctx context.Context, rawDB *sql.DB, specs []ObjectSpec, defs []domainschema.TableDefinition) error {
	// Merge selected pack spec relationships + substrate relationships (DashboardComponent)
	allRels := relationshipSemanticsFromSpecs(specs)
	allRels = append(allRels, domainschema.ManagedStandardRelationshipSemanticsList()...)

	// Combine standard object table defs with substrate defs
	// (the reconciler needs field metadata from BOTH to resolve FK columns)
	allDefs := append(defs, managedTableDefinitions()...)

	return ReconcileManagedRelationshipSemanticsWithDefs(ctx, rawDB, allRels, allDefs)
}

func relationshipSemanticsFromSpecs(specs []ObjectSpec) []domainschema.ManagedRelationshipSemantics {
	var result []domainschema.ManagedRelationshipSemantics
	for _, spec := range specs {
		for _, rel := range spec.ManagedRelationships {
			result = append(result, domainschema.ManagedRelationshipSemantics{
				ChildObjectAPIName:  spec.APIName,
				ParentObjectAPIName: rel.ParentObject,
				FieldAPIName:        rel.FieldAPIName,
				IsSharingParent:     rel.IsSharingParent,
				CascadeDelete:       rel.CascadeDelete,
			})
		}
	}
	return result
}

func strPtr(s string) *string {
	if s == "" {
		return nil
	}
	return &s
}