package packs

import (
	"bytes"
	"context"
	"crypto/sha1"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/nexuscrm/automation/flowauthoring"
	kernelconnector "github.com/nexuscrm/kernel/pkg/connector"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	seedrows "github.com/pingcap-inc/nexuscrm-contract/pkg/seedrows"
	"github.com/pingcap-inc/nexuscrm-controlplane/layoutownership"
)

type sourceBackedManagedComponent struct {
	packName           string
	manifestComponent  ManifestComponent
	payload            []byte
	expectedProjection func() (any, error)
	liveProjection     func(context.Context, VerifyDeps) (any, error)
	apply              func(context.Context, ApplyDeps) error
	verify             func(context.Context, VerifyDeps) error
	applyGroupKey      string
}

func (c *sourceBackedManagedComponent) Kind() ComponentKind {
	return c.manifestComponent.Kind
}

func (c *sourceBackedManagedComponent) PackName() string {
	return c.packName
}

func (c *sourceBackedManagedComponent) BusinessKey() string {
	return strings.TrimSpace(c.manifestComponent.BusinessKey)
}

func (c *sourceBackedManagedComponent) Manageability() Manageability {
	return c.manifestComponent.Manageability
}

func (c *sourceBackedManagedComponent) DependsOn() []ComponentReference {
	return append([]ComponentReference(nil), c.manifestComponent.DependsOn...)
}

func (c *sourceBackedManagedComponent) ApplyGroupKey() string {
	return strings.TrimSpace(c.applyGroupKey)
}

func (c *sourceBackedManagedComponent) ExpectedPayload() ([]byte, error) {
	return append([]byte(nil), c.payload...), nil
}

func (c *sourceBackedManagedComponent) ExpectedProjection() (any, error) {
	if c.expectedProjection == nil {
		return nil, nil
	}
	projection, err := c.expectedProjection()
	if err != nil {
		return nil, err
	}
	return clonePackComponentProjection(projection)
}

func (c *sourceBackedManagedComponent) LiveProjection(ctx context.Context, deps VerifyDeps) (any, error) {
	if c.liveProjection == nil {
		return nil, nil
	}
	projection, err := c.liveProjection(ctx, deps)
	if err != nil {
		return nil, err
	}
	return clonePackComponentProjection(projection)
}

func (c *sourceBackedManagedComponent) Apply(ctx context.Context, deps ApplyDeps) error {
	if c.apply == nil {
		return nil
	}
	return c.apply(ctx, deps)
}

func (c *sourceBackedManagedComponent) VerifyComponent(ctx context.Context, deps VerifyDeps) error {
	if c.verify != nil {
		return c.verify(ctx, deps)
	}
	return verifyPackComponentDefault(ctx, deps, c)
}

func newSourceBackedPermissionComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, error) {
	rows, err := seedrows.ParseStandardPermissions(payload, componentPayloadLabel(component))
	if err != nil {
		return nil, err
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return seedSystemPermissionRows(ctx, deps.DB, rows, true, packName)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return verifySystemPermissionRows(ctx, deps.DB, rows, packName)
		},
	}, nil
}

func trimmedStringPtr(value *string) *string {
	if value == nil {
		return nil
	}
	trimmed := strings.TrimSpace(*value)
	if trimmed == "" {
		return nil
	}
	return &trimmed
}

func newSourceBackedListFieldsComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, error) {
	expected, err := sourceBackedListFieldsProjection(payload, componentPayloadLabel(component))
	if err != nil {
		return nil, err
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		expectedProjection: func() (any, error) {
			return expected, nil
		},
		liveProjection: func(ctx context.Context, deps VerifyDeps) (any, error) {
			return liveSourceBackedListFieldsProjection(ctx, deps, expected)
		},
		apply: func(_ context.Context, deps ApplyDeps) error {
			return applyStandardListFieldsFromPayload(deps.DB, payload, componentPayloadLabel(component))
		},
	}, nil
}

type sourceBackedListFieldsRow struct {
	ObjectAPIName string   `json:"objectApiName"`
	Fields        []string `json:"fields"`
}

func sourceBackedListFieldsProjection(payload []byte, source string) ([]sourceBackedListFieldsRow, error) {
	var byObject map[string][]string
	if err := json.Unmarshal(payload, &byObject); err != nil {
		return nil, fmt.Errorf("parse %s: %w", source, err)
	}
	out := make([]sourceBackedListFieldsRow, 0, len(byObject))
	for objectAPIName, fields := range byObject {
		out = append(out, sourceBackedListFieldsRow{
			ObjectAPIName: objectAPIName,
			Fields:        append([]string{}, sourceBackedNormalizeStringSlice(fields)...),
		})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].ObjectAPIName < out[j].ObjectAPIName })
	return out, nil
}

func liveSourceBackedListFieldsProjection(ctx context.Context, deps VerifyDeps, expected []sourceBackedListFieldsRow) (any, error) {
	live := make([]sourceBackedListFieldsRow, 0, len(expected))
	for _, want := range expected {
		var raw sql.NullString
		var customized bool
		err := deps.DB.QueryRowContext(ctx, `
SELECT ListFields, ListFieldsCustomized
FROM _System_Object
WHERE ApiName = ? AND IsDeleted = 0
LIMIT 1`, want.ObjectAPIName).Scan(&raw, &customized)
		if err == sql.ErrNoRows {
			return nil, fmt.Errorf("list fields object %s is missing", want.ObjectAPIName)
		}
		if err != nil {
			return nil, fmt.Errorf("query list fields for %s: %w", want.ObjectAPIName, err)
		}
		if customized {
			live = append(live, want)
			continue
		}
		var fields []string
		if raw.Valid && strings.TrimSpace(raw.String) != "" {
			if err := json.Unmarshal([]byte(raw.String), &fields); err != nil {
				return nil, fmt.Errorf("%s ListFields is not JSON array: %w", want.ObjectAPIName, err)
			}
		}
		live = append(live, sourceBackedListFieldsRow{
			ObjectAPIName: want.ObjectAPIName,
			Fields:        append([]string{}, sourceBackedNormalizeStringSlice(fields)...),
		})
	}
	return live, nil
}

func newSourceBackedFieldMappingComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, error) {
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		apply: func(_ context.Context, deps ApplyDeps) error {
			return InitializeFieldMappingsFromPayloadForOwner(deps.DB, payload, packName)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return VerifyFieldMappingsFromPayloadForOwner(ctx, deps.DB, payload, packName)
		},
	}, nil
}

// newSourceBackedConnectorFieldMappingComponent constructs the per-component
// upsert-only apply/verify for a connector_field_mapping component.
//
// The returned declared slice is the union contribution from this component;
// the caller (buildComponents) accumulates these into
// p.connectorFieldMappingDeclarations so the pack-level sweeper receives the
// full union. This mirrors the connector_capability pattern exactly.
func newSourceBackedConnectorFieldMappingComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, []connector.ConnectorFieldMappingInput, error) {
	// BusinessKey contract: the first segment is the source/owner the rows
	// install under and MUST equal packName — the same value Parse persists as
	// Source and the resolver later queries by. Enforce it so a manifest cannot
	// declare one source (e.g. "other-connector/...") while the rows silently
	// install under packName and the declared source is dropped. Mirrors the
	// businessKey guard every other connector kind already has (connector_rate_limit,
	// webhook_secret, connector_inbound_route, connector_callout_profile); this
	// kind was the only one missing it.
	if err := validateConnectorFieldMappingBusinessKeySource(packName, component); err != nil {
		return nil, nil, err
	}
	// Parse at build time to (a) fail fast on invalid payload and (b) provide
	// the declared-row union to the pack-level sweeper without re-parsing at
	// apply time.
	declared, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload(payload, packName)
	if err != nil {
		return nil, nil, err
	}
	comp := &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		apply: func(_ context.Context, deps ApplyDeps) error {
			_, err := InitializeConnectorFieldMappingsFromPayloadForOwner(deps.DB, payload, packName)
			return err
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return VerifyConnectorFieldMappingsFromPayloadForOwner(ctx, deps.DB, payload, packName)
		},
	}
	return comp, declared, nil
}

func validateConnectorFieldMappingBusinessKeySource(packName string, component ManifestComponent) error {
	bk := strings.TrimSpace(component.BusinessKey)
	firstSeg := bk
	if idx := strings.Index(bk, "/"); idx >= 0 {
		firstSeg = bk[:idx]
	}
	if !strings.EqualFold(strings.TrimSpace(firstSeg), strings.TrimSpace(packName)) {
		return fmt.Errorf(
			"connector_field_mapping component businessKey %q first segment must equal pack name %q (the source/owner the rows install under)",
			component.BusinessKey, packName)
	}
	return nil
}

func newSourceBackedConnectorPushContractComponent(packName string, component ManifestComponent, payload []byte, flowPayloads map[string][]byte) (PackComponent, connector.ConnectorPushDurableMetadataContract, error) {
	contract, err := kernelconnector.ParseConnectorPushDurableMetadataContractFromPayload(payload, packName)
	if err != nil {
		return nil, connector.ConnectorPushDurableMetadataContract{}, err
	}
	if err := validateConnectorPushContractFlowSlot(contract, flowPayloads); err != nil {
		return nil, connector.ConnectorPushDurableMetadataContract{}, err
	}
	expectedBusinessKey := JoinComponentBusinessKey(
		contract.ConnectorOwnerKey,
		contract.FlowDeveloperName,
		contract.FlowActionName,
		contract.SourceObject,
		contract.DestinationSlotID,
	)
	if strings.TrimSpace(component.BusinessKey) != expectedBusinessKey {
		return nil, connector.ConnectorPushDurableMetadataContract{}, fmt.Errorf(
			"connector_push_contract manifest businessKey %q does not match assembled durable identity %q",
			component.BusinessKey,
			expectedBusinessKey,
		)
	}
	comp := &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		expectedProjection: func() (any, error) {
			return contract, nil
		},
		liveProjection: func(context.Context, VerifyDeps) (any, error) {
			return contract, nil
		},
	}
	return comp, contract, nil
}

func validateConnectorPushContractFlowSlot(contract connector.ConnectorPushDurableMetadataContract, flowPayloads map[string][]byte) error {
	flowName := strings.TrimSpace(contract.FlowDeveloperName)
	raw, ok := flowPayloads[flowName+".flow.json"]
	if !ok {
		return fmt.Errorf("connector_push_contract %s/%s references standard_flow %q that is not declared by this pack", flowName, strings.TrimSpace(contract.FlowActionName), flowName)
	}
	var meta models.FlowMetadata
	if err := json.Unmarshal(raw, &meta); err != nil {
		return fmt.Errorf("connector_push_contract %s/%s parse standard_flow: %w", flowName, strings.TrimSpace(contract.FlowActionName), err)
	}
	if meta.Start == nil || strings.TrimSpace(meta.Start.Object) != strings.TrimSpace(contract.SourceObject) {
		got := ""
		if meta.Start != nil {
			got = strings.TrimSpace(meta.Start.Object)
		}
		return fmt.Errorf("connector_push_contract %s/%s sourceObject %q does not match standard_flow start object %q", flowName, strings.TrimSpace(contract.FlowActionName), strings.TrimSpace(contract.SourceObject), got)
	}
	for _, action := range meta.ActionCalls {
		if strings.TrimSpace(action.Name) != strings.TrimSpace(contract.FlowActionName) {
			continue
		}
		if strings.TrimSpace(action.ActionType) != constants.ActionTypePushToConnector {
			return fmt.Errorf("connector_push_contract %s/%s action type %q is not %s", flowName, strings.TrimSpace(contract.FlowActionName), strings.TrimSpace(action.ActionType), constants.ActionTypePushToConnector)
		}
		return nil
	}
	return fmt.Errorf("connector_push_contract %s/%s references PushToConnector action that is not declared by the standard_flow", flowName, strings.TrimSpace(contract.FlowActionName))
}

// newSourceBackedChildCloneRulesComponent constructs the declared manifest
// component for child_clone_rules. The component payload is overlaid into the
// owning object_spec component before apply/verify, so _System_Object
// ChildCloneRules and ChildCloneRulesPack* sidecars still have one DB authority:
// object_spec. This component remains in the manifest/dependency/hash surface
// but does not write the DB directly.
func newSourceBackedChildCloneRulesComponent(packName string, component ManifestComponent, payload []byte, objectSpecIndex map[string]sourceBackedObjectSpecTarget) (PackComponent, error) {
	rulesByObject, err := parseSourceBackedChildCloneRulesPayload(component, payload)
	if err != nil {
		return nil, err
	}
	for objectAPIName := range rulesByObject {
		key := strings.ToLower(strings.TrimSpace(objectAPIName))
		target, ok := objectSpecIndex[key]
		if !ok {
			return nil, fmt.Errorf("child_clone_rules for %s requires declaring object_spec in the same pack", objectAPIName)
		}
		if !hasComponentDependency(component.DependsOn, ComponentKindObjectSpec, target.businessKey) {
			return nil, fmt.Errorf("child_clone_rules for %s requires dependsOn on object_spec %q", objectAPIName, target.businessKey)
		}
		if target.hasRawChildCloneRules {
			return nil, fmt.Errorf("child_clone_rules for %s conflicts with non-empty object_spec.childCloneRules in object_spec %q; author ChildCloneRules through exactly one pack surface", objectAPIName, target.businessKey)
		}
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		apply:             func(context.Context, ApplyDeps) error { return nil },
		verify:            func(context.Context, VerifyDeps) error { return nil },
	}, nil
}

func parseSourceBackedChildCloneRulesPayload(component ManifestComponent, payload []byte) (map[string][]models.ChildCloneRule, error) {
	var rulesByObject map[string][]models.ChildCloneRule
	if err := json.Unmarshal(payload, &rulesByObject); err != nil {
		return nil, fmt.Errorf("parse %s: %w", componentPayloadLabel(component), err)
	}
	seenObjects := map[string]string{}
	for objectAPIName, rules := range rulesByObject {
		key := strings.ToLower(strings.TrimSpace(objectAPIName))
		if key == "" {
			return nil, fmt.Errorf("child_clone_rules payload contains empty objectApiName")
		}
		if previous, ok := seenObjects[key]; ok {
			return nil, fmt.Errorf("child_clone_rules payload contains duplicate objectApiName %q and %q after canonicalization", previous, objectAPIName)
		}
		seenObjects[key] = objectAPIName
		if rules == nil {
			rulesByObject[objectAPIName] = []models.ChildCloneRule{}
		}
	}
	return rulesByObject, nil
}

func newSourceBackedStandardFlowComponent(
	packName string,
	component ManifestComponent,
	payload []byte,
	packFlowPayloads map[string][]byte,
	packObjectSpecs []ObjectSpec,
	formulaFunctionBundles []formula.FunctionBundle,
) (PackComponent, error) {
	flowName := sourceBackedFlowFileName(component)
	if _, err := LoadStandardFlowSeedsFromPayloads(map[string][]byte{flowName: payload}); err != nil {
		return nil, err
	}
	var meta models.FlowMetadata
	if jsonErr := json.Unmarshal(payload, &meta); jsonErr != nil {
		return nil, fmt.Errorf("parse flow metadata %s: %w", flowName, jsonErr)
	}
	if err := flowauthoring.ValidateNoLegacyEmailAlertActionCalls(meta); err != nil {
		return nil, fmt.Errorf("standard_flow %s: %w", flowName, err)
	}
	graph := buildFlowObjectFieldGraph(packObjectSpecs)
	if refErr := validateFlowFieldReferences(flowName, meta, graph); refErr != nil {
		return nil, refErr
	}
	if syntaxErr := validateFlowFilterFormulaSyntaxWithFunctionBundles(flowName, meta, packObjectSpecs, formulaFunctionBundles); syntaxErr != nil {
		return nil, syntaxErr
	}
	if err := models.ValidateConnectorSource(meta.Start); err != nil {
		return nil, fmt.Errorf("standard_flow %s: %w", flowName, err)
	}
	batchPayloads := map[string][]byte{}
	for name, raw := range packFlowPayloads {
		batchPayloads[name] = append([]byte(nil), raw...)
	}
	if len(batchPayloads) == 0 {
		batchPayloads[flowName] = append([]byte(nil), payload...)
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		applyGroupKey:     string(ComponentKindStandardFlow),
		apply: func(_ context.Context, deps ApplyDeps) error {
			return InitializeStandardFlowsFromPayloadsForOwner(deps.DB, batchPayloads, packName)
		},
		verify: func(_ context.Context, deps VerifyDeps) error {
			return VerifyStandardFlowsFromPayloadsForOwner(deps.DB, map[string][]byte{flowName: payload}, packName)
		},
	}, nil
}

func newSourceBackedInvocableActionComponent(packName string, component ManifestComponent, payload []byte, flowPayloads map[string][]byte) (PackComponent, error) {
	if len(flowPayloads) == 0 {
		return nil, fmt.Errorf("invocable actions require standard_flow components in the same pack")
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		apply: func(_ context.Context, deps ApplyDeps) error {
			return InitializeStandardActionsForOwnerFromPayloads(deps.DB, packName, payload, flowPayloads)
		},
		verify: func(_ context.Context, deps VerifyDeps) error {
			return VerifyStandardActionsForOwnerFromPayload(deps.DB, packName, payload)
		},
	}, nil
}

func newSourceBackedRelatedListRuleComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, []metadata.RelatedListRule, error) {
	var parsed sourceBackedRelatedListRulesPayload
	if err := json.Unmarshal(payload, &parsed); err != nil {
		return nil, nil, fmt.Errorf("parse %s: %w", componentPayloadLabel(component), err)
	}
	if err := metadata.ValidatePackRelatedListRules(parsed.Rules); err != nil {
		return nil, nil, err
	}
	return newRelatedListRuleManagedComponentFromRules(packName, component, payload, parsed.Rules), parsed.Rules, nil
}

// newRelatedListRuleManagedComponentFromRules builds the managed component from
// an already-parsed/validated metadata.RelatedListRule slice plus the RAW
// payload bytes and the engine ManifestComponent. It is shared by the internal
// build path (newSourceBackedRelatedListRuleComponent) and the engine
// implementation of cp.PresentationComponentBuildContext.AddRelatedListRuleComponent,
// guaranteeing both paths produce a byte-identical component (the payload field
// is the raw bytes, never re-marshalled). The verify closure validates the same
// rule slice, exactly as the internal builder does.
func newRelatedListRuleManagedComponentFromRules(packName string, component ManifestComponent, payload []byte, rules []metadata.RelatedListRule) PackComponent {
	verifyRules := append([]metadata.RelatedListRule(nil), rules...)
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		verify: func(_ context.Context, _ VerifyDeps) error {
			return metadata.ValidatePackRelatedListRules(verifyRules)
		},
	}
}

type sourceBackedRelatedListRulesPayload struct {
	Rules []metadata.RelatedListRule `json:"rules"`
}

func newSourceBackedPageLayoutComponent(packName string, component ManifestComponent, payload []byte, rules []metadata.RelatedListRule) (PackComponent, error) {
	if strings.TrimSpace(packName) == "" {
		return nil, fmt.Errorf("source-backed pack: empty manifest name is forbidden for page-layout component")
	}
	layout, err := sourceBackedPageLayoutFromPayload(payload, component.BusinessKey, componentPayloadLabel(component))
	if err != nil {
		return nil, err
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			next := layout
			if err := seedSourceBackedRelatedLists(deps.DB, &next, rules); err != nil {
				return err
			}
			if isSourceBackedDefaultLayout(next) {
				return applySourceBackedDefaultPageLayout(ctx, deps.DB, next, packName)
			}
			return applySourceBackedManagedLayoutRow(ctx, deps.DB, next, packName)
		},
		verify: func(_ context.Context, deps VerifyDeps) error {
			// No seedSourceBackedRelatedLists call here. With the two-axis split:
			//   - axis 1 strips RelatedLists from both expected and live before
			//     comparing, so non-RelatedLists content comes from `layout` (pre-
			//     seed). seedSourceBackedRelatedLists only mutates RelatedLists, so
			//     skipping it does not change any non-RelatedLists field.
			//   - axis 2 compares payload-authored RelatedLists (layout.RelatedLists
			//     pre-seed) against live by (ObjectAPIName, LookupField). The seed
			//     result is not the expected value for axis 2.
			// Removing the verify-time seed call also eliminates the live-DB read
			// whose non-determinism between apply time and verify time is the root
			// cause of issue #424 REPRO-B.
			if isSourceBackedDefaultLayout(layout) {
				if err := verifySourceBackedDefaultPageLayout(context.Background(), deps.DB, layout, packName, layout.ObjectAPIName); err != nil {
					return err
				}
				return verifyCuratedPageLayoutConfig(
					context.Background(), deps.DB, layout,
					layout.RelatedLists, // pre-seed, payload-authored
					rules,
					layout.ObjectAPIName,
				)
			}
			if err := verifySourceBackedManagedLayoutRow(context.Background(), deps.DB, layout, packName, layout.ObjectAPIName); err != nil {
				return err
			}
			return verifySourceBackedManagedLayoutConfig(
				context.Background(), deps.DB, layout, packName,
				layout.RelatedLists, // pre-seed, payload-authored
				rules,
				layout.ObjectAPIName,
			)
		},
	}, nil
}

func sourceBackedPageLayoutFromPayload(payload []byte, args ...string) (models.PageLayout, error) {
	businessKey := ""
	source := ""
	switch len(args) {
	case 1:
		source = args[0]
		if looksLikeSourceBackedPageLayoutBusinessKey(args[0]) {
			businessKey = args[0]
		}
	case 2:
		businessKey = args[0]
		source = args[1]
	default:
		return models.PageLayout{}, fmt.Errorf("parse page_layout payload: want source or businessKey+source")
	}
	var envelope struct {
		IsDefault *bool `json:"isDefault"`
	}
	if err := json.Unmarshal(payload, &envelope); err != nil {
		return models.PageLayout{}, fmt.Errorf("parse %s: %w", source, err)
	}
	layout, err := layoutownership.DecodePageLayoutConfig(payload)
	if err != nil {
		return layout, fmt.Errorf("parse %s: %w", source, err)
	}
	layout.ObjectAPIName = strings.TrimSpace(layout.ObjectAPIName)
	layout.LayoutName = strings.TrimSpace(layout.LayoutName)
	if layout.ObjectAPIName == "" {
		return layout, fmt.Errorf("parse %s: missing objectApiName", source)
	}
	componentDeclaresDefault := false
	componentObjectAPIName := ""
	componentLayoutKey := ""
	if objectAPIName, layoutKey, ok := splitSourceBackedPageLayoutBusinessKey(businessKey); ok {
		componentObjectAPIName = strings.TrimSpace(objectAPIName)
		componentLayoutKey = strings.TrimSpace(layoutKey)
		if strings.EqualFold(layoutKey, "default") {
			componentDeclaresDefault = true
		}
	}
	if componentObjectAPIName != "" && !strings.EqualFold(componentObjectAPIName, layout.ObjectAPIName) {
		return models.PageLayout{}, fmt.Errorf(
			"parse %s: page_layout objectApiName %q must match business key head %q",
			source, layout.ObjectAPIName, componentObjectAPIName,
		)
	}
	if layout.LayoutName == "" && componentLayoutKey != "" && !componentDeclaresDefault {
		layout.LayoutName = componentLayoutKey
	}
	if !componentDeclaresDefault && componentLayoutKey != "" && layout.LayoutName != "" && strings.TrimSpace(layout.LayoutName) != componentLayoutKey {
		return models.PageLayout{}, fmt.Errorf(
			"parse %s: non-default page_layout layoutName %q must match business key tail %q",
			source, layout.LayoutName, componentLayoutKey,
		)
	}
	switch {
	case envelope.IsDefault != nil:
		layout.IsDefault = *envelope.IsDefault
	case componentDeclaresDefault:
		layout.IsDefault = true
	case componentLayoutKey != "" && !componentDeclaresDefault:
		layout.IsDefault = false
	case layout.LayoutName != "" && !strings.EqualFold(layout.LayoutName, "default"):
		layout.IsDefault = false
	default:
		return models.PageLayout{}, fmt.Errorf("parse %s: non-default page_layout requires explicit isDefault", source)
	}
	if componentDeclaresDefault && !layout.IsDefault {
		return models.PageLayout{}, fmt.Errorf("parse %s: default page_layout business key requires isDefault=true", source)
	}
	if !componentDeclaresDefault && layout.IsDefault {
		return models.PageLayout{}, fmt.Errorf("parse %s: non-default page_layout business key requires isDefault=false", source)
	}
	if !layout.IsDefault && (layout.LayoutName == "" || strings.EqualFold(layout.LayoutName, "default")) {
		return layout, fmt.Errorf("parse %s: non-default page_layout requires non-empty non-reserved layoutName", source)
	}
	layout.ID = sourceBackedLayoutID(layout.ObjectAPIName, layout.LayoutName, layout.IsDefault)
	if !layout.IsDefault && strings.EqualFold(layout.ID, sourceBackedLayoutID(layout.ObjectAPIName, "", true)) {
		return layout, fmt.Errorf(
			"parse %s: non-default page_layout %s/%s normalizes to reserved default managed ID %s",
			source,
			layout.ObjectAPIName,
			layout.LayoutName,
			layout.ID,
		)
	}
	layout.Source = curatedLayoutSource
	layout.GenerationVersion = 0
	layout.IsCustomized = false
	layout.LayoutVersion = 0
	return layout, nil
}

func looksLikeSourceBackedPageLayoutBusinessKey(value string) bool {
	if strings.Contains(value, "[") || strings.Contains(value, "]") || strings.Contains(value, ".") {
		return false
	}
	_, _, ok := splitSourceBackedPageLayoutBusinessKey(value)
	return ok
}

func seedSourceBackedRelatedLists(db *sql.DB, layout *models.PageLayout, rules []metadata.RelatedListRule) error {
	if layout == nil || len(layout.RelatedLists) > 0 {
		return nil
	}
	nameFieldMap, err := loadNameFieldMap(db)
	if err != nil {
		return fmt.Errorf("load name field map: %w", err)
	}
	candidates, err := getRelatedListCandidates(db, layout.ObjectAPIName)
	if err != nil {
		return fmt.Errorf("get related list candidates: %w", err)
	}
	resolvedCandidates := resolveMetadataRelatedListCandidatesWithRules(layout.ObjectAPIName, nameFieldMap, candidates, rules)
	layout.RelatedLists = make([]models.RelatedListConfig, 0, len(resolvedCandidates))
	for _, candidate := range resolvedCandidates {
		if !candidate.IsRecommended {
			continue
		}
		layout.RelatedLists = append(layout.RelatedLists, models.RelatedListConfig{
			ID:             candidate.ID,
			Label:          candidate.Label,
			SecondaryLabel: candidate.SecondaryLabel,
			ObjectAPIName:  candidate.ObjectAPIName,
			LookupField:    candidate.LookupField,
			Fields:         append([]string(nil), candidate.Fields...),
		})
	}
	metadata.SortByCuratedOrderWithRules(layout.ObjectAPIName, layout.RelatedLists, rules)
	return nil
}

func applySourceBackedDefaultPageLayout(ctx context.Context, db *sql.DB, layout models.PageLayout, packName string) error {
	if _, err := backfillLegacySourceBackedCuratedDefaultLayoutID(ctx, db, packName, layout); err != nil {
		return fmt.Errorf("adopt legacy source-backed default layout %s/%s: %w", layout.ObjectAPIName, layout.LayoutName, err)
	}
	return applyCuratedPageLayout(ctx, db, layout, sourceBackedLayoutAssignmentID(layout.ObjectAPIName), layout.ObjectAPIName, packName)
}

func verifySourceBackedDefaultPageLayout(ctx context.Context, db *sql.DB, layout models.PageLayout, packName string, logName string) error {
	row, err := layoutownership.LoadPageLayoutRow(ctx, db, layout.ID)
	if err != nil {
		return fmt.Errorf("query source-backed default layout row %s: %w", logName, err)
	}
	if row != nil && !layoutownership.IsAdminOwnedPageLayout(row.IsCustomized, row.Source) {
		liveOwnerPack := strings.TrimSpace(row.OwnerPack)
		if liveOwnerPack == "" || !strings.EqualFold(liveOwnerPack, strings.TrimSpace(packName)) {
			return fmt.Errorf(
				"source-backed page_layout %s/default row %s OwnerPack = %q, refusing to verify for %q",
				layout.ObjectAPIName, row.ID, row.OwnerPack, packName,
			)
		}
	}
	return verifyCuratedPageLayout(db, layout, packName, logName)
}

func applySourceBackedManagedLayoutRow(ctx context.Context, db *sql.DB, layout models.PageLayout, packName string) error {
	row, err := layoutownership.LoadPageLayoutRow(ctx, db, layout.ID)
	if err != nil {
		return fmt.Errorf("query source-backed managed layout row %s/%s: %w", layout.ObjectAPIName, layout.LayoutName, err)
	}
	if row != nil && layoutownership.IsAdminOwnedPageLayout(row.IsCustomized, row.Source) {
		return nil
	}
	layout.IsDefault = false
	resolvedLayout, err := layoutownership.UpsertManagedPageLayoutResolved(ctx, db, layout, curatedLayoutSource, packName)
	if err != nil {
		return fmt.Errorf("upsert managed source-backed layout row %s/%s: %w", layout.ObjectAPIName, layout.LayoutName, err)
	}
	if err := layoutownership.ClearPageLayoutDefaultFlagByID(ctx, db, resolvedLayout.ID); err != nil {
		return fmt.Errorf("clear managed source-backed non-default flag %s/%s: %w", layout.ObjectAPIName, layout.LayoutName, err)
	}
	return nil
}

func verifySourceBackedManagedLayoutConfig(
	ctx context.Context,
	db *sql.DB,
	expected models.PageLayout,
	packName string,
	payloadAuthoredRelatedLists []models.RelatedListConfig,
	relatedListRules []metadata.RelatedListRule,
	logName string,
) error {
	row, err := resolveSourceBackedManagedLiveRow(ctx, db, expected, packName, logName)
	if err != nil {
		return err
	}
	if layoutownership.IsAdminOwnedPageLayout(row.IsCustomized, row.Source) {
		return nil
	}

	live, err := layoutownership.DecodePageLayoutConfig([]byte(row.ConfigJSON))
	if err != nil {
		return fmt.Errorf("parse source-backed managed %s layout config: %w", logName, err)
	}
	expectedLiveIdentity := expected
	expectedLiveIdentity.ID = row.ID
	if err := compareLayoutContentExcludingRelatedLists(expectedLiveIdentity, live); err != nil {
		return &curatedPageLayoutConfigDriftError{logName: logName}
	}
	if err := comparePayloadAuthoredRelatedLists(payloadAuthoredRelatedLists, live.RelatedLists); err != nil {
		return &curatedPageLayoutConfigDriftError{logName: logName}
	}
	if len(payloadAuthoredRelatedLists) == 0 {
		if err := checkLiveRelatedListsSeedSentinel(db, expected.ObjectAPIName, live.RelatedLists, relatedListRules); err != nil {
			return &curatedPageLayoutConfigDriftError{logName: logName}
		}
	}
	return nil
}

func resolveSourceBackedManagedLiveRow(ctx context.Context, db *sql.DB, expected models.PageLayout, packName, logName string) (*layoutownership.PageLayoutRow, error) {
	row, err := layoutownership.LoadPageLayoutRow(ctx, db, expected.ID)
	if err != nil {
		return nil, fmt.Errorf("query source-backed managed layout row %s: %w", logName, err)
	}
	if row != nil {
		if layoutownership.IsAdminOwnedPageLayout(row.IsCustomized, row.Source) {
			return row, nil
		}
		liveOwnerPack := strings.TrimSpace(row.OwnerPack)
		if liveOwnerPack == "" {
			return nil, fmt.Errorf(
				"source-backed page_layout %s/%s row %s OwnerPack = %q, refusing to verify for %q",
				expected.ObjectAPIName, expected.LayoutName, row.ID, row.OwnerPack, packName,
			)
		}
		if liveOwnerPack != "" && !strings.EqualFold(liveOwnerPack, strings.TrimSpace(packName)) {
			return nil, fmt.Errorf(
				"source-backed page_layout %s/%s row %s OwnerPack = %q, refusing to verify for %q",
				expected.ObjectAPIName, expected.LayoutName, row.ID, row.OwnerPack, packName,
			)
		}
	}
	businessKeyRow, err := layoutownership.LoadPageLayoutRowByBusinessKey(ctx, db, expected.ObjectAPIName, expected.LayoutName)
	if err != nil {
		return nil, fmt.Errorf("query source-backed managed layout row by business key %s: %w", logName, err)
	}
	if row != nil {
		if businessKeyRow != nil && !strings.EqualFold(strings.TrimSpace(businessKeyRow.ID), strings.TrimSpace(row.ID)) {
			return nil, fmt.Errorf(
				"source-backed page_layout %s/%s has duplicate live rows %s and %s for the same business key; refusing verify for %q until same-name duplicate state is repaired",
				expected.ObjectAPIName, expected.LayoutName, row.ID, businessKeyRow.ID, packName,
			)
		}
		return row, nil
	}
	if businessKeyRow == nil {
		return nil, &curatedPageLayoutMissingError{logName: logName, layoutID: expected.ID}
	}
	if layoutownership.IsAdminOwnedPageLayout(businessKeyRow.IsCustomized, businessKeyRow.Source) {
		return nil, &curatedPageLayoutMissingError{logName: logName, layoutID: expected.ID}
	}
	liveOwnerPack := strings.TrimSpace(businessKeyRow.OwnerPack)
	if liveOwnerPack != "" && !strings.EqualFold(liveOwnerPack, strings.TrimSpace(packName)) {
		return nil, fmt.Errorf(
			"source-backed page_layout %s/%s row %s OwnerPack = %q, refusing to verify for %q",
			expected.ObjectAPIName, expected.LayoutName, businessKeyRow.ID, businessKeyRow.OwnerPack, packName,
		)
	}
	return businessKeyRow, nil
}

func verifySourceBackedManagedLayoutRow(ctx context.Context, db *sql.DB, expected models.PageLayout, packName, logName string) error {
	row, err := resolveSourceBackedManagedLiveRow(ctx, db, expected, packName, logName)
	if err != nil {
		return err
	}
	if layoutownership.IsAdminOwnedPageLayout(row.IsCustomized, row.Source) {
		return nil
	}
	if row.IsDefault {
		target, err := layoutownership.LoadWildcardLayoutTarget(ctx, db, expected.ObjectAPIName)
		if err != nil {
			return fmt.Errorf("query source-backed managed layout wildcard target %s: %w", logName, err)
		}
		if target != nil &&
			target.LayoutID == row.ID &&
			target.RouteSource != nil &&
			strings.EqualFold(strings.TrimSpace(*target.RouteSource), models.LayoutAssignmentRouteSourceAdminAuthored) {
			return nil
		}
		return &curatedPageLayoutRoutingError{
			logName: logName,
			detail:  fmt.Sprintf("non-default source-backed layout row %s is incorrectly marked default", expected.ID),
		}
	}
	return nil
}

const (
	sourceBackedLayoutIDMaxLen       = 255
	sourceBackedLayoutIDHashLen      = 12
	sourceBackedLayoutIDPrefix       = "curated_"
	sourceBackedLayoutIDMinSlugLen   = 1
	sourceBackedLayoutIDHashSepLen   = 1
	sourceBackedLayoutIDObjectSepLen = 1
)

func sourceBackedLayoutID(objectAPIName string, layoutName string, isDefault bool) string {
	normalizedObject := strings.TrimSpace(objectAPIName)
	normalizedLayout := strings.TrimSpace(layoutName)
	if isDefault || normalizedLayout == "" || strings.EqualFold(normalizedLayout, "default") {
		return "curated_" + normalizedObject + "_default"
	}
	objectPart := normalizedObject
	layoutPart := canonicalIdentifierFragment(normalizedLayout)
	hashPart := sourceBackedLayoutIDHash(normalizedObject, normalizedLayout)
	maxObjectLen := sourceBackedLayoutIDMaxLen - len(sourceBackedLayoutIDPrefix) - sourceBackedLayoutIDObjectSepLen - sourceBackedLayoutIDHashSepLen - len(hashPart) - sourceBackedLayoutIDMinSlugLen
	if maxObjectLen < 1 {
		maxObjectLen = 1
	}
	if len(objectPart) > maxObjectLen {
		objectPart = objectPart[:maxObjectLen]
	}
	maxLayoutLen := sourceBackedLayoutIDMaxLen - len(sourceBackedLayoutIDPrefix) - len(objectPart) - sourceBackedLayoutIDObjectSepLen - sourceBackedLayoutIDHashSepLen - len(hashPart)
	if maxLayoutLen < 1 {
		maxLayoutLen = 1
	}
	if len(layoutPart) > maxLayoutLen {
		layoutPart = strings.Trim(layoutPart[:maxLayoutLen], "_")
	}
	if layoutPart == "" {
		layoutPart = "x"
	}
	return sourceBackedLayoutIDPrefix + objectPart + "_" + layoutPart + "_" + hashPart
}

func sourceBackedLayoutAssignmentID(objectAPIName string) string {
	return "curated_" + objectAPIName + "_assignment_wildcard"
}

func canonicalIdentifierFragment(value string) string {
	value = strings.TrimSpace(value)
	if value == "" {
		return "default"
	}
	var b strings.Builder
	b.Grow(len(value))
	lastUnderscore := false
	for _, r := range value {
		switch {
		case r >= 'A' && r <= 'Z':
			b.WriteRune(r)
			lastUnderscore = false
		case r >= 'a' && r <= 'z':
			b.WriteRune(r)
			lastUnderscore = false
		case r >= '0' && r <= '9':
			b.WriteRune(r)
			lastUnderscore = false
		default:
			if !lastUnderscore {
				b.WriteByte('_')
				lastUnderscore = true
			}
		}
	}
	out := strings.Trim(b.String(), "_")
	if out == "" {
		return "default"
	}
	return out
}

func sourceBackedLayoutIDHash(objectAPIName, layoutName string) string {
	sum := sha1.Sum([]byte(strings.TrimSpace(objectAPIName) + "\n" + strings.TrimSpace(layoutName)))
	return hex.EncodeToString(sum[:])[:sourceBackedLayoutIDHashLen]
}

func isSourceBackedDefaultLayout(layout models.PageLayout) bool {
	return layout.IsDefault
}

func newSourceBackedReportTypeComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, error) {
	parsed, err := sourceBackedReportTypesFromPayload(payload, componentPayloadLabel(component))
	if err != nil {
		return nil, err
	}
	return newReportTypeManagedComponentFromPayload(packName, component, payload, parsed), nil
}

// newReportTypeManagedComponentFromPayload builds the managed component from an
// already-parsed/canonicalized sourceBackedReportTypesPayload plus the RAW
// payload bytes and the engine ManifestComponent. It is shared by the internal
// build path (newSourceBackedReportTypeComponent) and the engine implementation
// of cp.PresentationComponentBuildContext.AddReportTypeComponent so both paths
// produce a byte-identical component (raw payload, never re-marshalled) with the
// same apply/verify closures over the same parsed seeds.
func newReportTypeManagedComponentFromPayload(packName string, component ManifestComponent, payload []byte, parsed sourceBackedReportTypesPayload) PackComponent {
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return applySourceBackedReportTypes(ctx, deps.DB, parsed, packName)
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			return verifySourceBackedReportTypes(ctx, deps.DB, parsed, packName)
		},
	}
}

type sourceBackedReportTypesPayload struct {
	ReportTypes []sourceBackedReportTypeSeed `json:"reportTypes"`
}

type sourceBackedReportTypePackBatch struct {
	declaredFullNames []string
}

func (b *sourceBackedReportTypePackBatch) append(payload sourceBackedReportTypesPayload) {
	if b == nil {
		return
	}
	for _, reportType := range payload.ReportTypes {
		b.declaredFullNames = append(b.declaredFullNames, reportType.FullName)
	}
}

type sourceBackedReportTypeSeed struct {
	FullName      string          `json:"fullName"`
	Label         string          `json:"label"`
	Description   string          `json:"description,omitempty"`
	BaseObject    string          `json:"baseObject"`
	Deployed      bool            `json:"deployed"`
	Autogenerated bool            `json:"autogenerated"`
	Sections      json.RawMessage `json:"sections,omitempty"`
	Categories    json.RawMessage `json:"categories,omitempty"`
	Join          json.RawMessage `json:"join,omitempty"`
	Metadata      json.RawMessage `json:"metadata,omitempty"`
}

func sourceBackedReportTypesFromPayload(payload []byte, source string) (sourceBackedReportTypesPayload, error) {
	var parsed sourceBackedReportTypesPayload
	if err := json.Unmarshal(payload, &parsed); err != nil {
		return parsed, fmt.Errorf("parse %s: %w", source, err)
	}
	seen := make(map[string]struct{}, len(parsed.ReportTypes))
	for i := range parsed.ReportTypes {
		rt := &parsed.ReportTypes[i]
		rt.FullName = strings.TrimSpace(rt.FullName)
		rt.Label = strings.TrimSpace(rt.Label)
		rt.BaseObject = strings.TrimSpace(rt.BaseObject)
		if rt.FullName == "" {
			return parsed, fmt.Errorf("report type at index %d missing fullName", i)
		}
		if rt.Label == "" {
			rt.Label = rt.FullName
		}
		if rt.BaseObject == "" {
			return parsed, fmt.Errorf("report type %s missing baseObject", rt.FullName)
		}
		key := strings.ToLower(rt.FullName)
		if _, ok := seen[key]; ok {
			return parsed, fmt.Errorf("duplicate report type %s", rt.FullName)
		}
		seen[key] = struct{}{}
	}
	sort.Slice(parsed.ReportTypes, func(i, j int) bool {
		return parsed.ReportTypes[i].FullName < parsed.ReportTypes[j].FullName
	})
	return parsed, nil
}

func applySourceBackedReportTypes(ctx context.Context, db *sql.DB, payload sourceBackedReportTypesPayload, ownerPack string) error {
	now := time.Now().UTC()
	for _, rt := range payload.ReportTypes {
		if err := assertSourceBackedReportTypeOwner(ctx, db, rt.FullName, ownerPack); err != nil {
			return err
		}
		if _, err := db.ExecContext(ctx, `
			INSERT INTO _System_ReportType
				(Id, FullName, Label, Description, BaseObject, Deployed, Autogenerated, Sections, Categories, `+"`Join`"+`, Metadata, OwnerPack, CreatedDate, LastModifiedDate, IsDeleted)
			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0)
			ON DUPLICATE KEY UPDATE
				Label = VALUES(Label),
				Description = VALUES(Description),
				BaseObject = VALUES(BaseObject),
				Deployed = VALUES(Deployed),
				Autogenerated = VALUES(Autogenerated),
				Sections = VALUES(Sections),
				Categories = VALUES(Categories),
				`+"`Join`"+` = VALUES(`+"`Join`"+`),
				Metadata = VALUES(Metadata),
				OwnerPack = VALUES(OwnerPack),
				LastModifiedDate = VALUES(LastModifiedDate),
				IsDeleted = 0`,
			sourceBackedReportTypeID(ownerPack, rt.FullName),
			rt.FullName,
			rt.Label,
			sourceBackedNullableString(rt.Description),
			rt.BaseObject,
			sourceBackedBoolInt(rt.Deployed),
			sourceBackedBoolInt(rt.Autogenerated),
			sourceBackedNullableRawJSON(rt.Sections),
			sourceBackedNullableRawJSON(rt.Categories),
			sourceBackedNullableRawJSON(rt.Join),
			sourceBackedNullableRawJSON(rt.Metadata),
			ownerPack,
			now,
			now,
		); err != nil {
			return fmt.Errorf("upsert report type %s: %w", rt.FullName, err)
		}
	}
	return nil
}

func verifySourceBackedReportTypes(ctx context.Context, db *sql.DB, payload sourceBackedReportTypesPayload, ownerPack string) error {
	for _, want := range payload.ReportTypes {
		var (
			label         string
			description   sql.NullString
			baseObject    string
			deployed      int
			autogenerated int
			sectionsRaw   sql.NullString
			categoriesRaw sql.NullString
			joinRaw       sql.NullString
			metadataRaw   sql.NullString
			gotOwner      sql.NullString
		)
		err := db.QueryRowContext(ctx, `
			SELECT Label, Description, BaseObject, Deployed, Autogenerated, Sections, Categories, `+"`Join`"+`, Metadata, OwnerPack
			FROM _System_ReportType
			WHERE FullName = ? AND COALESCE(IsDeleted, 0) = 0
			LIMIT 1`, want.FullName).Scan(&label, &description, &baseObject, &deployed, &autogenerated, &sectionsRaw, &categoriesRaw, &joinRaw, &metadataRaw, &gotOwner)
		if err == sql.ErrNoRows {
			return fmt.Errorf("report type %s missing", want.FullName)
		}
		if err != nil {
			return fmt.Errorf("query report type %s: %w", want.FullName, err)
		}
		if label != want.Label || sourceBackedNullStringValue(description) != want.Description || baseObject != want.BaseObject ||
			deployed != sourceBackedBoolInt(want.Deployed) || autogenerated != sourceBackedBoolInt(want.Autogenerated) ||
			!gotOwner.Valid || gotOwner.String != ownerPack {
			return fmt.Errorf("report type %s drifted", want.FullName)
		}
		if !sourceBackedJSONEqualString(metadataRaw, want.Metadata) {
			return fmt.Errorf("report type %s metadata drifted", want.FullName)
		}
		if !sourceBackedJSONEqualString(sectionsRaw, want.Sections) {
			return fmt.Errorf("report type %s sections drifted", want.FullName)
		}
		if !sourceBackedJSONEqualString(categoriesRaw, want.Categories) {
			return fmt.Errorf("report type %s categories drifted", want.FullName)
		}
		if !sourceBackedJSONEqualString(joinRaw, want.Join) {
			return fmt.Errorf("report type %s join drifted", want.FullName)
		}
	}
	return nil
}

func newSourceBackedReportTypePackSweeperComponent(packName string, batch *sourceBackedReportTypePackBatch) PackComponent {
	manifestComponent := ManifestComponent{
		Kind:          ComponentKindReportType,
		BusinessKey:   JoinComponentBusinessKey(packName, "@sweeper"),
		Manageability: ManageabilityLocked,
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: manifestComponent,
		apply: func(ctx context.Context, deps ApplyDeps) error {
			stale, err := queryStaleSourceBackedReportTypes(ctx, deps.DB, batch, packName)
			if err != nil {
				return err
			}
			now := time.Now().UTC()
			for _, fullName := range stale {
				if _, err := deps.DB.ExecContext(ctx, `
					UPDATE _System_ReportType
					SET LastModifiedDate = ?, IsDeleted = 1
					WHERE FullName = ? AND OwnerPack = ? AND COALESCE(IsDeleted, 0) = 0`, now, fullName, packName); err != nil {
					return fmt.Errorf("soft-delete stale report type %s owned by %s: %w", fullName, packName, err)
				}
			}
			return nil
		},
		verify: func(ctx context.Context, deps VerifyDeps) error {
			stale, err := queryStaleSourceBackedReportTypes(ctx, deps.DB, batch, packName)
			if err != nil {
				return err
			}
			if len(stale) > 0 {
				return fmt.Errorf("report type stale owner rows for pack %s: %s", packName, strings.Join(stale, ", "))
			}
			return nil
		},
	}
}

func queryStaleSourceBackedReportTypes(ctx context.Context, db *sql.DB, batch *sourceBackedReportTypePackBatch, ownerPack string) ([]string, error) {
	if db == nil {
		return nil, fmt.Errorf("database is required")
	}
	expected := map[string]struct{}{}
	if batch != nil {
		for _, fullName := range batch.declaredFullNames {
			expected[strings.ToLower(strings.TrimSpace(fullName))] = struct{}{}
		}
	}
	rows, err := db.QueryContext(ctx, `
		SELECT FullName
		FROM _System_ReportType
		WHERE OwnerPack = ? AND COALESCE(IsDeleted, 0) = 0
		ORDER BY FullName`, ownerPack)
	if err != nil {
		return nil, fmt.Errorf("query report types owned by %s for reconciliation: %w", ownerPack, err)
	}
	defer rows.Close()
	var stale []string
	for rows.Next() {
		var fullName string
		if err := rows.Scan(&fullName); err != nil {
			return nil, fmt.Errorf("scan report type owned by %s for reconciliation: %w", ownerPack, err)
		}
		if _, ok := expected[strings.ToLower(strings.TrimSpace(fullName))]; !ok {
			stale = append(stale, fullName)
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate report types owned by %s for reconciliation: %w", ownerPack, err)
	}
	return stale, nil
}

func assertSourceBackedReportTypeOwner(ctx context.Context, db *sql.DB, fullName string, ownerPack string) error {
	var owner sql.NullString
	err := db.QueryRowContext(ctx, `
		SELECT OwnerPack
		FROM _System_ReportType
		WHERE FullName = ? AND COALESCE(IsDeleted, 0) = 0
		LIMIT 1`, fullName).Scan(&owner)
	if err == sql.ErrNoRows {
		return nil
	}
	if err != nil {
		return fmt.Errorf("query report type owner %s: %w", fullName, err)
	}
	if owner.Valid && strings.TrimSpace(owner.String) != "" && owner.String != ownerPack {
		return fmt.Errorf("report type %s is owned by %s, refusing to overwrite with %s", fullName, owner.String, ownerPack)
	}
	return nil
}

func sourceBackedReportTypeID(ownerPack, fullName string) string {
	id := strings.ToLower(ownerPack + "_" + fullName)
	var b strings.Builder
	b.WriteString("reporttype_")
	for _, r := range id {
		switch {
		case r >= 'a' && r <= 'z', r >= '0' && r <= '9':
			b.WriteRune(r)
		default:
			b.WriteByte('_')
		}
	}
	return b.String()
}

func newSourceBackedUIActionComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, error) {
	source := componentPayloadLabel(component)
	actions, err := parseStandardActionPayload(payload, source)
	if err != nil {
		return nil, err
	}
	if err := validateStandardActionPayload(actions, packName, source, true); err != nil {
		return nil, err
	}
	expanded, err := expandStandardActionRows(actions, packName)
	if err != nil {
		return nil, err
	}
	// Expose the parsed action rows (INCLUDING Label) as the component's
	// projection. The raw-byte PayloadHash cannot see a parsed-then-reconstructed
	// drift; ExpectedProjection() makes that drift observable to a deep-compare.
	// The four black-box oracle surfaces never read ExpectedProjection, so adding
	// it moves no golden fixture.
	return newUIActionManagedComponentFromRows(packName, component, payload, sortedStandardActionRows(expanded)), nil
}

func validateStandardActionPayloadFromRaw(payload []byte, ownerPack string, source string, allowAppliesTo bool) error {
	actions, err := parseStandardActionPayload(payload, source)
	if err != nil {
		return err
	}
	return validateStandardActionPayload(actions, ownerPack, source, allowAppliesTo)
}

func sourceBackedFlowFileName(component ManifestComponent) string {
	if strings.TrimSpace(component.Path) != "" {
		return filepath.Base(strings.TrimSpace(component.Path))
	}
	return strings.TrimSpace(component.BusinessKey) + ".flow.json"
}

func sourceBackedNullStringPtr(value sql.NullString) *string {
	if !value.Valid {
		return nil
	}
	v := strings.TrimSpace(value.String)
	if v == "" {
		return nil
	}
	return &v
}

func sourceBackedNormalizeStringSlice(values []string) []string {
	if len(values) == 0 {
		return nil
	}
	return append([]string(nil), values...)
}

func sourceBackedBoolInt(value bool) int {
	if value {
		return 1
	}
	return 0
}

func sourceBackedNullableString(value string) any {
	if strings.TrimSpace(value) == "" {
		return nil
	}
	return value
}

func sourceBackedNullableRawJSON(raw json.RawMessage) any {
	if len(raw) == 0 || string(raw) == "null" {
		return nil
	}
	return string(raw)
}

func sourceBackedNullStringValue(value sql.NullString) string {
	if !value.Valid {
		return ""
	}
	return value.String
}

func sourceBackedJSONEqualString(got sql.NullString, want json.RawMessage) bool {
	if len(want) == 0 || string(want) == "null" {
		return !got.Valid || strings.TrimSpace(got.String) == ""
	}
	if !got.Valid {
		return false
	}
	gotCanonical, err := sourceBackedCanonicalJSON([]byte(got.String))
	if err != nil {
		return false
	}
	wantCanonical, err := sourceBackedCanonicalJSON(want)
	if err != nil {
		return false
	}
	return bytes.Equal(gotCanonical, wantCanonical)
}

func sourceBackedCanonicalJSON(raw []byte) ([]byte, error) {
	if err := rejectDuplicateJSONKeys(raw, "source_backed_metadata"); err != nil {
		return nil, fmt.Errorf("source_backed_metadata canonical JSON: %w", err)
	}
	dec := json.NewDecoder(bytes.NewReader(raw))
	dec.UseNumber()
	var decoded any
	if err := dec.Decode(&decoded); err != nil {
		return nil, err
	}
	if err := ensureJSONEOF(dec); err != nil {
		return nil, err
	}
	return json.Marshal(decoded)
}

// newSourceBackedWebhookSecretComponent parses a single webhook secret
// definition from the manifest component payload and returns a PackComponent
// that delegates apply/projection logic to the existing helpers in
// webhook_secret_component.go. Each manifest entry is one secret (not an
// envelope of multiple secrets).
func newSourceBackedWebhookSecretComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, error) {
	var def webhookSecretDefinition
	if err := json.Unmarshal(payload, &def); err != nil {
		return nil, fmt.Errorf("parse webhook_secret component %s: %w", componentPayloadLabel(component), err)
	}
	// Normalize at parse time so the BusinessKey comparison uses canonical values
	// and the inner component stores lowercase-trimmed identifiers throughout.
	normalizedPackName := strings.ToLower(strings.TrimSpace(packName))
	def.Name = strings.ToLower(strings.TrimSpace(def.Name))
	expectedBusinessKey := JoinComponentBusinessKey(normalizedPackName, def.Name)
	if strings.ToLower(strings.TrimSpace(component.BusinessKey)) != expectedBusinessKey {
		return nil, fmt.Errorf(
			"webhook_secret manifest businessKey %q does not match payload identity %q",
			component.BusinessKey,
			expectedBusinessKey,
		)
	}
	inner, err := NewWebhookSecretComponent(normalizedPackName, def)
	if err != nil {
		return nil, fmt.Errorf("webhook_secret component %s/%s: %w", packName, def.Name, err)
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		expectedProjection: func() (any, error) {
			return inner.ExpectedProjection()
		},
		liveProjection: func(ctx context.Context, deps VerifyDeps) (any, error) {
			return inner.LiveProjection(ctx, deps)
		},
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return inner.Apply(ctx, deps)
		},
	}, nil
}

// newSourceBackedConnectorCalloutProfileComponent parses a single
// connector_callout_profile component payload and wraps it in a
// sourceBackedManagedComponent.
func newSourceBackedConnectorCalloutProfileComponent(packName string, component ManifestComponent, payload []byte, allowLegacyMissingBatchIDResponseField bool) (PackComponent, error) {
	packComponent, _, err := newSourceBackedConnectorCalloutProfileComponentWithDefinition(
		packName,
		component,
		payload,
		allowLegacyMissingBatchIDResponseField,
	)
	return packComponent, err
}

func newSourceBackedConnectorCalloutProfileComponentWithDefinition(
	packName string,
	component ManifestComponent,
	payload []byte,
	allowLegacyMissingBatchIDResponseField bool,
) (PackComponent, connectorCalloutProfileDefinition, error) {
	var def connectorCalloutProfileDefinition
	if err := json.Unmarshal(payload, &def); err != nil {
		return nil, connectorCalloutProfileDefinition{}, fmt.Errorf("connector_callout_profile payload unmarshal: %w", err)
	}
	normalizedPackName := strings.TrimSpace(packName)
	def.ProfileName = strings.ToLower(strings.TrimSpace(def.ProfileName))
	expectedBusinessKey := JoinComponentBusinessKey(normalizedPackName, def.ProfileName)
	if strings.TrimSpace(component.BusinessKey) != expectedBusinessKey {
		return nil, connectorCalloutProfileDefinition{}, fmt.Errorf(
			"connector_callout_profile manifest businessKey %q does not match payload identity %q",
			component.BusinessKey,
			expectedBusinessKey,
		)
	}
	var opts []connectorCalloutProfileComponentOption
	if allowLegacyMissingBatchIDResponseField {
		opts = append(opts, withLegacyMissingBatchIDResponseFieldCompat())
	}
	inner, err := newConnectorCalloutProfileComponent(normalizedPackName, def, opts...)
	if err != nil {
		return nil, connectorCalloutProfileDefinition{}, fmt.Errorf("connector_callout_profile component %s/%s: %w", packName, def.ProfileName, err)
	}
	profileComponent, ok := inner.(*connectorCalloutProfileComponent)
	if !ok {
		return nil, connectorCalloutProfileDefinition{}, fmt.Errorf(
			"connector_callout_profile component %s/%s returned %T",
			packName,
			def.ProfileName,
			inner,
		)
	}
	normalizedDefinition := profileComponent.definition
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		expectedProjection: func() (any, error) {
			return inner.ExpectedProjection()
		},
		liveProjection: func(ctx context.Context, deps VerifyDeps) (any, error) {
			return inner.LiveProjection(ctx, deps)
		},
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return inner.Apply(ctx, deps)
		},
	}, normalizedDefinition, nil
}

// newSourceBackedConnectorRateLimitComponent parses a single connector_rate_limit
// component payload and wraps it in a sourceBackedManagedComponent.
func newSourceBackedConnectorRateLimitComponent(packName string, component ManifestComponent, payload []byte) (PackComponent, error) {
	var def connectorRateLimitDefinition
	if err := json.Unmarshal(payload, &def); err != nil {
		return nil, fmt.Errorf("connector_rate_limit payload unmarshal: %w", err)
	}
	// Normalize identifiers to lowercase to match the canonical storage form;
	// and the inner component stores lowercase-trimmed identifiers throughout.
	normalizedPackName := strings.ToLower(strings.TrimSpace(packName))
	def.EndpointClass = strings.ToLower(strings.TrimSpace(def.EndpointClass))
	expectedBusinessKey := JoinComponentBusinessKey(normalizedPackName, def.EndpointClass)
	if strings.ToLower(strings.TrimSpace(component.BusinessKey)) != expectedBusinessKey {
		return nil, fmt.Errorf(
			"connector_rate_limit manifest businessKey %q does not match payload identity %q",
			component.BusinessKey,
			expectedBusinessKey,
		)
	}
	inner, err := NewConnectorRateLimitComponent(normalizedPackName, def)
	if err != nil {
		return nil, fmt.Errorf("connector_rate_limit component %s/%s: %w", packName, def.EndpointClass, err)
	}
	return &sourceBackedManagedComponent{
		packName:          packName,
		manifestComponent: component,
		payload:           payload,
		expectedProjection: func() (any, error) {
			return inner.ExpectedProjection()
		},
		liveProjection: func(ctx context.Context, deps VerifyDeps) (any, error) {
			return inner.LiveProjection(ctx, deps)
		},
		apply: func(ctx context.Context, deps ApplyDeps) error {
			return inner.Apply(ctx, deps)
		},
	}, nil
}