core/scheduler.go

// Copyright (c) 2026 Jose Cely <me@jacs.guru>
// Use of this source code is governed by MIT-style
// license that can be found in the LICENSE file.

package core

import (
	"context"
	"fmt"
	"log"
	"time"

	"git.smarteching.com/goffee/core/logger"
	"git.smarteching.com/goffee/core/scheduler"
)

// Default values for SchedulerConfig
var DefaultSchedulerInterval = 10 * time.Second
var DefaultSchedulerRateLimit = 10

// globalSchedulerMux holds the active Schedulermux instance so controllers
// can access it without requiring a direct import chain.
var globalSchedulerMux *Schedulermux

// Schedulermux wraps the scheduler Registry and manages the lifecycle
// of a simple database-backed task scheduler.
type Schedulermux struct {
	registry  *scheduler.Registry
	store     *scheduler.Store
	semaphore *scheduler.PersistedSemaphore
	sched     *scheduler.Scheduler
}

// SchedulerInit initializes the scheduler internals (registry).
// The store must be set separately via SetStore before starting.
// The semaphore is created lazily during RunScheduler once the DB is available.
func (s *Schedulermux) SchedulerInit() {
	s.registry = scheduler.NewRegistry()
	globalSchedulerMux = s
}

// SetStore sets the store (backed by *gorm.DB) on the scheduler.
// This also runs AutoMigrate for the queue_items and processed_items tables.
func (s *Schedulermux) SetStore(st *scheduler.Store) {
	s.store = st
}

// AddWork registers a task handler function for a given task type.
func (s *Schedulermux) AddWork(taskType string, handler scheduler.HandlerFunc) {
	s.registry.Register(taskType, handler)
}

// GetStore returns the scheduler store, or nil if not initialized.
func (s *Schedulermux) GetStore() *scheduler.Store {
	return s.store
}

// RunScheduler starts the scheduler loop with the given configuration.
// It auto-migrates the scheduler_meta table, recovers the persisted semaphore
// state, creates the in-memory semaphore, and launches the scheduler goroutine.
func (s *Schedulermux) RunScheduler(config SchedulerConfig) {
	if s.store == nil {
		log.Fatal("scheduler: store is not set — call SetStore before RunScheduler")
	}

	interval := config.SchedulerInterval
	if interval <= 0 {
		interval = DefaultSchedulerInterval
	}

	rateLimit := config.SchedulerRateLimit
	if rateLimit < 0 {
		rateLimit = DefaultSchedulerRateLimit
	}

	ctx := context.Background()

	// Auto-migrate the scheduler_meta table for semaphore persistence.
	_ = s.store.DB().AutoMigrate(&scheduler.SchedulerMeta{})

	// Recover the persisted semaphore state (defaults to green).
	wasGreen := s.store.GetSemaphore(ctx)
	s.semaphore = scheduler.NewPersistedSemaphore(s.store, wasGreen)
	logger.ResolveLogger().Info(fmt.Sprintf("[scheduler] semaphore recovered: %s", map[bool]string{true: "green", false: "red"}[wasGreen]))

	// Recover stuck items from a previous crash before starting.
	recovered, err := s.store.RecoverStuck(ctx)
	if err != nil {
		logger.ResolveLogger().Error(fmt.Sprintf("[scheduler] recover stuck error: %v", err))
	} else if recovered > 0 {
		logger.ResolveLogger().Info(fmt.Sprintf("[scheduler] recovered %d stuck task(s) back to pending", recovered))
	}

	s.sched = scheduler.NewScheduler(
		scheduler.SchedulerConfig{
			Interval:  interval,
			RateLimit: rateLimit,
		},
		s.semaphore,
		s.store,
		s.registry,
	)

	s.sched.Start(ctx)
}

// StopScheduler gracefully stops the scheduler loop.
func (s *Schedulermux) StopScheduler() {
	if s.sched != nil {
		s.sched.Stop()
	}
}

// SemaphoreSetGreen allows task execution to proceed.
func (s *Schedulermux) SemaphoreSetGreen() {
	if s.semaphore != nil {
		s.semaphore.SetGreen()
	}
}

// SemaphoreSetRed blocks task execution.
func (s *Schedulermux) SemaphoreSetRed() {
	if s.semaphore != nil {
		s.semaphore.SetRed()
	}
}

// ResolveSchedulerStore returns the store from the global scheduler mux instance.
// Returns nil if the scheduler was never initialized.
func ResolveSchedulerStore() *scheduler.Store {
	if globalSchedulerMux == nil {
		return nil
	}
	return globalSchedulerMux.GetStore()
}

// SchedulerSemaphoreSetGreen enables task execution by setting the semaphore to green.
func SchedulerSemaphoreSetGreen() {
	if globalSchedulerMux != nil {
		globalSchedulerMux.SemaphoreSetGreen()
	}
}

// SchedulerSemaphoreSetRed blocks task execution by setting the semaphore to red.
func SchedulerSemaphoreSetRed() {
	if globalSchedulerMux != nil {
		globalSchedulerMux.SemaphoreSetRed()
	}
}

// SchedulerSemaphoreIsGreen returns whether the scheduler semaphore is currently green.
func SchedulerSemaphoreIsGreen() bool {
	if globalSchedulerMux == nil || globalSchedulerMux.semaphore == nil {
		return false
	}
	return globalSchedulerMux.semaphore.IsGreen()
}