fixed revie changes

Author: coolwednesday

docs/advanced-guide/using-publisher-subscriber/page.md

@@ -434,7 +434,7 @@ The following configs apply specifically to Redis Pub/Sub behavior. For base Red
 ---
 
 - `REDIS_STREAMS_BLOCK_TIMEOUT`
-- Block duration for stream reads using Redis `XREADGROUP`. This controls how long the consumer blocks waiting for new messages before the Redis call times out and retries. **Benefits of configuring this:** (1) **Resource efficiency** - Without blocking, consumers would constantly poll Redis, wasting CPU cycles and network bandwidth. Blocking allows Redis to push messages immediately when available. (2) **Latency vs CPU trade-off** - Lower values (e.g., `1s-2s`) provide faster message detection but increase CPU from frequent timeouts. Higher values (e.g., `10s-30s`) reduce CPU usage and network round-trips but may delay processing. (3) **Cost optimization** - In cloud/serverless environments, reducing CPU usage directly reduces costs. (4) **Battery efficiency** - Important for mobile/edge deployments where power consumption matters. Choose based on your latency requirements: real-time systems may use `1s-2s`, while batch processing can use `10s-30s`.
+- Block duration for stream reads using Redis `XREADGROUP`. Controls how long the consumer blocks waiting for new messages before timing out. Lower values (1s-2s) provide faster detection but increase CPU usage. Higher values (10s-30s) reduce CPU usage, ideal for batch processing.
 - `5s`
 - `2s` (low latency) or `30s` (low CPU)
 
@@ -498,7 +498,9 @@ docker run -d \
 > **Important**: If you are using **GoFr migrations** with **Redis** and also using **Redis Pub/Sub in Streams mode**, do not use the same Redis logical DB for both.
 > GoFr stores Redis migration state in a Redis **HASH** named `gofr_migrations`, while Redis Streams mode uses a Redis **STREAM** key for topics (including the PubSub migration topic `gofr_migrations`).
 > If both clients share the same DB, migrations can fail with `WRONGTYPE` errors.
-> Set `REDIS_PUBSUB_DB` to a different DB index (for example, `REDIS_DB=0` and `REDIS_PUBSUB_DB=1`).
+> By default, `REDIS_DB` is `0` and `REDIS_PUBSUB_DB` is `15` (highest default Redis database), so they are already separated. If you change `REDIS_DB` from the default, ensure `REDIS_PUBSUB_DB` is set to a different DB index (for example, `REDIS_DB=0` and `REDIS_PUBSUB_DB=1`).
+
+> **Note on `REDIS_STREAMS_BLOCK_TIMEOUT`**: This configuration controls how long the consumer blocks waiting for new messages using Redis `XREADGROUP` before the call times out and retries. **Benefits of configuring this:** (1) **Resource efficiency** - Without blocking, consumers would constantly poll Redis, wasting CPU cycles and network bandwidth. Blocking allows Redis to push messages immediately when available. (2) **Latency vs CPU trade-off** - Lower values (e.g., `1s-2s`) provide faster message detection but increase CPU from frequent timeouts. Higher values (e.g., `10s-30s`) reduce CPU usage and network round-trips but may delay processing. (3) **Cost optimization** - In cloud/serverless environments, reducing CPU usage directly reduces costs. (4) **Battery efficiency** - Important for mobile/edge deployments where power consumption matters. Choose based on your latency requirements: real-time systems may use `1s-2s`, while batch processing can use `10s-30s`.
 
 ### Azure Event Hubs
 GoFr supports Event Hubs starting gofr version v1.22.0.

docs/references/configs/page.md

@@ -422,7 +422,7 @@ This document lists all the configuration options supported by the GoFr framewor
 ---
 
 - REDIS_STREAMS_BLOCK_TIMEOUT
-- Blocking duration for reading new messages. Enables efficient message consumption by allowing Redis to push messages immediately when available, avoiding constant polling. Lower values (1s-2s) provide faster detection but increase CPU usage. Higher values (10s-30s) reduce CPU usage, ideal for batch processing.
+- Blocking duration for reading new messages using Redis `XREADGROUP`. Lower values (1s-2s) provide faster detection but increase CPU usage. Higher values (10s-30s) reduce CPU usage, ideal for batch processing.
 -  5s
 
 ---
@@ -434,7 +434,9 @@ This document lists all the configuration options supported by the GoFr framewor
 {% /table %}
 
 > If `REDIS_PUBSUB_MODE` is set to anything other than `streams` or `pubsub`, it falls back to `streams`.
-> If you are using GoFr migrations with Redis and Redis PubSub Streams mode together, set `REDIS_PUBSUB_DB` to a different DB than `REDIS_DB` to avoid `WRONGTYPE` errors on the `gofr_migrations` key.
+> If you are using GoFr migrations with Redis and Redis PubSub Streams mode together, set `REDIS_PUBSUB_DB` to a different DB than `REDIS_DB` to avoid `WRONGTYPE` errors on the `gofr_migrations` key. By default, `REDIS_DB` is `0` and `REDIS_PUBSUB_DB` is `15` (highest default Redis database), so they are already separated. If you change `REDIS_DB` from the default, ensure `REDIS_PUBSUB_DB` is set to a different DB index.
+
+> **Note on `REDIS_STREAMS_BLOCK_TIMEOUT`**: This configuration controls how long the consumer blocks waiting for new messages using Redis `XREADGROUP` before the call times out and retries. **Benefits of configuring this:** (1) **Resource efficiency** - Without blocking, consumers would constantly poll Redis, wasting CPU cycles and network bandwidth. Blocking allows Redis to push messages immediately when available. (2) **Latency vs CPU trade-off** - Lower values (e.g., `1s-2s`) provide faster message detection but increase CPU from frequent timeouts. Higher values (e.g., `10s-30s`) reduce CPU usage and network round-trips but may delay processing. (3) **Cost optimization** - In cloud/serverless environments, reducing CPU usage directly reduces costs. (4) **Battery efficiency** - Important for mobile/edge deployments where power consumption matters. Choose based on your latency requirements: real-time systems may use `1s-2s`, while batch processing can use `10s-30s`.
 
 ### Pub/Sub
 

pkg/gofr/container/container.go

@@ -38,6 +38,11 @@ import (
 	"gofr.dev/pkg/gofr/websocket"
 )
 
+const (
+	redisPubSubModeStreams = "streams"
+	redisPubSubModePubSub  = "pubsub"
+)
+
 // Container is a collection of all common application level concerns. Things like Logger, Connection Pool for Redis
 // etc. which is shared across is placed here.
 type Container struct {
@@ -322,6 +327,13 @@ func (c *Container) createKafkaPubSub(conf config.Config) {
 	}
 
 	partition, _ := strconv.Atoi(conf.GetOrDefault("PARTITION_SIZE", "0"))
+	// PUBSUB_OFFSET determines the starting position for message consumption in Kafka.
+	// This allows control over whether to read historical messages or only new ones:
+	// - Default value -1: Start from the latest offset (only consume new messages after consumer starts)
+	// - Value 0: Start from the earliest offset (read all historical messages from the beginning)
+	// - Positive value: Start from a specific offset position (useful for resuming from a known point)
+	// This is particularly important for scenarios like message replay, recovery from failures,
+	// or when you only want to process messages that arrive after the consumer is initialized.
 	offSet, _ := strconv.Atoi(conf.GetOrDefault("PUBSUB_OFFSET", "-1"))
 	batchSize, _ := strconv.Atoi(conf.GetOrDefault("KAFKA_BATCH_SIZE", strconv.Itoa(kafka.DefaultBatchSize)))
 	batchBytes, _ := strconv.Atoi(conf.GetOrDefault("KAFKA_BATCH_BYTES", strconv.Itoa(kafka.DefaultBatchBytes)))
@@ -366,11 +378,6 @@ func (c *Container) createRedisPubSub(conf config.Config) {
 	c.PubSub = redis.NewPubSub(conf, c.Logger, c.metricsManager)
 }
 
-const (
-	redisPubSubModeStreams = "streams"
-	redisPubSubModePubSub  = "pubsub"
-)
-
 func (c *Container) warnIfRedisPubSubSharesRedisDB(conf config.Config) {
 	// Warn (do not fail): if Redis PubSub (streams mode) shares the same Redis logical DB as the primary Redis datasource,
 	// GoFr migrations can later fail due to `gofr_migrations` key-type collision (HASH vs STREAM).

pkg/gofr/datasource/redis/health.go

@@ -57,7 +57,7 @@ func (ps *PubSub) Health() datasource.Health {
 	}
 
 	addr := fmt.Sprintf("%s:%d", ps.config.HostName, ps.config.Port)
-	res.Details["addr"] = addr
+	res.Details["host"] = addr
 
 	mode := ps.config.PubSubMode
 	if mode == "" {

pkg/gofr/datasource/redis/health_test.go

@@ -56,7 +56,7 @@ func TestPubSub_HealthDetails(t *testing.T) {
 	h := client.PubSub.Health()
 	assert.Equal(t, datasource.StatusUp, h.Status)
 	assert.Equal(t, "REDIS", h.Details["backend"])
-	assert.Equal(t, "localhost:6380", h.Details["addr"])
+	assert.Equal(t, "localhost:6380", h.Details["host"])
 	assert.Equal(t, "pubsub", h.Details["mode"])
 
 	assert.NoError(t, mock.ExpectationsWereMet())

pkg/gofr/datasource/redis/pubsub.go

@@ -44,15 +44,25 @@ func (ps *PubSub) Publish(ctx context.Context, topic string, message []byte) err
 
 // publishToChannel publishes a message to a Redis PubSub channel.
 func (ps *PubSub) publishToChannel(ctx context.Context, topic string, message []byte, span trace.Span) error {
+	start := time.Now()
 	err := ps.client.Publish(ctx, topic, message).Err()
+	end := time.Since(start)
+
 	if err != nil {
 		ps.logger.Errorf("failed to publish message to Redis channel '%s': %v", topic, err)
 		return err
 	}
 
-	traceID := span.SpanContext().TraceID().String()
 	addr := fmt.Sprintf("%s:%d", ps.config.HostName, ps.config.Port)
-	ps.logger.Debugf("PUB %s %s %s", topic, traceID, addr)
+	ps.logger.Debug(&pubsub.Log{
+		Mode:          "PUB",
+		CorrelationID: span.SpanContext().TraceID().String(),
+		MessageValue:  string(message),
+		Topic:         topic,
+		Host:          addr,
+		PubSubBackend: "REDIS",
+		Time:          end.Microseconds(),
+	})
 	ps.metrics.IncrementCounter(ctx, "app_pubsub_publish_success_count", "topic", topic)
 
 	return nil
@@ -70,15 +80,25 @@ func (ps *PubSub) publishToStream(ctx context.Context, topic string, message []b
 		args.Approx = true
 	}
 
+	start := time.Now()
 	_, err := ps.client.XAdd(ctx, args).Result()
+	end := time.Since(start)
+
 	if err != nil {
 		ps.logger.Errorf("failed to publish message to Redis stream '%s': %v", topic, err)
 		return err
 	}
 
-	traceID := span.SpanContext().TraceID().String()
 	addr := fmt.Sprintf("%s:%d", ps.config.HostName, ps.config.Port)
-	ps.logger.Debugf("PUB %s %s %s", topic, traceID, addr)
+	ps.logger.Debug(&pubsub.Log{
+		Mode:          "PUB",
+		CorrelationID: span.SpanContext().TraceID().String(),
+		MessageValue:  string(message),
+		Topic:         topic,
+		Host:          addr,
+		PubSubBackend: "REDIS",
+		Time:          end.Microseconds(),
+	})
 	ps.metrics.IncrementCounter(ctx, "app_pubsub_publish_success_count", "topic", topic)
 
 	return nil
@@ -103,11 +123,28 @@ func (ps *PubSub) Subscribe(ctx context.Context, topic string) (*pubsub.Message,
 	spanCtx, span := ps.tracer.Start(ctx, "redis-subscribe")
 	defer span.End()
 
-	ps.metrics.IncrementCounter(spanCtx, "app_pubsub_subscribe_total_count", "topic", topic)
+	// Determine mode and consumer group for metrics
+	mode := ps.config.PubSubMode
+	if mode == "" {
+		mode = modeStreams
+	}
+
+	var consumerGroup string
+	if mode == modeStreams && ps.config.PubSubStreamsConfig != nil {
+		consumerGroup = ps.config.PubSubStreamsConfig.ConsumerGroup
+	}
+
+	// Increment subscribe total count with consumer_group label if using streams
+	if consumerGroup != "" {
+		ps.metrics.IncrementCounter(spanCtx, "app_pubsub_subscribe_total_count", "topic", topic, "consumer_group", consumerGroup)
+	} else {
+		ps.metrics.IncrementCounter(spanCtx, "app_pubsub_subscribe_total_count", "topic", topic)
+	}
 
+	start := time.Now()
 	msgChan := ps.ensureSubscription(ctx, topic)
 
-	msg := ps.waitForMessage(ctx, spanCtx, span, topic, msgChan)
+	msg := ps.waitForMessage(ctx, spanCtx, span, topic, msgChan, start, consumerGroup)
 
 	return msg, nil
 }
@@ -175,15 +212,28 @@ func (ps *PubSub) ensureSubscription(_ context.Context, topic string) chan *pubs
 
 // waitForMessage waits for a message from the channel.
 func (ps *PubSub) waitForMessage(ctx context.Context, spanCtx context.Context, span trace.Span,
-	topic string, msgChan chan *pubsub.Message) *pubsub.Message {
+	topic string, msgChan chan *pubsub.Message, start time.Time, consumerGroup string) *pubsub.Message {
 	select {
 	case msg := <-msgChan:
-		ps.metrics.IncrementCounter(spanCtx, "app_pubsub_subscribe_success_count", "topic", topic)
+		// Increment subscribe success count with consumer_group label if using streams
+		if consumerGroup != "" {
+			ps.metrics.IncrementCounter(spanCtx, "app_pubsub_subscribe_success_count", "topic", topic, "consumer_group", consumerGroup)
+		} else {
+			ps.metrics.IncrementCounter(spanCtx, "app_pubsub_subscribe_success_count", "topic", topic)
+		}
 
 		if msg != nil {
-			traceID := span.SpanContext().TraceID().String()
+			end := time.Since(start)
 			addr := fmt.Sprintf("%s:%d", ps.config.HostName, ps.config.Port)
-			ps.logger.Debugf("SUB %s %s %s", topic, traceID, addr)
+			ps.logger.Debug(&pubsub.Log{
+				Mode:          "SUB",
+				CorrelationID: span.SpanContext().TraceID().String(),
+				MessageValue:  string(msg.Value),
+				Topic:         topic,
+				Host:          addr,
+				PubSubBackend: "REDIS",
+				Time:          end.Microseconds(),
+			})
 		}
 
 		return msg

pkg/gofr/datasource/redis/redis.go

@@ -188,12 +188,6 @@ func (r *Redis) Close() error {
 func NewPubSub(conf config.Config, logger datasource.Logger, metrics Metrics) pubsub.Client {
 	redisConfig := getRedisConfig(conf, logger)
 
-	// Always parse PubSub config for NewPubSub since we're explicitly creating a PubSub client.
-	// getRedisConfig() only parses PubSub config when PUBSUB_BACKEND=REDIS, but NewPubSub() can be
-	// called directly (not via container), so we need to ensure config is parsed here.
-	// If it was already parsed, parsePubSubConfig() will just set the same values (idempotent).
-	parsePubSubConfig(conf, redisConfig)
-
 	// if Hostname is not provided, we won't try to connect to Redis
 	if redisConfig.HostName == "" {
 		return nil

pkg/gofr/datasource/redis/redis_test.go

@@ -48,6 +48,7 @@ func TestRedis_QueryLogging(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
+	// Mock Redis server setup
 	s, err := miniredis.Run()
 	require.NoError(t, err)
 
@@ -73,6 +74,7 @@ func TestRedis_QueryLogging(t *testing.T) {
 		assert.Equal(t, "OK", result)
 	})
 
+	// Assertions
 	assert.Contains(t, result, "ping")
 	assert.Contains(t, result, "set key value ex 60")
 }
@@ -81,6 +83,7 @@ func TestRedis_PipelineQueryLogging(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
+	// Mock Redis server setup
 	s, err := miniredis.Run()
 	require.NoError(t, err)
 
@@ -90,6 +93,7 @@ func TestRedis_PipelineQueryLogging(t *testing.T) {
 	mockMetric.EXPECT().RecordHistogram(gomock.Any(), "app_redis_stats", gomock.Any(),
 		"hostname", gomock.Any(), "type", gomock.Any()).AnyTimes()
 
+	// Execute Redis pipeline
 	result := testutil.StdoutOutputForFunc(func() {
 		mockLogger := logging.NewMockLogger(logging.DEBUG)
 		client := NewClient(config.NewMockConfig(map[string]string{
@@ -99,13 +103,16 @@ func TestRedis_PipelineQueryLogging(t *testing.T) {
 
 		require.NoError(t, err)
 
+		// Pipeline execution
 		pipe := client.Pipeline()
 		setCmd := pipe.Set(t.Context(), "key1", "value1", 1*time.Minute)
 		getCmd := pipe.Get(t.Context(), "key1")
 
+		// Pipeline Exec should return a non-nil error
 		_, err = pipe.Exec(t.Context())
 		require.NoError(t, err)
 
+		// Retrieve results
 		setResult, err := setCmd.Result()
 		require.NoError(t, err)
 		assert.Equal(t, "OK", setResult)
@@ -115,6 +122,7 @@ func TestRedis_PipelineQueryLogging(t *testing.T) {
 		assert.Equal(t, "value1", getResult)
 	})
 
+	// Assertions
 	assert.Contains(t, result, "ping")
 	assert.Contains(t, result, "set key1 value1 ex 60: OK")
 }
@@ -123,11 +131,13 @@ func TestRedis_Close(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
+	// Mock Redis server setup
 	s, err := miniredis.Run()
 	require.NoError(t, err)
 
 	defer s.Close()
 
+	// Mock metrics setup
 	mockMetric := NewMockMetrics(ctrl)
 	mockMetric.EXPECT().RecordHistogram(gomock.Any(), "app_redis_stats", gomock.Any(), "hostname",
 		gomock.Any(), "type", gomock.Any()).AnyTimes()