86 lines
2 KiB
Go
86 lines
2 KiB
Go
package scheduler
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"github.com/riverqueue/river"
|
|
|
|
"github.com/nomadcode/nomadcode-core/internal/notification"
|
|
"github.com/nomadcode/nomadcode-core/internal/storage"
|
|
"github.com/nomadcode/nomadcode-core/internal/workflow"
|
|
)
|
|
|
|
type TaskJobArgs struct {
|
|
TaskID string `json:"task_id"`
|
|
}
|
|
|
|
func (TaskJobArgs) Kind() string {
|
|
return "task_run"
|
|
}
|
|
|
|
type TaskWorker struct {
|
|
river.WorkerDefaults[TaskJobArgs]
|
|
|
|
Store *storage.Store
|
|
Notifications *notification.Service
|
|
Logger *slog.Logger
|
|
}
|
|
|
|
func (w *TaskWorker) Work(ctx context.Context, job *river.Job[TaskJobArgs]) error {
|
|
taskID := job.Args.TaskID
|
|
if w.Logger != nil {
|
|
w.Logger.Info("task job started", "task_id", taskID)
|
|
}
|
|
|
|
task, err := w.Store.UpdateStatus(ctx, taskID, string(workflow.StatusRunning))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
select {
|
|
case <-time.After(500 * time.Millisecond):
|
|
case <-ctx.Done():
|
|
w.markFailed(taskID, ctx.Err())
|
|
return ctx.Err()
|
|
}
|
|
|
|
result := json.RawMessage(`{"message":"dummy task completed"}`)
|
|
task, err = w.Store.CompleteTask(ctx, taskID, result)
|
|
if err != nil {
|
|
w.markFailed(taskID, err)
|
|
return err
|
|
}
|
|
|
|
if w.Notifications != nil {
|
|
err = w.Notifications.NotifyTaskCompleted(ctx, notification.TaskNotification{
|
|
TaskID: task.ID,
|
|
Title: task.Title,
|
|
Status: task.Status,
|
|
Message: "dummy task completed",
|
|
})
|
|
if err != nil && w.Logger != nil {
|
|
w.Logger.Warn("task notification failed", "task_id", taskID, "error", err)
|
|
}
|
|
}
|
|
|
|
if w.Logger != nil {
|
|
w.Logger.Info("task job completed", "task_id", taskID)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (w *TaskWorker) markFailed(taskID string, err error) {
|
|
if err == nil || w.Store == nil {
|
|
return
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
|
|
if _, failErr := w.Store.FailTask(ctx, taskID, err.Error()); failErr != nil && w.Logger != nil {
|
|
w.Logger.Error("failed to mark task failed", "task_id", taskID, "error", failErr)
|
|
}
|
|
}
|