package runtimeapi

import (
	"context"
	"strconv"
	"testing"
	"time"

	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	approvalnotificationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalnotificationflowv1"
	approvalsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/approvalv1"
	permissionsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/effectivepermissionsv1"
	filessemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/filesv1"
	mailboxauthorizationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/mailboxauthorizationv1"
	metadataadminsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/metadataadminreadv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionstatev1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	searchsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/searchv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	catalogv1 "github.com/pingcap-inc/nexuscrm-contract/wire-go/gen/catalog/v1"
	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"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
)

func TestRuntimeAPIProcessMetadataAdminReadCrossesRealInterceptorChain(t *testing.T) {
	fixture := newServingFixture(t, 17)
	admission, err := NewMetadataAdminReadAdmission(MetadataAdminReadSelection{
		OrgID: testOrgID, AppInstallationID: testInstallationID,
		WorkloadID: testWorkloadID,
	})
	if err != nil {
		t.Fatal(err)
	}
	description := "Customer accounts"
	engine := &metadataAdminReadTestEngine{objects: []models.ObjectMetadata{{
		APIName: "Account", Description: &description, DeleteMode: "soft",
	}}}
	connection := startPlatformCapabilityProcess(
		t,
		fixture,
		metadataadminsemantic.CapabilityName,
		runtimegrpc.ResourceClassHeavyWork,
		func(config *ProcessConfig) {
			config.MetadataAdminReads = engine
			config.MetadataAdminReadAdmission = admission
			config.EffectivePermissions = &metadataAdminReadTestPermissions{allowed: true}
		},
	)
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(context.Background(), metadataadminsemantic.CapabilityName),
		900*time.Millisecond,
	)
	defer cancel()
	request := metadataadminsemantic.ObjectAdminIndexRequest{}
	wireRequest, err := mapping.ToWireReadObjectAdminIndexRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewMetadataAdminReadServiceClient(
		connection,
	).ReadObjectAdminIndex(ctx, wireRequest)
	if err != nil {
		t.Fatalf("ReadObjectAdminIndex = %v, decoded=%#v, engine calls=%d", err,
			mapping.FromPlatformStatusError(err, metadataadminsemantic.CapabilityName), engine.objectCalls)
	}
	response, err := mapping.FromWireReadObjectAdminIndexResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.Objects) != 1 || response.Objects[0].ObjectID != "Account" ||
		response.Objects[0].Description == nil || *response.Objects[0].Description != description ||
		response.Receipts.CatalogRevision != 0 || response.Receipts.SecurityEpoch != 1 ||
		engine.objectCalls != 1 {
		t.Fatalf("metadata admin-read process response = %#v, calls=%d", response, engine.objectCalls)
	}
}

func TestRuntimeAPIProcessApprovalNotificationFlowReadCrossesRealInterceptorChain(t *testing.T) {
	fixture := newServingFixture(t, 17)
	admission, err := NewApprovalNotificationFlowReadAdmission(
		ApprovalNotificationFlowReadSelection{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	engine := &approvalNotificationFlowReadTestEngine{definitions: []*models.FlowDefinition{{
		DeveloperName: "Notify_Approver", MasterLabel: "Notify Approver",
	}}}
	connection := startPlatformCapabilityProcess(
		t,
		fixture,
		approvalnotificationsemantic.CapabilityName,
		runtimegrpc.ResourceClassHeavyWork,
		func(config *ProcessConfig) {
			config.ApprovalNotificationFlows = engine
			config.ApprovalNotificationFlowReadAdmission = admission
		},
	)
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(context.Background(), approvalnotificationsemantic.CapabilityName),
		900*time.Millisecond,
	)
	defer cancel()
	request := approvalnotificationsemantic.Request{}
	wireRequest, err := mapping.ToWireReadApprovalNotificationFlowsRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewApprovalNotificationFlowReadServiceClient(
		connection,
	).ReadApprovalNotificationFlows(ctx, wireRequest)
	if err != nil {
		t.Fatalf("ReadApprovalNotificationFlows = %v, decoded=%#v, engine calls=%d", err,
			mapping.FromPlatformStatusError(err, approvalnotificationsemantic.CapabilityName), engine.calls)
	}
	response, err := mapping.FromWireReadApprovalNotificationFlowsResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.Flows) != 1 || response.Flows[0].Name != "Notify_Approver" ||
		response.Flows[0].Label != "Notify Approver" || response.Receipts.CatalogRevision != 0 ||
		response.Receipts.SecurityEpoch != 1 || engine.calls != 1 {
		t.Fatalf("approval notification Flow process response = %#v, calls=%d", response, engine.calls)
	}
}

func TestRuntimeAPIProcessMailboxAuthorizationCrossesRealInterceptorChain(
	t *testing.T,
) {
	view := validCatalogIntegrationBindingView(41)
	view.Receipts.SecurityEpoch = 1
	view.Lineage.ViewDigest = catalogsemantic.ComputeIntegrationBindingViewDigest(view)
	admission, err := NewMailboxAuthorizationAdmission(
		t.Context(),
		MailboxAuthorizationSelection{
			OrgID: view.OrgID, AppInstallationID: view.AppInstallationID,
			WorkloadID: view.AuthorizedWorkloadID,
		},
		&catalogConsumerViewTestEngine{integrationView: view},
	)
	if err != nil {
		t.Fatal(err)
	}
	store := &memoryMailboxHandleStore{}
	session := testRuntimeAuthorizingSessionState(testPrincipalID, 1)
	config := ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			mailboxauthorizationsemantic.CapabilityName: time.Second,
		},
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: testSecurityAdmission(t),
		ServingAdmission:  &ServingAdmission{},
		MailboxAuthorization: &MailboxAuthorizationProcessConfig{
			Admission: admission, Handles: store,
			Permissions: mailboxPermissionStub{allowed: true},
			Sessions: mailboxSessionStub{session: mailboxAuthorizationSession{
				User: session.User, Subject: session.Subject, AuthEpoch: session.AuthEpoch,
				PermissionFloor: session.PermissionFloor,
			}},
			RequiredPermission: MailboxPermissionRequirement{CustomPermission: "SendRecordEmail"},
		},
	}
	withoutAdmission := config
	withoutMailboxAdmission := *withoutAdmission.MailboxAuthorization
	withoutMailboxAdmission.Admission = nil
	withoutAdmission.MailboxAuthorization = &withoutMailboxAdmission
	if process, assembleErr := Assemble(withoutAdmission, &readerStub{}); assembleErr == nil || process != nil {
		t.Fatalf("mailbox process without Integration admission = %v, %v", process, assembleErr)
	}
	process, err := Assemble(config, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { stopProcess(t, process) })
	connection, err := grpc.NewClient(
		"passthrough:///"+process.GRPCAddress(),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = connection.Close() })
	client := runtimev1.NewConnectedMailboxAuthorizationServiceClient(connection)

	beginCtx, cancelBegin := context.WithTimeout(
		platformProtocolContext(context.Background(), mailboxauthorizationsemantic.CapabilityName),
		time.Second,
	)
	defer cancelBegin()
	begin := mailboxauthorizationsemantic.Request{
		Operation: mailboxauthorizationsemantic.OperationBegin,
	}
	wireBegin, err := mapping.ToWireAuthorizeConnectedMailboxRequest(beginCtx, begin)
	if err != nil {
		t.Fatal(err)
	}
	wireStarted, err := client.AuthorizeConnectedMailbox(beginCtx, wireBegin)
	if err != nil {
		t.Fatal(err)
	}
	started, err := mapping.FromWireAuthorizeConnectedMailboxResponse(begin, wireStarted)
	if err != nil || started.AuthorizedSubjectID != testPrincipalID ||
		started.ReauthorizationHandle == "" || store.mints != 1 {
		t.Fatalf("mailbox begin = %#v, mints=%d, error=%v", started, store.mints, err)
	}

	callbackCtx, cancelCallback := context.WithTimeout(
		platformWorkloadProtocolContext(context.Background(), mailboxauthorizationsemantic.CapabilityName),
		time.Second,
	)
	defer cancelCallback()
	callback := mailboxauthorizationsemantic.ReauthorizeRequest{
		ReauthorizationHandle: started.ReauthorizationHandle,
	}
	wireCallback, err := mapping.ToWireReauthorizeConnectedMailboxRequest(callbackCtx, callback)
	if err != nil {
		t.Fatal(err)
	}
	wireCompleted, err := client.ReauthorizeConnectedMailbox(callbackCtx, wireCallback)
	if err != nil {
		t.Fatal(err)
	}
	completed, err := mapping.FromWireReauthorizeConnectedMailboxResponse(callback, wireCompleted)
	if err != nil || completed.AuthorizedSubjectID != testPrincipalID || store.record.ConsumedAt == nil {
		t.Fatalf("mailbox callback = %#v, record=%#v, error=%v", completed, store.record, err)
	}
}

func TestRuntimeAPIProcessApprovalCapabilitiesRegisterOnExistingEndpoint(t *testing.T) {
	tests := []struct {
		name       string
		capability string
		execute    bool
	}{
		{name: "execute", capability: approvalsemantic.ExecuteCapabilityName, execute: true},
		{name: "read", capability: approvalsemantic.ReadCapabilityName},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			capabilities := []string{test.capability}
			if test.execute {
				capabilities = append(capabilities, mutationsemantic.CapabilityName)
			}
			snapshotArtifact, views := approvalAuthorityTestArtifactsForCapabilities(
				t, testOrgID, testInstallationID, capabilities, nil,
			)
			fixture := &servingFixture{admission: newApprovalServingFixture(t, snapshotArtifact, views)}
			store := &approvalServerTestStore{t: t}
			connection := startPlatformCapabilityProcess(
				t,
				fixture,
				test.capability,
				runtimegrpc.ResourceClassHeavyWork,
				func(config *ProcessConfig) {
					// A valid request below intentionally reaches this incomplete
					// engine and fails there. Reaching it proves Approval is
					// registered on the existing Platform listener; Unimplemented
					// would mean the process composition omitted the service.
					config.ApprovalEngine = &ApprovalExecutionEngine{}
					if test.execute {
						config.ApprovalInvocations = store
						config.ApprovalTransactions = &approvalServerTestRunner{store: store}
						config.ApprovalPermissionFence = &approvalServerTestPermissionFence{t: t}
					}
				},
			)
			ctx, cancel := context.WithTimeout(
				platformProtocolContext(context.Background(), test.capability),
				time.Second,
			)
			defer cancel()
			client := runtimev1.NewApprovalServiceClient(connection)
			var err error
			if test.execute {
				wireRequest, mapErr := mapping.ToWireExecuteApprovalRequest(ctx, approvalsemantic.ExecuteRequest{
					Operation: approvalsemantic.OperationSubmit, IdempotencyKey: "process-registration",
					ObjectID: "WorkItem", TargetRecordID: "record-1",
				})
				if mapErr != nil {
					t.Fatal(mapErr)
				}
				_, err = client.ExecuteApproval(ctx, wireRequest)
			} else {
				wireRequest, mapErr := mapping.ToWireReadApprovalRequest(
					ctx,
					approvalsemantic.ReadRequest{Kind: approvalsemantic.ReadPendingForCaller},
				)
				if mapErr != nil {
					t.Fatal(mapErr)
				}
				_, err = client.ReadApproval(ctx, wireRequest)
			}
			if status.Code(err) == codes.Unimplemented {
				t.Fatalf("Approval %s was not registered on the Runtime Platform endpoint", test.name)
			}
			if err == nil {
				t.Fatalf("incomplete test engine unexpectedly served Approval %s", test.name)
			}
		})
	}
}

func TestRuntimeAPIProcessCatalogConsumerViewCrossesRegisteredEndpoint(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 17, "io.nexuscrm.test.catalog-consumer.v1",
		catalogsemantic.AppMetadataCapabilityName, catalogsnapshot.ProjectionRecordQueryV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	appView := validCatalogAppMetadataView(17)
	appView.Lineage.SchemaGeneration = 8
	appView.Lineage.EngineGeneration = 9
	appView.Receipts.SchemaGeneration = 8
	appView.Receipts.EngineGeneration = 9
	appView.Receipts.SecurityEpoch = 1
	appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewDigest(appView)
	engine := &catalogConsumerViewTestEngine{appView: appView}
	connection := startPlatformCapabilityProcess(
		t,
		fixture,
		catalogsemantic.AppMetadataCapabilityName,
		runtimegrpc.ResourceClassCatalogWatch,
		func(config *ProcessConfig) { config.CatalogConsumerViews = engine },
	)
	ctx, cancel := context.WithTimeout(
		platformWorkloadProtocolContext(context.Background(), catalogsemantic.AppMetadataCapabilityName),
		time.Second,
	)
	defer cancel()
	response, err := catalogv1.NewConsumerViewServiceClient(connection).GetAppMetadataView(
		ctx,
		&catalogv1.GetAppMetadataViewRequest{},
	)
	if err != nil {
		t.Fatal(err)
	}
	view, err := mapping.FromWireAppMetadataView(response.GetView())
	if err != nil {
		t.Fatal(err)
	}
	if view.AppInstallationID != testInstallationID ||
		view.Lineage.ViewDigest != appView.Lineage.ViewDigest {
		t.Fatalf("Catalog process response = %#v", view)
	}
	requireCatalogConsumerScope(
		t,
		engine.appScopes[0],
		catalogsemantic.AppMetadataCapabilityName,
	)
}

func TestRuntimeAPIProcessStaticCatalogConsumerDoesNotShadowServingCatalogView(
	t *testing.T,
) {
	capability := catalogsemantic.AppMetadataCapabilityName
	fixture := newServingFixtureForCapability(
		t, 17, "io.nexuscrm.test.catalog-consumer.v1", capability,
		catalogsnapshot.ProjectionRecordQueryV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	appView := validCatalogAppMetadataView(17)
	appView.Lineage.SchemaGeneration = 8
	appView.Lineage.EngineGeneration = 9
	appView.Receipts.SchemaGeneration = 8
	appView.Receipts.EngineGeneration = 9
	appView.Receipts.SecurityEpoch = 1
	appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewDigest(appView)
	engine := &catalogConsumerViewTestEngine{appView: appView}
	staticAdmission, err := NewCatalogConsumerAdmission(
		[]CatalogConsumerSelection{{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID:   testWorkloadID,
			CapabilityID: catalogsemantic.IntegrationBindingCapabilityName,
		}},
	)
	if err != nil {
		t.Fatal(err)
	}
	connection := startPlatformCapabilityProcess(
		t, fixture, capability, runtimegrpc.ResourceClassCatalogWatch,
		func(config *ProcessConfig) {
			config.CatalogConsumerAdmission = staticAdmission
			config.CatalogConsumerViews = engine
		},
	)
	ctx, cancel := context.WithTimeout(
		platformWorkloadProtocolContext(context.Background(), capability),
		time.Second,
	)
	defer cancel()
	response, err := catalogv1.NewConsumerViewServiceClient(
		connection,
	).GetAppMetadataView(ctx, &catalogv1.GetAppMetadataViewRequest{})
	if err != nil {
		t.Fatal(err)
	}
	view, err := mapping.FromWireAppMetadataView(response.GetView())
	if err != nil {
		t.Fatal(err)
	}
	if view.AppInstallationID != testInstallationID ||
		view.Lineage.ViewDigest != appView.Lineage.ViewDigest {
		t.Fatalf("Catalog process response = %#v", view)
	}
	requireCatalogConsumerScope(t, engine.appScopes[0], capability)
}

func TestRuntimeAPIProcessStaticCatalogConsumerUsesOneEndpoint(
	t *testing.T,
) {
	capability := catalogsemantic.IntegrationBindingCapabilityName
	fixture := newServingFixtureForCapability(
		t,
		17,
		"io.nexuscrm.test.integration-binding.v1",
		capability,
		catalogsnapshot.ProjectionRecordQueryV1,
	)
	view := validCatalogIntegrationBindingView(17)
	engine := &catalogConsumerViewTestEngine{integrationView: view}
	admission, err := NewCatalogConsumerAdmission(
		[]CatalogConsumerSelection{{
			OrgID: testOrgID, AppInstallationID: testInstallationID,
			WorkloadID: testWorkloadID, CapabilityID: capability,
		}},
	)
	if err != nil {
		t.Fatal(err)
	}
	connection := startPlatformCapabilityProcess(
		t,
		fixture,
		capability,
		runtimegrpc.ResourceClassCatalogWatch,
		func(config *ProcessConfig) {
			// This request is admitted only by the static Catalog consumer
			// path. It must not borrow a ServingState metadata selection.
			config.ServingAdmission = nil
			config.CatalogConsumerAdmission = admission
			config.CatalogConsumerViews = engine
		},
	)
	ctx, cancel := context.WithTimeout(
		platformWorkloadProtocolContext(context.Background(), capability),
		time.Second,
	)
	defer cancel()
	response, err := catalogv1.NewConsumerViewServiceClient(
		connection,
	).GetIntegrationBindingView(
		ctx,
		&catalogv1.GetIntegrationBindingViewRequest{},
	)
	if err != nil {
		t.Fatal(err)
	}
	got, err := mapping.FromWireIntegrationBindingView(response.GetView())
	if err != nil {
		t.Fatal(err)
	}
	if got.AppInstallationID != testInstallationID ||
		got.Lineage.ViewDigest != view.Lineage.ViewDigest {
		t.Fatalf("Integration binding response = %#v", got)
	}
	requireCatalogConsumerScope(
		t,
		engine.integrationScopes[0],
		capability,
	)
}

func TestRuntimeAPIProcessCatalogConsumerViewV2CrossesRegisteredEndpoint(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 17, "io.nexuscrm.test.catalog-consumer.v2",
		catalogsemantic.AppMetadataV2CapabilityName, catalogsnapshot.ProjectionRecordQueryV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	appView := validCatalogAppMetadataViewV2(17)
	appView.Lineage.SchemaGeneration = 8
	appView.Lineage.EngineGeneration = 9
	appView.Receipts.SchemaGeneration = 8
	appView.Receipts.EngineGeneration = 9
	appView.Receipts.SecurityEpoch = 1
	appView.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(appView)
	engine := &catalogConsumerViewTestEngine{appViewV2: appView}
	connection := startPlatformCapabilityProcess(
		t,
		fixture,
		catalogsemantic.AppMetadataV2CapabilityName,
		runtimegrpc.ResourceClassCatalogWatch,
		func(config *ProcessConfig) { config.CatalogConsumerViews = engine },
	)
	ctx, cancel := context.WithTimeout(
		platformWorkloadProtocolContext(context.Background(), catalogsemantic.AppMetadataV2CapabilityName),
		time.Second,
	)
	defer cancel()
	response, err := catalogv1.NewConsumerViewServiceClient(connection).GetAppMetadataViewV2(
		ctx,
		&catalogv1.GetAppMetadataViewV2Request{},
	)
	if err != nil {
		t.Fatal(err)
	}
	view, err := mapping.FromWireAppMetadataViewV2(response.GetView())
	if err != nil {
		t.Fatal(err)
	}
	if view.AppInstallationID != testInstallationID ||
		view.Lineage.ViewDigest != appView.Lineage.ViewDigest {
		t.Fatalf("Catalog V2 process response = %#v", view)
	}
	requireCatalogConsumerScope(
		t,
		engine.appV2Scopes[0],
		catalogsemantic.AppMetadataV2CapabilityName,
	)
}

type platformProcessSearchEngine struct {
	spec       queryspec.SearchSpec
	candidates []string
}

func (e *platformProcessSearchEngine) SearchWithSpec(
	_ context.Context,
	spec queryspec.SearchSpec,
	_ *models.UserSession,
) ([]contract.Record, error) {
	e.spec = spec
	return []contract.Record{{"Id": "widget-1"}}, nil
}

func (e *platformProcessSearchEngine) ScopeOrder(
	_ context.Context,
	_ *models.UserSession,
	candidates []string,
	_ permissionstatev1.Version,
) ([]string, error) {
	e.candidates = append([]string(nil), candidates...)
	result := append([]string(nil), candidates...)
	for left, right := 0, len(result)-1; left < right; left, right = left+1, right-1 {
		result[left], result[right] = result[right], result[left]
	}
	return result, nil
}

func TestRuntimeAPIProcessSearchCrossesRegisteredEndpoint(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 17, "io.nexuscrm.test.search.v1",
		searchsemantic.CapabilityName, catalogsnapshot.ProjectionRecordQueryV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	engine := &platformProcessSearchEngine{}
	connection := startPlatformCapabilityProcess(
		t,
		fixture,
		searchsemantic.CapabilityName,
		runtimegrpc.ResourceClassHeavyWork,
		func(config *ProcessConfig) {
			config.SearchEngine = engine
			config.SearchScopeOrder = engine
		},
	)
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(context.Background(), searchsemantic.CapabilityName),
		time.Second,
	)
	defer cancel()
	request := searchsemantic.Request{
		ObjectIDs: []string{"Widget"}, Term: "proof",
		SearchFields: []string{"Id"}, SelectedFields: []string{"Id"}, Limit: 1,
	}
	wireRequest, err := mapping.ToWireSearchRequest(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewRecordSearchServiceClient(connection).SearchRecords(ctx, wireRequest)
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireSearchResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.Records) != 1 || response.Records[0].RecordID != "widget-1" ||
		engine.spec.ObjectAPIName != "Widget" {
		t.Fatalf("Search process response = %#v spec=%#v", response, engine.spec)
	}

	scopeRequest := searchsemantic.ScopeOrderRequest{
		ObjectIDs: []string{"Widget"},
	}
	wireScopeRequest, err := mapping.ToWireSearchScopeOrderRequest(
		ctx, scopeRequest,
	)
	if err != nil {
		t.Fatal(err)
	}
	wireScopeResponse, err := runtimev1.NewRecordSearchServiceClient(
		connection,
	).ReadSearchScopeOrder(ctx, wireScopeRequest)
	if err != nil {
		t.Fatal(err)
	}
	scopeResponse, err := mapping.FromWireSearchScopeOrderResponse(
		scopeRequest, wireScopeResponse,
	)
	if err != nil {
		t.Fatal(err)
	}
	if len(scopeResponse.ObjectIDs) != 1 ||
		scopeResponse.ObjectIDs[0] != "Widget" ||
		len(engine.candidates) != 1 || engine.candidates[0] != "Widget" {
		t.Fatalf(
			"Search scope process response = %#v candidates=%#v",
			scopeResponse, engine.candidates,
		)
	}
}

func TestRuntimeAPIProcessFilesCrossesRegisteredEndpoint(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 17, "io.nexuscrm.test.files.v1",
		filessemantic.CapabilityName, catalogsnapshot.ProjectionRecordMutationV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	engine := &authorizedFilesCreateTestEngine{t: t}
	connection := startPlatformCapabilityProcess(
		t,
		fixture,
		filessemantic.CapabilityName,
		runtimegrpc.ResourceClassFileStream,
		func(config *ProcessConfig) { config.FileEngine = engine },
	)
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(context.Background(), filessemantic.CapabilityName),
		time.Second,
	)
	defer cancel()
	stream, err := runtimev1.NewAuthorizedFilesServiceClient(connection).CreateFile(ctx)
	if err != nil {
		t.Fatal(err)
	}
	header, err := mapping.ToWireCreateFileChunk(filessemantic.CreateChunk{
		Header: &filessemantic.CreateHeader{
			IdempotencyKey: "upload-process-1", ParentObjectID: "Widget",
			ParentRecordID: "widget-1", FileName: "proof.txt",
			ContentType: "text/plain", ContentLength: 5,
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	data, err := mapping.ToWireCreateFileChunk(filessemantic.CreateChunk{Data: []byte("proof")})
	if err != nil {
		t.Fatal(err)
	}
	if err := stream.Send(header); err != nil {
		t.Fatal(err)
	}
	if err := stream.Send(data); err != nil {
		t.Fatal(err)
	}
	wireResponse, err := stream.CloseAndRecv()
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireCreateFileResponse(wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if string(engine.body) != "proof" || response.FileName != "proof.txt" || response.Size != 5 {
		t.Fatalf("Files process response = %#v body=%q", response, engine.body)
	}
}

func TestRuntimeAPIProcessEffectivePermissionsCrossesRegisteredEndpoint(t *testing.T) {
	fixture := newServingFixtureForCapabilityAndObject(
		t, 17, "io.nexuscrm.test.effective-permissions.v1",
		permissionsemantic.CapabilityName,
		catalogsnapshot.ProjectionEffectivePermissionsV1,
		&models.ObjectMetadata{ID: "object-widget", APIName: "Widget", Label: "Widget", PluralLabel: "Widgets",
			SharingModel: constants.SharingModelPublicReadWrite,
			Fields:       []models.FieldMetadata{{ID: "field-widget-name", APIName: "Name", Label: "Name", Type: constants.FieldTypeString, Createable: true, Updateable: true}},
		},
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	engineCalled := false
	engine := effectivePermissionEngineFunc(func(
		_ context.Context,
		viewer *models.UserSession,
		target string,
		floor permissionstatev1.Version,
	) (*contract.EffectivePermissionsResult, error) {
		engineCalled = true
		if viewer.ID != testPrincipalID || target != "user-2" ||
			floor.Validate() != nil {
			t.Fatalf("effective-permission authority = %#v/%q", viewer, target)
		}
		return &contract.EffectivePermissionsResult{
			ObjectPermissions: []contract.ObjectAccessProjection{{
				ResourceType: "Widget", CanRead: true,
			}},
			FieldPermissions: []contract.FieldAccessProjection{{
				ResourceType: "Widget", FieldName: "Name", CanRead: true, CanCreate: true, CanEdit: false,
			}},
		}, nil
	})
	connection := startPlatformCapabilityProcess(
		t,
		fixture,
		permissionsemantic.CapabilityName,
		runtimegrpc.ResourceClassHeavyWork,
		func(config *ProcessConfig) { config.EffectivePermissions = engine },
	)
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(
			context.Background(), permissionsemantic.CapabilityName,
		),
		time.Second,
	)
	defer cancel()
	request := permissionsemantic.Request{TargetPrincipalID: "user-2"}
	wireRequest, err := mapping.ToWireReadEffectivePermissionsRequest(
		ctx, request,
	)
	if err != nil {
		t.Fatal(err)
	}
	wireResponse, err := runtimev1.NewEffectivePermissionServiceClient(
		connection,
	).ReadEffectivePermissions(ctx, wireRequest)
	if err != nil {
		t.Fatalf("ReadEffectivePermissions() = %v; engine called = %v", err, engineCalled)
	}
	response, err := mapping.FromWireReadEffectivePermissionsResponse(
		wireResponse,
	)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.ObjectPermissions) != 1 ||
		response.ObjectPermissions[0].ResourceType != "Widget" ||
		!response.ObjectPermissions[0].CanRead ||
		response.Receipts.CatalogRevision != 17 {
		t.Fatalf("effective-permission process response = %#v", response)
	}
	if len(response.FieldPermissions) != 1 || !response.FieldPermissions[0].CanCreate || response.FieldPermissions[0].CanEdit {
		t.Fatalf("registered endpoint lost distinct field creation authority: %#v", response.FieldPermissions)
	}
}

func TestRuntimeAPIProcessEffectivePermissionsPropagatesCallerCancellation(t *testing.T) {
	fixture := newServingFixtureForCapability(
		t, 17, "io.nexuscrm.test.effective-permissions.v1",
		permissionsemantic.CapabilityName,
		catalogsnapshot.ProjectionEffectivePermissionsV1,
	)
	if err := fixture.state.Reconcile(t.Context(), fixture.selection); err != nil {
		t.Fatal(err)
	}
	started := make(chan struct{})
	canceled := make(chan struct{})
	engine := effectivePermissionEngineFunc(func(
		ctx context.Context,
		_ *models.UserSession,
		_ string,
		_ permissionstatev1.Version,
	) (*contract.EffectivePermissionsResult, error) {
		close(started)
		<-ctx.Done()
		close(canceled)
		return nil, ctx.Err()
	})
	connection := startPlatformCapabilityProcess(
		t,
		fixture,
		permissionsemantic.CapabilityName,
		runtimegrpc.ResourceClassHeavyWork,
		func(config *ProcessConfig) { config.EffectivePermissions = engine },
	)
	ctx, cancel := context.WithTimeout(
		platformProtocolContext(
			context.Background(), permissionsemantic.CapabilityName,
		),
		time.Second,
	)
	request, err := mapping.ToWireReadEffectivePermissionsRequest(
		ctx, permissionsemantic.Request{},
	)
	if err != nil {
		cancel()
		t.Fatal(err)
	}
	result := make(chan error, 1)
	go func() {
		_, err := runtimev1.NewEffectivePermissionServiceClient(
			connection,
		).ReadEffectivePermissions(ctx, request)
		result <- err
	}()
	select {
	case <-started:
	case <-time.After(time.Second):
		cancel()
		t.Fatal("effective-permission engine did not start")
	}
	cancel()
	select {
	case err := <-result:
		if status.Code(err) != codes.Canceled {
			t.Fatalf("canceled RPC = %v, want Canceled", err)
		}
	case <-time.After(time.Second):
		t.Fatal("canceled RPC did not return")
	}
	select {
	case <-canceled:
	case <-time.After(time.Second):
		t.Fatal("caller cancellation did not reach the permission owner")
	}
}

func startPlatformCapabilityProcess(
	t *testing.T,
	fixture *servingFixture,
	capability string,
	resourceClass runtimegrpc.ResourceClass,
	configure func(*ProcessConfig),
) *grpc.ClientConn {
	t.Helper()
	config := ProcessConfig{
		Version: "test-sha", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
		AdmissionLimit: 2, ServingEnabled: true,
		PlatformCapabilityMaxAdmissionLeases: map[string]time.Duration{
			capability: time.Second,
		},
		WorkloadAuthority: testWorkloadAuthority,
		SecurityAdmission: testSecurityAdmission(t),
		ServingAdmission:  fixture.admission,
	}
	if resourceClass != runtimegrpc.ResourceClassShortUnary {
		config.PlatformResourceClassAdmissionLimits = map[runtimegrpc.ResourceClass]int{
			resourceClass: 1,
		}
	}
	configure(&config)
	process, err := Assemble(config, &readerStub{})
	if err != nil {
		t.Fatal(err)
	}
	if err := process.Start(); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { stopProcess(t, process) })
	connection, err := grpc.NewClient(
		"passthrough:///"+process.GRPCAddress(),
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = connection.Close() })
	return connection
}

func platformProtocolContext(ctx context.Context, capability string) context.Context {
	ctx = platformWorkloadProtocolContext(ctx, capability)
	return withTestDelegation(
		ctx,
		mustSignTestDelegation(testClaimsFor(time.Now(), capability)),
	)
}

func platformWorkloadProtocolContext(ctx context.Context, capability string) context.Context {
	ctx = metadata.AppendToOutgoingContext(
		ctx,
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(searchsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(searchsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, capability,
	)
	return ctx
}