package automation

import (
	"context"
	"errors"
	"fmt"
	"strings"
	"sync/atomic"
	"testing"
	"time"

	"github.com/nexuscrm/metadata"
	"github.com/nexuscrm/metadata/catalogsnapshot"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

type schedulerWorkerRoleStoreStub struct{}

func TestSchedulerWorkerDiscoveryDoesNotAdvanceWithoutBoundCatalogAuthority(t *testing.T) {
	service, err := metadata.NewRuntimeMetadataService(
		func(context.Context) (*catalogsnapshot.RuntimeMetadataView, bool) {
			return nil, false
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	scheduler := NewSchedulerService(nil, nil, service, nil)
	if complete := scheduler.dispatchScheduledFlowsBetweenWithProgress(
		time.Now().UTC().Add(-time.Minute),
		time.Now().UTC(),
		make(map[string]struct{}),
	); complete {
		t.Fatal("scheduled discovery advanced without exact Catalog authority")
	}
}

func (schedulerWorkerRoleStoreStub) MetadataBinding() string { return "binding" }

func (schedulerWorkerRoleStoreStub) Claim(context.Context, ScheduledRunReservationKey, []byte) (ScheduledRunLease, error) {
	return ScheduledRunLease{Status: ScheduledRunLeaseDuplicate}, nil
}

func (schedulerWorkerRoleStoreStub) ClaimBatch(_ context.Context, acceptances []ScheduledRunAcceptance) (ScheduledRunBatchClaim, error) {
	leases := make([]ScheduledRunLease, len(acceptances))
	for i := range leases {
		leases[i].Status = ScheduledRunLeaseDuplicate
	}
	return ScheduledRunBatchClaim{Leases: leases}, nil
}

func (schedulerWorkerRoleStoreStub) ListRecoverable(context.Context, int) ([]ScheduledRunCandidate, error) {
	return nil, nil
}

func (schedulerWorkerRoleStoreStub) Reclaim(context.Context, ScheduledRunReservationKey) (ScheduledRunLease, error) {
	return ScheduledRunLease{Status: ScheduledRunLeaseDuplicate}, nil
}

func (schedulerWorkerRoleStoreStub) RefreshLease(
	_ context.Context,
	lease ScheduledRunLease,
) (ScheduledRunLease, error) {
	return lease, nil
}

func (schedulerWorkerRoleStoreStub) RecordPreparationFailure(
	context.Context,
	ScheduledRunLease,
	ScheduledRunPreparationFailureClass,
	string,
) error {
	return nil
}

func (schedulerWorkerRoleStoreStub) Execute(ctx context.Context, _ ScheduledRunLease, effect func(context.Context) error) error {
	return effect(ctx)
}

func (schedulerWorkerRoleStoreStub) Complete(context.Context, ScheduledRunLease, ScheduledRunExecutionStatus, string) error {
	return nil
}

type schedulerLeaseLossStoreStub struct {
	executeErr    error
	executeCalls  atomic.Int64
	completeCalls atomic.Int64
}

type schedulerAcceptanceStoreStub struct {
	claimEntered  chan struct{}
	allowClaim    chan struct{}
	completeCalls atomic.Int64
	completeErr   error
}

type schedulerRecoveryAuthorityRequest struct {
	binding    string
	definition string
	version    string
	entry      string
}

type schedulerRecoveryAuthorityStub struct {
	plan             ScheduledFlowExecutionPlan
	installExecution ScheduledFlowExecutionInstaller
	requests         chan schedulerRecoveryAuthorityRequest
	err              error
	permanent        bool
}

func (s schedulerRecoveryAuthorityStub) PrepareScheduledFlow(
	_ context.Context,
	binding string,
	definition string,
	version string,
	entry string,
) (PreparedScheduledFlowExecution, error) {
	s.requests <- schedulerRecoveryAuthorityRequest{
		binding: binding, definition: definition, version: version, entry: entry,
	}
	if s.err != nil {
		return PreparedScheduledFlowExecution{}, s.err
	}
	installExecution := s.installExecution
	if installExecution == nil {
		installExecution = func(ctx context.Context) (context.Context, error) {
			return ctx, nil
		}
	}
	return NewPreparedScheduledFlowExecution(s.plan, installExecution)
}

func (s schedulerRecoveryAuthorityStub) IsPermanentScheduledFlowAuthorityFailure(error) bool {
	return s.permanent
}

type schedulerRecoveryTOCTOUStoreStub struct {
	candidate    ScheduledRunCandidate
	lease        ScheduledRunLease
	failureClass ScheduledRunPreparationFailureClass
	failureError string
}

func (s *schedulerRecoveryTOCTOUStoreStub) MetadataBinding() string { return "current-binding" }
func (s *schedulerRecoveryTOCTOUStoreStub) Claim(
	context.Context,
	ScheduledRunReservationKey,
	[]byte,
) (ScheduledRunLease, error) {
	return ScheduledRunLease{}, errors.New("unexpected discovery claim")
}
func (s *schedulerRecoveryTOCTOUStoreStub) ClaimBatch(
	context.Context,
	[]ScheduledRunAcceptance,
) (ScheduledRunBatchClaim, error) {
	return ScheduledRunBatchClaim{}, errors.New("unexpected discovery batch claim")
}
func (s *schedulerRecoveryTOCTOUStoreStub) ListRecoverable(
	context.Context,
	int,
) ([]ScheduledRunCandidate, error) {
	return []ScheduledRunCandidate{s.candidate}, nil
}
func (s *schedulerRecoveryTOCTOUStoreStub) Reclaim(
	context.Context,
	ScheduledRunReservationKey,
) (ScheduledRunLease, error) {
	return s.lease, nil
}
func (s *schedulerRecoveryTOCTOUStoreStub) RefreshLease(
	_ context.Context,
	lease ScheduledRunLease,
) (ScheduledRunLease, error) {
	return lease, nil
}
func (s *schedulerRecoveryTOCTOUStoreStub) RecordPreparationFailure(
	_ context.Context,
	_ ScheduledRunLease,
	failureClass ScheduledRunPreparationFailureClass,
	lastError string,
) error {
	s.failureClass = failureClass
	s.failureError = lastError
	return nil
}
func (s *schedulerRecoveryTOCTOUStoreStub) Execute(
	context.Context,
	ScheduledRunLease,
	func(context.Context) error,
) error {
	return errors.New("unexpected effect execution")
}
func (s *schedulerRecoveryTOCTOUStoreStub) Complete(
	context.Context,
	ScheduledRunLease,
	ScheduledRunExecutionStatus,
	string,
) error {
	return errors.New("unexpected terminal completion")
}

func TestRecoveredScheduledRunClaimsBeforePreparationAndRejectsScanClaimAuthorityChange(t *testing.T) {
	def := &models.FlowDefinition{
		Id: "flow-def-recovery-authority", DeveloperName: "RecoveryAuthorityFlow",
		ActiveVersionId: "flow-version-recovery-authority",
	}
	version := scheduledFlowVersion("Daily", "2026-07-29", "10:00:00")
	version.Id = def.ActiveVersionId
	version.DefinitionId = def.Id
	plan := testScheduledFlowExecutionPlan(def, version)
	key, err := NewScheduledRunReservationKey(
		def.Id, version.Id, "daily:2026-07-29T10:00:00Z", "record-recovery-authority",
	)
	if err != nil {
		t.Fatal(err)
	}
	candidate := ScheduledRunCandidate{
		Key: key, EntryElement: "timed-entry-a", MetadataBinding: "row-binding-old-revision",
	}
	lease := testScheduledRunLease(time.Hour)
	lease.Key = key
	lease.EntryElement = "timed-entry-b"
	lease.MetadataBinding = candidate.MetadataBinding
	lease.CanonicalInputs = []byte(`{"$Record":{"Id":"record-recovery-authority"}}`)
	store := &schedulerRecoveryTOCTOUStoreStub{candidate: candidate, lease: lease}
	authorityRequests := make(chan schedulerRecoveryAuthorityRequest, 1)
	var executions atomic.Int64
	scheduler := &SchedulerService{
		fencedExecutionAuthority: schedulerRecoveryAuthorityStub{
			plan: plan, requests: authorityRequests,
		},
		executePreparedScheduledFlowFn: func(
			context.Context,
			ScheduledFlowExecutionPlan,
			map[string]interface{},
			*models.UserSession,
		) (*models.FlowExecutionResult, error) {
			executions.Add(1)
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		scheduledFlowWorkers: make(chan struct{}, 1),
		stopCh:               make(chan struct{}),
	}
	if !scheduler.tryDispatchRecoveredScheduledRun(
		t.Context(), store, candidate, newSystemAutomationUser(),
	) {
		t.Fatal("recovery dispatch was not accepted")
	}
	scheduler.wg.Wait()
	select {
	case request := <-authorityRequests:
		t.Fatalf("recovery opened authority before rejecting changed claimed row: %#v", request)
	default:
	}
	if executions.Load() != 0 {
		t.Fatalf("executions = %d, want zero after scan/claim entry mismatch", executions.Load())
	}
	if store.failureClass != ScheduledRunPreparationFailurePermanent ||
		!strings.Contains(store.failureError, "changed between recovery scan and claim") {
		t.Fatalf("preparation failure = %q/%q, want permanent changed-authority disposition",
			store.failureClass, store.failureError)
	}
}

func TestRecoveredScheduledRunDurablyClassifiesAuthorityPreparationFailure(t *testing.T) {
	key, err := NewScheduledRunReservationKey(
		"flow-def-authority-failure",
		"flow-version-authority-failure",
		"window-authority-failure",
		"record-authority-failure",
	)
	if err != nil {
		t.Fatal(err)
	}
	candidate := ScheduledRunCandidate{
		Key: key, EntryElement: "timed-entry", MetadataBinding: "row-binding",
	}
	tests := []struct {
		name      string
		permanent bool
		wantClass ScheduledRunPreparationFailureClass
	}{
		{
			name:      "temporarily missing artifact remains retryable",
			wantClass: ScheduledRunPreparationFailureRetryable,
		},
		{
			name:      "invalid immutable artifact is permanent",
			permanent: true,
			wantClass: ScheduledRunPreparationFailurePermanent,
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			lease := testScheduledRunLease(time.Hour)
			lease.Key = key
			lease.EntryElement = candidate.EntryElement
			lease.MetadataBinding = candidate.MetadataBinding
			lease.CanonicalInputs = []byte(`{"$Record":{"Id":"record-authority-failure"}}`)
			store := &schedulerRecoveryTOCTOUStoreStub{candidate: candidate, lease: lease}
			requests := make(chan schedulerRecoveryAuthorityRequest, 1)
			scheduler := &SchedulerService{
				fencedExecutionAuthority: schedulerRecoveryAuthorityStub{
					requests:  requests,
					err:       errors.New("immutable authority unavailable"),
					permanent: tt.permanent,
				},
				scheduledFlowWorkers: make(chan struct{}, 1),
				stopCh:               make(chan struct{}),
			}
			if !scheduler.tryDispatchRecoveredScheduledRun(
				t.Context(), store, candidate, newSystemAutomationUser(),
			) {
				t.Fatal("recovery dispatch was not accepted")
			}
			scheduler.wg.Wait()
			if store.failureClass != tt.wantClass ||
				!strings.Contains(store.failureError, "immutable authority unavailable") {
				t.Fatalf("preparation failure = %q/%q, want %q authority failure",
					store.failureClass, store.failureError, tt.wantClass)
			}
		})
	}
}

func TestDecodeScheduledRunInputsRehydratesFlowRecordCarriers(t *testing.T) {
	inputs, err := decodeScheduledRunInputs([]byte(
		`{"$Record":{"Id":"record-current"},"$Record__Prior":{"Id":"record-prior"},"other":{"Id":"ordinary-map"}}`,
	))
	if err != nil {
		t.Fatal(err)
	}
	if record, ok := inputs["$Record"].(models.SObject); !ok ||
		record[constants.FieldID] != "record-current" {
		t.Fatalf("$Record = %#v, want SObject carrier", inputs["$Record"])
	}
	if prior, ok := inputs["$Record__Prior"].(models.SObject); !ok ||
		prior[constants.FieldID] != "record-prior" {
		t.Fatalf("$Record__Prior = %#v, want SObject carrier", inputs["$Record__Prior"])
	}
	if _, ok := inputs["other"].(models.SObject); ok {
		t.Fatalf("ordinary map was retyped as SObject: %#v", inputs["other"])
	}
}

func (*schedulerAcceptanceStoreStub) MetadataBinding() string { return "binding" }

func (s *schedulerAcceptanceStoreStub) Claim(_ context.Context, _ ScheduledRunReservationKey, canonicalInputs []byte) (ScheduledRunLease, error) {
	lease := testScheduledRunLease(time.Hour)
	lease.CanonicalInputs = append([]byte(nil), canonicalInputs...)
	return lease, nil
}

func (s *schedulerAcceptanceStoreStub) ClaimBatch(_ context.Context, acceptances []ScheduledRunAcceptance) (ScheduledRunBatchClaim, error) {
	close(s.claimEntered)
	<-s.allowClaim
	leases := make([]ScheduledRunLease, len(acceptances))
	for i, acceptance := range acceptances {
		leases[i] = testScheduledRunLease(time.Hour)
		leases[i].ReservationID = fmt.Sprintf("reservation-%d", i)
		leases[i].CanonicalInputs = append([]byte(nil), acceptance.CanonicalInputs...)
	}
	return ScheduledRunBatchClaim{Leases: leases}, nil
}

func (*schedulerAcceptanceStoreStub) ListRecoverable(context.Context, int) ([]ScheduledRunCandidate, error) {
	return nil, nil
}

func (*schedulerAcceptanceStoreStub) Reclaim(context.Context, ScheduledRunReservationKey) (ScheduledRunLease, error) {
	return ScheduledRunLease{Status: ScheduledRunLeaseDuplicate}, nil
}

func (*schedulerAcceptanceStoreStub) RefreshLease(
	_ context.Context,
	lease ScheduledRunLease,
) (ScheduledRunLease, error) {
	return lease, nil
}

func (*schedulerAcceptanceStoreStub) RecordPreparationFailure(
	context.Context,
	ScheduledRunLease,
	ScheduledRunPreparationFailureClass,
	string,
) error {
	return nil
}

func (*schedulerAcceptanceStoreStub) Execute(ctx context.Context, _ ScheduledRunLease, effect func(context.Context) error) error {
	return effect(ctx)
}

func (s *schedulerAcceptanceStoreStub) Complete(context.Context, ScheduledRunLease, ScheduledRunExecutionStatus, string) error {
	s.completeCalls.Add(1)
	return s.completeErr
}

func TestFencedExecutionObservationLeavesFailedDispositionNonterminal(t *testing.T) {
	def := &models.FlowDefinition{
		Id: "flow-def-observation", DeveloperName: "ObservedFencedFlow",
		ActiveVersionId: "flow-version-observation",
	}
	version := scheduledFlowVersion("Daily", "2026-07-29", "10:00:00")
	version.Id = def.ActiveVersionId
	version.DefinitionId = def.Id
	store := &schedulerAcceptanceStoreStub{completeErr: errors.New("terminal write unavailable")}
	scheduler := &SchedulerService{
		executePreparedScheduledFlowFn: func(context.Context, ScheduledFlowExecutionPlan, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			return &models.FlowExecutionResult{IsSuccess: false}, nil
		},
	}
	_, err := scheduler.executeFencedScheduledVersion(
		t.Context(), store, testScheduledRunLease(time.Hour),
		testPreparedScheduledFlowExecution(testScheduledFlowExecutionPlan(def, version)),
		nil, newSystemAutomationUser(),
	)
	if err == nil {
		t.Fatal("failed terminal disposition was not returned")
	}
	if snapshot := scheduler.WorkerObservationSnapshot(); snapshot.ExecutionStarted != 1 || snapshot.ExecutionCompleted != 0 || snapshot.ExecutionFailed != 0 {
		t.Fatalf("worker observation=%+v", snapshot)
	}
}

func TestFencedDiscoveryDoesNotAdvanceBeforeDurableAcceptance(t *testing.T) {
	def := &models.FlowDefinition{
		Id: "flow-def-acceptance", DeveloperName: "AcceptanceBeforeCursor",
		ActiveVersionId: "flow-version-acceptance",
	}
	version := scheduledFlowVersion("Daily", "2026-07-29", "10:00:00")
	version.Id = def.ActiveVersionId
	version.DefinitionId = def.Id
	plan := testScheduledFlowExecutionPlan(def, version)
	store := &schedulerAcceptanceStoreStub{
		claimEntered: make(chan struct{}),
		allowClaim:   make(chan struct{}),
	}
	executed := make(chan struct{})
	scheduler := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{def}
		},
		getFlowDefinitionByIDFn: func(context.Context, string) *models.FlowDefinition { return def },
		getFlowVersionByIDFn:    func(context.Context, string) *models.FlowVersion { return version },
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{
				"$Record": map[string]interface{}{"Id": "record-acceptance"},
			}}, nil
		},
		fencedExecutionAuthority: scheduledFlowAuthorityStub{plan: plan},
		executePreparedScheduledFlowFn: func(context.Context, ScheduledFlowExecutionPlan, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			close(executed)
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		fencedReservationStoreFn: func() scheduledRunFencedStore { return store },
		scheduledFlowWorkers:     make(chan struct{}, 1),
		stopCh:                   make(chan struct{}),
	}
	accepted := map[string]struct{}{}
	dispatchDone := make(chan bool, 1)
	go func() {
		dispatchDone <- scheduler.dispatchScheduledFlowsBetweenWithProgress(
			time.Date(2026, 7, 29, 9, 59, 0, 0, time.UTC),
			time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC),
			accepted,
		)
	}()
	select {
	case <-store.claimEntered:
	case <-time.After(time.Second):
		t.Fatal("discovery did not reach durable Claim")
	}
	select {
	case <-dispatchDone:
		t.Fatal("discovery advanced before durable Claim completed")
	default:
	}
	if len(accepted) != 0 {
		t.Fatalf("accepted progress entries = %d before durable Claim, want zero", len(accepted))
	}
	close(store.allowClaim)
	select {
	case complete := <-dispatchDone:
		if !complete {
			t.Fatal("discovery did not complete after durable Claim")
		}
	case <-time.After(time.Second):
		t.Fatal("discovery did not return after durable Claim")
	}
	if len(accepted) != 1 {
		t.Fatalf("accepted progress entries = %d after durable Claim, want one", len(accepted))
	}
	select {
	case <-executed:
	case <-time.After(time.Second):
		t.Fatal("durably accepted occurrence was not executed")
	}
	scheduler.wg.Wait()
	if store.completeCalls.Load() != 0 {
		t.Fatalf("terminal completions = %d, want zero; success completion belongs to the effect transaction fence", store.completeCalls.Load())
	}
	if snapshot := scheduler.WorkerObservationSnapshot(); snapshot.AcceptanceAttempts != 1 || snapshot.AcceptanceEmpty != 0 || snapshot.OccurrencesAccepted != 1 || snapshot.ExecutionStarted != 1 || snapshot.ExecutionCompleted != 1 || snapshot.ExecutionFailed != 0 {
		t.Fatalf("worker observation=%+v", snapshot)
	}
}

func TestFencedDiscoveryDurablyAcceptsWhenExecutionLaneIsFull(t *testing.T) {
	def := &models.FlowDefinition{
		Id: "flow-def-capacity", DeveloperName: "AcceptanceBeforeCapacity",
		ActiveVersionId: "flow-version-capacity",
	}
	version := scheduledFlowVersion("Daily", "2026-07-29", "10:00:00")
	version.Id = def.ActiveVersionId
	version.DefinitionId = def.Id
	plan := testScheduledFlowExecutionPlan(def, version)
	store := &schedulerAcceptanceStoreStub{
		claimEntered: make(chan struct{}),
		allowClaim:   make(chan struct{}),
	}
	close(store.allowClaim)
	var executions atomic.Int64
	workers := make(chan struct{}, 1)
	workers <- struct{}{}
	scheduler := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{def}
		},
		getFlowDefinitionByIDFn: func(context.Context, string) *models.FlowDefinition { return def },
		getFlowVersionByIDFn:    func(context.Context, string) *models.FlowVersion { return version },
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{
				"$Record": map[string]interface{}{"Id": "record-capacity"},
			}}, nil
		},
		fencedExecutionAuthority: scheduledFlowAuthorityStub{plan: plan},
		executePreparedScheduledFlowFn: func(context.Context, ScheduledFlowExecutionPlan, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			executions.Add(1)
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		fencedReservationStoreFn: func() scheduledRunFencedStore { return store },
		scheduledFlowWorkers:     workers,
		stopCh:                   make(chan struct{}),
	}
	accepted := map[string]struct{}{}
	if complete := scheduler.dispatchScheduledFlowsBetweenWithProgress(
		time.Date(2026, 7, 29, 9, 59, 0, 0, time.UTC),
		time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC),
		accepted,
	); !complete {
		t.Fatal("fenced discovery reported incomplete only because the execution lane was full")
	}
	if len(accepted) != 1 {
		t.Fatalf("accepted progress entries = %d, want one durable occurrence", len(accepted))
	}
	if executions.Load() != 0 || store.completeCalls.Load() != 0 {
		t.Fatalf("executions/completions = %d/%d, want zero until lease recovery",
			executions.Load(), store.completeCalls.Load())
	}
	if snapshot := scheduler.WorkerObservationSnapshot(); snapshot.AcceptanceAttempts != 1 || snapshot.OccurrencesAccepted != 1 || snapshot.CapacityDeferrals != 1 || snapshot.ExecutionStarted != 0 {
		t.Fatalf("worker observation=%+v", snapshot)
	}
	<-workers
}

func (s *schedulerLeaseLossStoreStub) Claim(context.Context, ScheduledRunReservationKey, []byte) (ScheduledRunLease, error) {
	return ScheduledRunLease{}, nil
}

func (*schedulerLeaseLossStoreStub) MetadataBinding() string { return "binding" }

func (s *schedulerLeaseLossStoreStub) ClaimBatch(context.Context, []ScheduledRunAcceptance) (ScheduledRunBatchClaim, error) {
	return ScheduledRunBatchClaim{}, nil
}

func (s *schedulerLeaseLossStoreStub) ListRecoverable(context.Context, int) ([]ScheduledRunCandidate, error) {
	return nil, nil
}

func (s *schedulerLeaseLossStoreStub) Reclaim(context.Context, ScheduledRunReservationKey) (ScheduledRunLease, error) {
	return ScheduledRunLease{}, nil
}

func (s *schedulerLeaseLossStoreStub) RefreshLease(
	_ context.Context,
	lease ScheduledRunLease,
) (ScheduledRunLease, error) {
	return lease, nil
}

func (s *schedulerLeaseLossStoreStub) RecordPreparationFailure(
	context.Context,
	ScheduledRunLease,
	ScheduledRunPreparationFailureClass,
	string,
) error {
	return nil
}

func (s *schedulerLeaseLossStoreStub) Execute(
	ctx context.Context,
	_ ScheduledRunLease,
	effect func(context.Context) error,
) error {
	s.executeCalls.Add(1)
	if s.executeErr != nil {
		return s.executeErr
	}
	return effect(ctx)
}

func (s *schedulerLeaseLossStoreStub) Complete(context.Context, ScheduledRunLease, ScheduledRunExecutionStatus, string) error {
	s.completeCalls.Add(1)
	return nil
}

func TestSchedulerWorkerRoleUsesControllerContextAndExcludesPausedInterviews(t *testing.T) {
	var pausedCalls atomic.Int64
	var sawControllerContext atomic.Bool
	controllerCtx, cancel := context.WithCancel(t.Context())
	scheduler := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(callCtx context.Context) []*models.FlowDefinition {
			if callCtx == controllerCtx {
				sawControllerContext.Store(true)
			}
			return nil
		},
		listDuePausedInterviewsFn: func(context.Context, time.Time, int, *models.UserSession) ([]string, error) {
			pausedCalls.Add(1)
			return nil, nil
		},
		resumeInterviewFn: func(context.Context, string, *models.UserSession) error { return nil },
		fencedReservationStoreFn: func() scheduledRunFencedStore {
			return schedulerWorkerRoleStoreStub{}
		},
		fencedExecutionAuthority: scheduledFlowAuthorityStub{
			plan: testScheduledFlowExecutionPlan(
				&models.FlowDefinition{
					Id: "flow-definition", DeveloperName: "Scheduled",
					ActiveVersionId: "flow-version",
				},
				&models.FlowVersion{
					Id: "flow-version", DefinitionId: "flow-definition",
				},
			),
		},
	}
	scheduler.SetSchedulerIntervalForTest(5 * time.Millisecond)
	done := make(chan error, 1)
	go func() {
		done <- scheduler.RunWorkerRole(controllerCtx)
	}()
	time.Sleep(30 * time.Millisecond)
	cancel()
	select {
	case err := <-done:
		if err != nil {
			t.Fatalf("RunWorkerRole() error = %v", err)
		}
	case <-time.After(time.Second):
		t.Fatal("RunWorkerRole did not drain after controller cancellation")
	}
	if !sawControllerContext.Load() {
		t.Fatal("scheduler loop did not use the Worker controller context")
	}
	if pausedCalls.Load() != 0 {
		t.Fatalf("paused loop called %d times, want zero", pausedCalls.Load())
	}
}

func TestFencedScheduledExecutionLeavesOwnerCancellationForDurableReclaim(t *testing.T) {
	store := &schedulerLeaseLossStoreStub{}
	started := make(chan struct{})
	scheduler := &SchedulerService{
		executePreparedScheduledFlowFn: func(ctx context.Context, _ ScheduledFlowExecutionPlan, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			close(started)
			<-ctx.Done()
			return nil, ctx.Err()
		},
	}
	plan := testScheduledFlowExecutionPlan(
		&models.FlowDefinition{
			Id: "flow-definition", DeveloperName: "Scheduled",
			ActiveVersionId: "flow-version",
		},
		&models.FlowVersion{Id: "flow-version", DefinitionId: "flow-definition"},
	)
	parent, cancel := context.WithCancel(t.Context())
	done := make(chan error, 1)
	go func() {
		_, err := scheduler.executeFencedScheduledVersion(
			parent,
			store,
			testScheduledRunLease(time.Hour),
			testPreparedScheduledFlowExecution(plan),
			nil,
			&models.UserSession{},
		)
		done <- err
	}()
	<-started
	cancel()
	select {
	case err := <-done:
		if err == nil || !errors.Is(err, context.Canceled) {
			t.Fatalf("execution error = %v, want controller cancellation", err)
		}
	case <-time.After(time.Second):
		t.Fatal("execution did not drain after controller cancellation")
	}
	if store.completeCalls.Load() != 0 {
		t.Fatalf("terminal completions = %d, want zero so the effect-free attempt remains reclaimable after expiry", store.completeCalls.Load())
	}
}

func TestFencedScheduledExecutionInstallsPreparedAuthorityBeforeEffect(t *testing.T) {
	type exactAuthorityContextKey struct{}
	const exactAuthority = "durable-row-revision"
	store := &schedulerLeaseLossStoreStub{}
	effectCalls := 0
	scheduler := &SchedulerService{
		executePreparedScheduledFlowFn: func(
			ctx context.Context,
			_ ScheduledFlowExecutionPlan,
			_ map[string]interface{},
			_ *models.UserSession,
		) (*models.FlowExecutionResult, error) {
			effectCalls++
			if got, _ := ctx.Value(exactAuthorityContextKey{}).(string); got != exactAuthority {
				return nil, fmt.Errorf(
					"scheduled effect authority = %q, want %q",
					got,
					exactAuthority,
				)
			}
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
	}
	plan := testScheduledFlowExecutionPlan(
		&models.FlowDefinition{
			Id: "flow-definition", DeveloperName: "Scheduled",
			ActiveVersionId: "flow-version",
		},
		&models.FlowVersion{Id: "flow-version", DefinitionId: "flow-definition"},
	)
	prepared, err := NewPreparedScheduledFlowExecution(
		plan,
		func(ctx context.Context) (context.Context, error) {
			return context.WithValue(ctx, exactAuthorityContextKey{}, exactAuthority), nil
		},
	)
	if err != nil {
		t.Fatal(err)
	}
	result, err := scheduler.executeFencedScheduledVersion(
		t.Context(),
		store,
		testScheduledRunLease(time.Hour),
		prepared,
		nil,
		&models.UserSession{},
	)
	if err != nil {
		t.Fatal(err)
	}
	if result == nil || !result.IsSuccess || effectCalls != 1 {
		t.Fatalf("scheduled result = %#v, effect calls = %d", result, effectCalls)
	}
}

func TestFencedScheduledExecutionLeavesLeaseLossForDurableDisposition(t *testing.T) {
	store := &schedulerLeaseLossStoreStub{executeErr: ErrScheduledRunLeaseLost}
	scheduler := &SchedulerService{
		executePreparedScheduledFlowFn: func(ctx context.Context, _ ScheduledFlowExecutionPlan, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			return nil, errors.New("effect must not start after lease loss")
		},
	}
	plan := testScheduledFlowExecutionPlan(
		&models.FlowDefinition{
			Id: "flow-definition", DeveloperName: "Scheduled",
			ActiveVersionId: "flow-version",
		},
		&models.FlowVersion{Id: "flow-version", DefinitionId: "flow-definition"},
	)
	_, err := scheduler.executeFencedScheduledVersion(
		t.Context(),
		store,
		testScheduledRunLease(time.Hour),
		testPreparedScheduledFlowExecution(plan),
		nil,
		&models.UserSession{},
	)
	if !errors.Is(err, ErrScheduledRunLeaseLost) {
		t.Fatalf("execution error = %v, want lease loss", err)
	}
	if store.executeCalls.Load() != 1 {
		t.Fatalf("atomic execution calls = %d, want one", store.executeCalls.Load())
	}
	if store.completeCalls.Load() != 0 {
		t.Fatalf("terminal completions = %d, want zero after lease loss", store.completeCalls.Load())
	}
}

type scheduledFlowAuthorityStub struct {
	plan             ScheduledFlowExecutionPlan
	installExecution ScheduledFlowExecutionInstaller
	err              error
	permanent        bool
}

func (s scheduledFlowAuthorityStub) PrepareScheduledFlow(
	context.Context,
	string,
	string,
	string,
	string,
) (PreparedScheduledFlowExecution, error) {
	if s.err != nil {
		return PreparedScheduledFlowExecution{}, s.err
	}
	installExecution := s.installExecution
	if installExecution == nil {
		installExecution = func(ctx context.Context) (context.Context, error) {
			return ctx, nil
		}
	}
	return NewPreparedScheduledFlowExecution(s.plan, installExecution)
}

func (s scheduledFlowAuthorityStub) IsPermanentScheduledFlowAuthorityFailure(error) bool {
	return s.permanent
}

var errPermanentlyIneligibleScheduledFlow = errors.New("scheduled Flow launch shape is permanently ineligible")

type scheduledFlowAuthorityByDefinitionStub struct {
	plans       map[string]ScheduledFlowExecutionPlan
	permanentID string
	calls       *atomic.Int64
}

func (s scheduledFlowAuthorityByDefinitionStub) PrepareScheduledFlow(
	_ context.Context,
	_ string,
	definitionID string,
	_ string,
	_ string,
) (PreparedScheduledFlowExecution, error) {
	s.calls.Add(1)
	if definitionID == s.permanentID {
		return PreparedScheduledFlowExecution{}, errPermanentlyIneligibleScheduledFlow
	}
	return NewPreparedScheduledFlowExecution(s.plans[definitionID], func(ctx context.Context) (context.Context, error) {
		return ctx, nil
	})
}

func (scheduledFlowAuthorityByDefinitionStub) IsPermanentScheduledFlowAuthorityFailure(err error) bool {
	return errors.Is(err, errPermanentlyIneligibleScheduledFlow)
}

func TestFencedDiscoveryPermanentIneligibleFlowDoesNotStarveHealthyFlow(t *testing.T) {
	const (
		ineligibleDefinitionID = "flow-def-ineligible"
		healthyDefinitionID    = "flow-def-healthy"
	)
	ineligibleDefinition := &models.FlowDefinition{
		Id: ineligibleDefinitionID, DeveloperName: "IneligibleScheduled",
		ActiveVersionId: "flow-version-ineligible",
	}
	healthyDefinition := &models.FlowDefinition{
		Id: healthyDefinitionID, DeveloperName: "HealthyScheduled",
		ActiveVersionId: "flow-version-healthy",
	}
	ineligibleVersion := scheduledFlowVersion("Daily", "2026-07-29", "10:00:00")
	ineligibleVersion.Id = ineligibleDefinition.ActiveVersionId
	ineligibleVersion.DefinitionId = ineligibleDefinition.Id
	healthyVersion := scheduledFlowVersion("Daily", "2026-07-29", "10:00:00")
	healthyVersion.Id = healthyDefinition.ActiveVersionId
	healthyVersion.DefinitionId = healthyDefinition.Id
	var authorityCalls atomic.Int64
	scheduler := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{ineligibleDefinition, healthyDefinition}
		},
		getFlowVersionByIDFn: func(_ context.Context, id string) *models.FlowVersion {
			switch id {
			case ineligibleVersion.Id:
				return ineligibleVersion
			case healthyVersion.Id:
				return healthyVersion
			default:
				return nil
			}
		},
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{{
				"$Record": map[string]interface{}{"Id": "record-healthy"},
			}}, nil
		},
		fencedExecutionAuthority: scheduledFlowAuthorityByDefinitionStub{
			plans: map[string]ScheduledFlowExecutionPlan{
				healthyDefinitionID: testScheduledFlowExecutionPlan(healthyDefinition, healthyVersion),
			},
			permanentID: ineligibleDefinitionID,
			calls:       &authorityCalls,
		},
		fencedReservationStoreFn: func() scheduledRunFencedStore {
			return schedulerWorkerRoleStoreStub{}
		},
		scheduledFlowWorkers: make(chan struct{}, 1),
		stopCh:               make(chan struct{}),
	}
	accepted := make(map[string]struct{})
	if complete := scheduler.dispatchScheduledFlowsBetweenWithProgress(
		time.Date(2026, 7, 29, 9, 59, 0, 0, time.UTC),
		time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC),
		accepted,
	); !complete {
		t.Fatal("permanently ineligible Flow pinned the discovery cursor")
	}
	if len(accepted) != 2 {
		t.Fatalf("discovery progress entries = %d, want rejected and healthy occurrences", len(accepted))
	}
	if authorityCalls.Load() != 2 {
		t.Fatalf("authority calls = %d, want ineligible and later healthy Flow", authorityCalls.Load())
	}
}

func TestFencedDiscoveryRetryableAuthorityFailureRetainsCursor(t *testing.T) {
	def := &models.FlowDefinition{
		Id: "flow-def-retryable", DeveloperName: "RetryableScheduled",
		ActiveVersionId: "flow-version-retryable",
	}
	version := scheduledFlowVersion("Daily", "2026-07-29", "10:00:00")
	version.Id = def.ActiveVersionId
	version.DefinitionId = def.Id
	scheduler := &SchedulerService{
		listScheduledFlowDefinitionsFn: func(context.Context) []*models.FlowDefinition {
			return []*models.FlowDefinition{def}
		},
		getFlowVersionByIDFn: func(context.Context, string) *models.FlowVersion { return version },
		fencedExecutionAuthority: scheduledFlowAuthorityStub{
			err:       errors.New("immutable artifact store temporarily unavailable"),
			permanent: false,
		},
		fencedReservationStoreFn: func() scheduledRunFencedStore {
			return schedulerWorkerRoleStoreStub{}
		},
		scheduledFlowWorkers: make(chan struct{}, 1),
		stopCh:               make(chan struct{}),
	}
	accepted := make(map[string]struct{})
	if complete := scheduler.dispatchScheduledFlowsBetweenWithProgress(
		time.Date(2026, 7, 29, 9, 59, 0, 0, time.UTC),
		time.Date(2026, 7, 29, 10, 0, 0, 0, time.UTC),
		accepted,
	); complete {
		t.Fatal("retryable authority failure advanced the discovery cursor")
	}
	if len(accepted) != 0 {
		t.Fatalf("retryable discovery progress entries = %d, want zero", len(accepted))
	}
}

func testScheduledFlowExecutionPlan(
	definition *models.FlowDefinition,
	version *models.FlowVersion,
) ScheduledFlowExecutionPlan {
	definitionCopy := *definition
	versionCopy := *version
	versionCopy.Status = models.FlowStatusActive
	versionCopy.Metadata.RunAtomically = true
	if versionCopy.Metadata.Start == nil {
		versionCopy.Metadata.Start = &models.FlowStart{
			Schedule: &models.FlowSchedule{Frequency: "Daily"},
		}
	}
	return ScheduledFlowExecutionPlan{
		Definition: &definitionCopy,
		Root:       &versionCopy,
		Versions: map[string]*models.FlowVersion{
			strings.ToLower(definitionCopy.DeveloperName): &versionCopy,
		},
		MaxElementExecutions: 1,
	}
}

func testPreparedScheduledFlowExecution(
	plan ScheduledFlowExecutionPlan,
) PreparedScheduledFlowExecution {
	prepared, err := NewPreparedScheduledFlowExecution(
		plan,
		func(ctx context.Context) (context.Context, error) {
			return ctx, nil
		},
	)
	if err != nil {
		panic(err)
	}
	return prepared
}

func testScheduledRunLease(renewAfter time.Duration) ScheduledRunLease {
	return ScheduledRunLease{
		Status: ScheduledRunLeaseClaimed, ReservationID: "reservation-1",
		LeaseToken: "lease-1", Attempt: 1, LeaseExpiresAt: time.Now().Add(renewAfter),
		MetadataBinding: "{}",
	}
}