From 2fa5f2cf6fd1ef5a4c416b0405cd8119656128b2 Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Mon, 30 Mar 2026 00:31:00 +0100 Subject: [PATCH] redis counters (page views / active users) --- cmd/processor/main.go | 3 +- go.mod | 1 + go.sum | 2 + internal/consumer/consumer_test.go | 1 + internal/pipeline/pipeline.go | 9 ++++ internal/pipeline/pipeline_test.go | 1 + internal/storage/redis.go | 55 ++++++++++++++++++++++++ internal/storage/redis_test.go | 69 ++++++++++++++++++++++++++++++ 8 files changed, 140 insertions(+), 1 deletion(-) create mode 100644 internal/storage/redis.go create mode 100644 internal/storage/redis_test.go diff --git a/cmd/processor/main.go b/cmd/processor/main.go index e2a5ea8..e97b233 100644 --- a/cmd/processor/main.go +++ b/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") diff --git a/go.mod b/go.mod index 9ffd7fc..e2dfb3e 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 5817a0d..a0794e0 100644 --- a/go.sum +++ b/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= diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go index 49b598d..78e2e72 100644 --- a/internal/consumer/consumer_test.go +++ b/internal/consumer/consumer_test.go @@ -531,6 +531,7 @@ func testPipeline() *pipeline.Pipeline { processors.NewUserAgentParser(), nil, writer, + nil, &logger, ) } diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 79a3613..5dab46b 100644 --- a/internal/pipeline/pipeline.go +++ b/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,6 +26,7 @@ func NewPipeline( uaParser *processors.UserAgentParser, sessionMgr *processors.SessionManager, chWriter *storage.ClickHouseWriter, + redisStore *storage.RedisStore, logger *zerolog.Logger, ) *Pipeline { return &Pipeline{ @@ -32,6 +34,7 @@ func NewPipeline( 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 } diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index e45e9c9..eb39d3e 100644 --- a/internal/pipeline/pipeline_test.go +++ b/internal/pipeline/pipeline_test.go @@ -59,6 +59,7 @@ func TestProcessEventDerivesUUIDAndClampsScreenWidth(t *testing.T) { processors.NewUserAgentParser(), nil, writer, + nil, &logger, ) diff --git a/internal/storage/redis.go b/internal/storage/redis.go new file mode 100644 index 0000000..f37b44f --- /dev/null +++ b/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 +} diff --git a/internal/storage/redis_test.go b/internal/storage/redis_test.go new file mode 100644 index 0000000..b0b7e62 --- /dev/null +++ b/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()) +}