// deno-lint-ignore-file no-explicit-any import { GlobalOptions, isSuperset } from "./types.ts"; import { SEP, log, yamlStringify } from "./deps.ts"; import { colors, Command, Flow, FlowModule, FlowService, JobService, Table, yamlParse, } from "./deps.ts"; import { requireLogin, resolveWorkspace, validatePath } from "./context.ts"; import { resolve, track_job } from "./script.ts"; import { defaultFlowDefinition } from "./bootstrap/flow_bootstrap.ts"; export interface FlowFile { summary: string; description?: string; value: any; schema?: any; } const alreadySynced: string[] = []; export async function pushFlow( workspace: string, remotePath: string, localPath: string, message?: string ): Promise { if (alreadySynced.includes(localPath)) { return; } alreadySynced.push(localPath); let flow: Flow | undefined = undefined; try { flow = await FlowService.getFlowByPath({ workspace: workspace, path: remotePath, }); } catch { // flow doesn't exist } if (!localPath.endsWith(SEP)) { localPath += SEP; } const localFlowRaw = await Deno.readTextFile(localPath + "flow.yaml"); const localFlow = yamlParse(localFlowRaw) as FlowFile; function replaceInlineScripts(modules: FlowModule[]) { modules.forEach((m) => { if (m.value.type == "rawscript") { const path = m.value.content.split(" ")[1]; m.value.content = Deno.readTextFileSync(localPath + path); const lock = m.value.lock; if ( lock && typeof lock == "string" && lock.trimStart().startsWith("!inline ") ) { const path = lock.split(" ")[1]; try { m.value.lock = Deno.readTextFileSync(localPath + path); } catch { log.error(`Lock file ${path} not found`); } } } else if (m.value.type == "forloopflow") { replaceInlineScripts(m.value.modules); } else if (m.value.type == "branchall") { m.value.branches.forEach((b) => replaceInlineScripts(b.modules)); } else if (m.value.type == "branchone") { m.value.branches.forEach((b) => replaceInlineScripts(b.modules)); replaceInlineScripts(m.value.default); } }); } replaceInlineScripts(localFlow.value.modules); if (flow) { if (isSuperset(localFlow, flow)) { log.info(colors.green(`Flow ${remotePath} is up to date`)); return; } log.info(colors.bold.yellow(`Updating flow ${remotePath}...`)); await FlowService.updateFlow({ workspace: workspace, path: remotePath.replaceAll("\\", "/"), requestBody: { path: remotePath.replaceAll("\\", "/"), deployment_message: message, ...localFlow, }, }); } else { log.info(colors.bold.yellow("Creating new flow...")); await FlowService.createFlow({ workspace: workspace, requestBody: { path: remotePath.replaceAll("\\", "/"), deployment_message: message, ...localFlow, }, }); } } type Options = GlobalOptions; async function push(opts: Options, filePath: string, remotePath: string) { if (!validatePath(remotePath)) { return; } const workspace = await resolveWorkspace(opts); await requireLogin(opts); await pushFlow(workspace.workspaceId, remotePath, filePath); log.info(colors.bold.underline.green("Flow pushed")); } async function list(opts: GlobalOptions & { showArchived?: boolean }) { const workspace = await resolveWorkspace(opts); await requireLogin(opts); let page = 0; const perPage = 10; const total: Flow[] = []; while (true) { const res = await FlowService.listFlows({ workspace: workspace.workspaceId, page, perPage, showArchived: opts.showArchived ?? false, }); page += 1; total.push(...res); if (res.length < perPage) { break; } } new Table() .header(["path", "summary", "edited by"]) .padding(2) .border(true) .body(total.map((x) => [x.path, x.summary, x.edited_by])) .render(); } async function run( opts: GlobalOptions & { data?: string; silent: boolean; }, path: string ) { const workspace = await resolveWorkspace(opts); await requireLogin(opts); const input = opts.data ? await resolve(opts.data) : {}; const id = await JobService.runFlowByPath({ workspace: workspace.workspaceId, path, requestBody: input, }); let i = 0; while (true) { const jobInfo = await JobService.getJob({ workspace: workspace.workspaceId, id, }); if (jobInfo.flow_status!.modules.length <= i) { break; } const module = jobInfo.flow_status!.modules[i]; if (module.job) { if (!opts.silent) { log.info("====== Job " + (i + 1) + " ======"); await track_job(workspace.workspaceId, module.job); } } else { if (!opts.silent) { log.info(module.type); } await new Promise((resolve, _) => setTimeout(() => resolve(undefined), 100) ); continue; } i++; } if (!opts.silent) { log.info(colors.green.underline.bold("Flow ran to completion")); log.info("\n"); } const jobInfo = await JobService.getCompletedJob({ workspace: workspace.workspaceId, id, }); log.info(jobInfo.result ?? {}); } export function bootstrap( opts: GlobalOptions & { summary: string; description: string }, flowPath: string ) { if (!validatePath(flowPath)) { return; } const flowDirFullPath = `${flowPath}.flow`; Deno.mkdirSync(flowDirFullPath, { recursive: false }); const newFlowDefinition = defaultFlowDefinition(); if (opts.summary !== undefined) { newFlowDefinition.summary = opts.summary; } if (opts.description !== undefined) { newFlowDefinition.description = opts.description; } const newFlowDefinitionYaml = yamlStringify( newFlowDefinition as Record ); const flowYamlPath = `${flowDirFullPath}/flow.yaml`; Deno.writeTextFile(flowYamlPath, newFlowDefinitionYaml, { createNew: true }); } const command = new Command() .description("flow related commands") .option("--show-archived", "Enable archived scripts in output") .action(list as any) .command( "push", "push a local flow spec. This overrides any remote versions." ) .arguments(" ") .action(push as any) .command("run", "run a flow by path.") .arguments("") .option( "-d --data ", "Inputs specified as a JSON string or a file using @ or stdin using @-." ) .option( "-s --silent", "Do not ouput anything other then the final output. Useful for scripting." ) .action(run as any) .command("bootstrap", "create a new empty flow") .arguments("") .option("--summary ", "script summary") .option("--description ", "script description") .action(bootstrap as any); export default command;