package automation

import (
	"context"
	"fmt"
	"sort"
	"strings"

	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

type connectorInboundRegisterInstalledBinding struct {
	Generation        string
	Identity          ConnectorPushDurableIdentity
	FlowDeveloperName string
	FlowActionName    string
}

func (e *FlowExecutor) ResolveConnectorInboundRemoteIdentityRecoveryInstalledBindings(
	ctx context.Context,
	selector ConnectorInboundRemoteIdentityRecoverySelector,
) (ConnectorInboundRemoteIdentityRecoveryInstalledBindings, error) {
	selector, err := normalizeConnectorInboundRemoteIdentityRecoverySelector(selector)
	if err != nil {
		return ConnectorInboundRemoteIdentityRecoveryInstalledBindings{}, err
	}
	set, err := e.resolveConnectorInboundLockedInstalledContractSet(ctx, selector.ConnectorOwnerKey)
	if err != nil {
		return ConnectorInboundRemoteIdentityRecoveryInstalledBindings{}, err
	}
	bindings := make([]ConnectorInboundRemoteIdentityRecoveryBinding, 0)
	for i := range set.Contracts {
		contract := &set.Contracts[i]
		if strings.TrimSpace(contract.ConnectorOwnerKey) != selector.ConnectorOwnerKey ||
			strings.TrimSpace(contract.DestinationSlotID) != selector.DestinationSlotID {
			continue
		}
		if err := validateConnectorInboundInstalledContract(contract); err != nil {
			return ConnectorInboundRemoteIdentityRecoveryInstalledBindings{}, err
		}
		profileOwnerPack := strings.TrimSpace(contract.ProfileOwnerPack)
		if profileOwnerPack == "" {
			profileOwnerPack = strings.TrimSpace(contract.ConnectorOwnerKey)
		}
		expectedProfileInstallID := strings.TrimSpace(set.LockedPackInstallIDs[profileOwnerPack])
		if expectedProfileInstallID == "" {
			return ConnectorInboundRemoteIdentityRecoveryInstalledBindings{}, fmt.Errorf("locked installed recovery generation is missing profile owner %q", profileOwnerPack)
		}
		mappings, profile, err := e.resolveConnectorInboundInstalledSnapshotInputs(ctx, set.Mappings, contract, expectedProfileInstallID)
		if err != nil {
			return ConnectorInboundRemoteIdentityRecoveryInstalledBindings{}, err
		}
		identity := connectorPushDurableIdentityFromContract(contract, "")
		bindings = append(bindings, ConnectorInboundRemoteIdentityRecoveryBinding{
			EvidenceBinding: ConnectorInboundRemoteIDEvidenceBinding{
				SourceObject:  strings.TrimSpace(contract.SourceObject),
				RemoteIDField: strings.TrimSpace(contract.RemoteIDField),
			},
			Identity: identity,
			Snapshot: connectorPushDurableSnapshotFromMetadata(FlowExecutionIdentity{
				OwnerPack: strings.TrimSpace(contract.ConnectorOwnerKey),
			}, contract, mappings, profile, ""),
		})
	}
	if len(bindings) == 0 {
		return ConnectorInboundRemoteIdentityRecoveryInstalledBindings{}, fmt.Errorf("no installed connector contract owns destination %q", selector.DestinationSlotID)
	}
	return ConnectorInboundRemoteIdentityRecoveryInstalledBindings{
		Generation: set.Generation,
		Bindings:   bindings,
	}, nil
}

func (e *FlowExecutor) resolveConnectorInboundRegisterInstalledBinding(
	ctx context.Context,
	connectorOwnerKey, sourceObject, remoteIDField string,
) (connectorInboundRegisterInstalledBinding, error) {
	set, err := e.resolveConnectorInboundCurrentInstalledContractSet(ctx, connectorOwnerKey)
	if err != nil {
		return connectorInboundRegisterInstalledBinding{}, err
	}
	return selectConnectorInboundInstalledBinding(set, connectorOwnerKey, sourceObject, remoteIDField)
}

func selectConnectorInboundInstalledBinding(
	set ConnectorInboundRemoteIdentityInstalledContractSet,
	connectorOwnerKey, sourceObject, remoteIDField string,
) (connectorInboundRegisterInstalledBinding, error) {
	var matched *connector.ConnectorPushDurableMetadataContract
	for i := range set.Contracts {
		contract := &set.Contracts[i]
		if strings.TrimSpace(contract.ConnectorOwnerKey) != strings.TrimSpace(connectorOwnerKey) ||
			strings.TrimSpace(contract.SourceObject) != strings.TrimSpace(sourceObject) ||
			strings.TrimSpace(contract.RemoteIDField) != strings.TrimSpace(remoteIDField) {
			continue
		}
		if err := validateConnectorInboundInstalledContract(contract); err != nil {
			return connectorInboundRegisterInstalledBinding{}, err
		}
		if matched != nil {
			return connectorInboundRegisterInstalledBinding{}, fmt.Errorf("installed connector contract is ambiguous for %s/%s", sourceObject, remoteIDField)
		}
		matched = contract
	}
	if matched == nil {
		return connectorInboundRegisterInstalledBinding{}, fmt.Errorf("installed connector contract not found for %s/%s", sourceObject, remoteIDField)
	}
	return connectorInboundRegisterInstalledBinding{
		Generation:        set.Generation,
		Identity:          connectorPushDurableIdentityFromContract(matched, ""),
		FlowDeveloperName: strings.TrimSpace(matched.FlowDeveloperName),
		FlowActionName:    strings.TrimSpace(matched.FlowActionName),
	}, nil
}

// resolveConnectorIntegrationOutboundScope binds a transfer to the same
// installation and Integration workload that Catalog authorizes to create the
// outbound identity. A self-consistent persisted command is not scope authority.
func resolveConnectorIntegrationOutboundScope(ctx context.Context, binding connectorInboundRegisterInstalledBinding) (connectorIntegrationOutboundScope, error) {
	view, ok := metadata.RuntimeMetadataViewFromContext(ctx)
	if !ok {
		// Legacy callers can still transfer canonical V1 identities. The store
		// rejects V2 identities when this independently derived scope is absent.
		return connectorIntegrationOutboundScope{}, nil
	}
	if view.Profile() != catalogsnapshot.ProjectionRecordMutationV1 {
		return connectorIntegrationOutboundScope{}, fmt.Errorf("%w: remote identity transfer requires an exact record-mutation Catalog view", ErrConnectorPushDurableConflict)
	}
	identity := binding.Identity
	var matched []catalogsnapshot.OutboundIntentRoute
	for _, route := range view.OutboundIntentRoutes() {
		if (route.Operation == "" || route.Operation == outboundintentv1.OperationUpsert) &&
			identity.ActionType == constants.ActionTypePushToConnector &&
			route.ConnectorOwnerKey == identity.ConnectorOwnerKey &&
			route.ActionSlotID == identity.ActionSlotID &&
			route.SourceObject == identity.SourceObject &&
			route.DestinationSlotID == identity.DestinationSlotID &&
			route.FlowDeveloperName == binding.FlowDeveloperName &&
			route.FlowActionName == binding.FlowActionName {
			matched = append(matched, route)
		}
	}
	if len(matched) != 1 {
		return connectorIntegrationOutboundScope{}, fmt.Errorf("%w: remote identity transfer exact outbound route count is %d, want 1", ErrConnectorPushDurableConflict, len(matched))
	}
	scope := connectorIntegrationOutboundScope{
		orgID: view.OrgID(), appInstallationID: view.AppInstallationID(), authorizedWorkloadID: matched[0].AuthorizedWorkloadID,
	}
	if strings.TrimSpace(scope.orgID) == "" || strings.TrimSpace(scope.appInstallationID) == "" || strings.TrimSpace(scope.authorizedWorkloadID) == "" {
		return connectorIntegrationOutboundScope{}, fmt.Errorf("%w: remote identity transfer Catalog scope is incomplete", ErrConnectorPushDurableConflict)
	}
	return scope, nil
}

func (e *FlowExecutor) resolveConnectorRemoteIdentityTransferAdmissions(
	ctx context.Context,
	transfer ConnectorRemoteIdentityTransfer,
	hints []ConnectorRemoteIdentityTransferHint,
) ([]connectorRemoteIdentityTransferAdmission, error) {
	admissions := make(map[string]connectorRemoteIdentityTransferAdmission)
	ownerSet := make(map[string]struct{})
	remoteID := strings.TrimSpace(transfer.RemoteId)
	for _, hint := range hints {
		owner := strings.TrimSpace(hint.ConnectorOwnerKey)
		hintRemoteID := strings.TrimSpace(hint.RemoteID)
		blockingAttempt := hint.BlockingAttemptIdentity != (ConnectorPushDurableIdentity{})
		if owner == "" || strings.TrimSpace(hint.ConnectorNamespace) == "" || (hintRemoteID == "" && !blockingAttempt) {
			return nil, fmt.Errorf("connector remote identity transfer hint is incomplete")
		}
		if remoteID == "" && hintRemoteID != "" {
			remoteID = hintRemoteID
		}
		if hintRemoteID != "" && remoteID != hintRemoteID {
			return nil, fmt.Errorf("%w: transfer hints contain conflicting remote ids", ErrConnectorPushDurableConflict)
		}
		ownerSet[owner] = struct{}{}
	}
	owners := make([]string, 0, len(ownerSet))
	for owner := range ownerSet {
		owners = append(owners, owner)
	}
	sort.Strings(owners)
	if len(owners) == 0 {
		return nil, nil
	}
	sets, err := e.resolveConnectorInboundLockedInstalledContractSets(ctx, owners)
	if err != nil {
		return nil, fmt.Errorf("resolve transfer installed contracts: %w", err)
	}
	for _, hint := range hints {
		owner := strings.TrimSpace(hint.ConnectorOwnerKey)
		hintRemoteID := strings.TrimSpace(hint.RemoteID)
		set, ok := sets[owner]
		if !ok {
			return nil, fmt.Errorf("resolve transfer installed contracts for %s: locked set is missing", owner)
		}
		source, err := selectConnectorInboundInstalledBinding(set, owner, transfer.SourceObject, transfer.RemoteIDField)
		if err != nil {
			return nil, fmt.Errorf("resolve transfer source binding for %s: %w", owner, err)
		}
		target, err := selectConnectorInboundInstalledBinding(set, owner, transfer.TargetObject, transfer.RemoteIDField)
		if err != nil {
			return nil, fmt.Errorf("resolve transfer target binding for %s: %w", owner, err)
		}
		source.Identity.SourceRecordID = strings.TrimSpace(transfer.SourceRecordId)
		target.Identity.SourceRecordID = strings.TrimSpace(transfer.TargetRecordId)
		sourceScope, err := resolveConnectorIntegrationOutboundScope(ctx, source)
		if err != nil {
			return nil, fmt.Errorf("resolve transfer source scope for %s: %w", owner, err)
		}
		targetScope, err := resolveConnectorIntegrationOutboundScope(ctx, target)
		if err != nil {
			return nil, fmt.Errorf("resolve transfer target scope for %s: %w", owner, err)
		}
		if source.Identity.DestinationSlotID != target.Identity.DestinationSlotID {
			return nil, fmt.Errorf("%w: transfer source and target bindings do not share one remote destination", ErrConnectorPushDurableConflict)
		}
		blockingAttempt := hint.BlockingAttemptIdentity != (ConnectorPushDurableIdentity{})
		if blockingAttempt && hint.BlockingAttemptIdentity != source.Identity && hint.BlockingAttemptIdentity != target.Identity {
			return nil, fmt.Errorf("%w: unresolved create attempt does not match the current installed source or target identity", ErrConnectorPushDurableConflict)
		}
		if destination := strings.TrimSpace(hint.DestinationSlotID); destination != "" && destination != source.Identity.DestinationSlotID {
			return nil, fmt.Errorf("%w: active transfer authority does not match the current installed destination", ErrConnectorPushDurableConflict)
		}
		targetSnapshot, err := e.resolveConnectorRemoteIdentityTransferTargetSnapshot(ctx, set, target, transfer.RemoteIDField)
		if err != nil {
			return nil, fmt.Errorf("resolve transfer target snapshot for %s: %w", owner, err)
		}
		candidate := connectorRemoteIdentityTransferAdmission{
			ConnectorNamespace:     strings.TrimSpace(hint.ConnectorNamespace),
			RemoteID:               hintRemoteID,
			SourceIdentity:         source.Identity,
			TargetIdentity:         target.Identity,
			TargetSnapshot:         targetSnapshot,
			SourceIntegrationScope: sourceScope,
			TargetIntegrationScope: targetScope,
			UnresolvedCreateBlock:  blockingAttempt,
		}
		if existing, ok := admissions[owner]; ok {
			if existing.ConnectorNamespace != candidate.ConnectorNamespace ||
				!connectorPushPersistedRemoteLinkSnapshotEqual(existing.TargetSnapshot, candidate.TargetSnapshot) ||
				existing.SourceIdentity != candidate.SourceIdentity ||
				existing.TargetIdentity != candidate.TargetIdentity ||
				existing.SourceIntegrationScope != candidate.SourceIntegrationScope ||
				existing.TargetIntegrationScope != candidate.TargetIntegrationScope ||
				(existing.RemoteID != "" && candidate.RemoteID != "" && existing.RemoteID != candidate.RemoteID) {
				return nil, fmt.Errorf("%w: transfer has ambiguous installed admissions for owner %s", ErrConnectorPushDurableConflict, owner)
			}
			if existing.RemoteID == "" {
				existing.RemoteID = candidate.RemoteID
			}
			existing.UnresolvedCreateBlock = existing.UnresolvedCreateBlock || candidate.UnresolvedCreateBlock
			admissions[owner] = existing
			continue
		}
		admissions[owner] = candidate
	}
	result := make([]connectorRemoteIdentityTransferAdmission, 0, len(admissions))
	for _, admission := range admissions {
		result = append(result, admission)
	}
	sort.Slice(result, func(i, j int) bool {
		leftLinkKey, _ := result[i].SourceIdentity.LinkKey()
		rightLinkKey, _ := result[j].SourceIdentity.LinkKey()
		left := result[i].RemoteID + "\x00" + leftLinkKey
		right := result[j].RemoteID + "\x00" + rightLinkKey
		return left < right
	})
	return result, nil
}

// A protected conversion publishes target durable authority, not inbound
// authorship. It may execute before any inbound event, so its target snapshot
// must come from installed metadata and must not invent connector provenance.
func (e *FlowExecutor) resolveConnectorRemoteIdentityTransferTargetSnapshot(
	ctx context.Context,
	set ConnectorInboundRemoteIdentityInstalledContractSet,
	target connectorInboundRegisterInstalledBinding,
	remoteIDField string,
) (ConnectorPushDurableSnapshot, error) {
	if view, ok := metadata.RuntimeMetadataViewFromContext(ctx); ok &&
		!connectorInboundInstalledGenerationMatchesBoundView(view, set.Generation, target, remoteIDField) {
		return ConnectorPushDurableSnapshot{}, fmt.Errorf("%w: target installed contracts differ from the bound Catalog generation", ErrConnectorPushDurableConflict)
	}
	owner := target.Identity.ConnectorOwnerKey
	var contract *connector.ConnectorPushDurableMetadataContract
	var mappings []connector.ConnectorFieldMappingInput
	var profile *ConnectorPushProfile
	var err error
	if e.connectorInboundFinalizer != nil {
		contract, mappings, profile, err = e.resolveConnectorInboundFinalizeAuthority(ctx,
			FlowExecutionIdentity{OwnerPack: owner}, connectorInboundFinalizeSelectors{
				targetFlowDeveloperName: target.FlowDeveloperName,
				targetFlowActionName:    target.FlowActionName,
			})
		if err != nil {
			return ConnectorPushDurableSnapshot{}, err
		}
	} else {
		// Legacy composition uses the same generation-locked contract/profile
		// resolver as recovery. Independent Runtime uses the immutable closure above.
		for i := range set.Contracts {
			candidate := &set.Contracts[i]
			if connectorPushDurableIdentityFromContract(candidate, target.Identity.SourceRecordID) == target.Identity &&
				candidate.FlowDeveloperName == target.FlowDeveloperName && candidate.FlowActionName == target.FlowActionName {
				contract = candidate
				break
			}
		}
		if contract == nil {
			return ConnectorPushDurableSnapshot{}, fmt.Errorf("%w: target installed contract is missing", ErrConnectorPushDurableConflict)
		}
		profileOwner := strings.TrimSpace(contract.ProfileOwnerPack)
		if profileOwner == "" {
			profileOwner = owner
		}
		mappings, profile, err = e.resolveConnectorInboundInstalledSnapshotInputs(ctx, set.Mappings, contract, set.LockedPackInstallIDs[profileOwner])
		if err != nil {
			return ConnectorPushDurableSnapshot{}, err
		}
	}
	if connectorPushDurableIdentityFromContract(contract, target.Identity.SourceRecordID) != target.Identity ||
		strings.TrimSpace(contract.RemoteIDField) != strings.TrimSpace(remoteIDField) ||
		contract.FlowDeveloperName != target.FlowDeveloperName || contract.FlowActionName != target.FlowActionName {
		return ConnectorPushDurableSnapshot{}, fmt.Errorf("%w: target finalization authority differs from installed binding", ErrConnectorPushDurableConflict)
	}
	if generation := strings.TrimSpace(set.LockedPackInstallIDs[strings.TrimSpace(profile.OwnerPack)]); generation == "" ||
		generation != strings.TrimSpace(profile.CreatedPackInstallID) {
		return ConnectorPushDurableSnapshot{}, fmt.Errorf("%w: target profile incarnation changed after generation lock", ErrConnectorPushDurableConflict)
	}
	return connectorPushDurableSnapshotFromMetadata(FlowExecutionIdentity{OwnerPack: owner}, contract, mappings, profile, ""), nil
}

func connectorInboundInstalledGenerationMatchesBoundView(
	view *catalogsnapshot.RuntimeMetadataView,
	generation string,
	target connectorInboundRegisterInstalledBinding,
	remoteIDField string,
) bool {
	if view == nil {
		return false
	}
	generation = strings.TrimSpace(generation)
	if generation == "" {
		return false
	}
	// Runtime Worker identifies the complete immutable installed set by its
	// bound view digest. Runtime API's record-mutation adapter retains the
	// narrower installed contract generation. Both are Catalog-authored
	// generation identities, but they are not interchangeable strings.
	if generation == string(view.Digest()) {
		return true
	}
	matched := 0
	for _, authority := range view.ConnectorInboundFinalizationAuthorities(target.Identity.ConnectorOwnerKey) {
		contract := &authority.Contract
		if connectorPushDurableIdentityFromContract(contract, target.Identity.SourceRecordID) != target.Identity ||
			strings.TrimSpace(contract.FlowDeveloperName) != target.FlowDeveloperName ||
			strings.TrimSpace(contract.FlowActionName) != target.FlowActionName ||
			strings.TrimSpace(contract.RemoteIDField) != strings.TrimSpace(remoteIDField) {
			continue
		}
		if strings.TrimSpace(authority.ContractGeneration) != generation {
			return false
		}
		matched++
	}
	return matched == 1
}

func connectorRemoteIdentityTransferHintsEqual(left, right []ConnectorRemoteIdentityTransferHint) bool {
	if len(left) != len(right) {
		return false
	}
	for i := range left {
		if left[i] != right[i] {
			return false
		}
	}
	return true
}

func (e *FlowExecutor) resolveConnectorInboundCurrentInstalledContractSet(
	ctx context.Context,
	connectorOwnerKey string,
) (ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	if e == nil || e.connectorInboundInstalled == nil {
		return ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf("connector inbound installed contract resolver is not wired")
	}
	set, err := e.connectorInboundInstalled.ResolveConnectorInboundRemoteIdentityInstalledContracts(ctx, strings.TrimSpace(connectorOwnerKey))
	if err != nil {
		return ConnectorInboundRemoteIdentityInstalledContractSet{}, err
	}
	return validateConnectorInboundInstalledContractSet(set)
}

func (e *FlowExecutor) resolveConnectorInboundLockedInstalledContractSet(
	ctx context.Context,
	connectorOwnerKey string,
) (ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	if e == nil || e.connectorInboundInstalled == nil {
		return ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf("connector inbound installed contract resolver is not wired")
	}
	set, err := e.connectorInboundInstalled.ResolveConnectorInboundRemoteIdentityInstalledContractsInAmbientTx(ctx, strings.TrimSpace(connectorOwnerKey))
	if err != nil {
		return ConnectorInboundRemoteIdentityInstalledContractSet{}, err
	}
	return validateConnectorInboundInstalledContractSet(set)
}

func (e *FlowExecutor) resolveConnectorInboundLockedInstalledContractSets(
	ctx context.Context,
	connectorOwnerKeys []string,
) (map[string]ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	if e == nil || e.connectorInboundInstalled == nil {
		return nil, fmt.Errorf("connector inbound installed contract resolver is not wired")
	}
	sets, err := e.connectorInboundInstalled.ResolveConnectorInboundRemoteIdentityInstalledContractSetsInAmbientTx(ctx, connectorOwnerKeys)
	if err != nil {
		return nil, err
	}
	result := make(map[string]ConnectorInboundRemoteIdentityInstalledContractSet, len(sets))
	for _, owner := range connectorOwnerKeys {
		owner = strings.TrimSpace(owner)
		set, ok := sets[owner]
		if !ok {
			return nil, fmt.Errorf("connector inbound installed contract set for %q is missing", owner)
		}
		validated, err := validateConnectorInboundInstalledContractSet(set)
		if err != nil {
			return nil, fmt.Errorf("connector inbound installed contract set for %q: %w", owner, err)
		}
		result[owner] = validated
	}
	return result, nil
}

func validateConnectorInboundInstalledContractSet(
	set ConnectorInboundRemoteIdentityInstalledContractSet,
) (ConnectorInboundRemoteIdentityInstalledContractSet, error) {
	if strings.TrimSpace(set.Generation) == "" {
		return ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf("connector inbound installed contract generation is blank")
	}
	if len(set.Contracts) == 0 {
		return ConnectorInboundRemoteIdentityInstalledContractSet{}, fmt.Errorf("connector inbound installed contract set is empty")
	}
	return set, nil
}

func validateConnectorInboundInstalledContract(contract *connector.ConnectorPushDurableMetadataContract) error {
	if contract == nil ||
		strings.TrimSpace(contract.ConnectorOwnerKey) == "" ||
		strings.TrimSpace(contract.ActionSlotID) == "" ||
		strings.TrimSpace(contract.ActionType) != constants.ActionTypePushToConnector ||
		strings.TrimSpace(contract.SourceObject) == "" ||
		strings.TrimSpace(contract.DestinationSlotID) == "" ||
		strings.TrimSpace(contract.TargetObject) == "" ||
		strings.TrimSpace(contract.ProfileName) == "" ||
		strings.TrimSpace(contract.RemoteIDField) == "" {
		return fmt.Errorf("installed connector contract is incomplete or is not %s", constants.ActionTypePushToConnector)
	}
	return nil
}

func (e *FlowExecutor) resolveConnectorInboundInstalledSnapshotInputs(
	ctx context.Context,
	installedMappings []connector.ConnectorFieldMappingInput,
	contract *connector.ConnectorPushDurableMetadataContract,
	expectedProfileInstallID string,
) ([]connector.ConnectorFieldMappingInput, *ConnectorPushProfile, error) {
	if e.connectorPushProfiles == nil {
		return nil, nil, fmt.Errorf("connector inbound recovery profile resolver is not wired")
	}
	mappings, err := selectConnectorInboundInstalledMappings(installedMappings, contract)
	if err != nil {
		return nil, nil, fmt.Errorf("resolve installed outbound mappings: %w", err)
	}
	profileOwnerPack := strings.TrimSpace(contract.ProfileOwnerPack)
	if profileOwnerPack == "" {
		profileOwnerPack = strings.TrimSpace(contract.ConnectorOwnerKey)
	}
	profile, err := e.connectorPushProfiles.ResolveConnectorPushProfile(ctx, strings.TrimSpace(contract.ConnectorOwnerKey), profileOwnerPack, strings.TrimSpace(contract.ProfileName))
	if err != nil {
		return nil, nil, fmt.Errorf("resolve installed profile %s/%s: %w", profileOwnerPack, contract.ProfileName, err)
	}
	if err := validateConnectorInboundFinalizeProfile(profileOwnerPack, contract.ProfileName, profile); err != nil {
		return nil, nil, err
	}
	if strings.TrimSpace(profile.CreatedPackInstallID) != strings.TrimSpace(expectedProfileInstallID) {
		return nil, nil, fmt.Errorf("installed profile %s/%s incarnation changed after generation lock", profileOwnerPack, contract.ProfileName)
	}
	return mappings, profile, nil
}

func selectConnectorInboundInstalledMappings(
	installedMappings []connector.ConnectorFieldMappingInput,
	contract *connector.ConnectorPushDurableMetadataContract,
) ([]connector.ConnectorFieldMappingInput, error) {
	if contract == nil {
		return nil, fmt.Errorf("installed connector contract is nil")
	}
	owner := strings.TrimSpace(contract.ConnectorOwnerKey)
	sourceObject := strings.TrimSpace(contract.SourceObject)
	targetObject := strings.TrimSpace(contract.TargetObject)
	actionType := strings.TrimSpace(contract.ActionType)
	mappings := make([]connector.ConnectorFieldMappingInput, 0)
	for _, mapping := range installedMappings {
		if strings.TrimSpace(mapping.Source) != owner ||
			strings.TrimSpace(mapping.SourceObject) != sourceObject ||
			strings.TrimSpace(mapping.TargetObject) != targetObject ||
			strings.TrimSpace(mapping.ActionType) != actionType ||
			mapping.Direction != connector.DirectionOutbound {
			continue
		}
		mappings = append(mappings, mapping)
	}
	if len(mappings) == 0 {
		return nil, fmt.Errorf("no source-backed outbound mappings found for %s -> %s (%s)", sourceObject, targetObject, actionType)
	}
	if err := validateConnectorFieldMappingRuntimeTargetWrites(mappings); err != nil {
		return nil, err
	}
	if needsTargetRecord, err := connectorMappingsNeedAuthoritativeTargetRecord(mappings); err != nil {
		return nil, err
	} else if needsTargetRecord {
		return nil, fmt.Errorf("PushToConnector: outbound PATCH v1a does not support ConflictPolicy values that require authoritative remote target state")
	}
	return mappings, nil
}