package automation

import (
	"context"
	"reflect"
	"strings"
	"testing"

	"github.com/nexuscrm/kernel/pkg/flowactions"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	sharedcontract "github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	outboundintentv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
)

func TestRuntimeOutboundMailDescriptorIsHostOnlyAndProtected(t *testing.T) {
	descriptors := []struct {
		descriptor FlowActionCallRegistrationDescriptor
		actionType string
		handlerRef string
	}{
		{RuntimeOutboundMailFlowActionCallRegistrationDescriptor(), flowactions.ActionTypeEnqueueOutboundMail, runtimeOutboundMailHandlerRef},
		{RuntimeOutboundMailV3FlowActionCallRegistrationDescriptor(), flowactions.ActionTypeEnqueueOutboundMailV3, runtimeOutboundMailV3HandlerRef},
	}
	for _, test := range descriptors {
		descriptor := test.descriptor
		if descriptor.ActionType != test.actionType || descriptor.HandlerRef != test.handlerRef || !descriptor.SendsEmail ||
			descriptor.AtomicPolicy != FlowActionCallAtomicPolicySafe.MetadataValue() ||
			descriptor.PreviewPolicy != FlowActionCallPreviewPolicyReject.MetadataValue() ||
			descriptor.RequiresContext != FlowActionCallContextStandardActionProtected.MetadataValue() {
			t.Fatalf("outbound mail descriptor = %#v", descriptor)
		}
		for _, registration := range PlatformFlowActionCallRegistrations() {
			if normalizeFlowActionType(registration.ActionType) == normalizeFlowActionType(descriptor.ActionType) {
				t.Fatalf("generic platform registry owns Runtime-host action %q", registration.ActionType)
			}
		}
	}
	found := map[string]bool{}
	effectiveDescriptors, err := EffectiveFlowActionCallRegistrationDescriptors()
	if err != nil {
		t.Fatalf("EffectiveFlowActionCallRegistrationDescriptors: %v", err)
	}
	for _, candidate := range effectiveDescriptors {
		found[candidate.ActionType] = true
	}
	for _, test := range descriptors {
		if !found[normalizeFlowActionType(test.actionType)] {
			t.Fatalf("metadata-facing registration closure is missing %q", test.actionType)
		}
	}
}

type runtimeOutboundMailQueryCapture struct {
	ports.QueryService
	ctx    context.Context
	spec   queryspec.QuerySpec
	user   *models.UserSession
	result *models.QueryResult
}

func (q *runtimeOutboundMailQueryCapture) QueryWithSpec(
	ctx context.Context,
	spec queryspec.QuerySpec,
	user *models.UserSession,
) (*models.QueryResult, error) {
	q.ctx, q.spec, q.user = ctx, spec, user
	return q.result, nil
}

func TestRuntimeOutboundMailSourceReadUsesExactInvokingUserProjection(t *testing.T) {
	user := &models.UserSession{ID: "user-1", ProfileID: "profile-1"}
	query := &runtimeOutboundMailQueryCapture{result: &models.QueryResult{Records: []models.SObject{{
		constants.FieldID: "003-mail", "Email": "recipient@example.com", "EmailOptOut": int64(0),
		"SecretField": "must-not-escape",
	}}}}
	executor := &FlowExecutor{query: query}
	readCtx := constants.WithStandardActionInternalAccessScopes(
		context.Background(),
		"send-record-email",
		[]constants.StandardActionInternalAccessScope{{
			StandardActionID: "send-record-email", ObjectAPIName: "Contact", Operation: constants.PermRead,
		}},
	)
	record, err := executor.loadRuntimeOutboundMailRecord(
		runtimeOutboundMailInvokingUserReadContext(readCtx),
		"Contact",
		"003-mail",
		[]string{"Email", "EmailOptOut"},
		user,
	)
	if err != nil {
		t.Fatalf("loadRuntimeOutboundMailRecord: %v", err)
	}
	if query.user != user || constants.GetSecurityContext(query.ctx) != constants.UserContext {
		t.Fatalf("invoking-user query context = user %#v security %q", query.user, constants.GetSecurityContext(query.ctx))
	}
	if constants.HasStandardActionInternalAccessScopes(query.ctx) ||
		constants.IsScopedInternalAccess(query.ctx, "Contact", constants.PermRead) {
		t.Fatal("invoking-user source query retained standard-action internal READ authority")
	}
	if query.spec.ObjectAPIName != "Contact" ||
		!reflect.DeepEqual(query.spec.SelectFields, []string{constants.FieldID, "Email", "EmailOptOut"}) ||
		!reflect.DeepEqual(query.spec.Filter, queryspec.FieldPredicate(constants.FieldID, "=", "003-mail")) ||
		query.spec.Limit != 2 || !query.spec.SkipTotalSize {
		t.Fatalf("field-bounded source query = %#v", query.spec)
	}
	if _, leaked := record["SecretField"]; leaked || len(record) != 3 {
		t.Fatalf("field-bounded source record = %#v", record)
	}
}

func TestRuntimeOutboundMailEnqueuesBoundedIntentWithRequestReplayIdentity(t *testing.T) {
	harness := newRuntimeOutboundMailHarness(t)
	registration := RuntimeOutboundMailFlowActionCallRegistration()
	for attempt := 0; attempt < 2; attempt++ {
		next, err := registration.Handler(
			harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
			harness.action, harness.user,
		)
		if err != nil || next != "done" {
			t.Fatalf("attempt %d: next=%q err=%v", attempt, next, err)
		}
	}
	if len(harness.sink.attempts) != 2 ||
		harness.sink.attempts[0].IntentID != harness.sink.attempts[1].IntentID ||
		harness.sink.attempts[0].IdempotencyKey != harness.sink.attempts[1].IdempotencyKey {
		t.Fatalf("exact request replay identities = %#v", harness.sink.attempts)
	}
	intent := harness.sink.attempts[0]
	if harness.sink.scope != (persistence.RuntimeOutboundIntentScope{
		OrgID: "org-runtime-outbound", AppInstallationID: "app-installation",
		AuthorizedWorkloadID: "integration-service",
	}) || intent.Operation != outboundintentv1.OperationSendMail || intent.SourceObject != "Contact" ||
		intent.SourceRecordID != "003-mail" || intent.CatalogRevision != 41 || len(intent.Bindings) != 1 {
		t.Fatalf("scope=%#v intent=%#v", harness.sink.scope, intent)
	}
	payload, err := outboundintentv1.DecodeMailDeliveryV1(intent)
	if err != nil {
		t.Fatalf("DecodeMailDeliveryV1: %v", err)
	}
	if payload.AuthorizedActorID != "user-1" || len(payload.ToAddresses) != 1 ||
		payload.ToAddresses[0] != "recipient@example.com" || payload.Subject != " Subject " ||
		payload.TextBody != "Plain text body\n" {
		t.Fatalf("mail payload = %#v", payload)
	}
	if harness.interview.ContextData["intentId"] != intent.IntentID ||
		harness.interview.ContextData["deliveryState"] != persistence.RuntimeOutboundIntentPending {
		t.Fatalf("action outputs = %#v", harness.interview.ContextData)
	}

	// A changed payload under the same request identity keeps the same durable
	// keys, so the immutable repository rejects it as a conflicting replay.
	harness.interview.ContextData["subject"] = "Changed subject"
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err == nil {
		t.Fatal("same request identity with changed content was accepted")
	}
	conflict := harness.sink.attempts[2]
	if conflict.IntentID != intent.IntentID || conflict.IdempotencyKey != intent.IdempotencyKey ||
		conflict.PayloadDigest == intent.PayloadDigest {
		t.Fatalf("conflicting request identity = %#v, original %#v", conflict, intent)
	}

	harness.interview.ContextData["requestId"] = "mail-request-2"
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err != nil {
		t.Fatalf("distinct request: %v", err)
	}
	if harness.sink.attempts[3].IntentID == intent.IntentID ||
		harness.sink.attempts[3].IdempotencyKey == intent.IdempotencyKey {
		t.Fatalf("distinct request reused durable identity: %#v", harness.sink.attempts[3])
	}
}

func TestRuntimeOutboundMailV2PreservesRecipientClassesAndSanitizedHTML(t *testing.T) {
	view := runtimeOutboundIntentTestViewWithMailV2RouteConfigured(t, func(message *models.ObjectMetadata) {
		minLength := 10
		bccPattern := `^bcc@example\.com$`
		htmlPattern := `^<p>Hello <strong>world</strong></p>$`
		models.FindField(message, "CcAddress").MinLength = &minLength
		models.FindField(message, "BccAddress").Regex = &bccPattern
		models.FindField(message, "HtmlBody").Regex = &htmlPattern
	})
	harness := newRuntimeOutboundMailHarnessWithView(t, view)
	harness.interview.ContextData["cc"] = "cc-one@example.com; cc-two@example.com"
	harness.interview.ContextData["bcc"] = "bcc@example.com"
	harness.interview.ContextData["htmlBody"] = `<p onclick="steal()">Hello <strong>world</strong><script>bad()</script></p>`
	harness.action.InputParameters = append(harness.action.InputParameters,
		models.FlowActionParameter{Name: "cc", Value: map[string]interface{}{"elementReference": "cc"}},
		models.FlowActionParameter{Name: "bcc", Value: map[string]interface{}{"elementReference": "bcc"}},
		models.FlowActionParameter{Name: "htmlBody", Value: map[string]interface{}{"elementReference": "htmlBody"}},
	)
	registration := RuntimeOutboundMailFlowActionCallRegistration()
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err != nil {
		t.Fatalf("mail v2: %v", err)
	}
	if len(harness.sink.attempts) != 1 {
		t.Fatalf("mail v2 intents = %d, want 1", len(harness.sink.attempts))
	}
	payload, err := outboundintentv1.DecodeMailDeliveryV2(harness.sink.attempts[0])
	if err != nil {
		t.Fatalf("DecodeMailDeliveryV2: %v", err)
	}
	if !reflect.DeepEqual(payload.CcAddresses, []string{"cc-one@example.com", "cc-two@example.com"}) ||
		!reflect.DeepEqual(payload.BccAddresses, []string{"bcc@example.com"}) ||
		payload.HTMLBody != `<p>Hello <strong>world</strong></p>` {
		t.Fatalf("mail v2 payload = %#v", payload)
	}
}

type runtimeOutboundMailAttachmentMetadataStub struct {
	metadata map[string]sharedcontract.RecordAttachmentMetadata
	err      error
}

func (s *runtimeOutboundMailAttachmentMetadataStub) AuthorizeRecordAttachmentMetadata(_ context.Context, fileID string, _ *models.UserSession) (*sharedcontract.RecordAttachmentMetadata, bool, error) {
	if s.err != nil {
		return nil, true, s.err
	}
	value, ok := s.metadata[fileID]
	if !ok {
		return nil, false, nil
	}
	return &value, true, nil
}

func TestRuntimeOutboundMailV3FreezesAuthorizedAttachmentReferences(t *testing.T) {
	view := runtimeOutboundIntentTestViewWithMailV2Route(t)
	harness := newRuntimeOutboundMailHarnessWithView(t, view)
	harness.executor.standardActionDeps.RecordAttachmentMetadata = &runtimeOutboundMailAttachmentMetadataStub{metadata: map[string]sharedcontract.RecordAttachmentMetadata{
		"file-1": {
			FileID: "file-1", FileName: "営業資料.pdf", ContentType: "application/pdf",
			ContentLength: 10, ContentDigest: "sha256:" + strings.Repeat("a", 64),
		},
	}}
	harness.interview.ContextData["attachments"] = []interface{}{"file-1"}
	harness.action.InputParameters = append(harness.action.InputParameters,
		models.FlowActionParameter{Name: "attachments", Value: map[string]interface{}{"elementReference": "attachments"}},
	)
	harness.action.ActionType = flowactions.ActionTypeEnqueueOutboundMailV3
	registration := RuntimeOutboundMailV3FlowActionCallRegistration()
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err != nil {
		t.Fatalf("mail v3: %v", err)
	}
	if len(harness.sink.attempts) != 1 {
		t.Fatalf("intents = %d", len(harness.sink.attempts))
	}
	payload, err := outboundintentv1.DecodeMailDeliveryV3(harness.sink.attempts[0])
	if err != nil {
		t.Fatal(err)
	}
	if len(payload.Attachments) != 1 || payload.Attachments[0].FileID != "file-1" ||
		payload.Attachments[0].FileName != "営業資料.pdf" {
		t.Fatalf("attachments = %#v", payload.Attachments)
	}
}

func TestRuntimeOutboundMailV3RejectsUnreadableAttachmentBeforeEnqueue(t *testing.T) {
	harness := newRuntimeOutboundMailHarnessWithView(t, runtimeOutboundIntentTestViewWithMailV2Route(t))
	harness.executor.standardActionDeps.RecordAttachmentMetadata = &runtimeOutboundMailAttachmentMetadataStub{err: context.Canceled}
	harness.interview.ContextData["attachments"] = []interface{}{"file-1"}
	harness.action.InputParameters = append(harness.action.InputParameters,
		models.FlowActionParameter{Name: "attachments", Value: map[string]interface{}{"elementReference": "attachments"}},
	)
	harness.action.ActionType = flowactions.ActionTypeEnqueueOutboundMailV3
	registration := RuntimeOutboundMailV3FlowActionCallRegistration()
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err == nil {
		t.Fatal("unreadable attachment was accepted")
	}
	if len(harness.sink.attempts) != 0 {
		t.Fatalf("intents = %#v", harness.sink.attempts)
	}
}

func TestRuntimeOutboundMailV3RejectsAttachmentCompletionOverflowBeforeFileRead(t *testing.T) {
	view := runtimeOutboundIntentTestViewWithMailV2RouteConfigured(t, func(message *models.ObjectMetadata) {
		maxLength := 5
		models.FindField(message, "AttachmentIds").MaxLength = &maxLength
	})
	harness := newRuntimeOutboundMailHarnessWithView(t, view)
	// If capacity is checked at the owning pre-delivery boundary this sentinel
	// is never observed; the request fails before file authorization or enqueue.
	harness.executor.standardActionDeps.RecordAttachmentMetadata = &runtimeOutboundMailAttachmentMetadataStub{err: context.Canceled}
	harness.interview.ContextData["attachments"] = []interface{}{"file-too-long"}
	harness.action.InputParameters = append(harness.action.InputParameters,
		models.FlowActionParameter{Name: "attachments", Value: map[string]interface{}{"elementReference": "attachments"}},
	)
	harness.action.ActionType = flowactions.ActionTypeEnqueueOutboundMailV3
	registration := RuntimeOutboundMailV3FlowActionCallRegistration()
	_, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	)
	if err == nil || !strings.Contains(err.Error(), "attachmentIds does not satisfy") {
		t.Fatalf("attachment completion capacity error = %v", err)
	}
	if len(harness.sink.attempts) != 0 {
		t.Fatalf("overflowing attachment IDs created intents: %#v", harness.sink.attempts)
	}
}

func TestRuntimeOutboundMailLegacyActionRejectsAttachmentInputBeforeEnqueue(t *testing.T) {
	harness := newRuntimeOutboundMailHarnessWithView(t, runtimeOutboundIntentTestViewWithMailV2Route(t))
	harness.interview.ContextData["attachments"] = []interface{}{"file-1"}
	harness.action.InputParameters = append(harness.action.InputParameters,
		models.FlowActionParameter{Name: "attachments", Value: map[string]interface{}{"elementReference": "attachments"}},
	)
	registration := RuntimeOutboundMailFlowActionCallRegistration()
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err == nil || !strings.Contains(err.Error(), "action contract does not support attachments") {
		t.Fatalf("legacy attachment input error = %v", err)
	}
	if len(harness.sink.attempts) != 0 {
		t.Fatalf("legacy attachment input created intents: %#v", harness.sink.attempts)
	}
}

func TestRuntimeOutboundMailV2FailsClosedBeforeEnqueue(t *testing.T) {
	for _, test := range []struct {
		name          string
		v2View        bool
		cc            string
		bcc           string
		html          string
		configureView func(*models.ObjectMetadata)
	}{
		{name: "v2 input on v1 route", cc: "cc@example.com"},
		{name: "sanitized-away html on v1 route", html: "<script>bad()</script>"},
		{name: "duplicate across recipient classes", v2View: true, cc: "recipient@example.com"},
		{name: "header injection", v2View: true, cc: "cc@example.com\r\nBcc: stolen@example.com"},
		{name: "completion field capacity", v2View: true, cc: strings.Join([]string{
			"recipient-01-with-a-long-local-part@example.com", "recipient-02-with-a-long-local-part@example.com",
			"recipient-03-with-a-long-local-part@example.com", "recipient-04-with-a-long-local-part@example.com",
			"recipient-05-with-a-long-local-part@example.com", "recipient-06-with-a-long-local-part@example.com",
		}, ";")},
		{name: "empty completion value below minimum length", v2View: true, configureView: func(message *models.ObjectMetadata) {
			minLength := 1
			models.FindField(message, "CcAddress").MinLength = &minLength
		}},
		{name: "completion value does not match regex", v2View: true, bcc: "bcc@example.com", configureView: func(message *models.ObjectMetadata) {
			pattern := `^allowed@example\.com$`
			models.FindField(message, "BccAddress").Regex = &pattern
		}},
	} {
		t.Run(test.name, func(t *testing.T) {
			view := runtimeOutboundIntentTestViewWithMailRoute(t)
			if test.v2View {
				view = runtimeOutboundIntentTestViewWithMailV2RouteConfigured(t, test.configureView)
			}
			harness := newRuntimeOutboundMailHarnessWithView(t, view)
			harness.interview.ContextData["cc"] = test.cc
			harness.interview.ContextData["bcc"] = test.bcc
			harness.interview.ContextData["htmlBody"] = test.html
			harness.action.InputParameters = append(harness.action.InputParameters,
				models.FlowActionParameter{Name: "cc", Value: map[string]interface{}{"elementReference": "cc"}},
				models.FlowActionParameter{Name: "bcc", Value: map[string]interface{}{"elementReference": "bcc"}},
				models.FlowActionParameter{Name: "htmlBody", Value: map[string]interface{}{"elementReference": "htmlBody"}},
			)
			registration := RuntimeOutboundMailFlowActionCallRegistration()
			if _, err := registration.Handler(
				harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
				harness.action, harness.user,
			); err == nil {
				t.Fatal("invalid mail v2 request was accepted")
			}
			if len(harness.sink.attempts) != 0 {
				t.Fatalf("invalid mail v2 request created intents: %#v", harness.sink.attempts)
			}
		})
	}
}

func TestSanitizeRuntimeOutboundMailHTMLKeepsFormattingWithoutActiveContent(t *testing.T) {
	got, err := sanitizeRuntimeOutboundMailHTML(`<h2 style="color:red">Title</h2><a href="https://evil.invalid">link</a><iframe>bad</iframe>`)
	if err != nil || got != `<h2>Title</h2>link` {
		t.Fatalf("sanitized HTML = %q, %v", got, err)
	}
}

func TestRuntimeOutboundMailReplayProjectsBoundedTerminalReceipt(t *testing.T) {
	for _, test := range []struct {
		name        string
		result      outboundintentv1.MailDeliveryResultV1
		wantState   string
		wantFailure string
	}{
		{
			name: "delivered",
			result: outboundintentv1.MailDeliveryResultV1{
				Version: outboundintentv1.MailResultFormatV1, Outcome: outboundintentv1.MailOutcomeDelivered,
				RemoteMessageID: "gmail-message-1", SenderAddress: "sender@example.com",
			},
			wantState: string(outboundintentv1.MailOutcomeDelivered),
		},
		{
			name: "rejected",
			result: outboundintentv1.MailDeliveryResultV1{
				Version: outboundintentv1.MailResultFormatV1, Outcome: outboundintentv1.MailOutcomeRejected,
				FailureCode: outboundintentv1.MailFailureNotConnected,
			},
			wantState: string(outboundintentv1.MailOutcomeRejected), wantFailure: outboundintentv1.MailFailureNotConnected,
		},
		{
			name: "outcome unknown",
			result: outboundintentv1.MailDeliveryResultV1{
				Version: outboundintentv1.MailResultFormatV1, Outcome: outboundintentv1.MailOutcomeUnknown,
				FailureCode: outboundintentv1.MailFailureOutcomeUnknown,
			},
			wantState: string(outboundintentv1.MailOutcomeUnknown), wantFailure: outboundintentv1.MailFailureOutcomeUnknown,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			harness := newRuntimeOutboundMailHarness(t)
			registration := RuntimeOutboundMailFlowActionCallRegistration()
			if _, err := registration.Handler(
				harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
				harness.action, harness.user,
			); err != nil {
				t.Fatalf("initial enqueue: %v", err)
			}
			intent := harness.sink.attempts[0]
			request := runtimeOutboundMailResultTestRequest(t, intent, test.result)
			row := harness.sink.rows[intent.IdempotencyKey]
			row.AdmissionState = persistence.RuntimeOutboundIntentCompleted
			row.IntegrationCommandID = request.IntegrationCommandID
			row.ResultFormatVersion = request.ResultFormatVersion
			row.ResultDigest = request.ResultDigest
			row.CanonicalResult = request.CanonicalResult
			harness.sink.rows[intent.IdempotencyKey] = row

			if _, err := registration.Handler(
				harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
				harness.action, harness.user,
			); err != nil {
				t.Fatalf("terminal replay: %v", err)
			}
			if harness.interview.ContextData["deliveryState"] != test.wantState ||
				harness.interview.ContextData["failureCode"] != test.wantFailure {
				t.Fatalf("terminal action receipt = %#v", harness.interview.ContextData)
			}
		})
	}
}

func TestRuntimeOutboundMailReplayProjectsAdmissionRejectionAndRejectsCorruptTerminalReceipt(t *testing.T) {
	harness := newRuntimeOutboundMailHarness(t)
	registration := RuntimeOutboundMailFlowActionCallRegistration()
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err != nil {
		t.Fatalf("initial enqueue: %v", err)
	}
	intent := harness.sink.attempts[0]
	row := harness.sink.rows[intent.IdempotencyKey]
	row.AdmissionState = persistence.RuntimeOutboundIntentRejected
	row.RejectionCode = "route_unavailable"
	harness.sink.rows[intent.IdempotencyKey] = row
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err != nil {
		t.Fatalf("rejected replay: %v", err)
	}
	if harness.interview.ContextData["deliveryState"] != persistence.RuntimeOutboundIntentRejected ||
		harness.interview.ContextData["failureCode"] != "" {
		t.Fatalf("rejected action receipt = %#v", harness.interview.ContextData)
	}

	request := runtimeOutboundMailResultTestRequest(t, intent, outboundintentv1.MailDeliveryResultV1{
		Version: outboundintentv1.MailResultFormatV1, Outcome: outboundintentv1.MailOutcomeDelivered,
		RemoteMessageID: "gmail-message-1", SenderAddress: "sender@example.com",
	})
	row.AdmissionState = persistence.RuntimeOutboundIntentCompleted
	row.IntegrationCommandID = request.IntegrationCommandID
	row.ResultFormatVersion = request.ResultFormatVersion
	row.ResultDigest = request.ResultDigest
	row.CanonicalResult = request.CanonicalResult
	row.RejectionCode = "stale_admission_rejection"
	harness.sink.rows[intent.IdempotencyKey] = row
	delete(harness.interview.ContextData, "deliveryState")
	delete(harness.interview.ContextData, "failureCode")
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err == nil {
		t.Fatal("corrupt terminal receipt was projected")
	}
	if _, ok := harness.interview.ContextData["deliveryState"]; ok {
		t.Fatalf("corrupt terminal receipt published action outputs: %#v", harness.interview.ContextData)
	}
}

func TestRuntimeOutboundMailFailsClosedBeforeIntentCreation(t *testing.T) {
	for _, test := range []struct {
		name   string
		mutate func(*runtimeOutboundMailHarness)
	}{
		{"missing protected context", func(h *runtimeOutboundMailHarness) {
			h.ctx = constants.WithoutStandardActionExecutionContext(h.ctx)
			h.ctx = constants.WithoutStandardActionProtectedFlowExecution(h.ctx)
		}},
		{"foreign Flow route", func(h *runtimeOutboundMailHarness) {
			h.ctx = runtimeOutboundMailTestContext(h.view, FlowExecutionIdentity{
				APIName: "Foreign_SendRecordEmail", OwnerPack: "connector_a", VersionSource: flowSourceStandard,
			})
		}},
		{"unreadable recipient field", func(h *runtimeOutboundMailHarness) {
			h.executor.fieldReadAuthorizer = mapFieldReadAuthorizer{allowed: map[string]bool{
				"contact.emailoptout": true,
			}}
		}},
		{"unreadable source record", func(h *runtimeOutboundMailHarness) {
			h.executor.recordReadAuthorizer = staticRecordReadAuthorizer{allowed: false}
		}},
		{"opted-out recipient", func(h *runtimeOutboundMailHarness) {
			h.setSourceRecord(true, "recipient@example.com")
		}},
		{"invalid opt-out representation", func(h *runtimeOutboundMailHarness) {
			h.setSourceRecord(int64(2), "recipient@example.com")
		}},
		{"invalid recipient", func(h *runtimeOutboundMailHarness) {
			h.setSourceRecord(false, "Recipient <recipient@example.com>")
		}},
		{"invalid request identity", func(h *runtimeOutboundMailHarness) {
			h.interview.ContextData["requestId"] = strings.Repeat("x", outboundintentv1.MaxIdentifierBytes+1)
		}},
	} {
		t.Run(test.name, func(t *testing.T) {
			harness := newRuntimeOutboundMailHarness(t)
			test.mutate(harness)
			registration := RuntimeOutboundMailFlowActionCallRegistration()
			if _, err := registration.Handler(
				harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
				harness.action, harness.user,
			); err == nil {
				t.Fatal("invalid mail request was accepted")
			}
			if len(harness.sink.attempts) != 0 {
				t.Fatalf("invalid mail request created intents: %#v", harness.sink.attempts)
			}
		})
	}
}

func TestRuntimeOutboundMailAcceptsTiDBBooleanRepresentation(t *testing.T) {
	harness := newRuntimeOutboundMailHarness(t)
	harness.setSourceRecord(int64(0), "recipient@example.com")
	registration := RuntimeOutboundMailFlowActionCallRegistration()
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err != nil {
		t.Fatalf("int64 false opt-out value: %v", err)
	}
	if len(harness.sink.attempts) != 1 {
		t.Fatalf("int64 false opt-out value created %d intents, want 1", len(harness.sink.attempts))
	}
}

func TestRuntimeOutboundMailRejectsInvalidDurableReceipt(t *testing.T) {
	harness := newRuntimeOutboundMailHarness(t)
	harness.sink.invalidReceipt = true
	registration := RuntimeOutboundMailFlowActionCallRegistration()
	if _, err := registration.Handler(
		harness.ctx, harness.executor, harness.interview, &models.FlowMetadata{RunAtomically: true},
		harness.action, harness.user,
	); err == nil {
		t.Fatal("foreign durable receipt was accepted")
	}
	if _, ok := harness.interview.ContextData["intentId"]; ok {
		t.Fatalf("invalid receipt published action outputs: %#v", harness.interview.ContextData)
	}
}

type runtimeOutboundMailHarness struct {
	view      *catalogsnapshot.RuntimeMetadataView
	executor  *FlowExecutor
	sink      *runtimeOutboundMailTestSink
	ctx       context.Context
	interview *models.FlowInterview
	action    *models.FlowActionCall
	user      *models.UserSession
}

func newRuntimeOutboundMailHarness(t *testing.T) *runtimeOutboundMailHarness {
	t.Helper()
	return newRuntimeOutboundMailHarnessWithView(t, runtimeOutboundIntentTestViewWithMailRoute(t))
}

func newRuntimeOutboundMailHarnessWithView(t *testing.T, view *catalogsnapshot.RuntimeMetadataView) *runtimeOutboundMailHarness {
	t.Helper()
	var route catalogsnapshot.OutboundIntentRoute
	for _, candidate := range view.OutboundIntentRoutes() {
		if candidate.Operation == outboundintentv1.OperationSendMail {
			route = candidate
			break
		}
	}
	sink := &runtimeOutboundMailTestSink{}
	harness := &runtimeOutboundMailHarness{
		view: view,
		sink: sink,
		executor: &FlowExecutor{
			runtimeOutboundIntentSink: sink,
			fieldReadAuthorizer: mapFieldReadAuthorizer{allowed: map[string]bool{
				"contact.email": true, "contact.emailoptout": true,
			}},
			recordReadAuthorizer: staticRecordReadAuthorizer{allowed: true},
		},
		ctx: runtimeOutboundMailTestContext(view, FlowExecutionIdentity{
			APIName: route.FlowDeveloperName, DefinitionID: "mail-flow", VersionID: "mail-flow-v1",
			OwnerPack: route.ConnectorOwnerKey, VersionSource: flowSourceStandard,
		}),
		interview: &models.FlowInterview{Id: "mail-interview-1", ContextData: map[string]interface{}{
			"requestId": "mail-request-1", "recordId": "003-mail", "objectApiName": "Contact",
			"subject": " Subject ", "body": "Plain text body\n",
		}},
		action: &models.FlowActionCall{
			Name: route.FlowActionName, ActionType: flowactions.ActionTypeEnqueueOutboundMail,
			InputParameters: []models.FlowActionParameter{
				{Name: "requestId", Value: map[string]interface{}{"elementReference": "requestId"}},
				{Name: "recordId", Value: map[string]interface{}{"elementReference": "recordId"}},
				{Name: "objectApiName", Value: map[string]interface{}{"elementReference": "objectApiName"}},
				{Name: "subject", Value: map[string]interface{}{"elementReference": "subject"}},
				{Name: "body", Value: map[string]interface{}{"elementReference": "body"}},
			},
			OutputParameters: []models.FlowActionParameter{
				{Name: "intentId", Value: map[string]interface{}{"elementReference": "intentId"}},
				{Name: "deliveryState", Value: map[string]interface{}{"elementReference": "deliveryState"}},
				{Name: "failureCode", Value: map[string]interface{}{"elementReference": "failureCode"}},
			},
			Connector: &models.FlowConnector{TargetReference: "done"},
		},
		user: &models.UserSession{ID: "user-1", ProfileID: "profile-1"},
	}
	harness.setSourceRecord(false, "recipient@example.com")
	return harness
}

type runtimeOutboundMailTestSink struct {
	scope          persistence.RuntimeOutboundIntentScope
	attempts       []outboundintentv1.Intent
	rows           map[string]persistence.RuntimeOutboundIntentRow
	invalidReceipt bool
}

func (s *runtimeOutboundMailTestSink) Enqueue(
	_ context.Context,
	scope persistence.RuntimeOutboundIntentScope,
	intent outboundintentv1.Intent,
) (persistence.RuntimeOutboundIntentRow, bool, error) {
	s.scope = scope
	s.attempts = append(s.attempts, intent)
	if existing, ok := s.rows[intent.IdempotencyKey]; ok {
		if existing.Scope != scope || !reflect.DeepEqual(existing.Intent, intent) {
			return existing, true, persistence.ErrRuntimeOutboundIntentConflict
		}
		return existing, true, nil
	}
	row := persistence.RuntimeOutboundIntentRow{
		Scope: scope, Intent: intent, AdmissionState: persistence.RuntimeOutboundIntentPending,
	}
	if s.rows == nil {
		s.rows = map[string]persistence.RuntimeOutboundIntentRow{}
	}
	s.rows[intent.IdempotencyKey] = row
	if s.invalidReceipt {
		row.Scope.AuthorizedWorkloadID = "foreign-workload"
	}
	return row, false, nil
}

func (h *runtimeOutboundMailHarness) setSourceRecord(optedOut interface{}, address string) {
	h.executor.connectorSourceRecordLoaderFn = staticCommittedSourceRecordLoader{records: map[string]models.SObject{
		"contact/003-mail": {
			constants.FieldID: "003-mail", "Email": address, "EmailOptOut": optedOut,
		},
	}}.load
}

func runtimeOutboundMailTestContext(
	view *catalogsnapshot.RuntimeMetadataView,
	identity FlowExecutionIdentity,
) context.Context {
	ctx := pushConnectorContext(identity)
	ctx = metadata.WithRuntimeMetadataView(ctx, view)
	ctx = constants.WithStandardActionExecutionContext(ctx, "send-record-email")
	return constants.WithStandardActionProtectedFlowExecution(ctx, identity.APIName)
}