Developing with workflows #

The Workflow system is highly custimizable with your own custom code. You can develop your own Predicates, Tasks and Actions.

Predicate #

A predicate is an external validation involved in the triggering of a Workflow or rule.

In a predicate, it is possible to perform additional checks or to call external APIs in order to validate a Workflow or rule.

For each predicate, it is possible to define custom arguments that will be used for validation.

Each Workflow or rule can use several predicates in its actions.

Predicate Definition #

Predicates must be classes implementing the Predicate abstract class.

The super constructor must be called with the predicate name. This name will be used to refer to the Predicate in the Workflow definition.

Then the abstract Predicate.run method must be implemented.

This method takes 3 arguments:

workflowContext: Workflow execution context
object: Rule or Workflow instance
args: additional arguments defined for this predicate

It must return a promise resolving to a boolean value, indicating whether or not the predicate is validated.

The predicate properties description and argumentsDescription should also be defined. This allows to document the predicate and helps users to sort through them when configuring Workflows and rules.

Predicate Setup hooks #

It's possible te define setup and disposal hooks that will be executed when a new Workflow engine is created on an index.

For instance, these hooks can be used to set up collections:

Predicate.setup: executed when an engine is created on an index
Predicate.dispose: executed when an engine is deleted from an index

The hook method will receive the name of the index as an argument.

Predicate Usage #

Predicates must be registered to the plugin in order to be usable by Workflows and rules.

The WorkflowsPlugin.registerPredicate method allows to register new predicates.

Available predicates can be listed by using the Workflows/predicate:list API action.

Example: Predicate returning true if current time is a given hour

 Copied to clipboard!
import { JSONObject } from 'kuzzle';

import { Predicate, WorkflowContext, DynamicObject } from 'kuzzle-plugin-workflows';

export class HourPredicate extends Predicate {
  constructor () {
    super('check-if-its-hour');

    this.description = 'Returns true if the current time is a given hour';

    this.argumentsDescription = {
      hour: {
        type: 'number',
        description: 'Hour to check'
      }
    };
  }

  /**
   * A "hours" collection will be created when a workflow engine
   * is created on an index
   */
  async setup (index: string) {
    await this.sdk.collection.create(index, 'hours', {});
  }

  /**
   * The "hours" collection will be deleted when a workflow engine
   * is deleted from an index
   */
  async dispose (index: string) {
    await this.sdk.collection.delete(index, 'hours');
  }

  async run (workflowContext: WorkflowContext, initiator: DynamicObject, args: JSONObject) {
    const now = new Date();

    return now.getHours() === args.hour;
  }
}

Task #

A Task is a block of code to be executed within a Workflow or a Rule, once its conditions are fulfilled.

Tasks must be classes implementing the Task abstract class.

The super constructor must be called with the task name. This name will be used to refer to the task in the Workflow or Rule definition.

Then the abstract Task.run method has to be implemented.

This method takes 3 arguments:

workflowContext: Workflow execution context
object: Rule or Workflow instance
args: additional arguments defined for this task

It must return a promise resolving to the WorkflowContext instance received in the arguments.

The task properties description and argumentsDescription should also be defined. This allows to document tasks and helps users to sort through them when configuring Workflows and rules.

Task Setup hooks #

It's possible te define setup and disposal hooks that will be executed when a new Workflow engine is created on an index.

For instance, these hooks allow to setup data collections:

Task.setup: executed when an engine is created on an index
Task.dispose: executed when an engine is deleted from an index

The hook method will receive the name of the index as an argument.

Task Usage #

Tasks must be registered to the plugin in order to be usable by Workflows and rules.

The WorkflowsPlugin.registerTask method allows to register new tasks.

Available tasks can be listed by using the Workflows/task:list API action.

Example: A task used to create an alert in a specific collection

 Copied to clipboard!
import { JSONObject } from 'kuzzle';

import { Task , WorkflowContext, DynamicObject  } from 'kuzzle-plugin-workflows';

export class CreateAlertTask extends Task {
  constructor () {
    super('create-alert');

    this.description = 'Creates an alert in the engine\'s "alerts" collection';

    this.argumentsDescription = {
      _id: {
        description: 'ID of the created alert',
        type: 'string',
        optional: true,
      }
    };
  }

  /**
   * An "alerts" collection will be created when a workflow engine
   * is created on an index
   */
  async setup (index: string) {
    await this.sdk.collection.create(index, 'alerts', {});
  }

  /**
   * The "alerts" collection will be deleted when a workflow engine
   * is deleted from an index
   */
  async dispose (index: string) {
    await this.sdk.collection.delete(index, 'alerts');
  }

  /**
   * Creates a document in the "alerts" collection when the task is executed
   */
  async run (workflowContext: WorkflowContext, initiator: DynamicObject, args: JSONObject) {
    await this.secureSdk(object).document.createOrReplace(
      workflowContext.engineIndex,
      'alerts',
      args._id,
      {
        workflow: {
          id: workflowContext.workflow._id,
          name: workflowContext.workflow._source.name,
          description: workflowContext.workflow._source.description,
        },
        payload: {
          _id: workflowContext.payload._id,
          _source: workflowContext.payload._source,
        },
      });
  }
}

Workflow Context #

A WorkflowContext instance is an object which is used during the whole Workflow execution process.

This object will be passed to the predicates and tasks run methods.

It contains the following properties:

engineIndex: the name of the engine index
workflow: the Workflow being executed
payload: the extracted payload
props: additional custom information

Default Workflows and Rules #

It is possible to register default Workflows and rules that will be automatically copied into engine indexes when the application is starting or when a new engine is created.

This allows to define global Workflows and rules applied to all tenants.

You can use the following methods to register default Workflows and rules:

Security concerns #

Developers should be careful about security when developing custom predicates and tasks.

Those predicates and tasks will contain custom code that will be used by users inside Workflows and rules so developers need to be careful about privilege escalation or data leak between tenants.

Engine isolation #

Inside predicates and tasks, the Task.secureSdk SDK instance should be used (or Predicate.secureSdk for predicates), to get an impersonated SDK to ensure that users reusing a Workflow or a rule are allowed to execute those API actions.

 Copied to clipboard!
class DummyTask extends Task {
  constructor () {
    super('dummy-task');
  }

  async run (workflowContext: WorkflowContext, initiator: DynamicObject, args: JSONObject) {
    // don't
    await this.context.accessors.sdk.document.create(
      workflowContext.engineIndex,
      'alerts',
      {
        // ...
      });

    // do
    await this.secureSdk(object).document.create(
      workflowContext.engineIndex,
      'alerts',
      {
        // ...
      });
  }
}

Also the targeted index should never come from user input but developers should rather use the WorkflowContext.engineIndex property.

Avoid looping #

Workflow with a trigger of type notification may be triggered by actions executed by the EmbeddedSDK.

It means that if a Workflow action writes a document triggering a notification matching the same Workflow trigger, then you can enter an infinite loop.

To avoid this, calls to the document controller can use the silent option to skip realtime notifications, and thus preventing the same Workflow to be triggered again.

 Copied to clipboard!
class DummyTask extends Task {
  constructor () {
    super('dummy-task');
  }

  async run (workflowContext: WorkflowContext, initiator: DynamicObject, args: JSONObject) {
    await this.secureSdk(object).document.create(
      workflowContext.engineIndex,
      'devices',
      {
        // ...
      },
      {
        silent: true
      });
  }
}

Workflows with a trigger of type event use the internal event system and the EmbeddedSDK does not trigger internal events.

Workflows plugin has is own event system wich's used to activate / disable workflows / rules on creation / deletion They are constructed like this:

 Copied to clipboard!
  const event = `workflows:${initiatorType: "rule" | "workflow"}:${actionType: 'write' | 'delete'}`;

And can be triggered manually when using EmbeddedSDK with :

 Copied to clipboard!
  app.trigger<EventWorkflowsObjectDelete | EventWorkflowsObjectCreate>(
      `workflows:${initiatorType}:${actionType}`,
      index,
      [...documents]
  );

See full list of events

Edit this page on Github