From 096a2ce63a4dcfeee0c801a8533996173df7b99f Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Sun, 19 Apr 2026 22:22:34 +0100 Subject: [PATCH 1/2] Add support for custom event --- internal/consumer/consumer_test.go | 7 +++-- internal/models/event.go | 3 ++ internal/pipeline/pipeline.go | 1 + internal/pipeline/pipeline_test.go | 49 ++++++++++++++++++++++++++++++ internal/storage/clickhouse.go | 4 +-- 5 files changed, 60 insertions(+), 4 deletions(-) diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go index c2cae23..c6ead20 100644 --- a/internal/consumer/consumer_test.go +++ b/internal/consumer/consumer_test.go @@ -190,7 +190,8 @@ func TestInternalEventUnmarshal(t *testing.T) { "event_id": "evt-123", "site_id": "site-456", "visitor_id": "vis-789", - "event_type": "page_view", + "event_type": "custom", + "event_name": "test_new_track", "timestamp": 1700000000000, "url": "https://example.com/page", "referrer": "https://google.com", @@ -216,7 +217,8 @@ func TestInternalEventUnmarshal(t *testing.T) { assert.Equal(t, "evt-123", event.EventID) assert.Equal(t, "site-456", event.SiteID) assert.Equal(t, "vis-789", event.VisitorID) - assert.Equal(t, models.EventPageView, event.EventType) + assert.Equal(t, models.EventCustom, event.EventType) + assert.Equal(t, "test_new_track", event.EventName) assert.Equal(t, int64(1700000000000), event.Timestamp) assert.Equal(t, "https://example.com/page", event.URL) assert.Equal(t, "https://google.com", event.Referrer) @@ -248,6 +250,7 @@ func TestInternalEventUnmarshalMinimalEvent(t *testing.T) { require.NoError(t, err) assert.Equal(t, "evt-min", event.EventID) assert.Equal(t, models.EventClick, event.EventType) + assert.Empty(t, event.EventName) assert.Empty(t, event.Referrer) assert.Empty(t, event.UTMSource) assert.Empty(t, event.IP) diff --git a/internal/models/event.go b/internal/models/event.go index 8474342..e5f728c 100644 --- a/internal/models/event.go +++ b/internal/models/event.go @@ -14,6 +14,7 @@ const ( EventPageView EventType = "page_view" EventPageLeave EventType = "page_leave" EventClick EventType = "click" + EventCustom EventType = "custom" EventScroll EventType = "scroll" EventMediaPlay EventType = "media_play" EventMediaPause EventType = "media_pause" @@ -70,6 +71,7 @@ type InternalEvent struct { SiteID string `json:"site_id"` VisitorID string `json:"visitor_id"` EventType EventType `json:"event_type"` + EventName string `json:"event_name,omitempty"` Timestamp int64 `json:"timestamp"` URL string `json:"url"` Referrer string `json:"referrer,omitempty"` @@ -91,6 +93,7 @@ type ProcessedEvent struct { VisitorID string SessionID string EventType EventType + EventName string Timestamp time.Time URL string Referrer string diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index 5dab46b..a74ef7e 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -101,6 +101,7 @@ func (p *Pipeline) processEvent(ctx context.Context, raw *models.InternalEvent) VisitorID: raw.VisitorID, SessionID: sessionID, EventType: raw.EventType, + EventName: raw.EventName, Timestamp: time.UnixMilli(raw.Timestamp), URL: raw.URL, Referrer: raw.Referrer, diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index 673951c..69604fd 100644 --- a/internal/pipeline/pipeline_test.go +++ b/internal/pipeline/pipeline_test.go @@ -90,6 +90,9 @@ func TestProcessEventDerivesUUIDAndClampsScreenWidth(t *testing.T) { if processed.DeviceType != "mobile" { t.Fatalf("DeviceType = %q, want %q", processed.DeviceType, "mobile") } + if processed.EventName != "" { + t.Fatalf("EventName = %q, want empty", processed.EventName) + } if processed.Browser != "Safari" { t.Fatalf("Browser = %q, want %q", processed.Browser, "Safari") } @@ -101,6 +104,52 @@ func TestProcessEventDerivesUUIDAndClampsScreenWidth(t *testing.T) { } } +func TestProcessEventPreservesCustomEventName(t *testing.T) { + writer, err := storage.NewClickHouseWriter(storage.Options{ + Addrs: []string{"some-addr"}, + Database: "default", + Username: "default", + Password: "test-password", + MaxOpenConns: 10, + MaxIdleConns: 5, + ConnOpenStrategy: "in_order", + }) + if err != nil { + t.Fatalf("failed to create writer: %v", err) + } + logger := zerolog.Nop() + p := NewPipeline( + processors.NewAnonymizer("test-salt", nil), + processors.NewUserAgentParser(), + nil, + writer, + nil, + &logger, + ) + + processed, err := p.processEvent(context.Background(), &models.InternalEvent{ + EventID: invalidEventID, + SiteID: "site-1", + VisitorID: "visitor-1", + EventType: models.EventCustom, + EventName: "test_new_track", + Timestamp: 1700000000000, + URL: "https://example.com", + IP: "192.168.1.12", + UserAgent: "Mozilla/5.0", + }) + if err != nil { + t.Fatalf("processEvent error = %v", err) + } + + if processed.EventType != models.EventCustom { + t.Fatalf("EventType = %q, want %q", processed.EventType, models.EventCustom) + } + if processed.EventName != "test_new_track" { + t.Fatalf("EventName = %q, want %q", processed.EventName, "test_new_track") + } +} + func TestClampUint16(t *testing.T) { tests := []struct { name string diff --git a/internal/storage/clickhouse.go b/internal/storage/clickhouse.go index 415a857..e74d7e2 100644 --- a/internal/storage/clickhouse.go +++ b/internal/storage/clickhouse.go @@ -80,7 +80,7 @@ func parseConnOpenStrategy(value string) (clickhouse.ConnOpenStrategy, error) { func (w *ClickHouseWriter) WriteBatch(ctx context.Context, events []*models.ProcessedEvent) error { batch, err := w.conn.PrepareBatch(ctx, ` INSERT INTO events ( - event_id, site_id, visitor_id, session_id, event_type, timestamp, + event_id, site_id, visitor_id, session_id, event_type, event_name, timestamp, url, referrer, utm_source, utm_medium, utm_campaign, country, region, device_type, browser, os, screen_width, properties, ip_hash, consent_given @@ -92,7 +92,7 @@ func (w *ClickHouseWriter) WriteBatch(ctx context.Context, events []*models.Proc for _, e := range events { if err := batch.Append( - e.EventID, e.SiteID, e.VisitorID, e.SessionID, string(e.EventType), + e.EventID, e.SiteID, e.VisitorID, e.SessionID, string(e.EventType), e.EventName, e.Timestamp, e.URL, e.Referrer, e.UTMSource, e.UTMMedium, e.UTMCampaign, e.Country, e.Region, e.DeviceType, e.Browser, e.OS, e.ScreenWidth, From ced4afcee77b0b88a1dbca3e6bd09fb9cd0faa61 Mon Sep 17 00:00:00 2001 From: Ret2Hell Date: Sun, 19 Apr 2026 22:25:50 +0100 Subject: [PATCH 2/2] refactor pipeline test --- internal/pipeline/pipeline_test.go | 116 +++++++++++++---------------- 1 file changed, 50 insertions(+), 66 deletions(-) diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index 69604fd..22a1bb7 100644 --- a/internal/pipeline/pipeline_test.go +++ b/internal/pipeline/pipeline_test.go @@ -13,6 +13,44 @@ import ( const invalidEventID = "evt-123" +func newTestPipeline(t *testing.T) *Pipeline { + t.Helper() + + writer, err := storage.NewClickHouseWriter(storage.Options{ + Addrs: []string{"some-addr"}, + Database: "default", + Username: "default", + Password: "test-password", + MaxOpenConns: 10, + MaxIdleConns: 5, + ConnOpenStrategy: "in_order", + }) + if err != nil { + t.Fatalf("failed to create writer: %v", err) + } + + logger := zerolog.Nop() + return NewPipeline( + processors.NewAnonymizer("test-salt", nil), + processors.NewUserAgentParser(), + nil, + writer, + nil, + &logger, + ) +} + +func newTestEvent() *models.InternalEvent { + return &models.InternalEvent{ + EventID: invalidEventID, + SiteID: "site-1", + VisitorID: "visitor-1", + Timestamp: 1700000000000, + URL: "https://example.com", + IP: "192.168.1.12", + } +} + func TestParseUUIDKeepsValidUUID(t *testing.T) { expected := uuid.New() @@ -41,41 +79,13 @@ func TestParseUUIDDerivesDeterministicUUID(t *testing.T) { } func TestProcessEventDerivesUUIDAndClampsScreenWidth(t *testing.T) { - writer, err := storage.NewClickHouseWriter(storage.Options{ - Addrs: []string{"some-addr"}, - Database: "default", - Username: "default", - Password: "test-password", - MaxOpenConns: 10, - MaxIdleConns: 5, - ConnOpenStrategy: "in_order", - }) - if err != nil { - t.Fatalf("failed to create writer: %v", err) - } - logger := zerolog.Nop() - p := NewPipeline( - processors.NewAnonymizer("test-salt", nil), - processors.NewUserAgentParser(), - nil, - writer, - nil, - &logger, - ) + p := newTestPipeline(t) + event := newTestEvent() + event.EventType = models.EventPageView + event.UserAgent = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1" + event.SessionContext.ScreenWidth = 100000 - processed, err := p.processEvent(context.Background(), &models.InternalEvent{ - EventID: invalidEventID, - SiteID: "site-1", - VisitorID: "visitor-1", - EventType: models.EventPageView, - Timestamp: 1700000000000, - URL: "https://example.com", - IP: "192.168.1.12", - UserAgent: "Mozilla/5.0 (iPhone; CPU iPhone OS 17_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1", - SessionContext: models.SessionContext{ - ScreenWidth: 100000, - }, - }) + processed, err := p.processEvent(context.Background(), event) if err != nil { t.Fatalf("processEvent error = %v", err) } @@ -105,39 +115,13 @@ func TestProcessEventDerivesUUIDAndClampsScreenWidth(t *testing.T) { } func TestProcessEventPreservesCustomEventName(t *testing.T) { - writer, err := storage.NewClickHouseWriter(storage.Options{ - Addrs: []string{"some-addr"}, - Database: "default", - Username: "default", - Password: "test-password", - MaxOpenConns: 10, - MaxIdleConns: 5, - ConnOpenStrategy: "in_order", - }) - if err != nil { - t.Fatalf("failed to create writer: %v", err) - } - logger := zerolog.Nop() - p := NewPipeline( - processors.NewAnonymizer("test-salt", nil), - processors.NewUserAgentParser(), - nil, - writer, - nil, - &logger, - ) + p := newTestPipeline(t) + event := newTestEvent() + event.EventType = models.EventCustom + event.EventName = "test_new_track" + event.UserAgent = "Mozilla/5.0" - processed, err := p.processEvent(context.Background(), &models.InternalEvent{ - EventID: invalidEventID, - SiteID: "site-1", - VisitorID: "visitor-1", - EventType: models.EventCustom, - EventName: "test_new_track", - Timestamp: 1700000000000, - URL: "https://example.com", - IP: "192.168.1.12", - UserAgent: "Mozilla/5.0", - }) + processed, err := p.processEvent(context.Background(), event) if err != nil { t.Fatalf("processEvent error = %v", err) }