feat: add a full wipe command

.github/workflows/integration-test-cron.yml

@@ -127,6 +127,8 @@ jobs:
         with:
           repository: nextcloud/context_chat_backend
           path: context_chat_backend/
+          # todo: remove this
+          ref: feat/vector-db-wipe
           persist-credentials: false
 
       - name: Checkout viewer
@@ -410,6 +412,73 @@ jobs:
           kill -9 $WORKER1_PID
           kill -9 $WORKER2_PID
 
+      - name: Wipe and reindex only the files provider
+        run: |
+          ./occ context_chat:wipe --force --provider files__default --reindex
+
+          stats=$(./occ context_chat:stats --json)
+          echo "$stats"
+          indexed=$(echo "$stats" | jq '.vectordb_document_counts.files__default // 0')
+          if [ "$indexed" != "0" ]; then
+            echo "Error: expected no indexed files right after the files wipe, got $indexed"
+            exit 1
+          fi
+
+          echo "::group::Waiting up to 10 minutes for the files to be indexed again"
+          for i in {1..60}; do
+            stats=$(timeout 30 ./occ context_chat:stats --json || true)
+            indexed=$(echo "$stats" | jq '.vectordb_document_counts.files__default // 0' 2>/dev/null || echo 0)
+            echo "Attempt $i: indexed=$indexed"
+            if [ "$indexed" -ge 10 ]; then
+              echo "::endgroup::"
+              echo "Files reindexed, $indexed documents"
+              exit 0
+            fi
+            sleep 10
+          done
+          echo "::endgroup::"
+          echo "Error: files were not indexed again after the files wipe"
+          exit 1
+
+      - name: Wipe all the indexed data and check that the queues and the vector DB are empty
+        run: |
+          ./occ context_chat:wipe --force
+
+          stats=$(./occ context_chat:stats --json)
+          echo "$stats"
+          indexed=$(echo "$stats" | jq '[.vectordb_document_counts | to_entries[].value] | add // 0')
+          queued=$(echo "$stats" | jq '[.queued_documents_counts | to_entries[].value] | add // 0')
+          actions=$(echo "$stats" | jq '.queued_actions_count')
+          fs_events=$(echo "$stats" | jq '.queued_fs_events_count')
+          # the queued counts leave out locked items, check those separately
+          queued_locked=$(echo "$stats" | jq '[.queued_documents_locked_counts | to_entries[].value] | add // 0')
+          actions_locked=$(echo "$stats" | jq '.queued_actions_locked_count')
+          backend_available=$(echo "$stats" | jq '.backend_available')
+          echo "indexed=$indexed queued=$queued queued_locked=$queued_locked actions=$actions actions_locked=$actions_locked fs_events=$fs_events backend_available=$backend_available"
+          if [ "$backend_available" != "true" ] || [ "$indexed" != "0" ] || [ "$queued" != "0" ] || [ "$queued_locked" != "0" ] || [ "$actions" != "0" ] || [ "$actions_locked" != "0" ] || [ "$fs_events" != "0" ]; then
+            echo "Error: expected an empty vector DB and empty queues after the wipe"
+            exit 1
+          fi
+
+      - name: Reindex and check that documents get indexed again
+        run: |
+          ./occ context_chat:reindex
+          echo "::group::Waiting up to 10 minutes for documents to be indexed again"
+          for i in {1..60}; do
+            stats=$(timeout 30 ./occ context_chat:stats --json || true)
+            indexed=$(echo "$stats" | jq '.vectordb_document_counts.files__default // 0' 2>/dev/null || echo 0)
+            echo "Attempt $i: indexed=$indexed"
+            if [ "$indexed" -ge 10 ]; then
+              echo "Reindex works, $indexed documents indexed again"
+              echo "::endgroup::"
+              exit 0
+            fi
+            sleep 10
+          done
+          echo "::endgroup::"
+          echo "Error: documents were not indexed again after the reindex"
+          exit 1
+
       - name: Show nextcloud logs
         if: always()
         run: |

.github/workflows/integration-test-file-listener.yml

@@ -115,6 +115,8 @@ jobs:
         with:
           repository: nextcloud/context_chat_backend
           path: context_chat_backend/
+          # todo: remove this
+          ref: feat/vector-db-wipe
           persist-credentials: false
 
       - name: Checkout viewer

appinfo/info.xml

@@ -47,6 +47,7 @@ Refer to the [Context Chat Backend's readme](https://github.com/nextcloud/contex
 		<command>OCA\ContextChat\Command\Search</command>
 		<command>OCA\ContextChat\Command\Statistics</command>
 		<command>OCA\ContextChat\Command\Reindex</command>
+		<command>OCA\ContextChat\Command\Wipe</command>
 	</commands>
     <repair-steps>
         <install>

lib/Command/Reindex.php

@@ -9,43 +9,66 @@
 
 namespace OCA\ContextChat\Command;
 
-use OCA\ContextChat\BackgroundJobs\SchedulerJob;
-use OCP\BackgroundJob\IJobList;
+use OCA\ContextChat\Service\ReindexService;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 
 /**
  * Re-seed the one-shot filesystem crawl/index on demand.
  *
  * SchedulerJob -> StorageCrawlJob -> IndexerJob is seeded only by the <install> repair step and
  * removes itself once the initial crawl finishes. There is otherwise no way to re-enumerate
- * mounts (e.g. after installing on an instance whose files predate the app, or to recover a crawl
- * that did not complete) short of reinstalling the app. This command re-adds SchedulerJob so the
- * full enumeration runs again; it is a no-op if one is already scheduled, and already-indexed
- * files are skipped.
+ * mounts (e.g. to recover a crawl that did not complete, or after context_chat:wipe) short of
+ * reinstalling the app. This command re-adds SchedulerJob so the full enumeration runs again if one
+ * doesn't exist already.
+ * Content providers' initial import also gets scheduled (e.g. after context_chat:wipe) if it has not run yet.
  */
 class Reindex extends Command {
 
 	public function __construct(
-		private IJobList $jobList,
+		private ReindexService $reindexService,
 	) {
 		parent::__construct();
 	}
 
 	protected function configure() {
 		$this->setName('context_chat:reindex')
-			->setDescription('Schedule a full re-crawl of all the files in all the mounts. Indexed files are not re-indexed when compared against context_chat_backend\'s vector DB. Content providers are not re-indexed.');
+			->setDescription(
+				'Schedule a full re-crawl of all the files in all the mounts and the import of the content providers'
+				. ' that were not imported yet. Indexed files are not re-indexed when compared against context_chat_backend\'s vector DB.'
+			)
+			->addOption(
+				'provider',
+				'p',
+				InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY,
+				'Only reindex this provider, e.g. "files__default" (appId__providerId) for the files.'
+				. ' A named content provider is imported again even if it was imported before. Multiple values can be specified by repeating the option.',
+			);
 	}
 
 	protected function execute(InputInterface $input, OutputInterface $output): int {
-		if ($this->jobList->has(SchedulerJob::class, null)) {
-			$output->writeln('<comment>A full re-crawl is already scheduled; nothing to do.</comment>');
+		/** @var list<string> $providerKeys */
+		$providerKeys = $input->getOption('provider');
+		$scope = $providerKeys === [] ? null : $providerKeys;
+
+		if ($scope !== null) {
+			$known = $this->reindexService->getRegisteredProviderKeys();
+			$unknown = array_diff($scope, $known);
+			if ($unknown !== []) {
+				$output->writeln('<error>Unknown provider(s): ' . implode(', ', $unknown) . '</error>');
+				$output->writeln('Registered providers: ' . implode(', ', $known));
+				return 1;
+			}
+		}
+
+		if (!$this->reindexService->scheduleReindex($scope)) {
+			$output->writeln('<comment>The re-index is already scheduled; nothing to do.</comment>');
 			return 0;
 		}
 
-		$this->jobList->add(SchedulerJob::class);
-		$output->writeln('<info>Scheduled a full re-crawl. SchedulerJob will enumerate all mounts on its next run.</info>');
+		$output->writeln('<info>Scheduled the re-index. The background jobs will queue the documents on their next run.</info>');
 		return 0;
 	}
 }

lib/Command/Wipe.php

@@ -0,0 +1,140 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors
+ * SPDX-License-Identifier: AGPL-3.0-or-later
+ */
+
+namespace OCA\ContextChat\Command;
+
+use OCA\ContextChat\Service\ReindexService;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Helper\QuestionHelper;
+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 Wipe extends Command {
+
+	public function __construct(
+		private ReindexService $reindexService,
+	) {
+		parent::__construct();
+	}
+
+	protected function configure() {
+		$this->setName('context_chat:wipe')
+			->setDescription(
+				'Delete the indexed documents from context_chat_backend\'s vector DB and clear the queues.'
+				. ' With --provider option, only the provider\'s data is deleted, the actions queue, and the indexing statistics are left intact.'
+				. ' Nothing gets indexed again until context_chat:reindex is run or the --reindex option is used.'
+			)
+			->addOption(
+				'provider',
+				'p',
+				InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY,
+				'Only wipe this provider, e.g. "files__default" (appId__providerId) for the files. Multiple values can be specified by repeating the option.',
+			)
+			->addOption(
+				'reindex',
+				null,
+				InputOption::VALUE_NONE,
+				'Schedule a re-crawl of the files and a re-import of the content providers that were wiped',
+			)
+			->addOption(
+				'ignore-handed-out',
+				null,
+				InputOption::VALUE_NONE,
+				'Do not wait for the documents context_chat_backend is still indexing. Only use this when the backend was stopped while indexing them, otherwise they can be written to the vector DB after the wipe.',
+			)
+			->addOption(
+				'force',
+				'f',
+				InputOption::VALUE_NONE,
+				'Do not ask for confirmation',
+			);
+	}
+
+	protected function execute(InputInterface $input, OutputInterface $output): int {
+		/** @var list<string> $providerKeys */
+		$providerKeys = $input->getOption('provider');
+		$scope = $providerKeys === [] ? null : $providerKeys;
+
+		if ($scope !== null) {
+			try {
+				$known = $this->reindexService->getWipeableProviderKeys();
+			} catch (\RuntimeException $e) {
+				$output->writeln('<error>Could not list the indexed providers: ' . $e->getMessage() . '</error>');
+				return 1;
+			}
+			$unknown = array_diff($scope, $known);
+			if ($unknown !== []) {
+				$output->writeln('<error>Unknown provider(s): ' . implode(', ', $unknown) . '</error>');
+				$output->writeln('Known providers: ' . implode(', ', $known));
+				return 1;
+			}
+		}
+
+		$scopeText = $scope === null ? '' : ' of ' . implode(', ', $scope);
+
+		if (!$input->getOption('force')) {
+			/** @var QuestionHelper $helper */
+			$helper = $this->getHelper('question');
+			$question = new ConfirmationQuestion(
+				'All the indexed documents in Context Chat'
+				. $scopeText
+				. ' will be deleted. The original documents are not touched. Continue? [y/N] ',
+				false,
+			);
+			if (!$helper->ask($input, $output, $question)) {
+				$output->writeln('<comment>Aborted, nothing was deleted. Use --force to skip the confirmation.</comment>');
+				return 1;
+			}
+		}
+
+		try {
+			$this->reindexService->wipe(
+				$scope,
+				(bool)$input->getOption('ignore-handed-out'),
+				static function (string $message) use ($output): void {
+					$output->writeln('<info>' . $message . '...</info>');
+				},
+			);
+		} catch (\Throwable $e) {
+			$output->writeln('<error>Wipe failed. The command can be safely run again once the problem is fixed.</error>');
+			$output->writeln('<error>Error: ' . $e->getMessage() . '</error>');
+			$output->writeln('<error>' . $e->getTraceAsString() . '</error>');
+			return 1;
+		}
+		$output->writeln('<info>Deleted all the indexed documents' . $scopeText . ' and cleared the queues.</info>');
+
+		// providers that are not registered anymore cannot be indexed again
+		$reindexScope = $scope === null
+			? null
+			: array_values(array_intersect($scope, $this->reindexService->getRegisteredProviderKeys()));
+		if ($reindexScope === []) {
+			return 0;
+		}
+
+		if ($input->getOption('reindex')) {
+			try {
+				$this->reindexService->scheduleReindex($reindexScope);
+			} catch (\Throwable $e) {
+				$output->writeln('<error>Scheduling re-index failed. It can be tried again with context_chat:reindex.</error>');
+				$output->writeln('<error>Error: ' . $e->getMessage() . '</error>');
+				$output->writeln('<error>' . $e->getTraceAsString() . '</error>');
+				return 1;
+			}
+			$output->writeln('<info>Scheduled the re-index.</info>');
+		} elseif ($reindexScope === null) {
+			$output->writeln('Run context_chat:reindex to index the documents again.');
+		} else {
+			$output->writeln('Run context_chat:reindex --provider ' . implode(' --provider ', $reindexScope) . ' to index the documents again.');
+		}
+
+		return 0;
+	}
+}

lib/Controller/QueueController.php

@@ -17,6 +17,7 @@
 use OCA\ContextChat\Db\QueueContentItemMapper;
 use OCA\ContextChat\Db\QueueFile;
 use OCA\ContextChat\Db\QueueMapper;
+use OCA\ContextChat\Service\FetchGuard;
 use OCA\ContextChat\Service\ProviderConfigService;
 use OCA\ContextChat\Type\Source;
 use OCP\AppFramework\Http;
@@ -47,6 +48,7 @@ public function __construct(
 		private IJobList $jobList,
 		private ITimeFactory $timeFactory,
 		private QueueMapper $queueMapper,
+		private FetchGuard $fetchGuard,
 		string $corsMethods = 'PUT, POST, GET, DELETE, PATCH',
 		string $corsAllowedHeaders = 'Authorization, Content-Type, Accept, OCS-APIRequest',
 		int $corsMaxAge = 1728000,
@@ -106,6 +108,13 @@ public function getDocumentsQueueItems(
 		if ($n > $maxN) {
 			$n = $maxN;
 		}
+		// hand out nothing, the backend waits and asks again
+		if ($this->fetchGuard->isPaused()) {
+			return new DataResponse([
+				'files' => new \stdClass(),
+				'content_providers' => new \stdClass(),
+			]);
+		}
 		try {
 			$files = [];
 			while (count($files) < $n) {
@@ -216,6 +225,10 @@ public function countDocumentsQueueItems(QueueMapper $queueMapper, QueueContentI
 	#[ExAppRequired]
 	#[ApiRoute(verb: 'GET', url: '/queues/actions/')]
 	public function getActionsQueueItems(QueueActionMapper $queueActionMapper, int $n = 512) : DataResponse {
+		if ($this->fetchGuard->isPaused()) {
+			// stdClass is used to represent an empty object in the response
+			return new DataResponse(['actions' => new \stdClass()]);
+		}
 		try {
 			$actions = [];
 			while (count($actions) < $n) {

lib/Db/QueueContentItemMapper.php

@@ -71,6 +71,17 @@ public function removeFromQueue(array $ids): void {
 		}
 	}
 
+	/**
+	 * @throws \OCP\DB\Exception
+	 */
+	public function removeProvider(string $appId, string $providerId): void {
+		$qb = $this->db->getQueryBuilder();
+		$qb->delete($this->getTableName())
+			->where($qb->expr()->eq('app_id', $qb->createNamedParameter($appId)))
+			->andWhere($qb->expr()->eq('provider_id', $qb->createNamedParameter($providerId)))
+			->executeStatement();
+	}
+
 	/**
 	 * @throws \OCP\DB\Exception
 	 * @return array<string, int>