windmill/cli/windmill-utils-internal/src/deploy.ts

/**
 * Shared deploy logic for workspace fork/merge operations.
 *
 * Used by both the CLI (`wmill workspace merge`) and the frontend
 * (`CompareWorkspaces.svelte`). The caller provides a {@link DeployProvider}
 * that wraps the concrete API client (class-based for the frontend,
 * standalone functions for the CLI).
 */

// ---------------------------------------------------------------------------
// Types
// ---------------------------------------------------------------------------

export type DeployKind =
  | "script"
  | "flow"
  | "app"
  | "raw_app"
  | "resource"
  | "variable"
  | "resource_type"
  | "folder"
  | "schedule"
  | "datatable_migration"
  | "http_trigger"
  | "websocket_trigger"
  | "kafka_trigger"
  | "nats_trigger"
  | "postgres_trigger"
  | "mqtt_trigger"
  | "sqs_trigger"
  | "gcp_trigger"
  | "azure_trigger"
  | "email_trigger";

export const TRIGGER_KINDS = [
  "http_trigger",
  "websocket_trigger",
  "kafka_trigger",
  "nats_trigger",
  "postgres_trigger",
  "mqtt_trigger",
  "sqs_trigger",
  "gcp_trigger",
  "azure_trigger",
  "email_trigger",
] as const satisfies readonly DeployKind[];

export type TriggerDeployKind = (typeof TRIGGER_KINDS)[number];

export function isTriggerKind(kind: string): kind is TriggerDeployKind {
  return (TRIGGER_KINDS as readonly string[]).includes(kind);
}

/** True for any kind that deploys via a trigger or schedule API. */
export function isTriggerOrScheduleKind(
  kind: string
): kind is TriggerDeployKind | "schedule" {
  return kind === "schedule" || isTriggerKind(kind);
}

export interface DeployResult {
  success: boolean;
  error?: string;
}

/**
 * Abstraction over the generated API client.
 * Both the frontend (class-based services) and the CLI (standalone functions)
 * can satisfy this interface with a thin adapter.
 */
export interface DeployProvider {
  // Existence checks
  existsFlowByPath(p: { workspace: string; path: string }): Promise<boolean>;
  existsScriptByPath(p: { workspace: string; path: string }): Promise<boolean>;
  existsApp(p: { workspace: string; path: string }): Promise<boolean>;
  existsVariable(p: { workspace: string; path: string }): Promise<boolean>;
  existsResource(p: { workspace: string; path: string }): Promise<boolean>;
  existsResourceType(p: { workspace: string; path: string }): Promise<boolean>;
  existsFolder(p: { workspace: string; name: string }): Promise<boolean>;
  // Flows
  getFlowByPath(p: { workspace: string; path: string }): Promise<any>;
  createFlow(p: { workspace: string; requestBody: any }): Promise<any>;
  updateFlow(p: {
    workspace: string;
    path: string;
    requestBody: any;
  }): Promise<any>;
  archiveFlowByPath(p: {
    workspace: string;
    path: string;
    requestBody: any;
  }): Promise<any>;
  // Scripts
  getScriptByPath(p: { workspace: string; path: string }): Promise<any>;
  createScript(p: { workspace: string; requestBody: any }): Promise<any>;
  archiveScriptByPath(p: {
    workspace: string;
    path: string;
  }): Promise<any>;
  // Apps
  getAppByPath(p: { workspace: string; path: string }): Promise<any>;
  createApp(p: { workspace: string; requestBody: any }): Promise<any>;
  updateApp(p: {
    workspace: string;
    path: string;
    requestBody: any;
  }): Promise<any>;
  createAppRaw(p: { workspace: string; formData: any }): Promise<any>;
  updateAppRaw(p: {
    workspace: string;
    path: string;
    formData: any;
  }): Promise<any>;
  getPublicSecretOfLatestVersionOfApp(p: {
    workspace: string;
    path: string;
  }): Promise<any>;
  getRawAppData(p: {
    secretWithExtension: string;
    workspace: string;
  }): Promise<any>;
  deleteApp(p: { workspace: string; path: string }): Promise<any>;
  // Variables
  getVariable(p: {
    workspace: string;
    path: string;
    decryptSecret?: boolean;
  }): Promise<any>;
  createVariable(p: { workspace: string; requestBody: any }): Promise<any>;
  updateVariable(p: {
    workspace: string;
    path: string;
    requestBody: any;
    alreadyEncrypted?: boolean;
  }): Promise<any>;
  deleteVariable(p: { workspace: string; path: string }): Promise<any>;
  // Resources
  getResource(p: { workspace: string; path: string }): Promise<any>;
  createResource(p: { workspace: string; requestBody: any }): Promise<any>;
  updateResource(p: {
    workspace: string;
    path: string;
    requestBody: any;
  }): Promise<any>;
  deleteResource(p: { workspace: string; path: string }): Promise<any>;
  // Resource types
  getResourceType(p: { workspace: string; path: string }): Promise<any>;
  createResourceType(p: { workspace: string; requestBody: any }): Promise<any>;
  updateResourceType(p: {
    workspace: string;
    path: string;
    requestBody: any;
  }): Promise<any>;
  deleteResourceType(p: { workspace: string; path: string }): Promise<any>;
  // Folders
  getFolder(p: { workspace: string; name: string }): Promise<any>;
  createFolder(p: { workspace: string; requestBody: any }): Promise<any>;
  updateFolder(p: {
    workspace: string;
    name: string;
    requestBody: any;
  }): Promise<any>;
  deleteFolder(p: { workspace: string; name: string }): Promise<any>;
  // Datatable migrations. In the diff, an item's `path` is
  // `<datatable>/<timestamp>_<name>` (see `parseDatatableMigrationDeployPath`).
  listDatatableMigrations(p: { workspace: string }): Promise<any>;
  upsertDatatableMigration(p: {
    workspace: string;
    datatableName: string;
    requestBody: {
      timestamp: number;
      name: string;
      code_up: string;
      code_down?: string;
    };
  }): Promise<any>;
  deleteDatatableMigration(p: {
    workspace: string;
    datatableName: string;
    timestamp: number;
  }): Promise<any>;
  // Triggers — per-kind dispatch is delegated to the implementor so the shared
  // module doesn't need to know about each of the 9 trigger services.
  existsTriggerByKind(
    kind: TriggerDeployKind,
    p: { workspace: string; path: string }
  ): Promise<boolean>;
  /**
   * Get a trigger as it should be sent to create/update. Includes any kind-specific
   * transforms the implementor wants to apply before deploy (e.g. wiping GCP
   * subscription_id so the target workspace can create its own subscription).
   *
   * Operational-state handling (strip `mode`/`enabled` on update, pass through
   * on create) is applied by `deployItem` at the dispatch layer — implementors
   * should return the full row including `mode`/`enabled` and let the shared
   * dispatch decide.
   */
  getTriggerForDeploy(
    kind: TriggerDeployKind,
    p: { workspace: string; path: string; onBehalfOf?: string }
  ): Promise<any>;
  createTriggerByKind(
    kind: TriggerDeployKind,
    p: { workspace: string; requestBody: any }
  ): Promise<any>;
  updateTriggerByKind(
    kind: TriggerDeployKind,
    p: { workspace: string; path: string; requestBody: any }
  ): Promise<any>;
  deleteTriggerByKind(
    kind: TriggerDeployKind,
    p: { workspace: string; path: string }
  ): Promise<any>;
  /** Stripped trigger row used for the diff drawer (config fields only). */
  getTriggerValue(
    kind: TriggerDeployKind,
    p: { workspace: string; path: string }
  ): Promise<unknown>;
  /** Returns the trigger's `permissioned_as` for `--preserve-on-behalf-of`. */
  getTriggerPermissionedAs(
    kind: TriggerDeployKind,
    p: { workspace: string; path: string }
  ): Promise<string | undefined>;
  // Schedules
  existsSchedule(p: { workspace: string; path: string }): Promise<boolean>;
  getSchedule(p: { workspace: string; path: string }): Promise<any>;
  createSchedule(p: { workspace: string; requestBody: any }): Promise<any>;
  updateSchedule(p: {
    workspace: string;
    path: string;
    requestBody: any;
  }): Promise<any>;
  deleteSchedule(p: { workspace: string; path: string }): Promise<any>;
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/** Folder diff paths carry the `f/` prefix; folder API endpoints expect just the name. */
export function folderName(path: string): string {
  return path.replace(/^f\//, "");
}

/**
 * Strip operational state (`mode`, `enabled`) from a trigger/schedule payload
 * when deploying as an update; pass through unchanged on create.
 *
 * On update the strip lets the backend preserve the target row's existing
 * state via `is_mode_unspecified()` (triggers) or via `EditSchedule` lacking
 * `enabled` (schedules). On create there's no target state to preserve, so
 * the source's flag is sent through — fork-only items land with whatever
 * state the fork creator chose, and missing flags fall back to the backend
 * default of enabled (`BaseTriggerData::mode()` / schedule insert default).
 */
export function stripOperationalStateOnUpdate<T extends Record<string, any>>(
  payload: T,
  alreadyExists: boolean
): T {
  if (!alreadyExists) return payload;
  const { mode: _mode, enabled: _enabled, ...rest } = payload;
  return rest as T;
}

function getSubModules(flowModule: any): any[][] {
  const type = flowModule?.value?.type;
  if (type === "forloopflow" || type === "whileloopflow") {
    return [flowModule.value.modules ?? []];
  } else if (type === "branchall") {
    return (flowModule.value.branches ?? []).map(
      (branch: any) => branch.modules ?? []
    );
  } else if (type === "branchone") {
    return [
      ...(flowModule.value.branches ?? []).map((b: any) => b.modules ?? []),
      flowModule.value.default ?? [],
    ];
  } else if (type === "aiagent") {
    if (flowModule.value.tools) {
      return [
        flowModule.value.tools
          .filter(
            (t: any) =>
              t.value?.type === "script" || t.value?.type === "flow"
          )
          .map((t: any) => ({
            id: t.id,
            value: t.value,
            summary: t.summary,
          })),
      ];
    }
  }
  return [];
}

function getAllSubmodules(flowModule: any): any[] {
  return getSubModules(flowModule)
    .map((modules) => modules.flatMap((m: any) => [m, ...getAllSubmodules(m)]))
    .flat();
}

/** Recursively collect all modules from a flow definition, including the failure module. */
export function getAllModules(
  flowModules: any[],
  failureModule?: any
): any[] {
  return [
    ...flowModules,
    ...flowModules.flatMap((x) => getAllSubmodules(x)),
    ...(failureModule ? [failureModule] : []),
  ];
}

function toError(e: unknown): string {
  const err = e as { body?: string; message?: string };
  return err.body || err.message || String(e);
}

// A datatable-migration diff item's path is `<datatable>/<timestamp>_<name>`
// (mirrors the backend, e.g. `mydt/20260101000001_create_users`).
export function parseDatatableMigrationDeployPath(path: string): {
  datatable: string;
  timestamp: number;
  name: string;
} {
  const slash = path.indexOf("/");
  const underscore = slash >= 0 ? path.indexOf("_", slash + 1) : -1;
  if (slash < 0 || underscore < 0) {
    throw new Error(`Invalid datatable migration path: ${path}`);
  }
  const datatable = path.slice(0, slash);
  const timestamp = Number(path.slice(slash + 1, underscore));
  const name = path.slice(underscore + 1);
  if (!datatable || !Number.isFinite(timestamp) || !name) {
    throw new Error(`Invalid datatable migration path: ${path}`);
  }
  return { datatable, timestamp, name };
}

// The backend rejects `upsertDatatableMigration` when the target data table
// hasn't opted in to migrations. Turn that opaque 400 into an explicit,
// deploy-context message (falls back to the original error otherwise).
function asMigrationsDisabledError(e: unknown, datatable: string): unknown {
  const msg = (e as { body?: string; message?: string })?.body ?? ''
  if (typeof msg === "string" && /migrations are not enabled/i.test(msg)) {
    return new Error(
      `Data table '${datatable}' has not opted in to migrations on the target workspace; enable migrations for it there before deploying its migrations.`
    );
  }
  return e;
}

// ---------------------------------------------------------------------------
// checkItemExists
// ---------------------------------------------------------------------------

export async function checkItemExists(
  provider: DeployProvider,
  kind: DeployKind,
  path: string,
  workspace: string
): Promise<boolean> {
  if (kind === "flow") {
    return provider.existsFlowByPath({ workspace, path });
  } else if (kind === "script") {
    return provider.existsScriptByPath({ workspace, path });
  } else if (kind === "app" || kind === "raw_app") {
    return provider.existsApp({ workspace, path });
  } else if (kind === "variable") {
    return provider.existsVariable({ workspace, path });
  } else if (kind === "resource") {
    return provider.existsResource({ workspace, path });
  } else if (kind === "resource_type") {
    return provider.existsResourceType({ workspace, path });
  } else if (kind === "folder") {
    return provider.existsFolder({ workspace, name: folderName(path) });
  } else if (kind === "schedule") {
    return provider.existsSchedule({ workspace, path });
  } else if (kind === "datatable_migration") {
    const { datatable, timestamp } = parseDatatableMigrationDeployPath(path);
    const migrations = await provider.listDatatableMigrations({ workspace });
    return (migrations as { datatable: string; timestamp: number }[]).some(
      (m) => m.datatable === datatable && m.timestamp === timestamp
    );
  } else if (isTriggerKind(kind)) {
    return provider.existsTriggerByKind(kind, { workspace, path });
  }
  throw new Error(`Unknown kind: ${kind}`);
}

/**
 * Fetch one part of a raw app's compiled bundle.
 *
 * A bundle with no styles may have no `css` blob stored at all, which older
 * backends serve as a 404. Absent means empty, not broken — letting that reject
 * makes such an app permanently un-deployable. Only `.css` and only a 404 are
 * forgiven: swallowing auth/network failures would silently deploy the app with
 * its styles stripped, and a missing `.js` is a genuinely broken bundle.
 */
export async function getRawAppBundlePart(
  provider: DeployProvider,
  secret: string,
  ext: "js" | "css",
  workspace: string
): Promise<any> {
  try {
    return await provider.getRawAppData({
      secretWithExtension: `${secret}.${ext}`,
      workspace,
    });
  } catch (e: any) {
    if (ext === "css" && e?.status === 404) return "";
    throw e;
  }
}

// ---------------------------------------------------------------------------
// deployItem
// ---------------------------------------------------------------------------

export async function deployItem(
  provider: DeployProvider,
  kind: DeployKind,
  path: string,
  workspaceFrom: string,
  workspaceTo: string,
  onBehalfOf?: string
): Promise<DeployResult> {
  const preserveOnBehalfOf = onBehalfOf !== undefined;

  try {
    const alreadyExists = await checkItemExists(
      provider,
      kind,
      path,
      workspaceTo
    );

    if (kind === "flow") {
      const flow = await provider.getFlowByPath({
        workspace: workspaceFrom,
        path,
      });
      // Clear inline script hashes so the target workspace resolves by path
      getAllModules(
        flow.value?.modules ?? [],
        flow.value?.failure_module
      ).forEach((x: any) => {
        if (x.value?.type === "script" && x.value.hash != undefined) {
          x.value.hash = undefined;
        }
      });
      if (alreadyExists) {
        await provider.updateFlow({
          workspace: workspaceTo,
          path,
          requestBody: {
            ...flow,
            // Sent even when unset: the backend keeps a restriction an omitted field leaves out.
            job_token_scopes: flow.job_token_scopes ?? null,
            preserve_on_behalf_of: preserveOnBehalfOf,
            on_behalf_of_email: onBehalfOf,
            // Usernames are per-workspace, so the source's principal names nobody in
            // the target (or names a different person). Clearing it lets the backend
            // derive the target's own principal from the email above.
            on_behalf_of: undefined,
          },
        });
      } else {
        await provider.createFlow({
          workspace: workspaceTo,
          requestBody: {
            ...flow,
            // Sent even when unset: the backend keeps a restriction an omitted field leaves out.
            job_token_scopes: flow.job_token_scopes ?? null,
            preserve_on_behalf_of: preserveOnBehalfOf,
            on_behalf_of_email: onBehalfOf,
            // Usernames are per-workspace, so the source's principal names nobody in
            // the target (or names a different person). Clearing it lets the backend
            // derive the target's own principal from the email above.
            on_behalf_of: undefined,
          },
        });
      }
    } else if (kind === "script") {
      const script = await provider.getScriptByPath({
        workspace: workspaceFrom,
        path,
      });
      let parentHash: string | undefined;
      if (alreadyExists) {
        const existing = await provider.getScriptByPath({
          workspace: workspaceTo,
          path,
        });
        parentHash = existing.hash;
      }
      await provider.createScript({
        workspace: workspaceTo,
        requestBody: {
          ...script,
          lock: script.lock,
          job_token_scopes: script.job_token_scopes ?? null,
          parent_hash: parentHash,
          preserve_on_behalf_of: preserveOnBehalfOf,
          on_behalf_of_email: onBehalfOf,
          // See the flow branch: a source-workspace principal is never valid here.
          on_behalf_of: undefined,
        },
      });
    } else if (kind === "app" || kind === "raw_app") {
      const rawApp = await provider.getAppByPath({
        workspace: workspaceFrom,
        path,
      });
      // See the flow branch: a source-workspace principal is never valid here, and the
      // policy carries the app's in `on_behalf_of`. Clearing it lets the backend derive
      // the target's own from the address. A group travels as its synthetic
      // `group-*@windmill.dev` address, which an admin-created account holding it would
      // win: known and accepted, see `users::permissioned_as_from_email` in the backend.
      const app = {
        ...rawApp,
        policy: {
          ...rawApp.policy,
          on_behalf_of: undefined,
          on_behalf_of_email: onBehalfOf,
        },
      };
      if (alreadyExists) {
        if (app.raw_app) {
          const secret = await provider.getPublicSecretOfLatestVersionOfApp({
            workspace: workspaceFrom,
            path: app.path,
          });
          const js = await getRawAppBundlePart(
            provider,
            secret,
            "js",
            workspaceFrom
          );
          const css = await getRawAppBundlePart(
            provider,
            secret,
            "css",
            workspaceFrom
          );
          await provider.updateAppRaw({
            workspace: workspaceTo,
            path,
            formData: {
              app: { ...app, preserve_on_behalf_of: preserveOnBehalfOf },
              css,
              js,
            },
          });
        } else {
          await provider.updateApp({
            workspace: workspaceTo,
            path,
            requestBody: {
              ...app,
              preserve_on_behalf_of: preserveOnBehalfOf,
            },
          });
        }
      } else {
        if (app.raw_app) {
          const secret = await provider.getPublicSecretOfLatestVersionOfApp({
            workspace: workspaceFrom,
            path: app.path,
          });
          const js = await getRawAppBundlePart(
            provider,
            secret,
            "js",
            workspaceFrom
          );
          const css = await getRawAppBundlePart(
            provider,
            secret,
            "css",
            workspaceFrom
          );
          await provider.createAppRaw({
            workspace: workspaceTo,
            formData: {
              app: { ...app, preserve_on_behalf_of: preserveOnBehalfOf },
              css,
              js,
            },
          });
        } else {
          await provider.createApp({
            workspace: workspaceTo,
            requestBody: {
              ...app,
              preserve_on_behalf_of: preserveOnBehalfOf,
            },
          });
        }
      }
    } else if (kind === "variable") {
      const variable = await provider.getVariable({
        workspace: workspaceFrom,
        path,
        decryptSecret: true,
      });
      if (alreadyExists) {
        await provider.updateVariable({
          workspace: workspaceTo,
          path,
          requestBody: {
            path,
            value: variable.value ?? "",
            is_secret: variable.is_secret,
            description: variable.description ?? "",
            labels: variable.labels,
            // `null` clears; absent would keep the target's stored date.
            value_valid_until: variable.value_valid_until ?? null,
          },
          alreadyEncrypted: false,
        });
      } else {
        await provider.createVariable({
          workspace: workspaceTo,
          requestBody: {
            path,
            value: variable.value ?? "",
            is_secret: variable.is_secret,
            description: variable.description ?? "",
            labels: variable.labels,
            value_valid_until: variable.value_valid_until ?? null,
          },
        });
      }
    } else if (kind === "resource") {
      const resource = await provider.getResource({
        workspace: workspaceFrom,
        path,
      });
      let value = resource.value ?? "";
      // An agent's value carries the principal it runs as, as a trigger's row does: the source's
      // names nobody in the target, so it is replaced by the chosen one or dropped, leaving the
      // backend to make it the deployer's.
      const agent = resource.resource_type === "ai_agent" && typeof value === "object";
      if (agent) {
        const { on_behalf_of: _, ...rest } = value;
        value = onBehalfOf ? { ...rest, on_behalf_of: onBehalfOf } : rest;
      }
      const agentPreserve = agent ? { preserve_on_behalf_of: preserveOnBehalfOf } : {};
      if (alreadyExists) {
        await provider.updateResource({
          workspace: workspaceTo,
          path,
          requestBody: {
            path,
            value,
            description: resource.description ?? "",
            labels: resource.labels,
            ...agentPreserve,
          },
        });
      } else {
        await provider.createResource({
          workspace: workspaceTo,
          requestBody: {
            path,
            value,
            resource_type: resource.resource_type,
            description: resource.description ?? "",
            labels: resource.labels,
            ...agentPreserve,
          },
        });
      }
    } else if (kind === "resource_type") {
      const rt = await provider.getResourceType({
        workspace: workspaceFrom,
        path,
      });
      if (alreadyExists) {
        await provider.updateResourceType({
          workspace: workspaceTo,
          path,
          requestBody: {
            schema: rt.schema,
            description: rt.description ?? "",
          },
        });
      } else {
        await provider.createResourceType({
          workspace: workspaceTo,
          requestBody: {
            name: rt.name,
            schema: rt.schema,
            description: rt.description ?? "",
          },
        });
      }
    } else if (kind === "folder") {
      const name = folderName(path);
      const folder = await provider.getFolder({
        workspace: workspaceFrom,
        name,
      });
      if (alreadyExists) {
        await provider.updateFolder({
          workspace: workspaceTo,
          name,
          requestBody: {
            owners: folder.owners,
            extra_perms: folder.extra_perms,
            summary: folder.summary ?? undefined,
            // A folder stores cleared labels as NULL and reads back without the field, so an
            // absent value has to be sent as [] or the target keeps labels the source removed.
            labels: folder.labels ?? [],
          },
        });
      } else {
        await provider.createFolder({
          workspace: workspaceTo,
          requestBody: {
            name,
            owners: folder.owners,
            extra_perms: folder.extra_perms,
            summary: folder.summary ?? undefined,
            labels: folder.labels,
          },
        });
      }
    } else if (kind === "schedule") {
      const schedule = await provider.getSchedule({
        workspace: workspaceFrom,
        path,
      });
      // Operational-state handling — same shape as triggers below:
      //   - Update: strip `enabled` so the target's existing state is preserved
      //     (`EditSchedule` lacks `enabled` server-side, so this is also
      //     enforced by the type, but stripping keeps the intent explicit).
      //   - Create: pass `enabled` through so a fork-only schedule lands with
      //     the state the fork creator chose.
      const baseBody = stripOperationalStateOnUpdate(schedule, alreadyExists);
      const requestBody = {
        ...baseBody,
        permissioned_as: onBehalfOf,
        preserve_permissioned_as: preserveOnBehalfOf,
      };
      if (alreadyExists) {
        await provider.updateSchedule({
          workspace: workspaceTo,
          path,
          requestBody,
        });
      } else {
        await provider.createSchedule({
          workspace: workspaceTo,
          requestBody,
        });
      }
    } else if (isTriggerKind(kind)) {
      const triggerBody = await provider.getTriggerForDeploy(kind, {
        workspace: workspaceFrom,
        path,
        onBehalfOf,
      });
      // Strip operational state on update; pass through on create. The
      // implementor's `getTriggerForDeploy` is responsible only for kind-specific
      // transforms (e.g. GCP subscription_id wipe) — the operational-state
      // strip lives here so it stays consistent with the schedule branch and
      // with the legacy `kind === 'trigger'` path in the frontend.
      const requestBody = stripOperationalStateOnUpdate(triggerBody, alreadyExists);
      if (alreadyExists) {
        await provider.updateTriggerByKind(kind, {
          workspace: workspaceTo,
          path,
          requestBody,
        });
      } else {
        await provider.createTriggerByKind(kind, {
          workspace: workspaceTo,
          requestBody,
        });
      }
    } else if (kind === "datatable_migration") {
      const { datatable, timestamp } = parseDatatableMigrationDeployPath(path);
      const migrations = await provider.listDatatableMigrations({
        workspace: workspaceFrom,
      });
      const migration = (
        migrations as {
          datatable: string;
          timestamp: number;
          name: string;
          code_up: string;
          code_down?: string;
        }[]
      ).find((m) => m.datatable === datatable && m.timestamp === timestamp);
      if (!migration) {
        throw new Error(
          `Datatable migration ${path} not found in ${workspaceFrom}`
        );
      }
      try {
        await provider.upsertDatatableMigration({
          workspace: workspaceTo,
          datatableName: datatable,
          requestBody: {
            timestamp: migration.timestamp,
            name: migration.name,
            code_up: migration.code_up,
            ...(migration.code_down != null
              ? { code_down: migration.code_down }
              : {}),
          },
        });
      } catch (e) {
        throw asMigrationsDisabledError(e, datatable);
      }
    } else {
      throw new Error(`Unknown kind: ${kind}`);
    }

    return { success: true };
  } catch (e: unknown) {
    return { success: false, error: toError(e) };
  }
}

// ---------------------------------------------------------------------------
// deleteItemInWorkspace
// ---------------------------------------------------------------------------

/**
 * Delete/archive an item in a workspace.
 * Scripts and flows are archived (reversible). Other types are deleted.
 */
export async function deleteItemInWorkspace(
  provider: DeployProvider,
  kind: DeployKind,
  path: string,
  workspace: string
): Promise<DeployResult> {
  try {
    if (kind === "script") {
      await provider.archiveScriptByPath({ workspace, path });
    } else if (kind === "flow") {
      await provider.archiveFlowByPath({
        workspace,
        path,
        requestBody: { archived: true },
      });
    } else if (kind === "app" || kind === "raw_app") {
      await provider.deleteApp({ workspace, path });
    } else if (kind === "variable") {
      await provider.deleteVariable({ workspace, path });
    } else if (kind === "resource") {
      await provider.deleteResource({ workspace, path });
    } else if (kind === "resource_type") {
      await provider.deleteResourceType({ workspace, path });
    } else if (kind === "folder") {
      await provider.deleteFolder({ workspace, name: folderName(path) });
    } else if (kind === "schedule") {
      await provider.deleteSchedule({ workspace, path });
    } else if (kind === "datatable_migration") {
      const { datatable, timestamp } = parseDatatableMigrationDeployPath(path);
      await provider.deleteDatatableMigration({
        workspace,
        datatableName: datatable,
        timestamp,
      });
    } else if (isTriggerKind(kind)) {
      await provider.deleteTriggerByKind(kind, { workspace, path });
    } else {
      throw new Error(`Deletion not supported for kind: ${kind}`);
    }
    return { success: true };
  } catch (e: unknown) {
    return { success: false, error: toError(e) };
  }
}

// ---------------------------------------------------------------------------
// getOnBehalfOf
// ---------------------------------------------------------------------------

/**
 * Get the value of an item for diff comparison.
 * Returns a normalized representation suitable for JSON comparison.
 */
export async function getItemValue(
  provider: DeployProvider,
  kind: DeployKind,
  path: string,
  workspace: string
): Promise<unknown> {
  try {
    if (kind === "flow") {
      const flow = await provider.getFlowByPath({ workspace, path });
      getAllModules(flow.value?.modules ?? [], flow.value?.failure_module).forEach(
        (x: any) => {
          if (x.value?.type === "script" && x.value.hash != undefined) {
            x.value.hash = undefined;
          }
        }
      );
      return {
        summary: flow.summary,
        description: flow.description,
        value: flow.value,
        job_token_scopes: flow.job_token_scopes ?? null,
      };
    } else if (kind === "script") {
      const script = await provider.getScriptByPath({ workspace, path });
      return {
        content: script.content,
        lock: script.lock,
        schema: script.schema,
        summary: script.summary,
        language: script.language,
        job_token_scopes: script.job_token_scopes ?? null,
      };
    } else if (kind === "app" || kind === "raw_app") {
      return await provider.getAppByPath({ workspace, path });
    } else if (kind === "variable") {
      const variable = await provider.getVariable({
        workspace,
        path,
        decryptSecret: true,
      });
      return {
        value: variable.value,
        is_secret: variable.is_secret,
        description: variable.description,
        value_valid_until: variable.value_valid_until ?? undefined,
      };
    } else if (kind === "resource") {
      const resource = await provider.getResource({ workspace, path });
      // Each workspace resolves its own agent identity, which a deploy never copies.
      if (resource.resource_type === "ai_agent" && resource.value) {
        const { on_behalf_of: _, ...value } = resource.value;
        return value;
      }
      return resource.value;
    } else if (kind === "resource_type") {
      const rt = await provider.getResourceType({ workspace, path });
      return rt.schema;
    } else if (kind === "folder") {
      const folder = await provider.getFolder({
        workspace,
        name: folderName(path),
      });
      return {
        name: folder.name,
        owners: folder.owners,
        extra_perms: folder.extra_perms,
        summary: folder.summary,
      };
    } else if (kind === "schedule") {
      // Mirror the runtime-fields-ignore set used server-side so the diff drawer
      // matches the merge UI's "did config change?" semantics.
      const schedule = await provider.getSchedule({ workspace, path });
      return stripTriggerOrScheduleRuntimeFields(schedule);
    } else if (isTriggerKind(kind)) {
      const trigger = await provider.getTriggerValue(kind, { workspace, path });
      return stripTriggerOrScheduleRuntimeFields(trigger);
    } else if (kind === "datatable_migration") {
      // Surface the migration SQL so the diff drawer shows the up/down bodies a
      // reviewer needs to inspect before deploying.
      const { datatable, timestamp } = parseDatatableMigrationDeployPath(path);
      const migrations = (await provider.listDatatableMigrations({
        workspace,
      })) as {
        datatable: string;
        timestamp: number;
        name: string;
        code_up: string;
        code_down?: string;
      }[];
      const migration = migrations.find(
        (m) => m.datatable === datatable && m.timestamp === timestamp
      );
      if (migration) {
        return {
          name: migration.name,
          code_up: migration.code_up,
          code_down: migration.code_down ?? null,
        };
      }
    }
  } catch {
    // Item may not exist
  }
  return {};
}

/**
 * Strip runtime fields from a trigger or schedule row so the diff drawer
 * shows only config differences. Mirrors the backend's
 * `TRIGGER_COMPARE_IGNORE` constant.
 */
function stripTriggerOrScheduleRuntimeFields(row: unknown): unknown {
  if (!row || typeof row !== "object") return row;
  const ignore = new Set([
    "workspace_id",
    "edited_by",
    "edited_at",
    "email",
    "error",
    "enabled",
    "mode",
    "server_id",
    "last_server_ping",
    "extra_perms",
    "permissioned_as",
    // Server-managed (kept in sync with backend `TRIGGER_COMPARE_IGNORE`).
    "subscription_id",
    "push_auth_config",
  ]);
  const out: Record<string, unknown> = {};
  for (const [k, v] of Object.entries(row as Record<string, unknown>)) {
    if (!ignore.has(k)) out[k] = v;
  }
  return out;
}

/**
 * Fetch the on_behalf_of value for a deployable item.
 * Returns an email for flows/scripts/apps, a principal (u/username, g/group) for
 * triggers/schedules and agents, or undefined if not applicable.
 */
export async function getOnBehalfOf(
  provider: DeployProvider,
  kind: DeployKind,
  path: string,
  workspace: string
): Promise<string | undefined> {
  try {
    if (kind === "flow") {
      const flow = await provider.getFlowByPath({ workspace, path });
      return flow.on_behalf_of_email;
    } else if (kind === "script") {
      const script = await provider.getScriptByPath({ workspace, path });
      return script.on_behalf_of_email;
    } else if (kind === "app" || kind === "raw_app") {
      const app = await provider.getAppByPath({ workspace, path });
      return app.policy?.on_behalf_of_email;
    } else if (kind === "resource") {
      // Only an agent runs as anyone, and like a trigger it records the principal.
      const resource = await provider.getResource({ workspace, path });
      return resource.resource_type === "ai_agent"
        ? resource.value?.on_behalf_of
        : undefined;
    } else if (kind === "schedule") {
      const schedule = await provider.getSchedule({ workspace, path });
      return schedule.permissioned_as;
    } else if (isTriggerKind(kind)) {
      return await provider.getTriggerPermissionedAs(kind, { workspace, path });
    }
  } catch {
    // Item may not exist
  }
  return undefined;
}