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 <rogee@ipao.vip>

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")
+}

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)

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)
+}

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())
+}

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}

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}

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))
-}

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)

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) {

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))
 }
 

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),
 	}
-}
+}

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)
-}

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)
+}

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.

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