refactor(FileListener): Process file events asynchronously

lib/BackgroundJobs/ProcessAccessUpdatesJob.php

@@ -0,0 +1,61 @@
+<?php
+
+/*
+ * Copyright (c) 2021-2022 The Recognize contributors.
+ * This file is licensed under the Affero General Public License version 3 or later. See the COPYING file.
+ */
+declare(strict_types=1);
+namespace OCA\Recognize\BackgroundJobs;
+
+use OCA\Recognize\Classifiers\Audio\MusicnnClassifier;
+use OCA\Recognize\Classifiers\Images\ClusteringFaceClassifier;
+use OCA\Recognize\Classifiers\Images\ImagenetClassifier;
+use OCA\Recognize\Classifiers\Images\LandmarksClassifier;
+use OCA\Recognize\Classifiers\Video\MovinetClassifier;
+use OCA\Recognize\Db\AccessUpdateMapper;
+use OCA\Recognize\Db\QueueFile;
+use OCA\Recognize\Service\AccessUpdateService;
+use OCA\Recognize\Service\Logger;
+use OCA\Recognize\Service\QueueService;
+use OCA\Recognize\Service\StorageService;
+use OCA\Recognize\Service\TagManager;
+use OCP\AppFramework\Utility\ITimeFactory;
+use OCP\BackgroundJob\IJobList;
+use OCP\BackgroundJob\QueuedJob;
+use OCP\DB\Exception;
+use Psr\Log\LoggerInterface;
+
+final class ProcessAccessUpdatesJob extends QueuedJob {
+
+	public function __construct(
+		ITimeFactory $timeFactory,
+		private AccessUpdateService  $accessUpdateService,
+		private IJobList $jobList,
+		private AccessUpdateMapper $accessUpdateMapper,
+		private LoggerInterface $logger,
+	) {
+		parent::__construct($timeFactory);
+	}
+
+	/**
+	 * @param array{storage_id:int} $argument
+	 * @return void
+	 */
+	protected function run($argument): void {
+		$storageId = $argument['storage_id'];
+
+		$this->accessUpdateService->processAccessUpdates($storageId);
+		try {
+			$count = $this->accessUpdateMapper->countByStorageId($storageId);
+		} catch (Exception $e) {
+			$this->logger->error('Failed to count access updates' . $e->getMessage(), ['exception' => $e]);
+			$count = 1;
+		}
+		if ($count > 0) {
+			// Schedule next iteration
+			$this->jobList->add(self::class, [
+				'storage_id' => $storageId,
+			]);
+		}
+	}
+}

lib/Db/AccessUpdate.php

@@ -0,0 +1,49 @@
+<?php
+
+/*
+ * Copyright (c) 2022 The Recognize contributors.
+ * This file is licensed under the Affero General Public License version 3 or later. See the COPYING file.
+ */
+declare(strict_types=1);
+namespace OCA\Recognize\Db;
+
+use OCP\AppFramework\Db\Entity;
+
+/**
+ * Class AccessUpdate
+ *
+ * @package OCA\Recognize\Db
+ * @method int getStorageId()
+ * @method setStorageId(int $storageId)
+ * @method int getRootId()
+ * @method setRootId(int $rootId)
+ */
+final class AccessUpdate extends Entity {
+	protected $storageId;
+	protected $rootId;
+
+	/**
+	 * @var string[]
+	 */
+	public static array $columns = ['id', 'storage_id', 'root_id'];
+
+	/**
+	 * @var string[]
+	 */
+	public static array $fields = ['id', 'storageId', 'rootId'];
+
+	public function __construct() {
+		// add types in constructor
+		$this->addType('id', 'integer');
+		$this->addType('storageId', 'integer');
+		$this->addType('rootId', 'integer');
+	}
+
+	public function toArray(): array {
+		$array = [];
+		foreach (self::$fields as $field) {
+			$array[$field] = $this->{$field};
+		}
+		return $array;
+	}
+}

lib/Db/AccessUpdateMapper.php

@@ -0,0 +1,103 @@
+<?php
+
+/*
+ * Copyright (c) 2022 The Recognize contributors.
+ * This file is licensed under the Affero General Public License version 3 or later. See the COPYING file.
+ */
+declare(strict_types=1);
+namespace OCA\Recognize\Db;
+
+use OCA\Recognize\BackgroundJobs\ProcessAccessUpdatesJob;
+use OCP\AppFramework\Db\DoesNotExistException;
+use OCP\AppFramework\Db\MultipleObjectsReturnedException;
+use OCP\AppFramework\Db\QBMapper;
+use OCP\BackgroundJob\IJobList;
+use OCP\DB\Exception;
+use OCP\DB\QueryBuilder\IQueryBuilder;
+use OCP\IDBConnection;
+
+/**
+ * @psalm-extends QBMapper<AccessUpdate>
+ */
+final class AccessUpdateMapper extends QBMapper {
+	public function __construct(
+		IDBConnection $db,
+		private IJobList $jobList,
+	) {
+		parent::__construct($db, 'recognize_access_updates', AccessUpdate::class);
+		$this->db = $db;
+	}
+
+	/**
+	 * @throws \OCP\DB\Exception
+	 * @return list<\OCA\Recognize\Db\AccessUpdate>
+	 */
+	public function findByStorageId(int $storageId, int $limit = 0): array {
+		$qb = $this->db->getQueryBuilder();
+		$qb->selectDistinct(AccessUpdate::$columns)
+			->from($this->getTableName())
+			->where($qb->expr()->eq('storage_id', $qb->createPositionalParameter($storageId, IQueryBuilder::PARAM_INT)));
+		if ($limit > 0) {
+			$qb->setMaxResults($limit);
+		}
+		return $this->findEntities($qb);
+	}
+
+	/**
+	 * @param int $storageId
+	 * @return int
+	 * @throws Exception
+	 */
+	public function countByStorageId(int $storageId): int {
+		$qb = $this->db->getQueryBuilder();
+		$qb->select($qb->func()->count('id'))
+			->from($this->getTableName())
+			->where($qb->expr()->eq('storage_id', $qb->createPositionalParameter($storageId, IQueryBuilder::PARAM_INT)));
+		$result = $qb->executeQuery();
+		$count = $result->fetchOne();
+		$result->closeCursor();
+		if ($count === false) {
+			return 0;
+		}
+		return $count;
+	}
+
+	/**
+	 * @param int $storageId
+	 * @param int $rootId
+	 * @return AccessUpdate
+	 * @throws DoesNotExistException
+	 * @throws MultipleObjectsReturnedException
+	 * @throws Exception
+	 */
+	public function findByStorageIdAndRootId(int $storageId, int $rootId): AccessUpdate {
+		$qb = $this->db->getQueryBuilder();
+		$qb->selectDistinct(AccessUpdate::$columns)
+			->from($this->getTableName())
+			->where($qb->expr()->eq('storage_id', $qb->createPositionalParameter($storageId, IQueryBuilder::PARAM_INT)))
+			->andWhere($qb->expr()->eq('root_id', $qb->createPositionalParameter($rootId, IQueryBuilder::PARAM_INT)));
+		return $this->findEntity($qb);
+	}
+
+	/**
+	 * @param int $storageId
+	 * @param int $rootId
+	 * @return AccessUpdate
+	 * @throws Exception
+	 * @throws MultipleObjectsReturnedException
+	 */
+	public function insertAccessUpdate(int $storageId, int $rootId): AccessUpdate {
+		try {
+			$accessUpdate = $this->findByStorageIdAndRootId($storageId, $rootId);
+		} catch (DoesNotExistException $e) {
+			$accessUpdate = new AccessUpdate();
+			$accessUpdate->setStorageId($storageId);
+			$accessUpdate->setRootId($rootId);
+			$this->insert($accessUpdate);
+			if (!$this->jobList->has(ProcessAccessUpdatesJob::class, [ 'storage_id' => $storageId ])) {
+				$this->jobList->add(self::class, [ 'storage_id' => $storageId ]);
+			}
+		}
+		return $accessUpdate;
+	}
+}

lib/Hooks/FileListener.php

@@ -13,11 +13,13 @@
 use OCA\Recognize\Classifiers\Images\ImagenetClassifier;
 use OCA\Recognize\Classifiers\Video\MovinetClassifier;
 use OCA\Recognize\Constants;
+use OCA\Recognize\Db\AccessUpdateMapper;
 use OCA\Recognize\Db\FaceDetectionMapper;
 use OCA\Recognize\Db\QueueFile;
 use OCA\Recognize\Service\IgnoreService;
 use OCA\Recognize\Service\QueueService;
 use OCA\Recognize\Service\StorageService;
+use OCP\AppFramework\Db\MultipleObjectsReturnedException;
 use OCP\BackgroundJob\IJobList;
 use OCP\DB\Exception;
 use OCP\EventDispatcher\Event;
@@ -38,8 +40,6 @@
 use OCP\Files\Node;
 use OCP\Files\NotFoundException;
 use OCP\IGroupManager;
-use OCP\Share\Events\ShareCreatedEvent;
-use OCP\Share\Events\ShareDeletedEvent;
 use OCP\Share\IManager;
 use Psr\Log\LoggerInterface;
 
@@ -61,12 +61,10 @@ public function __construct(
 		private LoggerInterface $logger,
 		private QueueService $queue,
 		private IgnoreService $ignoreService,
-		private StorageService $storageService,
 		private IRootFolder $rootFolder,
 		private IUserMountCache $userMountCache,
-		private IManager $shareManager,
-		private IGroupManager $groupManager,
 		private IJobList $jobList,
+		private AccessUpdateMapper $accessUpdateMapper,
 	) {
 		$this->movingFromIgnoredTerritory = null;
 		$this->movingDirFromIgnoredTerritory = null;
@@ -110,60 +108,7 @@ public function handle(Event $event): void {
 				// Asynchronous, because we potentially recurse and this event needs to be handled fast
 				$this->onAccessUpdate($event->mountPoint->getStorageId(), $rootId);
 			}
-
-			if ($event instanceof ShareCreatedEvent) {
-				$share = $event->getShare();
-				$ownerId = $share->getShareOwner();
-				$node = $share->getNode();
-				$userIds = $this->getUsersWithFileAccess($node->getId());
-
-				if ($node->getType() === FileInfo::TYPE_FOLDER) {
-					$mount = $node->getMountPoint();
-					if ($mount->getNumericStorageId() === null) {
-						return;
-					}
-					$files = $this->storageService->getFilesInMount($mount->getNumericStorageId(), $node->getId(), [ClusteringFaceClassifier::MODEL_NAME], 0, 0);
-					foreach ($files as $fileInfo) {
-						foreach ($userIds as $userId) {
-							if ($userId === $ownerId) {
-								continue;
-							}
-							if (count($this->faceDetectionMapper->findByFileIdAndUser($node->getId(), $userId)) > 0) {
-								continue;
-							}
-							$this->faceDetectionMapper->copyDetectionsForFileFromUserToUser($fileInfo['fileid'], $ownerId, $userId);
-						}
-					}
-				} else {
-					foreach ($userIds as $userId) {
-						if ($userId === $ownerId) {
-							continue;
-						}
-						if (count($this->faceDetectionMapper->findByFileIdAndUser($node->getId(), $userId)) > 0) {
-							continue;
-						}
-						$this->faceDetectionMapper->copyDetectionsForFileFromUserToUser($node->getId(), $ownerId, $userId);
-					}
-				}
-			}
-			if ($event instanceof ShareDeletedEvent) {
-				$share = $event->getShare();
-				$node = $share->getNode();
-				$userIds = $this->getUsersWithFileAccess($node->getId());
-
-				if ($node->getType() === FileInfo::TYPE_FOLDER) {
-					$mount = $node->getMountPoint();
-					if ($mount->getNumericStorageId() === null) {
-						return;
-					}
-					$files = $this->storageService->getFilesInMount($mount->getNumericStorageId(), $node->getId(), [ClusteringFaceClassifier::MODEL_NAME], 0, 0);
-					foreach ($files as $fileInfo) {
-						$this->faceDetectionMapper->removeDetectionsForFileFromUsersNotInList($fileInfo['fileid'], $userIds);
-					}
-				} else {
-					$this->faceDetectionMapper->removeDetectionsForFileFromUsersNotInList($node->getId(), $userIds);
-				}
-			}
+
 			if ($event instanceof BeforeNodeRenamedEvent) {
 				$this->movingFromIgnoredTerritory = null;
 				$this->movingDirFromIgnoredTerritory = [];
@@ -537,39 +482,10 @@ private function resetIgnoreCache(Node $node) : void {
 	}
 
 	/**
-	 * @throws NotFoundException
-	 * @throws InvalidPathException
 	 * @throws Exception
+	 * @throws MultipleObjectsReturnedException
 	 */
 	private function onAccessUpdate(int $storageId, int $rootId): void {
-		$userIds = $this->getUsersWithFileAccess($rootId);
-		$files = $this->storageService->getFilesInMount($storageId, $rootId, [ClusteringFaceClassifier::MODEL_NAME], 0, 0);
-		$userIdsToScheduleClustering = [];
-		foreach ($files as $fileInfo) {
-			$node = current($this->rootFolder->getById($fileInfo['fileid'])) ?: null;
-			$ownerId = $node?->getOwner()?->getUID();
-			if ($ownerId === null) {
-				continue;
-			}
-			$detectionsForFile = $this->faceDetectionMapper->findByFileId($fileInfo['fileid']);
-			$userHasDetectionForFile = [];
-			foreach ($detectionsForFile as $detection) {
-				$userHasDetectionForFile[$detection->getUserId()] = true;
-			}
-			foreach ($userIds as $userId) {
-				if ($userId === $ownerId) {
-					continue;
-				}
-				if ($userHasDetectionForFile[$userId] ?? false) {
-					continue;
-				}
-				$this->faceDetectionMapper->copyDetectionsForFileFromUserToUser($fileInfo['fileid'], $ownerId, $userId);
-				$userIdsToScheduleClustering[$userId] = true;
-			}
-			$this->faceDetectionMapper->removeDetectionsForFileFromUsersNotInList($fileInfo['fileid'], $userIds);
-		}
-		foreach (array_keys($userIdsToScheduleClustering) as $userId) {
-			$this->jobList->add(ClusterFacesJob::class, ['userId' => $userId]);
-		}
+		$this->accessUpdateMapper->insertAccessUpdate($storageId, $rootId);
 	}
 }

lib/Migration/Version011000001Date20251215094821.php

@@ -0,0 +1,42 @@
+<?php
+
+/*
+ * Copyright (c) 2020-2025 The Recognize contributors.
+ * This file is licensed under the Affero General Public License version 3 or later. See the COPYING file.
+ */
+declare(strict_types=1);
+namespace OCA\Recognize\Migration;
+
+use Closure;
+use Doctrine\DBAL\Schema\SchemaException;
+use OCP\DB\ISchemaWrapper;
+use OCP\DB\Types;
+use OCP\Migration\IOutput;
+use OCP\Migration\SimpleMigrationStep;
+
+final class Version011000001Date20251215094821 extends SimpleMigrationStep {
+
+	/**
+	 * @param IOutput $output
+	 * @param Closure $schemaClosure The `\Closure` returns a `ISchemaWrapper`
+	 * @param array $options
+	 *
+	 * @return ?ISchemaWrapper
+	 * @throws SchemaException
+	 */
+	public function changeSchema(IOutput $output, Closure $schemaClosure, array $options) {
+		/** @var ISchemaWrapper $schema */
+		$schema = $schemaClosure();
+
+		$changed = false;
+		if (!$schema->hasTable('recognize_access_updates')) {
+			$table = $schema->createTable('recognize_access_updates');
+			$table->addColumn('id', Types::BIGINT, ['autoincrement' => true]);
+			$table->addColumn('storage_id', Types::BIGINT, ['notnull' => true]);
+			$table->addColumn('root_id', Types::BIGINT, ['notnull' => true]);
+			$table->addUniqueIndex(['storage_id', 'root_id'], 'recognize_au_unique');
+			$changed = true;
+		}
+		return $changed ? $schema : null;
+	}
+}