nextcloud/lib/private/Files/Cache/Propagator.php

<?php

declare(strict_types=1);

/**
 * SPDX-FileCopyrightText: 2016-2024 Nextcloud GmbH and Nextcloud contributors
 * SPDX-FileCopyrightText: 2016 ownCloud, Inc.
 * SPDX-License-Identifier: AGPL-3.0-only
 */
namespace OC\Files\Cache;

use OCP\DB\QueryBuilder\IQueryBuilder;
use OCP\Files\Cache\IPropagator;
use OCP\Files\Storage\IStorage;
use OCP\IDBConnection;
use OCP\Server;
use Override;
use Psr\Clock\ClockInterface;
use Psr\Log\LoggerInterface;

class Propagator implements IPropagator {
	private array $batch = [];
	private ClockInterface $clock;

	public function __construct(
		protected readonly IStorage $storage,
		private readonly IDBConnection $connection,
		private readonly array $ignore = [],
	) {
		$this->clock = Server::get(ClockInterface::class);
	}

	public function __destruct() {
		$this->commitBatch();
	}

	#[Override]
	public function propagateChange(string $internalPath, int $time, int $sizeDifference = 0): void {
		// Do not propagate changes in ignored paths
		foreach ($this->ignore as $ignore) {
			if (str_starts_with($internalPath, $ignore)) {
				return;
			}
		}

		$time = min($time, $this->clock->now()->getTimestamp());

		$parents = $this->getParents($internalPath);

		foreach ($parents as $parent) {
			$this->addToBatch($parent, $time, $sizeDifference);
		}
	}

	/**
	 * @return string[]
	 */
	protected function getParents(string $path): array {
		$parts = explode('/', $path);
		$parent = '';
		$parents = [];
		foreach ($parts as $part) {
			$parents[] = $parent;
			$parent = trim($parent . '/' . $part, '/');
		}
		return $parents;
	}

	#[Override]
	public function beginBatch(): void {
		Server::get(LoggerInterface::class)->debug('Propagated changes are committed in batches automatically, so it is no longer necessary to begin a batch manually.');
	}

	private function addToBatch(string $internalPath, int $time, int $sizeDifference): void {
		if (!isset($this->batch[$internalPath])) {
			$this->batch[$internalPath] = [
				'hash' => md5($internalPath),
				'time' => $time,
				'size' => $sizeDifference,
			];
		} else {
			$this->batch[$internalPath]['size'] += $sizeDifference;
			if ($time > $this->batch[$internalPath]['time']) {
				$this->batch[$internalPath]['time'] = $time;
			}
		}
	}

	#[Override]
	public function commitBatch(): void {
		// Ensure rows are always locked in the same order
		uasort($this->batch, static fn (array $a, array $b) => $a['hash'] <=> $b['hash']);

		try {
			$this->connection->beginTransaction();

			$storageId = $this->storage->getCache()->getNumericStorageId();

			if ($this->connection->getDatabaseProvider() !== IDBConnection::PLATFORM_SQLITE) {
				// Lock the rows before updating then with a SELECT FOR UPDATE
				// The select also allow us to fetch the fileid and then use these in the UPDATE
				// queries as a faster lookup than the path_hash
				$hashes = array_map(static fn (array $a): string => $a['hash'], $this->batch);

				foreach (array_chunk($hashes, 1000) as $hashesChunk) {
					$query = $this->connection->getQueryBuilder();
					$result = $query->select('fileid', 'path', 'path_hash', 'size')
						->from('filecache')
						->where($query->expr()->eq('storage', $query->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
						->andWhere($query->expr()->in('path_hash', $query->createNamedParameter($hashesChunk, IQueryBuilder::PARAM_STR_ARRAY)))
						->orderBy('path_hash')
						->forUpdate()
						->executeQuery();

					$query = $this->connection->getQueryBuilder();
					$query->update('filecache')
						->set('mtime', $query->func()->greatest('mtime', $query->createParameter('time')))
						->set('etag', $query->expr()->literal(uniqid()))
						->where($query->expr()->eq('storage', $query->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
						->andWhere($query->expr()->eq('fileid', $query->createParameter('fileid')));

					$queryWithSize = $this->connection->getQueryBuilder();
					$queryWithSize->update('filecache')
						->set('mtime', $queryWithSize->func()->greatest('mtime', $queryWithSize->createParameter('time')))
						->set('etag', $queryWithSize->expr()->literal(uniqid()))
						->set('size', $queryWithSize->func()->add('size', $queryWithSize->createParameter('size')))
						->where($queryWithSize->expr()->eq('storage', $queryWithSize->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
						->andWhere($queryWithSize->expr()->eq('fileid', $queryWithSize->createParameter('fileid')));

					while ($row = $result->fetchAssociative()) {
						$item = $this->batch[$row['path']];
						if ($item['size'] && $row['size'] > -1) {
							$queryWithSize->setParameter('fileid', $row['fileid'], IQueryBuilder::PARAM_INT)
								->setParameter('size', $item['size'], IQueryBuilder::PARAM_INT)
								->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT)
								->executeStatement();
						} else {
							$query->setParameter('fileid', $row['fileid'], IQueryBuilder::PARAM_INT)
								->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT)
								->executeStatement();
						}
					}
				}
			} else {
				// No FOR UPDATE support in Sqlite, but instead the whole table is locked
				$query = $this->connection->getQueryBuilder();
				$query->update('filecache')
					->set('mtime', $query->func()->greatest('mtime', $query->createParameter('time')))
					->set('etag', $query->expr()->literal(uniqid()))
					->where($query->expr()->eq('storage', $query->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
					->andWhere($query->expr()->eq('path_hash', $query->createParameter('hash')));

				$queryWithSize = $this->connection->getQueryBuilder();
				$queryWithSize->update('filecache')
					->set('mtime', $queryWithSize->func()->greatest('mtime', $queryWithSize->createParameter('time')))
					->set('etag', $queryWithSize->expr()->literal(uniqid()))
					->set('size', $queryWithSize->func()->add('size', $queryWithSize->createParameter('size')))
					->where($queryWithSize->expr()->eq('storage', $queryWithSize->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
					->andWhere($queryWithSize->expr()->eq('path_hash', $queryWithSize->createParameter('hash')));

				foreach ($this->batch as $item) {
					if ($item['size']) {
						$queryWithSize->setParameter('hash', $item['hash'], IQueryBuilder::PARAM_STR)
							->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT)
							->setParameter('size', $item['size'], IQueryBuilder::PARAM_INT)
							->executeStatement();
					} else {
						$query->setParameter('hash', $item['hash'], IQueryBuilder::PARAM_STR)
							->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT)
							->executeStatement();
					}
				}
			}

			$this->batch = [];

			$this->connection->commit();
		} catch (\Exception $e) {
			$this->connection->rollback();
			throw $e;
		}
	}
}