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