458 lines
12 KiB
Go
458 lines
12 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, securityOnly bool, 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, security_only, active, next_run)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
|
|
RETURNING id
|
|
`, agentID, name, action, schedTime, webhookURL, createdBy,
|
|
scheduleType, scheduleTime, weekday, monthday,
|
|
reboot, healthCheck, healthCheckTimeout, securityOnly, 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, securityOnly, 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, &securityOnly, &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.SecurityOnly = securityOnly.Valid && securityOnly.Bool
|
|
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, COALESCE(t.security_only, FALSE), 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) UpdateTask(id int, name, scheduleType, scheduleTime string, weekday, monthday *int,
|
|
reboot, healthCheck bool, healthCheckTimeout int, webhookURL string, active bool) (*models.Task, error) {
|
|
|
|
now := time.Now()
|
|
nextRun := CalculateNextRun(scheduleType, scheduleTime, weekday, monthday, now)
|
|
|
|
_, err := d.db.Exec(`
|
|
UPDATE tasks SET
|
|
name = $1, schedule_type = $2, schedule_time = $3,
|
|
schedule_weekday = $4, schedule_monthday = $5,
|
|
reboot = $6, health_check = $7, health_check_timeout = $8,
|
|
webhook_url = $9, active = $10, next_run = $11
|
|
WHERE id = $12
|
|
`, name, scheduleType, scheduleTime, weekday, monthday,
|
|
reboot, healthCheck, healthCheckTimeout, webhookURL, active, nextRun, id)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("Task aktualisieren: %w", err)
|
|
}
|
|
return d.GetTask(id)
|
|
}
|
|
|
|
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
|
|
}
|