tâches planifiées : prompts cron persistés, outil scheduler, conversations liées aux tâches, rattrapage des exécutions manquées au redémarrage (go-cron MissedRunOnce)
This commit is contained in:
1 parent
bedb6c98e7
commit
058370830f
71 files changed
+1627
-100
No files matched your search
@@ -0,0 +1,318 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,410 @@
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
_ "github.com/mattn/go-sqlite3"
|
||||
|
||||
"trankilou.fr/lassistanoque/backend/internal/adapter/database/turso"
|
||||
"trankilou.fr/lassistanoque/backend/internal/domain"
|
||||
)
|
||||
|
||||
type fakeChatRunner struct {
|
||||
prompts []string
|
||||
}
|
||||
|
||||
func (f *fakeChatRunner) NewChat(userID, teamID, agentID, taskID string) (*domain.Chat, error) {
|
||||
return &domain.Chat{ID: "chat-1", TeamID: teamID, UserID: userID, TaskID: taskID}, nil
|
||||
}
|
||||
|
||||
func (f *fakeChatRunner) AddChatMessage(
|
||||
ctx context.Context,
|
||||
userID, teamID, chatID, agentID, prompt string,
|
||||
params *domain.ChatParams,
|
||||
) (*domain.Message, error) {
|
||||
f.prompts = append(f.prompts, prompt)
|
||||
return &domain.Message{}, nil
|
||||
}
|
||||
|
||||
func newTestRepo(t *testing.T) domain.TaskRepository {
|
||||
t.Helper()
|
||||
// cache partagé : l'ordonnanceur tourne dans sa propre goroutine et
|
||||
// doit voir la même base que le test (une :memory: classique est
|
||||
// propre à chaque connexion du pool)
|
||||
db, err := sql.Open("sqlite3", "file::memory:?cache=shared&_busy_timeout=5000")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { db.Close() })
|
||||
|
||||
schema := `
|
||||
create table tasks (
|
||||
id text not null primary key, team_id text not null, model_id text not null,
|
||||
label text not null, prompt text not null, cron text not null,
|
||||
repeat integer not null default 1, status text not null, next_datetime numeric,
|
||||
last_datetime numeric,
|
||||
_date_created numeric not null default current_timestamp, _date_updated numeric,
|
||||
_version text not null);
|
||||
create table user_teams (
|
||||
id text not null primary key, user_id text not null, team_id text not null,
|
||||
administrator bool default 0, _date_created numeric not null default current_timestamp,
|
||||
_date_updated numeric, _version text not null);
|
||||
insert into user_teams (id, user_id, team_id, _date_created, _version)
|
||||
values ('ut1','user-1','team-1','2026-01-01 10:00:00','v');
|
||||
`
|
||||
if _, err := db.Exec(schema); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return turso.NewTursoTaskRepository(db)
|
||||
}
|
||||
|
||||
func TestCreateTaskValidation(t *testing.T) {
|
||||
svc := NewService(newTestRepo(t))
|
||||
|
||||
if _, err := svc.CreateTask("user-1", "team-1", &domain.Task{Label: "l", Prompt: "p", Cron: "invalid"}); err == nil {
|
||||
t.Errorf("invalid cron spec should fail")
|
||||
}
|
||||
if _, err := svc.CreateTask("user-1", "team-1", &domain.Task{Prompt: "p", Cron: "0 8 * * *"}); err == nil {
|
||||
t.Errorf("missing label should fail")
|
||||
}
|
||||
if _, err := svc.CreateTask("user-1", "team-1", &domain.Task{Label: "l", Cron: "0 8 * * *"}); err == nil {
|
||||
t.Errorf("missing prompt should fail")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCreateTaskRecurring(t *testing.T) {
|
||||
repo := newTestRepo(t)
|
||||
svc := NewService(repo)
|
||||
|
||||
created, err := svc.CreateTask("user-1", "team-1", &domain.Task{
|
||||
ModelID: "m1",
|
||||
Label: "Meteo du matin",
|
||||
Prompt: "Donne-moi la meteo",
|
||||
Cron: "0 8 * * *",
|
||||
Repeat: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if created.Status != domain.TaskStatusActive {
|
||||
t.Errorf("status = %s", created.Status)
|
||||
}
|
||||
if created.NextDatetime == nil || created.NextDatetime.Before(time.Now()) {
|
||||
t.Errorf("next datetime should be in the future, got %v", created.NextDatetime)
|
||||
}
|
||||
|
||||
tasks, err := svc.ListTasks("user-1", "team-1")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(tasks) != 1 || tasks[0].ID != created.ID {
|
||||
t.Fatalf("expected 1 task %s, got %v", created.ID, tasks)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOneShotExecution(t *testing.T) {
|
||||
repo := newTestRepo(t)
|
||||
svc := NewService(repo)
|
||||
fake := &fakeChatRunner{}
|
||||
svc.SetChatService(fake)
|
||||
|
||||
_, err := svc.CreateTask("user-1", "team-1", &domain.Task{
|
||||
Label: "Ponctuel",
|
||||
Prompt: "Fais quelque chose",
|
||||
Cron: "@every 1s",
|
||||
Repeat: false,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
svc.Start()
|
||||
defer svc.Stop()
|
||||
|
||||
// attend l'exécution
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if len(fake.prompts) > 0 {
|
||||
break
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
if len(fake.prompts) == 0 || fake.prompts[0] != "Fais quelque chose" {
|
||||
t.Fatalf("task did not execute, prompts=%v", fake.prompts)
|
||||
}
|
||||
|
||||
// statut "done" après une exécution sans répétition
|
||||
// (runTask persiste le statut juste après l'exécution : on l'attend)
|
||||
var done *domain.Task
|
||||
deadline = time.Now().Add(5 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
tasks, _ := repo.ListTasks("user-1", "team-1")
|
||||
done = nil
|
||||
for _, tk := range tasks {
|
||||
done = tk
|
||||
}
|
||||
if done != nil && done.Status == domain.TaskStatusDone {
|
||||
break
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
if done == nil || done.Status != domain.TaskStatusDone {
|
||||
t.Fatalf("expected status done, got %+v", done)
|
||||
}
|
||||
|
||||
// pas de deuxième exécution
|
||||
time.Sleep(2 * time.Second)
|
||||
if len(fake.prompts) != 1 {
|
||||
t.Fatalf("one-shot task ran %d times", len(fake.prompts))
|
||||
}
|
||||
}
|
||||
|
||||
func TestUpdateTask(t *testing.T) {
|
||||
repo := newTestRepo(t)
|
||||
svc := NewService(repo)
|
||||
|
||||
created, err := svc.CreateTask("user-1", "team-1", &domain.Task{
|
||||
Label: "Ancien", Prompt: "ancien prompt", Cron: "0 8 * * *", Repeat: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
updated, err := svc.UpdateTask("user-1", "team-1", &domain.Task{
|
||||
ID: created.ID, Label: "Nouveau", Prompt: "nouveau prompt", Cron: "0 6 * * *", Repeat: false,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if updated.Label != "Nouveau" || updated.Prompt != "nouveau prompt" ||
|
||||
updated.Cron != "0 6 * * *" || updated.Repeat {
|
||||
t.Fatalf("update not applied: %+v", updated)
|
||||
}
|
||||
if updated.NextDatetime == nil || updated.NextDatetime.Hour() != 6 {
|
||||
t.Fatalf("next datetime should be recalculated at 6am, got %v", updated.NextDatetime)
|
||||
}
|
||||
|
||||
// cron invalide refusé
|
||||
if _, err := svc.UpdateTask("user-1", "team-1", &domain.Task{
|
||||
ID: created.ID, Label: "x", Prompt: "y", Cron: "pasunecron", Repeat: true,
|
||||
}); err == nil {
|
||||
t.Errorf("invalid cron spec should fail")
|
||||
}
|
||||
|
||||
// une tâche done est réactivée par une mise à jour
|
||||
done := updated
|
||||
done.Status = domain.TaskStatusDone
|
||||
if _, err := repo.UpdateTask(done); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
reactivated, err := svc.UpdateTask("user-1", "team-1", &domain.Task{
|
||||
ID: created.ID, Label: "Reprise", Prompt: "p", Cron: "0 9 * * *", Repeat: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if reactivated.Status != domain.TaskStatusActive {
|
||||
t.Fatalf("done task should be reactivated, got %s", reactivated.Status)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeleteTask(t *testing.T) {
|
||||
repo := newTestRepo(t)
|
||||
svc := NewService(repo)
|
||||
|
||||
created, err := svc.CreateTask("user-1", "team-1", &domain.Task{
|
||||
Label: "l", Prompt: "p", Cron: "0 8 * * *", Repeat: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := svc.DeleteTask("user-1", "team-1", created.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tasks, _ := svc.ListTasks("user-1", "team-1")
|
||||
if len(tasks) != 0 {
|
||||
t.Fatalf("expected 0 tasks after delete, got %d", len(tasks))
|
||||
}
|
||||
}
|
||||
|
||||
// createMissedRecurringTask insère une tâche récurrente dont la dernière
|
||||
// exécution remonte à plus d'une heure, avec une occurrence toutes les 30
|
||||
// minutes : des exécutions ont été manquées, comme pendant un arrêt.
|
||||
func createMissedRecurringTask(t *testing.T, repo domain.TaskRepository, prompt string) *domain.Task {
|
||||
t.Helper()
|
||||
last := time.Now().Add(-65 * time.Minute)
|
||||
task := &domain.Task{
|
||||
TeamID: "team-1", ModelID: "m1",
|
||||
Label: "Manquee", Prompt: prompt,
|
||||
Cron: "@every 30m", Repeat: true,
|
||||
Status: domain.TaskStatusActive,
|
||||
LastDatetime: &last,
|
||||
}
|
||||
created, err := repo.CreateTask(task)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return created
|
||||
}
|
||||
|
||||
// createMissedOneShotTask insère une tâche ponctuelle active dont l'échéance
|
||||
// est passée, comme après un arrêt du serveur.
|
||||
func createMissedOneShotTask(t *testing.T, repo domain.TaskRepository, prompt string) *domain.Task {
|
||||
t.Helper()
|
||||
past := time.Now().Add(-2 * time.Hour)
|
||||
task := &domain.Task{
|
||||
TeamID: "team-1", ModelID: "m1",
|
||||
Label: "Manquee", Prompt: prompt,
|
||||
Cron: "@every 1h", Repeat: false,
|
||||
Status: domain.TaskStatusActive,
|
||||
NextDatetime: &past,
|
||||
}
|
||||
created, err := repo.CreateTask(task)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return created
|
||||
}
|
||||
|
||||
// waitPrompts attend que le fake chat runner ait enregistré n prompts.
|
||||
func waitPrompts(t *testing.T, fake *fakeChatRunner, n int) bool {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if len(fake.prompts) >= n {
|
||||
return true
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func TestStartCatchUpMissedRecurringTask(t *testing.T) {
|
||||
repo := newTestRepo(t)
|
||||
svc := NewService(repo)
|
||||
fake := &fakeChatRunner{}
|
||||
svc.SetChatService(fake)
|
||||
|
||||
// dernière exécution il y a 65 min, occurrence toutes les 30 min :
|
||||
// go-cron (WithPrev + MissedRunOnce) doit rattraper au démarrage
|
||||
createMissedRecurringTask(t, repo, "prompt rattrapage")
|
||||
|
||||
svc.Start()
|
||||
defer svc.Stop()
|
||||
|
||||
if !waitPrompts(t, fake, 1) {
|
||||
t.Fatalf("missed task did not execute at startup, prompts=%v", fake.prompts)
|
||||
}
|
||||
if fake.prompts[0] != "prompt rattrapage" {
|
||||
t.Fatalf("unexpected prompt %q", fake.prompts[0])
|
||||
}
|
||||
|
||||
// une seule exécution de rattrapage (pas une par occurrence manquée)
|
||||
time.Sleep(1 * time.Second)
|
||||
if len(fake.prompts) != 1 {
|
||||
t.Fatalf("catch-up ran %d times, expected 1", len(fake.prompts))
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartCatchUpMissedOneShotTask(t *testing.T) {
|
||||
repo := newTestRepo(t)
|
||||
svc := NewService(repo)
|
||||
fake := &fakeChatRunner{}
|
||||
svc.SetChatService(fake)
|
||||
|
||||
created := createMissedOneShotTask(t, repo, "ponctuel manque")
|
||||
|
||||
svc.Start()
|
||||
defer svc.Stop()
|
||||
|
||||
if !waitPrompts(t, fake, 1) {
|
||||
t.Fatalf("missed one-shot task did not execute at startup, prompts=%v", fake.prompts)
|
||||
}
|
||||
|
||||
// la tâche passe "done" et n'est pas replanifiée
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
var done *domain.Task
|
||||
for time.Now().Before(deadline) {
|
||||
done, _ = svc.GetTask("user-1", "team-1", created.ID)
|
||||
if done != nil && done.Status == domain.TaskStatusDone {
|
||||
break
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
if done == nil || done.Status != domain.TaskStatusDone {
|
||||
t.Fatalf("expected status done after catch-up, got %+v", done)
|
||||
}
|
||||
time.Sleep(1 * time.Second)
|
||||
if len(fake.prompts) != 1 {
|
||||
t.Fatalf("one-shot catch-up ran %d times", len(fake.prompts))
|
||||
}
|
||||
}
|
||||
|
||||
func TestStartDoesNotCatchUpFutureTask(t *testing.T) {
|
||||
repo := newTestRepo(t)
|
||||
svc := NewService(repo)
|
||||
fake := &fakeChatRunner{}
|
||||
svc.SetChatService(fake)
|
||||
|
||||
// dernière exécution récente, occurrence dans une heure :
|
||||
// rien à rattraper, pas d'exécution immédiate
|
||||
last := time.Now().Add(-10 * time.Minute)
|
||||
_, err := repo.CreateTask(&domain.Task{
|
||||
TeamID: "team-1", ModelID: "m1",
|
||||
Label: "Future", Prompt: "trop tot",
|
||||
Cron: "@every 1h", Repeat: true,
|
||||
Status: domain.TaskStatusActive,
|
||||
LastDatetime: &last,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
svc.Start()
|
||||
defer svc.Stop()
|
||||
|
||||
time.Sleep(1 * time.Second)
|
||||
if len(fake.prompts) != 0 {
|
||||
t.Fatalf("task with no missed occurrence should not run, prompts=%v", fake.prompts)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunTaskPersistsLastAndNextDatetime(t *testing.T) {
|
||||
repo := newTestRepo(t)
|
||||
svc := NewService(repo)
|
||||
fake := &fakeChatRunner{}
|
||||
svc.SetChatService(fake)
|
||||
|
||||
_, err := svc.CreateTask("user-1", "team-1", &domain.Task{
|
||||
Label: "Repetitive", Prompt: "encore", Cron: "@every 1s", Repeat: true,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
svc.Start()
|
||||
defer svc.Stop()
|
||||
|
||||
if !waitPrompts(t, fake, 1) {
|
||||
t.Fatalf("task did not execute, prompts=%v", fake.prompts)
|
||||
}
|
||||
|
||||
// après une exécution, la dernière exécution et la prochaine échéance
|
||||
// doivent être persistées (last_datetime nil à la création)
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
ok := false
|
||||
for time.Now().Before(deadline) {
|
||||
tasks, _ := repo.ListTasks("user-1", "team-1")
|
||||
if len(tasks) == 1 && tasks[0].LastDatetime != nil && tasks[0].NextDatetime != nil && tasks[0].NextDatetime.After(time.Now()) {
|
||||
ok = true
|
||||
break
|
||||
}
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}
|
||||
if !ok {
|
||||
t.Fatal("last_datetime/next_datetime should be persisted after each execution")
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user