X-Git-Url: https://git.mxchange.org/?a=blobdiff_plain;f=src%2FProtocol%2FActivityPub%2FQueue.php;h=9180de08f26d8322a52471a09d74e6fd1bff21c2;hb=6e3602591d52771f7f3c8c4c0d926a1690891730;hp=011101eed7cb06bded425f04321afb45c7b919d6;hpb=55b6a89c7c42d2bde3da3c849c8ba0c124b40867;p=friendica.git diff --git a/src/Protocol/ActivityPub/Queue.php b/src/Protocol/ActivityPub/Queue.php index 011101eed7..9180de08f2 100644 --- a/src/Protocol/ActivityPub/Queue.php +++ b/src/Protocol/ActivityPub/Queue.php @@ -1,6 +1,6 @@ $activity['id'], @@ -52,6 +56,7 @@ class Queue 'activity' => json_encode($activity, JSON_UNESCAPED_SLASHES | JSON_UNESCAPED_UNICODE | JSON_PRETTY_PRINT), 'received' => DateTimeFormat::utcNow(), 'push' => $push, + 'trust' => $trust_source, ]; if (!empty($activity['reply-to-id'])) { @@ -82,6 +87,18 @@ class Queue return $activity; } + /** + * Checks if an entry for a given url and type already exists + * + * @param string $url + * @param string $type + * @return boolean + */ + public static function exists(string $url, string $type): bool + { + return DBA::exists('inbox-entry', ['type' => $type, 'object-id' => $url]); + } + /** * Remove activity from the queue * @@ -124,57 +141,80 @@ class Queue return; } + Logger::debug('Delete inbox-entry', ['id' => $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); - DBA::delete('inbox-entry', ['id' => $entry['id']]); } /** * Set the worker id for the queue entry * - * @param array $activity - * @param int $wid + * @param int $entry_id + * @param int $wid * @return void */ - public static function setWorkerId(array $activity, int $wid) + public static function setWorkerId(int $entry_id, int $wid) { - if (empty($activity['entry-id']) || empty($wid)) { + if (empty($entry_id) || empty($wid)) { return; } - DBA::update('inbox-entry', ['wid' => $wid], ['id' => $activity['entry-id']]); + DBA::update('inbox-entry', ['wid' => $wid], ['id' => $entry_id]); } /** * Check if there is an assigned worker task * - * @param array $activity + * @param int $wid + * * @return bool */ - public static function hasWorker(array $activity = []): bool + public static function hasWorker(int $wid): bool { - if (empty($activity['worker-id'])) { + if (empty($wid)) { return false; } - return DBA::exists('workerqueue', ['id' => $activity['worker-id'], 'done' => false]); + return DBA::exists('workerqueue', ['id' => $wid, 'done' => false]); } /** * Process the activity with the given id * * @param integer $id - * @return void + * @param bool $fetch_parents + * + * @return bool */ - public static function process(int $id) + public static function process(int $id, bool $fetch_parents = true): bool { $entry = DBA::selectFirst('inbox-entry', [], ['id' => $id]); if (empty($entry)) { - return; + return false; } - 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']]); + 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; + if ($entry['wid'] != $wid) { + $workerqueue = DBA::selectFirst('workerqueue', ['pid'], ['id' => $entry['wid'], 'done' => false]); + if (!empty($workerqueue['pid']) && posix_kill($workerqueue['pid'], 0)) { + Logger::notice('Entry is already processed via another process.', ['current' => $wid, 'processor' => $entry['wid']]); + return false; + } + } + } + + 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']; @@ -184,7 +224,11 @@ class Queue $activity['worker-id'] = $entry['wid']; $activity['recursion-depth'] = 0; - $receivers = DBA::select('inbox-entry-receiver', ['uid'], ['queue-id' => $entry['id']]); + 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'])) { $activity['receiver'][] = $receiver['uid']; @@ -192,9 +236,11 @@ class Queue } DBA::close($receivers); - if (!Receiver::routeActivities($activity, $type, $push)) { + if (!Receiver::routeActivities($activity, $type, $push, $fetch_parents, $activity['receiver'][0] ?? 0)) { self::remove($activity); } + + return true; } /** @@ -204,15 +250,48 @@ class Queue */ public static function processAll() { - $entries = DBA::select('inbox-entry', ['id', 'type', 'object-type', 'object-id', 'in-reply-to-id'], ["`wid` IS NULL"], ['order' => ['id' => true]]); + $entries = DBA::select('inbox-entry', ['id', 'type', 'object-type', 'object-id', 'in-reply-to-id'], ["`trust` AND `wid` IS NULL"], ['order' => ['id' => true]]); while ($entry = DBA::fetch($entries)) { + // Don't process entries of items that are answer to non existing posts + if (!empty($entry['in-reply-to-id']) && !Post::exists(['uri' => $entry['in-reply-to-id']])) { + continue; + } // We don't need to process entries that depend on already existing entries. if (!empty($entry['in-reply-to-id']) && DBA::exists('inbox-entry', ["`id` != ? AND `object-id` = ?", $entry['id'], $entry['in-reply-to-id']])) { continue; } Logger::debug('Process leftover entry', $entry); - self::process($entry['id']); + self::process($entry['id'], false); + } + 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; } /** @@ -224,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')) { @@ -238,13 +321,80 @@ class Queue * Process all activities that are children of a given post url * * @param string $uri - * @return void + * @return int */ - public static function processReplyByUri(string $uri) + public static function processReplyByUri(string $uri): int { + $count = 0; $entries = DBA::select('inbox-entry', ['id'], ["`in-reply-to-id` = ? AND `object-id` != ?", $uri, $uri]); while ($entry = DBA::fetch($entries)) { - self::process($entry['id']); + $count += 1; + self::process($entry['id'], false); } + DBA::close($entries); + return $count; + } + + /** + * Checks if there are children of the given uri + * + * @param string $uri + * + * @return bool + */ + public static function hasChildren(string $uri): bool + { + return DBA::exists('inbox-entry', ["`in-reply-to-id` = ? AND `object-id` != ?", $uri, $uri]); + } + + /** + * Prepare the queue entry. + * This is a test function that is used solely for development. + * + * @param integer $id + * @return array + */ + public static function reprepareActivityById(int $id): array + { + $entry = DBA::selectFirst('inbox-entry', [], ['id' => $id]); + if (empty($entry)) { + return []; + } + + $receiver = DBA::selectFirst('inbox-entry-receiver', ['uid'], ['queue-id' => $id]); + if (!empty($receiver)) { + $uid = $receiver['uid']; + } else { + $uid = 0; + } + + $trust_source = $entry['trust']; + + $data = json_decode($entry['activity'], true); + $activity = json_decode($data['raw'], true); + + $ldactivity = JsonLD::compact($activity); + return [ + 'data' => Receiver::prepareObjectData($ldactivity, $uid, $entry['push'], $trust_source), + 'trust' => $trust_source + ]; + } + + /** + * Set the trust for all untrusted entries. + * This is a test function that is used solely for development. + * + * @return void + */ + public static function reprepareAll() + { + $entries = DBA::select('inbox-entry', ['id'], ["NOT `trust` AND `wid` IS NULL"], ['order' => ['id' => true]]); + while ($entry = DBA::fetch($entries)) { + $data = self::reprepareActivityById($entry['id'], false); + if ($data['trust']) { + DBA::update('inbox-entry', ['trust' => true], ['id' => $entry['id']]); + } + } + DBA::close($entries); } }