package persistence

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

	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/kernel/pkg/query"
	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/fieldcapabilities"
	"github.com/nexuscrm/shared/pkg/models"
)

// FieldWithContext removed (moved to domain/schema)

// SaveObjectMetadata upserts object metadata into _System_Object
func (r *SchemaRepository) SaveObjectMetadata(ctx context.Context, obj *models.ObjectMetadata, tx interface{}) error {
	candidates, err := prepareCatalogObjectIdentityCandidates([]*models.ObjectMetadata{obj})
	if err != nil {
		return err
	}
	const writerID = "nexuscrm-persistence/schema_metadata_persist.go#SchemaRepository.SaveObjectMetadata"
	if r.permissionStateDirectWriterShouldRun(ctx, writerID) {
		if err := r.requirePermissionStateOwnedExecutor(ctx, tx); err != nil {
			return err
		}
		return r.runPermissionStateDirectWriter(
			ctx,
			writerID,
			[]permissionStateDirectTarget{
				permissionStateDirectTargetFor(
					constants.TableObject,
					permissionStateDirectCaseInsensitiveEqual(constants.FieldAPIName),
					obj.APIName,
				),
			},
			func(txCtx context.Context) error {
				return r.SaveObjectMetadata(
					withPermissionStateDirectWriter(txCtx, writerID),
					obj,
					r.permissionStateOwnedTransaction(txCtx),
				)
			},
		)
	}

	if tx == nil {
		ownedTx, err := r.db.BeginTx(ctx, nil)
		if err != nil {
			return apperrors.NewInternalError("failed to begin object metadata identity transaction", err)
		}
		defer func() {
			_ = ownedTx.Rollback()
		}()
		if err := r.SaveObjectMetadata(ctx, obj, ownedTx); err != nil {
			return err
		}
		if err := ownedTx.Commit(); err != nil {
			return apperrors.NewInternalError("failed to commit object metadata identity transaction", err)
		}
		return nil
	}

	transaction, err := requireCatalogObjectIdentityTransaction(tx)
	if err != nil {
		return err
	}
	var exec Executor = transaction

	if err := assertCatalogObjectIdentityMappings(ctx, exec, candidates, true); err != nil {
		return err
	}
	claimValues, err := r.prepareCatalogObjectIdentityClaimValues(
		candidates,
		[]*models.ObjectMetadata{obj},
	)
	if err != nil {
		return err
	}
	claim := query.Insert(constants.TableObject, nil).
		ValuesBatch(claimValues).
		OnDuplicateKeyUpdate(map[string]interface{}{
			constants.FieldID: query.Raw("`" + constants.FieldID + "`"),
		}).
		Build()
	if _, err := exec.ExecContext(ctx, claim.SQL, claim.Params...); err != nil {
		return apperrors.NewInternalError("failed to claim Catalog object identity mappings", err)
	}
	if err := requireCatalogObjectIdentityMappings(ctx, exec, candidates); err != nil {
		return err
	}
	if err := r.ensureObjectKeyPrefix(ctx, exec, obj); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to assign key prefix for %s", obj.APIName), err)
	}
	if err := r.preservePackOwnedAccessPolicySidecarsForObjectMetadata(ctx, exec, obj); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to preserve AccessPolicy sidecars for %s", obj.APIName), err)
	}

	values, err := r.prepareObjectDBMap(obj)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to prepare object values for %s", obj.APIName), err)
	}

	// Prepare the mutable update map. ID and APIName are the immutable
	// Catalog identity established by the claim above.
	updates := make(map[string]interface{})
	for k, v := range values {
		if k != constants.FieldID &&
			k != constants.FieldCreatedDate &&
			k != constants.FieldSysObject_APIName {
			updates[k] = v
		}
	}
	updates[constants.FieldLastModifiedDate] = time.Now()

	qr := query.Insert(constants.TableObject, values).
		OnDuplicateKeyUpdate(updates).
		Build()

	_, err = exec.ExecContext(ctx, qr.SQL, qr.Params...)
	return err
}

// InsertObjectMetadata inserts object metadata into _System_Object (Strict - Fails on Unique Constraint)
func (r *SchemaRepository) InsertObjectMetadata(ctx context.Context, obj *models.ObjectMetadata, tx interface{}) error {
	candidates, err := prepareCatalogObjectIdentityCandidates([]*models.ObjectMetadata{obj})
	if err != nil {
		return err
	}
	const writerID = "nexuscrm-persistence/schema_metadata_persist.go#SchemaRepository.InsertObjectMetadata"
	if r.permissionStateDirectWriterShouldRun(ctx, writerID) {
		if err := r.requirePermissionStateOwnedExecutor(ctx, tx); err != nil {
			return err
		}
		return r.runPermissionStateDirectWriter(
			ctx,
			writerID,
			[]permissionStateDirectTarget{
				permissionStateDirectTargetFor(
					constants.TableObject,
					permissionStateDirectCaseInsensitiveEqual(constants.FieldAPIName),
					obj.APIName,
				),
			},
			func(txCtx context.Context) error {
				return r.InsertObjectMetadata(
					withPermissionStateDirectWriter(txCtx, writerID),
					obj,
					r.permissionStateOwnedTransaction(txCtx),
				)
			},
		)
	}

	if tx == nil {
		ownedTx, err := r.db.BeginTx(ctx, nil)
		if err != nil {
			return apperrors.NewInternalError("failed to begin strict object identity transaction", err)
		}
		defer func() {
			_ = ownedTx.Rollback()
		}()
		if err := r.InsertObjectMetadata(ctx, obj, ownedTx); err != nil {
			return err
		}
		if err := ownedTx.Commit(); err != nil {
			return apperrors.NewInternalError("failed to commit strict object identity transaction", err)
		}
		return nil
	}

	transaction, err := requireCatalogObjectIdentityTransaction(tx)
	if err != nil {
		return err
	}
	var exec Executor = transaction

	if err := assertCatalogObjectIdentityMappings(ctx, exec, candidates, false); err != nil {
		return err
	}
	if err := r.ensureObjectKeyPrefix(ctx, exec, obj); err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to assign key prefix for %s", obj.APIName), err)
	}

	values, err := r.prepareObjectDBMap(obj)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to prepare object values for %s", obj.APIName), err)
	}

	qr := query.Insert(constants.TableObject, values).Build()

	_, err = exec.ExecContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		if IsDuplicateKeyError(err) {
			return apperrors.NewConflictError("Object Metadata", "api_name", obj.APIName)
		}
		return err
	}
	return requireCatalogObjectIdentityMappings(ctx, exec, candidates)
}

// SaveFieldMetadataWithIDs upserts field metadata with explicit IDs.
// objectAPIName is needed to compute object-specific intrinsic capabilities.
func (r *SchemaRepository) SaveFieldMetadataWithIDs(ctx context.Context, field *models.FieldMetadata, objectID string, objectAPIName string, fieldID string, tx interface{}) error {
	// Admin/DDL paths do not carry an explicit capability declaration, so use
	// platform-universal intrinsic defaults. Pack batch writes use
	// CapabilitiesDeclared to persist app-pack authored capabilities.
	fieldCopy := *field
	field = &fieldCopy
	if err := validateCatalogWriteProtectionProjection(objectAPIName, field); err != nil {
		return err
	}
	if field.GeneratedForPolymorphicReference != "" {
		field.Createable, field.Updateable = fieldcapabilities.ApplyIntrinsicBounds(objectAPIName, field, false, false)
	} else {
		field.Createable, field.Updateable = fieldcapabilities.ComputeIntrinsic(objectAPIName, field)
	}
	field.OwnerKind = constants.FieldOwnerKindAdmin
	field.OwnerPack = nil
	var exec Executor
	if tx != nil {
		if e, ok := tx.(Executor); ok {
			exec = e
		}
	}
	if exec == nil {
		exec = r.db
	}

	if err := r.admitAdminFieldMetadataWriteByObjectID(ctx, exec, objectID, objectAPIName, field.APIName); err != nil {
		return err
	}

	if field.TrackHistory {
		var existing int
		err := exec.QueryRowContext(ctx, "SELECT 1 FROM _System_Field WHERE Id = ?", fieldID).Scan(&existing)
		if errors.Is(err, sql.ErrNoRows) {
			return apperrors.NewValidationError(constants.FieldSysField_TrackHistory, "new fields must be created with trackHistory disabled; enable it in a separate committed transition")
		}
		if err != nil {
			return apperrors.NewInternalError("check field history tracking create state", err)
		}
	}
	values, err := r.prepareFieldDBMap(field, objectID, fieldID)
	if err != nil {
		return apperrors.NewInternalError(fmt.Sprintf("failed to prepare field values for %s", field.APIName), err)
	}
	// Generic upsert may insert only the disabled initial state. Existing-row
	// TrackHistory is owned exclusively by ApplyHistoryTrackingTransition.
	values[constants.FieldSysField_TrackHistory] = false

	// Prepare Update Map — only include fields that should change on upsert.
	// Skip Label if it is empty to avoid overwriting existing labels with blanks.
	updates := make(map[string]interface{})
	for k, v := range values {
		if k == constants.FieldID || k == constants.FieldCreatedDate || k == constants.FieldObjectID || k == constants.FieldSysField_TrackHistory {
			continue
		}
		if k == constants.FieldSysField_Label {
			if s, ok := v.(string); ok && s == "" {
				continue
			}
		}
		updates[k] = v
	}
	// Preserve a catalog-owned declaration atomically even if a malformed
	// caller reaches this lower-level upsert with a stale field projection.
	updates[constants.FieldSysField_WriteProtectionPolicy] = preserveAuthorityOnlyPolicyUpdate(false)
	updates[constants.FieldSysField_Createable] = authorityOnlyCapabilityUpdate(constants.FieldSysField_Createable)
	updates[constants.FieldSysField_Updateable] = authorityOnlyCapabilityUpdate(constants.FieldSysField_Updateable)
	updates[constants.FieldLastModifiedDate] = time.Now()
	// Mark as user-customized so bootstrap label reconciliation won't overwrite.
	// This path is called from admin API / field editor, not from bootstrap.
	updates["IsCustomized"] = 1
	updates[constants.FieldSysField_OwnerKind] = query.Raw(fmt.Sprintf(
		"IF(COALESCE(`%s`, '') = '' OR `%s` = '%s', VALUES(`%s`), `%s`)",
		constants.FieldSysField_OwnerKind,
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindRuntime,
		constants.FieldSysField_OwnerKind,
		constants.FieldSysField_OwnerKind,
	))
	updates[constants.FieldSysField_OwnerPack] = query.Raw(fmt.Sprintf(
		"IF(VALUES(`%s`) = '%s' AND COALESCE(`%s`, '') NOT IN ('%s', '%s', '%s'), VALUES(`%s`), `%s`)",
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindAdmin,
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindPack,
		constants.FieldOwnerKindPackRetained,
		constants.FieldOwnerKindPackRemoving,
		constants.FieldSysField_OwnerPack,
		constants.FieldSysField_OwnerPack,
	))

	qr := query.Insert(constants.TableField, values).
		OnDuplicateKeyUpdate(updates).
		Build()

	_, err = exec.ExecContext(ctx, qr.SQL, qr.Params...)
	return err
}

// BatchSaveObjectMetadata inserts multiple objects in a single statement
func (r *SchemaRepository) BatchSaveObjectMetadata(ctx context.Context, objs []*models.ObjectMetadata, tx interface{}) error {
	return r.batchSaveObjectMetadata(ctx, objs, tx, true)
}

func (r *SchemaRepository) BatchSaveObjectMetadataWithCustomizationPolicy(ctx context.Context, objs []*models.ObjectMetadata, tx interface{}, preserveCustomized bool) error {
	return r.batchSaveObjectMetadata(ctx, objs, tx, preserveCustomized)
}

func (r *SchemaRepository) BatchSaveObjectMetadataWithoutChildCloneRules(ctx context.Context, objs []*models.ObjectMetadata, tx interface{}) error {
	return r.batchSaveObjectMetadata(ctx, objs, tx, true,
		constants.FieldSysObject_ChildCloneRules,
		constants.FieldSysObject_ChildCloneRulesCustomized,
		constants.FieldSysObject_ExclusiveSiblingFields,
		constants.FieldSysObject_ExclusiveSiblingFieldsCustomized,
	)
}

func (r *SchemaRepository) batchSaveObjectMetadata(ctx context.Context, objs []*models.ObjectMetadata, tx interface{}, preserveCustomized bool, skipUpdateFields ...string) error {
	if len(objs) == 0 {
		return nil
	}
	candidates, err := prepareCatalogObjectIdentityCandidates(objs)
	if err != nil {
		return err
	}
	const writerID = "nexuscrm-persistence/schema_metadata_persist.go#SchemaRepository.batchSaveObjectMetadata"
	if r.permissionStateDirectWriterShouldRun(ctx, writerID) {
		if err := r.requirePermissionStateOwnedExecutor(ctx, tx); err != nil {
			return err
		}
		args := make([]interface{}, 0, len(objs))
		for _, obj := range objs {
			args = append(args, obj.APIName)
		}
		return r.runPermissionStateDirectWriter(
			ctx,
			writerID,
			[]permissionStateDirectTarget{
				permissionStateDirectTargetFor(
					constants.TableObject,
					permissionStateDirectIn(constants.FieldAPIName, len(args)),
					args...,
				),
			},
			func(txCtx context.Context) error {
				return r.batchSaveObjectMetadata(
					withPermissionStateDirectWriter(txCtx, writerID),
					objs,
					r.permissionStateOwnedTransaction(txCtx),
					preserveCustomized,
					skipUpdateFields...,
				)
			},
		)
	}

	if tx == nil {
		ownedTx, err := r.db.BeginTx(ctx, nil)
		if err != nil {
			return apperrors.NewInternalError("failed to begin batch object identity transaction", err)
		}
		defer func() {
			_ = ownedTx.Rollback()
		}()
		if err := r.batchSaveObjectMetadata(ctx, objs, ownedTx, preserveCustomized, skipUpdateFields...); err != nil {
			return err
		}
		if err := ownedTx.Commit(); err != nil {
			return apperrors.NewInternalError("failed to commit batch object identity transaction", err)
		}
		return nil
	}

	transaction, err := requireCatalogObjectIdentityTransaction(tx)
	if err != nil {
		return err
	}
	var exec Executor = transaction
	if err := assertCatalogObjectIdentityMappings(ctx, exec, candidates, true); err != nil {
		return err
	}
	claimValues, err := r.prepareCatalogObjectIdentityClaimValues(candidates, objs)
	if err != nil {
		return err
	}
	claim := query.Insert(constants.TableObject, nil).
		ValuesBatch(claimValues).
		OnDuplicateKeyUpdate(map[string]interface{}{
			constants.FieldID: query.Raw("`" + constants.FieldID + "`"),
		}).
		Build()
	if _, err := exec.ExecContext(ctx, claim.SQL, claim.Params...); err != nil {
		return apperrors.NewInternalError("failed to claim Catalog object identity mappings", err)
	}
	if err := requireCatalogObjectIdentityMappings(ctx, exec, candidates); err != nil {
		return err
	}

	// Build map list for Batch Insert
	var batchValues []map[string]interface{}

	for _, obj := range objs {
		if err := r.ensureObjectKeyPrefix(ctx, exec, obj); err != nil {
			return err
		}
		if err := r.preservePackOwnedAccessPolicySidecarsForObjectMetadata(ctx, exec, obj); err != nil {
			return fmt.Errorf("preserve AccessPolicy sidecars for %s: %w", obj.APIName, err)
		}

		values, err := r.prepareObjectDBMap(obj)
		if err != nil {
			return err
		}
		batchValues = append(batchValues, values)
	}

	// Prepare Update Map (Generic fields to update on duplicate)
	// For Batch Upsert, the builder needs to know which columns to update.
	// We map the Logical Fields we want to update.
	// The values here are dummy values or raw SQL "VALUES(col)".
	// Use query.Raw to specify VALUES(physical_name) for ON DUPLICATE KEY UPDATE.

	updates := make(map[string]interface{})
	updatableFields := batchObjectMetadataUpdatableFields(skipUpdateFields...)

	r.configureObjectBatchUpdatesWithCustomizationPolicy(updates, updatableFields, preserveCustomized)

	qr := query.Insert(constants.TableObject, nil).
		ValuesBatch(batchValues).
		OnDuplicateKeyUpdate(updates).
		Build()

	_, err = exec.ExecContext(ctx, qr.SQL, qr.Params...)
	return err
}

func batchObjectMetadataUpdatableFields(skip ...string) []string {
	fields := []string{
		constants.FieldSysObject_Label, constants.FieldSysObject_PluralLabel,
		constants.FieldSysObject_KeyPrefix,
		constants.FieldSysObject_Icon, constants.FieldSysObject_Description,
		constants.FieldSysObject_IsCustom, constants.FieldSysObject_Searchable,
		constants.FieldSysObject_CustomSetting,
		constants.FieldSysObject_AppID,
		constants.FieldSysObject_ListFields, constants.FieldSysObject_ListFieldsCustomized, constants.FieldSysObject_DefaultChildRows, constants.FieldSysObject_DefaultChildRowsCustomized, constants.FieldSysObject_ChildCloneRules, constants.FieldSysObject_ChildCloneRulesCustomized, constants.FieldSysObject_ExclusiveSiblingFields, constants.FieldSysObject_ExclusiveSiblingFieldsCustomized, constants.FieldSysObject_PathField, constants.FieldSysObject_DisplayNameField, constants.FieldSysObject_LockRule, constants.FieldSysObject_LockRuleCustomized,
		constants.FieldSysObject_ReadSource,
		constants.FieldSysObject_ThemeColor, constants.FieldSysObject_TableType,
		constants.FieldSysObject_SharingModel,
		constants.FieldSysObject_EnableHierarchySharing,
		constants.FieldSysObject_RequiredWritePermission,
		constants.FieldSysObject_RequiredReadPermission,
		constants.FieldSysObject_AccessPolicy,
		constants.FieldSysObject_DeleteMode,
		constants.FieldLastModifiedDate,
	}
	if len(skip) == 0 {
		return fields
	}
	skipped := make(map[string]struct{}, len(skip))
	for _, field := range skip {
		if field != "" {
			skipped[field] = struct{}{}
		}
	}
	out := make([]string, 0, len(fields))
	for _, field := range fields {
		if _, ok := skipped[field]; ok {
			continue
		}
		out = append(out, field)
	}
	return out
}

func (r *SchemaRepository) ensureObjectKeyPrefix(ctx context.Context, exec Executor, obj *models.ObjectMetadata) error {
	if obj == nil || !obj.IsCustom || obj.KeyPrefix != nil && *obj.KeyPrefix != "" {
		return nil
	}
	rows, err := exec.QueryContext(ctx, `
		SELECT KeyPrefix
		FROM _System_Object
		WHERE IsDeleted = 0 AND KeyPrefix IS NOT NULL AND KeyPrefix <> ''`)
	if err != nil {
		return err
	}
	defer rows.Close()

	used := map[string]bool{}
	for rows.Next() {
		var prefix string
		if err := rows.Scan(&prefix); err != nil {
			return err
		}
		used[prefix] = true
	}
	if err := rows.Err(); err != nil {
		return err
	}

	next := 0
	prefix := nextAvailableCustomPrefix(used, &next)
	obj.KeyPrefix = &prefix
	return nil
}

func (r *SchemaRepository) configureObjectBatchUpdatesWithCustomizationPolicy(updates map[string]interface{}, fields []string, preserveCustomized bool) {
	guarded := map[string]struct{}{}
	if preserveCustomized {
		guarded = objectBatchCustomizedGuardFields()
	}
	for _, field := range fields {
		physical := field
		if field == constants.FieldLastModifiedDate {
			if preserveCustomized {
				updates[field] = query.Raw(fmt.Sprintf(
					"IF(`%s` = 0 OR `%s` IS NULL OR `%s` = 0 OR `%s` IS NULL, NOW(), `%s`)",
					constants.FieldSysObject_IsCustomized,
					constants.FieldSysObject_IsCustomized,
					constants.FieldSysObject_ListFieldsCustomized,
					constants.FieldSysObject_ListFieldsCustomized,
					physical,
				))
			} else {
				updates[field] = query.Raw("NOW()")
			}
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_ListFields {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_ListFieldsCustomized,
				constants.FieldSysObject_ListFieldsCustomized,
				physical,
				physical,
			))
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_ListFieldsCustomized {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_ListFieldsCustomized,
				constants.FieldSysObject_ListFieldsCustomized,
				physical,
				physical,
			))
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_LockRule {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_LockRuleCustomized,
				constants.FieldSysObject_LockRuleCustomized,
				physical,
				physical,
			))
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_LockRuleCustomized {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_LockRuleCustomized,
				constants.FieldSysObject_LockRuleCustomized,
				physical,
				physical,
			))
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_DefaultChildRows {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_DefaultChildRowsCustomized,
				constants.FieldSysObject_DefaultChildRowsCustomized,
				physical,
				physical,
			))
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_DefaultChildRowsCustomized {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_DefaultChildRowsCustomized,
				constants.FieldSysObject_DefaultChildRowsCustomized,
				physical,
				physical,
			))
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_ChildCloneRules {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_ChildCloneRulesCustomized,
				constants.FieldSysObject_ChildCloneRulesCustomized,
				physical,
				physical,
			))
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_ChildCloneRulesCustomized {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_ChildCloneRulesCustomized,
				constants.FieldSysObject_ChildCloneRulesCustomized,
				physical,
				physical,
			))
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_ExclusiveSiblingFields {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_ExclusiveSiblingFieldsCustomized,
				constants.FieldSysObject_ExclusiveSiblingFieldsCustomized,
				physical,
				physical,
			))
			continue
		}
		if preserveCustomized && field == constants.FieldSysObject_ExclusiveSiblingFieldsCustomized {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_ExclusiveSiblingFieldsCustomized,
				constants.FieldSysObject_ExclusiveSiblingFieldsCustomized,
				physical,
				physical,
			))
			continue
		}
		if _, ok := guarded[field]; ok {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.FieldSysObject_IsCustomized,
				constants.FieldSysObject_IsCustomized,
				physical,
				physical,
			))
			continue
		}
		updates[field] = query.Raw(fmt.Sprintf("VALUES(`%s`)", physical))
	}
}

func objectBatchCustomizedGuardFields() map[string]struct{} {
	// Object customization is currently row-scoped. Once an admin update marks
	// the row customized, batch seed re-apply must not become a second author
	// for mutable object metadata. Shape/classification columns that heal
	// physical storage, such as IsCustom and TableType, stay unguarded.
	//
	// LockRule, ChildCloneRules, and ExclusiveSiblingFields are intentionally absent from this
	// row-scoped set: they use column-scoped customization bits so that an
	// unrelated admin edit (label, icon, etc.) does not suppress pack re-apply
	// of executable metadata facts.
	//
	return map[string]struct{}{
		constants.FieldSysObject_Label:            {},
		constants.FieldSysObject_PluralLabel:      {},
		constants.FieldSysObject_Icon:             {},
		constants.FieldSysObject_Description:      {},
		constants.FieldSysObject_Searchable:       {},
		constants.FieldSysObject_AppID:            {},
		constants.FieldSysObject_ListFields:       {},
		constants.FieldSysObject_PathField:        {},
		constants.FieldSysObject_DisplayNameField: {},
		// ReadSource is a behavior-affecting pack-authored object fact. It stays
		// row-scoped with IsCustomized so defaults_only admin object edits take
		// ownership of the whole mutable object metadata surface consistently.
		constants.FieldSysObject_ReadSource:             {},
		constants.FieldSysObject_ThemeColor:             {},
		constants.FieldSysObject_SharingModel:           {},
		constants.FieldSysObject_EnableHierarchySharing: {},
		// LockRule is excluded: it uses the column-scoped LockRuleCustomized bit
		// (not the row-scoped IsCustomized guard) so that an unrelated admin edit
		// (label, icon, etc.) does not suppress pack re-apply of the LockRule fact.
		constants.FieldSysObject_RequiredWritePermission: {},
		constants.FieldSysObject_RequiredReadPermission:  {},
		constants.FieldSysObject_AccessPolicy:            {},
		constants.FieldSysObject_DeleteMode:              {},
	}
}

// configureBatchUpdatesGuarded behaves like configureBatchUpdates but wraps each
// column listed in `guarded` so the ON DUPLICATE KEY UPDATE only writes the new
// value when the existing row is not admin-customized (IsCustomized=0 or NULL).
// Customized rows keep their current value, and VerifyStandardFieldSpecDrift
// surfaces the resulting spec-vs-live divergence for operator review.
func (r *SchemaRepository) configureBatchUpdatesGuarded(updates map[string]interface{}, fields, guarded []string) {
	guardedSet := make(map[string]struct{}, len(guarded))
	for _, g := range guarded {
		guardedSet[g] = struct{}{}
	}
	for _, field := range fields {
		physical := field
		if field == constants.FieldLastModifiedDate {
			updates[field] = query.Raw("NOW()")
			continue
		}
		if field == constants.FieldSysField_WriteProtectionPolicy {
			_, preserveCustomized := guardedSet[field]
			updates[field] = preserveAuthorityOnlyPolicyUpdate(preserveCustomized)
			continue
		}
		if field == constants.FieldSysField_Createable || field == constants.FieldSysField_Updateable {
			updates[field] = authorityOnlyCapabilityUpdate(field)
			continue
		}
		if _, ok := guardedSet[field]; ok {
			updates[field] = query.Raw(fmt.Sprintf(
				"IF(`%s` = 0 OR `%s` IS NULL, VALUES(`%s`), `%s`)",
				constants.ColSysField_IsCustomized,
				constants.ColSysField_IsCustomized,
				physical, physical,
			))
			continue
		}
		updates[field] = query.Raw(fmt.Sprintf("VALUES(`%s`)", physical))
	}
}

// BatchSaveFieldMetadata inserts multiple fields in a single statement
func (r *SchemaRepository) BatchSaveFieldMetadata(ctx context.Context, fields []schema.FieldWithContext, tx interface{}) error {
	return r.batchSaveFieldMetadata(ctx, fields, tx, true, false)
}

func (r *SchemaRepository) BatchSaveFieldMetadataWithCustomizationPolicy(ctx context.Context, fields []schema.FieldWithContext, tx interface{}, preserveCustomized bool) error {
	return r.batchSaveFieldMetadata(ctx, fields, tx, preserveCustomized, true)
}

type historyTrackingBatchCandidate struct {
	fieldID              string
	desired              bool
	existed              bool
	needsTransition      bool
	expectedTrackHistory bool
	expectedRevision     int64
	expectedCustomized   bool
}

const historyTrackingStatePrefetchBatchSize = 50

func (r *SchemaRepository) batchSaveFieldMetadata(ctx context.Context, fields []schema.FieldWithContext, tx interface{}, preserveCustomized bool, applyHistoryTransitions bool) error {
	if len(fields) == 0 {
		return nil
	}
	for _, fc := range fields {
		if err := r.validateCatalogFieldContextWriteProtectionProjection(fc); err != nil {
			return err
		}
	}
	var exec Executor
	if tx != nil {
		if e, ok := tx.(Executor); ok {
			exec = e
		}
	}
	if exec == nil {
		exec = r.db
	}
	if applyHistoryTransitions && tx == nil {
		return apperrors.NewInternalError("history tracking batch reapply requires an active transaction executor", nil)
	}
	if applyHistoryTransitions {
		if _, rootDB := tx.(*sql.DB); rootDB {
			return apperrors.NewInternalError("history tracking batch reapply rejects the root database executor", nil)
		}
		// The sorted candidate loop below is itself the lock-order owner when a
		// schema caller supplies a raw active transaction instead of the generic
		// transaction manager. Preserve an outer coordinator when one exists.
		ctx = ports.EnsureHistorySerializationAttemptState(ctx)
	}

	candidates := make([]historyTrackingBatchCandidate, 0, len(fields))
	for _, fc := range fields {
		fieldID := fc.FieldID
		if fieldID == "" {
			fieldID = schema.GenerateFieldID(fc.TableName, fc.ColumnDefinition.Name)
		}
		desired := fc.Field != nil && fc.Field.TrackHistory
		candidate := historyTrackingBatchCandidate{fieldID: fieldID, desired: desired}
		if !applyHistoryTransitions && desired {
			var marker int
			err := exec.QueryRowContext(ctx, "SELECT 1 FROM _System_Field WHERE Id = ?", fieldID).Scan(&marker)
			switch {
			case err == nil:
				candidate.existed = true
			case errors.Is(err, sql.ErrNoRows):
				if desired {
					return apperrors.NewValidationError(constants.FieldSysField_TrackHistory, "new fields must be created with trackHistory disabled; enable it in a separate committed transition")
				}
			default:
				return apperrors.NewInternalError("check batch field history tracking create state", err)
			}
		}
		candidates = append(candidates, candidate)
	}
	sort.Slice(candidates, func(i, j int) bool { return candidates[i].fieldID < candidates[j].fieldID })
	if applyHistoryTransitions {
		for i := 0; i < len(candidates); i += historyTrackingStatePrefetchBatchSize {
			end := i + historyTrackingStatePrefetchBatchSize
			if end > len(candidates) {
				end = len(candidates)
			}
			fieldIDs := make([]string, 0, end-i)
			for _, candidate := range candidates[i:end] {
				fieldIDs = append(fieldIDs, candidate.fieldID)
			}
			states, err := loadHistoryTrackingStates(ctx, exec, fieldIDs)
			if err != nil {
				return err
			}
			for candidateIndex := i; candidateIndex < end; candidateIndex++ {
				candidate := &candidates[candidateIndex]
				state, exists := states[candidate.fieldID]
				if !exists {
					if candidate.desired {
						return apperrors.NewValidationError(constants.FieldSysField_TrackHistory, "new fields must be created with trackHistory disabled; enable it in a separate committed transition")
					}
					continue
				}
				candidate.existed = true
				candidate.expectedTrackHistory = state.trackHistory
				candidate.expectedRevision = state.revision
				candidate.expectedCustomized = state.customized
				if state.deleted {
					return apperrors.NewValidationError(constants.FieldSysField_TrackHistory, "cannot change history tracking for a deleted field")
				}
				if err := validateHistoryTrackingState(state); err != nil {
					return err
				}
				if preserveCustomized && state.customized {
					continue
				}
				legacyInitialization := state.trackHistory && state.intervals == 0 && candidate.desired
				candidate.needsTransition = legacyInitialization || state.trackHistory != candidate.desired
			}
		}
	}

	// Batch callers like init-l0 may register >1k fields in one pass. Preload
	// live column specs once so the gate keeps the same safety semantics
	// without issuing per-field INFORMATION_SCHEMA round-trips.
	tableNames := make([]string, 0, len(fields))
	for _, fc := range fields {
		tableNames = append(tableNames, fc.TableName)
	}
	liveSpecs, err := r.loadPhysicalColumnSpecsForTables(tableNames)
	if err != nil {
		return apperrors.NewInternalError("batch field gate: preload physical column specs", err)
	}

	for _, fc := range fields {
		spec, exists := liveSpecs.get(fc.TableName, fc.ColumnDefinition.Name)
		if err := r.gatePhysicalShapeBeforeMetadataWithPrefetchedSpec(fc.TableName, fc.ColumnDefinition, spec, exists); err != nil {
			return err
		}
	}

	if applyHistoryTransitions {
		// Ledger mutations are owned by ApplyHistoryTrackingTransition, which
		// changes the field revision in the same transaction. Claiming the exact
		// prefetched field state after the ledger read therefore detects an
		// already-stale no-op and makes any later concurrent transition conflict
		// at commit, including for optimistic transactions.
		claims := make([]historyTrackingNoOpClaim, 0, len(candidates))
		for _, candidate := range candidates {
			if !candidate.existed || candidate.needsTransition {
				continue
			}
			claims = append(claims, historyTrackingNoOpClaim{
				fieldID: candidate.fieldID, tracked: candidate.expectedTrackHistory,
				revision: candidate.expectedRevision, customized: candidate.expectedCustomized,
			})
		}
		for i := 0; i < len(claims); i += historyTrackingStatePrefetchBatchSize {
			end := i + historyTrackingStatePrefetchBatchSize
			if end > len(claims) {
				end = len(claims)
			}
			if err := claimHistoryTrackingNoOps(ctx, exec, claims[i:end]); err != nil {
				return err
			}
		}
	}

	// Process in batches of 50
	batchSize := 50
	for i := 0; i < len(fields); i += batchSize {
		end := i + batchSize
		if end > len(fields) {
			end = len(fields)
		}
		batch := fields[i:end]

		var batchValues []map[string]interface{}
		for _, fc := range batch {
			// Ensure IDs are set
			objectID := fc.ObjectID
			if objectID == "" {
				return apperrors.NewValidationError("ObjectID", fmt.Sprintf("ObjectID is required for field %s.%s — caller must resolve from _System_Object", fc.TableName, fc.ColumnDefinition.Name))
			}
			fieldID := fc.FieldID
			if fieldID == "" {
				fieldID = schema.GenerateFieldID(fc.TableName, fc.ColumnDefinition.Name)
			}

			var fieldMetadata *models.FieldMetadata

			if fc.Field != nil {
				// Use provided field metadata (preserves Logical APIName even if PhysicalName is hashed)
				fieldMetadata = fc.Field
			} else {
				fieldMetadata = r.buildFieldMetadataFromColumn(fc.ColumnDefinition)
			}
			if fc.CapabilitiesDeclared {
				fieldMetadata.Createable, fieldMetadata.Updateable = fieldcapabilities.ApplyIntrinsicBounds(
					fc.TableName,
					fieldMetadata,
					fieldMetadata.Createable,
					fieldMetadata.Updateable,
				)
			} else {
				fieldMetadata.Createable, fieldMetadata.Updateable = fieldcapabilities.ComputeIntrinsic(fc.TableName, fieldMetadata)
			}
			fieldCopy := *fieldMetadata
			fieldMetadata = &fieldCopy
			applyBatchFieldOwnership(fieldMetadata, fc)

			values, err := r.prepareFieldDBMap(fieldMetadata, objectID, fieldID)
			if err != nil {
				return err
			}
			// Existing-row TrackHistory never participates in generic upsert.
			// New rows are admitted only in the disabled initial state.
			values[constants.FieldSysField_TrackHistory] = false
			batchValues = append(batchValues, values)
		}

		updates := make(map[string]interface{})
		// NOTE: Label is intentionally excluded from updatableFields to prevent
		// overwriting user-customized labels during schema reconciliation (#355).
		// Labels are only set on INSERT (new fields), never overwritten on UPDATE.
		updatableFields := batchFieldMetadataUpdatableFields()
		// Spec-controlled columns that admins can override via the field-edit UI
		// (which writes through SaveFieldMetadataWithIDs and sets IsCustomized=1).
		// Intrinsic/server-managed columns (IsSystem, Createable, Updateable,
		// PhysicalName) still auto-heal from schema shape; admin-owned metadata
		// columns stay untouched on pack re-apply.
		var guardedFields []string
		if preserveCustomized {
			guardedFields = batchFieldMetadataGuardedFields()
		}
		r.configureBatchUpdatesGuarded(updates, updatableFields, guardedFields)
		r.configureBatchFieldOwnershipUpdates(updates)

		qr := query.Insert(constants.TableField, nil).
			ValuesBatch(batchValues).
			OnDuplicateKeyUpdate(updates).
			Build()

		if _, err := exec.ExecContext(ctx, qr.SQL, qr.Params...); err != nil {
			return apperrors.NewInternalError("batch field insert failed", err)
		}
	}

	if applyHistoryTransitions {
		for _, candidate := range candidates {
			if !candidate.existed || !candidate.needsTransition {
				continue
			}
			if _, err := r.ApplyHistoryTrackingTransition(ctx, candidate.fieldID, candidate.desired, tx, preserveCustomized, false); err != nil {
				return err
			}
		}
	}

	return nil
}

func applyBatchFieldOwnership(field *models.FieldMetadata, fc schema.FieldWithContext) {
	ownerKind := strings.TrimSpace(fc.OwnerKind)
	if ownerKind == "" {
		ownerKind = strings.TrimSpace(field.OwnerKind)
	}
	if ownerKind == "" {
		ownerKind = constants.FieldOwnerKindRuntime
	}
	ownerPack := strings.TrimSpace(fc.OwnerPack)
	if ownerPack == "" && field.OwnerPack != nil {
		ownerPack = strings.TrimSpace(*field.OwnerPack)
	}
	if ownerKind != constants.FieldOwnerKindPack {
		ownerPack = ""
	}
	field.OwnerKind = ownerKind
	if ownerPack == "" {
		field.OwnerPack = nil
		return
	}
	field.OwnerPack = &ownerPack
}

func (r *SchemaRepository) configureBatchFieldOwnershipUpdates(updates map[string]interface{}) {
	updates[constants.FieldSysField_OwnerKind] = query.Raw(fmt.Sprintf(
		"IF(COALESCE(`%s`, '') = '' OR `%s` = '%s', VALUES(`%s`), `%s`)",
		constants.FieldSysField_OwnerKind,
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindRuntime,
		constants.FieldSysField_OwnerKind,
		constants.FieldSysField_OwnerKind,
	))
	updates[constants.FieldSysField_OwnerPack] = query.Raw(fmt.Sprintf(
		"IF(VALUES(`%s`) = '%s' AND COALESCE(`%s`, '') <> '%s', VALUES(`%s`), IF(VALUES(`%s`) = '%s' AND COALESCE(`%s`, '') <> '%s' AND (COALESCE(`%s`, '') IN ('', '%s') OR COALESCE(`%s`, '') = ''), VALUES(`%s`), `%s`))",
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindAdmin,
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindPack,
		constants.FieldSysField_OwnerPack,
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindPack,
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindAdmin,
		constants.FieldSysField_OwnerKind,
		constants.FieldOwnerKindRuntime,
		constants.FieldSysField_OwnerPack,
		constants.FieldSysField_OwnerPack,
		constants.FieldSysField_OwnerPack,
	))
}

func batchFieldMetadataUpdatableFields() []string {
	return []string{
		constants.FieldSysField_Type,
		constants.FieldSysField_Required, constants.FieldSysField_IsUnique,
		constants.FieldSysField_DefaultValue, constants.FieldSysField_SiblingMaxParentField, constants.FieldSysField_HelpText,
		constants.FieldSysField_IsSystem, constants.FieldSysField_IsNameField,
		constants.FieldSysField_Createable, constants.FieldSysField_Updateable,
		constants.FieldSysField_Options, constants.FieldSysField_MinLength,
		constants.FieldSysField_MaxLength, constants.FieldReferenceTo,
		constants.FieldSysField_WritableReferenceTo,
		constants.FieldSysField_Formula, constants.FieldSysField_ReturnType,
		constants.FieldSysField_FormulaMode, constants.FieldSysField_FormulaStorageMode, constants.FieldSysField_ProjectionConfig,
		constants.FieldSysField_Validator, constants.FieldSysField_ValidatorConfig, constants.FieldSysField_ControllingField,
		constants.FieldSysField_PicklistDependency,
		constants.FieldSysField_RollupConfig, constants.FieldSysField_IsMasterDetail,
		constants.FieldSysField_IsPolymorphic, constants.FieldSysField_DeleteRule,
		constants.FieldSysField_RelationshipName, constants.FieldSysField_ChildRelationshipName, constants.ColSysField_PhysicalName,
		constants.FieldSysField_WriteProtectionPolicy,
		constants.FieldSysField_ReadProtectionPolicy,
		// Issue #1582 RFC v1.6 — Source/IsCompound/ComposedFrom removed.
		// Provenance is derived at read time by fieldprovenance.ClassifySource;
		// there is no stored column to overwrite on pack re-apply.
		constants.FieldLastModifiedDate,
	}
}

func batchFieldMetadataGuardedFields() []string {
	return []string{
		constants.FieldSysField_Type,
		constants.FieldSysField_Required,
		constants.FieldSysField_IsUnique,
		constants.FieldSysField_DefaultValue,
		constants.FieldSysField_SiblingMaxParentField,
		constants.FieldSysField_HelpText,
		constants.FieldSysField_IsNameField,
		constants.FieldSysField_Options,
		constants.FieldSysField_MinLength,
		constants.FieldSysField_MaxLength,
		constants.FieldReferenceTo,
		constants.FieldSysField_WritableReferenceTo,
		constants.FieldSysField_Formula,
		constants.FieldSysField_ReturnType,
		constants.FieldSysField_FormulaMode,
		constants.FieldSysField_FormulaStorageMode,
		constants.FieldSysField_ProjectionConfig,
		constants.FieldSysField_Validator,
		constants.FieldSysField_ValidatorConfig,
		constants.FieldSysField_ControllingField,
		constants.FieldSysField_PicklistDependency,
		constants.FieldSysField_RollupConfig,
		constants.FieldSysField_IsMasterDetail,
		constants.FieldSysField_IsPolymorphic,
		constants.FieldSysField_DeleteRule,
		constants.FieldSysField_RelationshipName,
		constants.FieldSysField_ChildRelationshipName,
		constants.FieldSysField_WriteProtectionPolicy,
		constants.FieldSysField_ReadProtectionPolicy,
	}
}

// PrepareFieldForBatch converts a column definition to FieldWithContext
func (r *SchemaRepository) PrepareFieldForBatch(tableName string, col schema.ColumnDefinition) schema.FieldWithContext {
	fc := schema.FieldWithContext{
		ColumnDefinition: col,
		TableName:        tableName,
		// IDs will be generated if missing inside BatchSaveFieldMetadata or caller must provide
	}
	if schema.IsPolymorphicDiscriminatorColumn(col) {
		field := r.buildFieldMetadataFromColumn(col)
		field.Createable = false
		field.Updateable = false
		fc.Field = field
		fc.CapabilitiesDeclared = true
	}
	return fc
}

func (r *SchemaRepository) prepareStrictCreateFieldsForBatch(def schema.TableDefinition, objectMeta *models.ObjectMetadata) []schema.FieldWithContext {
	batchFields := make([]schema.FieldWithContext, 0, len(def.Columns))
	for _, col := range def.Columns {
		fc := r.prepareStrictCreateFieldForBatch(def.TableName, col, objectMeta)
		fc.OwnerKind = constants.FieldOwnerKindAdmin
		batchFields = append(batchFields, fc)
	}
	return batchFields
}

func (r *SchemaRepository) prepareStrictCreateFieldForBatch(tableName string, col schema.ColumnDefinition, objectMeta *models.ObjectMetadata) schema.FieldWithContext {
	fc := r.PrepareFieldForBatch(tableName, col)
	if objectMeta != nil && objectMeta.ID != "" {
		fc.ObjectID = objectMeta.ID
	}

	if r.IsSystemColumn(col.Name) {
		return fc
	}

	authored := authoredFieldForColumn(objectMeta, col)
	if authored == nil {
		return fc
	}

	field := *authored
	if field.APIName == "" {
		field.APIName = col.Name
	}
	if field.PhysicalName == "" {
		if col.PhysicalName != "" {
			field.PhysicalName = col.PhysicalName
		} else {
			field.PhysicalName = col.Name
		}
	}
	if field.Label == "" && col.Label != "" {
		field.Label = col.Label
	}
	if field.Type == "" {
		field.Type = models.FieldType(r.mapSQLTypeToLogical(col.Type))
	}
	if col.GeneratedForPolymorphicReference != "" {
		field.GeneratedForPolymorphicReference = col.GeneratedForPolymorphicReference
		field.Createable = false
		field.Updateable = false
		fc.CapabilitiesDeclared = true
	}
	fc.Field = &field
	return fc
}

func authoredFieldForColumn(objectMeta *models.ObjectMetadata, col schema.ColumnDefinition) *models.FieldMetadata {
	if objectMeta == nil {
		return nil
	}
	for i := range objectMeta.Fields {
		field := &objectMeta.Fields[i]
		if strings.EqualFold(field.APIName, col.Name) {
			return field
		}
		if col.PhysicalName != "" && strings.EqualFold(field.APIName, col.PhysicalName) {
			return field
		}
		if field.PhysicalName != "" && strings.EqualFold(field.PhysicalName, col.Name) {
			return field
		}
		if col.PhysicalName != "" && field.PhysicalName != "" && strings.EqualFold(field.PhysicalName, col.PhysicalName) {
			return field
		}
	}
	return nil
}

// GetTableRegistry retrieves all registered tables
func (r *SchemaRepository) GetTableRegistry(ctx context.Context) ([]*schema.TableRegistryItem, error) {
	qr := query.From(constants.TableTable).
		Select([]string{
			constants.ColID, constants.FieldSysTable_TableName, constants.FieldSysTable_TableType,
			constants.FieldSysTable_SchemaVersion, constants.ColCreatedDate,
			constants.FieldSysTable_Category, constants.FieldSysTable_Description,
			constants.FieldSysTable_IsManaged, constants.FieldSysTable_CreatedBy,
			constants.ColLastModifiedDate,
		}).
		Where(constants.FieldIsDeleted, false).
		Build()

	rows, err := r.db.QueryContext(ctx, qr.SQL, qr.Params...)
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	var items []*schema.TableRegistryItem
	for rows.Next() {
		var t schema.TableRegistryItem
		var description sql.NullString
		var category sql.NullString
		var schemaVersion sql.NullString

		err := rows.Scan(
			&t.ID, &t.TableName, &t.TableType, &category, &description,
			&t.IsManaged, &schemaVersion, &t.CreatedBy, &t.CreatedDate, &t.LastModifiedDate,
		)
		if err != nil {
			return nil, err
		}

		if description.Valid {
			t.Description = description.String
		}
		if category.Valid {
			t.Category = category.String
		}
		if schemaVersion.Valid {
			t.SchemaVersion = schemaVersion.String
		}

		items = append(items, &t)
	}

	if err := rows.Err(); err != nil {
		return nil, apperrors.NewInternalError("error iterating table registry", err)
	}

	return items, nil
}