From 17244bcc9d5000f7a7040d680768d6a7cd69491a Mon Sep 17 00:00:00 2001 From: Rogee Date: Sun, 23 Aug 2026 22:28:43 +0800 Subject: [PATCH] HH-554: connect auto-assignment runtime policy (#135) * fix(HH-554): connect auto-assignment runtime policy * fix(HH-554): bound and coalesce assignment jobs * fix(HH-554): drain assignment backlog safely * fix(HH-554): hold assignment claim through retries --------- Co-authored-by: Rogee --- .../app/autoassignment_listener_test.go | 18 ++ backend/internal/app/bootstrap.go | 1 + .../internal/autoassignment/assignment_job.go | 121 +++++++ .../autoassignment/assignment_job_test.go | 298 ++++++++++++++++++ .../internal/autoassignment/capacity_test.go | 27 +- .../internal/autoassignment/coverage3_test.go | 34 -- .../internal/autoassignment/coverage4_test.go | 24 -- .../internal/autoassignment/coverage5_test.go | 4 - .../internal/autoassignment/coverage9_test.go | 50 ++- .../internal/autoassignment/coverage_test.go | 1 - backend/internal/autoassignment/listener.go | 90 ++++-- backend/internal/autoassignment/model.go | 100 ++---- .../autoassignment/runtime_policy_test.go | 93 ++++++ backend/internal/autoassignment/service.go | 198 ++++++++---- backend/internal/worker/worker.go | 36 ++- 15 files changed, 826 insertions(+), 269 deletions(-) create mode 100644 backend/internal/app/autoassignment_listener_test.go create mode 100644 backend/internal/autoassignment/assignment_job.go create mode 100644 backend/internal/autoassignment/assignment_job_test.go create mode 100644 backend/internal/autoassignment/runtime_policy_test.go diff --git a/backend/internal/app/autoassignment_listener_test.go b/backend/internal/app/autoassignment_listener_test.go new file mode 100644 index 00000000..d800a89a --- /dev/null +++ b/backend/internal/app/autoassignment_listener_test.go @@ -0,0 +1,18 @@ +package app + +import ( + "context" + "testing" + + "github.com/gochat/gochat/internal/autoassignment" + "github.com/gochat/gochat/internal/channel" + "github.com/stretchr/testify/require" +) + +func TestRegisterAutoAssignmentListener(t *testing.T) { + dispatcher := channel.NewDispatcher() + autoassignment.RegisterAutoAssignmentListener(dispatcher, nil, nil) + + err := dispatcher.Dispatch(context.Background(), &channel.ChannelEvent{Type: channel.EventConversationCreated, Data: map[string]any{}}) + require.ErrorContains(t, err, "conversation_id") +} diff --git a/backend/internal/app/bootstrap.go b/backend/internal/app/bootstrap.go index f276853b..71162356 100644 --- a/backend/internal/app/bootstrap.go +++ b/backend/internal/app/bootstrap.go @@ -706,6 +706,7 @@ func Bootstrap(env string) (*App, error) { campaignInternalSvc := campaign.NewCampaignService(db, channelDispatcher) campaignService := service.NewCampaignService(campaignInternalSvc, campaignRepo) assignmentInternalSvc := autoassignment.NewAssignmentService(db, rdb) + autoassignment.RegisterAutoAssignmentListener(channelDispatcher, db, rdb, workerPool) assignmentPolicyService := service.NewAssignmentPolicyService(assignmentPolicyRepo, inboxAssignmentPolicyRepo, assignmentInternalSvc) // SLA Policy service (M11 — SLA Policy + AppliedSLA + SlaEvent + inbox associations) diff --git a/backend/internal/autoassignment/assignment_job.go b/backend/internal/autoassignment/assignment_job.go new file mode 100644 index 00000000..ebf9e37f --- /dev/null +++ b/backend/internal/autoassignment/assignment_job.go @@ -0,0 +1,121 @@ +package autoassignment + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + "github.com/google/uuid" + "github.com/redis/go-redis/v9" +) + +const ( + TaskTypeAssignmentJob = "auto_assignment:assign_inbox" + assignmentInFlightTTL = 5 * time.Minute +) + +var errAssignmentJobUnavailable = errors.New("auto-assignment job requires worker and redis") + +type assignmentJobPayload struct { + InboxID uint `json:"inbox_id"` + AccountID uint `json:"account_id"` + Token string `json:"token"` +} + +// AssignmentJob durably processes one bounded inbox backlog batch. +type AssignmentJob struct { + service *AssignmentService + worker *worker.WorkerPool + redis *redis.Client +} + +func NewAssignmentJob(service *AssignmentService, workerPool *worker.WorkerPool, rdb *redis.Client) *AssignmentJob { + job := &AssignmentJob{service: service, worker: workerPool, redis: rdb} + if workerPool != nil { + workerPool.Register(TaskTypeAssignmentJob, job.perform) + workerPool.RegisterFailureHandler(TaskTypeAssignmentJob, job.afterFailure) + } + return job +} + +func (j *AssignmentJob) EnqueueForInbox(ctx context.Context, inboxID, accountID uint) (bool, error) { + if j.worker == nil || j.redis == nil { + return false, errAssignmentJobUnavailable + } + + payload := assignmentJobPayload{InboxID: inboxID, AccountID: accountID, Token: uuid.NewString()} + claimed, err := j.redis.SetNX(ctx, j.lockKey(inboxID), payload.Token, assignmentInFlightTTL).Result() + if err != nil || !claimed { + return false, err + } + + if _, err := j.worker.Enqueue(ctx, TaskTypeAssignmentJob, payload, worker.WithMaxAttempts(3)); err != nil { + _ = j.release(ctx, inboxID, payload.Token) + return false, err + } + return true, nil +} + +func (j *AssignmentJob) perform(ctx context.Context, backgroundJob *model.BackgroundJob) (err error) { + var payload assignmentJobPayload + if err := json.Unmarshal(backgroundJob.Payload, &payload); err != nil { + return worker.Permanent(fmt.Errorf("unmarshal auto-assignment job: %w", err)) + } + keepToken := false + defer func() { + if keepToken || err != nil { + return + } + _ = j.release(context.WithoutCancel(ctx), payload.InboxID, payload.Token) + }() + + assigned, err := j.service.AssignUnassignedConversations(ctx, payload.InboxID, payload.AccountID) + if err != nil || len(assigned) < assignmentBatchLimit { + return err + } + _, err = j.worker.Enqueue(ctx, TaskTypeAssignmentJob, payload, worker.WithMaxAttempts(3)) + keepToken = err == nil + return err +} + +func (j *AssignmentJob) afterFailure(ctx context.Context, backgroundJob *model.BackgroundJob, retryAfter time.Duration) error { + var payload assignmentJobPayload + if err := json.Unmarshal(backgroundJob.Payload, &payload); err != nil { + return fmt.Errorf("unmarshal auto-assignment failure payload: %w", err) + } + if backgroundJob.Status == model.BackgroundJobStatusDead { + return j.release(ctx, payload.InboxID, payload.Token) + } + if retryAfter < 0 { + retryAfter = 0 + } + return j.renew(ctx, payload.InboxID, payload.Token, assignmentInFlightTTL+retryAfter) +} + +func (j *AssignmentJob) renew(ctx context.Context, inboxID uint, token string, ttl time.Duration) error { + if j.redis == nil || token == "" { + return nil + } + return j.redis.Eval(ctx, + `if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("PEXPIRE", KEYS[1], ARGV[2]) else return 0 end`, + []string{j.lockKey(inboxID)}, token, ttl.Milliseconds(), + ).Err() +} + +func (j *AssignmentJob) release(ctx context.Context, inboxID uint, token string) error { + if j.redis == nil || token == "" { + return nil + } + return j.redis.Eval(ctx, + `if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1]) else return 0 end`, + []string{j.lockKey(inboxID)}, token, + ).Err() +} + +func (j *AssignmentJob) lockKey(inboxID uint) string { + return fmt.Sprintf("gochat:auto_assignment:in_flight:%d", inboxID) +} diff --git a/backend/internal/autoassignment/assignment_job_test.go b/backend/internal/autoassignment/assignment_job_test.go new file mode 100644 index 00000000..e0816966 --- /dev/null +++ b/backend/internal/autoassignment/assignment_job_test.go @@ -0,0 +1,298 @@ +package autoassignment + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/gochat/gochat/internal/channel" + "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + "github.com/stretchr/testify/require" + "gorm.io/gorm" +) + +func TestAssignmentServiceAssignsOnlyTeamMember(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, nonMember, inbox, conversation := seedAssignableConversation_Cov9(t, db) + teamMember := &model.User{AccountID: account.ID, Name: "team agent", Email: "team-agent@example.com", Password: "p", Active: true, Available: true} + require.NoError(t, db.Create(teamMember).Error) + require.NoError(t, db.Create(&model.AccountUser{AccountID: account.ID, UserID: teamMember.ID, Role: "agent"}).Error) + require.NoError(t, db.Create(&model.InboxMember{InboxID: inbox.ID, UserID: teamMember.ID, Role: "agent"}).Error) + team := &model.Team{AccountID: account.ID, Name: "support", AllowAutoAssignment: true} + require.NoError(t, db.Create(team).Error) + require.NoError(t, db.Create(&model.TeamMember{TeamID: team.ID, UserID: teamMember.ID}).Error) + require.NoError(t, db.Model(conversation).Update("team_id", team.ID).Error) + + assigned, err := NewAssignmentService(db, rdb).AssignConversation(context.Background(), conversation.ID, inbox.ID, account.ID) + require.NoError(t, err) + require.Equal(t, teamMember.ID, assigned) + require.NotEqual(t, nonMember.ID, assigned) +} + +func TestAssignmentServiceSkipsTeamWithoutAutoAssignment(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, agent, inbox, conversation := seedAssignableConversation_Cov9(t, db) + team := &model.Team{AccountID: account.ID, Name: "manual only", AllowAutoAssignment: true} + require.NoError(t, db.Create(team).Error) + require.NoError(t, db.Model(team).Update("allow_auto_assignment", false).Error) + require.NoError(t, db.Create(&model.TeamMember{TeamID: team.ID, UserID: agent.ID}).Error) + require.NoError(t, db.Model(conversation).Update("team_id", team.ID).Error) + + assigned, err := NewAssignmentService(db, rdb).AssignConversation(context.Background(), conversation.ID, inbox.ID, account.ID) + require.NoError(t, err) + require.Zero(t, assigned) +} + +func TestAssignmentServiceSkipsSoftDeletedMemberships(t *testing.T) { + t.Run("inbox member", func(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, agent, inbox, conversation := seedAssignableConversation_Cov9(t, db) + require.NoError(t, db.Where("inbox_id = ? AND user_id = ?", inbox.ID, agent.ID).Delete(&model.InboxMember{}).Error) + + assigned, err := NewAssignmentService(db, rdb).AssignConversation(context.Background(), conversation.ID, inbox.ID, account.ID) + require.NoError(t, err) + require.Zero(t, assigned) + }) + + t.Run("team member and team", func(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, agent, inbox, conversation := seedAssignableConversation_Cov9(t, db) + team := &model.Team{AccountID: account.ID, Name: "removed", AllowAutoAssignment: true} + require.NoError(t, db.Create(team).Error) + member := &model.TeamMember{TeamID: team.ID, UserID: agent.ID} + require.NoError(t, db.Create(member).Error) + require.NoError(t, db.Model(conversation).Update("team_id", team.ID).Error) + require.NoError(t, db.Delete(member).Error) + + service := NewAssignmentService(db, rdb) + assigned, err := service.AssignConversation(context.Background(), conversation.ID, inbox.ID, account.ID) + require.NoError(t, err) + require.Zero(t, assigned) + + require.NoError(t, db.Create(&model.TeamMember{TeamID: team.ID, UserID: agent.ID}).Error) + require.NoError(t, db.Delete(team).Error) + assigned, err = service.AssignConversation(context.Background(), conversation.ID, inbox.ID, account.ID) + require.NoError(t, err) + require.Zero(t, assigned) + }) +} + +func TestAssignmentJobDrainsBacklogPastBatchLimit(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, _, inbox, first := seedAssignableConversation_Cov9(t, db) + conversations := make([]model.Conversation, assignmentBatchLimit) + for i := range conversations { + conversations[i] = model.Conversation{ + AccountID: account.ID, + InboxID: inbox.ID, + ContactID: first.ContactID, + Status: string(model.ConversationStatusOpen), + } + } + require.NoError(t, db.Create(&conversations).Error) + policy := &model.AssignmentPolicy{AccountID: account.ID, Name: "batch", FairDistributionLimit: assignmentBatchLimit + 1, Enabled: true} + require.NoError(t, db.Create(policy).Error) + require.NoError(t, db.Create(&model.InboxAssignmentPolicy{InboxID: inbox.ID, AssignmentPolicyID: policy.ID}).Error) + + workerPool := worker.NewWorkerPool(db) + job := NewAssignmentJob(NewAssignmentService(db, rdb), workerPool, rdb) + enqueued, err := job.EnqueueForInbox(context.Background(), inbox.ID, account.ID) + require.NoError(t, err) + require.True(t, enqueued) + processed, err := workerPool.ProcessOne(context.Background()) + require.NoError(t, err) + require.True(t, processed) + require.EqualValues(t, 1, rdb.Exists(context.Background(), job.lockKey(inbox.ID)).Val(), "successor owns the original token") + processed, err = workerPool.ProcessOne(context.Background()) + require.NoError(t, err) + require.True(t, processed) + + var remaining int64 + require.NoError(t, db.Model(&model.Conversation{}). + Where("account_id = ? AND inbox_id = ? AND assignee_id IS NULL", account.ID, inbox.ID). + Count(&remaining).Error) + require.Zero(t, remaining) + require.EqualValues(t, 0, rdb.Exists(context.Background(), job.lockKey(inbox.ID)).Val()) + var jobCount int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeAssignmentJob).Count(&jobCount).Error) + require.EqualValues(t, 2, jobCount) +} + +func TestAssignmentListenerEnqueuesDurableJob(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, agent, inbox, conversation := seedAssignableConversation_Cov9(t, db) + workerPool := worker.NewWorkerPool(db) + listener := NewAutoAssignmentListener(db, rdb, workerPool) + event := channel.NewChannelEvent(channel.EventConversationCreated, channel.ChannelWebWidget, account.ID, inbox.ID) + event.ConversationID = conversation.ID + + require.NoError(t, listener.OnEvent(context.Background(), event)) + require.NoError(t, db.First(conversation, conversation.ID).Error) + require.Nil(t, conversation.AssigneeID, "event request must not scan the backlog synchronously") + + var backgroundJob model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeAssignmentJob).First(&backgroundJob).Error) + require.Equal(t, model.BackgroundJobStatusQueued, backgroundJob.Status) + var payload assignmentJobPayload + require.NoError(t, json.Unmarshal(backgroundJob.Payload, &payload)) + require.Equal(t, inbox.ID, payload.InboxID) + require.Equal(t, account.ID, payload.AccountID) + require.NotEmpty(t, payload.Token) + + require.NoError(t, listener.job.perform(context.Background(), &backgroundJob)) + require.NoError(t, db.First(conversation, conversation.ID).Error) + require.Equal(t, &agent.ID, conversation.AssigneeID) + require.EqualValues(t, 0, rdb.Exists(context.Background(), listener.job.lockKey(inbox.ID)).Val()) +} + +func TestAssignmentJobCoalescesConcurrentInboxEnqueues(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + workerPool := worker.NewWorkerPool(db) + job := NewAssignmentJob(NewAssignmentService(db, rdb), workerPool, rdb) + + const callers = 32 + start := make(chan struct{}) + errs := make(chan error, callers) + var created atomic.Int32 + var wg sync.WaitGroup + for i := 0; i < callers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + enqueued, err := job.EnqueueForInbox(context.Background(), 42, 7) + if err != nil { + errs <- err + return + } + if enqueued { + created.Add(1) + } + }() + } + close(start) + wg.Wait() + close(errs) + for err := range errs { + require.NoError(t, err) + } + require.EqualValues(t, 1, created.Load()) + + var count int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeAssignmentJob).Count(&count).Error) + require.EqualValues(t, 1, count) + require.NoError(t, job.release(context.Background(), 42, fmt.Sprintf("not-%s", rdb.Get(context.Background(), job.lockKey(42)).Val()))) + require.EqualValues(t, 1, rdb.Exists(context.Background(), job.lockKey(42)).Val(), "a stale job must not release a newer claim") +} + +func TestAssignmentJobKeepsTokenUntilRetryDeadLetters(t *testing.T) { + db, _ := setupFullAADB_Cov9(t) + redisServer, rdb := setupTestRedis(t) + account, _, inbox, _ := seedAssignableConversation_Cov9(t, db) + now := time.Now() + workerPool := worker.NewWorkerPoolWithOptions(db, + worker.WithNow(func() time.Time { return now }), + worker.WithBackoff(func(int) time.Duration { return time.Hour }), + ) + job := NewAssignmentJob(NewAssignmentService(db, rdb), workerPool, rdb) + transient := errors.New("temporary database failure") + require.NoError(t, db.Callback().Query().Before("gorm:query").Register("test:auto_assignment_transient", func(tx *gorm.DB) { + if tx.Statement.Table == "inboxes" { + tx.AddError(transient) + } + })) + t.Cleanup(func() { _ = db.Callback().Query().Remove("test:auto_assignment_transient") }) + + enqueued, err := job.EnqueueForInbox(context.Background(), inbox.ID, account.ID) + require.NoError(t, err) + require.True(t, enqueued) + processed, err := workerPool.ProcessOne(context.Background()) + require.True(t, processed) + require.ErrorIs(t, err, transient) + token := rdb.Get(context.Background(), job.lockKey(inbox.ID)).Val() + require.NotEmpty(t, token) + require.Greater(t, redisServer.TTL(job.lockKey(inbox.ID)), time.Hour) + redisServer.FastForward(assignmentInFlightTTL + time.Minute) + + enqueued, err = job.EnqueueForInbox(context.Background(), inbox.ID, account.ID) + require.NoError(t, err) + require.False(t, enqueued, "an event during retry backoff must coalesce") + var jobCount int64 + require.NoError(t, db.Model(&model.BackgroundJob{}).Where("job_type = ?", TaskTypeAssignmentJob).Count(&jobCount).Error) + require.EqualValues(t, 1, jobCount) + require.Equal(t, token, rdb.Get(context.Background(), job.lockKey(inbox.ID)).Val()) + + for range 2 { + now = now.Add(time.Hour) + processed, err = workerPool.ProcessOne(context.Background()) + require.True(t, processed) + require.ErrorIs(t, err, transient) + } + var backgroundJob model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeAssignmentJob).First(&backgroundJob).Error) + require.Equal(t, model.BackgroundJobStatusDead, backgroundJob.Status) + require.EqualValues(t, 0, rdb.Exists(context.Background(), job.lockKey(inbox.ID)).Val()) +} + +func TestAssignmentJobReleasesTokenOnlyAfterDeadIsPersisted(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, _, inbox, _ := seedAssignableConversation_Cov9(t, db) + workerPool := worker.NewWorkerPool(db) + job := NewAssignmentJob(NewAssignmentService(db, rdb), workerPool, rdb) + transient := errors.New("temporary database failure") + require.NoError(t, db.Callback().Query().Before("gorm:query").Register("test:auto_assignment_terminal_failure", func(tx *gorm.DB) { + if tx.Statement.Table == "inboxes" { + tx.AddError(transient) + } + })) + t.Cleanup(func() { _ = db.Callback().Query().Remove("test:auto_assignment_terminal_failure") }) + + enqueued, err := job.EnqueueForInbox(context.Background(), inbox.ID, account.ID) + require.NoError(t, err) + require.True(t, enqueued) + require.NoError(t, db.Model(&model.BackgroundJob{}). + Where("job_type = ?", TaskTypeAssignmentJob). + Update("attempts", 2).Error) + + deadUpdateStarted := make(chan struct{}) + allowDeadUpdate := make(chan struct{}) + var once sync.Once + require.NoError(t, db.Callback().Update().Before("gorm:update").Register("test:pause_assignment_dead", func(tx *gorm.DB) { + updates, ok := tx.Statement.Dest.(map[string]any) + if !ok || updates["status"] != model.BackgroundJobStatusDead { + return + } + once.Do(func() { close(deadUpdateStarted) }) + <-allowDeadUpdate + })) + t.Cleanup(func() { _ = db.Callback().Update().Remove("test:pause_assignment_dead") }) + + result := make(chan error, 1) + go func() { + processed, processErr := workerPool.ProcessOne(context.Background()) + if !processed { + result <- errors.New("expected terminal job to be processed") + return + } + result <- processErr + }() + <-deadUpdateStarted + + enqueued, err = job.EnqueueForInbox(context.Background(), inbox.ID, account.ID) + require.NoError(t, err) + require.False(t, enqueued, "an event before dead persistence must coalesce") + require.EqualValues(t, 1, rdb.Exists(context.Background(), job.lockKey(inbox.ID)).Val()) + + close(allowDeadUpdate) + require.ErrorIs(t, <-result, transient) + var backgroundJob model.BackgroundJob + require.NoError(t, db.Where("job_type = ?", TaskTypeAssignmentJob).First(&backgroundJob).Error) + require.Equal(t, model.BackgroundJobStatusDead, backgroundJob.Status) + require.EqualValues(t, 0, rdb.Exists(context.Background(), job.lockKey(inbox.ID)).Val()) +} diff --git a/backend/internal/autoassignment/capacity_test.go b/backend/internal/autoassignment/capacity_test.go index ed12d3d6..81950403 100644 --- a/backend/internal/autoassignment/capacity_test.go +++ b/backend/internal/autoassignment/capacity_test.go @@ -26,6 +26,8 @@ func setupCapacityAssignmentDB(t *testing.T) *gorm.DB { &model.InboxMember{}, &model.Contact{}, &model.Conversation{}, + &model.AssignmentPolicy{}, + &model.InboxAssignmentPolicy{}, &model.AgentCapacityPolicy{}, &model.InboxCapacityLimit{}, )) @@ -38,7 +40,7 @@ func TestAssignmentService_SkipsAgentsAtInboxCapacity(t *testing.T) { svc := NewAssignmentService(db, rdb) ctx := context.Background() - account := &model.Account{Name: "Capacity Account"} + account := &model.Account{Name: "Capacity Account", FeatureFlags: `{"advanced_assignment":true}`} require.NoError(t, db.Create(account).Error) inbox := &model.Inbox{AccountID: account.ID, Name: "Capacity Inbox", ChannelType: string(model.InboxChannelTypeWebWidget), EnableAutoAssignment: true} require.NoError(t, db.Create(inbox).Error) @@ -68,6 +70,29 @@ func TestAssignmentService_SkipsAgentsAtInboxCapacity(t *testing.T) { assert.Equal(t, agentWithCapacity.ID, agentID) } +func TestAssignmentServiceIgnoresCapacityWhenAdvancedAssignmentIsDisabled(t *testing.T) { + _, rdb := setupTestRedis(t) + db := setupCapacityAssignmentDB(t) + account := &model.Account{Name: "Default Assignment"} + require.NoError(t, db.Create(account).Error) + inbox := &model.Inbox{AccountID: account.ID, Name: "Inbox", ChannelType: string(model.InboxChannelTypeWebWidget), EnableAutoAssignment: true} + require.NoError(t, db.Create(inbox).Error) + agent := createCapacityUser(t, db, account.ID, "default@test.com") + require.NoError(t, db.Create(&model.InboxMember{InboxID: inbox.ID, UserID: agent.ID}).Error) + policy := &model.AgentCapacityPolicy{AccountID: account.ID, Name: "Excluded", ExclusionRules: []byte(`{}`)} + require.NoError(t, db.Create(policy).Error) + require.NoError(t, db.Model(&model.AccountUser{}).Where("account_id = ? AND user_id = ?", account.ID, agent.ID).Update("agent_capacity_policy_id", policy.ID).Error) + require.NoError(t, db.Create(&model.InboxCapacityLimit{AgentCapacityPolicyID: policy.ID, InboxID: inbox.ID, ConversationLimit: 0}).Error) + contact := &model.Contact{AccountID: account.ID, Name: "Contact"} + require.NoError(t, db.Create(contact).Error) + conversation := &model.Conversation{AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, Status: string(model.ConversationStatusOpen)} + require.NoError(t, db.Create(conversation).Error) + + assigned, err := NewAssignmentService(db, rdb).AssignConversation(context.Background(), conversation.ID, inbox.ID, account.ID) + require.NoError(t, err) + require.Equal(t, agent.ID, assigned) +} + func createCapacityUser(t *testing.T, db *gorm.DB, accountID uint, email string) *model.User { t.Helper() user := &model.User{AccountID: accountID, Name: email, Email: email, Password: "hashed", Active: true, Available: true} diff --git a/backend/internal/autoassignment/coverage3_test.go b/backend/internal/autoassignment/coverage3_test.go index 860a2bd6..c294d034 100644 --- a/backend/internal/autoassignment/coverage3_test.go +++ b/backend/internal/autoassignment/coverage3_test.go @@ -8,40 +8,6 @@ import ( "github.com/stretchr/testify/require" ) -func TestEffectivePolicy_All_Cov3(t *testing.T) { - t.Skip("assertion mismatch") - // Both nil - require.Equal(t, AssignmentPolicyType(""), EffectivePolicy(nil, nil)) - // Account only - ap := &AssignmentPolicy{Policy: "round_robin"} - require.Equal(t, AssignmentPolicyType("round_robin"), EffectivePolicy(ap, nil)) - // Inbox overrides - ip := &InboxAssignmentPolicy{Policy: "lowest_load"} - require.Equal(t, AssignmentPolicyType("lowest_load"), EffectivePolicy(ap, ip)) - // Inbox only - require.Equal(t, AssignmentPolicyType("lowest_load"), EffectivePolicy(nil, ip)) -} - -func TestEffectiveLimit_All_Cov3(t *testing.T) { - t.Skip("assertion mismatch") - require.Equal(t, 0, EffectiveLimit(nil, nil)) - ap := &AssignmentPolicy{FairDistributionLimit: 5} - require.Equal(t, 5, EffectiveLimit(ap, nil)) - ip := &InboxAssignmentPolicy{FairDistributionLimit: 10} - require.Equal(t, 10, EffectiveLimit(ap, ip)) - require.Equal(t, 10, EffectiveLimit(nil, ip)) -} - -func TestEffectiveWindow_All_Cov3(t *testing.T) { - t.Skip("assertion mismatch") - require.Equal(t, 0, EffectiveWindow(nil, nil)) - ap := &AssignmentPolicy{FairDistributionWindow: 60} - require.Equal(t, 60, EffectiveWindow(ap, nil)) - ip := &InboxAssignmentPolicy{FairDistributionWindow: 120} - require.Equal(t, 120, EffectiveWindow(ap, ip)) - require.Equal(t, 120, EffectiveWindow(nil, ip)) -} - func TestAutoAssignmentListener_GetConversation_Cov3(t *testing.T) { db := setupAADB_Cov2(t) l := &AutoAssignmentListener{db: db} diff --git a/backend/internal/autoassignment/coverage4_test.go b/backend/internal/autoassignment/coverage4_test.go index 0d3b8d96..7886f2ec 100644 --- a/backend/internal/autoassignment/coverage4_test.go +++ b/backend/internal/autoassignment/coverage4_test.go @@ -154,27 +154,3 @@ func TestAutoAssignmentListener_OnConversationUnassigned_Success_Cov4(t *testing defer func() { _ = recover() }() _ = l.onConversationUnassigned(context.Background(), nil) } - -func TestEffectivePolicy_All_Cov4(t *testing.T) { - t.Skip("assertion mismatch") - require.Equal(t, AssignmentPolicyType(""), EffectivePolicy(nil, nil)) - ap := &AssignmentPolicy{Policy: "round_robin"} - require.Equal(t, AssignmentPolicyType("round_robin"), EffectivePolicy(ap, nil)) - ip := &InboxAssignmentPolicy{Policy: "lowest_load"} - require.Equal(t, AssignmentPolicyType("lowest_load"), EffectivePolicy(ap, ip)) - require.Equal(t, AssignmentPolicyType("lowest_load"), EffectivePolicy(nil, ip)) -} - -func TestEffectiveLimit_All_Cov4(t *testing.T) { - t.Skip("assertion mismatch") - require.Equal(t, 0, EffectiveLimit(nil, nil)) - ap := &AssignmentPolicy{FairDistributionLimit: 5} - require.Equal(t, 5, EffectiveLimit(ap, nil)) -} - -func TestEffectiveWindow_All_Cov4(t *testing.T) { - t.Skip("assertion mismatch") - require.Equal(t, 0, EffectiveWindow(nil, nil)) - ap := &AssignmentPolicy{FairDistributionWindow: 60} - require.Equal(t, 60, EffectiveWindow(ap, nil)) -} diff --git a/backend/internal/autoassignment/coverage5_test.go b/backend/internal/autoassignment/coverage5_test.go index 8d955187..987ff00c 100644 --- a/backend/internal/autoassignment/coverage5_test.go +++ b/backend/internal/autoassignment/coverage5_test.go @@ -29,10 +29,6 @@ func TestConversationQueryModel_TableName_Cov5(t *testing.T) { assert.Equal(t, "conversations", ConversationQueryModel{}.TableName()) } -func TestAssignmentPolicy_TableName_Cov5(t *testing.T) { - assert.Equal(t, "assignment_policies", AssignmentPolicy{}.TableName()) -} - func TestNewLowestLoadSelector_Cov5(t *testing.T) { s := NewLowestLoadSelector(nil) assert.NotNil(t, s) diff --git a/backend/internal/autoassignment/coverage9_test.go b/backend/internal/autoassignment/coverage9_test.go index 1305d2ae..65cc81fb 100644 --- a/backend/internal/autoassignment/coverage9_test.go +++ b/backend/internal/autoassignment/coverage9_test.go @@ -28,7 +28,8 @@ func migrateFullAADB_Cov9(t *testing.T, db *gorm.DB) { &model.Account{}, &model.User{}, &model.AccountUser{}, &model.Inbox{}, &model.InboxMember{}, &model.Contact{}, &model.ContactInbox{}, &model.Conversation{}, &model.Message{}, - &AssignmentPolicy{}, &InboxAssignmentPolicy{}, + &model.Team{}, &model.TeamMember{}, &model.BackgroundJob{}, + &model.AssignmentPolicy{}, &model.InboxAssignmentPolicy{}, &model.AgentCapacityPolicy{}, &model.InboxCapacityLimit{}, )) } @@ -110,29 +111,16 @@ func seedAssignableConversation_Cov9(t *testing.T, db *gorm.DB) (*model.Account, return acc, agent, inbox, conv } -func TestEffectivePolicyLimitWindow_ActiveInactive_Cov9(t *testing.T) { - activeAccount := &AssignmentPolicy{Policy: PolicyLongestWaiting, FairDistributionLimit: 11, FairDistributionWindow: 111, Active: true} - inactiveAccount := &AssignmentPolicy{Policy: PolicyLowestLoad, FairDistributionLimit: 22, FairDistributionWindow: 222, Active: false} - activeInbox := &InboxAssignmentPolicy{Policy: PolicyLowestLoad, FairDistributionLimit: 33, FairDistributionWindow: 333, Active: true} - inactiveInbox := &InboxAssignmentPolicy{Policy: PolicyLowestLoad, FairDistributionLimit: 44, FairDistributionWindow: 444, Active: false} +func TestEffectivePolicyLimitWindow_Cov9(t *testing.T) { + policy := &model.AssignmentPolicy{AssignmentOrder: 1, FairDistributionLimit: 11, FairDistributionWindow: 111, Enabled: true} - require.Equal(t, PolicyRoundRobin, EffectivePolicy(nil, nil)) - require.Equal(t, PolicyLongestWaiting, EffectivePolicy(activeAccount, nil)) - require.Equal(t, PolicyRoundRobin, EffectivePolicy(inactiveAccount, nil)) - require.Equal(t, PolicyLowestLoad, EffectivePolicy(activeAccount, activeInbox)) - require.Equal(t, PolicyLongestWaiting, EffectivePolicy(activeAccount, inactiveInbox)) - - require.Equal(t, 5, EffectiveLimit(nil, nil)) - require.Equal(t, 11, EffectiveLimit(activeAccount, nil)) - require.Equal(t, 22, EffectiveLimit(inactiveAccount, nil)) - require.Equal(t, 33, EffectiveLimit(activeAccount, activeInbox)) - require.Equal(t, 11, EffectiveLimit(activeAccount, inactiveInbox)) - - require.Equal(t, 300, EffectiveWindow(nil, nil)) - require.Equal(t, 111, EffectiveWindow(activeAccount, nil)) - require.Equal(t, 222, EffectiveWindow(inactiveAccount, nil)) - require.Equal(t, 333, EffectiveWindow(activeAccount, activeInbox)) - require.Equal(t, 111, EffectiveWindow(activeAccount, inactiveInbox)) + require.Equal(t, PolicyRoundRobin, EffectivePolicy(nil, false)) + require.Equal(t, PolicyRoundRobin, EffectivePolicy(policy, false)) + require.Equal(t, PolicyLowestLoad, EffectivePolicy(policy, true)) + require.Equal(t, 5, EffectiveLimit(nil)) + require.Equal(t, 11, EffectiveLimit(policy)) + require.Equal(t, 300, EffectiveWindow(nil)) + require.Equal(t, 111, EffectiveWindow(policy)) } func TestAssignmentService_AssignConversation_Success_Cov9(t *testing.T) { @@ -172,7 +160,7 @@ func TestAssignmentServiceRejectsCandidateDeactivatedAfterSelection(t *testing.T db, rdb := setupFullAADB_Cov9(t) account, agent, inbox, conversation := seedAssignableConversation_Cov9(t, db) svc := NewAssignmentService(db, rdb) - agents, err := svc.getEligibleAgents(context.Background(), inbox.ID, account.ID) + agents, err := svc.getEligibleAgents(context.Background(), inbox.ID, account.ID, nil, false) require.NoError(t, err) require.Equal(t, []uint{agent.ID}, agents) @@ -314,7 +302,7 @@ func TestAssignmentServiceExcludesNonAgentInboxMembers(t *testing.T) { Where("account_id = ? AND user_id = ?", account.ID, agent.ID). Update("role", "member").Error) - agents, err := NewAssignmentService(db, rdb).getEligibleAgents(context.Background(), inbox.ID, account.ID) + agents, err := NewAssignmentService(db, rdb).getEligibleAgents(context.Background(), inbox.ID, account.ID, nil, false) require.NoError(t, err) require.Empty(t, agents) } @@ -323,20 +311,20 @@ func TestAssignmentService_GetAndPolicyHelpers_Cov9(t *testing.T) { db, rdb := setupFullAADB_Cov9(t) acc, _, inbox, conv := seedAssignableConversation_Cov9(t, db) svc := NewAssignmentService(db, rdb) - policy := &AssignmentPolicy{AccountID: acc.ID, Policy: PolicyLowestLoad, FairDistributionLimit: 7, FairDistributionWindow: 77, Active: true} + policy := &model.AssignmentPolicy{AccountID: acc.ID, Name: "Runtime", AssignmentOrder: 1, FairDistributionLimit: 7, FairDistributionWindow: 77, Enabled: true} require.NoError(t, db.Create(policy).Error) - inboxPolicy := &InboxAssignmentPolicy{AccountID: acc.ID, InboxID: inbox.ID, Policy: PolicyLongestWaiting, FairDistributionLimit: 8, FairDistributionWindow: 88, Active: true} - require.NoError(t, db.Create(inboxPolicy).Error) + require.NoError(t, db.Create(&model.InboxAssignmentPolicy{InboxID: inbox.ID, AssignmentPolicyID: policy.ID}).Error) loadedInbox, err := svc.getInbox(context.Background(), inbox.ID) require.NoError(t, err) require.Equal(t, inbox.ID, loadedInbox.ID) - convs, err := svc.findUnassignedConversations(context.Background(), inbox.ID, acc.ID) + convs, err := svc.findUnassignedConversations(context.Background(), inbox.ID, acc.ID, policy) require.NoError(t, err) require.NotEmpty(t, convs) _ = conv - require.Equal(t, PolicyLowestLoad, svc.getAccountPolicy(context.Background(), acc.ID).Policy) - require.Equal(t, PolicyLongestWaiting, svc.getInboxPolicy(context.Background(), inbox.ID).Policy) + loadedPolicy, err := svc.getInboxPolicy(context.Background(), acc.ID, inbox.ID) + require.NoError(t, err) + require.Equal(t, policy.ID, loadedPolicy.ID) } func TestAssignmentService_AgentHasInboxCapacity_Limited_Cov9(t *testing.T) { diff --git a/backend/internal/autoassignment/coverage_test.go b/backend/internal/autoassignment/coverage_test.go index 7b70c489..4a6ffe26 100644 --- a/backend/internal/autoassignment/coverage_test.go +++ b/backend/internal/autoassignment/coverage_test.go @@ -37,7 +37,6 @@ func TestConversationQueryModel_TableName(t *testing.T) { func TestAssignmentPolicyType_Constants(t *testing.T) { assert.Equal(t, "round_robin", string(PolicyRoundRobin)) - assert.Equal(t, "longest_waiting", string(PolicyLongestWaiting)) assert.Equal(t, "lowest_load", string(PolicyLowestLoad)) } diff --git a/backend/internal/autoassignment/listener.go b/backend/internal/autoassignment/listener.go index aa8b50c6..1f4fc4be 100644 --- a/backend/internal/autoassignment/listener.go +++ b/backend/internal/autoassignment/listener.go @@ -21,26 +21,36 @@ import ( "github.com/gochat/gochat/internal/channel" "github.com/gochat/gochat/internal/dispatch" "github.com/gochat/gochat/internal/model" + "github.com/gochat/gochat/internal/worker" + applogger "github.com/gochat/gochat/pkg/logger" "github.com/redis/go-redis/v9" "gorm.io/gorm" - applogger "github.com/gochat/gochat/pkg/logger" ) // AutoAssignmentListener listens for conversation events and triggers // auto-assignment. type AutoAssignmentListener struct { - db *gorm.DB - redis *redis.Client - service *AssignmentService + db *gorm.DB + service *AssignmentService + job *AssignmentJob } // NewAutoAssignmentListener creates a new AutoAssignmentListener. -func NewAutoAssignmentListener(db *gorm.DB, rdb *redis.Client) *AutoAssignmentListener { - return &AutoAssignmentListener{ +func NewAutoAssignmentListener(db *gorm.DB, rdb *redis.Client, workers ...*worker.WorkerPool) *AutoAssignmentListener { + listener := &AutoAssignmentListener{ db: db, - redis: rdb, service: NewAssignmentService(db, rdb), } + if len(workers) > 0 && workers[0] != nil { + listener.job = NewAssignmentJob(listener.service, workers[0], rdb) + } + return listener +} + +// RegisterAutoAssignmentListener registers the listener once in the application's +// name-keyed dispatcher. +func RegisterAutoAssignmentListener(dispatcher *channel.Dispatcher, db *gorm.DB, rdb *redis.Client, workers ...*worker.WorkerPool) { + dispatcher.Register(NewAutoAssignmentListener(db, rdb, workers...)) } // Name returns the unique identifier for this listener. @@ -66,28 +76,32 @@ func (l *AutoAssignmentListener) OnEvent(ctx context.Context, event *channel.Cha // onConversationCreated handles new conversation events. // It triggers auto-assignment for the conversation's inbox. func (l *AutoAssignmentListener) onConversationCreated(ctx context.Context, event *channel.ChannelEvent) error { - conversationID, err := dispatch.ExtractConversationID(event.Data) + conversationID, err := eventConversationID(event) if err != nil { return fmt.Errorf("extract conversation_id: %w", err) } - inboxID, err := dispatch.ExtractInboxID(event.Data) + inboxID, err := eventInboxID(event) if err != nil { return fmt.Errorf("extract inbox_id: %w", err) } - accountID, err := dispatch.ExtractAccountID(event.Data) + accountID, err := eventAccountID(event) if err != nil { return fmt.Errorf("extract account_id: %w", err) } applogger.L().Infof("auto-assignment: new conversation %d in inbox %d", conversationID, inboxID) - assignedAgent, err := l.service.AssignConversation(ctx, conversationID, inboxID, accountID) + assignedAgent, err := l.assignInbox(ctx, conversationID, inboxID, accountID) if err != nil { applogger.L().Errorf("auto-assignment failed for conversation %d: %v", conversationID, err) return err } + if l.job != nil { + applogger.L().Infof("auto-assignment: inbox %d backlog job queued or coalesced", inboxID) + return nil + } if assignedAgent > 0 { applogger.L().Infof("auto-assignment: conversation %d assigned to agent %d", conversationID, assignedAgent) @@ -101,7 +115,7 @@ func (l *AutoAssignmentListener) onConversationCreated(ctx context.Context, even // onConversationOpened handles reopened conversation events. // It triggers auto-assignment only if the conversation has no assignee. func (l *AutoAssignmentListener) onConversationOpened(ctx context.Context, event *channel.ChannelEvent) error { - conversationID, err := dispatch.ExtractConversationID(event.Data) + conversationID, err := eventConversationID(event) if err != nil { return fmt.Errorf("extract conversation_id: %w", err) } @@ -117,19 +131,19 @@ func (l *AutoAssignmentListener) onConversationOpened(ctx context.Context, event return nil // already assigned } - inboxID, err := dispatch.ExtractInboxID(event.Data) + inboxID, err := eventInboxID(event) if err != nil { inboxID = conv.InboxID // fallback to conversation's inbox } - accountID, err := dispatch.ExtractAccountID(event.Data) + accountID, err := eventAccountID(event) if err != nil { accountID = conv.AccountID // fallback to conversation's account } applogger.L().Infof("auto-assignment: reopened conversation %d in inbox %d", conversationID, inboxID) - assignedAgent, err := l.service.AssignConversation(ctx, conversationID, inboxID, accountID) + assignedAgent, err := l.assignInbox(ctx, conversationID, inboxID, accountID) if err != nil { applogger.L().Errorf("auto-assignment failed for conversation %d: %v", conversationID, err) return err @@ -145,24 +159,24 @@ func (l *AutoAssignmentListener) onConversationOpened(ctx context.Context, event // onConversationUnassigned handles unassigned conversation events. // It triggers re-assignment for the conversation. func (l *AutoAssignmentListener) onConversationUnassigned(ctx context.Context, event *channel.ChannelEvent) error { - conversationID, err := dispatch.ExtractConversationID(event.Data) + conversationID, err := eventConversationID(event) if err != nil { return fmt.Errorf("extract conversation_id: %w", err) } - inboxID, err := dispatch.ExtractInboxID(event.Data) + inboxID, err := eventInboxID(event) if err != nil { return fmt.Errorf("extract inbox_id: %w", err) } - accountID, err := dispatch.ExtractAccountID(event.Data) + accountID, err := eventAccountID(event) if err != nil { return fmt.Errorf("extract account_id: %w", err) } applogger.L().Infof("auto-assignment: unassigned conversation %d in inbox %d", conversationID, inboxID) - assignedAgent, err := l.service.AssignConversation(ctx, conversationID, inboxID, accountID) + assignedAgent, err := l.assignInbox(ctx, conversationID, inboxID, accountID) if err != nil { applogger.L().Errorf("auto-assignment re-assign failed for conversation %d: %v", conversationID, err) return err @@ -184,6 +198,42 @@ func (l *AutoAssignmentListener) getConversation(ctx context.Context, id uint) ( return &conv, nil } +func (l *AutoAssignmentListener) assignInbox(ctx context.Context, conversationID, inboxID, accountID uint) (uint, error) { + if l.job != nil { + _, err := l.job.EnqueueForInbox(ctx, inboxID, accountID) + return 0, err + } + if _, err := l.service.AssignUnassignedConversations(ctx, inboxID, accountID); err != nil { + return 0, err + } + conversation, err := l.getConversation(ctx, conversationID) + if err != nil || conversation.AssigneeID == nil { + return 0, err + } + return *conversation.AssigneeID, nil +} + +func eventConversationID(event *channel.ChannelEvent) (uint, error) { + if event.ConversationID != 0 { + return event.ConversationID, nil + } + return dispatch.ExtractConversationID(event.Data) +} + +func eventInboxID(event *channel.ChannelEvent) (uint, error) { + if event.InboxID != 0 { + return event.InboxID, nil + } + return dispatch.ExtractInboxID(event.Data) +} + +func eventAccountID(event *channel.ChannelEvent) (uint, error) { + if event.AccountID != 0 { + return event.AccountID, nil + } + return dispatch.ExtractAccountID(event.Data) +} + // EventNames returns the event names this listener subscribes to. // Used when registering with the EventDispatcher. func EventNames() []string { @@ -192,4 +242,4 @@ func EventNames() []string { string(channel.EventConversationOpened), string(channel.EventConversationUnassigned), } -} \ No newline at end of file +} diff --git a/backend/internal/autoassignment/model.go b/backend/internal/autoassignment/model.go index 3d6eea26..43b944d8 100644 --- a/backend/internal/autoassignment/model.go +++ b/backend/internal/autoassignment/model.go @@ -1,20 +1,6 @@ package autoassignment -// GORM models for the auto-assignment system. -// -// Reference: Chatwoot AutoAssignment pattern -// - AssignmentPolicy: global policy for how conversations are assigned -// (round_robin, longest_waiting). Has fair_distribution_limit and -// fair_distribution_window for rate limiting per agent. -// - InboxAssignmentPolicy: per-inbox override of the global policy. -// Each inbox can have its own assignment policy and rate limits. -// -// Pattern: gochat models embed model.Base (ID, CreatedAt, UpdatedAt, DeletedAt), -// use TableName() method, and use gorm struct tags. - -import ( - "github.com/gochat/gochat/internal/model" -) +import "github.com/gochat/gochat/internal/model" // AssignmentPolicyType defines the type of auto-assignment policy. type AssignmentPolicyType string @@ -24,79 +10,33 @@ const ( // Reference: Chatwoot InboxRoundRobinService PolicyRoundRobin AssignmentPolicyType = "round_robin" - // PolicyLongestWaiting assigns conversations to the agent with the - // longest idle time since their last assignment. - // Reference: Chatwoot "longest_waiting" policy (planned feature) - PolicyLongestWaiting AssignmentPolicyType = "longest_waiting" - // PolicyLowestLoad assigns conversations to the agent with the // fewest currently open conversations. - // Reference: Chatwoot "least_busy" concept — agent with lowest workload. + // This implements Chatwoot's balanced assignment order. PolicyLowestLoad AssignmentPolicyType = "lowest_load" ) -// AssignmentPolicy represents the global auto-assignment policy for an account. -// Reference: Chatwoot AssignmentPolicy model -// - Defines how unassigned conversations are distributed among agents -// - fair_distribution_limit: max assignments per agent per window (default 5) -// - fair_distribution_window: time window in seconds for rate limiting (default 300 = 5 min) -type AssignmentPolicy struct { - model.Base - AccountID uint `gorm:"index;not null" json:"account_id"` - Policy AssignmentPolicyType `gorm:"size:50;default:round_robin" json:"policy"` - FairDistributionLimit int `gorm:"default:5" json:"fair_distribution_limit"` - FairDistributionWindow int `gorm:"default:300" json:"fair_distribution_window"` // seconds - Active bool `gorm:"default:true" json:"active"` +// EffectivePolicy maps the persisted Chatwoot assignment order to its runtime +// selector. Balanced assignment remains feature-gated like upstream Chatwoot. +func EffectivePolicy(policy *model.AssignmentPolicy, advancedAssignment bool) AssignmentPolicyType { + if policy != nil && policy.AssignmentOrder == 1 && advancedAssignment { + return PolicyLowestLoad + } + return PolicyRoundRobin } -func (AssignmentPolicy) TableName() string { return "assignment_policies" } - -// InboxAssignmentPolicy represents a per-inbox override of the global assignment policy. -// Reference: Chatwoot InboxAssignmentPolicy model -// - Each inbox can override the account-level policy -// - If no inbox-specific policy exists, the account policy is used -type InboxAssignmentPolicy struct { - model.Base - AccountID uint `gorm:"index;not null" json:"account_id"` - InboxID uint `gorm:"uniqueIndex;not null" json:"inbox_id"` - Policy AssignmentPolicyType `gorm:"size:50;default:round_robin" json:"policy"` - FairDistributionLimit int `gorm:"default:5" json:"fair_distribution_limit"` - FairDistributionWindow int `gorm:"default:300" json:"fair_distribution_window"` // seconds - Active bool `gorm:"default:true" json:"active"` +// EffectiveLimit returns the linked policy's fair-distribution limit. +func EffectiveLimit(policy *model.AssignmentPolicy) int { + if policy != nil && policy.FairDistributionLimit > 0 { + return policy.FairDistributionLimit + } + return 5 } -func (InboxAssignmentPolicy) TableName() string { return "inbox_assignment_policies" } - -// EffectivePolicy returns the effective assignment policy for a given inbox. -// If the inbox has a specific policy, it is used; otherwise the account policy is used. -func EffectivePolicy(accountPolicy *AssignmentPolicy, inboxPolicy *InboxAssignmentPolicy) AssignmentPolicyType { - if inboxPolicy != nil && inboxPolicy.Active { - return inboxPolicy.Policy +// EffectiveWindow returns the linked policy's rate-limit window in seconds. +func EffectiveWindow(policy *model.AssignmentPolicy) int { + if policy != nil && policy.FairDistributionWindow > 0 { + return policy.FairDistributionWindow } - if accountPolicy != nil && accountPolicy.Active { - return accountPolicy.Policy - } - return PolicyRoundRobin // default + return 300 } - -// EffectiveLimit returns the effective fair distribution limit for a given inbox. -func EffectiveLimit(accountPolicy *AssignmentPolicy, inboxPolicy *InboxAssignmentPolicy) int { - if inboxPolicy != nil && inboxPolicy.Active && inboxPolicy.FairDistributionLimit > 0 { - return inboxPolicy.FairDistributionLimit - } - if accountPolicy != nil && accountPolicy.FairDistributionLimit > 0 { - return accountPolicy.FairDistributionLimit - } - return 5 // default -} - -// EffectiveWindow returns the effective fair distribution window (in seconds). -func EffectiveWindow(accountPolicy *AssignmentPolicy, inboxPolicy *InboxAssignmentPolicy) int { - if inboxPolicy != nil && inboxPolicy.Active && inboxPolicy.FairDistributionWindow > 0 { - return inboxPolicy.FairDistributionWindow - } - if accountPolicy != nil && accountPolicy.FairDistributionWindow > 0 { - return accountPolicy.FairDistributionWindow - } - return 300 // default (5 minutes) -} \ No newline at end of file diff --git a/backend/internal/autoassignment/runtime_policy_test.go b/backend/internal/autoassignment/runtime_policy_test.go new file mode 100644 index 00000000..73e2ca6b --- /dev/null +++ b/backend/internal/autoassignment/runtime_policy_test.go @@ -0,0 +1,93 @@ +package autoassignment + +import ( + "context" + "testing" + "time" + + "github.com/gochat/gochat/internal/channel" + "github.com/gochat/gochat/internal/model" + "github.com/stretchr/testify/require" +) + +func TestAutoAssignmentListenerUsesChannelEventIDs(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, agent, inbox, conversation := seedAssignableConversation_Cov9(t, db) + + event := channel.NewChannelEvent(channel.EventConversationCreated, channel.ChannelWebWidget, account.ID, inbox.ID) + event.ConversationID = conversation.ID + require.NoError(t, NewAutoAssignmentListener(db, rdb).OnEvent(context.Background(), event)) + + require.NoError(t, db.First(conversation, conversation.ID).Error) + require.Equal(t, &agent.ID, conversation.AssigneeID) +} + +func TestAssignmentServiceUsesLinkedCurrentPolicy(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, firstAgent, inbox, conversation := seedAssignableConversation_Cov9(t, db) + secondAgent := &model.User{AccountID: account.ID, Name: "second", Email: "second-policy@example.com", Password: "p", Active: true, Available: true} + require.NoError(t, db.Create(secondAgent).Error) + require.NoError(t, db.Create(&model.AccountUser{AccountID: account.ID, UserID: secondAgent.ID, Role: "agent"}).Error) + require.NoError(t, db.Create(&model.InboxMember{InboxID: inbox.ID, UserID: secondAgent.ID, Role: "agent", AvailabilityStatus: "online"}).Error) + + contact := &model.Contact{AccountID: account.ID, Name: "load", Email: "load-policy@example.com"} + require.NoError(t, db.Create(contact).Error) + load := &model.Conversation{AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, AssigneeID: &firstAgent.ID, Status: string(model.ConversationStatusOpen)} + require.NoError(t, db.Create(load).Error) + + require.NoError(t, db.Model(account).Update("feature_flags", `{"advanced_assignment":true}`).Error) + policy := &model.AssignmentPolicy{AccountID: account.ID, Name: "Runtime", AssignmentOrder: 1, FairDistributionLimit: 100, FairDistributionWindow: 3600, Enabled: true} + require.NoError(t, db.Create(policy).Error) + require.NoError(t, db.Create(&model.InboxAssignmentPolicy{InboxID: inbox.ID, AssignmentPolicyID: policy.ID}).Error) + + service := NewAssignmentService(db, rdb) + assigned, err := service.AssignConversation(context.Background(), conversation.ID, inbox.ID, account.ID) + require.NoError(t, err) + require.Equal(t, secondAgent.ID, assigned, "balanced policy should choose the least-loaded agent") + + policy.AssignmentOrder = 0 + require.NoError(t, db.Save(policy).Error) + next := &model.Conversation{AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, Status: string(model.ConversationStatusOpen)} + require.NoError(t, db.Create(next).Error) + assigned, err = service.AssignConversation(context.Background(), next.ID, inbox.ID, account.ID) + require.NoError(t, err) + require.Equal(t, firstAgent.ID, assigned, "round-robin policy should use the queue order") +} + +func TestListenerAppliesConversationPriorityToInboxBacklog(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, agent, inbox, newest := seedAssignableConversation_Cov9(t, db) + now := time.Now().Unix() + require.NoError(t, db.Model(newest).Update("last_activity_at", now).Error) + + contact := &model.Contact{AccountID: account.ID, Name: "old", Email: "old-policy@example.com"} + require.NoError(t, db.Create(contact).Error) + oldActivity := now - int64((2 * time.Hour).Seconds()) + oldest := &model.Conversation{AccountID: account.ID, InboxID: inbox.ID, ContactID: contact.ID, Status: string(model.ConversationStatusOpen), LastActivityAt: &oldActivity} + require.NoError(t, db.Create(oldest).Error) + policy := &model.AssignmentPolicy{AccountID: account.ID, Name: "Oldest first", ConversationPriority: 1, FairDistributionLimit: 1, FairDistributionWindow: 3600, Enabled: true} + require.NoError(t, db.Create(policy).Error) + require.NoError(t, db.Create(&model.InboxAssignmentPolicy{InboxID: inbox.ID, AssignmentPolicyID: policy.ID}).Error) + + event := channel.NewChannelEvent(channel.EventConversationCreated, channel.ChannelWebWidget, account.ID, inbox.ID) + event.ConversationID = newest.ID + require.NoError(t, NewAutoAssignmentListener(db, rdb).OnEvent(context.Background(), event)) + + require.NoError(t, db.First(oldest, oldest.ID).Error) + require.NoError(t, db.First(newest, newest.ID).Error) + require.Equal(t, &agent.ID, oldest.AssigneeID) + require.Nil(t, newest.AssigneeID) +} + +func TestDisabledLinkedPolicyStopsAssignment(t *testing.T) { + db, rdb := setupFullAADB_Cov9(t) + account, _, inbox, conversation := seedAssignableConversation_Cov9(t, db) + policy := &model.AssignmentPolicy{AccountID: account.ID, Name: "Paused", Enabled: true} + require.NoError(t, db.Create(policy).Error) + require.NoError(t, db.Model(policy).Update("enabled", false).Error) + require.NoError(t, db.Create(&model.InboxAssignmentPolicy{InboxID: inbox.ID, AssignmentPolicyID: policy.ID}).Error) + + assigned, err := NewAssignmentService(db, rdb).AssignConversation(context.Background(), conversation.ID, inbox.ID, account.ID) + require.NoError(t, err) + require.Zero(t, assigned) +} diff --git a/backend/internal/autoassignment/service.go b/backend/internal/autoassignment/service.go index 7b809942..b597e4d9 100644 --- a/backend/internal/autoassignment/service.go +++ b/backend/internal/autoassignment/service.go @@ -16,8 +16,11 @@ package autoassignment import ( "context" + "encoding/json" "errors" "fmt" + "strings" + "time" "github.com/gochat/gochat/internal/model" "github.com/gochat/gochat/internal/repository" @@ -26,6 +29,8 @@ import ( "gorm.io/gorm" ) +const assignmentBatchLimit = 100 + // AssignmentService handles auto-assignment of conversations to agents. type AssignmentService struct { db *gorm.DB @@ -70,8 +75,21 @@ func (s *AssignmentService) AssignUnassignedConversations(ctx context.Context, i return nil, nil } - // Step 2: Find unassigned open conversations - conversations, err := s.findUnassignedConversations(ctx, inboxID, accountID) + // Step 2: Load the policy associated with this inbox. + policy, err := s.getInboxPolicy(ctx, accountID, inboxID) + if err != nil { + return nil, fmt.Errorf("get assignment policy: %w", err) + } + if policy != nil && !policy.Enabled { + return nil, nil + } + advancedAssignment, err := s.advancedAssignmentEnabled(ctx, accountID) + if err != nil { + return nil, fmt.Errorf("get advanced assignment feature: %w", err) + } + + // Step 3: Find unassigned open conversations in policy order. + conversations, err := s.findUnassignedConversations(ctx, inboxID, accountID, policy) if err != nil { return nil, fmt.Errorf("find unassigned: %w", err) } @@ -79,30 +97,24 @@ func (s *AssignmentService) AssignUnassignedConversations(ctx context.Context, i return nil, nil } - // Step 3: Get eligible agents - agents, err := s.getEligibleAgents(ctx, inboxID, accountID) - if err != nil { - return nil, fmt.Errorf("get eligible agents: %w", err) - } - if len(agents) == 0 { - applogger.L().Infof("no eligible agents for inbox %d", inboxID) - return nil, nil - } + // Step 4: Resolve selector and rate limits from the linked policy. + selector := EffectivePolicy(policy, advancedAssignment) + limit := EffectiveLimit(policy) + window := EffectiveWindow(policy) - // Step 4: Get assignment policy and limits - accountPolicy := s.getAccountPolicy(ctx, accountID) - inboxPolicy := s.getInboxPolicy(ctx, inboxID) - policy := EffectivePolicy(accountPolicy, inboxPolicy) - limit := EffectiveLimit(accountPolicy, inboxPolicy) - window := EffectiveWindow(accountPolicy, inboxPolicy) - - // Step 5: Reset round-robin queue if members changed - s.roundRobin.SyncQueue(ctx, inboxID, agents) - - // Step 6: Assign each conversation + // Step 5: Assign each conversation using its team-scoped candidates. assignedIDs := make([]uint, 0) for _, conv := range conversations { - agentID, err := s.selectAgent(ctx, inboxID, agents, policy, limit, window) + agents, err := s.getEligibleAgents(ctx, inboxID, accountID, conv.TeamID, advancedAssignment) + if err != nil { + return assignedIDs, fmt.Errorf("get eligible agents for conversation %d: %w", conv.ID, err) + } + if len(agents) == 0 { + continue + } + + s.roundRobin.SyncQueue(ctx, inboxID, agents) + agentID, err := s.selectAgent(ctx, inboxID, agents, selector, limit, window) if err != nil { applogger.L().Warnf("failed to select agent for conversation %d: %v", conv.ID, err) continue @@ -150,8 +162,31 @@ func (s *AssignmentService) AssignConversation(ctx context.Context, conversation return 0, nil } + policy, err := s.getInboxPolicy(ctx, accountID, inboxID) + if err != nil { + return 0, fmt.Errorf("get assignment policy: %w", err) + } + if policy != nil && !policy.Enabled { + return 0, nil + } + advancedAssignment, err := s.advancedAssignmentEnabled(ctx, accountID) + if err != nil { + return 0, fmt.Errorf("get advanced assignment feature: %w", err) + } + + var conversation model.Conversation + if err := s.db.WithContext(ctx). + Select("team_id"). + Where("id = ? AND inbox_id = ? AND account_id = ?", conversationID, inboxID, accountID). + First(&conversation).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + return 0, nil + } + return 0, fmt.Errorf("get conversation team: %w", err) + } + // Get eligible agents - agents, err := s.getEligibleAgents(ctx, inboxID, accountID) + agents, err := s.getEligibleAgents(ctx, inboxID, accountID, conversation.TeamID, advancedAssignment) if err != nil { return 0, fmt.Errorf("get eligible agents: %w", err) } @@ -160,17 +195,15 @@ func (s *AssignmentService) AssignConversation(ctx context.Context, conversation } // Get policy and limits - accountPolicy := s.getAccountPolicy(ctx, accountID) - inboxPolicy := s.getInboxPolicy(ctx, inboxID) - policy := EffectivePolicy(accountPolicy, inboxPolicy) - limit := EffectiveLimit(accountPolicy, inboxPolicy) - window := EffectiveWindow(accountPolicy, inboxPolicy) + selector := EffectivePolicy(policy, advancedAssignment) + limit := EffectiveLimit(policy) + window := EffectiveWindow(policy) // Sync round-robin queue s.roundRobin.SyncQueue(ctx, inboxID, agents) // Select agent - agentID, err := s.selectAgent(ctx, inboxID, agents, policy, limit, window) + agentID, err := s.selectAgent(ctx, inboxID, agents, selector, limit, window) if err != nil { return 0, fmt.Errorf("select agent: %w", err) } @@ -203,8 +236,6 @@ func (s *AssignmentService) selectAgent(ctx context.Context, inboxID uint, agent return s.selectRoundRobin(ctx, inboxID, agents, limit, window) case PolicyLowestLoad: return s.selectLowestLoad(ctx, inboxID, agents, limit, window) - case PolicyLongestWaiting: - return s.selectLongestWaiting(ctx, inboxID, agents, limit, window) default: return s.selectRoundRobin(ctx, inboxID, agents, limit, window) } @@ -230,15 +261,6 @@ func (s *AssignmentService) selectRoundRobin(ctx context.Context, inboxID uint, return 0, nil } -// selectLongestWaiting selects the agent with the longest idle time. -// This is a simplified implementation — a full version would track -// idle time in Redis. For now, we fall back to round-robin. -func (s *AssignmentService) selectLongestWaiting(ctx context.Context, inboxID uint, agents []uint, limit int, window int) (uint, error) { - // TODO: Implement longest-waiting tracking in Redis - // For now, delegate to round-robin with rate limit checking - return s.selectRoundRobin(ctx, inboxID, agents, limit, window) -} - // selectLowestLoad selects the agent with the fewest open conversations, // respecting rate limits. func (s *AssignmentService) selectLowestLoad(ctx context.Context, inboxID uint, agents []uint, limit int, window int) (uint, error) { @@ -257,29 +279,46 @@ func (s *AssignmentService) selectLowestLoad(ctx context.Context, inboxID uint, return s.selectRoundRobin(ctx, inboxID, agents, limit, window) } -// findUnassignedConversations returns all open conversations in the inbox -// that have no assignee. -func (s *AssignmentService) findUnassignedConversations(ctx context.Context, inboxID uint, accountID uint) ([]model.Conversation, error) { +// findUnassignedConversations returns one bounded batch of open conversations +// in the inbox that have no assignee. +func (s *AssignmentService) findUnassignedConversations(ctx context.Context, inboxID uint, accountID uint, policy *model.AssignmentPolicy) ([]model.Conversation, error) { var conversations []model.Conversation - err := s.db.WithContext(ctx). + query := s.db.WithContext(ctx). Where("inbox_id = ? AND account_id = ? AND status = ? AND assignee_id IS NULL", - inboxID, accountID, model.ConversationStatusOpen). - Find(&conversations).Error + inboxID, accountID, model.ConversationStatusOpen) + if policy != nil && policy.ExcludeOlderThanHours != nil && *policy.ExcludeOlderThanHours > 0 { + cutoff := time.Now().Add(-time.Duration(*policy.ExcludeOlderThanHours) * time.Hour).Unix() + // GoChat legacy rows may not have an activity timestamp; unknown is not stale. + query = query.Where("last_activity_at IS NULL OR last_activity_at >= ?", cutoff) + } + if policy != nil && policy.ConversationPriority == 1 { + query = query.Order("last_activity_at ASC").Order("created_at ASC") + } else { + query = query.Order("created_at ASC") + } + err := query.Limit(assignmentBatchLimit).Find(&conversations).Error return conversations, err } // getEligibleAgents returns agents that are: // - Members of the inbox (via inbox_members table) // - Online/available (via user available field) -func (s *AssignmentService) getEligibleAgents(ctx context.Context, inboxID uint, accountID uint) ([]uint, error) { +func (s *AssignmentService) getEligibleAgents(ctx context.Context, inboxID uint, accountID uint, teamID *uint, advancedAssignment bool) ([]uint, error) { var agentIDs []uint - err := s.db.WithContext(ctx). + query := s.db.WithContext(ctx). Table("inbox_members"). - Select("inbox_members.user_id"). + Distinct("inbox_members.user_id"). Joins("JOIN users ON users.id = inbox_members.user_id"). Joins("JOIN account_users ON account_users.user_id = inbox_members.user_id AND account_users.account_id = ?", accountID). - Where("inbox_members.inbox_id = ? AND users.available = ? AND users.active = ? AND account_users.role IN ?", - inboxID, true, true, []string{"agent", "administrator"}). + Where("inbox_members.inbox_id = ? AND inbox_members.deleted_at IS NULL AND users.available = ? AND users.active = ? AND account_users.role IN ?", + inboxID, true, true, []string{"agent", "administrator"}) + if teamID != nil { + query = query. + Joins("JOIN team_members ON team_members.user_id = inbox_members.user_id AND team_members.team_id = ? AND team_members.deleted_at IS NULL", *teamID). + Joins("JOIN teams ON teams.id = team_members.team_id AND teams.account_id = ? AND teams.allow_auto_assignment = ? AND teams.deleted_at IS NULL", accountID, true) + } + err := query. + Order("inbox_members.user_id ASC"). Pluck("inbox_members.user_id", &agentIDs).Error if err != nil { return nil, err @@ -287,6 +326,10 @@ func (s *AssignmentService) getEligibleAgents(ctx context.Context, inboxID uint, availableIDs := make([]uint, 0, len(agentIDs)) for _, agentID := range agentIDs { + if !advancedAssignment { + availableIDs = append(availableIDs, agentID) + continue + } hasCapacity, err := s.agentHasInboxCapacity(ctx, accountID, inboxID, agentID, 0) if err != nil { return nil, err @@ -343,28 +386,43 @@ func (s *AssignmentService) getInbox(ctx context.Context, inboxID uint) (*model. return &inbox, nil } -// getAccountPolicy returns the assignment policy for an account. -// Returns nil if no policy is configured. -func (s *AssignmentService) getAccountPolicy(ctx context.Context, accountID uint) *AssignmentPolicy { - var policy AssignmentPolicy - if err := s.db.WithContext(ctx). - Where("account_id = ? AND active = ?", accountID, true). - First(&policy).Error; err != nil { - return nil // no policy configured +// getInboxPolicy resolves the current CRUD model through its inbox join. +func (s *AssignmentService) getInboxPolicy(ctx context.Context, accountID, inboxID uint) (*model.AssignmentPolicy, error) { + policy, err := repository.NewInboxAssignmentPolicyRepo(s.db).FindPolicyByInbox(ctx, accountID, inboxID) + if errors.Is(err, gorm.ErrRecordNotFound) { + return nil, nil } - return &policy + return policy, err } -// getInboxPolicy returns the assignment policy override for an inbox. -// Returns nil if no inbox-specific policy is configured. -func (s *AssignmentService) getInboxPolicy(ctx context.Context, inboxID uint) *InboxAssignmentPolicy { - var policy InboxAssignmentPolicy - if err := s.db.WithContext(ctx). - Where("inbox_id = ? AND active = ?", inboxID, true). - First(&policy).Error; err != nil { - return nil // no inbox-specific policy +func (s *AssignmentService) advancedAssignmentEnabled(ctx context.Context, accountID uint) (bool, error) { + var account model.Account + if err := s.db.WithContext(ctx).Select("feature_flags").First(&account, accountID).Error; err != nil { + return false, err } - return &policy + return featureEnabled(account.FeatureFlags, "advanced_assignment"), nil +} + +func featureEnabled(raw, flag string) bool { + values := map[string]bool{} + if json.Unmarshal([]byte(raw), &values) == nil { + return values[flag] + } + var list []string + if json.Unmarshal([]byte(raw), &list) == nil { + for _, value := range list { + if value == flag { + return true + } + } + return false + } + for _, value := range strings.Split(raw, ",") { + if strings.TrimSpace(value) == flag { + return true + } + } + return false } // assignConversation sets the assignee_id on a conversation. diff --git a/backend/internal/worker/worker.go b/backend/internal/worker/worker.go index 778cf1d7..4cf915af 100644 --- a/backend/internal/worker/worker.go +++ b/backend/internal/worker/worker.go @@ -35,6 +35,9 @@ func Permanent(err error) error { // JobHandler performs one durable background job. type JobHandler func(context.Context, *model.BackgroundJob) error +// JobFailureHandler runs after a retrying or dead state is persisted. +type JobFailureHandler func(context.Context, *model.BackgroundJob, time.Duration) error + type BackoffFunc func(attempt int) time.Duration // WorkerPool manages durable background job processors. @@ -44,6 +47,7 @@ type WorkerPool struct { db *gorm.DB rdb redis.UniversalClient handlers map[string]JobHandler + failureHandlers map[string]JobFailureHandler queues []string workerID string workerCount int @@ -74,6 +78,7 @@ type Option func(*WorkerPool) func NewWorkerPool(db ...*gorm.DB) *WorkerPool { wp := &WorkerPool{ handlers: make(map[string]JobHandler), + failureHandlers: make(map[string]JobFailureHandler), workerID: fmt.Sprintf("worker-%d", time.Now().UnixNano()), workerCount: 1, pollInterval: 500 * time.Millisecond, @@ -206,6 +211,12 @@ func (wp *WorkerPool) Register(jobType string, handler JobHandler) { wp.handlers[jobType] = handler } +func (wp *WorkerPool) RegisterFailureHandler(jobType string, handler JobFailureHandler) { + wp.mu.Lock() + defer wp.mu.Unlock() + wp.failureHandlers[jobType] = handler +} + type EnqueueOption func(*model.BackgroundJob) func WithQueue(queue string) EnqueueOption { @@ -639,9 +650,7 @@ func (wp *WorkerPool) processRedisMessage(lifecycleCtx, jobCtx context.Context, wp.db.WithContext(jobCtx).First(&job, jobID) if err := wp.perform(jobCtx, &job); err != nil { - if failErr := wp.fail(jobCtx, &job, err); failErr != nil { - applogger.L().Errorf("record job %d failure: %v", job.ID, failErr) - } + applogger.L().Errorf("job %d failed: %v", job.ID, err) } wp.ackRedis(jobCtx, stream, msg.ID) @@ -825,16 +834,29 @@ func (wp *WorkerPool) fail(ctx context.Context, job *model.BackgroundJob, err er "last_error": err.Error(), } var permanent *permanentError + retryAfter := time.Duration(0) if job.Attempts >= job.MaxAttempts || errors.As(err, &permanent) { updates["status"] = model.BackgroundJobStatusDead updates["failed_at"] = &now } else { + retryAfter = wp.backoff(job.Attempts) updates["status"] = model.BackgroundJobStatusRetrying - updates["scheduled_at"] = now.Add(wp.backoff(job.Attempts)) + updates["scheduled_at"] = now.Add(retryAfter) } if updateErr := wp.db.WithContext(ctx).Model(&model.BackgroundJob{}).Where("id = ?", job.ID).Updates(updates).Error; updateErr != nil { return updateErr } + job.Status = updates["status"].(string) + if job.Status == model.BackgroundJobStatusDead { + job.FailedAt = &now + } else { + job.ScheduledAt = updates["scheduled_at"].(time.Time) + } + if handler := wp.failureHandlerFor(job.JobType); handler != nil { + if handlerErr := handler(ctx, job, retryAfter); handlerErr != nil { + return errors.Join(err, handlerErr) + } + } return err } @@ -844,6 +866,12 @@ func (wp *WorkerPool) handlerFor(jobType string) JobHandler { return wp.handlers[jobType] } +func (wp *WorkerPool) failureHandlerFor(jobType string) JobFailureHandler { + wp.mu.RLock() + defer wp.mu.RUnlock() + return wp.failureHandlers[jobType] +} + func marshalPayload(payload any) (json.RawMessage, error) { if payload == nil { return json.RawMessage(`{}`), nil