feat: chlogstore supports multi-deployment settings

internal/config/config.go

@@ -473,10 +473,11 @@ func (c *Config) ToMigratorOpts() migrator.MigrationOpts {
 			URL: c.PostgresURL,
 		},
 		CH: migrator.MigrationOptsCH{
-			Addr:     c.ClickHouse.Addr,
-			Username: c.ClickHouse.Username,
-			Password: c.ClickHouse.Password,
-			Database: c.ClickHouse.Database,
+			Addr:         c.ClickHouse.Addr,
+			Username:     c.ClickHouse.Username,
+			Password:     c.ClickHouse.Password,
+			Database:     c.ClickHouse.Database,
+			DeploymentID: c.DeploymentID,
 		},
 	}
 }

internal/logstore/chlogstore/chlogstore.go

@@ -15,13 +15,23 @@ import (
 )
 
 type logStoreImpl struct {
-	chDB clickhouse.DB
+	chDB            clickhouse.DB
+	eventsTable     string
+	deliveriesTable string
 }
 
 var _ driver.LogStore = (*logStoreImpl)(nil)
 
-func NewLogStore(chDB clickhouse.DB) driver.LogStore {
-	return &logStoreImpl{chDB: chDB}
+func NewLogStore(chDB clickhouse.DB, deploymentID string) driver.LogStore {
+	prefix := ""
+	if deploymentID != "" {
+		prefix = deploymentID + "_"
+	}
+	return &logStoreImpl{
+		chDB:            chDB,
+		eventsTable:     prefix + "events",
+		deliveriesTable: prefix + "deliveries",
+	}
 }
 
 func (s *logStoreImpl) ListEvent(ctx context.Context, req driver.ListEventRequest) (driver.ListEventResponse, error) {
@@ -114,11 +124,11 @@ func (s *logStoreImpl) ListEvent(ctx context.Context, req driver.ListEventReques
 			event_time,
 			metadata,
 			data
-		FROM events
+		FROM %s
 		WHERE %s
 		%s
 		LIMIT %d
-	`, whereClause, orderByClause, limit+1)
+	`, s.eventsTable, whereClause, orderByClause, limit+1)
 
 	rows, err := s.chDB.Query(ctx, query, args...)
 	if err != nil {
@@ -384,11 +394,11 @@ func (s *logStoreImpl) ListDeliveryEvent(ctx context.Context, req driver.ListDel
 			response_data,
 			manual,
 			attempt
-		FROM deliveries
+		FROM %s
 		WHERE %s
 		%s
 		LIMIT %d
-	`, whereClause, orderByClause, limit+1)
+	`, s.deliveriesTable, whereClause, orderByClause, limit+1)
 
 	rows, err := s.chDB.Query(ctx, query, args...)
 	if err != nil {
@@ -582,9 +592,9 @@ func (s *logStoreImpl) RetrieveEvent(ctx context.Context, req driver.RetrieveEve
 			event_time,
 			metadata,
 			data
-		FROM deliveries
+		FROM %s
 		WHERE %s
-		LIMIT 1`, whereClause)
+		LIMIT 1`, s.deliveriesTable, whereClause)
 
 	rows, err := s.chDB.Query(ctx, query, args...)
 	if err != nil {
@@ -658,9 +668,9 @@ func (s *logStoreImpl) RetrieveDeliveryEvent(ctx context.Context, req driver.Ret
 			response_data,
 			manual,
 			attempt
-		FROM deliveries
+		FROM %s
 		WHERE %s
-		LIMIT 1`, whereClause)
+		LIMIT 1`, s.deliveriesTable, whereClause)
 
 	rows, err := s.chDB.Query(ctx, query, args...)
 	if err != nil {
@@ -766,9 +776,9 @@ func (s *logStoreImpl) InsertManyDeliveryEvent(ctx context.Context, deliveryEven
 	}
 
 	eventBatch, err := s.chDB.PrepareBatch(ctx,
-		`INSERT INTO events (
+		fmt.Sprintf(`INSERT INTO %s (
 			event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data
-		)`,
+		)`, s.eventsTable),
 	)
 	if err != nil {
 		return fmt.Errorf("prepare events batch failed: %w", err)
@@ -803,10 +813,10 @@ func (s *logStoreImpl) InsertManyDeliveryEvent(ctx context.Context, deliveryEven
 	}
 
 	deliveryBatch, err := s.chDB.PrepareBatch(ctx,
-		`INSERT INTO deliveries (
+		fmt.Sprintf(`INSERT INTO %s (
 			event_id, tenant_id, destination_id, topic, eligible_for_retry, event_time, metadata, data,
 			delivery_id, delivery_event_id, status, delivery_time, code, response_data, manual, attempt
-		)`,
+		)`, s.deliveriesTable),
 	)
 	if err != nil {
 		return fmt.Errorf("prepare deliveries batch failed: %w", err)

internal/logstore/chlogstore/chlogstore_test.go

@@ -21,8 +21,9 @@ func TestConformance(t *testing.T) {
 }
 
 type harness struct {
-	chDB   clickhouse.DB
-	closer func()
+	chDB         clickhouse.DB
+	deploymentID string
+	closer       func()
 }
 
 func setupClickHouseConnection(t *testing.T) clickhouse.DB {
@@ -75,12 +76,71 @@ func (h *harness) Close() {
 
 func (h *harness) FlushWrites(ctx context.Context) error {
 	// Force ClickHouse to merge parts and deduplicate rows on both tables
-	if err := h.chDB.Exec(ctx, "OPTIMIZE TABLE events FINAL"); err != nil {
+	eventsTable := "events"
+	deliveriesTable := "deliveries"
+	if h.deploymentID != "" {
+		eventsTable = h.deploymentID + "_events"
+		deliveriesTable = h.deploymentID + "_deliveries"
+	}
+	if err := h.chDB.Exec(ctx, "OPTIMIZE TABLE "+eventsTable+" FINAL"); err != nil {
 		return err
 	}
-	return h.chDB.Exec(ctx, "OPTIMIZE TABLE deliveries FINAL")
+	return h.chDB.Exec(ctx, "OPTIMIZE TABLE "+deliveriesTable+" FINAL")
 }
 
 func (h *harness) MakeDriver(ctx context.Context) (driver.LogStore, error) {
-	return NewLogStore(h.chDB), nil
+	return NewLogStore(h.chDB, h.deploymentID), nil
+}
+
+func TestConformance_WithDeploymentID(t *testing.T) {
+	testutil.CheckIntegrationTest(t)
+	t.Parallel()
+
+	drivertest.RunConformanceTests(t, newHarnessWithDeploymentID)
+}
+
+func newHarnessWithDeploymentID(ctx context.Context, t *testing.T) (drivertest.Harness, error) {
+	t.Helper()
+
+	chDB := setupClickHouseConnectionWithDeploymentID(t, "mydeployment")
+
+	return &harness{
+		chDB:         chDB,
+		deploymentID: "mydeployment",
+		closer: func() {
+			chDB.Close()
+		},
+	}, nil
+}
+
+func setupClickHouseConnectionWithDeploymentID(t *testing.T, deploymentID string) clickhouse.DB {
+	t.Helper()
+	t.Cleanup(testinfra.Start(t))
+
+	chConfig := testinfra.NewClickHouseConfig(t)
+
+	chDB, err := clickhouse.New(&chConfig)
+	require.NoError(t, err)
+
+	ctx := context.Background()
+	m, err := migrator.New(migrator.MigrationOpts{
+		CH: migrator.MigrationOptsCH{
+			Addr:         chConfig.Addr,
+			Username:     chConfig.Username,
+			Password:     chConfig.Password,
+			Database:     chConfig.Database,
+			DeploymentID: deploymentID,
+		},
+	})
+	require.NoError(t, err)
+	_, _, err = m.Up(ctx, -1)
+	require.NoError(t, err)
+
+	defer func() {
+		sourceErr, dbErr := m.Close(ctx)
+		require.NoError(t, sourceErr)
+		require.NoError(t, dbErr)
+	}()
+
+	return chDB
 }

internal/logstore/logstore.go

@@ -29,8 +29,9 @@ type LogStore interface {
 }
 
 type DriverOpts struct {
-	CH clickhouse.DB
-	PG *pgxpool.Pool
+	CH           clickhouse.DB
+	PG           *pgxpool.Pool
+	DeploymentID string
 }
 
 func (d *DriverOpts) Close() error {
@@ -45,7 +46,7 @@ func (d *DriverOpts) Close() error {
 
 func NewLogStore(ctx context.Context, driverOpts DriverOpts) (LogStore, error) {
 	if driverOpts.CH != nil {
-		return chlogstore.NewLogStore(driverOpts.CH), nil
+		return chlogstore.NewLogStore(driverOpts.CH, driverOpts.DeploymentID), nil
 	}
 	if driverOpts.PG != nil {
 		return pglogstore.NewLogStore(driverOpts.PG), nil
@@ -60,12 +61,15 @@ func NewMemLogStore() LogStore {
 }
 
 type Config struct {
-	ClickHouse *clickhouse.ClickHouseConfig
-	Postgres   *string
+	ClickHouse   *clickhouse.ClickHouseConfig
+	Postgres     *string
+	DeploymentID string
 }
 
 func MakeDriverOpts(cfg Config) (DriverOpts, error) {
-	driverOpts := DriverOpts{}
+	driverOpts := DriverOpts{
+		DeploymentID: cfg.DeploymentID,
+	}
 
 	if cfg.ClickHouse != nil {
 		chDB, err := clickhouse.New(cfg.ClickHouse)

internal/migrator/migrations/clickhouse/000001_init.down.sql

@@ -1,2 +1,2 @@
-DROP TABLE IF EXISTS deliveries;
-DROP TABLE IF EXISTS events;
+DROP TABLE IF EXISTS {deployment_prefix}deliveries;
+DROP TABLE IF EXISTS {deployment_prefix}events;

internal/migrator/migrations/clickhouse/000001_init.up.sql

@@ -2,7 +2,7 @@
 -- Each row represents a unique event (ReplacingMergeTree deduplicates by ORDER BY)
 -- Enables O(limit) event listing without GROUP BY
 
-CREATE TABLE IF NOT EXISTS events (
+CREATE TABLE IF NOT EXISTS {deployment_prefix}events (
     event_id String,
     tenant_id String,
     destination_id String,
@@ -25,7 +25,7 @@ ORDER BY (event_time, event_id);
 -- Each row represents a delivery attempt for an event
 -- Stateless queries: no GROUP BY, no aggregation, direct row access
 
-CREATE TABLE IF NOT EXISTS deliveries (
+CREATE TABLE IF NOT EXISTS {deployment_prefix}deliveries (
     -- Event fields
     event_id String,
     tenant_id String,