Scheduling

// cronParser accepts standard 5-field expressions (minute hour dom month dow).
var cronParser = cron.NewParser(
	cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow,
)

// NextRun returns the next fire time after "from". Shared with the API so schedule
// edits compute next_run_at the same way the scheduler does.
func NextRun(expr string, from time.Time) (time.Time, error) {
	sched, err := cronParser.Parse(expr)
	if err != nil { return time.Time{}, err }
	return sched.Next(from), nil
}

func Valid(expr string) bool { _, err := cronParser.Parse(expr); return err == nil }

// NextRuns returns the next n fire times — powers the builder's live preview.
func NextRuns(expr string, from time.Time, n int) ([]time.Time, error) { /* loops sched.Next */ }

if !scheduler.Valid(req.Cron) {
	s.fail(w, http.StatusBadRequest, errors.New("invalid cron expression (expected 5 fields: minute hour day month weekday)"))
	return
}
next := pgtype.Timestamptz{} // NULL when disabled → never selected by ListDueSchedules
if req.Enabled {
	t, err := scheduler.NextRun(req.Cron, time.Now())
	if err != nil { s.fail(w, http.StatusBadRequest, err); return }
	next = pgtype.Timestamptz{Time: t, Valid: true}
}
s.q.UpsertSchedule(r.Context(), sqlc.UpsertScheduleParams{
	PipelineID: id, Cron: req.Cron, Enabled: req.Enabled, DryRun: req.DryRun, NextRunAt: next, WorkspaceID: ws,
})

func (s *Scheduler) Run(ctx context.Context) {
	t := time.NewTicker(s.interval) // 30s
	defer t.Stop()
	s.tick(ctx) // prompt initial tick: catch schedules that came due while down
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			s.tick(ctx)
		}
	}
}

// TickOnce runs a single pass on demand — the external heartbeat calls this so due
// schedules still fire when the periodic goroutine is paused. State is all in
// Postgres, so an on-demand tick and the periodic one are equivalent.
func (s *Scheduler) TickOnce(ctx context.Context) { s.tick(ctx) }

func (s *Scheduler) tick(ctx context.Context) {
	due, _ := s.q.ListDueSchedules(ctx) // next_run_at <= now
	now := time.Now()
	for _, sc := range due {
		next, err := NextRun(sc.Cron, now)
		if err != nil {
			// A bad expression should never fire forever — log and skip advancing.
			s.log.Error("scheduler: bad cron", "pipeline", sc.PipelineID, "cron", sc.Cron, "err", err)
			continue
		}
		fired, err := s.disp.FireSchedule(ctx, sc.PipelineID, sc.DryRun)
		// Advance the cadence REGARDLESS: a skipped or failed fire must not stall the
		// schedule; the next occurrence runs once the runtime is back.
		s.q.AdvanceSchedule(ctx, sqlc.AdvanceScheduleParams{PipelineID: sc.PipelineID, NextRunAt: /* next */})
		s.log.Info("scheduler: ticked", "pipeline", sc.PipelineID, "fired", fired, "next", next)
	}
}

// Dispatcher fires a due pipeline through the SAME placement-aware orchestration a
// manual run gets. fired=false (no run created) when the bound runtime is offline —
// the dispatcher notifies the owners instead of leaving a failed run for every tick.
type Dispatcher interface {
	FireSchedule(ctx context.Context, pipelineID uuid.UUID, dryRun bool) (fired bool, err error)
}

func (s *Server) FireSchedule(ctx context.Context, pipelineID uuid.UUID, dryRun bool) (bool, error) {
	pipe, _ := s.q.GetPipelineForWorker(ctx, pipelineID)
	src, _ := s.q.GetConnection(ctx, /* source */)
	dst, _ := s.q.GetConnection(ctx, /* dest */)
	placement := derivePlacement(src.Placement, /* … */, dst.Placement, /* … */)

	// Skip-when-offline: if the stage that runs FIRST on a runtime is offline,
	// don't create a run destined to fail — notify the owners and return fired=false.
	if !placement.Extract.Cloud {
		if reachable, name := s.agentReachable(ctx, ws, placement.Extract.AgentID); !reachable {
			s.notifyScheduleSkipped(ctx, ws, pipe, name)
			return false, nil
		}
	} else if !placement.Load.Cloud {
		if reachable, name := s.agentReachable(ctx, ws, placement.Load.AgentID); !reachable {
			s.notifyScheduleSkipped(ctx, ws, pipe, name)
			return false, nil
		}
	}

	run, _ := s.q.CreateRun(ctx, sqlc.CreateRunParams{PipelineID: pipelineID, DryRun: dryRun})
	return true, s.orchestrateRun(ctx, ws, pipe, run) // identical routing to a manual run
}

members, _ := s.q.ListWorkspaceMembers(ctx, ws)
body := fmt.Sprintf(
	"%q didn't run — its runtime %q is offline. It runs on the next schedule once the runtime is back.",
	pipe.Name, agentName,
)
for _, m := range members {
	if m.Role != "owner" { continue } // owners only
	s.notify(ctx, s.q, m.UserID, &ws, "schedule.skipped", "Scheduled run skipped", body,
		map[string]any{"pipelineId": pipe.ID.String()})
}

// POST /internal/scheduler/tick — bearer token from SETU_TICK_TOKEN.
func (s *Server) schedulerTick(w http.ResponseWriter, r *http.Request) {
	want := strings.TrimSpace(os.Getenv("SETU_TICK_TOKEN"))
	if want == "" {
		http.NotFound(w, r) // no token configured → endpoint doesn't exist
		return
	}
	got := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
	if subtle.ConstantTimeCompare([]byte(got), []byte(want)) != 1 {
		s.unauthenticated(w)
		return
	}
	scheduler.New(s.q, s, s.log).TickOnce(r.Context()) // one pass, same as the periodic tick
	w.WriteHeader(http.StatusNoContent)
}

on:
  schedule:
    - cron: "*/5 * * * *" # every 5 min (GitHub's minimum granularity)
  workflow_dispatch: {}

concurrency:            # don't pile up ticks if a cold-start run is still going
  group: scheduler-heartbeat
  cancel-in-progress: false

jobs:
  tick:
    runs-on: ubuntu-latest
    steps:
      - name: Ping scheduler tick
        env:
          TICK_TOKEN: ${{ secrets.SETU_TICK_TOKEN }}
          API_URL: ${{ vars.SETU_API_URL }}
        run: |
          url="${API_URL:-https://setu-api-f776.onrender.com}"
          curl -fsS -X POST "$url/internal/scheduler/tick" \
            -H "Authorization: Bearer ${TICK_TOKEN}" \
            --max-time 120 --retry 4 --retry-all-errors --retry-delay 10 \
            -o /dev/null -w "tick -> HTTP %{http_code}\n"