]> git.mxchange.org Git - friendica.git/blobdiff - src/Protocol/ActivityPub/Queue.php
Merge branch 'post-reason' of github.com:annando/friendica into post-reason
[friendica.git] / src / Protocol / ActivityPub / Queue.php
index d5a308b109d3fd34287daccb6edcae97cba17a76..95a4cfa0d4557ecd5a34e33ef8ae5d413ac751dc 100644 (file)
@@ -24,7 +24,9 @@ namespace Friendica\Protocol\ActivityPub;
 use Friendica\Core\Logger;
 use Friendica\Database\Database;
 use Friendica\Database\DBA;
+use Friendica\DI;
 use Friendica\Util\DateTimeFormat;
+use Friendica\Util\JsonLD;
 
 /**
  * This class handles the processing of incoming posts
@@ -41,7 +43,7 @@ class Queue
         * @param boolean $push
         * @return array
         */
-       public static function add(array $activity, string $type, int $uid, string $http_signer, bool $push): array
+       public static function add(array $activity, string $type, int $uid, string $http_signer, bool $push, bool $trust_source): array
        {
                $fields = [
                        'activity-id' => $activity['id'],
@@ -51,12 +53,19 @@ 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'])) {
                        $fields['in-reply-to-id'] = $activity['reply-to-id'];
                }
 
+               if (!empty($activity['context'])) {
+                       $fields['conversation'] = $activity['context'];
+               } elseif (!empty($activity['conversation'])) {
+                       $fields['conversation'] = $activity['conversation'];
+               }
+
                if (!empty($activity['object_object_type'])) {
                        $fields['object-object-type'] = $activity['object_object_type'];
                }
@@ -89,6 +98,37 @@ class Queue
                DBA::delete('inbox-entry', ['id' => $activity['entry-id']]);
        }
 
+       /**
+        * Delete all entries that depend on the given worker id
+        *
+        * @param integer $wid
+        * @return void
+        */
+       public static function deleteByWorkerId(int $wid)
+       {
+               $entries = DBA::select('inbox-entry', ['id'], ['wid' => $wid]);
+               while ($entry = DBA::fetch($entries)) {
+                       self::deleteById($entry['id']);
+               }
+               DBA::close($entries);
+       }
+
+       /**
+        * Delete recursively an entry and all their children
+        *
+        * @param integer $id
+        * @return void
+        */
+       public static function deleteById(int $id)
+       {
+               $entry = DBA::selectFirst('inbox-entry', ['id', 'object-id'], ['id' => $id]);
+               if (empty($entry)) {
+                       return;
+               }
+
+               DBA::delete('inbox-entry', ['id' => $entry['id']]);
+       }
+
        /**
         * Set the worker id for the queue entry
         *
@@ -122,13 +162,14 @@ class Queue
         * Process the activity with the given id
         *
         * @param integer $id
-        * @return void
+        *
+        * @return bool
         */
-       public static function process(int $id)
+       public static function process(int $id): 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']]);
@@ -137,12 +178,23 @@ class Queue
                $type     = $entry['type'];
                $push     = $entry['push'];
 
-               $activity['entry-id']  = $entry['id'];
-               $activity['worker-id'] = $entry['wid'];
+               $activity['entry-id']        = $entry['id'];
+               $activity['worker-id']       = $entry['wid'];
+               $activity['recursion-depth'] = 0;
+
+               $receivers = DBA::select('inbox-entry-receiver', ['uid'], ['queue-id' => $entry['id']]);
+               while ($receiver = DBA::fetch($receivers)) {
+                       if (!in_array($receiver['uid'], $activity['receiver'])) {
+                               $activity['receiver'][] = $receiver['uid'];
+                       }
+               }
+               DBA::close($receivers);
 
                if (!Receiver::routeActivities($activity, $type, $push)) {
                        self::remove($activity);
                }
+
+               return true;
        }
 
        /**
@@ -152,10 +204,16 @@ class Queue
         */
        public static function processAll()
        {
-               $entries = DBA::select('inbox-entry', ['id', 'type', 'object-type'], [], ['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)) {
+                       // 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']);
                }
+               DBA::close($entries);
        }
 
        /**
@@ -165,20 +223,97 @@ class Queue
         */
        public static function clear()
        {
-               DBA::delete('inbox-entry', ["`wid` IS NULL AND `received` < ?", DateTimeFormat::utc('now - 4 hours')]);
+               // 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')]);
+
+               // Optimizing this table only last seconds
+               if (DI::config()->get('system', 'optimize_tables')) {
+                       Logger::info('Optimize start');
+                       DBA::e("OPTIMIZE TABLE `inbox-entry`");
+                       Logger::info('Optimize end');
+               }
        }
 
        /**
         * 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
        {
-               $entries = DBA::select('inbox-entry', ['id'], ['in-reply-to-id' => $uri], ['order' => ['id' => true]]);
+               $count = 0;
+               $entries = DBA::select('inbox-entry', ['id'], ["`in-reply-to-id` = ? AND `object-id` != ?", $uri, $uri]);
                while ($entry = DBA::fetch($entries)) {
+                       $count += 1;
                        self::process($entry['id']);
                }
+               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);
+
        }
 }