Rework duplicate entry handling for tags

lib/Db/MessageMapper.php

@@ -65,8 +65,7 @@ class MessageMapper extends QBMapper {
 	public function __construct(IDBConnection $db,
 								ITimeFactory $timeFactory,
 								TagMapper $tagMapper,
-								PerformanceLogger $performanceLogger
-								) {
+								PerformanceLogger $performanceLogger) {
 		parent::__construct($db, 'mail_messages');
 		$this->timeFactory = $timeFactory;
 		$this->tagMapper = $tagMapper;
@@ -120,19 +119,19 @@ public function findHighestUid(Mailbox $mailbox): ?int {
 		return $max;
 	}
 
-	public function findByUserId(string $uid, int $id): Message {
+	public function findByUserId(string $userId, int $id): Message {
 		$query = $this->db->getQueryBuilder();
 
 		$query->select('m.*')
 			->from($this->getTableName(), 'm')
 			->join('m', 'mail_mailboxes', 'mb', $query->expr()->eq('m.mailbox_id', 'mb.id', IQueryBuilder::PARAM_INT))
 			->join('m', 'mail_accounts', 'a', $query->expr()->eq('mb.account_id', 'a.id', IQueryBuilder::PARAM_INT))
 			->where(
-				$query->expr()->eq('a.user_id', $query->createNamedParameter($uid)),
+				$query->expr()->eq('a.user_id', $query->createNamedParameter($userId)),
 				$query->expr()->eq('m.id', $query->createNamedParameter($id, IQueryBuilder::PARAM_INT), IQueryBuilder::PARAM_INT)
 			);
 
-		$results = $this->findRelatedData($this->findEntities($query));
+		$results = $this->findRelatedData($this->findEntities($query), $userId);
 		if (empty($results)) {
 			throw new DoesNotExistException("Message $id does not exist");
 		}
@@ -364,7 +363,7 @@ public function updateBulk(Account $account, Message ...$messages): array {
 				$query->expr()->eq('mailbox_id', $query->createParameter('mailbox_id'))
 			));
 		// get all tags before the loop and create a mapping [message_id => [tag,...]]
-		$tags = $this->tagMapper->getAllTagsForMessages($messages);
+		$tags = $this->tagMapper->getAllTagsForMessages($messages, $account->getUserId());
 		$perf->step("Selected Tags for all messages");
 		foreach ($messages as $message) {
 			if (empty($message->getUpdatedFields())) {
@@ -552,7 +551,7 @@ public function findThread(Account $account, int $messageId): array {
 				$qb->expr()->isNotNull('m1.thread_root_id')
 			)
 			->orderBy('sent_at', 'desc');
-		return $this->findRelatedData($this->findEntities($selectMessages));
+		return $this->findRelatedData($this->findEntities($selectMessages), $account->getUserId());
 	}
 
 	/**
@@ -808,16 +807,15 @@ public function findByUids(Mailbox $mailbox, array $uids): array {
 				$qb->expr()->in('uid', $qb->createNamedParameter($uids, IQueryBuilder::PARAM_INT_ARRAY))
 			)
 			->orderBy('sent_at', 'desc');
-
-		return $this->findRelatedData($this->findEntities($select));
+		return $this->findRecipients($this->findEntities($select));
 	}
 
 	/**
 	 * @param int[] $ids
 	 *
 	 * @return Message[]
 	 */
-	public function findByIds(array $ids): array {
+	public function findByIds(string $userId, array $ids): array {
 		if (empty($ids)) {
 			return [];
 		}
@@ -831,7 +829,7 @@ public function findByIds(array $ids): array {
 			)
 			->orderBy('sent_at', 'desc');
 
-		return $this->findRelatedData($this->findEntities($select));
+		return $this->findRelatedData($this->findEntities($select), $userId);
 	}
 
 	/**
@@ -888,9 +886,9 @@ private function findRecipients(array $messages): array {
 	 * @param Message[] $messages
 	 * @return Message[]
 	 */
-	public function findRelatedData(array $messages): array {
+	public function findRelatedData(array $messages, string $userId): array {
 		$messages = $this->findRecipients($messages);
-		$tags = $this->tagMapper->getAllTagsForMessages($messages);
+		$tags = $this->tagMapper->getAllTagsForMessages($messages, $userId);
 		/** @var Message $message */
 		$messages = array_map(function ($message) use ($tags) {
 			$message->setTags($tags[$message->getMessageId()] ?? []);
@@ -927,7 +925,10 @@ public function findNewIds(Mailbox $mailbox, array $ids): array {
 		return $this->findIds($select);
 	}
 
-	public function findChanged(Mailbox $mailbox, int $since): array {
+	/**
+	 * Currently unused
+	 */
+	public function findChanged(Account $account, Mailbox $mailbox, int $since): array {
 		$qb = $this->db->getQueryBuilder();
 
 		$select = $qb
@@ -937,8 +938,8 @@ public function findChanged(Mailbox $mailbox, int $since): array {
 				$qb->expr()->eq('mailbox_id', $qb->createNamedParameter($mailbox->getId(), IQueryBuilder::PARAM_INT)),
 				$qb->expr()->gt('updated_at', $qb->createNamedParameter($since, IQueryBuilder::PARAM_INT))
 			);
-
-		return $this->findRelatedData($this->findEntities($select));
+		// TODO: change this to findRelatedData
+		return $this->findRelatedData($this->findEntities($select), $account->getUserId());
 	}
 
 	/**
@@ -947,7 +948,7 @@ public function findChanged(Mailbox $mailbox, int $since): array {
 	 *
 	 * @return Message[]
 	 */
-	public function findLatestMessages(array $mailboxIds, int $limit): array {
+	public function findLatestMessages(string $userId, array $mailboxIds, int $limit): array {
 		$qb = $this->db->getQueryBuilder();
 
 		$select = $qb
@@ -961,7 +962,7 @@ public function findLatestMessages(array $mailboxIds, int $limit): array {
 			->orderBy('sent_at', 'desc')
 			->setMaxResults($limit);
 
-		return $this->findRelatedData($this->findEntities($select));
+		return $this->findRelatedData($this->findEntities($select), $userId);
 	}
 
 	public function deleteOrphans(): void {

lib/Db/TagMapper.php

@@ -25,11 +25,8 @@
 
 namespace OCA\Mail\Db;
 
-use Doctrine\DBAL\Exception\UniqueConstraintViolationException;
-use OCP\DB\Exception;
-use Throwable;
 use function array_map;
-
+use function array_chunk;
 use OCP\AppFramework\Db\DoesNotExistException;
 use OCP\AppFramework\Db\Entity;
 use OCP\AppFramework\Db\QBMapper;
@@ -83,7 +80,7 @@ public function getTagForUser(int $id, string $userId): Entity {
 	 * @return Tag[]
 	 * @throws DoesNotExistException
 	 */
-	public function getAllTagForUser(string $userId): array {
+	public function getAllTagsForUser(string $userId): array {
 		$qb = $this->db->getQueryBuilder();
 		$qb->select('*')
 			->from($this->getTableName())
@@ -93,17 +90,6 @@ public function getAllTagForUser(string $userId): array {
 		return $this->findEntities($qb);
 	}
 
-	/**
-	 * @throws DoesNotExistException
-	 */
-	public function getTag(int $id): Entity {
-		$qb = $this->db->getQueryBuilder();
-		$qb->select('*')
-			->from($this->getTableName())
-			->where($qb->expr()->eq('id', $qb->createNamedParameter($id, IQueryBuilder::PARAM_INT)));
-		return $this->findEntity($qb);
-	}
-
 	/**
 	 * Tag a message in the DB
 	 *
@@ -122,26 +108,7 @@ public function tagMessage(Tag $tag, string $messageId, string $userId): void {
 		$qb->insert('mail_message_tags')
 		   ->setValue('imap_message_id', $qb->createNamedParameter($messageId))
 		   ->setValue('tag_id', $qb->createNamedParameter($tag->getId(), IQueryBuilder::PARAM_INT));
-		try {
-			$qb->execute();
-		} catch (Throwable $e) {
-			/**
-			 * @psalm-suppress all
-			 */
-			if (class_exists(Exception::class) && ($e instanceof Exception) && $e->getReason() === Exception::REASON_UNIQUE_CONSTRAINT_VIOLATION) {
-				// OK -> ignore
-				return;
-			}
-			/**
-			 * @psalm-suppress all
-			 */
-			if (class_exists(UniqueConstraintViolationException::class) && ($e instanceof UniqueConstraintViolationException)) {
-				// OK -> ignore
-				return;
-			}
-
-			throw $e;
-		}
+		$qb->execute();
 	}
 
 	/**
@@ -159,21 +126,24 @@ public function untagMessage(Tag $tag, string $messageId): void {
 
 	/**
 	 * @param Message[] $messages
+	 * @param string $userId
 	 * @return Tag[][]
 	 */
-	public function getAllTagsForMessages(array $messages): array {
+	public function getAllTagsForMessages(array $messages, string $userId): array {
 		$ids = array_map(function (Message $message) {
 			return $message->getMessageId();
 		}, $messages);
 
 		$qb = $this->db->getQueryBuilder();
-		$tagsQuery = $qb->select('t.*', 'mt.imap_message_id')
+		$tagsQuery = $qb->selectDistinct(['t.*', 'mt.imap_message_id'])
 			->from($this->getTableName(), 't')
 			->join('t', 'mail_message_tags', 'mt', $qb->expr()->eq('t.id', 'mt.tag_id', IQueryBuilder::PARAM_INT))
 			->where(
-				$qb->expr()->in('mt.imap_message_id', $qb->createNamedParameter($ids, IQueryBuilder::PARAM_STR_ARRAY))
+				$qb->expr()->in('mt.imap_message_id', $qb->createNamedParameter($ids, IQueryBuilder::PARAM_STR_ARRAY)),
+				$qb->expr()->eq('t.user_id', $qb->createNamedParameter($userId, IQueryBuilder::PARAM_STR))
 			);
 		$queryResult = $tagsQuery->execute();
+
 		$tags = [];
 		while (($row = $queryResult->fetch()) !== false) {
 			$messageId = $row['imap_message_id'];
@@ -240,12 +210,60 @@ public function createDefaultTags(MailAccount $account): void {
 			}
 			$tags[] = $tag;
 		}
-		$dbTags = $this->getAllTagForUser($account->getUserId());
+		$dbTags = $this->getAllTagsForUser($account->getUserId());
 		$toInsert = array_udiff($tags, $dbTags, function (Tag $a, Tag $b) {
 			return strcmp($a->getImapLabel(), $b->getImapLabel());
 		});
 		foreach ($toInsert as $entity) {
 			$this->insert($entity);
 		}
 	}
+
+	public function deleteDuplicates(): void {
+		$qb = $this->db->getQueryBuilder();
+		$qb->select('mt2.id')
+		->from('mail_message_tags', 'mt2')
+		->join('mt2','mail_message_tags', 'mt1', $qb->expr()->andX(
+			$qb->expr()->gt('mt1.id', 'mt2.id'),
+			$qb->expr()->eq('mt1.imap_message_id', 'mt2.imap_message_id'),
+			$qb->expr()->eq('mt1.tag_id', 'mt2.tag_id')
+			)
+		);
+		$result = $qb->execute();
+		$rows = $result->fetchAll();
+		$result->closeCursor();
+		$ids = array_unique(array_map(function ($row) {
+			return $row['id'];
+		}, $rows));
+
+		$deleteQB = $this->db->getQueryBuilder();
+		foreach (array_chunk($ids, 1000) as $chunk) {
+			$deleteQB->delete('mail_message_tags')
+				->where($deleteQB->expr()->in('id', $deleteQB->createNamedParameter($chunk, IQueryBuilder::PARAM_INT_ARRAY), IQueryBuilder::PARAM_INT_ARRAY));
+			$deleteQB->execute();
+		}
+	}
+
+	public function deleteOrphans(): void {
+		$qb = $this->db->getQueryBuilder();
+
+		$qb->select('mt.id')
+		->from('mail_message_tags', 'mt')
+		->leftJoin('mt', $this->getTableName(), 't', $qb->expr()->eq('t.id', 'mt.tag_id'))
+		->where($qb->expr()->isNull('t.id'));
+		$result = $qb->execute();
+		$rows = $result->fetchAll();
+		$result->closeCursor();
+
+		$ids = array_map(function (array $row) {
+			return $row['id'];
+		}, $rows);
+
+		$deleteQB = $this->db->getQueryBuilder();
+		foreach (array_chunk($ids, 1000) as $chunk) {
+			$deleteQB->delete('mail_message_tags')
+				->where($deleteQB->expr()->in('id', $deleteQB->createNamedParameter($chunk, IQueryBuilder::PARAM_INT_ARRAY), IQueryBuilder::PARAM_INT_ARRAY));
+			$deleteQB->execute();
+		}
+	}
 }

lib/Migration/Version1100Date20210304143008.php

@@ -78,13 +78,6 @@ public function changeSchema(IOutput $output, Closure $schemaClosure, array $opt
 				'length' => 4,
 			]);
 			$tagsMessageTable->setPrimaryKey(['id']);
-			$tagsMessageTable->addUniqueIndex(
-				[
-					'imap_message_id',
-					'tag_id',
-				],
-				'mail_msg_tag_id_idx'
-			);
 		}
 		return $schema;
 	}

lib/Migration/Version1100Date20210326103929.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace OCA\Mail\Migration;
+
+use Closure;
+use OCP\DB\ISchemaWrapper;
+use OCP\Migration\IOutput;
+use OCP\Migration\SimpleMigrationStep;
+
+/**
+ * @link https://github.com/nextcloud/mail/issues/4833
+ */
+class Version1100Date20210326103929 extends SimpleMigrationStep {
+
+	/**
+	 * @param IOutput $output
+	 * @param Closure $schemaClosure The `\Closure` returns a `ISchemaWrapper`
+	 * @param array $options
+	 * @return null|ISchemaWrapper
+	 */
+	public function changeSchema(IOutput $output, Closure $schemaClosure, array $options): ?ISchemaWrapper {
+		$schema = $schemaClosure();
+
+		if ($schema->hasTable('mail_message_tags')) {
+			$table = $schema->getTable('mail_message_tags');
+			if ($table->hasIndex('mail_msg_tag_id_idx')) {
+				$table->dropIndex('mail_msg_tag_id_idx');
+			}
+			if (!$table->hasIndex('mail_msg_imap_id_idx')) {
+				$table->addIndex(['imap_message_id'], 'mail_msg_imap_id_idx', [], ['lengths' => [128]]);
+			}
+		}
+		return $schema;
+	}
+}

lib/Service/Classification/ImportanceClassifier.php

@@ -164,7 +164,7 @@ public function train(Account $account, LoggerInterface $logger): void {
 			return $mailbox->getId();
 		}, $incomingMailboxes);
 		$messages = array_filter(
-			$this->messageMapper->findLatestMessages($mailboxIds, self::MAX_TRAINING_SET_SIZE),
+			$this->messageMapper->findLatestMessages($account->getUserId(), $mailboxIds, self::MAX_TRAINING_SET_SIZE),
 			[$this, 'filterMessageHasSenderEmail']
 		);
 		$importantMessages = array_filter($messages, function (Message $message) {

lib/Service/CleanupService.php

@@ -29,6 +29,7 @@
 use OCA\Mail\Db\CollectedAddressMapper;
 use OCA\Mail\Db\MailboxMapper;
 use OCA\Mail\Db\MessageMapper;
+use OCA\Mail\Db\TagMapper;
 
 class CleanupService {
 
@@ -44,20 +45,27 @@ class CleanupService {
 	/** @var CollectedAddressMapper */
 	private $collectedAddressMapper;
 
+	/** @var TagMapper */
+	private $tagMapper;
+
 	public function __construct(AliasMapper $aliasMapper,
 								MailboxMapper $mailboxMapper,
 								MessageMapper $messageMapper,
-								CollectedAddressMapper $collectedAddressMapper) {
+								CollectedAddressMapper $collectedAddressMapper,
+								TagMapper $tagMapper) {
 		$this->aliasMapper = $aliasMapper;
 		$this->mailboxMapper = $mailboxMapper;
 		$this->messageMapper = $messageMapper;
 		$this->collectedAddressMapper = $collectedAddressMapper;
+		$this->tagMapper = $tagMapper;
 	}
 
 	public function cleanUp(): void {
 		$this->aliasMapper->deleteOrphans();
 		$this->mailboxMapper->deleteOrphans();
 		$this->messageMapper->deleteOrphans();
 		$this->collectedAddressMapper->deleteOrphans();
+		$this->tagMapper->deleteOrphans();
+		$this->tagMapper->deleteDuplicates();
 	}
 }