package catalogreconcile

import (
	"context"
	"errors"
	"slices"
	"strings"
	"testing"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	consumerv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	outboundintentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	"github.com/pingcap-inc/nexuscrm-controlplane/catalogapp"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

type connectorExecutionReader struct {
	*markedReader
	effective     []connector.ConnectorFieldMappingInput
	packAuthored  map[string][]connector.ConnectorFieldMappingInput
	mappingByKey  map[string]ConnectorFieldMappingSelection
	mappingErr    error
	mappingCalls  int
	resolved      ConnectorPushContractSelection
	resolveErr    error
	resolvedCalls int
	profileByKey  map[string]ConnectorCalloutProfileSelection
	profileErr    error
	profileCalls  int
}

func TestConnectorPushBindingPublishesConsumerOnlyOutboundIntentProjection(t *testing.T) {
	selection := MetadataSelection{
		RuntimeCapabilityIDs: []string{recordmutationv1.CapabilityName},
		IntegrationBindings: []catalogapp.IntegrationBindingSelection{{
			Binding: consumerv1.IntegrationBinding{Kind: consumerv1.BindingConnectorPushContract},
		}},
	}
	capabilities := metadataCapabilityIDs(selection)
	if !slices.Contains(capabilities, outboundintentsemantic.CapabilityName) {
		t.Fatalf("connector-push capability closure = %v, want %q", capabilities, outboundintentsemantic.CapabilityName)
	}
	if profile, err := projectionProfile(outboundintentsemantic.CapabilityName); err != nil ||
		profile != catalogsnapshot.ProjectionRecordMutationV1 {
		t.Fatalf("outbound-intent projection = %q, %v", profile, err)
	}
}

func (r *connectorExecutionReader) ResolveConnectorFieldMappings(
	ctx context.Context,
	binding consumerv1.IntegrationBinding,
) (ConnectorFieldMappingSelection, error) {
	if r.markedReader != nil && r.markedReader.t != nil {
		r.markedReader.requireTransaction(ctx)
	}
	r.mappingCalls++
	return r.mappingByKey[binding.BusinessKey], r.mappingErr
}

func (r *connectorExecutionReader) GetAllInboundConnectorFieldMappings(
	ctx context.Context,
) ([]connector.ConnectorFieldMappingInput, error) {
	if r.markedReader != nil && r.markedReader.t != nil {
		r.markedReader.requireTransaction(ctx)
	}
	return r.effective, nil
}

func (r *connectorExecutionReader) GetAllPackInboundConnectorFieldMappingsForOwner(
	ctx context.Context,
	owner string,
) ([]connector.ConnectorFieldMappingInput, error) {
	if r.markedReader != nil && r.markedReader.t != nil {
		r.markedReader.requireTransaction(ctx)
	}
	return r.packAuthored[owner], nil
}

func (r *connectorExecutionReader) ResolveConnectorPushContract(
	ctx context.Context,
	_ consumerv1.IntegrationBinding,
) (ConnectorPushContractSelection, error) {
	if r.markedReader != nil && r.markedReader.t != nil {
		r.markedReader.requireTransaction(ctx)
	}
	r.resolvedCalls++
	return r.resolved, r.resolveErr
}

func (r *connectorExecutionReader) ResolveConnectorCalloutProfile(
	ctx context.Context,
	binding consumerv1.IntegrationBinding,
) (ConnectorCalloutProfileSelection, error) {
	if r.markedReader != nil && r.markedReader.t != nil {
		r.markedReader.requireTransaction(ctx)
	}
	r.profileCalls++
	return r.profileByKey[binding.BusinessKey], r.profileErr
}

func TestAppendConnectorExecutionDefinitionsPublishesOnlySelectedExactAuthority(t *testing.T) {
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		effective: []connector.ConnectorFieldMappingInput{
			{Source: "connector-a", SourceObject: "RemoteA", TargetObject: "Account", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"},
			{Source: "connector-a", SourceObject: "RemoteOther", TargetObject: "Contact", SourceField: "id", TargetField: "OtherRemoteId", Direction: "inbound"},
			{Source: "connector-b", SourceObject: "RemoteB", TargetObject: "Account", SourceField: "id", TargetField: "OtherId", Direction: "inbound"},
		},
		packAuthored: map[string][]connector.ConnectorFieldMappingInput{
			"connector-a": {
				{Source: "connector-a", SourceObject: "RemoteA", TargetObject: "Account", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"},
				{Source: "connector-a", SourceObject: "RemoteOther", TargetObject: "Contact", SourceField: "id", TargetField: "OtherRemoteId", Direction: "inbound"},
			},
		},
		mappingByKey: map[string]ConnectorFieldMappingSelection{
			"mapping/account": {
				Generation: "install-a",
				Mappings:   []connector.ConnectorFieldMappingInput{{Source: "connector-a", SourceObject: "RemoteA", TargetObject: "Account", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"}},
				OutboundMappings: []connector.ConnectorFieldMappingInput{{
					Source: "connector-a", SourceObject: "RemoteA", TargetObject: "Account",
					ActionType: constants.ActionTypePushToConnector, SourceField: "name", TargetField: "Name",
					Direction: connector.DirectionOutbound,
				}},
			},
		},
		profileByKey: map[string]ConnectorCalloutProfileSelection{
			"connector-a/default": testConnectorCalloutProfileSelection("install-a"),
		},
		resolved: ConnectorPushContractSelection{
			Generation: "install-a",
			Contract: connector.ConnectorPushDurableMetadataContract{
				ConnectorOwnerKey:  "connector-a",
				ActionSlotID:       "upsert-account",
				ActionType:         constants.ActionTypePushToConnector,
				FlowDeveloperName:  "Remote_To_Account",
				FlowActionName:     "PushAccount",
				SourceObject:       "RemoteA",
				DestinationSlotID:  "account",
				TargetObject:       "Account",
				CreatePathTemplate: "/crm/v3/objects/accounts",
				UpdatePathTemplate: "/crm/v3/objects/accounts/{remoteId}",
				ResponseIDPath:     "id",
				ProfileOwnerPack:   "connector-a",
				ProfileName:        "default",
				RemoteIDField:      "RemoteId",
			},
		},
	}
	mapping := catalogapp.IntegrationBindingSelection{AuthorizedWorkloadID: "integration-a", Binding: consumerv1.IntegrationBinding{
		Kind:               consumerv1.BindingConnectorFieldMapping,
		OwnerPack:          "connector-a",
		PackInstallationID: "install-a",
		PackPayloadDigest:  "sha256:pack-a",
		BusinessKey:        "mapping/account",
	}}
	push := catalogapp.IntegrationBindingSelection{
		AuthorizedWorkloadID: "integration-a",
		Binding: consumerv1.IntegrationBinding{
			Kind:               consumerv1.BindingConnectorPushContract,
			OwnerPack:          "connector-a",
			PackInstallationID: "install-a",
			PackPayloadDigest:  "sha256:pack-a",
			BusinessKey:        "push/account",
		},
	}
	profile := catalogapp.IntegrationBindingSelection{AuthorizedWorkloadID: "integration-a", Binding: consumerv1.IntegrationBinding{
		Kind:               consumerv1.BindingConnectorCalloutProfile,
		OwnerPack:          "connector-a",
		PackInstallationID: "install-a",
		PackPayloadDigest:  "sha256:pack-a",
		BusinessKey:        "connector-a/default",
		CredentialRefs:     []string{"credential-a"},
	}}
	duplicatePush := push
	duplicatePush.AuthorizedWorkloadID = "integration-b"
	duplicateMapping := mapping
	duplicateMapping.AuthorizedWorkloadID = "integration-b"
	duplicateProfile := profile
	duplicateProfile.AuthorizedWorkloadID = "integration-b"
	input := catalogsnapshot.CompileInput{}

	err := appendConnectorExecutionDefinitions(
		context.Background(),
		reader,
		catalogsnapshot.CatalogRevision(41),
		[]catalogapp.IntegrationBindingSelection{
			mapping, profile, push,
			duplicateMapping, duplicateProfile, duplicatePush,
		},
		&input,
	)
	if err != nil {
		t.Fatal(err)
	}
	if reader.resolvedCalls != 1 {
		t.Fatalf("exact connector push contract resolved %d times, want 1", reader.resolvedCalls)
	}
	if reader.mappingCalls != 1 {
		t.Fatalf("exact connector field mapping resolved %d times, want 1", reader.mappingCalls)
	}
	if reader.profileCalls != 2 {
		t.Fatalf("workload-scoped connector profiles resolved %d times, want 2", reader.profileCalls)
	}
	if len(input.MutationDefinitions) != 4 {
		t.Fatalf("connector mutation definitions = %#v", input.MutationDefinitions)
	}
	for index, definition := range input.MutationDefinitions {
		if definition.CatalogRevision != 41 {
			t.Fatalf("definition[%d] revision = %d, want 41", index, definition.CatalogRevision)
		}
	}
	packAuthored := input.MutationDefinitions[0].ConnectorInboundFieldMappingRule
	if packAuthored == nil || !packAuthored.PackAuthored || packAuthored.TargetObject != "Account" {
		t.Fatalf("pack-authored connector mapping = %#v", packAuthored)
	}
	effective := input.MutationDefinitions[1].ConnectorInboundFieldMappingRule
	if effective == nil || effective.Source != "connector-a" || effective.PackAuthored || effective.TargetField != "RemoteId" {
		t.Fatalf("effective connector mapping = %#v", effective)
	}
	authorship := input.MutationDefinitions[2].ConnectorInboundAuthorship
	if authorship == nil || authorship.ConnectorOwnerKey != "connector-a" ||
		authorship.Generation != "install-a" || authorship.ActionSlotID != "upsert-account" {
		t.Fatalf("connector authorship = %#v", authorship)
	}
	finalization := input.MutationDefinitions[3].ConnectorInboundFinalization
	if finalization == nil || finalization.Profile.CredentialRef != "credential-a" ||
		len(finalization.OutboundMappings) != 1 || finalization.OutboundMappings[0].Direction != connector.DirectionOutbound {
		t.Fatalf("connector finalization authority = %#v", finalization)
	}
	if len(input.OutboundIntentRoutes) != 2 ||
		input.OutboundIntentRoutes[0].Route.AuthorizedWorkloadID != "integration-a" ||
		input.OutboundIntentRoutes[1].Route.AuthorizedWorkloadID != "integration-b" {
		t.Fatalf("workload-scoped outbound intent routes = %#v", input.OutboundIntentRoutes)
	}
}

func TestValidateConnectorPushOutboundAuthorityClosesDealCompanyOperation(t *testing.T) {
	contract := connector.ConnectorPushDurableMetadataContract{
		TargetObject:       "deals",
		CreatePathTemplate: "/crm/v3/objects/deals",
		UpdatePathTemplate: "/crm/v3/objects/deals/{remoteId}",
		Associations: []connector.ConnectorPushAssociationContract{{
			TargetObject:      "companies",
			AssociationMethod: "PUT",
			AssociationPath:   "/crm/v3/objects/deals/{fromId}/associations/companies/{toId}/5",
		}},
	}
	capabilities := []packs.ConnectorCalloutProfileOutboundCapability{
		{Operation: "create", TargetObject: "deals", Method: "POST", PathTemplate: "/crm/v3/objects/deals"},
		{Operation: "update", TargetObject: "deals", Method: "PATCH", PathTemplate: "/crm/v3/objects/deals/{remoteId}"},
		{Operation: "associate", TargetObject: "deals", AssociationTargetObject: "companies", Method: "PUT", PathTemplate: "/crm/v3/objects/deals/{fromId}/associations/companies/{toId}/5"},
	}
	if err := validateConnectorPushOutboundAuthority(contract, capabilities); err != nil {
		t.Fatalf("complete deal-company authority rejected: %v", err)
	}
	withAlternatePathFirst := append(
		[]packs.ConnectorCalloutProfileOutboundCapability{{
			Operation: "create", TargetObject: "deals", Method: "POST", PathTemplate: "/crm/v4/objects/deals",
		}},
		capabilities...,
	)
	if err := validateConnectorPushOutboundAuthority(contract, withAlternatePathFirst); err != nil {
		t.Fatalf("later exact path authority rejected after alternate path: %v", err)
	}

	tests := []struct {
		name   string
		mutate func(*connector.ConnectorPushDurableMetadataContract)
		want   string
	}{
		{
			name: "target object",
			mutate: func(value *connector.ConnectorPushDurableMetadataContract) {
				value.TargetObject = "tickets"
			},
			want: `does not explicitly authorize create target "tickets"`,
		},
		{
			name: "create path family",
			mutate: func(value *connector.ConnectorPushDurableMetadataContract) {
				value.CreatePathTemplate = "/crm/v4/objects/deals"
			},
			want: "contract requires POST \"/crm/v4/objects/deals\"",
		},
		{
			name: "association target",
			mutate: func(value *connector.ConnectorPushDurableMetadataContract) {
				value.Associations[0].TargetObject = "contacts"
			},
			want: `association target "contacts"`,
		},
		{
			name: "association method",
			mutate: func(value *connector.ConnectorPushDurableMetadataContract) {
				value.Associations[0].AssociationMethod = "POST"
			},
			want: "contract requires POST",
		},
		{
			name: "association path family",
			mutate: func(value *connector.ConnectorPushDurableMetadataContract) {
				value.Associations[0].AssociationPath = "/crm/v4/objects/deals/{fromId}/associations/companies/{toId}/5"
			},
			want: "contract requires PUT \"/crm/v4/objects/deals",
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			mutated := contract
			mutated.Associations = append(
				[]connector.ConnectorPushAssociationContract(nil),
				contract.Associations...,
			)
			test.mutate(&mutated)
			err := validateConnectorPushOutboundAuthority(mutated, capabilities)
			if err == nil || !strings.Contains(err.Error(), test.want) {
				t.Fatalf("error = %v, want %q", err, test.want)
			}
		})
	}
}

func TestAppendConnectorExecutionDefinitionsRejectsLiveMappingDrift(t *testing.T) {
	exact := connector.ConnectorFieldMappingInput{
		Source: "connector-a", SourceObject: "Remote", TargetObject: "Account",
		ActionType: "upsert", SourceField: "id", TargetField: "RemoteId", Direction: "inbound",
	}
	drifted := exact
	drifted.TargetField = "OtherRemoteId"
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		mappingByKey: map[string]ConnectorFieldMappingSelection{
			"mapping/account": {Generation: "install-a", Mappings: []connector.ConnectorFieldMappingInput{exact}},
		},
		packAuthored: map[string][]connector.ConnectorFieldMappingInput{"connector-a": {drifted}},
	}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(), reader, 1,
		[]catalogapp.IntegrationBindingSelection{{Binding: consumerv1.IntegrationBinding{
			Kind: consumerv1.BindingConnectorFieldMapping, OwnerPack: "connector-a",
			PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a", BusinessKey: "mapping/account",
		}}},
		&input,
	)
	if err == nil || !strings.Contains(err.Error(), "drifted from exact generation") {
		t.Fatalf("live mapping drift error = %v", err)
	}
	if len(input.MutationDefinitions) != 0 {
		t.Fatalf("drifted mapping published definitions: %#v", input.MutationDefinitions)
	}
}

func TestAppendConnectorExecutionDefinitionsPreservesMappingOnlyInboundRequirement(t *testing.T) {
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		mappingByKey: map[string]ConnectorFieldMappingSelection{
			"mapping/account": {
				Generation: "install-a",
				OutboundMappings: []connector.ConnectorFieldMappingInput{{
					Source: "connector-a", SourceObject: "Account", TargetObject: "remote-account",
					ActionType: constants.ActionTypePushToConnector, SourceField: "Name", TargetField: "name",
					Direction: connector.DirectionOutbound,
				}},
			},
		},
	}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(), reader, 1,
		[]catalogapp.IntegrationBindingSelection{{Binding: consumerv1.IntegrationBinding{
			Kind: consumerv1.BindingConnectorFieldMapping, OwnerPack: "connector-a",
			PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a", BusinessKey: "mapping/account",
		}}},
		&input,
	)
	if err == nil || !strings.Contains(err.Error(), "no inbound closure and no selected push contract") {
		t.Fatalf("mapping-only outbound closure error = %v", err)
	}
	if len(input.MutationDefinitions) != 0 {
		t.Fatalf("mapping-only outbound closure published definitions: %#v", input.MutationDefinitions)
	}
}

func TestReadEffectiveMetadataPublishesConnectorExecutionFactsInOneRevision(t *testing.T) {
	account := validObject("Account")
	account.Fields = append(account.Fields,
		models.FieldMetadata{ID: "field-account-remote-id", APIName: "RemoteId", Label: "Remote ID", Type: constants.FieldTypeString, PhysicalName: "remote_id"},
		models.FieldMetadata{ID: "field-account-name", APIName: "Name", Label: "Name", Type: constants.FieldTypeString, PhysicalName: "name"},
		models.FieldMetadata{ID: "field-account-archived", APIName: "Archived", Label: "Archived", Type: constants.FieldTypeBoolean, PhysicalName: "archived"},
	)
	reader := &connectorExecutionReader{
		markedReader: &markedReader{t: t, objects: []*models.ObjectMetadata{account}},
		effective: []connector.ConnectorFieldMappingInput{{
			Source: "connector-a", SourceObject: "remote-account", TargetObject: "Account",
			ActionType: "upsert", SourceField: "name", TargetField: "Name",
			Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite,
		}},
		packAuthored: map[string][]connector.ConnectorFieldMappingInput{"connector-a": {{
			Source: "connector-a", SourceObject: "remote-account", TargetObject: "Account",
			ActionType: "upsert", SourceField: "name", TargetField: "Name",
			Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite,
		}}},
		mappingByKey: map[string]ConnectorFieldMappingSelection{
			"mapping/account": {
				Generation: "install-a",
				Mappings: []connector.ConnectorFieldMappingInput{{
					Source: "connector-a", SourceObject: "remote-account", TargetObject: "Account",
					ActionType: "upsert", SourceField: "name", TargetField: "Name",
					Direction: connector.DirectionInbound, ConflictPolicy: connector.ConflictPolicyOverwrite,
				}},
				OutboundMappings: []connector.ConnectorFieldMappingInput{{
					Source: "connector-a", SourceObject: "Account", TargetObject: "remote-account",
					ActionType: constants.ActionTypePushToConnector, SourceField: "Name", TargetField: "name",
					Direction: connector.DirectionOutbound,
				}},
			},
		},
		profileByKey: map[string]ConnectorCalloutProfileSelection{
			"connector-a/default": testConnectorCalloutProfileSelection("install-a"),
		},
		resolved: ConnectorPushContractSelection{
			Generation: "install-a",
			RecoverySourceEligibility: &connector.ConnectorPushRecoverySourceEligibilityContract{
				RequiredFalseBooleanFields: []string{"Archived"},
			},
			Contract: connector.ConnectorPushDurableMetadataContract{
				ConnectorOwnerKey:  "connector-a",
				ActionSlotID:       "push-account",
				ActionType:         constants.ActionTypePushToConnector,
				FlowDeveloperName:  "Account_To_Remote",
				FlowActionName:     "PushAccount",
				SourceObject:       "Account",
				DestinationSlotID:  "remote-account",
				TargetObject:       "remote-account",
				CreatePathTemplate: "/crm/v3/objects/accounts",
				UpdatePathTemplate: "/crm/v3/objects/accounts/{remoteId}",
				ResponseIDPath:     "id",
				ProfileOwnerPack:   "connector-a",
				ProfileName:        "default",
				RemoteIDField:      "RemoteId",
			},
		},
	}
	bindings := []catalogapp.IntegrationBindingSelection{
		{AuthorizedWorkloadID: "integration-service", Binding: consumerv1.IntegrationBinding{
			Kind: consumerv1.BindingConnectorFieldMapping, OwnerPack: "connector-a",
			PackInstallationID: "install-a", PackPayloadDigest: connectorExecutionTestPackDigest(), BusinessKey: "mapping/account", DefinitionVersion: 1,
		}},
		{AuthorizedWorkloadID: "integration-service", Binding: consumerv1.IntegrationBinding{
			Kind: consumerv1.BindingConnectorCalloutProfile, OwnerPack: "connector-a",
			PackInstallationID: "install-a", PackPayloadDigest: connectorExecutionTestPackDigest(), BusinessKey: "connector-a/default", DefinitionVersion: 1,
			CredentialRefs: []string{"credential-a"},
		}},
		{AuthorizedWorkloadID: "integration-service", Binding: consumerv1.IntegrationBinding{
			Kind: consumerv1.BindingConnectorPushContract, OwnerPack: "connector-a",
			PackInstallationID: "install-a", PackPayloadDigest: connectorExecutionTestPackDigest(), BusinessKey: "push/account", DefinitionVersion: 1,
		}},
	}
	selection := MetadataSelection{
		OrgID: "org-test", AppInstallationID: "installation-test",
		RuntimeCapabilityID: recordmutationv1.CapabilityName,
		IntegrationBindings: bindings,
		CatalogRevision:     41, SchemaGeneration: 42, EngineGeneration: 43,
	}
	effective, err := ReadEffectiveMetadata(
		context.Background(),
		func(ctx context.Context, fn func(context.Context) error) error {
			return fn(context.WithValue(ctx, transactionMarker{}, "one-snapshot"))
		},
		reader,
		selection,
		nil,
	)
	if err != nil {
		t.Fatal(err)
	}
	artifact, err := catalogsnapshot.CompileEffectiveMetadata(effective)
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(artifact)
	if err != nil {
		t.Fatal(err)
	}
	viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
		AppInstallationID: selection.AppInstallationID,
		CapabilityID:      recordmutationv1.CapabilityName,
	})
	if err != nil {
		t.Fatal(err)
	}
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	routes := view.OutboundIntentRoutes()
	if len(routes) != 1 || routes[0].AuthorizedWorkloadID != "integration-service" || len(routes[0].Bindings) != 3 {
		t.Fatalf("outbound intent routes = %#v", routes)
	}
	mappings := view.InboundConnectorFieldMappingRulesForTarget("Account")
	packMappings := 0
	for _, mapping := range mappings {
		if mapping.PackAuthored {
			packMappings++
		}
	}
	if len(mappings) != 2 || packMappings != 1 || mappings[0].Source != "connector-a" || mappings[0].TargetField != "Name" {
		t.Fatalf("projected connector mappings = %#v", mappings)
	}
	authorship := view.ConnectorInboundAuthorshipBindings("connector-a")
	if len(authorship) != 1 || authorship[0].Generation != "install-a" ||
		authorship[0].RemoteIDField != "RemoteId" {
		t.Fatalf("projected connector authorship = %#v", authorship)
	}
	finalization := view.ConnectorInboundFinalizationAuthorities("connector-a")
	if len(finalization) != 1 || finalization[0].Profile.CredentialRef != "credential-a" ||
		len(finalization[0].OutboundMappings) != 1 ||
		finalization[0].RecoverySourceEligibility == nil ||
		!slices.Equal(finalization[0].RecoverySourceEligibility.RequiredFalseBooleanFields, []string{"Archived"}) {
		t.Fatalf("projected connector finalization authority = %#v", finalization)
	}
	if err := view.VerifySource(snapshot); err != nil {
		t.Fatalf("connector execution view source: %v", err)
	}
}

func connectorExecutionTestPackDigest() string {
	return "nexus-pack-sha256-v1:" + strings.Repeat("a", 64)
}

func testConnectorCalloutProfileSelection(generation string) ConnectorCalloutProfileSelection {
	return ConnectorCalloutProfileSelection{
		Generation: generation,
		OutboundCapabilities: []packs.ConnectorCalloutProfileOutboundCapability{
			{Operation: "create", TargetObject: "Account", Method: "POST", PathTemplate: "/crm/v3/objects/accounts"},
			{Operation: "update", TargetObject: "Account", Method: "PATCH", PathTemplate: "/crm/v3/objects/accounts/{remoteId}"},
			{Operation: "create", TargetObject: "remote-account", Method: "POST", PathTemplate: "/crm/v3/objects/accounts"},
			{Operation: "update", TargetObject: "remote-account", Method: "PATCH", PathTemplate: "/crm/v3/objects/accounts/{remoteId}"},
		},
		Profile: catalogsnapshot.ConnectorInboundFinalizationProfile{
			Generation: generation,
			OwnerPack:  "connector-a", ProfileName: "default",
			CredentialRef: "credential-a", EndpointClass: "crm-write",
			MaxAttempts: 3, BaseDelayNanos: 1_000_000, Multiplier: 2, MaxDelayNanos: 10_000_000,
			RetryableStatusCodes: []int{429, 503},
		},
	}
}

func TestAppendConnectorExecutionDefinitionsFailsClosedWithoutConnectorReader(t *testing.T) {
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(),
		&markedReader{},
		1,
		[]catalogapp.IntegrationBindingSelection{{Binding: consumerv1.IntegrationBinding{
			Kind:      consumerv1.BindingConnectorFieldMapping,
			OwnerPack: "connector-a",
		}}},
		&input,
	)
	if err == nil || !strings.Contains(err.Error(), "connector execution metadata reader") {
		t.Fatalf("missing connector reader error = %v", err)
	}
	if len(input.MutationDefinitions) != 0 {
		t.Fatalf("missing connector reader published definitions: %#v", input.MutationDefinitions)
	}
}

func TestValidateMetadataSelectionRejectsConnectorExecutionWithoutMutationProjection(t *testing.T) {
	base := MetadataSelection{
		OrgID: "org-test", AppInstallationID: "installation-test",
		RuntimeCapabilityIDs: []string{commandsemantic.CapabilityName},
		CatalogRevision:      1, SchemaGeneration: 1, EngineGeneration: 1,
	}
	selection := base
	selection.IntegrationBindings = []catalogapp.IntegrationBindingSelection{{
		Binding: consumerv1.IntegrationBinding{Kind: consumerv1.BindingConnectorFieldMapping, OwnerPack: "connector-a"},
	}}
	if err := validateSelection(selection); err == nil || !strings.Contains(err.Error(), "requires a record-mutation") {
		t.Fatalf("field-mapping command-only connector execution selection error = %v", err)
	}

	push := base
	push.IntegrationBindings = []catalogapp.IntegrationBindingSelection{{
		Binding: consumerv1.IntegrationBinding{Kind: consumerv1.BindingConnectorPushContract, OwnerPack: "connector-a"},
	}}
	if err := validateSelection(push); err != nil {
		t.Fatalf("connector push should acquire its consumer-only mutation projection: %v", err)
	}

	allowed := base
	allowed.IntegrationBindings = []catalogapp.IntegrationBindingSelection{{
		Binding: consumerv1.IntegrationBinding{
			Kind:      consumerv1.BindingConnectorInboundRoute,
			OwnerPack: "connector-a",
		},
	}}
	if err := validateSelection(allowed); err != nil {
		t.Fatalf("route-only command selection should remain valid: %v", err)
	}
}

func TestAppendConnectorExecutionDefinitionsRejectsMixedMappingOwner(t *testing.T) {
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		mappingByKey: map[string]ConnectorFieldMappingSelection{
			"mapping/account": {
				Generation: "install-a",
				Mappings: []connector.ConnectorFieldMappingInput{{
					Source: "connector-a", SourceObject: "Remote", TargetObject: "Account",
					SourceField: "id", TargetField: "RemoteId", Direction: "inbound",
				}},
			},
		},
		packAuthored: map[string][]connector.ConnectorFieldMappingInput{
			"connector-a": {{Source: "Connector-A", SourceObject: "Remote", TargetObject: "Account", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"}},
		},
	}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(), reader, 1,
		[]catalogapp.IntegrationBindingSelection{{Binding: consumerv1.IntegrationBinding{
			Kind:               consumerv1.BindingConnectorFieldMapping,
			OwnerPack:          "connector-a",
			PackInstallationID: "install-a",
			PackPayloadDigest:  "sha256:pack-a",
			BusinessKey:        "mapping/account",
		}}},
		&input,
	)
	if err == nil || !strings.Contains(err.Error(), "returned source") {
		t.Fatalf("mixed mapping owner error = %v", err)
	}
}

func TestAppendConnectorExecutionDefinitionsOrdersCaseDistinctOwnersExactly(t *testing.T) {
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		effective: []connector.ConnectorFieldMappingInput{
			{Source: "Connector-A", SourceObject: "RemoteA", TargetObject: "Account", ActionType: "upsert", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"},
			{Source: "connector-a", SourceObject: "RemoteB", TargetObject: "Contact", ActionType: "upsert", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"},
		},
		packAuthored: map[string][]connector.ConnectorFieldMappingInput{
			"Connector-A": {{Source: "Connector-A", SourceObject: "RemoteA", TargetObject: "Account", ActionType: "upsert", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"}},
			"connector-a": {{Source: "connector-a", SourceObject: "RemoteB", TargetObject: "Contact", ActionType: "upsert", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"}},
		},
		mappingByKey: map[string]ConnectorFieldMappingSelection{
			"mapping/account": {Generation: "install-upper", Mappings: []connector.ConnectorFieldMappingInput{{Source: "Connector-A", SourceObject: "RemoteA", TargetObject: "Account", ActionType: "upsert", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"}}},
			"mapping/contact": {Generation: "install-lower", Mappings: []connector.ConnectorFieldMappingInput{{Source: "connector-a", SourceObject: "RemoteB", TargetObject: "Contact", ActionType: "upsert", SourceField: "id", TargetField: "RemoteId", Direction: "inbound"}}},
		},
	}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(), reader, 1,
		[]catalogapp.IntegrationBindingSelection{
			{Binding: consumerv1.IntegrationBinding{Kind: consumerv1.BindingConnectorFieldMapping, OwnerPack: "connector-a", PackInstallationID: "install-lower", PackPayloadDigest: "sha256:lower", BusinessKey: "mapping/contact"}},
			{Binding: consumerv1.IntegrationBinding{Kind: consumerv1.BindingConnectorFieldMapping, OwnerPack: "Connector-A", PackInstallationID: "install-upper", PackPayloadDigest: "sha256:upper", BusinessKey: "mapping/account"}},
		},
		&input,
	)
	if err != nil {
		t.Fatal(err)
	}
	if len(input.MutationDefinitions) != 4 ||
		input.MutationDefinitions[0].ConnectorInboundFieldMappingRule.Source != "Connector-A" ||
		input.MutationDefinitions[1].ConnectorInboundFieldMappingRule.Source != "connector-a" ||
		input.MutationDefinitions[2].ConnectorInboundFieldMappingRule.Source != "Connector-A" ||
		input.MutationDefinitions[3].ConnectorInboundFieldMappingRule.Source != "connector-a" {
		t.Fatalf("case-distinct owner order = %#v", input.MutationDefinitions)
	}
}

func TestAppendConnectorExecutionDefinitionsPropagatesExactGenerationFailure(t *testing.T) {
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		resolveErr:   errors.New("installed generation is stale"),
	}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(), reader, 1,
		[]catalogapp.IntegrationBindingSelection{{Binding: consumerv1.IntegrationBinding{
			Kind:               consumerv1.BindingConnectorPushContract,
			OwnerPack:          "connector-a",
			PackInstallationID: "install-a",
			PackPayloadDigest:  "sha256:pack-a",
			BusinessKey:        "push/account",
		}}},
		&input,
	)
	if err == nil || !strings.Contains(err.Error(), "installed generation is stale") {
		t.Fatalf("exact generation failure = %v", err)
	}
	if len(input.MutationDefinitions) != 0 {
		t.Fatalf("failed exact generation published definitions: %#v", input.MutationDefinitions)
	}
}

func TestAppendConnectorExecutionDefinitionsRejectsUnauthorizedProfileBeforePublication(t *testing.T) {
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		profileByKey: map[string]ConnectorCalloutProfileSelection{
			"connector-a/default": testConnectorCalloutProfileSelection("install-a"),
		},
	}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(), reader, 1,
		[]catalogapp.IntegrationBindingSelection{
			{Binding: consumerv1.IntegrationBinding{
				Kind: consumerv1.BindingConnectorCalloutProfile, OwnerPack: "connector-a",
				PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a", BusinessKey: "connector-a/default",
				CredentialRefs: []string{"different-credential"},
			}},
			{Binding: consumerv1.IntegrationBinding{
				Kind: consumerv1.BindingConnectorPushContract, OwnerPack: "connector-a",
				PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a", BusinessKey: "push/account",
			}},
		},
		&input,
	)
	if err == nil || !strings.Contains(err.Error(), "mixed or unauthorized authority") {
		t.Fatalf("unauthorized profile error = %v", err)
	}
	if len(input.MutationDefinitions) != 0 {
		t.Fatalf("unauthorized profile published definitions: %#v", input.MutationDefinitions)
	}
}

func TestAppendConnectorExecutionDefinitionsValidatesEveryWorkloadProfileBinding(t *testing.T) {
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		profileByKey: map[string]ConnectorCalloutProfileSelection{
			"connector-a/default": testConnectorCalloutProfileSelection("install-a"),
		},
	}
	valid := catalogapp.IntegrationBindingSelection{
		AuthorizedWorkloadID: "integration-a",
		Binding: consumerv1.IntegrationBinding{
			Kind: consumerv1.BindingConnectorCalloutProfile, OwnerPack: "connector-a",
			PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a", BusinessKey: "connector-a/default",
			CredentialRefs: []string{"credential-a"},
		},
	}
	unauthorized := valid
	unauthorized.AuthorizedWorkloadID = "integration-b"
	unauthorized.Binding.CredentialRefs = []string{"different-credential"}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(), reader, 1,
		[]catalogapp.IntegrationBindingSelection{
			valid,
			unauthorized,
			{Binding: consumerv1.IntegrationBinding{
				Kind: consumerv1.BindingConnectorPushContract, OwnerPack: "connector-a",
				PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a", BusinessKey: "push/account",
			}},
		},
		&input,
	)
	if err == nil || !strings.Contains(err.Error(), "mixed or unauthorized authority") {
		t.Fatalf("per-workload profile authorization error = %v", err)
	}
	if reader.profileCalls != 2 {
		t.Fatalf("profile bindings validated = %d, want 2", reader.profileCalls)
	}
	if len(input.MutationDefinitions) != 0 {
		t.Fatalf("unauthorized workload profile published definitions: %#v", input.MutationDefinitions)
	}
}

func TestAppendConnectorExecutionDefinitionsRejectsCrossWorkloadProfileJoin(t *testing.T) {
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		profileByKey: map[string]ConnectorCalloutProfileSelection{
			"connector-a/default": testConnectorCalloutProfileSelection("install-a"),
		},
		resolved: ConnectorPushContractSelection{
			Generation: "install-a",
			Contract: connector.ConnectorPushDurableMetadataContract{
				ConnectorOwnerKey: "connector-a", ActionSlotID: "push-account",
				ActionType:        constants.ActionTypePushToConnector,
				FlowDeveloperName: "Remote_To_Account", FlowActionName: "PushAccount",
				SourceObject: "RemoteA", DestinationSlotID: "account", TargetObject: "Account",
				ProfileOwnerPack: "connector-a", ProfileName: "default", RemoteIDField: "RemoteId",
			},
		},
	}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(), reader, 1,
		[]catalogapp.IntegrationBindingSelection{
			{AuthorizedWorkloadID: "integration-b", Binding: consumerv1.IntegrationBinding{
				Kind: consumerv1.BindingConnectorCalloutProfile, OwnerPack: "connector-a",
				PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a",
				BusinessKey: "connector-a/default", CredentialRefs: []string{"credential-a"},
			}},
			{AuthorizedWorkloadID: "integration-a", Binding: consumerv1.IntegrationBinding{
				Kind: consumerv1.BindingConnectorPushContract, OwnerPack: "connector-a",
				PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a",
				BusinessKey: "push/account",
			}},
		},
		&input,
	)
	if err == nil || !strings.Contains(err.Error(), `workload "integration-a" has no exact selected callout profile`) {
		t.Fatalf("cross-workload profile join error = %v", err)
	}
	if len(input.MutationDefinitions) != 0 {
		t.Fatalf("cross-workload profile join published definitions: %#v", input.MutationDefinitions)
	}
}

func TestAppendConnectorExecutionDefinitionsRejectsCrossWorkloadOutboundMappingJoin(t *testing.T) {
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		mappingByKey: map[string]ConnectorFieldMappingSelection{
			"mapping/account": {
				Generation: "install-a",
				OutboundMappings: []connector.ConnectorFieldMappingInput{{
					Source: "connector-a", SourceObject: "RemoteA", TargetObject: "Account",
					ActionType: constants.ActionTypePushToConnector, SourceField: "name", TargetField: "Name",
					Direction: connector.DirectionOutbound,
				}},
			},
		},
		profileByKey: map[string]ConnectorCalloutProfileSelection{
			"connector-a/default": testConnectorCalloutProfileSelection("install-a"),
		},
		resolved: ConnectorPushContractSelection{
			Generation: "install-a",
			Contract: connector.ConnectorPushDurableMetadataContract{
				ConnectorOwnerKey: "connector-a", ActionSlotID: "push-account",
				ActionType:        constants.ActionTypePushToConnector,
				FlowDeveloperName: "Remote_To_Account", FlowActionName: "PushAccount",
				SourceObject: "RemoteA", DestinationSlotID: "account", TargetObject: "Account",
				ProfileOwnerPack: "connector-a", ProfileName: "default", RemoteIDField: "RemoteId",
			},
		},
	}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(
		context.Background(), reader, 1,
		[]catalogapp.IntegrationBindingSelection{
			{AuthorizedWorkloadID: "integration-b", Binding: consumerv1.IntegrationBinding{
				Kind: consumerv1.BindingConnectorFieldMapping, OwnerPack: "connector-a",
				PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a",
				BusinessKey: "mapping/account",
			}},
			{AuthorizedWorkloadID: "integration-a", Binding: consumerv1.IntegrationBinding{
				Kind: consumerv1.BindingConnectorCalloutProfile, OwnerPack: "connector-a",
				PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a",
				BusinessKey: "connector-a/default", CredentialRefs: []string{"credential-a"},
			}},
			{AuthorizedWorkloadID: "integration-a", Binding: consumerv1.IntegrationBinding{
				Kind: consumerv1.BindingConnectorPushContract, OwnerPack: "connector-a",
				PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a",
				BusinessKey: "push/account",
			}},
		},
		&input,
	)
	if err == nil || !strings.Contains(err.Error(), "no exact selected outbound mapping closure") {
		t.Fatalf("cross-workload outbound mapping join error = %v", err)
	}
	if len(input.MutationDefinitions) != 0 {
		t.Fatalf("cross-workload outbound mapping join published definitions: %#v", input.MutationDefinitions)
	}
}

func TestAppendConnectorExecutionDefinitionsDoesNotPublishPartialConnectorFactsWhenFinalizationClosureIsMissing(t *testing.T) {
	inbound := connector.ConnectorFieldMappingInput{
		Source: "connector-a", SourceObject: "RemoteA", TargetObject: "Account",
		ActionType: constants.ActionTypePushToConnector, SourceField: "id", TargetField: "RemoteId",
		Direction: connector.DirectionInbound,
	}
	reader := &connectorExecutionReader{
		markedReader: &markedReader{},
		effective:    []connector.ConnectorFieldMappingInput{inbound},
		packAuthored: map[string][]connector.ConnectorFieldMappingInput{"connector-a": {inbound}},
		mappingByKey: map[string]ConnectorFieldMappingSelection{"mapping/account": {Generation: "install-a", Mappings: []connector.ConnectorFieldMappingInput{inbound}}},
		profileByKey: map[string]ConnectorCalloutProfileSelection{"connector-a/default": testConnectorCalloutProfileSelection("install-a")},
		resolved: ConnectorPushContractSelection{
			Generation: "install-a",
			Contract: connector.ConnectorPushDurableMetadataContract{
				ConnectorOwnerKey: "connector-a", ActionSlotID: "push-account",
				ActionType:        constants.ActionTypePushToConnector,
				FlowDeveloperName: "Remote_To_Account", FlowActionName: "PushAccount",
				SourceObject: "RemoteA", DestinationSlotID: "account", TargetObject: "Account",
				ProfileOwnerPack: "connector-a", ProfileName: "default", RemoteIDField: "RemoteId",
			},
		},
	}
	bindings := []catalogapp.IntegrationBindingSelection{
		{Binding: consumerv1.IntegrationBinding{Kind: consumerv1.BindingConnectorFieldMapping, OwnerPack: "connector-a", PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a", BusinessKey: "mapping/account"}},
		{Binding: consumerv1.IntegrationBinding{Kind: consumerv1.BindingConnectorCalloutProfile, OwnerPack: "connector-a", PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a", BusinessKey: "connector-a/default", CredentialRefs: []string{"credential-a"}}},
		{Binding: consumerv1.IntegrationBinding{Kind: consumerv1.BindingConnectorPushContract, OwnerPack: "connector-a", PackInstallationID: "install-a", PackPayloadDigest: "sha256:pack-a", BusinessKey: "push/account"}},
	}
	input := catalogsnapshot.CompileInput{}
	err := appendConnectorExecutionDefinitions(context.Background(), reader, 1, bindings, &input)
	if err == nil || !strings.Contains(err.Error(), "no exact selected outbound mapping closure") {
		t.Fatalf("missing outbound closure error = %v", err)
	}
	if len(input.MutationDefinitions) != 0 {
		t.Fatalf("missing finalization closure published partial connector definitions: %#v", input.MutationDefinitions)
	}
}