package automation

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"

	"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/fieldcapabilities"
	"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"
)

// RuntimeOutboundIntentSink is the only write seam used by the standalone
// Runtime PushToConnector registration. Implementations must join the ambient
// Runtime transaction and must not perform provider I/O.
type RuntimeOutboundIntentSink interface {
	Enqueue(
		context.Context,
		persistence.RuntimeOutboundIntentScope,
		outboundintentv1.Intent,
	) (persistence.RuntimeOutboundIntentRow, bool, error)
}

// RuntimeOutboundResultCommitter closes result authorization, local
// projection and the terminal receipt over one caller-owned Runtime
// transaction. The projection helper is deliberately not exported: callers
// cannot bypass the persisted ACCEPTED Integration command fence.
type RuntimeOutboundResultCommitter struct {
	executor *FlowExecutor
	intents  *persistence.RuntimeOutboundIntentRepository
}

func NewRuntimeOutboundResultCommitter(
	executor *FlowExecutor,
	intents *persistence.RuntimeOutboundIntentRepository,
) (*RuntimeOutboundResultCommitter, error) {
	if executor == nil || intents == nil {
		return nil, fmt.Errorf("Runtime outbound result: executor and intent repository are required")
	}
	return &RuntimeOutboundResultCommitter{executor: executor, intents: intents}, nil
}

func (c *RuntimeOutboundResultCommitter) Commit(
	ctx context.Context,
	view *catalogsnapshot.RuntimeMetadataView,
	scope persistence.RuntimeOutboundIntentScope,
	request outboundintentv1.CommitResultRequest,
) (bool, error) {
	if c == nil || c.executor == nil || c.intents == nil || ctx == nil {
		return false, fmt.Errorf("Runtime outbound result: committer is unavailable")
	}
	if view == nil || view.Profile() != catalogsnapshot.ProjectionRecordMutationV1 ||
		view.OrgID() != scope.OrgID || view.AppInstallationID() != scope.AppInstallationID ||
		uint64(view.Lineage().CatalogRevision) != request.CatalogRevision {
		return false, fmt.Errorf("Runtime outbound result: request scope and exact Catalog view do not match")
	}
	duplicate, err := c.intents.CommitResultWithProjection(
		ctx, scope, request,
		func(ctx context.Context, row persistence.RuntimeOutboundIntentRow) error {
			return c.executor.commitRuntimeOutboundResultProjection(
				ctx, view, scope.AuthorizedWorkloadID, row.Intent, request,
			)
		},
	)
	if err != nil {
		return false, fmt.Errorf("Runtime outbound result: close accepted projection: %w", err)
	}
	return duplicate, nil
}

// EnableRuntimeOutboundIntents is called only by the standalone Runtime
// composition. Keeping registration here makes it impossible for the generic
// mutation graph constructor to silently acquire an Integration/provider
// effect owner.
func (e *FlowExecutor) EnableRuntimeOutboundIntents(sink RuntimeOutboundIntentSink) error {
	if e == nil || sink == nil {
		return fmt.Errorf("Runtime outbound intent: executor and sink are required")
	}
	if e.runtimeOutboundIntentSink != nil {
		return fmt.Errorf("Runtime outbound intent: sink is already configured")
	}
	e.runtimeOutboundIntentSink = sink
	e.RegisterFlowActionCalls(
		RuntimeOutboundIntentFlowActionCallRegistration(),
		RuntimeOutboundMailFlowActionCallRegistration(),
		RuntimeOutboundMailV3FlowActionCallRegistration(),
	)
	return nil
}

// commitRuntimeOutboundResultProjection applies the only Runtime-owned consequence of an
// Integration provider delivery: an exact Catalog-authorized compatibility
// projection onto the source record. It performs no provider I/O and accepts
// no caller-selected object or field. It is reachable only after the committer
// locks and verifies the persisted ACCEPTED Integration command.
func (e *FlowExecutor) commitRuntimeOutboundResultProjection(
	ctx context.Context,
	view *catalogsnapshot.RuntimeMetadataView,
	authorizedWorkloadID string,
	intent outboundintentv1.Intent,
	request outboundintentv1.CommitResultRequest,
) error {
	if e == nil || view == nil || view.Profile() != catalogsnapshot.ProjectionRecordMutationV1 {
		return fmt.Errorf("Runtime outbound result: exact record-mutation Catalog view is required")
	}
	if err := outboundintentv1.ValidateCommitResultRequestForIntent(intent, request); err != nil {
		return fmt.Errorf("Runtime outbound result: result does not match the immutable intent: %w", err)
	}
	if uint64(view.Lineage().CatalogRevision) != intent.CatalogRevision {
		return fmt.Errorf("Runtime outbound result: Catalog revision is invalid")
	}
	// A validated no-effect receipt closes only the accepted intent. It must
	// never read or mutate a source-record projection or invent a remote ID.
	if request.ResultFormatVersion == outboundintentv1.ResourceNoEffectResultFormatV1 {
		return nil
	}
	if intent.Operation == outboundintentv1.OperationSendMail {
		return e.commitRuntimeOutboundMailResultProjection(
			ctx, view, authorizedWorkloadID, intent, request,
		)
	}
	remoteID := strings.TrimSpace(request.RemoteID)
	route, authority, err := resolveRuntimeOutboundResultAuthority(view, authorizedWorkloadID, intent)
	if err != nil {
		return err
	}
	contract := authority.Contract
	if strings.TrimSpace(contract.RemoteIDField) == "" {
		return fmt.Errorf("Runtime outbound result: exact route has no compatibility projection field")
	}
	identity := ConnectorPushDurableIdentity{
		ConnectorOwnerKey: intent.ConnectorOwnerKey,
		ActionSlotID:      intent.ActionSlotID,
		ActionType:        contract.ActionType,
		SourceObject:      intent.SourceObject,
		SourceRecordID:    intent.SourceRecordID,
		DestinationSlotID: intent.DestinationSlotID,
	}
	linkKey, err := identity.LinkKey()
	if err != nil {
		return fmt.Errorf("Runtime outbound result: derive immutable link identity: %w", err)
	}
	return e.writeRuntimeOutboundResultProjection(ctx, view, ConnectorPushMirrorWrite{
		SourceObject: intent.SourceObject, SourceRecordID: intent.SourceRecordID,
		RemoteIDField: contract.RemoteIDField, RemoteID: remoteID,
		RemoteIDProvenanceField:   contract.RemoteIDProvenanceField,
		RemoteIDProvenance:        linkKey,
		ConnectorSyncOrigin:       route.ConnectorOwnerKey,
		SuppressedAutomationTuple: intent.SourceObject + "/" + intent.SourceRecordID,
	})
}

func (e *FlowExecutor) writeRuntimeOutboundResultProjection(
	ctx context.Context,
	view *catalogsnapshot.RuntimeMetadataView,
	write ConnectorPushMirrorWrite,
) error {
	if e == nil || e.persistence == nil || view == nil {
		return fmt.Errorf("Runtime outbound result: projection persistence and Catalog view are required")
	}
	remoteIDField := strings.TrimSpace(write.RemoteIDField)
	provenanceField := strings.TrimSpace(write.RemoteIDProvenanceField)
	if provenanceField != "" && strings.EqualFold(remoteIDField, provenanceField) {
		return fmt.Errorf("Runtime outbound result: remote identity and provenance fields must be distinct")
	}
	object, ok := view.Object(strings.TrimSpace(write.SourceObject))
	if !ok || object == nil {
		return fmt.Errorf("Runtime outbound result: source object is outside the exact Catalog view")
	}
	validateField := func(fieldAPIName, role string) error {
		field := models.FindField(object, strings.TrimSpace(fieldAPIName))
		if !runtimeOutboundResultProjectionFieldAllowed(field) {
			return fmt.Errorf("Runtime outbound result: %s field %s.%s is not an updateable persisted string in the exact Catalog view", role, object.APIName, strings.TrimSpace(fieldAPIName))
		}
		return nil
	}
	if err := validateField(remoteIDField, "remote identity"); err != nil {
		return err
	}
	if provenanceField != "" {
		if err := validateField(provenanceField, "provenance"); err != nil {
			return err
		}
	}
	ctx = metadata.WithRuntimeMetadataView(ctx, view)
	current, err := e.persistence.GetRaw(ctx, write.SourceObject, write.SourceRecordID)
	if err != nil {
		return fmt.Errorf("Runtime outbound result: read current compatibility projection: %w", err)
	}
	if current == nil {
		return fmt.Errorf("Runtime outbound result: source record is missing")
	}
	remoteAlreadyProjected := trimmedString(current[remoteIDField]) == write.RemoteID
	provenanceAlreadyProjected := provenanceField == "" ||
		trimmedString(current[provenanceField]) == write.RemoteIDProvenance
	if remoteAlreadyProjected && provenanceAlreadyProjected {
		return nil
	}
	updates := models.Record{remoteIDField: write.RemoteID}
	if provenanceField != "" {
		updates[provenanceField] = write.RemoteIDProvenance
	}
	// Unlike the legacy compatibility-mirror path, this result projection must
	// remain in the caller-owned transaction that locked the ACCEPTED intent.
	// The terminal receipt is written only after this update succeeds.
	writeCtx := constants.WithoutStandardActionExecutionContext(ctx)
	writeCtx = constants.WithoutStandardActionInternalAccessScopes(writeCtx)
	writeCtx = constants.WithoutStandardActionProtectedFlowExecution(writeCtx)
	writeCtx = constants.WithSecurityContext(writeCtx, constants.SystemContext, "Runtime outbound result projection")
	if strings.TrimSpace(write.ConnectorSyncOrigin) != "" {
		writeCtx = context.WithValue(writeCtx, constants.ContextKeyConnectorSyncOrigin, strings.TrimSpace(write.ConnectorSyncOrigin))
	}
	if strings.TrimSpace(write.SuppressedAutomationTuple) != "" {
		writeCtx = context.WithValue(writeCtx, constants.ContextKeySuppressRecordTriggeredAutomation, strings.TrimSpace(write.SuppressedAutomationTuple))
	}
	if err := e.persistence.Update(writeCtx, write.SourceObject, write.SourceRecordID, updates, newSystemAutomationUser()); err != nil {
		return fmt.Errorf("Runtime outbound result: write compatibility projection: %w", err)
	}
	return nil
}

func runtimeOutboundResultProjectionFieldAllowed(field *models.FieldMetadata) bool {
	if field == nil || !field.Updateable || field.IsSystem || field.IsNameField || field.IsCompound || field.IsVirtual ||
		(field.Formula != nil && strings.TrimSpace(*field.Formula) != "") || field.ProjectionConfig != nil ||
		!runtimeOutboundResultProjectionSourceAllowed(field.Source) ||
		connectorPushTechnicalProjectionFieldNameProtected(field.APIName) {
		return false
	}
	_, policyUpdateable := fieldcapabilities.ApplyWriteProtectionPolicy(
		false, field.Updateable, field.WriteProtectionPolicy,
	)
	if !policyUpdateable {
		return false
	}
	normalizedType := constants.NormalizeFieldType(string(field.Type))
	return normalizedType == constants.FieldTypeString &&
		normalizedType != constants.FieldTypeAutonumber && normalizedType != constants.FieldTypeRollupSummary
}

func runtimeOutboundResultProjectionSourceAllowed(source string) bool {
	switch strings.TrimSpace(source) {
	case constants.FieldSourceManaged,
		constants.FieldSourceStandardSpec,
		constants.FieldSourceTenantSpec,
		constants.FieldSourceCustom,
		constants.FieldSourceBootstrapSeed:
		return true
	default:
		return false
	}
}

func resolveRuntimeOutboundResultAuthority(
	view *catalogsnapshot.RuntimeMetadataView,
	authorizedWorkloadID string,
	intent outboundintentv1.Intent,
) (catalogsnapshot.OutboundIntentRoute, catalogsnapshot.ConnectorInboundFinalizationAuthority, error) {
	if intent.Operation != outboundintentv1.OperationUpsert {
		return catalogsnapshot.OutboundIntentRoute{}, catalogsnapshot.ConnectorInboundFinalizationAuthority{},
			fmt.Errorf("Runtime outbound result: operation %q has no UPSERT projection authority", intent.Operation)
	}
	var routes []catalogsnapshot.OutboundIntentRoute
	for _, route := range view.OutboundIntentRoutes() {
		operation := route.Operation
		if operation == "" {
			operation = outboundintentv1.OperationUpsert
		}
		if operation == intent.Operation &&
			route.AuthorizedWorkloadID == strings.TrimSpace(authorizedWorkloadID) &&
			route.ConnectorOwnerKey == intent.ConnectorOwnerKey &&
			route.FlowDeveloperName == intent.FlowDeveloperName &&
			route.FlowActionName == intent.FlowActionName &&
			route.SourceObject == intent.SourceObject &&
			route.ActionSlotID == intent.ActionSlotID &&
			route.DestinationSlotID == intent.DestinationSlotID &&
			runtimeOutboundIntentBindingsEqual(route.Bindings, intent.Bindings) {
			routes = append(routes, route)
		}
	}
	if len(routes) != 1 {
		return catalogsnapshot.OutboundIntentRoute{}, catalogsnapshot.ConnectorInboundFinalizationAuthority{},
			fmt.Errorf("Runtime outbound result: exact outbound route count is %d, want 1", len(routes))
	}
	route := routes[0]
	var authorities []catalogsnapshot.ConnectorInboundFinalizationAuthority
	for _, authority := range view.ConnectorInboundFinalizationAuthorities(route.ConnectorOwnerKey) {
		contract := authority.Contract
		if contract.FlowDeveloperName == route.FlowDeveloperName &&
			contract.FlowActionName == route.FlowActionName &&
			contract.SourceObject == route.SourceObject &&
			contract.ActionSlotID == route.ActionSlotID &&
			contract.DestinationSlotID == route.DestinationSlotID {
			authorities = append(authorities, authority)
		}
	}
	if len(authorities) != 1 {
		return catalogsnapshot.OutboundIntentRoute{}, catalogsnapshot.ConnectorInboundFinalizationAuthority{},
			fmt.Errorf("Runtime outbound result: exact projection authority count is %d, want 1", len(authorities))
	}
	return route, authorities[0], nil
}

func runtimeOutboundIntentBindingsEqual(
	left []catalogsnapshot.IntegrationBindingReference,
	right []outboundintentv1.BindingReference,
) bool {
	if len(left) != len(right) {
		return false
	}
	for index := range left {
		if string(left[index].Kind) != string(right[index].Kind) ||
			left[index].OwnerPack != right[index].OwnerPack ||
			left[index].PackInstallationID != right[index].PackInstallationID ||
			left[index].PackPayloadDigest != right[index].PackPayloadDigest ||
			left[index].BusinessKey != right[index].BusinessKey ||
			left[index].DefinitionVersion != right[index].DefinitionVersion {
			return false
		}
	}
	return true
}

// RuntimeOutboundIntentFlowActionCallRegistration preserves the existing L1
// action contract while changing only the standalone Runtime composition's
// mechanism: it records a neutral intent and never dispatches a provider call.
// The atomic policy stays aligned with already-published pack metadata. This
// PR does not broaden the launch Flow eligibility contract.
func RuntimeOutboundIntentFlowActionCallRegistration() FlowActionCallRegistration {
	descriptor := ConnectorPushFlowActionCallRegistrationDescriptor()
	return FlowActionCallRegistration{
		ActionType:    descriptor.ActionType,
		HandlerRef:    descriptor.HandlerRef,
		AtomicPolicy:  FlowActionCallAtomicPolicyReject,
		PreviewPolicy: FlowActionCallPreviewPolicyReject,
		AtomicRejectText: func(*models.FlowActionCall) string {
			return "flows executing atomically cannot contain PushToConnector action calls under the published launch profile"
		},
		Handler: func(ctx context.Context, e *FlowExecutor, interview *models.FlowInterview, flowMeta *models.FlowMetadata, action *models.FlowActionCall, user *models.UserSession) (string, error) {
			return e.executeRuntimeOutboundIntent(ctx, interview, flowMeta, action, user)
		},
	}
}

func (e *FlowExecutor) executeRuntimeOutboundIntent(
	ctx context.Context,
	interview *models.FlowInterview,
	flowMeta *models.FlowMetadata,
	action *models.FlowActionCall,
	user *models.UserSession,
) (string, error) {
	fault := func(err error) (string, error) {
		if action != nil && action.FaultConnector != nil {
			return e.routeFault(ctx, interview, action.FaultConnector.TargetReference, err)
		}
		return "", err
	}
	if e == nil || e.runtimeOutboundIntentSink == nil {
		return fault(fmt.Errorf("PushToConnector: Runtime outbound intent sink is not wired"))
	}
	identity, ok := flowRuntimeContextFromCtx(ctx).CurrentFlow()
	if !ok || !flowIdentityIsPackOwnedStandard(identity) {
		return fault(fmt.Errorf("PushToConnector: exact uncustomized pack-owned Flow identity is required"))
	}
	view, ok := metadata.RuntimeMetadataViewFromContext(ctx)
	if !ok || view == nil || view.Profile() != catalogsnapshot.ProjectionRecordMutationV1 {
		return fault(fmt.Errorf("PushToConnector: exact request-bound record-mutation Catalog view is required"))
	}
	formulaEngine, err := runtimeOutboundIntentFormulaEngine(view)
	if err != nil {
		return fault(err)
	}
	params, err := e.resolvePushToConnectorParams(ctx, interview, flowMeta, action, user)
	if err != nil {
		return fault(err)
	}
	route, authority, err := resolveRuntimeOutboundIntentAuthority(view, identity, action, params)
	if err != nil {
		return fault(err)
	}
	contract := &authority.Contract
	if err := validatePushToConnectorContractDrift(ctx, flowMeta, action, params, contract); err != nil {
		return fault(err)
	}
	sameOriginWrite := connectorSyncOriginMatchesPushToConnectorContract(ctx, identity.OwnerPack, contract)
	writeDisposition := outboundintentv1.UpsertWriteCreateOrUpdate
	connectorOriginSystemAuthorization := false
	switch {
	case sameOriginWrite && !connectorPushAllowsSameOriginPatchExisting(contract):
		return connectorPushNext(action), nil
	case sameOriginWrite:
		if err := requirePushToConnectorPostCommitRootExecution(ctx, flowMeta); err != nil {
			return fault(fmt.Errorf("PushToConnector: same-origin update-existing execution is not eligible: %w", err))
		}
		if user == nil {
			return fault(fmt.Errorf("PushToConnector: same-origin update-existing execution requires originating authority"))
		}
		writeDisposition = outboundintentv1.UpsertWriteUpdateExistingOnly
		connectorOriginSystemAuthorization = connectorPushSystemUser(user)
	default:
		if err := requirePushToConnectorSynchronousUserExecution(ctx, flowMeta, user); err != nil {
			return fault(err)
		}
	}
	mappings, err := selectPushToConnectorPayloadMappings(authority.OutboundMappings, params.PayloadSourceFields)
	if err != nil {
		return fault(err)
	}
	if len(params.PayloadSourceFields) > 0 {
		// An action-scoped sparse projection cannot authorize remote creation,
		// including when the originating execution is an ordinary user save.
		writeDisposition = outboundintentv1.UpsertWriteUpdateExistingOnly
	}
	if !connectorOriginSystemAuthorization && (e.fieldReadAuthorizer == nil || e.recordReadAuthorizer == nil) {
		return fault(fmt.Errorf("PushToConnector: invoking-user field and record read authorization must be wired"))
	}
	sourceRecordID, err := trustedPushToConnectorRootRecordID(ctx, params.SourceRecord)
	if err != nil {
		return fault(err)
	}
	current, err := e.loadRuntimeOutboundIntentRecord(ctx, params.SourceObject, sourceRecordID)
	if err != nil {
		return fault(err)
	}
	requiredFields, err := connectorPushRequiredSourceFields(
		mappings,
		connectorPushAssociationLookupFields(contract.Associations),
		formulaEngine,
	)
	if err != nil {
		return fault(err)
	}
	if !connectorOriginSystemAuthorization {
		if err := e.authorizePushToConnectorSourceFieldsWithEngine(
			ctx,
			params.SourceObject,
			mappings,
			connectorPushAssociationLookupFields(contract.Associations),
			user,
			formulaEngine,
		); err != nil {
			return fault(err)
		}
		if err := e.authorizePushToConnectorSourceRecord(ctx, params.SourceObject, current, user); err != nil {
			return fault(err)
		}
	}
	payloadRecord, err := e.pushToConnectorPayloadSourceRecord(ctx, params.SourceObject, current, requiredFields)
	if err != nil {
		return fault(err)
	}
	mapped, err := buildConnectorFieldMappingUpdatesWithEngine(
		formulaEngine,
		payloadRecord,
		models.Record{},
		mappings,
		nil,
		false,
		nil,
	)
	if err != nil {
		return fault(fmt.Errorf("PushToConnector: build provider-neutral mapped fields: %w", err))
	}
	if len(mapped) == 0 {
		return fault(fmt.Errorf("PushToConnector: outbound mapping produced an empty desired state"))
	}
	associations, err := e.runtimeOutboundIntentAssociations(
		ctx, current, contract.Associations, user, connectorOriginSystemAuthorization,
	)
	if err != nil {
		return fault(err)
	}
	mappedFields := make(map[string]json.RawMessage, len(mapped))
	for fieldName, value := range mapped {
		encoded, err := json.Marshal(value)
		if err != nil {
			return fault(fmt.Errorf("PushToConnector: encode mapped field %q: %w", fieldName, err))
		}
		mappedFields[fieldName] = encoded
	}
	payload, err := json.Marshal(outboundintentv1.UpsertDesiredStateV2{
		Version: 2, WriteDisposition: writeDisposition,
		TargetObject: strings.TrimSpace(contract.TargetObject),
		BodyRoot:     strings.TrimSpace(contract.BodyRoot), MappedFields: mappedFields,
		Associations: associations,
	})
	if err != nil {
		return fault(fmt.Errorf("PushToConnector: encode provider-neutral outbound intent: %w", err))
	}
	payloadSum := sha256.Sum256(payload)
	payloadDigest := "sha256:" + hex.EncodeToString(payloadSum[:])
	flowInvocationID := strings.TrimSpace(interview.Id)
	if flowInvocationID == "" {
		return fault(fmt.Errorf("PushToConnector: Flow invocation identity is unavailable"))
	}
	lineage := view.Lineage()
	identitySum := sha256.Sum256([]byte(strings.Join([]string{
		view.OrgID(), view.AppInstallationID(), strconv.FormatUint(uint64(lineage.CatalogRevision), 10),
		route.AuthorizedWorkloadID, route.ConnectorOwnerKey, route.FlowDeveloperName,
		route.FlowActionName, route.SourceObject, sourceRecordID, route.ActionSlotID,
		route.DestinationSlotID, flowInvocationID, payloadDigest,
	}, "\x00")))
	identityDigest := hex.EncodeToString(identitySum[:])
	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,
		}
	}
	intent := outboundintentv1.Intent{
		IntentID:             "outbound-intent-" + identityDigest,
		IdempotencyKey:       "outbound-state-" + identityDigest,
		PayloadFormatVersion: 2, PayloadDigest: payloadDigest,
		CatalogRevision: uint64(lineage.CatalogRevision), Operation: outboundintentv1.OperationUpsert,
		SourceObject: route.SourceObject, SourceRecordID: sourceRecordID,
		ConnectorOwnerKey: route.ConnectorOwnerKey,
		FlowDeveloperName: route.FlowDeveloperName, FlowActionName: route.FlowActionName,
		ActionSlotID: route.ActionSlotID, DestinationSlotID: route.DestinationSlotID,
		Bindings: bindings, CanonicalPayload: payload,
	}
	if err := outboundintentv1.ValidateIntent(intent); err != nil {
		return fault(fmt.Errorf("PushToConnector: validate provider-neutral outbound intent: %w", err))
	}
	if _, err := outboundintentv1.DecodeUpsertDesiredStateV2(intent); err != nil {
		return fault(fmt.Errorf("PushToConnector: validate provider-neutral desired state: %w", err))
	}
	if _, _, err := e.runtimeOutboundIntentSink.Enqueue(ctx, persistence.RuntimeOutboundIntentScope{
		OrgID: view.OrgID(), AppInstallationID: view.AppInstallationID(),
		AuthorizedWorkloadID: route.AuthorizedWorkloadID,
	}, intent); err != nil {
		return fault(fmt.Errorf("PushToConnector: enqueue provider-neutral outbound intent: %w", err))
	}
	return connectorPushNext(action), nil
}

func runtimeOutboundIntentFormulaEngine(view *catalogsnapshot.RuntimeMetadataView) (*formula.Engine, error) {
	if view == nil || view.Profile() != catalogsnapshot.ProjectionRecordMutationV1 {
		return nil, fmt.Errorf("PushToConnector: exact formula authority is unavailable")
	}
	engine := formula.NewEngine(formula.WithoutRuntimeFunctionBundles())
	if err := engine.RegisterFunctionBundles(view.FormulaFunctionBundles()); err != nil {
		return nil, fmt.Errorf("PushToConnector: install exact Catalog formula declarations: %w", err)
	}
	return engine, nil
}

func resolveRuntimeOutboundIntentAuthority(
	view *catalogsnapshot.RuntimeMetadataView,
	identity FlowExecutionIdentity,
	action *models.FlowActionCall,
	params pushToConnectorParams,
) (catalogsnapshot.OutboundIntentRoute, catalogsnapshot.ConnectorInboundFinalizationAuthority, error) {
	if view == nil || action == nil {
		return catalogsnapshot.OutboundIntentRoute{}, catalogsnapshot.ConnectorInboundFinalizationAuthority{}, fmt.Errorf("PushToConnector: route authority is incomplete")
	}
	var matchedRoutes []catalogsnapshot.OutboundIntentRoute
	for _, route := range view.OutboundIntentRoutes() {
		if (route.Operation == "" || route.Operation == outboundintentv1.OperationUpsert) &&
			route.ConnectorOwnerKey == strings.TrimSpace(identity.OwnerPack) &&
			route.FlowDeveloperName == strings.TrimSpace(identity.APIName) &&
			route.FlowActionName == strings.TrimSpace(action.Name) &&
			route.SourceObject == strings.TrimSpace(params.SourceObject) &&
			route.DestinationSlotID == strings.TrimSpace(params.DestinationSlotID) {
			matchedRoutes = append(matchedRoutes, route)
		}
	}
	if len(matchedRoutes) != 1 {
		return catalogsnapshot.OutboundIntentRoute{}, catalogsnapshot.ConnectorInboundFinalizationAuthority{}, fmt.Errorf("PushToConnector: exact outbound intent route count is %d, want 1", len(matchedRoutes))
	}
	route := matchedRoutes[0]
	var matchedAuthorities []catalogsnapshot.ConnectorInboundFinalizationAuthority
	for _, candidate := range view.ConnectorInboundFinalizationAuthorities(route.ConnectorOwnerKey) {
		contract := candidate.Contract
		if contract.FlowDeveloperName == route.FlowDeveloperName &&
			contract.FlowActionName == route.FlowActionName &&
			contract.SourceObject == route.SourceObject &&
			contract.ActionSlotID == route.ActionSlotID &&
			contract.DestinationSlotID == route.DestinationSlotID {
			matchedAuthorities = append(matchedAuthorities, candidate)
		}
	}
	if len(matchedAuthorities) != 1 {
		return catalogsnapshot.OutboundIntentRoute{}, catalogsnapshot.ConnectorInboundFinalizationAuthority{}, fmt.Errorf("PushToConnector: exact outbound execution authority count is %d, want 1", len(matchedAuthorities))
	}
	return route, matchedAuthorities[0], nil
}

func (e *FlowExecutor) loadRuntimeOutboundIntentRecord(ctx context.Context, objectName, recordID string) (models.Record, error) {
	var (
		record models.Record
		err    error
	)
	if e.connectorSourceRecordLoaderFn != nil {
		record, err = e.connectorSourceRecordLoaderFn(ctx, objectName, recordID)
	} else if e.persistence != nil {
		record, err = e.persistence.GetRaw(ctx, objectName, recordID)
	} else {
		return nil, fmt.Errorf("Runtime outbound intent: transaction-bound source record loader is not wired")
	}
	if err != nil {
		return nil, fmt.Errorf("Runtime outbound intent: load transaction-bound source record %s/%s: %w", objectName, recordID, err)
	}
	if record == nil {
		return nil, fmt.Errorf("Runtime outbound intent: transaction-bound source record %s/%s was not found", objectName, recordID)
	}
	if got := trimmedString(record[constants.FieldID]); got != "" && got != recordID {
		return nil, fmt.Errorf("Runtime outbound intent: source record identity drift")
	}
	record[constants.FieldID] = recordID
	models.StampObjectType(record, objectName)
	return record, nil
}

func (e *FlowExecutor) runtimeOutboundIntentAssociations(
	ctx context.Context,
	source models.Record,
	contracts []connector.ConnectorPushAssociationContract,
	user *models.UserSession,
	connectorOriginSystemAuthorization bool,
) ([]outboundintentv1.AssociationSourceV1, error) {
	result := make([]outboundintentv1.AssociationSourceV1, 0, len(contracts))
	for index, contract := range contracts {
		if contract.Junction != nil {
			return nil, fmt.Errorf("PushToConnector: outbound intent payload v1 does not support junction association resolution")
		}
		lookupField := strings.TrimSpace(contract.SourceLookupField)
		lookupValue, ok := lookupRecordFieldValue(source, lookupField)
		if !ok {
			return nil, fmt.Errorf("PushToConnector: association lookup field %q is absent from the transaction-bound source row", lookupField)
		}
		recordID := trimmedString(lookupValue)
		if recordID == "" {
			continue
		}
		related, err := e.loadRuntimeOutboundIntentRecord(ctx, strings.TrimSpace(contract.SourceObject), recordID)
		if err != nil {
			return nil, err
		}
		if !connectorOriginSystemAuthorization {
			if err := e.authorizePushToConnectorSourceRecord(ctx, strings.TrimSpace(contract.SourceObject), related, user); err != nil {
				return nil, err
			}
		}
		result = append(result, outboundintentv1.AssociationSourceV1{
			ContractIndex: uint32(index), SourceObject: strings.TrimSpace(contract.SourceObject),
			SourceRecordID: recordID,
		})
	}
	return result, nil
}