package runtimeworker

import (
	"context"
	"database/sql"
	"fmt"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/automation"
	"github.com/nexuscrm/backend/internal/managedruntime"
	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/persistence"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	catalogsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/catalog/consumerv1"
	commandsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	flowsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowv1"
	mutationsemantic "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/recordmutationv1"
)

const integrationAppInstallationID = "0198c7a1-2f31-7abc-8def-1234567890ab"

type workerAssignmentPolicyEligibilityStagerProbe struct {
	view catalogsnapshot.Digest
}

func (*workerAssignmentPolicyEligibilityStagerProbe) Stage(
	context.Context,
	*sql.Tx,
	[]permissions.AssignmentPolicyEligibilityMutation,
) error {
	return nil
}

func newTestMutationArtifactBinder(
	store catalogsnapshot.Store,
	scope persistence.RuntimeWorkScope,
	compatibility catalogsnapshot.Compatibility,
) (*mutationArtifactBinder, error) {
	return newMutationArtifactBinder(
		store,
		scope,
		compatibility,
		func(
			context.Context,
			*catalogsnapshot.RuntimeMetadataView,
		) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error) {
			return nil, nil
		},
	)
}

func TestMutationArtifactBinderReopensOnlyExactScopedView(t *testing.T) {
	store, view := workerMutationView(t)
	scope := persistence.RuntimeWorkScope{
		OrgID: view.OrgID(), AppInstallationID: view.AppInstallationID(), RuntimeCapabilityID: view.CapabilityID(),
	}
	lineage := view.Lineage()
	binder, err := newTestMutationArtifactBinder(store, scope, catalogsnapshot.Compatibility{
		SchemaGeneration: lineage.SchemaGeneration, EngineGeneration: lineage.EngineGeneration,
	})
	if err != nil {
		t.Fatal(err)
	}
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(view)
	if err != nil {
		t.Fatal(err)
	}
	encoded, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		t.Fatal(err)
	}
	install, err := binder.bindEncoded(context.Background(), encoded)
	if err != nil {
		t.Fatal(err)
	}
	executionCtx, err := install(context.Background())
	if err != nil {
		t.Fatalf("install durable execution envelope: %v", err)
	}
	bound, ok := metadata.RuntimeMetadataViewFromContext(executionCtx)
	if !ok || bound.Digest() != view.Digest() {
		t.Fatalf("bound view = %#v, %v", bound, ok)
	}
	firstEngine, engineBound, err := metadata.RuntimeFormulaEngineFromContext(executionCtx)
	if err != nil || !engineBound {
		t.Fatalf("first Runtime formula engine = %p/%v/%v, want bound engine", firstEngine, engineBound, err)
	}
	secondEngine, engineBound, err := metadata.RuntimeFormulaEngineFromContext(executionCtx)
	if err != nil || !engineBound || secondEngine != firstEngine {
		t.Fatalf("second Runtime formula engine = %p/%v/%v, want execution-cached %p", secondEngine, engineBound, err, firstEngine)
	}

	outside := binding
	outside.OrgID = "another-org"
	if _, err := binder.open(context.Background(), outside); err == nil {
		t.Fatal("cross-scope durable binding was accepted")
	}
	wrongCompatibility, err := newTestMutationArtifactBinder(store, scope, catalogsnapshot.Compatibility{
		SchemaGeneration: lineage.SchemaGeneration, EngineGeneration: lineage.EngineGeneration + 1,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := wrongCompatibility.open(context.Background(), binding); err == nil {
		t.Fatal("incompatible durable view was accepted")
	}
}

func TestMutationArtifactBinderPinsLifecycleBindingToExactDurableView(t *testing.T) {
	store, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	views := make([]*catalogsnapshot.RuntimeMetadataView, 0, 2)
	for _, revision := range []catalogsnapshot.CatalogRevision{21, 22} {
		snapshotArtifact, mutationArtifact, view, _, _ :=
			compileWorkerMutationArtifactsForRevision(t, revision)
		if err := store.Put(t.Context(), snapshotArtifact); err != nil {
			t.Fatal(err)
		}
		if err := store.Put(t.Context(), mutationArtifact); err != nil {
			t.Fatal(err)
		}
		views = append(views, view)
	}
	scope := persistence.RuntimeWorkScope{
		OrgID: views[0].OrgID(), AppInstallationID: views[0].AppInstallationID(),
		RuntimeCapabilityID: views[0].CapabilityID(),
	}
	lineage := views[0].Lineage()
	prepared := make(map[catalogsnapshot.Digest]int)
	stagers := make(map[catalogsnapshot.Digest]*workerAssignmentPolicyEligibilityStagerProbe)
	binder, err := newMutationArtifactBinder(
		store,
		scope,
		catalogsnapshot.Compatibility{
			SchemaGeneration: lineage.SchemaGeneration,
			EngineGeneration: lineage.EngineGeneration,
		},
		func(
			_ context.Context,
			view *catalogsnapshot.RuntimeMetadataView,
		) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error) {
			digest := view.Digest()
			prepared[digest]++
			stager := &workerAssignmentPolicyEligibilityStagerProbe{view: digest}
			stagers[digest] = stager
			return stager, nil
		},
	)
	if err != nil {
		t.Fatal(err)
	}

	for _, view := range views {
		binding, err := catalogsnapshot.RuntimeWorkBindingFromView(view)
		if err != nil {
			t.Fatal(err)
		}
		encoded, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
		if err != nil {
			t.Fatal(err)
		}
		for attempt := 0; attempt < 2; attempt++ {
			install, err := binder.bindEncoded(t.Context(), encoded)
			if err != nil {
				t.Fatal(err)
			}
			executionCtx, err := install(t.Context())
			if err != nil {
				t.Fatal(err)
			}
			got, ok := managedruntime.AssignmentPolicyEligibilityAmbientStagerFromContext(executionCtx)
			if !ok || got != stagers[view.Digest()] {
				t.Fatalf("view %s lifecycle stager = %#v, %v", view.Digest(), got, ok)
			}
		}
		if prepared[view.Digest()] != 1 {
			t.Fatalf(
				"view %s lifecycle preparations = %d, want one content-addressed preparation",
				view.Digest(),
				prepared[view.Digest()],
			)
		}
	}
	if stagers[views[0].Digest()] == stagers[views[1].Digest()] {
		t.Fatal("different immutable mutation views shared one lifecycle stager")
	}
}

func TestMutationArtifactBinderCoalescesConcurrentImmutableReopen(t *testing.T) {
	store, mutationView, flowView := workerMutationAndFlowViews(t)
	counted := &countingArtifactStore{Store: store, gets: make(map[catalogsnapshot.ArtifactKind]int)}
	scope := persistence.RuntimeWorkScope{
		OrgID: mutationView.OrgID(), AppInstallationID: mutationView.AppInstallationID(),
		RuntimeCapabilityID: mutationView.CapabilityID(),
	}
	lineage := mutationView.Lineage()
	binder, err := newTestMutationArtifactBinder(counted, scope, catalogsnapshot.Compatibility{
		SchemaGeneration: lineage.SchemaGeneration,
		EngineGeneration: lineage.EngineGeneration,
	})
	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)
	}
	encoded, err := catalogsnapshot.EncodeRuntimeWorkBindingSet(catalogsnapshot.RuntimeWorkBindingSet{
		Primary:  mutationBinding,
		FlowWork: &flowBinding,
	})
	if err != nil {
		t.Fatal(err)
	}

	const callers = 24
	start := make(chan struct{})
	errs := make(chan error, callers)
	var workers sync.WaitGroup
	workers.Add(callers)
	for range callers {
		go func() {
			defer workers.Done()
			<-start
			install, bindErr := binder.bindEncoded(context.Background(), encoded)
			if bindErr == nil {
				_, bindErr = install(context.Background())
			}
			errs <- bindErr
		}()
	}
	close(start)
	workers.Wait()
	close(errs)
	for err := range errs {
		if err != nil {
			t.Fatalf("concurrent immutable bind failed: %v", err)
		}
	}

	if got := counted.count(catalogsnapshot.ArtifactCatalogSnapshot); got != 1 {
		t.Fatalf("Catalog snapshot physical loads = %d, want 1", got)
	}
	if got := counted.count(catalogsnapshot.ArtifactRuntimeMetadataView); got != 2 {
		t.Fatalf("Runtime view physical loads = %d, want one per exact view (2)", got)
	}
	if _, err := binder.bindEncoded(context.Background(), encoded); err != nil {
		t.Fatalf("warm immutable bind failed: %v", err)
	}
	if got := counted.count(catalogsnapshot.ArtifactCatalogSnapshot); got != 1 {
		t.Fatalf("warm Catalog snapshot physical loads = %d, want 1", got)
	}
	if got := counted.count(catalogsnapshot.ArtifactRuntimeMetadataView); got != 2 {
		t.Fatalf("warm Runtime view physical loads = %d, want 2", got)
	}
}

type countingArtifactStore struct {
	catalogsnapshot.Store
	mu   sync.Mutex
	gets map[catalogsnapshot.ArtifactKind]int
}

func (s *countingArtifactStore) Get(
	ctx context.Context,
	kind catalogsnapshot.ArtifactKind,
	digest catalogsnapshot.Digest,
) (catalogsnapshot.Artifact, error) {
	s.mu.Lock()
	s.gets[kind]++
	s.mu.Unlock()
	return s.Store.Get(ctx, kind, digest)
}

func (s *countingArtifactStore) count(kind catalogsnapshot.ArtifactKind) int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.gets[kind]
}

func TestCommandArtifactBinderInstallsExactFlowAndMutationSiblings(t *testing.T) {
	store, commandView, mutationView, flowView := workerRuntimeFlowCommandViews(t)
	scope := persistence.RuntimeWorkScope{
		OrgID: commandView.OrgID(), AppInstallationID: commandView.AppInstallationID(),
		RuntimeCapabilityID: commandView.CapabilityID(),
	}
	lineage := commandView.Lineage()
	stager := &workerAssignmentPolicyEligibilityStagerProbe{view: mutationView.Digest()}
	binder, err := newMutationArtifactBinder(
		store,
		scope,
		catalogsnapshot.Compatibility{
			SchemaGeneration: lineage.SchemaGeneration,
			EngineGeneration: lineage.EngineGeneration,
		},
		func(
			_ context.Context,
			view *catalogsnapshot.RuntimeMetadataView,
		) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error) {
			if view.Digest() != mutationView.Digest() {
				t.Fatalf(
					"Runtime Flow command lifecycle prepared from %s, want mutation sibling %s",
					view.Digest(),
					mutationView.Digest(),
				)
			}
			return stager, nil
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(commandView)
	if err != nil {
		t.Fatal(err)
	}
	encoded, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		t.Fatal(err)
	}
	authority, err := (commandViewBinderAdapter{binder: binder}).OpenCommand(
		t.Context(),
		encoded,
		RuntimeFlowCommandKind,
		RuntimeFlowCommandPayloadFormatVersion,
	)
	if err != nil {
		t.Fatal(err)
	}
	if !authority.AllowsCommand(
		RuntimeFlowCommandKind,
		RuntimeFlowCommandPayloadFormatVersion,
	) {
		t.Fatal("exact command registration was not preserved")
	}
	namespace, allowed := authority.CommandExecutionContext(
		RuntimeFlowCommandKind,
		RuntimeFlowCommandPayloadFormatVersion,
		"integration-a",
		"ExactChild",
		"version-child",
	)
	if !allowed || namespace != "connector-a" {
		t.Fatal("workload-scoped command definition authority was not preserved")
	}
	if _, allowed = authority.CommandExecutionContext(
		RuntimeFlowCommandKind,
		RuntimeFlowCommandPayloadFormatVersion,
		"integration-other",
		"ExactChild",
		"version-child",
	); allowed {
		t.Fatal("workload-scoped command definition authority was not preserved")
	}
	executionCtx, err := authority.Bind(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	boundMutation, ok := metadata.RuntimeMetadataViewFromContext(executionCtx)
	if !ok || boundMutation.Digest() != mutationView.Digest() {
		t.Fatalf("mutation view = %#v, %v", boundMutation, ok)
	}
	boundFlow, ok := metadata.RuntimeFlowWorkViewFromContext(executionCtx)
	if !ok || boundFlow.Digest() != flowView.Digest() {
		t.Fatalf("Flow view = %#v, %v", boundFlow, ok)
	}
	boundStager, ok := managedruntime.AssignmentPolicyEligibilityAmbientStagerFromContext(executionCtx)
	if !ok || boundStager != stager {
		t.Fatalf("Runtime Flow command lifecycle stager = %#v, %v", boundStager, ok)
	}
	v3Authority, err := (commandViewBinderAdapter{binder: binder}).OpenCommand(
		t.Context(),
		encoded,
		RuntimeFlowCommandKind,
		RuntimeFlowCommandPayloadFormatVersionV3,
	)
	if err != nil {
		t.Fatal(err)
	}
	if !v3Authority.AllowsCommand(
		RuntimeFlowCommandKind,
		RuntimeFlowCommandPayloadFormatVersionV3,
	) {
		t.Fatal("exact V3 command registration was not preserved")
	}
	v3ExecutionCtx, err := v3Authority.Bind(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	v3Mutation, ok := metadata.RuntimeMetadataViewFromContext(v3ExecutionCtx)
	if !ok || v3Mutation.Digest() != mutationView.Digest() {
		t.Fatalf("V3 mutation view = %#v, %v", v3Mutation, ok)
	}
	v3Flow, ok := metadata.RuntimeFlowWorkViewFromContext(v3ExecutionCtx)
	if !ok || v3Flow.Digest() != flowView.Digest() {
		t.Fatalf("V3 Flow view = %#v, %v", v3Flow, ok)
	}
	v3Stager, ok := managedruntime.AssignmentPolicyEligibilityAmbientStagerFromContext(v3ExecutionCtx)
	if !ok || v3Stager != stager {
		t.Fatalf("V3 Runtime Flow command lifecycle stager = %#v, %v", v3Stager, ok)
	}

	missingFlowStore, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	for _, artifactIdentity := range []struct {
		kind   catalogsnapshot.ArtifactKind
		digest catalogsnapshot.Digest
	}{
		{catalogsnapshot.ArtifactCatalogSnapshot, binding.CatalogSnapshotDigest},
		{catalogsnapshot.ArtifactRuntimeMetadataView, binding.RuntimeMetadataViewDigest},
		{catalogsnapshot.ArtifactRuntimeMetadataView, mutationView.Digest()},
	} {
		artifact, getErr := store.Get(
			t.Context(),
			artifactIdentity.kind,
			artifactIdentity.digest,
		)
		if getErr != nil {
			t.Fatal(getErr)
		}
		if err := missingFlowStore.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	missingFlowBinder, err := newTestMutationArtifactBinder(
		missingFlowStore,
		scope,
		catalogsnapshot.Compatibility{
			SchemaGeneration: lineage.SchemaGeneration,
			EngineGeneration: lineage.EngineGeneration,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := (commandViewBinderAdapter{
		binder: missingFlowBinder,
	}).OpenCommand(
		t.Context(),
		encoded,
		RuntimeFlowCommandKind,
		RuntimeFlowCommandPayloadFormatVersion,
	); err == nil || isPermanentMetadataBindingError(err) {
		t.Fatalf("missing Flow sibling error = %v, want retryable fail closed", err)
	}
}

func TestCommandArtifactBinderKeepsConnectorActionsInFlowSibling(t *testing.T) {
	store, commandView, _, flowView := workerRuntimeFlowCommandViewsWithMetadata(
		t,
		[]byte(
			`{"runAtomically":true,"startElementReference":"ApplyMappings","actionCalls":[{"name":"ApplyMappings","actionName":"applyConnectorFieldMappings","actionType":"applyConnectorFieldMappings"}]}`,
		),
	)
	scope := persistence.RuntimeWorkScope{
		OrgID: commandView.OrgID(), AppInstallationID: commandView.AppInstallationID(),
		RuntimeCapabilityID: commandView.CapabilityID(),
	}
	lineage := commandView.Lineage()
	binder, err := newTestMutationArtifactBinder(store, scope, catalogsnapshot.Compatibility{
		SchemaGeneration: lineage.SchemaGeneration,
		EngineGeneration: lineage.EngineGeneration,
	})
	if err != nil {
		t.Fatal(err)
	}
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(commandView)
	if err != nil {
		t.Fatal(err)
	}
	encoded, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		t.Fatal(err)
	}
	authority, err := (commandViewBinderAdapter{binder: binder}).OpenCommand(
		t.Context(),
		encoded,
		RuntimeFlowCommandKind,
		RuntimeFlowCommandPayloadFormatVersion,
	)
	if err != nil {
		t.Fatal(err)
	}
	executionCtx, err := authority.Bind(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	boundFlow, ok := metadata.RuntimeFlowWorkViewFromContext(executionCtx)
	if !ok || boundFlow.Digest() != flowView.Digest() {
		t.Fatalf("connector Flow view = %#v, %v", boundFlow, ok)
	}
	version, ok := boundFlow.ActiveFlowVersion("ExactChild")
	if !ok || version == nil || len(version.Metadata.ActionCalls) != 1 ||
		version.Metadata.ActionCalls[0].ActionType !=
			"applyConnectorFieldMappings" {
		t.Fatalf("connector Flow closure = %#v, %v", version, ok)
	}
}

func workerRuntimeFlowCommandViews(t testing.TB) (
	catalogsnapshot.Store,
	*catalogsnapshot.RuntimeMetadataView,
	*catalogsnapshot.RuntimeMetadataView,
	*catalogsnapshot.RuntimeMetadataView,
) {
	return workerRuntimeFlowCommandViewsWithMetadata(
		t,
		[]byte(
			`{"runAtomically":true,"startElementReference":"Assign","assignments":[{"name":"Assign"}]}`,
		),
	)
}

func workerRuntimeFlowCommandViewsWithMetadata(
	t testing.TB,
	metadataRaw []byte,
) (
	catalogsnapshot.Store,
	*catalogsnapshot.RuntimeMetadataView,
	*catalogsnapshot.RuntimeMetadataView,
	*catalogsnapshot.RuntimeMetadataView,
) {
	return workerRuntimeFlowCommandViewsWithMetadataClosure(
		t,
		metadataRaw,
		"",
		nil,
		nil,
	)
}

func workerRuntimeFlowCommandViewsWithMetadataClosure(
	t testing.TB,
	metadataRaw []byte,
	flowOwnerPack string,
	extraFields []models.FieldMetadata,
	extraMutationDefinitions []catalogsnapshot.MutationDefinitionInput,
) (
	catalogsnapshot.Store,
	*catalogsnapshot.RuntimeMetadataView,
	*catalogsnapshot.RuntimeMetadataView,
	*catalogsnapshot.RuntimeMetadataView,
) {
	t.Helper()
	revision := catalogsnapshot.CatalogRevision(11)
	commandComponentID := "component.command.runtime-flow"
	commandV2ComponentID := "component.command.runtime-flow-v2"
	commandV3ComponentID := "component.command.runtime-flow-v3"
	workKeyPrefix := "a01"
	retriableMetadataRaw := []byte(`{
  "runAtomically": true,
  "startElementReference": "Retry",
  "actionCalls": [{
    "name": "Retry",
    "actionName": "setRetriable",
    "actionType": "setRetriable"
  }]
}`)
	dependencyMetadataRaw := []byte(`{
  "runAtomically": true,
  "startElementReference": "WaitForIdentity",
  "actionCalls": [{
    "name": "WaitForIdentity",
    "actionName": "setIdentityConvergenceDependency",
    "actionType": "setIdentityConvergenceDependency",
    "inputParameters": [
      {"name":"objectApiName","value":{"stringValue":"Lead"}},
      {"name":"recordId","value":{"elementReference":"targetId"}},
      {"name":"destinationSlotId","value":{"stringValue":"destination-1"}}
    ]
  }]
}`)
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(
		catalogsnapshot.CompileInput{
			OrgID: "org-test",
			Lineage: catalogsnapshot.Lineage{
				CatalogRevision:  revision,
				SchemaGeneration: 1,
				EngineGeneration: 1,
			},
			Objects: []catalogsnapshot.ObjectInput{{
				CatalogRevision: revision,
				Metadata: &models.ObjectMetadata{
					ID: "object-work", APIName: "Work",
					Label: "Work", PluralLabel: "Work", KeyPrefix: &workKeyPrefix,
					SharingModel: constants.SharingModelPublicReadWrite,
					Fields: append([]models.FieldMetadata{{
						ID: "field-work-id", APIName: constants.FieldID,
						Label: "ID", Type: constants.FieldTypeString,
						PhysicalName: constants.FieldID, IsSystem: true,
					}}, extraFields...),
				},
			}},
			CommandRegistrations: []catalogsnapshot.CommandRegistrationInput{
				{
					CatalogRevision:   revision,
					ComponentID:       commandComponentID,
					DefinitionVersion: "1",
					ContentDigest: catalogsnapshot.ComputeDigest(
						[]byte(RuntimeFlowCommandKind + "/v1"),
					),
					CommandKind:          RuntimeFlowCommandKind,
					PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersion,
					ExecutionBindings: []catalogsnapshot.CommandExecutionBinding{{
						AuthorizedWorkloadID: "integration-a",
						DefinitionRef:        "ExactChild",
						DefinitionVersion:    "version-child",
						ContextNamespace:     "connector-a",
					}, {
						AuthorizedWorkloadID: "integration-a",
						DefinitionRef:        "RetryChild",
						DefinitionVersion:    "version-retry-child",
						ContextNamespace:     "connector-a",
					}},
				},
				{
					CatalogRevision:   revision,
					ComponentID:       commandV2ComponentID,
					DefinitionVersion: "2",
					ContentDigest: catalogsnapshot.ComputeDigest(
						[]byte(RuntimeFlowCommandKind + "/v2"),
					),
					CommandKind:          RuntimeFlowCommandKind,
					PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersionV2,
					ExecutionBindings: []catalogsnapshot.CommandExecutionBinding{{
						AuthorizedWorkloadID: "integration-a",
						DefinitionRef:        "DependencyChild",
						DefinitionVersion:    "version-dependency-child",
						ContextNamespace:     "connector-a",
					}, {
						AuthorizedWorkloadID: "integration-a",
						DefinitionRef:        "ExactChild",
						DefinitionVersion:    "version-child",
						ContextNamespace:     "connector-a",
					}},
				},
				{
					CatalogRevision:   revision,
					ComponentID:       commandV3ComponentID,
					DefinitionVersion: "3",
					ContentDigest: catalogsnapshot.ComputeDigest(
						[]byte(RuntimeFlowCommandKind + "/v3"),
					),
					CommandKind:          RuntimeFlowCommandKind,
					PayloadFormatVersion: RuntimeFlowCommandPayloadFormatVersionV3,
					ExecutionBindings: []catalogsnapshot.CommandExecutionBinding{{
						AuthorizedWorkloadID: "integration-a",
						DefinitionRef:        "ExactChild",
						DefinitionVersion:    "version-child",
						ContextNamespace:     "connector-a",
					}, {
						AuthorizedWorkloadID: "integration-a",
						DefinitionRef:        "RetryChild",
						DefinitionVersion:    "version-retry-child",
						ContextNamespace:     "connector-a",
					}},
				},
			},
			MutationDefinitions: append([]catalogsnapshot.MutationDefinitionInput{
				{
					CatalogRevision: revision,
					FlowDefinition: &models.FlowDefinition{
						Id: "definition-dependency-child", DeveloperName: "DependencyChild",
						ProcessType:     "AutoLaunchedFlow",
						ActiveVersionId: "version-dependency-child",
					},
				},
				{
					CatalogRevision: revision,
					FlowVersion: &models.FlowVersion{
						Id: "version-dependency-child", DefinitionId: "definition-dependency-child",
						VersionNumber: 1, Status: models.FlowStatusActive,
						ProcessType: "AutoLaunchedFlow", Source: "standard",
						DefinitionOwnerPack: flowOwnerPack,
						MetadataRaw:         dependencyMetadataRaw,
					},
				},
				{
					CatalogRevision: revision,
					FlowDefinition: &models.FlowDefinition{
						Id: "definition-child", DeveloperName: "ExactChild",
						ProcessType:     "AutoLaunchedFlow",
						ActiveVersionId: "version-child",
					},
				},
				{
					CatalogRevision: revision,
					FlowVersion: &models.FlowVersion{
						Id: "version-child", DefinitionId: "definition-child",
						VersionNumber: 1, Status: models.FlowStatusActive,
						ProcessType: "AutoLaunchedFlow", Source: "standard",
						DefinitionOwnerPack: flowOwnerPack,
						MetadataRaw:         append([]byte(nil), metadataRaw...),
					},
				},
				{
					CatalogRevision: revision,
					FlowDefinition: &models.FlowDefinition{
						Id: "definition-retry-child", DeveloperName: "RetryChild",
						ProcessType:     "AutoLaunchedFlow",
						ActiveVersionId: "version-retry-child",
					},
				},
				{
					CatalogRevision: revision,
					FlowVersion: &models.FlowVersion{
						Id: "version-retry-child", DefinitionId: "definition-retry-child",
						VersionNumber: 1, Status: models.FlowStatusActive,
						ProcessType: "AutoLaunchedFlow", Source: "standard",
						DefinitionOwnerPack: flowOwnerPack,
						MetadataRaw:         retriableMetadataRaw,
					},
				},
			}, extraMutationDefinitions...),
			Projections: []catalogsnapshot.RuntimeProjectionInput{
				{
					CatalogRevision:        revision,
					AppInstallationID:      "installation-test",
					CapabilityID:           commandsemantic.CapabilityName,
					Profile:                catalogsnapshot.ProjectionCommandAcceptanceV1,
					DefinitionComponentIDs: []string{commandComponentID, commandV2ComponentID, commandV3ComponentID},
				},
				{
					CatalogRevision:   revision,
					AppInstallationID: "installation-test",
					CapabilityID:      mutationsemantic.CapabilityName,
					Profile:           catalogsnapshot.ProjectionRecordMutationV1,
					ObjectAPINames:    []string{"Work"},
				},
				{
					CatalogRevision:   revision,
					AppInstallationID: "installation-test",
					CapabilityID:      flowsemantic.CapabilityName,
					Profile:           catalogsnapshot.ProjectionRecordMutationV1,
					ObjectAPINames:    []string{"Work"},
				},
			},
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	store, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	if err := store.Put(t.Context(), snapshotArtifact); err != nil {
		t.Fatal(err)
	}
	views := make([]*catalogsnapshot.RuntimeMetadataView, 0, 3)
	for _, capabilityID := range []string{
		commandsemantic.CapabilityName,
		mutationsemantic.CapabilityName,
		flowsemantic.CapabilityName,
	} {
		artifact, projectErr := catalogsnapshot.ProjectRuntimeMetadataView(
			snapshot,
			catalogsnapshot.ProjectionKey{
				AppInstallationID: "installation-test",
				CapabilityID:      capabilityID,
			},
		)
		if projectErr != nil {
			t.Fatal(projectErr)
		}
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
		view, openErr := catalogsnapshot.OpenRuntimeMetadataView(artifact)
		if openErr != nil {
			t.Fatal(openErr)
		}
		views = append(views, view)
	}
	return store, views[0], views[1], views[2]
}

func TestMutationArtifactBinderInstallsExactFlowWorkViewWithoutSelectingCurrent(t *testing.T) {
	store, mutationView, flowView := workerMutationAndFlowViews(t)
	scope := persistence.RuntimeWorkScope{
		OrgID: mutationView.OrgID(), AppInstallationID: mutationView.AppInstallationID(),
		RuntimeCapabilityID: mutationView.CapabilityID(),
	}
	lineage := mutationView.Lineage()
	binder, err := newTestMutationArtifactBinder(store, scope, catalogsnapshot.Compatibility{
		SchemaGeneration: lineage.SchemaGeneration, EngineGeneration: lineage.EngineGeneration,
	})
	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)
	}
	encoded, err := catalogsnapshot.EncodeRuntimeWorkBindingSet(catalogsnapshot.RuntimeWorkBindingSet{
		Primary: mutationBinding, FlowWork: &flowBinding,
	})
	if err != nil {
		t.Fatal(err)
	}
	install, err := binder.bindEncoded(t.Context(), encoded)
	if err != nil {
		t.Fatal(err)
	}
	executionCtx, err := install(t.Context())
	if err != nil {
		t.Fatal(err)
	}
	boundMutation, ok := metadata.RuntimeMetadataViewFromContext(executionCtx)
	if !ok || boundMutation.Digest() != mutationView.Digest() {
		t.Fatalf("mutation view = %#v, %v", boundMutation, ok)
	}
	boundFlow, ok := metadata.RuntimeFlowWorkViewFromContext(executionCtx)
	if !ok || boundFlow.Digest() != flowView.Digest() {
		t.Fatalf("Flow work view = %#v, %v", boundFlow, ok)
	}

	wrongCapability := flowBinding
	wrongCapability.RuntimeCapabilityID = "runtime.other.v1"
	wrongEncoded, err := catalogsnapshot.EncodeRuntimeWorkBindingSet(catalogsnapshot.RuntimeWorkBindingSet{
		Primary: mutationBinding, FlowWork: &wrongCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := binder.bindEncoded(t.Context(), wrongEncoded); !isPermanentMetadataBindingError(err) {
		t.Fatalf("wrong Flow capability bind error = %v, want permanent fail closed", err)
	}
}

func TestScheduledFlowArtifactBinderPreparesOnlyExactImmutableClosure(t *testing.T) {
	store, mutationView, flowView := workerScheduledFlowView(t)
	scope := persistence.RuntimeWorkScope{
		OrgID: flowView.OrgID(), AppInstallationID: flowView.AppInstallationID(),
		RuntimeCapabilityID: flowView.CapabilityID(),
	}
	lineage := flowView.Lineage()
	stager := &workerAssignmentPolicyEligibilityStagerProbe{view: mutationView.Digest()}
	binder, err := newMutationArtifactBinder(
		store,
		scope,
		catalogsnapshot.Compatibility{
			SchemaGeneration: lineage.SchemaGeneration,
			EngineGeneration: lineage.EngineGeneration,
		},
		func(
			_ context.Context,
			view *catalogsnapshot.RuntimeMetadataView,
		) (managedruntime.AssignmentPolicyEligibilityAmbientStager, error) {
			if view.Digest() != mutationView.Digest() {
				t.Fatalf(
					"scheduled Flow lifecycle prepared from %s, want mutation sibling %s",
					view.Digest(),
					mutationView.Digest(),
				)
			}
			return stager, nil
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	binding, err := catalogsnapshot.RuntimeWorkBindingFromView(flowView)
	if err != nil {
		t.Fatal(err)
	}
	encoded, err := catalogsnapshot.EncodeRuntimeWorkBinding(binding)
	if err != nil {
		t.Fatal(err)
	}
	prepared, err := binder.bindScheduledFlow(
		t.Context(),
		encoded,
		"definition-scheduled",
		"version-scheduled",
		"",
	)
	if err != nil {
		t.Fatal(err)
	}
	if err := prepared.Validate(); err != nil {
		t.Fatal(err)
	}
	plan := prepared.ScheduledFlowExecutionPlan
	if plan.Definition.DeveloperName != "ScheduledRoot" ||
		plan.Root.Id != "version-scheduled" ||
		len(plan.Versions) != 2 ||
		plan.Versions["exactchild"].Id != "version-child" {
		t.Fatalf("scheduled plan = %#v", plan)
	}
	_, startupView := workerMutationView(t)
	executionCtx, err := prepared.InstallExecution(
		metadata.WithRuntimeMetadataView(t.Context(), startupView),
	)
	if err != nil {
		t.Fatal(err)
	}
	boundView, ok := metadata.RuntimeMetadataViewFromContext(executionCtx)
	if !ok || boundView.Digest() != mutationView.Digest() {
		t.Fatalf(
			"scheduled mutation view = %#v, %v; want same-revision mutation view %s",
			boundView,
			ok,
			mutationView.Digest(),
		)
	}
	boundFlowView, ok := metadata.RuntimeFlowWorkViewFromContext(executionCtx)
	if !ok || boundFlowView.Digest() != flowView.Digest() {
		t.Fatalf(
			"scheduled Flow view = %#v, %v; want durable-row Flow view %s",
			boundFlowView,
			ok,
			flowView.Digest(),
		)
	}
	boundStager, ok := managedruntime.AssignmentPolicyEligibilityAmbientStagerFromContext(executionCtx)
	if !ok || boundStager != stager {
		t.Fatalf("scheduled Flow lifecycle stager = %#v, %v", boundStager, ok)
	}
	if automation.AfterCommitScheduledFlowExecutionAuthorizesVersion(executionCtx, plan.Root.Id) {
		t.Fatal("ordinary scheduled Flow execution received AsyncAfterCommit authority")
	}
	asyncPlan := plan
	asyncPlan.AsyncAfterCommit = true
	mutationScope := scope
	mutationScope.RuntimeCapabilityID = mutationView.CapabilityID()
	asyncExecutionCtx, err := installScheduledFlowExecution(
		t.Context(), asyncPlan, mutationView, flowView, mutationScope, nil,
	)
	if err != nil {
		t.Fatal(err)
	}
	if !automation.AfterCommitScheduledFlowExecutionAuthorizesVersion(asyncExecutionCtx, plan.Root.Id) ||
		automation.AfterCommitScheduledFlowExecutionAuthorizesVersion(asyncExecutionCtx, "other-version") {
		t.Fatal("scheduled execution marker did not bind exactly the prepared root FlowVersion")
	}

	missingSiblingStore, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	for _, artifactIdentity := range []struct {
		kind   catalogsnapshot.ArtifactKind
		digest catalogsnapshot.Digest
	}{
		{kind: catalogsnapshot.ArtifactCatalogSnapshot, digest: binding.CatalogSnapshotDigest},
		{kind: catalogsnapshot.ArtifactRuntimeMetadataView, digest: binding.RuntimeMetadataViewDigest},
	} {
		artifact, getErr := store.Get(t.Context(), artifactIdentity.kind, artifactIdentity.digest)
		if getErr != nil {
			t.Fatal(getErr)
		}
		if err := missingSiblingStore.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	missingSiblingBinder, err := newTestMutationArtifactBinder(
		missingSiblingStore,
		scope,
		catalogsnapshot.Compatibility{
			SchemaGeneration: lineage.SchemaGeneration,
			EngineGeneration: lineage.EngineGeneration,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := missingSiblingBinder.bindScheduledFlow(
		t.Context(),
		encoded,
		"definition-scheduled",
		"version-scheduled",
		"",
	); err == nil || isPermanentMetadataBindingError(err) {
		t.Fatalf("missing published mutation view error = %v, want retryable fail closed", err)
	}

	outside := binding
	outside.AppInstallationID = "installation-other"
	outsideEncoded, err := catalogsnapshot.EncodeRuntimeWorkBinding(outside)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := binder.bindScheduledFlow(
		t.Context(),
		outsideEncoded,
		"definition-scheduled",
		"version-scheduled",
		"",
	); !isPermanentMetadataBindingError(err) {
		t.Fatalf("cross-installation bind error = %v, want permanent fail-closed authority", err)
	}
}

func workerMutationView(t *testing.T) (catalogsnapshot.Store, *catalogsnapshot.RuntimeMetadataView) {
	t.Helper()
	snapshotArtifact, viewArtifact, view, _, _ := compileWorkerMutationArtifacts(t)
	store, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	if err := store.Put(context.Background(), snapshotArtifact); err != nil {
		t.Fatal(err)
	}
	if err := store.Put(context.Background(), viewArtifact); err != nil {
		t.Fatal(err)
	}
	return store, view
}

func workerMutationAndFlowViews(t *testing.T) (catalogsnapshot.Store, *catalogsnapshot.RuntimeMetadataView, *catalogsnapshot.RuntimeMetadataView) {
	t.Helper()
	snapshotArtifact, mutationArtifact, mutationView, flowArtifact, flowView := compileWorkerMutationArtifacts(t)
	store, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	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)
		}
	}
	return store, mutationView, flowView
}

func workerScheduledFlowView(t testing.TB, extraFields ...models.FieldMetadata) (
	catalogsnapshot.Store,
	*catalogsnapshot.RuntimeMetadataView,
	*catalogsnapshot.RuntimeMetadataView,
) {
	t.Helper()
	revision := catalogsnapshot.CatalogRevision(9)
	fields := append([]models.FieldMetadata{{
		ID: "field-work-id", APIName: constants.FieldID, Label: "ID",
		Type: constants.FieldTypeString, PhysicalName: "id", IsSystem: true,
	}}, extraFields...)
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(catalogsnapshot.CompileInput{
		OrgID: "org-test",
		Lineage: catalogsnapshot.Lineage{
			CatalogRevision: revision, SchemaGeneration: 1, EngineGeneration: 1,
		},
		Objects: []catalogsnapshot.ObjectInput{{
			CatalogRevision: revision,
			Metadata: &models.ObjectMetadata{
				ID: "object-work", APIName: "Work", Label: "Work", PluralLabel: "Work",
				SharingModel: constants.SharingModelPublicReadWrite,
				Fields:       fields,
			},
		}},
		MutationDefinitions: []catalogsnapshot.MutationDefinitionInput{
			{
				CatalogRevision: revision,
				FlowDefinition: &models.FlowDefinition{
					Id: "definition-scheduled", DeveloperName: "ScheduledRoot",
					ProcessType: "AutoLaunchedFlow", ActiveVersionId: "version-scheduled",
				},
			},
			{
				CatalogRevision: revision,
				FlowVersion: &models.FlowVersion{
					Id: "version-scheduled", DefinitionId: "definition-scheduled",
					VersionNumber: 1, Status: models.FlowStatusActive,
					ProcessType: "AutoLaunchedFlow",
					MetadataRaw: []byte(`{"runAtomically":true,"start":{"schedule":{"frequency":"Daily"},"connector":{"targetReference":"CallChild"}},"subflows":[{"name":"CallChild","flowName":"ExactChild"}]}`),
				},
			},
			{
				CatalogRevision: revision,
				FlowDefinition: &models.FlowDefinition{
					Id: "definition-child", DeveloperName: "ExactChild",
					ProcessType: "AutoLaunchedFlow", ActiveVersionId: "version-child",
					OwnerPack: "connector-a",
				},
			},
			{
				CatalogRevision: revision,
				FlowVersion: &models.FlowVersion{
					Id: "version-child", DefinitionId: "definition-child",
					VersionNumber: 1, Status: models.FlowStatusActive,
					ProcessType:         "AutoLaunchedFlow",
					DefinitionOwnerPack: "connector-a",
					MetadataRaw:         []byte(`{"runAtomically":true,"startElementReference":"Assign","assignments":[{"name":"Assign"}]}`),
				},
			},
		},
		Projections: []catalogsnapshot.RuntimeProjectionInput{
			{
				CatalogRevision: revision, AppInstallationID: "installation-test",
				CapabilityID: mutationsemantic.CapabilityName,
				Profile:      catalogsnapshot.ProjectionRecordMutationV1,
				ObjectAPINames: []string{
					"Work",
				},
			},
			{
				CatalogRevision: revision, AppInstallationID: "installation-test",
				CapabilityID: flowsemantic.CapabilityName,
				Profile:      catalogsnapshot.ProjectionRecordMutationV1,
				ObjectAPINames: []string{
					"Work",
				},
			},
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	mutationArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(
		snapshot,
		catalogsnapshot.ProjectionKey{
			AppInstallationID: "installation-test",
			CapabilityID:      mutationsemantic.CapabilityName,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	mutationView, err := catalogsnapshot.OpenRuntimeMetadataView(mutationArtifact)
	if err != nil {
		t.Fatal(err)
	}
	flowArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(
		snapshot,
		catalogsnapshot.ProjectionKey{
			AppInstallationID: "installation-test",
			CapabilityID:      flowsemantic.CapabilityName,
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	flowView, err := catalogsnapshot.OpenRuntimeMetadataView(flowArtifact)
	if err != nil {
		t.Fatal(err)
	}
	store, err := catalogsnapshot.NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	if err := store.Put(t.Context(), snapshotArtifact); err != nil {
		t.Fatal(err)
	}
	for _, artifact := range []catalogsnapshot.Artifact{mutationArtifact, flowArtifact} {
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatal(err)
		}
	}
	return store, mutationView, flowView
}

func compileWorkerMutationArtifacts(t *testing.T) (
	catalogsnapshot.Artifact,
	catalogsnapshot.Artifact,
	*catalogsnapshot.RuntimeMetadataView,
	catalogsnapshot.Artifact,
	*catalogsnapshot.RuntimeMetadataView,
) {
	t.Helper()
	return compileWorkerMutationArtifactsForRevision(
		t,
		catalogsnapshot.CatalogRevision(7),
	)
}

func compileWorkerMutationArtifactsForRevision(
	t *testing.T,
	revision catalogsnapshot.CatalogRevision,
) (
	catalogsnapshot.Artifact,
	catalogsnapshot.Artifact,
	*catalogsnapshot.RuntimeMetadataView,
	catalogsnapshot.Artifact,
	*catalogsnapshot.RuntimeMetadataView,
) {
	t.Helper()
	resultType := constants.SchemaFieldType("integer")
	childKeyPrefix := "a01"
	scheduledTargetKeyPrefix := "a02"
	discoveryOccurrence := time.Now().UTC().Add(-10 * time.Second)
	input := catalogsnapshot.CompileInput{
		OrgID: "org-test",
		Lineage: catalogsnapshot.Lineage{
			CatalogRevision: revision, SchemaGeneration: 1, EngineGeneration: 1,
		},
		Objects: []catalogsnapshot.ObjectInput{
			{CatalogRevision: revision, Metadata: &models.ObjectMetadata{
				ID: "object-parent", APIName: "Parent", SharingModel: constants.SharingModelPublicReadWrite,
				Fields: []models.FieldMetadata{
					{ID: "field-parent-id", APIName: constants.FieldID, PhysicalName: constants.FieldID, Type: constants.FieldTypeString, IsSystem: true},
					{ID: "field-parent-count", APIName: "ChildCount", PhysicalName: "ChildCount", Type: constants.FieldTypeRollupSummary,
						RollupConfig: &models.RollupConfig{SummaryObject: "Child", RelationshipField: "ParentId", CalcType: "COUNT", MaintenanceMode: "eventual", ResultType: &resultType}},
					{ID: "field-parent-modified-by", APIName: constants.FieldLastModifiedByID, PhysicalName: constants.FieldLastModifiedByID, Type: constants.FieldTypeReference, IsSystem: true},
					{ID: "field-parent-modified", APIName: constants.FieldLastModifiedDate, PhysicalName: constants.FieldLastModifiedDate, Type: constants.FieldTypeDatetime, IsSystem: true},
				},
			}},
			{CatalogRevision: revision, Metadata: &models.ObjectMetadata{
				ID: "object-child", APIName: "Child", KeyPrefix: &childKeyPrefix, SharingModel: constants.SharingModelPublicReadWrite,
				Fields: []models.FieldMetadata{
					{ID: "field-child-id", APIName: constants.FieldID, PhysicalName: constants.FieldID, Type: constants.FieldTypeString, IsSystem: true},
					{ID: "field-child-parent", APIName: "ParentId", PhysicalName: "ParentId", Type: constants.FieldTypeReference, ReferenceTo: []string{"Parent"}},
					{ID: "field-child-deleted", APIName: constants.FieldIsDeleted, PhysicalName: constants.FieldIsDeleted, Type: constants.FieldTypeBoolean, IsSystem: true},
				},
			}},
			{CatalogRevision: revision, Metadata: &models.ObjectMetadata{
				ID: "object-scheduled-target", APIName: "ScheduledTarget", KeyPrefix: &scheduledTargetKeyPrefix,
				SharingModel: constants.SharingModelPublicReadWrite,
				Fields: []models.FieldMetadata{
					{ID: "field-scheduled-target-id", APIName: constants.FieldID, PhysicalName: constants.FieldID, Type: constants.FieldTypeString, IsSystem: true},
					{
						ID: "field-scheduled-target-status", APIName: "Status", PhysicalName: "Status",
						Type: constants.FieldTypeString, Createable: true, Updateable: true,
					},
				},
			}},
		},
		MutationDefinitions: []catalogsnapshot.MutationDefinitionInput{
			{CatalogRevision: revision, FlowDefinition: &models.FlowDefinition{
				Id: "definition-scheduled-integration", DeveloperName: "ScheduledIntegration",
				ProcessType: "AutoLaunchedFlow", TriggerType: "Scheduled",
				ActiveVersionId: "version-scheduled-integration",
			}},
			{CatalogRevision: revision, FlowVersion: &models.FlowVersion{
				Id: "version-scheduled-integration", DefinitionId: "definition-scheduled-integration",
				VersionNumber: 1, Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
				MetadataRaw: []byte(`{"start":{"object":"ScheduledTarget","schedule":{"frequency":"Daily","startDate":"2099-01-01","startTime":"12:00:00","timezoneSidKey":"UTC"},"connector":{"targetReference":"UpdateScheduled"}},"recordUpdates":[{"name":"UpdateScheduled","object":"ScheduledTarget","inputReference":"$Record","inputAssignments":[{"field":"Status","value":{"stringValue":"scheduled-complete"}}]}]}`),
			}},
			{CatalogRevision: revision, FlowDefinition: &models.FlowDefinition{
				Id: "definition-scheduled-rollback", DeveloperName: "ScheduledRollback",
				ProcessType: "AutoLaunchedFlow", TriggerType: "Scheduled",
				ActiveVersionId: "version-scheduled-rollback",
			}},
			{CatalogRevision: revision, FlowVersion: &models.FlowVersion{
				Id: "version-scheduled-rollback", DefinitionId: "definition-scheduled-rollback",
				VersionNumber: 1, Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
				MetadataRaw: []byte(fmt.Sprintf(
					`{"start":{"object":"ScheduledTarget","schedule":{"frequency":"Daily","startDate":%q,"startTime":%q,"timezoneSidKey":"UTC"},"connector":{"targetReference":"UpdateBeforeFailure"}},"recordUpdates":[{"name":"UpdateBeforeFailure","object":"ScheduledTarget","inputReference":"$Record","inputAssignments":[{"field":"Status","value":{"stringValue":"must-roll-back"}}],"connector":{"targetReference":"FailAfterUpdate"}}],"customErrors":[{"name":"FailAfterUpdate","customErrorMessages":[{"errorMessage":"expected scheduled rollback","statusCode":"FLOW_TEST_ROLLBACK"}]}]}`,
					discoveryOccurrence.Format("2006-01-02"),
					discoveryOccurrence.Format("15:04:05"),
				)),
			}},
			{CatalogRevision: revision, TriggerSubscription: &catalogsnapshot.TriggerSubscription{
				ID: "trigger-timed-integration", KeyPrefix: "a02", EventType: "RecordAfterSave",
				RecordTriggerType: "Update", FlowDefinitionID: "definition-timed-integration",
				FlowVersionID: "version-timed-integration", IsActive: true,
			}},
			{CatalogRevision: revision, FlowDefinition: &models.FlowDefinition{
				Id: "definition-timed-integration", DeveloperName: "TimedIntegration",
				ProcessType: "AutoLaunchedFlow", TriggerType: "RecordAfterSave",
				TriggerObjectOrEvent: "ScheduledTarget", ActiveVersionId: "version-timed-integration",
			}},
			{CatalogRevision: revision, FlowVersion: &models.FlowVersion{
				Id: "version-timed-integration", DefinitionId: "definition-timed-integration",
				VersionNumber: 1, Status: models.FlowStatusActive, ProcessType: "AutoLaunchedFlow",
				MetadataRaw: []byte(`{"start":{"object":"ScheduledTarget","triggerType":"RecordAfterSave","recordTriggerType":"CreateAndUpdate","scheduledPaths":[{"name":"OneMinuteLater","timeSource":"RecordTriggerEvent","offsetNumber":1,"offsetUnit":"Minutes","connector":{"targetReference":"UpdateTimed"}}]},"recordUpdates":[{"name":"UpdateTimed","object":"ScheduledTarget","inputReference":"$Record","inputAssignments":[{"field":"Status","value":{"stringValue":"timed-complete"}}]}]}`),
			}},
		},
		Projections: []catalogsnapshot.RuntimeProjectionInput{
			{
				CatalogRevision: revision, AppInstallationID: integrationAppInstallationID,
				CapabilityID: "runtime.records.mutate.v1", Profile: catalogsnapshot.ProjectionRecordMutationV1,
				ObjectAPINames: []string{"Child", "Parent", "ScheduledTarget"},
			},
			{
				CatalogRevision: revision, AppInstallationID: integrationAppInstallationID,
				CapabilityID: flowsemantic.CapabilityName, Profile: catalogsnapshot.ProjectionRecordMutationV1,
				ObjectAPINames: []string{"Child", "Parent", "ScheduledTarget"},
			},
			{
				CatalogRevision: revision, AppInstallationID: integrationAppInstallationID,
				CapabilityID:   catalogsemantic.AppMetadataV2CapabilityName,
				Profile:        catalogsnapshot.ProjectionRecordQueryV1,
				ObjectAPINames: []string{"Child", "Parent", "ScheduledTarget"},
			},
		},
	}
	snapshotArtifact, err := catalogsnapshot.CompileCatalogSnapshot(input)
	if err != nil {
		t.Fatal(err)
	}
	snapshot, err := catalogsnapshot.OpenCatalogSnapshot(snapshotArtifact)
	if err != nil {
		t.Fatal(err)
	}
	viewArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
		AppInstallationID: integrationAppInstallationID, CapabilityID: "runtime.records.mutate.v1",
	})
	if err != nil {
		t.Fatal(err)
	}
	view, err := catalogsnapshot.OpenRuntimeMetadataView(viewArtifact)
	if err != nil {
		t.Fatal(err)
	}
	flowArtifact, err := catalogsnapshot.ProjectRuntimeMetadataView(snapshot, catalogsnapshot.ProjectionKey{
		AppInstallationID: integrationAppInstallationID, CapabilityID: flowsemantic.CapabilityName,
	})
	if err != nil {
		t.Fatal(err)
	}
	flowView, err := catalogsnapshot.OpenRuntimeMetadataView(flowArtifact)
	if err != nil {
		t.Fatal(err)
	}
	return snapshotArtifact, viewArtifact, view, flowArtifact, flowView
}

func mustRuntimeWorkScopeDigest(t *testing.T, scope persistence.RuntimeWorkScope) string {
	t.Helper()
	digest, err := scope.Digest()
	if err != nil {
		t.Fatal(err)
	}
	return digest
}