redis counters (page views / active users)

cmd/processor/main.go

@@ -137,6 +137,7 @@ func main() {
 	anonymizer := processors.NewAnonymizer(cfg.AnonymizationSalt)
 	uaParser := processors.NewUserAgentParser()
 	redisClient := newRedisClient(ctx, cfg.RedisURL, logger)
+	redisStore := storage.NewRedisStore(redisClient)
 
 	defer func() {
 		//nolint:errcheck
@@ -149,7 +150,7 @@ func main() {
 		return
 	}
 
-	processorPipeline := pipeline.NewPipeline(anonymizer, uaParser, sessionMgr, chWriter, logger)
+	processorPipeline := pipeline.NewPipeline(anonymizer, uaParser, sessionMgr, chWriter, redisStore, logger)
 	dlqProducer, err := newDLQProducer(cfg, logger)
 	if err != nil {
 		logger.Error().Err(err).Msg("failed to initialize dlq producer")

go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/go-faster/city v1.0.1 // indirect
 	github.com/go-faster/errors v0.7.1 // indirect
+	github.com/go-redis/redismock/v9 v9.2.0 // indirect
 	github.com/klauspost/compress v1.18.5 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect

go.sum

@@ -21,6 +21,8 @@ github.com/go-faster/city v1.0.1 h1:4WAxSZ3V2Ws4QRDrscLEDcibJY8uf41H6AhXDrNDcGw=
 github.com/go-faster/city v1.0.1/go.mod h1:jKcUJId49qdW3L1qKHH/3wPeUstCVpVSXTM6vO3VcTw=
 github.com/go-faster/errors v0.7.1 h1:MkJTnDoEdi9pDabt1dpWf7AA8/BaSYZqibYyhZ20AYg=
 github.com/go-faster/errors v0.7.1/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo=
+github.com/go-redis/redismock/v9 v9.2.0 h1:ZrMYQeKPECZPjOj5u9eyOjg8Nnb0BS9lkVIZ6IpsKLw=
+github.com/go-redis/redismock/v9 v9.2.0/go.mod h1:18KHfGDK4Y6c2R0H38EUGWAdc7ZQS9gfYxc94k7rWT0=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=

internal/consumer/consumer_test.go

@@ -531,6 +531,7 @@ func testPipeline() *pipeline.Pipeline {
 		processors.NewUserAgentParser(),
 		nil,
 		writer,
+		nil,
 		&logger,
 	)
 }

internal/pipeline/pipeline.go

@@ -17,6 +17,7 @@ type Pipeline struct {
 	uaParser   *processors.UserAgentParser
 	sessionMgr *processors.SessionManager
 	chWriter   *storage.ClickHouseWriter
+	redisStore *storage.RedisStore
 	logger     *zerolog.Logger
 }
 
@@ -25,13 +26,15 @@ func NewPipeline(
 	uaParser *processors.UserAgentParser,
 	sessionMgr *processors.SessionManager,
 	chWriter *storage.ClickHouseWriter,
+	redisStore *storage.RedisStore,
 	logger *zerolog.Logger,
 ) *Pipeline {
 	return &Pipeline{
 		anonymizer: anonymizer,
 		uaParser:   uaParser,
 		sessionMgr: sessionMgr,
 		chWriter:   chWriter,
+		redisStore: redisStore,
 		logger:     logger,
 	}
 }
@@ -58,6 +61,12 @@ func (p *Pipeline) Process(ctx context.Context, events []*models.InternalEvent)
 		return fmt.Errorf("clickhouse write: %w", err)
 	}
 
+	// Step 6: Update Redis counters
+	if err := p.redisStore.UpdateCounters(ctx, processed); err != nil {
+		// Redis counter failure is non-fatal —> log but don't fail the pipeline
+		p.logger.Error().Err(err).Msg("redis counter update failed")
+	}
+
 	return nil
 }
 

internal/pipeline/pipeline_test.go

@@ -59,6 +59,7 @@ func TestProcessEventDerivesUUIDAndClampsScreenWidth(t *testing.T) {
 		processors.NewUserAgentParser(),
 		nil,
 		writer,
+		nil,
 		&logger,
 	)
 

internal/storage/redis.go

@@ -0,0 +1,55 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/Tanilytics/processing/internal/models"
+	"github.com/redis/go-redis/v9"
+)
+
+type RedisStore struct {
+	client *redis.Client
+}
+
+func NewRedisStore(client *redis.Client) *RedisStore {
+	return &RedisStore{client: client}
+}
+
+func (r *RedisStore) UpdateCounters(ctx context.Context, events []*models.ProcessedEvent) error {
+	pipe := r.client.Pipeline()
+	now := time.Now().UTC()
+	hourKey := now.Format("2006-01-02-15") // YYYY-MM-DD-HH
+
+	for _, e := range events {
+		if e.EventType == models.EventPageView {
+			// Increment hourly page view counter
+			pvKey := fmt.Sprintf("rt:pageviews:%s:%s", e.SiteID, hourKey)
+			pipe.Incr(ctx, pvKey)
+			pipe.Expire(ctx, pvKey, 25*time.Hour)
+
+			// Add to HyperLogLog for unique visitors
+			hllKey := fmt.Sprintf("rt:visitors:%s:%s", e.SiteID, hourKey)
+			pipe.PFAdd(ctx, hllKey, e.VisitorID)
+			pipe.Expire(ctx, hllKey, 25*time.Hour)
+		}
+
+		// Track active users (sorted set: visitor_id → timestamp)
+		activeKey := fmt.Sprintf("rt:active:%s", e.SiteID)
+		pipe.ZAdd(ctx, activeKey, redis.Z{
+			Score:  float64(e.Timestamp.UnixMilli()),
+			Member: e.VisitorID,
+		})
+		// Remove visitors inactive for > 5 minutes
+		cutoff := float64(now.Add(-5 * time.Minute).UnixMilli())
+		pipe.ZRemRangeByScore(ctx, activeKey, "-inf", fmt.Sprintf("%f", cutoff))
+		pipe.Expire(ctx, activeKey, 10*time.Minute)
+	}
+
+	_, err := pipe.Exec(ctx)
+	if err != nil {
+		return fmt.Errorf("redis pipeline: %w", err)
+	}
+	return nil
+}

internal/storage/redis_test.go

@@ -0,0 +1,69 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/Tanilytics/processing/internal/models"
+	redismock "github.com/go-redis/redismock/v9"
+	"github.com/redis/go-redis/v9"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRedisStoreUpdateCountersForPageView(t *testing.T) {
+	client, mock := redismock.NewClientMock()
+	store := NewRedisStore(client)
+
+	now := time.Now().UTC()
+	hourKey := now.Format("2006-01-02-15")
+	eventTS := now.Add(-time.Minute)
+	event := &models.ProcessedEvent{
+		SiteID:    "site-1",
+		VisitorID: "visitor-1",
+		EventType: models.EventPageView,
+		Timestamp: eventTS,
+	}
+
+	pvKey := fmt.Sprintf("rt:pageviews:%s:%s", event.SiteID, hourKey)
+	hllKey := fmt.Sprintf("rt:visitors:%s:%s", event.SiteID, hourKey)
+	activeKey := fmt.Sprintf("rt:active:%s", event.SiteID)
+
+	mock.ExpectIncr(pvKey).SetVal(1)
+	mock.ExpectExpire(pvKey, 25*time.Hour).SetVal(true)
+	mock.ExpectPFAdd(hllKey, event.VisitorID).SetVal(1)
+	mock.ExpectExpire(hllKey, 25*time.Hour).SetVal(true)
+	mock.ExpectZAdd(activeKey, redis.Z{Score: float64(eventTS.UnixMilli()), Member: event.VisitorID}).SetVal(1)
+	mock.Regexp().ExpectZRemRangeByScore(activeKey, "-inf", `^[0-9]+\.000000$`).SetVal(0)
+	mock.ExpectExpire(activeKey, 10*time.Minute).SetVal(true)
+
+	err := store.UpdateCounters(context.Background(), []*models.ProcessedEvent{event})
+	require.NoError(t, err)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}
+
+func TestRedisStoreUpdateCountersSkipsPageViewCountersForOtherEvents(t *testing.T) {
+	client, mock := redismock.NewClientMock()
+	store := NewRedisStore(client)
+
+	now := time.Now().UTC()
+	eventTS := now.Add(-time.Minute)
+	event := &models.ProcessedEvent{
+		SiteID:    "site-1",
+		VisitorID: "visitor-2",
+		EventType: models.EventClick,
+		Timestamp: eventTS,
+	}
+
+	activeKey := fmt.Sprintf("rt:active:%s", event.SiteID)
+
+	mock.ExpectZAdd(activeKey, redis.Z{Score: float64(eventTS.UnixMilli()), Member: event.VisitorID}).SetVal(1)
+	mock.Regexp().ExpectZRemRangeByScore(activeKey, "-inf", `^[0-9]+\.000000$`).SetVal(0)
+	mock.ExpectExpire(activeKey, 10*time.Minute).SetVal(true)
+
+	err := store.UpdateCounters(context.Background(), []*models.ProcessedEvent{event})
+	require.NoError(t, err)
+	assert.NoError(t, mock.ExpectationsWereMet())
+}