diff --git a/cmd/scheduler.go b/cmd/scheduler.go new file mode 100644 index 0000000..94bbf04 --- /dev/null +++ b/cmd/scheduler.go @@ -0,0 +1,308 @@ +// Copyright (c) 2026 Jose Cely +// 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) +}