package scheduler import ( "context" "testing" "time" "github.com/alexei/tinyforge/internal/store" "github.com/alexei/tinyforge/internal/workload/plugin" ) // newTestStore opens an in-memory SQLite store. Each test gets its own // DSN so parallel runs do not collide on shared cache databases. func newTestStore(t *testing.T) *store.Store { t.Helper() st, err := store.New(":memory:") if err != nil { t.Fatalf("open store: %v", err) } t.Cleanup(func() { _ = st.Close() }) return st } func seedScheduleTrigger(t *testing.T, st *store.Store, name, interval, lastFired string) store.Trigger { t.Helper() trg, err := st.CreateTrigger(store.Trigger{ Kind: "schedule", Name: name, Config: `{"interval":"` + interval + `"}`, LastFiredAt: lastFired, }) if err != nil { t.Fatalf("CreateTrigger: %v", err) } return trg } func TestShouldFire(t *testing.T) { st := newTestStore(t) now := time.Date(2026, 5, 16, 12, 0, 0, 0, time.UTC) s := New(st, func(context.Context, store.Trigger, plugin.InboundEvent) error { return nil }, 0) cases := []struct { name string interval string lastFired string want bool }{ {"never fired fires", "1h", "", true}, {"window not yet elapsed", "1h", now.Add(-30 * time.Minute).Format(time.RFC3339), false}, {"window exactly elapsed fires", "1h", now.Add(-1 * time.Hour).Format(time.RFC3339), true}, {"window long elapsed fires", "24h", now.Add(-48 * time.Hour).Format(time.RFC3339), true}, {"bad interval suppressed", "banana", "", false}, {"bad last_fired_at treated as never", "1h", "not-a-timestamp", true}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { trg := store.Trigger{ Config: `{"interval":"` + tc.interval + `"}`, LastFiredAt: tc.lastFired, } got := s.shouldFire(trg, now) if got != tc.want { t.Fatalf("shouldFire = %v, want %v", got, tc.want) } }) } } func TestTickOnce_FiresOverdueTriggers(t *testing.T) { st := newTestStore(t) now := time.Date(2026, 5, 16, 12, 0, 0, 0, time.UTC) // Three triggers: one overdue, one not yet due, one never-fired. overdue := seedScheduleTrigger(t, st, "overdue", "1h", now.Add(-2*time.Hour).Format(time.RFC3339)) notDue := seedScheduleTrigger(t, st, "notdue", "1h", now.Add(-30*time.Minute).Format(time.RFC3339)) never := seedScheduleTrigger(t, st, "never", "1h", "") fired := make(map[string]int) s := New(st, func(_ context.Context, trg store.Trigger, _ plugin.InboundEvent) error { fired[trg.Name]++ return nil }, 0) s.clock = func() time.Time { return now } s.TickOnce(context.Background()) if fired["overdue"] != 1 { t.Errorf("overdue should fire once, got %d", fired["overdue"]) } if fired["notdue"] != 0 { t.Errorf("notdue should not fire, got %d", fired["notdue"]) } if fired["never"] != 1 { t.Errorf("never should fire once on first tick, got %d", fired["never"]) } // last_fired_at must advance for everyone we dispatched. for _, id := range []string{overdue.ID, never.ID} { row, err := st.GetTriggerByID(id) if err != nil { t.Fatalf("GetTriggerByID(%s): %v", id, err) } if row.LastFiredAt == "" { t.Errorf("last_fired_at not persisted for %s", row.Name) } } // not-due trigger's last_fired_at must NOT have changed. row, err := st.GetTriggerByID(notDue.ID) if err != nil { t.Fatalf("GetTriggerByID(notdue): %v", err) } if row.LastFiredAt != notDue.LastFiredAt { t.Errorf("notdue last_fired_at changed: was %q now %q", notDue.LastFiredAt, row.LastFiredAt) } } func TestTickOnce_DispatchErrorDoesNotWedgeOthers(t *testing.T) { st := newTestStore(t) now := time.Date(2026, 5, 16, 12, 0, 0, 0, time.UTC) broken := seedScheduleTrigger(t, st, "broken", "1h", "") seedScheduleTrigger(t, st, "healthy", "1h", "") fired := map[string]int{} s := New(st, func(_ context.Context, trg store.Trigger, _ plugin.InboundEvent) error { fired[trg.Name]++ if trg.Name == "broken" { return context.Canceled } return nil }, 0) s.clock = func() time.Time { return now } s.TickOnce(context.Background()) if fired["broken"] != 1 { t.Errorf("broken should be attempted once, got %d", fired["broken"]) } if fired["healthy"] != 1 { t.Errorf("healthy should fire once, got %d", fired["healthy"]) } // Core persist-before-dispatch invariant: even though the broken // trigger's dispatcher returned an error, last_fired_at must have // advanced. Otherwise the scheduler would re-fire it on every tick. row, err := st.GetTriggerByID(broken.ID) if err != nil { t.Fatalf("GetTriggerByID(broken): %v", err) } if row.LastFiredAt == "" { t.Fatalf("broken trigger last_fired_at must advance even on dispatch error") } // And: a second TickOnce at the same `now` must not re-fire broken. s.TickOnce(context.Background()) if fired["broken"] != 1 { t.Errorf("broken refired after persist; got %d (want 1)", fired["broken"]) } } func TestTickOnce_PersistsLastFiredBeforeDispatch(t *testing.T) { // Documented behavior: last_fired_at is persisted before the // dispatcher runs so a panicking match cannot wedge a tight loop. st := newTestStore(t) now := time.Date(2026, 5, 16, 12, 0, 0, 0, time.UTC) trg := seedScheduleTrigger(t, st, "tick", "1h", "") dispatched := false s := New(st, func(_ context.Context, t store.Trigger, _ plugin.InboundEvent) error { // At dispatch time the column must already be set. row, err := st.GetTriggerByID(t.ID) if err != nil { return err } dispatched = row.LastFiredAt != "" return nil }, 0) s.clock = func() time.Time { return now } s.TickOnce(context.Background()) if !dispatched { t.Fatalf("last_fired_at must be persisted before dispatcher runs") } row, err := st.GetTriggerByID(trg.ID) if err != nil { t.Fatalf("get: %v", err) } if row.LastFiredAt != now.Format(time.RFC3339) { t.Errorf("last_fired_at = %q, want %q", row.LastFiredAt, now.Format(time.RFC3339)) } } func TestLifecycle_StartStopIdempotent(t *testing.T) { // Start + Stop are wrapped in sync.Once. A second call must be a // no-op (no panic on double-cancel, no goroutine leak from double- // Start). This guards the shutdown path that runs Stop from both // defer and the signal-handler block in cmd/server/main.go. st := newTestStore(t) noop := func(context.Context, store.Trigger, plugin.InboundEvent) error { return nil } s := New(st, noop, 100*time.Millisecond) s.Start(context.Background()) s.Start(context.Background()) // second call: no goroutine spawned s.Stop() s.Stop() // second call: no panic on closing already-cancelled context } func TestNew_ClampsInterval(t *testing.T) { st := newTestStore(t) noop := func(context.Context, store.Trigger, plugin.InboundEvent) error { return nil } if got := New(st, noop, 0).tickInterval; got != 30*time.Second { t.Errorf("default = %s, want 30s", got) } if got := New(st, noop, 1*time.Hour).tickInterval; got != 5*time.Minute { t.Errorf("clamped = %s, want 5m", got) } if got := New(st, noop, 2*time.Minute).tickInterval; got != 2*time.Minute { t.Errorf("passthrough = %s, want 2m", got) } }