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

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

226 lines 7.2 KB

Raw Download Zip

1	<?php
2	/**
3	* Queue item repository.
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\Interfaces\QueueItemRepository;
13	use SeQura\Core\Infrastructure\ORM\QueryFilter\QueryFilter;
14	use SeQura\Core\Infrastructure\ORM\Utility\IndexHelper;
15	use SeQura\Core\Infrastructure\TaskExecution\Exceptions\QueueItemSaveException;
16	use SeQura\Core\Infrastructure\TaskExecution\Interfaces\Priority;
17	use SeQura\Core\Infrastructure\TaskExecution\QueueItem;
18	use SeQura\WC\Dto\Table_Index;
19	use SeQura\WC\Dto\Table_Index_Column;
20
21	/**
22	* Queue item repository.
23	*/
24	class Queue_Item_Repository extends Repository implements QueueItemRepository {
25
26	/**
27	* Returns unprefixed table name.
28	*/
29	protected function get_unprefixed_table_name(): string {
30	return 'sequra_queue';
31	}
32
33	/**
34	* Finds list of earliest queued queue items per queue. Following list of criteria for searching must be satisfied:
35	* - Queue must be without already running queue items
36	* - For one queue only one (oldest queued) item should be returned
37	*
38	* @param int $priority Queue item priority.
39	* @param int $limit Result set limit. By default max 10 earliest queue items will be returned.
40	*
41	* @return Entity[] Found queue item list
42	*/
43	public function findOldestQueuedItems( $priority, $limit = 10 ) {
44	if ( ! $this->table_exists() \|\| Priority::NORMAL !== $priority ) {
45	return array();
46	}
47
48	/**
49	* Entity object.
50	*
51	* @var Entity $entity
52	*/
53	$entity = new $this->entity_class();
54	$type = $this->escape_value( $entity->getConfig()->getType() );
55	$index_map = IndexHelper::mapFieldsToIndexes( $entity );
56
57	$status_index = 'index_' . $index_map['status'];
58	$queue_name_index = 'index_' . $index_map['queueName'];
59
60	$running_queues_query = "SELECT $queue_name_index FROM `{$this->get_table_name()}` q2 WHERE q2.`$status_index` = '"
61	. QueueItem::IN_PROGRESS . "' AND q2.`type` = $type";
62
63	$sql = "SELECT queueTable.*
64	FROM (
65	SELECT $queue_name_index, MIN(id) AS id
66	FROM `{$this->get_table_name()}` AS q
67	WHERE q.`type` = $type AND q.`$status_index` = '" . QueueItem::QUEUED . "' AND q.`$queue_name_index` NOT IN ($running_queues_query)
68	GROUP BY `$queue_name_index` LIMIT $limit
69	) AS queueView
70	INNER JOIN `{$this->get_table_name()}` as queueTable
71	ON queueView.id = queueTable.id";
72
73	$result = $this->db->get_results( $sql, ARRAY_A );
74	if ( ! is_array( $result ) ) {
75	$result = array();
76	}
77	$pending_items = $limit - count( $result );
78
79	if ( $pending_items > 0 && $this->table_exists( true ) ) {
80	$legacy_result = $this->db->get_results( str_replace( $this->get_table_name(), $this->get_legacy_table_name(), $sql ), ARRAY_A );
81	if ( is_array( $legacy_result ) ) {
82	$length = count( $legacy_result );
83	for ( $i = 0; $i < $length && $pending_items > 0; $i++ ) {
84	$result[] = $legacy_result[ $i ];
85	--$pending_items;
86	}
87	}
88	}
89
90	return $this->translateToEntities( $result );
91	}
92
93	/**
94	* Creates or updates given queue item. If queue item id is not set, new queue item will be created otherwise update will be performed.
95	*
96	* @param QueueItem $queue_item Item to save.
97	* @param mixed[] $additional_where List of key/value pairs that must be satisfied upon saving queue item.
98	* Key is queue item property and value is condition value for that property.
99	*
100	* @return int Id of saved queue item.
101	* @throws QueueItemSaveException If queue item could not be saved.
102	*/
103	public function saveWithCondition( QueueItem $queue_item, array $additional_where = array() ): int {
104	if ( ! $this->table_exists() ) {
105	return -1;
106	}
107	$item_id = null;
108	try {
109	$queue_item_id = $queue_item->getId();
110	if ( null === $queue_item_id \|\| $queue_item_id <= 0 ) {
111	$item_id = $this->save( $queue_item );
112	} else {
113	$filter = $this->build_query_filter(
114	array_merge( $additional_where, array( 'id' => $queue_item->getId() ) )
115	);
116
117	if ( null === $this->selectOne( $filter ) ) {
118	throw new QueueItemSaveException( \esc_html( 'Failed to save queue item, update condition(s) not met.' ) );
119	}
120	$item_id = $this->save( $queue_item );
121	}
122	} catch ( \Exception $exception ) {
123	throw new QueueItemSaveException(
124	\esc_html( 'Failed to save queue item with id: ' . $item_id ),
125	0,
126	$exception // phpcs:ignore WordPress.Security.EscapeOutput.ExceptionNotEscaped
127	);
128	}
129
130	return $item_id;
131	}
132
133	/**
134	* Updates status of a batch of queue items.
135	*
136	* @param mixed[] $ids
137	* @param string $status
138	*
139	* @return void
140	*/
141	public function batchStatusUpdate( array $ids, $status ): void {
142	// Not used in this implementation.
143	}
144
145	/**
146	* Builds query filter from conditions array.
147	*
148	* @noinspection PhpDocMissingThrowsInspection
149	*
150	* @param mixed[]$conditions Array of conditions.
151	*
152	* @return QueryFilter Query filter object.
153	*/
154	private function build_query_filter( array $conditions ) {
155	$filter = new QueryFilter();
156	$filter->setOffset( 0 );
157	$filter->setLimit( 1 );
158	foreach ( $conditions as $column => $value ) {
159	if ( null === $value ) {
160	$filter->where( $column, 'IS' );
161	} else {
162	$filter->where( $column, '=', $value );
163	}
164	}
165
166	return $filter;
167	}
168
169	/**
170	* Get the index column name that stores the store ID.
171	*
172	* @return string Index column name or empty string if not applicable.
173	*/
174	protected function get_store_id_index_column(): string {
175	return '';
176	}
177
178	/**
179	* Get a list of indexes that are required for the table.
180	*
181	* @return Table_Index[] The list of indexes.
182	*/
183	public function get_required_indexes() {
184
185	$type_col = new Table_Index_Column( 'type', 64 );
186	$index_1_col = new Table_Index_Column( 'index_1', 64 );
187	$index_2_col = new Table_Index_Column( 'index_2', 64 );
188	$index_3_col = new Table_Index_Column( 'index_3', 64 );
189	$index_4_col = new Table_Index_Column( 'index_4', 64 );
190
191	return array_merge(
192	parent::get_required_indexes(),
193	array(
194	new Table_Index( $this->get_table_name() . '_type_index_1', array( $type_col, $index_1_col ) ),
195	new Table_Index( $this->get_table_name() . '_type_index_2', array( $type_col, $index_2_col ) ),
196	new Table_Index( $this->get_table_name() . '_type_index_3', array( $type_col, $index_3_col ) ),
197	new Table_Index( $this->get_table_name() . '_type_index_4', array( $type_col, $index_4_col ) ),
198	)
199	);
200	}
201
202	/**
203	* Get the SQL statement to create the table without the indexes definition.
204	* Resulting string should include an additional %s placeholder for the indexes.
205	*
206	* @return string The SQL statement to create the table.
207	*/
208	protected function get_create_table_sql() {
209	$charset_collate = $this->db->get_charset_collate();
210	return "CREATE TABLE {$this->get_table_name()} (
211	`id` BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
212	`type` VARCHAR(255),
213	`index_1` VARCHAR(127),
214	`index_2` VARCHAR(127),
215	`index_3` VARCHAR(127),
216	`index_4` VARCHAR(127),
217	`index_5` VARCHAR(127),
218	`index_6` BIGINT UNSIGNED,
219	`index_7` BIGINT UNSIGNED,
220	`index_8` BIGINT UNSIGNED,
221	`index_9` BIGINT UNSIGNED,
222	`data` LONGTEXT,
223	PRIMARY KEY (id) %s) $charset_collate;";
224	}
225	}
226