Merge branch 'automatisch:main' into feature/signalwire-integration

packages/backend/bin/database/utils.ts

@@ -12,6 +12,8 @@ export async function createUser(
   const userParams = {
     email,
     password,
+    fullName: 'Initial admin',
+    role: 'admin',
   };
 
   try {

packages/backend/package.json

@@ -58,6 +58,7 @@
     "oauth-1.0a": "^2.2.6",
     "objection": "^3.0.0",
     "pg": "^8.7.1",
+    "stripe": "^11.13.0",
     "winston": "^3.7.1"
   },
   "contributors": [

packages/backend/src/apps/filter/actions/continue/index.ts

@@ -0,0 +1,79 @@
+import defineAction from '../../../../helpers/define-action';
+
+type TGroupItem = {
+  key: string;
+  operator: keyof TOperators;
+  value: string;
+  id: string;
+}
+
+type TGroup = Record<'and', TGroupItem[]>;
+
+const isEqual = (a: string, b: string) => a === b;
+const isNotEqual = (a: string, b: string) => !isEqual(a, b)
+const isGreaterThan = (a: string, b: string) => Number(a) > Number(b);
+const isLessThan = (a: string, b: string) => Number(a) < Number(b);
+const isGreaterThanOrEqual = (a: string, b: string) => Number(a) >= Number(b);
+const isLessThanOrEqual = (a: string, b: string) => Number(a) <= Number(b);
+const contains = (a: string, b: string) => a.includes(b);
+const doesNotContain = (a: string, b: string) => !contains(a, b);
+
+type TOperatorFunc = (a: string, b: string) => boolean;
+
+type TOperators = {
+  equal: TOperatorFunc;
+  not_equal: TOperatorFunc;
+  greater_than: TOperatorFunc;
+  less_than: TOperatorFunc;
+  greater_than_or_equal: TOperatorFunc;
+  less_than_or_equal: TOperatorFunc;
+  contains: TOperatorFunc;
+  not_contains: TOperatorFunc;
+};
+
+const operators: TOperators = {
+  'equal': isEqual,
+  'not_equal': isNotEqual,
+  'greater_than': isGreaterThan,
+  'less_than': isLessThan,
+  'greater_than_or_equal': isGreaterThanOrEqual,
+  'less_than_or_equal': isLessThanOrEqual,
+  'contains': contains,
+  'not_contains': doesNotContain,
+};
+
+const operate = (operation: keyof TOperators, a: string, b: string) => {
+  return operators[operation](a, b);
+};
+
+export default defineAction({
+  name: 'Continue if conditions match',
+  key: 'continueIfMatches',
+  description: 'Let the execution continue if the conditions match',
+  arguments: [],
+
+  async run($) {
+    const orGroups = $.step.parameters.or as TGroup[];
+
+    const matchingGroups = orGroups.reduce((groups, group) => {
+      const matchingConditions = group.and
+        .filter((condition) => operate(condition.operator, condition.key, condition.value));
+
+      if (matchingConditions.length) {
+        return groups.concat([{ and: matchingConditions }]);
+      }
+
+      return groups;
+    }, []);
+
+    if (matchingGroups.length === 0) {
+      $.execution.exit();
+    }
+
+    $.setActionItem({
+      raw: {
+        or: matchingGroups,
+      }
+    });
+  },
+});

packages/backend/src/apps/filter/actions/index.ts

@@ -0,0 +1,3 @@
+import continueIfMatches from './continue';
+
+export default [continueIfMatches];

packages/backend/src/apps/filter/assets/favicon.svg

@@ -0,0 +1,8 @@
+<svg width="800px" height="800px" viewBox="0 0 512 512" version="1.1" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink">
+  <g id="Page-1" stroke="none" stroke-width="1" fill="none" fill-rule="evenodd">
+    <g id="Shape" fill="#000000" transform="translate(42.666667, 85.333333)">
+      <path d="M3.55271368e-14,1.42108547e-14 L191.565013,234.666667 L192,234.666667 L192,384 L234.666667,384 L234.666667,234.666667 L426.666667,1.42108547e-14 L3.55271368e-14,1.42108547e-14 Z M214.448,192 L211.81248,192 L89.9076267,42.6666667 L336.630187,42.6666667 L214.448,192 Z">
+      </path>
+    </g>
+  </g>
+</svg>

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

@@ -0,0 +1,14 @@
+import defineApp from '../../helpers/define-app';
+import actions from './actions';
+
+export default defineApp({
+  name: 'Filter',
+  key: 'filter',
+  iconUrl: '{BASE_URL}/apps/filter/assets/favicon.svg',
+  authDocUrl: 'https://automatisch.io/docs/apps/filter/connection',
+  supportsConnections: false,
+  baseUrl: '',
+  apiBaseUrl: '',
+  primaryColor: '001F52',
+  actions,
+});

packages/backend/src/apps/ntfy/common/add-auth-header.ts

@@ -1,8 +1,8 @@
 import { TBeforeRequest } from '@automatisch/types';
 
 const addAuthHeader: TBeforeRequest = ($, requestConfig) => {
-  if ($.auth.data.apiBaseUrl) {
-    requestConfig.baseURL = $.auth.data.apiBaseUrl as string;
+  if ($.auth.data.serverUrl) {
+    requestConfig.baseURL = $.auth.data.serverUrl as string;
   }
 
   if ($.auth.data?.username && $.auth.data?.password) {

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

@@ -51,25 +51,20 @@ export default defineAction({
           value: false,
         },
       ],
-    },
-    {
-      label: 'Bot name',
-      key: 'botName',
-      type: 'string' as const,
-      required: true,
-      value: 'Automatisch',
-      description:
-        'Specify the bot name which appears as a bold username above the message inside Slack. Defaults to Automatisch.',
-      variables: true,
-    },
-    {
-      label: 'Bot icon',
-      key: 'botIcon',
-      type: 'string' as const,
-      required: false,
-      description:
-        'Either an image url or an emoji available to your team (surrounded by :). For example, https://example.com/icon_256.png or :robot_face:',
-      variables: true,
+      additionalFields: {
+        type: 'query',
+        name: 'getDynamicFields',
+        arguments: [
+          {
+            name: 'key',
+            value: 'listFieldsAfterSendAsBot',
+          },
+          {
+            name: 'parameters.sendAsBot',
+            value: '{parameters.sendAsBot}',
+          },
+        ],
+      },
     },
   ],
 

packages/backend/src/apps/slack/dynamic-fields/index.ts

@@ -0,0 +1,3 @@
+import listFieldsAfterSendAsBot from './send-as-bot';
+
+export default [listFieldsAfterSendAsBot];

packages/backend/src/apps/slack/dynamic-fields/send-as-bot/index.ts

@@ -0,0 +1,32 @@
+import { IGlobalVariable } from '@automatisch/types';
+
+export default {
+  name: 'List fields after send as bot',
+  key: 'listFieldsAfterSendAsBot',
+
+  async run($: IGlobalVariable) {
+    if ($.step.parameters.sendAsBot) {
+      return [
+        {
+          label: 'Bot name',
+          key: 'botName',
+          type: 'string' as const,
+          required: true,
+          value: 'Automatisch',
+          description:
+            'Specify the bot name which appears as a bold username above the message inside Slack. Defaults to Automatisch.',
+          variables: true,
+        },
+        {
+          label: 'Bot icon',
+          key: 'botIcon',
+          type: 'string' as const,
+          required: false,
+          description:
+            'Either an image url or an emoji available to your team (surrounded by :). For example, https://example.com/icon_256.png or :robot_face:',
+          variables: true,
+        },
+      ];
+    }
+  },
+};

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

@@ -3,6 +3,7 @@ import addAuthHeader from './common/add-auth-header';
 import actions from './actions';
 import auth from './auth';
 import dynamicData from './dynamic-data';
+import dynamicFields from './dynamic-fields';
 
 export default defineApp({
   name: 'Slack',
@@ -17,4 +18,5 @@ export default defineApp({
   auth,
   actions,
   dynamicData,
+  dynamicFields,
 });

packages/backend/src/config/app.ts

@@ -38,6 +38,11 @@ type AppConfig = {
   smtpUser: string;
   smtpPassword: string;
   fromEmail: string;
+  isCloud: boolean;
+  stripeSecretKey: string;
+  stripeSigningSecret: string;
+  stripeStarterPriceKey: string;
+  stripeGrowthPriceKey: string;
   licenseKey: string;
 };
 
@@ -104,6 +109,11 @@ const appConfig: AppConfig = {
   smtpUser: process.env.SMTP_USER,
   smtpPassword: process.env.SMTP_PASSWORD,
   fromEmail: process.env.FROM_EMAIL,
+  isCloud: process.env.AUTOMATISCH_CLOUD === 'true',
+  stripeSecretKey: process.env.STRIPE_SECRET_KEY,
+  stripeSigningSecret: process.env.STRIPE_SIGNING_SECRET,
+  stripeStarterPriceKey: process.env.STRIPE_STARTER_PRICE_KEY,
+  stripeGrowthPriceKey: process.env.STRIPE_GROWTH_PRICE_KEY,
   licenseKey: process.env.LICENSE_KEY,
 };
 

packages/backend/src/controllers/stripe/webhooks.ee.ts

@@ -0,0 +1,23 @@
+import { Response } from 'express';
+import { IRequest } from '@automatisch/types';
+import Billing from '../../helpers/billing/index.ee';
+import appConfig from '../../config/app';
+import logger from '../../helpers/logger';
+
+export default async (request: IRequest, response: Response) => {
+  const signature = request.headers['stripe-signature'];
+
+  try {
+    const event = Billing.stripe.webhooks.constructEvent(
+      request.rawBody,
+      signature,
+      appConfig.stripeSigningSecret
+    );
+
+    await Billing.handleWebhooks(event);
+    return response.sendStatus(200);
+  } catch (error) {
+    logger.error(`Webhook Error: ${error.message}`);
+    return response.sendStatus(400);
+  }
+};

packages/backend/src/db/migrations/20230301211751_add_full_name_to_users.ts

@@ -0,0 +1,15 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+  return knex.schema.table('users', async (table) => {
+    table.string('full_name');
+
+    await knex('users').update({ full_name: 'Initial admin' });
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  return knex.schema.table('users', (table) => {
+    table.dropColumn('full_name');
+  });
+}

packages/backend/src/db/migrations/20230303134548_create_payment_plans.ts

@@ -0,0 +1,24 @@
+import { Knex } from 'knex';
+import appConfig from '../../config/app';
+
+export async function up(knex: Knex): Promise<void> {
+  if (!appConfig.isCloud) return;
+
+  return knex.schema.createTable('payment_plans', (table) => {
+    table.uuid('id').primary().defaultTo(knex.raw('gen_random_uuid()'));
+    table.string('name').notNullable();
+    table.integer('task_count').notNullable();
+    table.uuid('user_id').references('id').inTable('users');
+    table.string('stripe_customer_id');
+    table.string('stripe_subscription_id');
+    table.timestamp('current_period_started_at').nullable();
+    table.timestamp('current_period_ends_at').nullable();
+    table.timestamp('deleted_at').nullable();
+    table.timestamps(true, true);
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  if (!appConfig.isCloud) return;
+  return knex.schema.dropTable('payment_plans');
+}

packages/backend/src/db/migrations/20230303180902_create_usage_data.ts

@@ -0,0 +1,20 @@
+import { Knex } from 'knex';
+import appConfig from '../../config/app';
+
+export async function up(knex: Knex): Promise<void> {
+  if (!appConfig.isCloud) return;
+
+  return knex.schema.createTable('usage_data', (table) => {
+    table.uuid('id').primary().defaultTo(knex.raw('gen_random_uuid()'));
+    table.uuid('user_id').references('id').inTable('users');
+    table.string('consumed_task_count').notNullable();
+    table.timestamp('next_reset_at').nullable();
+    table.timestamp('deleted_at').nullable();
+    table.timestamps(true, true);
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  if (!appConfig.isCloud) return;
+  return knex.schema.dropTable('usage_data');
+}

packages/backend/src/db/migrations/20230306103149_alter_consumed_task_count_of_usage_data.ts

@@ -0,0 +1,18 @@
+import { Knex } from 'knex';
+import appConfig from '../../config/app';
+
+export async function up(knex: Knex): Promise<void> {
+  if (!appConfig.isCloud) return;
+
+  return knex.schema.alterTable('usage_data', (table) => {
+    table.integer('consumed_task_count').notNullable().alter();
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  if (!appConfig.isCloud) return;
+
+  return knex.schema.alterTable('usage_data', (table) => {
+    table.string('consumed_task_count').notNullable().alter();
+  });
+}

packages/backend/src/errors/already-processed.ts

@@ -0,0 +1,3 @@
+import BaseError from './base';
+
+export default class AlreadyProcessedError extends BaseError { }

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

@@ -13,6 +13,7 @@ import createStep from './mutations/create-step';
 import updateStep from './mutations/update-step';
 import deleteStep from './mutations/delete-step';
 import createUser from './mutations/create-user.ee';
+import deleteUser from './mutations/delete-user.ee';
 import updateUser from './mutations/update-user';
 import forgotPassword from './mutations/forgot-password.ee';
 import resetPassword from './mutations/reset-password.ee';
@@ -34,6 +35,7 @@ const mutationResolvers = {
   updateStep,
   deleteStep,
   createUser,
+  deleteUser,
   updateUser,
   forgotPassword,
   resetPassword,

packages/backend/src/graphql/mutations/create-user.ee.ts

@@ -1,14 +1,17 @@
 import User from '../../models/user';
+import Billing from '../../helpers/billing/index.ee';
+import appConfig from '../../config/app';
 
 type Params = {
   input: {
+    fullName: string;
     email: string;
     password: string;
   };
 };
 
 const createUser = async (_parent: unknown, params: Params) => {
-  const { email, password } = params.input;
+  const { fullName, email, password } = params.input;
 
   const existingUser = await User.query().findOne({ email });
 
@@ -17,11 +20,16 @@ const createUser = async (_parent: unknown, params: Params) => {
   }
 
   const user = await User.query().insert({
+    fullName,
     email,
     password,
     role: 'user',
   });
 
+  if (appConfig.isCloud) {
+    Billing.createSubscription(user);
+  }
+
   return user;
 };
 

packages/backend/src/graphql/mutations/delete-user.ee.ts

@@ -0,0 +1,22 @@
+import Context from '../../types/express/context';
+import deleteUserQueue from '../../queues/delete-user.ee';
+import { Duration } from 'luxon';
+
+const deleteUser = async (_parent: unknown, params: never, context: Context) => {
+  const id = context.currentUser.id;
+
+  await context.currentUser.$query().delete();
+
+  const jobName = `Delete user - ${id}`;
+  const jobPayload = { id };
+  const millisecondsFor30Days = Duration.fromObject({ days: 30 }).toMillis();
+  const jobOptions = {
+    delay: millisecondsFor30Days
+  };
+
+  await deleteUserQueue.add(jobName, jobPayload, jobOptions);
+
+  return true;
+};
+
+export default deleteUser;

packages/backend/src/graphql/mutations/forgot-password.ee.ts

@@ -1,3 +1,4 @@
+import appConfig from '../../config/app';
 import User from '../../models/user';
 import emailQueue from '../../queues/email';
 import {
@@ -30,6 +31,8 @@ const forgotPassword = async (_parent: unknown, params: Params) => {
     template: 'reset-password-instructions',
     params: {
       token: user.resetPasswordToken,
+      webAppUrl: appConfig.webAppUrl,
+      fullName: user.fullName,
     },
   };
 
@@ -40,7 +43,7 @@ const forgotPassword = async (_parent: unknown, params: Params) => {
 
   await emailQueue.add(jobName, jobPayload, jobOptions);
 
-  return;
+  return true;
 };
 
 export default forgotPassword;

packages/backend/src/graphql/mutations/reset-password.ee.ts

@@ -24,7 +24,7 @@ const resetPassword = async (_parent: unknown, params: Params) => {
 
   await user.resetPassword(password);
 
-  return;
+  return true;
 };
 
 export default resetPassword;

packages/backend/src/graphql/queries/get-automatisch-info.ts

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

packages/backend/src/graphql/queries/get-dynamic-fields.ts

@@ -0,0 +1,48 @@
+import { IDynamicFields, IJSONObject } from '@automatisch/types';
+import Context from '../../types/express/context';
+import App from '../../models/app';
+import globalVariable from '../../helpers/global-variable';
+
+type Params = {
+  stepId: string;
+  key: string;
+  parameters: IJSONObject;
+};
+
+const getDynamicFields = async (
+  _parent: unknown,
+  params: Params,
+  context: Context
+) => {
+  const step = await context.currentUser
+    .$relatedQuery('steps')
+    .withGraphFetched({
+      connection: true,
+      flow: true,
+    })
+    .findById(params.stepId);
+
+  if (!step) return null;
+
+  const connection = step.connection;
+
+  if (!connection || !step.appKey) return null;
+
+  const app = await App.findOneByKey(step.appKey);
+  const $ = await globalVariable({ connection, app, flow: step.flow, step });
+
+  const command = app.dynamicFields.find(
+    (data: IDynamicFields) => data.key === params.key
+  );
+
+  for (const parameterKey in params.parameters) {
+    const parameterValue = params.parameters[parameterKey];
+    $.step.parameters[parameterKey] = parameterValue;
+  }
+
+  const additionalFields = await command.run($) || [];
+
+  return additionalFields;
+};
+
+export default getDynamicFields;

packages/backend/src/graphql/queries/get-license.ee.ts

@@ -1,11 +0,0 @@
-import checkLicense from '../../helpers/check-license.ee';
-
-const getLicense = async () => {
-  const license = await checkLicense();
-
-  return {
-    type: license ? 'ee' : 'ce',
-  };
-};
-
-export default getLicense;

packages/backend/src/graphql/queries/get-payment-portal-url.ee.ts

@@ -0,0 +1,16 @@
+import appConfig from '../../config/app';
+import Context from '../../types/express/context';
+import Billing from '../../helpers/billing/index.ee';
+
+const getPaymentPortalUrl = async (
+  _parent: unknown,
+  _params: unknown,
+  context: Context
+) => {
+  if (!appConfig.isCloud) return;
+
+  const url = Billing.createPaymentPortalUrl(context.currentUser);
+  return { url };
+};
+
+export default getPaymentPortalUrl;

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

@@ -0,0 +1,30 @@
+import appConfig from '../../config/app';
+import Context from '../../types/express/context';
+
+const getUsageData = async (
+  _parent: unknown,
+  _params: unknown,
+  context: Context
+) => {
+  if (!appConfig.isCloud) return;
+
+  const usageData = await context.currentUser
+    .$relatedQuery('usageData')
+    .throwIfNotFound();
+
+  const paymentPlan = await context.currentUser
+    .$relatedQuery('paymentPlan')
+    .throwIfNotFound();
+
+  const computedUsageData = {
+    name: paymentPlan.name,
+    allowedTaskCount: paymentPlan.taskCount,
+    consumedTaskCount: usageData.consumedTaskCount,
+    remainingTaskCount: paymentPlan.taskCount - usageData.consumedTaskCount,
+    nextResetAt: usageData.nextResetAt,
+  };
+
+  return computedUsageData;
+};
+
+export default getUsageData;

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

@@ -9,8 +9,11 @@ import getExecution from './queries/get-execution';
 import getExecutions from './queries/get-executions';
 import getExecutionSteps from './queries/get-execution-steps';
 import getDynamicData from './queries/get-dynamic-data';
+import getDynamicFields from './queries/get-dynamic-fields';
 import getCurrentUser from './queries/get-current-user';
-import getLicense from './queries/get-license.ee';
+import getUsageData from './queries/get-usage-data.ee';
+import getPaymentPortalUrl from './queries/get-payment-portal-url.ee';
+import getAutomatischInfo from './queries/get-automatisch-info';
 import healthcheck from './queries/healthcheck';
 
 const queryResolvers = {
@@ -25,8 +28,11 @@ const queryResolvers = {
   getExecutions,
   getExecutionSteps,
   getDynamicData,
+  getDynamicFields,
   getCurrentUser,
-  getLicense,
+  getUsageData,
+  getPaymentPortalUrl,
+  getAutomatischInfo,
   healthcheck,
 };
 

packages/backend/src/graphql/schema.graphql

@@ -28,8 +28,15 @@ type Query {
     key: String!
     parameters: JSONObject
   ): JSONObject
+  getDynamicFields(
+    stepId: String!
+    key: String!
+    parameters: JSONObject
+  ): [SubstepArgument]
   getCurrentUser: User
-  getLicense: GetLicense
+  getUsageData: GetUsageData
+  getPaymentPortalUrl: GetPaymentPortalUrl
+  getAutomatischInfo: GetAutomatischInfo
   healthcheck: AppHealth
 }
 
@@ -49,6 +56,7 @@ type Mutation {
   updateStep(input: UpdateStepInput): Step
   deleteStep(input: DeleteStepInput): Step
   createUser(input: CreateUserInput): User
+  deleteUser: Boolean
   updateUser(input: UpdateUserInput): User
   forgotPassword(input: ForgotPasswordInput): Boolean
   resetPassword(input: ResetPasswordInput): Boolean
@@ -65,38 +73,64 @@ directive @specifiedBy(
   url: String!
 ) on SCALAR
 
+type Trigger {
+  name: String
+  key: String
+  description: String
+  pollInterval: Int
+  type: String
+  substeps: [Substep]
+}
+
 type Action {
   name: String
   key: String
   description: String
-  substeps: [ActionSubstep]
+  substeps: [Substep]
 }
 
-type ActionSubstep {
+type Substep {
   key: String
   name: String
-  arguments: [ActionSubstepArgument]
+  arguments: [SubstepArgument]
 }
 
-type ActionSubstepArgument {
+type SubstepArgument {
   label: String
   key: String
   type: String
   description: String
   required: Boolean
   variables: Boolean
-  options: [ArgumentOption]
-  source: ActionSubstepArgumentSource
+  options: [SubstepArgumentOption]
+  source: SubstepArgumentSource
+  additionalFields: SubstepArgumentAdditionalFields
   dependsOn: [String]
 }
 
-type ActionSubstepArgumentSource {
-  type: String
-  name: String
-  arguments: [ActionSubstepArgumentSourceArgument]
+type SubstepArgumentOption {
+  label: String
+  value: JSONObject
 }
 
-type ActionSubstepArgumentSourceArgument {
+type SubstepArgumentSource {
+  type: String
+  name: String
+  arguments: [SubstepArgumentSourceArgument]
+}
+
+type SubstepArgumentSourceArgument {
+  name: String
+  value: String
+}
+
+type SubstepArgumentAdditionalFields {
+  type: String
+  name: String
+  arguments: [SubstepArgumentAdditionalFieldsArgument]
+}
+
+type SubstepArgumentAdditionalFieldsArgument {
   name: String
   value: String
 }
@@ -198,7 +232,7 @@ type Field {
   description: String
   docUrl: String
   clickToCopy: Boolean
-  options: [ArgumentOption]
+  options: [SubstepArgumentOption]
 }
 
 type FlowConnection {
@@ -304,6 +338,7 @@ input DeleteStepInput {
 }
 
 input CreateUserInput {
+  fullName: String!
   email: String!
   password: String!
 }
@@ -394,52 +429,11 @@ input StepInput {
   previousStep: PreviousStepInput
 }
 
-type Trigger {
-  name: String
-  key: String
-  description: String
-  pollInterval: Int
-  type: String
-  substeps: [TriggerSubstep]
-}
-
-type TriggerSubstep {
-  key: String
-  name: String
-  arguments: [TriggerSubstepArgument]
-}
-
-type TriggerSubstepArgument {
-  label: String
-  key: String
-  type: String
-  description: String
-  required: Boolean
-  variables: Boolean
-  source: TriggerSubstepArgumentSource
-  dependsOn: [String]
-  options: [ArgumentOption]
-}
-
-type TriggerSubstepArgumentSource {
-  type: String
-  name: String
-  arguments: [TriggerSubstepArgumentSourceArgument]
-}
-
-type ArgumentOption {
-  label: String
-  value: JSONObject
-}
-
-type TriggerSubstepArgumentSourceArgument {
-  name: String
-  value: String
-}
-
 type User {
   id: String
+  fullName: String
   email: String
+  role: String
   createdAt: String
   updatedAt: String
 }
@@ -471,8 +465,20 @@ type AppHealth {
   version: String
 }
 
-type GetLicense {
-  type: String
+type GetAutomatischInfo {
+  isCloud: Boolean
+}
+
+type GetUsageData {
+  name: String
+  allowedTaskCount: Int
+  consumedTaskCount: Int
+  remainingTaskCount: Int
+  nextResetAt: String
+}
+
+type GetPaymentPortalUrl {
+  url: String
 }
 
 schema {

packages/backend/src/helpers/authentication.ts

@@ -24,6 +24,7 @@ const authentication = shield(
   {
     Query: {
       '*': isAuthenticated,
+      getAutomatischInfo: allow,
       healthcheck: allow,
     },
     Mutation: {
@@ -31,6 +32,7 @@ const authentication = shield(
       login: allow,
       createUser: allow,
       forgotPassword: allow,
+      resetPassword: allow,
     },
   },
   {

packages/backend/src/helpers/billing/index.ee.ts

@@ -0,0 +1,100 @@
+import Stripe from 'stripe';
+import User from '../../models/user';
+import PaymentPlan from '../../models/payment-plan.ee';
+import UsageData from '../../models/usage-data.ee';
+import appConfig from '../../config/app';
+import handleWebhooks from './webhooks.ee';
+
+const plans = [
+  {
+    price: appConfig.stripeStarterPriceKey,
+    name: 'Starter',
+    taskCount: 1000,
+    default: true,
+  },
+  {
+    price: appConfig.stripeGrowthPriceKey,
+    name: 'Growth',
+    taskCount: 10000,
+    default: false,
+  },
+];
+
+const stripe = new Stripe(appConfig.stripeSecretKey, {
+  apiVersion: '2022-11-15',
+});
+
+const createStripeCustomer = async (user: User) => {
+  const params: Stripe.CustomerCreateParams = {
+    email: user.email,
+    name: user.fullName,
+    description: `User ID: ${user.id}`,
+  };
+
+  return await stripe.customers.create(params);
+};
+
+const defaultPlan = plans.find((plan) => plan.default);
+
+const createStripeSubscription = async (
+  user: User,
+  stripeCustomer: Stripe.Customer
+) => {
+  const params: Stripe.SubscriptionCreateParams = {
+    customer: stripeCustomer.id,
+    items: [{ price: defaultPlan.price }],
+  };
+
+  return await stripe.subscriptions.create(params);
+};
+
+const createSubscription = async (user: User) => {
+  const stripeCustomer = await createStripeCustomer(user);
+  const stripeSubscription = await createStripeSubscription(
+    user,
+    stripeCustomer
+  );
+
+  await PaymentPlan.query().insert({
+    name: defaultPlan.name,
+    taskCount: defaultPlan.taskCount,
+    userId: user.id,
+    stripeCustomerId: stripeCustomer.id,
+    stripeSubscriptionId: stripeSubscription.id,
+    currentPeriodStartedAt: new Date(
+      stripeSubscription.current_period_start * 1000
+    ).toISOString(),
+    currentPeriodEndsAt: new Date(
+      stripeSubscription.current_period_end * 1000
+    ).toISOString(),
+  });
+
+  await UsageData.query().insert({
+    userId: user.id,
+    consumedTaskCount: 0,
+    nextResetAt: new Date(
+      stripeSubscription.current_period_end * 1000
+    ).toISOString(),
+  });
+};
+
+const createPaymentPortalUrl = async (user: User) => {
+  const paymentPlan = await user.$relatedQuery('paymentPlan');
+
+  const userSession = await stripe.billingPortal.sessions.create({
+    customer: paymentPlan.stripeCustomerId,
+    return_url: 'https://cloud.automatisch.io',
+  });
+
+  return userSession.url;
+};
+
+const billing = {
+  createSubscription,
+  createPaymentPortalUrl,
+  handleWebhooks,
+  stripe,
+  plans,
+};
+
+export default billing;

packages/backend/src/helpers/billing/webhooks.ee.ts

@@ -0,0 +1,42 @@
+import Stripe from 'stripe';
+import PaymentPlan from '../../models/payment-plan.ee';
+import Billing from './index.ee';
+
+const handleWebhooks = async (event: Stripe.Event) => {
+  const trackedWebhookTypes = [
+    'customer.subscription.created',
+    'customer.subscription.updated',
+    'customer.subscription.deleted',
+  ];
+
+  if (!trackedWebhookTypes.includes(event.type)) {
+    return;
+  }
+
+  await updatePaymentPlan(event);
+};
+
+const updatePaymentPlan = async (event: Stripe.Event) => {
+  const subscription = event.data.object as Stripe.Subscription;
+  const priceKey = subscription.items.data[0].plan.id;
+  const plan = Billing.plans.find((plan) => plan.price === priceKey);
+
+  const paymentPlan = await PaymentPlan.query().findOne({
+    stripe_customer_id: subscription.customer,
+  });
+
+  await paymentPlan.$query().patchAndFetch({
+    name: plan.name,
+    taskCount: plan.taskCount,
+    stripeSubscriptionId: subscription.id,
+  });
+
+  const user = await paymentPlan.$relatedQuery('user');
+  const usageData = await user.$relatedQuery('usageData');
+
+  await usageData.$query().patchAndFetch({
+    nextResetAt: new Date(subscription.current_period_end * 1000).toISOString(),
+  });
+};
+
+export default handleWebhooks;

packages/backend/src/helpers/compute-parameters.ts

@@ -38,6 +38,13 @@ export default function computeParameters(
       };
     }
 
+    if (Array.isArray(value)) {
+      return {
+        ...result,
+        [key]: value.map(item => computeParameters(item, executionSteps)),
+      };
+    }
+
     return {
       ...result,
       [key]: value,

packages/backend/src/helpers/create-bull-board-handler.ts

@@ -4,6 +4,8 @@ import { BullMQAdapter } from '@bull-board/api/bullMQAdapter';
 import flowQueue from '../queues/flow';
 import triggerQueue from '../queues/trigger';
 import actionQueue from '../queues/action';
+import emailQueue from '../queues/email';
+import deleteUserQueue from '../queues/delete-user.ee';
 import appConfig from '../config/app';
 
 const serverAdapter = new ExpressAdapter();
@@ -21,6 +23,8 @@ const createBullBoardHandler = async (serverAdapter: ExpressAdapter) => {
       new BullMQAdapter(flowQueue),
       new BullMQAdapter(triggerQueue),
       new BullMQAdapter(actionQueue),
+      new BullMQAdapter(emailQueue),
+      new BullMQAdapter(deleteUserQueue),
     ],
     serverAdapter: serverAdapter,
   });

packages/backend/src/helpers/global-variable.ts

@@ -13,6 +13,7 @@ import {
   IRequest,
 } from '@automatisch/types';
 import EarlyExitError from '../errors/early-exit';
+import AlreadyProcessedError from '../errors/already-processed';
 
 type GlobalVariableOptions = {
   connection?: Connection;
@@ -77,6 +78,9 @@ const globalVariable = async (
     execution: {
       id: execution?.id,
       testRun,
+      exit: () => {
+        throw new EarlyExitError();
+      }
     },
     lastExecutionStep: (await step?.getLastExecutionStep())?.toJSON(),
     triggerOutput: {
@@ -93,7 +97,7 @@ const globalVariable = async (
         !$.execution.testRun
       ) {
         // early exit as we do not want to process duplicate items in actual executions
-        throw new EarlyExitError();
+        throw new AlreadyProcessedError();
       }
 
       $.triggerOutput.data.push(triggerItem);

packages/backend/src/helpers/graphql-instance.ts

@@ -28,7 +28,7 @@ const graphQLInstance = graphqlHTTP({
       delete (error.originalError as HttpError).response;
     }
 
-    return error.originalError;
+    return error;
   },
 });
 

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

@@ -40,8 +40,9 @@ export default function createHttpClient({
   instance.interceptors.response.use(
     (response) => response,
     async (error) => {
-      const { config } = error;
-      const { status } = error.response;
+      const { config, response } = error;
+      // Do not destructure `status` from `error.response` because it might not exist
+      const status = response?.status;
 
       if (
         // TODO: provide a `shouldRefreshToken` function in the app

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

@@ -1,9 +1,10 @@
 import type { QueryContext } from 'objection';
+import { IJSONObject } from '@automatisch/types';
+import appConfig from '../config/app';
 import Base from './base';
 import Execution from './execution';
 import Step from './step';
 import Telemetry from '../helpers/telemetry';
-import { IJSONObject } from '@automatisch/types';
 
 class ExecutionStep extends Base {
   id!: string;
@@ -14,6 +15,7 @@ class ExecutionStep extends Base {
   errorDetails: IJSONObject;
   status: 'success' | 'failure';
   step: Step;
+  execution?: Execution;
 
   static tableName = 'execution_steps';
 
@@ -57,6 +59,18 @@ class ExecutionStep extends Base {
   async $afterInsert(queryContext: QueryContext) {
     await super.$afterInsert(queryContext);
     Telemetry.executionStepCreated(this);
+
+    if (appConfig.isCloud) {
+      const execution = await this.$relatedQuery('execution');
+
+      if (!execution.testRun && !this.isFailed) {
+        const flow = await execution.$relatedQuery('flow');
+        const user = await flow.$relatedQuery('user');
+        const usageData = await user.$relatedQuery('usageData');
+
+        await usageData.increaseConsumedTaskCountByOne();
+      }
+    }
   }
 }
 

packages/backend/src/models/execution.ts

@@ -10,6 +10,7 @@ class Execution extends Base {
   testRun: boolean;
   internalId: string;
   executionSteps: ExecutionStep[];
+  flow?: Flow;
 
   static tableName = 'executions';
 

packages/backend/src/models/flow.ts

@@ -3,6 +3,7 @@ import type { ModelOptions, QueryContext } from 'objection';
 import ExtendedQueryBuilder from './query-builder';
 import Base from './base';
 import Step from './step';
+import User from './user';
 import Execution from './execution';
 import Telemetry from '../helpers/telemetry';
 
@@ -15,6 +16,7 @@ class Flow extends Base {
   published_at: string;
   remoteWebhookId: string;
   executions?: Execution[];
+  user?: User;
 
   static tableName = 'flows';
 
@@ -51,6 +53,14 @@ class Flow extends Base {
         to: 'executions.flow_id',
       },
     },
+    user: {
+      relation: Base.HasOneRelation,
+      modelClass: User,
+      join: {
+        from: 'flows.user_id',
+        to: 'users.id',
+      },
+    },
   });
 
   async lastInternalId() {

packages/backend/src/models/payment-plan.ee.ts

@@ -0,0 +1,53 @@
+import Base from './base';
+import User from './user';
+
+class PaymentPlan extends Base {
+  id!: string;
+  name!: string;
+  taskCount: number;
+  userId!: string;
+  stripeCustomerId!: string;
+  stripeSubscriptionId!: string;
+  currentPeriodStartedAt!: string;
+  currentPeriodEndsAt!: string;
+  user?: User;
+
+  static tableName = 'payment_plans';
+
+  static jsonSchema = {
+    type: 'object',
+    required: [
+      'name',
+      'taskCount',
+      'userId',
+      'stripeCustomerId',
+      'stripeSubscriptionId',
+      'currentPeriodStartedAt',
+      'currentPeriodEndsAt',
+    ],
+
+    properties: {
+      id: { type: 'string', format: 'uuid' },
+      name: { type: 'string' },
+      taskCount: { type: 'integer' },
+      userId: { type: 'string', format: 'uuid' },
+      stripeCustomerId: { type: 'string' },
+      stripeSubscriptionId: { type: 'string' },
+      currentPeriodStartedAt: { type: 'string' },
+      currentPeriodEndsAt: { type: 'string' },
+    },
+  };
+
+  static relationMappings = () => ({
+    user: {
+      relation: Base.BelongsToOneRelation,
+      modelClass: User,
+      join: {
+        from: 'payment_plans.user_id',
+        to: 'users.id',
+      },
+    },
+  });
+}
+
+export default PaymentPlan;

packages/backend/src/models/step.ts

@@ -149,6 +149,22 @@ class Step extends Base {
 
     return command;
   }
+
+  async getSetupFields() {
+    let setupSupsteps;
+
+    if (this.isTrigger) {
+      setupSupsteps = (await this.getTriggerCommand()).substeps;
+    } else {
+      setupSupsteps = (await this.getActionCommand()).substeps;
+    }
+
+    const existingArguments = setupSupsteps.find(
+      (substep) => substep.key === 'chooseTrigger'
+    ).arguments;
+
+    return existingArguments;
+  }
 }
 
 export default Step;

packages/backend/src/models/usage-data.ee.ts

@@ -0,0 +1,41 @@
+import { raw } from 'objection';
+import Base from './base';
+import User from './user';
+
+class UsageData extends Base {
+  id!: string;
+  userId!: string;
+  consumedTaskCount!: number;
+  nextResetAt!: string;
+
+  static tableName = 'usage_data';
+
+  static jsonSchema = {
+    type: 'object',
+    required: ['userId', 'consumedTaskCount', 'nextResetAt'],
+
+    properties: {
+      id: { type: 'string', format: 'uuid' },
+      userId: { type: 'string', format: 'uuid' },
+      consumedTaskCount: { type: 'integer' },
+      nextResetAt: { type: 'string' },
+    },
+  };
+
+  static relationMappings = () => ({
+    user: {
+      relation: Base.BelongsToOneRelation,
+      modelClass: User,
+      join: {
+        from: 'usage_data.user_id',
+        to: 'users.id',
+      },
+    },
+  });
+
+  async increaseConsumedTaskCountByOne() {
+    return await this.$query().patch({ consumedTaskCount: raw('consumed_task_count + 1') });
+  }
+}
+
+export default UsageData;

packages/backend/src/models/user.ts

@@ -6,9 +6,12 @@ import Step from './step';
 import Execution from './execution';
 import bcrypt from 'bcrypt';
 import crypto from 'crypto';
+import PaymentPlan from './payment-plan.ee';
+import UsageData from './usage-data.ee';
 
 class User extends Base {
   id!: string;
+  fullName!: string;
   email!: string;
   password!: string;
   role: string;
@@ -18,15 +21,18 @@ class User extends Base {
   flows?: Flow[];
   steps?: Step[];
   executions?: Execution[];
+  paymentPlan?: PaymentPlan;
+  usageData?: UsageData;
 
   static tableName = 'users';
 
   static jsonSchema = {
     type: 'object',
-    required: ['email', 'password'],
+    required: ['fullName', 'email', 'password'],
 
     properties: {
       id: { type: 'string', format: 'uuid' },
+      fullName: { type: 'string', minLength: 1 },
       email: { type: 'string', format: 'email', minLength: 1, maxLength: 255 },
       password: { type: 'string', minLength: 1, maxLength: 255 },
       role: { type: 'string', enum: ['admin', 'user'] },
@@ -74,6 +80,22 @@ class User extends Base {
         to: 'executions.flow_id',
       },
     },
+    paymentPlan: {
+      relation: Base.HasOneRelation,
+      modelClass: PaymentPlan,
+      join: {
+        from: 'payment_plans.user_id',
+        to: 'users.id',
+      },
+    },
+    usageData: {
+      relation: Base.HasOneRelation,
+      modelClass: UsageData,
+      join: {
+        from: 'usage_data.user_id',
+        to: 'users.id',
+      },
+    },
   });
 
   login(password: string) {

packages/backend/src/queues/delete-user.ee.ts

@@ -0,0 +1,25 @@
+import process from 'process';
+import { Queue } from 'bullmq';
+import redisConfig from '../config/redis';
+import logger from '../helpers/logger';
+
+const CONNECTION_REFUSED = 'ECONNREFUSED';
+
+const redisConnection = {
+  connection: redisConfig,
+};
+
+const deleteUserQueue = new Queue('delete-user', redisConnection);
+
+process.on('SIGTERM', async () => {
+  await deleteUserQueue.close();
+});
+
+deleteUserQueue.on('error', (err) => {
+  if ((err as any).code === CONNECTION_REFUSED) {
+    logger.error('Make sure you have installed Redis and it is running.', err);
+    process.exit();
+  }
+});
+
+export default deleteUserQueue;

packages/backend/src/routes/index.ts

@@ -1,10 +1,12 @@
 import { Router } from 'express';
 import graphQLInstance from '../helpers/graphql-instance';
 import webhooksRouter from './webhooks';
+import stripeRouter from './stripe.ee';
 
 const router = Router();
 
 router.use('/graphql', graphQLInstance);
 router.use('/webhooks', webhooksRouter);
+router.use('/stripe', stripeRouter);
 
 export default router;

packages/backend/src/routes/stripe.ee.ts

@@ -0,0 +1,23 @@
+import express, { Router } from 'express';
+import multer from 'multer';
+import { IRequest } from '@automatisch/types';
+import appConfig from '../config/app';
+import stripeWebhooksAction from '../controllers/stripe/webhooks.ee';
+
+const router = Router();
+const upload = multer();
+
+router.use(upload.none());
+
+router.use(
+  express.text({
+    limit: appConfig.requestBodySizeLimit,
+    verify(req, res, buf) {
+      (req as IRequest).rawBody = buf;
+    },
+  })
+);
+
+router.post('/webhooks', stripeWebhooksAction);
+
+export default router;

packages/backend/src/services/action.ts

@@ -5,6 +5,8 @@ import ExecutionStep from '../models/execution-step';
 import computeParameters from '../helpers/compute-parameters';
 import globalVariable from '../helpers/global-variable';
 import HttpError from '../errors/http';
+import EarlyExitError from '../errors/early-exit';
+import AlreadyProcessedError from '../errors/already-processed';
 
 type ProcessActionOptions = {
   flowId: string;
@@ -44,13 +46,19 @@ export const processAction = async (options: ProcessActionOptions) => {
   try {
     await actionCommand.run($);
   } catch (error) {
-    if (error instanceof HttpError) {
-      $.actionOutput.error = error.details;
-    } else {
-      try {
-        $.actionOutput.error = JSON.parse(error.message);
-      } catch {
-        $.actionOutput.error = { error: error.message };
+    const shouldEarlyExit = error instanceof EarlyExitError;
+    const shouldNotProcess = error instanceof AlreadyProcessedError;
+    const shouldNotConsiderAsError = shouldEarlyExit || shouldNotProcess;
+
+    if (!shouldNotConsiderAsError) {
+      if (error instanceof HttpError) {
+        $.actionOutput.error = error.details;
+      } else {
+        try {
+          $.actionOutput.error = JSON.parse(error.message);
+        } catch {
+          $.actionOutput.error = { error: error.message };
+        }
       }
     }
   }

packages/backend/src/services/flow.ts

@@ -1,6 +1,7 @@
 import Flow from '../models/flow';
 import globalVariable from '../helpers/global-variable';
 import EarlyExitError from '../errors/early-exit';
+import AlreadyProcessedError from '../errors/already-processed';
 import HttpError from '../errors/http';
 
 type ProcessFlowOptions = {
@@ -29,7 +30,11 @@ export const processFlow = async (options: ProcessFlowOptions) => {
       await triggerCommand.run($);
     }
   } catch (error) {
-    if (error instanceof EarlyExitError === false) {
+    const shouldEarlyExit = error instanceof EarlyExitError;
+    const shouldNotProcess = error instanceof AlreadyProcessedError;
+    const shouldNotConsiderAsError = shouldEarlyExit || shouldNotProcess;
+
+    if (!shouldNotConsiderAsError) {
       if (error instanceof HttpError) {
         $.triggerOutput.error = error.details;
       } else {

packages/backend/src/views/emails/reset-password-instructions.ee.hbs

@@ -1,16 +1,23 @@
 <html lang="en">
 <head>
     <meta charset="UTF-8">
-    <title>Title</title>
+    <title>Reset password instructions</title>
 </head>
 <body>
-  Hello {{ email }}
+  <p>
+    Hello {{ fullName }},
+  </p>
 
-  Someone has requested a link to change your password, and you can do this through the link below.
+  <p>
+    Someone has requested a link to change your password, and you can do this through the link below.
+  </p>
 
-  <a href="/reset-password">Change my password</a>
+  <p>
+    <a href="{{ webAppUrl }}/reset-password?token={{ token }}">Change my password</a>
+  </p>
 
-  If you didn't request this, please ignore this email.
-  Your password won't change until you access the link above and create a new one.
+  <p>
+    If you didn't request this, please ignore this email. Your password won't change until you access the link above and create a new one.
+  </p>
 </body>
 </html>

packages/backend/src/worker.ts

@@ -4,6 +4,7 @@ import './workers/flow';
 import './workers/trigger';
 import './workers/action';
 import './workers/email';
+import './workers/delete-user.ee';
 import telemetry from './helpers/telemetry';
 
 telemetry.setServiceType('worker');

packages/backend/src/workers/action.ts

@@ -21,7 +21,7 @@ const DEFAULT_DELAY_DURATION = 0;
 export const worker = new Worker(
   'action',
   async (job) => {
-    const { stepId, flowId, executionId, computedParameters } = await processAction(
+    const { stepId, flowId, executionId, computedParameters, executionStep } = await processAction(
       job.data as JobData
     );
 
@@ -48,6 +48,10 @@ export const worker = new Worker(
       jobOptions.delay = delayAsMilliseconds(step.key, computedParameters);
     }
 
+    if (step.appKey === 'filter' && !executionStep.dataOut) {
+      return;
+    }
+
     await actionQueue.add(jobName, jobPayload, jobOptions);
   },
   { connection: redisConfig }

packages/backend/src/workers/delete-user.ee.ts

@@ -0,0 +1,44 @@
+import { Worker } from 'bullmq';
+import redisConfig from '../config/redis';
+import logger from '../helpers/logger';
+import User from '../models/user';
+import Execution from '../models/execution';
+import ExecutionStep from '../models/execution-step';
+
+export const worker = new Worker(
+  'delete-user',
+  async (job) => {
+    const { id } = job.data;
+
+    const user = await User.query().findById(id).throwIfNotFound();
+
+    const executionIds = (
+      await user.$relatedQuery('executions').select('executions.id')
+    ).map((execution: Execution) => execution.id);
+
+    await ExecutionStep.query().hardDelete().whereIn('execution_id', executionIds);
+    await user.$relatedQuery('executions').hardDelete();
+    await user.$relatedQuery('steps').hardDelete();
+    await user.$relatedQuery('flows').hardDelete();
+    await user.$relatedQuery('connections').hardDelete();
+
+    await user.$query().hardDelete();
+  },
+  { connection: redisConfig }
+);
+
+worker.on('completed', (job) => {
+  logger.info(
+    `JOB ID: ${job.id} - The user with the ID of '${job.data.id}' has been deleted!`
+  );
+});
+
+worker.on('failed', (job, err) => {
+  logger.info(
+    `JOB ID: ${job.id} - The user with the ID of '${job.data.id}' has failed to be deleted! ${err.message}`
+  );
+});
+
+process.on('SIGTERM', async () => {
+  await worker.close();
+});

packages/backend/src/workers/email.ts

@@ -8,13 +8,13 @@ import appConfig from '../config/app';
 export const worker = new Worker(
   'email',
   async (job) => {
-    const { email, subject, templateName, params } = job.data;
+    const { email, subject, template, params } = job.data;
 
     await mailer.sendMail({
       to: email,
       from: appConfig.fromEmail,
       subject: subject,
-      html: compileEmail(templateName, params),
+      html: compileEmail(template, params),
     });
   },
   { connection: redisConfig }