Merge pull request #60 from zenaton/scheduling_alfred

Scheduling with alfred and new syntax.

Author: MrYawe

CHANGELOG.md

@@ -15,6 +15,10 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 
 ### Changed
 
+- Changed scheduling requests that pass now through Alfred (GraphQL API).
+
+- Changed scheduling syntax from `x.repeat("* * * * *").schedule()` to `x.shedule("* * * * *")`.
+
 ### Deprecated
 
 ### Removed

package-lock.json

@@ -28,7 +28,7 @@
   "dependencies": {
     "@babel/runtime": "7.3.1",
     "axios": "0.18.1",
-    "cron-parser": "2.11.0",
+    "graphql-request": "^1.8.2",
     "moment-timezone": "0.5.23",
     "uuid": "^3.3.2"
   },

package.json

@@ -1,6 +1,7 @@
 const uuidv4 = require("uuid/v4");
 const workflowManager = require("./Workflows/WorkflowManager");
 const http = require("./Services/Http");
+const graphQL = require("./Services/GraphQL");
 const serializer = require("./Services/Serializer");
 const { version } = require("../infos");
 const { init, credentials } = require("../client");
@@ -10,6 +11,8 @@ const ZENATON_WORKER_URL = "http://localhost";
 const DEFAULT_WORKER_PORT = 4001;
 const WORKER_API_VERSION = "v_newton";
 
+const ZENATON_GATEWAY_URL = "https://gateway.zenaton.com/api";
+
 const APP_ENV = "app_env";
 const APP_ID = "app_id";
 const API_TOKEN = "api_token";
@@ -25,7 +28,6 @@ const ATTR_INITIAL_LIB_VERSION = "initial_library_version";
 const ATTR_CODE_PATH_VERSION = "code_path_version";
 const ATTR_MODE = "mode";
 const ATTR_MAX_PROCESSING_TIME = "maxProcessingTime";
-const ATTR_SCHEDULING_CRON = "scheduling_cron";
 
 const PROG = "Javascript";
 const INITIAL_LIB_VERSION = version;
@@ -113,6 +115,14 @@ module.exports = class Client {
     return `${host}${path}`;
   }
 
+  getGatewayUrl() {
+    const host = process.env.ZENATON_GATEWAY_URL
+      ? process.env.ZENATON_GATEWAY_URL
+      : ZENATON_GATEWAY_URL;
+
+    return host;
+  }
+
   /**
    * Start a task instance
    */
@@ -136,19 +146,24 @@ module.exports = class Client {
   }
 
   async startScheduledTask(task) {
-    const url = this.getWorkerUrlNew("scheduling/tasks");
-
-    // schedule task
-    const body = this.getBodyForTask(task);
-
-    const params = Object.assign(
-      {
-        [ATTR_SCHEDULING_CRON]: task.scheduling.cron,
+    const endpoint = this.getGatewayUrl();
+    const taskBody = this.getBodyForTask(task);
+    const mutation = graphQL.mutations.createTaskSchedule;
+    const variables = {
+      createTaskScheduleInput: {
+        intentId: taskBody[ATTR_INTENT_ID],
+        environmentName: credentials.appEnv,
+        cron: task.scheduling.cron,
+        taskName: taskBody[ATTR_NAME],
+        programmingLanguage: taskBody[ATTR_PROG].toUpperCase(),
+        properties: taskBody[ATTR_DATA],
+        codePathVersion: taskBody[ATTR_CODE_PATH_VERSION],
+        initialLibraryVersion: taskBody[ATTR_INITIAL_LIB_VERSION],
       },
-      this.getAppEnv(),
-    );
+    };
 
-    return http.post(url, body, { params });
+    const res = await graphQL.request(endpoint, mutation, variables);
+    return res.createTaskSchedule;
   }
 
   /**
@@ -174,19 +189,25 @@ module.exports = class Client {
   }
 
   async startScheduledWorkflow(flow) {
-    const url = this.getWorkerUrlNew("scheduling/instances");
-
-    // schedule workflow
-    const body = this.getBodyForWorkflow(flow);
-
-    const params = Object.assign(
-      {
-        [ATTR_SCHEDULING_CRON]: flow.scheduling.cron,
+    const endpoint = this.getGatewayUrl();
+    const workflowBody = this.getBodyForWorkflow(flow);
+    const mutation = graphQL.mutations.createWorkflowSchedule;
+    const variables = {
+      createWorkflowScheduleInput: {
+        intentId: workflowBody[ATTR_INTENT_ID],
+        environmentName: credentials.appEnv,
+        cron: flow.scheduling.cron,
+        workflowName: workflowBody[ATTR_NAME],
+        canonicalName: workflowBody[ATTR_CANONICAL] || workflowBody[ATTR_NAME],
+        programmingLanguage: workflowBody[ATTR_PROG].toUpperCase(),
+        properties: workflowBody[ATTR_DATA],
+        codePathVersion: workflowBody[ATTR_CODE_PATH_VERSION],
+        initialLibraryVersion: workflowBody[ATTR_INITIAL_LIB_VERSION],
       },
-      this.getAppEnv(),
-    );
+    };
 
-    return http.post(url, body, { params });
+    const res = await graphQL.request(endpoint, mutation, variables);
+    return res.createWorkflowSchedule;
   }
 
   /**

src/async/Client.js

@@ -73,9 +73,7 @@ module.exports = class Engine {
           ? this.client.startWorkflow(job)
           : this.client.startTask(job);
 
-        await handler;
-
-        return undefined;
+        return handler;
       });
 
       // return results

src/async/Engine/Engine.js

@@ -0,0 +1,98 @@
+const { GraphQLClient } = require("graphql-request");
+const {
+  ExternalZenatonError,
+  InternalZenatonError,
+  ZenatonError,
+} = require("../../Errors");
+const { credentials } = require("../../client");
+
+function getError(err) {
+  // Validation errors
+  if (err.response && err.response.errors && err.response.errors.length > 0) {
+    const message = err.response.errors
+      .map((graphqlError) => {
+        const path = graphqlError.path ? `(${graphqlError.path}) ` : "";
+        const errorMessage = graphqlError.message || "Unknown error";
+        return `${path}${errorMessage}`;
+      })
+      .join("\n");
+    return new ExternalZenatonError(message);
+  }
+
+  // Internal Server Error
+  if (err.response && err.response.status >= 500) {
+    return new InternalZenatonError(
+      `Please contact Zenaton support - ${err.message}`,
+    );
+  }
+
+  return new ZenatonError(err.message);
+}
+
+async function request(endpoint, query, variables) {
+  try {
+    const graphQLClient = new GraphQLClient(endpoint, {
+      headers: {
+        Accept: "application/json",
+        "Content-Type": "application/json",
+        "app-id": credentials.appId,
+        "api-token": credentials.apiToken,
+      },
+    });
+
+    return graphQLClient.request(query, variables);
+  } catch (err) {
+    throw getError(err);
+  }
+}
+
+const mutations = {
+  createWorkflowSchedule: `
+    mutation ($createWorkflowScheduleInput: CreateWorkflowScheduleInput!) {
+      createWorkflowSchedule(input: $createWorkflowScheduleInput) {
+        schedule {
+          id
+          name
+          cron
+          insertedAt
+          updatedAt
+          target {
+            ... on WorkflowTarget {
+              name
+              type
+              canonicalName
+              programmingLanguage
+              properties
+              codePathVersion
+              initialLibraryVersion
+            }
+          }
+        }
+      }
+    }`,
+  createTaskSchedule: `
+    mutation ($createTaskScheduleInput: CreateTaskScheduleInput!) {
+      createTaskSchedule(input: $createTaskScheduleInput) {
+        schedule {
+          id
+          name
+          cron
+          insertedAt
+          updatedAt
+          target {
+            ... on TaskTarget {
+              name
+              type
+              programmingLanguage
+              properties
+              codePathVersion
+              initialLibraryVersion
+            }
+          }
+        }
+      }
+    }`,
+};
+
+module.exports.request = request;
+module.exports.mutations = mutations;

src/async/Services/GraphQL.js

@@ -1,3 +1,4 @@
 module.exports.http = require("./Http");
+module.exports.graphQL = require("./GraphQL");
 module.exports.serializer = require("./Serializer");
 module.exports.Trait = require("./Trait");

src/async/Services/index.js

@@ -1,6 +1,4 @@
-const CronParser = require("cron-parser");
 const TaskContext = require("../Runtime/Contexts/TaskContext");
-
 const Engine = require("../Engine/Engine");
 const { ZenatonError } = require("../../Errors");
 
@@ -27,14 +25,24 @@ module.exports = class AbstractTask {
   }
 
   // asynchronous execution within a workflow
-  async schedule() {
+  async schedule(cron) {
+    if (typeof cron !== "string" || cron === "") {
+      throw new ZenatonError(
+        "Param passed to 'schedule' function must be a non empty string",
+      );
+    }
+
+    this.scheduling = this.scheduling || {};
+    this.scheduling.cron = cron;
+
     const result = await new Engine().dispatch([this]);
-    return result[0].then(() => undefined);
+    return result[0].then((res) => res);
   }
 
   // asynchronous execution within a workflow
-  async dispatch(...args) {
-    return this.schedule(...args);
+  async dispatch() {
+    const result = await new Engine().dispatch([this]);
+    return result[0].then(() => undefined);
   }
 
   // synchronous execution within a workflow
@@ -43,27 +51,6 @@ module.exports = class AbstractTask {
     return result[0];
   }
 
-  repeat(cron) {
-    if (typeof cron !== "string") {
-      throw new ZenatonError(
-        "Param passed to 'repeat' function must be a string",
-      );
-    }
-
-    try {
-      CronParser.parseExpression(cron);
-    } catch (err) {
-      throw new ZenatonError(
-        "Param passed to 'repeat' function is not a proper CRON expression",
-      );
-    }
-
-    this.scheduling = this.scheduling || {};
-    this.scheduling.cron = cron;
-
-    return this;
-  }
-
   static methods() {
     return ["handle", "maxProcessingTime", "onFailureRetryDelay"];
   }

src/async/Tasks/AbstractTask.js

@@ -1,6 +1,4 @@
-const CronParser = require("cron-parser");
 const WorkflowContext = require("../Runtime/Contexts/WorkflowContext");
-
 const Engine = require("../Engine/Engine");
 const { ZenatonError } = require("../../Errors");
 
@@ -27,14 +25,24 @@ module.exports = class AbstractWorkflow {
   }
 
   // asynchronous execution within a workflow
-  async schedule() {
+  async schedule(cron) {
+    if (typeof cron !== "string" || cron === "") {
+      throw new ZenatonError(
+        "Param passed to 'schedule' function must be a non empty string",
+      );
+    }
+
+    this.scheduling = this.scheduling || {};
+    this.scheduling.cron = cron;
+
     const result = await new Engine().dispatch([this]);
-    return result[0].then(() => undefined);
+    return result[0].then((res) => res);
   }
 
   // asynchronous execution within a workflow
-  async dispatch(...args) {
-    return this.schedule(...args);
+  async dispatch() {
+    const result = await new Engine().dispatch([this]);
+    return result[0].then(() => undefined);
   }
 
   // synchronous execution within a workflow
@@ -43,27 +51,6 @@ module.exports = class AbstractWorkflow {
     return result[0];
   }
 
-  repeat(cron) {
-    if (typeof cron !== "string") {
-      throw new ZenatonError(
-        "Param passed to 'repeat' function must be a string",
-      );
-    }
-
-    try {
-      CronParser.parseExpression(cron);
-    } catch (err) {
-      throw new ZenatonError(
-        "Param passed to 'repeat' function is not a proper CRON expression",
-      );
-    }
-
-    this.scheduling = this.scheduling || {};
-    this.scheduling.cron = cron;
-
-    return this;
-  }
-
   static methods() {
     return [
       "handle",