package tenantoverlay

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"path/filepath"
	"sort"
	"strings"

	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	domainschema "github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/records"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
)

// applyInsertRules seeds tenant-authored insert-inheritance metadata plus the
// retained compatibility child-clone surface:
//   - _System_ObjectFieldMapping rows with ActionType='insertInheritFromLookup'
//     (from object_field_mappings.json)
//   - _System_ObjectFieldMapping rows with other seed-authored ActionType values
//   - _System_Object.ChildCloneRules JSON column (custom objects from
//     tenant_objects_spec.json; standard-object overrides from
//     object_child_clone_rules.json)
//
// Idempotent: repeated apply-tenant runs produce no drift; fields stamped
// by previous applies are overwritten only when the file changes.
func (p *Pack) applyInsertRules(ctx context.Context, db *sql.DB, specs []packs.ObjectSpec, childCloneRulePackOwnership map[string]childCloneRulePackOwnership) error {
	if err := p.applyObjectFieldMappingsOverlay(ctx, db); err != nil {
		return fmt.Errorf("tenant object field mappings: %w", err)
	}
	if err := p.applyObjectChildCloneRulesOverlay(ctx, db, specs, childCloneRulePackOwnership); err != nil {
		return fmt.Errorf("tenant object child clone rules: %w", err)
	}
	return nil
}

// applyObjectFieldMappingsOverlay routes tenant-seeded
// object-field-mapping rows through the shared owner-scoped writer. The
// tenant owner is part of the reconcile WHERE clause, so a partial overlay
// can add rows without pruning platform-standard mappings.
func (p *Pack) applyObjectFieldMappingsOverlay(ctx context.Context, db *sql.DB) error {
	objectFieldMappingsJSON, ok, err := p.readOptional(ObjectFieldMappingsFile)
	if err != nil {
		return err
	}
	if !ok {
		return nil
	}
	var groups []connector.FieldMappingGroup
	if err := json.Unmarshal(objectFieldMappingsJSON, &groups); err != nil {
		return fmt.Errorf("parse tenant object_field_mappings.json: %w", err)
	}
	if len(groups) == 0 {
		return nil
	}
	return packs.UpsertFieldMappingGroupsForOwner(ctx, db, groups, p.objectFieldMappingOwner())
}

func (p *Pack) objectFieldMappingOwner() string {
	if p == nil || strings.TrimSpace(p.Dir) == "" {
		return "tenant:overlay"
	}
	base := filepath.Base(filepath.Clean(p.Dir))
	if base == "." || base == string(filepath.Separator) || strings.TrimSpace(base) == "" {
		return "tenant:overlay"
	}
	return "tenant:" + base
}

// applyObjectChildCloneRulesOverlay updates `_System_Object.ChildCloneRules`
// from retained compatibility L2 tenant overlay payload and stamps
// ChildCloneRulesCustomized so defaults-only pack reapply/verify preserves the
// tenant-owned value. New create-from-source authoring is Flow-owned under
// issue #979; this path only keeps existing/transitional child-clone metadata
// safe while it remains supported.
func (p *Pack) applyObjectChildCloneRulesOverlay(ctx context.Context, db *sql.DB, specs []packs.ObjectSpec, childCloneRulePackOwnership ...map[string]childCloneRulePackOwnership) error {
	ownership := firstChildCloneRulePackOwnership(childCloneRulePackOwnership)
	keep := tenantObjectSpecChildCloneRuleObjects(specs)
	rulesByObject, ok, err := p.readObjectChildCloneRulesFile()
	if err != nil {
		return err
	}
	if !ok {
		return clearMissingChildCloneRuleOverlays(ctx, db, keep, ObjectChildCloneRulesFile, ownership)
	}
	if err := rejectCustomObjectChildCloneRulesFileObjects(specs, rulesByObject); err != nil {
		return err
	}
	if err := rejectLockedObjectChildCloneRuleOverlays(rulesByObject, ownership); err != nil {
		return err
	}
	for apiName := range rulesByObject {
		keep[strings.ToLower(strings.TrimSpace(apiName))] = true
	}
	if err := applyChildCloneRulesOverlayMap(ctx, db, rulesByObject, ObjectChildCloneRulesFile); err != nil {
		return err
	}
	return clearMissingChildCloneRuleOverlays(ctx, db, keep, ObjectChildCloneRulesFile, ownership)
}

func applyChildCloneRulesOverlayMap(ctx context.Context, db *sql.DB, rulesByObject map[string][]models.ChildCloneRule, source string) error {
	if len(rulesByObject) == 0 {
		return nil
	}
	for apiName, rules := range rulesByObject {
		objectID := domainschema.GenerateObjectID(apiName)
		if rules == nil {
			rules = []models.ChildCloneRule{}
		}
		encoded, err := json.Marshal(rules)
		if err != nil {
			return fmt.Errorf("encode ChildCloneRules for %s: %w", apiName, err)
		}
		query := fmt.Sprintf(
			"UPDATE `%s` SET `%s` = ?, `%s` = 1, `%s` = NOW() WHERE `%s` = ?",
			constants.TableObject,
			constants.FieldSysObject_ChildCloneRules,
			constants.FieldSysObject_ChildCloneRulesCustomized,
			constants.FieldLastModifiedDate,
			constants.ColID,
		)
		res, err := db.ExecContext(ctx, query, string(encoded), objectID)
		if err != nil {
			return fmt.Errorf("update ChildCloneRules for %s: %w", apiName, err)
		}
		affected, _ := res.RowsAffected()
		if affected == 0 {
			return fmt.Errorf(
				"ChildCloneRules overlay %s for %q updated 0 rows (expected id=%s on %s) -- "+
					"parent object row missing. Check apply-tenant ordering: "+
					"tenant_objects_spec.json apply must precede applyInsertRules.",
				source, apiName, objectID, constants.TableObject,
			)
		}
		log.Printf("   ✅ ChildCloneRules seeded on %s (%d rules)", apiName, len(rules))
	}
	return nil
}

func tenantObjectSpecChildCloneRuleObjects(specs []packs.ObjectSpec) map[string]bool {
	keep := make(map[string]bool)
	for _, spec := range specs {
		if !isTenantCustomObjectSpec(spec) {
			continue
		}
		if !tenantObjectSpecDeclaresChildCloneRules(spec) {
			continue
		}
		key := strings.ToLower(strings.TrimSpace(spec.APIName))
		if key != "" {
			keep[key] = true
		}
	}
	return keep
}

func tenantObjectSpecChildCloneRules(specs []packs.ObjectSpec) map[string][]models.ChildCloneRule {
	rulesByObject := make(map[string][]models.ChildCloneRule)
	for _, spec := range specs {
		if !isTenantCustomObjectSpec(spec) {
			continue
		}
		if !tenantObjectSpecDeclaresChildCloneRules(spec) {
			continue
		}
		rulesByObject[spec.APIName] = append([]models.ChildCloneRule(nil), spec.ChildCloneRules...)
	}
	return rulesByObject
}

func tenantObjectSpecDeclaresChildCloneRules(spec packs.ObjectSpec) bool {
	return spec.ChildCloneRules != nil
}

func (p *Pack) readObjectChildCloneRulesFile() (map[string][]models.ChildCloneRule, bool, error) {
	objectChildCloneRulesJSON, ok, err := p.readOptional(ObjectChildCloneRulesFile)
	if err != nil {
		return nil, false, err
	}
	if !ok {
		return nil, false, nil
	}
	rulesByObject := map[string][]models.ChildCloneRule{}
	if err := json.Unmarshal(objectChildCloneRulesJSON, &rulesByObject); err != nil {
		return nil, false, fmt.Errorf("tenant object_child_clone_rules.json is not parseable: %w", err)
	}
	return rulesByObject, true, nil
}

func (p *Pack) validateObjectChildCloneRulesFilePayload(ctx context.Context, deps packs.ApplyDeps, specs []packs.ObjectSpec, childCloneRulePackOwnership map[string]childCloneRulePackOwnership) error {
	rulesByObject, ok, err := p.readObjectChildCloneRulesFile()
	if err != nil || !ok {
		return err
	}
	if err := rejectCustomObjectChildCloneRulesFileObjects(specs, rulesByObject); err != nil {
		return err
	}
	if err := rejectLockedObjectChildCloneRuleOverlays(rulesByObject, childCloneRulePackOwnership); err != nil {
		return err
	}
	return validateObjectChildCloneRulesFilePreflight(ctx, deps, specs, rulesByObject)
}

func rejectCustomObjectChildCloneRulesFileObjects(specs []packs.ObjectSpec, rulesByObject map[string][]models.ChildCloneRule) error {
	customSpecs := tenantObjectSpecChildCloneRuleObjects(specs)
	for apiName := range rulesByObject {
		trimmed := strings.TrimSpace(apiName)
		key := strings.ToLower(trimmed)
		if key == "" {
			continue
		}
		if customSpecs[key] || strings.HasSuffix(key, "__c") {
			return fmt.Errorf(
				"%s declares compatibility childCloneRules for custom object %s; custom-object compatibility rules belong in %s, while %s is reserved for standard-object L2 compatibility overrides",
				ObjectChildCloneRulesFile,
				trimmed,
				ObjectsSpecFile,
				ObjectChildCloneRulesFile,
			)
		}
	}
	return nil
}

func rejectLockedObjectChildCloneRuleOverlays(rulesByObject map[string][]models.ChildCloneRule, childCloneRulePackOwnership map[string]childCloneRulePackOwnership) error {
	for apiName := range rulesByObject {
		key := strings.ToLower(strings.TrimSpace(apiName))
		ownership, ok := childCloneRulePackOwnership[key]
		if !ok {
			return fmt.Errorf(
				"%s declares compatibility childCloneRules for standard object %s, but _System_Object has no persisted ChildCloneRules pack ownership after active-pack ownership backfill; reapply the installed source-backed pack before retained L2 overrides",
				ObjectChildCloneRulesFile,
				strings.TrimSpace(apiName),
			)
		}
		if ownership.Manageability != packs.ManageabilityLocked {
			continue
		}
		return fmt.Errorf(
			"%s declares compatibility childCloneRules for locked pack object %s owned by %s object_spec %s; locked object_spec childCloneRules are not L2-overridable",
			ObjectChildCloneRulesFile,
			strings.TrimSpace(apiName),
			ownership.PackName,
			ownership.BusinessKey,
		)
	}
	return nil
}

func validateObjectChildCloneRulesFilePreflight(ctx context.Context, deps packs.ApplyDeps, specs []packs.ObjectSpec, rulesByObject map[string][]models.ChildCloneRule) error {
	if len(rulesByObject) == 0 {
		return nil
	}
	var fallback records.DefaultChildRowsSchemaSnapshot
	if deps.Metadata != nil {
		fallback = deps.Metadata
	}
	snapshot := newTenantChildCloneRulePreflightSnapshot(specs, fallback)
	parentSchemas := make([]*models.ObjectMetadata, 0, len(rulesByObject))
	var missing []string
	for apiName, rules := range rulesByObject {
		trimmed := strings.TrimSpace(apiName)
		if trimmed == "" {
			missing = append(missing, "<empty object apiName>")
			continue
		}
		schema := snapshot.GetSchema(ctx, trimmed)
		if schema == nil {
			missing = append(missing, trimmed)
			continue
		}
		clone := models.CloneWithAdditionalFields(schema, nil)
		clone.ChildCloneRules = append([]models.ChildCloneRule(nil), rules...)
		parentSchemas = append(parentSchemas, clone)
	}
	if len(missing) > 0 {
		sort.Strings(missing)
		return fmt.Errorf("tenant %s preflight: object(s) not found in metadata: %s", ObjectChildCloneRulesFile, strings.Join(missing, ", "))
	}
	if err := packs.ValidateObjectChildCloneRules(ctx, snapshot, parentSchemas); err != nil {
		return fmt.Errorf("tenant %s preflight: %w", ObjectChildCloneRulesFile, err)
	}
	return nil
}

type tenantChildCloneRulePreflightSnapshot struct {
	specs    map[string]*models.ObjectMetadata
	fallback records.DefaultChildRowsSchemaSnapshot
}

func newTenantChildCloneRulePreflightSnapshot(specs []packs.ObjectSpec, fallback records.DefaultChildRowsSchemaSnapshot) tenantChildCloneRulePreflightSnapshot {
	return tenantChildCloneRulePreflightSnapshot{
		specs:    packs.BuildObjectMetadataIndex(specs),
		fallback: fallback,
	}
}

func (s tenantChildCloneRulePreflightSnapshot) GetSchema(ctx context.Context, apiName string) *models.ObjectMetadata {
	key := strings.ToLower(strings.TrimSpace(apiName))
	overlay := s.specs[key]
	var base *models.ObjectMetadata
	if s.fallback != nil {
		base = s.fallback.GetSchema(ctx, apiName)
	}
	if base != nil {
		var additions []models.FieldMetadata
		if overlay != nil {
			additions = append([]models.FieldMetadata(nil), overlay.Fields...)
		}
		clone := models.CloneWithAdditionalFields(base, additions)
		if overlay != nil && len(overlay.ChildCloneRules) > 0 {
			clone.ChildCloneRules = append([]models.ChildCloneRule(nil), overlay.ChildCloneRules...)
		}
		return clone
	}
	if overlay != nil {
		return models.CloneWithAdditionalFields(overlay, nil)
	}
	return nil
}

func clearMissingChildCloneRuleOverlays(ctx context.Context, db *sql.DB, keep map[string]bool, source string, childCloneRulePackOwnership ...map[string]childCloneRulePackOwnership) error {
	ownership := firstChildCloneRulePackOwnership(childCloneRulePackOwnership)
	clearKeep := make(map[string]bool, len(keep))
	for key, value := range keep {
		if value {
			clearKeep[strings.ToLower(strings.TrimSpace(key))] = true
		}
	}
	keepClause, args := childCloneRuleKeepExclusion(clearKeep)
	if err := rejectUnresolvedSidecarlessChildCloneRules(ctx, db, source, keepClause, args); err != nil {
		return err
	}
	if _, err := restoreMissingPackChildCloneRuleOverlays(ctx, db, keep, ownership, source); err != nil {
		return err
	}
	query := fmt.Sprintf(
		"UPDATE `%s` SET `%s` = NULL, `%s` = 0, `%s` = NOW() WHERE COALESCE(`%s`, 0) <> 0",
		constants.TableObject,
		constants.FieldSysObject_ChildCloneRules,
		constants.FieldSysObject_ChildCloneRulesCustomized,
		constants.FieldLastModifiedDate,
		constants.FieldSysObject_ChildCloneRulesCustomized,
	)
	query += keepClause
	res, err := db.ExecContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("clear stale ChildCloneRules overlays for %s: %w", source, err)
	}
	affected, _ := res.RowsAffected()
	if affected > 0 {
		log.Printf("   ✅ cleared %d stale ChildCloneRules overlay rows for %s", affected, source)
	}
	return nil
}

func childCloneRuleKeepExclusion(clearKeep map[string]bool) (string, []interface{}) {
	args := make([]interface{}, 0, len(clearKeep))
	if len(clearKeep) > 0 {
		keys := make([]string, 0, len(clearKeep))
		for key := range clearKeep {
			if strings.TrimSpace(key) != "" {
				keys = append(keys, strings.ToLower(strings.TrimSpace(key)))
			}
		}
		sort.Strings(keys)
		if len(keys) > 0 {
			placeholders := make([]string, 0, len(keys))
			for _, key := range keys {
				placeholders = append(placeholders, "?")
				args = append(args, key)
			}
			return " AND LOWER(`" + constants.FieldSysObject_APIName + "`) NOT IN (" + strings.Join(placeholders, ",") + ")", args
		}
	}
	return "", args
}

func rejectUnresolvedSidecarlessChildCloneRules(ctx context.Context, db *sql.DB, source string, keepClause string, args []interface{}) error {
	query := fmt.Sprintf(
		"SELECT `%s` FROM `%s` WHERE COALESCE(`%s`, 0) = 0 AND `%s` IS NOT NULL "+
			"AND TRIM(`%s`) <> '' AND TRIM(`%s`) <> '[]' AND COALESCE(`%s`, '') = '' AND COALESCE(`%s`, '') = ''",
		constants.FieldSysObject_APIName,
		constants.TableObject,
		constants.FieldSysObject_IsDeleted,
		constants.FieldSysObject_ChildCloneRules,
		constants.FieldSysObject_ChildCloneRules,
		constants.FieldSysObject_ChildCloneRules,
		constants.FieldSysObject_ChildCloneRulesPackName,
		constants.FieldSysObject_ChildCloneRulesPackBusinessKey,
	)
	query += keepClause
	query += " ORDER BY `" + constants.FieldSysObject_APIName + "`"
	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return fmt.Errorf("query unresolved sidecar-less ChildCloneRules rows for %s: %w", source, err)
	}
	defer rows.Close()
	var unresolved []string
	for rows.Next() {
		var apiName string
		if err := rows.Scan(&apiName); err != nil {
			return fmt.Errorf("scan unresolved sidecar-less ChildCloneRules rows for %s: %w", source, err)
		}
		unresolved = append(unresolved, apiName)
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("iterate unresolved sidecar-less ChildCloneRules rows for %s: %w", source, err)
	}
	if len(unresolved) == 0 {
		return nil
	}
	return fmt.Errorf(
		"cannot safely clean unresolved sidecar-less ChildCloneRules rows while applying %s: %s. "+
			"Run the active-pack ChildCloneRules ownership backfill/reconcile before apply-tenant, "+
			"or explicitly reauthor/clear the tenant-owned rule through current metadata before removing the legacy source.",
		source,
		strings.Join(unresolved, ", "),
	)
}

func firstChildCloneRulePackOwnership(values []map[string]childCloneRulePackOwnership) map[string]childCloneRulePackOwnership {
	if len(values) == 0 {
		return nil
	}
	return values[0]
}

func mergeChildCloneRulePackOwnership(left, right map[string]childCloneRulePackOwnership) map[string]childCloneRulePackOwnership {
	if len(left) == 0 {
		return right
	}
	if len(right) == 0 {
		return left
	}
	merged := make(map[string]childCloneRulePackOwnership, len(left)+len(right))
	for key, value := range left {
		merged[key] = value
	}
	for key, value := range right {
		merged[key] = value
	}
	return merged
}

func restoreMissingPackChildCloneRuleOverlays(ctx context.Context, db *sql.DB, keep map[string]bool, childCloneRulePackOwnership map[string]childCloneRulePackOwnership, source string) (map[string]bool, error) {
	if len(childCloneRulePackOwnership) == 0 {
		return nil, nil
	}
	restored := make(map[string]bool)
	keys := make([]string, 0, len(childCloneRulePackOwnership))
	for key := range childCloneRulePackOwnership {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	for _, key := range keys {
		if keep[strings.ToLower(strings.TrimSpace(key))] {
			continue
		}
		ownership := childCloneRulePackOwnership[key]
		if ownership.MissingPackOwnership {
			continue
		}
		var encoded interface{}
		if len(ownership.Rules) > 0 {
			raw, err := json.Marshal(ownership.Rules)
			if err != nil {
				return nil, fmt.Errorf("encode pack ChildCloneRules for %s: %w", ownership.ObjectAPIName, err)
			}
			encoded = string(raw)
		}
		query := fmt.Sprintf(
			"UPDATE `%s` SET `%s` = ?, `%s` = 0, `%s` = NOW() WHERE COALESCE(`%s`, 0) <> 0 AND LOWER(`%s`) = ?",
			constants.TableObject,
			constants.FieldSysObject_ChildCloneRules,
			constants.FieldSysObject_ChildCloneRulesCustomized,
			constants.FieldLastModifiedDate,
			constants.FieldSysObject_ChildCloneRulesCustomized,
			constants.FieldSysObject_APIName,
		)
		res, err := db.ExecContext(ctx, query, encoded, key)
		if err != nil {
			return nil, fmt.Errorf("restore pack ChildCloneRules for %s after removing %s override: %w", ownership.ObjectAPIName, source, err)
		}
		affected, _ := res.RowsAffected()
		if affected > 0 {
			log.Printf("   ✅ restored %s ChildCloneRules from pack %s after removing %s overlay", ownership.ObjectAPIName, ownership.PackName, source)
			restored[key] = true
		}
	}
	return restored, nil
}

type childCloneRulePackOwnership struct {
	ObjectAPIName        string
	PackName             string
	BusinessKey          string
	Manageability        packs.Manageability
	Rules                []models.ChildCloneRule
	MissingPackOwnership bool
}

func (p *Pack) loadChildCloneRulePackOwnershipForApply(ctx context.Context, db *sql.DB) (map[string]childCloneRulePackOwnership, error) {
	if db == nil {
		return nil, nil
	}
	targets := map[string]bool{}
	if rulesByObject, ok, err := p.readObjectChildCloneRulesFile(); err != nil {
		return nil, err
	} else if ok {
		for apiName := range rulesByObject {
			key := strings.ToLower(strings.TrimSpace(apiName))
			if key != "" {
				targets[key] = true
			}
		}
	}
	return loadChildCloneRulePackOwnershipRows(ctx, db, targets)
}

func loadCustomChildCloneRulePackOwnershipForApply(ctx context.Context, db *sql.DB, specs []packs.ObjectSpec) (map[string]childCloneRulePackOwnership, error) {
	if db == nil {
		return nil, nil
	}
	targets := map[string]bool{}
	for _, spec := range specs {
		if !isTenantCustomObjectSpec(spec) {
			continue
		}
		key := strings.ToLower(strings.TrimSpace(spec.APIName))
		if key != "" {
			targets[key] = true
		}
	}
	return loadCustomChildCloneRulePackOwnershipRows(ctx, db, targets)
}

func loadChildCloneRulePackOwnershipRows(ctx context.Context, db *sql.DB, targets map[string]bool) (map[string]childCloneRulePackOwnership, error) {
	if db == nil {
		return nil, nil
	}
	query := fmt.Sprintf(
		"SELECT `%s`, COALESCE(`%s`, ''), COALESCE(`%s`, ''), COALESCE(`%s`, ''), `%s` FROM `%s` "+
			"WHERE COALESCE(`%s`, 0) = 0 AND COALESCE(`%s`, 0) = 0 AND (COALESCE(`%s`, 0) <> 0",
		constants.FieldSysObject_APIName,
		constants.FieldSysObject_ChildCloneRulesPackName,
		constants.FieldSysObject_ChildCloneRulesPackBusinessKey,
		constants.FieldSysObject_ChildCloneRulesPackManageability,
		constants.FieldSysObject_ChildCloneRulesPackValue,
		constants.TableObject,
		constants.FieldSysObject_IsDeleted,
		constants.FieldSysObject_IsCustom,
		constants.FieldSysObject_ChildCloneRulesCustomized,
	)
	args := make([]interface{}, 0, len(targets))
	keys := make([]string, 0, len(targets))
	for key := range targets {
		key = strings.ToLower(strings.TrimSpace(key))
		if key != "" {
			keys = append(keys, key)
		}
	}
	sort.Strings(keys)
	if len(keys) > 0 {
		placeholders := make([]string, 0, len(keys))
		for _, key := range keys {
			placeholders = append(placeholders, "?")
			args = append(args, key)
		}
		query += " OR LOWER(`" + constants.FieldSysObject_APIName + "`) IN (" + strings.Join(placeholders, ",") + ")"
	}
	query += ") ORDER BY `" + constants.FieldSysObject_APIName + "`"

	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("query ChildCloneRules pack ownership rows: %w", err)
	}
	defer rows.Close()

	ownership, err := scanChildCloneRulePackOwnershipRows(rows)
	if err != nil {
		return nil, err
	}
	if len(ownership) == 0 {
		return nil, nil
	}
	return ownership, nil
}

func loadCustomChildCloneRulePackOwnershipRows(ctx context.Context, db *sql.DB, targets map[string]bool) (map[string]childCloneRulePackOwnership, error) {
	if db == nil {
		return nil, nil
	}
	query := fmt.Sprintf(
		"SELECT o.`%s`, COALESCE(o.`%s`, ''), COALESCE(o.`%s`, ''), COALESCE(o.`%s`, ''), o.`%s`, o.`%s`, COALESCE(o.`%s`, 0), "+
			"COALESCE((SELECT COUNT(*) FROM `%s` f WHERE f.`%s` = o.`%s` AND COALESCE(f.`%s`, 0) = 0 AND COALESCE(f.`%s`, '') IN ('%s', '%s') AND COALESCE(f.`%s`, '') <> ''), 0) "+
			"FROM `%s` o WHERE COALESCE(o.`%s`, 0) = 0 AND COALESCE(o.`%s`, 0) <> 0 AND (COALESCE(o.`%s`, 0) <> 0",
		constants.FieldSysObject_APIName,
		constants.FieldSysObject_ChildCloneRulesPackName,
		constants.FieldSysObject_ChildCloneRulesPackBusinessKey,
		constants.FieldSysObject_ChildCloneRulesPackManageability,
		constants.FieldSysObject_ChildCloneRulesPackValue,
		constants.FieldSysObject_ChildCloneRules,
		constants.FieldSysObject_ChildCloneRulesCustomized,
		constants.TableField,
		constants.FieldSysField_ObjectID,
		constants.FieldID,
		constants.FieldSysField_IsDeleted,
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindPack,
		constants.FieldOwnerKindPackRemoving,
		constants.FieldSysField_OwnerPack,
		constants.TableObject,
		constants.FieldSysObject_IsDeleted,
		constants.FieldSysObject_IsCustom,
		constants.FieldSysObject_ChildCloneRulesCustomized,
	)
	args := make([]interface{}, 0, len(targets))
	keys := make([]string, 0, len(targets))
	for key := range targets {
		key = strings.ToLower(strings.TrimSpace(key))
		if key != "" {
			keys = append(keys, key)
		}
	}
	sort.Strings(keys)
	if len(keys) > 0 {
		placeholders := make([]string, 0, len(keys))
		for _, key := range keys {
			placeholders = append(placeholders, "?")
			args = append(args, key)
		}
		query += " OR LOWER(o.`" + constants.FieldSysObject_APIName + "`) IN (" + strings.Join(placeholders, ",") + ")"
	}
	query += ") ORDER BY o.`" + constants.FieldSysObject_APIName + "`"

	rows, err := db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("query custom ChildCloneRules pack ownership rows: %w", err)
	}
	defer rows.Close()

	ownership, err := scanCustomChildCloneRulePackOwnershipRows(rows)
	if err != nil {
		return nil, err
	}
	if len(ownership) == 0 {
		return nil, nil
	}
	return ownership, nil
}

func scanCustomChildCloneRulePackOwnershipRows(rows *sql.Rows) (map[string]childCloneRulePackOwnership, error) {
	ownership := map[string]childCloneRulePackOwnership{}
	for rows.Next() {
		var apiName, packName, businessKey, manageability string
		var rawRules, liveRules sql.NullString
		var customized, packOwnedFieldCount int
		if err := rows.Scan(&apiName, &packName, &businessKey, &manageability, &rawRules, &liveRules, &customized, &packOwnedFieldCount); err != nil {
			return nil, fmt.Errorf("scan custom ChildCloneRules pack ownership row: %w", err)
		}
		apiName = strings.TrimSpace(apiName)
		packName = strings.TrimSpace(packName)
		businessKey = strings.TrimSpace(businessKey)
		manageability = strings.TrimSpace(manageability)
		if apiName == "" {
			continue
		}
		key := strings.ToLower(apiName)
		if packName == "" {
			if customized == 0 && (packOwnedFieldCount > 0 || (liveRules.Valid && strings.TrimSpace(liveRules.String) != "")) {
				ownership[key] = childCloneRulePackOwnership{
					ObjectAPIName:        apiName,
					MissingPackOwnership: true,
				}
			}
			continue
		}
		var rules []models.ChildCloneRule
		if rawRules.Valid && strings.TrimSpace(rawRules.String) != "" {
			if err := json.Unmarshal([]byte(rawRules.String), &rules); err != nil {
				return nil, fmt.Errorf("parse custom pack ChildCloneRules ownership value for %s: %w", apiName, err)
			}
		}
		if prev, exists := ownership[key]; exists {
			return nil, fmt.Errorf(
				"object %s childCloneRules ownership declared by both %s object_spec %s and %s object_spec %s",
				apiName,
				prev.PackName,
				prev.BusinessKey,
				packName,
				businessKey,
			)
		}
		ownership[key] = childCloneRulePackOwnership{
			ObjectAPIName: apiName,
			PackName:      packName,
			BusinessKey:   businessKey,
			Manageability: packs.Manageability(manageability),
			Rules:         append([]models.ChildCloneRule(nil), rules...),
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate custom ChildCloneRules pack ownership rows: %w", err)
	}
	return ownership, nil
}

func scanChildCloneRulePackOwnershipRows(rows *sql.Rows) (map[string]childCloneRulePackOwnership, error) {
	ownership := map[string]childCloneRulePackOwnership{}
	for rows.Next() {
		var apiName, packName, businessKey, manageability string
		var rawRules sql.NullString
		if err := rows.Scan(&apiName, &packName, &businessKey, &manageability, &rawRules); err != nil {
			return nil, fmt.Errorf("scan ChildCloneRules pack ownership row: %w", err)
		}
		apiName = strings.TrimSpace(apiName)
		packName = strings.TrimSpace(packName)
		businessKey = strings.TrimSpace(businessKey)
		manageability = strings.TrimSpace(manageability)
		if apiName == "" || packName == "" {
			continue
		}
		var rules []models.ChildCloneRule
		if rawRules.Valid && strings.TrimSpace(rawRules.String) != "" {
			if err := json.Unmarshal([]byte(rawRules.String), &rules); err != nil {
				return nil, fmt.Errorf("parse pack ChildCloneRules ownership value for %s: %w", apiName, err)
			}
		}
		key := strings.ToLower(apiName)
		if prev, exists := ownership[key]; exists {
			return nil, fmt.Errorf(
				"object %s childCloneRules ownership declared by both %s object_spec %s and %s object_spec %s",
				apiName,
				prev.PackName,
				prev.BusinessKey,
				packName,
				businessKey,
			)
		}
		ownership[key] = childCloneRulePackOwnership{
			ObjectAPIName: apiName,
			PackName:      packName,
			BusinessKey:   businessKey,
			Manageability: packs.Manageability(manageability),
			Rules:         append([]models.ChildCloneRule(nil), rules...),
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate ChildCloneRules pack ownership rows: %w", err)
	}
	return ownership, nil
}

// assertTenantInsertRulesClean runs the pure-schema sanity pass after
// apply: every referenced object / field must resolve. This is a lighter
// check than the platform-level ValidateObjectChildCloneRules +
// ValidateInsertInheritFromLookupRules functions — those need a live
// metadata service, which this tenant overlay does have via ApplyDeps,
// so the two callers can coexist.
func (p *Pack) assertTenantInsertRulesSurfacesLoadable() error {
	// Parse the insert-rule files purely to catch JSON schema
	// drift at apply time (e.g. someone renamed a field). Detailed
	// cross-schema checks run in the platform validator pass immediately
	// after.
	if objectFieldMappingsJSON, ok, err := p.readOptional(ObjectFieldMappingsFile); err != nil {
		return err
	} else if ok {
		var groups []connector.FieldMappingGroup
		if err := json.Unmarshal(objectFieldMappingsJSON, &groups); err != nil {
			return fmt.Errorf("tenant object_field_mappings.json is not parseable as FieldMappingGroup: %w", err)
		}
		for _, g := range groups {
			if strings.TrimSpace(g.SourceObject) == "" || strings.TrimSpace(g.TargetObject) == "" {
				return fmt.Errorf("tenant object_field_mappings.json group missing sourceObject/targetObject: %+v", g)
			}
			if g.ActionType == "" {
				return fmt.Errorf("tenant object_field_mappings.json group missing actionType: %s→%s", g.SourceObject, g.TargetObject)
			}
		}
	}
	if objectChildCloneRulesJSON, ok, err := p.readOptional(ObjectChildCloneRulesFile); err != nil {
		return err
	} else if ok {
		rules := map[string][]models.ChildCloneRule{}
		if err := json.Unmarshal(objectChildCloneRulesJSON, &rules); err != nil {
			return fmt.Errorf("tenant object_child_clone_rules.json is not parseable: %w", err)
		}
	}
	return nil
}

// VerifyInsertRulesLanded is the declared-vs-DB closure check for the
// tenant metadata surfaces introduced by the insert-inheritance cutover.
// Invoked only from controlplane.ApplyTenant (NOT from generic
// ApplyPacks — platform boundary, round-7 reviewer finding #1).
//
// Why this exists separately from ValidateInsertInheritFromLookupRules
// (packs/insert_rules_validator.go): that validator inspects the rows
// that *currently exist* in DB and rejects any that reference missing
// metadata. It cannot detect a silent write-miss ("seed declared X, DB
// has no X") because "no row in DB" is indistinguishable from "seed
// declared nothing." The round-3 reviewer finding #2 is exactly this
// blind spot.
//
// _System_ObjectFieldMapping is admin-editable metadata. DB is the live
// authority; seed is a default template; IsCustomized=1 rows are
// first-class admin overrides preserved by the convergent upsert.
// Closure here is BUSINESS-KEY EXISTENCE only — it asserts every
// seed-declared tenant-owned (SourceObject, TargetObject, ActionType,
// SourceField) has an active, non-deleted row in DB. Values
// (TargetField / DefaultValue / SourceLookupField) are NOT compared —
// admin edits must survive. See verifyObjectFieldMappingsLanded.
func (p *Pack) VerifyInsertRulesLanded(ctx context.Context, db *sql.DB) error {
	if db == nil {
		return nil
	}
	specs, err := p.parseTenantObjectsSpec()
	if err != nil {
		return fmt.Errorf("closure: parse tenant objects spec: %w", err)
	}
	if err := p.verifyChildCloneRulesLanded(ctx, db, specs); err != nil {
		return err
	}
	if err := p.verifyObjectFieldMappingsLanded(ctx, db); err != nil {
		return err
	}
	return nil
}

// verifyChildCloneRulesLanded reads the L2 seed map and asserts that every
// declared apiName has the canonical rules plus ChildCloneRulesCustomized=1 in
// _System_Object.
func (p *Pack) verifyChildCloneRulesLanded(ctx context.Context, db *sql.DB, specs []packs.ObjectSpec) error {
	if err := verifyChildCloneRulesSeedLanded(ctx, db, tenantObjectSpecChildCloneRules(specs), ObjectsSpecFile); err != nil {
		return err
	}
	objectChildCloneRulesJSON, ok, err := p.readOptional(ObjectChildCloneRulesFile)
	if err != nil {
		return err
	}
	if !ok {
		return nil
	}
	seed := map[string][]models.ChildCloneRule{}
	if err := json.Unmarshal(objectChildCloneRulesJSON, &seed); err != nil {
		return fmt.Errorf("closure: parse object_child_clone_rules.json: %w", err)
	}
	if err := rejectCustomObjectChildCloneRulesFileObjects(specs, seed); err != nil {
		return err
	}
	return verifyChildCloneRulesSeedLanded(ctx, db, seed, ObjectChildCloneRulesFile)
}

func verifyChildCloneRulesSeedLanded(ctx context.Context, db *sql.DB, seed map[string][]models.ChildCloneRule, source string) error {
	for apiName, seedRules := range seed {
		objectID := domainschema.GenerateObjectID(apiName)
		query := fmt.Sprintf(
			"SELECT `%s`, COALESCE(`%s`, 0) FROM `%s` WHERE `%s` = ?",
			constants.FieldSysObject_ChildCloneRules,
			constants.FieldSysObject_ChildCloneRulesCustomized,
			constants.TableObject,
			constants.ColID,
		)
		var stored sql.NullString
		var customized int
		row := db.QueryRowContext(ctx, query, objectID)
		if err := row.Scan(&stored, &customized); err != nil {
			if err == sql.ErrNoRows {
				return fmt.Errorf(
					"ChildCloneRules closure failed: object %q (id=%s) has no row in %s -- "+
						"apply-tenant ordering is broken or the object was renamed",
					apiName, objectID, constants.TableObject,
				)
			}
			return fmt.Errorf("closure: query ChildCloneRules for %s: %w", apiName, err)
		}
		if customized == 0 {
			return fmt.Errorf(
				"ChildCloneRules closure failed for %s in %s: %s is 0, want 1 for L2-owned overlay payload",
				apiName, source, constants.FieldSysObject_ChildCloneRulesCustomized,
			)
		}

		want, err := canonicalChildCloneRules(seedRules)
		if err != nil {
			return fmt.Errorf("closure: canonicalize seed ChildCloneRules for %s in %s: %w", apiName, source, err)
		}
		got, err := canonicalChildCloneRulesJSON(stored)
		if err != nil {
			return fmt.Errorf("closure: canonicalize DB ChildCloneRules for %s: %w", apiName, err)
		}
		if !bytes.Equal(got, want) {
			return fmt.Errorf(
				"ChildCloneRules closure failed for %s in %s: DB=%s seed=%s. "+
					"Re-run apply-tenant or inspect _System_Object.ChildCloneRules for drift.",
				apiName, source, string(got), string(want),
			)
		}
	}
	return nil
}

func canonicalChildCloneRulesJSON(raw sql.NullString) ([]byte, error) {
	if !raw.Valid || strings.TrimSpace(raw.String) == "" {
		return canonicalChildCloneRules(nil)
	}
	var rules []models.ChildCloneRule
	if err := json.Unmarshal([]byte(raw.String), &rules); err != nil {
		return nil, err
	}
	return canonicalChildCloneRules(rules)
}

func canonicalChildCloneRules(rules []models.ChildCloneRule) ([]byte, error) {
	if len(rules) == 0 {
		return []byte("[]"), nil
	}
	type normalizedRule struct {
		ParentLookupField string            `json:"parentLookupField"`
		SourceObject      string            `json:"sourceObject"`
		SourceLookupField string            `json:"sourceLookupField"`
		SourceIdField     string            `json:"sourceIdField,omitempty"`
		HydrateSource     bool              `json:"hydrateSource,omitempty"`
		TargetObject      string            `json:"targetObject"`
		TargetLookupField string            `json:"targetLookupField"`
		FieldMappings     map[string]string `json:"fieldMappings"`
	}
	normalized := make([]normalizedRule, 0, len(rules))
	for _, rule := range rules {
		fieldMappings := map[string]string{}
		for key, value := range rule.FieldMappings {
			fieldMappings[key] = value
		}
		normalized = append(normalized, normalizedRule{
			ParentLookupField: rule.ParentLookupField,
			SourceObject:      rule.SourceObject,
			SourceLookupField: rule.SourceLookupField,
			SourceIdField:     rule.SourceIdField,
			HydrateSource:     rule.HydrateSource,
			TargetObject:      rule.TargetObject,
			TargetLookupField: rule.TargetLookupField,
			FieldMappings:     fieldMappings,
		})
	}
	sort.Slice(normalized, func(i, j int) bool {
		left, _ := json.Marshal(normalized[i])
		right, _ := json.Marshal(normalized[j])
		return bytes.Compare(left, right) < 0
	})
	return json.Marshal(normalized)
}

// verifyObjectFieldMappingsLanded asserts, for every row declared in
// the seed, that an authoritative row exists in DB with the matching
// BUSINESS UNIQUE KEY — (Source, SourceObject, TargetObject, ActionType,
// SourceField). Nothing else is compared.
//
// Why business-key existence, NOT value equality (round-7 reviewer
// finding #2 correction):
//
//  1. _System_ObjectFieldMapping is admin-editable metadata. An admin
//     can edit TargetField / DefaultValue / SourceLookupField and stamp
//     IsCustomized=1; the convergent upsert (field_mappings.go ON
//     DUPLICATE KEY UPDATE) deliberately preserves those edits. If
//     closure compared values, the admin override would trigger a
//     closure failure on the next apply-tenant — breaking the
//     admin-edits-are-first-class contract of
//     _System_ObjectFieldMapping.
//
//  2. The generated Id encodes Source owner, SourceField, and TargetField,
//     but the unique key is only (Source, SourceObject, TargetObject,
//     ActionType, SourceField). When the seed changes TargetField for an
//     existing owner-owned row, ON DUPLICATE KEY UPDATE collides on the
//     unique key and updates TargetField on the old primary-key row; the
//     row's Id does NOT change (primary keys are immutable). A closure that
//     looked up by seed-derived Id would miss the row and report a false
//     negative.
//
// Closure's job reduces to: "did every seed-declared tenant-owned
// business key land in DB as an active, non-deleted row?" That's
// sufficient to catch silent write-miss (RowsAffected==0 on the upsert)
// and owner-stamping regressions, and non-fragile under admin overrides.
//
// Soft-deleted (IsDeleted=1) and inactive (IsActive=0) rows are
// excluded — those are tombstones / disabled rows and do not count
// as "landed".
func (p *Pack) verifyObjectFieldMappingsLanded(ctx context.Context, db *sql.DB) error {
	objectFieldMappingsJSON, ok, err := p.readOptional(ObjectFieldMappingsFile)
	if err != nil {
		return err
	}
	if !ok {
		return nil
	}
	var groups []connector.FieldMappingGroup
	if err := json.Unmarshal(objectFieldMappingsJSON, &groups); err != nil {
		return fmt.Errorf("closure: parse object_field_mappings.json: %w", err)
	}

	const existenceQuery = `
		SELECT 1 FROM _System_ObjectFieldMapping
		WHERE SourceObject = ? AND TargetObject = ? AND ActionType = ?
		  AND SourceField = ? AND Source = ? AND IsActive = 1 AND IsDeleted = 0
		LIMIT 1`

	for _, g := range groups {
		for _, m := range g.Mappings {
			var found int
			err := db.QueryRowContext(ctx, existenceQuery,
				g.SourceObject, g.TargetObject, g.ActionType, m.SourceField, p.objectFieldMappingOwner(),
			).Scan(&found)
			if err == sql.ErrNoRows {
				return fmt.Errorf(
					"ObjectFieldMapping closure failed on %s→%s(%s) SourceField=%q: "+
						"no active, non-deleted tenant-owned row for this business unique key. "+
						"Either the convergent upsert failed silently on a previous apply, "+
						"or a rogue delete left the surface incomplete. Reapply required.",
					g.SourceObject, g.TargetObject, g.ActionType, m.SourceField,
				)
			}
			if err != nil {
				return fmt.Errorf("closure: query mapping %s→%s(%s) SourceField=%q: %w",
					g.SourceObject, g.TargetObject, g.ActionType, m.SourceField, err)
			}
		}
	}
	return nil
}