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.
Falls back to legacy PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME and
PEERDB_CLICKHOUSE_GCS_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, env, 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")
@@ -183,13 +194,19 @@ func createS3StagingStore(
 func createGCSStagingStore(
 	ctx context.Context,
 	env map[string]string,
+	unifiedBucketName 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)
+	// Prefer unified bucket name, fall back to legacy GCS-specific env var.
+	gcsBucketName := unifiedBucketName
+	if gcsBucketName == "" {
+		var err 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")
+		return nil, errors.New("staging bucket name not set for gcs provider")
 	}
 
 	deploymentUID := internal.PeerDBDeploymentUID()

flow/internal/dynamicconf.go

@@ -182,9 +182,17 @@ var DynamicSettings = [...]*protos.DynamicSetting{
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_AFTER_RESUME,
 		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
 	},
+	{
+		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,
+		TargetForSetting: protos.DynconfTarget_CLICKHOUSE,
+	},
 	{
 		Name:             "PEERDB_CLICKHOUSE_GCS_BUCKET_NAME",
-		Description:      "GCS bucket to store Avro files for mirrors with ClickHouse target",
+		Description:      "GCS bucket to store Avro files for mirrors with ClickHouse target (legacy, prefer PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME)",
 		DefaultValue:     "",
 		ValueType:        protos.DynconfValueType_STRING,
 		ApplyMode:        protos.DynconfApplyMode_APPLY_MODE_IMMEDIATE,
@@ -730,6 +738,10 @@ func PeerDBClickHouseStagingProvider(ctx context.Context, env map[string]string)
 	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_PROVIDER")
 }
 
+func PeerDBClickHouseStagingBucketName(ctx context.Context, env map[string]string) (string, error) {
+	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME")
+}
+
 func PeerDBClickHouseGCSBucketName(ctx context.Context, env map[string]string) (string, error) {
 	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_GCS_BUCKET_NAME")
 }