From c35fc4a2fb870105fd2244730b53e8eb91c6745a Mon Sep 17 00:00:00 2001
From: Kai Jellinghaus <contact@kaij.tech>
Date: Wed, 23 Nov 2022 21:24:36 +0100
Subject: [PATCH] Improve Script Logic & Enable Flow run

---
 cli/flow.ts   | 49 +++++++++++++++++++++++++++++++++++++++++++++++--
 cli/script.ts | 41 +++++++++++++++++++++++++++++++++++++----
 2 files changed, 84 insertions(+), 6 deletions(-)

diff --git a/cli/flow.ts b/cli/flow.ts
index d5933bf5e0..e77a6f0e38 100644
--- a/cli/flow.ts
+++ b/cli/flow.ts
@@ -1,5 +1,8 @@
 import { Command } from "https://deno.land/x/cliffy@v0.25.4/command/command.ts";
-import { FlowService } from "https://deno.land/x/windmill@v1.41.0/mod.ts";
+import {
+  FlowService,
+  JobService,
+} from "https://deno.land/x/windmill@v1.41.0/mod.ts";
 import { GlobalOptions } from "./types.ts";
 import {
   Flow,
@@ -8,6 +11,8 @@ import {
 import { colors } from "https://deno.land/x/cliffy@v0.25.4/ansi/colors.ts";
 import { getContext } from "./context.ts";
 import { Table } from "https://deno.land/x/cliffy@v0.25.4/table/table.ts";
+import { track_job } from "./script.ts";
+import { resolve } from "https://deno.land/std@0.141.0/path/win32.ts";
 
 type Options = GlobalOptions;
 
@@ -100,6 +105,42 @@ async function list(opts: GlobalOptions & { showArchived?: boolean }) {
     )
     .render();
 }
+async function run(
+  opts: GlobalOptions & {
+    input: Record<string, any>;
+  },
+  path: string
+) {
+  const { workspace } = await getContext(opts);
+  const id = await JobService.runFlowByPath({
+    workspace,
+    path,
+    requestBody: opts.input,
+  });
+
+  let i = 0;
+  while (true) {
+    const jobInfo = await JobService.getJob({ workspace, id });
+    if (jobInfo.flow_status!.modules.length <= i) {
+      break;
+    }
+    const module = jobInfo.flow_status!.modules[i];
+
+    if (module.job) {
+      console.log("====== Job " + (i + 1) + " ======");
+      await track_job(workspace, module.job);
+    } else {
+      console.log(module.type);
+      await new Promise((resolve, _) =>
+        setTimeout(() => resolve(undefined), 100)
+      );
+      continue;
+    }
+    i++;
+  }
+
+  console.log(colors.green.underline.bold("Flow ran to completion"));
+}
 
 const command = new Command()
   .description("flow related commands")
@@ -110,6 +151,10 @@ const command = new Command()
     "push a local flow spec. This overrides any remote versions."
   )
   .arguments("<file_path:string> <remote_path:string>")
-  .action(push as any);
+  .action(push as any)
+  .command("run", "run a flow by path.")
+  .arguments("<path:string>")
+  .option("--input.* <input>", "Inputs to pass to the script")
+  .action(run as any);
 
 export default command;
diff --git a/cli/script.ts b/cli/script.ts
index cbca12beed..a11b1f9260 100644
--- a/cli/script.ts
+++ b/cli/script.ts
@@ -4,6 +4,7 @@ import { GlobalOptions } from "./types.ts";
 import { colors } from "https://deno.land/x/cliffy@v0.25.4/ansi/colors.ts";
 import { getContext } from "./context.ts";
 import {
+  Job,
   JobService,
   Script,
 } from "https://deno.land/x/windmill@v1.41.0/windmill-api/index.ts";
@@ -171,6 +172,7 @@ async function list(opts: GlobalOptions & { showArchived?: boolean }) {
     )
     .render();
 }
+
 async function run(
   opts: GlobalOptions & {
     input: Record<string, any>;
@@ -187,11 +189,22 @@ async function run(
   track_job(workspace, id);
 }
 
-async function track_job(workspace: string, id: string) {
-  console.log(colors.yellow("Waiting for Job to start..."));
+export async function track_job(workspace: string, id: string) {
+  try {
+    const result = await JobService.getCompletedJob({ workspace, id });
+
+    console.log(result.logs);
+    console.log(colors.bold.underline.green("Job Completed"));
+    return;
+  } catch {
+    /* ignore */
+  }
+
+  console.log(colors.yellow("Waiting for Job " + id + " to start..."));
 
   let logOffset = 0;
   let running = false;
+  let retry = 0;
   while (true) {
     let updates: {
       running?: boolean | undefined;
@@ -206,7 +219,12 @@ async function track_job(workspace: string, id: string) {
         running,
       });
     } catch {
-      break;
+      retry++;
+      if (retry > 3) {
+        console.log("failed to get job updated. skipping log streaming.");
+        break;
+      }
+      continue;
     }
 
     if (!running && updates.running === true) {
@@ -220,6 +238,7 @@ async function track_job(workspace: string, id: string) {
     }
 
     if (updates.completed === true) {
+      console.log("completed");
       running = false;
       break;
     }
@@ -231,8 +250,22 @@ async function track_job(workspace: string, id: string) {
       );
     }
   }
+  await new Promise((resolve, _) => setTimeout(() => resolve(undefined), 1000));
 
-  console.log(colors.bold.underline.green("Job Completed"));
+  try {
+    const final_job = await JobService.getCompletedJob({ workspace, id });
+    if ((final_job.logs?.length ?? -1) > logOffset) {
+      console.log(final_job.logs!.substring(logOffset));
+    }
+
+    if (final_job.success) {
+      console.log(colors.bold.underline.green("Job Completed"));
+    } else {
+      console.log(colors.bold.underline.red("Job Completed"));
+    }
+  } catch {
+    console.log("Job appears to have completed, but no data can be retrieved");
+  }
 }
 
 async function show(opts: GlobalOptions, path: string) {