Fixing problem with promisc flag during event listen

api/main.py

@@ -46,7 +46,7 @@
 from pydantic import BaseModel
 from jose import jwt
 from jose.exceptions import JWTError
 from kernelci.api.models import (
     Node,
     Hierarchy,
     PublishEvent,
@@ -1412,7 +1412,13 @@
     """Listen messages from a subscribed Pub/Sub channel"""
     metrics.add('http_requests_total', 1)
     try:
-        return await pubsub.listen(sub_id, user.username)
+        result = await pubsub.listen(sub_id, user.username)
+        # Debug: print result type and check for bytes
+        print(f"[DEBUG listen] result type: {type(result)}")
+        if result:
+            for k, v in result.items():
+                print(f"[DEBUG listen] {k}: {type(v)} = {repr(v)[:100]}")
+        return result
     except KeyError as error:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -1423,6 +1429,10 @@
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=f"Error while listening to sub id {sub_id}: {str(error)}"
         ) from error
+    except Exception as error:
+        print(f"[DEBUG listen] Unexpected: {type(error).__name__}: {error}")
+        traceback.print_exc()
+        raise
 
 
 @app.post('/publish/{channel}')

api/pubsub_mongo.py

@@ -295,6 +295,21 @@ async def _update_subscriber_state(self, subscriber_id: str,
             {'$set': update}
         )
 
+    @staticmethod
+    def _decode_redis_message(msg: Dict) -> Dict:
+        """Decode Redis message bytes to strings for JSON serialization"""
+        return {
+            'type': msg.get('type'),
+            'pattern': (msg.get('pattern').decode('utf-8')
+                        if msg.get('pattern') else None),
+            'channel': (msg['channel'].decode('utf-8')
+                        if isinstance(msg['channel'], bytes)
+                        else msg['channel']),
+            'data': (msg['data'].decode('utf-8')
+                     if isinstance(msg['data'], bytes)
+                     else msg['data']),
+        }
+
     def _eventhistory_to_cloudevent(self, event: Dict) -> str:
         """Convert eventhistory document to CloudEvent JSON string
 
@@ -547,10 +562,10 @@ async def listen(self, sub_id: int,
 
             # Filter by owner if not promiscuous
             if sub.promiscuous:
-                return msg
+                return self._decode_redis_message(msg)
             if 'owner' in msg_data and msg_data['owner'] != sub.user:
                 continue
-            return msg
+            return self._decode_redis_message(msg)
 
     async def publish(self, channel: str, message: str):
         """Publish a message on a channel (Redis only, no durability)"""