Feature: Recurring jobs with professional scheduling dialog

This commit is contained in:
cynfo3000
2026-03-05 09:39:41 +01:00
parent f4a28463b7
commit 721fd5008f
8 changed files with 505 additions and 330 deletions
+1
View File
@@ -86,6 +86,7 @@ func (h *Handler) SetupRoutes(mux *http.ServeMux, hub *ws.Hub) {
mux.HandleFunc("POST /api/v1/tasks", h.createTask(hub))
mux.HandleFunc("DELETE /api/v1/tasks/{id}", h.deleteTask)
mux.HandleFunc("POST /api/v1/tasks/{id}/cancel", h.cancelTask)
mux.HandleFunc("PUT /api/v1/tasks/{id}/toggle", h.toggleTask)
// WebSocket und Tunnel-Routes
h.setupTunnelRoutes(mux, hub)
+24 -23
View File
@@ -11,6 +11,7 @@ import (
"time"
"github.com/cynfo/rmm-backend/db"
"github.com/cynfo/rmm-backend/models"
"github.com/cynfo/rmm-backend/ws"
)
@@ -44,46 +45,46 @@ func (s *TaskScheduler) processDueTasks() {
}
for _, task := range tasks {
go s.executeTask(task.ID, task.AgentID, task.Action, task.WebhookURL, task.AgentName, task.CustomerName, task.ScheduledAt)
go s.executeTask(task)
}
}
func (s *TaskScheduler) executeTask(taskID int, agentID, action, webhookURL, agentName, customerName, scheduledAt string) {
log.Printf("Scheduler: Starte Task #%d (%s) fuer Agent %s", taskID, action, agentID)
func (s *TaskScheduler) executeTask(task models.Task) {
log.Printf("Scheduler: Starte Task #%d (%s) fuer Agent %s", task.ID, task.Action, task.AgentID)
// Set status to running
s.db.UpdateTaskStatus(taskID, "running", nil)
s.db.UpdateTaskStatus(task.ID, "running", nil)
var result interface{}
var taskErr error
switch action {
switch task.Action {
case "backup":
result, taskErr = s.executeBackup(agentID)
result, taskErr = s.executeBackup(task.AgentID)
case "update":
result, taskErr = s.executeUpdate(agentID)
result, taskErr = s.executeUpdate(task.AgentID, task.Reboot)
default:
taskErr = fmt.Errorf("unbekannte Aktion: %s", action)
taskErr = fmt.Errorf("unbekannte Aktion: %s", task.Action)
}
// Update status
status := "completed"
if taskErr != nil {
status = "failed"
success := taskErr == nil
if !success {
result = map[string]string{"error": taskErr.Error()}
log.Printf("Scheduler: Task #%d fehlgeschlagen: %v", taskID, taskErr)
log.Printf("Scheduler: Task #%d fehlgeschlagen: %v", task.ID, taskErr)
} else {
log.Printf("Scheduler: Task #%d erfolgreich abgeschlossen", taskID)
log.Printf("Scheduler: Task #%d erfolgreich abgeschlossen", task.ID)
}
s.db.UpdateTaskStatus(taskID, status, result)
// Use CompleteRecurringTask for proper recurring handling
s.db.CompleteRecurringTask(task.ID, task.ScheduleType, task.ScheduleTime,
task.ScheduleWeekday, task.ScheduleMonthday, success, result)
// Re-read task for webhook payload
task, _ := s.db.GetTask(taskID)
fullTask, _ := s.db.GetTask(task.ID)
// Send webhook if configured
if webhookURL != "" && task != nil {
s.sendWebhook(taskID, webhookURL, task)
if task.WebhookURL != "" && fullTask != nil {
s.sendWebhook(task.ID, task.WebhookURL, fullTask)
}
}
@@ -121,7 +122,6 @@ func (s *TaskScheduler) executeBackup(agentID string) (interface{}, error) {
return resp, fmt.Errorf("ungueltiges Response-Format")
}
// Save backup to DB (same logic as triggerBackup handler)
configB64, _ := data["config"].(string)
hash, _ := data["hash"].(string)
sizeFloat, _ := data["size"].(float64)
@@ -155,7 +155,7 @@ func (s *TaskScheduler) executeBackup(agentID string) (interface{}, error) {
}
}
func (s *TaskScheduler) executeUpdate(agentID string) (interface{}, error) {
func (s *TaskScheduler) executeUpdate(agentID string, reboot bool) (interface{}, error) {
if !s.hub.IsAgentConnected(agentID) {
return nil, fmt.Errorf("Agent nicht verbunden")
}
@@ -190,7 +190,7 @@ func (s *TaskScheduler) executeUpdate(agentID string) (interface{}, error) {
"type": "command",
"id": updateCmdID,
"command": "update",
"params": map[string]interface{}{"reboot": false},
"params": map[string]interface{}{"reboot": reboot},
}
updateCh := s.hub.RegisterResponseWaiter(updateCmdID)
@@ -209,9 +209,10 @@ func (s *TaskScheduler) executeUpdate(agentID string) (interface{}, error) {
return resp, fmt.Errorf("Update fehlgeschlagen: %s", errMsg)
}
return map[string]interface{}{
"message": "Update ausgefuehrt",
"check_result": checkResult,
"message": "Update ausgefuehrt",
"check_result": checkResult,
"update_result": resp,
"reboot": reboot,
}, nil
case <-time.After(900 * time.Second):
return nil, fmt.Errorf("Timeout beim Update")
+52 -15
View File
@@ -68,10 +68,19 @@ func (h *Handler) getTask(w http.ResponseWriter, r *http.Request) {
func (h *Handler) createTask(hub *ws.Hub) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req struct {
AgentID string `json:"agent_id"`
Action string `json:"action"`
ScheduledAt string `json:"scheduled_at"`
WebhookURL string `json:"webhook_url"`
AgentID string `json:"agent_id"`
Name string `json:"name"`
Action string `json:"action"`
ScheduledAt string `json:"scheduled_at"`
ScheduleType string `json:"schedule_type"`
ScheduleTime string `json:"schedule_time"`
ScheduleWeekday *int `json:"schedule_weekday"`
ScheduleMonthday *int `json:"schedule_monthday"`
Reboot bool `json:"reboot"`
HealthCheck bool `json:"health_check"`
HealthCheckTimeout int `json:"health_check_timeout"`
WebhookURL string `json:"webhook_url"`
Active *bool `json:"active"`
}
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "Ungueltige Anfrage")
@@ -83,26 +92,33 @@ func (h *Handler) createTask(hub *ws.Hub) http.HandlerFunc {
return
}
// Validate action
if req.Action != "backup" && req.Action != "update" {
writeError(w, http.StatusBadRequest, "Ungueltige Aktion (backup, update)")
return
}
// Check agent exists
agent, _, err := h.db.GetAgent(req.AgentID)
if err != nil || agent == nil {
writeError(w, http.StatusNotFound, "Agent nicht gefunden")
return
}
// Get user from context
createdBy := "api-key"
if claims, ok := r.Context().Value(UserContextKey).(*JWTClaims); ok {
createdBy = claims.Username
}
task, err := h.db.CreateTask(req.AgentID, req.Action, req.ScheduledAt, req.WebhookURL, createdBy)
active := true
if req.Active != nil {
active = *req.Active
}
task, err := h.db.CreateTask(
req.AgentID, req.Action, req.ScheduledAt, req.WebhookURL, createdBy,
req.Name, req.ScheduleType, req.ScheduleTime,
req.ScheduleWeekday, req.ScheduleMonthday,
req.Reboot, req.HealthCheck, req.HealthCheckTimeout, active,
)
if err != nil {
log.Printf("Task erstellen fehlgeschlagen: %v", err)
writeError(w, http.StatusInternalServerError, "Task erstellen fehlgeschlagen: "+err.Error())
@@ -111,13 +127,11 @@ func (h *Handler) createTask(hub *ws.Hub) http.HandlerFunc {
h.auditLog(r, "task.create", "task", strconv.Itoa(task.ID), agent.Name, req.Action)
// If scheduled_at is in the past or now, trigger immediate execution
if req.ScheduledAt == "" {
// Already scheduled for now, scheduler will pick it up within 30s
} else {
schedTime, err := time.Parse(time.RFC3339, req.ScheduledAt)
if err == nil && !schedTime.After(time.Now()) {
// In the past — scheduler picks it up
// If once and scheduled_at is in the past or now, scheduler picks it up within 30s
if req.ScheduledAt != "" && req.ScheduleType == "once" {
if schedTime, err := time.Parse(time.RFC3339, req.ScheduledAt); err == nil && !schedTime.After(time.Now()) {
// scheduler will pick it up
_ = schedTime
}
}
@@ -125,6 +139,29 @@ func (h *Handler) createTask(hub *ws.Hub) http.HandlerFunc {
}
}
// PUT /api/v1/tasks/{id}/toggle
func (h *Handler) toggleTask(w http.ResponseWriter, r *http.Request) {
idStr := r.PathValue("id")
id, err := strconv.Atoi(idStr)
if err != nil {
writeError(w, http.StatusBadRequest, "Ungueltige Task-ID")
return
}
task, err := h.db.ToggleTask(id)
if err != nil {
writeError(w, http.StatusInternalServerError, "Fehler beim Umschalten")
return
}
if task == nil {
writeError(w, http.StatusNotFound, "Task nicht gefunden")
return
}
h.auditLog(r, "task.toggle", "task", idStr, "", "")
writeJSON(w, http.StatusOK, task)
}
// POST /api/v1/tasks/{id}/cancel
func (h *Handler) cancelTask(w http.ResponseWriter, r *http.Request) {
idStr := r.PathValue("id")
+15
View File
@@ -245,6 +245,21 @@ func (d *Database) migrate() error {
)`)
d.db.Exec(`CREATE INDEX IF NOT EXISTS idx_tasks_status_scheduled ON tasks(status, scheduled_at)`)
// Tasks: recurring jobs columns
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS name TEXT DEFAULT ''`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_type TEXT DEFAULT 'once'`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_time TEXT DEFAULT '02:00'`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_weekday INT`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS schedule_monthday INT`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS reboot BOOLEAN DEFAULT FALSE`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS health_check BOOLEAN DEFAULT FALSE`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS health_check_timeout INT DEFAULT 600`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS active BOOLEAN DEFAULT TRUE`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS last_run TIMESTAMPTZ`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS next_run TIMESTAMPTZ`)
d.db.Exec(`ALTER TABLE tasks ADD COLUMN IF NOT EXISTS run_count INT DEFAULT 0`)
d.db.Exec(`CREATE INDEX IF NOT EXISTS idx_tasks_active_next_run ON tasks(active, next_run) WHERE active = TRUE`)
// Retention Policy (90 Tage)
d.setupRetention()
+222 -119
View File
@@ -10,73 +10,168 @@ import (
"github.com/cynfo/rmm-backend/models"
)
func (d *Database) CreateTask(agentID, action, scheduledAt, webhookURL, createdBy string) (*models.Task, error) {
// Validate action
// CalculateNextRun computes the next run time based on schedule parameters.
func CalculateNextRun(scheduleType, scheduleTime string, weekday, monthday *int, after time.Time) time.Time {
h, m := 2, 0
fmt.Sscanf(scheduleTime, "%d:%d", &h, &m)
loc := after.Location()
switch scheduleType {
case "daily":
next := time.Date(after.Year(), after.Month(), after.Day(), h, m, 0, 0, loc)
if !next.After(after) {
next = next.AddDate(0, 0, 1)
}
return next
case "weekly":
wd := 1 // Monday default
if weekday != nil {
wd = *weekday
}
next := time.Date(after.Year(), after.Month(), after.Day(), h, m, 0, 0, loc)
// Find next occurrence of weekday (0=Sun, 1=Mon, ..., 6=Sat)
targetWd := time.Weekday(wd % 7)
for next.Weekday() != targetWd || !next.After(after) {
next = next.AddDate(0, 0, 1)
}
return next
case "monthly":
md := 1
if monthday != nil {
md = *monthday
}
if md < 1 {
md = 1
}
if md > 28 {
md = 28
}
next := time.Date(after.Year(), after.Month(), md, h, m, 0, 0, loc)
if !next.After(after) {
next = time.Date(after.Year(), after.Month()+1, md, h, m, 0, 0, loc)
}
return next
default: // "once"
return after
}
}
func (d *Database) CreateTask(agentID, action, scheduledAt, webhookURL, createdBy string,
name, scheduleType, scheduleTime string, weekday, monthday *int,
reboot, healthCheck bool, healthCheckTimeout int, active bool) (*models.Task, error) {
validActions := map[string]bool{"backup": true, "update": true}
if !validActions[action] {
return nil, fmt.Errorf("ungueltige Aktion: %s", action)
}
var schedTime time.Time
var err error
if scheduledAt == "" {
schedTime = time.Now()
} else {
schedTime, err = time.Parse(time.RFC3339, scheduledAt)
if err != nil {
return nil, fmt.Errorf("ungueltiges Datum: %w", err)
}
if scheduleType == "" {
scheduleType = "once"
}
if scheduleTime == "" {
scheduleTime = "02:00"
}
if healthCheckTimeout <= 0 {
healthCheckTimeout = 600
}
var task models.Task
err = d.db.QueryRow(`
INSERT INTO tasks (agent_id, action, scheduled_at, webhook_url, created_by)
VALUES ($1, $2, $3, $4, $5)
RETURNING id, agent_id, action, status, scheduled_at, webhook_url, COALESCE(created_by, ''), created_at
`, agentID, action, schedTime, webhookURL, createdBy).Scan(
&task.ID, &task.AgentID, &task.Action, &task.Status,
&schedTime, &task.WebhookURL, &task.CreatedBy, &schedTime,
)
now := time.Now()
var schedTime time.Time
if scheduleType == "once" {
if scheduledAt == "" {
schedTime = now
} else {
var err error
schedTime, err = time.Parse(time.RFC3339, scheduledAt)
if err != nil {
return nil, fmt.Errorf("ungueltiges Datum: %w", err)
}
}
} else {
schedTime = now
}
var nextRun time.Time
if scheduleType == "once" {
nextRun = schedTime
} else {
nextRun = CalculateNextRun(scheduleType, scheduleTime, weekday, monthday, now)
}
var taskID int
err := d.db.QueryRow(`
INSERT INTO tasks (agent_id, name, action, scheduled_at, webhook_url, created_by,
schedule_type, schedule_time, schedule_weekday, schedule_monthday,
reboot, health_check, health_check_timeout, active, next_run)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
RETURNING id
`, agentID, name, action, schedTime, webhookURL, createdBy,
scheduleType, scheduleTime, weekday, monthday,
reboot, healthCheck, healthCheckTimeout, active, nextRun).Scan(&taskID)
if err != nil {
return nil, fmt.Errorf("Task erstellen: %w", err)
}
// Re-read cleanly
return d.GetTask(task.ID)
return d.GetTask(taskID)
}
func (d *Database) GetTask(id int) (*models.Task, error) {
func scanTask(scan func(dest ...interface{}) error) (*models.Task, error) {
var t models.Task
var scheduledAt, createdAt time.Time
var startedAt, finishedAt sql.NullTime
var startedAt, finishedAt, lastRun, nextRun sql.NullTime
var resultJSON sql.NullString
var webhookURL, webhookResponse, createdBy sql.NullString
var agentName, customerName sql.NullString
var name, scheduleType, scheduleTime sql.NullString
var scheduleWeekday, scheduleMonthday sql.NullInt32
var reboot, healthCheck, active sql.NullBool
var healthCheckTimeout, runCount sql.NullInt32
err := d.db.QueryRow(`
SELECT t.id, t.agent_id, COALESCE(a.name, ''), COALESCE(c.name, ''),
t.action, t.status, t.scheduled_at, t.started_at, t.finished_at,
t.result, t.webhook_url, t.webhook_sent, t.webhook_response,
t.created_by, t.created_at
FROM tasks t
LEFT JOIN agents a ON a.id = t.agent_id
LEFT JOIN customers c ON c.id = a.customer_id
WHERE t.id = $1
`, id).Scan(
err := scan(
&t.ID, &t.AgentID, &agentName, &customerName,
&t.Action, &t.Status, &scheduledAt, &startedAt, &finishedAt,
&name, &t.Action, &t.Status, &scheduledAt, &startedAt, &finishedAt,
&resultJSON, &webhookURL, &t.WebhookSent, &webhookResponse,
&createdBy, &createdAt,
&scheduleType, &scheduleTime, &scheduleWeekday, &scheduleMonthday,
&reboot, &healthCheck, &healthCheckTimeout, &active, &lastRun, &nextRun, &runCount,
)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
t.AgentName = agentName.String
t.CustomerName = customerName.String
t.Name = name.String
t.ScheduleType = scheduleType.String
if t.ScheduleType == "" {
t.ScheduleType = "once"
}
t.ScheduleTime = scheduleTime.String
if t.ScheduleTime == "" {
t.ScheduleTime = "02:00"
}
if scheduleWeekday.Valid {
v := int(scheduleWeekday.Int32)
t.ScheduleWeekday = &v
}
if scheduleMonthday.Valid {
v := int(scheduleMonthday.Int32)
t.ScheduleMonthday = &v
}
t.Reboot = reboot.Valid && reboot.Bool
t.HealthCheck = healthCheck.Valid && healthCheck.Bool
t.HealthCheckTimeout = 600
if healthCheckTimeout.Valid {
t.HealthCheckTimeout = int(healthCheckTimeout.Int32)
}
t.Active = !active.Valid || active.Bool
t.RunCount = int(runCount.Int32)
t.ScheduledAt = scheduledAt.Format(time.RFC3339)
t.CreatedAt = createdAt.Format(time.RFC3339)
if startedAt.Valid {
@@ -87,6 +182,14 @@ func (d *Database) GetTask(id int) (*models.Task, error) {
s := finishedAt.Time.Format(time.RFC3339)
t.FinishedAt = &s
}
if lastRun.Valid {
s := lastRun.Time.Format(time.RFC3339)
t.LastRun = &s
}
if nextRun.Valid {
s := nextRun.Time.Format(time.RFC3339)
t.NextRun = &s
}
if resultJSON.Valid {
var r interface{}
json.Unmarshal([]byte(resultJSON.String), &r)
@@ -105,6 +208,29 @@ func (d *Database) GetTask(id int) (*models.Task, error) {
return &t, nil
}
const taskSelectCols = `t.id, t.agent_id, COALESCE(a.name, ''), COALESCE(c.name, ''),
t.name, t.action, t.status, t.scheduled_at, t.started_at, t.finished_at,
t.result, t.webhook_url, t.webhook_sent, t.webhook_response,
t.created_by, t.created_at,
t.schedule_type, t.schedule_time, t.schedule_weekday, t.schedule_monthday,
t.reboot, t.health_check, t.health_check_timeout, t.active, t.last_run, t.next_run, t.run_count`
const taskFromJoin = `FROM tasks t
LEFT JOIN agents a ON a.id = t.agent_id
LEFT JOIN customers c ON c.id = a.customer_id`
func (d *Database) GetTask(id int) (*models.Task, error) {
row := d.db.QueryRow(fmt.Sprintf("SELECT %s %s WHERE t.id = $1", taskSelectCols, taskFromJoin), id)
t, err := scanTask(row.Scan)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
return t, nil
}
func (d *Database) GetTasks(limit, offset int, agentID, status, action string) ([]models.Task, int, error) {
if limit <= 0 {
limit = 50
@@ -135,24 +261,14 @@ func (d *Database) GetTasks(limit, offset int, agentID, status, action string) (
whereClause = "WHERE " + strings.Join(where, " AND ")
}
// Count
var total int
countQuery := "SELECT COUNT(*) FROM tasks t " + whereClause
if err := d.db.QueryRow(countQuery, args...).Scan(&total); err != nil {
return nil, 0, err
}
// Data
query := fmt.Sprintf(`
SELECT t.id, t.agent_id, COALESCE(a.name, ''), COALESCE(c.name, ''),
t.action, t.status, t.scheduled_at, t.started_at, t.finished_at,
t.result, t.webhook_url, t.webhook_sent, t.webhook_response,
t.created_by, t.created_at
FROM tasks t
LEFT JOIN agents a ON a.id = t.agent_id
LEFT JOIN customers c ON c.id = a.customer_id
%s ORDER BY t.created_at DESC LIMIT $%d OFFSET $%d
`, whereClause, argIdx, argIdx+1)
query := fmt.Sprintf("SELECT %s %s %s ORDER BY t.created_at DESC LIMIT $%d OFFSET $%d",
taskSelectCols, taskFromJoin, whereClause, argIdx, argIdx+1)
args = append(args, limit, offset)
rows, err := d.db.Query(query, args...)
@@ -163,50 +279,11 @@ func (d *Database) GetTasks(limit, offset int, agentID, status, action string) (
var tasks []models.Task
for rows.Next() {
var t models.Task
var scheduledAt, createdAt time.Time
var startedAt, finishedAt sql.NullTime
var resultJSON sql.NullString
var webhookURL, webhookResponse, createdBy sql.NullString
var agentName, customerName sql.NullString
if err := rows.Scan(
&t.ID, &t.AgentID, &agentName, &customerName,
&t.Action, &t.Status, &scheduledAt, &startedAt, &finishedAt,
&resultJSON, &webhookURL, &t.WebhookSent, &webhookResponse,
&createdBy, &createdAt,
); err != nil {
t, err := scanTask(rows.Scan)
if err != nil {
return nil, 0, err
}
t.AgentName = agentName.String
t.CustomerName = customerName.String
t.ScheduledAt = scheduledAt.Format(time.RFC3339)
t.CreatedAt = createdAt.Format(time.RFC3339)
if startedAt.Valid {
s := startedAt.Time.Format(time.RFC3339)
t.StartedAt = &s
}
if finishedAt.Valid {
s := finishedAt.Time.Format(time.RFC3339)
t.FinishedAt = &s
}
if resultJSON.Valid {
var r interface{}
json.Unmarshal([]byte(resultJSON.String), &r)
t.Result = r
}
if webhookURL.Valid {
t.WebhookURL = webhookURL.String
}
if webhookResponse.Valid {
t.WebhookResponse = webhookResponse.String
}
if createdBy.Valid {
t.CreatedBy = createdBy.String
}
tasks = append(tasks, t)
tasks = append(tasks, *t)
}
if tasks == nil {
tasks = []models.Task{}
@@ -245,8 +322,44 @@ func (d *Database) UpdateTaskStatus(id int, status string, result interface{}) e
}
}
// CompleteRecurringTask handles post-execution for recurring tasks.
func (d *Database) CompleteRecurringTask(id int, scheduleType, scheduleTime string, weekday, monthday *int, success bool, result interface{}) error {
var resultJSON *string
if result != nil {
b, err := json.Marshal(result)
if err == nil {
s := string(b)
resultJSON = &s
}
}
now := time.Now()
if scheduleType == "once" {
status := "completed"
if !success {
status = "failed"
}
_, err := d.db.Exec(
`UPDATE tasks SET status = $1, finished_at = $2, result = $3, active = FALSE,
last_run = $2, run_count = run_count + 1 WHERE id = $4`,
status, now, resultJSON, id,
)
return err
}
// Recurring: reset to scheduled, compute next_run
nextRun := CalculateNextRun(scheduleType, scheduleTime, weekday, monthday, now)
_, err := d.db.Exec(
`UPDATE tasks SET status = 'scheduled', finished_at = $1, result = $2,
last_run = $1, next_run = $3, run_count = run_count + 1,
started_at = NULL WHERE id = $4`,
now, resultJSON, nextRun, id,
)
return err
}
func (d *Database) SetTaskWebhookResult(id int, sent bool, response string) error {
// Truncate response to 500 chars
if len(response) > 500 {
response = response[:500]
}
@@ -258,15 +371,11 @@ func (d *Database) SetTaskWebhookResult(id int, sent bool, response string) erro
}
func (d *Database) GetDueTasks() ([]models.Task, error) {
rows, err := d.db.Query(`
SELECT t.id, t.agent_id, COALESCE(a.name, ''), COALESCE(c.name, ''),
t.action, t.status, t.scheduled_at, t.webhook_url, t.created_at
FROM tasks t
LEFT JOIN agents a ON a.id = t.agent_id
LEFT JOIN customers c ON c.id = a.customer_id
WHERE t.status = 'scheduled' AND t.scheduled_at <= NOW()
ORDER BY t.scheduled_at ASC
`)
query := fmt.Sprintf(`SELECT %s %s
WHERE t.active = TRUE AND t.next_run IS NOT NULL AND t.next_run <= NOW() AND t.status != 'running'
ORDER BY t.next_run ASC`, taskSelectCols, taskFromJoin)
rows, err := d.db.Query(query)
if err != nil {
return nil, err
}
@@ -274,29 +383,23 @@ func (d *Database) GetDueTasks() ([]models.Task, error) {
var tasks []models.Task
for rows.Next() {
var t models.Task
var scheduledAt, createdAt time.Time
var webhookURL sql.NullString
var agentName, customerName sql.NullString
if err := rows.Scan(
&t.ID, &t.AgentID, &agentName, &customerName,
&t.Action, &t.Status, &scheduledAt, &webhookURL, &createdAt,
); err != nil {
t, err := scanTask(rows.Scan)
if err != nil {
return nil, err
}
t.AgentName = agentName.String
t.CustomerName = customerName.String
t.ScheduledAt = scheduledAt.Format(time.RFC3339)
t.CreatedAt = createdAt.Format(time.RFC3339)
if webhookURL.Valid {
t.WebhookURL = webhookURL.String
}
tasks = append(tasks, t)
tasks = append(tasks, *t)
}
return tasks, rows.Err()
}
func (d *Database) ToggleTask(id int) (*models.Task, error) {
_, err := d.db.Exec("UPDATE tasks SET active = NOT active WHERE id = $1", id)
if err != nil {
return nil, err
}
return d.GetTask(id)
}
func (d *Database) CancelTask(id int) error {
res, err := d.db.Exec(
"UPDATE tasks SET status = 'cancelled' WHERE id = $1 AND status = 'scheduled'",
+27 -15
View File
@@ -1,19 +1,31 @@
package models
type Task struct {
ID int `json:"id"`
AgentID string `json:"agent_id"`
AgentName string `json:"agent_name,omitempty"`
CustomerName string `json:"customer_name,omitempty"`
Action string `json:"action"`
Status string `json:"status"`
ScheduledAt string `json:"scheduled_at"`
StartedAt *string `json:"started_at,omitempty"`
FinishedAt *string `json:"finished_at,omitempty"`
Result interface{} `json:"result,omitempty"`
WebhookURL string `json:"webhook_url,omitempty"`
WebhookSent bool `json:"webhook_sent"`
WebhookResponse string `json:"webhook_response,omitempty"`
CreatedBy string `json:"created_by,omitempty"`
CreatedAt string `json:"created_at"`
ID int `json:"id"`
AgentID string `json:"agent_id"`
AgentName string `json:"agent_name,omitempty"`
CustomerName string `json:"customer_name,omitempty"`
Name string `json:"name"`
Action string `json:"action"`
Status string `json:"status"`
ScheduleType string `json:"schedule_type"`
ScheduleTime string `json:"schedule_time"`
ScheduleWeekday *int `json:"schedule_weekday,omitempty"`
ScheduleMonthday *int `json:"schedule_monthday,omitempty"`
ScheduledAt string `json:"scheduled_at"`
StartedAt *string `json:"started_at,omitempty"`
FinishedAt *string `json:"finished_at,omitempty"`
Result interface{} `json:"result,omitempty"`
Reboot bool `json:"reboot"`
HealthCheck bool `json:"health_check"`
HealthCheckTimeout int `json:"health_check_timeout"`
WebhookURL string `json:"webhook_url,omitempty"`
WebhookSent bool `json:"webhook_sent"`
WebhookResponse string `json:"webhook_response,omitempty"`
Active bool `json:"active"`
LastRun *string `json:"last_run,omitempty"`
NextRun *string `json:"next_run,omitempty"`
RunCount int `json:"run_count"`
CreatedBy string `json:"created_by,omitempty"`
CreatedAt string `json:"created_at"`
}