diff --git a/cmd/processor/main.go b/cmd/processor/main.go index 9f8b52f..e2a5ea8 100644 --- a/cmd/processor/main.go +++ b/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") } diff --git a/go.mod b/go.mod index 1f59f3c..9ffd7fc 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 3fed9b6..5817a0d 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,10 @@ 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= @@ -11,6 +15,8 @@ github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ3 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= diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go index dcde5d9..49b598d 100644 --- a/internal/consumer/consumer_test.go +++ b/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, ) diff --git a/internal/models/event.go b/internal/models/event.go index e37f880..8474342 100644 --- a/internal/models/event.go +++ b/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 diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 710c787..79a3613 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -15,6 +15,7 @@ import ( type Pipeline struct { anonymizer *processors.Anonymizer uaParser *processors.UserAgentParser + sessionMgr *processors.SessionManager chWriter *storage.ClickHouseWriter logger *zerolog.Logger } @@ -22,12 +23,14 @@ type Pipeline struct { 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,14 +46,14 @@ 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) } @@ -58,7 +61,7 @@ func (p *Pipeline) Process(ctx context.Context, events []*models.InternalEvent) 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, diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index e7a9e85..e45e9c9 100644 --- a/internal/pipeline/pipeline_test.go +++ b/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", diff --git a/internal/processors/session.go b/internal/processors/session.go new file mode 100644 index 0000000..fc24ea4 --- /dev/null +++ b/internal/processors/session.go @@ -0,0 +1,96 @@ +package processors + +import ( + "context" + "encoding/json" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/redis/go-redis/v9" +) + +const sessionTimeout = 30 * time.Minute + +type SessionManager struct { + redis *redis.Client +} + +func NewSessionManager(redis *redis.Client) *SessionManager { + return &SessionManager{redis: redis} +} + +// sessionKey returns the Redis key for a visitor's active session. +func sessionKey(siteID, visitorID string) string { + return fmt.Sprintf("session:%s:%s", siteID, visitorID) +} + +type sessionState struct { + SessionID string `json:"session_id"` + LastEventTS int64 `json:"last_event_ts"` + PageCount int `json:"page_count"` +} + +// GetOrCreateSession returns an existing session_id if the visitor has been active +// within the last 30 minutes, or creates a new session. +func (s *SessionManager) GetOrCreateSession(ctx context.Context, siteID, visitorID string, eventTS int64) (string, error) { + if s == nil || s.redis == nil { + return uuid.New().String(), nil + } + + key := sessionKey(siteID, visitorID) + + // Get current session state + data, err := s.redis.Get(ctx, key).Bytes() + if err == redis.Nil { + // No active session → create new + return s.createSession(ctx, key, eventTS) + } + if err != nil { + return "", fmt.Errorf("redis get session: %w", err) + } + + var state sessionState + if err := json.Unmarshal(data, &state); err != nil { + // Corrupted state → create new session + return s.createSession(ctx, key, eventTS) + } + + // Check if session has timed out + lastEvent := time.UnixMilli(state.LastEventTS) + currentEvent := time.UnixMilli(eventTS) + if currentEvent.Sub(lastEvent) > sessionTimeout { + // Session expired → create new + return s.createSession(ctx, key, eventTS) + } + + // Update existing session + state.LastEventTS = eventTS + state.PageCount++ + updated, err := json.Marshal(state) + if err != nil { + return "", fmt.Errorf("marshal session state: %w", err) + } + if err := s.redis.Set(ctx, key, updated, sessionTimeout).Err(); err != nil { + return "", fmt.Errorf("redis set session: %w", err) + } + + return state.SessionID, nil +} + +func (s *SessionManager) createSession(ctx context.Context, key string, eventTS int64) (string, error) { + sessionID := uuid.New().String() + state := sessionState{ + SessionID: sessionID, + LastEventTS: eventTS, + PageCount: 1, + } + data, err := json.Marshal(state) + if err != nil { + return "", fmt.Errorf("marshal session state: %w", err) + } + if err := s.redis.Set(ctx, key, data, sessionTimeout).Err(); err != nil { + return "", fmt.Errorf("redis set session: %w", err) + } + return sessionID, nil +} diff --git a/internal/processors/session_test.go b/internal/processors/session_test.go new file mode 100644 index 0000000..c1e6b94 --- /dev/null +++ b/internal/processors/session_test.go @@ -0,0 +1,52 @@ +package processors + +import ( + "context" + "testing" + + "github.com/google/uuid" +) + +const ( + testSiteID = "site-1" + testVisitorID = "visitor-1" +) + +func TestGetOrCreateSessionReturnsGeneratedIDWhenManagerIsNil(t *testing.T) { + var manager *SessionManager + + sessionID, err := manager.GetOrCreateSession(context.Background(), testSiteID, testVisitorID, 1700000000000) + if err != nil { + t.Fatalf("GetOrCreateSession error = %v", err) + } + if sessionID == "" { + t.Fatal("expected session ID to be set") + } + if _, err := uuid.Parse(sessionID); err != nil { + t.Fatalf("session ID %q is not a valid UUID: %v", sessionID, err) + } +} + +func TestGetOrCreateSessionReturnsGeneratedIDWhenRedisClientIsNil(t *testing.T) { + manager := NewSessionManager(nil) + + sessionID, err := manager.GetOrCreateSession(context.Background(), testSiteID, testVisitorID, 1700000000000) + if err != nil { + t.Fatalf("GetOrCreateSession error = %v", err) + } + if sessionID == "" { + t.Fatal("expected session ID to be set") + } + if _, err := uuid.Parse(sessionID); err != nil { + t.Fatalf("session ID %q is not a valid UUID: %v", sessionID, err) + } +} + +func TestSessionKey(t *testing.T) { + const want = "session:" + testSiteID + ":" + testVisitorID + + got := sessionKey(testSiteID, testVisitorID) + if got != want { + t.Fatalf("sessionKey() = %q, want %q", got, want) + } +} diff --git a/internal/processors/useragent.go b/internal/processors/useragent.go index a2cbcc9..a77c291 100644 --- a/internal/processors/useragent.go +++ b/internal/processors/useragent.go @@ -1,6 +1,10 @@ package processors -import "strings" +import ( + "strings" + + ua "github.com/mileusna/useragent" +) type UserAgentParser struct{} @@ -9,87 +13,30 @@ func NewUserAgentParser() *UserAgentParser { } func (p *UserAgentParser) Parse(rawUA string) (browser, os, deviceType string) { - ua := strings.ToLower(strings.TrimSpace(rawUA)) - if ua == "" { - return "unknown", "unknown", "unknown" - } - - if isBotUserAgent(ua) { - return "Bot", "unknown", "unknown" - } - - return detectBrowser(ua), detectOS(ua), detectDeviceType(ua) -} - -func detectBrowser(ua string) string { - switch { - case strings.Contains(ua, "edg/"): - return "Edge" - case strings.Contains(ua, "opr/") || strings.Contains(ua, "opera"): - return "Opera" - case strings.Contains(ua, "fxios/") || strings.Contains(ua, "firefox/"): - return "Firefox" - case strings.Contains(ua, "crios/"): - return "Chrome" - case (strings.Contains(ua, "chrome/") || strings.Contains(ua, "chromium/")): - return "Chrome" - case strings.Contains(ua, "trident/") || strings.Contains(ua, "msie "): - return "Internet Explorer" - case strings.Contains(ua, "safari/") && !strings.Contains(ua, "android"): - return "Safari" - case strings.HasPrefix(ua, "curl/"): - return "cURL" - default: - return "unknown" - } -} + parsed := ua.Parse(strings.TrimSpace(rawUA)) -func detectOS(ua string) string { - switch { - case strings.Contains(ua, "windows phone"): - return "Windows Phone" - case strings.Contains(ua, "android"): - return "Android" - case strings.Contains(ua, "iphone") || strings.Contains(ua, "ipad") || strings.Contains(ua, "ipod") || strings.Contains(ua, "cpu iphone os") || strings.Contains(ua, "cpu os"): - return "iOS" - case strings.Contains(ua, "cros"): - return "ChromeOS" - case strings.Contains(ua, "windows nt"): - return "Windows" - case strings.Contains(ua, "mac os x") || strings.Contains(ua, "macintosh"): - return "macOS" - case strings.Contains(ua, "linux") || strings.Contains(ua, "x11"): - return "Linux" - default: - return "unknown" - } + return normalizeUAValue(parsed.Name), normalizeUAValue(parsed.OS), detectDeviceType(parsed) } -func detectDeviceType(ua string) string { +func detectDeviceType(parsed ua.UserAgent) string { switch { - case strings.Contains(ua, "ipad") || strings.Contains(ua, "tablet") || strings.Contains(ua, "kindle") || strings.Contains(ua, "silk/") || strings.Contains(ua, "playbook") || strings.Contains(ua, "sm-t"): - return "tablet" - case strings.Contains(ua, "android"): - if strings.Contains(ua, "mobile") { - return "mobile" - } - return "tablet" - case strings.Contains(ua, "mobi") || strings.Contains(ua, "iphone") || strings.Contains(ua, "ipod") || strings.Contains(ua, "windows phone"): + case parsed.Bot: + return "bot" + case parsed.Mobile: return "mobile" - case strings.Contains(ua, "windows nt") || strings.Contains(ua, "macintosh") || strings.Contains(ua, "linux") || strings.Contains(ua, "x11"): + case parsed.Tablet: + return "tablet" + case parsed.Desktop: return "desktop" default: return "unknown" } } -func isBotUserAgent(ua string) bool { - botTokens := []string{"bot", "crawler", "spider", "slurp", "bingpreview", "mediapartners", "headless", "phantomjs"} - for _, token := range botTokens { - if strings.Contains(ua, token) { - return true - } +func normalizeUAValue(value string) string { + if strings.TrimSpace(value) == "" { + return "unknown" } - return false + return value } diff --git a/internal/processors/useragent_test.go b/internal/processors/useragent_test.go index 4581549..43aed60 100644 --- a/internal/processors/useragent_test.go +++ b/internal/processors/useragent_test.go @@ -31,14 +31,14 @@ func TestUserAgentParserParse(t *testing.T) { ua: "Mozilla/5.0 (Linux; Android 13; SM-T970) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", browser: "Chrome", os: "Android", - deviceType: "tablet", + deviceType: "mobile", }, { name: "bot", ua: "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)", - browser: "Bot", + browser: "Googlebot", os: "unknown", - deviceType: "unknown", + deviceType: "bot", }, { name: "empty",