package task import ( "context" "fmt" "log/slog" "sync" "time" gocron "github.com/netresearch/go-cron" "trankilou.fr/lassistanoque/backend/internal/domain" ) // ChatRunner est la partie du service de chat nécessaire à l'exécution // d'une tâche (interface pour découplage et tests). type ChatRunner interface { NewChat(userID string, teamID string, agentID string, taskID string) (*domain.Chat, error) AddChatMessage( ctx context.Context, userID string, teamID string, chatID string, agentID string, prompt string, params *domain.ChatParams, ) (*domain.Message, error) } // Service planifie l'exécution de tâches (prompt dans une nouvelle // conversation) selon une spécification cron, avec ou sans répétition. // Les tâches sont persistées : elles sont replanifiées au démarrage et // leurs exécutions manquées pendant un arrêt sont rattrapées (go-cron // MissedRunOnce pour les répétées, manuellement pour les ponctuelles). type Service struct { repo domain.TaskRepository chat ChatRunner cron *gocron.Cron mu sync.Mutex entries map[string]gocron.EntryID } func NewService(repo domain.TaskRepository) *Service { return &Service{ repo: repo, cron: gocron.New(), entries: make(map[string]gocron.EntryID), } } // SetChatService est appelé après la construction : le service de chat a // besoin du service d'outils, dont l'outil scheduler qui a besoin d'ici. func (s *Service) SetChatService(chatService ChatRunner) { s.chat = chatService } // Start replanifie les tâches actives persistées et démarre l'ordonnanceur. // Les exécutions manquées pendant l'arrêt sont rattrapées : // - tâches répétées : par go-cron (WithPrev + MissedRunOnce), à partir de // la dernière exécution persistée ; // - tâches ponctuelles : go-cron ignore le rattrapage des entrées // run-once, elles sont exécutées ici si leur échéance est passée. func (s *Service) Start() { tasks, err := s.repo.ListActiveTasks() if err != nil { slog.Error("task.Start.ListActiveTasks", "error", err) } now := time.Now() catchups := 0 for _, t := range tasks { if !t.Repeat && t.NextDatetime != nil && t.NextDatetime.Before(now) { // l'échéance persistée est dans le passé : l'exécution a été // manquée pendant l'arrêt (next_datetime n'est plus mis à jour // après l'exécution d'une tâche ponctuelle) slog.Info("task catch-up", "id", t.ID, "label", t.Label, "due", t.NextDatetime.Format(time.RFC3339)) go s.runTask(t.ID) catchups++ // l'exécution passera la tâche "done" : rien à replanifier continue } s.schedule(t) } s.cron.Start() slog.Info("task scheduler started", "tasks", len(tasks), "catchups", catchups) } // Stop arrête l'ordonnanceur. func (s *Service) Stop() { s.cron.Stop() } // CreateTask valide la spécification cron, persiste la tâche et la planifie. func (s *Service) CreateTask(userID string, teamID string, task *domain.Task) (*domain.Task, error) { if task.Label == "" { return nil, fmt.Errorf("'label' is required") } if task.Prompt == "" { return nil, fmt.Errorf("'prompt' is required") } schedule, err := gocron.ParseStandard(task.Cron) if err != nil { return nil, fmt.Errorf("invalid cron spec %q: %w", task.Cron, err) } next := schedule.Next(time.Now()) if next.IsZero() { return nil, fmt.Errorf("cron spec %q has no next execution", task.Cron) } task.TeamID = teamID task.Status = domain.TaskStatusActive nextTime := next task.NextDatetime = &nextTime created, err := s.repo.CreateTask(task) if err != nil { return nil, err } s.schedule(created) return created, nil } // ListTasks retourne les tâches d'une équipe accessible à l'utilisateur. func (s *Service) ListTasks(userID string, teamID string) ([]*domain.Task, error) { return s.repo.ListTasks(userID, teamID) } // GetTask retourne une tâche d'une équipe accessible à l'utilisateur. func (s *Service) GetTask(userID string, teamID string, id string) (*domain.Task, error) { return s.repo.GetTaskByID(userID, teamID, id) } // UpdateTask met à jour les champs éditables (label, prompt, cron, repeat) // et replanifie la tâche. Une tâche terminée est réactivée. func (s *Service) UpdateTask(userID string, teamID string, task *domain.Task) (*domain.Task, error) { existing, err := s.repo.GetTaskByID(userID, teamID, task.ID) if err != nil { return nil, err } if task.Label == "" { return nil, fmt.Errorf("'label' is required") } if task.Prompt == "" { return nil, fmt.Errorf("'prompt' is required") } schedule, err := gocron.ParseStandard(task.Cron) if err != nil { return nil, fmt.Errorf("invalid cron spec %q: %w", task.Cron, err) } if schedule.Next(time.Now()).IsZero() { return nil, fmt.Errorf("cron spec %q has no next execution", task.Cron) } existing.Label = task.Label existing.Prompt = task.Prompt existing.Cron = task.Cron existing.Repeat = task.Repeat existing.Status = domain.TaskStatusActive nextTime := schedule.Next(time.Now()) existing.NextDatetime = &nextTime updated, err := s.repo.UpdateTask(existing) if err != nil { return nil, err } s.unschedule(updated.ID) s.schedule(updated) return updated, nil } // DeleteTask supprime une tâche et déplanifie son entrée. func (s *Service) DeleteTask(userID string, teamID string, id string) error { if err := s.repo.DeleteTask(userID, teamID, id); err != nil { return err } s.unschedule(id) return nil } // schedule ajoute une tâche à l'ordonnanceur et met à jour sa prochaine // exécution en base. Les tâches sans répétition utilisent AddOnceFunc : // une seule exécution, l'entrée disparaît ensuite. func (s *Service) schedule(t *domain.Task) { s.mu.Lock() defer s.mu.Unlock() if t.Status != domain.TaskStatusActive { return } // idempotent : replanifier une tâche déjà planifiée remplace son entrée if entryID, ok := s.entries[t.ID]; ok { s.cron.Remove(entryID) delete(s.entries, t.ID) } run := func() { s.runTask(t.ID) } var entryID gocron.EntryID var err error if t.Repeat { // rattrapage des exécutions manquées pendant un arrêt : go-cron // compare la dernière exécution aux occurrences passées. Sans // dernière exécution connue, il part de la création de la tâche. // Aucune période de grâce : un rattrapage reste dû quel que soit // l'âge de l'occurrence manquée. prev := t.DateCreated if t.LastDatetime != nil { prev = *t.LastDatetime } entryID, err = s.cron.AddFunc(t.Cron, run, gocron.WithPrev(prev), gocron.WithMissedPolicy(gocron.MissedRunOnce), ) } else { // go-cron ignore le rattrapage des entrées run-once : une tâche // ponctuelle dont l'échéance est passée est rattrapée dans Start() entryID, err = s.cron.AddOnceFunc(t.Cron, run) } if err != nil { slog.Error("task.schedule", "id", t.ID, "cron", t.Cron, "error", err) return } s.entries[t.ID] = entryID } func (s *Service) unschedule(taskID string) { s.mu.Lock() defer s.mu.Unlock() if entryID, ok := s.entries[taskID]; ok { s.cron.Remove(entryID) delete(s.entries, taskID) } } // runTask exécute une tâche : nouvelle conversation portant le prompt de la // tâche, avec le modèle enregistré. Après une exécution sans répétition, // la tâche passe au statut "done". func (s *Service) runTask(taskID string) { if s.chat == nil { slog.Error("task.runTask: chat service not set") return } // relit la tâche fraîche depuis la base tasks, err := s.repo.ListActiveTasks() if err != nil { slog.Error("task.runTask.ListActiveTasks", "error", err) return } var task *domain.Task for _, t := range tasks { if t.ID == taskID { task = t break } } if task == nil { // tâche supprimée entre-temps s.unschedule(taskID) return } userID, err := s.repo.FirstUserOfTeam(task.TeamID) if err != nil { slog.Error("task.runTask.FirstUserOfTeam", "team", task.TeamID, "error", err) return } // la conversation créée est liée à la tâche (navigation tâche <-> conversations) chatModel, err := s.chat.NewChat(userID, task.TeamID, "", task.ID) if err != nil { slog.Error("task.runTask.NewChat", "error", err) return } _, err = s.chat.AddChatMessage( context.Background(), userID, task.TeamID, chatModel.ID, "", task.Prompt, &domain.ChatParams{ModelID: task.ModelID}, ) if err != nil { slog.Error("task.runTask.AddChatMessage", "error", err) return } slog.Info("task executed", "id", task.ID, "label", task.Label, "chat", chatModel.ID) // persiste la dernière exécution : c'est ce marqueur qui permet à // go-cron de détecter, au redémarrage, les occurrences manquées now := time.Now() task.LastDatetime = &now if !task.Repeat { task.Status = domain.TaskStatusDone next := time.Time{} task.NextDatetime = &next if _, err := s.repo.UpdateTask(task); err != nil { slog.Error("task.runTask.UpdateTask", "error", err) } s.unschedule(task.ID) } else { // persiste aussi la prochaine échéance (information d'affichage) if schedule, err := gocron.ParseStandard(task.Cron); err == nil { next := schedule.Next(now) task.NextDatetime = &next } if _, err := s.repo.UpdateTask(task); err != nil { slog.Error("task.runTask.UpdateTask", "error", err) } } }