feat: add tenant status check to scheduler executeJob (#2152)

* feat: add tenant status check to scheduler executeJob

* fix: apply tenant status check in catch-up path and strengthen empty-tenant test

* fix: correct catch-up skipped record timestamp and executedCount tracking

* test: add catch-up path coverage for tenant status checker

---------

Co-authored-by: Ben Coombs <bjcoombs@users.noreply.github.com>

Author: bjcoombs

shared/platform/scheduler/catchup.go

@@ -115,8 +115,8 @@ func (s *CronScheduler) catchUpSchedule(ctx context.Context, sched Schedule, now
 			// Too old to execute: record as MISSED for audit trail.
 			s.recordMissedWindow(schedCtx, sched, nextTime)
 			missedCount++
-		} else {
-			// Within catch-up age: execute.
+		} else if s.catchUpWindowEligible(schedCtx, sched, nextTime) {
+			// Within catch-up age and tenant is active: execute.
 			s.executeCatchUpWindow(schedCtx, sched, nextTime)
 			executedCount++
 		}
@@ -140,6 +140,45 @@ func (s *CronScheduler) catchUpSchedule(ctx context.Context, sched Schedule, now
 // itself blocks until catch-up completes or the context is cancelled, so the
 // lifecycle already knows work is in progress. Context cancellation (from Stop)
 // is checked between iterations in catchUpSchedule.
+// catchUpWindowEligible checks tenant status before executing a catch-up window.
+// Unlike tenantIsEligible used in executeJob, this records the skipped execution
+// with the actual catch-up window timestamp rather than time.Now().
+func (s *CronScheduler) catchUpWindowEligible(ctx context.Context, sched Schedule, scheduledAt time.Time) bool {
+	if s.statusChecker == nil || sched.TenantID == "" {
+		return true
+	}
+	active, err := s.statusChecker.IsActive(ctx, sched.TenantID)
+	if err != nil {
+		s.logger.Error("failed to check tenant status for catch-up",
+			"schedule_id", sched.ID,
+			"error", err)
+		return true // fail open
+	}
+	if !active {
+		s.logger.Info("skipping catch-up window for inactive tenant",
+			"schedule_id", sched.ID,
+			"tenant_id", sched.TenantID,
+			"scheduled_at", scheduledAt)
+		if s.store != nil {
+			exec := Execution{
+				ID:            uuid.New(),
+				SchedulerName: s.config.Name,
+				ScheduleID:    sched.ID,
+				ScheduledAt:   scheduledAt,
+				Status:        ExecutionStatusSkipped,
+				ErrorMessage:  strPtr("tenant not active"),
+			}
+			if err := s.store.RecordExecution(ctx, exec); err != nil {
+				s.logger.Error("failed to record catch-up skipped execution",
+					"schedule_id", sched.ID,
+					"error", err)
+			}
+		}
+		return false
+	}
+	return true
+}
+
 func (s *CronScheduler) executeCatchUpWindow(ctx context.Context, sched Schedule, scheduledAt time.Time) {
 	// Acquire per-schedule lock (consistent with normal executeJob).
 	if s.lock != nil {

shared/platform/scheduler/catchup_test.go

@@ -2,6 +2,7 @@ package scheduler_test
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"sync"
 	"testing"
@@ -14,6 +15,8 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+var errStatusUnavailable = errors.New("status service unavailable")
+
 // secondsParser is a cron.Parser matching the seconds-level cron runner used in tests.
 var secondsParser = cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
 
@@ -412,6 +415,223 @@ func TestCatchUp_MultipleSchedules_IndependentCatchUp(t *testing.T) {
 	s.Stop()
 }
 
+func TestCatchUp_TenantStatusChecker_ActiveTenant_CatchUpExecutes(t *testing.T) {
+	cronExpr := "0 */10 * * * *"
+	now := time.Now().UTC()
+	lastExecTime := now.Add(-30 * time.Minute)
+
+	store := &stubExecutionStore{}
+	_ = store.RecordExecution(context.Background(), scheduler.Execution{
+		SchedulerName: "test-scheduler",
+		ScheduleID:    "sched-1",
+		ScheduledAt:   lastExecTime,
+		Status:        scheduler.ExecutionStatusCompleted,
+	})
+
+	provider := &stubProvider{
+		schedules: []scheduler.Schedule{
+			{ID: "sched-1", CronExpr: cronExpr, TenantID: "tenant1"},
+		},
+	}
+	executor := &stubExecutor{executeCh: make(chan struct{}, 100)}
+	lock := &stubLock{acquired: true}
+	checker := &stubTenantStatusChecker{active: true}
+
+	expected := expectedWindowCount(t, cronExpr, lastExecTime, now)
+	require.Greater(t, expected, 0)
+
+	s := newTestScheduler(provider, executor, lock,
+		scheduler.CronSchedulerConfig{
+			Name:            "test-scheduler",
+			RefreshInterval: time.Hour,
+			ShutdownTimeout: 2 * time.Second,
+			MaxCatchUpAge:   time.Hour,
+		},
+		scheduler.WithCronExecutionStore(store),
+		scheduler.WithTenantStatusChecker(checker),
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		_ = s.Start(ctx)
+	}()
+
+	err := await.New().AtMost(5 * time.Second).PollInterval(50 * time.Millisecond).Until(func() bool {
+		return executor.callCount.Load() >= int32(expected)
+	})
+	require.NoError(t, err)
+
+	assert.GreaterOrEqual(t, int(executor.callCount.Load()), expected)
+
+	cancel()
+	s.Stop()
+}
+
+func TestCatchUp_TenantStatusChecker_InactiveTenant_CatchUpSkipped(t *testing.T) {
+	cronExpr := "0 */10 * * * *"
+	now := time.Now().UTC()
+	lastExecTime := now.Add(-30 * time.Minute)
+
+	store := &stubExecutionStore{}
+	_ = store.RecordExecution(context.Background(), scheduler.Execution{
+		SchedulerName: "test-scheduler",
+		ScheduleID:    "sched-1",
+		ScheduledAt:   lastExecTime,
+		Status:        scheduler.ExecutionStatusCompleted,
+	})
+
+	provider := &stubProvider{
+		schedules: []scheduler.Schedule{
+			{ID: "sched-1", CronExpr: cronExpr, TenantID: "inactive-tenant"},
+		},
+	}
+	executor := &stubExecutor{}
+	lock := &stubLock{acquired: true}
+	checker := &stubTenantStatusChecker{active: false}
+
+	expected := expectedWindowCount(t, cronExpr, lastExecTime, now)
+	require.Greater(t, expected, 0)
+
+	s := newTestScheduler(provider, executor, lock,
+		scheduler.CronSchedulerConfig{
+			Name:            "test-scheduler",
+			RefreshInterval: time.Hour,
+			ShutdownTimeout: 2 * time.Second,
+			MaxCatchUpAge:   time.Hour,
+		},
+		scheduler.WithCronExecutionStore(store),
+		scheduler.WithTenantStatusChecker(checker),
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		_ = s.Start(ctx)
+	}()
+
+	// Wait until all catch-up windows are recorded as SKIPPED (one per window)
+	err := await.New().AtMost(5 * time.Second).PollInterval(50 * time.Millisecond).Until(func() bool {
+		count := 0
+		for _, e := range store.getExecutions() {
+			if e.Status == scheduler.ExecutionStatusSkipped &&
+				e.ErrorMessage != nil && *e.ErrorMessage == "tenant not active" {
+				count++
+			}
+		}
+		return count >= expected
+	})
+	require.NoError(t, err)
+
+	assert.Equal(t, int32(0), executor.callCount.Load(),
+		"executor must not be called for inactive tenant during catch-up")
+
+	cancel()
+	s.Stop()
+}
+
+func TestCatchUp_TenantStatusChecker_CheckError_CatchUpProceeds(t *testing.T) {
+	cronExpr := "0 */10 * * * *"
+	now := time.Now().UTC()
+	lastExecTime := now.Add(-30 * time.Minute)
+
+	store := &stubExecutionStore{}
+	_ = store.RecordExecution(context.Background(), scheduler.Execution{
+		SchedulerName: "test-scheduler",
+		ScheduleID:    "sched-1",
+		ScheduledAt:   lastExecTime,
+		Status:        scheduler.ExecutionStatusCompleted,
+	})
+
+	provider := &stubProvider{
+		schedules: []scheduler.Schedule{
+			{ID: "sched-1", CronExpr: cronExpr, TenantID: "tenant1"},
+		},
+	}
+	executor := &stubExecutor{executeCh: make(chan struct{}, 100)}
+	lock := &stubLock{acquired: true}
+	checker := &stubTenantStatusChecker{err: errStatusUnavailable}
+
+	expected := expectedWindowCount(t, cronExpr, lastExecTime, now)
+	require.Greater(t, expected, 0)
+
+	s := newTestScheduler(provider, executor, lock,
+		scheduler.CronSchedulerConfig{
+			Name:            "test-scheduler",
+			RefreshInterval: time.Hour,
+			ShutdownTimeout: 2 * time.Second,
+			MaxCatchUpAge:   time.Hour,
+		},
+		scheduler.WithCronExecutionStore(store),
+		scheduler.WithTenantStatusChecker(checker),
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		_ = s.Start(ctx)
+	}()
+
+	// Fail open: catch-up proceeds despite status check error
+	err := await.New().AtMost(5 * time.Second).PollInterval(50 * time.Millisecond).Until(func() bool {
+		return executor.callCount.Load() >= int32(expected)
+	})
+	require.NoError(t, err)
+
+	assert.GreaterOrEqual(t, int(executor.callCount.Load()), expected)
+
+	cancel()
+	s.Stop()
+}
+
+func TestCatchUp_TenantStatusChecker_NoChecker_CatchUpProceeds(t *testing.T) {
+	cronExpr := "0 */10 * * * *"
+	now := time.Now().UTC()
+	lastExecTime := now.Add(-30 * time.Minute)
+
+	store := &stubExecutionStore{}
+	_ = store.RecordExecution(context.Background(), scheduler.Execution{
+		SchedulerName: "test-scheduler",
+		ScheduleID:    "sched-1",
+		ScheduledAt:   lastExecTime,
+		Status:        scheduler.ExecutionStatusCompleted,
+	})
+
+	provider := &stubProvider{
+		schedules: []scheduler.Schedule{
+			{ID: "sched-1", CronExpr: cronExpr, TenantID: "tenant1"},
+		},
+	}
+	executor := &stubExecutor{executeCh: make(chan struct{}, 100)}
+	lock := &stubLock{acquired: true}
+
+	expected := expectedWindowCount(t, cronExpr, lastExecTime, now)
+	require.Greater(t, expected, 0)
+
+	// No WithTenantStatusChecker option
+	s := newTestScheduler(provider, executor, lock,
+		scheduler.CronSchedulerConfig{
+			Name:            "test-scheduler",
+			RefreshInterval: time.Hour,
+			ShutdownTimeout: 2 * time.Second,
+			MaxCatchUpAge:   time.Hour,
+		},
+		scheduler.WithCronExecutionStore(store),
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		_ = s.Start(ctx)
+	}()
+
+	err := await.New().AtMost(5 * time.Second).PollInterval(50 * time.Millisecond).Until(func() bool {
+		return executor.callCount.Load() >= int32(expected)
+	})
+	require.NoError(t, err)
+
+	assert.GreaterOrEqual(t, int(executor.callCount.Load()), expected)
+
+	cancel()
+	s.Stop()
+}
+
 func TestCatchUp_LockNotAcquired_SkipsCatchUp(t *testing.T) {
 	// When the catch-up lock cannot be acquired, no catch-up should happen.
 	store := &stubExecutionStore{}