package metadata

import (
	"context"
	"errors"
	"fmt"
	"log"
	"slices"
	"strings"
	"time"

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedEvents "github.com/nexuscrm/shared/pkg/events"
	"github.com/nexuscrm/shared/pkg/models"
)

// wrapStrictCreateError preserves a typed ConflictError from the strict-create
// chokepoint (CreateTableWithStrictMetadata) instead of blanket-wrapping it in
// an InternalError. A ConflictError carries the orphan-blocked-create
// remediation and maps to a client-visible 409; the metadataAuthoringError
// contract sanitizes InternalError to an opaque 500 (its IsInternal check runs
// first), which would hide both the status and the remediation (#3484). Any
// other error keeps the original InternalError wrap.
func wrapStrictCreateError(err error, msg string) error {
	var conflict *apperrors.ConflictError
	if errors.As(err, &conflict) {
		return conflict
	}
	return apperrors.NewInternalError(msg, err)
}

func rejectReadSourceAuthoring(cfg *models.ObjectReadSourceConfig) error {
	if cfg == nil {
		return nil
	}
	return apperrors.NewValidationError(constants.FieldSysObject_ReadSource, "readSource is not writable via generic metadata schema APIs")
}

func rejectExclusiveSiblingFieldsAuthoring(schema *models.ObjectMetadata) error {
	if schema == nil {
		return nil
	}
	if schema.ExclusiveSiblingFields != nil || schema.ExclusiveSiblingFieldsCustomized {
		return apperrors.NewValidationError(constants.FieldSysObject_ExclusiveSiblingFields, "exclusiveSiblingFields is not writable via generic metadata schema APIs; use validated pack or tenant overlay metadata")
	}
	return nil
}

// validateDisplayNameFieldUpdate checks that the incoming DisplayNameField
// value resolves to a real, label-renderable field on the existing object
// schema. It accepts direct field API names plus relationship-name and one-hop
// relationship child paths backed by reference/masterdetail fields.
// Relationship-backed display fields must be single-target because the current
// list-name hydration path resolves lookup labels from ReferenceTo[0]. This
// prevents an admin PATCH from writing an arbitrary string or unsupported
// relationship shape that would later break list-view queries or render the
// wrong related label.
func validateDisplayNameFieldUpdate(value string, obj *models.ObjectMetadata, targetSchema func(string) *models.ObjectMetadata) error {
	if value == "" {
		return nil
	}
	trimmed := strings.TrimSpace(value)
	if trimmed == "" {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			"displayNameField is whitespace-only; use exact empty string to clear it, omit it to leave it unchanged, or set it to a field ApiName that exists on this object")
	}
	if trimmed != value {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			fmt.Sprintf("displayNameField %q has leading or trailing whitespace; value must be byte-identical to its trimmed form", value))
	}
	childField, hasChildPath := displayNameChildPath(trimmed)
	if hasChildPath && !displayNameDotPathSegmentsAreCanonical(trimmed) {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			fmt.Sprintf("displayNameField %q has padded relationship path segment(s); parent and child segments must be byte-identical to their trimmed forms", trimmed))
	}
	resolved := resolveDisplayNameFieldMetadata(trimmed, obj)
	if resolved == nil {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			fmt.Sprintf("displayNameField %q does not resolve to any field, relationshipName, or one-hop relationship child path on this object", trimmed))
	}
	resolvedType := constants.NormalizeFieldType(string(resolved.Type))
	if !constants.IsDisplayNameFieldRenderableType(resolvedType) {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			fmt.Sprintf("displayNameField %q resolves to a field of type %q, which is not a label-renderable type; accepted types are reference, masterdetail, string, textarea, email, phone, url, autonumber", trimmed, resolved.Type))
	}
	if constants.IsReferenceType(resolvedType) {
		if resolved.IsPolymorphic || len(resolved.ReferenceTo) != 1 {
			return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
				fmt.Sprintf("displayNameField %q resolves to relationship field %q with referenceTo=%v and isPolymorphic=%t; reference/masterdetail display-name fields must have exactly one concrete target because list-name hydration resolves lookup labels from ReferenceTo[0]", trimmed, resolved.APIName, resolved.ReferenceTo, resolved.IsPolymorphic))
		}
		if resolved.RelationshipName == nil || strings.TrimSpace(*resolved.RelationshipName) == "" {
			return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
				fmt.Sprintf("displayNameField %q resolves to relationship field %q, but that field has no relationshipName; reference/masterdetail display-name fields must declare the RelationshipName used by describe/list expansion", trimmed, resolved.APIName))
		}
		if *resolved.RelationshipName != strings.TrimSpace(*resolved.RelationshipName) {
			return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
				fmt.Sprintf("displayNameField %q resolves to relationship field %q with padded relationshipName %q; relationshipName must be byte-identical to its trimmed form", trimmed, resolved.APIName, *resolved.RelationshipName))
		}
		if hasChildPath {
			if err := validateDisplayNameRelationshipChild(trimmed, childField, resolved, targetSchema); err != nil {
				return err
			}
		}
	}
	return nil
}

func displayNameChildPath(value string) (string, bool) {
	if strings.Count(value, ".") != 1 {
		return "", false
	}
	parts := strings.SplitN(value, ".", 2)
	child := strings.TrimSpace(parts[1])
	return child, child != ""
}

func displayNameDotPathSegmentsAreCanonical(value string) bool {
	if strings.Count(value, ".") != 1 {
		return true
	}
	parts := strings.SplitN(value, ".", 2)
	return parts[0] == strings.TrimSpace(parts[0]) && parts[1] == strings.TrimSpace(parts[1])
}

func validateDisplayNameRelationshipChild(value string, childField string, referenceField *models.FieldMetadata, targetSchema func(string) *models.ObjectMetadata) error {
	if !isDisplayNameChildAPIName(childField) {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			fmt.Sprintf("displayNameField %q has relationship child %q that is not a legal field ApiName", value, childField))
	}
	if referenceField == nil || len(referenceField.ReferenceTo) != 1 {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			fmt.Sprintf("displayNameField %q cannot resolve child field %q because its parent relationship target is not single-target", value, childField))
	}
	targetAPIName := strings.TrimSpace(referenceField.ReferenceTo[0])
	var target *models.ObjectMetadata
	if targetSchema != nil {
		target = targetSchema(targetAPIName)
	}
	if target == nil {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			fmt.Sprintf("displayNameField %q references target object %q, but that object schema is unavailable for child-field validation", value, targetAPIName))
	}
	child := models.FindField(target, childField)
	if child == nil || child.APIName != childField {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			fmt.Sprintf("displayNameField %q references child field %q, but %s has no case-sensitive ApiName match", value, childField, targetAPIName))
	}
	childType := constants.NormalizeFieldType(string(child.Type))
	if !constants.IsDisplayNameRelationshipChildType(childType) {
		return apperrors.NewValidationError(constants.FieldSysObject_DisplayNameField,
			fmt.Sprintf("displayNameField %q references child field %q of type %q, which is not a scalar label/search type; accepted child types are string, textarea, email, phone, url, autonumber", value, childField, child.Type))
	}
	return nil
}

func isDisplayNameChildAPIName(apiName string) bool {
	if apiName == "" {
		return false
	}
	for _, c := range apiName {
		if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' {
			continue
		}
		return false
	}
	return true
}

func resolveDisplayNameFieldMetadata(value string, obj *models.ObjectMetadata) *models.FieldMetadata {
	if obj == nil {
		return nil
	}
	if strings.Count(value, ".") == 1 {
		parts := strings.SplitN(value, ".", 2)
		parentRaw := parts[0]
		childRaw := parts[1]
		parent := strings.TrimSpace(parentRaw)
		child := strings.TrimSpace(childRaw)
		if parent == "" || child == "" || strings.Contains(child, ".") {
			return nil
		}
		if parent != parentRaw || child != childRaw {
			return nil
		}
		return resolveDisplayNameReferenceFieldMetadata(parent, obj)
	}
	if strings.Contains(value, ".") {
		return nil
	}
	for i := range obj.Fields {
		if obj.Fields[i].APIName == value {
			return &obj.Fields[i]
		}
	}
	return resolveDisplayNameReferenceFieldMetadata(value, obj)
}

func resolveDisplayNameReferenceFieldMetadata(parent string, obj *models.ObjectMetadata) *models.FieldMetadata {
	for i := range obj.Fields {
		field := &obj.Fields[i]
		if field.APIName == parent {
			if constants.IsReferenceType(constants.NormalizeFieldType(string(field.Type))) {
				return field
			}
			return nil
		}
	}
	for i := range obj.Fields {
		field := &obj.Fields[i]
		if !constants.IsReferenceType(constants.NormalizeFieldType(string(field.Type))) || field.RelationshipName == nil {
			continue
		}
		if *field.RelationshipName == parent {
			return field
		}
	}
	return nil
}

func (ms *MetadataService) validateCreateDisplayNameField(ctx context.Context, schema *models.ObjectMetadata) error {
	if schema == nil || schema.DisplayNameField == nil {
		return nil
	}
	return validateDisplayNameFieldUpdate(*schema.DisplayNameField, schema, func(targetAPIName string) *models.ObjectMetadata {
		if strings.EqualFold(targetAPIName, schema.APIName) {
			return schema
		}
		target, err := ms.repo.GetSchemaByAPIName(ctx, targetAPIName)
		if err != nil {
			return nil
		}
		return target
	})
}

// --- Schema CRUD Methods ---

// CreateSchema creates a new custom object schema and physical table
func (ms *MetadataService) CreateSchema(ctx context.Context, schema *models.ObjectMetadata) error {
	ms.mu.Lock()
	defer ms.mu.Unlock()
	if schema == nil {
		return apperrors.NewValidationError("schema", "object schema is required")
	}
	// The row ID is an incarnation identity, not the portable API-name key.
	// Ignore caller input so delete/recreate cannot satisfy an immutable Catalog
	// reference to the previous interactive object.
	mintInteractiveObjectIncarnation(schema)

	if err := rejectReadSourceAuthoring(schema.ReadSource); err != nil {
		return err
	}
	if err := rejectExclusiveSiblingFieldsAuthoring(schema); err != nil {
		return err
	}
	if err := rejectTrackedFieldsOnObjectCreate(schema); err != nil {
		return err
	}

	// Capture original labels before any processing/defaults
	originalPlural := schema.PluralLabel
	originalLabel := schema.Label

	// Validate Object Metadata
	if ms.validationSvc != nil {
		if err := ms.validationSvc.ValidateObjectMetadata(schema); err != nil {
			return err
		}
	}

	// Prepare Table Definition (Validation, Enrichment, Mapping).
	// Must run before LockRule validation so that system fields (OwnerId,
	// CreatedDate, etc.) are present in schema.Fields when
	// ValidateLockRuleExpression checks field references.
	def, _, err := ms.PrepareTableDefinition(ctx, schema)
	if err != nil {
		return err
	}

	if err := ms.validateCreateDisplayNameField(ctx, schema); err != nil {
		return err
	}

	if schema.LockRule != nil && *schema.LockRule != "" {
		if err := formula.ValidateLockRuleExpression(*schema.LockRule, formula.NewEngine(), schema); err != nil {
			return apperrors.NewValidationError("LockRule", err.Error())
		}
	}

	// Check if metadata exists in DB
	existing, err := ms.repo.GetSchemaByAPIName(ctx, schema.APIName)
	if err == nil && existing != nil {
		return apperrors.NewConflictError("Object Metadata", "api_name", schema.APIName)
	}

	// Restore original labels for strict insert (if provided/valid)
	// This ensures we persist exactly what the user provided, overriding any "Humanization" done by PrepareTableDefinition
	if originalLabel != "" {
		schema.Label = originalLabel
	}
	if originalPlural != "" {
		schema.PluralLabel = originalPlural
	}

	// Create Table (Strict Mode: Fails if metadata conflicts, or if a physical
	// table of this name already exists with no live metadata — an orphan block).
	if err := ms.schemaMgr.CreateTableWithStrictMetadata(ctx, def, schema); err != nil {
		return wrapStrictCreateError(err, "failed to create schema via SchemaManager")
	}

	ms.ensureRelationshipFKIndexes(def.TableName, schema.Fields)

	// --- L2 DEFAULT LAYOUT SCAFFOLD ---
	defaultLayout := ms.GenerateDefaultLayout(schema)

	// Persist an explicit admin-authored scaffold, then route the wildcard
	// default through the layout-assignment SSoT. This is object-create
	// authoring, not L0 read-time generation.
	if err := ms.repo.SaveLayout(ctx, &defaultLayout); err != nil {
		log.Printf("⚠️ Failed to create default layout scaffold for %s: %v", schema.APIName, err)
	} else if err := ms.repo.SetDefaultLayout(ctx, schema.APIName, defaultLayout.ID, nil); err != nil {
		log.Printf("⚠️ Failed to assign default layout scaffold for %s: %v", schema.APIName, err)
	} else {
		log.Printf("✅ Created default layout scaffold for %s", schema.APIName)
	}

	ms.invalidateCacheLocked()

	// Publish Event
	ms.publishMetadataEvent(ctx, sharedEvents.ObjectCreated, sharedEvents.SchemaEventPayload{
		EventType:     sharedEvents.ObjectCreated,
		ObjectAPIName: schema.APIName,
		Timestamp:     time.Now().Format(time.RFC3339),
	})

	return nil
}

func rejectTrackedFieldsOnObjectCreate(schema *models.ObjectMetadata) error {
	for i := range schema.Fields {
		if schema.Fields[i].TrackHistory {
			return apperrors.NewValidationError(constants.FieldSysField_TrackHistory, "history tracking must be enabled after object creation")
		}
	}
	return nil
}

// ensureRelationshipFKIndexes indexes the FK column of every non-system
// relationship (lookup / master-detail) field on a freshly materialized table,
// so related-list counts, lookups, joins, and RLS subqueries do not full-scan
// the child table. The set of columns is derived from field Type via
// constants.IsReferenceType — generic, never an SObject-name branch. The index
// is a performance optimization, so a failure is logged and tolerated rather
// than failing object/field creation; the bootstrap reconcile
// (EnsureStandardIndexes) is the backstop.
func (ms *MetadataService) ensureRelationshipFKIndexes(tableName string, fields []models.FieldMetadata) {
	for i := range fields {
		f := &fields[i]
		if f.IsSystem || !constants.IsReferenceType(constants.NormalizeFieldType(string(f.Type))) {
			continue
		}
		// Index the exact physical column the table builder created. buildColumnDDL
		// names columns from ColumnDefinition.Name (= APIName); PhysicalName is a
		// read-time alias, not the created column name.
		if err := ms.schemaMgr.EnsureColumnIndex(tableName, f.APIName); err != nil {
			log.Printf("⚠️ Failed to index relationship FK %s.%s: %v", tableName, f.APIName, err)
		}
	}
}

// CreateSchemaOptimized creates a new object schema using batch metadata registration
// This is faster than CreateSchema for objects with many fields
func (ms *MetadataService) CreateSchemaOptimized(ctx context.Context, schema *models.ObjectMetadata) error {
	ms.mu.Lock()
	defer ms.mu.Unlock()
	if schema == nil {
		return apperrors.NewValidationError("schema", "object schema is required")
	}
	mintInteractiveObjectIncarnation(schema)

	if err := rejectReadSourceAuthoring(schema.ReadSource); err != nil {
		return err
	}
	if err := rejectExclusiveSiblingFieldsAuthoring(schema); err != nil {
		return err
	}
	if err := rejectTrackedFieldsOnObjectCreate(schema); err != nil {
		return err
	}

	// Validate Object Metadata
	if ms.validationSvc != nil {
		if err := ms.validationSvc.ValidateObjectMetadata(schema); err != nil {
			return err
		}
	}

	// Prepare Table Definition.
	// Must run before LockRule validation so that system fields (OwnerId,
	// CreatedDate, etc.) are present in schema.Fields when
	// ValidateLockRuleExpression checks field references.
	def, _, err := ms.PrepareTableDefinition(ctx, schema)
	if err != nil {
		return err
	}

	if err := ms.validateCreateDisplayNameField(ctx, schema); err != nil {
		return err
	}

	if schema.LockRule != nil && *schema.LockRule != "" {
		if err := formula.ValidateLockRuleExpression(*schema.LockRule, formula.NewEngine(), schema); err != nil {
			return apperrors.NewValidationError("LockRule", err.Error())
		}
	}

	// Check if metadata exists in DB
	existing, err := ms.repo.GetSchemaByAPIName(ctx, schema.APIName)
	if err == nil && existing != nil {
		return apperrors.NewConflictError("Object Metadata", "api_name", schema.APIName)
	}

	// Create table with STRICT metadata registration (Fails on Unique Constraint,
	// or on an orphan physical table with no live metadata).
	// Note: PrepareTableDefinition updates schema with defaults, so we pass it back
	if err := ms.schemaMgr.CreateTableWithStrictMetadata(ctx, def, schema); err != nil {
		return wrapStrictCreateError(err, "failed to create schema")
	}

	ms.ensureRelationshipFKIndexes(def.TableName, schema.Fields)

	// Create explicit L2/admin default layout scaffold (same as CreateSchema).
	defaultLayout := ms.GenerateDefaultLayout(schema)
	if err := ms.repo.SaveLayout(ctx, &defaultLayout); err != nil {
		log.Printf("⚠️ Failed to insert default layout scaffold for %s: %v", schema.APIName, err)
	} else if err := ms.repo.SetDefaultLayout(ctx, schema.APIName, defaultLayout.ID, nil); err != nil {
		log.Printf("⚠️ Failed to assign default layout scaffold for %s: %v", schema.APIName, err)
	}

	ms.invalidateCacheLocked()

	// Publish Event
	ms.publishMetadataEvent(ctx, sharedEvents.ObjectCreated, sharedEvents.SchemaEventPayload{
		EventType:     sharedEvents.ObjectCreated,
		ObjectAPIName: schema.APIName,
		Timestamp:     time.Now().Format(time.RFC3339),
	})

	return nil
}

func mintInteractiveObjectIncarnation(object *models.ObjectMetadata) {
	if object != nil {
		object.ID = GenerateID()
	}
}

// InvalidateCache/invalidateCacheLocked moved to metadata_service.go

// UpdateSchema updates an existing object schema
func (ms *MetadataService) UpdateSchema(ctx context.Context, apiName string, updates *models.ObjectMetadata) error {
	ms.mu.Lock()
	defer ms.mu.Unlock()

	if err := rejectReadSourceAuthoring(updates.ReadSource); err != nil {
		return err
	}
	if err := rejectExclusiveSiblingFieldsAuthoring(updates); err != nil {
		return err
	}

	// Check if exists
	obj, err := ms.repo.GetSchemaByAPIName(ctx, apiName)
	if err != nil || obj == nil {
		return apperrors.NewNotFoundError("Object", apiName)
	}
	apiName = obj.APIName

	// Prepare updates by modifying the existing object in-memory
	if updates.Label != "" && obj.Label != updates.Label {
		obj.Label = updates.Label
		obj.IsCustomized = true
	}
	if updates.PluralLabel != "" && obj.PluralLabel != updates.PluralLabel {
		obj.PluralLabel = updates.PluralLabel
		obj.IsCustomized = true
	}
	if updates.Description != nil {
		if obj.Description == nil || *obj.Description != *updates.Description {
			obj.IsCustomized = true
		}
		obj.Description = updates.Description
	}
	if updates.PathField != nil {
		if obj.PathField == nil || *obj.PathField != *updates.PathField {
			obj.IsCustomized = true
		}
		obj.PathField = updates.PathField
	}
	if updates.DisplayNameField != nil {
		if err := validateDisplayNameFieldUpdate(*updates.DisplayNameField, obj, func(targetAPIName string) *models.ObjectMetadata {
			if strings.EqualFold(targetAPIName, obj.APIName) {
				return obj
			}
			target, err := ms.repo.GetSchemaByAPIName(ctx, targetAPIName)
			if err != nil {
				return nil
			}
			return target
		}); err != nil {
			return err
		}
		if *updates.DisplayNameField == "" {
			obj.IsCustomized = true
			obj.DisplayNameField = nil
		} else {
			if obj.DisplayNameField == nil || *obj.DisplayNameField != *updates.DisplayNameField {
				obj.IsCustomized = true
			}
			obj.DisplayNameField = updates.DisplayNameField
		}
	}
	if updates.LockRule != nil {
		if obj.LockRule == nil || *obj.LockRule != *updates.LockRule {
			obj.IsCustomized = true
			obj.LockRuleCustomized = true
		}
		if exprStr := *updates.LockRule; exprStr != "" {
			if err := formula.ValidateLockRuleExpression(exprStr, formula.NewEngine(), obj); err != nil {
				return apperrors.NewValidationError("LockRule", err.Error())
			}
		}
		obj.LockRule = updates.LockRule
	}
	if len(updates.ListFields) > 0 {
		if !slices.Equal(obj.ListFields, updates.ListFields) {
			obj.IsCustomized = true
			obj.ListFieldsCustomized = true
		}
		obj.ListFields = updates.ListFields
	}
	if updates.Icon != "" && obj.Icon != updates.Icon {
		obj.Icon = updates.Icon
		obj.IsCustomized = true
	}
	if updates.AppID != nil {
		if obj.AppID == nil || *obj.AppID != *updates.AppID {
			obj.IsCustomized = true
		}
		obj.AppID = updates.AppID
	}
	if updates.SharingModel != "" {
		if obj.SharingModel != updates.SharingModel {
			obj.SharingModel = updates.SharingModel
			// Object-level customization currently protects managed OWD reconcile.
			// Only mark customized when the sharing model itself actually changes.
			obj.IsCustomized = true
		}
	}
	if updates.ThemeColor != nil {
		if obj.ThemeColor == nil || *obj.ThemeColor != *updates.ThemeColor {
			obj.IsCustomized = true
		}
		obj.ThemeColor = updates.ThemeColor
	}

	// Use helper to persist changes
	if err := ms.schemaMgr.SaveObjectMetadata(ctx, obj, nil); err != nil {
		return apperrors.NewInternalError("failed to update object", err)
	}

	ms.invalidateCacheLocked()

	// Publish Event
	ms.publishMetadataEvent(ctx, sharedEvents.ObjectUpdated, sharedEvents.SchemaEventPayload{
		EventType:     sharedEvents.ObjectUpdated,
		ObjectAPIName: apiName,
		Timestamp:     time.Now().Format(time.RFC3339),
	})

	return nil
}

// DeleteSchema deletes a custom object schema and drops its table
func (ms *MetadataService) DeleteSchema(ctx context.Context, apiName string) error {
	ms.mu.Lock()
	defer ms.mu.Unlock()

	// Resolve the object. A real DB error is surfaced as-is.
	obj, err := ms.repo.GetSchemaByAPIName(ctx, apiName)
	if err != nil {
		return apperrors.NewInternalError("failed to resolve object for delete", err)
	}
	if obj == nil {
		// No live metadata. This may be a retry of a delete whose physical DROP
		// failed AFTER the metadata was committed (#3484), leaving a stranded
		// orphan table with no metadata. Because the strict create path fails
		// closed on an orphan, and the controlplane adopt paths re-run
		// CREATE TABLE IF NOT EXISTS, reconcile it by dropping any lingering table
		// so neither can rebind to it. DropTable is idempotent here (its metadata
		// deletes match nothing and DROP TABLE IF EXISTS clears the orphan).
		//
		// This DROP is destructive, so it is fenced by two guards before firing:
		//
		//  1. Managed-substrate guard (constants.IsManagedSystemTable, case-
		//     insensitive on BOTH the _System_ prefix and the managed list). The
		//     raw HasPrefix(apiName, "_System_") check this replaces was case-
		//     SENSITIVE and missed unprefixed substrate (User, Profile,
		//     ObjectPermissions) entirely; combined with the interactive delete
		//     route lowercasing the API name and TiDB's case-insensitive table
		//     matching, a DELETE of e.g. "_System_InstalledPack" (any casing)
		//     would otherwise DROP the live pack-install registry.
		//
		//  2. ReadSource corroboration: a physical table can legitimately back a
		//     LIVE object under a different logical name (ReadSource aliasing). If
		//     any live schema resolves its read table to this name, the "no
		//     metadata ⇒ droppable orphan" assumption is false — refuse to drop.
		//
		// The reconcile is a destructive admin action, so it is audit-logged
		// whether or not it drops anything.
		//
		// Single-writer boundary: ms.mu serializes this reconcile against
		// CreateSchema in-process, and the strict plain-CREATE closes cross-
		// replica create-vs-create. The residual reconcile-DROP-vs-remote-create
		// window across replicas is NOT closed transactionally (TiDB DDL auto-
		// commits; pessimistic locks do not gap-lock absent unique keys). This PR
		// assumes single-writer interactive authoring; a DB-level authoring name
		// lock is the named follow-up if multi-replica authoring is supported.
		if ms.shouldReconcileOrphanTable(ctx, apiName) {
			log.Printf("🧹 DeleteSchema reconcile: dropping stranded orphan table %q (no live metadata)", apiName)
			reconcileErr := ms.schemaMgr.DropTable(apiName)
			ms.invalidateCacheLocked()
			if reconcileErr != nil {
				return wrapDropTableError(reconcileErr, "failed to reconcile orphaned table for "+apiName)
			}
		}
		return apperrors.NewNotFoundError("Object", apiName)
	}
	apiName = obj.APIName

	// Unregister metadata and drop the physical table. DropTable (#3484) removes
	// the object's metadata in a transaction FIRST, then drops the table — so if
	// the physical DROP fails, the metadata is already gone and any cached copy
	// is now stale. Invalidate the cache regardless of the outcome so a phantom
	// object cannot survive; only a fully clean delete publishes the event.
	dropErr := ms.schemaMgr.DropTable(apiName)
	ms.invalidateCacheLocked()
	if dropErr != nil {
		// An inbound-FK-blocked delete surfaces as a typed ConflictError naming
		// the blocking constraint(s); pass it through unwrapped so the constraint
		// names and remediation reach the API client instead of a sanitized 500.
		return wrapDropTableError(dropErr, "failed to drop table and schema")
	}

	// Publish Event
	ms.publishMetadataEvent(ctx, sharedEvents.ObjectDeleted, sharedEvents.SchemaEventPayload{
		EventType:     sharedEvents.ObjectDeleted,
		ObjectAPIName: apiName,
		Timestamp:     time.Now().Format(time.RFC3339),
	})

	return nil
}

// shouldReconcileOrphanTable decides whether the DeleteSchema no-live-metadata
// branch may DROP the physical table of apiName. It returns false (refuse to
// drop) when apiName is managed substrate, or when a live object's ReadSource
// resolves its physical read table to apiName — in either case the "no metadata
// ⇒ droppable orphan" assumption does not hold. See DeleteSchema for the full
// rationale and boundaries (#3484).
func (ms *MetadataService) shouldReconcileOrphanTable(ctx context.Context, apiName string) bool {
	if constants.IsManagedSystemTable(apiName) {
		log.Printf("🛑 DeleteSchema reconcile refused: %q is managed system substrate, never dropped", apiName)
		return false
	}
	// ReadSource corroboration REQUIRES a loaded schema cache. GetSchemas
	// swallows a cache-init failure into an EMPTY slice, which would make the
	// loop below see zero live objects and wrongly authorize the destructive
	// DROP — dropping the physical read table of a LIVE ReadSource-aliased
	// object on a transient snapshot-build error (cold cache is the norm here:
	// invalidateCacheLocked fires after every DropTable). Fail CLOSED: if the
	// cache cannot load, refuse to reconcile (#3484).
	if err := ms.ensureCoreCacheInitialized(ctx); err != nil {
		log.Printf("🛑 DeleteSchema reconcile refused: core schema cache failed to initialize while checking %q: %v", apiName, err)
		return false
	}
	// ReadSource corroboration: a live object may physically read from a table
	// whose name differs from its own logical API name. Dropping such a table
	// would destroy data serving a LIVE object.
	for _, live := range ms.GetSchemas(ctx) {
		physical, err := live.ResolveReadSourceTable()
		if err != nil {
			// A schema whose read source cannot resolve is already broken; do not
			// let it silently authorize a destructive drop — refuse conservatively.
			log.Printf("🛑 DeleteSchema reconcile refused: could not resolve read source for %q while checking %q: %v", live.APIName, apiName, err)
			return false
		}
		if strings.EqualFold(physical, apiName) {
			log.Printf("🛑 DeleteSchema reconcile refused: %q is the physical read table of live object %q", apiName, live.APIName)
			return false
		}
	}
	// Platform-provenance proof: the reconcile authorizes a destructive DROP for
	// a metadata-absent name, but "no metadata" alone does NOT prove the table
	// was ever a platform object. The proof is the DROP-PENDING tombstone — a
	// _System_Table row with IsDeleted=1 that DropTable writes ONLY when a
	// physical DROP failed after the metadata purge. A successful drop deletes
	// the registry row inside the metadata transaction and never writes the
	// marker, so no residue of a successful delete can later authorize
	// destruction of an unrelated same-name table; and a raw operator/ETL/backup
	// table sharing the name — including a same-shape CREATE TABLE ... LIKE
	// copy — never has the marker and is never dropped: DeleteSchema stays a
	// fail-closed 404 no-op, exactly the base behavior. Fail CLOSED on probe
	// error: a probe that cannot run must never authorize a destructive drop.
	dropPending, err := ms.schemaMgr.HasDropPendingTombstone(apiName)
	if err != nil {
		log.Printf("🛑 DeleteSchema reconcile refused: could not probe the drop-pending tombstone for %q: %v", apiName, err)
		return false
	}
	if !dropPending {
		log.Printf("🛑 DeleteSchema reconcile refused: %q has no live metadata and no drop-pending tombstone (not a platform orphan); not dropping", apiName)
		return false
	}
	return true
}

// wrapDropTableError preserves a typed ConflictError from DropTable (an inbound-
// FK-blocked delete names the referencing constraint(s), or an orphan-block
// remediation) instead of blanket-wrapping it in an InternalError, so the 409
// status and the constraint names reach the API client (#3484).
func wrapDropTableError(err error, msg string) error {
	var conflict *apperrors.ConflictError
	if errors.As(err, &conflict) {
		return conflict
	}
	return apperrors.NewInternalError(msg, err)
}

// Field CRUD methods are split across:
// - field_create.go (CreateField)
// - field_update.go (UpdateField)
// - field_delete.go (DeleteField)
// - field_batch.go  (BatchCreateFields)