X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=src%2FProtocol%2FActivityPub%2FQueue.php;h=9180de08f26d8322a52471a09d74e6fd1bff21c2;hb=5f5298125530857f009ca236139853954511a94e;hp=2fa95897c67f6cd0aeba351a06487d42c5cb9de0;hpb=196a1de7f21c3b7dd7f2212646b52af047033048;p=friendica.git diff --git a/src/Protocol/ActivityPub/Queue.php b/src/Protocol/ActivityPub/Queue.php index 2fa95897c6..9180de08f2 100644 --- a/src/Protocol/ActivityPub/Queue.php +++ b/src/Protocol/ActivityPub/Queue.php @@ -1,6 +1,6 @@ $entry['id']]); + DBA::delete('inbox-entry', ['id' => $entry['id']]); + + $children = DBA::select('inbox-entry', ['id'], ['in-reply-to-id' => $entry['object-id']]); + while ($child = DBA::fetch($children)) { + self::deleteById($child['id']); + } + DBA::close($children); } /** @@ -187,6 +197,11 @@ class Queue return false; } + if (!self::isProcessable($id)) { + Logger::debug('Other queue entries need to be processed first.', ['id' => $id]); + return false; + } + if (!empty($entry['wid'])) { $worker = DI::app()->getQueue(); $wid = $worker['id'] ?? 0; @@ -199,7 +214,7 @@ class Queue } } - Logger::debug('Processing queue entry', ['id' => $entry['id'], 'type' => $entry['type'], 'object-type' => $entry['object-type'], 'uri' => $entry['object-id'], 'in-reply-to' => $entry['in-reply-to-id']]); + Logger::debug('Processing queue entry', ['id' => $entry['id'], 'type' => $entry['type'], 'object-type' => $entry['object-type'], 'uri' => $entry['object-id'], 'in-reply-to' => $entry['in-reply-to-id'], 'callstack' => System::callstack(20)]); $activity = json_decode($entry['activity'], true); $type = $entry['type']; @@ -209,6 +224,10 @@ class Queue $activity['worker-id'] = $entry['wid']; $activity['recursion-depth'] = 0; + if (empty($activity['thread-children-type'])) { + $activity['thread-children-type'] = $type; + } + $receivers = DBA::select('inbox-entry-receiver', ['uid'], ["`queue-id` = ? AND `uid` != ?", $entry['id'], 0]); while ($receiver = DBA::fetch($receivers)) { if (!in_array($receiver['uid'], $activity['receiver'])) { @@ -217,7 +236,7 @@ class Queue } DBA::close($receivers); - if (!Receiver::routeActivities($activity, $type, $push, $fetch_parents)) { + if (!Receiver::routeActivities($activity, $type, $push, $fetch_parents, $activity['receiver'][0] ?? 0)) { self::remove($activity); } @@ -247,6 +266,34 @@ class Queue DBA::close($entries); } + public static function isProcessable(int $id): bool + { + $entry = DBA::selectFirst('inbox-entry', [], ['id' => $id]); + if (empty($entry)) { + return false; + } + + if (!empty($entry['object-id']) && Post::exists(['uri' => $entry['object-id']])) { + // The object already exists, so processing can be done + return true; + } + + if (!empty($entry['conversation'])) { + $conv_id = ItemURI::getIdByURI($entry['conversation'], false); + if (DBA::exists('post-thread', ['conversation-id' => $conv_id])) { + // We have got the conversation in the system, so the post can be processed + return true; + } + } + + if (!empty($entry['object-id']) && !empty($entry['in-reply-to-id']) && ($entry['object-id'] != $entry['in-reply-to-id']) && DBA::exists('inbox-entry', ['object-id' => $entry['in-reply-to-id']])) { + // This entry belongs to some other entry that should be processed first + return false; + } + + return true; + } + /** * Clear old activities * @@ -256,7 +303,11 @@ class Queue { // We delete all entries that aren't associated with a worker entry after seven days. // The other entries are deleted when the worker deferred for too long. - DBA::delete('inbox-entry', ["`wid` IS NULL AND `received` < ?", DateTimeFormat::utc('now - 7 days')]); + $entries = DBA::select('inbox-entry', ['id'], ["`wid` IS NULL AND `received` < ?", DateTimeFormat::utc('now - 7 days')]); + while ($entry = DBA::fetch($entries)) { + self::deleteById($entry['id']); + } + DBA::close($entries); // Optimizing this table only last seconds if (DI::config()->get('system', 'optimize_tables')) {