refactor(staging): move S3 bucket name fallback into PeerDBClickHouseStagingBucketName

Centralize the fallback from PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME to
the legacy PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME in the config lookup
function, simplifying createS3StagingStore.

Author: whites11

flow/connectors/clickhouse/staging.go

@@ -81,21 +81,13 @@ func createS3StagingStore(
 	}
 
 	if awsBucketPath == "" {
+		if unifiedBucketName == "" {
+			return nil, nil, errors.New("PeerDB ClickHouse Bucket Name not set")
+		}
 		deploymentUID := internal.PeerDBDeploymentUID()
 		flowName, _ := ctx.Value(shared.FlowNameKey).(string)
 		bucketPathSuffix := fmt.Sprintf("%s/%s", url.PathEscape(deploymentUID), url.PathEscape(flowName))
-		// 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")
-		}
-		awsBucketPath = fmt.Sprintf("s3://%s/%s", awsBucketName, bucketPathSuffix)
+		awsBucketPath = fmt.Sprintf("s3://%s/%s", unifiedBucketName, bucketPathSuffix)
 	}
 
 	// S3 with session tokens requires ClickHouse >= 24.3.1

flow/internal/dynamicconf.go

@@ -731,7 +731,15 @@ func PeerDBClickHouseStagingProvider(ctx context.Context, env map[string]string)
 }
 
 func PeerDBClickHouseStagingBucketName(ctx context.Context, env map[string]string) (string, error) {
-	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME")
+	name, err := dynLookup(ctx, env, "PEERDB_CLICKHOUSE_STAGING_BUCKET_NAME")
+	if err != nil {
+		return "", err
+	}
+	if name != "" {
+		return name, nil
+	}
+	// Fall back to legacy S3-specific env var.
+	return dynLookup(ctx, env, "PEERDB_CLICKHOUSE_AWS_S3_BUCKET_NAME")
 }
 
 func PeerDBS3UuidPrefix(ctx context.Context, env map[string]string) (bool, error) {