multica/server/internal/scheduler/plans_for_scope_test.go

package scheduler

import (
	"context"
	"strings"
	"sync/atomic"
	"testing"
	"time"
)

// TestJobSpecValidatePlansForScopeRelaxesCadence covers the relaxed
// JobSpec.validate() rule: when PlansForScope is set, Cadence is
// optional because the hook owns plan_time selection (used by the
// Autopilot schedule job, where each trigger's cron expression is
// arbitrary and does not fit a single Cadence grid).
func TestJobSpecValidatePlansForScopeRelaxesCadence(t *testing.T) {
	base := JobSpec{
		Name:              "hook_validate",
		RunTimeout:        time.Minute,
		StaleTimeout:      2 * time.Minute,
		HeartbeatInterval: 30 * time.Second,
		MaxAttempts:       1,
		Scopes:            StaticScopes(ScopeGlobal),
		Handler: func(ctx context.Context, in HandlerInput) (HandlerResult, error) {
			return HandlerResult{}, nil
		},
	}

	t.Run("cadence still required without hook", func(t *testing.T) {
		j := base
		err := j.validate()
		if err == nil {
			t.Fatalf("expected validate error when both Cadence and PlansForScope are unset")
		}
		if !strings.Contains(err.Error(), "cadence must be > 0") {
			t.Fatalf("expected cadence error, got %v", err)
		}
	})

	t.Run("cadence optional when hook is set", func(t *testing.T) {
		j := base
		j.PlansForScope = func(ctx context.Context, scope Scope, now time.Time, latest LatestPlanInfo) ([]time.Time, error) {
			return nil, nil
		}
		if err := j.validate(); err != nil {
			t.Fatalf("expected validate to pass with PlansForScope and no Cadence; got %v", err)
		}
	})

	t.Run("every_plan max_plans_per_tick still required without hook", func(t *testing.T) {
		j := base
		j.Cadence = 5 * time.Minute
		j.CatchUpMode = CatchUpEveryPlan
		err := j.validate()
		if err == nil || !strings.Contains(err.Error(), "max_plans_per_tick") {
			t.Fatalf("expected max_plans_per_tick error for every_plan without hook, got %v", err)
		}
	})

	t.Run("every_plan max_plans_per_tick optional with hook", func(t *testing.T) {
		j := base
		j.CatchUpMode = CatchUpEveryPlan // legal but ignored when hook is set
		j.PlansForScope = func(ctx context.Context, scope Scope, now time.Time, latest LatestPlanInfo) ([]time.Time, error) {
			return nil, nil
		}
		if err := j.validate(); err != nil {
			t.Fatalf("hook-driven jobs may leave max_plans_per_tick=0; got %v", err)
		}
	})

	t.Run("other invariants still fire", func(t *testing.T) {
		// RunTimeout > 0 must still be enforced even when the hook is set.
		j := base
		j.PlansForScope = func(ctx context.Context, scope Scope, now time.Time, latest LatestPlanInfo) ([]time.Time, error) {
			return nil, nil
		}
		j.RunTimeout = 0
		err := j.validate()
		if err == nil || !strings.Contains(err.Error(), "run_timeout") {
			t.Fatalf("expected run_timeout invariant to still fire with hook, got %v", err)
		}
	})
}

// TestManagerPlansForScopeHookDrivesPlans verifies end-to-end that a
// JobSpec.PlansForScope hook fully replaces the Cadence planner:
//
//   - The hook is invoked with a fresh LatestPlanInfo{Found:false} on
//     first tick, then with the prior plan on the second tick.
//   - Every plan_time the hook returns is claimed and finalised
//     SUCCESS by the manager — proof the hook is wired through the
//     same tryClaim/runClaimed lease path as Cadence-driven jobs.
//   - MaxPlansPerTick caps an over-eager hook without erroring.
//   - Cadence=0 is legal when PlansForScope is set.
//
// Lives in the integration tier (requires DATABASE_URL) because it
// exercises the manager's actual SQL primitives, not a stub.
func TestManagerPlansForScopeHookDrivesPlans(t *testing.T) {
	pool := integrationPool(t)
	job := newTestJobSpec(uniqueJobName(t, "hook_planner"))
	t.Cleanup(func() { cleanupExecutions(t, pool, job.Name) })

	ctx := context.Background()
	now, err := dbNow(ctx, pool)
	if err != nil {
		t.Fatalf("dbNow: %v", err)
	}

	// Hook-driven job: arbitrary plan_times, no Cadence.
	job.Cadence = 0
	job.CatchUpMode = CatchUpLatestOnly // ignored when hook is set, but exercise the path
	job.MaxPlansPerTick = 2             // safety cap; we'll return 3 from the hook below

	// Plan_times the hook will offer in tick 1. Deliberately NOT on a
	// uniform Cadence grid so the Cadence planner would never produce
	// them — proving the hook bypasses FloorPlan entirely.
	t0 := now.Add(-3 * time.Minute).Truncate(time.Second)
	t1 := now.Add(-2*time.Minute - 17*time.Second).Truncate(time.Second)
	t2 := now.Add(-1 * time.Minute).Truncate(time.Second)
	t3 := now.Add(-15 * time.Second).Truncate(time.Second) // beyond MaxPlansPerTick; should be dropped

	var hookCalls atomic.Int32
	var lastSeenFound atomic.Bool
	var lastSeenPlan atomic.Pointer[time.Time]

	job.PlansForScope = func(ctx context.Context, scope Scope, ts time.Time, latest LatestPlanInfo) ([]time.Time, error) {
		hookCalls.Add(1)
		lastSeenFound.Store(latest.Found)
		if latest.Found {
			pt := latest.PlanTime
			lastSeenPlan.Store(&pt)
		} else {
			lastSeenPlan.Store(nil)
		}
		// Realistic hook contract: return only plan_times strictly
		// after the most recent stored one. This is exactly what the
		// Autopilot scheduler will do — compute cron occurrences in
		// (latest.PlanTime, now]. We always *could* return everything
		// and rely on tryClaim's conflict-no-op for idempotency, but
		// returning fewer plans is cheaper and exercises the
		// LatestPlanInfo plumbing.
		all := []time.Time{t0, t1, t2, t3}
		var out []time.Time
		for _, p := range all {
			if !latest.Found || p.After(latest.PlanTime) {
				out = append(out, p)
			}
		}
		return out, nil
	}

	mgr := NewManager(pool, Options{RunnerID: "hook-planner-runner"})
	if err := mgr.Register(*job); err != nil {
		t.Fatalf("register hook-driven job: %v", err)
	}

	// Tick 1: no prior plan, hook returns all 4 plans, MaxPlansPerTick
	// truncates to {t0, t1}.
	if err := mgr.RunOnce(ctx); err != nil {
		t.Fatalf("runOnce tick 1: %v", err)
	}
	if hookCalls.Load() == 0 {
		t.Fatalf("hook was never called on tick 1")
	}
	if lastSeenFound.Load() {
		t.Fatalf("first tick should see LatestPlanInfo{Found:false}")
	}

	rows := dumpJobRows(t, pool, job.Name)
	if len(rows) != 2 {
		t.Fatalf("expected 2 plans claimed (MaxPlansPerTick=2), got %d: %+v", len(rows), rows)
	}
	for _, r := range rows {
		if r.Status != "SUCCESS" {
			t.Fatalf("hook-claimed plan should finish SUCCESS, got %q at %s", r.Status, r.PlanTime)
		}
	}
	wantPlans := []time.Time{t0.UTC(), t1.UTC()}
	for i, r := range rows {
		if !r.PlanTime.Equal(wantPlans[i]) {
			t.Fatalf("plan[%d] = %s; want %s (MaxPlansPerTick should preserve hook order)",
				i, r.PlanTime.Format(time.RFC3339Nano), wantPlans[i].Format(time.RFC3339Nano))
		}
	}

	// Tick 2: latest stored plan is t1. The hook now returns only the
	// plans strictly after t1 — i.e. {t2, t3}. MaxPlansPerTick=2 lets
	// both through this tick. The hook must see Found=true with the
	// latest plan_time (t1, picked by latestPlan ORDER BY plan_time
	// DESC).
	hookCalls.Store(0)
	if err := mgr.RunOnce(ctx); err != nil {
		t.Fatalf("runOnce tick 2: %v", err)
	}
	if hookCalls.Load() == 0 {
		t.Fatalf("hook was never called on tick 2")
	}
	if !lastSeenFound.Load() {
		t.Fatalf("second tick should see LatestPlanInfo{Found:true}")
	}
	if got := lastSeenPlan.Load(); got == nil || !got.Equal(t1.UTC()) {
		var gotStr string
		if got != nil {
			gotStr = got.Format(time.RFC3339Nano)
		}
		t.Fatalf("second tick's LatestPlanInfo.PlanTime should be the most recent stored plan; want %s got %s",
			t1.UTC().Format(time.RFC3339Nano), gotStr)
	}

	rows = dumpJobRows(t, pool, job.Name)
	if len(rows) != 4 {
		t.Fatalf("expected 4 plans after tick 2 (tick 1 left {t0,t1}, tick 2 adds {t2,t3}), got %d: %+v",
			len(rows), rows)
	}
	for _, r := range rows {
		if r.Status != "SUCCESS" {
			t.Fatalf("all plans should be SUCCESS after tick 2, got %q at %s", r.Status, r.PlanTime)
		}
	}
}

// TestManagerPlansForScopeHookEmptyIsNoOp covers the "nothing due"
// path: an empty plan slice must not error, must not block subsequent
// scopes, and must not produce any sys_cron_executions row.
func TestManagerPlansForScopeHookEmptyIsNoOp(t *testing.T) {
	pool := integrationPool(t)
	job := newTestJobSpec(uniqueJobName(t, "hook_empty"))
	t.Cleanup(func() { cleanupExecutions(t, pool, job.Name) })

	job.Cadence = 0
	var hookCalls atomic.Int32
	job.PlansForScope = func(ctx context.Context, scope Scope, now time.Time, latest LatestPlanInfo) ([]time.Time, error) {
		hookCalls.Add(1)
		return nil, nil
	}

	mgr := NewManager(pool, Options{RunnerID: "hook-empty-runner"})
	if err := mgr.Register(*job); err != nil {
		t.Fatalf("register: %v", err)
	}
	if err := mgr.RunOnce(context.Background()); err != nil {
		t.Fatalf("runOnce: %v", err)
	}
	if hookCalls.Load() == 0 {
		t.Fatalf("hook was never called for the registered job")
	}
	rows := dumpJobRows(t, pool, job.Name)
	if len(rows) != 0 {
		t.Fatalf("empty hook output must not create rows; got %d", len(rows))
	}
}