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/BackgroundJobs/InitialContentImportJob.php

@@ -11,6 +11,7 @@
 
 use OCA\ContextChat\Logger;
 use OCA\ContextChat\Public\IContentProvider as OCAIContentProvider;
+use OCA\ContextChat\Service\FetchGuard;
 use OCA\ContextChat\Service\ProviderConfigService;
 use OCP\App\IAppManager;
 use OCP\AppFramework\Utility\ITimeFactory;
@@ -29,6 +30,7 @@ public function __construct(
 		private IUserManager $userMan,
 		ITimeFactory $timeFactory,
 		private ?string $userId,
+		private FetchGuard $fetchGuard,
 	) {
 		parent::__construct($timeFactory);
 	}
@@ -56,16 +58,26 @@ protected function run($argument): void {
 			return;
 		}
 
-		$registeredProviders = $this->providerConfig->getProviders();
 		$identifier = ProviderConfigService::getConfigKey($providerObj->getAppId(), $providerObj->getId());
-		if (!isset($registeredProviders[$identifier])
-			|| $registeredProviders[$identifier]['isInitiated']
-		) {
-			$this->logger->info('[InitialContentImportJob] Provider has already been initiated, skipping content import', ['provider' => $identifier]);
-			return;
-		}
+		$running = $this->fetchGuard->jobStarted($identifier, $this->getId(), FetchGuard::IMPORT_RUNNING_TTL);
+		try {
+			// a wipe of the provider removes this job, a re-index after it adds it again
+			if ($this->fetchGuard->isPaused($identifier)) {
+				return;
+			}
 
-		$providerObj->triggerInitialImport();
-		$this->providerConfig->updateProvider($providerObj->getAppId(), $providerObj->getId(), $argument, true);
+			$registeredProviders = $this->providerConfig->getProviders();
+			if (!isset($registeredProviders[$identifier])
+				|| $registeredProviders[$identifier]['isInitiated']
+			) {
+				$this->logger->info('[InitialContentImportJob] Provider has already been initiated, skipping content import', ['provider' => $identifier]);
+				return;
+			}
+
+			$providerObj->triggerInitialImport();
+			$this->providerConfig->updateProvider($providerObj->getAppId(), $providerObj->getId(), $argument, isInitiated: true);
+		} finally {
+			$this->fetchGuard->jobEnded($running);
+		}
 	}
 }

lib/BackgroundJobs/SchedulerJob.php

@@ -11,6 +11,8 @@
 namespace OCA\ContextChat\BackgroundJobs;
 
 use OCA\ContextChat\Logger;
+use OCA\ContextChat\Service\FetchGuard;
+use OCA\ContextChat\Service\ProviderConfigService;
 use OCA\ContextChat\Service\StorageService;
 use OCP\AppFramework\Services\IAppConfig;
 use OCP\AppFramework\Utility\ITimeFactory;
@@ -26,6 +28,7 @@ public function __construct(
 		private IJobList $jobList,
 		private StorageService $storageService,
 		private IAppConfig $appConfig,
+		private FetchGuard $fetchGuard,
 	) {
 		parent::__construct($timeFactory);
 	}
@@ -34,17 +37,27 @@ public function __construct(
 	 * @throws Exception
 	 */
 	protected function run($argument): void {
-		$this->appConfig->setAppValueInt('last_indexed_time', 0, lazy: true);
-		foreach ($this->storageService->getMounts() as $mount) {
-			$this->logger->debug('Scheduling StorageCrawlJob storage_id=' . $mount['storage_id'] . ' root_id=' . $mount['root_id' ] . 'override_root=' . $mount['overridden_root']);
-			$this->jobList->add(StorageCrawlJob::class, [
-				'storage_id' => $mount['storage_id'],
-				'root_id' => $mount['root_id' ],
-				'overridden_root' => $mount['overridden_root'],
-				'last_file_id' => 0,
-			]);
+		$filesKey = ProviderConfigService::getDefaultProviderKey();
+		$running = $this->fetchGuard->jobStarted($filesKey, $this->getId());
+		try {
+			// a wipe of the files removes this job, a re-index after it adds it again
+			if ($this->fetchGuard->isPaused($filesKey)) {
+				return;
+			}
+			foreach ($this->storageService->getMounts() as $mount) {
+				$this->logger->debug('Scheduling StorageCrawlJob storage_id=' . $mount['storage_id'] . ' root_id=' . $mount['root_id' ] . 'override_root=' . $mount['overridden_root']);
+				$this->jobList->add(StorageCrawlJob::class, [
+					'storage_id' => $mount['storage_id'],
+					'root_id' => $mount['root_id' ],
+					'overridden_root' => $mount['overridden_root'],
+					'last_file_id' => 0,
+				]);
+			}
+
+			$this->jobList->remove(self::class);
+			$this->appConfig->setAppValueInt('last_indexed_time', 0, lazy: true);
+		} finally {
+			$this->fetchGuard->jobEnded($running);
 		}
-
-		$this->jobList->remove(self::class);
 	}
 }

lib/BackgroundJobs/StorageCrawlJob.php

@@ -13,6 +13,8 @@
 use OCA\ContextChat\Db\QueueFile;
 use OCA\ContextChat\Logger;
 use OCA\ContextChat\Service\DiagnosticService;
+use OCA\ContextChat\Service\FetchGuard;
+use OCA\ContextChat\Service\ProviderConfigService;
 use OCA\ContextChat\Service\QueueService;
 use OCA\ContextChat\Service\StorageService;
 use OCP\AppFramework\Services\IAppConfig;
@@ -33,6 +35,7 @@ public function __construct(
 		private StorageService $storageService,
 		private DiagnosticService $diagnosticService,
 		private IAppConfig $appConfig,
+		private FetchGuard $fetchGuard,
 	) {
 		parent::__construct($timeFactory);
 	}
@@ -50,7 +53,13 @@ protected function run($argument): void {
 		// Remove current iteration
 		$this->jobList->remove(self::class, $argument);
 
+		$filesKey = ProviderConfigService::getDefaultProviderKey();
+		$running = $this->fetchGuard->jobStarted($filesKey, $this->getId());
 		try {
+			// wipe of everything/for files is running, skip the crawl
+			if ($this->fetchGuard->isPaused($filesKey)) {
+				return;
+			}
 			$this->diagnosticService->sendJobStart(static::class, $this->getId());
 			$this->diagnosticService->sendHeartbeat(static::class, $this->getId());
 
@@ -64,6 +73,7 @@ protected function run($argument): void {
 				$queueFile->setFileId($fileId);
 				$queueFile->setUpdate(false);
 				$this->diagnosticService->sendHeartbeat(static::class, $this->getId());
+				$this->fetchGuard->jobAlive($running);
 				try {
 					$dbFile = $this->queue->insertIntoQueue($queueFile);
 					$lastSuccessfulDbId = $dbFile->getId();
@@ -96,6 +106,7 @@ protected function run($argument): void {
 			}
 		} finally {
 			$this->diagnosticService->sendJobEnd(static::class, $this->getId());
+			$this->fetchGuard->jobEnded($running);
 		}
 	}
 

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;
 	}
 }