Addressed PR comments, further isolated legacy behavior

jonesphillip · jonesphillip · commit 344beeeac7ee · 2025-09-16T10:32:55.000-07:00
diff --git a/packages/wrangler/src/__tests__/pipelines.test.ts b/packages/wrangler/src/__tests__/pipelines.test.ts
@@ -381,7 +381,13 @@ describe("wrangler pipelines", () => {
 			await expect(
 				runWrangler(`pipelines create my_pipeline --sql "${sql}"`)
 			).rejects.toThrowErrorMatchingInlineSnapshot(
-				`[Error: An authentication error has occurred creating a pipeline with this version of Wrangler. Please try: npm install wrangler@4.36.0]`
+				`
+				[Error: Your account does not have access to the new Pipelines API. To use the legacy Pipelines API, please run:
+
+				npx wrangler@4.36.0 pipelines create my_pipeline
+
+				This will use an older version of Wrangler that supports the legacy API.]
+			`
 			);
 
 			expect(validateRequest.count).toBe(1);
@@ -985,7 +991,7 @@ describe("wrangler pipelines", () => {
 		it("should create stream with default settings", async () => {
 			setIsTTY(true);
 			mockConfirm({
-				text: "No schema file provided. Create stream without a schema (unstructured JSON)?",
+				text: "No schema file provided. Do you want to create stream without a schema (unstructured JSON)?",
 				result: true,
 			});
 
diff --git a/packages/wrangler/src/pipelines/cli/create.ts b/packages/wrangler/src/pipelines/cli/create.ts
@@ -23,10 +23,12 @@ export const pipelinesCreateCommand = createCommand({
 		sql: {
 			describe: "Inline SQL query for the pipeline",
 			type: "string",
+			conflicts: "sql-file",
 		},
 		"sql-file": {
 			describe: "Path to file containing SQL query for the pipeline",
 			type: "string",
+			conflicts: "sql",
 		},
 	},
 	positionalArgs: ["pipeline"],
@@ -35,13 +37,6 @@ export const pipelinesCreateCommand = createCommand({
 		await requireAuth(config);
 		const pipelineName = args.pipeline;
 
-		if (!args.sql && !args.sqlFile) {
-			throw new UserError("Either --sql or --sql-file must be provided");
-		}
-		if (args.sql && args.sqlFile) {
-			throw new UserError("Cannot specify both --sql and --sql-file");
-		}
-
 		let sql: string;
 		if (args.sql) {
 			sql = args.sql;
@@ -54,7 +49,6 @@ export const pipelinesCreateCommand = createCommand({
 				);
 			}
 		} else {
-			// This should never happen due to earlier validation
 			throw new UserError("Either --sql or --sql-file must be provided");
 		}
 
@@ -112,7 +106,7 @@ export const pipelinesCreateCommand = createCommand({
 			if (error instanceof APIError && error.code === 10000) {
 				// Show error when no access to v1 Pipelines API
 				throw new UserError(
-					`An authentication error has occurred creating a pipeline with this version of Wrangler. Please try: npm install wrangler@4.36.0`
+					`Your account does not have access to the new Pipelines API. To use the legacy Pipelines API, please run:\n\nnpx wrangler@4.36.0 pipelines create ${pipelineName}\n\nThis will use an older version of Wrangler that supports the legacy API.`
 				);
 			}
 			throw error;
@@ -131,7 +125,7 @@ export const pipelinesCreateCommand = createCommand({
 
 			if (streamTable) {
 				const stream = await getStream(config, streamTable.id);
-				displayUsageExamples(stream);
+				displayUsageExamples(stream, config);
 			} else {
 				logger.log(
 					`\nRun 'wrangler pipelines get ${pipeline.id}' to view full details.`
diff --git a/packages/wrangler/src/pipelines/cli/delete.ts b/packages/wrangler/src/pipelines/cli/delete.ts
@@ -35,7 +35,8 @@ export const pipelinesDeleteCommand = createCommand({
 
 			if (!args.force) {
 				const confirmedDelete = await confirm(
-					`Are you sure you want to delete the pipeline '${pipeline.name}' (${pipelineId})?`
+					`Are you sure you want to delete the pipeline '${pipeline.name}' (${pipelineId})?`,
+					{ fallbackValue: false }
 				);
 				if (!confirmedDelete) {
 					logger.log("Delete cancelled.");
diff --git a/packages/wrangler/src/pipelines/cli/get.ts b/packages/wrangler/src/pipelines/cli/get.ts
@@ -53,7 +53,7 @@ export const pipelinesGetCommand = createCommand({
 		}
 
 		if (args.format === "json") {
-			logger.log(JSON.stringify(pipeline, null, 2));
+			logger.json(pipeline);
 			return;
 		}
 
diff --git a/packages/wrangler/src/pipelines/cli/legacy-helpers.ts b/packages/wrangler/src/pipelines/cli/legacy-helpers.ts
@@ -5,11 +5,11 @@ import { APIError } from "../../parse";
 import {
 	authorizeR2Bucket,
 	BYTES_PER_MB,
-	formatPipelinePretty,
 	getAccountR2Endpoint,
 } from "../index";
 import {
 	deletePipeline,
+	formatPipelinePretty,
 	getPipeline,
 	listPipelines,
 	updatePipeline,
diff --git a/packages/wrangler/src/pipelines/cli/list.ts b/packages/wrangler/src/pipelines/cli/list.ts
@@ -46,7 +46,7 @@ export const pipelinesListCommand = createCommand({
 						legacyPipelines: legacyPipelines,
 					}
 				: newPipelines || [];
-			logger.log(JSON.stringify(result, null, 2));
+			logger.json(result);
 			return;
 		}
 
diff --git a/packages/wrangler/src/pipelines/cli/setup.ts b/packages/wrangler/src/pipelines/cli/setup.ts
@@ -48,7 +48,7 @@ export const pipelinesSetupCommand = createCommand({
 	},
 	args: {
 		name: {
-			describe: "Pipeline name (optional - will prompt if not provided)",
+			describe: "Pipeline name",
 			type: "string",
 		},
 	},
@@ -792,7 +792,7 @@ async function createPipelineIfNeeded(
 		logger.log("\n✨ Setup complete!");
 
 		if (created.stream) {
-			displayUsageExamples(created.stream);
+			displayUsageExamples(created.stream, config);
 		}
 	} catch (error) {
 		logger.error(
diff --git a/packages/wrangler/src/pipelines/cli/sinks/create.ts b/packages/wrangler/src/pipelines/cli/sinks/create.ts
@@ -1,5 +1,5 @@
 import { createCommand } from "../../../core/create-command";
-import { UserError } from "../../../errors";
+import { CommandLineArgsError, UserError } from "../../../errors";
 import { logger } from "../../../logger";
 import { bucketFormatMessage, isValidR2BucketName } from "../../../r2/helpers";
 import { requireAuth } from "../../../user";
@@ -67,17 +67,14 @@ export const pipelinesSinksCreateCommand = createCommand({
 			type: "string",
 		},
 		"roll-size": {
-			describe: "Roll file when size reaches (e.g., 100MB, 1GB)",
-			type: "string",
-			default:
-				SINK_DEFAULTS.rolling_policy.file_size_bytes === 0
-					? undefined
-					: `${SINK_DEFAULTS.rolling_policy.file_size_bytes}`,
+			describe: "Roll file size in MB",
+			type: "number",
+			default: SINK_DEFAULTS.rolling_policy.file_size_bytes / (1024 * 1024),
 		},
 		"roll-interval": {
-			describe: "Roll file when time reaches (e.g., 5m, 1h)",
-			type: "string",
-			default: `${SINK_DEFAULTS.rolling_policy.interval_seconds}s`,
+			describe: "Roll file interval in seconds",
+			type: "number",
+			default: SINK_DEFAULTS.rolling_policy.interval_seconds,
 		},
 		"access-key-id": {
 			describe:
@@ -105,32 +102,37 @@ export const pipelinesSinksCreateCommand = createCommand({
 			type: "string",
 		},
 	},
-	async handler(args, { config }) {
-		const accountId = await requireAuth(config);
-		const sinkName = args.sink;
+	validateArgs: (args) => {
 		const sinkType = parseSinkType(args.type);
 
 		if (!isValidR2BucketName(args.bucket)) {
-			throw new UserError(
+			throw new CommandLineArgsError(
 				`The bucket name "${args.bucket}" is invalid. ${bucketFormatMessage}`
 			);
 		}
 
 		if (sinkType === "r2_data_catalog") {
 			if (!args.namespace) {
-				throw new UserError(
+				throw new CommandLineArgsError(
 					"--namespace is required for r2-data-catalog sinks"
 				);
 			}
 			if (!args.table) {
-				throw new UserError("--table is required for r2-data-catalog sinks");
+				throw new CommandLineArgsError(
+					"--table is required for r2-data-catalog sinks"
+				);
 			}
 			if (!args.catalogToken) {
-				throw new UserError(
+				throw new CommandLineArgsError(
 					"--catalog-token is required for r2-data-catalog sinks"
 				);
 			}
 		}
+	},
+	async handler(args, { config }) {
+		const accountId = await requireAuth(config);
+		const sinkName = args.sink;
+		const sinkType = parseSinkType(args.type);
 
 		const sinkConfig: CreateSinkRequest = {
 			name: sinkName,
@@ -182,23 +184,10 @@ export const pipelinesSinksCreateCommand = createCommand({
 				SINK_DEFAULTS.rolling_policy.interval_seconds;
 
 			if (args.rollSize) {
-				// Parse file size (e.g., "100MB" -> bytes)
-				const sizeMatch = args.rollSize.match(/^(\d+)(MB|GB)?$/i);
-				if (sizeMatch) {
-					const size = parseInt(sizeMatch[1]);
-					const unit = sizeMatch[2]?.toUpperCase() || "MB";
-					file_size_bytes =
-						unit === "GB" ? size * 1024 * 1024 * 1024 : size * 1024 * 1024;
-				}
+				file_size_bytes = args.rollSize * 1024 * 1024;
 			}
 			if (args.rollInterval) {
-				// Parse interval (e.g., "300s" or "5m" -> seconds)
-				const intervalMatch = args.rollInterval.match(/^(\d+)([sm])?$/i);
-				if (intervalMatch) {
-					const interval = parseInt(intervalMatch[1]);
-					const unit = intervalMatch[2]?.toLowerCase() || "s";
-					interval_seconds = unit === "m" ? interval * 60 : interval;
-				}
+				interval_seconds = args.rollInterval;
 			}
 
 			sinkConfig.config.rolling_policy = {
diff --git a/packages/wrangler/src/pipelines/cli/sinks/delete.ts b/packages/wrangler/src/pipelines/cli/sinks/delete.ts
@@ -31,7 +31,8 @@ export const pipelinesSinksDeleteCommand = createCommand({
 
 		if (!args.force) {
 			const confirmedDelete = await confirm(
-				`Are you sure you want to delete the sink '${sink.name}' (${args.sink})?`
+				`Are you sure you want to delete the sink '${sink.name}' (${args.sink})?`,
+				{ fallbackValue: false }
 			);
 			if (!confirmedDelete) {
 				logger.log("Delete cancelled.");
diff --git a/packages/wrangler/src/pipelines/cli/sinks/get.ts b/packages/wrangler/src/pipelines/cli/sinks/get.ts
@@ -31,7 +31,7 @@ export const pipelinesSinksGetCommand = createCommand({
 		const rawSink = await getSink(config, args.sink);
 
 		if (args.json) {
-			logger.log(JSON.stringify(rawSink, null, 2));
+			logger.json(rawSink);
 			return;
 		}
 
diff --git a/packages/wrangler/src/pipelines/cli/sinks/list.ts b/packages/wrangler/src/pipelines/cli/sinks/list.ts
@@ -40,7 +40,7 @@ export const pipelinesSinksListCommand = createCommand({
 		});
 
 		if (args.json) {
-			logger.log(JSON.stringify(sinks, null, 2));
+			logger.log(sinks);
 			return;
 		}
 
diff --git a/packages/wrangler/src/pipelines/cli/streams/create.ts b/packages/wrangler/src/pipelines/cli/streams/create.ts
@@ -36,45 +36,50 @@ export const pipelinesStreamsCreateCommand = createCommand({
 			type: "boolean",
 			default: true,
 		},
-		"cors-origins": {
-			describe: "Comma-separated CORS origins",
+		"cors-origin": {
+			describe: "CORS origin",
 			type: "string",
+			array: true,
 		},
 	},
 	async handler(args, { config }) {
 		await requireAuth(config);
 		const streamName = args.stream;
 
-		let corsOrigins: string[] | undefined;
-		if (args.corsOrigins) {
-			corsOrigins =
-				args.corsOrigins === "*"
-					? ["*"]
-					: args.corsOrigins.split(",").map((origin) => origin.trim());
-		}
+		const corsOrigins = args.corsOrigin;
 
 		let schema: { fields: SchemaField[] } | undefined;
 		if (args.schemaFile) {
+			let schemaContent: string;
+			let parsedSchema: { fields: SchemaField[] };
+
 			try {
-				const schemaContent = readFileSync(args.schemaFile, "utf-8");
-				const parsedSchema = parseJSON(schemaContent, args.schemaFile) as {
+				schemaContent = readFileSync(args.schemaFile, "utf-8");
+			} catch (error) {
+				throw new UserError(
+					`Failed to read schema file '${args.schemaFile}': ${error instanceof Error ? error.message : String(error)}`
+				);
+			}
+
+			try {
+				parsedSchema = parseJSON(schemaContent, args.schemaFile) as {
 					fields: SchemaField[];
 				};
-
-				if (!parsedSchema || !Array.isArray(parsedSchema.fields)) {
-					throw new UserError("Schema file must contain a 'fields' array");
-				}
-
-				schema = parsedSchema;
 			} catch (error) {
 				throw new UserError(
-					`Failed to read or parse schema file '${args.schemaFile}': ${error instanceof Error ? error.message : String(error)}`
+					`Failed to parse schema file '${args.schemaFile}': ${error instanceof Error ? error.message : String(error)}`
 				);
 			}
+
+			if (!parsedSchema || !Array.isArray(parsedSchema.fields)) {
+				throw new UserError("Schema file must contain a 'fields' array");
+			}
+
+			schema = parsedSchema;
 		} else {
 			// No schema file provided - confirm with user
 			const confirmNoSchema = await confirm(
-				"No schema file provided. Create stream without a schema (unstructured JSON)?",
+				"No schema file provided. Do you want to create stream without a schema (unstructured JSON)?",
 				{ defaultValue: false }
 			);
 
@@ -92,7 +97,8 @@ export const pipelinesStreamsCreateCommand = createCommand({
 				enabled: args.httpEnabled,
 				authentication: args.httpAuth,
 				...(args.httpEnabled &&
-					corsOrigins && { cors: { origins: corsOrigins } }),
+					corsOrigins &&
+					corsOrigins.length > 0 && { cors: { origins: corsOrigins } }),
 			},
 			worker_binding: {
 				enabled: true,
diff --git a/packages/wrangler/src/pipelines/cli/streams/delete.ts b/packages/wrangler/src/pipelines/cli/streams/delete.ts
@@ -31,7 +31,8 @@ export const pipelinesStreamsDeleteCommand = createCommand({
 
 		if (!args.force) {
 			const confirmedDelete = await confirm(
-				`Are you sure you want to delete the stream '${stream.name}' (${args.stream})?`
+				`Are you sure you want to delete the stream '${stream.name}' (${args.stream})?`,
+				{ fallbackValue: false }
 			);
 			if (!confirmedDelete) {
 				logger.log("Delete cancelled.");
diff --git a/packages/wrangler/src/pipelines/cli/streams/get.ts b/packages/wrangler/src/pipelines/cli/streams/get.ts
@@ -29,7 +29,7 @@ export const pipelinesStreamsGetCommand = createCommand({
 		const stream = await getStream(config, args.stream);
 
 		if (args.json) {
-			logger.log(JSON.stringify(stream, null, 2));
+			logger.json(stream);
 			return;
 		}
 
diff --git a/packages/wrangler/src/pipelines/cli/streams/list.ts b/packages/wrangler/src/pipelines/cli/streams/list.ts
@@ -40,7 +40,7 @@ export const pipelinesStreamsListCommand = createCommand({
 		});
 
 		if (args.json) {
-			logger.log(JSON.stringify(streams, null, 2));
+			logger.json(streams);
 			return;
 		}
 
diff --git a/packages/wrangler/src/pipelines/cli/streams/utils.ts b/packages/wrangler/src/pipelines/cli/streams/utils.ts
diff --git a/packages/wrangler/src/pipelines/client.ts b/packages/wrangler/src/pipelines/client.ts
diff --git a/packages/wrangler/src/pipelines/index.ts b/packages/wrangler/src/pipelines/index.ts
diff --git a/packages/wrangler/src/pipelines/legacy-client.ts b/packages/wrangler/src/pipelines/legacy-client.ts

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ export const pipelinesGetCommand = createCommand({`
`53`	`53`	`}`
`54`	`54`
`55`	`55`	`if (args.format === "json") {`
`56`		`- logger.log(JSON.stringify(pipeline, null, 2));`
	`56`	`+ logger.json(pipeline);`
`57`	`57`	`return;`
`58`	`58`	`}`
`59`	`59`
Original file line number	Diff line number	Diff line change
`@@ -46,7 +46,7 @@ export const pipelinesListCommand = createCommand({`
`46`	`46`	`legacyPipelines: legacyPipelines,`
`47`	`47`	`}`
`48`	`48`	`: newPipelines \|\| [];`
`49`		`- logger.log(JSON.stringify(result, null, 2));`
	`49`	`+ logger.json(result);`
`50`	`50`	`return;`
`51`	`51`	`}`
`52`	`52`
Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,7 @@ export const pipelinesSinksGetCommand = createCommand({`
`31`	`31`	`const rawSink = await getSink(config, args.sink);`
`32`	`32`
`33`	`33`	`if (args.json) {`
`34`		`- logger.log(JSON.stringify(rawSink, null, 2));`
	`34`	`+ logger.json(rawSink);`
`35`	`35`	`return;`
`36`	`36`	`}`
`37`	`37`