Skip to content

Commit 4514992

Browse files
pierre-bpierre-b
committed
auto start task
1 parent 0ca632a commit 4514992

5 files changed

Lines changed: 110 additions & 7 deletions

File tree

‎TODO.md‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
# TODO
22

3-
- add refresh on broadcast page + message + webhooks + lists + contacts
43
- test reply to from templates in transactional api
54
- test list welcome email / unsub email
65
- test schedule sending

‎internal/service/broadcast_service.go‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1062,6 +1062,24 @@ func (s *BroadcastService) SelectWinner(ctx context.Context, workspaceID, broadc
10621062
task.NextRunAfter = &nextRunAfter
10631063
task.Status = domain.TaskStatusPending
10641064

1065-
return s.taskRepo.Update(ctx, workspaceID, task)
1065+
if updateErr := s.taskRepo.Update(ctx, workspaceID, task); updateErr != nil {
1066+
return updateErr
1067+
}
1068+
1069+
// Immediately trigger task execution after winner selection
1070+
// Note: We always trigger here since winner selection should immediately resume sending
1071+
go func() {
1072+
// Small delay to ensure transaction is committed
1073+
time.Sleep(100 * time.Millisecond)
1074+
if execErr := s.taskService.ExecutePendingTasks(context.Background(), 1); execErr != nil {
1075+
s.logger.WithFields(map[string]interface{}{
1076+
"broadcast_id": broadcastID,
1077+
"task_id": task.ID,
1078+
"error": execErr.Error(),
1079+
}).Error("Failed to trigger immediate task execution after winner selection")
1080+
}
1081+
}()
1082+
1083+
return nil
10661084
})
10671085
}

‎internal/service/broadcast_service_test.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -342,8 +342,14 @@ func TestBroadcastService_SelectWinner_SetsWinnerAndResumesTask(t *testing.T) {
342342
d.taskRepo.EXPECT().GetTaskByBroadcastID(ctx, workspaceID, broadcastID).Return(task, nil)
343343
d.taskRepo.EXPECT().Update(ctx, workspaceID, gomock.Any()).Return(nil)
344344

345+
// Expect ExecutePendingTasks to be called in goroutine (may happen after test completes)
346+
d.taskService.EXPECT().ExecutePendingTasks(gomock.Any(), 1).Return(nil).AnyTimes()
347+
345348
err := d.svc.SelectWinner(ctx, workspaceID, broadcastID, winner)
346349
require.NoError(t, err)
350+
351+
// Give goroutine time to complete
352+
time.Sleep(200 * time.Millisecond)
347353
}
348354

349355
func TestBroadcastService_SetTaskService_SetsField(t *testing.T) {

‎internal/service/task_service.go‎

Lines changed: 64 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,9 @@ type TaskService struct {
2828
processors map[string]domain.TaskProcessor
2929
lock sync.RWMutex
3030
apiEndpoint string
31+
// autoExecuteImmediate controls whether tasks are automatically executed when set to immediate
32+
// This is mainly used to disable auto-execution during testing
33+
autoExecuteImmediate bool
3134
}
3235

3336
// WithTransaction executes a function within a transaction
@@ -47,14 +50,23 @@ func (s *TaskService) WithTransaction(ctx context.Context, fn func(*sql.Tx) erro
4750
func NewTaskService(repository domain.TaskRepository, logger logger.Logger, authService *AuthService, apiEndpoint string) *TaskService {
4851

4952
return &TaskService{
50-
repo: repository,
51-
logger: logger,
52-
authService: authService,
53-
processors: make(map[string]domain.TaskProcessor),
54-
apiEndpoint: apiEndpoint,
53+
repo: repository,
54+
logger: logger,
55+
authService: authService,
56+
processors: make(map[string]domain.TaskProcessor),
57+
apiEndpoint: apiEndpoint,
58+
autoExecuteImmediate: true, // Enable auto-execution by default
5559
}
5660
}
5761

62+
// SetAutoExecuteImmediate sets whether tasks should be automatically executed immediately
63+
// This is mainly used for testing to disable auto-execution
64+
func (s *TaskService) SetAutoExecuteImmediate(enabled bool) {
65+
s.lock.Lock()
66+
defer s.lock.Unlock()
67+
s.autoExecuteImmediate = enabled
68+
}
69+
5870
// RegisterProcessor registers a task processor for a specific task type
5971
func (s *TaskService) RegisterProcessor(processor domain.TaskProcessor) {
6072
s.lock.Lock()
@@ -692,6 +704,21 @@ func (s *TaskService) handleBroadcastScheduled(ctx context.Context, payload doma
692704
}).Error("Failed to update task for scheduled broadcast")
693705
return updateErr
694706
}
707+
708+
// Immediately trigger task execution after the transaction commits
709+
if s.autoExecuteImmediate {
710+
go func() {
711+
// Small delay to ensure transaction is committed
712+
time.Sleep(100 * time.Millisecond)
713+
if execErr := s.ExecutePendingTasks(context.Background(), 1); execErr != nil {
714+
s.logger.WithFields(map[string]interface{}{
715+
"broadcast_id": broadcastID,
716+
"task_id": existingTask.ID,
717+
"error": execErr.Error(),
718+
}).Error("Failed to trigger immediate task execution")
719+
}
720+
}()
721+
}
695722
}
696723

697724
return nil
@@ -755,6 +782,21 @@ func (s *TaskService) handleBroadcastScheduled(ctx context.Context, payload doma
755782
"workspace_id": payload.WorkspaceID,
756783
}).Info("Successfully created task for scheduled broadcast")
757784

785+
// If the broadcast is set to send immediately, trigger task execution
786+
if sendNow && status == string(domain.BroadcastStatusSending) && s.autoExecuteImmediate {
787+
// Immediately trigger task execution after the transaction commits
788+
go func() {
789+
// Small delay to ensure transaction is committed
790+
time.Sleep(100 * time.Millisecond)
791+
if execErr := s.ExecutePendingTasks(context.Background(), 1); execErr != nil {
792+
s.logger.WithFields(map[string]interface{}{
793+
"broadcast_id": broadcastID,
794+
"error": execErr.Error(),
795+
}).Error("Failed to trigger immediate task execution for new task")
796+
}
797+
}()
798+
}
799+
758800
return nil
759801
})
760802

@@ -873,6 +915,23 @@ func (s *TaskService) handleBroadcastResumed(ctx context.Context, payload domain
873915
"broadcast_id": broadcastID,
874916
"task_id": task.ID,
875917
}).Info("Successfully resumed task for resumed broadcast")
918+
919+
// Check if broadcast should start immediately
920+
startNow, _ := payload.Data["start_now"].(bool)
921+
if startNow && s.autoExecuteImmediate {
922+
// Immediately trigger task execution
923+
go func() {
924+
// Small delay to ensure transaction is committed
925+
time.Sleep(100 * time.Millisecond)
926+
if execErr := s.ExecutePendingTasks(context.Background(), 1); execErr != nil {
927+
s.logger.WithFields(map[string]interface{}{
928+
"broadcast_id": broadcastID,
929+
"task_id": task.ID,
930+
"error": execErr.Error(),
931+
}).Error("Failed to trigger immediate task execution for resumed broadcast")
932+
}
933+
}()
934+
}
876935
}
877936
}
878937

‎internal/service/task_service_test.go‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@ func TestTaskService_ExecuteTask(t *testing.T) {
3737
mockLogger.EXPECT().Error(gomock.Any()).AnyTimes()
3838

3939
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
40+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
4041

4142
// Setup transaction mocking for all tests
4243
mockRepo.EXPECT().
@@ -171,6 +172,7 @@ func TestTaskService_ExecuteTask(t *testing.T) {
171172

172173
// Create a new task service instance for this test
173174
procTaskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
175+
procTaskService.SetAutoExecuteImmediate(false) // Disable for testing
174176

175177
// Register a processor for the task type
176178
mockProcessor := mocks.NewMockTaskProcessor(procCtrl)
@@ -292,6 +294,7 @@ func TestTaskService_ExecuteTask(t *testing.T) {
292294

293295
// Create a new task service instance for this test
294296
procTaskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
297+
procTaskService.SetAutoExecuteImmediate(false) // Disable for testing
295298

296299
// Register a processor for the task type
297300
mockProcessor := mocks.NewMockTaskProcessor(procCtrl)
@@ -354,6 +357,7 @@ func TestTaskService_ExecuteTask(t *testing.T) {
354357

355358
// Create a new task service instance for this test
356359
procTaskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
360+
procTaskService.SetAutoExecuteImmediate(false) // Disable for testing
357361

358362
// Register a processor for the task type
359363
mockProcessor := mocks.NewMockTaskProcessor(procCtrl)
@@ -417,6 +421,7 @@ func TestTaskService_CreateTask(t *testing.T) {
417421
mockLogger.EXPECT().Info(gomock.Any()).AnyTimes()
418422

419423
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
424+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
420425

421426
t.Run("Sets default values when not provided", func(t *testing.T) {
422427
// Setup
@@ -523,6 +528,7 @@ func TestTaskService_ListTasks(t *testing.T) {
523528
mockLogger.EXPECT().Info(gomock.Any()).AnyTimes()
524529

525530
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
531+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
526532

527533
t.Run("Returns tasks with pagination info", func(t *testing.T) {
528534
// Setup
@@ -632,6 +638,7 @@ func TestTaskService_GetTask(t *testing.T) {
632638
mockLogger.EXPECT().Info(gomock.Any()).AnyTimes()
633639

634640
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
641+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
635642

636643
t.Run("Returns task when found", func(t *testing.T) {
637644
// Setup
@@ -696,6 +703,7 @@ func TestTaskService_DeleteTask(t *testing.T) {
696703
mockLogger.EXPECT().Info(gomock.Any()).AnyTimes()
697704

698705
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
706+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
699707

700708
t.Run("Deletes task successfully", func(t *testing.T) {
701709
// Setup
@@ -750,6 +758,7 @@ func TestTaskService_RegisterProcessor(t *testing.T) {
750758
mockLogger.EXPECT().Info(gomock.Any()).AnyTimes()
751759

752760
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
761+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
753762

754763
t.Run("Registers processor for supported task types", func(t *testing.T) {
755764
// Create a processor that only supports certain task types
@@ -831,6 +840,9 @@ func TestTaskService_BroadcastEventHandlers(t *testing.T) {
831840
mockEventBus.EXPECT().Subscribe(domain.EventBroadcastFailed, gomock.Any()).Times(1)
832841
mockEventBus.EXPECT().Subscribe(domain.EventBroadcastCancelled, gomock.Any()).Times(1)
833842

843+
// Disable auto-execution for testing to avoid goroutine issues
844+
taskService.SetAutoExecuteImmediate(false)
845+
834846
// Subscribe to events
835847
taskService.SubscribeToBroadcastEvents(mockEventBus)
836848

@@ -944,6 +956,7 @@ func TestTaskService_ExecutePendingTasks(t *testing.T) {
944956
t.Run("Uses HTTP execution when API endpoint is configured", func(t *testing.T) {
945957
// Create TaskService with API endpoint
946958
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
959+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
947960

948961
// Setup
949962
ctx := context.Background()
@@ -1004,6 +1017,7 @@ func TestTaskService_ExecutePendingTasks(t *testing.T) {
10041017

10051018
// Create TaskService without API endpoint
10061019
taskService := NewTaskService(localRepo, localLogger, mockAuthService, "")
1020+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
10071021

10081022
// Setup
10091023
ctx := context.Background()
@@ -1058,6 +1072,7 @@ func TestTaskService_ExecutePendingTasks(t *testing.T) {
10581072
t.Run("Handles GetNextBatch error", func(t *testing.T) {
10591073
// Create TaskService
10601074
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
1075+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
10611076

10621077
// Setup
10631078
ctx := context.Background()
@@ -1080,6 +1095,7 @@ func TestTaskService_ExecutePendingTasks(t *testing.T) {
10801095
t.Run("Uses default maxTasks when 0 is provided", func(t *testing.T) {
10811096
// Create TaskService
10821097
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
1098+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
10831099

10841100
// Setup
10851101
ctx := context.Background()
@@ -1115,6 +1131,7 @@ func TestTaskService_HandleBroadcastResumed(t *testing.T) {
11151131
mockLogger.EXPECT().Error(gomock.Any()).AnyTimes()
11161132

11171133
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
1134+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
11181135

11191136
t.Run("Successfully resumes a task for resumed broadcast", func(t *testing.T) {
11201137
// Setup
@@ -1264,6 +1281,7 @@ func TestTaskService_HandleBroadcastSent(t *testing.T) {
12641281
mockLogger.EXPECT().Error(gomock.Any()).AnyTimes()
12651282

12661283
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
1284+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
12671285

12681286
t.Run("Successfully completes a task for sent broadcast", func(t *testing.T) {
12691287
// Setup
@@ -1406,6 +1424,7 @@ func TestTaskService_HandleBroadcastFailed(t *testing.T) {
14061424
mockLogger.EXPECT().Error(gomock.Any()).AnyTimes()
14071425

14081426
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
1427+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
14091428

14101429
t.Run("Successfully marks task as failed for failed broadcast", func(t *testing.T) {
14111430
// Setup
@@ -1589,6 +1608,7 @@ func TestTaskService_HandleBroadcastCancelled(t *testing.T) {
15891608
mockLogger.EXPECT().Error(gomock.Any()).AnyTimes()
15901609

15911610
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
1611+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
15921612

15931613
t.Run("Successfully marks task as failed for cancelled broadcast", func(t *testing.T) {
15941614
// Setup
@@ -1732,6 +1752,7 @@ func TestTaskService_HandleBroadcastScheduledExtended(t *testing.T) {
17321752
mockLogger.EXPECT().Error(gomock.Any()).AnyTimes()
17331753

17341754
taskService := NewTaskService(mockRepo, mockLogger, mockAuthService, apiEndpoint)
1755+
taskService.SetAutoExecuteImmediate(false) // Disable for testing
17351756

17361757
t.Run("Updates existing task when found for immediate sending", func(t *testing.T) {
17371758
// Setup

0 commit comments

Comments
 (0)