package autoassignment // RoundRobinSelector implements Redis-backed round-robin agent selection. // // Reference: Chatwoot InboxRoundRobinService // - Uses Redis to maintain a per-inbox queue of agent IDs // - Push/pop/rotate agents through the queue // - Reset queue if members changed (SyncQueue method) // - RoundRobinSelector wraps InboxRoundRobinService for single agent selection // // In gochat, we use Redis lists to implement the round-robin queue: // - Key format: gochat:round_robin:{inboxID} // - On Next(): pop from left, push to right (rotate) // - On SyncQueue(): rebuild the list from current inbox members import ( "context" "fmt" "strconv" "github.com/redis/go-redis/v9" applogger "github.com/gochat/gochat/pkg/logger" ) const ( // roundRobinKeyPrefix is the Redis key prefix for round-robin queues. roundRobinKeyPrefix = "gochat:round_robin:" ) // RoundRobinSelector implements Redis-backed round-robin agent selection // for a given inbox. type RoundRobinSelector struct { redis *redis.Client } // NewRoundRobinSelector creates a new RoundRobinSelector. func NewRoundRobinSelector(rdb *redis.Client) *RoundRobinSelector { return &RoundRobinSelector{redis: rdb} } // key returns the Redis key for the given inbox's round-robin queue. func (rr *RoundRobinSelector) key(inboxID uint) string { return fmt.Sprintf("%s%d", roundRobinKeyPrefix, inboxID) } // Next returns the next agent ID from the round-robin queue. // It pops from the left of the list and pushes to the right (rotate), // so each agent gets a turn before any agent gets a second turn. // // Returns 0 if the queue is empty. func (rr *RoundRobinSelector) Next(ctx context.Context, inboxID uint) (uint, error) { key := rr.key(inboxID) // Pop from left, push to right (rotate) result, err := rr.redis.LPop(ctx, key).Result() if err == redis.Nil { // Queue is empty return 0, nil } if err != nil { return 0, fmt.Errorf("redis LPop: %w", err) } // Push the same agent to the right (rotate back) if err := rr.redis.RPush(ctx, key, result).Err(); err != nil { return 0, fmt.Errorf("redis RPush: %w", err) } agentID, err := strconv.ParseUint(result, 10, 32) if err != nil { return 0, fmt.Errorf("parse agent ID: %w", err) } return uint(agentID), nil } // AvailableAgent selects the next agent from the round-robin queue that is also // in the allowedAgentIDs set. // 1:1 Chatwoot: InboxRoundRobinService#available_agent(allowed_agent_ids:) // Logic: queue.intersection(allowed_agent_ids).pop → pop_push_to_queue func (rr *RoundRobinSelector) AvailableAgent(ctx context.Context, inboxID uint, allowedAgentIDs []uint) (uint, error) { if len(allowedAgentIDs) == 0 { return 0, nil } key := rr.key(inboxID) // Get current queue queue, err := rr.redis.LRange(ctx, key, 0, -1).Result() if err != nil { return 0, fmt.Errorf("redis LRange: %w", err) } if len(queue) == 0 { return 0, nil } // Build allowed set for intersection allowedSet := make(map[string]bool, len(allowedAgentIDs)) for _, id := range allowedAgentIDs { allowedSet[strconv.FormatUint(uint64(id), 10)] = true } // Find last agent in queue that is in allowed set (Chatwoot: intersection.pop) var selected string for i := len(queue) - 1; i >= 0; i-- { if allowedSet[queue[i]] { selected = queue[i] break } } if selected == "" { return 0, nil // no eligible agent } // pop_push_to_queue: remove from current position, push to end rr.redis.LRem(ctx, key, 1, selected) rr.redis.RPush(ctx, key, selected) agentID, err := strconv.ParseUint(selected, 10, 32) if err != nil { return 0, fmt.Errorf("parse agent ID: %w", err) } return uint(agentID), nil } // SyncQueue rebuilds the round-robin queue for the given inbox. // It compares the current queue with the provided agent list, // and resets the queue if the members have changed. // // Reference: Chatwoot InboxRoundRobinService.reset_round_robin_for_inbox // - Called when inbox members change (agent added/removed) // - Ensures only current members are in the rotation func (rr *RoundRobinSelector) SyncQueue(ctx context.Context, inboxID uint, agents []uint) { key := rr.key(inboxID) // Check current queue current, err := rr.redis.LRange(ctx, key, 0, -1).Result() if err != nil && err != redis.Nil { applogger.L().Warnf("redis LRange error for inbox %d: %v", inboxID, err) } // Build expected queue expected := make([]string, len(agents)) for i, agentID := range agents { expected[i] = strconv.FormatUint(uint64(agentID), 10) } // Check if queue needs reset needsReset := len(current) != len(expected) if !needsReset { // Check if all expected agents are in current queue currentSet := make(map[string]bool) for _, id := range current { currentSet[id] = true } for _, id := range expected { if !currentSet[id] { needsReset = true break } } } if needsReset { // Delete old queue and create new one rr.redis.Del(ctx, key) if len(expected) > 0 { values := make([]interface{}, len(expected)) for i, v := range expected { values[i] = v } if err := rr.redis.RPush(ctx, key, values...).Err(); err != nil { applogger.L().Warnf("redis RPush error for inbox %d: %v", inboxID, err) } } applogger.L().Infof("round-robin queue synced for inbox %d (%d agents)", inboxID, len(agents)) } } // ClearQueue removes the round-robin queue for the given inbox. func (rr *RoundRobinSelector) ClearQueue(ctx context.Context, inboxID uint) error { return rr.redis.Del(ctx, rr.key(inboxID)).Err() } // QueueLength returns the number of agents in the round-robin queue. func (rr *RoundRobinSelector) QueueLength(ctx context.Context, inboxID uint) (int64, error) { return rr.redis.LLen(ctx, rr.key(inboxID)).Result() }