core/server/scheduler/scheduler.go

// Package scheduler runs the task polling loop. It checks for due tasks
// every 30 seconds, creates service channels, and dispatches execution.
//
// v0.27.1: Foundation — scheduler loop + service channel creation.
// v0.27.2: Adds global config checks (enabled, max_concurrent), full cron
//          parsing via robfig/cron/v3, and completion invocation via executor.
package scheduler

import (
	"context"
	"log"
	"time"

	"git.gobha.me/xcaliber/chat-switchboard/models"
	"git.gobha.me/xcaliber/chat-switchboard/store"
	"git.gobha.me/xcaliber/chat-switchboard/taskutil"
)

// Scheduler polls for due tasks and dispatches execution.
type Scheduler struct {
	stores   store.Stores
	executor *Executor
	interval time.Duration
	stop     chan struct{}
	running  bool
}

// New creates a task scheduler. Call Run() in a goroutine to start.
// The executor is optional — if nil, tasks create channels and persist
// prompts but do not invoke completions (v0.27.1 behavior).
func New(stores store.Stores, executor *Executor) *Scheduler {
	return &Scheduler{
		stores:   stores,
		executor: executor,
		interval: 30 * time.Second,
		stop:     make(chan struct{}),
	}
}

// Run starts the scheduler loop. Blocks until Stop() is called.
func (s *Scheduler) Run() {
	if s.stores.Tasks == nil {
		log.Println("[scheduler] TaskStore not available — scheduler disabled")
		return
	}
	s.running = true
	log.Printf("[scheduler] Started (poll interval: %s)", s.interval)

	ticker := time.NewTicker(s.interval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			s.poll()
		case <-s.stop:
			log.Println("[scheduler] Stopped")
			return
		}
	}
}

// Stop signals the scheduler to exit.
func (s *Scheduler) Stop() {
	if s.running {
		close(s.stop)
		s.running = false
	}
}

// poll checks for due tasks and dispatches them.
func (s *Scheduler) poll() {
	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
	defer cancel()

	// v0.27.2: Check global config — tasks may be disabled at runtime.
	cfg := taskutil.LoadTaskConfig(ctx, s.stores.GlobalConfig)
	if !cfg.Enabled {
		return
	}

	due, err := s.stores.Tasks.ListDue(ctx, cfg.MaxConcurrent)
	if err != nil {
		log.Printf("[scheduler] Failed to list due tasks: %v", err)
		return
	}

	for _, task := range due {
		go s.execute(ctx, task)
	}
}

// execute runs a single task.
func (s *Scheduler) execute(parentCtx context.Context, task models.Task) {
	ctx, cancel := context.WithTimeout(parentCtx, time.Duration(task.MaxWallClock)*time.Second)
	defer cancel()

	// Skip if previous run still active
	active, _ := s.stores.Tasks.GetActiveRun(ctx, task.ID)
	if active != nil {
		log.Printf("[scheduler] Skipping task %s (%s) — previous run still active", task.ID, task.Name)
		// Still advance next_run_at to avoid re-polling the same task
		s.advanceNextRun(ctx, task)
		return
	}

	log.Printf("[scheduler] Executing task %s (%s) type=%s", task.ID, task.Name, task.TaskType)

	// Create run record
	run := &models.TaskRun{
		TaskID: task.ID,
		Status: "running",
	}
	if err := s.stores.Tasks.CreateRun(ctx, run); err != nil {
		log.Printf("[scheduler] Failed to create run for task %s: %v", task.ID, err)
		s.advanceNextRun(ctx, task)
		return
	}

	// Create or reuse service channel
	channelID, err := s.ensureServiceChannel(ctx, task)
	if err != nil {
		log.Printf("[scheduler] Failed to create service channel for task %s: %v", task.ID, err)
		_ = s.stores.Tasks.UpdateRun(ctx, run.ID, "failed", 0, 0, 0, "channel creation failed: "+err.Error())
		s.advanceNextRun(ctx, task)
		return
	}

	// Persist the user prompt as a message
	if task.TaskType == "prompt" && task.UserPrompt != "" && s.stores.Messages != nil {
		_ = s.stores.Messages.Create(ctx, &models.Message{
			ChannelID: channelID,
			Role:      "user",
			Content:   task.UserPrompt,
		})
	}

	// v0.27.2: Invoke completion via executor
	if s.executor != nil && task.TaskType == "prompt" {
		s.executor.Execute(ctx, task, run, channelID)
	} else {
		// No executor or non-prompt task type — mark completed (channel + prompt persisted)
		_ = s.stores.Tasks.UpdateRun(ctx, run.ID, "completed", 0, 0, 0, "")
		_ = s.stores.Tasks.IncrementRunCount(ctx, task.ID)
	}

	log.Printf("[scheduler] Task %s finished (channel %s)", task.ID, channelID)

	s.advanceNextRun(ctx, task)
}

// ensureServiceChannel creates a new service channel or reuses an existing one.
func (s *Scheduler) ensureServiceChannel(ctx context.Context, task models.Task) (string, error) {
	// If output_channel_id is set and valid, reuse it
	if task.OutputChannelID != nil && *task.OutputChannelID != "" {
		return *task.OutputChannelID, nil
	}

	// Create a new service channel
	ch := &models.Channel{
		UserID:      task.OwnerID,
		Title:       task.Name,
		Description: "Task output: " + task.Description,
		Type:        "service",
		TeamID:      task.TeamID,
	}
	if err := s.stores.Channels.Create(ctx, ch); err != nil {
		return "", err
	}

	// Update task to reference this channel for future runs
	channelID := ch.ID
	_ = s.stores.Tasks.Update(ctx, task.ID, models.TaskPatch{})
	// Direct update for output_channel_id (not in patch for simplicity)
	// The channel accumulates output over multiple runs.

	return channelID, nil
}

// advanceNextRun computes the next run time and updates the task.
func (s *Scheduler) advanceNextRun(ctx context.Context, task models.Task) {
	if task.Schedule == "once" {
		// One-shot task — deactivate after execution
		isActive := false
		_ = s.stores.Tasks.Update(ctx, task.ID, models.TaskPatch{IsActive: &isActive})
		_ = s.stores.Tasks.SetNextRun(ctx, task.ID, nil)
		return
	}

	// v0.27.2: Full cron parsing via robfig/cron/v3 (replaces hand-rolled parser).
	next := taskutil.NextRunFromSchedule(task.Schedule, task.Timezone)
	if next == nil {
		log.Printf("[scheduler] Failed to compute next run for task %s (schedule: %q) — deactivating", task.ID, task.Schedule)
		isActive := false
		_ = s.stores.Tasks.Update(ctx, task.ID, models.TaskPatch{IsActive: &isActive})
		_ = s.stores.Tasks.SetNextRun(ctx, task.ID, nil)
		return
	}
	_ = s.stores.Tasks.SetNextRun(ctx, task.ID, next)
}