Support explicit ClickHouse credential sources and LZ4

README.md

@@ -147,9 +147,32 @@ Settings can be set via environment variables or loaded directly from Infisical
 | `METRICS_PORT` | `9090` | Prometheus metrics and health check port |
 | `ADMIN_TOKEN` | **required, no default** | Token for admin provisioning RPCs. Min 32 chars. Server refuses to start without it. |
 | `LUMEN_DEV` | `false` | Development only. Generates an ephemeral `ADMIN_TOKEN` instead of failing. |
-| `CLICKHOUSE_DSN` | `clickhouse://127.0.0.1:9001/lumen` | Native ClickHouse connection string |
+| `CLICKHOUSE_DSN` | `clickhouse://127.0.0.1:9000/lumen?dial_timeout=10s&compress=true` | Native ClickHouse connection string |
+| `CLICKHOUSE_DSN_SOURCE` | `infisical` | DSN selection: `infisical` or `environment`. See below. |
+| `CLICKHOUSE_COMPRESSION` | unset | Set `lz4` to override DSN compression. Unset preserves the DSN settings. |
 | `POSTGRES_DSN` | `postgres://postgres:postgres@localhost:5433/lumen?sslmode=disable` | Postgres connection string |
 
+### Managed connections and compression
+
+By default, a nonempty Infisical `CLICKHOUSE_DSN` overrides the environment value.
+If Infisical does not supply one, Lumen uses the environment value, then the development default. Explicit `infisical` mode keeps this behavior.
+
+For a platform-provided connection binding, set:
+
+```sh
+CLICKHOUSE_DSN_SOURCE=environment
+CLICKHOUSE_COMPRESSION=lz4
+```
+
+The platform must supply a nonempty `CLICKHOUSE_DSN`. Environment mode ignores the Infisical DSN and refuses startup when the environment DSN is empty.
+It never substitutes a different credential source. Other Infisical settings continue to load normally.
+These two selector variables come from the environment, not Infisical.
+
+`lz4` replaces any DSN compression method and level with LZ4 at the driver default level.
+Leave `CLICKHOUSE_COMPRESSION` unset to preserve all DSN compression settings, including disabled compression or another codec.
+Unsupported values stop startup. The override preserves credentials, database selection, TLS verification, timeouts and pool settings.
+Continue to configure verified TLS and the managed database's public CA. Compression does not enable or disable TLS.
+
 ### Infisical Secret Loading
 
 If `INFISICAL_CLIENT_ID` and `INFISICAL_CLIENT_SECRET` (or `INFISICAL_TOKEN`) are set, Lumen automatically fetches configuration secrets directly into memory at startup without writing them to process environment variables.

app/app.go

@@ -72,7 +72,7 @@ func Run(ctx context.Context, opts Options) error {
 		return err
 	}
 	chClient, err := connectWithRetry(initCtx, time.Second, func(attempt context.Context) (*ch.Client, error) {
-		return ch.NewClientContext(attempt, cfg.ClickHouseDSN)
+		return ch.NewClientContextWithOptions(attempt, cfg.ClickHouseDSN, ch.ClientOptions{Compression: cfg.ClickHouseCompression})
 	})
 	if err != nil {
 		return fmt.Errorf("ClickHouse startup connection failed: %w", err)

internal/ch/ch.go

@@ -71,9 +71,15 @@ func NewClient(dsn string) (*Client, error) {
 
 // NewClientContext bounds connection setup by the caller deadline.
 func NewClientContext(parent context.Context, dsn string) (*Client, error) {
-	opts, err := clickhouse.ParseDSN(dsn)
+	return NewClientContextWithOptions(parent, dsn, ClientOptions{})
+}
+
+// NewClientContextWithOptions applies explicit transport options before connecting.
+// Empty options preserve the connection settings in the DSN.
+func NewClientContextWithOptions(parent context.Context, dsn string, options ClientOptions) (*Client, error) {
+	opts, err := parseClientOptions(dsn, options)
 	if err != nil {
-		return nil, fmt.Errorf("failed to parse clickhouse dsn: %w", err)
+		return nil, err
 	}
 
 	database := opts.Auth.Database

internal/ch/options.go

@@ -0,0 +1,29 @@
+package ch
+
+import (
+	"errors"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+)
+
+// ClientOptions contains explicit overrides for DSN connection settings.
+type ClientOptions struct {
+	// Compression accepts empty or lz4. Empty preserves the DSN compression settings.
+	// lz4 replaces the DSN method and level with LZ4 at the driver default level.
+	Compression string
+}
+
+func parseClientOptions(dsn string, overrides ClientOptions) (*clickhouse.Options, error) {
+	if overrides.Compression != "" && overrides.Compression != "lz4" {
+		return nil, errors.New("ClickHouse compression override must be empty or lz4")
+	}
+	opts, err := clickhouse.ParseDSN(dsn)
+	if err != nil {
+		// Driver parse errors can contain the credential-bearing DSN.
+		return nil, errors.New("CLICKHOUSE_DSN is invalid. Check the connection settings")
+	}
+	if overrides.Compression == "lz4" {
+		opts.Compression = &clickhouse.Compression{Method: clickhouse.CompressionLZ4}
+	}
+	return opts, nil
+}

internal/ch/options_test.go

@@ -0,0 +1,88 @@
+package ch
+
+import (
+	"context"
+	"reflect"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+)
+
+func TestClientOptionsPreserveDSNWhenCompressionIsUnset(t *testing.T) {
+	for _, query := range []string{"", "?compress=false", "?compress=true", "?compress=zstd&compress_level=4", "?compress=lz4&compress_level=3"} {
+		t.Run(query, func(t *testing.T) {
+			dsn := "clickhouse://test-user:test-password@localhost:9440/app" + query
+			expected, err := clickhouse.ParseDSN(dsn)
+			if err != nil {
+				t.Fatal(err)
+			}
+			actual, err := parseClientOptions(dsn, ClientOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !reflect.DeepEqual(actual, expected) {
+				t.Fatal("empty override changed DSN options")
+			}
+		})
+	}
+}
+
+func TestClientOptionsLZ4OverridesOnlyCompression(t *testing.T) {
+	for _, query := range []string{"", "&compress=false", "&compress=zstd&compress_level=7", "&compress=lz4&compress_level=3"} {
+		t.Run(query, func(t *testing.T) {
+			dsn := "clickhouse://test-user:test-password@database.internal:9440/app?secure=true&skip_verify=false&tls_server_name=database.internal&dial_timeout=8s&read_timeout=17s&max_open_conns=9&max_idle_conns=3&block_buffer_size=4&max_execution_time=10" + query
+			before, err := clickhouse.ParseDSN(dsn)
+			if err != nil {
+				t.Fatal(err)
+			}
+			opts, err := parseClientOptions(dsn, ClientOptions{Compression: "lz4"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if opts.Compression == nil || opts.Compression.Method != clickhouse.CompressionLZ4 || opts.Compression.Level != 0 {
+				t.Fatal("LZ4 did not replace the DSN compression method and level")
+			}
+			if opts.TLS == nil || opts.TLS.InsecureSkipVerify || opts.TLS.ServerName != "database.internal" {
+				t.Fatal("compression changed verified TLS")
+			}
+			if opts.Auth.Database != "app" || opts.Auth.Username != "test-user" || opts.Auth.Password != "test-password" || opts.DialTimeout != 8*time.Second || opts.ReadTimeout != 17*time.Second {
+				t.Fatal("compression changed connection settings")
+			}
+			before.Compression = opts.Compression
+			if !reflect.DeepEqual(opts, before) {
+				t.Fatal("compression changed unrelated driver options")
+			}
+		})
+	}
+}
+
+func TestClientOptionsRejectInvalidOverridesBeforeConnecting(t *testing.T) {
+	for _, value := range []string{"zstd", "none", "LZ4", " lz4"} {
+		if _, err := parseClientOptions("clickhouse://127.0.0.1:1/app", ClientOptions{Compression: value}); err == nil {
+			t.Fatal("accepted unsupported override")
+		}
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	if _, err := NewClientContextWithOptions(ctx, "clickhouse://127.0.0.1:1/app", ClientOptions{Compression: "unsupported"}); err == nil || !strings.Contains(err.Error(), "compression override") {
+		t.Fatal("invalid override reached the connection path")
+	}
+}
+
+func TestClientOptionsDoNotExposeInvalidDSNCredentials(t *testing.T) {
+	const password = "private-password-marker"
+	for _, dsn := range []string{
+		"clickhouse://user:" + password + "@bad host/app",
+		"clickhouse://user:" + password + "@host/app?dial_timeout=" + password,
+	} {
+		_, err := parseClientOptions(dsn, ClientOptions{Compression: "lz4"})
+		if err == nil {
+			t.Fatal("accepted invalid DSN")
+		}
+		if strings.Contains(err.Error(), password) || strings.Contains(err.Error(), dsn) {
+			t.Fatal("parse error exposed connection credentials")
+		}
+	}
+}

internal/config/clickhouse_options_test.go

@@ -0,0 +1,98 @@
+package config
+
+import (
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestClickHouseDSNSourcePrecedence(t *testing.T) {
+	const envDSN = "clickhouse://env-user:env-secret@environment:9440/app?secure=true"
+	const vaultDSN = "clickhouse://vault-user:vault-secret@infisical:9440/app?secure=true"
+	for _, tc := range []struct{ name, source, environment, secret, want string }{
+		{"default keeps infisical first", "", envDSN, vaultDSN, vaultDSN},
+		{"explicit infisical first", "infisical", envDSN, vaultDSN, vaultDSN},
+		{"default falls back to environment", "", envDSN, "", envDSN},
+		{"explicit infisical keeps legacy fallback", "infisical", envDSN, "", envDSN},
+		{"environment overrides secret", "environment", envDSN, vaultDSN, envDSN},
+		{"environment missing ignores secret", "environment", "", vaultDSN, ""},
+		{"environment missing has no development fallback", "environment", "", "", ""},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Setenv("CLICKHOUSE_DSN_SOURCE", tc.source)
+			t.Setenv("CLICKHOUSE_DSN", tc.environment)
+			t.Setenv("CLICKHOUSE_COMPRESSION", "")
+			cfg := loadWithSecrets(map[string]string{"CLICKHOUSE_DSN": tc.secret, "POSTGRES_DSN": "postgres://secret-only/db"})
+			if cfg.ClickHouseDSN != tc.want {
+				t.Fatal("selected an unexpected credential source")
+			}
+			if cfg.PostgresDSN != "postgres://secret-only/db" {
+				t.Fatal("changed unrelated secret loading")
+			}
+			if os.Getenv("CLICKHOUSE_DSN") != tc.environment {
+				t.Fatal("secret loading changed the process environment")
+			}
+			cfg.AdminToken = strings.Repeat("a", minAdminTokenLen)
+			err := cfg.Validate()
+			wantErr := tc.source == "environment" && tc.environment == ""
+			if (err != nil) != wantErr {
+				t.Fatalf("validation mismatch: %v", err)
+			}
+			if err != nil && (strings.Contains(err.Error(), "vault-secret") || strings.Contains(err.Error(), "env-secret")) {
+				t.Fatal("error exposed credentials")
+			}
+		})
+	}
+}
+
+func TestClickHouseConfigurationDefaults(t *testing.T) {
+	t.Setenv("CLICKHOUSE_DSN_SOURCE", "")
+	t.Setenv("CLICKHOUSE_DSN", "")
+	t.Setenv("CLICKHOUSE_COMPRESSION", "")
+	cfg := loadWithSecrets(nil)
+	if cfg.ClickHouseDSNSource != "infisical" || cfg.ClickHouseCompression != "" {
+		t.Fatal("changed the default configuration")
+	}
+	if cfg.ClickHouseDSN != "clickhouse://127.0.0.1:9000/lumen?dial_timeout=10s&compress=true" {
+		t.Fatal("changed the legacy development DSN")
+	}
+}
+
+func TestClickHouseExplicitOptionsRejectInvalidValuesInDevMode(t *testing.T) {
+	for _, tc := range []struct{ name, source, dsn, compression string }{
+		{"unknown source", "automatic", "clickhouse://host/app", ""},
+		{"source is case sensitive", "Environment", "clickhouse://host/app", ""},
+		{"blank environment DSN", "environment", " \t\n", ""},
+		{"unsupported compression", "environment", "clickhouse://host/app", "zstd"},
+		{"compression is case sensitive", "environment", "clickhouse://host/app", "LZ4"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg := baseConfig()
+			cfg.AdminToken = ""
+			cfg.DevMode = true
+			cfg.ClickHouseDSNSource = tc.source
+			cfg.ClickHouseDSN = tc.dsn
+			cfg.ClickHouseCompression = tc.compression
+			if err := cfg.Validate(); err == nil {
+				t.Fatal("development mode bypassed connection option validation")
+			}
+			if cfg.AdminToken != "" {
+				t.Fatal("generated a token before rejecting invalid options")
+			}
+		})
+	}
+}
+
+func TestClickHouseCompressionIsExplicitEnvironmentOption(t *testing.T) {
+	t.Setenv("CLICKHOUSE_COMPRESSION", "lz4")
+	t.Setenv("CLICKHOUSE_DSN_SOURCE", "environment")
+	t.Setenv("CLICKHOUSE_DSN", "clickhouse://bound/app?compress=zstd")
+	cfg := loadWithSecrets(map[string]string{"CLICKHOUSE_COMPRESSION": "none", "CLICKHOUSE_DSN_SOURCE": "infisical", "CLICKHOUSE_DSN": "clickhouse://other/app"})
+	if cfg.ClickHouseCompression != "lz4" || cfg.ClickHouseDSNSource != "environment" || cfg.ClickHouseDSN != "clickhouse://bound/app?compress=zstd" {
+		t.Fatal("Infisical replaced the selected source or explicit compression option")
+	}
+	cfg.AdminToken = strings.Repeat("a", minAdminTokenLen)
+	if err := cfg.Validate(); err != nil {
+		t.Fatal(err)
+	}
+}

internal/config/config.go

@@ -9,6 +9,7 @@ import (
 	"log"
 	"os"
 	"strconv"
+	"strings"
 )
 
 // minAdminTokenLen is the shortest ADMIN_TOKEN we will accept. The admin port
@@ -28,15 +29,17 @@ var knownWeakAdminTokens = map[string]bool{
 
 // Config holds runtime configuration settings for Lumen.
 type Config struct {
-	IngestPort    int    // Port for public telemetry ingest (Connect + gRPC)
-	AdminPort     int    // Port for internal control plane RPCs
-	MetricsPort   int    // Port for Prometheus metrics and health endpoints
-	AdminToken    string // Internal authorization token for Admin RPCs
-	ClickHouseDSN string // Connection DSN for ClickHouse cluster
-	PostgresDSN   string // Connection DSN for Postgres control plane DB
-	GeoIPDBPath   string // Optional file path to MaxMind GeoLite2-City.mmdb
-	DevMode       bool   // When true, generate an ephemeral admin token instead of failing
-	CHHost        string // Hostname for ClickHouse (without scheme), for multi-tenant host prefixing
+	IngestPort            int    // Port for public telemetry ingest (Connect + gRPC)
+	AdminPort             int    // Port for internal control plane RPCs
+	MetricsPort           int    // Port for Prometheus metrics and health endpoints
+	AdminToken            string // Internal authorization token for Admin RPCs
+	ClickHouseDSN         string // Connection DSN for ClickHouse cluster
+	ClickHouseDSNSource   string // infisical preserves legacy precedence. environment requires the environment DSN
+	ClickHouseCompression string // Empty preserves the DSN. lz4 replaces its compression method and level
+	PostgresDSN           string // Connection DSN for Postgres control plane DB
+	GeoIPDBPath           string // Optional file path to MaxMind GeoLite2-City.mmdb
+	DevMode               bool   // When true, generate an ephemeral admin token instead of failing
+	CHHost                string // Hostname for ClickHouse (without scheme), for multi-tenant host prefixing
 }
 
 // Load loads configuration parameters from environment variables or Infisical directly into memory.
@@ -46,20 +49,30 @@ type Config struct {
 // set it. Callers must run Validate, which rejects an empty or weak token unless
 // LUMEN_DEV is set.
 func Load() *Config {
+	return loadWithSecrets(FetchInfisicalSecrets())
+}
+
+func loadWithSecrets(secrets map[string]string) *Config {
 	cfg := &Config{
-		IngestPort:    getEnvAsInt("INGEST_PORT", 50051),
-		AdminPort:     getEnvAsInt("ADMIN_PORT", 50052),
-		MetricsPort:   getEnvAsInt("METRICS_PORT", 9090),
-		AdminToken:    getEnv("ADMIN_TOKEN", ""),
-		DevMode:       getEnvAsBool("LUMEN_DEV", false),
-		ClickHouseDSN: getEnv("CLICKHOUSE_DSN", "clickhouse://127.0.0.1:9000/lumen?dial_timeout=10s&compress=true"),
-		PostgresDSN:   getEnv("POSTGRES_DSN", "postgres://postgres:postgres@localhost:5432/lumen?sslmode=disable"),
-		GeoIPDBPath:   getEnv("GEOIP_DB_PATH", ""),
-		CHHost:        getEnv("CH_HOST", "localhost"),
+		IngestPort:            getEnvAsInt("INGEST_PORT", 50051),
+		AdminPort:             getEnvAsInt("ADMIN_PORT", 50052),
+		MetricsPort:           getEnvAsInt("METRICS_PORT", 9090),
+		AdminToken:            getEnv("ADMIN_TOKEN", ""),
+		DevMode:               getEnvAsBool("LUMEN_DEV", false),
+		ClickHouseDSNSource:   getEnv("CLICKHOUSE_DSN_SOURCE", "infisical"),
+		ClickHouseCompression: os.Getenv("CLICKHOUSE_COMPRESSION"),
+		ClickHouseDSN:         getEnv("CLICKHOUSE_DSN", "clickhouse://127.0.0.1:9000/lumen?dial_timeout=10s&compress=true"),
+		PostgresDSN:           getEnv("POSTGRES_DSN", "postgres://postgres:postgres@localhost:5432/lumen?sslmode=disable"),
+		GeoIPDBPath:           getEnv("GEOIP_DB_PATH", ""),
+		CHHost:                getEnv("CH_HOST", "localhost"),
+	}
+
+	if cfg.ClickHouseDSNSource == "environment" {
+		// Do not substitute the development DSN when the selected environment value is absent.
+		cfg.ClickHouseDSN = os.Getenv("CLICKHOUSE_DSN")
 	}
 
 	// Apply secrets directly from Infisical in memory without exposing to process environment
-	secrets := FetchInfisicalSecrets()
 	if secrets != nil {
 		if val, ok := secrets["INGEST_PORT"]; ok && val != "" {
 			if port, err := strconv.Atoi(val); err == nil {
@@ -79,7 +92,7 @@ func Load() *Config {
 		if val, ok := secrets["ADMIN_TOKEN"]; ok && val != "" {
 			cfg.AdminToken = val
 		}
-		if val, ok := secrets["CLICKHOUSE_DSN"]; ok && val != "" {
+		if val, ok := secrets["CLICKHOUSE_DSN"]; ok && val != "" && cfg.ClickHouseDSNSource != "environment" {
 			cfg.ClickHouseDSN = val
 		}
 		if val, ok := secrets["POSTGRES_DSN"]; ok && val != "" {
@@ -100,6 +113,9 @@ func Load() *Config {
 // development-only values when LUMEN_DEV is set. Callers must treat a returned
 // error as fatal.
 func (c *Config) Validate() error {
+	if err := c.validateClickHouseOptions(); err != nil {
+		return err
+	}
 	if c.AdminToken == "" {
 		if !c.DevMode {
 			return errors.New(
@@ -137,6 +153,23 @@ func (c *Config) Validate() error {
 	return c.validatePorts()
 }
 
+// validateClickHouseOptions also runs before the development-mode early return.
+func (c *Config) validateClickHouseOptions() error {
+	switch c.ClickHouseDSNSource {
+	case "", "infisical": // A zero-value Config preserves legacy callers.
+	case "environment":
+		if strings.TrimSpace(c.ClickHouseDSN) == "" {
+			return errors.New("CLICKHOUSE_DSN must be nonempty when CLICKHOUSE_DSN_SOURCE is environment")
+		}
+	default:
+		return errors.New("CLICKHOUSE_DSN_SOURCE must be environment or infisical")
+	}
+	if c.ClickHouseCompression != "" && c.ClickHouseCompression != "lz4" {
+		return errors.New("CLICKHOUSE_COMPRESSION must be empty or lz4")
+	}
+	return nil
+}
+
 func (c *Config) validatePorts() error {
 	ports := map[string]int{
 		"INGEST_PORT":  c.IngestPort,