//go:build runtime_worker_integration

package runtimeworker

import (
	"bytes"
	"context"
	"crypto/ed25519"
	"database/sql"
	"encoding/json"
	"os"
	"sort"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/nexuscrm/audit"
	"github.com/nexuscrm/automation"
	runtimeapi "github.com/nexuscrm/backend/internal/runtimeapi"
	database "github.com/nexuscrm/db"
	eventfeature "github.com/nexuscrm/events"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/records"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/pingcap-inc/nexuscrm-api-host/runtimegrpc"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/appauthority"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/delegationv1"
	"github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/permissionstatev1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
	"github.com/pingcap-inc/nexuscrm-contract/wire-go/envelope"
	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/credentials/insecure"
	"google.golang.org/grpc/metadata"
)

const (
	m1cOrgID             = "org-test"
	m1cInstallationID    = integrationAppInstallationID
	m1cWorkloadID        = "sfcrm-app-service"
	m1cRuntimeAudience   = "runtime-m1c"
	m1cPrincipalID       = "user-m1c"
	m1cProfileID         = "profile-m1c"
	m1cSessionID         = "session-m1c"
	m1cSigningKeyID      = "m1c-key"
	m1cMutationAppCap    = "io.nexuscrm.sfcrm.records-mutation.v1"
	m1cFlowAppCap        = "io.nexuscrm.sfcrm.flow-execution.v1"
	m1cScheduledRecordID = "a02000000000999AAA"
)

var m1cPrivateKey = ed25519.NewKeyFromSeed(bytes.Repeat([]byte{0x51}, ed25519.SeedSize))

// TestM1CMutationToTimedFlowCrossProcessWalkingSlice starts before the durable
// row: one real mutation RPC crosses the sole Runtime API listener, Runtime API
// stops, and the independently composed record-event and scheduled roles
// finish the immutable, revision-bound work through TiDB.
func TestM1CMutationToTimedFlowCrossProcessWalkingSlice(t *testing.T) {
	if os.Getenv("TIDB_HOST") == "" {
		t.Skip("M1-C walking slice requires the declared TiDB environment")
	}
	db, cleanup := openWorkerScratchDB(t)
	defer cleanup()
	prepareM1CRuntimeMutationSchema(t, db)

	control, err := persistence.NewRuntimeWorkOwnershipControl(db)
	if err != nil {
		t.Fatal(err)
	}
	var eventOwner persistence.RuntimeWorkOwnership
	for _, activation := range []persistence.RuntimeWorkOwnershipActivation{
		{
			OperationID: "m1c-activate-audit", WorkFamily: audit.AuditHistoryFanoutWorkFamily,
			DeploymentID: "m1c-audit", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "m1c-activate-events", WorkFamily: eventfeature.RuntimeRecordEventDispatchWorkFamily,
			DeploymentID: "m1c-events", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "m1c-activate-rollups", WorkFamily: records.RuntimeRollupDeltaWorkFamily,
			DeploymentID: "m1c-rollups", WorkerReleaseDigest: integrationWorkerDigest,
		},
		{
			OperationID: "m1c-activate-commands", WorkFamily: persistence.RuntimeRecordCommandLifecycleWorkFamily,
			DeploymentID: "m1c-commands", WorkerReleaseDigest: integrationWorkerDigest,
		},
	} {
		owner := activateWorker(t, control, activation)
		if activation.WorkFamily == eventfeature.RuntimeRecordEventDispatchWorkFamily {
			eventOwner = owner
		}
	}
	scheduledOwner := activateWorker(t, control, persistence.RuntimeWorkOwnershipActivation{
		OperationID: "m1c-activate-scheduled", WorkFamily: automation.RuntimeScheduledFlowWorkFamily,
		DeploymentID: "m1c-scheduled", WorkerReleaseDigest: integrationWorkerDigest,
	})
	ageScheduledFlowOwnershipCutover(t, db, scheduledOwner)

	snapshotArtifact, mutationArtifact, mutationView, flowArtifact, flowView :=
		compileWorkerMutationArtifacts(t)
	storeRoot := t.TempDir()
	store, err := catalogsnapshot.NewDirectoryStore(storeRoot)
	if err != nil {
		t.Fatal(err)
	}
	for _, artifact := range []catalogsnapshot.Artifact{
		snapshotArtifact, mutationArtifact, flowArtifact,
	} {
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	appRelease := catalogsnapshot.ComputeDigest([]byte("m1c-app-release"))
	serviceRelease := catalogsnapshot.ComputeDigest([]byte("m1c-runtime-release"))
	scheduledReceipt, err := catalogsnapshot.BuildServingCompatibilityReceipt(
		catalogsnapshot.ServingCompatibilityReceiptInput{
			OrgID: m1cOrgID, AppInstallationID: m1cInstallationID,
			AppReleaseDigest: appRelease, AppCapabilityID: integrationScheduledAppCapability,
			RuntimeCapabilityID: flowView.CapabilityID(), WorkloadID: "runtime-scheduled-worker",
			ServiceReleaseDigest: serviceRelease, OwnershipEpoch: scheduledOwner.OwnershipEpoch,
			CatalogSnapshotDigest: snapshotArtifact.Digest(), RuntimeMetadataViewDigest: flowArtifact.Digest(),
			Lineage: flowView.Lineage(), SecurityEpoch: 1,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := store.Put(t.Context(), scheduledReceipt); err != nil {
		t.Fatal(err)
	}
	seedScheduledCatalogCapability(
		t, db, flowView, appRelease, serviceRelease,
		scheduledReceipt.Digest(), scheduledOwner.OwnershipEpoch,
	)

	if _, err := db.Exec(`CREATE TABLE ScheduledTarget (
Id VARCHAR(191) PRIMARY KEY,
Status VARCHAR(191) NULL,
IsDeleted TINYINT(1) NOT NULL DEFAULT 0,
LastModifiedById VARCHAR(191) NULL,
LastModifiedDate DATETIME(6) NULL
)`); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Exec(
		`INSERT INTO ScheduledTarget (Id, Status) VALUES (?, 'before')`,
		m1cScheduledRecordID,
	); err != nil {
		t.Fatal(err)
	}

	scheduledConfig := integrationMetadataWorkerConfig(scheduledOwner, storeRoot, flowView)
	scheduledConfig.Role.AppCapabilityID = integrationScheduledAppCapability
	scheduledConfig.Role.ExpectedCompatibilityReceiptDigest = string(scheduledReceipt.Digest())
	scheduledConfig.Role.PollInterval = "50ms"
	scheduledConfig.Role.LeaseDuration = "10s"
	mismatchedEventConfig := integrationMetadataWorkerConfig(eventOwner, storeRoot, mutationView)
	mismatchedScheduledRole := scheduledConfig.Role
	mismatchedScheduledRole.DeploymentID = "m1c-scheduled-other"
	mismatchedEventConfig.TimedFlowOwner = &mismatchedScheduledRole
	_, mismatchedDB, err := assembleProduction(
		t.Context(),
		integrationWorkerSourceRevision,
		mismatchedEventConfig,
		integrationPermissionStateFactory(t),
	)
	if mismatchedDB != nil {
		_ = mismatchedDB.Close()
	}
	if err == nil || !strings.Contains(err.Error(), "exact scheduled Flow owner is not ready") {
		t.Fatalf("mismatched scheduled owner error = %v", err)
	}
	scheduledProcess, scheduledDB, err := assembleProduction(
		t.Context(),
		integrationWorkerSourceRevision,
		scheduledConfig,
		integrationPermissionStateFactory(t),
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := scheduledProcess.Start(); err != nil {
		_ = scheduledDB.Close()
		t.Fatal(err)
	}
	defer stopWorker(t, scheduledProcess, scheduledDB)
	assertWorkerOperationalSurface(t, scheduledProcess, scheduledOwner.WorkFamily)

	servingAdmission := newM1CServingAdmission(
		t, store, snapshotArtifact, mutationArtifact, flowArtifact, appRelease, serviceRelease,
	)
	apiProcess, err := runtimeapi.AssembleProductionWithConnection(
		t.Context(),
		runtimeapi.ProcessConfig{
			Version: "m1c-integration", AdminAddress: "127.0.0.1:0", GRPCAddress: "127.0.0.1:0",
			AdmissionLimit: runtimeapi.DefaultAdmissionLimit,
			WorkloadAuthority: func(context.Context, string) (runtimegrpc.VerifiedCaller, error) {
				return runtimegrpc.VerifiedCaller(m1cWorkloadID), nil
			},
			SecurityAdmission:              runtimeM1CSecurityAdmission(t),
			ServingAdmission:               servingAdmission,
			ServingEnabled:                 true,
			TransportCredentials:           insecure.NewCredentials(),
			RecordMutationAttemptCap:       5 * time.Second,
			RecordMutationOutcomeRetention: time.Hour,
			PermissionStateCapabilities:    []string{"permission_state.v1"},
			PermissionStateReady:           func() bool { return true },
		},
		database.NewConnectionFromDB(db),
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := apiProcess.Start(); err != nil {
		t.Fatal(err)
	}
	apiAddress := apiProcess.GRPCAddress()
	if apiAddress == "" {
		t.Fatal("Runtime API did not expose its one private gRPC endpoint")
	}
	mutateM1CScheduledTarget(t, apiAddress)
	stopM1CProcess(t, apiProcess)

	var eventID, eventCapability string
	var eventPayload []byte
	var eventCount, reservationCount int
	var preWorkerStatus string
	if err := db.QueryRow(`SELECT
(SELECT Id FROM _System_RecordEventOutbox WHERE RecordKey = ? ORDER BY CreatedDate LIMIT 1),
(SELECT RuntimeCapabilityId FROM _System_RecordEventOutbox WHERE RecordKey = ? ORDER BY CreatedDate LIMIT 1),
(SELECT Payload FROM _System_RecordEventOutbox WHERE RecordKey = ? ORDER BY CreatedDate LIMIT 1),
(SELECT COUNT(*) FROM _System_RecordEventOutbox WHERE RecordKey = ?),
(SELECT Status FROM ScheduledTarget WHERE Id = ?),
(SELECT COUNT(*) FROM _System_ScheduledRunReservation
  WHERE FlowDefinitionId = 'definition-timed-integration'
    AND InputKey = ?)`,
		"ScheduledTarget/"+m1cScheduledRecordID,
		"ScheduledTarget/"+m1cScheduledRecordID,
		"ScheduledTarget/"+m1cScheduledRecordID,
		"ScheduledTarget/"+m1cScheduledRecordID,
		m1cScheduledRecordID,
		m1cScheduledRecordID,
	).Scan(
		&eventID,
		&eventCapability,
		&eventPayload,
		&eventCount,
		&preWorkerStatus,
		&reservationCount,
	); err != nil {
		t.Fatal(err)
	}
	if eventCapability != mutationsemantic.CapabilityName || eventCount != 1 {
		t.Fatalf("Runtime API event capability/count = %q/%d", eventCapability, eventCount)
	}
	if preWorkerStatus != "triggered" || reservationCount != 0 {
		t.Fatalf(
			"before record-event Worker: record status/reservations = %q/%d, want triggered/0 (no synchronous or process-local timed fallback)",
			preWorkerStatus,
			reservationCount,
		)
	}
	var eventEnvelope eventfeature.InternalRecordEventEnvelope
	if err := json.Unmarshal(eventPayload, &eventEnvelope); err != nil {
		t.Fatal(err)
	}
	eventBindings, err := catalogsnapshot.OpenRuntimeWorkBindingSet(
		eventEnvelope.InternalMetadata["runtime_work_metadata_binding"],
	)
	if err != nil {
		t.Fatal(err)
	}
	mutationBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(mutationView)
	if err != nil {
		t.Fatal(err)
	}
	flowBinding, err := catalogsnapshot.RuntimeWorkBindingFromView(flowView)
	if err != nil {
		t.Fatal(err)
	}
	if eventBindings.Primary != mutationBinding ||
		eventBindings.FlowWork == nil ||
		*eventBindings.FlowWork != flowBinding {
		t.Fatalf("Runtime API event bindings = %#v, want one mutation/Flow revision", eventBindings)
	}
	if _, err := db.Exec(`UPDATE _System_RecordEventOutbox
SET CreatedDate = UTC_TIMESTAMP(6) - INTERVAL 2 MINUTE,
    LastModifiedDate = UTC_TIMESTAMP(6) - INTERVAL 2 MINUTE
WHERE Id = ?`, eventID); err != nil {
		t.Fatal(err)
	}

	eventConfig := integrationMetadataWorkerConfig(eventOwner, storeRoot, mutationView)
	eventConfig.Role.ProcessedRetention = "1h"
	eventConfig.Role.ProcessedCleanupInterval = "1m"
	scheduledRole := scheduledConfig.Role
	eventConfig.TimedFlowOwner = &scheduledRole
	eventProcess, eventDB, err := assembleProduction(
		t.Context(),
		integrationWorkerSourceRevision,
		eventConfig,
		integrationPermissionStateFactory(t),
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := eventProcess.Start(); err != nil {
		_ = eventDB.Close()
		t.Fatal(err)
	}
	assertWorkerOperationalSurface(t, eventProcess, eventOwner.WorkFamily)
	waitForTimedHandoffWalkingSlice(t, db, eventID, m1cScheduledRecordID)
	stopWorker(t, eventProcess, eventDB)
}

func newM1CServingAdmission(
	t *testing.T,
	store catalogsnapshot.Store,
	snapshotArtifact, mutationArtifact, flowArtifact catalogsnapshot.Artifact,
	appRelease, serviceRelease catalogsnapshot.Digest,
) *runtimeapi.ServingAdmission {
	t.Helper()
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	appMetadataArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(
		snapshot,
		catalogsnapshot.ProjectionKey{
			AppInstallationID: m1cInstallationID,
			CapabilityID:      catalogsemantic.AppMetadataV2CapabilityName,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := store.Put(t.Context(), appMetadataArtifact); err != nil {
		t.Fatal(err)
	}
	flowView, err := catalogsnapshot.OpenRuntimeMetadataView(flowArtifact)
	if err != nil {
		t.Fatal(err)
	}
	appMetadataView := newM1CAppMetadataView(t, flowView, appRelease)
	type capability struct {
		appID   string
		runtime string
		view    catalogsnapshot.Artifact
		epoch   uint64
	}
	capabilities := []capability{
		{appID: m1cMutationAppCap, runtime: mutationsemantic.CapabilityName, view: mutationArtifact, epoch: 11},
		{appID: m1cFlowAppCap, runtime: flowArtifactRuntimeCapability(t, flowArtifact), view: flowArtifact, epoch: 12},
		{
			appID: "io.nexuscrm.sfcrm.app-metadata.v2", runtime: catalogsemantic.AppMetadataV2CapabilityName,
			view: appMetadataArtifact, epoch: 13,
		},
	}
	bindings := make(map[string]runtimeapi.ServingHandlerBinding, len(capabilities))
	selections := make([]runtimeapi.ServingSelection, 0, len(capabilities))
	for _, capability := range capabilities {
		receipt, err := catalogsnapshot.BuildServingCompatibilityReceipt(
			catalogsnapshot.ServingCompatibilityReceiptInput{
				OrgID: m1cOrgID, AppInstallationID: m1cInstallationID,
				AppReleaseDigest: appRelease, AppCapabilityID: capability.appID,
				RuntimeCapabilityID: capability.runtime, WorkloadID: m1cWorkloadID,
				ServiceReleaseDigest: serviceRelease, OwnershipEpoch: capability.epoch,
				CatalogSnapshotDigest:     snapshotArtifact.Digest(),
				RuntimeMetadataViewDigest: capability.view.Digest(),
				Lineage: catalogsnapshot.Lineage{
					CatalogRevision: 7, SchemaGeneration: 1, EngineGeneration: 1,
				},
				SecurityEpoch: 1,
			},
		)
		if err != nil {
			t.Fatal(err)
		}
		if err := store.Put(t.Context(), receipt); err != nil {
			t.Fatal(err)
		}
		bindings[capability.appID] = runtimeapi.ServingHandlerBinding{
			OrgID: m1cOrgID, AppInstallationID: m1cInstallationID,
			AppReleaseDigest: string(appRelease), CapabilityID: capability.appID,
			WorkloadID: m1cWorkloadID, ServiceReleaseDigest: string(serviceRelease),
			OwnershipEpoch: capability.epoch, CompatibilityReceiptDigest: string(receipt.Digest()),
		}
		selections = append(selections, runtimeapi.ServingSelection{
			OrgID: m1cOrgID, AppInstallationID: m1cInstallationID,
			ExpectedWorkloadID: m1cWorkloadID, AppCapabilityID: capability.appID,
			RuntimeCapabilityID: capability.runtime, ExpectedServiceReleaseDigest: serviceRelease,
			ExpectedOwnershipEpoch:             capability.epoch,
			ExpectedCompatibilityReceiptDigest: receipt.Digest(),
			LocalCompatibility: catalogsnapshot.Compatibility{
				SchemaGeneration: 1, EngineGeneration: 1,
			},
			SecurityEpoch: 1,
		})
	}
	state, err := runtimeapi.NewServingState(runtimeapi.ServingStateConfig{
		Store: store,
		Catalog: func(
			_ context.Context,
			installationID, capabilityID string,
		) (runtimeapi.ServingInstallation, runtimeapi.ServingHandlerBinding, bool, error) {
			binding, found := bindings[capabilityID]
			if installationID != m1cInstallationID || !found {
				return runtimeapi.ServingInstallation{}, runtimeapi.ServingHandlerBinding{}, false, nil
			}
			return runtimeapi.ServingInstallation{
				OrgID: m1cOrgID, AppInstallationID: m1cInstallationID,
				ActiveReleaseDigest: string(appRelease), CatalogRevision: 7,
			}, binding, true, nil
		},
		AppMetadata: m1cAppMetadataEngine{view: appMetadataView},
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := state.ReconcileAll(t.Context(), selections); err != nil {
		t.Fatal(err)
	}
	admission, err := runtimeapi.NewServingAdmission(state)
	if err != nil {
		t.Fatal(err)
	}
	return admission
}

type m1cAppMetadataEngine struct {
	view catalogsemantic.AppMetadataViewV2
}

func (engine m1cAppMetadataEngine) GetAppMetadataViewV2(
	context.Context,
	runtimeapi.CatalogConsumerScope,
	catalogsemantic.GetRequest,
) (catalogsemantic.AppMetadataViewV2, error) {
	return engine.view, nil
}

func newM1CAppMetadataView(
	t *testing.T,
	runtimeView *catalogsnapshot.RuntimeMetadataView,
	appRelease catalogsnapshot.Digest,
) catalogsemantic.AppMetadataViewV2 {
	t.Helper()
	const componentKind = "metadata.pack.standard-flow.v1"
	const namespaceID = "io.neutral"
	components := make([]catalogsemantic.AppMetadataComponent, 0, 3)
	for _, definitionRef := range []string{"ScheduledIntegration", "ScheduledRollback", "TimedIntegration"} {
		version, ok := runtimeView.ActiveFlowVersion(definitionRef)
		if !ok {
			t.Fatalf("M1-C Runtime view has no active Flow %q", definitionRef)
		}
		behavior := append([]byte(nil), version.MetadataRaw...)
		contentDigest, err := appauthority.ComponentDigest(componentKind, 1, behavior)
		if err != nil {
			t.Fatal(err)
		}
		components = append(components, catalogsemantic.AppMetadataComponent{
			ComponentID:   appauthority.ComponentID(namespaceID, componentKind, definitionRef),
			ComponentKind: componentKind, ComponentFormatVersion: 1, NamespaceID: namespaceID,
			BusinessKey: definitionRef, DefinitionVersion: strconv.Itoa(version.VersionNumber),
			ContentDigest: contentDigest, CanonicalBehavior: behavior,
		})
	}
	sort.Slice(components, func(i, j int) bool {
		return components[i].ComponentID < components[j].ComponentID
	})
	lineage := runtimeView.Lineage()
	view := catalogsemantic.AppMetadataViewV2{
		OrgID: m1cOrgID, AppInstallationID: m1cInstallationID,
		AppID: "io.nexuscrm.neutral", AppReleaseDigest: string(appRelease),
		Lineage: catalogsemantic.ViewLineage{
			CatalogRevision:      uint64(lineage.CatalogRevision),
			SchemaGeneration:     uint64(lineage.SchemaGeneration),
			EngineGeneration:     uint64(lineage.EngineGeneration),
			SourceSnapshotDigest: string(runtimeView.SourceSnapshotDigest()),
		},
		AppComponents: components,
		Receipts: catalogsemantic.Receipts{
			ProtocolVersion: "1.0", CapabilityVersion: catalogsemantic.AppMetadataV2CapabilityName,
			CatalogRevision:  uint64(lineage.CatalogRevision),
			SchemaGeneration: uint64(lineage.SchemaGeneration),
			EngineGeneration: uint64(lineage.EngineGeneration), SecurityEpoch: 1,
		},
	}
	view.Lineage.ViewDigest = catalogsemantic.ComputeAppMetadataViewV2Digest(view)
	if err := catalogsemantic.ValidateAppMetadataViewV2(view); err != nil {
		t.Fatalf("M1-C AppMetadata V2 fixture is invalid: %v", err)
	}
	return view
}

func flowArtifactRuntimeCapability(t *testing.T, artifact catalogsnapshot.Artifact) string {
	t.Helper()
	view, err := catalogsnapshot.OpenRuntimeMetadataView(artifact)
	if err != nil {
		t.Fatal(err)
	}
	return view.CapabilityID()
}

func runtimeM1CSecurityAdmission(t *testing.T) *runtimeapi.SecurityAdmission {
	t.Helper()
	verifier, err := delegationv1.NewVerifier(delegationv1.VerifierConfig{
		Keys: []delegationv1.VerificationKey{{
			SigningKeyID: m1cSigningKeyID,
			PublicKey:    m1cPrivateKey.Public().(ed25519.PublicKey),
			State:        delegationv1.KeyStateCurrent,
		}},
		SupportedCapabilities: []string{mutationsemantic.CapabilityName},
	})
	if err != nil {
		t.Fatal(err)
	}
	now := time.Now
	state, err := runtimeapi.NewRuntimeSecurityState(t.Context(), runtimeapi.RuntimeSecurityStateConfig{
		Now: now,
		Replica: func(context.Context) (runtimeapi.RuntimeReplicaSecurityState, bool, error) {
			return runtimeapi.RuntimeReplicaSecurityState{
				OrgID: m1cOrgID, RuntimeAudience: m1cRuntimeAudience, SecurityEpoch: 1,
				LeaseSecurityEpoch: 1, LeaseExpiresAt: now().Add(time.Hour),
			}, true, nil
		},
		Admission: func(_ context.Context, claims delegationv1.Claims) (runtimeapi.RuntimeAdmissionSecurityState, error) {
			return runtimeapi.RuntimeAdmissionSecurityState{
				StateFound: true, InstallationFound: claims.AppInstallationID == m1cInstallationID,
				InstallationActive: true, Capabilities: []string{mutationsemantic.CapabilityName},
				SecurityEpoch: 1, LeaseFound: true, LeaseSecurityEpoch: 1,
				LeaseExpiresAt: now().Add(time.Hour),
			}, nil
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	subject, err := permissionstatev1.NewSubject(m1cPrincipalID, m1cProfileID)
	if err != nil {
		t.Fatal(err)
	}
	lineage, err := permissionstatev1.LineageFromBytes(
		[]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16},
	)
	if err != nil {
		t.Fatal(err)
	}
	version, err := permissionstatev1.NewVersion(lineage, 1)
	if err != nil {
		t.Fatal(err)
	}
	admission, err := runtimeapi.NewSecurityAdmission(runtimeapi.SecurityAdmissionConfig{
		Verifier: verifier,
		Bindings: func(_ context.Context, caller runtimegrpc.VerifiedCaller) (runtimeapi.RuntimeCallerBinding, bool, error) {
			return runtimeapi.RuntimeCallerBinding{CallerWorkloadID: m1cWorkloadID},
				caller == runtimegrpc.VerifiedCaller(m1cWorkloadID), nil
		},
		Sessions: func(context.Context, string, string) (*runtimeapi.RuntimeSessionState, error) {
			return &runtimeapi.RuntimeSessionState{
				User:      models.UserSession{ID: m1cPrincipalID, ProfileID: m1cProfileID},
				AuthEpoch: 1, Subject: subject, PermissionFloor: version,
			}, nil
		},
		State: state,
	})
	if err != nil {
		t.Fatal(err)
	}
	return admission
}

func mutateM1CScheduledTarget(t *testing.T, address string) {
	t.Helper()
	connection, err := grpc.NewClient(
		"passthrough:///"+address,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithDefaultCallOptions(
			grpc.MaxCallSendMsgSize(envelope.MaxRequestWireBytes),
			grpc.MaxCallRecvMsgSize(envelope.MaxResponseWireBytes),
			grpc.WaitForReady(false),
		),
	)
	if err != nil {
		t.Fatal(err)
	}
	defer connection.Close()
	statusValue := "triggered"
	request := mutationsemantic.Request{
		CommandID: "m1c-trigger-timed-flow", BatchMode: mutationsemantic.BatchAtomic,
		Operations: []mutationsemantic.Operation{{
			Kind: mutationsemantic.OperationUpdate, ObjectID: "ScheduledTarget",
			RecordID: m1cScheduledRecordID,
			Fields: []mutationsemantic.FieldValue{{
				Name: "Status", Kind: mutationsemantic.ValueKindString, String: &statusValue,
			}},
		}},
	}
	ctx := metadata.AppendToOutgoingContext(
		context.Background(),
		envelope.ProtocolMajorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMajor), 10),
		envelope.ProtocolMinorMetadataKey, strconv.FormatUint(uint64(mutationsemantic.ProtocolMinor), 10),
		envelope.CapabilityMetadataKey, mutationsemantic.CapabilityName,
	)
	now := time.Now()
	token, err := delegationv1.Sign(m1cSigningKeyID, m1cPrivateKey, delegationv1.Claims{
		DelegationID: "m1c-delegation", OrgID: m1cOrgID, PrincipalID: m1cPrincipalID,
		SessionID: m1cSessionID, CallerWorkloadID: m1cWorkloadID,
		AppInstallationID: m1cInstallationID, RuntimeAudience: m1cRuntimeAudience,
		Capabilities: []string{mutationsemantic.CapabilityName}, AuthEpoch: 1, SecurityEpoch: 1,
		IssuedAt: now.Add(-time.Second).Unix(), ExpiresAt: now.Add(time.Minute).Unix(),
	})
	if err != nil {
		t.Fatal(err)
	}
	ctx = metadata.AppendToOutgoingContext(ctx, envelope.RuntimeDelegationMetadataKey, token)
	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	frames, err := mapping.EncodeMutationFrames(ctx, request)
	if err != nil {
		t.Fatal(err)
	}
	stream, err := runtimev1.NewRecordMutationsServiceClient(connection).MutateRecords(ctx)
	if err != nil {
		t.Fatal(err)
	}
	for _, frame := range frames {
		if err := stream.Send(frame); err != nil {
			t.Fatal(err)
		}
	}
	wireResponse, err := stream.CloseAndRecv()
	if err != nil {
		t.Fatal(err)
	}
	response, err := mapping.FromWireMutationResponse(request, wireResponse)
	if err != nil {
		t.Fatal(err)
	}
	if len(response.Results) != 1 || !response.Results[0].Success ||
		response.Results[0].RecordID != m1cScheduledRecordID {
		t.Fatalf("mutation response = %#v", response)
	}
}

func prepareM1CRuntimeMutationSchema(t *testing.T, db *sql.DB) {
	t.Helper()
	for _, statement := range []string{
		`CREATE TABLE ProcessDefinition (
			Id VARCHAR(255) PRIMARY KEY,
			LockType VARCHAR(64) NULL
		)`,
		`CREATE TABLE ProcessInstance (
			Id VARCHAR(255) PRIMARY KEY,
			ProcessDefinitionId VARCHAR(255) NOT NULL,
			TargetObjectId VARCHAR(255) NOT NULL,
			Status VARCHAR(64) NOT NULL,
			CreatedDate DATETIME(6) NOT NULL
		)`,
		`CREATE TABLE PermissionSet (
			Id VARCHAR(255) PRIMARY KEY,
			ProfileId VARCHAR(255) NULL,
			IsOwnedByProfile BOOL NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE PermissionSetAssignment (
			PermissionSetId VARCHAR(255) NOT NULL,
			AssigneeId VARCHAR(255) NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE PermissionSetGroup (
			Id VARCHAR(255) PRIMARY KEY,
			IsActive BOOL NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE PermissionSetGroupComponent (
			PermissionSetId VARCHAR(255) NOT NULL,
			GroupId VARCHAR(255) NOT NULL
		)`,
		`CREATE TABLE PermissionSetGroupAssignment (
			GroupId VARCHAR(255) NOT NULL,
			AssigneeId VARCHAR(255) NOT NULL,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE ObjectPermissions (
			ParentId VARCHAR(255) NOT NULL,
			SobjectType VARCHAR(255) NOT NULL,
			PermissionsRead BOOL NOT NULL DEFAULT 0,
			PermissionsCreate BOOL NOT NULL DEFAULT 0,
			PermissionsEdit BOOL NOT NULL DEFAULT 0,
			PermissionsDelete BOOL NOT NULL DEFAULT 0,
			PermissionsViewAllRecords BOOL NOT NULL DEFAULT 0,
			PermissionsModifyAllRecords BOOL NOT NULL DEFAULT 0,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`CREATE TABLE FieldPermissions (
			ParentId VARCHAR(255) NOT NULL,
			SobjectType VARCHAR(255) NOT NULL,
			Field VARCHAR(255) NOT NULL,
			PermissionsRead BOOL NOT NULL DEFAULT 0,
			PermissionsEdit BOOL NOT NULL DEFAULT 0,
			IsDeleted BOOL NOT NULL DEFAULT 0
		)`,
		`INSERT INTO PermissionSet
			(Id, ProfileId, IsOwnedByProfile, IsDeleted)
			VALUES ('m1c-profile-permissions', 'profile-m1c', 1, 0)`,
		`INSERT INTO ObjectPermissions
			(ParentId, SobjectType, PermissionsRead, PermissionsEdit)
			VALUES ('m1c-profile-permissions', 'scheduledtarget', 1, 1)`,
		`INSERT INTO FieldPermissions
			(ParentId, SobjectType, Field, PermissionsRead, PermissionsEdit)
			VALUES
			('m1c-profile-permissions', 'scheduledtarget', 'Id', 1, 0),
			('m1c-profile-permissions', 'scheduledtarget', 'Status', 1, 1)`,
	} {
		if _, err := db.Exec(statement); err != nil {
			t.Fatal(err)
		}
	}
}

func waitForTimedHandoffWalkingSlice(t *testing.T, db *sql.DB, eventID, recordID string) {
	t.Helper()
	deadline := time.Now().Add(30 * time.Second)
	for {
		var eventStatus, recordStatus string
		var eventLastError, reservationStatus sql.NullString
		err := db.QueryRow(`SELECT
(SELECT Status FROM _System_RecordEventOutbox WHERE Id = ?),
(SELECT ErrorMessage FROM _System_RecordEventOutbox WHERE Id = ?),
(SELECT Status FROM ScheduledTarget WHERE Id = ?),
(SELECT Status FROM _System_ScheduledRunReservation
  WHERE FlowDefinitionId = 'definition-timed-integration'
    AND InputKey = ?
  ORDER BY CreatedDate
  LIMIT 1)`,
			eventID,
			eventID,
			recordID,
			recordID,
		).Scan(&eventStatus, &eventLastError, &recordStatus, &reservationStatus)
		if err == nil && eventStatus == "failed" {
			t.Fatalf("timed handoff event failed: %s", eventLastError.String)
		}
		if err == nil &&
			eventStatus == "processed" &&
			recordStatus == "timed-complete" &&
			reservationStatus.Valid &&
			reservationStatus.String == "completed" {
			return
		}
		if time.Now().After(deadline) {
			t.Fatalf(
				"timed handoff did not converge: event=%q record=%q reservation=%q valid=%v err=%v",
				eventStatus,
				recordStatus,
				reservationStatus.String,
				reservationStatus.Valid,
				err,
			)
		}
		time.Sleep(20 * time.Millisecond)
	}
}

func stopM1CProcess(t *testing.T, process interface{ Stop(context.Context) error }) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := process.Stop(ctx); err != nil {
		t.Fatal(err)
	}
}