sequra / src / Repositories / class-queue-item-repository.php

class-queue-item-repository.php in seQura 3.1.1, at src/Repositories/class-queue-item-repository.php

260 lines 7.9 KB

Raw Download Zip

1	<?php
2	/**
3	* Settings
4	*
5	* @package SeQura/WC
6	* @subpackage SeQura/WC/Repositories
7	*/
8
9	namespace SeQura\WC\Repositories;
10
11	use SeQura\Core\Infrastructure\ORM\Entity;
12	use SeQura\Core\Infrastructure\ORM\Exceptions\QueryFilterInvalidParamException;
13	use SeQura\Core\Infrastructure\ORM\Interfaces\QueueItemRepository;
14	use SeQura\Core\Infrastructure\ORM\QueryFilter\QueryFilter;
15	use SeQura\Core\Infrastructure\ORM\Utility\IndexHelper;
16	use SeQura\Core\Infrastructure\TaskExecution\Exceptions\QueueItemSaveException;
17	use SeQura\Core\Infrastructure\TaskExecution\Interfaces\Priority;
18	use SeQura\Core\Infrastructure\TaskExecution\QueueItem;
19
20	/**
21	* Class Base_Repository
22	*/
23	class Queue_Item_Repository extends Repository implements QueueItemRepository {
24
25	/**
26	* Returns unprefixed table name.
27	*/
28	protected function get_unprefixed_table_name(): string {
29	return 'sequra_queue';
30	}
31
32	/**
33	* Finds list of earliest queued queue items per queue. Following list of criteria for searching must be satisfied:
34	* - Queue must be without already running queue items
35	* - For one queue only one (oldest queued) item should be returned
36	*
37	* @param int $priority Queue item priority.
38	* @param int $limit Result set limit. By default max 10 earliest queue items will be returned.
39	*
40	* @return Entity[] Found queue item list
41	*/
42	public function findOldestQueuedItems( $priority, $limit = 10 ) {
43	if ( ! $this->table_exists() \|\| Priority::NORMAL !== $priority ) {
44	return array();
45	}
46
47	/**
48	* Entity object.
49	*
50	* @var Entity $entity
51	*/
52	$entity = new $this->entity_class();
53	$type = $this->escape_value( $entity->getConfig()->getType() );
54	$index_map = IndexHelper::mapFieldsToIndexes( $entity );
55
56	$status_index = 'index_' . $index_map['status'];
57	$queue_name_index = 'index_' . $index_map['queueName'];
58
59	$running_queues_query = "SELECT $queue_name_index FROM `{$this->get_table_name()}` q2 WHERE q2.`$status_index` = '"
60	. QueueItem::IN_PROGRESS . "' AND q2.`type` = $type";
61
62	$sql = "SELECT queueTable.*
63	FROM (
64	SELECT $queue_name_index, MIN(id) AS id
65	FROM `{$this->get_table_name()}` AS q
66	WHERE q.`type` = $type AND q.`$status_index` = '" . QueueItem::QUEUED . "' AND q.`$queue_name_index` NOT IN ($running_queues_query)
67	GROUP BY `$queue_name_index` LIMIT $limit
68	) AS queueView
69	INNER JOIN `{$this->get_table_name()}` as queueTable
70	ON queueView.id = queueTable.id";
71
72	$result = $this->db->get_results( $sql, ARRAY_A );
73	if ( ! is_array( $result ) ) {
74	return array();
75	}
76
77	return $this->translateToEntities( $result );
78	}
79
80	/**
81	* Creates or updates given queue item. If queue item id is not set, new queue item will be created otherwise update will be performed.
82	*
83	* @param QueueItem $queue_item Item to save.
84	* @param mixed[] $additional_where List of key/value pairs that must be satisfied upon saving queue item.
85	* Key is queue item property and value is condition value for that property.
86	*
87	* @return int Id of saved queue item.
88	* @throws QueueItemSaveException If queue item could not be saved.
89	*/
90	public function saveWithCondition( QueueItem $queue_item, array $additional_where = array() ): int {
91	if ( ! $this->table_exists() ) {
92	return -1;
93	}
94	$item_id = null;
95	try {
96	$queue_item_id = $queue_item->getId();
97	if ( null === $queue_item_id \|\| $queue_item_id <= 0 ) {
98	$item_id = $this->save( $queue_item );
99	} else {
100	$this->update_queue_item( $queue_item, $additional_where );
101	$item_id = $queue_item_id;
102	}
103	} catch ( \Exception $exception ) {
104	throw new QueueItemSaveException(
105	\esc_html( 'Failed to save queue item with id: ' . $item_id ),
106	0,
107	$exception // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped
108	);
109	}
110
111	return $item_id;
112	}
113
114	/**
115	* Updates status of a batch of queue items.
116	*
117	* @param mixed[] $ids
118	* @param string $status
119	*
120	* @return void
121	*/
122	public function batchStatusUpdate( array $ids, $status ): void {
123	// Not used in this implementation.
124	}
125
126	/**
127	* Updates database record with data from provided $queueItem.
128	*
129	* @param QueueItem $queue_item Queue item.
130	* @param mixed[] $conditions Array of update conditions.
131	*
132	* @throws QueueItemSaveException Queue item save exception.
133	* @throws QueryFilterInvalidParamException If filter condition is invalid.
134	*/
135	private function update_queue_item( QueueItem $queue_item, array $conditions = array() ): void {
136	$conditions = array_merge( $conditions, array( 'id' => $queue_item->getId() ) );
137
138	$item = $this->select_for_update( $conditions );
139	$this->check_if_record_exists( $item );
140
141	if ( null !== $item ) {
142	$this->update_with_condition( $queue_item, $conditions );
143	}
144	}
145
146	/**
147	* Executes select query for update.
148	*
149	* @param mixed[]$conditions Array of update conditions.
150	*
151	* @return QueueItem\|null First found entity or NULL.
152	* @throws QueryFilterInvalidParamException If filter condition is invalid.
153	*/
154	private function select_for_update( array $conditions ) {
155	/**
156	* Entity object.
157	*
158	* @var Entity $entity
159	*/
160	$entity = new $this->entity_class();
161	$type = $entity->getConfig()->getType();
162	$field_index_map = IndexHelper::mapFieldsToIndexes( $entity );
163
164	$filter = $this->build_query_filter( $conditions );
165
166	$query = "SELECT * FROM {$this->get_table_name()} WHERE type = '$type' ";
167	$query .= $this->apply_query_filter( $filter, $field_index_map );
168	$query .= ' FOR UPDATE';
169
170	$raw_results = $this->db->get_results( $query, ARRAY_A );
171
172	if ( ! is_array( $raw_results ) ) {
173	return null;
174	}
175
176	/**
177	* Entities
178	*
179	* @var QueueItem[] $entities
180	*/
181	$entities = $this->translateToEntities( $raw_results );
182
183	return ! empty( $entities ) ? $entities[0] : null;
184	}
185
186	/**
187	* Builds query filter from conditions array.
188	*
189	* @noinspection PhpDocMissingThrowsInspection
190	*
191	* @param mixed[]$conditions Array of conditions.
192	*
193	* @return QueryFilter Query filter object.
194	*/
195	private function build_query_filter( array $conditions ) {
196	$filter = new QueryFilter();
197	$filter->setOffset( 0 );
198	$filter->setLimit( 1 );
199	foreach ( $conditions as $column => $value ) {
200	if ( null === $value ) {
201	$filter->where( $column, 'IS' );
202	} else {
203	$filter->where( $column, '=', $value );
204	}
205	}
206
207	return $filter;
208	}
209
210	/**
211	* Validates if item exists.
212	*
213	* @param QueueItem $item Queue item.
214	*
215	* @throws QueueItemSaveException Queue item save exception.
216	*/
217	private function check_if_record_exists( QueueItem $item = null ): void {
218	if ( null === $item ) {
219	$message = 'Failed to save queue item, update condition(s) not met.';
220	throw new QueueItemSaveException( esc_html( $message ) );
221	}
222	}
223
224	/**
225	* Updates single record.
226	*
227	* @param QueueItem $item Queue item.
228	* @param mixed[] $conditions List of simple search filters as key-value pair to find records to update.
229	*
230	* @return bool TRUE if operation succeeded; otherwise, FALSE.
231	*
232	* @throws \InvalidArgumentException Invalid argument.
233	*/
234	private function update_with_condition( QueueItem $item, array $conditions ) {
235	$field_index_map = IndexHelper::mapFieldsToIndexes( $item );
236	$prepared = $this->prepare_entity_for_storage( $item );
237
238	$indexed_conditions = array();
239	foreach ( $conditions as $key => $value ) {
240	if ( 'id' === $key ) {
241	$indexed_conditions[ $key ] = intval( $value );
242	} else {
243	$indexed_conditions[ 'index_' . $field_index_map[ $key ] ] = IndexHelper::castFieldValue( $value, gettype( $value ) );
244	}
245	}
246
247	// Only one record should be updated.
248	return 1 === $this->db->update( $this->get_table_name(), $prepared, $indexed_conditions );
249	}
250
251	/**
252	* Get the index column name that stores the store ID.
253	*
254	* @return string Index column name or empty string if not applicable.
255	*/
256	protected function get_store_id_index_column(): string {
257	return '';
258	}
259	}
260