// deno-lint-ignore-file no-explicit-any import { GlobalOptions } from "./types.ts"; import { requireLogin, resolveWorkspace, validatePath } from "./context.ts"; import { colors, Command, JobService, log, NewScript, readAll, Script, ScriptService, Table, writeAllSync, yamlParse, yamlStringify, } from "./deps.ts"; import { deepEqual } from "./utils.ts"; import { defaultScriptMetadata, scriptBootstrapCode, } from "./bootstrap/script_bootstrap.ts"; import { instantiate as instantiateWasm, parse_bash, parse_bigquery, parse_deno, parse_go, parse_graphql, parse_mssql, parse_mysql, parse_powershell, parse_python, parse_snowflake, parse_sql, } from "./wasm/windmill_parser_wasm.generated.js"; import { Workspace } from "./workspace.ts"; import { SchemaProperty } from "./bootstrap/common.ts"; export interface ScriptFile { parent_hash?: string; summary: string; description: string; schema?: any; is_template?: boolean; lock?: Array; kind?: "script" | "failure" | "trigger" | "command" | "approval"; } type PushOptions = GlobalOptions; async function push(opts: PushOptions, filePath: string) { const workspace = await resolveWorkspace(opts); if (!validatePath(filePath)) { return; } const fstat = await Deno.stat(filePath); if (!fstat.isFile) { throw new Error("file path must refer to a file."); } if (filePath.endsWith(".script.json") || filePath.endsWith(".script.yaml")) { throw Error( "Cannot push a script metadata file, point to the script content file instead (.py, .ts, .go|.sh)" ); } await requireLogin(opts); await handleFile(filePath, workspace.workspaceId, []); log.info(colors.bold.underline.green(`Script ${filePath} pushed`)); } export async function handleScriptMetadata( path: string, workspace: string, alreadySynced: string[] ): Promise { if (path.endsWith(".script.json") || path.endsWith(".script.yaml")) { const contentPath = await findContentFile(path); return handleFile(contentPath, workspace, alreadySynced); } else { return false; } } async function parseMetadataFile( scriptPath: string ): Promise<{ isJson: boolean; payload: any } | undefined> { let metadataFilePath = scriptPath + ".script.json"; try { await Deno.stat(metadataFilePath); return { payload: JSON.parse(await Deno.readTextFile(metadataFilePath)), isJson: true, }; } catch { try { metadataFilePath = scriptPath + ".script.yaml"; await Deno.stat(metadataFilePath); return { payload: yamlParse(await Deno.readTextFile(metadataFilePath)), isJson: false, }; } catch { // no metadata file at all. Create it metadataFilePath = scriptPath + ".script.yaml"; const scriptInitialMetadata = defaultScriptMetadata(); const scriptInitialMetadataYaml = yamlStringify( scriptInitialMetadata as Record ); Deno.writeTextFile(metadataFilePath, scriptInitialMetadataYaml, { createNew: true, }); return { payload: scriptInitialMetadata, isJson: false, }; } } } export async function handleFile( path: string, workspace: string, alreadySynced: string[], message?: string ): Promise { if ( !path.includes(".inline_script.") && (path.endsWith(".ts") || path.endsWith(".py") || path.endsWith(".go") || path.endsWith(".sh") || path.endsWith(".sql") || path.endsWith(".gql") || path.endsWith(".ps1")) ) { if (alreadySynced.includes(path)) { return true; } log.debug(`Processing local script ${path}`); alreadySynced.push(path); const remotePath = path .substring(0, path.indexOf(".")) .replaceAll("\\", "/"); const typed = (await parseMetadataFile(remotePath))?.payload; const language = inferContentTypeFromFilePath(path); let remote = undefined; try { remote = await ScriptService.getScriptByPath({ workspace, path: remotePath.replaceAll("\\", "/"), }); log.debug(`Script ${remotePath} exists on remote`); } catch { log.debug(`Script ${remotePath} does not exist on remote`); } const content = await Deno.readTextFile(path); if (remote) { if (content === remote.content) { if ( typed == undefined || (typed.description === remote.description && typed.summary === remote.summary && (typed.is_template ?? false) === (remote.is_template ?? false) && typed.kind == remote.kind && !remote.archived && (remote?.lock ?? "").trim() == (Array.isArray(typed.lock) ? typed.lock.join("\n") : typed?.lock ?? "" ).trim() && deepEqual(typed.schema, remote.schema) && typed.tag == remote.tag && (typed.ws_error_handler_muted ?? false) == remote.ws_error_handler_muted && typed.dedicated_worker == remote.dedicated_worker && typed.cache_ttl == remote.cache_ttl && typed.concurrency_time_window_s == remote.concurrency_time_window_s && typed.concurrent_limit == remote.concurrent_limit) ) { log.info(colors.green(`Script ${remotePath} is up to date`)); return true; } } log.info( colors.yellow.bold(`Creating script with a parent ${remotePath}`) ); await ScriptService.createScript({ workspace, requestBody: { content, description: typed?.description ?? "", language: language as NewScript.language, path: remotePath.replaceAll("\\", "/"), summary: typed?.summary ?? "", is_template: typed?.is_template, kind: typed?.kind, lock: typed?.lock, parent_hash: remote.hash, schema: typed?.schema, tag: typed?.tag, ws_error_handler_muted: typed?.ws_error_handler_muted, dedicated_worker: typed?.dedicated_worker, cache_ttl: typed?.cache_ttl, concurrency_time_window_s: typed?.concurrency_time_window_s, concurrent_limit: typed?.concurrent_limit, deployment_message: message, }, }); } else { log.info( colors.yellow.bold(`Creating script without parent ${remotePath}`) ); // no parent hash await ScriptService.createScript({ workspace: workspace, requestBody: { content, description: typed?.description ?? "", language: language as NewScript.language, path: remotePath.replaceAll("\\", "/"), summary: typed?.summary ?? "", is_template: typed?.is_template, kind: typed?.kind, lock: typed?.lock, parent_hash: undefined, schema: typed?.schema, tag: typed?.tag, ws_error_handler_muted: typed?.ws_error_handler_muted, dedicated_worker: typed?.dedicated_worker, cache_ttl: typed?.cache_ttl, concurrency_time_window_s: typed?.concurrency_time_window_s, concurrent_limit: typed?.concurrent_limit, deployment_message: message, }, }); } return true; } return false; } export async function findContentFile(filePath: string) { const candidates = filePath.endsWith("script.json") ? [ filePath.replace(".script.json", ".fetch.ts"), filePath.replace(".script.json", ".bun.ts"), filePath.replace(".script.json", ".ts"), filePath.replace(".script.json", ".py"), filePath.replace(".script.json", ".go"), filePath.replace(".script.json", ".sh"), filePath.replace(".script.json", "pg.sql"), filePath.replace(".script.json", "my.sql"), filePath.replace(".script.json", "bq.sql"), filePath.replace(".script.json", "sf.sql"), filePath.replace(".script.json", ".gql"), filePath.replace(".script.json", ".ps1"), ] : [ filePath.replace(".script.yaml", ".fetch.ts"), filePath.replace(".script.yaml", ".bun.ts"), filePath.replace(".script.yaml", ".ts"), filePath.replace(".script.yaml", ".py"), filePath.replace(".script.yaml", ".go"), filePath.replace(".script.yaml", ".sh"), filePath.replace(".script.yaml", "pg.sql"), filePath.replace(".script.yaml", "bq.sql"), filePath.replace(".script.yaml", "sf.sql"), filePath.replace(".script.yaml", ".gql"), filePath.replace(".script.yaml", ".ps1"), ]; const validCandidates = ( await Promise.all( candidates.map((x) => { return Deno.stat(x) .catch(() => undefined) .then((x) => x?.isFile) .then((e) => { return { path: x, file: e }; }); }) ) ) .filter((x) => x.file) .map((x) => x.path); if (validCandidates.length > 1) { throw new Error( "No content path given and more than one candidate found: " + validCandidates.join(", ") ); } if (validCandidates.length < 1) { throw new Error("No content path given and no content file found."); } return validCandidates[0]; } type ScriptLanguage = | "python3" | "deno" | "bun" | "nativets" | "go" | "bash" | "powershell" | "postgresql" | "mysql" | "bigquery" | "snowflake" | "mssql" | "graphql"; export function filePathExtensionFromContentType( language: ScriptLanguage ): string { if (language === "python3") { return ".py"; } else if (language === "nativets") { return ".fetch.ts"; } else if (language === "bun") { return ".bun.ts"; } else if (language === "deno") { return ".ts"; } else if (language === "go") { return ".go"; } else if (language === "mysql") { return ".my.sql"; } else if (language === "bigquery") { return ".bq.sql"; } else if (language === "snowflake") { return ".sf.sql"; } else if (language === "mssql") { return ".ms.sql"; } else if (language === "postgresql") { return ".pg.sql"; } else if (language === "graphql") { return ".gql"; } else if (language === "bash") { return ".sh"; } else if (language === "powershell") { return ".ps1"; } else { throw new Error("Invalid language: " + language); } } export function inferContentTypeFromFilePath( contentPath: string ): ScriptLanguage { if (contentPath.endsWith(".py")) { return "python3"; } else if (contentPath.endsWith("fetch.ts")) { return "nativets"; } else if (contentPath.endsWith("bun.ts")) { return "bun"; } else if (contentPath.endsWith(".ts")) { return "deno"; } else if (contentPath.endsWith(".go")) { return "go"; } else if (contentPath.endsWith(".my.sql")) { return "mysql"; } else if (contentPath.endsWith(".bq.sql")) { return "bigquery"; } else if (contentPath.endsWith(".sf.sql")) { return "snowflake"; } else if (contentPath.endsWith(".ms.sql")) { return "mssql"; } else if (contentPath.endsWith(".pg.sql")) { return "postgresql"; } else if (contentPath.endsWith(".gql")) { return "graphql"; } else if (contentPath.endsWith(".sh")) { return "bash"; } else if (contentPath.endsWith(".ps1")) { return "powershell"; } else { throw new Error( "Invalid language: " + contentPath.substring(contentPath.lastIndexOf(".")) ); } } //////////////////////////////////////////////////////////////////////////////////////////// // below functions copied from Windmill's FE inferArgs function. TODO: refactor // //////////////////////////////////////////////////////////////////////////////////////////// export function inferSchema( language: ScriptLanguage, content: string, currentSchema: any ) { let inferedSchema: any; if (language === "python3") { inferedSchema = JSON.parse(parse_python(content)); } else if (language === "nativets") { inferedSchema = JSON.parse(parse_deno(content)); } else if (language === "bun") { inferedSchema = JSON.parse(parse_deno(content)); } else if (language === "deno") { inferedSchema = JSON.parse(parse_deno(content)); } else if (language === "go") { inferedSchema = JSON.parse(parse_go(content)); } else if (language === "mysql") { inferedSchema = JSON.parse(parse_mysql(content)); inferedSchema.args = [ { name: "database", typ: { resource: "mysql" } }, ...inferedSchema.args, ]; } else if (language === "bigquery") { inferedSchema = JSON.parse(parse_bigquery(content)); inferedSchema.args = [ { name: "database", typ: { resource: "bigquery" } }, ...inferedSchema.args, ]; } else if (language === "snowflake") { inferedSchema = JSON.parse(parse_snowflake(content)); inferedSchema.args = [ { name: "database", typ: { resource: "snowflake" } }, ...inferedSchema.args, ]; } else if (language === "mssql") { inferedSchema = JSON.parse(parse_mssql(content)); inferedSchema.args = [ { name: "database", typ: { resource: "ms_sql_server" } }, ...inferedSchema.args, ]; } else if (language === "postgresql") { inferedSchema = JSON.parse(parse_sql(content)); inferedSchema.args = [ { name: "database", typ: { resource: "postgresql" } }, ...inferedSchema.args, ]; } else if (language === "graphql") { inferedSchema = JSON.parse(parse_graphql(content)); inferedSchema.args = [ { name: "api", typ: { resource: "graphql" } }, ...inferedSchema.args, ]; } else if (language === "bash") { inferedSchema = JSON.parse(parse_bash(content)); } else if (language === "powershell") { inferedSchema = JSON.parse(parse_powershell(content)); } else { throw new Error("Invalid language: " + language); } if (inferedSchema.type == "Invalid") { throw new Error( `Script invalid, it cannot be parsed to infer schema. Error was: ${inferedSchema.error}` ); } currentSchema.required = []; const oldProperties = JSON.parse(JSON.stringify(currentSchema.properties)); currentSchema.properties = {}; for (const arg of inferedSchema.args) { if (!(arg.name in oldProperties)) { currentSchema.properties[arg.name] = { description: "", type: "" }; } else { currentSchema.properties[arg.name] = oldProperties[arg.name]; } currentSchema.properties[arg.name] = sortObject( currentSchema.properties[arg.name] ); argSigToJsonSchemaType(arg.typ, currentSchema.properties[arg.name]); currentSchema.properties[arg.name].default = arg.default; if (!arg.has_default && !currentSchema.required.includes(arg.name)) { currentSchema.required.push(arg.name); } } return currentSchema; } function sortObject(obj: any): any { return Object.keys(obj) .sort() .reduce( (acc, key) => ({ ...acc, [key]: obj[key], }), {} ); } function argSigToJsonSchemaType( typ: | string | { resource: string | null } | { list: | string | { str: any } | { object: { key: string; typ: any }[] } | null; } | { str: string[] | null } | { object: { key: string; typ: any }[] }, oldS: SchemaProperty ): void { const newS: SchemaProperty = { type: "" }; if (typ === "int") { newS.type = "integer"; } else if (typ === "float") { newS.type = "number"; } else if (typ === "bool") { newS.type = "boolean"; } else if (typ === "email") { newS.type = "string"; newS.format = "email"; } else if (typ === "sql") { newS.type = "string"; newS.format = "sql"; } else if (typ === "yaml") { newS.type = "string"; newS.format = "yaml"; } else if (typ === "bytes") { newS.type = "string"; newS.contentEncoding = "base64"; } else if (typ === "datetime") { newS.type = "string"; newS.format = "date-time"; } else if (typeof typ !== "string" && `object` in typ) { newS.type = "object"; if (typ.object) { const properties: Record = {}; for (const prop of typ.object) { properties[prop.key] = { type: undefined }; argSigToJsonSchemaType(prop.typ, properties[prop.key]); } newS.properties = properties; } } else if (typeof typ !== "string" && `str` in typ) { newS.type = "string"; if (typ.str) { newS.enum = typ.str; } } else if (typeof typ !== "string" && `resource` in typ) { newS.type = "object"; newS.format = `resource-${typ.resource}`; } else if (typeof typ !== "string" && `list` in typ) { newS.type = "array"; if (typ.list === "int" || typ.list === "float") { newS.items = { type: "number" }; } else if (typ.list === "bytes") { newS.items = { type: "string", contentEncoding: "base64" }; } else if (typ.list == "string") { newS.items = { type: "string" }; } else if (typ.list && typeof typ.list == "object" && "str" in typ.list) { newS.items = { type: "string", enum: typ.list.str }; } else { newS.items = { type: "object" }; } } else { newS.type = "object"; } if (oldS.type != newS.type) { for (const prop of Object.getOwnPropertyNames(newS)) { if (prop != "description") { delete oldS[prop]; } } } else if (oldS.format == "date-time" && newS.format != "date-time") { delete oldS.format; } else if (oldS.items?.type != newS.items?.type) { delete oldS.items; } Object.assign(oldS, newS); if (oldS.format?.startsWith("resource-") && newS.type != "object") { oldS.format = undefined; } } //////////////////////////////////////////////////////////////////////////////////////////// // end of refactoring TODO // //////////////////////////////////////////////////////////////////////////////////////////// async function list(opts: GlobalOptions & { showArchived?: boolean }) { const workspace = await resolveWorkspace(opts); await requireLogin(opts); let page = 0; const perPage = 10; const total: Script[] = []; while (true) { const res = await ScriptService.listScripts({ workspace: workspace.workspaceId, page, perPage, showArchived: opts.showArchived ?? false, }); page += 1; total.push(...res); if (res.length < perPage) { break; } } new Table() .header(["path", "summary", "language", "created by"]) .padding(2) .border(true) .body(total.map((x) => [x.path, x.summary, x.language, x.created_by])) .render(); } export async function resolve(input: string): Promise> { if (!input) { throw new Error("No data given"); } if (input == "@-") { input = new TextDecoder().decode(await readAll(Deno.stdin)); } if (input[0] == "@") { input = await Deno.readTextFile(input.substring(1)); } try { return JSON.parse(input); } catch (e) { console.error("Impossible to parse input as JSON", input); throw e; } } 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.runScriptByPath({ workspace: workspace.workspaceId, path, requestBody: input, }); if (!opts.silent) { await track_job(workspace.workspaceId, id); } while (true) { try { const result = ( await JobService.getCompletedJob({ workspace: workspace.workspaceId, id, }) ).result ?? {}; log.info(result); break; } catch { new Promise((resolve, _) => setTimeout(() => resolve(undefined), 100)); } } } export async function track_job(workspace: string, id: string) { try { const result = await JobService.getCompletedJob({ workspace, id }); log.info(result.logs); log.info("\n"); log.info(colors.bold.underline.green("Job Completed")); log.info("\n"); return; } catch { /* ignore */ } log.info(colors.yellow("Waiting for Job " + id + " to start...")); let logOffset = 0; let running = false; let retry = 0; while (true) { let updates: { running?: boolean | undefined; completed?: boolean | undefined; new_logs?: string | undefined; }; try { updates = await JobService.getJobUpdates({ workspace, id, logOffset, running, }); } catch { retry++; if (retry > 3) { log.info("failed to get job updated. skipping log streaming."); break; } continue; } if (!running && updates.running === true) { running = true; log.info(colors.green("Job running. Streaming logs...")); } if (updates.new_logs) { writeAllSync(Deno.stdout, new TextEncoder().encode(updates.new_logs)); logOffset += updates.new_logs.length; } if (updates.completed === true) { running = false; break; } if (running && updates.running === false) { running = false; log.info(colors.yellow("Job suspended. Waiting for it to continue...")); } } await new Promise((resolve, _) => setTimeout(() => resolve(undefined), 1000)); try { const final_job = await JobService.getCompletedJob({ workspace, id }); if ((final_job.logs?.length ?? -1) > logOffset) { log.info(final_job.logs!.substring(logOffset)); } log.info("\n"); if (final_job.success) { log.info(colors.bold.underline.green("Job Completed")); } else { log.info(colors.bold.underline.red("Job Completed")); } log.info("\n"); } catch { log.info("Job appears to have completed, but no data can be retrieved"); } } async function show(opts: GlobalOptions, path: string) { const workspace = await resolveWorkspace(opts); await requireLogin(opts); const s = await ScriptService.getScriptByPath({ workspace: workspace.workspaceId, path, }); log.info(colors.underline(s.path)); if (s.description) log.info(s.description); log.info(""); log.info(s.content); } async function bootstrap( opts: GlobalOptions & { summary: string; description: string }, scriptPath: string, language: ScriptLanguage ) { if (!validatePath(scriptPath)) { return; } const scriptInitialCode = scriptBootstrapCode[language]; if (scriptInitialCode === undefined) { throw new Error("Language unknown"); } const extension = filePathExtensionFromContentType(language); const scriptCodeFileFullPath = scriptPath + extension; const scriptMetadataFileFullPath = scriptPath + ".script.yaml"; try { await Deno.stat(scriptCodeFileFullPath); await Deno.stat(scriptMetadataFileFullPath); throw new Error("File already exists in repository"); } catch { // file does not exist, we can continue } const scriptMetadata = defaultScriptMetadata(); if (opts.summary !== undefined) { scriptMetadata.summary = opts.summary; } if (opts.description !== undefined) { scriptMetadata.description = opts.description; } const scriptInitialMetadataYaml = yamlStringify( scriptMetadata as Record ); Deno.writeTextFile(scriptCodeFileFullPath, scriptInitialCode, { createNew: true, }); Deno.writeTextFile(scriptMetadataFileFullPath, scriptInitialMetadataYaml, { createNew: true, }); } async function generateMetadata( opts: GlobalOptions & { lockOnly: boolean; schemaOnly: boolean }, scriptPath: string ) { if (!validatePath(scriptPath)) { return; } const workspace = await resolveWorkspace(opts); await requireLogin(opts); const language = inferContentTypeFromFilePath(scriptPath); // read script metadata file const remotePath = scriptPath .substring(0, scriptPath.indexOf(".")) .replaceAll("\\", "/"); const metadataWithType = await parseMetadataFile(remotePath); if (metadataWithType === undefined) { throw new Error("Script metadata file does not exist at this path"); } // read script content const scriptContent = await Deno.readTextFile(scriptPath); const metadataParsedContent = metadataWithType?.payload as Record< string, any >; if (!opts.lockOnly) { await updateScriptSchema(scriptContent, language, metadataParsedContent); } if (!opts.schemaOnly) { await updateScriptLock( workspace, scriptContent, language, remotePath, metadataParsedContent ); } let metaPath = remotePath + ".script.yaml"; let newMetadataContent = yamlStringify(metadataParsedContent); if (metadataWithType.isJson) { metaPath = remotePath + ".script.json"; newMetadataContent = JSON.stringify(metadataParsedContent); } Deno.writeTextFile(metaPath, newMetadataContent); } async function updateScriptSchema( scriptContent: string, language: ScriptLanguage, metadataContent: Record ): Promise { // infer schema from script content and update it inplace await instantiateWasm(); const newSchema = inferSchema( language, scriptContent, metadataContent.schema ); metadataContent.schema = newSchema; } async function updateScriptLock( workspace: Workspace, scriptContent: string, language: ScriptLanguage, remotePath: string, metadataContent: Record ): Promise { // generate the script lock running a dependency job in Windmill and update it inplace // TODO: update this once the client is released const rawResponse = await fetch( `${workspace.remote}api/w/${workspace.workspaceId}/jobs/run/dependencies`, { method: "POST", headers: { Cookie: `token=${workspace.token}`, "Content-Type": "application/json", }, body: JSON.stringify({ raw_scripts: [ { raw_code: scriptContent, language: language, script_path: remotePath, }, ], entrypoint: remotePath, }), } ); try { const response = await rawResponse.json(); const lock = response.lock; if (lock === undefined) { throw new Error( `Failed to generate lockfile. Full response was: ${JSON.stringify( response )}` ); } metadataContent.lock = lock; } catch { throw new Error( `Failed to generate lockfile. Status was: ${rawResponse.statusText}` ); } } const command = new Command() .description("script related commands") .option("--show-archived", "Enable archived scripts in output") .action(list as any) .command( "push", "push a local script spec. This overrides any remote versions. Use the script file (.ts, .js, .py, .sh)" ) .arguments("") .action(push as any) .command("show", "show a scripts content") .arguments("") .action(show as any) .command("run", "run a script by path") .arguments("") .option( "-d --data ", "Inputs specified as a JSON string or a file using @ or stdin using @-." ) .option( "-s --silent", "Do not output anything other then the final output. Useful for scripting." ) .action(run as any) .command("bootstrap", "create a new script") .arguments(" ") .option("--summary ", "script summary") .option("--description ", "script description") .action(bootstrap as any) .command( "generate-metadata", "re-generate the metadata file updating the lock and the script schema" ) .arguments("") .option("--lock-only", "re-generate only the lock") .option("--schema-only", "re-generate only script schema") .action(generateMetadata as any); export default command;