package automation

import (
	"bytes"
	"context"
	"errors"
	"log"
	"strings"
	"testing"
	"time"

	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/stretchr/testify/assert"
)

func TestShouldRunFlow(t *testing.T) {
	svc := &SchedulerService{}

	tests := []struct {
		name     string
		version  *models.FlowVersion
		now      time.Time
		expected bool
	}{
		{
			name: "nil Start returns false",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{},
			},
			now:      time.Now(),
			expected: false,
		},
		{
			name: "nil Schedule returns false",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{},
				},
			},
			now:      time.Now(),
			expected: false,
		},
		{
			name: "Daily - exact match",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Daily",
							StartTime: "14:30",
						},
					},
				},
			},
			now:      time.Date(2025, 6, 15, 14, 30, 0, 0, time.UTC),
			expected: true,
		},
		{
			name: "Daily - wrong time",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Daily",
							StartTime: "14:30",
						},
					},
				},
			},
			now:      time.Date(2025, 6, 15, 10, 0, 0, 0, time.UTC),
			expected: false,
		},
		{
			name: "Once - within window",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Once",
							StartDate: "2025-06-15",
							StartTime: "14:30",
						},
					},
				},
			},
			now:      time.Date(2025, 6, 15, 14, 31, 0, 0, time.UTC),
			expected: true,
		},
		{
			name: "Once - before scheduled time",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Once",
							StartDate: "2025-06-15",
							StartTime: "14:30",
						},
					},
				},
			},
			now:      time.Date(2025, 6, 15, 14, 0, 0, 0, time.UTC),
			expected: false,
		},
		{
			name: "Once - well after scheduled time",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Once",
							StartDate: "2025-06-15",
							StartTime: "14:30",
						},
					},
				},
			},
			now:      time.Date(2025, 6, 15, 15, 0, 0, 0, time.UTC),
			expected: false,
		},
		{
			name: "Once - missing startTime",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Once",
							StartDate: "2025-06-15",
						},
					},
				},
			},
			now:      time.Date(2025, 6, 15, 14, 30, 0, 0, time.UTC),
			expected: false,
		},
		{
			name: "Weekly - correct day and time",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Weekly",
							StartDate: "2025-06-09", // Monday
							StartTime: "09:00",
						},
					},
				},
			},
			// A Monday at 09:00
			now:      time.Date(2025, 6, 16, 9, 0, 0, 0, time.UTC),
			expected: true,
		},
		{
			name: "Weekly - wrong day",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Weekly",
							StartDate: "2025-06-09", // Monday
							StartTime: "09:00",
						},
					},
				},
			},
			// A Tuesday at 09:00
			now:      time.Date(2025, 6, 17, 9, 0, 0, 0, time.UTC),
			expected: false,
		},
		{
			name: "Weekly - missing startDate",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Weekly",
							StartTime: "09:00",
						},
					},
				},
			},
			now:      time.Date(2025, 6, 16, 9, 0, 0, 0, time.UTC),
			expected: false,
		},
		{
			name: "before start date returns false",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Daily",
							StartDate: "2025-12-01",
							StartTime: "09:00",
						},
					},
				},
			},
			now:      time.Date(2025, 6, 15, 9, 0, 0, 0, time.UTC),
			expected: false,
		},
		{
			name: "unknown frequency returns false",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Monthly",
							StartTime: "09:00",
						},
					},
				},
			},
			now:      time.Now(),
			expected: false,
		},
		{
			name: "time with seconds parsed correctly",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Daily",
							StartTime: "14:30:00",
						},
					},
				},
			},
			now:      time.Date(2025, 6, 15, 14, 30, 0, 0, time.UTC),
			expected: true,
		},
		// --- issue #2232: schedule startTime is UTC-anchored regardless of host TZ ---
		{
			name: "Daily UTC-anchored schedule fires when host clock is non-UTC and UTC matches",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Daily",
							StartTime: "05:00:00.000Z",
						},
					},
				},
			},
			// 13:00 in UTC+8 == 05:00 UTC: should fire.
			now:      time.Date(2025, 6, 15, 13, 0, 0, 0, time.FixedZone("UTC+8", 8*3600)),
			expected: true,
		},
		{
			name: "Daily UTC-anchored schedule does not fire when local clock matches but UTC does not",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Daily",
							StartTime: "05:00",
						},
					},
				},
			},
			// 05:00 in UTC+8 == 21:00 UTC the previous day: must NOT fire.
			now:      time.Date(2025, 6, 15, 5, 0, 0, 0, time.FixedZone("UTC+8", 8*3600)),
			expected: false,
		},
		{
			name: "Weekly day boundary uses UTC weekday not local weekday",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Weekly",
							StartDate: "2025-06-09", // Monday
							StartTime: "23:00",
						},
					},
				},
			},
			// 07:00 Tuesday in UTC+8 == 23:00 Monday UTC: Monday in UTC, must fire.
			now:      time.Date(2025, 6, 17, 7, 0, 0, 0, time.FixedZone("UTC+8", 8*3600)),
			expected: true,
		},
		{
			name: "Once window evaluated in UTC",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Once",
							StartDate: "2025-06-15",
							StartTime: "14:30:00.000Z",
						},
					},
				},
			},
			// 22:31 in UTC+8 on the 15th == 14:31 UTC on the 15th: inside the 1-min window.
			now:      time.Date(2025, 6, 15, 22, 31, 0, 0, time.FixedZone("UTC+8", 8*3600)),
			expected: true,
		},
		{
			name: "Daily HH:MM bare string still treated as UTC (back-compat contract)",
			version: &models.FlowVersion{
				Metadata: models.FlowMetadata{
					Start: &models.FlowStart{
						Schedule: &models.FlowSchedule{
							Frequency: "Daily",
							StartTime: "09:00",
						},
					},
				},
			},
			// 17:00 UTC+8 == 09:00 UTC: should fire.
			now:      time.Date(2025, 6, 15, 17, 0, 0, 0, time.FixedZone("UTC+8", 8*3600)),
			expected: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := svc.shouldRunFlow(tt.version, tt.now)
			assert.Equal(t, tt.expected, result)
		})
	}
}

func TestSchedulerService_StartStopIdempotent(t *testing.T) {
	svc := &SchedulerService{
		stopCh: make(chan struct{}),
	}

	// Stop without start should be safe
	svc.StopScheduler()

	// Double start should be safe
	svc.StartScheduler()
	svc.StartScheduler() // Should not panic or deadlock

	// Stop
	svc.StopScheduler()

	// Double stop should be safe
	svc.StopScheduler()
}

func TestExecuteScheduledDefinition_RunsOncePerInputSet(t *testing.T) {
	executedInputs := make([]map[string]interface{}, 0)
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{
				{"$Record": models.SObject{"Id": "001", "Name": "Acme"}},
				{"$Record": models.SObject{"Id": "002", "Name": "Globex"}},
			}, nil
		},
		executeFlowVersionFn: func(_ context.Context, _ *models.FlowVersion, _ string, inputs map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			executedInputs = append(executedInputs, inputs)
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
	}

	svc.executeScheduledDefinition(
		context.Background(),
		&models.FlowDefinition{DeveloperName: "Scheduled_Accounts"},
		&models.FlowVersion{},
		&models.UserSession{ID: "system"},
	)

	assert.Len(t, executedInputs, 2)
	assert.Equal(t, models.SObject{"Id": "001", "Name": "Acme"}, executedInputs[0]["$Record"])
	assert.Equal(t, models.SObject{"Id": "002", "Name": "Globex"}, executedInputs[1]["$Record"])
}

func TestExecuteScheduledDefinition_SkipsExecutionWhenNoInputsMatch(t *testing.T) {
	executeCalls := 0
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return []map[string]interface{}{}, nil
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			executeCalls++
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
	}

	svc.executeScheduledDefinition(
		context.Background(),
		&models.FlowDefinition{DeveloperName: "Scheduled_Accounts"},
		&models.FlowVersion{},
		&models.UserSession{ID: "system"},
	)

	assert.Equal(t, 0, executeCalls)
}

func TestExecuteScheduledDefinition_UsesSystemContext(t *testing.T) {
	loadCtxLevel := constants.UserContext
	execCtxLevel := constants.UserContext

	svc := &SchedulerService{
		loadScheduledInputsFn: func(ctx context.Context, _ *models.FlowVersion, _ *models.UserSession) ([]map[string]interface{}, error) {
			loadCtxLevel = constants.GetSecurityContext(ctx)
			return []map[string]interface{}{{"$Record": models.SObject{"Id": "001"}}}, nil
		},
		executeFlowVersionFn: func(ctx context.Context, _ *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			execCtxLevel = constants.GetSecurityContext(ctx)
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
	}

	svc.executeScheduledDefinition(
		context.Background(),
		&models.FlowDefinition{DeveloperName: "Scheduled_Accounts"},
		&models.FlowVersion{},
		&models.UserSession{ID: "system"},
	)

	assert.Equal(t, constants.SystemContext, loadCtxLevel)
	assert.Equal(t, constants.SystemContext, execCtxLevel)
}

// TestExecuteScheduledDefinition_BacksOffAfterRepeatedFailures is the
// regression test for issue #2199. Before the fix, the scheduler re-attempts
// every input on every tick even when prior ticks consistently failed for
// the same flow definition — a single misconfigured Scheduled flow with N
// input rows produces N failures per minute in perpetuity, drowning log
// signal and accumulating cost.
//
// After the fix, repeated all-failed ticks must trigger a backoff ladder
// (1m → 5m → 15m → 1h → 24h) keyed by FlowDefinition.Id, so the scheduler
// stops storming and the data-issue becomes diagnosable.
func TestExecuteScheduledDefinition_BacksOffAfterRepeatedFailures(t *testing.T) {
	callCount := 0
	inputs := []map[string]interface{}{
		{"$Record": models.SObject{"Id": "001"}},
		{"$Record": models.SObject{"Id": "002"}},
		{"$Record": models.SObject{"Id": "003"}},
	}
	base := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC)
	now := base
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return inputs, nil
		},
		executeFlowVersionFn: func(_ context.Context, _ *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			callCount++
			return nil, errors.New(`FlowVersionId: referenced Flow record "X" is not accessible or does not exist`)
		},
		nowFn: func() time.Time { return now },
	}
	def := &models.FlowDefinition{Id: "def1", DeveloperName: "Scheduled_Accounts"}

	// Simulate 30 ticks at 1-minute intervals. Without backoff this is
	// 30 × 3 inputs = 90 attempts; with the backoff ladder
	// (1m, 5m, 15m, 1h, 24h) only ticks 0, 1, 6, 21 should execute within
	// the first 30 minutes (≈12 attempts).
	for i := 0; i < 30; i++ {
		now = base.Add(time.Duration(i) * time.Minute)
		svc.executeScheduledDefinition(
			context.Background(),
			def,
			&models.FlowVersion{},
			&models.UserSession{ID: "system"},
		)
	}

	// Math: with the ladder [1m, 5m, 15m, 1h, 24h] and 1-minute ticks,
	// ticks 0, 1, 6, 21 execute within the first 30 minutes — 4 ticks ×
	// 3 inputs = 12 calls. Exact bound (not a band) so a regression that
	// over-suppresses (callCount=0) or under-suppresses (≥30) both fail.
	assert.Equal(t, 12, callCount,
		"scheduler must back off after consecutive failures (issue #2199); got %d calls", callCount)
}

// TestExecuteScheduledDefinition_ResetsBackoffOnSuccess proves the back-off
// state does not leak across recoveries: once any tick succeeds for a flow
// definition, the consecutive-failure counter resets and subsequent ticks
// run without artificial delay.
func TestExecuteScheduledDefinition_ResetsBackoffOnSuccess(t *testing.T) {
	calls := 0
	failNext := true
	inputs := []map[string]interface{}{{"$Record": models.SObject{"Id": "001"}}}
	base := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC)
	now := base
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return inputs, nil
		},
		executeFlowVersionFn: func(_ context.Context, _ *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			calls++
			if failNext {
				return nil, errors.New("simulated FK validator failure")
			}
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		nowFn: func() time.Time { return now },
	}
	def := &models.FlowDefinition{Id: "def-reset", DeveloperName: "Scheduled_X"}

	// 3 failing ticks → backoff state advances.
	for i := 0; i < 3; i++ {
		now = base.Add(time.Duration(i) * time.Minute)
		svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	}
	failingCalls := calls
	assert.Greater(t, failingCalls, 0)

	// Skip far enough ahead to escape any backoff window, then succeed once.
	failNext = false
	now = base.Add(2 * time.Hour)
	svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	postSuccessCalls := calls
	assert.Equal(t, failingCalls+1, postSuccessCalls, "success tick must execute, not skip")

	// Immediately afterwards, another tick (failing again) must also run —
	// proving the backoff counter reset to zero on success.
	failNext = true
	now = base.Add(2*time.Hour + time.Minute)
	svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	assert.Equal(t, postSuccessCalls+1, calls, "next tick after success must run (no stale backoff state)")
}

// TestExecuteScheduledDefinition_BackoffPerDefinitionId proves two flows
// with distinct Ids back off independently — one storming flow does not
// silence a healthy sibling that shares the scheduler tick.
func TestExecuteScheduledDefinition_BackoffPerDefinitionId(t *testing.T) {
	calls := map[string]int{}
	// executeFlowVersionFn receives the flow API name (DeveloperName), not Id.
	failingFor := "BadFlow"
	inputs := []map[string]interface{}{{"$Record": models.SObject{"Id": "001"}}}
	base := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC)
	now := base
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return inputs, nil
		},
		executeFlowVersionFn: func(_ context.Context, _ *models.FlowVersion, name string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			calls[name]++
			if name == failingFor {
				return nil, errors.New("simulated failure for one def only")
			}
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		nowFn: func() time.Time { return now },
	}
	bad := &models.FlowDefinition{Id: "def-bad", DeveloperName: "BadFlow"}
	good := &models.FlowDefinition{Id: "def-good", DeveloperName: "GoodFlow"}

	// 10 ticks at 1-minute intervals.
	for i := 0; i < 10; i++ {
		now = base.Add(time.Duration(i) * time.Minute)
		svc.executeScheduledDefinition(context.Background(), bad, &models.FlowVersion{}, &models.UserSession{ID: "system"})
		svc.executeScheduledDefinition(context.Background(), good, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	}

	// Healthy flow runs every tick (10), broken flow is throttled (≤ 4
	// within first 10 minutes per ladder).
	assert.Equal(t, 10, calls["GoodFlow"], "healthy flow must run every tick regardless of sibling backoff")
	assert.LessOrEqual(t, calls["BadFlow"], 4, "broken flow must back off independently")
	assert.Greater(t, calls["BadFlow"], 0, "broken flow must still attempt at least once")
}

// TestExecuteScheduledDefinition_LogsOncePerLadderTransition proves the
// stage-transition log line is emitted exactly once per ladder advance —
// a single broken flow must not produce one log per failed input.
func TestExecuteScheduledDefinition_LogsOncePerLadderTransition(t *testing.T) {
	var buf bytes.Buffer
	prevWriter := log.Writer()
	prevFlags := log.Flags()
	log.SetOutput(&buf)
	log.SetFlags(0)
	defer func() {
		log.SetOutput(prevWriter)
		log.SetFlags(prevFlags)
	}()

	inputs := []map[string]interface{}{
		{"$Record": models.SObject{"Id": "001"}},
		{"$Record": models.SObject{"Id": "002"}},
		{"$Record": models.SObject{"Id": "003"}},
	}
	base := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC)
	now := base
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return inputs, nil
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			return nil, errors.New("boom")
		},
		nowFn: func() time.Time { return now },
	}
	def := &models.FlowDefinition{Id: "def-log", DeveloperName: "LogTransitionFlow"}

	for i := 0; i < 30; i++ {
		now = base.Add(time.Duration(i) * time.Minute)
		svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	}

	transitions := strings.Count(buf.String(), "consecutive tick(s); next retry deferred")
	// 4 executed ticks × 1 transition each = 4 stage transitions in 30 min.
	assert.Equal(t, 4, transitions,
		"each ladder advance must log exactly once; got %d transition lines", transitions)
}

// TestExecuteScheduledDefinition_SoftFailureCountsAsFailure proves a flow
// that returns IsSuccess=false (no Go-level error) still engages the
// backoff ladder. Without this, validation-error storms inside the result
// payload would silently re-create the original retry-storm shape.
func TestExecuteScheduledDefinition_SoftFailureCountsAsFailure(t *testing.T) {
	calls := 0
	inputs := []map[string]interface{}{{"$Record": models.SObject{"Id": "001"}}}
	base := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC)
	now := base
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return inputs, nil
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			calls++
			return &models.FlowExecutionResult{IsSuccess: false, Errors: []models.FlowExecutionError{{Message: "soft fail"}}}, nil
		},
		nowFn: func() time.Time { return now },
	}
	def := &models.FlowDefinition{Id: "def-soft", DeveloperName: "SoftFailFlow"}

	for i := 0; i < 10; i++ {
		now = base.Add(time.Duration(i) * time.Minute)
		svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	}

	assert.LessOrEqual(t, calls, 4, "soft IsSuccess=false failures must still engage backoff (issue #2199)")
}

// TestExecuteScheduledDefinition_LoadFailureCountsAsFailure proves a
// persistently failing loadScheduledInputs (the early-return at the top of
// executeScheduledDefinition) also engages backoff — otherwise that whole
// failure mode escapes the protection.
func TestExecuteScheduledDefinition_LoadFailureCountsAsFailure(t *testing.T) {
	loadCalls := 0
	base := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC)
	now := base
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			loadCalls++
			return nil, errors.New("simulated load failure")
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			t.Fatal("executeFlowVersionFn should not be called when load fails")
			return nil, nil
		},
		nowFn: func() time.Time { return now },
	}
	def := &models.FlowDefinition{Id: "def-load", DeveloperName: "LoadFailFlow"}

	for i := 0; i < 10; i++ {
		now = base.Add(time.Duration(i) * time.Minute)
		svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	}

	assert.LessOrEqual(t, loadCalls, 4, "load failures must engage backoff just like exec failures")
}

// TestExecuteScheduledDefinition_EmptyInputsAreNeutral proves an empty
// inputSets result (no matching records) is treated as neutral: it neither
// advances nor resets the backoff ladder, since nothing was attempted.
func TestExecuteScheduledDefinition_EmptyInputsAreNeutral(t *testing.T) {
	executions := 0
	failNext := true
	base := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC)
	now := base
	emptyInputs := false
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			if emptyInputs {
				return []map[string]interface{}{}, nil
			}
			return []map[string]interface{}{{"$Record": models.SObject{"Id": "001"}}}, nil
		},
		executeFlowVersionFn: func(context.Context, *models.FlowVersion, string, map[string]interface{}, *models.UserSession) (*models.FlowExecutionResult, error) {
			executions++
			if failNext {
				return nil, errors.New("fail")
			}
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		nowFn: func() time.Time { return now },
	}
	def := &models.FlowDefinition{Id: "def-empty", DeveloperName: "EmptyInputsFlow"}

	// First two ticks fail with non-empty inputs → ladder reaches stage 2 (5m).
	for i := 0; i < 2; i++ {
		now = base.Add(time.Duration(i) * time.Minute)
		svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	}
	failingExecutions := executions

	// Next tick has empty inputs — should be NEUTRAL (no ladder change).
	emptyInputs = true
	now = base.Add(6 * time.Minute) // past 5m backoff
	svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	assert.Equal(t, failingExecutions, executions, "empty-inputs tick must not run executor")

	// Now go back to non-empty + failing. If the empty tick had reset the
	// state we'd see consec=1 (1m backoff); if it had advanced we'd see
	// consec=4 (1h backoff). NEUTRAL means consec=3 (15m backoff).
	emptyInputs = false
	failNext = true
	now = base.Add(7 * time.Minute) // past prev 5m backoff but still in scope
	svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{}, &models.UserSession{ID: "system"})
	// One more execution should have happened, advancing to stage 3 (15m).
	assert.Equal(t, failingExecutions+1, executions)

	// Verify the next-retry timer matches stage 3 (15 min) — proves empty
	// inputs were neutral, not a reset (would be 1m) or advance (would be 1h).
	svc.retryStateMu.Lock()
	state := svc.retryState["def-empty"]
	svc.retryStateMu.Unlock()
	if assert.NotNil(t, state) {
		assert.Equal(t, 3, state.consecutiveFailures, "empty inputs must be neutral; expected consec=3 after 2+1 failures")
	}
}

// TestExecuteScheduledDefinition_NewActiveVersionResetsBackoff is a
// regression test for the Stage-2 codex finding on PR #2211: when an admin
// fixes a broken Scheduled flow by activating a new FlowVersion, the
// scheduler's retry-backoff state — which is keyed by FlowDefinition only
// in the original fix — would suppress the corrected version for up to
// 24h. The Admin Self-Service path (admin metadata change → propagates
// without code/ops intervention) requires that the new version Id
// participates in the retry key so the corrected flow gets attempted on
// the next tick regardless of the prior version's failure history.
func TestExecuteScheduledDefinition_NewActiveVersionResetsBackoff(t *testing.T) {
	calls := 0
	failingVersionId := "ver-broken"
	currentVersionId := failingVersionId
	inputs := []map[string]interface{}{{"$Record": models.SObject{"Id": "001"}}}
	base := time.Date(2026, 4, 28, 12, 0, 0, 0, time.UTC)
	now := base
	svc := &SchedulerService{
		loadScheduledInputsFn: func(context.Context, *models.FlowVersion, *models.UserSession) ([]map[string]interface{}, error) {
			return inputs, nil
		},
		executeFlowVersionFn: func(_ context.Context, v *models.FlowVersion, _ string, _ map[string]interface{}, _ *models.UserSession) (*models.FlowExecutionResult, error) {
			calls++
			if v != nil && v.Id == failingVersionId {
				return nil, errors.New("simulated FlowVersion failure")
			}
			return &models.FlowExecutionResult{IsSuccess: true}, nil
		},
		nowFn: func() time.Time { return now },
	}
	def := &models.FlowDefinition{Id: "def-version-flip", DeveloperName: "VersionFlipFlow"}

	// Drive the ladder to its terminal stage (24h) by simulating many
	// consecutive failures on the broken version.
	for i := 0; i < 30; i++ {
		now = base.Add(time.Duration(i) * time.Minute)
		svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{Id: currentVersionId}, &models.UserSession{ID: "system"})
	}
	callsAtTerminalStage := calls
	assert.GreaterOrEqual(t, callsAtTerminalStage, 4, "ladder must have advanced past first stage")

	// Confirm the broken-version key is parked in 24h backoff.
	svc.retryStateMu.Lock()
	brokenState := svc.retryState["def-version-flip:"+failingVersionId]
	svc.retryStateMu.Unlock()
	if !assert.NotNil(t, brokenState, "broken-version retry state should exist") {
		t.FailNow()
	}
	// After 30 ticks at 1m intervals the ladder lands on stage 4 (1h
	// backoff): tick 0 → 1m, tick 1 → 5m, tick 6 → 15m, tick 21 → 1h.
	// Stage 5 (24h) needs another full hour, well outside the test window.
	// What matters is the backoff window is now ≥1h, more than long enough
	// to mask the version-flip recovery without this fix.
	assert.GreaterOrEqual(t, brokenState.consecutiveFailures, 4,
		"ladder should have advanced to ≥1h within 30 minutes of failures")
	assert.True(t, brokenState.nextRetryAt.After(now.Add(30*time.Minute)),
		"backoff window must extend well past 31 minutes so version-flip below is meaningful")

	// Admin activates a fixed version — only 1 minute later, well inside
	// the broken version's 24h backoff window.
	currentVersionId = "ver-fixed"
	now = base.Add(31 * time.Minute)
	svc.executeScheduledDefinition(context.Background(), def, &models.FlowVersion{Id: currentVersionId}, &models.UserSession{ID: "system"})

	// The fixed version must execute despite the broken version's backoff
	// being far from elapsed. Without the version in the retry key, this
	// tick would be skipped and the assertion fails.
	assert.Equal(t, callsAtTerminalStage+1, calls,
		"new active version must NOT inherit prior version's backoff window (issue #2199 follow-up)")

	// And the fixed version's success must be recorded under its own key,
	// not pollute the broken version's state.
	svc.retryStateMu.Lock()
	fixedState := svc.retryState["def-version-flip:ver-fixed"]
	stillBrokenState := svc.retryState["def-version-flip:"+failingVersionId]
	svc.retryStateMu.Unlock()
	if assert.NotNil(t, fixedState) {
		assert.Equal(t, 0, fixedState.consecutiveFailures, "fixed version should have a clean state")
	}
	assert.NotNil(t, stillBrokenState, "broken version's state should remain untouched (process-local debug aid)")
}

func TestNewSystemAutomationUser_IdentityOnly(t *testing.T) {
	user := newSystemAutomationUser()

	if assert.NotNil(t, user) {
		assert.Equal(t, constants.SystemUserID, user.ID)
		assert.Equal(t, constants.SystemSchedulerUserName, user.Name)
		if assert.NotNil(t, user.Email) {
			assert.Equal(t, constants.SystemUserEmail, *user.Email)
		}
		// The user session must NOT carry elevated permissions — security
		// bypass is on the context (SystemContext), not the user identity.
		assert.Empty(t, user.ProfileID, "system user should not have an admin profile — security comes from ctx")
		assert.False(t, user.IsSystemAdmin, "system user should not set IsSystemAdmin — security comes from ctx")
		assert.Empty(t, user.Scope, "Scope is enforced by HTTP middleware, not relevant for scheduler")
	}
}