TaskAdder.php — MailPoet – Newsletters, Email Marketing, and Automation 5.34.2

mailpoet / lib / Cron / CliCommands / TaskAdder.php

mailpoet / lib / Cron / CliCommands Last commit date

ClaimedTaskRunner.php 1 month ago Cli.php 1 month ago CronCommand.php 1 month ago DaemonRunner.php 1 month ago ExecutionLimitOverride.php 1 month ago ScheduledTaskResolver.php 1 month ago ScheduledTasksLister.php 1 month ago TaskAdder.php 1 month ago TaskCanceller.php 1 month ago TaskRunner.php 1 month ago TaskTrigger.php 1 month ago WorkerTypesCatalog.php 1 month ago index.php 1 month ago

TaskAdder.php

155 lines

1	<?php declare(strict_types = 1);
2
3	namespace MailPoet\Cron\CliCommands;
4
5	if (!defined('ABSPATH')) exit;
6
7
8	use Exception;
9	use InvalidArgumentException;
10	use MailPoet\Entities\ScheduledTaskEntity;
11	use MailPoet\Newsletter\Sending\ScheduledTasksRepository;
12	use MailPoetVendor\Carbon\Carbon;
13
14	/**
15	* Adds a new scheduled task for a standard cron worker, optionally claiming and running it in-process.
16	*
17	* Only standard CronWorkerInterface workers are addable (see WorkerTypesCatalog::getAddableTypes());
18	* mailing 'sending'/'stats_notification' rows are created by app flows and are rejected. --run drives a
19	* freshly claimed row through ClaimedTaskRunner, shared with `cron run --task-id`.
20	*
21	* See doc/wp-cli-cron-commands.md for command behaviour.
22	*/
23	class TaskAdder {
24	const PRIORITY_MAP = [
25	'high' => ScheduledTaskEntity::PRIORITY_HIGH,
26	'medium' => ScheduledTaskEntity::PRIORITY_MEDIUM,
27	'low' => ScheduledTaskEntity::PRIORITY_LOW,
28	];
29
30	private WorkerTypesCatalog $workerTypesCatalog;
31
32	private ScheduledTasksRepository $scheduledTasksRepository;
33
34	private ClaimedTaskRunner $claimedTaskRunner;
35
36	public function __construct(
37	WorkerTypesCatalog $workerTypesCatalog,
38	ScheduledTasksRepository $scheduledTasksRepository,
39	ClaimedTaskRunner $claimedTaskRunner
40	) {
41	$this->workerTypesCatalog = $workerTypesCatalog;
42	$this->scheduledTasksRepository = $scheduledTasksRepository;
43	$this->claimedTaskRunner = $claimedTaskRunner;
44	}
45
46	/**
47	* @return array{id: int, type: string, action: string, message: string, run: array{completed: bool, limit_reached: bool, message: string}\|null}
48	*/
49	public function add(string $type, ?string $at, ?int $in, string $priority, bool $force, bool $run): array {
50	$this->workerTypesCatalog->assertAddableType($type);
51	$priorityValue = $this->resolvePriority($priority);
52
53	if ($run && ($at !== null \|\| $in !== null)) {
54	throw new InvalidArgumentException('--run cannot be combined with --at or --in. A claimed task runs immediately.');
55	}
56
57	if ($run) {
58	return $this->claimAndRun($type, $priorityValue);
59	}
60
61	$scheduledAt = $this->resolveScheduledAt($at, $in);
62
63	if (!$force) {
64	$existing = $this->scheduledTasksRepository->findScheduledTask($type);
65	if ($existing instanceof ScheduledTaskEntity) {
66	$existingId = (int)$existing->getId();
67	return [
68	'id' => $existingId,
69	'type' => $type,
70	'action' => 'duplicate',
71	'message' => sprintf("A task of type '%s' is already scheduled as task %d. Use --force to add another.", $type, $existingId),
72	'run' => null,
73	];
74	}
75	}
76
77	$task = $this->createTask($type, $priorityValue, $scheduledAt);
78
79	return [
80	'id' => (int)$task->getId(),
81	'type' => $type,
82	'action' => 'created',
83	'message' => sprintf("Added task %d (%s), scheduled for %s, priority %s.", $task->getId(), $type, $scheduledAt->format('Y-m-d H:i:s'), $priority),
84	'run' => null,
85	];
86	}
87
88	private function createTask(string $type, int $priority, Carbon $scheduledAt): ScheduledTaskEntity {
89	$task = new ScheduledTaskEntity();
90	$task->setType($type);
91	$task->setStatus(ScheduledTaskEntity::STATUS_SCHEDULED);
92	$task->setPriority($priority);
93	$task->setScheduledAt($scheduledAt);
94	$this->scheduledTasksRepository->persist($task);
95	$this->scheduledTasksRepository->flush();
96	return $task;
97	}
98
99	/**
100	* Claims a fresh row (status STATUS_CLI) and processes exactly that one task. The duplicate check is
101	* intentionally skipped: the claim is a fresh, independently-owned row, never a "scheduled
102	* duplicate", so --run ignores --force entirely.
103	*
104	* @return array{id: int, type: string, action: string, message: string, run: array{completed: bool, limit_reached: bool, message: string}}
105	*/
106	private function claimAndRun(string $type, int $priority): array {
107	$worker = $this->workerTypesCatalog->getWorkerByType($type);
108	if ($worker === null) {
109	// Unreachable: assertAddableType already restricts to types with a standard worker.
110	throw new InvalidArgumentException("Task type '{$type}' has no runnable worker.");
111	}
112
113	$task = $this->claimedTaskRunner->claimNew($type, $priority);
114	$taskId = (int)$task->getId();
115
116	$runResult = $this->claimedTaskRunner->run($worker, $task);
117
118	return [
119	'id' => $taskId,
120	'type' => $type,
121	'action' => 'claimed',
122	'message' => sprintf("Claimed task %d (%s) and ran it in this process.", $taskId, $type),
123	'run' => $runResult,
124	];
125	}
126
127	private function resolveScheduledAt(?string $at, ?int $in): Carbon {
128	if ($at !== null && $in !== null) {
129	throw new InvalidArgumentException('--at and --in cannot be used together. Pick one.');
130	}
131
132	if ($at !== null) {
133	try {
134	return Carbon::parse($at)->millisecond(0);
135	} catch (Exception $e) {
136	throw new InvalidArgumentException("Could not parse --at value '{$at}'. Use a date/time like '2026-01-01 09:00' or 'tomorrow 8am'.");
137	}
138	}
139
140	if ($in !== null) {
141	return Carbon::now()->millisecond(0)->addSeconds($in);
142	}
143
144	return Carbon::now()->millisecond(0);
145	}
146
147	private function resolvePriority(string $priority): int {
148	if (!isset(self::PRIORITY_MAP[$priority])) {
149	$valid = implode(', ', array_keys(self::PRIORITY_MAP));
150	throw new InvalidArgumentException("Invalid priority '{$priority}'. Valid values: {$valid}.");
151	}
152	return self::PRIORITY_MAP[$priority];
153	}
154	}
155