better index of streams

Signed-off-by: Maxence Lange <maxence@artificial-owl.com>

11 files changed, 278 insertions, 296 deletions

diff --git a/lib/Command/CheckInstall.php b/lib/Command/CheckInstall.php
index 00eb3447..4a2cd6a8 100644
--- a/lib/Command/CheckInstall.php
+++ b/lib/Command/CheckInstall.php
@@ -34,14 +34,20 @@ namespace OCA\Social\Command;
 use daita\MySmallPhpTools\Traits\TArrayTools;
 use Exception;
 use OC\Core\Command\Base;
+use OCA\Social\Db\StreamDestRequest;
+use OCA\Social\Db\StreamRequest;
+use OCA\Social\Db\StreamTagsRequest;
+use OCA\Social\Service\CacheActorService;
 use OCA\Social\Service\CheckService;
 use OCA\Social\Service\ConfigService;
 use OCA\Social\Service\MiscService;
 use OCA\Social\Service\PushService;
 use OCP\IUserManager;
+use Symfony\Component\Console\Helper\ProgressBar;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Question\ConfirmationQuestion;
 
 
 class CheckInstall extends Base {
@@ -53,12 +59,27 @@ class CheckInstall extends Base {
 	/** @var IUserManager */
 	private $userManager;
 
+	/** @var StreamRequest */
+	private $streamRequest;
+
+	/** @var CacheActorService */
+	private $cacheActorService;
+
+	/** @var StreamDestRequest */
+	private $streamDestRequest;
+
+	/** @var StreamTagsRequest */
+	private $streamTagsRequest;
+
 	/** @var CheckService */
 	private $checkService;
 
 	/** @var */
 	private $configService;
 
+	/** @var PushService */
+	private $pushService;
+
 	/** @var MiscService */
 	private $miscService;
 
@@ -67,26 +88,34 @@ class CheckInstall extends Base {
 	 * CacheUpdate constructor.
 	 *
 	 * @param IUserManager $userManager
+	 * @param StreamRequest $streamRequest
+	 * @param StreamDestRequest $streamDestRequest
+	 * @param StreamTagsRequest $streamTagsRequest
+	 * @param CacheActorService $cacheActorService
 	 * @param CheckService $checkService
 	 * @param ConfigService $configService
-	 * @param MiscService $miscService
 	 * @param PushService $pushService
+	 * @param MiscService $miscService
 	 */
 	public function __construct(
-		IUserManager $userManager, CheckService $checkService, ConfigService $configService,
-		MiscService $miscService, PushService $pushService
+		IUserManager $userManager, StreamRequest $streamRequest, StreamDestRequest $streamDestRequest,
+		StreamTagsRequest $streamTagsRequest, CacheActorService $cacheActorService,
+		CheckService $checkService, ConfigService $configService, PushService $pushService,
+		MiscService $miscService
 	) {
 		parent::__construct();
 
 		$this->userManager = $userManager;
+		$this->streamRequest = $streamRequest;
+		$this->streamDestRequest = $streamDestRequest;
+		$this->streamTagsRequest = $streamTagsRequest;
+		$this->cacheActorService = $cacheActorService;
 		$this->checkService = $checkService;
 		$this->configService = $configService;
 		$this->miscService = $miscService;
 		$this->pushService = $pushService;
 	}
 
-	/** @var PushService */
-	private $pushService;
 
 	/**
 	 *
@@ -94,6 +123,7 @@ class CheckInstall extends Base {
 	protected function configure() {
 		parent::configure();
 		$this->setName('social:check:install')
+			 ->addOption('index', '', InputOption::VALUE_NONE, 'regenerate your index')
 			 ->addOption(
 				 'push', '', InputOption::VALUE_REQUIRED,
 				 'a local account used to test integration to Nextcloud Push',
@@ -110,12 +140,16 @@ class CheckInstall extends Base {
 	 * @throws Exception
 	 */
 	protected function execute(InputInterface $input, OutputInterface $output) {
-		$result = $this->checkService->checkInstallationStatus();
+		if ($this->askRegenerateIndex($input, $output)) {
+			return;
+		}
 
 		if ($this->checkPushApp($input, $output)) {
 			return;
 		}
 
+		$result = $this->checkService->checkInstallationStatus();
+
 		$output->writeln('- ' . $this->getInt('invalidFollowers', $result, 0) . ' invalid followers removed');
 		$output->writeln('- ' . $this->getInt('invalidNotes', $result, 0) . ' invalid notes removed');
 
@@ -150,5 +184,62 @@ class CheckInstall extends Base {
 		return true;
 	}
 
+
+	/**
+	 * @param InputInterface $input
+	 * @param OutputInterface $output
+	 *
+	 * @return bool
+	 */
+	private function askRegenerateIndex(InputInterface $input, OutputInterface $output): bool {
+		if (!$input->getOption('index')) {
+			return false;
+		}
+
+		$helper = $this->getHelper('question');
+		$output->writeln('<error>This command will regenerate the index of the Social App.</error>');
+		$output->writeln(
+			'<error>This operation can takes a while, and the Social App might not be stable during the process.</error>'
+		);
+		$output->writeln('');
+		$question = new ConfirmationQuestion(
+			'<info>Do you confirm this operation?</info> (y/N) ', false, '/^(y|Y)/i'
+		);
+
+		if (!$helper->ask($input, $output, $question)) {
+			return true;
+		}
+
+		$this->streamDestRequest->emptyStreamDest();
+		$this->streamTagsRequest->emptyStreamTags();
+		$this->regenerateIndex($input, $output);
+
+		return true;
+	}
+
+
+	/**
+	 * @param InputInterface $input
+	 * @param OutputInterface $output
+	 */
+	private function regenerateIndex(InputInterface $input, OutputInterface $output) {
+		$streams = $this->streamRequest->getAll();
+		$progressBar = new ProgressBar($output, count($streams));
+		$progressBar->start();
+
+		foreach ($streams as $stream) {
+			try {
+				$this->streamDestRequest->generateStreamDest($stream);
+				$this->streamTagsRequest->generateStreamTags($stream);
+			} catch (Exception $e) {
+				echo '-- ' . get_class($e) . ' - ' . $e->getMessage() . ' - ' . json_encode($stream) . "\n";
+			}
+			$progressBar->advance();
+		}
+
+		$progressBar->finish();
+		$output->writeln('');
+	}
+
 }
 
diff --git a/lib/Controller/LocalController.php b/lib/Controller/LocalController.php
index 89976cd1..b9f9b1bc 100644
--- a/lib/Controller/LocalController.php
+++ b/lib/Controller/LocalController.php
@@ -379,7 +379,7 @@ class LocalController extends Controller {
 	public function streamHome($since = 0, int $limit = 5): DataResponse {
 		try {
 			$this->initViewer(true);
-			$posts = $this->streamService->getStreamHome($this->viewer, $since, $limit);
+			$posts = $this->streamService->getStreamHome($since, $limit);
 
 			return $this->success($posts);
 		} catch (Exception $e) {
@@ -445,7 +445,7 @@ class LocalController extends Controller {
 	public function streamDirect(int $since = 0, int $limit = 5): DataResponse {
 		try {
 			$this->initViewer(true);
-			$posts = $this->streamService->getStreamDirect($this->viewer, $since, $limit);
+			$posts = $this->streamService->getStreamDirect($since, $limit);
 
 			return $this->success($posts);
 		} catch (Exception $e) {
diff --git a/lib/Db/SocialFiltersQueryBuilder.php b/lib/Db/SocialFiltersQueryBuilder.php
index e0d2e7c2..d0aca550 100644
--- a/lib/Db/SocialFiltersQueryBuilder.php
+++ b/lib/Db/SocialFiltersQueryBuilder.php
@@ -43,7 +43,7 @@ class SocialFiltersQueryBuilder extends SocialLimitsQueryBuilder {
 
 
 	/**
-	 * @param IQueryBuilder $qb
+	 *
 	 */
 	public function filterDuplicate() {
 		if (!$this->hasViewer()) {
diff --git a/lib/Db/SocialLimitsQueryBuilder.php b/lib/Db/SocialLimitsQueryBuilder.php
index e3cb83ae..1af01eed 100644
--- a/lib/Db/SocialLimitsQueryBuilder.php
+++ b/lib/Db/SocialLimitsQueryBuilder.php
@@ -387,5 +387,6 @@ class SocialLimitsQueryBuilder extends SocialCrossQueryBuilder {
 
 		$this->andWhere($orX);
 	}
+
 }
 
diff --git a/lib/Db/StreamDestRequest.php b/lib/Db/StreamDestRequest.php
index 78194f0d..6e0c7b25 100644
--- a/lib/Db/StreamDestRequest.php
+++ b/lib/Db/StreamDestRequest.php
@@ -33,7 +33,14 @@ namespace OCA\Social\Db;
 
 use daita\MySmallPhpTools\Traits\TStringTools;
 use Doctrine\DBAL\Exception\UniqueConstraintViolationException;
+use Exception;
+use OCA\Social\Model\ActivityPub\Internal\SocialAppNotification;
 use OCA\Social\Model\ActivityPub\Stream;
+use OCA\Social\Service\CacheActorService;
+use OCA\Social\Service\ConfigService;
+use OCA\Social\Service\MiscService;
+use OCP\IDBConnection;
+use OCP\ILogger;
 
 
 /**
@@ -47,17 +54,47 @@ class StreamDestRequest extends StreamDestRequestBuilder {
 	use TStringTools;
 
 
+	/** @var CacheActorService */
+	private $cacheActorService;
+
+
+	/**
+	 * StreamDestRequest constructor.
+	 *
+	 * @param IDBConnection $connection
+	 * @param ILogger $logger
+	 * @param CacheActorService $cacheActorService
+	 * @param ConfigService $configService
+	 * @param MiscService $miscService
+	 */
+	public function __construct(
+		IDBConnection $connection, ILogger $logger, CacheActorService $cacheActorService,
+		ConfigService $configService, MiscService $miscService
+	) {
+		parent::__construct($connection, $logger, $configService, $miscService);
+
+		$this->cacheActorService = $cacheActorService;
+	}
+
+
 	/**
-	 * @return int
+	 * @param string $streamId
+	 * @param string $actorId
+	 * @param string $type
+	 * @param string $subType
 	 */
-	public function countStreamDest(): int {
-		$qb = $this->countStreamDestSelectSql();
+	public function create(string $streamId, string $actorId, string $type, string $subType = '') {
+		$qb = $this->getStreamDestInsertSql();
 
-		$cursor = $qb->execute();
-		$data = $cursor->fetch();
-		$cursor->closeCursor();
+		$qb->setValue('stream_id', $qb->createNamedParameter($qb->prim($streamId)));
+		$qb->setValue('actor_id', $qb->createNamedParameter($qb->prim($actorId)));
+		$qb->setValue('type', $qb->createNamedParameter($type));
+		$qb->setValue('subtype', $qb->createNamedParameter($subType));
 
-		return $this->getInt('count', $data, 0);
+		try {
+			$qb->execute();
+		} catch (UniqueConstraintViolationException $e) {
+		}
 	}
 
 
@@ -65,6 +102,24 @@ class StreamDestRequest extends StreamDestRequestBuilder {
 	 * @param Stream $stream
 	 */
 	public function generateStreamDest(Stream $stream) {
+		if ($this->generateStreamNotification($stream)) {
+			return;
+		}
+
+		if ($this->generateStreamDirect($stream)) {
+			return;
+		}
+
+		$this->generateStreamHome($stream);
+	}
+
+
+	/**
+	 * @param Stream $stream
+	 *
+	 * @return bool
+	 */
+	private function generateStreamHome(Stream $stream): bool {
 		$recipients =
 			[
 				'to' => array_merge($stream->getToAll(), [$stream->getAttributedTo()]),
@@ -77,32 +132,76 @@ class StreamDestRequest extends StreamDestRequestBuilder {
 					continue;
 				}
 
-				$qb = $this->getStreamDestInsertSql();
-				$streamId = $qb->prim($stream->getId());
-				$qb->setValue('stream_id', $qb->createNamedParameter($streamId));
-				$qb->setValue('actor_id', $qb->createNamedParameter($qb->prim($actorId)));
-				$qb->setValue('type', $qb->createNamedParameter('recipient'));
-				$qb->setValue('subtype', $qb->createNamedParameter($subtype));
-				try {
-					$qb->execute();
-				} catch (UniqueConstraintViolationException $e) {
-					\OC::$server->getLogger()
-								->log(1, 'Social - Duplicate recipient on Stream ' . json_encode($stream));
-				}
+				$this->create($stream->getId(), $actorId, 'recipient', $subtype);
 			}
 		}
+
+		return true;
 	}
 
 
 	/**
+	 * @param Stream $stream
 	 *
+	 * @return bool
 	 */
-	public function generateRandomDest() {
-		$qb = $this->getStreamDestInsertSql();
+	private function generateStreamDirect(Stream $stream): bool {
+		try {
+			$author = $this->cacheActorService->getFromId($stream->getAttributedTo());
+		} catch (Exception $e) {
+			return false;
+		}
 
-		$qb->setValue('actor_id', $qb->createNamedParameter($this->uuid()));
-		$qb->setValue('stream_id', $qb->createNamedParameter($this->uuid()));
-		$qb->setValue('type', $qb->createNamedParameter('unk'));
+		$all = array_merge(
+			$stream->getToAll(), [$stream->getAttributedTo()], $stream->getCcArray(), $stream->getBccArray()
+		);
+
+		foreach ($all as $item) {
+			if ($item === Stream::CONTEXT_PUBLIC || $item === $author->getFollowers()) {
+				return false;
+			}
+		}
+
+		foreach ($all as $actorId) {
+			if ($actorId === '') {
+				continue;
+			}
+
+			$this->create($stream->getId(), $actorId, 'dm');
+		}
+
+		return true;
+	}
+
+
+	/**
+	 * @param Stream $stream
+	 *
+	 * @return bool
+	 */
+	private function generateStreamNotification(Stream $stream): bool {
+		if ($stream->getType() !== SocialAppNotification::TYPE) {
+			return false;
+		}
+
+		foreach ($stream->getToAll() as $actorId) {
+			if ($actorId === '') {
+				continue;
+			}
+
+			$this->create($stream->getId(), $actorId, 'notif');
+		}
+
+		return true;
+	}
+
+
+	/**
+	 *
+	 */
+	public function emptyStreamDest() {
+		$qb = $this->dbConnection->getQueryBuilder();
+		$qb->delete(self::TABLE_STREAM_DEST);
 
 		$qb->execute();
 	}
diff --git a/lib/Db/StreamRequest.php b/lib/Db/StreamRequest.php
index 225d060e..9a2d9f0f 100644
--- a/lib/Db/StreamRequest.php
+++ b/lib/Db/StreamRequest.php
@@ -44,6 +44,7 @@ use OCA\Social\Model\ActivityPub\Internal\SocialAppNotification;
 use OCA\Social\Model\ActivityPub\Object\Document;
 use OCA\Social\Model\ActivityPub\Object\Note;
 use OCA\Social\Model\ActivityPub\Stream;
+use OCA\Social\Service\CacheActorService;
 use OCA\Social\Service\ConfigService;
 use OCA\Social\Service\MiscService;
 use OCP\DB\QueryBuilder\IQueryBuilder;
@@ -59,6 +60,9 @@ use OCP\ILogger;
 class StreamRequest extends StreamRequestBuilder {
 
 
+	/** @var CacheActorService */
+	private $cacheActorService;
+
 	/** @var StreamDestRequest */
 	private $streamDestRequest;
 
@@ -71,16 +75,20 @@ class StreamRequest extends StreamRequestBuilder {
 	 *
 	 * @param IDBConnection $connection
 	 * @param ILogger $logger
+	 * @param CacheActorService $cacheActorService
 	 * @param StreamDestRequest $streamDestRequest
+	 * @param StreamTagsRequest $streamTagsRequest
 	 * @param ConfigService $configService
 	 * @param MiscService $miscService
 	 */
 	public function __construct(
-		IDBConnection $connection, ILogger $logger, StreamDestRequest $streamDestRequest,
-		StreamTagsRequest $streamTagsRequest, ConfigService $configService, MiscService $miscService
+		IDBConnection $connection, ILogger $logger, CacheActorService $cacheActorService,
+		StreamDestRequest $streamDestRequest, StreamTagsRequest $streamTagsRequest,
+		ConfigService $configService, MiscService $miscService
 	) {
 		parent::__construct($connection, $logger, $configService, $miscService);
 
+		$this->cacheActorService = $cacheActorService;
 		$this->streamDestRequest = $streamDestRequest;
 		$this->streamTagsRequest = $streamTagsRequest;
 	}
@@ -358,22 +366,20 @@ class StreamRequest extends StreamRequestBuilder {
 	 *  * Own posts,
 	 *  * Followed accounts
 	 *
-	 * @param Person $actor
 	 * @param int $since
 	 * @param int $limit
 	 *
 	 * @return Stream[]
 	 * @throws DateTimeException
 	 */
-	public function getTimelineHome(Person $actor, int $since = 0, int $limit = 5): array {
+	public function getTimelineHome(int $since = 0, int $limit = 5): array {
 		$qb = $this->getStreamSelectSql();
-		$expr = $qb->expr();
 
-		$qb->linkToCacheActors('ca', 'f.object_id_prim');
+		$qb->filterType(SocialAppNotification::TYPE);
 		$qb->limitPaginate($since, $limit);
 
-		$qb->andWhere($qb->exprLimitToDBField('type', SocialAppNotification::TYPE, false));
 		$qb->limitToViewer('sd', 'f', false);
+		$this->timelineHomeLinkCacheActor($qb, 'ca', 'f');
 
 		$qb->leftJoinStreamAction('sa');
 		$qb->filterDuplicate();
@@ -403,7 +409,7 @@ class StreamRequest extends StreamRequestBuilder {
 		$qb->limitPaginate($since, $limit);
 
 		$qb->selectDestFollowing('sd', '');
-		$qb->limitToDest($actor->getId(), 'recipient', '', 'sd');
+		$qb->limitToDest($actor->getId(), 'notif', '', 'sd');
 		$qb->limitToType(SocialAppNotification::TYPE);
 
 		$qb->linkToCacheActors('ca', 's.attributed_to_prim');
@@ -447,14 +453,13 @@ class StreamRequest extends StreamRequestBuilder {
 	 *  * Private message.
 	 *  - group messages. (not yet)
 	 *
-	 * @param Person $actor
 	 * @param int $since
 	 * @param int $limit
 	 *
 	 * @return Stream[]
 	 * @throws DateTimeException
 	 */
-	public function getTimelineDirect(Person $actor, int $since = 0, int $limit = 5): array {
+	public function getTimelineDirect(int $since = 0, int $limit = 5): array {
 		$qb = $this->getStreamSelectSql();
 
 		$qb->filterType(SocialAppNotification::TYPE);
@@ -462,13 +467,9 @@ class StreamRequest extends StreamRequestBuilder {
 
 		$qb->linkToCacheActors('ca', 's.attributed_to_prim');
 
+		$viewer = $qb->getViewer();
 		$qb->selectDestFollowing('sd', '');
-		$qb->innerJoinSteamDest('recipient', 'id_prim', 'sd', 's');
-		$qb->limitToDest($actor->getId(), 'recipient', '', 'sd');
-
-		$qb->filterDest(ACore::CONTEXT_PUBLIC);
-		$qb->filterDest($actor->getFollowers());
-		$qb->andWhere($qb->exprLimitToDBFieldInt('hidden_on_timeline', 1, 's'));
+		$qb->limitToDest($viewer->getId(), 'dm', '', 'sd');
 
 		return $this->getStreamsFromRequest($qb);
 	}
diff --git a/lib/Db/StreamRequestBuilder.php b/lib/Db/StreamRequestBuilder.php
index 4ee8db35..f13d4403 100644
--- a/lib/Db/StreamRequestBuilder.php
+++ b/lib/Db/StreamRequestBuilder.php
@@ -27,25 +27,21 @@ declare(strict_types=1);
  *
  */
 
+
 namespace OCA\Social\Db;
 
 
 use daita\MySmallPhpTools\Exceptions\CacheItemNotFoundException;
 use daita\MySmallPhpTools\Exceptions\RowNotFoundException;
 use daita\MySmallPhpTools\Traits\TArrayTools;
-use Doctrine\DBAL\Query\QueryBuilder;
 use OCA\Social\AP;
 use OCA\Social\Exceptions\InvalidResourceException;
 use OCA\Social\Exceptions\ItemUnknownException;
 use OCA\Social\Exceptions\SocialAppConfigException;
 use OCA\Social\Exceptions\StreamNotFoundException;
-use OCA\Social\Model\ActivityPub\ACore;
-use OCA\Social\Model\ActivityPub\Actor\Person;
 use OCA\Social\Model\ActivityPub\Object\Announce;
 use OCA\Social\Model\ActivityPub\Stream;
 use OCA\Social\Model\InstancePath;
-use OCP\DB\QueryBuilder\ICompositeExpression;
-use OCP\DB\QueryBuilder\IQueryBuilder;
 
 
 /**
@@ -139,232 +135,29 @@ class StreamRequestBuilder extends CoreRequestBuilder {
 
 
 	/**
-	 * @param IQueryBuilder $qb
-	 */
-	protected function limitToViewer(IQueryBuilder $qb) {
-		$actor = $this->viewer;
-
-		// TODO - rewrite this request to use stream_dest !
-		if ($this->viewer === null) {
-			$qb->andWhere($this->exprLimitToRecipient($qb, ACore::CONTEXT_PUBLIC, false));
-
-			return;
-		}
-
-		$on = $this->exprJoinFollowing($qb, $actor);
-		$on->add($this->exprLimitToRecipient($qb, ACore::CONTEXT_PUBLIC, false));
-		$on->add($this->exprLimitToRecipient($qb, $actor->getId(), true));
-		$qb->join($this->defaultSelectAlias, CoreRequestBuilder::TABLE_FOLLOWS, 'f', $on);
-	}
-
-
-	/**
-	 * @param IQueryBuilder $qb
-	 * @param Person $actor
-	 */
-	protected function leftJoinFollowing(IQueryBuilder $qb, Person $actor) {
-		if ($qb->getType() !== QueryBuilder::SELECT) {
-			return;
-		}
-
-		$on = $this->exprJoinFollowing($qb, $actor);
-		$qb->selectAlias('f.object_id', 'following_actor_id');
-		$qb->leftJoin($this->defaultSelectAlias, CoreRequestBuilder::TABLE_FOLLOWS, 'f', $on);
-	}
-
-
-	/**
-	 * @param IQueryBuilder $qb
-	 * @param Person $actor
-	 *
-	 * @return ICompositeExpression
-	 * @deprecated - use the new table social_stream_dest
-	 */
-	protected function exprJoinFollowing(IQueryBuilder $qb, Person $actor) {
-		$expr = $qb->expr();
-		$func = $qb->func();
-		$pf = $this->defaultSelectAlias . '.';
-
-		$on = $expr->orX();
-
-		// list of possible recipient as a follower (to, to_array, cc, ...)
-		$recipientFields = $expr->orX();
-		$recipientFields->add($expr->eq($func->lower($pf . 'to'), $func->lower('f.follow_id')));
-		$recipientFields->add($this->exprFieldWithinJsonFormat($qb, 'to_array', 'f.follow_id'));
-		$recipientFields->add($this->exprFieldWithinJsonFormat($qb, 'cc', 'f.follow_id'));
-		$recipientFields->add($this->exprFieldWithinJsonFormat($qb, 'bcc', 'f.follow_id'));
-
-		// all possible follow, but linked by followers (actor_id) and accepted follow
-		$crossFollows = $expr->andX();
-		$crossFollows->add($recipientFields);
-		$crossFollows->add(
-			$this->exprLimitToDBField($qb, 'actor_id', $actor->getId(), true, false, 'f')
-		);
-		$crossFollows->add($this->exprLimitToDBFieldInt($qb, 'accepted', 1, 'f'));
-		$on->add($crossFollows);
-
-		return $on;
-	}
-
-
-	/**
-	 * @param IQueryBuilder $qb
-	 * @param string $field
-	 * @param string $fieldRight
+	 * @param SocialQueryBuilder $qb
 	 * @param string $alias
-	 *
-	 * @return string
+	 * @param string $aliasFollow
 	 */
-	protected function exprFieldWithinJsonFormat(
-		IQueryBuilder $qb, string $field, string $fieldRight, string $alias = ''
+	protected function timelineHomeLinkCacheActor(
+		SocialQueryBuilder $qb, string $alias = 'ca', string $aliasFollow = 'f'
 	) {
-		$func = $qb->func();
-		$expr = $qb->expr();
-
-		if ($alias === '') {
-			$alias = $this->defaultSelectAlias;
-		}
-
-		$concat = $func->concat(
-			$qb->createNamedParameter('%"'),
-			$func->concat($fieldRight, $qb->createNamedParameter('"%'))
-		);
-
-		return $expr->iLike($alias . '.' . $field, $concat);
-	}
+		$qb->linkToCacheActors($alias);
 
-
-	/**
-	 * @param IQueryBuilder $qb
-	 * @param string $field
-	 * @param string $value
-	 *
-	 * @return string
-	 */
-	protected function exprValueWithinJsonFormat(IQueryBuilder $qb, string $field, string $value): string {
-		$dbConn = $this->getConnection();
 		$expr = $qb->expr();
+		$orX = $expr->orX();
 
-		return $expr->iLike(
-			$field,
-			$qb->createNamedParameter('%"' . $dbConn->escapeLikeParameter($value) . '"%')
-		);
-	}
-
-
-	/**
-	 * @param IQueryBuilder $qb
-	 * @param string $field
-	 * @param string $value
-	 *
-	 * @return string
-	 */
-	protected function exprValueNotWithinJsonFormat(IQueryBuilder $qb, string $field, string $value): string {
-		$dbConn = $this->getConnection();
-		$expr = $qb->expr();
-		$func = $qb->func();
-
-		return $expr->notLike(
-			$func->lower($field),
-			$qb->createNamedParameter(
-				'%"' . $func->lower($dbConn->escapeLikeParameter($value)) . '"%'
-			)
-		);
-	}
-
-
-	/**
-	 * @param IQueryBuilder $qb
-	 * @param string $recipient
-	 * @param bool $asAuthor
-	 * @param array $type
-	 *
-	 * @deprecated
-	 */
-	protected function limitToRecipient(
-		IQueryBuilder &$qb, string $recipient, bool $asAuthor = false, array $type = []
-	) {
-		$qb->andWhere($this->exprLimitToRecipient($qb, $recipient, $asAuthor, $type));
-	}
-
-
-	/**
-	 * @param IQueryBuilder $qb
-	 * @param string $recipient
-	 * @param bool $asAuthor
-	 * @param array $type
-	 *
-	 * @return ICompositeExpression
-	 * @deprecated
-	 */
-	protected function exprLimitToRecipient(
-		IQueryBuilder &$qb, string $recipient, bool $asAuthor = false, array $type = []
-	): ICompositeExpression {
-
-		$expr = $qb->expr();
-		$limit = $expr->orX();
-
-		if ($asAuthor === true) {
-			$func = $qb->func();
-			$limit->add(
-				$expr->eq(
-					$func->lower('attributed_to'),
-					$func->lower($qb->createNamedParameter($recipient))
-				)
-			);
-		}
-
-		if ($type === []) {
-			$type = ['to', 'cc', 'bcc'];
-		}
-
-		$this->addLimitToRecipient($qb, $limit, $type, $recipient);
-
-		return $limit;
-	}
-
-
-	/**
-	 * @param IQueryBuilder $qb
-	 * @param ICompositeExpression $limit
-	 * @param array $type
-	 * @param string $to
-	 */
-	private function addLimitToRecipient(
-		IQueryBuilder $qb, ICompositeExpression &$limit, array $type, string $to
-	) {
-
-		$expr = $qb->expr();
-		if (in_array('to', $type)) {
-			$limit->add($expr->eq('to', $qb->createNamedParameter($to)));
-			$limit->add($this->exprValueWithinJsonFormat($qb, 'to_array', $to));
-		}
-
-		if (in_array('cc', $type)) {
-			$limit->add($this->exprValueWithinJsonFormat($qb, 'cc', $to));
-		}
-
-		if (in_array('bcc', $type)) {
-			$limit->add($this->exprValueWithinJsonFormat($qb, 'bcc', $to));
-		}
-	}
-
-
-	/**
-	 * @param IQueryBuilder $qb
-	 * @param string $recipient
-	 */
-	protected function filterRecipient(IQueryBuilder &$qb, string $recipient) {
-
-		$expr = $qb->expr();
-		$filter = $expr->andX();
+		$follow = $expr->andX();
+		$follow->add($expr->eq($aliasFollow . '.type', $qb->createNamedParameter('Follow')));
+		$follow->add($expr->eq($alias . '.id_prim', $aliasFollow . '.object_id_prim'));
+		$orX->add($follow);
 
-		$filter->add($expr->neq('to', $qb->createNamedParameter($recipient)));
-		$filter->add($this->exprValueNotWithinJsonFormat($qb, 'to_array', $recipient));
-		$filter->add($this->exprValueNotWithinJsonFormat($qb, 'cc', $recipient));
-		$filter->add($this->exprValueNotWithinJsonFormat($qb, 'bcc', $recipient));
+		$loopback = $expr->andX();
+		$loopback->add($expr->eq($aliasFollow . '.type', $qb->createNamedParameter('Loopback')));
+		$loopback->add($expr->eq($alias . '.id_prim', $qb->getDefaultSelectAlias() . '.attributed_to_prim'));
+		$orX->add($loopback);
 
-		$qb->andWhere($filter);
+		$qb->andWhere($orX);
 	}
 
 
@@ -401,7 +194,6 @@ class StreamRequestBuilder extends CoreRequestBuilder {
 
 	/**
 	 * @param array $data
-	 * @param string $as
 	 *
 	 * @return Stream
 	 * @throws ItemUnknownException
diff --git a/lib/Db/StreamTagsRequest.php b/lib/Db/StreamTagsRequest.php
index 6a7d687f..c8504c03 100644
--- a/lib/Db/StreamTagsRequest.php
+++ b/lib/Db/StreamTagsRequest.php
@@ -73,5 +73,16 @@ class StreamTagsRequest extends StreamTagsRequestBuilder {
 		}
 	}
 
+
+	/**
+	 *
+	 */
+	public function emptyStreamTags() {
+		$qb = $this->dbConnection->getQueryBuilder();
+		$qb->delete(self::TABLE_STREAM_TAGS);
+
+		$qb->execute();
+	}
+
 }
 
diff --git a/lib/Interfaces/Object/AnnounceInterface.php b/lib/Interfaces/Object/AnnounceInterface.php
index 182232c2..ea2487be 100644
--- a/lib/Interfaces/Object/AnnounceInterface.php
+++ b/lib/Interfaces/Object/AnnounceInterface.php
@@ -204,11 +204,10 @@ class AnnounceInterface implements IActivityPubInterface {
 	 * @throws Exception
 	 */
 	public function save(ACore $item) {
-
 		/** @var Announce $item */
+
 		try {
-			$knownItem =
-				$this->streamRequest->getStreamByObjectId($item->getObjectId(), Announce::TYPE);
+			$knownItem = $this->streamRequest->getStreamByObjectId($item->getObjectId(), Announce::TYPE);
 
 			if ($item->hasActor()) {
 				$actor = $item->getActor();
@@ -216,6 +215,7 @@ class AnnounceInterface implements IActivityPubInterface {
 				$actor = $this->cacheActorService->getFromId($item->getActorId());
 			}
 
+			$knownItem->setAttributedTo($actor->getId());
 			if (!$knownItem->hasCc($actor->getFollowers())) {
 				$knownItem->addCc($actor->getFollowers(