command scheduler

scheduler:processed List processed task executions
  scheduler:queue     List queued scheduler tasks
  scheduler:semaphore Get or set the scheduler semaphore state
  scheduler:truncate  Empty the scheduler queue and processed logs

cmd/scheduler.go

@@ -0,0 +1,308 @@
+// 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 cmd
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/joho/godotenv"
+	"github.com/spf13/cobra"
+	"gorm.io/gorm"
+)
+
+// ─── Helper ──────────────────────────────────────────────────────────────────
+
+// table names (must match core/scheduler/scheduler.go)
+const (
+	queueTable         = "queue_items"
+	processedTable     = "processed_items"
+	schedulerMetaTable = "scheduler_meta"
+	semaphoreKey       = "semaphore"
+	semaphoreDefault   = "green"
+)
+
+// loadEnvAndDB loads the env file for the given mode and returns a connected *gorm.DB.
+func loadEnvAndDB(envMode string) (*gorm.DB, error) {
+	var envFile string
+	switch envMode {
+	case "dev":
+		envFile = ".env-dev"
+	case "prod":
+		envFile = ".env"
+	default:
+		return nil, fmt.Errorf("environment must be 'dev' or 'prod', got '%s'", envMode)
+	}
+
+	envPath := filepath.Join(".", envFile)
+	if _, err := os.Stat(envPath); os.IsNotExist(err) {
+		return nil, fmt.Errorf("environment file '%s' not found in current directory", envFile)
+	}
+
+	if err := godotenv.Load(envPath); err != nil {
+		return nil, fmt.Errorf("error loading '%s': %v", envFile, err)
+	}
+
+	return connectDatabase()
+}
+
+// ─── scheduler:queue ─────────────────────────────────────────────────────────
+
+var SchedulerQueueCmd = &cobra.Command{
+	Use:   "scheduler:queue [dev|prod]",
+	Short: "List queued scheduler tasks",
+	Long: `Lists all pending and processing tasks in the scheduler queue.
+
+Example:
+  goffee scheduler:queue dev
+  goffee scheduler:queue prod
+`,
+	Args: cobra.ExactArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		db, err := loadEnvAndDB(args[0])
+		if err != nil {
+			fmt.Printf("Error: %v\n", err)
+			return
+		}
+
+		type QueueItem struct {
+			ID        uint      `gorm:"column:id"`
+			TaskType  string    `gorm:"column:task_type"`
+			Payload   string    `gorm:"column:payload"`
+			Status    string    `gorm:"column:status"`
+			Priority  int       `gorm:"column:priority"`
+			MaxRuns   int       `gorm:"column:max_runs"`
+			Thread    int       `gorm:"column:thread"`
+			CreatedAt time.Time `gorm:"column:created_at"`
+		}
+
+		var items []QueueItem
+		result := db.Table(queueTable).
+			Where("status IN ?", []string{"pending", "processing"}).
+			Order("priority DESC, id ASC").
+			Find(&items)
+		if result.Error != nil {
+			fmt.Printf("Error querying queue: %v\n", result.Error)
+			return
+		}
+
+		if len(items) == 0 {
+			fmt.Println("Queue is empty.")
+			return
+		}
+
+		fmt.Printf("%-6s %-18s %-12s %-8s %-8s %-7s %-6s  %s\n",
+			"ID", "TYPE", "STATUS", "PRIORITY", "MAXRUNS", "THREAD", "CREATED", "PAYLOAD")
+		fmt.Println("------ ------------------ ------------ -------- -------- ------- ------  ------------------------------")
+		for _, it := range items {
+			trunc := it.Payload
+			if len(trunc) > 40 {
+				trunc = trunc[:37] + "..."
+			}
+			fmt.Printf("%-6d %-18s %-12s %-8d %-8d %-7d %-6s  %s\n",
+				it.ID, it.TaskType, it.Status, it.Priority, it.MaxRuns, it.Thread,
+				it.CreatedAt.Format("15:04:05"), trunc)
+		}
+		fmt.Printf("\nTotal: %d queued task(s)\n", len(items))
+	},
+}
+
+// ─── scheduler:processed ─────────────────────────────────────────────────────
+
+var SchedulerProcessedCmd = &cobra.Command{
+	Use:   "scheduler:processed [dev|prod] [limit]",
+	Short: "List processed task executions",
+	Long: `Lists the most recent processed task executions.
+
+The limit defaults to 10. Pass a number to change it.
+
+Example:
+  goffee scheduler:processed dev        # last 10
+  goffee scheduler:processed prod       # last 10
+  goffee scheduler:processed dev 100    # last 100
+  goffee scheduler:processed prod 50    # last 50
+`,
+	Args: cobra.RangeArgs(1, 2),
+	Run: func(cmd *cobra.Command, args []string) {
+		db, err := loadEnvAndDB(args[0])
+		if err != nil {
+			fmt.Printf("Error: %v\n", err)
+			return
+		}
+
+		limit := 10
+		if len(args) == 2 {
+			n, err := fmt.Sscanf(args[1], "%d", &limit)
+			if err != nil || n != 1 || limit < 1 {
+				fmt.Printf("Error: invalid limit '%s' — must be a positive integer\n", args[1])
+				return
+			}
+		}
+
+		type ProcessedItem struct {
+			ID          uint      `gorm:"column:id"`
+			QueueItemID uint      `gorm:"column:queue_item_id"`
+			TaskType    string    `gorm:"column:task_type"`
+			Status      string    `gorm:"column:status"`
+			ErrorMsg    string    `gorm:"column:error_msg"`
+			DurationMs  int64     `gorm:"column:duration_ms"`
+			StartedAt   time.Time `gorm:"column:started_at"`
+			FinishedAt  time.Time `gorm:"column:finished_at"`
+		}
+
+		var items []ProcessedItem
+		result := db.Table(processedTable).
+			Order("id DESC").
+			Limit(limit).
+			Find(&items)
+		if result.Error != nil {
+			fmt.Printf("Error querying processed items: %v\n", result.Error)
+			return
+		}
+
+		if len(items) == 0 {
+			fmt.Println("No processed executions found.")
+			return
+		}
+
+		fmt.Printf("%-6s %-14s %-18s %-8s %-10s %-8s  %s\n",
+			"ID", "QUEUE_ITEM_ID", "TYPE", "STATUS", "DURATION", "STARTED", "ERROR")
+		fmt.Println("------ -------------- ------------------ -------- ---------- --------  ------------------------------")
+		for _, it := range items {
+			errMsg := it.ErrorMsg
+			if len(errMsg) > 35 {
+				errMsg = errMsg[:32] + "..."
+			}
+			fmt.Printf("%-6d %-14d %-18s %-8s %-10s %-8s  %s\n",
+				it.ID, it.QueueItemID, it.TaskType, it.Status,
+				fmt.Sprintf("%dms", it.DurationMs),
+				it.StartedAt.Format("15:04:05"), errMsg)
+		}
+	},
+}
+
+// ─── scheduler:semaphore ─────────────────────────────────────────────────────
+
+var SchedulerSemaphoreCmd = &cobra.Command{
+	Use:   "scheduler:semaphore [dev|prod] [green|red]",
+	Short: "Get or set the scheduler semaphore state",
+	Long: `View or change the scheduler semaphore state.
+
+Without a state argument, prints the current state.
+With "green" or "red", sets the semaphore accordingly.
+
+⚠ WARNING: changing the state only updates the database. The running
+application keeps the semaphore in memory and will NOT reflect the
+change until you restart it.
+
+Example:
+  goffee scheduler:semaphore dev        # show current state
+  goffee scheduler:semaphore dev red    # block after restart
+  goffee scheduler:semaphore prod green # allow after restart
+`,
+	Args: cobra.RangeArgs(1, 2),
+	Run: func(cmd *cobra.Command, args []string) {
+		db, err := loadEnvAndDB(args[0])
+		if err != nil {
+			fmt.Printf("Error: %v\n", err)
+			return
+		}
+
+		// Ensure the meta table exists.
+		_ = db.Table(schedulerMetaTable).AutoMigrate(&struct {
+			Key   string `gorm:"primaryKey;size:64"`
+			Value string `gorm:"size:16;not null"`
+		}{})
+
+		if len(args) == 2 {
+			state := args[1]
+			if state != "green" && state != "red" {
+				fmt.Printf("Error: state must be 'green' or 'red', got '%s'\n", state)
+				return
+			}
+			var count int64
+			db.Table(schedulerMetaTable).Where("key = ?", semaphoreKey).Count(&count)
+			if count == 0 {
+				_ = db.Table(schedulerMetaTable).Create(map[string]interface{}{
+					"key": semaphoreKey, "value": state,
+				})
+			} else {
+				_ = db.Table(schedulerMetaTable).
+					Where("key = ?", semaphoreKey).
+					Update("value", state)
+			}
+			fmt.Printf("Semaphore set to %s (restart the application for it to take effect)\n", state)
+		} else {
+			var value string
+			err := db.Table(schedulerMetaTable).
+				Where("key = ?", semaphoreKey).
+				Select("value").
+				Take(&value).Error
+			if err != nil {
+				fmt.Printf("Semaphore: %s (default)\n", semaphoreDefault)
+			} else {
+				fmt.Printf("Semaphore: %s\n", value)
+			}
+		}
+	},
+}
+
+// ─── scheduler:truncate ──────────────────────────────────────────────────────
+
+var SchedulerTruncateCmd = &cobra.Command{
+	Use:   "scheduler:truncate [dev|prod]",
+	Short: "Empty the scheduler queue and processed logs",
+	Long: `Deletes all rows from queue_items and processed_items tables.
+Use with caution — this action cannot be undone.
+
+Example:
+  goffee scheduler:truncate dev
+  goffee scheduler:truncate prod
+`,
+	Args: cobra.ExactArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		db, err := loadEnvAndDB(args[0])
+		if err != nil {
+			fmt.Printf("Error: %v\n", err)
+			return
+		}
+
+		// Delete all processed items first (child table), then queue items.
+		procResult := db.Exec(fmt.Sprintf("DELETE FROM %s", processedTable))
+		if procResult.Error != nil {
+			fmt.Printf("Error truncating processed_items: %v\n", procResult.Error)
+			return
+		}
+
+		queueResult := db.Exec(fmt.Sprintf("DELETE FROM %s", queueTable))
+		if queueResult.Error != nil {
+			fmt.Printf("Error truncating queue_items: %v\n", queueResult.Error)
+			return
+		}
+
+		// Reset semaphore to green (default).
+		var count int64
+		db.Table(schedulerMetaTable).Where("key = ?", semaphoreKey).Count(&count)
+		if count > 0 {
+			db.Table(schedulerMetaTable).
+				Where("key = ?", semaphoreKey).
+				Update("value", semaphoreDefault)
+		}
+
+		fmt.Printf("Truncated %d processed item(s) and %d queue item(s). Semaphore reset to %s.\n",
+			procResult.RowsAffected, queueResult.RowsAffected, semaphoreDefault)
+	},
+}
+
+// ─── Registration ────────────────────────────────────────────────────────────
+
+func init() {
+	rootCmd.AddCommand(SchedulerQueueCmd)
+	rootCmd.AddCommand(SchedulerProcessedCmd)
+	rootCmd.AddCommand(SchedulerSemaphoreCmd)
+	rootCmd.AddCommand(SchedulerTruncateCmd)
+}