Session Stitching

cmd/processor/main.go

@@ -18,33 +18,28 @@ import (
 	"github.com/Tanilytics/processing/internal/producer"
 	"github.com/Tanilytics/processing/internal/server"
 	"github.com/Tanilytics/processing/internal/storage"
+	"github.com/redis/go-redis/v9"
 	"github.com/rs/zerolog"
 )
 
-const clickhouseWriterShutdownErrorMsg = "clickhouse writer shutdown error"
-
-func main() {
-	// 1. Load configuration
-	cfg := config.LoadProcessorConfig()
-
-	// 2. Init structured logger
-	logger := observability.NewLogger("processor-service", cfg.LogLevel)
-	logger.Info().
-		Str("port", cfg.Port).
-		Strs("brokers", cfg.RedpandaBrokers).
-		Strs("clickhouse_addrs", cfg.ClickhouseAddrs).
-		Str("redis_url", cfg.RedisURL).
-		Msg("starting processor service")
+func newRedisClient(ctx context.Context, redisURL string, logger *zerolog.Logger) *redis.Client {
+	redisOpts, err := redis.ParseURL(redisURL)
+	if err != nil {
+		logger.Error().Err(err).Msg("failed to parse redis URL")
+		os.Exit(1)
+	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	redisClient := redis.NewClient(redisOpts)
+	if err := redisClient.Ping(ctx).Err(); err != nil {
+		logger.Error().Err(err).Msg("failed to connect to redis")
+		os.Exit(1)
+	}
 
-	// 3. Create HTTP server
-	app := server.NewServer(cfg.Port)
+	logger.Info().Msg("connected to redis")
+	return redisClient
+}
 
-	// 4. Initialize event consumer
-	anonymizer := processors.NewAnonymizer(cfg.AnonymizationSalt)
-	uaParser := processors.NewUserAgentParser()
+func newClickHouseWriter(cfg config.ProcessorConfig, logger *zerolog.Logger) (*storage.ClickHouseWriter, bool) {
 	chWriter, err := storage.NewClickHouseWriter(storage.Options{
 		Addrs:            cfg.ClickhouseAddrs,
 		Database:         cfg.ClickhouseDatabase,
@@ -57,11 +52,14 @@ func main() {
 	})
 	if err != nil {
 		logger.Error().Err(err).Msg("failed to initialize clickhouse writer")
-		return
+		return nil, false
 	}
 
-	processorPipeline := pipeline.NewPipeline(anonymizer, uaParser, chWriter, logger)
-	dlqProducer, err := producer.NewRedpandaProducer(
+	return chWriter, true
+}
+
+func newDLQProducer(cfg config.ProcessorConfig, logger *zerolog.Logger) (*producer.RedpandaProducer, error) {
+	return producer.NewRedpandaProducer(
 		cfg.RedpandaBrokers,
 		producer.ConnectionOptions{
 			SASL: producer.SASLOptions{
@@ -80,15 +78,15 @@ func main() {
 		},
 		logger,
 	)
-	if err != nil {
-		logger.Error().Err(err).Msg("failed to initialize dlq producer")
-		if closeErr := chWriter.Close(); closeErr != nil {
-			logger.Error().Err(closeErr).Msg(clickhouseWriterShutdownErrorMsg)
-		}
-		return
-	}
+}
 
-	eventConsumer, err := consumer.NewEventConsumer(
+func newEventConsumer(
+	cfg config.ProcessorConfig,
+	dlqProducer *producer.RedpandaProducer,
+	processorPipeline *pipeline.Pipeline,
+	logger *zerolog.Logger,
+) (*consumer.EventConsumer, error) {
+	return consumer.NewEventConsumer(
 		cfg.RedpandaBrokers,
 		consumer.ConnectionOptions{
 			GroupID: cfg.ConsumerGroup,
@@ -114,12 +112,59 @@ func main() {
 		processorPipeline,
 		logger,
 	)
+}
+
+func main() {
+	// 1. Load configuration
+	cfg := config.LoadProcessorConfig()
+
+	// 2. Init structured logger
+	logger := observability.NewLogger("processor-service", cfg.LogLevel)
+	logger.Info().
+		Str("port", cfg.Port).
+		Strs("brokers", cfg.RedpandaBrokers).
+		Strs("clickhouse_addrs", cfg.ClickhouseAddrs).
+		Str("redis_url", cfg.RedisURL).
+		Msg("starting processor service")
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// 3. Create HTTP server
+	app := server.NewServer(cfg.Port)
+
+	// 4. Initialize event consumer
+	anonymizer := processors.NewAnonymizer(cfg.AnonymizationSalt)
+	uaParser := processors.NewUserAgentParser()
+	redisClient := newRedisClient(ctx, cfg.RedisURL, logger)
+
+	defer func() {
+		//nolint:errcheck
+		redisClient.Close()
+	}()
+
+	sessionMgr := processors.NewSessionManager(redisClient)
+	chWriter, ok := newClickHouseWriter(cfg, logger)
+	if !ok {
+		return
+	}
+
+	processorPipeline := pipeline.NewPipeline(anonymizer, uaParser, sessionMgr, chWriter, logger)
+	dlqProducer, err := newDLQProducer(cfg, logger)
+	if err != nil {
+		logger.Error().Err(err).Msg("failed to initialize dlq producer")
+		//nolint:errcheck
+		chWriter.Close()
+		return
+	}
+
+	eventConsumer, err := newEventConsumer(cfg, dlqProducer, processorPipeline, logger)
 	if err != nil {
 		logger.Error().Err(err).Msg("failed to initialize consumer")
+		//nolint:errcheck
 		dlqProducer.Close()
-		if closeErr := chWriter.Close(); closeErr != nil {
-			logger.Error().Err(closeErr).Msg(clickhouseWriterShutdownErrorMsg)
-		}
+		//nolint:errcheck
+		chWriter.Close()
 		return
 	}
 
@@ -157,11 +202,12 @@ func main() {
 	cancel()
 	gracefulShutdown(app, logger)
 	consumerWG.Wait()
+	//nolint:errcheck
 	eventConsumer.Close()
+	//nolint:errcheck
 	dlqProducer.Close()
-	if err := chWriter.Close(); err != nil {
-		logger.Error().Err(err).Msg(clickhouseWriterShutdownErrorMsg)
-	}
+	//nolint:errcheck
+	chWriter.Close()
 
 	logger.Info().Msg("shutdown complete")
 }

go.mod

@@ -5,6 +5,8 @@ go 1.26.0
 require (
 	github.com/ClickHouse/clickhouse-go/v2 v2.43.0
 	github.com/google/uuid v1.6.0
+	github.com/mileusna/useragent v1.3.5
+	github.com/redis/go-redis/v9 v9.18.0
 	github.com/rs/zerolog v1.34.0
 	github.com/stretchr/testify v1.11.1
 	github.com/twmb/franz-go v1.20.7
@@ -15,6 +17,7 @@ require (
 	github.com/andybalholm/brotli v1.2.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	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/klauspost/compress v1.18.5 // indirect
@@ -30,6 +33,7 @@ require (
 	github.com/twmb/franz-go/pkg/kmsg v1.12.0 // indirect
 	go.opentelemetry.io/otel v1.42.0 // indirect
 	go.opentelemetry.io/otel/trace v1.42.0 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/crypto v0.48.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect

go.sum

@@ -4,13 +4,19 @@ github.com/ClickHouse/clickhouse-go/v2 v2.43.0 h1:fUR05TrF1GyvLDa/mAQjkx7KbgwdLR
 github.com/ClickHouse/clickhouse-go/v2 v2.43.0/go.mod h1:o6jf7JM/zveWC/PP277BLxjHy5KjnGX/jfljhM4s34g=
 github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ=
 github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 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=
@@ -30,6 +36,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
 github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
 github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4=
+github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
@@ -43,6 +51,8 @@ github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovk
 github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
 github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA=
 github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/mileusna/useragent v1.3.5 h1:SJM5NzBmh/hO+4LGeATKpaEX9+b4vcGg2qXGLiNGDws=
+github.com/mileusna/useragent v1.3.5/go.mod h1:3d8TOmwL/5I8pJjyVDteHtgDGcefrFUX4ccGOMKNYYc=
 github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc=
 github.com/paulmach/orb v0.12.0 h1:z+zOwjmG3MyEEqzv92UN49Lg1JFYx0L9GpGKNVDKk1s=
 github.com/paulmach/orb v0.12.0/go.mod h1:5mULz1xQfs3bmQm63QEJA6lNGujuRafwA5S/EnuLaLU=
@@ -53,6 +63,8 @@ github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsK
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs=
+github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0=
 github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
@@ -80,11 +92,15 @@ github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3i
 github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0=
+github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA=
 go.mongodb.org/mongo-driver v1.11.4/go.mod h1:PTSz5yu21bkT/wXpkS7WR5f0ddqw5quethTUn9WM+2g=
 go.opentelemetry.io/otel v1.42.0 h1:lSQGzTgVR3+sgJDAU/7/ZMjN9Z+vUip7leaqBKy4sho=
 go.opentelemetry.io/otel v1.42.0/go.mod h1:lJNsdRMxCUIWuMlVJWzecSMuNjE7dOYyWlqOXWkdqCc=
 go.opentelemetry.io/otel/trace v1.42.0 h1:OUCgIPt+mzOnaUTpOQcBiM/PLQ/Op7oq6g4LenLmOYY=
 go.opentelemetry.io/otel/trace v1.42.0/go.mod h1:f3K9S+IFqnumBkKhRJMeaZeNk9epyhnCmQh/EysQCdc=
+go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
+go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=

internal/consumer/consumer_test.go

@@ -529,6 +529,7 @@ func testPipeline() *pipeline.Pipeline {
 	return pipeline.NewPipeline(
 		processors.NewAnonymizer("test-salt"),
 		processors.NewUserAgentParser(),
+		nil,
 		writer,
 		&logger,
 	)

internal/models/event.go

@@ -99,7 +99,7 @@ type ProcessedEvent struct {
 	UTMCampaign  string
 	Country      string
 	Region       string
-	DeviceType   string // desktop, mobile, tablet
+	DeviceType   string // desktop, mobile, tablet, bot
 	Browser      string
 	OS           string
 	ScreenWidth  uint16

internal/pipeline/pipeline.go

@@ -15,19 +15,22 @@ import (
 type Pipeline struct {
 	anonymizer *processors.Anonymizer
 	uaParser   *processors.UserAgentParser
+	sessionMgr *processors.SessionManager
 	chWriter   *storage.ClickHouseWriter
 	logger     *zerolog.Logger
 }
 
 func NewPipeline(
 	anonymizer *processors.Anonymizer,
 	uaParser *processors.UserAgentParser,
+	sessionMgr *processors.SessionManager,
 	chWriter *storage.ClickHouseWriter,
 	logger *zerolog.Logger,
 ) *Pipeline {
 	return &Pipeline{
 		anonymizer: anonymizer,
 		uaParser:   uaParser,
+		sessionMgr: sessionMgr,
 		chWriter:   chWriter,
 		logger:     logger,
 	}
@@ -43,22 +46,22 @@ func (p *Pipeline) Process(ctx context.Context, events []*models.InternalEvent)
 		default:
 		}
 
-		pe, err := p.processEvent(raw)
+		pe, err := p.processEvent(ctx, raw)
 		if err != nil {
 			return err
 		}
 		processed = append(processed, pe)
 	}
 
-	// Step 4: Batch write to ClickHouse
+	// Step 5: Batch write to ClickHouse
 	if err := p.chWriter.WriteBatch(ctx, processed); err != nil {
 		return fmt.Errorf("clickhouse write: %w", err)
 	}
 
 	return nil
 }
 
-func (p *Pipeline) processEvent(raw *models.InternalEvent) (*models.ProcessedEvent, error) {
+func (p *Pipeline) processEvent(ctx context.Context, raw *models.InternalEvent) (*models.ProcessedEvent, error) {
 	// Step 1: Anonymize IP
 	ipHash, country, region := p.anonymizer.Anonymize(raw.IP)
 
@@ -71,11 +74,23 @@ func (p *Pipeline) processEvent(raw *models.InternalEvent) (*models.ProcessedEve
 		p.logger.Debug().Str("event_id", raw.EventID).Str("derived_uuid", eventID.String()).Msg("derived uuid from event id")
 	}
 
+	// Step 4: Session stitching
+	sessionID, err := p.sessionMgr.GetOrCreateSession(ctx, raw.SiteID, raw.VisitorID, raw.Timestamp)
+	if err != nil {
+		p.logger.Error().
+			Err(err).
+			Str("site_id", raw.SiteID).
+			Str("visitor_id", raw.VisitorID).
+			Msg("session stitching failed")
+		// Use a fallback session ID so we don't lose the event
+		sessionID = "unknown"
+	}
+
 	return &models.ProcessedEvent{
 		EventID:      eventID,
 		SiteID:       raw.SiteID,
 		VisitorID:    raw.VisitorID,
-		SessionID:    "unknown",
+		SessionID:    sessionID,
 		EventType:    raw.EventType,
 		Timestamp:    time.UnixMilli(raw.Timestamp),
 		URL:          raw.URL,

internal/pipeline/pipeline_test.go

@@ -1,6 +1,7 @@
 package pipeline
 
 import (
+	"context"
 	"testing"
 
 	"github.com/Tanilytics/processing/internal/models"
@@ -56,11 +57,12 @@ func TestProcessEventDerivesUUIDAndClampsScreenWidth(t *testing.T) {
 	p := NewPipeline(
 		processors.NewAnonymizer("test-salt"),
 		processors.NewUserAgentParser(),
+		nil,
 		writer,
 		&logger,
 	)
 
-	processed, err := p.processEvent(&models.InternalEvent{
+	processed, err := p.processEvent(context.Background(), &models.InternalEvent{
 		EventID:   invalidEventID,
 		SiteID:    "site-1",
 		VisitorID: "visitor-1",