Skip to content

Commit 0a7892a

Browse files
committed
fix phase transition
1 parent bb2e8b1 commit 0a7892a

7 files changed

Lines changed: 215 additions & 20 deletions

File tree

console/src/services/api/task.ts

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ export interface SendBroadcastState {
1111
failed_count: number
1212
channel_type: string
1313
recipient_offset: number
14-
end_offset: number
1514
}
1615

1716
export interface TaskState {

console/src/services/api/types.ts

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -562,7 +562,6 @@ export interface SendBroadcastState {
562562
failed_count: number
563563
channel_type: string
564564
recipient_offset: number
565-
end_offset: number
566565
}
567566

568567
export interface TaskState {

internal/domain/task.go

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -72,7 +72,6 @@ type SendBroadcastState struct {
7272
FailedCount int `json:"failed_count"`
7373
ChannelType string `json:"channel_type"`
7474
RecipientOffset int64 `json:"recipient_offset"`
75-
EndOffset int64 `json:"end_offset"`
7675
// New fields for A/B testing phases
7776
Phase string `json:"phase"` // "test", "winner", or "single"
7877
TestPhaseCompleted bool `json:"test_phase_completed"`

internal/domain/task_test.go

Lines changed: 1 addition & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,6 @@ func TestTaskState_Value(t *testing.T) {
5959
FailedCount: 10,
6060
ChannelType: "email",
6161
RecipientOffset: 750,
62-
EndOffset: 1000,
6362
},
6463
}
6564
value, err := state.Value()
@@ -84,7 +83,6 @@ func TestTaskState_Value(t *testing.T) {
8483
assert.Equal(t, float64(10), broadcastMap["failed_count"])
8584
assert.Equal(t, "email", broadcastMap["channel_type"])
8685
assert.Equal(t, float64(750), broadcastMap["recipient_offset"])
87-
assert.Equal(t, float64(1000), broadcastMap["end_offset"])
8886
})
8987
}
9088

@@ -134,8 +132,7 @@ func TestTaskState_Scan(t *testing.T) {
134132
"sent_count": 300,
135133
"failed_count": 5,
136134
"channel_type": "email",
137-
"recipient_offset": 300,
138-
"end_offset": 500
135+
"recipient_offset": 300
139136
}
140137
}`)
141138

@@ -151,7 +148,6 @@ func TestTaskState_Scan(t *testing.T) {
151148
assert.Equal(t, 5, state.SendBroadcast.FailedCount)
152149
assert.Equal(t, "email", state.SendBroadcast.ChannelType)
153150
assert.Equal(t, int64(300), state.SendBroadcast.RecipientOffset)
154-
assert.Equal(t, int64(500), state.SendBroadcast.EndOffset)
155151
})
156152

157153
t.Run("invalid type", func(t *testing.T) {
@@ -469,7 +465,6 @@ func TestCreateTaskRequest_Validate(t *testing.T) {
469465
FailedCount: 0,
470466
ChannelType: "email",
471467
RecipientOffset: 0,
472-
EndOffset: 1000,
473468
},
474469
}
475470

@@ -658,7 +653,6 @@ func TestCreateTaskRequest_Validate(t *testing.T) {
658653
FailedCount: 0,
659654
ChannelType: "sms",
660655
RecipientOffset: 0,
661-
EndOffset: 500,
662656
},
663657
}
664658

internal/service/broadcast/orchestrator.go

Lines changed: 15 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -647,20 +647,21 @@ func (o *BroadcastOrchestrator) Process(ctx context.Context, task *domain.Task,
647647
// Determine which templates to load based on phase
648648
var templateIDs []string
649649

650-
if broadcastState.Phase == "test" {
650+
switch broadcastState.Phase {
651+
case "test":
651652
// Load all variations for testing
652653
templateIDs = make([]string, len(broadcast.TestSettings.Variations))
653654
for i, variation := range broadcast.TestSettings.Variations {
654655
templateIDs[i] = variation.TemplateID
655656
}
656-
} else if broadcastState.Phase == "winner" {
657+
case "winner":
657658
// Load only the winning template
658659
if broadcast.WinningTemplate != "" {
659660
templateIDs = []string{broadcast.WinningTemplate}
660661
} else {
661662
return false, fmt.Errorf("winner phase but no winning template selected")
662663
}
663-
} else {
664+
default:
664665
// Single template broadcast
665666
templateIDs = make([]string, len(broadcast.TestSettings.Variations))
666667
for i, variation := range broadcast.TestSettings.Variations {
@@ -710,16 +711,21 @@ func (o *BroadcastOrchestrator) Process(ctx context.Context, task *domain.Task,
710711
// All phases use the same RecipientOffset for continuity
711712
currentOffset = int(broadcastState.RecipientOffset)
712713

714+
// If a winner has already been selected manually while test is running, transition immediately
713715
if broadcastState.Phase == "test" {
714-
// If a winner has already been selected manually while test is running, transition immediately
715716
if broadcast.WinningTemplate != "" || broadcast.Status == domain.BroadcastStatusWinnerSelected {
716717
broadcastState.Phase = "winner"
717718
}
719+
}
720+
721+
// Set recipient limit based on current phase (after potential transition)
722+
switch broadcastState.Phase {
723+
case "test":
718724
recipientLimit = broadcastState.TestPhaseRecipientCount
719-
} else if broadcastState.Phase == "winner" {
725+
case "winner":
720726
// Winner phase processes remaining recipients after test phase
721727
recipientLimit = broadcastState.TotalRecipients
722-
} else {
728+
default:
723729
// Single template - process all recipients
724730
recipientLimit = broadcastState.TotalRecipients
725731
}
@@ -961,7 +967,8 @@ func (o *BroadcastOrchestrator) Process(ctx context.Context, task *domain.Task,
961967
if allDone {
962968
var statusMessage string
963969

964-
if broadcastState.Phase == "winner" || broadcastState.Phase == "single" {
970+
switch broadcastState.Phase {
971+
case "winner", "single":
965972
// Winner phase or single template complete - mark as sent
966973
broadcast.Status = domain.BroadcastStatusSent
967974
broadcast.UpdatedAt = time.Now().UTC()
@@ -976,7 +983,7 @@ func (o *BroadcastOrchestrator) Process(ctx context.Context, task *domain.Task,
976983
}
977984

978985
statusMessage = "sent"
979-
} else if broadcastState.Phase == "test" {
986+
case "test":
980987
// Test phase complete - should have been handled by handleTestPhaseCompletion
981988
// This shouldn't happen, but handle it gracefully
982989
o.logger.WithField("broadcast_id", broadcastState.BroadcastID).Warn("Test phase marked as complete in final processing - this should have been handled earlier")

internal/service/broadcast/orchestrator_test.go

Lines changed: 199 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1745,6 +1745,205 @@ func TestBroadcastOrchestrator_Process_AutoWinnerEvaluationPath(t *testing.T) {
17451745
assert.True(t, done)
17461746
}
17471747

1748+
// TestBroadcastOrchestrator_Process_ABTestWinnerPhaseProcessesRemainingRecipients
1749+
// This test reproduces and verifies the fix for the bug where the winner phase
1750+
// would not process remaining recipients after test phase completion.
1751+
// Scenario: 2 recipients, 50% test phase (1 recipient), winner selection, then winner phase should process remaining 1 recipient.
1752+
func TestBroadcastOrchestrator_Process_ABTestWinnerPhaseProcessesRemainingRecipients(t *testing.T) {
1753+
ctrl := gomock.NewController(t)
1754+
defer ctrl.Finish()
1755+
1756+
// Create mocks
1757+
mockMessageSender := mocks.NewMockMessageSender(ctrl)
1758+
mockBroadcastRepo := domainmocks.NewMockBroadcastRepository(ctrl)
1759+
mockTemplateRepo := domainmocks.NewMockTemplateRepository(ctrl)
1760+
mockContactRepo := domainmocks.NewMockContactRepository(ctrl)
1761+
mockTaskRepo := domainmocks.NewMockTaskRepository(ctrl)
1762+
mockWorkspaceRepo := domainmocks.NewMockWorkspaceRepository(ctrl)
1763+
mockLogger := pkgmocks.NewMockLogger(ctrl)
1764+
mockTimeProvider := mocks.NewMockTimeProvider(ctrl)
1765+
1766+
// Mock time provider
1767+
base := time.Date(2023, 1, 1, 12, 0, 0, 0, time.UTC)
1768+
mockTimeProvider.EXPECT().Now().Return(base).AnyTimes()
1769+
mockTimeProvider.EXPECT().Since(gomock.Any()).Return(time.Minute).AnyTimes()
1770+
1771+
// Setup workspace with email provider
1772+
workspace := &domain.Workspace{
1773+
ID: "workspace-123",
1774+
Settings: domain.WorkspaceSettings{
1775+
SecretKey: "secret-key",
1776+
EmailTrackingEnabled: true,
1777+
MarketingEmailProviderID: "marketing-provider-id",
1778+
},
1779+
Integrations: []domain.Integration{
1780+
{
1781+
ID: "marketing-provider-id",
1782+
Type: domain.IntegrationTypeEmail,
1783+
EmailProvider: domain.EmailProvider{
1784+
Kind: domain.EmailProviderKindSES,
1785+
SES: &domain.AmazonSESSettings{
1786+
AccessKey: "access-key",
1787+
SecretKey: "secret-key",
1788+
Region: "us-east-1",
1789+
},
1790+
},
1791+
},
1792+
},
1793+
}
1794+
mockWorkspaceRepo.EXPECT().GetByID(gomock.Any(), "workspace-123").Return(workspace, nil).AnyTimes()
1795+
1796+
// Setup broadcast with A/B testing enabled
1797+
bcast := &domain.Broadcast{
1798+
ID: "broadcast-123",
1799+
WorkspaceID: "workspace-123",
1800+
Audience: domain.AudienceSettings{Lists: []string{"list-1"}},
1801+
Status: domain.BroadcastStatusWinnerSelected, // Winner already selected
1802+
WinningTemplate: "template-B", // Winner is template B
1803+
TestSettings: domain.BroadcastTestSettings{
1804+
Enabled: true,
1805+
SamplePercentage: 50, // 50% test phase
1806+
Variations: []domain.BroadcastVariation{
1807+
{TemplateID: "template-A"},
1808+
{TemplateID: "template-B"},
1809+
},
1810+
},
1811+
}
1812+
mockBroadcastRepo.EXPECT().GetBroadcast(gomock.Any(), "workspace-123", "broadcast-123").Return(bcast, nil).AnyTimes()
1813+
1814+
// Setup templates - we need to mock both in case the phase transition doesn't work initially
1815+
templateA := &domain.Template{
1816+
ID: "template-A",
1817+
Email: &domain.EmailTemplate{
1818+
Subject: "Test Subject A",
1819+
SenderID: "sender-1",
1820+
VisualEditorTree: &notifuse_mjml.MJMLBlock{
1821+
BaseBlock: notifuse_mjml.BaseBlock{
1822+
ID: "root",
1823+
Type: notifuse_mjml.MJMLComponentMjml,
1824+
},
1825+
},
1826+
},
1827+
}
1828+
templateB := &domain.Template{
1829+
ID: "template-B",
1830+
Email: &domain.EmailTemplate{
1831+
Subject: "Test Subject B",
1832+
SenderID: "sender-1",
1833+
VisualEditorTree: &notifuse_mjml.MJMLBlock{
1834+
BaseBlock: notifuse_mjml.BaseBlock{
1835+
ID: "root",
1836+
Type: notifuse_mjml.MJMLComponentMjml,
1837+
},
1838+
},
1839+
},
1840+
}
1841+
1842+
// Mock template loading - might load all variations first, then just winner
1843+
mockTemplateRepo.EXPECT().GetTemplateByID(gomock.Any(), "workspace-123", "template-A", int64(0)).Return(templateA, nil).AnyTimes()
1844+
mockTemplateRepo.EXPECT().GetTemplateByID(gomock.Any(), "workspace-123", "template-B", int64(0)).Return(templateB, nil).AnyTimes()
1845+
1846+
// Setup recipients: winner phase should fetch from offset=1 (after test phase processed 1 recipient)
1847+
// This is the key part of the test - ensuring the winner phase processes the remaining recipient
1848+
recipient := &domain.ContactWithList{
1849+
Contact: &domain.Contact{
1850+
Email: "recipient2@example.com",
1851+
},
1852+
ListID: "list-1",
1853+
}
1854+
mockContactRepo.EXPECT().GetContactsForBroadcast(
1855+
gomock.Any(),
1856+
"workspace-123",
1857+
bcast.Audience,
1858+
1, // limit: remaining recipients in phase
1859+
1, // offset: should be 1 (after test phase processed first recipient)
1860+
).Return([]*domain.ContactWithList{recipient}, nil)
1861+
1862+
// Mock successful sending
1863+
mockMessageSender.EXPECT().SendBatch(
1864+
gomock.Any(),
1865+
"workspace-123",
1866+
"secret-key",
1867+
true,
1868+
"broadcast-123",
1869+
[]*domain.ContactWithList{recipient},
1870+
gomock.Any(), // templates
1871+
gomock.Any(), // email provider
1872+
gomock.Any(), // timeout
1873+
).Return(1, 0, nil) // 1 sent, 0 failed
1874+
1875+
// Mock task state saving
1876+
mockTaskRepo.EXPECT().SaveState(gomock.Any(), "workspace-123", "task-123", gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
1877+
1878+
// Mock final broadcast status update to "sent"
1879+
mockBroadcastRepo.EXPECT().UpdateBroadcast(gomock.Any(), gomock.Any()).DoAndReturn(func(_ context.Context, b *domain.Broadcast) error {
1880+
assert.Equal(t, domain.BroadcastStatusSent, b.Status)
1881+
return nil
1882+
})
1883+
1884+
// Mock logger calls
1885+
mockLogger.EXPECT().WithField(gomock.Any(), gomock.Any()).Return(mockLogger).AnyTimes()
1886+
mockLogger.EXPECT().WithFields(gomock.Any()).Return(mockLogger).AnyTimes()
1887+
mockLogger.EXPECT().Info(gomock.Any()).AnyTimes()
1888+
mockLogger.EXPECT().Debug(gomock.Any()).AnyTimes()
1889+
mockLogger.EXPECT().Error(gomock.Any()).AnyTimes()
1890+
1891+
// Create orchestrator
1892+
config := &broadcast.Config{
1893+
FetchBatchSize: 100,
1894+
MaxProcessTime: 30 * time.Second,
1895+
ProgressLogInterval: 5 * time.Second,
1896+
}
1897+
orchestrator := broadcast.NewBroadcastOrchestrator(
1898+
mockMessageSender,
1899+
mockBroadcastRepo,
1900+
mockTemplateRepo,
1901+
mockContactRepo,
1902+
mockTaskRepo,
1903+
mockWorkspaceRepo,
1904+
nil, // abTestEvaluator not needed for this test
1905+
mockLogger,
1906+
config,
1907+
mockTimeProvider,
1908+
)
1909+
1910+
// Create task that simulates resuming after test phase completion
1911+
// This is the key: RecipientOffset=1 means test phase already processed 1 recipient
1912+
// Phase="test" but winner is already selected, so should transition to "winner"
1913+
task := &domain.Task{
1914+
ID: "task-123",
1915+
WorkspaceID: "workspace-123",
1916+
Type: "send_broadcast",
1917+
BroadcastID: stringPtr("broadcast-123"),
1918+
State: &domain.TaskState{
1919+
SendBroadcast: &domain.SendBroadcastState{
1920+
BroadcastID: "broadcast-123",
1921+
TotalRecipients: 2, // Total of 2 recipients
1922+
TestPhaseRecipientCount: 1, // Test phase processes 1 recipient (50% of 2)
1923+
WinnerPhaseRecipientCount: 1, // Winner phase should process remaining 1 recipient
1924+
RecipientOffset: 1, // Test phase already processed 1 recipient
1925+
SentCount: 1, // Test phase sent to 1 recipient
1926+
FailedCount: 0,
1927+
Phase: "test", // Should transition to "winner" when processing starts
1928+
TestPhaseCompleted: true,
1929+
},
1930+
},
1931+
}
1932+
1933+
// Process the task
1934+
ctx := context.Background()
1935+
timeout := time.Now().Add(30 * time.Second)
1936+
done, err := orchestrator.Process(ctx, task, timeout)
1937+
1938+
// Verify results
1939+
require.NoError(t, err)
1940+
assert.True(t, done, "Task should be marked as done")
1941+
assert.Equal(t, "winner", task.State.SendBroadcast.Phase, "Phase should be 'winner'")
1942+
assert.Equal(t, int64(2), task.State.SendBroadcast.RecipientOffset, "Should have processed 2 recipients total")
1943+
assert.Equal(t, 2, task.State.SendBroadcast.SentCount, "Should have sent to 2 recipients total")
1944+
assert.Equal(t, 100.0, task.Progress, "Task progress should be 100%")
1945+
}
1946+
17481947
// Helper function to create string pointer
17491948
func stringPtr(s string) *string {
17501949
return &s

tests/testutil/factory.go

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -896,7 +896,6 @@ func (tdf *TestDataFactory) CreateSendBroadcastTask(workspaceID, broadcastID str
896896
FailedCount: 0,
897897
ChannelType: "email",
898898
RecipientOffset: 0,
899-
EndOffset: 100,
900899
Phase: "single",
901900
},
902901
}
@@ -926,7 +925,6 @@ func (tdf *TestDataFactory) CreateTaskWithABTesting(workspaceID, broadcastID str
926925
FailedCount: 0,
927926
ChannelType: "email",
928927
RecipientOffset: 0,
929-
EndOffset: 1000,
930928
Phase: "test",
931929
TestPhaseCompleted: false,
932930
TestPhaseRecipientCount: 100, // 10% for A/B testing

0 commit comments

Comments
 (0)