Feature: Task Scheduler - planned tasks with webhook support

2026-03-05 08:59:53 +01:00
parent e1a3246075
commit 7908ae26a0
9 changed files with 1036 additions and 1 deletions
@@ -80,6 +80,13 @@ func (h *Handler) SetupRoutes(mux *http.ServeMux, hub *ws.Hub) {
 	mux.HandleFunc("POST /api/v1/installer/upload", h.uploadInstaller)
 	mux.HandleFunc("GET /api/v1/installer/download", h.downloadInstaller)

+	// Task-Routes
+	mux.HandleFunc("GET /api/v1/tasks", h.listTasks)
+	mux.HandleFunc("GET /api/v1/tasks/{id}", h.getTask)
+	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)
+
 	// WebSocket und Tunnel-Routes
 	h.setupTunnelRoutes(mux, hub)
 }
@@ -0,0 +1,244 @@
+package api
+
+import (
+	"bytes"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"time"
+
+	"github.com/cynfo/rmm-backend/db"
+	"github.com/cynfo/rmm-backend/ws"
+)
+
+type TaskScheduler struct {
+	db  *db.Database
+	hub *ws.Hub
+}
+
+func NewTaskScheduler(database *db.Database, hub *ws.Hub) *TaskScheduler {
+	return &TaskScheduler{db: database, hub: hub}
+}
+
+func (s *TaskScheduler) Run() {
+	log.Println("Task-Scheduler gestartet (30s Intervall)")
+	ticker := time.NewTicker(30 * time.Second)
+	defer ticker.Stop()
+
+	// Run once immediately
+	s.processDueTasks()
+
+	for range ticker.C {
+		s.processDueTasks()
+	}
+}
+
+func (s *TaskScheduler) processDueTasks() {
+	tasks, err := s.db.GetDueTasks()
+	if err != nil {
+		log.Printf("Scheduler: Fehler beim Laden faelliger Tasks: %v", err)
+		return
+	}
+
+	for _, task := range tasks {
+		go s.executeTask(task.ID, task.AgentID, task.Action, task.WebhookURL, task.AgentName, task.CustomerName, task.ScheduledAt)
+	}
+}
+
+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)
+
+	// Set status to running
+	s.db.UpdateTaskStatus(taskID, "running", nil)
+
+	var result interface{}
+	var taskErr error
+
+	switch action {
+	case "backup":
+		result, taskErr = s.executeBackup(agentID)
+	case "update":
+		result, taskErr = s.executeUpdate(agentID)
+	default:
+		taskErr = fmt.Errorf("unbekannte Aktion: %s", action)
+	}
+
+	// Update status
+	status := "completed"
+	if taskErr != nil {
+		status = "failed"
+		result = map[string]string{"error": taskErr.Error()}
+		log.Printf("Scheduler: Task #%d fehlgeschlagen: %v", taskID, taskErr)
+	} else {
+		log.Printf("Scheduler: Task #%d erfolgreich abgeschlossen", taskID)
+	}
+
+	s.db.UpdateTaskStatus(taskID, status, result)
+
+	// Re-read task for webhook payload
+	task, _ := s.db.GetTask(taskID)
+
+	// Send webhook if configured
+	if webhookURL != "" && task != nil {
+		s.sendWebhook(taskID, webhookURL, task)
+	}
+}
+
+func (s *TaskScheduler) executeBackup(agentID string) (interface{}, error) {
+	if !s.hub.IsAgentConnected(agentID) {
+		return nil, fmt.Errorf("Agent nicht verbunden")
+	}
+
+	cmdID := generateID()
+	cmd := map[string]interface{}{
+		"type":    "command",
+		"id":      cmdID,
+		"command": "backup",
+		"params":  map[string]interface{}{},
+	}
+
+	responseCh := s.hub.RegisterResponseWaiter(cmdID)
+	defer s.hub.UnregisterResponseWaiter(cmdID)
+
+	cmdJSON, _ := json.Marshal(cmd)
+	if err := s.hub.SendToAgent(agentID, cmdJSON); err != nil {
+		return nil, fmt.Errorf("Command senden fehlgeschlagen: %w", err)
+	}
+
+	select {
+	case resp := <-responseCh:
+		respStatus, _ := resp["status"].(string)
+		if respStatus != "ok" {
+			errMsg, _ := resp["error"].(string)
+			return resp, fmt.Errorf("Backup fehlgeschlagen: %s", errMsg)
+		}
+
+		data, ok := resp["data"].(map[string]interface{})
+		if !ok {
+			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)
+		size := int(sizeFloat)
+
+		configXML, err := base64.StdEncoding.DecodeString(configB64)
+		if err != nil {
+			return nil, fmt.Errorf("Base64-Dekodierung fehlgeschlagen")
+		}
+
+		backupID, isNew, err := s.db.SaveConfigBackup(agentID, hash, size, string(configXML))
+		if err != nil {
+			return nil, fmt.Errorf("Backup speichern fehlgeschlagen: %w", err)
+		}
+
+		msg := "Backup gespeichert"
+		if !isNew {
+			msg = "Config unveraendert (Hash identisch)"
+		}
+
+		return map[string]interface{}{
+			"message":   msg,
+			"backup_id": backupID,
+			"sha256":    hash,
+			"size":      size,
+			"new":       isNew,
+		}, nil
+
+	case <-time.After(60 * time.Second):
+		return nil, fmt.Errorf("Timeout: Agent hat nicht rechtzeitig geantwortet")
+	}
+}
+
+func (s *TaskScheduler) executeUpdate(agentID string) (interface{}, error) {
+	if !s.hub.IsAgentConnected(agentID) {
+		return nil, fmt.Errorf("Agent nicht verbunden")
+	}
+
+	// First: update_check
+	checkCmdID := generateID()
+	checkCmd := map[string]interface{}{
+		"type":    "command",
+		"id":      checkCmdID,
+		"command": "update_check",
+	}
+
+	checkCh := s.hub.RegisterResponseWaiter(checkCmdID)
+	defer s.hub.UnregisterResponseWaiter(checkCmdID)
+
+	checkJSON, _ := json.Marshal(checkCmd)
+	if err := s.hub.SendToAgent(agentID, checkJSON); err != nil {
+		return nil, fmt.Errorf("Update-Check senden fehlgeschlagen: %w", err)
+	}
+
+	var checkResult map[string]interface{}
+	select {
+	case resp := <-checkCh:
+		checkResult = resp
+	case <-time.After(120 * time.Second):
+		return nil, fmt.Errorf("Timeout beim Update-Check")
+	}
+
+	// Then: update
+	updateCmdID := generateID()
+	updateCmd := map[string]interface{}{
+		"type":    "command",
+		"id":      updateCmdID,
+		"command": "update",
+		"params":  map[string]interface{}{"reboot": false},
+	}
+
+	updateCh := s.hub.RegisterResponseWaiter(updateCmdID)
+	defer s.hub.UnregisterResponseWaiter(updateCmdID)
+
+	updateJSON, _ := json.Marshal(updateCmd)
+	if err := s.hub.SendToAgent(agentID, updateJSON); err != nil {
+		return nil, fmt.Errorf("Update-Command senden fehlgeschlagen: %w", err)
+	}
+
+	select {
+	case resp := <-updateCh:
+		respStatus, _ := resp["status"].(string)
+		if respStatus != "ok" {
+			errMsg, _ := resp["error"].(string)
+			return resp, fmt.Errorf("Update fehlgeschlagen: %s", errMsg)
+		}
+		return map[string]interface{}{
+			"message":      "Update ausgefuehrt",
+			"check_result": checkResult,
+			"update_result": resp,
+		}, nil
+	case <-time.After(900 * time.Second):
+		return nil, fmt.Errorf("Timeout beim Update")
+	}
+}
+
+func (s *TaskScheduler) sendWebhook(taskID int, url string, task interface{}) {
+	payload, err := json.Marshal(task)
+	if err != nil {
+		log.Printf("Scheduler: Webhook-Payload fuer Task #%d serialisieren fehlgeschlagen: %v", taskID, err)
+		s.db.SetTaskWebhookResult(taskID, false, fmt.Sprintf("JSON-Fehler: %v", err))
+		return
+	}
+
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Post(url, "application/json", bytes.NewReader(payload))
+	if err != nil {
+		log.Printf("Scheduler: Webhook fuer Task #%d fehlgeschlagen: %v", taskID, err)
+		s.db.SetTaskWebhookResult(taskID, false, fmt.Sprintf("Fehler: %v", err))
+		return
+	}
+	defer resp.Body.Close()
+
+	body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+	response := fmt.Sprintf("HTTP %d: %s", resp.StatusCode, string(body))
+
+	sent := resp.StatusCode >= 200 && resp.StatusCode < 300
+	s.db.SetTaskWebhookResult(taskID, sent, response)
+	log.Printf("Scheduler: Webhook fuer Task #%d gesendet (Status: %d)", taskID, resp.StatusCode)
+}
@@ -0,0 +1,162 @@
+package api
+
+import (
+	"encoding/json"
+	"log"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/cynfo/rmm-backend/ws"
+)
+
+// GET /api/v1/tasks
+func (h *Handler) listTasks(w http.ResponseWriter, r *http.Request) {
+	limit := 50
+	offset := 0
+	agentID := r.URL.Query().Get("agent_id")
+	status := r.URL.Query().Get("status")
+	action := r.URL.Query().Get("action")
+
+	if l := r.URL.Query().Get("limit"); l != "" {
+		if n, err := strconv.Atoi(l); err == nil && n > 0 {
+			limit = n
+		}
+	}
+	if o := r.URL.Query().Get("offset"); o != "" {
+		if n, err := strconv.Atoi(o); err == nil && n >= 0 {
+			offset = n
+		}
+	}
+
+	tasks, total, err := h.db.GetTasks(limit, offset, agentID, status, action)
+	if err != nil {
+		log.Printf("Tasks laden fehlgeschlagen: %v", err)
+		writeError(w, http.StatusInternalServerError, "Fehler beim Laden")
+		return
+	}
+
+	writeJSON(w, http.StatusOK, map[string]interface{}{
+		"data":  tasks,
+		"total": total,
+	})
+}
+
+// GET /api/v1/tasks/{id}
+func (h *Handler) getTask(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.GetTask(id)
+	if err != nil {
+		writeError(w, http.StatusInternalServerError, "Fehler beim Laden")
+		return
+	}
+	if task == nil {
+		writeError(w, http.StatusNotFound, "Task nicht gefunden")
+		return
+	}
+
+	writeJSON(w, http.StatusOK, task)
+}
+
+// POST /api/v1/tasks
+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"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			writeError(w, http.StatusBadRequest, "Ungueltige Anfrage")
+			return
+		}
+
+		if req.AgentID == "" || req.Action == "" {
+			writeError(w, http.StatusBadRequest, "agent_id und action sind erforderlich")
+			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)
+		if err != nil {
+			log.Printf("Task erstellen fehlgeschlagen: %v", err)
+			writeError(w, http.StatusInternalServerError, "Task erstellen fehlgeschlagen: "+err.Error())
+			return
+		}
+
+		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
+			}
+		}
+
+		writeJSON(w, http.StatusCreated, task)
+	}
+}
+
+// POST /api/v1/tasks/{id}/cancel
+func (h *Handler) cancelTask(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
+	}
+
+	if err := h.db.CancelTask(id); err != nil {
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	h.auditLog(r, "task.cancel", "task", idStr, "", "")
+	writeJSON(w, http.StatusOK, map[string]string{"message": "Task abgebrochen"})
+}
+
+// DELETE /api/v1/tasks/{id}
+func (h *Handler) deleteTask(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
+	}
+
+	if err := h.db.DeleteTask(id); err != nil {
+		writeError(w, http.StatusBadRequest, err.Error())
+		return
+	}
+
+	h.auditLog(r, "task.delete", "task", idStr, "", "")
+	writeJSON(w, http.StatusOK, map[string]string{"message": "Task geloescht"})
+}