package runtimeapi

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"io"
	"net/http"
	"strings"
	"testing"
	"time"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	sharedcontract "github.com/nexuscrm/shared/pkg/contract"
	"github.com/pingcap-inc/nexuscrm-api-host/processhttp"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	commonsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
	outboundintentsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/outboundintentv1"
	recordsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordsv1"
	runtimev1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/runtime/v1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/mapping"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	grpcmetadata "google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
)

func TestOutboundIntentAdmissionBindsExactCurrentConsumerAuthority(t *testing.T) {
	view := outboundIntentTestView(t, 7)
	current := commandConsumerIntegrationView(view)
	engine := &catalogConsumerViewTestEngine{integrationView: current}
	admission, err := NewOutboundIntentAdmission(t.Context(), OutboundIntentSelection{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		WorkloadID: testWorkloadID, RuntimeMetadataView: view,
	}, engine)
	if err != nil {
		t.Fatal(err)
	}
	bound, _, err := bindIntegrationConsumerAuthority(
		outboundIntentTestContext(t, testWorkloadID, 3),
		integrationConsumerSelection{
			orgID: testOrgID, appInstallationID: testInstallationID,
			workloadID: testWorkloadID, capability: outboundintentsemantic.CapabilityName,
			runtimeView: view,
		},
		admission.prepared().current,
	)
	if err != nil {
		t.Fatal(err)
	}
	scope, ok := runtimeExecutionScopeFromContext(bound)
	if !ok || scope.view != view || scope.workloadID != testWorkloadID ||
		scope.capability != outboundintentsemantic.CapabilityName ||
		scope.receipts.CatalogRevision != 7 ||
		scope.receipts.CapabilityVersion != outboundintentsemantic.CapabilityName {
		t.Fatalf("outbound admission scope = %#v", scope)
	}
	if len(engine.integrationScopes) != 1 {
		t.Fatalf("Integration authority reads = %d, want 1", len(engine.integrationScopes))
	}
	if _, _, err := bindIntegrationConsumerAuthority(
		outboundIntentTestContext(t, testWorkloadID, 3),
		integrationConsumerSelection{
			orgID: testOrgID, appInstallationID: testInstallationID,
			workloadID: testWorkloadID, capability: outboundintentsemantic.CapabilityName,
			runtimeView: view,
		},
		admission.prepared().current,
	); err != nil {
		t.Fatal(err)
	}
	if len(engine.integrationScopes) != 1 {
		t.Fatalf("repeated admission authority reads = %d, want 1 total", len(engine.integrationScopes))
	}

	for name, mutate := range map[string]func(*catalogsemantic.IntegrationBindingView){
		"mixed revision": func(current *catalogsemantic.IntegrationBindingView) {
			current.Lineage.CatalogRevision++
			current.Receipts.CatalogRevision++
			current.Lineage.ViewDigest = catalogsemantic.ComputeIntegrationBindingViewDigest(*current)
		},
		"mixed source": func(current *catalogsemantic.IntegrationBindingView) {
			current.Lineage.SourceSnapshotDigest = "sha256:" + strings.Repeat("b", 64)
			current.Lineage.ViewDigest = catalogsemantic.ComputeIntegrationBindingViewDigest(*current)
		},
	} {
		t.Run(name, func(t *testing.T) {
			changed := current
			mutate(&changed)
			_, err := prepareIntegrationConsumerAuthority(
				t.Context(),
				integrationConsumerSelection{
					orgID: testOrgID, appInstallationID: testInstallationID,
					workloadID: testWorkloadID, capability: outboundintentsemantic.CapabilityName,
					runtimeView: view,
				},
				&catalogConsumerViewTestEngine{integrationView: changed},
			)
			var failure *recordsemantic.Failure
			if !errors.As(err, &failure) || failure.Category != recordsemantic.FailureStaleState ||
				failure.StaleAxis != recordsemantic.StaleAxisCatalog {
				t.Fatalf("mixed authority error = %#v", err)
			}
		})
	}
}

func TestOutboundIntentServerClosesListAdmissionAndResultInOwnedTransactions(t *testing.T) {
	view := outboundIntentTestView(t, 7)
	intent := outboundIntentTestIntent(7)
	engine := &outboundIntentTestEngine{rows: []persistence.RuntimeOutboundIntentRow{{Intent: intent}}}
	committer := &outboundResultTestCommitter{}
	runner := &outboundIntentTestTransactionRunner{}
	server, err := NewOutboundIntentServer(engine, committer, runner)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(outboundIntentAdmittedContext(t, view), time.Second)
	defer cancel()

	generationRequest := outboundintentsemantic.GenerationRequest{AfterGeneration: 6}
	wireGeneration, err := mapping.ToWireReadOutboundIntentGenerationRequest(ctx, generationRequest)
	if err != nil {
		t.Fatal(err)
	}
	wireGenerationResponse, err := server.ReadOutboundIntentGeneration(ctx, wireGeneration)
	if err != nil {
		t.Fatal(err)
	}
	generation, err := mapping.FromWireReadOutboundIntentGenerationResponse(
		generationRequest,
		wireGenerationResponse,
	)
	if err != nil || generation.Generation != 7 ||
		engine.generationScope.AuthorizedWorkloadID != testWorkloadID {
		t.Fatalf("generation=%#v scope=%#v err=%v", generation, engine.generationScope, err)
	}

	listRequest := outboundintentsemantic.ListRequest{Limit: 4}
	wireList, err := mapping.ToWireListPendingOutboundIntentsRequest(ctx, listRequest)
	if err != nil {
		t.Fatal(err)
	}
	wireListed, err := server.ListPendingOutboundIntents(ctx, wireList)
	if err != nil {
		t.Fatal(err)
	}
	listed, err := mapping.FromWireListPendingOutboundIntentsResponse(listRequest, wireListed)
	if err != nil || len(listed.Intents) != 1 || listed.Intents[0].IntentID != intent.IntentID ||
		engine.listScope.AuthorizedWorkloadID != testWorkloadID {
		t.Fatalf("listed=%#v scope=%#v err=%v", listed, engine.listScope, err)
	}

	ackRequest := outboundintentsemantic.AcknowledgeRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest,
		CatalogRevision: 7, Outcome: outboundintentsemantic.AdmissionAccepted,
		IntegrationCommandID: "integration-command-1",
	}
	wireAck, err := mapping.ToWireAcknowledgeOutboundIntentRequest(ctx, ackRequest)
	if err != nil {
		t.Fatal(err)
	}
	wireAcked, err := server.AcknowledgeOutboundIntent(ctx, wireAck)
	if err != nil {
		t.Fatal(err)
	}
	acked, err := mapping.FromWireAcknowledgeOutboundIntentResponse(ackRequest, wireAcked)
	if err != nil || acked.IntentID != intent.IntentID || !engine.ackInTransaction {
		t.Fatalf("ack=%#v inTx=%v err=%v", acked, engine.ackInTransaction, err)
	}

	commitRequest := outboundintentsemantic.CommitResultRequest{
		IntentID: intent.IntentID, PayloadDigest: intent.PayloadDigest,
		CatalogRevision: 7, IntegrationCommandID: "integration-command-1",
		RemoteID: "remote-1",
	}
	wireCommit, err := mapping.ToWireCommitOutboundResultRequest(ctx, commitRequest)
	if err != nil {
		t.Fatal(err)
	}
	wireCommitted, err := server.CommitOutboundResult(ctx, wireCommit)
	if err != nil {
		t.Fatal(err)
	}
	committed, err := mapping.FromWireCommitOutboundResultResponse(commitRequest, wireCommitted)
	if err != nil || committed.IntentID != intent.IntentID || !committer.inTransaction ||
		committer.view != view || runner.calls != 2 {
		t.Fatalf("commit=%#v committer=%#v transactions=%d err=%v", committed, committer, runner.calls, err)
	}
}

func TestOutboundIntentServerStreamsOnlyExactAcceptedAttachment(t *testing.T) {
	view := outboundIntentTestView(t, 7)
	content := []byte("attachment-content")
	sum := sha256.Sum256(content)
	ref := outboundintentsemantic.MailAttachmentV1{
		FileID: "file-1", FileName: "資料.pdf", ContentType: "application/pdf",
		ContentLength: uint64(len(content)), ContentDigest: "sha256:" + hex.EncodeToString(sum[:]),
	}
	engine := &outboundIntentTestEngine{attachmentRef: ref}
	files := &outboundIntentTestFileEngine{content: content}
	server, err := NewOutboundIntentServer(engine, &outboundResultTestCommitter{}, &outboundIntentTestTransactionRunner{}, files)
	if err != nil {
		t.Fatal(err)
	}
	request := outboundintentsemantic.OpenMailAttachmentRequest{
		IntentID: "intent-1", PayloadDigest: "sha256:" + strings.Repeat("a", 64),
		CatalogRevision: 7, IntegrationCommandID: "command-1", AttachmentIndex: 0,
	}
	ctx, cancel := context.WithTimeout(outboundIntentAdmittedContext(t, view), time.Second)
	defer cancel()
	wireRequest, err := mapping.ToWireOpenOutboundMailAttachmentRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	stream := &outboundIntentAttachmentTestStream{ctx: ctx}
	if err := server.OpenOutboundMailAttachment(wireRequest, stream); err != nil {
		t.Fatalf("OpenOutboundMailAttachment: %v chunks=%d request=%#v", err, len(stream.chunks), engine.attachmentRequest)
	}
	state, err := outboundintentsemantic.NewMailAttachmentStreamState(ref)
	if err != nil {
		t.Fatal(err)
	}
	var got []byte
	for _, wireChunk := range stream.chunks {
		chunk, err := mapping.FromWireOpenOutboundMailAttachmentChunk(wireChunk)
		if err != nil || state.Observe(chunk) != nil {
			t.Fatalf("chunk: %v", err)
		}
		got = append(got, chunk.Data...)
	}
	if err := state.Complete(); err != nil || !bytes.Equal(got, content) || engine.attachmentRequest != request {
		t.Fatalf("complete=%v got=%q request=%#v", err, got, engine.attachmentRequest)
	}
}

func TestOutboundIntentServerClassifiesFrozenAttachmentFailuresAsInternal(t *testing.T) {
	view := outboundIntentTestView(t, 7)
	ref := outboundintentsemantic.MailAttachmentV1{
		FileID: "file-1", FileName: "file.pdf", ContentType: "application/pdf",
		ContentLength: 1, ContentDigest: "sha256:" + strings.Repeat("a", 64),
	}
	engine := &outboundIntentTestEngine{attachmentRef: ref}
	server, err := NewOutboundIntentServer(
		engine,
		&outboundResultTestCommitter{},
		&outboundIntentTestTransactionRunner{},
		&outboundIntentTestFileEngine{err: errors.New("storage unavailable")},
	)
	if err != nil {
		t.Fatal(err)
	}
	request := outboundintentsemantic.OpenMailAttachmentRequest{
		IntentID: "intent-1", PayloadDigest: "sha256:" + strings.Repeat("b", 64),
		CatalogRevision: 7, IntegrationCommandID: "command-1", AttachmentIndex: 0,
	}
	ctx, cancel := context.WithTimeout(outboundIntentAdmittedContext(t, view), time.Second)
	defer cancel()
	wireRequest, err := mapping.ToWireOpenOutboundMailAttachmentRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	err = server.OpenOutboundMailAttachment(
		wireRequest,
		&outboundIntentAttachmentTestStream{ctx: ctx},
	)
	decoded := mapping.FromPlatformStatusError(err, outboundintentsemantic.CapabilityName)
	var failure *outboundintentsemantic.Failure
	if !errors.As(decoded, &failure) || failure.Category != commonsemantic.FailureInternal {
		t.Fatalf("frozen storage failure = %v, want internal", decoded)
	}
}

func TestOutboundIntentServerClosesFrozenReaderOnMetadataMismatch(t *testing.T) {
	view := outboundIntentTestView(t, 7)
	ref := outboundintentsemantic.MailAttachmentV1{
		FileID: "file-1", FileName: "file.pdf", ContentType: "text/plain",
		ContentLength: 1, ContentDigest: "sha256:" + strings.Repeat("a", 64),
	}
	reader := &outboundIntentTestReadCloser{Reader: strings.NewReader("x")}
	engine := &outboundIntentTestEngine{attachmentRef: ref}
	server, err := NewOutboundIntentServer(
		engine,
		&outboundResultTestCommitter{},
		&outboundIntentTestTransactionRunner{},
		&outboundIntentTestFileEngine{download: &sharedcontract.FileDownload{
			Reader: reader, FileName: ref.FileName, Size: 2, ContentType: ref.ContentType,
		}},
	)
	if err != nil {
		t.Fatal(err)
	}
	request := outboundintentsemantic.OpenMailAttachmentRequest{
		IntentID: "intent-1", PayloadDigest: "sha256:" + strings.Repeat("b", 64),
		CatalogRevision: 7, IntegrationCommandID: "command-1", AttachmentIndex: 0,
	}
	ctx, cancel := context.WithTimeout(outboundIntentAdmittedContext(t, view), time.Second)
	defer cancel()
	wireRequest, err := mapping.ToWireOpenOutboundMailAttachmentRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	err = server.OpenOutboundMailAttachment(
		wireRequest,
		&outboundIntentAttachmentTestStream{ctx: ctx},
	)
	if err == nil || !reader.closed {
		t.Fatalf("metadata mismatch err=%v reader.closed=%v, want rejected and closed", err, reader.closed)
	}
}

func TestOutboundIntentServerOmittedOperationSetRemainsExplicitV1(t *testing.T) {
	view := outboundIntentTestView(t, 7)
	upsert := outboundIntentTestIntent(7)
	mail := outboundIntentTestIntent(7)
	mail.IntentID = "mail-intent"
	mail.Operation = outboundintentsemantic.OperationSendMail
	mail.PayloadFormatVersion = outboundintentsemantic.MailPayloadFormatV2
	engine := &outboundIntentTestEngine{rows: []persistence.RuntimeOutboundIntentRow{
		{Intent: upsert}, {Intent: mail},
	}}
	server, err := NewOutboundIntentServer(
		engine, &outboundResultTestCommitter{}, &outboundIntentTestTransactionRunner{},
	)
	if err != nil {
		t.Fatal(err)
	}
	request := outboundintentsemantic.ListRequest{Limit: 2}
	if err := outboundintentsemantic.ValidateListResponseFor(request, outboundintentsemantic.ListResponse{
		Intents: []outboundintentsemantic.Intent{mail},
	}); err == nil {
		t.Fatal("published neutral contract accepted SEND_MAIL for an omitted operation-set version")
	}
	ctx, cancel := context.WithTimeout(outboundIntentAdmittedContext(t, view), time.Second)
	defer cancel()
	wireRequest, err := mapping.ToWireListPendingOutboundIntentsRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := server.ListPendingOutboundIntents(ctx, wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireListPendingOutboundIntentsResponse(request, wireResponse)
	if err != nil || len(response.Intents) != 1 || response.Intents[0].IntentID != upsert.IntentID {
		t.Fatalf("omitted operation-set response = %#v, err=%v", response, err)
	}
}

func TestOutboundIntentServerFailsClosedOnMixedRevisionAndCancellation(t *testing.T) {
	view := outboundIntentTestView(t, 7)
	engine := &outboundIntentTestEngine{}
	server, err := NewOutboundIntentServer(
		engine, &outboundResultTestCommitter{}, &outboundIntentTestTransactionRunner{},
	)
	if err != nil {
		t.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(outboundIntentAdmittedContext(t, view), time.Second)
	defer cancel()
	request := outboundintentsemantic.AcknowledgeRequest{
		IntentID: "intent-1", PayloadDigest: "sha256:" + strings.Repeat("a", 64),
		CatalogRevision: 6, Outcome: outboundintentsemantic.AdmissionAccepted,
		IntegrationCommandID: "integration-command-1",
	}
	wire, err := mapping.ToWireAcknowledgeOutboundIntentRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := server.AcknowledgeOutboundIntent(ctx, wire); status.Code(err) != codes.FailedPrecondition {
		t.Fatalf("mixed revision status = %v, want FailedPrecondition", err)
	}
	if engine.ackCalls != 0 {
		t.Fatalf("mixed revision reached persistence: %d", engine.ackCalls)
	}

	canceled, cancelNow := context.WithCancel(outboundIntentAdmittedContext(t, view))
	cancelNow()
	if _, err := server.AcknowledgeOutboundIntent(canceled, wire); status.Code(err) != codes.Canceled {
		t.Fatalf("canceled status = %v, want Canceled", err)
	}
}

func TestRuntimeAPIProcessRegistersOutboundIntentOnTheExistingEndpointOnly(t *testing.T) {
	view := outboundIntentTestView(t, 7)
	attachmentContent := []byte("process-attachment-content")
	attachmentSum := sha256.Sum256(attachmentContent)
	attachmentRef := outboundintentsemantic.MailAttachmentV1{
		FileID: "file-1", FileName: "process.pdf", ContentType: "application/pdf",
		ContentLength: uint64(len(attachmentContent)),
		ContentDigest: "sha256:" + hex.EncodeToString(attachmentSum[:]),
	}
	engine := &outboundIntentTestEngine{rows: []persistence.RuntimeOutboundIntentRow{{
		Intent: outboundIntentTestIntent(7),
	}}, attachmentRef: attachmentRef}
	current := commandConsumerIntegrationView(view)
	current.Receipts.SecurityEpoch = 1
	admission, err := NewOutboundIntentAdmission(t.Context(), OutboundIntentSelection{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		WorkloadID: testWorkloadID, RuntimeMetadataView: view,
	}, &catalogConsumerViewTestEngine{integrationView: current})
	if err != nil {
		t.Fatal(err)
	}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		OutboundIntentAdmission: admission, OutboundIntents: engine,
		OutboundResultCommitter:    &outboundResultTestCommitter{},
		OutboundIntentTransactions: &outboundIntentTestTransactionRunner{},
		OutboundMailAttachments:    &outboundIntentTestFileEngine{content: attachmentContent},
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			outboundintentsemantic.CapabilityName: time.Second,
		},
		PlatformResourceClassAdmissionLimits: map[runtimegrpc.ResourceClass]int{
			runtimegrpc.ResourceClassFileStream: 1,
		},
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	defer stopProcess(t, process)
	assertRuntimeProbe(t, "http://"+process.AdminAddress()+processhttp.ReadyPath, http.StatusOK)
	capabilities := readRuntimeMetrics(t, "http://"+process.AdminAddress()+processhttp.CapabilitiesPath)
	if !strings.Contains(capabilities, outboundintentsemantic.CapabilityName) ||
		strings.Contains(capabilities, "runtime.external-effects.dispatch.v1") {
		t.Fatalf("Runtime endpoint capabilities = %s", capabilities)
	}

	connection, err := grpc.NewClient(
		"passthrough:///"+process.GRPCAddress(),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	ctx, cancel := context.WithTimeout(
		platformWorkloadProtocolContext(context.Background(), outboundintentsemantic.CapabilityName),
		time.Second,
	)
	defer cancel()
	request := outboundintentsemantic.ListRequest{Limit: 1}
	wireRequest, err := mapping.ToWireListPendingOutboundIntentsRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewOutboundIntentServiceClient(connection).
		ListPendingOutboundIntents(ctx, wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireListPendingOutboundIntentsResponse(request, wireResponse)
	if err != nil || len(response.Intents) != 1 {
		t.Fatalf("cross-process response=%#v err=%v", response, err)
	}
	openRequest := outboundintentsemantic.OpenMailAttachmentRequest{
		IntentID: "intent-1", PayloadDigest: "sha256:" + strings.Repeat("a", 64),
		CatalogRevision: 7, IntegrationCommandID: "command-1", AttachmentIndex: 0,
	}
	wireOpenRequest, err := mapping.ToWireOpenOutboundMailAttachmentRequest(ctx, openRequest)
	if err != nil {
		t.Fatal(err)
	}
	attachmentStream, err := runtimev1.NewOutboundIntentServiceClient(connection).
		OpenOutboundMailAttachment(ctx, wireOpenRequest)
	if err != nil {
		t.Fatal(err)
	}
	state, err := outboundintentsemantic.NewMailAttachmentStreamState(attachmentRef)
	if err != nil {
		t.Fatal(err)
	}
	var streamed []byte
	for {
		wireChunk, receiveErr := attachmentStream.Recv()
		if errors.Is(receiveErr, io.EOF) {
			break
		}
		if receiveErr != nil {
			t.Fatalf("receive attachment: %v", receiveErr)
		}
		chunk, mapErr := mapping.FromWireOpenOutboundMailAttachmentChunk(wireChunk)
		if mapErr != nil || state.Observe(chunk) != nil {
			t.Fatalf("attachment chunk: %v", mapErr)
		}
		streamed = append(streamed, chunk.Data...)
	}
	if err := state.Complete(); err != nil || !bytes.Equal(streamed, attachmentContent) ||
		engine.attachmentRequest != openRequest {
		t.Fatalf("attachment complete=%v content=%q request=%#v", err, streamed, engine.attachmentRequest)
	}
}

func TestRuntimeAPIProcessKeepsLegacyUnaryOutboundCompositionWithoutAttachmentFeature(t *testing.T) {
	view := outboundIntentTestView(t, 7)
	current := commandConsumerIntegrationView(view)
	current.Receipts.SecurityEpoch = 1
	admission, err := NewOutboundIntentAdmission(t.Context(), OutboundIntentSelection{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		WorkloadID: testWorkloadID, RuntimeMetadataView: view,
	}, &catalogConsumerViewTestEngine{integrationView: current})
	if err != nil {
		t.Fatal(err)
	}
	process, err := Assemble(ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		WorkloadAuthority: testWorkloadAuthority, SecurityAdmission: testSecurityAdmission(t),
		OutboundIntentAdmission: admission, OutboundIntents: &outboundIntentTestEngine{},
		OutboundResultCommitter:    &outboundResultTestCommitter{},
		OutboundIntentTransactions: &outboundIntentTestTransactionRunner{},
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			outboundintentsemantic.CapabilityName: time.Second,
		},
	}, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	stopProcess(t, process)
}

type outboundIntentTransactionContextKey struct{}

type outboundIntentTestTransactionRunner struct{ calls int }

func (r *outboundIntentTestTransactionRunner) RunInPessimisticTransaction(
	ctx context.Context,
	fn func(context.Context) error,
) error {
	r.calls++
	return fn(context.WithValue(ctx, outboundIntentTransactionContextKey{}, true))
}

type outboundIntentTestEngine struct {
	rows              []persistence.RuntimeOutboundIntentRow
	generationScope   persistence.RuntimeOutboundIntentScope
	listScope         persistence.RuntimeOutboundIntentScope
	ackCalls          int
	ackInTransaction  bool
	attachmentRef     outboundintentsemantic.MailAttachmentV1
	attachmentRequest outboundintentsemantic.OpenMailAttachmentRequest
}

func (e *outboundIntentTestEngine) ReadGeneration(
	_ context.Context,
	scope persistence.RuntimeOutboundIntentScope,
) (uint64, error) {
	e.generationScope = scope
	return 7, nil
}

func (e *outboundIntentTestEngine) ListPending(
	_ context.Context,
	scope persistence.RuntimeOutboundIntentScope,
	_ uint64,
	_ uint32,
) ([]persistence.RuntimeOutboundIntentRow, error) {
	e.listScope = scope
	return e.rows, nil
}

func (e *outboundIntentTestEngine) OpenMailAttachmentReference(
	_ context.Context,
	_ persistence.RuntimeOutboundIntentScope,
	request outboundintentsemantic.OpenMailAttachmentRequest,
) (outboundintentsemantic.MailAttachmentV1, error) {
	e.attachmentRequest = request
	return e.attachmentRef, nil
}

func (e *outboundIntentTestEngine) Acknowledge(
	ctx context.Context,
	_ persistence.RuntimeOutboundIntentScope,
	_ outboundintentsemantic.AcknowledgeRequest,
) (persistence.RuntimeOutboundIntentRow, bool, error) {
	e.ackCalls++
	e.ackInTransaction, _ = ctx.Value(outboundIntentTransactionContextKey{}).(bool)
	return persistence.RuntimeOutboundIntentRow{}, false, nil
}

type outboundIntentTestFileEngine struct {
	content  []byte
	err      error
	download *sharedcontract.FileDownload
}

func (e *outboundIntentTestFileEngine) OpenFrozenRecordAttachment(_ context.Context, expected sharedcontract.RecordAttachmentMetadata) (*sharedcontract.FileDownload, error) {
	if e.err != nil {
		return nil, e.err
	}
	if e.download != nil {
		return e.download, nil
	}
	return &sharedcontract.FileDownload{
		Reader: io.NopCloser(bytes.NewReader(e.content)), FileName: expected.FileName,
		Size: int64(len(e.content)), ContentType: expected.ContentType,
	}, nil
}

type outboundIntentTestReadCloser struct {
	io.Reader
	closed bool
}

func (r *outboundIntentTestReadCloser) Close() error {
	r.closed = true
	return nil
}

type outboundIntentAttachmentTestStream struct {
	ctx    context.Context
	chunks []*runtimev1.OpenOutboundMailAttachmentResponse
}

func (s *outboundIntentAttachmentTestStream) Send(value *runtimev1.OpenOutboundMailAttachmentResponse) error {
	s.chunks = append(s.chunks, value)
	return nil
}
func (s *outboundIntentAttachmentTestStream) SetHeader(grpcmetadata.MD) error  { return nil }
func (s *outboundIntentAttachmentTestStream) SendHeader(grpcmetadata.MD) error { return nil }
func (s *outboundIntentAttachmentTestStream) SetTrailer(grpcmetadata.MD)       {}
func (s *outboundIntentAttachmentTestStream) Context() context.Context         { return s.ctx }
func (s *outboundIntentAttachmentTestStream) SendMsg(any) error                { return nil }
func (s *outboundIntentAttachmentTestStream) RecvMsg(any) error                { return io.EOF }

var _ runtimev1.OutboundIntentService_OpenOutboundMailAttachmentServer = (*outboundIntentAttachmentTestStream)(nil)

type outboundResultTestCommitter struct {
	inTransaction bool
	view          *catalogsnapshot.RuntimeMetadataView
}

func (c *outboundResultTestCommitter) Commit(
	ctx context.Context,
	view *catalogsnapshot.RuntimeMetadataView,
	_ persistence.RuntimeOutboundIntentScope,
	_ outboundintentsemantic.CommitResultRequest,
) (bool, error) {
	c.inTransaction, _ = ctx.Value(outboundIntentTransactionContextKey{}).(bool)
	c.view = view
	return false, nil
}

func outboundIntentTestView(t *testing.T, revision uint64) *catalogsnapshot.RuntimeMetadataView {
	t.Helper()
	_, artifact := servingArtifacts(
		t, revision, outboundintentsemantic.CapabilityName,
		catalogsnapshot.ProjectionRecordMutationV1,
	)
	view, err := catalogsnapshot.OpenRuntimeMetadataView(artifact)
	if err != nil {
		t.Fatal(err)
	}
	return view
}

func outboundIntentTestContext(t *testing.T, workloadID string, securityEpoch uint64) context.Context {
	t.Helper()
	return context.WithValue(t.Context(), trustedWorkloadCallContextKey{}, TrustedWorkloadCall{
		orgID: testOrgID, callerWorkloadID: workloadID,
		capability: outboundintentsemantic.CapabilityName, securityEpoch: securityEpoch,
	})
}

func outboundIntentAdmittedContext(
	t *testing.T,
	view *catalogsnapshot.RuntimeMetadataView,
) context.Context {
	t.Helper()
	lineage := view.Lineage()
	ctx := outboundIntentTestContext(t, testWorkloadID, 3)
	return context.WithValue(ctx, runtimeExecutionScopeContextKey{}, RuntimeExecutionScope{
		view: view, orgID: testOrgID, appInstallationID: testInstallationID,
		workloadID: testWorkloadID, capability: outboundintentsemantic.CapabilityName,
		receipts: commonsemantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: outboundintentsemantic.CapabilityName,
			CatalogRevision:  uint64(lineage.CatalogRevision),
			SchemaGeneration: uint64(lineage.SchemaGeneration),
			EngineGeneration: uint64(lineage.EngineGeneration), SecurityEpoch: 3,
		},
	})
}

func outboundIntentTestIntent(revision uint64) outboundintentsemantic.Intent {
	payload := []byte(`{"version":1,"targetObject":"contact","mappedFields":{"name":"test"}}`)
	sum := sha256.Sum256(payload)
	return outboundintentsemantic.Intent{
		IntentID: "intent-1", IdempotencyKey: "intent-key-1",
		PayloadFormatVersion: 1, PayloadDigest: "sha256:" + hex.EncodeToString(sum[:]),
		CatalogRevision: revision, Operation: outboundintentsemantic.OperationUpsert,
		SourceObject: "Widget", SourceRecordID: "widget-1", ConnectorOwnerKey: "connector.test",
		FlowDeveloperName: "PushWidget", FlowActionName: "PushToConnector",
		ActionSlotID: "action-1", DestinationSlotID: "destination-1",
		Bindings: []outboundintentsemantic.BindingReference{{
			Kind: catalogsemantic.BindingConnectorPushContract, OwnerPack: "connector.test",
			PackInstallationID: "pack-installation-1",
			PackPayloadDigest:  "nexus-pack-sha256-v1:" + strings.Repeat("a", 64),
			BusinessKey:        "push/widget", DefinitionVersion: 1,
		}},
		CanonicalPayload: payload,
	}
}