package catalogreconcile

import (
	"context"
	"fmt"
	"reflect"
	"slices"
	"sort"
	"strings"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	"github.com/pingcap-inc/nexuscrm-controlplane/catalogapp"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// ConnectorExecutionMetadataReader resolves the connector facts selected by
// one Catalog operation. Implementations read the effective mapping rows and
// exact installed source generation under the same transaction as the rest of
// the Catalog snapshot. Runtime never receives this resolver.
type ConnectorExecutionMetadataReader interface {
	GetAllInboundConnectorFieldMappings(context.Context) ([]connector.ConnectorFieldMappingInput, error)
	GetAllPackInboundConnectorFieldMappingsForOwner(context.Context, string) ([]connector.ConnectorFieldMappingInput, error)
	ResolveConnectorFieldMappings(
		context.Context,
		consumerv1.IntegrationBinding,
	) (ConnectorFieldMappingSelection, error)
	ResolveConnectorPushContract(
		context.Context,
		consumerv1.IntegrationBinding,
	) (ConnectorPushContractSelection, error)
	ResolveConnectorCalloutProfile(
		context.Context,
		consumerv1.IntegrationBinding,
	) (ConnectorCalloutProfileSelection, error)
}

// ConnectorFieldMappingSelection is one exact source-backed mapping component
// after its installed generation and payload digest have been verified.
type ConnectorFieldMappingSelection struct {
	Generation       string
	Mappings         []connector.ConnectorFieldMappingInput
	OutboundMappings []connector.ConnectorFieldMappingInput
}

// ConnectorPushContractSelection is one exact source-backed component after
// its installed generation and payload digest have been verified.
type ConnectorPushContractSelection struct {
	Generation                string
	Contract                  connector.ConnectorPushDurableMetadataContract
	RecoverySourceEligibility *connector.ConnectorPushRecoverySourceEligibilityContract
}

// ConnectorCalloutProfileSelection is the app-blind runtime profile subset and
// provider-authored outbound allowlist from one exact immutable connector-pack
// generation.
type ConnectorCalloutProfileSelection struct {
	Generation           string
	Profile              catalogsnapshot.ConnectorInboundFinalizationProfile
	OutboundCapabilities []packs.ConnectorCalloutProfileOutboundCapability
}

func appendConnectorExecutionDefinitions(
	ctx context.Context,
	reader MetadataReader,
	revision catalogsnapshot.CatalogRevision,
	bindings []catalogapp.IntegrationBindingSelection,
	input *catalogsnapshot.CompileInput,
) error {
	mappingSelections, pushSelections := connectorExecutionSelections(bindings)
	if len(mappingSelections) == 0 && len(pushSelections) == 0 {
		return nil
	}
	connectorReader, ok := reader.(ConnectorExecutionMetadataReader)
	if !ok {
		return fmt.Errorf("connector-backed mutation projection requires the connector execution metadata reader")
	}
	pendingDefinitions := make([]catalogsnapshot.MutationDefinitionInput, 0)
	appendDefinition := func(value catalogsnapshot.MutationDefinitionInput) {
		value.CatalogRevision = revision
		pendingDefinitions = append(pendingDefinitions, value)
	}
	profilesByWorkload := map[string]map[string]ConnectorCalloutProfileSelection{}
	if len(pushSelections) != 0 {
		var err error
		profilesByWorkload, err = resolveConnectorFinalizationProfiles(
			ctx,
			connectorReader,
			connectorFinalizationProfileSelections(bindings),
		)
		if err != nil {
			return err
		}
	}
	selectedOutboundMappingsByWorkload := make(map[string][]connectorFinalizationMappingSelection)

	if len(mappingSelections) != 0 {
		type exactMapping struct {
			owner string
			value connector.ConnectorFieldMappingInput
		}
		exactByKey := make(map[string]exactMapping)
		resolvedBindings := make(map[string]ConnectorFieldMappingSelection, len(mappingSelections))
		owners := make(map[string]struct{})
		for _, selection := range mappingSelections {
			binding := selection.Binding
			identity := integrationBindingComponentIdentity(binding)
			resolved, alreadyResolved := resolvedBindings[identity]
			if !alreadyResolved {
				var err error
				resolved, err = connectorReader.ResolveConnectorFieldMappings(ctx, binding)
				if err != nil {
					return fmt.Errorf(
						"resolve connector field mapping %s/%s: %w",
						binding.OwnerPack,
						binding.BusinessKey,
						err,
					)
				}
				resolvedBindings[identity] = resolved
			}
			if resolved.Generation != binding.PackInstallationID {
				return fmt.Errorf("connector field mapping %s/%s returned mixed generation", binding.OwnerPack, binding.BusinessKey)
			}
			if len(resolved.Mappings) == 0 && len(pushSelections) == 0 {
				return fmt.Errorf(
					"connector field mapping %s/%s has no inbound closure and no selected push contract",
					binding.OwnerPack,
					binding.BusinessKey,
				)
			}
			for _, value := range resolved.OutboundMappings {
				if value.Source != binding.OwnerPack {
					return fmt.Errorf("connector outbound field mapping owner %q returned source %q", binding.OwnerPack, value.Source)
				}
				selectedOutboundMappingsByWorkload[selection.AuthorizedWorkloadID] = append(
					selectedOutboundMappingsByWorkload[selection.AuthorizedWorkloadID],
					connectorFinalizationMappingSelection{Generation: resolved.Generation, Mapping: value},
				)
			}
			if alreadyResolved {
				continue
			}
			owners[binding.OwnerPack] = struct{}{}
			for _, value := range resolved.Mappings {
				if value.Source != binding.OwnerPack {
					return fmt.Errorf("connector field mapping owner %q returned source %q", binding.OwnerPack, value.Source)
				}
				key := connectorFieldMappingIdentity(value)
				if _, duplicate := exactByKey[key]; duplicate {
					return fmt.Errorf("connector field mapping identity %q is selected by multiple components", key)
				}
				exactByKey[key] = exactMapping{owner: binding.OwnerPack, value: value}
			}
		}

		ownerNames := make([]string, 0, len(owners))
		for owner := range owners {
			ownerNames = append(ownerNames, owner)
		}
		sort.Strings(ownerNames)
		exactKeys := make([]string, 0, len(exactByKey))
		for key := range exactByKey {
			exactKeys = append(exactKeys, key)
		}
		sort.Strings(exactKeys)
		for _, owner := range ownerNames {
			live, err := connectorReader.GetAllPackInboundConnectorFieldMappingsForOwner(ctx, owner)
			if err != nil {
				return fmt.Errorf("read pack-authored inbound connector mappings for %q: %w", owner, err)
			}
			liveByKey := make(map[string]connector.ConnectorFieldMappingInput, len(live))
			for _, value := range live {
				if value.Source != owner {
					return fmt.Errorf("pack-authored inbound connector mapping owner %q returned source %q", owner, value.Source)
				}
				liveByKey[connectorFieldMappingIdentity(value)] = value
			}
			for _, key := range exactKeys {
				expected := exactByKey[key]
				if expected.owner != owner {
					continue
				}
				value, found := liveByKey[key]
				if !found || value != expected.value {
					return fmt.Errorf("pack-authored inbound connector mapping %q drifted from exact generation", key)
				}
				rule := connectorFieldMappingRule(expected.value, true)
				appendDefinition(catalogsnapshot.MutationDefinitionInput{
					ConnectorInboundFieldMappingRule: &rule,
				})
			}
		}

		effective, err := connectorReader.GetAllInboundConnectorFieldMappings(ctx)
		if err != nil {
			return fmt.Errorf("read effective inbound connector mappings: %w", err)
		}
		seenEffective := make(map[string]struct{}, len(exactByKey))
		for index := range effective {
			value := effective[index]
			key := connectorFieldMappingIdentity(value)
			if _, selected := exactByKey[key]; !selected {
				continue
			}
			seenEffective[key] = struct{}{}
			rule := connectorFieldMappingRule(value, false)
			appendDefinition(catalogsnapshot.MutationDefinitionInput{
				ConnectorInboundFieldMappingRule: &rule,
			})
		}
		for _, key := range exactKeys {
			if _, found := seenEffective[key]; !found {
				return fmt.Errorf("effective inbound connector mapping %q is missing", key)
			}
		}
	}

	publishedAuthorship := make(map[string]catalogsnapshot.ConnectorInboundAuthorshipBinding)
	publishedFinalization := make(map[string]catalogsnapshot.ConnectorInboundFinalizationAuthority)
	resolvedPushBindings := make(map[string]ConnectorPushContractSelection, len(pushSelections))
	for index := range pushSelections {
		selection := pushSelections[index]
		binding := selection.Binding
		bindingIdentity := integrationBindingComponentIdentity(binding)
		resolved, alreadyResolved := resolvedPushBindings[bindingIdentity]
		if !alreadyResolved {
			var err error
			resolved, err = connectorReader.ResolveConnectorPushContract(ctx, binding)
			if err != nil {
				return fmt.Errorf("resolve connector push contract %s/%s: %w", binding.OwnerPack, binding.BusinessKey, err)
			}
			resolvedPushBindings[bindingIdentity] = resolved
		}
		if resolved.Generation != binding.PackInstallationID {
			return fmt.Errorf("connector push contract %s/%s returned mixed generation", binding.OwnerPack, binding.BusinessKey)
		}
		contract := resolved.Contract
		profileOwner := strings.TrimSpace(contract.ProfileOwnerPack)
		if profileOwner == "" {
			profileOwner = contract.ConnectorOwnerKey
		}
		profile, ok := profilesByWorkload[selection.AuthorizedWorkloadID][connectorFinalizationProfileIdentity(
			profileOwner,
			contract.ProfileName,
		)]
		if !ok {
			return fmt.Errorf(
				"connector push contract %s/%s workload %q has no exact selected callout profile %s/%s",
				binding.OwnerPack,
				binding.BusinessKey,
				selection.AuthorizedWorkloadID,
				profileOwner,
				contract.ProfileName,
			)
		}
		if err := validateConnectorPushOutboundAuthority(
			contract,
			profile.OutboundCapabilities,
		); err != nil {
			return fmt.Errorf(
				"connector push contract %s/%s is incompatible with callout profile %s/%s: %w",
				binding.OwnerPack,
				binding.BusinessKey,
				profileOwner,
				contract.ProfileName,
				err,
			)
		}
		outbound, err := connectorFinalizationOutboundMappings(
			contract,
			resolved.Generation,
			selectedOutboundMappingsByWorkload[selection.AuthorizedWorkloadID],
		)
		if err != nil {
			return fmt.Errorf("connector push contract %s/%s: %w", binding.OwnerPack, binding.BusinessKey, err)
		}
		if len(outbound) == 0 {
			return fmt.Errorf(
				"connector push contract %s/%s has no exact selected outbound mapping closure",
				binding.OwnerPack,
				binding.BusinessKey,
			)
		}
		authorship := catalogsnapshot.ConnectorInboundAuthorshipBinding{
			ConnectorOwnerKey: contract.ConnectorOwnerKey,
			Generation:        resolved.Generation,
			ActionSlotID:      contract.ActionSlotID,
			ActionType:        contract.ActionType,
			SourceObject:      contract.SourceObject,
			DestinationSlotID: contract.DestinationSlotID,
			TargetObject:      contract.TargetObject,
			ProfileOwnerPack:  profileOwner,
			ProfileName:       contract.ProfileName,
			RemoteIDField:     contract.RemoteIDField,
		}
		finalization := catalogsnapshot.ConnectorInboundFinalizationAuthority{
			ContractGeneration: resolved.Generation,
			Contract:           contract,
			Profile:            profile.Profile,
			OutboundMappings:   outbound,
		}
		if resolved.RecoverySourceEligibility != nil {
			finalization.RecoverySourceEligibility = &connector.ConnectorPushRecoverySourceEligibilityContract{
				RequiredFalseBooleanFields: append(
					[]string(nil),
					resolved.RecoverySourceEligibility.RequiredFalseBooleanFields...,
				),
			}
		}
		authorshipKey := connectorFinalizationAuthorshipIdentity(authorship)
		if previous, duplicate := publishedAuthorship[authorshipKey]; duplicate {
			if !reflect.DeepEqual(previous, authorship) {
				return fmt.Errorf(
					"connector push contract %s/%s has conflicting workload-scoped authorship authority",
					binding.OwnerPack,
					binding.BusinessKey,
				)
			}
		} else {
			publishedAuthorship[authorshipKey] = authorship
			appendDefinition(catalogsnapshot.MutationDefinitionInput{
				ConnectorInboundAuthorship: &authorship,
			})
		}
		finalizationKey := connectorFinalizationAuthorityIdentity(finalization)
		if previous, duplicate := publishedFinalization[finalizationKey]; duplicate {
			if !reflect.DeepEqual(previous, finalization) {
				return fmt.Errorf(
					"connector push contract %s/%s has conflicting workload-scoped finalization authority",
					binding.OwnerPack,
					binding.BusinessKey,
				)
			}
		} else {
			publishedFinalization[finalizationKey] = finalization
			appendDefinition(catalogsnapshot.MutationDefinitionInput{
				ConnectorInboundFinalization: &finalization,
			})
		}
		intentBindings, err := connectorOutboundIntentBindingReferences(
			bindings,
			selection,
			contract,
			resolved.Generation,
			profile.Profile,
		)
		if err != nil {
			return fmt.Errorf(
				"connector push contract %s/%s outbound intent route: %w",
				binding.OwnerPack,
				binding.BusinessKey,
				err,
			)
		}
		input.OutboundIntentRoutes = append(
			input.OutboundIntentRoutes,
			catalogsnapshot.OutboundIntentRouteInput{
				CatalogRevision: revision,
				Route: catalogsnapshot.OutboundIntentRoute{
					AuthorizedWorkloadID: selection.AuthorizedWorkloadID,
					ConnectorOwnerKey:    contract.ConnectorOwnerKey,
					FlowDeveloperName:    contract.FlowDeveloperName,
					FlowActionName:       contract.FlowActionName,
					SourceObject:         contract.SourceObject,
					ActionSlotID:         contract.ActionSlotID,
					DestinationSlotID:    contract.DestinationSlotID,
					Bindings:             intentBindings,
				},
			},
		)
	}
	input.MutationDefinitions = append(input.MutationDefinitions, pendingDefinitions...)
	return nil
}

func validateConnectorPushOutboundAuthority(
	contract connector.ConnectorPushDurableMetadataContract,
	capabilities []packs.ConnectorCalloutProfileOutboundCapability,
) error {
	required := make([]packs.ConnectorCalloutProfileOutboundCapability, 0, 2+len(contract.Associations))
	if strings.TrimSpace(contract.CreatePathTemplate) != "" {
		required = append(required, packs.ConnectorCalloutProfileOutboundCapability{
			Operation:    "create",
			TargetObject: contract.TargetObject,
			Method:       "POST",
			PathTemplate: contract.CreatePathTemplate,
		})
	}
	if strings.TrimSpace(contract.UpdatePathTemplate) != "" {
		required = append(required, packs.ConnectorCalloutProfileOutboundCapability{
			Operation:    "update",
			TargetObject: contract.TargetObject,
			Method:       "PATCH",
			PathTemplate: contract.UpdatePathTemplate,
		})
	}
	for _, association := range contract.Associations {
		if strings.TrimSpace(association.AssociationPath) == "" {
			continue
		}
		required = append(required, packs.ConnectorCalloutProfileOutboundCapability{
			Operation:               "associate",
			TargetObject:            contract.TargetObject,
			AssociationTargetObject: association.TargetObject,
			Method:                  association.AssociationMethod,
			PathTemplate:            association.AssociationPath,
		})
	}
	for _, expected := range required {
		found := false
		var incompatible *packs.ConnectorCalloutProfileOutboundCapability
		for _, actual := range capabilities {
			if actual.Operation != expected.Operation ||
				actual.TargetObject != expected.TargetObject ||
				actual.AssociationTargetObject != expected.AssociationTargetObject {
				continue
			}
			if actual.Method == expected.Method && actual.PathTemplate == expected.PathTemplate {
				found = true
				break
			}
			if incompatible == nil {
				candidate := actual
				incompatible = &candidate
			}
		}
		if found {
			continue
		}
		if incompatible != nil {
			return fmt.Errorf(
				"%s authority for target %q association target %q declares %s %q, contract requires %s %q",
				expected.Operation,
				expected.TargetObject,
				expected.AssociationTargetObject,
				incompatible.Method,
				incompatible.PathTemplate,
				expected.Method,
				expected.PathTemplate,
			)
		}
		return fmt.Errorf(
			"profile does not explicitly authorize %s target %q association target %q with %s %q",
			expected.Operation,
			expected.TargetObject,
			expected.AssociationTargetObject,
			expected.Method,
			expected.PathTemplate,
		)
	}
	return nil
}

func connectorOutboundIntentBindingReferences(
	all []catalogapp.IntegrationBindingSelection,
	push catalogapp.IntegrationBindingSelection,
	contract connector.ConnectorPushDurableMetadataContract,
	contractGeneration string,
	profile catalogsnapshot.ConnectorInboundFinalizationProfile,
) ([]catalogsnapshot.IntegrationBindingReference, error) {
	values := make([]catalogsnapshot.IntegrationBindingReference, 0)
	for _, selection := range all {
		if selection.AuthorizedWorkloadID != push.AuthorizedWorkloadID {
			continue
		}
		binding := selection.Binding
		selected := false
		switch binding.Kind {
		case consumerv1.BindingConnectorPushContract:
			selected = integrationBindingComponentIdentity(binding) ==
				integrationBindingComponentIdentity(push.Binding)
		case consumerv1.BindingConnectorCalloutProfile:
			selected = binding.OwnerPack == profile.OwnerPack &&
				binding.PackInstallationID == profile.Generation &&
				binding.BusinessKey == profile.OwnerPack+"/"+profile.ProfileName
		case consumerv1.BindingConnectorFieldMapping:
			selected = binding.OwnerPack == contract.ConnectorOwnerKey &&
				binding.PackInstallationID == contractGeneration
		}
		if !selected {
			continue
		}
		values = append(values, catalogsnapshot.IntegrationBindingReference{
			Kind: binding.Kind, OwnerPack: binding.OwnerPack,
			PackInstallationID: binding.PackInstallationID,
			PackPayloadDigest:  binding.PackPayloadDigest,
			BusinessKey:        binding.BusinessKey,
			DefinitionVersion:  binding.DefinitionVersion,
		})
	}
	if len(values) < 3 {
		return nil, fmt.Errorf(
			"exact push, callout profile, and field-mapping bindings are required",
		)
	}
	sort.Slice(values, func(left, right int) bool {
		leftKey := strings.Join([]string{
			string(values[left].Kind), values[left].OwnerPack,
			values[left].PackInstallationID, values[left].BusinessKey,
		}, "\x00")
		rightKey := strings.Join([]string{
			string(values[right].Kind), values[right].OwnerPack,
			values[right].PackInstallationID, values[right].BusinessKey,
		}, "\x00")
		return leftKey < rightKey
	})
	return values, nil
}

func resolveConnectorFinalizationProfiles(
	ctx context.Context,
	reader ConnectorExecutionMetadataReader,
	selections []catalogapp.IntegrationBindingSelection,
) (map[string]map[string]ConnectorCalloutProfileSelection, error) {
	result := make(map[string]map[string]ConnectorCalloutProfileSelection)
	for _, selection := range selections {
		binding := selection.Binding
		// Every workload-scoped binding carries its own credential
		// authorization. Resolve and validate every binding before deduplicating
		// the resulting immutable profile authority; otherwise an authorized
		// binding could mask a later binding that does not authorize the
		// selected credential.
		resolved, err := reader.ResolveConnectorCalloutProfile(ctx, binding)
		if err != nil {
			return nil, fmt.Errorf(
				"resolve connector callout profile %s/%s: %w",
				binding.OwnerPack,
				binding.BusinessKey,
				err,
			)
		}
		profile := resolved.Profile
		if resolved.Generation != binding.PackInstallationID ||
			profile.Generation != binding.PackInstallationID ||
			profile.OwnerPack != binding.OwnerPack ||
			binding.BusinessKey != binding.OwnerPack+"/"+profile.ProfileName ||
			!slices.Contains(binding.CredentialRefs, profile.CredentialRef) {
			return nil, fmt.Errorf(
				"connector callout profile %s/%s returned mixed or unauthorized authority",
				binding.OwnerPack,
				binding.BusinessKey,
			)
		}
		key := connectorFinalizationProfileIdentity(
			profile.OwnerPack,
			profile.ProfileName,
		)
		byIdentity := result[selection.AuthorizedWorkloadID]
		if byIdentity == nil {
			byIdentity = make(map[string]ConnectorCalloutProfileSelection)
			result[selection.AuthorizedWorkloadID] = byIdentity
		}
		if previous, duplicate := byIdentity[key]; duplicate {
			if !equalConnectorCalloutProfileSelection(previous, resolved) {
				return nil, fmt.Errorf(
					"connector callout profile %s/%s has conflicting exact authority",
					profile.OwnerPack,
					profile.ProfileName,
				)
			}
			continue
		}
		byIdentity[key] = resolved
	}
	return result, nil
}

func equalConnectorFinalizationProfile(
	left catalogsnapshot.ConnectorInboundFinalizationProfile,
	right catalogsnapshot.ConnectorInboundFinalizationProfile,
) bool {
	return left.Generation == right.Generation &&
		left.OwnerPack == right.OwnerPack &&
		left.ProfileName == right.ProfileName &&
		left.CredentialRef == right.CredentialRef &&
		left.EndpointClass == right.EndpointClass &&
		left.MaxAttempts == right.MaxAttempts &&
		left.BaseDelayNanos == right.BaseDelayNanos &&
		left.Multiplier == right.Multiplier &&
		left.MaxDelayNanos == right.MaxDelayNanos &&
		slices.Equal(left.RetryableStatusCodes, right.RetryableStatusCodes)
}

func equalConnectorCalloutProfileSelection(
	left ConnectorCalloutProfileSelection,
	right ConnectorCalloutProfileSelection,
) bool {
	return left.Generation == right.Generation &&
		equalConnectorFinalizationProfile(left.Profile, right.Profile) &&
		slices.Equal(left.OutboundCapabilities, right.OutboundCapabilities)
}

func connectorFinalizationProfileSelections(
	values []catalogapp.IntegrationBindingSelection,
) []catalogapp.IntegrationBindingSelection {
	result := make([]catalogapp.IntegrationBindingSelection, 0)
	for _, value := range values {
		if value.Binding.Kind == consumerv1.BindingConnectorCalloutProfile {
			result = append(result, value)
		}
	}
	return result
}

func connectorFinalizationProfileIdentity(ownerPack string, profileName string) string {
	return ownerPack + "\x00" + profileName
}

func connectorFinalizationOutboundMappings(
	contract connector.ConnectorPushDurableMetadataContract,
	contractGeneration string,
	values []connectorFinalizationMappingSelection,
) ([]connector.ConnectorFieldMappingInput, error) {
	result := make([]connector.ConnectorFieldMappingInput, 0)
	for _, selected := range values {
		value := selected.Mapping
		if value.Source == contract.ConnectorOwnerKey &&
			strings.EqualFold(value.SourceObject, contract.SourceObject) &&
			strings.EqualFold(value.TargetObject, contract.TargetObject) &&
			strings.EqualFold(value.ActionType, contract.ActionType) {
			if selected.Generation != contractGeneration {
				return nil, fmt.Errorf("outbound mapping closure has mixed connector generation")
			}
			result = append(result, value)
		}
	}
	return result, nil
}

type connectorFinalizationMappingSelection struct {
	Generation string
	Mapping    connector.ConnectorFieldMappingInput
}

func connectorExecutionSelections(
	values []catalogapp.IntegrationBindingSelection,
) ([]catalogapp.IntegrationBindingSelection, []catalogapp.IntegrationBindingSelection) {
	mappingSelections := make([]catalogapp.IntegrationBindingSelection, 0)
	pushSelections := make([]catalogapp.IntegrationBindingSelection, 0)
	for _, value := range values {
		binding := value.Binding
		switch binding.Kind {
		case consumerv1.BindingConnectorFieldMapping:
			mappingSelections = append(mappingSelections, value)
		case consumerv1.BindingConnectorPushContract:
			pushSelections = append(pushSelections, value)
		}
	}
	return mappingSelections, pushSelections
}

func integrationBindingComponentIdentity(value consumerv1.IntegrationBinding) string {
	return strings.Join([]string{
		value.OwnerPack,
		value.PackInstallationID,
		value.PackPayloadDigest,
		value.BusinessKey,
	}, "\x00")
}

func connectorFinalizationAuthorshipIdentity(
	value catalogsnapshot.ConnectorInboundAuthorshipBinding,
) string {
	return strings.Join([]string{
		strings.ToLower(value.ConnectorOwnerKey),
		strings.ToLower(value.SourceObject),
		strings.ToLower(value.RemoteIDField),
	}, "\x00")
}

func connectorFinalizationAuthorityIdentity(
	value catalogsnapshot.ConnectorInboundFinalizationAuthority,
) string {
	return strings.Join([]string{
		strings.ToLower(value.Contract.ConnectorOwnerKey),
		strings.ToLower(value.Contract.FlowDeveloperName),
		strings.ToLower(value.Contract.FlowActionName),
	}, "\x00")
}

func connectorFieldMappingIdentity(value connector.ConnectorFieldMappingInput) string {
	return strings.Join([]string{
		value.Source,
		value.SourceObject,
		value.TargetObject,
		value.ActionType,
		value.SourceField,
		value.Direction,
	}, "\x00")
}

func connectorFieldMappingRule(
	value connector.ConnectorFieldMappingInput,
	packAuthored bool,
) catalogsnapshot.ConnectorInboundFieldMappingRule {
	return catalogsnapshot.ConnectorInboundFieldMappingRule{
		Source:              value.Source,
		PackAuthored:        packAuthored,
		SourceObject:        value.SourceObject,
		TargetObject:        value.TargetObject,
		ActionType:          value.ActionType,
		SourceField:         value.SourceField,
		TargetField:         value.TargetField,
		Direction:           value.Direction,
		TransformExpression: value.TransformExpression,
		ConflictPolicy:      value.ConflictPolicy,
		DefaultValue:        value.DefaultValue,
		ValueMap:            value.ValueMap,
		OnUnmappedValue:     value.OnUnmappedValue,
		SortOrder:           value.SortOrder,
	}
}