#Расширение типов узлов

Тип узла, по сути, является операционной инструкцией. Различные инструкции представляют собой разные операции, выполняемые в рабочем процессе.

Аналогично триггерам, расширение типов узлов также делится на две части: серверную и клиентскую. Серверная часть должна реализовать логику для зарегистрированной инструкции, а клиентская — предоставить конфигурацию интерфейса для параметров узла, где находится эта инструкция.

#Серверная часть

#Простейшая инструкция узла

Основное содержимое инструкции — это функция, то есть метод run в классе инструкции должен быть реализован для выполнения логики инструкции. В этой функции можно выполнять любые необходимые операции, например, операции с базой данных, файловые операции, вызовы сторонних API и так далее.

Все инструкции должны быть унаследованы от базового класса Instruction. Простейшая инструкция требует лишь реализации функции run:

import { Instruction, JOB_STATUS } from '@nocobase/plugin-workflow';

export class MyInstruction extends Instruction {
  run(node, input, processor) {
    console.log('my instruction runs!');
    return {
      status: JOB_STATUS.RESOVLED,
    };
  }
}

И зарегистрируйте эту инструкцию в плагине рабочего процесса:

export default class MyPlugin extends Plugin {
  load() {
    // get workflow plugin instance
    const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);

    // register instruction
    workflowPlugin.registerInstruction('my-instruction', MyInstruction);
  }
}

Значение статуса (status) в возвращаемом объекте инструкции является обязательным и должно быть одним из значений константы JOB_STATUS. Это значение определяет дальнейшее направление обработки данного узла в рабочем процессе. Обычно используется JOB_STATUS.RESOVLED, что означает успешное выполнение узла и продолжение выполнения последующих узлов. Если необходимо заранее сохранить результирующее значение, вы также можете вызвать метод processor.saveJob и вернуть его возвращаемый объект. Исполнитель сгенерирует запись о результате выполнения на основе этого объекта.

#Результирующее значение узла

Если имеется конкретный результат выполнения, особенно данные, подготовленные для использования последующими узлами, его можно вернуть через свойство result и сохранить в объекте задачи узла:

import { Instruction, JOB_STATUS } from '@nocobase/plugin-workflow';

export class RandomStringInstruction extends Instruction {
  run(node, input, processor) {
    // customized config from node
    const { digit = 1 } = node.config;
    const result = `${Math.round(10 ** digit * Math.random())}`.padStart(
      digit,
      '0',
    );
    return {
      status: JOB_STATUS.RESOVLED,
      result,
    };
  },
};

Здесь node.config — это элемент конфигурации узла, который может быть любым необходимым значением. Он будет сохранен как поле типа JSON в соответствующей записи узла в базе данных.

#Обработка ошибок инструкции

Если в процессе выполнения могут возникнуть исключения, вы можете перехватить их заранее и вернуть статус ошибки:

import { JOB_STATUS } from '@nocobase/plugin-workflow';

export const errorInstruction = {
  run(node, input, processor) {
    try {
      throw new Error('exception');
    } catch (error) {
      return {
        status: JOB_STATUS.ERROR,
        result: error,
      };
    }
  },
};

Если предсказуемые исключения не будут перехвачены, движок рабочего процесса автоматически перехватит их и вернет статус ошибки, чтобы предотвратить сбой программы из-за необработанных исключений.

#Асинхронные узлы

Когда узлу необходимо дождаться завершения внешней операции перед продолжением рабочего процесса (например, HTTP-запросы, колбэки платёжных систем сторонних сервисов или другие длительные либо не возвращающие результат немедленно операции), задачу следует сначала сохранить со статусом JOB_STATUS.PENDING для приостановки текущего выполнения, а затем возобновить её через resume после завершения операции. Любая инструкция, использующая логику приостановки, должна также реализовывать метод resume; в противном случае рабочий процесс не сможет быть возобновлён.

Рекомендуемый паттерн реализации:

import { Instruction, JOB_STATUS, FlowNodeModel, IJob } from '@nocobase/plugin-workflow';

export class AsyncInstruction extends Instruction {
  async run(node: FlowNodeModel, prevJob, processor) {
    // 1. Сохраняем задачу в статусе ожидания и запоминаем её id
    const { id } = processor.saveJob({
      status: JOB_STATUS.PENDING,
      nodeId: node.id,
      nodeKey: node.key,
      upstreamId: prevJob?.id ?? null,
    });

    // 2. Явно вызываем exit(), чтобы сбросить задачу в БД и зафиксировать транзакцию
    await processor.exit();

    // 3. Запускаем асинхронную операцию (транзакция уже зафиксирована, соединение с БД не удерживается)
    const jobDone: IJob = { status: JOB_STATUS.PENDING };
    try {
      const result = await someAsyncOperation(node.config);
      jobDone.status = JOB_STATUS.RESOLVED;
      jobDone.result = result;
    } catch (error) {
      jobDone.status = JOB_STATUS.FAILED;
      jobDone.result = { message: error.message };
    } finally {
      // 4. Перезапрашиваем задачу из БД; не используем кешированный объект из памяти
      const job = await this.workflow.app.db.getRepository('jobs').findOne({
        filterByTk: id,
      });
      job.set(jobDone);

      // 5. Уведомляем движок рабочего процесса о возобновлении выполнения, входим в поток resume
      this.workflow.resume(job);
    }
    // 6. Ничего не возвращаем (void); исполнитель немедленно завершится
  }

  async resume(node: FlowNodeModel, job, processor) {
    // Задача уже получила финальный статус в run(), просто возвращаем её
    return job;
  }
}

Несколько ключевых деталей:

Почему нужно явно вызывать processor.exit(), а не возвращать объект задачи в состоянии ожидания?
return { status: PENDING } немедленно завершает функцию run, после чего выполнение любого кода становится невозможным. Вызов await processor.exit() лишь фиксирует транзакцию и выходит из контекста базы данных, тогда как сама функция продолжает выполняться. Это позволяет использовать await для длительной операции в том же теле функции и вызвать resume по её завершении. Если пропустить exit() и напрямую await-ить длительную операцию перед возвратом, транзакция базы данных будет удерживаться долгое время, вызывая конкуренцию блокировок, а запись задачи не будет сохранена до момента фиксации транзакции после завершения операции.

Почему нужно перезапрашивать задачу, а не использовать объект, возвращённый saveJob?
Объект, возвращённый saveJob, — это экземпляр модели в памяти, привязанный к исходной транзакции. После вызова processor.exit() эта транзакция зафиксирована и закрыта. Прямое изменение данного экземпляра и вызов resume приведёт к аномалиям состояния ORM (устаревшие ссылки на транзакцию, несогласованность состояния и т.д.). Перезапрос из базы данных по id гарантирует получение чистого экземпляра, не привязанного ни к какой транзакции.

Почему функция run ничего не возвращает (void)?
processor.exit() уже был вызван вручную. Когда исполнитель получает void, он вызывает exit(true) и немедленно завершается без какой-либо избыточной обработки. Если бы в этот момент был возвращён IJob, исполнитель попытался бы сохранить и зафиксировать данные повторно, что привело бы к ошибкам. Подробности см. в разделе о типах возвращаемых значений run/resume.

Для сценариев, требующих внешних колбэков (например, результаты платежа, полученные через webhook), применяется тот же подход: вызвать processor.exit() до регистрации колбэка, чтобы запись задачи была в базе данных до обратного вызова внешней системы. В колбэке перезапросить задачу по id и затем вызвать this.workflow.resume(job).

Полный пример из реального проекта: RequestInstruction.ts (узел HTTP-запроса, использующий этот паттерн в ветке асинхронного рабочего процесса)

#Статус результата узла

Статус выполнения узла влияет на успех или неудачу всего рабочего процесса. Обычно, при отсутствии ветвлений, сбой одного узла напрямую приводит к сбою всего рабочего процесса. Наиболее распространенный сценарий: если узел успешно выполнен, он переходит к следующему узлу в таблице узлов, пока не останется последующих узлов, после чего весь рабочий процесс завершается со статусом успеха.

Если в процессе выполнения какой-либо узел возвращает статус сбоя, движок будет обрабатывать это по-разному в зависимости от следующих двух ситуаций:

1. Узел, возвращающий статус сбоя, находится в основном рабочем процессе, то есть не находится ни в одном из ветвящихся процессов, запущенных вышестоящими узлами. В этом случае весь основной рабочий процесс будет считаться неудачным, и процесс завершится.

2. Узел, возвращающий статус сбоя, находится внутри ветвящегося процесса. В этом случае ответственность за определение следующего состояния процесса передается узлу, который открыл ветвь. Внутренняя логика этого узла определяет состояние последующего процесса, и это решение рекурсивно распространяется вверх до основного рабочего процесса.

В конечном итоге следующее состояние всего рабочего процесса определяется на узлах основного рабочего процесса. Если узел в основном рабочем процессе возвращает сбой, то весь рабочий процесс завершается со статусом сбоя.

Если какой-либо узел после выполнения возвращает статус "ожидание", то весь процесс выполнения будет временно прерван и приостановлен, ожидая события, определенного соответствующим узлом, для возобновления выполнения процесса. Например, узел "Ручной ввод": при достижении этого узла процесс приостанавливается со статусом "ожидание", ожидая ручного вмешательства для принятия решения о продолжении. Если введенный вручную статус — "одобрено", то выполнение продолжается по последующим узлам процесса; в противном случае обрабатывается согласно предыдущей логике сбоя.

Дополнительные статусы возврата инструкций можно найти в разделе "Справочник по API рабочих процессов".

#Типы возвращаемых значений run/resume и поведение исполнителя

Полное определение типа возвращаемого значения методов run и resume:

type InstructionResult = IJob | Promise<IJob> | Promise<void> | Promise<null> | null | void;

После вызова инструкции исполнитель (Processor) выполняет различную логику обработки в зависимости от типа возвращаемого значения. Существует три случая.

#1. Возврат объекта задачи IJob

Наиболее распространённый случай. Возвращается объект с обязательным полем status и необязательным полем result. Исполнитель сохраняет его как запись задачи узла и определяет дальнейший ход выполнения по значению status:

• JOB_STATUS.RESOLVED: Узел успешно выполнен; продолжает выполнение следующего узла при его наличии, иначе рабочий процесс завершается
• JOB_STATUS.PENDING: Узел переходит в состояние ожидания; текущий контекст выполнения останавливается, ожидая внешнего события для запуска resume
• Другие статусы ошибок (FAILED, ERROR и т.д.): Передаются родительскому узлу ветки или напрямую завершают весь рабочий процесс

Этот путь является полным путём фиксации транзакции — исполнитель сохраняет запись задачи, записывает в базу данных и фиксирует транзакцию.

Пример: ConditionInstruction.ts (возвращает объект job напрямую при отсутствии ветки; при наличии ветки см. случай void ниже)

#2. Возврат null

При возврате null исполнитель вызывает processor.exit() (без аргументов), что приводит к: сбросу текущих ожидающих задач в базу данных и фиксации транзакции, но без обновления общего статуса выполнения.

Такое использование характерно для метода resume узлов управления ветками: ветка завершилась и статус задачи родительского узла необходимо обновить и сохранить (например, записать «ветка N завершена»), но другие ветки ещё выполняются, и общее выполнение должно оставаться в статусе STARTED, ожидая оставшихся веток — возврат null выходит из текущего контекста resume без влияния на общий статус выполнения.

Пример: ParallelInstruction.ts

• Строка 117: Параллельный узел уже завершён досрочно (resolved/rejected); игнорирует последующие resume веток и возвращает null напрямую
• Строка 135: Некоторые ветки ещё не завершены (PENDING); сохраняет текущий прогресс и возвращает null для продолжения ожидания других веток

#3. Возврат void (без возврата, т.е. неявный undefined)

При возврате void (функция не содержит явного оператора return, или путь выполнения завершается без возвращаемого значения) исполнитель вызывает processor.exit(true), что приводит к немедленному возврату без выполнения каких-либо операций с базой данных.

Этот паттерн используется исключительно в сценариях, когда инструкция взяла на себя управление планированием выполнения: инструкция вручную запускает подпроцесс через processor.run(), и цепочка выполнения подпроцесса самостоятельно обработает запись в базу данных и фиксацию транзакции при завершении. Исполнитель не должен повторно обрабатывать данные.

Типичные примеры:

• ConditionInstruction.ts#L67: При наличии ветки вручную вызывает processor.run(branchNode, savedJob), затем функция завершается, неявно возвращая void
• ParallelInstruction.ts#L108: Перебирает все ветки и вызывает processor.run(branch, job) для каждой, затем функция завершается, неявно возвращая void

:::warn{title=Предупреждение} Если перед возвратом void был вызван processor.saveJob(), эти записи задач не будут записаны в базу данных текущим исполнителем. Они временно хранятся в списке задач исполнителя (в памяти) и будут сброшены в базу данных через exit(), вызванный при завершении подвыполнения, запущенного processor.run(). Поэтому при использовании данного паттерна необходимо убедиться, что существует путь подвыполнения, который завершается в штатном режиме для сохранения этих записей. Планирование ветвящихся рабочих процессов обладает определённой сложностью; требует тщательного проектирования и всестороннего тестирования. :::

Сравнительная таблица трёх возвращаемых значений:

#Узнать больше

Определения различных параметров для типов узлов см. в разделе "Справочник по API рабочих процессов".

#Клиентская часть

Аналогично триггерам, форма конфигурации для инструкции (типа узла) должна быть реализована на клиентской части.

#Простейшая инструкция узла

Все инструкции должны быть унаследованы от базового класса Instruction. Связанные свойства и методы используются для настройки и использования узла.

Например, если нам нужно предоставить интерфейс конфигурации для узла типа "случайная числовая строка" (randomString), определенного выше на серверной части, который имеет параметр digit, представляющий количество цифр в случайном числе, мы будем использовать поле ввода числа в форме конфигурации для получения пользовательского ввода.

import WorkflowPlugin, { Instruction, VariableOption } from '@nocobase/workflow/client';

class MyInstruction extends Instruction {
  title = 'Random number string';
  type = 'randomString';
  group = 'extended';
  fieldset = {
    'digit': {
      type: 'number',
      title: 'Digit',
      name: 'digit',
      'x-decorator': 'FormItem',
      'x-component': 'InputNumber',
      'x-component-props': {
        min: 1,
        max: 10,
      },
      default: 6,
    },
  };
  useVariables(node, options): VariableOption {
    return {
      value: node.key,
      label: node.title,
    };
  }
}

export default class MyPlugin extends Plugin {
  load() {
    // get workflow plugin instance
    const workflowPlugin = this.app.getPlugin<WorkflowPlugin>(WorkflowPlugin);

    // register instruction
    workflowPlugin.registerInstruction('randomString', MyInstruction);
  }
}

#Предоставление результатов узла в качестве переменных

Вы могли заметить метод useVariables в приведенном выше примере. Если вам нужно использовать результат узла (часть result) в качестве переменной для последующих узлов, вам необходимо реализовать этот метод в унаследованном классе инструкции и вернуть объект, соответствующий типу VariableOption. Этот объект служит структурным описанием результата выполнения узла, предоставляя сопоставление имен переменных для выбора и использования в последующих узлах.

Тип VariableOption определяется следующим образом:

export type VariableOption = {
  value?: string;
  label?: string;
  children?: VariableOption[] | null;
  [key: string]: any;
};

Ключевым является свойство value, которое представляет собой сегментированное значение пути имени переменной. label используется для отображения в интерфейсе, а children — для представления многоуровневой структуры переменной, что применяется, когда результат узла является глубоко вложенным объектом.

Используемая переменная внутри системы представляется в виде строкового шаблона пути, разделенного точками, например, {{jobsMapByNodeKey.2dw92cdf.abc}}. Здесь jobsMapByNodeKey представляет собой набор результатов всех узлов (внутренне определен, не требует обработки), 2dw92cdf — это key узла, а abc — это пользовательское свойство в объекте результата узла.

Кроме того, поскольку результат узла также может быть простым значением, при предоставлении переменных узла первый уровень обязательно должен быть описанием самого узла:

{
  value: node.key,
  label: node.title,
}

То есть, первый уровень — это key и заголовок узла. Например, в ссылке на код для узла вычисления, при использовании результата этого узла, параметры интерфейса будут следующими:

Когда результат узла представляет собой сложный объект, вы можете использовать children для дальнейшего описания вложенных свойств. Например, пользовательская инструкция может возвращать следующие данные JSON:

{
  "message": "ok",
  "data": {
    "id": 1,
    "name": "test",
  }
}

Тогда его можно вернуть с помощью метода useVariables следующим образом:

useVariables(node, options): VariableOption {
  return {
    value: node.key,
    label: node.title,
    children: [
      {
        value: 'message',
        label: 'Message',
      },
      {
        value: 'data',
        label: 'Data',
        children: [
          {
            value: 'id',
            label: 'ID',
          },
          {
            value: 'name',
            label: 'Name',
          },
        ],
      },
    ],
  };
}

Таким образом, в последующих узлах вы сможете использовать следующий интерфейс для выбора переменных:

#Доступность узла

По умолчанию в рабочий процесс можно добавлять любые узлы. Однако в некоторых случаях узел может быть неприменим в определенных типах рабочих процессов или ветвей. В таких ситуациях вы можете настроить доступность узла с помощью isAvailable:

// 类型定义
export abstract class Instruction {
  isAvailable?(ctx: NodeAvailableContext): boolean;
}

export type NodeAvailableContext = {
  // 工作流插件实例
  engine: WorkflowPlugin;
  // 工作流实例
  workflow: object;
  // 上游节点
  upstream: object;
  // 是否是分支节点（分支编号）
  branchIndex: number;
};

Метод isAvailable возвращает true, если узел доступен, и false, если он недоступен. Параметр ctx содержит контекстную информацию текущего узла, которую можно использовать для определения его доступности.

При отсутствии особых требований реализовывать метод isAvailable не нужно, так как узлы по умолчанию доступны. Наиболее распространенный сценарий, требующий настройки, — это когда узел может быть ресурсоемкой операцией и не подходит для выполнения в синхронном рабочем процессе. Вы можете использовать метод isAvailable для ограничения его использования. Например:

isAvailable({ engine, workflow, upstream, branchIndex }) {
  return !engine.isWorkflowSync(workflow);
}

#Узнать больше

Определения различных параметров для типов узлов см. в разделе "Справочник по API рабочих процессов".