Cavemanon/piratepoet
7
Fork 0
Code Pull Requests Actions Releases 16 Activity

Use "automation" instead of "workflow"

Browse Source 
[MAILPOET-4793]
This commit is contained in:
Jan Jakes
2022-11-10 10:24:31 +03:00
committed by David Remer
parent d199c3768a
commit 9d55d3f134
210 changed files with 3677 additions and 3659 deletions
Download Patch File Download Diff File Expand all files Collapse all files

128
mailpoet/lib/Automation/Engine/Control/StepHandler.php
View File

@ -4,13 +4,13 @@ namespace MailPoet\Automation\Engine\Control;
use Exception;
use MailPoet\Automation\Engine\Control\Steps\ActionStepRunner;
use MailPoet\Automation\Engine\Data\Automation;
use MailPoet\Automation\Engine\Data\AutomationRun;
use MailPoet\Automation\Engine\Data\AutomationRunLog;
use MailPoet\Automation\Engine\Data\Step;
use MailPoet\Automation\Engine\Data\StepRunArgs;
use MailPoet\Automation\Engine\Data\StepValidationArgs;
use MailPoet\Automation\Engine\Data\SubjectEntry;
use MailPoet\Automation\Engine\Data\Workflow;
use MailPoet\Automation\Engine\Data\WorkflowRun;
use MailPoet\Automation\Engine\Data\WorkflowRunLog;
use MailPoet\Automation\Engine\Exceptions;
use MailPoet\Automation\Engine\Exceptions\InvalidStateException;
use MailPoet\Automation\Engine\Hooks;
@ -18,9 +18,9 @@ use MailPoet\Automation\Engine\Integration\Action;
use MailPoet\Automation\Engine\Integration\Payload;
use MailPoet\Automation\Engine\Integration\Subject;
use MailPoet\Automation\Engine\Registry;
use MailPoet\Automation\Engine\Storage\WorkflowRunLogStorage;
use MailPoet\Automation\Engine\Storage\WorkflowRunStorage;
use MailPoet\Automation\Engine\Storage\WorkflowStorage;
use MailPoet\Automation\Engine\Storage\AutomationRunLogStorage;
use MailPoet\Automation\Engine\Storage\AutomationRunStorage;
use MailPoet\Automation\Engine\Storage\AutomationStorage;
use MailPoet\Automation\Engine\WordPress;
use Throwable;
@ -37,17 +37,17 @@ class StepHandler {
/** @var WordPress */
private $wordPress;
/** @var WorkflowRunStorage */
private $workflowRunStorage;
/** @var AutomationRunStorage */
private $automationRunStorage;
/** @var WorkflowStorage */
private $workflowStorage;
/** @var AutomationStorage */
private $automationStorage;
/** @var array<string, StepRunner> */
private $stepRunners = [];
/** @var WorkflowRunLogStorage */
private $workflowRunLogStorage;
/** @var AutomationRunLogStorage */
private $automationRunLogStorage;
/** @var Hooks */
private $hooks;
@ -61,9 +61,9 @@ class StepHandler {
Hooks $hooks,
SubjectLoader $subjectLoader,
WordPress $wordPress,
WorkflowRunStorage $workflowRunStorage,
WorkflowRunLogStorage $workflowRunLogStorage,
WorkflowStorage $workflowStorage,
AutomationRunStorage $automationRunStorage,
AutomationRunLogStorage $automationRunLogStorage,
AutomationStorage $automationStorage,
Registry $registry
) {
$this->actionScheduler = $actionScheduler;
@ -71,14 +71,14 @@ class StepHandler {
$this->hooks = $hooks;
$this->subjectLoader = $subjectLoader;
$this->wordPress = $wordPress;
$this->workflowRunStorage = $workflowRunStorage;
$this->workflowRunLogStorage = $workflowRunLogStorage;
$this->workflowStorage = $workflowStorage;
$this->automationRunStorage = $automationRunStorage;
$this->automationRunLogStorage = $automationRunLogStorage;
$this->automationStorage = $automationStorage;
$this->registry = $registry;
}
public function initialize(): void {
$this->wordPress->addAction(Hooks::WORKFLOW_STEP, [$this, 'handle']);
$this->wordPress->addAction(Hooks::AUTOMATION_STEP, [$this, 'handle']);
$this->addStepRunner(Step::TYPE_ACTION, $this->actionStepRunner);
$this->wordPress->doAction(Hooks::STEP_RUNNER_INITIALIZE, [$this]);
}
@ -110,57 +110,57 @@ class StepHandler {
try {
$this->handleStep($args);
} catch (Throwable $e) {
$status = $e instanceof InvalidStateException && $e->getErrorCode() === 'mailpoet_automation_workflow_not_active' ? WorkflowRun::STATUS_CANCELLED : WorkflowRun::STATUS_FAILED;
$this->workflowRunStorage->updateStatus((int)$args['workflow_run_id'], $status);
$this->postProcessWorkflowRun((int)$args['workflow_run_id']);
$status = $e instanceof InvalidStateException && $e->getErrorCode() === 'mailpoet_automation_automation_not_active' ? AutomationRun::STATUS_CANCELLED : AutomationRun::STATUS_FAILED;
$this->automationRunStorage->updateStatus((int)$args['automation_run_id'], $status);
$this->postProcessAutomationRun((int)$args['automation_run_id']);
if (!$e instanceof Exception) {
throw new Exception($e->getMessage(), intval($e->getCode()), $e);
}
throw $e;
}
$this->postProcessWorkflowRun((int)$args['workflow_run_id']);
$this->postProcessAutomationRun((int)$args['automation_run_id']);
}
private function handleStep(array $args): void {
$workflowRunId = $args['workflow_run_id'];
$automationRunId = $args['automation_run_id'];
$stepId = $args['step_id'];
$workflowRun = $this->workflowRunStorage->getWorkflowRun($workflowRunId);
if (!$workflowRun) {
throw Exceptions::workflowRunNotFound($workflowRunId);
$automationRun = $this->automationRunStorage->getAutomationRun($automationRunId);
if (!$automationRun) {
throw Exceptions::automationRunNotFound($automationRunId);
}
if ($workflowRun->getStatus() !== WorkflowRun::STATUS_RUNNING) {
throw Exceptions::workflowRunNotRunning($workflowRunId, $workflowRun->getStatus());
if ($automationRun->getStatus() !== AutomationRun::STATUS_RUNNING) {
throw Exceptions::automationRunNotRunning($automationRunId, $automationRun->getStatus());
}
$workflow = $this->workflowStorage->getWorkflow($workflowRun->getWorkflowId(), $workflowRun->getVersionId());
if (!$workflow) {
throw Exceptions::workflowVersionNotFound($workflowRun->getWorkflowId(), $workflowRun->getVersionId());
$automation = $this->automationStorage->getAutomation($automationRun->getAutomationId(), $automationRun->getVersionId());
if (!$automation) {
throw Exceptions::automationVersionNotFound($automationRun->getAutomationId(), $automationRun->getVersionId());
}
if (!in_array($workflow->getStatus(), [Workflow::STATUS_ACTIVE, Workflow::STATUS_DEACTIVATING], true)) {
throw Exceptions::workflowNotActive($workflowRun->getWorkflowId());
if (!in_array($automation->getStatus(), [Automation::STATUS_ACTIVE, Automation::STATUS_DEACTIVATING], true)) {
throw Exceptions::automationNotActive($automationRun->getAutomationId());
}
// complete workflow run
// complete automation run
if (!$stepId) {
$this->workflowRunStorage->updateStatus($workflowRunId, WorkflowRun::STATUS_COMPLETE);
$this->automationRunStorage->updateStatus($automationRunId, AutomationRun::STATUS_COMPLETE);
return;
}
$stepData = $workflow->getStep($stepId);
$stepData = $automation->getStep($stepId);
if (!$stepData) {
throw Exceptions::workflowStepNotFound($stepId);
throw Exceptions::automationStepNotFound($stepId);
}
$step = $this->registry->getStep($stepData->getKey());
$stepType = $stepData->getType();
if (isset($this->stepRunners[$stepType])) {
$log = new WorkflowRunLog($workflowRun->getId(), $stepData->getId());
$log = new AutomationRunLog($automationRun->getId(), $stepData->getId());
try {
$requiredSubjects = $step instanceof Action ? $step->getSubjectKeys() : [];
$subjectEntries = $this->getSubjectEntries($workflowRun, $requiredSubjects);
$args = new StepRunArgs($workflow, $workflowRun, $stepData, $subjectEntries);
$validationArgs = new StepValidationArgs($workflow, $stepData, array_map(function (SubjectEntry $entry) {
$subjectEntries = $this->getSubjectEntries($automationRun, $requiredSubjects);
$args = new StepRunArgs($automation, $automationRun, $stepData, $subjectEntries);
$validationArgs = new StepValidationArgs($automation, $stepData, array_map(function (SubjectEntry $entry) {
return $entry->getSubject();
}, $subjectEntries));
$this->stepRunners[$stepType]->run($args, $validationArgs);
@ -171,11 +171,11 @@ class StepHandler {
throw $e;
} finally {
try {
$this->hooks->doWorkflowStepAfterRun($log);
$this->hooks->doAutomationStepAfterRun($log);
} catch (Throwable $e) {
// Ignore integration errors
}
$this->workflowRunLogStorage->createWorkflowRunLog($log);
$this->automationRunLogStorage->createAutomationRunLog($log);
}
} else {
throw new InvalidStateException();
@ -184,33 +184,33 @@ class StepHandler {
$nextStep = $stepData->getNextSteps()[0] ?? null;
$nextStepArgs = [
[
'workflow_run_id' => $workflowRunId,
'automation_run_id' => $automationRunId,
'step_id' => $nextStep ? $nextStep->getId() : null,
],
];
$this->workflowRunStorage->updateNextStep($workflowRunId, $nextStep ? $nextStep->getId() : null);
$this->automationRunStorage->updateNextStep($automationRunId, $nextStep ? $nextStep->getId() : null);
// next step scheduled by action
if ($this->actionScheduler->hasScheduledAction(Hooks::WORKFLOW_STEP, $nextStepArgs)) {
if ($this->actionScheduler->hasScheduledAction(Hooks::AUTOMATION_STEP, $nextStepArgs)) {
return;
}
// no need to schedule a new step if the next step is null, complete the run
if (!$nextStep) {
$this->workflowRunStorage->updateStatus($workflowRunId, WorkflowRun::STATUS_COMPLETE);
$this->automationRunStorage->updateStatus($automationRunId, AutomationRun::STATUS_COMPLETE);
return;
}
// enqueue next step
$this->actionScheduler->enqueue(Hooks::WORKFLOW_STEP, $nextStepArgs);
$this->actionScheduler->enqueue(Hooks::AUTOMATION_STEP, $nextStepArgs);
// TODO: allow long-running steps (that are not done here yet)
}
/** @return SubjectEntry<Subject<Payload>>[] */
private function getSubjectEntries(WorkflowRun $workflowRun, array $requiredSubjectKeys): array {
private function getSubjectEntries(AutomationRun $automationRun, array $requiredSubjectKeys): array {
$subjectDataMap = [];
foreach ($workflowRun->getSubjects() as $data) {
foreach ($automationRun->getSubjects() as $data) {
$subjectDataMap[$data->getKey()] = array_merge($subjectDataMap[$data->getKey()] ?? [], [$data]);
}
@ -218,7 +218,7 @@ class StepHandler {
foreach ($requiredSubjectKeys as $key) {
$subjectData = $subjectDataMap[$key] ?? null;
if (!$subjectData) {
throw Exceptions::subjectDataNotFound($key, $workflowRun->getId());
throw Exceptions::subjectDataNotFound($key, $automationRun->getId());
}
foreach ($subjectData as $data) {
$subjectEntries[] = $this->subjectLoader->getSubjectEntry($data);
@ -227,26 +227,26 @@ class StepHandler {
return $subjectEntries;
}
private function postProcessWorkflowRun(int $workflowRunId): void {
$workflowRun = $this->workflowRunStorage->getWorkflowRun($workflowRunId);
if (!$workflowRun) {
private function postProcessAutomationRun(int $automationRunId): void {
$automationRun = $this->automationRunStorage->getAutomationRun($automationRunId);
if (!$automationRun) {
return;
}
$workflow = $this->workflowStorage->getWorkflow($workflowRun->getWorkflowId());
if (!$workflow) {
$automation = $this->automationStorage->getAutomation($automationRun->getAutomationId());
if (!$automation) {
return;
}
$this->postProcessWorkflow($workflow);
$this->postProcessAutomation($automation);
}
private function postProcessWorkflow(Workflow $workflow): void {
if ($workflow->getStatus() === Workflow::STATUS_DEACTIVATING) {
$activeRuns = $this->workflowRunStorage->getCountForWorkflow($workflow, WorkflowRun::STATUS_RUNNING);
private function postProcessAutomation(Automation $automation): void {
if ($automation->getStatus() === Automation::STATUS_DEACTIVATING) {
$activeRuns = $this->automationRunStorage->getCountForAutomation($automation, AutomationRun::STATUS_RUNNING);
// Set a deactivating Workflow to draft once all workflow runs are finished.
// Set a deactivating Automation to draft once all automation runs are finished.
if ($activeRuns === 0) {
$workflow->setStatus(Workflow::STATUS_DRAFT);
$this->workflowStorage->updateWorkflow($workflow);
$automation->setStatus(Automation::STATUS_DRAFT);
$this->automationStorage->updateAutomation($automation);
}
}
}