feat(clickhouse): add unified PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME env var

Add provider-agnostic PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME that works
with PEERDB_CLICKHOUSE_STAGING_PROVIDER to configure staging storage.
GCS uses the unified var exclusively. S3 falls back to the legacy
PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME when the unified var is not set.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: whites11

flow/connectors/clickhouse/clickhouse.go

@@ -117,19 +117,26 @@ func createStagingStore(
 		return nil, nil, fmt.Errorf("failed to get staging provider: %w", err)
 	}
 
+	// Prefer unified PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME, fall back to provider-specific env vars.
+	bucketName, err := internal.PeerDBClickHouseStagingBucketName(ctx, env)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to get staging bucket name: %w", err)
+	}
+
 	switch strings.ToLower(provider) {
 	case "gcs":
-		store, err := createGCSStagingStore(ctx, env)
+		store, err := createGCSStagingStore(ctx, bucketName)
 		return store, nil, err
 	default:
-		return createS3StagingStore(ctx, env, config)
+		return createS3StagingStore(ctx, env, config, bucketName)
 	}
 }
 
 func createS3StagingStore(
 	ctx context.Context,
 	env map[string]string,
 	config *protos.ClickhouseConfig,
+	unifiedBucketName string,
 ) (utils.StagingStore, *utils.ClickHouseS3Credentials, error) {
 	var awsConfig utils.PeerAWSCredentials
 	var awsBucketPath string
@@ -157,9 +164,13 @@ func createS3StagingStore(
 		deploymentUID := internal.PeerDBDeploymentUID()
 		flowName, _ := ctx.Value(shared.FlowNameKey).(string)
 		bucketPathSuffix := fmt.Sprintf("%s/%s", url.PathEscape(deploymentUID), url.PathEscape(flowName))
-		awsBucketName, err := internal.PeerDBClickHouseAWSS3BucketName(ctx, env)
-		if err != nil {
-			return nil, nil, fmt.Errorf("failed to get PeerDB ClickHouse Bucket Name: %w", err)
+		// Prefer unified bucket name, fall back to legacy S3-specific env var.
+		awsBucketName := unifiedBucketName
+		if awsBucketName == "" {
+			awsBucketName, err = internal.PeerDBClickHouseAWSS3BucketName(ctx, env)
+			if err != nil {
+				return nil, nil, fmt.Errorf("failed to get PeerDB ClickHouse Bucket Name: %w", err)
+			}
 		}
 		if awsBucketName == "" {
 			return nil, nil, errors.New("PeerDB ClickHouse Bucket Name not set")
@@ -182,20 +193,16 @@ func createS3StagingStore(
 
 func createGCSStagingStore(
 	ctx context.Context,
-	env map[string]string,
+	bucketName string,
 ) (utils.StagingStore, error) {
-	gcsBucketName, err := internal.PeerDBClickHouseGCSBucketName(ctx, env)
-	if err != nil {
-		return nil, fmt.Errorf("failed to get GCS bucket name: %w", err)
-	}
-	if gcsBucketName == "" {
-		return nil, errors.New("PEERDB_CLICKHOUSE_GCS_BUCKET_NAME not set but staging provider is gcs")
+	if bucketName == "" {
+		return nil, errors.New("PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME must be set when staging provider is gcs")
 	}
 
 	deploymentUID := internal.PeerDBDeploymentUID()
 	flowName, _ := ctx.Value(shared.FlowNameKey).(string)
 	bucketPath := fmt.Sprintf("gs://%s/%s/%s",
-		gcsBucketName, url.PathEscape(deploymentUID), url.PathEscape(flowName))
+		bucketName, url.PathEscape(deploymentUID), url.PathEscape(flowName))
 
 	return utils.NewGCSStagingStore(ctx, bucketPath)
 }

flow/internal/dynamicconf.go

@@ -183,8 +183,8 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
 	},
 	{
-		Name:             "PEERDB_CLICKHOUSE_GCS_BUCKET_NAME",
-		Description:      "GCS bucket to store Avro files for mirrors with ClickHouse target",
+		Name:             "PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME",
+		Description:      "Staging bucket name for ClickHouse mirrors (provider-agnostic, preferred over legacy env vars)",
 		DefaultValue:     "",
 		ValueType:        protos.DynconfValueType_STRING,
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
@@ -730,10 +730,11 @@ func PeerDBClickHouseStagingProvider(ctx context.Context, env map[string]string)
 	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_PROVIDER")
 }
 
-func PeerDBClickHouseGCSBucketName(ctx context.Context, env map[string]string) (string, error) {
-	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_GCS_BUCKET_NAME")
+func PeerDBClickHouseStagingBucketName(ctx context.Context, env map[string]string) (string, error) {
+	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME")
 }
 
+
 func PeerDBS3UuidPrefix(ctx context.Context, env map[string]string) (bool, error) {
 	return dynamicConfBool(ctx, env, "PEERDB_S3_UUID_PREFIX")
 }