consumer group creation

Author: yaananth

backend/redis/queue.go

@@ -43,20 +43,17 @@ func newTaskQueue[T any](rdb redis.UniversalClient, tasktype string) (*taskQueue
 		workerName: uuid.NewString(),
 	}
 
-	// Create the consumer group
-	_, err := rdb.XGroupCreateMkStream(context.Background(), tq.streamKey, tq.groupName, "0").Result()
-	if err != nil {
-		// Ugly, check since there is no UPSERT for consumer groups. Might replace with a script
-		// using XINFO & XGROUP CREATE atomically
-		if err.Error() != "BUSYGROUP Consumer Group name already exists" {
-			return nil, fmt.Errorf("creating task queue: %w", err)
-		}
-	}
-
 	// Pre-load script
 	cmds := map[string]*redis.StringCmd{
-		"enqueueCmd":  enqueueCmd.Load(context.Background(), rdb),
-		"completeCmd": completeCmd.Load(context.Background(), rdb),
+		"createGroupCmd": createGroupCmd.Load(context.Background(), rdb),
+		"enqueueCmd":     enqueueCmd.Load(context.Background(), rdb),
+		"completeCmd":    completeCmd.Load(context.Background(), rdb),
+	}
+
+	// Create the consumer group
+	_, err := createGroupCmd.Run(context.Background(), rdb, []string{tq.streamKey, tq.groupName}).Result()
+	if err != nil {
+		return nil, fmt.Errorf("creating task queue: %w", err)
 	}
 
 	for name, cmd := range cmds {
@@ -91,6 +88,23 @@ var enqueueCmd = redis.NewScript(
 	return true
 `)
 
+var createGroupCmd = redis.NewScript(`
+	local streamKey = KEYS[1]
+	local groupName = KEYS[2]
+	local status, groups = pcall(redis.call, 'XINFO', 'GROUPS', streamKey)
+
+	if status then
+		for i = 1, #groups do
+			if groups[i].name == groupName then
+				return false
+			end
+		end
+	end
+
+	redis.call('XGROUP', 'CREATE', streamKey, groupName, '0', 'MKSTREAM')
+	return true
+`)
+
 func (q *taskQueue[T]) Enqueue(ctx context.Context, p redis.Pipeliner, id string, data *T) error {
 	ds, err := json.Marshal(data)
 	if err != nil {