windmill/cli/src/types.ts

import { colors } from "@cliffy/ansi/colors";
import * as Diff from "diff";
import * as log from "./core/log.ts";
import * as path from "node:path";
import { sep as SEP } from "node:path";
import { stringify as yamlStringify } from "yaml";
import { yamlParseContent } from "./utils/yaml.ts";
import { isDbtDescriptorPath } from "./utils/resource_folders.ts";
import { pushApp } from "./commands/app/app.ts";
import { pushFolder } from "./commands/folder/folder.ts";
import { pushFlow } from "./commands/flow/flow.ts";
import { pushResource } from "./commands/resource/resource.ts";
import { pushResourceType } from "./commands/resource-type/resource-type.ts";
import { pushVariable } from "./commands/variable/variable.ts";
import { yamlOptions } from "./commands/sync/sync.ts";
import { showDiffs } from "./core/conf.ts";
import { deepEqual, isFileResource, isFilesetResource, isWorkspaceDependencies, readTextFileSync } from "./utils/utils.ts";
import { pushSchedule } from "./commands/schedule/schedule.ts";
import { pushWorkspaceUser } from "./commands/user/user.ts";
import { pushGroup } from "./commands/user/user.ts";
import { pushWorkspaceDependencies } from "./commands/dependencies/dependencies.ts";
import {
  pushWorkspaceSettings,
  pushWorkspaceKey,
  PushWorkspaceKeyOptions,
} from "./core/settings.ts";
import { pushTrigger, pushNativeTrigger } from "./commands/trigger/trigger.ts";
import { pushRawApp } from "./commands/app/raw_apps.ts";
import type { PermissionedAsContext } from "./core/permissioned_as.ts";
import {
  isFlowPath,
  isAppPath,
  isRawAppPath,
  extractResourceName,
  buildFolderPath,
  isScriptModulePath,
} from "./utils/resource_folders.ts";

export interface DifferenceCreate {
  type: "CREATE";
  path: (string | number)[];
  value: any;
}

export interface DifferenceRemove {
  type: "REMOVE";
  path: (string | number)[];
  oldValue: any;
}

export interface DifferenceChange {
  type: "CHANGE";
  path: (string | number)[];
  value: any;
  oldValue: any;
}

export type Difference = DifferenceCreate | DifferenceRemove | DifferenceChange;

export const TRIGGER_TYPES = [
  "http",
  "websocket",
  "kafka",
  "nats",
  "postgres",
  "mqtt",
  "amqp",
  "sqs",
  "gcp",
  "azure",
  "email",
] as const;

export const NATIVE_TRIGGER_SERVICES = ["nextcloud"] as const;
export type NativeTriggerService = (typeof NATIVE_TRIGGER_SERVICES)[number];

export type GlobalOptions = {
  baseUrl: string | undefined;
  workspace: string | undefined;
  token: string | undefined;
  configDir: string | undefined;
};

export function isSuperset(
  subset: Record<string, any>,
  superset: Record<string, any>
): boolean {
  return Object.keys(subset).every((key) => {
    const eq = deepEqual(subset[key], superset[key]);
    if (!eq && showDiffs) {
      const sub = subset[key];
      const supers = superset[key];
      if (!supers) {
        log.info(`Key ${key} not found in remote`);
      } else {
        log.info(`Found diff for ${key}:`);
        showDiff(
          yamlStringify(sub, yamlOptions),
          yamlStringify(supers, yamlOptions)
        );
      }
    }
    return eq;
  });
}

export function showDiff(local: string, remote: string) {
  let finalString = "";
  if (local?.length > 20000 || remote?.length > 20000) {
    log.info("Diff too large to display");
    return;
  }

  for (const part of Diff.diffLines(local ?? "", remote ?? "")) {
    if (part.removed) {
      // print red if removed without newline
      finalString += `\x1b[31m${part.value}\x1b[0m`;
    } else if (part.added) {
      // print green if added
      finalString += `\x1b[32m${part.value}\x1b[0m`;
    } else {
      let lines = part.value.split("\n");

      if (lines.length > 4) {
        lines = lines.slice(0, 2);
        lines.push("...");
        lines = lines.concat(part.value.split("\n").slice(-2));
      }
      // print white if unchanged
      finalString += `\x1b[37m${lines.join("\n")}\x1b[0m`;
    }
  }
  log.info(finalString);
}

export function showConflict(path: string, local: string, remote: string) {
  log.info(colors.yellow(`- ${path}`));
  let isEncryptionKey = false;
  try {
    isEncryptionKey = getTypeStrFromPath(path) === "encryption_key";
  } catch {
    // ignore
  }
  if (isEncryptionKey) {
    showDiff(redactEncryptionKey(local), redactEncryptionKey(remote));
  } else {
    showDiff(local, remote);
  }
  log.info("\x1b[31mlocal\x1b[31m - \x1b[32mremote\x1b[32m");
  log.info("\n");
}

// Reveal only the first 5 chars of the key so a rotation is still visible in
// the diff (different prefixes), without leaking the whole secret to stdout.
// The remaining chars are replaced with `*`, preserving length so the diff
// keeps showing whether the key length changed.
export function redactEncryptionKey(content: string): string {
  if (!content) return content;
  // The encryption_key payload is JSON-encoded (a quoted string). Parse it so
  // we redact the key value itself, then re-serialize to JSON to preserve the
  // file's shape; fall back to raw redaction if parsing fails.
  try {
    const parsed = JSON.parse(content);
    if (typeof parsed === "string") {
      return JSON.stringify(redactString(parsed));
    }
  } catch {
    // not JSON — treat content as the raw key
  }
  return redactString(content);
}

function redactString(s: string): string {
  if (s.length <= 5) return s;
  return s.slice(0, 5) + "*".repeat(s.length - 5);
}

export interface PushObjOptions {
  /** Optional commit/update message */
  message?: string;
  /** The original local file path (used for branch-specific resource file resolution) */
  originalLocalPath?: string;
  /** Identity to attribute the push to, for the types that carry one */
  permissionedAsContext?: PermissionedAsContext;
  /** Whether the item is workspace-specific */
  wsSpecific?: boolean;
  /** encryption_key push: non-interactive flag and explicit re-encryption choice */
  keyPushOpts?: PushWorkspaceKeyOptions;
  /** TypeScript runtime a bare `.ts` denotes, for raw-app runnables */
  defaultTs?: "bun" | "deno";
}

/**
 * Pushes an object to the workspace server based on its type
 * @param workspace - The workspace ID to push to
 * @param p - The server path (base path for branch-specific items)
 * @param befObj - The previous object state (for updates)
 * @param newObj - The new object state to push
 * @param plainSecrets - Whether to store secrets in plain text
 * @param alreadySynced - Array to track already synced items
 * @param opts - Per-type extras; see PushObjOptions
 */
export async function pushObj(
  workspace: string,
  p: string,
  befObj: any,
  newObj: any,
  plainSecrets: boolean,
  alreadySynced: string[],
  opts: PushObjOptions = {},
) {
  const {
    message,
    originalLocalPath,
    permissionedAsContext,
    wsSpecific,
    keyPushOpts,
    defaultTs,
  } = opts;
  const typeEnding = getTypeStrFromPath(p);

  if (typeEnding === "app") {
    const appName = extractResourceName(p, "app");
    if (!appName) {
      throw new Error(`Could not extract app name from path: ${p}`);
    }
    await pushApp(workspace, appName, buildFolderPath(appName, "app"), message, permissionedAsContext);
  } else if (typeEnding === "raw_app") {
    const rawAppName = extractResourceName(p, "raw_app");
    if (!rawAppName) {
      throw new Error(`Could not extract raw app name from path: ${p}`);
    }
    await pushRawApp(workspace, rawAppName, buildFolderPath(rawAppName, "raw_app"), message, defaultTs);
  } else if (typeEnding === "folder") {
    await pushFolder(workspace, p, befObj, newObj);
  } else if (typeEnding === "variable") {
    await pushVariable(workspace, p, befObj, newObj, plainSecrets, wsSpecific);
  } else if (typeEnding === "flow") {
    const flowName = extractResourceName(p, "flow");
    if (!flowName) {
      throw new Error(`Could not extract flow name from path: ${p}`);
    }
    await pushFlow(workspace, flowName, buildFolderPath(flowName, "flow"), message, permissionedAsContext);
  } else if (typeEnding === "resource") {
    if (!alreadySynced.includes(p)) {
      alreadySynced.push(p);
      await pushResource(workspace, p, befObj, newObj, originalLocalPath || p, wsSpecific, true);
    }
  } else if (typeEnding === "resource-type") {
    await pushResourceType(workspace, p, befObj, newObj);
  } else if (typeEnding === "schedule") {
    await pushSchedule(workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "http_trigger") {
    await pushTrigger("http", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "websocket_trigger") {
    await pushTrigger("websocket", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "kafka_trigger") {
    await pushTrigger("kafka", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "nats_trigger") {
    await pushTrigger("nats", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "postgres_trigger") {
    await pushTrigger("postgres", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "mqtt_trigger") {
    await pushTrigger("mqtt", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "amqp_trigger") {
    await pushTrigger("amqp", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "sqs_trigger") {
    await pushTrigger("sqs", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "gcp_trigger") {
    await pushTrigger("gcp", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "azure_trigger") {
    await pushTrigger("azure", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "email_trigger") {
    await pushTrigger("email", workspace, p, befObj, newObj, permissionedAsContext);
  } else if (typeEnding === "native_trigger") {
    await pushNativeTrigger(workspace, p, befObj, newObj);
  } else if (typeEnding === "user") {
    await pushWorkspaceUser(workspace, p, befObj, newObj);
  } else if (typeEnding === "group") {
    await pushGroup(workspace, p, befObj, newObj);
  } else if (typeEnding === "workspace_dependencies") {
    await pushWorkspaceDependencies(workspace, p, befObj, newObj);
  } else if (typeEnding === "settings") {
    await pushWorkspaceSettings(workspace, p, befObj, newObj);
  } else if (typeEnding === "encryption_key") {
    await pushWorkspaceKey(workspace, p, befObj, newObj, keyPushOpts);
  } else {
    throw new Error(
      `The item ${p} has an unrecognized type ending ${typeEnding}`
    );
  }
}

export function parseFromPath(p: string, content: string): any {
  return isWorkspaceDependencies(p)
    ? content
    : p.endsWith(".yaml")
      ? yamlParseContent(p, content)
      : p.endsWith(".json")
        ? JSON.parse(content)
        : content;
}
export function parseFromFile(p: string): any {
  if (p.endsWith(".json")) {
    return JSON.parse(readTextFileSync(p));
  } else if (p.endsWith(".yaml") || p.endsWith(".yml")) {
    return yamlParseContent(p, readTextFileSync(p));
  } else {
    throw new Error("Could not read file " + p);
  }
}
/**
 * Parse a `migrations/datatable/<datatable>/<timestamp>_<name>.(up|down).sql`
 * path into its parts. Returns undefined for any other path.
 */
export function parseDatatableMigrationPath(p: string):
  | { datatable: string; timestamp: number; name: string; kind: "up" | "down" }
  | undefined {
  const parts = p.split("/");
  if (
    parts[0] !== "migrations" ||
    parts[1] !== "datatable" ||
    parts.length !== 4
  )
    return undefined;
  const m = parts[3].match(/^(\d+)_(.*)\.(up|down)\.sql$/);
  if (!m) return undefined;
  return {
    datatable: parts[2],
    timestamp: Number(m[1]),
    name: m[2],
    kind: m[3] as "up" | "down",
  };
}

export function isDatatableMigrationPath(p: string): boolean {
  return parseDatatableMigrationPath(p) !== undefined;
}

export function getTypeStrFromPath(
  p: string
):
  | "datatable_migration"
  | "script"
  | "variable"
  | "flow"
  | "resource"
  | "resource-type"
  | "folder"
  | "app"
  | "raw_app"
  | "schedule"
  | "http_trigger"
  | "websocket_trigger"
  | "kafka_trigger"
  | "nats_trigger"
  | "postgres_trigger"
  | "mqtt_trigger"
  | "amqp_trigger"
  | "sqs_trigger"
  | "gcp_trigger"
  | "azure_trigger"
  | "email_trigger"
  | "native_trigger"
  | "user"
  | "group"
  | "settings"
  | "encryption_key"
  | "workspace_dependencies" {
  if (isDatatableMigrationPath(p)) {
    return "datatable_migration";
  }
  if (isScriptModulePath(p)) {
    return "script";
  }
  if (isFlowPath(p)) {
    return "flow";
  }
  if (isAppPath(p)) {
    return "app";
  }
  if (isRawAppPath(p)) {
    return "raw_app";
  }
  if (p.startsWith("dependencies" + SEP)) {
    return "workspace_dependencies";
  }
  if (isFileResource(p) || isFilesetResource(p)) {
    return "resource";
  }
  const parsed = path.parse(p);
  if (
    parsed.ext == ".go" ||
    parsed.ext == ".ts" ||
    parsed.ext == ".sh" ||
    parsed.ext == ".py" ||
    parsed.ext == ".sql" ||
    parsed.ext == ".gql" ||
    parsed.ext == ".ps1" ||
    parsed.ext == ".js" ||
    parsed.ext == ".php" ||
    parsed.ext == ".rs" ||
    parsed.ext == ".cs" ||
    parsed.ext == ".nu" ||
    parsed.ext == ".java" ||
    parsed.ext == ".rb" ||
    parsed.ext == ".r" ||
    // for related places search: ADD_NEW_LANG
    (parsed.ext == ".yml" && parsed.name.split(".").pop() == "playbook") ||
    // A dbt descriptor is `<project>__dbt/wm_dbt.yaml`. Without this it reads
    // as one of the CLI's own `.yaml` metadata files and a pull writes the
    // script's metadata and lock but never its content.
    isDbtDescriptorPath(p)
  ) {
    return "script";
  }
  // Match folder.meta (base) or folder.<branch>.meta (branch-specific)
  if (parsed.name === "folder.meta" || /^folder\.[^.]+\.meta$/.test(parsed.name)) {
    return "folder";
  }
  // Match settings (base) or settings.<branch> (branch-specific)
  if (parsed.name === "settings" || /^settings\.[^.]+$/.test(parsed.name)) {
    return "settings";
  }
  // Match encryption_key (base) or encryption_key.<branch> (branch-specific)
  if (parsed.name === "encryption_key" || /^encryption_key\.[^.]+$/.test(parsed.name)) {
    return "encryption_key";
  }

  const typeEnding = parsed.name.split(".").at(-1);
  // Check for native trigger: {service}_native_trigger pattern
  if (typeEnding?.endsWith("_native_trigger")) {
    return "native_trigger";
  }
  if (
    typeEnding === "script" ||
    typeEnding === "variable" ||
    typeEnding === "resource" ||
    typeEnding === "resource-type" ||
    typeEnding === "app" ||
    typeEnding === "schedule" ||
    typeEnding === "http_trigger" ||
    typeEnding === "websocket_trigger" ||
    typeEnding === "kafka_trigger" ||
    typeEnding === "nats_trigger" ||
    typeEnding === "postgres_trigger" ||
    typeEnding === "mqtt_trigger" ||
    typeEnding === "amqp_trigger" ||
    typeEnding === "sqs_trigger" ||
    typeEnding === "gcp_trigger" ||
    typeEnding === "azure_trigger" ||
    typeEnding === "email_trigger" ||
    typeEnding === "user" ||
    typeEnding === "group" ||
    typeEnding === "settings" ||
    typeEnding === "encryption_key"
  ) {
    return typeEnding;
  } else {
    throw new Error("Could not infer type of path " + JSON.stringify(parsed));
  }
}

export function removeType(str: string, type: string) {
  // Normalize path for cross-platform compatibility and convert to forward slashes for API consistency
  const normalizedStr = path.normalize(str).replaceAll(SEP, "/");

  if (
    normalizedStr.endsWith("." + type + ".yaml") ||
    normalizedStr.endsWith("." + type + ".json")
  ) {
    return normalizedStr.slice(0, normalizedStr.length - type.length - 6);
  }
  // Accept clean paths without the type suffix (e.g. "f/folder/name" instead of "f/folder/name.schedule.yaml")
  if (normalizedStr.includes("." + type)) {
    log.debug(`Path '${str}' contains '.${type}' but doesn't end with '.${type}.(yaml|json)' — treating as clean path`);
  }
  return normalizedStr;
}

/**
 * Extracts native trigger info from a path like:
 * u/admin/script.flow.12345.nextcloud_native_trigger.json
 * Returns { scriptPath: "u/admin/script", isFlow: true, externalId: "12345", serviceName: "nextcloud" }
 */
export function extractNativeTriggerInfo(p: string): {
  scriptPath: string;
  isFlow: boolean;
  externalId: string;
  serviceName: string;
} | null {
  // Remove extension (.json or .yaml)
  const normalizedPath = path.normalize(p).replaceAll(SEP, "/");
  const withoutExt = normalizedPath.replace(/\.(json|yaml)$/, "");

  // Match pattern: {script_path}.{flow|script}.{external_id}.{service}_native_trigger
  const match = withoutExt.match(/^(.+)\.(flow|script)\.([^.]+)\.(\w+)_native_trigger$/);
  if (!match) {
    return null;
  }

  return {
    scriptPath: match[1],
    isFlow: match[2] === "flow",
    externalId: match[3],
    serviceName: match[4],
  };
}

export function removePathPrefix(str: string, prefix: string) {
  // Normalize paths for cross-platform compatibility and convert to forward slashes for API consistency
  const normalizedStr = path.normalize(str).replaceAll(SEP, "/");
  const normalizedPrefix = path.normalize(prefix).replaceAll(SEP, "/");

  // Handle exact match case
  if (normalizedStr === normalizedPrefix) {
    return "";
  }

  if (!normalizedStr.startsWith(normalizedPrefix + "/")) {
    throw new Error(str + " does not start with " + prefix);
  }
  return normalizedStr.slice(normalizedPrefix.length + 1);
}