feat: Add error details logic into execution steps

feat: show poll interval in trigger step

docker/compose/docker-compose.yml

@@ -0,0 +1,47 @@
+version: "3.9"
+services:
+  automatisch:
+    build:
+      context: ../images/wait-for-postgres
+      network: host
+    ports:
+      - "3000:3000"
+    depends_on:
+      - postgres
+      - redis
+    environment:
+      - HOST=localhost
+      - PROTOCOL=http
+      - PORT=3000
+      - APP_ENV=production
+      - REDIS_HOST=redis
+      - POSTGRES_HOST=postgres
+      - POSTGRES_DATABASE=automatisch
+      - POSTGRES_USERNAME=automatisch_user
+    volumes:
+      - automatisch_storage:/automatisch/storage
+  worker:
+    build:
+      context: ../images/plain
+      network: host
+    depends_on:
+      - automatisch
+    environment:
+      - APP_ENV=production
+      - REDIS_HOST=redis
+      - POSTGRES_HOST=postgres
+      - POSTGRES_DATABASE=automatisch
+      - POSTGRES_USERNAME=automatisch_user
+    command: automatisch start-worker --env-file /automatisch/storage/.env
+    volumes:
+      - automatisch_storage:/automatisch/storage
+  postgres:
+    image: "postgres:14.5"
+    environment:
+      POSTGRES_HOST_AUTH_METHOD: trust
+      POSTGRES_DB: automatisch
+      POSTGRES_USER: automatisch_user
+  redis:
+    image: "redis:7.0.4"
+volumes:
+  automatisch_storage:

docker/images/plain/Dockerfile

@@ -0,0 +1,11 @@
+# syntax=docker/dockerfile:1
+FROM node:16
+WORKDIR /automatisch
+
+# npm registry for dev purposes
+RUN npm config set fetch-retry-maxtimeout 5000
+RUN npm config set fetch-retry-mintimeout 3000
+RUN npm set registry http://localhost:5000
+# npm registry for dev purposes
+
+RUN yarn global add @automatisch/cli

docker/images/wait-for-postgres/Dockerfile

@@ -0,0 +1,21 @@
+# syntax=docker/dockerfile:1
+FROM node:16
+WORKDIR /automatisch
+
+RUN apt-get update && apt-get install -y postgresql-client
+COPY ./wait-for-postgres.sh /automatisch/wait-for-postgres.sh
+
+# npm registry for dev purposes
+RUN npm config set fetch-retry-maxtimeout 5000
+RUN npm config set fetch-retry-mintimeout 3000
+RUN npm set registry http://localhost:5000
+# npm registry for dev purposes
+
+RUN mkdir -p /automatisch/storage
+RUN touch /automatisch/storage/.env
+RUN echo "ENCRYPTION_KEY=$(openssl rand -base64 36)" >> /automatisch/storage/.env
+RUN echo "APP_SECRET_KEY=$(openssl rand -base64 36)" >> /automatisch/storage/.env
+RUN yarn global add @automatisch/cli
+
+EXPOSE 3000
+CMD sh /automatisch/wait-for-postgres.sh automatisch start --env-file=/automatisch/storage/.env

docker/images/wait-for-postgres/wait-for-postgres.sh

@@ -0,0 +1,11 @@
+#!/bin/sh
+
+set -e
+
+until psql -h "$POSTGRES_HOST" -U "$POSTGRES_USERNAME" -d "$POSTGRES_HOST" -c '\q'; do
+  >&2 echo "Postgres is unavailable - sleeping"
+  sleep 1
+done
+
+>&2 echo "Postgres is up - executing command"
+exec "$@"

packages/backend/.env-example

@@ -13,3 +13,4 @@ ENCRYPTION_KEY=sample-encryption-key
 APP_SECRET_KEY=sample-app-secret-key
 REDIS_PORT=6379
 REDIS_HOST=127.0.0.1
+ENABLE_BULLMQ_DASHBOARD=false

packages/backend/bin/database/utils.ts

@@ -12,8 +12,14 @@ export async function createUser(email = 'user@automatisch.io', password = 'samp
   };
 
   try {
-    const user = await User.query().insertAndFetch(userParams);
-    logger.info(`User has been saved: ${user.email}`);
+    const userCount = await User.query().resultSize();
+
+    if (userCount === 0) {
+      const user = await User.query().insertAndFetch(userParams);
+      logger.info(`User has been saved: ${user.email}`);
+    } else {
+      logger.info('No need to seed a user.');
+    }
   } catch (err) {
     if ((err as any).nativeError.code !== UNIQUE_VIOLATION_CODE) {
       throw err;

packages/backend/database-utils.d.ts

@@ -0,0 +1 @@
+export * from './dist/bin/database/utils';

packages/backend/database-utils.js

@@ -0,0 +1,2 @@
+/* eslint-disable */
+module.exports = require('./dist/bin/database/utils');

packages/backend/database.d.ts

@@ -1,2 +1 @@
-export * as utils from './dist/bin/database/utils';
-export * as database from './dist/src/config/database';
+export * from './dist/src/config/database';

packages/backend/database.js

@@ -1,3 +1,2 @@
 /* eslint-disable */
-module.exports.utils = require('./dist/bin/database/utils');
-module.exports.database = require('./dist/src/config/database');
+module.exports = require('./dist/src/config/database');

packages/backend/package.json

@@ -3,7 +3,7 @@
   "version": "0.1.0",
   "description": "> TODO: description",
   "scripts": {
-    "dev": "nodemon --watch 'src/**/*.ts' --watch 'bin/**/*.ts' --exec 'ts-node' src/server.ts --ext ts,json",
+    "dev": "ts-node-dev src/server.ts",
     "worker": "nodemon --watch 'src/**/*.ts' --exec 'ts-node' src/worker.ts",
     "build": "tsc && yarn copy-statics",
     "build:watch": "nodemon --watch 'src/**/*.ts' --watch 'bin/**/*.ts' --exec yarn build --ext ts",
@@ -16,7 +16,8 @@
     "db:migration:create": "knex migrate:make",
     "db:rollback": "knex migrate:rollback",
     "db:migrate": "knex migrate:latest",
-    "copy-statics": "copyfiles src/**/*.{graphql,json,svg} dist"
+    "copy-statics": "copyfiles src/**/*.{graphql,json,svg} dist",
+    "prepack": "yarn build"
   },
   "dependencies": {
     "@automatisch/web": "0.1.0",
@@ -24,7 +25,6 @@
     "@gitbeaker/node": "^35.6.0",
     "@graphql-tools/graphql-file-loader": "^7.3.4",
     "@graphql-tools/load": "^7.5.2",
-    "@octokit/oauth-methods": "^1.2.6",
     "@rudderstack/rudder-sdk-node": "^1.1.2",
     "@slack/bolt": "3.10.0",
     "@types/luxon": "^2.3.1",
@@ -53,6 +53,7 @@
     "luxon": "2.3.1",
     "morgan": "^1.10.0",
     "nodemailer": "6.7.0",
+    "oauth-1.0a": "^2.2.6",
     "objection": "^3.0.0",
     "octokit": "^1.7.1",
     "pg": "^8.7.1",
@@ -75,14 +76,19 @@
     "test": "__tests__"
   },
   "files": [
+    "dist",
     "bin",
     "src",
     "server.js",
     "server.d.ts",
+    "worker.js",
+    "worker.d.ts",
     "logger.js",
     "logger.d.ts",
     "database.js",
-    "database.d.ts"
+    "database.d.ts",
+    "database-utils.js",
+    "database-utils.d.ts"
   ],
   "repository": {
     "type": "git",
@@ -109,7 +115,8 @@
     "ava": "^3.15.0",
     "nodemon": "^2.0.13",
     "sinon": "^11.1.2",
-    "ts-node": "^10.2.1"
+    "ts-node": "^10.2.1",
+    "ts-node-dev": "^1.1.8"
   },
   "ava": {
     "files": [

packages/backend/src/app.ts

@@ -15,13 +15,13 @@ import {
 } from './helpers/create-bull-board-handler';
 import injectBullBoardHandler from './helpers/inject-bull-board-handler';
 
-if (appConfig.appEnv === 'development') {
+if (appConfig.enableBullMQDashboard) {
   createBullBoardHandler(serverAdapter);
 }
 
 const app = express();
 
-if (appConfig.appEnv === 'development') {
+if (appConfig.enableBullMQDashboard) {
   injectBullBoardHandler(app, serverAdapter);
 }
 

packages/backend/src/apps/discord/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/discord/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/discord",
   "primaryColor": "5865f2",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "oAuthRedirectUrl",

packages/backend/src/apps/firebase/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/firebase/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/firebase",
   "primaryColor": "ffca28",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "oAuthRedirectUrl",

packages/backend/src/apps/flickr/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/flickr/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/flickr",
   "primaryColor": "000000",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "oAuthRedirectUrl",
@@ -222,8 +223,8 @@
       "description": "Triggers when you favorite a photo.",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "testStep",
@@ -237,8 +238,8 @@
       "description": "Triggers when you add a new photo in an album.",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -275,8 +276,8 @@
       "description": "Triggers when you add a new photo.",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "testStep",
@@ -290,8 +291,8 @@
       "description": "Triggers when you create a new album.",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "testStep",

packages/backend/src/apps/github/authentication.ts

@@ -4,35 +4,19 @@ import type {
   IField,
   IJSONObject,
 } from '@automatisch/types';
-import {
-  getWebFlowAuthorizationUrl,
-  exchangeWebFlowCode,
-  checkToken,
-} from '@octokit/oauth-methods';
+import HttpClient from '../../helpers/http-client';
+import { URLSearchParams } from 'url';
 
 export default class Authentication implements IAuthentication {
   appData: IApp;
   connectionData: IJSONObject;
-  scopes: string[] = [
-    'read:org',
-    'repo',
-    'user',
-  ];
-  client: {
-    getWebFlowAuthorizationUrl: typeof getWebFlowAuthorizationUrl;
-    exchangeWebFlowCode: typeof exchangeWebFlowCode;
-    checkToken: typeof checkToken;
-  };
+  scopes: string[] = ['read:org', 'repo', 'user'];
+  client: HttpClient;
 
   constructor(appData: IApp, connectionData: IJSONObject) {
     this.connectionData = connectionData;
     this.appData = appData;
-
-    this.client = {
-      getWebFlowAuthorizationUrl,
-      exchangeWebFlowCode,
-      checkToken,
-    };
+    this.client = new HttpClient({ baseURL: 'https://github.com' });
   }
 
   get oauthRedirectUrl(): string {
@@ -42,26 +26,28 @@ export default class Authentication implements IAuthentication {
   }
 
   async createAuthData(): Promise<{ url: string }> {
-    const { url } = await this.client.getWebFlowAuthorizationUrl({
-      clientType: 'oauth-app',
-      clientId: this.connectionData.consumerKey as string,
-      redirectUrl: this.oauthRedirectUrl,
-      scopes: this.scopes,
+    const searchParams = new URLSearchParams({
+      client_id: this.connectionData.consumerKey as string,
+      redirect_uri: this.oauthRedirectUrl,
+      scope: this.scopes.join(','),
     });
 
+    const url = `https://github.com/login/oauth/authorize?${searchParams.toString()}`;
+
     return {
-      url: url,
+      url,
     };
   }
 
   async verifyCredentials() {
-    const { data } = await this.client.exchangeWebFlowCode({
-      clientType: 'oauth-app',
-      clientId: this.connectionData.consumerKey as string,
-      clientSecret: this.connectionData.consumerSecret as string,
-      code: this.connectionData.oauthVerifier as string,
+    const response = await this.client.post('/login/oauth/access_token', {
+      client_id: this.connectionData.consumerKey,
+      client_secret: this.connectionData.consumerSecret,
+      code: this.connectionData.oauthVerifier,
     });
 
+    const data = Object.fromEntries(new URLSearchParams(response.data));
+
     this.connectionData.accessToken = data.access_token;
 
     const tokenInfo = await this.getTokenInfo();
@@ -78,12 +64,23 @@ export default class Authentication implements IAuthentication {
   }
 
   async getTokenInfo() {
-    return this.client.checkToken({
-      clientType: 'oauth-app',
-      clientId: this.connectionData.consumerKey as string,
-      clientSecret: this.connectionData.consumerSecret as string,
-      token: this.connectionData.accessToken as string,
-    });
+    const basicAuthToken = Buffer.from(
+      this.connectionData.consumerKey + ':' + this.connectionData.consumerSecret
+    ).toString('base64');
+
+    const headers = {
+      Authorization: `Basic ${basicAuthToken}`,
+    };
+
+    const body = {
+      access_token: this.connectionData.accessToken,
+    };
+
+    return await this.client.post(
+      `https://api.github.com/applications/${this.connectionData.consumerKey}/token`,
+      body,
+      { headers }
+    );
   }
 
   async isStillVerified() {

packages/backend/src/apps/github/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/github/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/github",
   "primaryColor": "000000",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "oAuthRedirectUrl",
@@ -19,26 +20,26 @@
     },
     {
       "key": "consumerKey",
-      "label": "Consumer Key",
+      "label": "Client ID",
       "type": "string",
       "required": true,
       "readOnly": false,
       "value": null,
       "placeholder": null,
       "description": null,
-      "docUrl": "https://automatisch.io/docs/github#consumer-key",
+      "docUrl": "https://automatisch.io/docs/github#client-id",
       "clickToCopy": false
     },
     {
       "key": "consumerSecret",
-      "label": "Consumer Secret",
+      "label": "Client Secret",
       "type": "string",
       "required": true,
       "readOnly": false,
       "value": null,
       "placeholder": null,
       "description": null,
-      "docUrl": "https://automatisch.io/docs/github#consumer-secret",
+      "docUrl": "https://automatisch.io/docs/github#client-secret",
       "clickToCopy": false
     }
   ],
@@ -222,8 +223,8 @@
       "description": "Triggers when a new repository is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "testStep",
@@ -237,8 +238,8 @@
       "description": "Triggers when a new organization is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "testStep",
@@ -252,8 +253,8 @@
       "description": "Triggers when a new branch is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -290,8 +291,8 @@
       "description": "Triggers when a new notification is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -329,8 +330,8 @@
       "description": "Triggers when a new pull request is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -367,8 +368,8 @@
       "description": "Triggers when a new watcher is added to a repo",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -405,8 +406,8 @@
       "description": "Triggers when a new milestone is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -443,8 +444,8 @@
       "description": "Triggers when a new commit comment is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -481,8 +482,8 @@
       "description": "Triggers when a new label is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -519,8 +520,8 @@
       "description": "Triggers when a new collaborator is added to a repo",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -557,8 +558,8 @@
       "description": "Triggers when a new release is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -595,8 +596,8 @@
       "description": "Triggers when a new commit is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -656,8 +657,8 @@
       "description": "Triggers when a new issue is created",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",

packages/backend/src/apps/gitlab/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/gitlab/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/gitlab",
   "primaryColor": "2DAAE1",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "oAuthRedirectUrl",

packages/backend/src/apps/postgresql/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/postgresql/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/postgresql",
   "primaryColor": "2DAAE1",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "host",

packages/backend/src/apps/scheduler/info.json

@@ -3,7 +3,9 @@
   "key": "scheduler",
   "iconUrl": "{BASE_URL}/apps/scheduler/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/scheduler",
+  "authDocUrl": "https://automatisch.io/docs/connections/scheduler",
   "primaryColor": "0059F7",
+  "supportsConnections": false,
   "requiresAuthentication": false,
   "triggers": [
     {

packages/backend/src/apps/slack/actions.ts

@@ -1,13 +1,15 @@
 import SendMessageToChannel from './actions/send-message-to-channel';
-import { IJSONObject } from '@automatisch/types';
+import FindMessage from './actions/find-message';
+import SlackClient from './client';
 
 export default class Actions {
+  client: SlackClient;
   sendMessageToChannel: SendMessageToChannel;
+  findMessage: FindMessage;
 
-  constructor(connectionData: IJSONObject, parameters: IJSONObject) {
-    this.sendMessageToChannel = new SendMessageToChannel(
-      connectionData,
-      parameters
-    );
+  constructor(client: SlackClient) {
+    this.client = client;
+    this.sendMessageToChannel = new SendMessageToChannel(client);
+    this.findMessage = new FindMessage(client);
   }
 }

packages/backend/src/apps/slack/actions/find-message.ts

@@ -0,0 +1,26 @@
+import SlackClient from '../client';
+
+export default class FindMessage {
+  client: SlackClient;
+
+  constructor(client: SlackClient) {
+    this.client = client;
+  }
+
+  async run() {
+    const parameters = this.client.step.parameters;
+    const query = parameters.query as string;
+    const sortBy = parameters.sortBy as string;
+    const sortDirection = parameters.sortDirection as string;
+    const count = 1;
+
+    const messages = await this.client.findMessages.run(
+      query,
+      sortBy,
+      sortDirection,
+      count,
+    );
+
+    return messages;
+  }
+}

packages/backend/src/apps/slack/actions/send-message-to-channel.ts

@@ -1,21 +1,18 @@
-import { WebClient } from '@slack/web-api';
-import { IJSONObject } from '@automatisch/types';
+import SlackClient from '../client';
 
 export default class SendMessageToChannel {
-  client: WebClient;
-  parameters: IJSONObject;
+  client: SlackClient;
 
-  constructor(connectionData: IJSONObject, parameters: IJSONObject) {
-    this.client = new WebClient(connectionData.accessToken as string);
-    this.parameters = parameters;
+  constructor(client: SlackClient) {
+    this.client = client;
   }
 
   async run() {
-    const result = await this.client.chat.postMessage({
-      channel: this.parameters.channel as string,
-      text: this.parameters.message as string,
-    });
+    const channelId = this.client.step.parameters.channel as string;
+    const text = this.client.step.parameters.message as string;
 
-    return result;
+    const message = await this.client.postMessageToChannel.run(channelId, text);
+
+    return message;
   }
 }

packages/backend/src/apps/slack/authentication.ts

@@ -1,36 +1,33 @@
-import type { IAuthentication, IApp, IJSONObject } from '@automatisch/types';
-import { WebClient } from '@slack/web-api';
+import type { IAuthentication, IJSONObject } from '@automatisch/types';
+import SlackClient from './client';
 
 export default class Authentication implements IAuthentication {
-  appData: IApp;
-  connectionData: IJSONObject;
-  client: WebClient;
+  client: SlackClient;
 
-  constructor(appData: IApp, connectionData: IJSONObject) {
-    this.client = new WebClient();
+  static requestOptions: IJSONObject = {
+    headers: {
+      'Content-Type': 'application/x-www-form-urlencoded',
+    },
+  };
 
-    this.connectionData = connectionData;
-    this.appData = appData;
+  constructor(client: SlackClient) {
+    this.client = client;
   }
 
   async verifyCredentials() {
-    const { bot_id: botId, user: screenName } = await this.client.auth.test({
-      token: this.connectionData.accessToken as string,
-    });
+    const { bot_id: botId, user: screenName } =
+      await this.client.verifyAccessToken.run();
 
     return {
       botId,
       screenName,
-      token: this.connectionData.accessToken,
+      token: this.client.connection.formattedData.accessToken,
     };
   }
 
   async isStillVerified() {
     try {
-      await this.client.auth.test({
-        token: this.connectionData.accessToken as string,
-      });
-
+      await this.client.verifyAccessToken.run();
       return true;
     } catch (error) {
       return false;

packages/backend/src/apps/slack/client/endpoints/find-messages.ts

@@ -0,0 +1,44 @@
+import SlackClient from '../index';
+
+export default class FindMessages {
+  client: SlackClient;
+
+  constructor(client: SlackClient) {
+    this.client = client;
+  }
+
+  async run(query: string, sortBy: string, sortDirection: string, count = 1) {
+    const headers = {
+      Authorization: `Bearer ${this.client.connection.formattedData.accessToken}`,
+    };
+
+    const params = {
+      query,
+      sort: sortBy,
+      sort_dir: sortDirection,
+      count,
+    };
+
+    const response = await this.client.httpClient.get('/search.messages', {
+      headers,
+      params,
+    });
+
+    const data = response.data;
+
+    if (!data.ok) {
+      if (data.error === 'missing_scope') {
+        throw new Error(
+          `Error occured while finding messages; ${data.error}: ${data.needed}`
+        );
+      }
+
+      throw new Error(`Error occured while finding messages; ${data.error}`);
+    }
+
+    const messages = data.messages.matches;
+    const message = messages?.[0];
+
+    return message;
+  }
+}

packages/backend/src/apps/slack/client/endpoints/post-message-to-channel.ts

@@ -0,0 +1,34 @@
+import SlackClient from '../index';
+
+export default class PostMessageToChannel {
+  client: SlackClient;
+
+  constructor(client: SlackClient) {
+    this.client = client;
+  }
+
+  async run(channelId: string, text: string) {
+    const headers = {
+      Authorization: `Bearer ${this.client.connection.formattedData.accessToken}`,
+    };
+
+    const params = {
+      channel: channelId,
+      text,
+    };
+
+    const response = await this.client.httpClient.post(
+      '/chat.postMessage',
+      params,
+      { headers }
+    );
+
+    if (response.data.ok === 'false') {
+      throw new Error(
+        `Error occured while posting a message to channel: ${response.data.error}`
+      );
+    }
+
+    return response.data.message;
+  }
+}

packages/backend/src/apps/slack/client/endpoints/verify-access-token.ts

@@ -0,0 +1,35 @@
+import { IJSONObject } from '@automatisch/types';
+import qs from 'qs';
+import SlackClient from '../index';
+
+export default class VerifyAccessToken {
+  client: SlackClient;
+
+  static requestOptions: IJSONObject = {
+    headers: {
+      'Content-Type': 'application/x-www-form-urlencoded',
+    },
+  };
+
+  constructor(client: SlackClient) {
+    this.client = client;
+  }
+
+  async run() {
+    const response = await this.client.httpClient.post(
+      '/auth.test',
+      qs.stringify({
+        token: this.client.connection.formattedData.accessToken,
+      }),
+      VerifyAccessToken.requestOptions
+    );
+
+    if (response.data.ok === false) {
+      throw new Error(
+        `Error occured while verifying credentials: ${response.data.error}.(More info: https://api.slack.com/methods/auth.test#errors)`
+      );
+    }
+
+    return response.data;
+  }
+}

packages/backend/src/apps/slack/client/index.ts

@@ -0,0 +1,29 @@
+import { IFlow, IStep, IConnection } from '@automatisch/types';
+import HttpClient from '../../../helpers/http-client';
+import VerifyAccessToken from './endpoints/verify-access-token';
+import PostMessageToChannel from './endpoints/post-message-to-channel';
+import FindMessages from './endpoints/find-messages';
+
+export default class SlackClient {
+  flow: IFlow;
+  step: IStep;
+  connection: IConnection;
+  httpClient: HttpClient;
+
+  verifyAccessToken: VerifyAccessToken;
+  postMessageToChannel: PostMessageToChannel;
+  findMessages: FindMessages;
+
+  static baseUrl = 'https://slack.com/api';
+
+  constructor(connection: IConnection, flow?: IFlow, step?: IStep) {
+    this.connection = connection;
+    this.flow = flow;
+    this.step = step;
+
+    this.httpClient = new HttpClient({ baseURL: SlackClient.baseUrl });
+    this.verifyAccessToken = new VerifyAccessToken(this);
+    this.postMessageToChannel = new PostMessageToChannel(this);
+    this.findMessages = new FindMessages(this);
+  }
+}

packages/backend/src/apps/slack/data.ts

@@ -1,10 +1,12 @@
 import ListChannels from './data/list-channels';
-import { IJSONObject } from '@automatisch/types';
+import SlackClient from './client';
 
 export default class Data {
+  client: SlackClient;
   listChannels: ListChannels;
 
-  constructor(connectionData: IJSONObject) {
-    this.listChannels = new ListChannels(connectionData);
+  constructor(client: SlackClient) {
+    this.client = client;
+    this.listChannels = new ListChannels(client);
   }
 }

packages/backend/src/apps/slack/data/list-channels.ts

@@ -1,17 +1,27 @@
-import type { IJSONObject } from '@automatisch/types';
-import { WebClient } from '@slack/web-api';
+import { IJSONObject } from '@automatisch/types';
+import SlackClient from '../client';
 
 export default class ListChannels {
-  client: WebClient;
+  client: SlackClient;
 
-  constructor(connectionData: IJSONObject) {
-    this.client = new WebClient(connectionData.accessToken as string);
+  constructor(client: SlackClient) {
+    this.client = client;
   }
 
   async run() {
-    const { channels } = await this.client.conversations.list();
+    const response = await this.client.httpClient.get('/conversations.list', {
+      headers: {
+        Authorization: `Bearer ${this.client.connection.formattedData.accessToken}`,
+      },
+    });
 
-    return channels.map((channel) => {
+    if (response.data.ok === 'false') {
+      throw new Error(
+        `Error occured while fetching slack channels: ${response.data.error}`
+      );
+    }
+
+    return response.data.channels.map((channel: IJSONObject) => {
       return {
         value: channel.id,
         name: channel.name,

packages/backend/src/apps/slack/index.ts

@@ -1,28 +1,30 @@
 import {
   IService,
   IAuthentication,
-  IApp,
-  IJSONObject,
+  IConnection,
+  IFlow,
+  IStep,
 } from '@automatisch/types';
 import Authentication from './authentication';
 import Triggers from './triggers';
 import Actions from './actions';
 import Data from './data';
+import SlackClient from './client';
 
 export default class Slack implements IService {
+  client: SlackClient;
+
   authenticationClient: IAuthentication;
   triggers: Triggers;
   actions: Actions;
   data: Data;
 
-  constructor(
-    appData: IApp,
-    connectionData: IJSONObject,
-    parameters: IJSONObject
-  ) {
-    this.authenticationClient = new Authentication(appData, connectionData);
-    this.data = new Data(connectionData);
-    this.triggers = new Triggers(connectionData, parameters);
-    this.actions = new Actions(connectionData, parameters);
+  constructor(connection: IConnection, flow?: IFlow, step?: IStep) {
+    this.client = new SlackClient(connection, flow, step);
+
+    this.authenticationClient = new Authentication(this.client);
+    // this.triggers = new Triggers(this.client);
+    this.actions = new Actions(this.client);
+    this.data = new Data(this.client);
   }
 }

packages/backend/src/apps/slack/info.json

@@ -3,7 +3,9 @@
   "key": "slack",
   "iconUrl": "{BASE_URL}/apps/slack/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/slack",
+  "authDocUrl": "https://automatisch.io/docs/connections/slack",
   "primaryColor": "2DAAE1",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "accessToken",
@@ -14,7 +16,6 @@
       "value": null,
       "placeholder": null,
       "description": "Access token of slack that Automatisch will connect to.",
-      "docUrl": "https://automatisch.io/docs/slack#access-token",
       "clickToCopy": false
     }
   ],
@@ -101,11 +102,12 @@
     {
       "name": "New message posted to a channel",
       "key": "newMessageToChannel",
+      "pollInterval": 15,
       "description": "Triggers when a new message is posted to a channel",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -163,8 +165,8 @@
       "description": "Send a message to a specific channel you specify.",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "setupAction",
@@ -203,6 +205,73 @@
           "name": "Test action"
         }
       ]
+    },
+    {
+      "name": "Find message",
+      "key": "findMessage",
+      "description": "Find a Slack message using the Slack Search feature.",
+      "substeps": [
+        {
+          "key": "chooseConnection",
+          "name": "Choose connection"
+        },
+        {
+          "key": "setupAction",
+          "name": "Set up action",
+          "arguments": [
+            {
+              "label": "Search Query",
+              "key": "query",
+              "type": "string",
+              "required": true,
+              "description": "Search query to use for finding matching messages. See the Slack Search Documentation for more information on constructing a query.",
+              "variables": true
+            },
+            {
+              "label": "Sort by",
+              "key": "sortBy",
+              "type": "dropdown",
+              "description": "Sort messages by their match strength or by their date. Default is score.",
+              "required": true,
+              "value": "score",
+              "variables": false,
+              "options": [
+                {
+                  "label": "Match strength",
+                  "value": "score"
+                },
+                {
+                  "label": "Message date time",
+                  "value": "timestamp"
+                }
+              ]
+            },
+            {
+              "label": "Sort direction",
+              "key": "sortDirection",
+              "type": "dropdown",
+              "description": "Sort matching messages in ascending or descending order. Default is descending.",
+              "required": true,
+              "value": "desc",
+              "variables": false,
+              "options": [
+                {
+                  "label": "Descending (newest or best match first)",
+                  "value": "desc"
+                },
+                {
+                  "label": "Ascending (oldest or worst match first)",
+                  "value": "asc"
+                }
+              ]
+            }
+          ]
+        },
+        {
+          "key": "testStep",
+          "name": "Test action"
+        }
+      ]
     }
   ]
 }

packages/backend/src/apps/smtp/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/smtp/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/smtp",
   "primaryColor": "2DAAE1",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "host",

packages/backend/src/apps/twilio/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/twilio/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/twilio",
   "primaryColor": "f22f46",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "accountSid",

packages/backend/src/apps/twitch/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/twitch/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/twitch",
   "primaryColor": "2DAAE1",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "oAuthRedirectUrl",

packages/backend/src/apps/twitter/actions.ts

@@ -1,10 +1,12 @@
+import TwitterClient from './client';
 import CreateTweet from './actions/create-tweet';
-import { IJSONObject } from '@automatisch/types';
 
 export default class Actions {
+  client: TwitterClient;
   createTweet: CreateTweet;
 
-  constructor(connectionData: IJSONObject, parameters: IJSONObject) {
-    this.createTweet = new CreateTweet(connectionData, parameters);
+  constructor(client: TwitterClient) {
+    this.client = client;
+    this.createTweet = new CreateTweet(client);
   }
 }

packages/backend/src/apps/twitter/actions/create-tweet.ts

@@ -1,23 +1,17 @@
-import TwitterApi, { TwitterApiTokens } from 'twitter-api-v2';
-import { IJSONObject } from '@automatisch/types';
+import TwitterClient from '../client';
 
 export default class CreateTweet {
-  client: TwitterApi;
-  parameters: IJSONObject;
+  client: TwitterClient;
 
-  constructor(connectionData: IJSONObject, parameters: IJSONObject) {
-    this.client = new TwitterApi({
-      appKey: connectionData.consumerKey,
-      appSecret: connectionData.consumerSecret,
-      accessToken: connectionData.accessToken,
-      accessSecret: connectionData.accessSecret,
-    } as TwitterApiTokens);
-
-    this.parameters = parameters;
+  constructor(client: TwitterClient) {
+    this.client = client;
   }
 
   async run() {
-    const tweet = await this.client.v1.tweet(this.parameters.tweet as string);
+    const tweet = await this.client.createTweet.run(
+      this.client.step.parameters.tweet as string
+    );
+
     return tweet;
   }
 }

packages/backend/src/apps/twitter/authentication.ts

@@ -1,65 +1,50 @@
-import type {
-  IAuthentication,
-  IApp,
-  IField,
-  IJSONObject,
-} from '@automatisch/types';
-import TwitterApi, { TwitterApiTokens } from 'twitter-api-v2';
+import type { IAuthentication, IField } from '@automatisch/types';
+import { URLSearchParams } from 'url';
+import TwitterClient from './client';
 
 export default class Authentication implements IAuthentication {
-  appData: IApp;
-  connectionData: IJSONObject;
-  client: TwitterApi;
+  client: TwitterClient;
 
-  constructor(appData: IApp, connectionData: IJSONObject) {
-    this.appData = appData;
-    this.connectionData = connectionData;
-
-    const clientParams = {
-      appKey: connectionData.consumerKey,
-      appSecret: connectionData.consumerSecret,
-      accessToken: connectionData.accessToken,
-      accessSecret: connectionData.accessSecret,
-    } as TwitterApiTokens;
-
-    this.client = new TwitterApi(clientParams);
+  constructor(client: TwitterClient) {
+    this.client = client;
   }
 
   async createAuthData() {
-    const appFields = this.appData.fields.find(
+    const appFields = this.client.connection.appData.fields.find(
       (field: IField) => field.key == 'oAuthRedirectUrl'
     );
     const callbackUrl = appFields.value;
 
-    const authLink = await this.client.generateAuthLink(callbackUrl);
+    const response = await this.client.oauthRequestToken.run(callbackUrl);
+    const responseData = Object.fromEntries(new URLSearchParams(response.data));
 
     return {
-      url: authLink.url,
-      accessToken: authLink.oauth_token,
-      accessSecret: authLink.oauth_token_secret,
+      url: `${TwitterClient.baseUrl}/oauth/authorize?oauth_token=${responseData.oauth_token}`,
+      accessToken: responseData.oauth_token,
+      accessSecret: responseData.oauth_token_secret,
     };
   }
 
   async verifyCredentials() {
-    const verifiedCredentials = await this.client.login(
-      this.connectionData.oauthVerifier as string
-    );
+    const response = await this.client.verifyAccessToken.run();
+    const responseData = Object.fromEntries(new URLSearchParams(response.data));
 
     return {
-      consumerKey: this.connectionData.consumerKey,
-      consumerSecret: this.connectionData.consumerSecret,
-      accessToken: verifiedCredentials.accessToken,
-      accessSecret: verifiedCredentials.accessSecret,
-      userId: verifiedCredentials.userId,
-      screenName: verifiedCredentials.screenName,
+      consumerKey: this.client.connection.formattedData.consumerKey as string,
+      consumerSecret: this.client.connection.formattedData
+        .consumerSecret as string,
+      accessToken: responseData.oauth_token,
+      accessSecret: responseData.oauth_token_secret,
+      userId: responseData.user_id,
+      screenName: responseData.screen_name,
     };
   }
 
   async isStillVerified() {
     try {
-      await this.client.currentUser();
+      await this.client.getCurrentUser.run();
       return true;
-    } catch {
+    } catch (error) {
       return false;
     }
   }

packages/backend/src/apps/twitter/client/endpoints/create-tweet.ts

@@ -0,0 +1,40 @@
+import TwitterClient from '../index';
+
+export default class CreateTweet {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(text: string) {
+    try {
+      const token = {
+        key: this.client.connection.formattedData.accessToken as string,
+        secret: this.client.connection.formattedData.accessSecret as string,
+      };
+
+      const requestData = {
+        url: `${TwitterClient.baseUrl}/2/tweets`,
+        method: 'POST',
+      };
+
+      const authHeader = this.client.oauthClient.toHeader(
+        this.client.oauthClient.authorize(requestData, token)
+      );
+
+      const response = await this.client.httpClient.post(
+        `/2/tweets`,
+        { text },
+        { headers: { ...authHeader } }
+      );
+
+      const tweet = response.data.data;
+
+      return tweet;
+    } catch (error) {
+      const errorMessage = error.response.data.detail;
+      throw new Error(`Error occured while creating a tweet: ${errorMessage}`);
+    }
+  }
+}

packages/backend/src/apps/twitter/client/endpoints/get-current-user.ts

@@ -0,0 +1,35 @@
+import TwitterClient from '../index';
+
+export default class GetCurrentUser {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run() {
+    const token = {
+      key: this.client.connection.formattedData.accessToken as string,
+      secret: this.client.connection.formattedData.accessSecret as string,
+    };
+
+    const requestPath = '/2/users/me';
+
+    const requestData = {
+      url: `${TwitterClient.baseUrl}${requestPath}`,
+      method: 'GET',
+    };
+
+    const authHeader = this.client.oauthClient.toHeader(
+      this.client.oauthClient.authorize(requestData, token)
+    );
+
+    const response = await this.client.httpClient.get(requestPath, {
+      headers: { ...authHeader },
+    });
+
+    const currentUser = response.data.data;
+
+    return currentUser;
+  }
+}

packages/backend/src/apps/twitter/client/endpoints/get-user-by-username.ts

@@ -0,0 +1,45 @@
+import { IJSONObject } from '@automatisch/types';
+import TwitterClient from '../index';
+
+export default class GetUserByUsername {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(username: string) {
+    const token = {
+      key: this.client.connection.formattedData.accessToken as string,
+      secret: this.client.connection.formattedData.accessSecret as string,
+    };
+
+    const requestPath = `/2/users/by/username/${username}`;
+
+    const requestData = {
+      url: `${TwitterClient.baseUrl}${requestPath}`,
+      method: 'GET',
+    };
+
+    const authHeader = this.client.oauthClient.toHeader(
+      this.client.oauthClient.authorize(requestData, token)
+    );
+
+    const response = await this.client.httpClient.get(requestPath, {
+      headers: { ...authHeader },
+    });
+
+    if (response.data?.errors) {
+      const errorMessages = response.data.errors
+        .map((error: IJSONObject) => error.detail)
+        .join(' ');
+
+      throw new Error(
+        `Error occured while fetching user data: ${errorMessages}`
+      );
+    }
+
+    const user = response.data.data;
+    return user;
+  }
+}

packages/backend/src/apps/twitter/client/endpoints/get-user-followers.ts

@@ -0,0 +1,70 @@
+import { IJSONObject } from '@automatisch/types';
+import { URLSearchParams } from 'url';
+import TwitterClient from '../index';
+import omitBy from 'lodash/omitBy';
+import isEmpty from 'lodash/isEmpty';
+
+export default class GetUserFollowers {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(userId: string, lastInternalId?: string) {
+    const token = {
+      key: this.client.connection.formattedData.accessToken as string,
+      secret: this.client.connection.formattedData.accessSecret as string,
+    };
+
+    let response;
+    const followers: IJSONObject[] = [];
+
+    do {
+      const params: IJSONObject = {
+        pagination_token: response?.data?.meta?.next_token,
+      };
+
+      const queryParams = new URLSearchParams(omitBy(params, isEmpty));
+
+      const requestPath = `/2/users/${userId}/followers${
+        queryParams.toString() ? `?${queryParams.toString()}` : ''
+      }`;
+
+      const requestData = {
+        url: `${TwitterClient.baseUrl}${requestPath}`,
+        method: 'GET',
+      };
+
+      const authHeader = this.client.oauthClient.toHeader(
+        this.client.oauthClient.authorize(requestData, token)
+      );
+
+      response = await this.client.httpClient.get(requestPath, {
+        headers: { ...authHeader },
+      });
+
+      if (response.data.meta.result_count > 0) {
+        response.data.data.forEach((tweet: IJSONObject) => {
+          if (!lastInternalId || Number(tweet.id) > Number(lastInternalId)) {
+            followers.push(tweet);
+          } else {
+            return;
+          }
+        });
+      }
+    } while (response.data.meta.next_token && lastInternalId);
+
+    if (response.data?.errors) {
+      const errorMessages = response.data.errors
+        .map((error: IJSONObject) => error.detail)
+        .join(' ');
+
+      throw new Error(
+        `Error occured while fetching user data: ${errorMessages}`
+      );
+    }
+
+    return followers;
+  }
+}

packages/backend/src/apps/twitter/client/endpoints/get-user-tweets.ts

@@ -0,0 +1,71 @@
+import { IJSONObject } from '@automatisch/types';
+import { URLSearchParams } from 'url';
+import TwitterClient from '../index';
+import omitBy from 'lodash/omitBy';
+import isEmpty from 'lodash/isEmpty';
+
+export default class GetUserTweets {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(userId: string, lastInternalId?: string) {
+    const token = {
+      key: this.client.connection.formattedData.accessToken as string,
+      secret: this.client.connection.formattedData.accessSecret as string,
+    };
+
+    let response;
+    const tweets: IJSONObject[] = [];
+
+    do {
+      const params: IJSONObject = {
+        since_id: lastInternalId,
+        pagination_token: response?.data?.meta?.next_token,
+      };
+
+      const queryParams = new URLSearchParams(omitBy(params, isEmpty));
+
+      const requestPath = `/2/users/${userId}/tweets${
+        queryParams.toString() ? `?${queryParams.toString()}` : ''
+      }`;
+
+      const requestData = {
+        url: `${TwitterClient.baseUrl}${requestPath}`,
+        method: 'GET',
+      };
+
+      const authHeader = this.client.oauthClient.toHeader(
+        this.client.oauthClient.authorize(requestData, token)
+      );
+
+      response = await this.client.httpClient.get(requestPath, {
+        headers: { ...authHeader },
+      });
+
+      if (response.data.meta.result_count > 0) {
+        response.data.data.forEach((tweet: IJSONObject) => {
+          if (!lastInternalId || Number(tweet.id) > Number(lastInternalId)) {
+            tweets.push(tweet);
+          } else {
+            return;
+          }
+        });
+      }
+    } while (response.data.meta.next_token && lastInternalId);
+
+    if (response.data?.errors) {
+      const errorMessages = response.data.errors
+        .map((error: IJSONObject) => error.detail)
+        .join(' ');
+
+      throw new Error(
+        `Error occured while fetching user data: ${errorMessages}`
+      );
+    }
+
+    return tweets;
+  }
+}

packages/backend/src/apps/twitter/client/endpoints/oauth-request-token.ts

@@ -0,0 +1,42 @@
+import { IJSONObject } from '@automatisch/types';
+import TwitterClient from '../index';
+
+export default class OAuthRequestToken {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(callbackUrl: string) {
+    try {
+      const requestData = {
+        url: `${TwitterClient.baseUrl}/oauth/request_token`,
+        method: 'POST',
+        data: { oauth_callback: callbackUrl },
+      };
+
+      const authHeader = this.client.oauthClient.toHeader(
+        this.client.oauthClient.authorize(requestData)
+      );
+
+      const response = await this.client.httpClient.post(
+        `/oauth/request_token`,
+        null,
+        {
+          headers: { ...authHeader },
+        }
+      );
+
+      return response;
+    } catch (error) {
+      const errorMessages = error.response.data.errors
+        .map((error: IJSONObject) => error.message)
+        .join(' ');
+
+      throw new Error(
+        `Error occured while verifying credentials: ${errorMessages}`
+      );
+    }
+  }
+}

packages/backend/src/apps/twitter/client/endpoints/search-tweets.ts

@@ -0,0 +1,70 @@
+import { IJSONObject } from '@automatisch/types';
+import { URLSearchParams } from 'url';
+import TwitterClient from '../index';
+import omitBy from 'lodash/omitBy';
+import isEmpty from 'lodash/isEmpty';
+import qs from 'qs';
+
+export default class SearchTweets {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(searchTerm: string, lastInternalId?: string) {
+    const token = {
+      key: this.client.connection.formattedData.accessToken as string,
+      secret: this.client.connection.formattedData.accessSecret as string,
+    };
+
+    let response;
+    const tweets: IJSONObject[] = [];
+
+    do {
+      const params: IJSONObject = {
+        query: searchTerm,
+        since_id: lastInternalId,
+        pagination_token: response?.data?.meta?.next_token,
+      };
+
+      const queryParams = qs.stringify(omitBy(params, isEmpty));
+
+      const requestPath = `/2/tweets/search/recent${
+        queryParams.toString() ? `?${queryParams.toString()}` : ''
+      }`;
+
+      const requestData = {
+        url: `${TwitterClient.baseUrl}${requestPath}`,
+        method: 'GET',
+      };
+
+      const authHeader = this.client.oauthClient.toHeader(
+        this.client.oauthClient.authorize(requestData, token)
+      );
+
+      response = await this.client.httpClient.get(requestPath, {
+        headers: { ...authHeader },
+      });
+
+      console.log(response);
+
+      if (response.data.meta.result_count > 0) {
+        response.data.data.forEach((tweet: IJSONObject) => {
+          if (!lastInternalId || Number(tweet.id) > Number(lastInternalId)) {
+            tweets.push(tweet);
+          } else {
+            return;
+          }
+        });
+      }
+    } while (response.data.meta.next_token && lastInternalId);
+
+    if (response.data?.errors) {
+      const errors = response.data.errors;
+      return { errors, data: tweets };
+    }
+
+    return { data: tweets };
+  }
+}

packages/backend/src/apps/twitter/client/endpoints/verify-access-token.ts

@@ -0,0 +1,20 @@
+import TwitterClient from '../index';
+
+export default class VerifyAccessToken {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run() {
+    try {
+      return await this.client.httpClient.post(
+        `/oauth/access_token?oauth_verifier=${this.client.connection.formattedData.oauthVerifier}&oauth_token=${this.client.connection.formattedData.accessToken}`,
+        null
+      );
+    } catch (error) {
+      throw new Error(error.response.data);
+    }
+  }
+}

packages/backend/src/apps/twitter/client/index.ts

@@ -0,0 +1,64 @@
+import { IFlow, IStep, IConnection } from '@automatisch/types';
+import OAuth from 'oauth-1.0a';
+import crypto from 'crypto';
+import HttpClient from '../../../helpers/http-client';
+import OAuthRequestToken from './endpoints/oauth-request-token';
+import VerifyAccessToken from './endpoints/verify-access-token';
+import GetCurrentUser from './endpoints/get-current-user';
+import GetUserByUsername from './endpoints/get-user-by-username';
+import GetUserTweets from './endpoints/get-user-tweets';
+import CreateTweet from './endpoints/create-tweet';
+import SearchTweets from './endpoints/search-tweets';
+import GetUserFollowers from './endpoints/get-user-followers';
+
+export default class TwitterClient {
+  flow: IFlow;
+  step: IStep;
+  connection: IConnection;
+  oauthClient: OAuth;
+  httpClient: HttpClient;
+
+  oauthRequestToken: OAuthRequestToken;
+  verifyAccessToken: VerifyAccessToken;
+  getCurrentUser: GetCurrentUser;
+  getUserByUsername: GetUserByUsername;
+  getUserTweets: GetUserTweets;
+  createTweet: CreateTweet;
+  searchTweets: SearchTweets;
+  getUserFollowers: GetUserFollowers;
+
+  static baseUrl = 'https://api.twitter.com';
+
+  constructor(connection: IConnection, flow?: IFlow, step?: IStep) {
+    this.connection = connection;
+    this.flow = flow;
+    this.step = step;
+
+    this.httpClient = new HttpClient({ baseURL: TwitterClient.baseUrl });
+
+    const consumerData = {
+      key: this.connection.formattedData.consumerKey as string,
+      secret: this.connection.formattedData.consumerSecret as string,
+    };
+
+    this.oauthClient = new OAuth({
+      consumer: consumerData,
+      signature_method: 'HMAC-SHA1',
+      hash_function(base_string, key) {
+        return crypto
+          .createHmac('sha1', key)
+          .update(base_string)
+          .digest('base64');
+      },
+    });
+
+    this.oauthRequestToken = new OAuthRequestToken(this);
+    this.verifyAccessToken = new VerifyAccessToken(this);
+    this.getCurrentUser = new GetCurrentUser(this);
+    this.getUserByUsername = new GetUserByUsername(this);
+    this.getUserTweets = new GetUserTweets(this);
+    this.createTweet = new CreateTweet(this);
+    this.searchTweets = new SearchTweets(this);
+    this.getUserFollowers = new GetUserFollowers(this);
+  }
+}

packages/backend/src/apps/twitter/index.ts

@@ -1,25 +1,27 @@
 import {
   IService,
   IAuthentication,
-  IApp,
-  IJSONObject,
+  IFlow,
+  IStep,
+  IConnection,
 } from '@automatisch/types';
 import Authentication from './authentication';
 import Triggers from './triggers';
 import Actions from './actions';
+import TwitterClient from './client';
 
 export default class Twitter implements IService {
+  client: TwitterClient;
+
   authenticationClient: IAuthentication;
   triggers: Triggers;
   actions: Actions;
 
-  constructor(
-    appData: IApp,
-    connectionData: IJSONObject,
-    parameters: IJSONObject
-  ) {
-    this.authenticationClient = new Authentication(appData, connectionData);
-    this.triggers = new Triggers(connectionData, parameters);
-    this.actions = new Actions(connectionData, parameters);
+  constructor(connection: IConnection, flow?: IFlow, step?: IStep) {
+    this.client = new TwitterClient(connection, flow, step);
+
+    this.authenticationClient = new Authentication(this.client);
+    this.triggers = new Triggers(this.client);
+    this.actions = new Actions(this.client);
   }
 }

packages/backend/src/apps/twitter/info.json

@@ -3,7 +3,9 @@
   "key": "twitter",
   "iconUrl": "{BASE_URL}/apps/twitter/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/twitter",
+  "authDocUrl": "https://automatisch.io/docs/connections/twitter",
   "primaryColor": "2DAAE1",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "oAuthRedirectUrl",
@@ -14,31 +16,28 @@
       "value": "{WEB_APP_URL}/app/twitter/connections/add",
       "placeholder": null,
       "description": "When asked to input an OAuth callback or redirect URL in Twitter OAuth, enter the URL above.",
-      "docUrl": "https://automatisch.io/docs/twitter#oauth-redirect-url",
       "clickToCopy": true
     },
     {
       "key": "consumerKey",
-      "label": "Consumer Key",
+      "label": "API Key",
       "type": "string",
       "required": true,
       "readOnly": false,
       "value": null,
       "placeholder": null,
       "description": null,
-      "docUrl": "https://automatisch.io/docs/twitter#consumer-key",
       "clickToCopy": false
     },
     {
       "key": "consumerSecret",
-      "label": "Consumer Secret",
+      "label": "API Secret",
       "type": "string",
       "required": true,
       "readOnly": false,
       "value": null,
       "placeholder": null,
       "description": null,
-      "docUrl": "https://automatisch.io/docs/twitter#consumer-secret",
       "clickToCopy": false
     }
   ],
@@ -217,13 +216,14 @@
   ],
   "triggers": [
     {
-      "name": "My Tweet",
-      "key": "myTweet",
+      "name": "My Tweets",
+      "key": "myTweets",
+      "pollInterval": 15,
       "description": "Will be triggered when you tweet something new.",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "testStep",
@@ -232,13 +232,14 @@
       ]
     },
     {
-      "name": "User Tweet",
-      "key": "userTweet",
+      "name": "User Tweets",
+      "key": "userTweets",
+      "pollInterval": 15,
       "description": "Will be triggered when a specific user tweet something new.",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -259,13 +260,14 @@
       ]
     },
     {
-      "name": "Search Tweet",
-      "key": "searchTweet",
+      "name": "Search Tweets",
+      "key": "searchTweets",
+      "pollInterval": 15,
       "description": "Will be triggered when any user tweet something containing a specific keyword, phrase, username or hashtag.",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseTrigger",
@@ -284,6 +286,22 @@
           "name": "Test trigger"
         }
       ]
+    },
+    {
+      "name": "New follower of me",
+      "key": "myFollowers",
+      "pollInterval": 15,
+      "description": "Will be triggered when you have a new follower.",
+      "substeps": [
+        {
+          "key": "chooseConnection",
+          "name": "Choose connection"
+        },
+        {
+          "key": "testStep",
+          "name": "Test trigger"
+        }
+      ]
     }
   ],
   "actions": [
@@ -293,8 +311,8 @@
       "description": "Will create a tweet.",
       "substeps": [
         {
-          "key": "chooseAccount",
-          "name": "Choose account"
+          "key": "chooseConnection",
+          "name": "Choose connection"
         },
         {
           "key": "chooseAction",

packages/backend/src/apps/twitter/triggers.ts

@@ -1,13 +1,21 @@
-import { IJSONObject } from '@automatisch/types';
-import MyTweet from './triggers/my-tweet';
-import SearchTweet from './triggers/search-tweet';
+import TwitterClient from './client';
+import UserTweets from './triggers/user-tweets';
+import SearchTweets from './triggers/search-tweets';
+import MyTweets from './triggers/my-tweets';
+import MyFollowers from './triggers/my-followers';
 
 export default class Triggers {
-  myTweet: MyTweet;
-  searchTweet: SearchTweet;
+  client: TwitterClient;
+  userTweets: UserTweets;
+  searchTweets: SearchTweets;
+  myTweets: MyTweets;
+  myFollowers: MyFollowers;
 
-  constructor(connectionData: IJSONObject, parameters: IJSONObject) {
-    this.myTweet = new MyTweet(connectionData);
-    this.searchTweet = new SearchTweet(connectionData, parameters);
+  constructor(client: TwitterClient) {
+    this.client = client;
+    this.userTweets = new UserTweets(client);
+    this.searchTweets = new SearchTweets(client);
+    this.myTweets = new MyTweets(client);
+    this.myFollowers = new MyFollowers(client);
   }
 }

packages/backend/src/apps/twitter/triggers/my-followers.ts

@@ -0,0 +1,28 @@
+import TwitterClient from '../client';
+
+export default class MyFollowers {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(lastInternalId: string) {
+    return this.getFollowers(lastInternalId);
+  }
+
+  async testRun() {
+    return this.getFollowers();
+  }
+
+  async getFollowers(lastInternalId?: string) {
+    const { username } = await this.client.getCurrentUser.run();
+    const user = await this.client.getUserByUsername.run(username as string);
+
+    const tweets = await this.client.getUserFollowers.run(
+      user.id,
+      lastInternalId
+    );
+    return tweets;
+  }
+}

packages/backend/src/apps/twitter/triggers/my-tweet.ts

@@ -1,25 +0,0 @@
-import TwitterApi, { TwitterApiTokens } from 'twitter-api-v2';
-import { IJSONObject } from '@automatisch/types';
-
-export default class MyTweet {
-  client: TwitterApi;
-
-  constructor(connectionData: IJSONObject) {
-    this.client = new TwitterApi({
-      appKey: connectionData.consumerKey,
-      appSecret: connectionData.consumerSecret,
-      accessToken: connectionData.accessToken,
-      accessSecret: connectionData.accessSecret,
-    } as TwitterApiTokens);
-  }
-
-  async run() {
-    const response = await this.client.currentUser();
-    const username = response.screen_name;
-
-    const userTimeline = await this.client.v1.userTimelineByUsername(username);
-    const fetchedTweets = userTimeline.tweets;
-
-    return fetchedTweets;
-  }
-}

packages/backend/src/apps/twitter/triggers/my-tweets.ts

@@ -0,0 +1,25 @@
+import TwitterClient from '../client';
+
+export default class MyTweets {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(lastInternalId: string) {
+    return this.getTweets(lastInternalId);
+  }
+
+  async testRun() {
+    return this.getTweets();
+  }
+
+  async getTweets(lastInternalId?: string) {
+    const { username } = await this.client.getCurrentUser.run();
+    const user = await this.client.getUserByUsername.run(username as string);
+
+    const tweets = await this.client.getUserTweets.run(user.id, lastInternalId);
+    return tweets;
+  }
+}

packages/backend/src/apps/twitter/triggers/search-tweet.ts

@@ -1,58 +0,0 @@
-import TwitterApi, { TwitterApiTokens } from 'twitter-api-v2';
-import { IJSONObject } from '@automatisch/types';
-
-export default class SearchTweet {
-  client: TwitterApi;
-  parameters: IJSONObject;
-
-  constructor(connectionData: IJSONObject, parameters: IJSONObject) {
-    this.client = new TwitterApi({
-      appKey: connectionData.consumerKey,
-      appSecret: connectionData.consumerSecret,
-      accessToken: connectionData.accessToken,
-      accessSecret: connectionData.accessSecret,
-    } as TwitterApiTokens);
-
-    this.parameters = parameters;
-  }
-
-  async run(startTime: Date) {
-    const tweets = [];
-
-    const response = await this.client.v2.search(
-      this.parameters.searchTerm as string,
-      {
-        max_results: 50,
-        'tweet.fields': 'created_at',
-      }
-    );
-
-    for await (const tweet of response.data.data) {
-      if (new Date(tweet.created_at).getTime() <= startTime.getTime()) {
-        break;
-      }
-
-      tweets.push(tweet);
-
-      if (response.data.meta.next_token) {
-        await response.fetchNext();
-      }
-    }
-
-    return tweets;
-  }
-
-  async testRun() {
-    const response = await this.client.v2.search(
-      this.parameters.searchTerm as string,
-      {
-        max_results: 10,
-        'tweet.fields': 'created_at',
-      }
-    );
-
-    const mostRecentTweet = response.data.data[0];
-
-    return [mostRecentTweet];
-  }
-}

packages/backend/src/apps/twitter/triggers/search-tweets.ts

@@ -0,0 +1,26 @@
+import TwitterClient from '../client';
+
+export default class SearchTweets {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(lastInternalId: string) {
+    return this.getTweets(lastInternalId);
+  }
+
+  async testRun() {
+    return this.getTweets();
+  }
+
+  async getTweets(lastInternalId?: string) {
+    const tweets = await this.client.searchTweets.run(
+      this.client.step.parameters.searchTerm as string,
+      lastInternalId
+    );
+
+    return tweets;
+  }
+}

packages/backend/src/apps/twitter/triggers/user-tweets.ts

@@ -0,0 +1,27 @@
+import TwitterClient from '../client';
+
+export default class UserTweets {
+  client: TwitterClient;
+
+  constructor(client: TwitterClient) {
+    this.client = client;
+  }
+
+  async run(lastInternalId: string) {
+    return this.getTweets(lastInternalId);
+  }
+
+  async testRun() {
+    return this.getTweets();
+  }
+
+  async getTweets(lastInternalId?: string) {
+    const user = await this.client.getUserByUsername.run(
+      this.client.step.parameters.username as string
+    );
+
+    const tweets = await this.client.getUserTweets.run(user.id, lastInternalId);
+
+    return tweets;
+  }
+}

packages/backend/src/apps/typeform/authentication.ts

@@ -5,14 +5,12 @@ import type {
   IJSONObject,
 } from '@automatisch/types';
 import { URLSearchParams } from 'url';
-import axios, { AxiosInstance } from 'axios';
+import HttpClient from '../../helpers/http-client';
 
 export default class Authentication implements IAuthentication {
   appData: IApp;
   connectionData: IJSONObject;
-  client: AxiosInstance = axios.create({
-    baseURL: 'https://api.typeform.com',
-  });
+  client: HttpClient;
 
   scope: string[] = [
     'forms:read',
@@ -27,6 +25,7 @@ export default class Authentication implements IAuthentication {
   constructor(appData: IApp, connectionData: IJSONObject) {
     this.connectionData = connectionData;
     this.appData = appData;
+    this.client = new HttpClient({ baseURL: 'https://api.typeform.com' });
   }
 
   get oauthRedirectUrl() {

packages/backend/src/apps/typeform/info.json

@@ -4,6 +4,7 @@
   "iconUrl": "{BASE_URL}/apps/typeform/assets/favicon.svg",
   "docUrl": "https://automatisch.io/docs/typeform",
   "primaryColor": "5865f2",
+  "supportsConnections": true,
   "fields": [
     {
       "key": "oAuthRedirectUrl",

packages/backend/src/config/app.ts

@@ -13,6 +13,7 @@ type AppConfig = {
   postgresHost: string;
   postgresUsername: string;
   postgresPassword?: string;
+  version: string;
   postgresEnableSsl: boolean;
   baseUrl: string;
   encryptionKey: string;
@@ -20,12 +21,14 @@ type AppConfig = {
   serveWebAppSeparately: boolean;
   redisHost: string;
   redisPort: number;
+  enableBullMQDashboard: boolean;
 };
 
 const host = process.env.HOST || 'localhost';
 const protocol = process.env.PROTOCOL || 'http';
 const port = process.env.PORT || '3000';
-const serveWebAppSeparately = process.env.SERVE_WEB_APP_SEPARATELY === 'true' ? true : false;
+const serveWebAppSeparately =
+  process.env.SERVE_WEB_APP_SEPARATELY === 'true' ? true : false;
 
 let webAppUrl = `${protocol}://${host}:${port}`;
 if (serveWebAppSeparately) {
@@ -42,8 +45,9 @@ const appConfig: AppConfig = {
   port,
   appEnv: appEnv,
   isDev: appEnv === 'development',
+  version: process.env.npm_package_version,
   postgresDatabase: process.env.POSTGRES_DATABASE || 'automatisch_development',
-  postgresPort: parseInt(process.env.POSTGRES_PORT|| '5432'),
+  postgresPort: parseInt(process.env.POSTGRES_PORT || '5432'),
   postgresHost: process.env.POSTGRES_HOST || 'localhost',
   postgresUsername:
     process.env.POSTGRES_USERNAME || 'automatisch_development_user',
@@ -54,6 +58,8 @@ const appConfig: AppConfig = {
   serveWebAppSeparately,
   redisHost: process.env.REDIS_HOST || '127.0.0.1',
   redisPort: parseInt(process.env.REDIS_PORT || '6379'),
+  enableBullMQDashboard:
+    process.env.ENABLE_BULLMQ_DASHBOARD === 'true' ? true : false,
   baseUrl,
   webAppUrl,
 };

packages/backend/src/config/database.ts

@@ -1,4 +1,8 @@
 import process from 'process';
+// The following two lines are required to get count values as number.
+// More info: https://github.com/knex/knex/issues/387#issuecomment-51554522
+import pg from 'pg';
+pg.types.setTypeParser(20, 'text', parseInt);
 import knex from 'knex';
 import type { Knex } from 'knex';
 import knexConfig from '../../knexfile';
@@ -8,10 +12,12 @@ export const client: Knex = knex(knexConfig);
 
 const CONNECTION_REFUSED = 'ECONNREFUSED';
 
-client.raw('SELECT 1')
-  .catch((err) => {
-    if (err.code === CONNECTION_REFUSED) {
-      logger.error('Make sure you have installed PostgreSQL and it is running.', err);
-      process.exit();
-    }
-  });
+client.raw('SELECT 1').catch((err) => {
+  if (err.code === CONNECTION_REFUSED) {
+    logger.error(
+      'Make sure you have installed PostgreSQL and it is running.',
+      err
+    );
+    process.exit();
+  }
+});

packages/backend/src/db/migrations/20220727104324_add_draft_column_to_connections.ts

@@ -0,0 +1,13 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+  return knex.schema.table('connections', (table) => {
+    table.boolean('draft').defaultTo(true);
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  return knex.schema.table('connections', (table) => {
+    table.dropColumn('draft');
+  });
+}

packages/backend/src/db/migrations/20220817121241_add_published_at_column_to_flows.ts

@@ -0,0 +1,13 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+  return knex.schema.table('flows', (table) => {
+    table.timestamp('published_at').nullable();
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  return knex.schema.table('flows', (table) => {
+    table.dropColumn('published_at');
+  });
+}

packages/backend/src/db/migrations/20220823171017_add_internal_id_to_executions.ts

@@ -0,0 +1,13 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+  return knex.schema.table('executions', (table) => {
+    table.string('internal_id');
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  return knex.schema.table('executions', (table) => {
+    table.dropColumn('internal_id');
+  });
+}

packages/backend/src/db/migrations/20220904160521_add_raw_error_to_execution_steps.ts

@@ -0,0 +1,13 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+  return knex.schema.table('execution_steps', (table) => {
+    table.jsonb('error_details');
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  return knex.schema.table('execution_steps', (table) => {
+    table.dropColumn('error_details');
+  });
+}

packages/backend/src/graphql/mutations/create-auth-data.ts

@@ -1,5 +1,5 @@
 import Context from '../../types/express/context';
-import App from '../../models/app';
+import axios from 'axios';
 
 type Params = {
   input: {
@@ -20,13 +20,20 @@ const createAuthData = async (
     .throwIfNotFound();
 
   const appClass = (await import(`../../apps/${connection.key}`)).default;
-  const appData = App.findOneByKey(connection.key);
 
-  if (!connection.formattedData) { return null; }
+  if (!connection.formattedData) {
+    return null;
+  }
 
-  const appInstance = new appClass(appData, connection.formattedData);
+  const appInstance = new appClass(connection);
   const authLink = await appInstance.authenticationClient.createAuthData();
 
+  try {
+    await axios.get(authLink.url);
+  } catch (error) {
+    throw new Error('Error occured while creating authorization URL!');
+  }
+
   await connection.$query().patch({
     formattedData: {
       ...connection.formattedData,

packages/backend/src/graphql/mutations/create-connection.ts

@@ -13,19 +13,12 @@ const createConnection = async (
   params: Params,
   context: Context
 ) => {
-  const app = App.findOneByKey(params.input.key);
+  App.findOneByKey(params.input.key);
 
-  const connection = await context.currentUser
-    .$relatedQuery('connections')
-    .insert({
-      key: params.input.key,
-      formattedData: params.input.formattedData,
-    });
-
-  return {
-    ...connection,
-    app,
-  };
+  return await context.currentUser.$relatedQuery('connections').insert({
+    key: params.input.key,
+    formattedData: params.input.formattedData,
+  });
 };
 
 export default createConnection;

packages/backend/src/graphql/mutations/create-flow.ts

@@ -4,6 +4,7 @@ import Context from '../../types/express/context';
 type Params = {
   input: {
     triggerAppKey: string;
+    connectionId: string;
   };
 };
 
@@ -12,17 +13,32 @@ const createFlow = async (
   params: Params,
   context: Context
 ) => {
+  const connectionId = params?.input?.connectionId;
   const appKey = params?.input?.triggerAppKey;
 
   const flow = await context.currentUser.$relatedQuery('flows').insert({
     name: 'Name your flow',
   });
 
+  if (connectionId) {
+    await context.currentUser
+      .$relatedQuery('connections')
+      .findById(connectionId)
+      .throwIfNotFound();
+  }
+
   await Step.query().insert({
     flowId: flow.id,
     type: 'trigger',
     position: 1,
     appKey,
+    connectionId
+  });
+
+  await Step.query().insert({
+    flowId: flow.id,
+    type: 'action',
+    position: 2
   });
 
   return flow;

packages/backend/src/graphql/mutations/update-flow-status.ts

@@ -36,9 +36,13 @@ const updateFlowStatus = async (
   const interval = trigger.interval;
   const repeatOptions = {
     cron: interval || EVERY_15_MINUTES_CRON,
-  }
+  };
 
   if (flow.active) {
+    flow = await flow.$query().patchAndFetch({
+      published_at: new Date().toISOString(),
+    });
+
     await processorQueue.add(
       JOB_NAME,
       { flowId: flow.id },
@@ -49,7 +53,7 @@ const updateFlowStatus = async (
     );
   } else {
     const repeatableJobs = await processorQueue.getRepeatableJobs();
-    const job = repeatableJobs.find(job => job.id === flow.id);
+    const job = repeatableJobs.find((job) => job.id === flow.id);
 
     await processorQueue.removeRepeatableByKey(job.key);
   }

packages/backend/src/graphql/mutations/verify-connection.ts

@@ -20,9 +20,9 @@ const verifyConnection = async (
     .throwIfNotFound();
 
   const appClass = (await import(`../../apps/${connection.key}`)).default;
-  const appData = App.findOneByKey(connection.key);
+  const app = App.findOneByKey(connection.key);
 
-  const appInstance = new appClass(appData, connection.formattedData);
+  const appInstance = new appClass(connection);
   const verifiedCredentials =
     await appInstance.authenticationClient.verifyCredentials();
 
@@ -32,9 +32,13 @@ const verifyConnection = async (
       ...verifiedCredentials,
     },
     verified: true,
+    draft: false,
   });
 
-  return connection;
+  return {
+    ...connection,
+    app,
+  };
 };
 
 export default verifyConnection;

packages/backend/src/graphql/queries/get-app-connections.ts

@@ -1,27 +0,0 @@
-import App from '../../models/app';
-import Context from '../../types/express/context';
-
-type Params = {
-  key: string;
-};
-
-const getAppConnections = async (
-  _parent: unknown,
-  params: Params,
-  context: Context
-) => {
-  const app = App.findOneByKey(params.key);
-
-  const connections = await context.currentUser
-    .$relatedQuery('connections')
-    .where({
-      key: params.key,
-    });
-
-  return connections.map((connection) => ({
-    ...connection,
-    app,
-  }));
-};
-
-export default getAppConnections;

packages/backend/src/graphql/queries/get-app.ts

@@ -11,9 +11,15 @@ const getApp = async (_parent: unknown, params: Params, context: Context) => {
   if (context.currentUser) {
     const connections = await context.currentUser
       .$relatedQuery('connections')
+      .select('connections.*')
+      .fullOuterJoinRelated('steps')
       .where({
-        key: params.key,
-      });
+        'connections.key': params.key,
+        'connections.draft': false,
+      })
+      .countDistinct('steps.flow_id as flowCount')
+      .groupBy('connections.id')
+      .orderBy('created_at', 'desc');
 
     return {
       ...app,

packages/backend/src/graphql/queries/get-connected-apps.ts

@@ -16,22 +16,42 @@ const getConnectedApps = async (
   const connections = await context.currentUser
     .$relatedQuery('connections')
     .select('connections.key')
+    .where({ draft: false })
     .count('connections.id as count')
-    .where({ verified: true })
     .groupBy('connections.key');
 
+  const flows = await context.currentUser
+    .$relatedQuery('flows')
+    .withGraphJoined('steps')
+    .orderBy('created_at', 'desc');
+
+  const duplicatedUsedApps = flows
+    .map((flow) => flow.steps.map((step) => step.appKey))
+    .flat()
+    .filter(Boolean);
+
   const connectionKeys = connections.map((connection) => connection.key);
+  const usedApps = [...new Set([...duplicatedUsedApps, ...connectionKeys])];
 
   apps = apps
-    .filter((app: IApp) => connectionKeys.includes(app.key))
+    .filter((app: IApp) => {
+      return usedApps.includes(app.key);
+    })
     .map((app: IApp) => {
       const connection = connections.find(
         (connection) => (connection as IConnection).key === app.key
       );
 
-      if (connection) {
-        app.connectionCount = connection.count;
-      }
+      app.connectionCount = connection?.count || 0;
+      app.flowCount = 0;
+
+      flows.forEach((flow) => {
+        const usedFlow = flow.steps.find((step) => step.appKey === app.key);
+
+        if (usedFlow) {
+          app.flowCount += 1;
+        }
+      });
 
       return app;
     });

packages/backend/src/graphql/queries/get-data.ts

@@ -1,5 +1,4 @@
 import { IJSONObject } from '@automatisch/types';
-import App from '../../models/app';
 import Context from '../../types/express/context';
 
 type Params = {
@@ -11,7 +10,10 @@ type Params = {
 const getData = async (_parent: unknown, params: Params, context: Context) => {
   const step = await context.currentUser
     .$relatedQuery('steps')
-    .withGraphFetched('connection')
+    .withGraphFetched({
+      connection: true,
+      flow: true,
+    })
     .findById(params.stepId);
 
   if (!step) return null;
@@ -20,10 +22,9 @@ const getData = async (_parent: unknown, params: Params, context: Context) => {
 
   if (!connection || !step.appKey) return null;
 
-  const appData = App.findOneByKey(step.appKey);
   const AppClass = (await import(`../../apps/${step.appKey}`)).default;
+  const appInstance = new AppClass(connection, step.flow, step);
 
-  const appInstance = new AppClass(appData, connection.formattedData, params.parameters);
   const command = appInstance.data[params.key];
   const fetchedData = await command.run();
 

packages/backend/src/graphql/queries/get-execution.ts

@@ -0,0 +1,25 @@
+import Context from '../../types/express/context';
+
+type Params = {
+  executionId: string;
+};
+
+const getExecution = async (
+  _parent: unknown,
+  params: Params,
+  context: Context
+) => {
+  const execution = await context.currentUser
+    .$relatedQuery('executions')
+    .withGraphFetched({
+      flow: {
+        steps: true
+      }
+    })
+    .findById(params.executionId)
+    .throwIfNotFound();
+
+  return execution;
+};
+
+export default getExecution;

packages/backend/src/graphql/queries/get-executions.ts

@@ -13,7 +13,11 @@ const getExecutions = async (
 ) => {
   const executions = context.currentUser
     .$relatedQuery('executions')
-    .withGraphFetched('flow')
+    .withGraphFetched({
+      flow: {
+        steps: true
+      }
+    })
     .orderBy('created_at', 'desc');
 
   return paginate(executions, params.limit, params.offset);

packages/backend/src/graphql/queries/get-flows.ts

@@ -1,22 +1,42 @@
 import Context from '../../types/express/context';
+import paginate from '../../helpers/pagination';
 
 type Params = {
   appKey?: string;
+  connectionId?: string;
+  name?: string;
+  limit: number;
+  offset: number;
 };
 
 const getFlows = async (_parent: unknown, params: Params, context: Context) => {
   const flowsQuery = context.currentUser
     .$relatedQuery('flows')
-    .withGraphJoined('[steps.[connection]]')
-    .orderBy('created_at', 'desc');
+    .joinRelated({
+      steps: true
+    })
+    .withGraphFetched({
+      steps: {
+        connection: true
+      }
+    })
+    .where((builder) => {
+      if (params.connectionId) {
+        builder.where('steps.connection_id', params.connectionId);
+      }
 
-  if (params.appKey) {
-    flowsQuery.where('steps.app_key', params.appKey);
-  }
+      if (params.name) {
+        builder.where('flows.name', 'ilike', `%${params.name}%`);
+      }
 
-  const flows = await flowsQuery;
+      if (params.appKey) {
+        builder.where('steps.app_key', params.appKey);
+      }
+    })
+    .groupBy('flows.id')
+    .orderBy('updated_at', 'desc');
 
-  return flows;
+  return paginate(flowsQuery, params.limit, params.offset);
 };
 
 export default getFlows;

packages/backend/src/graphql/queries/healthcheck.ts

@@ -0,0 +1,9 @@
+import appConfig from '../../config/app';
+
+const healthcheck = () => {
+  return {
+    version: appConfig.version,
+  }
+};
+
+export default healthcheck;

packages/backend/src/graphql/queries/test-connection.ts

@@ -1,5 +1,4 @@
 import Context from '../../types/express/context';
-import App from '../../models/app';
 
 type Params = {
   id: string;
@@ -19,9 +18,8 @@ const testConnection = async (
     .throwIfNotFound();
 
   const appClass = (await import(`../../apps/${connection.key}`)).default;
-  const appData = App.findOneByKey(connection.key);
+  const appInstance = new appClass(connection);
 
-  const appInstance = new appClass(appData, connection.formattedData);
   const isStillVerified =
     await appInstance.authenticationClient.isStillVerified();
 

packages/backend/src/graphql/query-resolvers.ts

@@ -1,29 +1,31 @@
 import getApps from './queries/get-apps';
 import getApp from './queries/get-app';
 import getConnectedApps from './queries/get-connected-apps';
-import getAppConnections from './queries/get-app-connections';
 import testConnection from './queries/test-connection';
 import getFlow from './queries/get-flow';
 import getFlows from './queries/get-flows';
 import getStepWithTestExecutions from './queries/get-step-with-test-executions';
+import getExecution from './queries/get-execution';
 import getExecutions from './queries/get-executions';
 import getExecutionSteps from './queries/get-execution-steps';
 import getData from './queries/get-data';
 import getCurrentUser from './queries/get-current-user';
+import healthcheck from './queries/healthcheck';
 
 const queryResolvers = {
   getApps,
   getApp,
   getConnectedApps,
-  getAppConnections,
   testConnection,
   getFlow,
   getFlows,
   getStepWithTestExecutions,
+  getExecution,
   getExecutions,
   getExecutionSteps,
   getData,
   getCurrentUser,
+  healthcheck,
 };
 
 export default queryResolvers;

packages/backend/src/graphql/schema.graphql

@@ -2,11 +2,17 @@ type Query {
   getApps(name: String, onlyWithTriggers: Boolean): [App]
   getApp(key: AvailableAppsEnumType!): App
   getConnectedApps(name: String): [App]
-  getAppConnections(key: AvailableAppsEnumType!): [Connection]
   testConnection(id: String!): Connection
   getFlow(id: String!): Flow
-  getFlows(appKey: String): [Flow]
+  getFlows(
+    limit: Int!
+    offset: Int!
+    appKey: String
+    connectionId: String
+    name: String
+  ): FlowConnection
   getStepWithTestExecutions(stepId: String!): [Step]
+  getExecution(executionId: String!): Execution
   getExecutions(limit: Int!, offset: Int!): ExecutionConnection
   getExecutionSteps(
     executionId: String!
@@ -15,6 +21,7 @@ type Query {
   ): ExecutionStepConnection
   getData(stepId: String!, key: String!, parameters: JSONObject): JSONObject
   getCurrentUser: User
+  healthcheck: AppHealth
 }
 
 type Mutation {
@@ -66,6 +73,7 @@ type ActionSubstepArgument {
   description: String
   required: Boolean
   variables: Boolean
+  options: [ActionSubstepArgumentOption]
   source: ActionSubstepArgumentSource
   dependsOn: [String]
 }
@@ -76,6 +84,11 @@ type ActionSubstepArgumentSource {
   arguments: [ActionSubstepArgumentSourceArgument]
 }
 
+type ActionSubstepArgumentOption {
+  label: String
+  value: JSONObject
+}
+
 type ActionSubstepArgumentSourceArgument {
   name: String
   value: String
@@ -85,9 +98,12 @@ type App {
   name: String
   key: String
   connectionCount: Int
+  flowCount: Int
   iconUrl: String
   docUrl: String
+  authDocUrl: String
   primaryColor: String
+  supportsConnections: Boolean
   fields: [Field]
   authenticationSteps: [AuthenticationStep]
   reconnectionSteps: [ReconnectionStep]
@@ -142,6 +158,7 @@ enum AvailableAppsEnumType {
   twitter
   typeform
   slack
+  scheduler
 }
 
 type Connection {
@@ -151,6 +168,7 @@ type Connection {
   verified: Boolean
   app: App
   createdAt: String
+  flowCount: Int
 }
 
 type ConnectionData {
@@ -187,11 +205,22 @@ type Field {
   clickToCopy: Boolean
 }
 
+type FlowConnection {
+  edges: [FlowEdge]
+  pageInfo: PageInfo
+}
+
+type FlowEdge {
+  node: Flow
+}
+
 type Flow {
   id: String
   name: String
   active: Boolean
   steps: [Step]
+  createdAt: String
+  updatedAt: String
 }
 
 type Execution {
@@ -230,6 +259,7 @@ input DeleteConnectionInput {
 
 input CreateFlowInput {
   triggerAppKey: String
+  connectionId: String
 }
 
 input UpdateFlowInput {
@@ -319,6 +349,7 @@ type Step {
   previousStepId: String
   key: String
   appKey: String
+  iconUrl: String
   type: StepEnumType
   parameters: JSONObject
   connection: Connection
@@ -356,6 +387,7 @@ type Trigger {
   name: String
   key: String
   description: String
+  pollInterval: Int
   substeps: [TriggerSubstep]
 }
 
@@ -423,6 +455,10 @@ type ExecutionStepConnection {
   pageInfo: PageInfo
 }
 
+type AppHealth {
+  version: String
+}
+
 schema {
   query: Query
   mutation: Mutation

packages/backend/src/helpers/authentication.ts

@@ -24,6 +24,7 @@ const authentication = shield(
   {
     Query: {
       '*': isAuthenticated,
+      healthcheck: allow,
     },
     Mutation: {
       '*': isAuthenticated,

packages/backend/src/helpers/http-client/index.ts

@@ -0,0 +1,21 @@
+import axios, { AxiosInstance } from 'axios';
+import { IJSONObject, IHttpClientParams } from '@automatisch/types';
+
+export default class HttpClient {
+  instance: AxiosInstance;
+
+  constructor(params: IHttpClientParams) {
+    this.instance = axios.create({
+      baseURL: params.baseURL,
+      validateStatus: () => true,
+    });
+  }
+
+  async get(path: string, options?: IJSONObject) {
+    return await this.instance.get(path, options);
+  }
+
+  async post(path: string, body: IJSONObject | string, options?: IJSONObject) {
+    return await this.instance.post(path, body, options);
+  }
+}

packages/backend/src/helpers/telemetry/index.ts

@@ -125,7 +125,7 @@ class Telemetry {
 
   diagnosticInfo() {
     this.track('diagnosticInfo', {
-      automatischVersion: process.env.npm_package_version,
+      automatischVersion: appConfig.version,
       serveWebAppSeparately: appConfig.serveWebAppSeparately,
       operatingSystem: {
         type: os.type(),
@@ -139,7 +139,7 @@ class Telemetry {
       },
     });
 
-    setTimeout(this.diagnosticInfo, SIX_HOURS_IN_MILLISECONDS);
+    setTimeout(() => this.diagnosticInfo(), SIX_HOURS_IN_MILLISECONDS);
   }
 }
 

packages/backend/src/models/app.ts

@@ -7,10 +7,15 @@ class App {
   static folderPath = join(__dirname, '../apps');
   static list = fs.readdirSync(this.folderPath);
 
-  static findAll(name?: string): IApp[] {
-    if (!name) return this.list.map((name) => this.findOneByName(name));
+  // Temporaryly restrict the apps we expose until
+  // their actions/triggers are implemented!
+  static temporaryList = ['slack', 'twitter', 'scheduler'];
 
-    return this.list
+  static findAll(name?: string): IApp[] {
+    if (!name)
+      return this.temporaryList.map((name) => this.findOneByName(name));
+
+    return this.temporaryList
       .filter((app) => app.includes(name.toLowerCase()))
       .map((name) => this.findOneByName(name));
   }

packages/backend/src/models/base.ts

@@ -31,9 +31,9 @@ class Base extends Model {
   }
 
   async $beforeUpdate(opt: ModelOptions, queryContext: QueryContext): Promise<void> {
-    await super.$beforeUpdate(opt, queryContext);
-
     this.updatedAt = new Date().toISOString();
+
+    await super.$beforeUpdate(opt, queryContext);
   }
 }
 

packages/backend/src/models/connection.ts

@@ -3,6 +3,8 @@ import type { RelationMappings } from 'objection';
 import { AES, enc } from 'crypto-js';
 import Base from './base';
 import User from './user';
+import Step from './step';
+import App from './app';
 import appConfig from '../config/app';
 import { IJSONObject } from '@automatisch/types';
 import Telemetry from '../helpers/telemetry';
@@ -14,7 +16,9 @@ class Connection extends Base {
   formattedData?: IJSONObject;
   userId!: string;
   verified = false;
+  draft: boolean;
   count?: number;
+  flowCount?: number;
 
   static tableName = 'connections';
 
@@ -29,6 +33,7 @@ class Connection extends Base {
       formattedData: { type: 'object' },
       userId: { type: 'string', format: 'uuid' },
       verified: { type: 'boolean' },
+      draft: { type: 'boolean' },
     },
   };
 
@@ -41,8 +46,20 @@ class Connection extends Base {
         to: 'users.id',
       },
     },
+    steps: {
+      relation: Base.HasManyRelation,
+      modelClass: Step,
+      join: {
+        from: 'connections.id',
+        to: 'steps.connection_id',
+      },
+    },
   });
 
+  get appData() {
+    return App.findOneByKey(this.key);
+  }
+
   encryptData(): void {
     if (!this.eligibleForEncryption()) return;
 

packages/backend/src/models/execution-step.ts

@@ -10,6 +10,7 @@ class ExecutionStep extends Base {
   stepId!: string;
   dataIn!: Record<string, unknown>;
   dataOut!: Record<string, unknown>;
+  errorDetails: Record<string, unknown>;
   status = 'failure';
   step: Step;
 
@@ -25,6 +26,7 @@ class ExecutionStep extends Base {
       dataIn: { type: 'object' },
       dataOut: { type: 'object' },
       status: { type: 'string', enum: ['success', 'failure'] },
+      errorDetails: { type: ['object', 'null'] },
     },
   };
 

packages/backend/src/models/execution.ts

@@ -8,6 +8,7 @@ class Execution extends Base {
   id!: string;
   flowId!: string;
   testRun = false;
+  internalId: string;
   executionSteps: ExecutionStep[] = [];
 
   static tableName = 'executions';
@@ -19,6 +20,7 @@ class Execution extends Base {
       id: { type: 'string', format: 'uuid' },
       flowId: { type: 'string', format: 'uuid' },
       testRun: { type: 'boolean' },
+      internalId: { type: 'string' },
     },
   };
 

packages/backend/src/models/flow.ts

@@ -1,5 +1,5 @@
 import { ValidationError } from 'objection';
-import type { ModelOptions, QueryContext } from 'objection';
+import type { ModelOptions, QueryContext, QueryBuilder } from 'objection';
 import Base from './base';
 import Step from './step';
 import Execution from './execution';
@@ -9,17 +9,19 @@ class Flow extends Base {
   id!: string;
   name!: string;
   userId!: string;
-  active = false;
+  active: boolean;
   steps?: [Step];
+  published_at: string;
 
   static tableName = 'flows';
 
   static jsonSchema = {
     type: 'object',
+    required: ['name'],
 
     properties: {
       id: { type: 'string', format: 'uuid' },
-      name: { type: 'string' },
+      name: { type: 'string', minLength: 1 },
       userId: { type: 'string', format: 'uuid' },
       active: { type: 'boolean' },
     },
@@ -33,6 +35,9 @@ class Flow extends Base {
         from: 'flows.id',
         to: 'steps.flow_id',
       },
+      filter(builder: QueryBuilder<Step>) {
+        builder.orderBy('position', 'asc');
+      },
     },
     executions: {
       relation: Base.HasManyRelation,
@@ -44,7 +49,20 @@ class Flow extends Base {
     },
   });
 
-  async $beforeUpdate(opt: ModelOptions): Promise<void> {
+  async lastInternalId() {
+    const lastExecution = await this.$relatedQuery('executions')
+      .orderBy('created_at', 'desc')
+      .first();
+
+    return lastExecution ? (lastExecution as Execution).internalId : null;
+  }
+
+  async $beforeUpdate(
+    opt: ModelOptions,
+    queryContext: QueryContext
+  ): Promise<void> {
+    await super.$beforeUpdate(opt, queryContext);
+
     if (!this.active) return;
 
     const oldFlow = opt.old as Flow;

packages/backend/src/models/step.ts

@@ -6,6 +6,7 @@ import Connection from './connection';
 import ExecutionStep from './execution-step';
 import type { IStep } from '@automatisch/types';
 import Telemetry from '../helpers/telemetry';
+import appConfig from '../config/app';
 
 class Step extends Base {
   id!: string;
@@ -40,6 +41,10 @@ class Step extends Base {
     },
   };
 
+  static get virtualAttributes() {
+    return ['iconUrl'];
+  }
+
   static relationMappings = () => ({
     flow: {
       relation: Base.BelongsToOneRelation,
@@ -67,6 +72,16 @@ class Step extends Base {
     },
   });
 
+  get iconUrl() {
+    if (!this.appKey) return null;
+
+    return `${appConfig.baseUrl}/apps/${this.appKey}/assets/favicon.svg`;
+  }
+
+  get appData() {
+    return App.findOneByKey(this.appKey);
+  }
+
   async $afterInsert(queryContext: QueryContext) {
     await super.$afterInsert(queryContext);
     Telemetry.stepCreated(this);
@@ -84,15 +99,13 @@ class Step extends Base {
   async getTrigger() {
     if (!this.isTrigger) return null;
 
-    const { appKey, connection, key, parameters = {} } = this;
+    const { appKey, key } = this;
+
+    const connection = await this.$relatedQuery('connection');
+    const flow = await this.$relatedQuery('flow');
 
-    const appData = App.findOneByKey(appKey);
     const AppClass = (await import(`../apps/${appKey}`)).default;
-    const appInstance = new AppClass(
-      appData,
-      connection?.formattedData,
-      parameters,
-    );
+    const appInstance = new AppClass(connection, flow, this);
     const command = appInstance.triggers[key];
 
     return command;

packages/backend/src/queues/processor.ts

@@ -10,7 +10,11 @@ const redisConnection = {
 };
 
 const processorQueue = new Queue('processor', redisConnection);
-new QueueScheduler('processor', redisConnection);
+const queueScheduler = new QueueScheduler('processor', redisConnection);
+
+process.on('SIGTERM', async () => {
+  await queueScheduler.close();
+});
 
 processorQueue.on('error', (err) => {
   if ((err as any).code === CONNECTION_REFUSED) {

packages/backend/src/services/processor.ts

@@ -1,9 +1,9 @@
 import get from 'lodash.get';
-import App from '../models/app';
 import Flow from '../models/flow';
 import Step from '../models/step';
 import Execution from '../models/execution';
 import ExecutionStep from '../models/execution-step';
+import { IJSONObject } from '@automatisch/types';
 
 type ExecutionSteps = Record<string, ExecutionStep>;
 
@@ -33,18 +33,44 @@ class Processor {
 
     const triggerStep = steps.find((step) => step.type === 'trigger');
     // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-    let initialTriggerData = await this.getInitialTriggerData(triggerStep!);
+    const initialTriggerData = await this.getInitialTriggerData(triggerStep!);
+
+    if (initialTriggerData.data.length === 0) {
+      const lastInternalId = await this.flow.lastInternalId();
+
+      const executionData: Partial<Execution> = {
+        flowId: this.flow.id,
+        testRun: this.testRun,
+      };
+
+      if (lastInternalId) {
+        executionData.internalId = lastInternalId;
+      }
+
+      await Execution.query().insert(executionData);
+
+      return;
+    }
 
     if (this.testRun) {
-      initialTriggerData = [initialTriggerData[0]];
+      initialTriggerData.data = [initialTriggerData.data[0]];
+    }
+
+    if (initialTriggerData.data.length > 1) {
+      initialTriggerData.data = initialTriggerData.data.sort(
+        (item: IJSONObject, nextItem: IJSONObject) => {
+          return (item.id as number) - (nextItem.id as number);
+        }
+      );
     }
 
     const executions: Execution[] = [];
 
-    for await (const data of initialTriggerData) {
+    for await (const data of initialTriggerData.data) {
       const execution = await Execution.query().insert({
         flowId: this.flow.id,
         testRun: this.testRun,
+        internalId: data.id,
       });
 
       executions.push(execution);
@@ -56,16 +82,7 @@ class Processor {
       for await (const step of steps) {
         if (!step.appKey) continue;
 
-        const appData = App.findOneByKey(step.appKey);
-
-        const {
-          appKey,
-          connection,
-          key,
-          type,
-          parameters: rawParameters = {},
-          id,
-        } = step;
+        const { appKey, key, type, parameters: rawParameters = {}, id } = step;
 
         const isTrigger = type === 'trigger';
         const AppClass = (await import(`../apps/${appKey}`)).default;
@@ -75,11 +92,9 @@ class Processor {
           priorExecutionSteps
         );
 
-        const appInstance = new AppClass(
-          appData,
-          connection?.formattedData,
-          computedParameters
-        );
+        step.parameters = computedParameters;
+
+        const appInstance = new AppClass(step.connection, this.flow, step);
 
         if (!isTrigger && key) {
           const command = appInstance.actions[key];
@@ -103,6 +118,23 @@ class Processor {
       }
     }
 
+    if (initialTriggerData.errors) {
+      const execution = await Execution.query().insert({
+        flowId: this.flow.id,
+        testRun: this.testRun,
+        internalId: null,
+      });
+
+      await ExecutionStep.query().insert({
+        executionId: execution.id,
+        stepId: steps[0].id,
+        status: 'failure',
+        dataIn: steps[0].parameters,
+        dataOut: null,
+        errorDetails: initialTriggerData.errors,
+      });
+    }
+
     if (!this.testRun) return;
 
     const lastExecutionStepFromFirstExecution = await executions[0]
@@ -114,37 +146,21 @@ class Processor {
   }
 
   async getInitialTriggerData(step: Step) {
-    if (!step.appKey) return null;
+    if (!step.appKey || !step.key) return null;
 
-    const appData = App.findOneByKey(step.appKey);
-    const { appKey, connection, key, parameters: rawParameters = {} } = step;
+    const AppClass = (await import(`../apps/${step.appKey}`)).default;
+    const appInstance = new AppClass(step.connection, this.flow, step);
 
-    if (!key) return null;
+    const command = appInstance.triggers[step.key];
 
-    const AppClass = (await import(`../apps/${appKey}`)).default;
-    const appInstance = new AppClass(
-      appData,
-      connection?.formattedData,
-      rawParameters
-    );
-
-    const lastExecutionStep = await step
-      .$relatedQuery('executionSteps')
-      .orderBy('created_at', 'desc')
-      .first();
-
-    const lastExecutionStepCreatedAt = lastExecutionStep?.createdAt as string;
-    const flow = (await step.$relatedQuery('flow')) as Flow;
-
-    const command = appInstance.triggers[key];
-
-    const startTime = new Date(lastExecutionStepCreatedAt || flow.updatedAt);
     let fetchedData;
 
+    const lastInternalId = await this.flow.lastInternalId();
+
     if (this.testRun) {
-      fetchedData = await command.testRun(startTime);
+      fetchedData = await command.testRun();
     } else {
-      fetchedData = await command.run(startTime);
+      fetchedData = await command.run(lastInternalId);
     }
 
     return fetchedData;
@@ -163,7 +179,10 @@ class Processor {
           .map((part: string) => {
             const isVariable = part.match(Processor.variableRegExp);
             if (isVariable) {
-              const stepIdAndKeyPath = part.replace(/{{step.|}}/g, '') as string;
+              const stepIdAndKeyPath = part.replace(
+                /{{step.|}}/g,
+                ''
+              ) as string;
               const [stepId, ...keyPaths] = stepIdAndKeyPath.split('.');
               const keyPath = keyPaths.join('.');
               const executionStep = executionSteps[stepId.toString() as string];

packages/backend/src/worker.ts

@@ -1,2 +1,2 @@
 import './config/orm';
-import './workers/processor';
+export { worker } from './workers/processor';

packages/backend/src/workers/processor.ts

@@ -4,7 +4,7 @@ import redisConfig from '../config/redis';
 import Flow from '../models/flow';
 import logger from '../helpers/logger';
 
-const worker = new Worker(
+export const worker = new Worker(
   'processor',
   async (job) => {
     const flow = await Flow.query().findById(job.data.flowId).throwIfNotFound();
@@ -24,3 +24,7 @@ worker.on('failed', (job, err) => {
     `JOB ID: ${job.id} - FLOW ID: ${job.data.flowId} has failed with ${err.message}`
   );
 });
+
+process.on('SIGTERM', async () => {
+  await worker.close();
+});

packages/backend/worker.d.ts

@@ -0,0 +1 @@
+export * from './dist/src/worker';

packages/backend/worker.js

@@ -0,0 +1,2 @@
+/* eslint-disable */
+module.exports = require('./dist/src/worker.js');

packages/cli/src/commands/start-worker.ts

@@ -0,0 +1,49 @@
+import { readFileSync } from 'fs';
+import { Command, Flags } from '@oclif/core';
+import * as dotenv from 'dotenv';
+
+export default class StartWorker extends Command {
+  static description = 'Run automatisch worker';
+
+  static flags = {
+    env: Flags.string({
+      multiple: true,
+      char: 'e',
+    }),
+    'env-file': Flags.string(),
+  }
+
+  async prepareEnvVars(): Promise<void> {
+    const { flags } = await this.parse(StartWorker);
+
+    if (flags['env-file']) {
+      const envFile = readFileSync(flags['env-file'], 'utf8');
+      const envConfig = dotenv.parse(envFile);
+
+      for (const key in envConfig) {
+        const value = envConfig[key];
+        process.env[key] = value;
+      }
+    }
+
+    if (flags.env) {
+      for (const env of flags.env) {
+        const [key, value] = env.split('=');
+        process.env[key] = value;
+      }
+    }
+
+    // must serve until more customization is introduced
+    delete process.env.SERVE_WEB_APP_SEPARATELY;
+  }
+
+  async runWorker(): Promise<void> {
+    await import('@automatisch/backend/worker');
+  }
+
+  async run(): Promise<void> {
+    await this.prepareEnvVars();
+
+    await this.runWorker();
+  }
+}

packages/cli/src/commands/start.ts

@@ -13,6 +13,10 @@ export default class Start extends Command {
     'env-file': Flags.string(),
   }
 
+  get isProduction() {
+    return process.env.APP_ENV === 'production';
+  }
+
   async prepareEnvVars(): Promise<void> {
     const { flags } = await this.parse(Start);
 
@@ -38,7 +42,7 @@ export default class Start extends Command {
   }
 
   async createDatabaseAndUser(): Promise<void> {
-    const { utils } = await import('@automatisch/backend/database');
+    const utils = await import('@automatisch/backend/database-utils');
 
     await utils.createDatabaseAndUser(
       process.env.POSTGRES_DATABASE,
@@ -48,7 +52,7 @@ export default class Start extends Command {
 
   async runMigrationsIfNeeded(): Promise<void> {
     const { logger } = await import('@automatisch/backend/logger');
-    const { database } = await import('@automatisch/backend/database');
+    const database = await import('@automatisch/backend/database');
     const migrator = database.client.migrate;
 
     const [, pendingMigrations] = await migrator.list();
@@ -66,7 +70,7 @@ export default class Start extends Command {
   }
 
   async seedUser(): Promise<void> {
-    const { utils } = await import('@automatisch/backend/database');
+    const utils = await import('@automatisch/backend/database-utils');
 
     await utils.createUser();
   }
@@ -78,7 +82,9 @@ export default class Start extends Command {
   async run(): Promise<void> {
     await this.prepareEnvVars();
 
-    await this.createDatabaseAndUser();
+    if (!this.isProduction) {
+      await this.createDatabaseAndUser();
+    }
 
     await this.runMigrationsIfNeeded();
 

packages/docs/.gitignore

@@ -1,20 +0,0 @@
-# Dependencies
-/node_modules
-
-# Production
-/build
-
-# Generated files
-.docusaurus
-.cache-loader
-
-# Misc
-.DS_Store
-.env.local
-.env.development.local
-.env.test.local
-.env.production.local
-
-npm-debug.log*
-yarn-debug.log*
-yarn-error.log*

packages/docs/README.md

@@ -1,33 +0,0 @@
-# Website
-
-This website is built using [Docusaurus 2](https://docusaurus.io/), a modern static website generator.
-
-### Installation
-
-```
-$ yarn
-```
-
-### Local Development
-
-```
-$ yarn start
-```
-
-This command starts a local development server and opens up a browser window. Most changes are reflected live without having to restart the server.
-
-### Build
-
-```
-$ yarn build
-```
-
-This command generates static content into the `build` directory and can be served using any static contents hosting service.
-
-### Deployment
-
-```
-$ GIT_USER=<Your GitHub username> USE_SSH=true yarn deploy
-```
-
-If you are using GitHub pages for hosting, this command is a convenient way to build the website and push to the `gh-pages` branch.