rmm2/backend/db/tasks.go
cynfo3000 d4b6ca86e5 feat: Task Health Check + customer_number in webhook payload
- Task model: CustomerNumber field (K-Nummer aus customers.number)
- DB: taskSelectCols joined c.number, scanTask scans CustomerNumber
- Hub: WaitForAgentOnline() polls until agent reconnects or timeout
- Scheduler: executeUpdateWithHealthCheck() waits for agent reconnect after update
  - health_check: {agent_online: bool, checked_at: RFC3339} in result
  - Status: completed_with_warning wenn Update OK aber Agent offline bleibt
- Frontend: Task-Tabelle zeigt Status-Badge (lauft/fehler/warnung/abgebr.)
2026-03-05 10:53:08 +01:00

436 lines
11 KiB
Go

package db
import (
"database/sql"
"encoding/json"
"fmt"
"strings"
"time"
"github.com/cynfo/rmm-backend/models"
)
// 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)
}
if scheduleType == "" {
scheduleType = "once"
}
if scheduleTime == "" {
scheduleTime = "02:00"
}
if healthCheckTimeout <= 0 {
healthCheckTimeout = 600
}
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)
}
return d.GetTask(taskID)
}
func scanTask(scan func(dest ...interface{}) error) (*models.Task, error) {
var t models.Task
var scheduledAt, createdAt time.Time
var startedAt, finishedAt, lastRun, nextRun sql.NullTime
var resultJSON sql.NullString
var webhookURL, webhookResponse, createdBy sql.NullString
var agentName, customerName, customerNumber 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 := scan(
&t.ID, &t.AgentID, &agentName, &customerName, &customerNumber,
&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 != nil {
return nil, err
}
t.AgentName = agentName.String
t.CustomerName = customerName.String
t.CustomerNumber = customerNumber.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 {
s := startedAt.Time.Format(time.RFC3339)
t.StartedAt = &s
}
if finishedAt.Valid {
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)
t.Result = r
}
if webhookURL.Valid {
t.WebhookURL = webhookURL.String
}
if webhookResponse.Valid {
t.WebhookResponse = webhookResponse.String
}
if createdBy.Valid {
t.CreatedBy = createdBy.String
}
return &t, nil
}
const taskSelectCols = `t.id, t.agent_id, COALESCE(a.name, ''), COALESCE(c.name, ''), COALESCE(c.number, ''),
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
}
where := []string{}
args := []interface{}{}
argIdx := 1
if agentID != "" {
where = append(where, fmt.Sprintf("t.agent_id = $%d", argIdx))
args = append(args, agentID)
argIdx++
}
if status != "" {
where = append(where, fmt.Sprintf("t.status = $%d", argIdx))
args = append(args, status)
argIdx++
}
if action != "" {
where = append(where, fmt.Sprintf("t.action = $%d", argIdx))
args = append(args, action)
argIdx++
}
whereClause := ""
if len(where) > 0 {
whereClause = "WHERE " + strings.Join(where, " AND ")
}
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
}
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...)
if err != nil {
return nil, 0, err
}
defer rows.Close()
var tasks []models.Task
for rows.Next() {
t, err := scanTask(rows.Scan)
if err != nil {
return nil, 0, err
}
tasks = append(tasks, *t)
}
if tasks == nil {
tasks = []models.Task{}
}
return tasks, total, rows.Err()
}
func (d *Database) UpdateTaskStatus(id int, status string, 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().UTC()
switch status {
case "running":
_, err := d.db.Exec(
"UPDATE tasks SET status = $1, started_at = $2 WHERE id = $3",
status, now, id,
)
return err
case "completed", "failed":
_, err := d.db.Exec(
"UPDATE tasks SET status = $1, finished_at = $2, result = $3 WHERE id = $4",
status, now, resultJSON, id,
)
return err
default:
_, err := d.db.Exec("UPDATE tasks SET status = $1 WHERE id = $2", status, id)
return err
}
}
// 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 {
if len(response) > 500 {
response = response[:500]
}
_, err := d.db.Exec(
"UPDATE tasks SET webhook_sent = $1, webhook_response = $2 WHERE id = $3",
sent, response, id,
)
return err
}
func (d *Database) GetDueTasks() ([]models.Task, error) {
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
}
defer rows.Close()
var tasks []models.Task
for rows.Next() {
t, err := scanTask(rows.Scan)
if err != nil {
return nil, err
}
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'",
id,
)
if err != nil {
return err
}
rows, _ := res.RowsAffected()
if rows == 0 {
return fmt.Errorf("Task nicht gefunden oder nicht mehr im Status 'scheduled'")
}
return nil
}
// UpdateTaskStatusDirect sets only the status field (for post-completion overrides).
func (d *Database) UpdateTaskStatusDirect(id int, status string) error {
_, err := d.db.Exec("UPDATE tasks SET status = $1 WHERE id = $2", status, id)
return err
}
func (d *Database) DeleteTask(id int) error {
res, err := d.db.Exec("DELETE FROM tasks WHERE id = $1", id)
if err != nil {
return err
}
rows, _ := res.RowsAffected()
if rows == 0 {
return fmt.Errorf("Task nicht gefunden")
}
return nil
}