package jobs import ( "context" "errors" "fmt" "time" log "github.com/go-admin-team/go-admin-core/logger" "github.com/google/uuid" "gorm.io/gorm" "go-admin/app/jobs/models" ) const ( executionErrorTargetMissing = "JOB_TARGET_NOT_FOUND" executionErrorExecFailed = "JOB_EXECUTION_FAILED" executionErrorHTTPFailed = "JOB_HTTP_FAILED" executionErrorPanicked = "JOB_EXECUTION_PANICKED" executionErrorInterrupted = "JOB_INTERRUPTED" ) type executionFailure struct { code string message string cause error } func (failure *executionFailure) Error() string { if failure.cause != nil { return failure.cause.Error() } return failure.message } func newExecutionFailure(code, message string, cause error) error { return &executionFailure{code: code, message: message, cause: cause} } func runWithExecutionLog(db *gorm.DB, core JobCore, execute func() error) (executionErr error) { startedAt := time.Now().UTC() record := &models.SysJobExecutionLog{ ExecutionID: uuid.NewString(), JobID: core.JobId, JobNameSnapshot: core.Name, InvokeTargetSnapshot: core.InvokeTarget, TriggerType: models.JobTriggerScheduled, Status: models.JobExecutionRunning, StartedAt: startedAt, } created := false if db != nil { if err := db.WithContext(context.Background()).Create(record).Error; err != nil { log.Errorf("[Job] execution log create failed job_id=%d: %v", core.JobId, err) } else { created = true } } defer func() { if recovered := recover(); recovered != nil { executionErr = newExecutionFailure(executionErrorPanicked, "任务执行异常中断", nil) if created { finishExecutionLog(db, core.JobId, record, startedAt, executionErr) } return } if created { finishExecutionLog(db, core.JobId, record, startedAt, executionErr) } }() executionErr = execute() return executionErr } func finishExecutionLog(db *gorm.DB, jobID int, record *models.SysJobExecutionLog, startedAt time.Time, executionErr error) { finishedAt := time.Now().UTC() updates := map[string]any{ "status": models.JobExecutionSucceeded, "finished_at": finishedAt, "duration_ms": finishedAt.Sub(startedAt).Milliseconds(), "error_code": "", "error_message": "", } if executionErr != nil { code, message := publicExecutionFailure(executionErr) updates["status"] = models.JobExecutionFailed updates["error_code"] = code updates["error_message"] = message } if err := db.WithContext(context.Background()).Model(&models.SysJobExecutionLog{}). Where("id = ? AND status = ?", record.ID, models.JobExecutionRunning).Updates(updates).Error; err != nil { log.Errorf("[Job] execution log finish failed job_id=%d execution_id=%s: %v", jobID, record.ExecutionID, err) } } func publicExecutionFailure(err error) (string, string) { var failure *executionFailure if errors.As(err, &failure) { return failure.code, failure.message } return executionErrorExecFailed, "任务执行失败,请查看受控服务日志" } // RecoverInterruptedExecutionLogs closes invocations left running by the // previous process. Production currently runs one scheduler per database. func RecoverInterruptedExecutionLogs(db *gorm.DB) error { if db == nil { return nil } now := time.Now().UTC() var records []models.SysJobExecutionLog if err := db.WithContext(context.Background()).Where("status = ?", models.JobExecutionRunning).Find(&records).Error; err != nil { return err } return db.WithContext(context.Background()).Transaction(func(tx *gorm.DB) error { for _, record := range records { duration := now.Sub(record.StartedAt).Milliseconds() if duration < 0 { duration = 0 } if err := tx.Model(&models.SysJobExecutionLog{}). Where("id = ? AND status = ?", record.ID, models.JobExecutionRunning).Updates(map[string]any{ "status": models.JobExecutionInterrupted, "finished_at": now, "duration_ms": duration, "error_code": executionErrorInterrupted, "error_message": "服务重启前任务未完成", }).Error; err != nil { return err } } return nil }) } func missingExecutionTarget(target string) error { return newExecutionFailure(executionErrorTargetMissing, "任务调用目标未注册", fmt.Errorf("job target %q is not registered", target)) }