Scheduler

Goffee ships with a lightweight, database-backed scheduler. Unlike the task queue (which requires Redis and is built on Asynq), the scheduler stores its tasks directly in the database using GORM. This makes it a good fit when you want durable, in-database background jobs without running an extra service.

It is designed for autonomy: the scheduler runs in-process, polls for pending tasks, and can be toggled on or off at any time via a global "semaphore" kill switch that even persists across restarts.

How it works

• Tasks are stored as rows in the queue_items table.
• The scheduler polls the table on a fixed interval (a "tick").
• On each tick it claims a batch of pending tasks (highest priority first), marks them as processing, and runs the matching handler.
• The result of every execution is written to the immutable processed_items table.
• Tasks can run once, a fixed number of times, or repeat indefinitely.
• A single task can also run its repetitions sequentially or concurrently.
• Any external process can enqueue work simply by inserting a row into queue_items — no Go code required.

Task execution flow

enqueue -> queue_items (pending) -> tick claims batch -> processing -> handler -> processed_items -> delete or re-queue

Enabling the scheduler

The scheduler is disabled by default. Open config/scheduler.go and set EnableScheduler to true:

#file: config/scheduler.go

package config

import (
    "time"

    "git.smarteching.com/goffee/core/v2"
)

func GetSchedulerConfig() core.SchedulerConfig {
    return core.SchedulerConfig{
        // Enable the scheduler system
        EnableScheduler: true,

        // How often the scheduler polls the database for new tasks
        SchedulerInterval: 10 * time.Second,

        // Max tasks to process per tick (0 = unlimited)
        SchedulerRateLimit: 5,
    }
}

Note: the scheduler requires the database to be enabled (EnableGorm: true in config/gorm.go), because it uses the same GORM connection.

Registering task handlers

Task handlers are functions of type scheduler.HandlerFunc:

func(ctx context.Context, item *scheduler.QueueItem) error

Return a non-nil error to record the execution with an error status. Return nil to record it as ok.

Open register-scheduler.go and register your handlers with AddWork, then set the store and start the scheduler:

#file: register-scheduler.go

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"

    "git.smarteching.com/goffee/core/v2"
    "git.smarteching.com/goffee/core/v2/scheduler"
    "git.smarteching.com/goffee/[myapp]/config"
)

// handleSendEmail processes a "send_email" task.
func handleSendEmail(ctx context.Context, item *scheduler.QueueItem) error {
    var payload struct {
        To      string `json:"to"`
        Subject string `json:"subject"`
    }
    if err := json.Unmarshal([]byte(item.Payload), &payload); err != nil {
        return fmt.Errorf("invalid send_email payload: %w", err)
    }
    log.Printf("[scheduler] [send_email] -> to=%s subject=%q", payload.To, payload.Subject)
    time.Sleep(2000 * time.Millisecond) // simulate I/O
    return nil
}

func registerScheduler() {
    var sched = new(core.Schedulermux)
    sched.SchedulerInit()

    //########################################
    //#   scheduler task registration    #####
    //########################################

    // Register your scheduler task handlers here ...
    sched.AddWork("send_email", handleSendEmail)

    //########################################
    // Set the store (DB-backed) and start
    //########################################

    // Create the store using the application's GORM DB
    st, err := scheduler.NewStore(core.ResolveGorm())
    if err != nil {
        panic(fmt.Sprintf("scheduler: failed to create store: %v", err))
    }
    sched.SetStore(st)

    // Start scheduler server in background, DO NOT TOUCH
    go sched.RunScheduler(config.GetSchedulerConfig())
}

The scheduler is wired up automatically in main.go:

#file: main.go

if config.GetSchedulerConfig().EnableScheduler == true {
    registerScheduler()
}

Enqueuing tasks

From a controller, use the store exposed by core.ResolveSchedulerStore() and call Enqueue:

store.Enqueue(ctx, taskType string, payload string, priority int, maxRuns int, thread int)

Parameter	Meaning
taskType	The task type registered with AddWork
payload	A string (usually JSON) passed to the handler via item.Payload
priority	Higher runs first within a tick
maxRuns	1 = run once, N = run N times, -1 = repeat indefinitely
thread	0 = sequential per type, 1 = concurrent per type

Here is how a sample controller enqueues three tasks:

#file: controllers/schedsample.go

func SchedulerSample(c *core.Context) *core.Response {
    store := core.ResolveSchedulerStore()
    if store == nil {
        return c.Response.SetStatusCode(500).Json(`{"message": "scheduler store not available"}`)
    }

    ctx := context.Background()

    // A task that runs once, sequentially
    emailPayload, _ := json.Marshal(map[string]interface{}{
        "to":      "user@example.com",
        "subject": "Hello from the scheduler!",
    })
    item1, err := store.Enqueue(ctx, "send_email", string(emailPayload), 0, 1, 0)
    // ...

    // A task that runs 3 times, concurrently
    item2, err := store.Enqueue(ctx, "send_email", string(emailPayload), 0, 3, 1)
    // ...

    // A high-priority task that repeats indefinitely, sequentially
    item3, err := store.Enqueue(ctx, "send_email", string(emailPayload), 10, -1, 0)
    // ...

    return c.Response.Json(fmt.Sprintf(
        `{"message": "Enqueued 3 scheduler tasks", "tasks": [%d, %d, %d]}`,
        item1.ID, item2.ID, item3.ID,
    ))
}

Enqueuing from SQL

Because tasks live in the database, any program can enqueue work by inserting a row:

INSERT INTO queue_items (task_type, payload, priority, max_runs, thread)
VALUES ('send_email', '{"to":"user@example.com"}', 0, 1, 0);

The semaphore (kill switch)

The scheduler has a global semaphore that lets you pause and resume task execution without restarting the app:

• GREEN — tasks are executed.
• RED — the scheduler ticks but skips execution entirely.

The state is persisted in the scheduler_meta table, so it survives restarts. You can control it from code, from a controller, or from the CLI.

From a controller:

// Flip the semaphore state
if core.SchedulerSemaphoreIsGreen() {
    core.SchedulerSemaphoreSetRed()   // block execution
} else {
    core.SchedulerSemaphoreSetGreen() // allow execution
}

Managing the scheduler from the CLI

The goffee cli provides commands to inspect and manage the scheduler tables. They must be run from the project directory and operate on the database configured in the selected environment (dev uses .env-dev, prod uses .env):

goffee scheduler:queue     dev          # list pending/processing tasks
goffee scheduler:processed dev 20       # list the last 20 processed executions
goffee scheduler:semaphore dev          # show the semaphore state
goffee scheduler:semaphore dev red      # block execution (takes effect after restart)
goffee scheduler:truncate  dev          # empty the queue and processed logs

Database tables

The scheduler uses three tables, which are created automatically. If you prefer to control migrations yourself, add them to run-auto-migrations.go:

#file: run-auto-migrations.go

db.AutoMigrate(&scheduler.QueueItem{})
db.AutoMigrate(&scheduler.ProcessedItem{})
db.AutoMigrate(&scheduler.SchedulerMeta{})

• queue_items — pending and processing tasks.
• processed_items — an immutable log of every execution (status, error, duration).
• scheduler_meta — key/value metadata, currently used for the persisted semaphore state.

Reliability notes

• Atomic claiming — tasks are claimed with SELECT ... FOR UPDATE SKIP LOCKED, so multiple scheduler instances never double-execute the same task.
• Crash recovery — on startup, any tasks left in the processing state (e.g. after a crash) are reset to pending via RecoverStuck.
• No retries by default — a failed execution is recorded with an error status, but the task is not automatically retried unless it is configured with maxRuns > 1 or -1.