debug redis queue

Author: tangg555

src/memos/mem_scheduler/task_schedule_modules/redis_queue.py

@@ -150,7 +150,7 @@ def put(
             logger.error(f"Failed to add message to Redis queue: {e}")
             raise
 
-    def ack_message(self, user_id, mem_cube_id, redis_message_id):
+    def ack_message(self, user_id, mem_cube_id, redis_message_id) -> None:
         stream_key = self.get_stream_key(user_id=user_id, mem_cube_id=mem_cube_id)
 
         self.redis.xack(stream_key, self.consumer_group, redis_message_id)
@@ -296,7 +296,13 @@ def get_stream_keys(self) -> list[str]:
             return []
 
         try:
-            return self._redis_conn.scan_iter(f"{self.stream_key_prefix}:*")
+            # Use match parameter and decode byte strings to regular strings
+            stream_keys = [
+                key.decode("utf-8") if isinstance(key, bytes) else key
+                for key in self._redis_conn.scan_iter(match=f"{self.stream_key_prefix}:*")
+            ]
+            logger.debug(f"get stream_keys from redis: {stream_keys}")
+            return stream_keys
         except Exception as e:
             logger.error(f"Failed to list Redis stream keys: {e}")
             return []

src/memos/mem_scheduler/task_schedule_modules/task_queue.py

@@ -35,6 +35,22 @@ def __init__(
 
         self.disabled_handlers = disabled_handlers
 
+    def ack_message(
+        self,
+        user_id,
+        mem_cube_id,
+        redis_message_id,
+    ) -> None:
+        if not isinstance(self.memos_message_queue, SchedulerRedisQueue):
+            logger.warning("ack_message is only supported for Redis queues")
+            return
+
+        self.memos_message_queue.ack_message(
+            user_id=user_id,
+            mem_cube_id=mem_cube_id,
+            redis_message_id=redis_message_id,
+        )
+
     def debug_mode_on(self):
         self.memos_message_queue.stream_key_prefix = (
             f"debug_mode:{self.memos_message_queue.stream_key_prefix}"