Merge pull request #1128 from trheyi/main

Implement job management functionality and update job model

job/data.go

@@ -0,0 +1,116 @@
+package job
+
+import (
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/kun/maps"
+)
+
+// ========================
+// Jobs methods
+// ========================
+
+// ListJobs list jobs
+func ListJobs(param model.QueryParam, page int, pagesize int) (maps.MapStrAny, error) {
+	return nil, nil
+}
+
+// GetActiveJobs get active jobs
+func GetActiveJobs() ([]*Job, error) {
+	return nil, nil
+}
+
+// CountJobs count jobs
+func CountJobs(param model.QueryParam) (int, error) {
+	return 0, nil
+}
+
+// SaveJob save job
+func SaveJob(job *Job) error {
+	return nil
+}
+
+// RemoveJobs remove jobs
+func RemoveJobs(ids []string) error {
+	return nil
+}
+
+// GetJob get job
+func GetJob(id string) (*Job, error) {
+	return nil, nil
+}
+
+// ========================
+// Categories methods
+// ========================
+
+// GetCategories get categories
+func GetCategories(param model.QueryParam) ([]*Category, error) {
+	return nil, nil
+}
+
+// CountCategories count categories
+func CountCategories(param model.QueryParam) (int, error) {
+	return 0, nil
+}
+
+// RemoveCategories remove categories
+func RemoveCategories(ids []string) error {
+	return nil
+}
+
+// SaveCategory save category
+func SaveCategory(category *Category) error {
+	return nil
+}
+
+// ========================
+// Logs methods
+// ========================
+
+// ListLogs get logs
+func ListLogs(id string, param model.QueryParam, page int, pagesize int) (maps.MapStrAny, error) {
+	return nil, nil
+}
+
+// SaveLog save log
+func SaveLog(log *Log) error {
+	return nil
+}
+
+// RemoveLogs remove logs
+func RemoveLogs(ids []string) error {
+	return nil
+}
+
+// ========================
+// Executions methods
+// ========================
+
+// GetExecutions get executions
+func GetExecutions(id string) ([]*Execution, error) {
+	return nil, nil
+}
+
+// CountExecutions count executions
+func CountExecutions(id string, param model.QueryParam) (int, error) {
+	return 0, nil
+}
+
+// RemoveExecutions remove executions
+func RemoveExecutions(ids []string) error {
+	return nil
+}
+
+// GetExecution get execution
+func GetExecution(id string, param model.QueryParam) (*Execution, error) {
+	return nil, nil
+}
+
+// ========================
+// Live progress methods
+// ========================
+
+// GetProgress get progress with callback
+func GetProgress(id string, cb func(progress *Progress)) (*Progress, error) {
+	return nil, nil
+}

job/execution.go

@@ -0,0 +1,63 @@
+package job
+
+// Add add a new execution to the job
+func (j *Job) Add(priority int, handler HandlerFunc) error {
+	return nil
+}
+
+// GetExecutions get executions
+func (j *Job) GetExecutions() ([]*Execution, error) {
+	return nil, nil
+}
+
+// GetExecution get execution
+func (j *Job) GetExecution(id string) (*Execution, error) {
+	return nil, nil
+}
+
+// Log log
+func (e *Execution) Log(level LogLevel, format string, args ...interface{}) error {
+	return nil
+}
+
+// Info info log
+func (e *Execution) Info(format string, args ...interface{}) error {
+	return e.Log(Info, format, args...)
+}
+
+// Debug debug log
+func (e *Execution) Debug(format string, args ...interface{}) error {
+	return e.Log(Debug, format, args...)
+}
+
+// Warn warn log
+func (e *Execution) Warn(format string, args ...interface{}) error {
+	return e.Log(Warn, format, args...)
+}
+
+// Error error log
+func (e *Execution) Error(format string, args ...interface{}) error {
+	return e.Log(Error, format, args...)
+}
+
+// Fatal fatal log
+func (e *Execution) Fatal(format string, args ...interface{}) error {
+	return e.Log(Fatal, format, args...)
+}
+
+// Panic panic log
+func (e *Execution) Panic(format string, args ...interface{}) error {
+	return e.Log(Panic, format, args...)
+}
+
+// Trace trace log
+func (e *Execution) Trace(format string, args ...interface{}) error {
+	return e.Log(Trace, format, args...)
+}
+
+// SetProgress set the progress
+func (e *Execution) SetProgress(progress int, message string) error {
+	p := e.Job.Progress()
+	p.Set(progress, message)
+	return nil
+}

job/interfaces.go

@@ -0,0 +1,6 @@
+package job
+
+// ProgressManager the progress manager
+type ProgressManager interface {
+	Set(progress int, message string) error
+}

job/job.go

@@ -1 +1,114 @@
 package job
+
+import jsoniter "github.com/json-iterator/go"
+
+// Once create a new job
+func Once(mode ModeType, data map[string]interface{}) (*Job, error) {
+	data["mode"] = mode
+	data["schedule_type"] = ScheduleTypeOnce
+	raw, err := jsoniter.Marshal(data)
+	if err != nil {
+		return nil, err
+	}
+	return makeJob(raw)
+}
+
+// Cron create a new job
+func Cron(mode ModeType, data map[string]interface{}, expression string) (*Job, error) {
+	data["mode"] = mode
+	data["schedule_type"] = ScheduleTypeCron
+	data["schedule_expression"] = expression
+	raw, err := jsoniter.Marshal(data)
+	if err != nil {
+		return nil, err
+	}
+	return makeJob(raw)
+}
+
+// Daemon create a new job
+func Daemon(mode ModeType, data map[string]interface{}) (*Job, error) {
+	data["mode"] = mode
+	data["schedule_type"] = ScheduleTypeDaemon
+	raw, err := jsoniter.Marshal(data)
+	if err != nil {
+		return nil, err
+	}
+	return makeJob(raw)
+}
+
+// Start start the job
+func (j *Job) Start() error {
+	return nil
+}
+
+// Cancel cancel the job
+func (j *Job) Cancel() error {
+	return nil
+}
+
+// SetData set the data of the job
+func (j *Job) SetData(data map[string]interface{}) *Job {
+	return j
+}
+
+// SetConfig set the config of the job
+func (j *Job) SetConfig(config map[string]interface{}) *Job {
+	j.Config = config
+	return j
+}
+
+// SetName set the name of the job
+func (j *Job) SetName(name string) *Job {
+	j.Name = name
+	return j
+}
+
+// SetDescription set the description of the job
+func (j *Job) SetDescription(description string) *Job {
+	j.Description = &description
+	return j
+}
+
+// SetCategory set the category of the job
+func (j *Job) SetCategory(category string) *Job {
+	j.CategoryID = category
+	return j
+}
+
+// SetMaxWorkerNums set the max worker nums of the job
+func (j *Job) SetMaxWorkerNums(maxWorkerNums int) *Job {
+	j.MaxWorkerNums = maxWorkerNums
+	return j
+}
+
+// SetStatus set the status of the job
+func (j *Job) SetStatus(status string) *Job {
+	j.Status = status
+	return j
+}
+
+// SetMaxRetryCount set the max retry count of the job
+func (j *Job) SetMaxRetryCount(maxRetryCount int) *Job {
+	j.MaxRetryCount = maxRetryCount
+	return j
+}
+
+// SetDefaultTimeout set the default timeout of the job
+func (j *Job) SetDefaultTimeout(defaultTimeout int) *Job {
+	j.DefaultTimeout = &defaultTimeout
+	return j
+}
+
+// SetMode set the mode of the job
+func (j *Job) SetMode(mode ModeType) {
+	j.Mode = mode
+}
+
+func makeJob(data []byte) (*Job, error) {
+	var job Job
+	err := jsoniter.Unmarshal(data, &job)
+	if err != nil {
+		return nil, err
+	}
+	return &job, nil
+}

job/job_test.go

@@ -0,0 +1,171 @@
+package job_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/yaoapp/yao/job"
+)
+
+// TestOnce test once job
+func TestOnceGoroutine(t *testing.T) {
+	test, err := job.Once(job.GOROUTINE, map[string]interface{}{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	test.Add(1, TestHandler)
+	test.Start()
+}
+
+func TestOnceProcess(t *testing.T) {
+	test, err := job.Once(job.PROCESS, map[string]interface{}{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	test.Add(1, TestHandler)
+	test.Start()
+}
+
+func TestCronGoroutine(t *testing.T) {
+	test, err := job.Cron(job.GOROUTINE, map[string]interface{}{}, "0 0 * * *")
+	if err != nil {
+		t.Fatal(err)
+	}
+	test.Add(1, TestHandler)
+	test.Start()
+}
+
+func TestCronProcess(t *testing.T) {
+	test, err := job.Cron(job.PROCESS, map[string]interface{}{}, "0 0 * * *")
+	if err != nil {
+		t.Fatal(err)
+	}
+	test.Add(1, TestHandler)
+	test.Start()
+}
+
+// TestDaemonGoroutine tests daemon job with goroutine mode using Ticker handler
+func TestDaemonGoroutine(t *testing.T) {
+	test, err := job.Daemon(job.GOROUTINE, map[string]interface{}{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	test.Add(1, TestDaemonHandler)
+	test.Start()
+}
+
+// TestDaemonProcess tests daemon job with process mode using Ticker handler
+func TestDaemonProcess(t *testing.T) {
+	test, err := job.Daemon(job.PROCESS, map[string]interface{}{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	test.Add(1, TestDaemonHandler)
+	test.Start()
+}
+
+// TestDaemonFastGoroutine tests fast daemon job with goroutine mode for quick testing
+func TestDaemonFastGoroutine(t *testing.T) {
+	test, err := job.Daemon(job.GOROUTINE, map[string]interface{}{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	test.Add(1, TestDaemonHandlerFast)
+	test.Start()
+}
+
+// TestDaemonFastProcess tests fast daemon job with process mode for quick testing
+func TestDaemonFastProcess(t *testing.T) {
+	test, err := job.Daemon(job.PROCESS, map[string]interface{}{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	test.Add(1, TestDaemonHandlerFast)
+	test.Start()
+}
+
+func TestHandler(ctx context.Context, execution *job.Execution) error {
+	execution.SetProgress(50, "Progress 50%")
+	execution.Info("Progress 50%")
+	time.Sleep(100 * time.Millisecond)
+	execution.SetProgress(100, "Progress 100%")
+	execution.Info("Progress 100%")
+	time.Sleep(200 * time.Millisecond)
+	execution.SetProgress(100, "Progress 100%")
+	execution.Info("Progress 100%")
+	return nil
+}
+
+func TestDaemonHandler(ctx context.Context, execution *job.Execution) error {
+	// Build a daemon handler using Ticker that executes tasks every 5 seconds continuously
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+
+	counter := 0
+
+	execution.Info("Daemon handler started, running continuously...")
+	execution.SetProgress(0, "Daemon initialized and ready")
+
+	for {
+		select {
+		case <-ctx.Done():
+			// Context cancelled, graceful shutdown
+			execution.Info("Daemon handler received cancellation signal after %d iterations, exiting...", counter)
+			execution.SetProgress(100, fmt.Sprintf("Daemon stopped gracefully after %d iterations", counter))
+			return ctx.Err()
+		case <-ticker.C:
+			counter++
+
+			// Daemon doesn't need specific completion progress, show running status instead
+			execution.SetProgress(50, fmt.Sprintf("Running - completed %d iterations", counter))
+			execution.Info("Daemon tick %d: Processing periodic task...", counter)
+
+			// Simulate periodic tasks execution
+			// e.g.: cleanup temp files, health checks, data synchronization, etc.
+			time.Sleep(500 * time.Millisecond) // Simulate task execution time
+
+			execution.Debug("Daemon iteration %d completed successfully", counter)
+
+			// Output statistics every 10 iterations
+			if counter%10 == 0 {
+				execution.Info("Daemon health check: %d iterations completed, still running...", counter)
+			}
+		}
+	}
+}
+
+// TestDaemonHandlerFast fast testing version of daemon handler for testing (executes every 500ms)
+func TestDaemonHandlerFast(ctx context.Context, execution *job.Execution) error {
+	// Use shorter interval for testing
+	ticker := time.NewTicker(500 * time.Millisecond)
+	defer ticker.Stop()
+
+	counter := 0
+
+	execution.Info("Fast daemon handler started for testing, running continuously...")
+	execution.SetProgress(0, "Fast daemon initialized")
+
+	for {
+		select {
+		case <-ctx.Done():
+			execution.Info("Fast daemon handler stopped after %d iterations", counter)
+			execution.SetProgress(100, fmt.Sprintf("Fast daemon stopped after %d iterations", counter))
+			return ctx.Err()
+		case <-ticker.C:
+			counter++
+
+			execution.SetProgress(50, fmt.Sprintf("Fast daemon: %d iterations", counter))
+			execution.Debug("Fast daemon tick %d: Quick task execution", counter)
+
+			// Quick task simulation
+			time.Sleep(50 * time.Millisecond)
+
+			// Output info every 5 iterations (due to higher frequency)
+			if counter%5 == 0 {
+				execution.Info("Fast daemon: %d iterations completed", counter)
+			}
+		}
+	}
+}

job/progress.go

@@ -0,0 +1,14 @@
+package job
+
+// Progress the progress manager struct
+type Progress struct{}
+
+// Progress Progress manager
+func (j *Job) Progress() ProgressManager {
+	return &Progress{}
+}
+
+// Set set the progress
+func (p *Progress) Set(progress int, message string) error {
+	return nil
+}

job/types.go

@@ -0,0 +1,171 @@
+package job
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+)
+
+// ScheduleType the schedule type
+type ScheduleType string
+
+// ScheduleType constants
+const (
+	ScheduleTypeOnce   ScheduleType = "once"
+	ScheduleTypeCron   ScheduleType = "cron"
+	ScheduleTypeDaemon ScheduleType = "daemon"
+)
+
+// ModeType tye execution mode
+type ModeType string
+
+// ModeType constants
+const (
+	GOROUTINE ModeType = "GOROUTINE" // Execute using Go goroutine (lightweight, fast)
+	PROCESS   ModeType = "PROCESS"   // Independent process isolated
+)
+
+// LogLevel the log level
+type LogLevel uint8
+
+// These are the different logging levels. You can set the logging level to log
+// on your instance of logger, obtained with `logrus.New()`.
+const (
+	// PanicLevel level, highest level of severity. Logs and then calls panic with the
+	// message passed to Debug, Info, ...
+	Panic LogLevel = iota
+	// FatalLevel level. Logs and then calls `logger.Exit(1)`. It will exit even if the
+	// logging level is set to Panic.
+	Fatal
+	// ErrorLevel level. Logs. Used for errors that should definitely be noted.
+	// Commonly used for hooks to send errors to an error tracking service.
+	Error
+	// WarnLevel level. Non-critical entries that deserve eyes.
+	Warn
+	// InfoLevel level. General operational entries about what's going on inside the
+	// application.
+	Info
+	// DebugLevel level. Usually only enabled when debugging. Very verbose logging.
+	Debug
+	// TraceLevel level. Designates finer-grained informational events than the Debug.
+	Trace
+)
+
+// HandlerFunc the job handler function
+type HandlerFunc func(ctx context.Context, execution *Execution) error
+
+// Job represents the main job entity
+type Job struct {
+	ID                 uint                   `json:"id"`
+	JobID              string                 `json:"job_id"`
+	Name               string                 `json:"name"`
+	Icon               *string                `json:"icon,omitempty"`        // nullable: true
+	Description        *string                `json:"description,omitempty"` // nullable: true
+	CategoryID         string                 `json:"category_id"`
+	MaxWorkerNums      int                    `json:"max_worker_nums"`               // default: 1
+	Status             string                 `json:"status"`                        // default: "draft"
+	Mode               ModeType               `json:"mode"`                          // default: "goroutine"
+	ScheduleType       string                 `json:"schedule_type"`                 // default: "once"
+	ScheduleExpression *string                `json:"schedule_expression,omitempty"` // nullable: true
+	MaxRetryCount      int                    `json:"max_retry_count"`               // default: 0
+	DefaultTimeout     *int                   `json:"default_timeout,omitempty"`     // nullable: true
+	Priority           int                    `json:"priority"`                      // default: 0
+	CreatedBy          string                 `json:"created_by"`
+	NextRunAt          *time.Time             `json:"next_run_at,omitempty"`          // nullable: true
+	LastRunAt          *time.Time             `json:"last_run_at,omitempty"`          // nullable: true
+	CurrentExecutionID *string                `json:"current_execution_id,omitempty"` // nullable: true
+	Config             map[string]interface{} `json:"config,omitempty"`               // nullable: true
+	Sort               int                    `json:"sort"`                           // default: 0
+	Enabled            bool                   `json:"enabled"`                        // default: true
+	System             bool                   `json:"system"`                         // default: false
+	Readonly           bool                   `json:"readonly"`                       // default: false
+	CreatedAt          time.Time              `json:"created_at"`
+	UpdatedAt          time.Time              `json:"updated_at"`
+
+	// Relationships
+	Category   *Category   `json:"category,omitempty"`
+	Executions []Execution `json:"executions,omitempty"`
+	Logs       []Log       `json:"logs,omitempty"`
+
+	ctx    context.Context
+	cancel context.CancelFunc
+}
+
+// Category represents job categories for organization
+type Category struct {
+	ID          uint      `json:"id"`
+	CategoryID  string    `json:"category_id"`
+	Name        string    `json:"name"`
+	Icon        *string   `json:"icon,omitempty"`        // nullable: true
+	Description *string   `json:"description,omitempty"` // nullable: true
+	Sort        int       `json:"sort"`                  // default: 0
+	System      bool      `json:"system"`                // default: false
+	Enabled     bool      `json:"enabled"`               // default: true
+	Readonly    bool      `json:"readonly"`              // default: false
+	CreatedAt   time.Time `json:"created_at"`
+	UpdatedAt   time.Time `json:"updated_at"`
+
+	// Relationships
+	Jobs []Job `json:"jobs,omitempty"`
+}
+
+// Execution represents individual job execution instances
+type Execution struct {
+	ID                uint             `json:"id"`
+	ExecutionID       string           `json:"execution_id"`
+	JobID             string           `json:"job_id"`
+	Status            string           `json:"status"` // default: "queued"
+	TriggerCategory   string           `json:"trigger_category"`
+	TriggerSource     *string          `json:"trigger_source,omitempty"`      // nullable: true
+	TriggerContext    *json.RawMessage `json:"trigger_context,omitempty"`     // nullable: true
+	ScheduledAt       *time.Time       `json:"scheduled_at,omitempty"`        // nullable: true
+	WorkerID          *string          `json:"worker_id,omitempty"`           // nullable: true
+	ProcessID         *string          `json:"process_id,omitempty"`          // nullable: true
+	RetryAttempt      int              `json:"retry_attempt"`                 // default: 0
+	ParentExecutionID *string          `json:"parent_execution_id,omitempty"` // nullable: true
+	StartedAt         *time.Time       `json:"started_at,omitempty"`          // nullable: true
+	EndedAt           *time.Time       `json:"ended_at,omitempty"`            // nullable: true
+	TimeoutSeconds    *int             `json:"timeout_seconds,omitempty"`     // nullable: true
+	Duration          *int             `json:"duration,omitempty"`            // nullable: true
+	Progress          int              `json:"progress"`                      // default: 0
+	ConfigSnapshot    *json.RawMessage `json:"config_snapshot,omitempty"`     // nullable: true
+	Result            *json.RawMessage `json:"result,omitempty"`              // nullable: true
+	ErrorInfo         *json.RawMessage `json:"error_info,omitempty"`          // nullable: true
+	StackTrace        *string          `json:"stack_trace,omitempty"`         // nullable: true
+	Metrics           *json.RawMessage `json:"metrics,omitempty"`             // nullable: true
+	Context           *json.RawMessage `json:"context,omitempty"`             // nullable: true
+	CreatedAt         time.Time        `json:"created_at"`
+	UpdatedAt         time.Time        `json:"updated_at"`
+
+	// Relationships
+	Job             *Job        `json:"job,omitempty"`
+	ParentExecution *Execution  `json:"parent_execution,omitempty"`
+	ChildExecutions []Execution `json:"child_executions,omitempty"`
+	Logs            []Log       `json:"logs,omitempty"`
+}
+
+// Log represents job execution logs and events
+type Log struct {
+	ID          uint             `json:"id"`
+	JobID       string           `json:"job_id"`
+	Level       string           `json:"level"` // default: "info"
+	Message     string           `json:"message"`
+	Context     *json.RawMessage `json:"context,omitempty"`      // nullable: true
+	Source      *string          `json:"source,omitempty"`       // nullable: true
+	ExecutionID *string          `json:"execution_id,omitempty"` // nullable: true
+	Step        *string          `json:"step,omitempty"`         // nullable: true
+	Progress    *int             `json:"progress,omitempty"`     // nullable: true
+	Duration    *int             `json:"duration,omitempty"`     // nullable: true
+	ErrorCode   *string          `json:"error_code,omitempty"`   // nullable: true
+	StackTrace  *string          `json:"stack_trace,omitempty"`  // nullable: true
+	WorkerID    *string          `json:"worker_id,omitempty"`    // nullable: true
+	ProcessID   *string          `json:"process_id,omitempty"`   // nullable: true
+	Timestamp   time.Time        `json:"timestamp"`              // default: "now()"
+	Sequence    int              `json:"sequence"`               // default: 0
+	CreatedAt   time.Time        `json:"created_at"`
+	UpdatedAt   time.Time        `json:"updated_at"`
+
+	// Relationships
+	Job       *Job       `json:"job,omitempty"`
+	Execution *Execution `json:"execution,omitempty"`
+}

job/types/interfaces.go

@@ -1,12 +0,0 @@
-package types
-
-import "context"
-
-// Job interface
-type Job interface {
-	Run(ctx context.Context) error
-	AddTask(ctx context.Context, task Task) error
-}
-
-// Task interface
-type Task func(ctx context.Context, job Job) error

job/types/types.go

@@ -1 +0,0 @@
-package types

yao/models/job/job.mod.yao

@@ -78,15 +78,15 @@
       "index": true
     },
     {
-      "name": "process_type",
+      "name": "mode",
       "type": "enum",
-      "label": "Process Type",
+      "label": "Mode",
-      "comment": "Job execution process type",
+      "comment": "Job execution mode",
       "option": [
-        "goroutine", // Execute using Go goroutine (lightweight, fast)
+        "GOROUTINE", // Execute using Go goroutine (lightweight, fast)
-        "process" // Execute as independent process (isolated, heavyweight)
+        "PROCESS" // Execute as independent process (isolated, heavyweight)
       ],
-      "default": "goroutine",
+      "default": "GOROUTINE",
       "nullable": false,
       "index": true
     },