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) 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 }