|
18 | 18 | use craft\events\IndexKeywordsEvent; |
19 | 19 | use craft\events\SearchEvent; |
20 | 20 | use craft\helpers\ArrayHelper; |
| 21 | +use craft\helpers\Component as ComponentHelper; |
21 | 22 | use craft\helpers\Db; |
22 | 23 | use craft\helpers\ElementHelper; |
| 24 | +use craft\helpers\Queue; |
23 | 25 | use craft\helpers\Search as SearchHelper; |
24 | 26 | use craft\helpers\StringHelper; |
| 27 | +use craft\queue\jobs\UpdateSearchIndex; |
25 | 28 | use craft\search\SearchQuery; |
26 | 29 | use craft\search\SearchQueryTerm; |
27 | 30 | use craft\search\SearchQueryTermGroup; |
28 | 31 | use Throwable; |
29 | 32 | use yii\base\Component; |
30 | | -use yii\db\Exception; |
| 33 | +use yii\base\Exception; |
| 34 | +use yii\db\Exception as DbException; |
31 | 35 | use yii\db\Expression; |
32 | 36 | use yii\db\Schema; |
33 | 37 |
|
@@ -197,6 +201,157 @@ public function indexElementAttributes(ElementInterface $element, ?array $fieldH |
197 | 201 | return true; |
198 | 202 | } |
199 | 203 |
|
| 204 | + /** |
| 205 | + * Queues up an element to be indexed. |
| 206 | + * |
| 207 | + * @param ElementInterface $element |
| 208 | + * @param string[] $fieldHandles |
| 209 | + * @since 4.15.0 |
| 210 | + */ |
| 211 | + public function queueIndexElement(ElementInterface $element, array $fieldHandles): void |
| 212 | + { |
| 213 | + $this->createOrUpdateIndexJob($element, $fieldHandles); |
| 214 | + |
| 215 | + Queue::push(new UpdateSearchIndex([ |
| 216 | + 'elementType' => get_class($element), |
| 217 | + 'elementId' => $element->id, |
| 218 | + 'siteId' => $element->siteId, |
| 219 | + 'queued' => true, |
| 220 | + ]), 2048); |
| 221 | + } |
| 222 | + |
| 223 | + private function createOrUpdateIndexJob(ElementInterface $element, array $fieldHandles): void |
| 224 | + { |
| 225 | + $jobId = $this->pendingIndexJobId($element->id, $element->siteId); |
| 226 | + |
| 227 | + if ($jobId) { |
| 228 | + if (empty($fieldHandles)) { |
| 229 | + // nothing more to do |
| 230 | + return; |
| 231 | + } |
| 232 | + |
| 233 | + // get a lock on the job and make sure it still exists && is pending |
| 234 | + $mutex = Craft::$app->getMutex(); |
| 235 | + $lockName = "searchqueue:$jobId"; |
| 236 | + |
| 237 | + if ($mutex->acquire($lockName)) { |
| 238 | + try { |
| 239 | + if ($this->isIndexJobPending($jobId)) { |
| 240 | + foreach ($fieldHandles as $fieldHandle) { |
| 241 | + Db::upsert(Table::SEARCHINDEXQUEUE_FIELDS, [ |
| 242 | + 'jobId' => $jobId, |
| 243 | + 'fieldHandle' => $fieldHandle, |
| 244 | + ]); |
| 245 | + } |
| 246 | + return; |
| 247 | + } |
| 248 | + } finally { |
| 249 | + $mutex->release($lockName); |
| 250 | + } |
| 251 | + } |
| 252 | + } |
| 253 | + |
| 254 | + $db = Craft::$app->getDb(); |
| 255 | + $db->transaction(function() use ($element, $fieldHandles, $db) { |
| 256 | + Db::insert(Table::SEARCHINDEXQUEUE, [ |
| 257 | + 'elementId' => $element->id, |
| 258 | + 'siteId' => $element->siteId, |
| 259 | + ], $db); |
| 260 | + $jobId = $db->getLastInsertID(Table::SEARCHINDEXQUEUE); |
| 261 | + $fieldData = array_map(fn(string $fieldHandle) => [$jobId, $fieldHandle], $fieldHandles); |
| 262 | + Db::batchInsert(Table::SEARCHINDEXQUEUE_FIELDS, ['jobId', 'fieldHandle'], $fieldData, $db); |
| 263 | + }); |
| 264 | + } |
| 265 | + |
| 266 | + /** |
| 267 | + * Indexes the attributes of a given element, only if it’s queued. |
| 268 | + * |
| 269 | + * @param int $elementId |
| 270 | + * @param int $siteId |
| 271 | + * @param class-string<ElementInterface>|null $elementType |
| 272 | + * @since 4.15.0 |
| 273 | + */ |
| 274 | + public function indexElementIfQueued(int $elementId, int $siteId, ?string $elementType = null): void |
| 275 | + { |
| 276 | + $jobId = $this->pendingIndexJobId($elementId, $siteId); |
| 277 | + |
| 278 | + if (!$jobId) { |
| 279 | + // no pending jobs |
| 280 | + return; |
| 281 | + } |
| 282 | + |
| 283 | + // get a lock on the job and then mark it as reserved, |
| 284 | + // so there's no chance another process reserves it/modifies it, overlapping with this one |
| 285 | + $mutex = Craft::$app->getMutex(); |
| 286 | + $lockName = "searchqueue:$jobId"; |
| 287 | + if (!$mutex->acquire($lockName, 5)) { |
| 288 | + throw new Exception("Unable to acquire a mutex lock for search queue job $jobId"); |
| 289 | + } |
| 290 | + |
| 291 | + try { |
| 292 | + if (!Db::update(Table::SEARCHINDEXQUEUE, ['reserved' => true], ['id' => $jobId])) { |
| 293 | + // another process must be handling the same job |
| 294 | + return; |
| 295 | + } |
| 296 | + } finally { |
| 297 | + $mutex->release($lockName); |
| 298 | + } |
| 299 | + |
| 300 | + try { |
| 301 | + // Figure out which fields need to be updated for the element |
| 302 | + $fieldHandles = (new Query()) |
| 303 | + ->select(['fieldHandle']) |
| 304 | + ->from(Table::SEARCHINDEXQUEUE_FIELDS) |
| 305 | + ->where(['jobId' => $jobId]) |
| 306 | + ->column(); |
| 307 | + |
| 308 | + $elementType ??= Craft::$app->getElements()->getElementTypeById($elementId); |
| 309 | + |
| 310 | + if ($elementType && ComponentHelper::validateComponentClass($elementType, ElementInterface::class)) { |
| 311 | + $element = $elementType::find() |
| 312 | + ->drafts(null) |
| 313 | + ->provisionalDrafts(null) |
| 314 | + ->id($elementId) |
| 315 | + ->siteId($siteId) |
| 316 | + ->status(null) |
| 317 | + ->one(); |
| 318 | + |
| 319 | + if ($elementId) { |
| 320 | + $this->indexElementAttributes($element, $fieldHandles); |
| 321 | + } |
| 322 | + } |
| 323 | + |
| 324 | + Db::delete(Table::SEARCHINDEXQUEUE, ['id' => $jobId]); |
| 325 | + } catch (Throwable $e) { |
| 326 | + Db::update(Table::SEARCHINDEXQUEUE, ['reserved' => false], ['id' => $jobId]); |
| 327 | + throw $e; |
| 328 | + } |
| 329 | + } |
| 330 | + |
| 331 | + private function pendingIndexJobId(int $elementId, int $siteId): ?int |
| 332 | + { |
| 333 | + return (new Query()) |
| 334 | + ->select('id') |
| 335 | + ->from(Table::SEARCHINDEXQUEUE) |
| 336 | + ->where([ |
| 337 | + 'elementId' => $elementId, |
| 338 | + 'siteId' => $siteId, |
| 339 | + 'reserved' => false, |
| 340 | + ]) |
| 341 | + ->scalar(); |
| 342 | + } |
| 343 | + |
| 344 | + private function isIndexJobPending(int $jobId): bool |
| 345 | + { |
| 346 | + $job = (new Query()) |
| 347 | + ->select('reserved') |
| 348 | + ->from(Table::SEARCHINDEXQUEUE) |
| 349 | + ->where(['id' => $jobId]) |
| 350 | + ->one(); |
| 351 | + |
| 352 | + return $job && !$job['reserved']; |
| 353 | + } |
| 354 | + |
200 | 355 | /** |
201 | 356 | * Returns whether we should search for the resulting elements up front via [[searchElements()]], |
202 | 357 | * rather than supply a subquery which should be applied to the main element query via [[createDbQuery()]]. |
@@ -488,7 +643,7 @@ private function _indexKeywords(ElementInterface $element, string $keywords, ?st |
488 | 643 | try { |
489 | 644 | Db::insert(Table::SEARCHINDEX, $columns); |
490 | 645 | return; |
491 | | - } catch (Exception $e) { |
| 646 | + } catch (DbException $e) { |
492 | 647 | if (str_contains($e->getPrevious()?->getMessage(), 'deadlock')) { |
493 | 648 | // A gap lock was probably hit. Try again in one second |
494 | 649 | // https://github.com/craftcms/cms/issues/15221 |
|
0 commit comments