Repository: redis/redis-py
File: redis/event.py (Python)

Code before:
class ExportStreamingLagMetric(EventListenerInterface):
    """
    Listener that exports streaming lag metric per stream message.
    """
    def listen(self, event: OnStreamMessageReceivedEvent):
        now = datetime.now().timestamp()

        # RESP3
        if type(event.response) == dict:
            for stream_name, stream_messages in event.response.items():
                for messages in stream_messages:
                    for message in messages:
                        message_id, message = message
                        message_id = str_if_bytes(message_id)
                        timestamp, _ = message_id.split("-")
                        lag_seconds = now - int(timestamp) / 1000

                        record_streaming_lag(
                            lag_seconds=lag_seconds,
                            stream_name=str_if_bytes(stream_name),
                            consumer_group=event.consumer_group,
                            consumer_name=event.consumer_name,
                        )
        else:
            # RESP 2
            for stream_entry in event.response:
                stream_name = str_if_bytes(stream_entry[0])

                for message in stream_entry[1]:
                    message_id, message = message
                    message_id = str_if_bytes(message_id)
                    timestamp, _ = message_id.split("-")
                    lag_seconds = now - int(timestamp) / 1000

                    record_streaming_lag(
                        lag_seconds=lag_seconds,
                        stream_name=stream_name,
                        consumer_group=event.consumer_group,
                        consumer_name=event.consumer_name,
                    )


Review comment (bug):
The nested iteration structure is incorrect for RESP3 responses. Based on the xreadgroup response format, for RESP3 the structure is a dict where each value is a single list of messages, not a list of lists. The current code iterates with `for messages in stream_messages: for message in messages:` which adds an extra level of nesting. It should be `for message in stream_messages:` directly. This will cause an error when trying to unpack message tuples.
```suggestion
                for message in stream_messages:
                    message_id, message = message
                    message_id = str_if_bytes(message_id)
                    timestamp, _ = message_id.split("-")
                    lag_seconds = now - int(timestamp) / 1000

                    record_streaming_lag(
                        lag_seconds=lag_seconds,
                        stream_name=str_if_bytes(stream_name),
                        consumer_group=event.consumer_group,
                        consumer_name=event.consumer_name,
                    )
```

Code after:
class ExportStreamingLagMetric(EventListenerInterface):
    """
    Listener that exports streaming lag metric per stream message.
    """
    def listen(self, event: OnStreamMessageReceivedEvent):
        now = datetime.now().timestamp()

        if not event.response:
            return

        # RESP3
        if isinstance(event.response, dict):
            for stream_name, stream_messages in event.response.items():
                for messages in stream_messages:
                    for message in messages:
                        message_id, message = message
                        message_id = str_if_bytes(message_id)
                        timestamp, _ = message_id.split("-")
                        lag_seconds = now - int(timestamp) / 1000

                        record_streaming_lag(
                            lag_seconds=lag_seconds,
                            stream_name=str_if_bytes(stream_name),
                            consumer_group=event.consumer_group,
                            consumer_name=event.consumer_name,
                        )
        else:
            # RESP 2
            for stream_entry in event.response:
                stream_name = str_if_bytes(stream_entry[0])

                for message in stream_entry[1]:
                    message_id, message = message
                    message_id = str_if_bytes(message_id)
                    timestamp, _ = message_id.split("-")
                    lag_seconds = now - int(timestamp) / 1000

                    record_streaming_lag(
                        lag_seconds=lag_seconds,
                        stream_name=stream_name,
                        consumer_group=event.consumer_group,
                        consumer_name=event.consumer_name,
                    )