package automation

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"strings"
	"testing"
	"time"

	kernelconnector "github.com/nexuscrm/kernel/pkg/connector"
	"github.com/nexuscrm/kernel/pkg/flowactions"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

type captureRuntimeOutboundIntentSink struct {
	scope   persistence.RuntimeOutboundIntentScope
	intents []outboundintentv1.Intent
}

func (s *captureRuntimeOutboundIntentSink) Enqueue(
	_ context.Context,
	scope persistence.RuntimeOutboundIntentScope,
	intent outboundintentv1.Intent,
) (persistence.RuntimeOutboundIntentRow, bool, error) {
	duplicate := len(s.intents) != 0
	s.scope = scope
	s.intents = append(s.intents, intent)
	return persistence.RuntimeOutboundIntentRow{
		Scope: scope, Intent: intent, AdmissionState: persistence.RuntimeOutboundIntentPending,
	}, duplicate, nil
}

func TestWithRuntimeOutboundIntentSinkInstallsNeutralActionRegistration(t *testing.T) {
	sink := &captureRuntimeOutboundIntentSink{}
	exec := NewFlowExecutor(nil, nil, nil, nil, nil, nil, nil,
		WithRuntimeOutboundIntentSink(sink),
	)
	if exec.runtimeOutboundIntentSink != sink {
		t.Fatal("Runtime outbound intent sink was not installed")
	}
	registration, ok := exec.flowActionCallRegistry().Lookup(constants.ActionTypePushToConnector)
	if !ok || registration.HandlerRef != ConnectorPushFlowActionCallRegistrationDescriptor().HandlerRef ||
		registration.Handler == nil {
		t.Fatalf("Runtime outbound intent action registration = %#v, found=%v", registration, ok)
	}
	mailRegistration, ok := exec.flowActionCallRegistry().Lookup(flowactions.ActionTypeEnqueueOutboundMail)
	if !ok || mailRegistration.HandlerRef != runtimeOutboundMailHandlerRef || mailRegistration.Handler == nil {
		t.Fatalf("Runtime outbound mail action registration = %#v, found=%v", mailRegistration, ok)
	}
	mailV3Registration, ok := exec.flowActionCallRegistry().Lookup(flowactions.ActionTypeEnqueueOutboundMailV3)
	if !ok || mailV3Registration.HandlerRef != runtimeOutboundMailV3HandlerRef || mailV3Registration.Handler == nil {
		t.Fatalf("Runtime outbound mail v3 action registration = %#v, found=%v", mailV3Registration, ok)
	}
	if err := exec.EnableRuntimeOutboundIntents(&captureRuntimeOutboundIntentSink{}); err == nil {
		t.Fatal("duplicate Runtime outbound intent owner was accepted")
	}
}

func TestRuntimeOutboundIntentActionUsesExactViewAndStableDesiredStateIdentity(t *testing.T) {
	view := runtimeOutboundIntentTestViewWithMailRoute(t)
	if bundles := view.FormulaFunctionBundles(); len(bundles) != 1 || len(bundles[0].Functions) != 1 || bundles[0].Functions[0].Name != "PACK_PREFIX" {
		t.Fatalf("exact Runtime formula bundles = %#v", bundles)
	}
	sink := &captureRuntimeOutboundIntentSink{}
	exec := &FlowExecutor{
		runtimeOutboundIntentSink: sink,
		connectorSourceRecordLoaderFn: staticCommittedSourceRecordLoader{records: map[string]models.SObject{
			"contact/003-test": {constants.FieldID: "003-test", "Email": "a@example.com"},
		}}.load,
		fieldReadAuthorizer:  mapFieldReadAuthorizer{allowed: map[string]bool{"contact.email": true}},
		recordReadAuthorizer: staticRecordReadAuthorizer{allowed: true},
		formulaEngine:        formulaEngineForPushTests(),
	}
	ctx := metadata.WithRuntimeMetadataView(packOwnedPushConnectorContext(), view)
	action := pushConnectorAction(map[string]interface{}{
		"profileName": "default", "sourceObject": "Contact", "targetObject": "remoteContact",
		"sourceRecord":       map[string]interface{}{"elementReference": "$Record"},
		"actionType":         constants.ActionTypePushToConnector,
		"updatePathTemplate": "/objects/{remoteId}",
	})
	for attempt := 0; attempt < 2; attempt++ {
		next, err := exec.executeRuntimeOutboundIntent(
			ctx, pushConnectorInterview(models.SObject{constants.FieldID: "003-test", "Email": "a@example.com"}),
			pushConnectorAfterSaveFlowMeta(), action, &models.UserSession{ID: "user-1", ProfileID: "profile-1"},
		)
		if err != nil || next != "done" {
			t.Fatalf("attempt %d: next=%q err=%v", attempt, next, err)
		}
	}
	if len(sink.intents) != 2 || sink.intents[0].IntentID != sink.intents[1].IntentID ||
		sink.intents[0].IdempotencyKey != sink.intents[1].IdempotencyKey {
		t.Fatalf("replay identities = %#v", sink.intents)
	}
	intent := sink.intents[0]
	if sink.scope != (persistence.RuntimeOutboundIntentScope{
		OrgID: "org-runtime-outbound", AppInstallationID: "app-installation",
		AuthorizedWorkloadID: "integration-service",
	}) || intent.CatalogRevision != 41 || intent.FlowDeveloperName != "Connector_Push" ||
		intent.FlowActionName != "Push" || len(intent.Bindings) != 3 {
		t.Fatalf("scope=%#v intent=%#v", sink.scope, intent)
	}
	payload, err := outboundintentv1.DecodeUpsertDesiredStateV2(intent)
	if err != nil {
		t.Fatal(err)
	}
	if payload.WriteDisposition != outboundintentv1.UpsertWriteCreateOrUpdate ||
		payload.TargetObject != "remoteContact" || string(payload.MappedFields["email"]) != `"a@example.com-probe"` || len(payload.Associations) != 0 {
		t.Fatalf("payload=%#v", payload)
	}
}

func TestRuntimeOutboundIntentIdentityIsStablePerInvocationButNotPerDesiredStateForever(t *testing.T) {
	view := runtimeOutboundIntentTestView(t)
	sink := &captureRuntimeOutboundIntentSink{}
	exec := &FlowExecutor{
		runtimeOutboundIntentSink: sink,
		connectorSourceRecordLoaderFn: staticCommittedSourceRecordLoader{records: map[string]models.SObject{
			"contact/003-test": {constants.FieldID: "003-test", "Email": "a@example.com"},
		}}.load,
		fieldReadAuthorizer:  mapFieldReadAuthorizer{allowed: map[string]bool{"contact.email": true}},
		recordReadAuthorizer: staticRecordReadAuthorizer{allowed: true},
		formulaEngine:        formulaEngineForPushTests(),
	}
	ctx := metadata.WithRuntimeMetadataView(packOwnedPushConnectorContext(), view)
	action := pushConnectorAction(map[string]interface{}{
		"profileName": "default", "sourceObject": "Contact", "targetObject": "remoteContact",
		"sourceRecord": map[string]interface{}{"elementReference": "$Record"},
		"actionType":   constants.ActionTypePushToConnector, "updatePathTemplate": "/objects/{remoteId}",
	})
	for _, invocationID := range []string{"flow-invocation-1", "flow-invocation-2"} {
		interview := pushConnectorInterview(models.SObject{constants.FieldID: "003-test", "Email": "a@example.com"})
		interview.Id = invocationID
		if _, err := exec.executeRuntimeOutboundIntent(
			ctx, interview, pushConnectorAfterSaveFlowMeta(), action,
			&models.UserSession{ID: "user-1", ProfileID: "profile-1"},
		); err != nil {
			t.Fatalf("invocation %s: %v", invocationID, err)
		}
	}
	if len(sink.intents) != 2 || sink.intents[0].IntentID == sink.intents[1].IntentID ||
		sink.intents[0].IdempotencyKey == sink.intents[1].IdempotencyKey {
		t.Fatalf("distinct Flow invocations collapsed into permanent desired-state identity: %#v", sink.intents)
	}
}

func TestRuntimeOutboundIntentActionEnforcesExecutionAndSameOriginPolicy(t *testing.T) {
	action := pushConnectorAction(map[string]interface{}{
		"profileName": "default", "sourceObject": "Contact", "targetObject": "remoteContact",
		"sourceRecord": map[string]interface{}{"elementReference": "$Record"},
		"actionType":   constants.ActionTypePushToConnector, "updatePathTemplate": "/objects/{remoteId}",
	})

	t.Run("non-root execution", func(t *testing.T) {
		view := runtimeOutboundIntentTestView(t)
		exec := &FlowExecutor{runtimeOutboundIntentSink: &captureRuntimeOutboundIntentSink{}}
		ctx := pushConnectorContext(FlowExecutionIdentity{
			APIName: "Connector_Push", DefinitionID: "flow-def", VersionID: "flow-version",
			OwnerPack: "connector_a", VersionSource: flowSourceStandard,
		})
		ctx = metadata.WithRuntimeMetadataView(ctx, view)
		if _, err := exec.executeRuntimeOutboundIntent(
			ctx, pushConnectorInterview(models.SObject{constants.FieldID: "003-test"}),
			pushConnectorAfterSaveFlowMeta(), action, &models.UserSession{ID: "user-1"},
		); err == nil {
			t.Fatal("non-root execution emitted an outbound intent")
		}
	})

	t.Run("same-origin suppress", func(t *testing.T) {
		view := runtimeOutboundIntentTestView(t)
		sink := &captureRuntimeOutboundIntentSink{}
		exec := &FlowExecutor{runtimeOutboundIntentSink: sink}
		ctx := metadata.WithRuntimeMetadataView(packOwnedPushConnectorContext(), view)
		ctx = context.WithValue(ctx, constants.ContextKeyConnectorSyncOrigin, "connector_a")
		if next, err := exec.executeRuntimeOutboundIntent(
			ctx, pushConnectorInterview(models.SObject{constants.FieldID: "003-test"}),
			pushConnectorAfterSaveFlowMeta(), action, newSystemAutomationUser(),
		); err != nil || next != "done" || len(sink.intents) != 0 {
			t.Fatalf("same-origin suppress next=%q intents=%d err=%v", next, len(sink.intents), err)
		}
	})

	t.Run("same-origin update existing", func(t *testing.T) {
		view := runtimeOutboundIntentTestViewWithSameOriginPatch(t)
		sink := &captureRuntimeOutboundIntentSink{}
		exec := &FlowExecutor{
			runtimeOutboundIntentSink: sink,
			connectorSourceRecordLoaderFn: staticCommittedSourceRecordLoader{records: map[string]models.SObject{
				"contact/003-test": {constants.FieldID: "003-test", "Email": "a@example.com"},
			}}.load,
			formulaEngine: formulaEngineForPushTests(),
		}
		ctx := metadata.WithRuntimeMetadataView(packOwnedPushConnectorContext(), view)
		ctx = context.WithValue(ctx, constants.ContextKeyConnectorSyncOrigin, "connector_a")
		if next, err := exec.executeRuntimeOutboundIntent(
			ctx, pushConnectorInterview(models.SObject{constants.FieldID: "003-test", "Email": "a@example.com"}),
			pushConnectorAfterSaveFlowMeta(), action, newSystemAutomationUser(),
		); err != nil || next != "done" || len(sink.intents) != 1 {
			t.Fatalf("same-origin update next=%q intents=%d err=%v", next, len(sink.intents), err)
		}
		payload, err := outboundintentv1.DecodeUpsertDesiredStateV2(sink.intents[0])
		if err != nil || payload.WriteDisposition != outboundintentv1.UpsertWriteUpdateExistingOnly {
			t.Fatalf("same-origin desired state=%#v err=%v", payload, err)
		}
	})
}

func TestRuntimeOutboundIntentActionFailsClosedWithoutExactView(t *testing.T) {
	exec := &FlowExecutor{runtimeOutboundIntentSink: &captureRuntimeOutboundIntentSink{}}
	_, err := exec.executeRuntimeOutboundIntent(
		packOwnedPushConnectorContext(), pushConnectorInterview(models.SObject{constants.FieldID: "003-test"}),
		pushConnectorAfterSaveFlowMeta(), pushConnectorAction(map[string]interface{}{}),
		&models.UserSession{ID: "user-1"},
	)
	if err == nil {
		t.Fatal("missing request-bound Catalog view was accepted")
	}
}

// The registered action owns the provider-neutral desired state, not database
// delivery. A capture sink observes that boundary while the real sealed Catalog
// and committed-record reader determine which fields the action may emit.
func TestRuntimeOutboundIntentPayloadSelection(t *testing.T) {
	for _, tc := range []struct {
		name              string
		selection         interface{}
		sameOrigin        bool
		systemUser        bool
		suppress          bool
		missingUnselected bool
		denyField         string
		denyRecord        bool
		withoutAuthority  bool
		association       bool
		wantError         bool
	}{
		{name: "same-origin selected payload omits shell fields", selection: "OwnerShadow", sameOrigin: true, systemUser: true},
		{name: "user selected payload cannot create", selection: "OwnerShadow"},
		{name: "selected payload does not require omitted source fields", selection: "OwnerShadow", missingUnselected: true},
		{name: "unselected fields do not require read permission", selection: "OwnerShadow", denyField: "ShellLabel"},
		{name: "selected fields still require read permission", selection: "OwnerShadow", denyField: "OwnerShadow", wantError: true},
		{name: "selected payload still requires record access", selection: "OwnerShadow", denyRecord: true, wantError: true},
		{name: "selected payload still requires invoking authority", selection: "OwnerShadow", withoutAuthority: true, wantError: true},
		{name: "unmapped selection fails before emission", selection: "UnmappedField", wantError: true},
		{name: "omitted selection retains full mapping"},
		{name: "blank selection retains full mapping", selection: "  "},
		{name: "full mapping retains unselected field authorization", denyField: "ShellLabel", wantError: true},
		{name: "selection does not override same-origin suppression", selection: "OwnerShadow", sameOrigin: true, systemUser: true, suppress: true},
		{name: "selected payload retains association identity", selection: "OwnerShadow", association: true},
		{name: "selected payload retains association field authorization", selection: "OwnerShadow", association: true, denyField: "ParentId", wantError: true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			policy := connector.ConnectorPushSameOriginWritePolicyAllowPatchExisting
			if tc.suppress {
				policy = ""
			}
			view := runtimeOutboundIntentTestViewWithPolicyAndMailRoute(t, policy, false, false, nil, func(input *catalogsnapshot.CompileInput) {
				object := input.Objects[0].Metadata
				for _, field := range []string{"OwnerShadow", "ShellLabel", "ParentId"} {
					object.Fields = append(object.Fields, models.FieldMetadata{
						ID: "contact-" + field, APIName: field, Label: field,
						Type: constants.FieldTypeString, PhysicalName: field,
					})
				}
				for index := range input.MutationDefinitions {
					authority := input.MutationDefinitions[index].ConnectorInboundFinalization
					if authority == nil {
						continue
					}
					mapping := authority.OutboundMappings[0]
					mapping.SourceField, mapping.TargetField, mapping.TransformExpression = "OwnerShadow", "owner", ""
					authority.OutboundMappings = append(authority.OutboundMappings, mapping)
					mapping.SourceField, mapping.TargetField, mapping.TransformExpression = "ShellLabel", "label", "PACK_PREFIX(ShellLabel)"
					authority.OutboundMappings = append(authority.OutboundMappings, mapping)
					if tc.association {
						association := pushConnectorContractWithAssociation().Associations[0]
						association.SourceLookupField, association.SourceObject = "ParentId", "Contact"
						association.TargetActionSlotID = authority.Contract.ActionSlotID
						association.TargetDestinationSlotID = authority.Contract.DestinationSlotID
						association.TargetObject = authority.Contract.TargetObject
						authority.Contract.Associations = []connector.ConnectorPushAssociationContract{association}
						contracts := []connector.ConnectorPushDurableMetadataContract{authority.Contract}
						if err := kernelconnector.ValidateConnectorPushDurableMetadataContracts(contracts, authority.OutboundMappings); err != nil {
							t.Fatalf("resolve fixture association contract: %v", err)
						}
						authority.Contract = contracts[0]
					}
				}
			})
			record := models.SObject{
				constants.FieldID: "003-test", "OwnerShadow": "assigned-owner",
				"Email": "a@example.com", "ShellLabel": "temporary shell", "ParentId": "003-parent",
			}
			if tc.missingUnselected {
				delete(record, "Email")
				delete(record, "ShellLabel")
			}
			allowed := map[string]bool{
				"contact.email": true, "contact.ownershadow": true,
				"contact.shelllabel": true, "contact.parentid": true,
			}
			delete(allowed, "contact."+strings.ToLower(tc.denyField))
			sink := &captureRuntimeOutboundIntentSink{}
			exec := &FlowExecutor{
				runtimeOutboundIntentSink: sink,
				connectorSourceRecordLoaderFn: staticCommittedSourceRecordLoader{records: map[string]models.SObject{
					"contact/003-test": record, "contact/003-parent": {constants.FieldID: "003-parent"},
				}}.load,
				fieldReadAuthorizer:  mapFieldReadAuthorizer{allowed: allowed},
				recordReadAuthorizer: staticRecordReadAuthorizer{allowed: !tc.denyRecord},
				formulaEngine:        formulaEngineForPushTests(),
			}
			ctx := metadata.WithRuntimeMetadataView(packOwnedPushConnectorContext(), view)
			if tc.sameOrigin {
				ctx = context.WithValue(ctx, constants.ContextKeyConnectorSyncOrigin, "connector_a")
			}
			user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}
			if tc.systemUser {
				user = newSystemAutomationUser()
				exec.fieldReadAuthorizer, exec.recordReadAuthorizer = nil, nil
			}
			if tc.withoutAuthority {
				user = nil
			}
			params := map[string]interface{}{
				"profileName": "default", "sourceObject": "Contact", "targetObject": "remoteContact",
				"sourceRecord": map[string]interface{}{"elementReference": "$Record"},
				"actionType":   constants.ActionTypePushToConnector, "updatePathTemplate": "/objects/{remoteId}",
			}
			interview := pushConnectorInterview(models.SObject{constants.FieldID: "003-test", "OwnerShadow": "stale-interview-owner"})
			if tc.selection != nil {
				interview.ContextData["selectedPayloadFields"] = tc.selection
				params["payloadSourceFields"] = map[string]interface{}{"elementReference": "selectedPayloadFields"}
			}
			registration := RuntimeOutboundIntentFlowActionCallRegistration()
			next, err := registration.Handler(ctx, exec, interview, pushConnectorAfterSaveFlowMeta(), pushConnectorAction(params), user)
			if tc.wantError {
				if err == nil || len(sink.intents) != 0 {
					t.Fatalf("rejected selection emitted %d intents, err=%v", len(sink.intents), err)
				}
				return
			}
			if tc.suppress {
				if err != nil || next != "done" || len(sink.intents) != 0 {
					t.Fatalf("suppression next=%q intents=%d err=%v", next, len(sink.intents), err)
				}
				return
			}
			if err != nil || next != "done" || len(sink.intents) != 1 {
				t.Fatalf("registered action next=%q intents=%d err=%v", next, len(sink.intents), err)
			}
			intent := sink.intents[0]
			if sink.scope != (persistence.RuntimeOutboundIntentScope{OrgID: "org-runtime-outbound", AppInstallationID: "app-installation", AuthorizedWorkloadID: "integration-service"}) ||
				intent.CatalogRevision != 41 || intent.SourceRecordID != "003-test" || len(intent.Bindings) != 3 {
				t.Fatalf("selection changed the authorized intent scope: %#v %#v", sink.scope, intent)
			}
			payload, err := outboundintentv1.DecodeUpsertDesiredStateV2(intent)
			if err != nil {
				t.Fatal(err)
			}
			selected, _ := tc.selection.(string)
			wantFields := map[string]string{"owner": `"assigned-owner"`}
			wantDisposition := outboundintentv1.UpsertWriteUpdateExistingOnly
			if strings.TrimSpace(selected) == "" {
				wantFields["email"], wantFields["label"] = `"a@example.com-probe"`, `"temporary shell-probe"`
				wantDisposition = outboundintentv1.UpsertWriteCreateOrUpdate
			}
			if payload.WriteDisposition != wantDisposition || len(payload.MappedFields) != len(wantFields) {
				t.Fatalf("desired state disposition=%s fields=%v, want disposition=%s fields=%v", payload.WriteDisposition, payload.MappedFields, wantDisposition, wantFields)
			}
			for field, want := range wantFields {
				if string(payload.MappedFields[field]) != want {
					t.Fatalf("mapped %s=%s, want %s", field, payload.MappedFields[field], want)
				}
			}
			if tc.association {
				if len(payload.Associations) != 1 || payload.Associations[0] != (outboundintentv1.AssociationSourceV1{ContractIndex: 0, SourceObject: "Contact", SourceRecordID: "003-parent"}) {
					t.Fatalf("selected payload lost contract association: %#v", payload.Associations)
				}
			} else if len(payload.Associations) != 0 {
				t.Fatalf("selected payload invented associations: %#v", payload.Associations)
			}
		})
	}
}

func TestCommitRuntimeOutboundResultUsesExactCatalogProjectionAndSuppressesEcho(t *testing.T) {
	view := runtimeOutboundIntentTestView(t)
	persistence := &capturePushMirrorPersistence{stubPersistenceCapture: &stubPersistenceCapture{rawRecords: map[string]models.SObject{
		"Contact/003-test": {constants.FieldID: "003-test"},
	}}}
	exec := &FlowExecutor{
		persistence: persistence,
		fieldMetaLookup: func(objectAPIName, fieldAPIName string) *models.FieldMetadata {
			if objectAPIName == "Contact" && fieldAPIName == "RemoteKey__c" {
				return &models.FieldMetadata{APIName: "RemoteKey__c", Type: constants.FieldTypeString, Updateable: true}
			}
			return nil
		},
	}
	intent := runtimeOutboundIntentTestIntent(view)
	request := runtimeOutboundIntentTestCommitRequest(intent, "remote-42")
	if err := exec.commitRuntimeOutboundResultProjection(t.Context(), view, "integration-service", intent, request); err != nil {
		t.Fatal(err)
	}
	if len(persistence.calls) != 1 {
		t.Fatalf("projection calls = %d, want 1", len(persistence.calls))
	}
	call := persistence.calls[0]
	if call.objectName != "Contact" || call.id != "003-test" ||
		call.updates["RemoteKey__c"] != "remote-42" || call.detached ||
		call.securityContext != constants.SystemContext || call.userID != constants.SystemUserID ||
		call.connectorSyncOrigin != "connector_a" || call.suppressedRow != "Contact/003-test" {
		t.Fatalf("projection call = %#v", call)
	}
	persistence.rawRecords["Contact/003-test"] = models.SObject{constants.FieldID: "003-test", "RemoteKey__c": "remote-42"}
	if err := exec.commitRuntimeOutboundResultProjection(t.Context(), view, "integration-service", intent, request); err != nil {
		t.Fatal(err)
	}
	if len(persistence.calls) != 1 {
		t.Fatalf("idempotent retry performed %d writes, want 1", len(persistence.calls))
	}
}

func TestRuntimeOutboundResultProjectionRejectsIntrinsicallyReadonlyFields(t *testing.T) {
	formulaText := "Email"
	for name, field := range map[string]*models.FieldMetadata{
		"formula": {
			APIName: "RemoteKey__c", Type: constants.FieldTypeString,
			Updateable: true, Source: constants.FieldSourceCustom, Formula: &formulaText,
		},
		"projection": {
			APIName: "RemoteKey__c", Type: constants.FieldTypeString,
			Updateable: true, Source: constants.FieldSourceCustom, ProjectionConfig: &models.FieldProjectionConfig{},
		},
		"rollup": {
			APIName: "RemoteKey__c", Type: constants.FieldTypeRollupSummary,
			Updateable: true, Source: constants.FieldSourceCustom,
		},
		"write protection": {
			APIName: "RemoteKey__c", Type: constants.FieldTypeString,
			Updateable: true, Source: constants.FieldSourceCustom,
			WriteProtectionPolicy: constants.FieldWriteProtectionTrustedOnly,
		},
		"discovered source": {
			APIName: "RemoteKey__c", Type: constants.FieldTypeString,
			Updateable: true, Source: constants.FieldSourceDiscovered,
		},
		"missing source": {
			APIName: "RemoteKey__c", Type: constants.FieldTypeString, Updateable: true,
		},
		"unknown source": {
			APIName: "RemoteKey__c", Type: constants.FieldTypeString,
			Updateable: true, Source: "future_provenance",
		},
	} {
		t.Run(name, func(t *testing.T) {
			if runtimeOutboundResultProjectionFieldAllowed(field) {
				t.Fatalf("intrinsically read-only field was accepted: %#v", field)
			}
		})
	}
}

func TestRuntimeOutboundResultProjectionRejectsAliasedIdentityAndProvenanceFields(t *testing.T) {
	view := runtimeOutboundIntentTestView(t)
	persistence := &capturePushMirrorPersistence{stubPersistenceCapture: &stubPersistenceCapture{rawRecords: map[string]models.SObject{
		"Contact/003-test": {constants.FieldID: "003-test"},
	}}}
	exec := &FlowExecutor{persistence: persistence}
	err := exec.writeRuntimeOutboundResultProjection(t.Context(), view, ConnectorPushMirrorWrite{
		SourceObject: "Contact", SourceRecordID: "003-test",
		RemoteIDField: "RemoteKey__c", RemoteID: "remote-42",
		RemoteIDProvenanceField: "remotekey__C", RemoteIDProvenance: "link-42",
	})
	if err == nil || len(persistence.calls) != 0 {
		t.Fatalf("aliased result projection = err %v writes %d", err, len(persistence.calls))
	}
}

func TestCommitRuntimeOutboundResultFailsClosedOnBindingOrRevisionDrift(t *testing.T) {
	view := runtimeOutboundIntentTestView(t)
	persistence := &capturePushMirrorPersistence{stubPersistenceCapture: &stubPersistenceCapture{}}
	exec := &FlowExecutor{persistence: persistence, fieldMetaLookup: func(_, _ string) *models.FieldMetadata {
		return &models.FieldMetadata{APIName: "RemoteKey__c", Type: constants.FieldTypeString, Updateable: true}
	}}
	for name, mutate := range map[string]func(*outboundintentv1.Intent){
		"revision": func(intent *outboundintentv1.Intent) { intent.CatalogRevision++ },
		"binding":  func(intent *outboundintentv1.Intent) { intent.Bindings[0].BusinessKey = "connector_a/other" },
	} {
		t.Run(name, func(t *testing.T) {
			intent := runtimeOutboundIntentTestIntent(view)
			mutate(&intent)
			request := runtimeOutboundIntentTestCommitRequest(intent, "remote-42")
			if err := exec.commitRuntimeOutboundResultProjection(t.Context(), view, "integration-service", intent, request); err == nil {
				t.Fatal("drifted result projection was accepted")
			}
		})
	}
	if len(persistence.calls) != 0 {
		t.Fatalf("drifted authority performed %d writes", len(persistence.calls))
	}
}

func TestResolveRuntimeOutboundResultAuthorityRejectsNonUpsertOperation(t *testing.T) {
	view := runtimeOutboundIntentTestViewWithMailRoute(t)
	if routes := view.OutboundIntentRoutes(); len(routes) != 2 || routes[0].Operation != outboundintentv1.OperationSendMail {
		t.Fatalf("mail result-isolation fixture routes = %#v", routes)
	}
	intent := runtimeOutboundIntentTestIntent(view)
	intent.Operation = outboundintentv1.OperationSendMail
	if _, _, err := resolveRuntimeOutboundResultAuthority(view, "integration-service", intent); err == nil {
		t.Fatal("SEND_MAIL intent reached UPSERT result projection authority")
	}
}

func runtimeOutboundIntentTestIntent(view *catalogsnapshot.RuntimeMetadataView) outboundintentv1.Intent {
	route := view.OutboundIntentRoutes()[0]
	bindings := make([]outboundintentv1.BindingReference, len(route.Bindings))
	for index, binding := range route.Bindings {
		bindings[index] = outboundintentv1.BindingReference{
			Kind: binding.Kind, OwnerPack: binding.OwnerPack, PackInstallationID: binding.PackInstallationID,
			PackPayloadDigest: binding.PackPayloadDigest, BusinessKey: binding.BusinessKey,
			DefinitionVersion: binding.DefinitionVersion,
		}
	}
	payload := outboundintentv1.UpsertDesiredStateV1{Version: 1, TargetObject: "remoteContact", MappedFields: map[string]json.RawMessage{"email": json.RawMessage(`"a@example.com"`)}}
	canonical, _ := json.Marshal(payload)
	digest := sha256.Sum256(canonical)
	return outboundintentv1.Intent{
		IntentID: "intent-1", IdempotencyKey: "intent-key-1", PayloadFormatVersion: 1,
		PayloadDigest: "sha256:" + hex.EncodeToString(digest[:]), CatalogRevision: uint64(view.Lineage().CatalogRevision),
		Operation: outboundintentv1.OperationUpsert, SourceObject: route.SourceObject,
		SourceRecordID: "003-test", ConnectorOwnerKey: route.ConnectorOwnerKey,
		FlowDeveloperName: route.FlowDeveloperName, FlowActionName: route.FlowActionName,
		ActionSlotID: route.ActionSlotID, DestinationSlotID: route.DestinationSlotID,
		Bindings: bindings, CanonicalPayload: canonical,
	}
}

func runtimeOutboundIntentTestCommitRequest(
	intent outboundintentv1.Intent,
	remoteID string,
) outboundintentv1.CommitResultRequest {
	return outboundintentv1.CommitResultRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest,
		CatalogRevision:      intent.CatalogRevision,
		IntegrationCommandID: "integration-command-1", RemoteID: remoteID,
	}
}

func runtimeOutboundIntentTestView(t *testing.T) *catalogsnapshot.RuntimeMetadataView {
	return runtimeOutboundIntentTestViewWithPolicy(t, "")
}

func runtimeOutboundIntentTestViewWithSameOriginPatch(t *testing.T) *catalogsnapshot.RuntimeMetadataView {
	return runtimeOutboundIntentTestViewWithPolicy(t, connector.ConnectorPushSameOriginWritePolicyAllowPatchExisting)
}

func runtimeOutboundIntentTestViewWithPolicy(t *testing.T, sameOriginWritePolicy string) *catalogsnapshot.RuntimeMetadataView {
	return runtimeOutboundIntentTestViewWithPolicyAndMailRoute(t, sameOriginWritePolicy, false, false, nil)
}

func runtimeOutboundIntentTestViewWithMailRoute(t *testing.T) *catalogsnapshot.RuntimeMetadataView {
	return runtimeOutboundIntentTestViewWithPolicyAndMailRoute(t, "", true, false, nil)
}

func runtimeOutboundIntentTestViewWithMailV2Route(t *testing.T) *catalogsnapshot.RuntimeMetadataView {
	return runtimeOutboundIntentTestViewWithMailV2RouteConfigured(t, nil)
}

func runtimeOutboundIntentTestViewWithMailV2RouteConfigured(
	t *testing.T,
	configure func(*models.ObjectMetadata),
) *catalogsnapshot.RuntimeMetadataView {
	return runtimeOutboundIntentTestViewWithPolicyAndMailRoute(t, "", true, true, configure)
}

func runtimeOutboundIntentTestViewWithPolicyAndMailRoute(
	t *testing.T,
	sameOriginWritePolicy string,
	includeMailRoute bool,
	mailV2 bool,
	configureMailV2 func(*models.ObjectMetadata),
	configure ...func(*catalogsnapshot.CompileInput),
) *catalogsnapshot.RuntimeMetadataView {
	t.Helper()
	revision := catalogsnapshot.CatalogRevision(41)
	contract := *defaultPushConnectorContract()
	contract.SameOriginWritePolicy = sameOriginWritePolicy
	mappings := defaultPushConnectorMappings()
	bundle, err := formula.ParseFunctionBundlePayload([]byte(`{
		"functions": [{
			"name": "PACK_PREFIX",
			"minArgs": 1,
			"maxArgs": 1,
			"return": {"kind": "fixed", "type": "string"},
			"args": [{"type": "string", "nullable": true}],
			"evaluator": {
				"kind": "formula_expr_v1",
				"params": ["value"],
				"body": "value + \"-probe\""
			}
		}]
	}`))
	if err != nil {
		t.Fatalf("parse formula bundle: %v", err)
	}
	mappings[0].TransformExpression = "PACK_PREFIX(Email)"
	input := catalogsnapshot.CompileInput{
		OrgID:   "org-runtime-outbound",
		Lineage: catalogsnapshot.Lineage{CatalogRevision: revision, SchemaGeneration: 7, EngineGeneration: 9},
		Objects: []catalogsnapshot.ObjectInput{{CatalogRevision: revision, Metadata: &models.ObjectMetadata{
			ID: "contact-object", APIName: "Contact", Label: "Contact", PluralLabel: "Contacts",
			SharingModel: constants.SharingModelPublicReadWrite,
			Fields: []models.FieldMetadata{
				{ID: "contact-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeString, PhysicalName: "Id"},
				{ID: "contact-email", APIName: "Email", Label: "Email", Type: constants.FieldTypeString, PhysicalName: "Email"},
				{ID: "contact-remote-key", APIName: "RemoteKey__c", Label: "Remote Key", Type: constants.FieldTypeString, PhysicalName: "RemoteKey__c", Source: constants.FieldSourceCustom, Updateable: true},
			},
		}}},
		FormulaFunctionBundles: []catalogsnapshot.FormulaFunctionBundleInput{{
			CatalogRevision: revision,
			Bundle:          bundle,
		}},
		MutationDefinitions: []catalogsnapshot.MutationDefinitionInput{
			{CatalogRevision: revision, ConnectorInboundAuthorship: &catalogsnapshot.ConnectorInboundAuthorshipBinding{
				ConnectorOwnerKey: contract.ConnectorOwnerKey, Generation: "contract-installation",
				ActionSlotID: contract.ActionSlotID, ActionType: contract.ActionType,
				SourceObject: contract.SourceObject, DestinationSlotID: contract.DestinationSlotID,
				TargetObject: contract.TargetObject, ProfileOwnerPack: contract.ProfileOwnerPack,
				ProfileName: contract.ProfileName, RemoteIDField: contract.RemoteIDField,
			}},
			{CatalogRevision: revision, ConnectorInboundFinalization: &catalogsnapshot.ConnectorInboundFinalizationAuthority{
				ContractGeneration: "contract-installation", Contract: contract,
				Profile: catalogsnapshot.ConnectorInboundFinalizationProfile{
					Generation: "profile-installation", OwnerPack: contract.ProfileOwnerPack,
					ProfileName: contract.ProfileName, CredentialRef: "credential-ref",
					EndpointClass: "crm.objects.write", MaxAttempts: 3,
					BaseDelayNanos: int64(time.Second), Multiplier: 2, MaxDelayNanos: int64(10 * time.Second),
					RetryableStatusCodes: []int{429, 503},
				},
				OutboundMappings: mappings,
			}},
		},
		OutboundIntentRoutes: []catalogsnapshot.OutboundIntentRouteInput{{
			CatalogRevision: revision,
			Route: catalogsnapshot.OutboundIntentRoute{
				AuthorizedWorkloadID: "integration-service", ConnectorOwnerKey: contract.ConnectorOwnerKey,
				FlowDeveloperName: contract.FlowDeveloperName, FlowActionName: contract.FlowActionName,
				SourceObject: contract.SourceObject, ActionSlotID: contract.ActionSlotID,
				DestinationSlotID: contract.DestinationSlotID,
				Bindings: []catalogsnapshot.IntegrationBindingReference{
					{Kind: "CONNECTOR_CALLOUT_PROFILE", OwnerPack: "connector_a", PackInstallationID: "profile-installation", PackPayloadDigest: "nexus-pack-sha256-v1:" + string(bytes.Repeat([]byte{'a'}, 64)), BusinessKey: "connector_a/default", DefinitionVersion: 1},
					{Kind: "CONNECTOR_FIELD_MAPPING", OwnerPack: "connector_a", PackInstallationID: "contract-installation", PackPayloadDigest: "nexus-pack-sha256-v1:" + string(bytes.Repeat([]byte{'b'}, 64)), BusinessKey: "Contact/remoteContact/PushToConnector", DefinitionVersion: 1},
					{Kind: "CONNECTOR_PUSH_CONTRACT", OwnerPack: "connector_a", PackInstallationID: "contract-installation", PackPayloadDigest: "nexus-pack-sha256-v1:" + string(bytes.Repeat([]byte{'b'}, 64)), BusinessKey: "connector_a/Connector_Push/Push/Contact/remote-contact", DefinitionVersion: 1},
				},
			},
		}},
		Projections: []catalogsnapshot.RuntimeProjectionInput{{
			CatalogRevision: revision, AppInstallationID: "app-installation",
			CapabilityID: "runtime.records.mutate.v1", Profile: catalogsnapshot.ProjectionRecordMutationV1,
			ObjectAPINames: []string{"Contact"},
		}},
	}
	if includeMailRoute {
		input.Objects[0].Metadata.Fields = append(input.Objects[0].Metadata.Fields, models.FieldMetadata{
			ID: "contact-email-opt-out", APIName: "EmailOptOut", Label: "Email Opt Out",
			Type: constants.FieldTypeBoolean, PhysicalName: "EmailOptOut",
		})
		legacyRoute := input.OutboundIntentRoutes[0].Route
		completion := runtimeOutboundMailCompletionTestAuthority()
		if mailV2 {
			completion.Message.CcAddressField = "CcAddress"
			completion.Message.BccAddressField = "BccAddress"
			completion.Message.HTMLBodyField = "HtmlBody"
			completion.Message.AttachmentIDsField = "AttachmentIds"
		}
		input.OutboundIntentRoutes = append(input.OutboundIntentRoutes, catalogsnapshot.OutboundIntentRouteInput{
			CatalogRevision: revision,
			Route: catalogsnapshot.OutboundIntentRoute{
				AuthorizedWorkloadID: legacyRoute.AuthorizedWorkloadID,
				ConnectorOwnerKey:    legacyRoute.ConnectorOwnerKey, FlowDeveloperName: legacyRoute.FlowDeveloperName,
				FlowActionName: legacyRoute.FlowActionName, SourceObject: legacyRoute.SourceObject,
				ActionSlotID: legacyRoute.ActionSlotID, DestinationSlotID: legacyRoute.DestinationSlotID,
				Operation: outboundintentv1.OperationSendMail,
				Bindings: []catalogsnapshot.IntegrationBindingReference{{
					Kind: "CONNECTOR_CALLOUT_PROFILE", OwnerPack: "connector_a",
					PackInstallationID: "profile-installation", PackPayloadDigest: "nexus-pack-sha256-v1:" + string(bytes.Repeat([]byte{'a'}, 64)),
					BusinessKey: "connector_a/default", DefinitionVersion: 1,
				}},
				MailCompletion: completion,
			},
		})
		input.Objects = append(input.Objects, runtimeOutboundMailTestObjects(revision)...)
		if mailV2 {
			message := input.Objects[len(input.Objects)-1].Metadata
			message.Fields = append(message.Fields,
				runtimeOutboundMailField("message-cc", "CcAddress", constants.FieldTypeString),
				runtimeOutboundMailField("message-bcc", "BccAddress", constants.FieldTypeString),
				runtimeOutboundMailField("message-html", "HtmlBody", constants.FieldTypeRichtextarea),
				runtimeOutboundMailField("message-attachments", "AttachmentIds", constants.FieldTypeTextarea),
			)
			if configureMailV2 != nil {
				configureMailV2(message)
			}
		}
		input.Projections[0].ObjectAPINames = append(input.Projections[0].ObjectAPINames, "ActivityLog", "MessageLog")
	}
	for _, apply := range configure {
		apply(&input)
	}
	artifact, err := catalogsnapshot.CompileCatalogSnapshot(input)
	if err != nil {
		t.Fatalf("compile Catalog snapshot: %v", err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(artifact)
	if err != nil {
		t.Fatal(err)
	}
	viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
		AppInstallationID: "app-installation", CapabilityID: "runtime.records.mutate.v1",
	})
	if err != nil {
		t.Fatal(err)
	}
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	return view
}

func runtimeOutboundMailCompletionTestAuthority() *catalogsnapshot.OutboundMailCompletionProjection {
	return &catalogsnapshot.OutboundMailCompletionProjection{
		RecipientAddressField: "Email", RecipientOptOutField: "EmailOptOut",
		Activity: catalogsnapshot.OutboundMailActivityProjection{
			ObjectAPIName: "ActivityLog", SourceRecordField: "SourceRecordId", SubjectField: "Subject",
			ActivityDateField: "ActivityDate", StatusField: "Status", PriorityField: "Priority",
			BodyField: "Body", OwnerField: "OwnerId", TypeField: "Type", SubtypeField: "Subtype",
			FixedSubject: "Mail", FixedStatus: "Completed", FixedPriority: "Normal",
			FixedType: "Mail", FixedSubtype: "Outbound",
		},
		Message: catalogsnapshot.OutboundMailMessageProjection{
			ObjectAPIName: "MessageLog", ActivityIDField: "ActivityId", FromAddressField: "FromAddress",
			ToAddressField: "ToAddress", SubjectField: "Subject", TextBodyField: "TextBody",
			IncomingField: "Incoming", StatusField: "Status", MessageDateField: "MessageDate",
			FixedStatus: "Delivered",
		},
	}
}

func runtimeOutboundMailTestObjects(revision catalogsnapshot.CatalogRevision) []catalogsnapshot.ObjectInput {
	return []catalogsnapshot.ObjectInput{
		{CatalogRevision: revision, Metadata: &models.ObjectMetadata{
			ID: "activity-log", APIName: "ActivityLog", Label: "Activity Log", PluralLabel: "Activity Logs",
			SharingModel: constants.SharingModelPublicReadWrite,
			Fields: []models.FieldMetadata{
				runtimeOutboundMailReferenceField("activity-source", "SourceRecordId", "Contact"),
				runtimeOutboundMailField("activity-subject", "Subject", constants.FieldTypeString),
				runtimeOutboundMailField("activity-date", "ActivityDate", constants.FieldTypeDate),
				runtimeOutboundMailField("activity-status", "Status", constants.FieldTypeString),
				runtimeOutboundMailField("activity-priority", "Priority", constants.FieldTypeString),
				runtimeOutboundMailField("activity-body", "Body", constants.FieldTypeTextarea),
				runtimeOutboundMailField("activity-owner", "OwnerId", constants.FieldTypeReference),
				runtimeOutboundMailField("activity-type", "Type", constants.FieldTypeString),
				runtimeOutboundMailField("activity-subtype", "Subtype", constants.FieldTypeString),
			},
		}},
		{CatalogRevision: revision, Metadata: &models.ObjectMetadata{
			ID: "message-log", APIName: "MessageLog", Label: "Message Log", PluralLabel: "Message Logs",
			SharingModel: constants.SharingModelPublicReadWrite,
			Fields: []models.FieldMetadata{
				runtimeOutboundMailReferenceField("message-activity", "ActivityId", "ActivityLog"),
				runtimeOutboundMailField("message-from", "FromAddress", constants.FieldTypeEmail),
				runtimeOutboundMailField("message-to", "ToAddress", constants.FieldTypeEmail),
				runtimeOutboundMailField("message-subject", "Subject", constants.FieldTypeString),
				runtimeOutboundMailField("message-body", "TextBody", constants.FieldTypeTextarea),
				runtimeOutboundMailField("message-incoming", "Incoming", constants.FieldTypeBoolean),
				runtimeOutboundMailField("message-status", "Status", constants.FieldTypeString),
				runtimeOutboundMailField("message-date", "MessageDate", constants.FieldTypeDatetime),
			},
		}},
	}
}

func runtimeOutboundMailField(id, name string, fieldType constants.SchemaFieldType) models.FieldMetadata {
	return models.FieldMetadata{ID: id, APIName: name, Label: name, Type: fieldType, Createable: true}
}

func runtimeOutboundMailReferenceField(id, name, target string) models.FieldMetadata {
	field := runtimeOutboundMailField(id, name, constants.FieldTypeReference)
	field.ReferenceTo = []string{target}
	return field
}