feat(device): run a directory of flows as a bundle

src/commands/device.ts

@@ -1,5 +1,6 @@
 import { defineCommand } from 'citty';
 import { existsSync, readFileSync, statSync, writeFileSync } from 'node:fs';
+import * as path from 'node:path';
 import { resolve as resolvePath } from 'node:path';
 
 import { apiFlags } from '../config/flags/api.flags.js';
@@ -19,6 +20,7 @@ import {
 import { platformOfAppFile, uploadLocalApp } from '../utils/app-upload.js';
 import { refreshAuth, resolveAuth } from '../utils/auth.js';
 import { openBrowser } from '../utils/browser.js';
+import { uploadFlowBundle } from '../utils/device-bundle.js';
 import {
   type DeviceArtifactFile,
   downloadArtifactFiles,
@@ -650,6 +652,10 @@ const screenshotSub = defineCommand({
   },
 });
 
+function isDirectory(p: string): boolean {
+  return existsSync(p) && statSync(p).isDirectory();
+}
+
 /** `--env KEY=VALUE`, given once or several times. */
 export function parseEnvFlags(value: unknown): Record<string, string> | undefined {
   if (value === undefined || value === false) return undefined;
@@ -680,19 +686,23 @@ async function waitForCommand(ctx: Ctx, id: string, commandId: string): Promise<
 }
 
 const runSub = defineCommand({
-  meta: { name: 'run', description: 'Run a Maestro flow on the device, whole, and wait for the result' },
+  meta: { name: 'run', description: 'Run a Maestro flow (or a directory of flows) on the device, and wait for the result' },
   args: {
     ...commonArgs,
     id: idArg,
     flow: {
       type: 'positional',
       required: false,
-      description: 'A flow file (.yaml)',
+      description: 'A flow file (.yaml), or a directory of flows (uploaded as a bundle)',
     },
     yaml: {
       type: 'string',
       description: 'The flow inline, instead of a file',
     },
+    entry: {
+      type: 'string',
+      description: 'With a directory: the one flow to run, as a path inside it (default: all of them)',
+    },
     env: {
       type: 'string',
       description: 'A variable for the flow, KEY=VALUE, used as ${KEY}. Repeatable. Sent sealed, never stored in plain text',
@@ -703,20 +713,21 @@ const runSub = defineCommand({
     await runDeviceCommand(json, async () => {
       let id = args.id as string | undefined;
       let flow = args.flow as string | undefined;
-      if (id && !flow && /\.ya?ml$/i.test(id)) [id, flow] = [undefined, id];
+      if (id && !flow && (/\.ya?ml$/i.test(id) || isDirectory(id))) [id, flow] = [undefined, id];
       const inline = recoverFlagValue('--yaml', args.yaml as string | undefined, rawArgs);
       if (Boolean(flow) === Boolean(inline)) {
-        throw new CliError('Give a flow file or --yaml, one of the two.');
+        throw new CliError('Give a flow file, a directory of flows, or --yaml: one of the three.');
       }
 
-      let yaml = inline;
-      if (flow) {
-        if (existsSync(flow) && statSync(flow).isDirectory()) {
-          throw new CliError(
-            'Running a directory of flows is not supported yet. Run one flow file at a time.',
-          );
-        }
+      const dir = flow && isDirectory(flow) ? flow : undefined;
+      const entry = args.entry as string | undefined;
+      if (entry && !dir) throw new CliError('--entry picks a flow inside a directory; give a directory.');
+      if (entry && (path.isAbsolute(entry) || entry.split(/[\\/]/).includes('..'))) {
+        throw new CliError('--entry is a path inside the directory, e.g. flows/login.yaml.');
+      }
 
+      let yaml = inline;
+      if (flow && !dir) {
         try {
           yaml = readFileSync(flow, 'utf8');
         } catch (error) {
@@ -728,11 +739,23 @@ const runSub = defineCommand({
       const env = parseEnvFlags(repeated.length > 0 ? repeated : args.env);
       const ctx = await context(args);
       id = await resolveSessionId(ctx.apiUrl, ctx.auth, id);
-      ctx.progress(ui.running(`Running ${flow ?? 'the flow'} on ${colors.highlight(id)}…`));
-      let response = await DeviceGateway.run(ctx.apiUrl, ctx.auth, id, {
-        yaml: yaml!,
-        ...(env && { env }),
-      });
+
+      let body: Parameters<typeof DeviceGateway.run>[3];
+      if (dir) {
+        ctx.progress(ui.running(`Uploading the flows in ${colors.highlight(dir)}…`));
+        const bundle = await uploadFlowBundle({ apiUrl: ctx.apiUrl, auth: ctx.auth, dir, log: ctx.progress });
+        ctx.progress(ui.success(`Uploaded ${bundle.files} files`));
+        body = {
+          flowUploadId: bundle.id,
+          ...(entry && { flow: entry.split(path.sep).join('/') }),
+          ...(env && { env }),
+        };
+      } else {
+        body = { yaml: yaml!, ...(env && { env }) };
+      }
+
+      ctx.progress(ui.running(`Running ${entry ?? flow ?? 'the flow'} on ${colors.highlight(id)}…`));
+      let response = await DeviceGateway.run(ctx.apiUrl, ctx.auth, id, body);
       if (response.status === 'running') response = await waitForCommand(ctx, id, response.command_id);
 
       const result = response.result ?? { success: false };

src/gateways/device-gateway.ts

@@ -1,6 +1,8 @@
 import type { AuthContext } from '../types/domain/auth.types.js';
 import type {
   DeviceArtifactsPage,
+  DeviceBundleUploadUrls,
+  DeviceFlowBundle,
   DeviceCatalogue,
   DeviceCommandResult,
   DeviceInstalledApp,
@@ -120,6 +122,14 @@ export const DeviceGateway = {
     return request(baseUrl, auth, 'GET', `${sessionPath(id)}/artifacts?${query}`);
   },
 
+  bundleUploadUrls(
+    baseUrl: string,
+    auth: AuthContext,
+    bytes: number,
+  ): Promise<DeviceBundleUploadUrls> {
+    return request(baseUrl, auth, 'POST', '/device/bundles/upload-url', { bytes });
+  },
+
   command(
     baseUrl: string,
     auth: AuthContext,
@@ -181,11 +191,28 @@ export const DeviceGateway = {
     return request(baseUrl, auth, 'GET', '/device/models');
   },
 
+  registerBundle(
+    baseUrl: string,
+    auth: AuthContext,
+    body: {
+      backblaze: boolean;
+      bytes: number;
+      enc?: unknown;
+      id: string;
+      sha: string;
+      supabase: boolean;
+    },
+  ): Promise<{ bundle: DeviceFlowBundle }> {
+    return request(baseUrl, auth, 'POST', '/device/bundles', body);
+  },
+
   run(
     baseUrl: string,
     auth: AuthContext,
     id: string,
-    body: { env?: Record<string, string>; yaml: string },
+    body:
+      | { env?: Record<string, string>; flow?: string; flowUploadId: string }
+      | { env?: Record<string, string>; yaml: string },
   ): Promise<DeviceRunResponse> {
     return request(baseUrl, auth, 'POST', `${sessionPath(id)}/run`, body);
   },

src/types/domain/device-session.types.ts

@@ -152,3 +152,19 @@ export interface DeviceArtifactsPage {
   recording: DeviceArtifactLink | null;
   session: DeviceSession;
 }
+
+/** POST /device/bundles/upload-url: where to PUT the zip (both, when both are given). */
+export interface DeviceBundleUploadUrls {
+  id: string;
+  mirror: { path: string; signedUrl: string } | null;
+  upload: { path: string; signedUrl: string } | null;
+}
+
+/** A registered flow bundle. `run` takes its id as `flowUploadId`. */
+export interface DeviceFlowBundle {
+  bytes: number;
+  created_at: string;
+  encrypted: boolean;
+  id: string;
+  sha: string;
+}

src/utils/device-bundle.ts

@@ -0,0 +1,129 @@
+import { createHash } from 'node:crypto';
+import { readdirSync } from 'node:fs';
+import * as path from 'node:path';
+
+import { DeviceGateway } from '../gateways/device-gateway.js';
+import { compressFilesFromRelativePath } from '../methods.js';
+import type { AuthContext } from '../types/domain/auth.types.js';
+import type { DeviceFlowBundle } from '../types/domain/device-session.types.js';
+import { detectCiContext } from './ci.js';
+import { CliError, getCliVersion } from './cli.js';
+import { fetchCompatibilityData } from './compatibility.js';
+import {
+  checkKekForEncryption,
+  encryptFlowBuffer,
+  resolveEncryption,
+  resolveKekPublicKey,
+} from './envelope.js';
+
+/** The API's limit on a bundle (dcd plan 2.5b). */
+export const MAX_BUNDLE_BYTES = 50 * 1024 * 1024;
+
+/** Never bundled: version control, dependencies, and other dot-folders. */
+function skipped(relativePath: string): boolean {
+  return relativePath
+    .split(path.sep)
+    .some((part) => part === 'node_modules' || (part.startsWith('.') && part !== '.'));
+}
+
+/** Every file under `dir`, relative to it, in a stable order. */
+export function bundleFiles(dir: string): string[] {
+  return readdirSync(dir, { recursive: true, withFileTypes: true })
+    .filter((entry) => entry.isFile())
+    .map((entry) => path.relative(dir, path.join(entry.parentPath, entry.name)))
+    .filter((file) => !skipped(file))
+    .sort();
+}
+
+/** PUT the zip to one signed URL. The URL is the credential. */
+async function put(url: string, body: Buffer): Promise<boolean> {
+  for (let attempt = 1; attempt <= 2; attempt++) {
+    try {
+      const res = await fetch(url, {
+        method: 'PUT',
+        headers: { 'content-type': 'application/zip' },
+        body: new Uint8Array(body),
+      });
+      await res.body?.cancel();
+      if (res.ok) return true;
+      if (res.status >= 400 && res.status < 500 && res.status !== 408 && res.status !== 429) {
+        return false;
+      }
+    } catch {
+      // A network error: try once more.
+    }
+  }
+
+  return false;
+}
+
+export interface UploadFlowBundleOptions {
+  apiUrl: string;
+  auth: AuthContext;
+  dir: string;
+  log: (message: string) => void;
+}
+
+/**
+ * Zip a directory of flows and upload it as a flow bundle: ask the API for
+ * upload URLs, PUT the zip to both copies, then register it with its
+ * SHA-256. Encrypted like `dcd cloud`'s flow zips when the org's default
+ * (or DCD_ENCRYPT) says so. Entries are relative to `dir`, so a run's
+ * `flow` is a path inside it, e.g. `login.yaml`.
+ */
+export async function uploadFlowBundle(
+  opts: UploadFlowBundleOptions,
+): Promise<DeviceFlowBundle & { files: number }> {
+  const { apiUrl, auth, dir, log } = opts;
+  const files = bundleFiles(dir);
+  if (!files.some((f) => /\.ya?ml$/i.test(f))) {
+    throw new CliError(`No flow files (.yaml) in ${dir}.`);
+  }
+
+  const zip = await compressFilesFromRelativePath(dir, files, '');
+
+  let serverDefault: boolean | undefined;
+  try {
+    const ci = detectCiContext();
+    const compatibility = await fetchCompatibilityData(apiUrl, auth, {
+      cliVersion: getCliVersion(),
+      ciProvider: ci.provider,
+      ciWrapperVersion: ci.wrapperVersion,
+    });
+    serverDefault = compatibility.encryption?.defaultOn;
+  } catch {
+    serverDefault = undefined;
+  }
+
+  const encryption = checkKekForEncryption(resolveEncryption({ serverDefault }), apiUrl);
+  if (encryption.notice) log(encryption.notice);
+  const kek = encryption.enabled ? resolveKekPublicKey(apiUrl) : null;
+  const sealed = kek ? encryptFlowBuffer(zip, kek) : null;
+  const body = sealed ? sealed.ciphertext : zip;
+
+  if (body.length > MAX_BUNDLE_BYTES) {
+    throw new CliError(
+      `${dir} zips to ${(body.length / 1024 / 1024).toFixed(1)} MB; a bundle can be at most ${MAX_BUNDLE_BYTES / 1024 / 1024} MB.`,
+    );
+  }
+
+  const sha = createHash('sha256').update(body).digest('hex');
+  const urls = await DeviceGateway.bundleUploadUrls(apiUrl, auth, body.length);
+  const [supabase, backblaze] = await Promise.all([
+    urls.upload ? put(urls.upload.signedUrl, body) : Promise.resolve(false),
+    urls.mirror ? put(urls.mirror.signedUrl, body) : Promise.resolve(false),
+  ]);
+  if (!supabase && !backblaze) {
+    throw new Error('The flow bundle could not be uploaded. Please retry.');
+  }
+
+  const { bundle } = await DeviceGateway.registerBundle(apiUrl, auth, {
+    id: urls.id,
+    sha,
+    bytes: body.length,
+    supabase,
+    backblaze,
+    ...(sealed && { enc: sealed.enc }),
+  });
+  return { ...bundle, files: files.length };
+}