diff --git a/.github/workflows/integration-test.yml b/.github/workflows/integration-test.yml index 0c8d6dd9..41f8993d 100644 --- a/.github/workflows/integration-test.yml +++ b/.github/workflows/integration-test.yml @@ -32,8 +32,18 @@ jobs: php-versions: [ '8.1' ] databases: [ 'pgsql' ] server-versions: [ 'stable30', 'stable31', 'master' ] - - name: Integration test on ${{ matrix.server-versions }} php@${{ matrix.php-versions }} + encryption: ['on', 'off'] + exclude: + - server-versions: stable30 + databases: pgsql + php-versions: 8.1 + encryption: on + - server-versions: stable31 + databases: pgsql + php-versions: 8.1 + encryption: on + + name: Integration test โ˜๏ธ${{ matrix.server-versions }} ๐Ÿ˜${{ matrix.php-versions }} ๐Ÿ”:${{ matrix.encryption }} env: MYSQL_PORT: 4444 @@ -136,6 +146,12 @@ jobs: - name: Enable context_chat, app_api and testing run: ./occ app:enable -vvv -f ${{ env.APP_NAME }} app_api testing + - name: Enable encryption + if: ${{ matrix.encryption == 'on' }} + run: | + ./occ app:enable -vvv -f encryption + ./occ encryption:enable + - name: Checkout documentation uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 with: @@ -164,6 +180,34 @@ jobs: cd .. rm -rf documentation + - name: Reduce number of files + run: | + cd data/admin/files + LIMIT=100 + + # Find all files in the directory and its subdirectories + FILES=($(find papers/ -type f)) + + # Count the number of files + FILE_COUNT=${#FILES[@]} + + # Check if the current number of files exceeds the limit + if [ "$FILE_COUNT" -le "$LIMIT" ]; then + echo "No files need to be deleted. Current count: $FILE_COUNT" + exit 0 + fi + + # Calculate how many files to delete + FILES_TO_DELETE=$((FILE_COUNT - LIMIT)) + + # Sort files by modification time (oldest first) and delete the oldest ones + for FILE in $(ls -t "${FILES[@]}" | tail -n "$FILES_TO_DELETE"); do + echo "Deleting: $FILE" + rm "$FILE" + done + + echo "Deleted $FILES_TO_DELETE files. Current count: $((FILE_COUNT - FILES_TO_DELETE))" + - name: Setup python 3.11 uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 with: @@ -181,6 +225,7 @@ jobs: python3 -u ./main_em.py > em_backend_logs 2>&1 & python3 -u ./main.py > backend_logs 2>&1 & echo $! > ../pid.txt # Save the process ID (PID) + sleep 60 # Wait for the backend to get ready - name: Register backend run: | @@ -219,10 +264,20 @@ jobs: - name: Run the prompts run: | - ./occ background-job:worker 'OC\TaskProcessing\SynchronousBackgroundJob' & - ./occ background-job:worker 'OC\TaskProcessing\SynchronousBackgroundJob' & - ./occ context_chat:prompt admin "Which factors are taken into account for the Ethical AI Rating?" - ./occ context_chat:prompt admin "Welche Faktoren beeinflussen das Ethical AI Rating?" + ./occ background-job:worker --stop_after=2m 'OC\TaskProcessing\SynchronousBackgroundJob' & + ./occ background-job:worker --stop_after=2m 'OC\TaskProcessing\SynchronousBackgroundJob' & + OUT1=$(./occ context_chat:prompt admin "Which factors are taken into account for the Ethical AI Rating?") + echo "$OUT1" + echo '--------------------------------------------------' + OUT2=$(./occ context_chat:prompt admin "Welche Faktoren beeinflussen das Ethical AI Rating?") + echo "$OUT2" + echo '--------------------------------------------------' + + OUT3=$(./occ context_chat:search admin "Welche Faktoren beeinflussen das Ethical AI Rating?") + echo "$OUT3" + echo "$OUT1" | grep -q "If all of these points are met, we give a Green label." || exit 1 + echo "$OUT2" | grep -q "If all of these points are met, we give a Green label." || exit 1 + echo "$OUT3" | grep -q "overview.rst" || exit 1 - name: Check python memory usage run: | @@ -283,8 +338,18 @@ jobs: php-versions: [ '8.1' ] databases: [ 'pgsql' ] server-versions: [ 'stable30', 'stable31', 'master' ] - - name: Integration test with cron on ${{ matrix.server-versions }} php@${{ matrix.php-versions }} + encryption: ['on', 'off'] + exclude: + - server-versions: stable30 + databases: pgsql + php-versions: 8.1 + encryption: on + - server-versions: stable31 + databases: pgsql + php-versions: 8.1 + encryption: on + + name: Integration test with cron โ˜๏ธ${{ matrix.server-versions }} ๐Ÿ˜${{ matrix.php-versions }} ๐Ÿ”:${{ matrix.encryption }} env: MYSQL_PORT: 4444 @@ -387,6 +452,12 @@ jobs: - name: Enable app_api and testing run: ./occ app:enable -vvv -f app_api testing + - name: Enable encryption + if: ${{ matrix.encryption == 'on' }} + run: | + ./occ app:enable -vvv -f encryption + ./occ encryption:enable + - name: Setup python 3.11 uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 with: @@ -404,6 +475,7 @@ jobs: python3 -u ./main_em.py > em_backend_logs 2>&1 & python3 -u ./main.py > backend_logs 2>&1 & echo $! > ../pid.txt # Save the process ID (PID) + sleep 60 # Wait for the backend to get ready - name: Register backend run: | @@ -442,6 +514,34 @@ jobs: cd .. rm -rf documentation + - name: Reduce number of files + run: | + cd data/admin/files + LIMIT=100 + + # Find all files in the directory and its subdirectories + FILES=($(find papers/ -type f)) + + # Count the number of files + FILE_COUNT=${#FILES[@]} + + # Check if the current number of files exceeds the limit + if [ "$FILE_COUNT" -le "$LIMIT" ]; then + echo "No files need to be deleted. Current count: $FILE_COUNT" + exit 0 + fi + + # Calculate how many files to delete + FILES_TO_DELETE=$((FILE_COUNT - LIMIT)) + + # Sort files by modification time (oldest first) and delete the oldest ones + for FILE in $(ls -t "${FILES[@]}" | tail -n "$FILES_TO_DELETE"); do + echo "Deleting: $FILE" + rm "$FILE" + done + + echo "Deleted $FILES_TO_DELETE files. Current count: $((FILE_COUNT - FILES_TO_DELETE))" + - name: Run files scan run: | ./occ files:scan admin # Do the scan before enabling context chat @@ -454,6 +554,8 @@ jobs: # Run cron in speed mode: Set interval to 0 minutes ./occ config:app:set --value 30 --type integer context_chat indexing_job_interval # 30 seconds ./occ config:app:set --value 10 --type integer context_chat crawl_job_interval # 10 seconds + ./occ config:app:set --value 10 --type integer context_chat action_job_interval # 10 seconds + ./occ config:app:set --value 10 --type integer context_chat fs_listener_job_interval # 10 seconds for i in {1..100}; do php cron.php & # Starting with stable31 we can use -v here for better visibility wait @@ -489,10 +591,20 @@ jobs: - name: Run the prompts run: | - ./occ background-job:worker 'OC\TaskProcessing\SynchronousBackgroundJob' & - ./occ background-job:worker 'OC\TaskProcessing\SynchronousBackgroundJob' & - ./occ context_chat:prompt admin "Which factors are taken into account for the Ethical AI Rating?" - ./occ context_chat:prompt admin "Welche Faktoren beeinflussen das Ethical AI Rating?" + ./occ background-job:worker --stop_after=2m 'OC\TaskProcessing\SynchronousBackgroundJob' & + ./occ background-job:worker --stop_after=2m 'OC\TaskProcessing\SynchronousBackgroundJob' & + OUT1=$(./occ context_chat:prompt admin "Which factors are taken into account for the Ethical AI Rating?") + echo "$OUT1" + echo '--------------------------------------------------' + OUT2=$(./occ context_chat:prompt admin "Welche Faktoren beeinflussen das Ethical AI Rating?") + echo "$OUT2" + echo '--------------------------------------------------' + + OUT3=$(./occ context_chat:search admin "Welche Faktoren beeinflussen das Ethical AI Rating?") + echo "$OUT3" + echo "$OUT1" | grep -q "If all of these points are met, we give a Green label." || exit 1 + echo "$OUT2" | grep -q "If all of these points are met, we give a Green label." || exit 1 + echo "$OUT3" | grep -q "overview.rst" || exit 1 - name: Show nextcloud logs if: always() @@ -522,8 +634,32 @@ jobs: php-versions: [ '8.1' ] databases: [ 'pgsql' ] server-versions: [ 'stable30', 'stable31', 'master' ] - - name: Integration test with file listener on ${{ matrix.server-versions }} php@${{ matrix.php-versions }} + encryption: ['on', 'off'] + file-deletion-method: [ 'occ', 'os' ] + exclude: + - server-versions: stable30 + file-deletion-method: os + databases: pgsql + php-versions: 8.1 + encryption: on + - server-versions: stable31 + file-deletion-method: os + databases: pgsql + php-versions: 8.1 + encryption: on + - server-versions: stable30 + file-deletion-method: os + databases: pgsql + php-versions: 8.1 + encryption: off + - server-versions: stable31 + file-deletion-method: os + databases: pgsql + php-versions: 8.1 + encryption: off + + + name: Integration test with file listener โ˜๏ธ${{ matrix.server-versions }} ๐Ÿ˜${{ matrix.php-versions }} ๐Ÿ—‘๏ธ${{ matrix.file-deletion-method }} ๐Ÿ”:${{ matrix.encryption }} env: MYSQL_PORT: 4444 @@ -626,6 +762,12 @@ jobs: - name: Enable app_api and testing run: ./occ app:enable -vvv -f app_api testing + - name: Enable encryption + if: ${{ matrix.encryption == 'on' }} + run: | + ./occ app:enable -vvv -f encryption + ./occ encryption:enable + - name: Setup python 3.11 uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 with: @@ -643,6 +785,7 @@ jobs: python3 -u ./main_em.py > em_backend_logs 2>&1 & python3 -u ./main.py > backend_logs 2>&1 & echo $! > ../pid.txt # Save the process ID (PID) + sleep 60 # Wait for the backend to get ready - name: Register backend run: | @@ -658,6 +801,8 @@ jobs: # Run cron in speed mode ./occ config:app:set --value 30 --type integer context_chat indexing_job_interval # 30 seconds ./occ config:app:set --value 10 --type integer context_chat crawl_job_interval # 10 seconds + ./occ config:app:set --value 10 --type integer context_chat action_job_interval # 10 seconds + ./occ config:app:set --value 10 --type integer context_chat fs_listener_job_interval # 10 seconds # Run normal indexing jobs which will only pick up welcome.txt etc for i in {1..10}; do php cron.php & # Starting with stable31 we can use -v here for better visibility @@ -694,9 +839,37 @@ jobs: cd .. rm -rf documentation + - name: Reduce number of files + run: | + cd data/admin/files + LIMIT=100 + + # Find all files in the directory and its subdirectories + FILES=($(find papers/ -type f)) + + # Count the number of files + FILE_COUNT=${#FILES[@]} + + # Check if the current number of files exceeds the limit + if [ "$FILE_COUNT" -le "$LIMIT" ]; then + echo "No files need to be deleted. Current count: $FILE_COUNT" + exit 0 + fi + + # Calculate how many files to delete + FILES_TO_DELETE=$((FILE_COUNT - LIMIT)) + + # Sort files by modification time (oldest first) and delete the oldest ones + for FILE in $(ls -t "${FILES[@]}" | tail -n "$FILES_TO_DELETE"); do + echo "Deleting: $FILE" + rm "$FILE" + done + + echo "Deleted $FILES_TO_DELETE files. Current count: $((FILE_COUNT - FILES_TO_DELETE))" + - name: Run files scan run: | - ./occ files:scan admin # Do the scan before enabling context chat + ./occ files:scan admin # We do the scan after enabling context chat here, so we can test the file listeners - name: Run indexer cron run: | @@ -735,10 +908,125 @@ jobs: - name: Run the prompts run: | - ./occ background-job:worker 'OC\TaskProcessing\SynchronousBackgroundJob' & - ./occ background-job:worker 'OC\TaskProcessing\SynchronousBackgroundJob' & - ./occ context_chat:prompt admin "Which factors are taken into account for the Ethical AI Rating?" - ./occ context_chat:prompt admin "Welche Faktoren beeinflussen das Ethical AI Rating?" + ./occ background-job:worker --stop_after=2m 'OC\TaskProcessing\SynchronousBackgroundJob' & + ./occ background-job:worker --stop_after=2m 'OC\TaskProcessing\SynchronousBackgroundJob' & + set +e + + # Check for user admin: Should be there + OUT_ADMIN_ENGLISH_PROMPT_AI=$(./occ context_chat:prompt admin "Which factors are taken into account for the Ethical AI Rating?") + OUT_ADMIN_GERMAN_PROMPT_AI=$(./occ context_chat:prompt admin "Welche Faktoren beeinflussen das Ethical AI Rating?") + OUT_ADMIN_SEARCH_AI=$(./occ context_chat:search admin "Welche Faktoren beeinflussen das Ethical AI Rating?") + + # Check for user test: Shouldn't be there + OUT_TEST_ENGLISH_PROMPT_AI=$(./occ context_chat:prompt test "Which factors are taken into account for the Ethical AI Rating?") + OUT_TEST_GERMAN_PROMPT_AI=$(./occ context_chat:prompt test "Welche Faktoren beeinflussen das Ethical AI Rating?") + OUT_TEST_SEARCH_AI=$(./occ context_chat:search test "Welche Faktoren beeinflussen das Ethical AI Rating?") + + # Check for user test: Shouldn't be there, because it was not shared yet + OUT_TEST_PROMPT_JOBS=$(./occ context_chat:prompt test "How do I set my TimedJob to be time insensitive?") + OUT_TEST_SEARCH_JOBS=$(./occ context_chat:search test "How do I set my TimedJob to be time insensitive?") + + echo "OUT_ADMIN_ENGLISH_PROMPT_AI" + echo "$OUT_ADMIN_ENGLISH_PROMPT_AI" + echo "OUT_ADMIN_GERMAN_PROMPT_AI" + echo "$OUT_ADMIN_GERMAN_PROMPT_AI" + echo "OUT_ADMIN_SEARCH_AI" + echo "$OUT_ADMIN_SEARCH_AI" + echo "OUT_TEST_ENGLISH_PROMPT_AI" + echo "$OUT_TEST_ENGLISH_PROMPT_AI" + echo "OUT_TEST_GERMAN_PROMPT_AI" + echo "$OUT_TEST_GERMAN_PROMPT_AI" + echo "OUT_TEST_SEARCH_AI" + echo "$OUT_TEST_SEARCH_AI" + echo "OUT_TEST_PROMPT_JOBS" + echo "$OUT_TEST_PROMPT_JOBS" + echo "OUT_TEST_SEARCH_JOBS" + echo "$OUT_TEST_SEARCH_JOBS" + + echo "$OUT_ADMIN_ENGLISH_PROMPT_AI" | grep -q "If all of these points are met, we give a Green label." && echo 'โœ… Admin does see AI overview' || exit 1 + echo "$OUT_ADMIN_GERMAN_PROMPT_AI" | grep -q "If all of these points are met, we give a Green label." && echo 'โœ… Admin does see AI overview, after german question' || exit 1 + echo "$OUT_ADMIN_SEARCH_AI" | grep -q "overview.rst" && echo 'โœ… Admin does see AI overview, when using search' || exit 1 + echo "$OUT_TEST_ENGLISH_PROMPT_AI" | grep -q -v "If all of these points are met, we give a Green label." && echo 'โœ… Test user does not see AI overview' || exit 1 + echo "$OUT_TEST_GERMAN_PROMPT_AI" | grep -q -v "If all of these points are met, we give a Green label." && echo 'โœ…Test user does not see AI overview, after german question' || exit 1 + echo "$OUT_TEST_SEARCH_AI" | grep -q -v "overview.rst" && echo 'โœ…Test user does not see AI overview, when using search' || exit 1 + echo "$OUT_TEST_PROMPT_JOBS" | grep -q -v "background jobs" && echo 'โœ…Test user does not see BackgroundJobs docs' || exit 1 + echo "$OUT_TEST_SEARCH_JOBS" | grep -q -v "backgroundjobs.md" && echo 'โœ…Test user does not see BackgroundJobs docs' || exit 1 + + - name: Remove some files using occ + if: ${{ matrix.file-deletion-method == 'occ'}} + run: | + ./occ files:delete admin/files/admin_manual + + - name: Remove some files using os + if: ${{ matrix.file-deletion-method == 'os'}} + run: | + cd data/admin/files + rm -rf ./admin_manual # this + files:scan sadly doesn't work, currently see https://github.com/nextcloud/context_chat/issues/153 + cd ../../.. + ./occ files:scan admin # We run the scan again to check if the files are successfully removed from the vectordb + + - name: Share some files + run: | + OC_PASS=test ./occ user:add --password-from-env -- test + # share developer_manual to test user + curl -X POST -u 'admin:password' -H 'OCS-APIRequest: true' 'http://localhost:8080/ocs/v2.php/apps/files_sharing/api/v1/shares?shareType=0&shareWith=test&path=developer_manual' + # list files to trigger mounting of shares + curl -X PROPFIND -H "Depth: 0" -u 'test:test' 'http://localhost:8080/remote.php/webdav/' + + - name: Run indexer cron + run: | + for i in {1..100}; do + php cron.php & # Starting with stable31 we can use -v here for better visibility + wait + ./occ context_chat:stats + done + + - name: Run the prompts again + run: | + ./occ background-job:worker --stop_after=2m 'OC\TaskProcessing\SynchronousBackgroundJob' & + ./occ background-job:worker --stop_after=2m 'OC\TaskProcessing\SynchronousBackgroundJob' & + + set +e + + # Check for user admin: Shouldn't be there + OUT_ADMIN_ENGLISH_PROMPT_AI=$(./occ context_chat:prompt admin "Which factors are taken into account for the Ethical AI Rating?") + OUT_ADMIN_GERMAN_PROMPT_AI=$(./occ context_chat:prompt admin "Welche Faktoren beeinflussen das Ethical AI Rating?") + OUT_ADMIN_SEARCH_AI=$(./occ context_chat:search admin "Welche Faktoren beeinflussen das Ethical AI Rating?") + + # Check for user test: Shouldn't be there + OUT_TEST_ENGLISH_PROMPT_AI=$(./occ context_chat:prompt test "Which factors are taken into account for the Ethical AI Rating?") + OUT_TEST_GERMAN_PROMPT_AI=$(./occ context_chat:prompt test "Welche Faktoren beeinflussen das Ethical AI Rating?") + OUT_TEST_SEARCH_AI=$(./occ context_chat:search test "Welche Faktoren beeinflussen das Ethical AI Rating?") + + # Check for user test: Should be there, because it was shared + OUT_TEST_PROMPT_JOBS=$(./occ context_chat:prompt test "How do I set my TimedJob to be time insensitive?") + OUT_TEST_SEARCH_JOBS=$(./occ context_chat:search test "How do I set my TimedJob to be time insensitive?") + + echo "OUT_ADMIN_ENGLISH_PROMPT_AI" + echo "$OUT_ADMIN_ENGLISH_PROMPT_AI" + echo "OUT_ADMIN_GERMAN_PROMPT_AI" + echo "$OUT_ADMIN_GERMAN_PROMPT_AI" + echo "OUT_ADMIN_SEARCH_AI" + echo "$OUT_ADMIN_SEARCH_AI" + echo "OUT_TEST_ENGLISH_PROMPT_AI" + echo "$OUT_TEST_ENGLISH_PROMPT_AI" + echo "OUT_TEST_GERMAN_PROMPT_AI" + echo "$OUT_TEST_GERMAN_PROMPT_AI" + echo "OUT_TEST_SEARCH_AI" + echo "$OUT_TEST_SEARCH_AI" + echo "OUT_TEST_PROMPT_JOBS" + echo "$OUT_TEST_PROMPT_JOBS" + echo "OUT_TEST_SEARCH_JOBS" + echo "$OUT_TEST_SEARCH_JOBS" + + echo "$OUT_ADMIN_ENGLISH_PROMPT_AI" | grep -q -v "If all of these points are met, we give a Green label." && echo 'โœ… Admin does not see AI overview anymore' || exit 1 + echo "$OUT_ADMIN_GERMAN_PROMPT_AI" | grep -q -v "If all of these points are met, we give a Green label." && echo 'โœ… Admin does not see AI overview anymore, after german question' || exit 1 + echo "$OUT_ADMIN_SEARCH_AI" | grep -q -v "overview.rst" && echo 'โœ… Admin does not see AI overview anymore, when using search' || exit 1 + echo "$OUT_TEST_ENGLISH_PROMPT_AI" | grep -q -v "If all of these points are met, we give a Green label." && echo 'โœ… Test user does not see AI overview' || exit 1 + echo "$OUT_TEST_GERMAN_PROMPT_AI" | grep -q -v "If all of these points are met, we give a Green label." && echo 'โœ…Test user does not see AI overview, after german question' || exit 1 + echo "$OUT_TEST_SEARCH_AI" | grep -q -v "overview.rst" && echo 'โœ…Test user does not see AI overview, when using search' || exit 1 + echo "$OUT_TEST_PROMPT_JOBS" | grep -q "background jobs" && echo 'โœ…Test user does see BackgroundJobs docs' || exit 1 + echo "$OUT_TEST_SEARCH_JOBS" | grep -q "backgroundjobs.md" && echo 'โœ…Test user does see BackgroundJobs docs' || exit 1 - name: Show nextcloud logs if: always() @@ -757,3 +1045,18 @@ jobs: cat context_chat_backend/em_backend_logs || echo "No embedding server logs" echo '--------------------------------------------------' tail -v -n +1 context_chat_backend/persistent_storage/logs/* || echo "No logs in logs directory" + + summary: + permissions: + contents: none + runs-on: ubuntu-latest-low + needs: [scan-test, cron-test, listener-test] + + if: always() + + # This is the summary, we just avoid to rename it so that branch protection rules still match + name: integration-test + + steps: + - name: Summary status + run: if ${{ needs.scan-test.rest != 'success' && needs.cron-test.result != 'success' && needs.listener-test.result != 'success' }}; then exit 1; fi \ No newline at end of file diff --git a/appinfo/info.xml b/appinfo/info.xml index 543cb85d..dc131af0 100644 --- a/appinfo/info.xml +++ b/appinfo/info.xml @@ -45,6 +45,7 @@ Refer to the [Context Chat Backend's readme](https://github.com/nextcloud/contex OCA\ContextChat\BackgroundJobs\SchedulerJob OCA\ContextChat\BackgroundJobs\FileSystemListenerJob + OCA\ContextChat\BackgroundJobs\ActionJob OCA\ContextChat\BackgroundJobs\RotateLogsJob diff --git a/lib/BackgroundJobs/ActionJob.php b/lib/BackgroundJobs/ActionJob.php index e3c2393d..06685e9e 100644 --- a/lib/BackgroundJobs/ActionJob.php +++ b/lib/BackgroundJobs/ActionJob.php @@ -17,9 +17,11 @@ use OCP\App\IAppManager; use OCP\AppFramework\Utility\ITimeFactory; use OCP\BackgroundJob\IJobList; -use OCP\BackgroundJob\QueuedJob; +use OCP\BackgroundJob\TimedJob; +use OCP\DB\Exception; +use OCP\IConfig; -class ActionJob extends QueuedJob { +class ActionJob extends TimedJob { private const BATCH_SIZE = 1000; public function __construct( @@ -30,8 +32,15 @@ public function __construct( private Logger $logger, private DiagnosticService $diagnosticService, private IAppManager $appManager, + private IConfig $config, ) { parent::__construct($timeFactory); + $this->setAllowParallelRuns(false); + $this->setInterval($this->getJobInterval()); + } + + private function getJobInterval(): int { + return intval($this->config->getAppValue('context_chat', 'action_job_interval', (string)(5 * 60))); // 5 minutes } protected function run($argument): void { @@ -40,15 +49,20 @@ protected function run($argument): void { return; } - $this->diagnosticService->sendJobStart(static::class, $this->getId()); - $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); - $entities = $this->actionMapper->getFromQueue(static::BATCH_SIZE); + try { + $this->diagnosticService->sendJobStart(static::class, $this->getId()); + $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); + try { + $entities = $this->actionMapper->getFromQueue(static::BATCH_SIZE); + } catch (Exception $e) { + $this->logger->warning('Error fetching actions in action Job : ' . $e->getMessage(), ['exception' => $e]); + return; + } - if (empty($entities)) { - return; - } + if (empty($entities)) { + return; + } - try { foreach ($entities as $entity) { $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); @@ -118,13 +132,10 @@ protected function run($argument): void { } } } catch (\Throwable $e) { - // schedule in 5mins - $this->jobList->scheduleAfter(static::class, $this->time->getTime() + 5 * 60); + $this->logger->warning('Error in action Job : ' . $e->getMessage(), ['exception' => $e]); throw $e; + } finally { + $this->diagnosticService->sendJobEnd(static::class, $this->getId()); } - - // schedule in 5mins - $this->jobList->scheduleAfter(static::class, $this->time->getTime() + 5 * 60); - $this->diagnosticService->sendJobEnd(static::class, $this->getId()); } } diff --git a/lib/BackgroundJobs/FileSystemListenerJob.php b/lib/BackgroundJobs/FileSystemListenerJob.php index f4ac31cf..eb7d8e47 100644 --- a/lib/BackgroundJobs/FileSystemListenerJob.php +++ b/lib/BackgroundJobs/FileSystemListenerJob.php @@ -16,9 +16,10 @@ use OCA\ContextChat\Type\FsEventType; use OCP\App\IAppManager; use OCP\AppFramework\Utility\ITimeFactory; -use OCP\BackgroundJob\IJobList; use OCP\BackgroundJob\TimedJob; +use OCP\DB\Exception; use OCP\Files\IRootFolder; +use OCP\IConfig; class FileSystemListenerJob extends TimedJob { private const BATCH_SIZE = 500; @@ -26,16 +27,20 @@ class FileSystemListenerJob extends TimedJob { public function __construct( ITimeFactory $timeFactory, private FsEventMapper $fsEventMapper, - private IJobList $jobList, private Logger $logger, private DiagnosticService $diagnosticService, private IAppManager $appManager, private FsEventService $fsEventService, private IRootFolder $rootFolder, + private IConfig $config, ) { parent::__construct($timeFactory); $this->allowParallelRuns = false; - $this->setInterval(5 * 60); // 5 minutes + $this->setInterval($this->getJobInterval()); + } + + private function getJobInterval(): int { + return intval($this->config->getAppValue('context_chat', 'fs_listener_job_interval', (string)(5 * 60))); // 5 minutes } protected function run($argument): void { @@ -44,45 +49,60 @@ protected function run($argument): void { return; } - $this->diagnosticService->sendJobStart(static::class, $this->getId()); - $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); - $fsEvents = $this->fsEventMapper->getFromQueue(static::BATCH_SIZE); - - if (empty($fsEvents)) { - return; - } - - foreach ($fsEvents as $fsEvent) { + try { + $this->diagnosticService->sendJobStart(static::class, $this->getId()); $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); - try { - $node = current($this->rootFolder->getUserFolder($fsEvent->getUserId())->getById($fsEvent->getNodeId())); - } catch (\Exception $e) { - $this->logger->warning('Error retrieving node for fs event "' . $fsEvent->getType() . '": ' . $e->getMessage(), ['exception' => $e]); - $node = false; + $fsEvents = $this->fsEventMapper->getFromQueue(static::BATCH_SIZE); + } catch (Exception $e) { + $this->logger->warning('Error fetching fs events: ' . $e->getMessage(), ['exception' => $e]); + return; } - if ($node === false) { - $this->logger->warning('Node with ID ' . $fsEvent->getNodeId() . ' not found for fs event "' . $fsEvent->getType() . '"'); - $this->fsEventMapper->delete($fsEvent); - continue; + + if (empty($fsEvents)) { + return; } - try { - switch ($fsEvent->getTypeObject()) { - case FsEventType::CREATE: - $this->fsEventService->onInsert($node); - break; - case FsEventType::ACCESS_UPDATE_DECL: - $this->fsEventService->onAccessUpdateDecl($node); - break; - } + foreach ($fsEvents as $fsEvent) { $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); - $this->fsEventMapper->delete($fsEvent); - } catch (\RuntimeException $e) { - $this->logger->warning('Error handling fs event "' . $fsEvent->getType() . '": ' . $e->getMessage(), ['exception' => $e]); + + try { + $node = current($this->rootFolder->getUserFolder($fsEvent->getUserId())->getById($fsEvent->getNodeId())); + } catch (\Exception $e) { + $this->logger->warning('Error retrieving node for fs event "' . $fsEvent->getType() . '": ' . $e->getMessage(), ['exception' => $e]); + $node = false; + } + if ($node === false) { + $this->logger->warning('Node with ID ' . $fsEvent->getNodeId() . ' not found for fs event "' . $fsEvent->getType() . '"'); + try { + $this->fsEventMapper->delete($fsEvent); + } catch (Exception $e) { + $this->logger->warning('Error deleting fs event "' . $fsEvent->getType() . '": ' . $e->getMessage(), ['exception' => $e]); + } + continue; + } + + try { + switch ($fsEvent->getTypeObject()) { + case FsEventType::CREATE: + $this->fsEventService->onInsert($node); + break; + case FsEventType::ACCESS_UPDATE_DECL: + $this->fsEventService->onAccessUpdateDecl($node); + break; + } + $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); + } catch (\Throwable $e) { + $this->logger->warning('Error handling fs event "' . $fsEvent->getType() . '": ' . $e->getMessage(), ['exception' => $e]); + } + try { + $this->fsEventMapper->delete($fsEvent); + } catch (Exception $e) { + $this->logger->warning('Error deleting fs event "' . $fsEvent->getType() . '": ' . $e->getMessage(), ['exception' => $e]); + } } + } finally { + $this->diagnosticService->sendJobEnd(static::class, $this->getId()); } - - $this->diagnosticService->sendJobEnd(static::class, $this->getId()); } } diff --git a/lib/BackgroundJobs/IndexerJob.php b/lib/BackgroundJobs/IndexerJob.php index d439ba1e..f93dd226 100644 --- a/lib/BackgroundJobs/IndexerJob.php +++ b/lib/BackgroundJobs/IndexerJob.php @@ -103,31 +103,31 @@ public function run($argument): void { return; } - $this->diagnosticService->sendJobStart(static::class, $this->getId()); - $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); + try { + $this->diagnosticService->sendJobStart(static::class, $this->getId()); + $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); - // Setup Filesystem for a users that can access this mount - $mounts = array_values(array_filter($this->userMountCache->getMountsForStorageId($this->storageId), function (ICachedMountInfo $mount) { - return $mount->getRootId() === $this->rootId; - })); + // Setup Filesystem for a users that can access this mount + $mounts = array_values(array_filter($this->userMountCache->getMountsForStorageId($this->storageId), function (ICachedMountInfo $mount) { + return $mount->getRootId() === $this->rootId; + })); - if (count($mounts) > 0) { - \OC_Util::setupFS($mounts[0]->getUser()->getUID()); - } + if (count($mounts) > 0) { + \OC_Util::setupFS($mounts[0]->getUser()->getUID()); + } - try { - $this->logger->debug('[IndexerJob] Running indexing', ['storageId' => $this->storageId, 'rootId' => $this->rootId]); - $this->index($files); - } catch (\RuntimeException $e) { - $this->logger->warning('[IndexerJob] Temporary problem with indexing', ['exception' => $e, 'storageId' => $this->storageId, 'rootId' => $this->rootId]); - } catch (\ErrorException $e) { - $this->logger->warning('[IndexerJob] Problem with indexing', ['exception' => $e, 'storageId' => $this->storageId, 'rootId' => $this->rootId]); - $this->logger->info('[IndexerJob] Removing ' . static::class . ' with argument ' . var_export($argument, true) . 'from oc_jobs'); - $this->jobList->remove(static::class, $argument); - throw $e; - } + try { + $this->logger->debug('[IndexerJob] Running indexing', ['storageId' => $this->storageId, 'rootId' => $this->rootId]); + $this->index($files); + } catch (\RuntimeException $e) { + $this->logger->warning('[IndexerJob] Temporary problem with indexing', ['exception' => $e, 'storageId' => $this->storageId, 'rootId' => $this->rootId]); + } catch (\ErrorException $e) { + $this->logger->warning('[IndexerJob] Problem with indexing', ['exception' => $e, 'storageId' => $this->storageId, 'rootId' => $this->rootId]); + $this->logger->info('[IndexerJob] Removing ' . static::class . ' with argument ' . var_export($argument, true) . 'from oc_jobs'); + $this->jobList->remove(static::class, $argument); + throw $e; + } - try { // If there is at least one file left in the queue, reschedule this job $files = $this->queue->getFromQueue($this->storageId, $this->rootId, 1); $indexerJobCount = $this->getJobCount(IndexerJob::class); @@ -138,13 +138,14 @@ public function run($argument): void { $this->setInitialIndexCompletion(); } elseif (count($files) === 0) { $this->logger->debug('[IndexerJob] No files left in queue, but we keep the job around to wait for potential StorageCrawlJob instances to finish'); - } } catch (Exception $e) { $this->logger->error('[IndexerJob] Cannot retrieve items from queue', ['exception' => $e, 'storageId' => $this->storageId, 'rootId' => $this->rootId]); - return; + } catch (\Throwable $e) { + $this->logger->error('[IndexerJob] Failure during job run', ['exception' => $e, 'storageId' => $this->storageId, 'rootId' => $this->rootId]); + } finally { + $this->diagnosticService->sendJobEnd(static::class, $this->getId()); } - $this->diagnosticService->sendJobEnd(static::class, $this->getId()); } protected function getBatchSize(): int { diff --git a/lib/BackgroundJobs/StorageCrawlJob.php b/lib/BackgroundJobs/StorageCrawlJob.php index 617c40c1..a10577a9 100644 --- a/lib/BackgroundJobs/StorageCrawlJob.php +++ b/lib/BackgroundJobs/StorageCrawlJob.php @@ -51,45 +51,48 @@ protected function run($argument): void { // Remove current iteration $this->jobList->remove(self::class, $argument); - $this->diagnosticService->sendJobStart(static::class, $this->getId()); - $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); - - $i = 0; - foreach ($this->storageService->getFilesInMount($storageId, $overrideRoot ?? $rootId, $lastFileId, self::BATCH_SIZE) as $fileId) { - $queueFile = new QueueFile(); - $queueFile->setStorageId($storageId); - $queueFile->setRootId($rootId); - $queueFile->setFileId($fileId); - $queueFile->setUpdate(false); + try { + $this->diagnosticService->sendJobStart(static::class, $this->getId()); $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); - try { - $this->queue->insertIntoQueue($queueFile); - } catch (Exception $e) { - $this->logger->error('[StorageCrawlJob] Failed to add file to queue', [ - 'fileId' => $fileId, - 'exception' => $e, + + $i = 0; + foreach ($this->storageService->getFilesInMount($storageId, $overrideRoot ?? $rootId, $lastFileId, self::BATCH_SIZE) as $fileId) { + $queueFile = new QueueFile(); + $queueFile->setStorageId($storageId); + $queueFile->setRootId($rootId); + $queueFile->setFileId($fileId); + $queueFile->setUpdate(false); + $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); + try { + $this->queue->insertIntoQueue($queueFile); + } catch (Exception $e) { + $this->logger->error('[StorageCrawlJob] Failed to add file to queue', [ + 'fileId' => $fileId, + 'exception' => $e, + 'storage_id' => $storageId, + 'root_id' => $rootId, + 'override_root' => $overrideRoot, + 'last_file_id' => $lastFileId + ]); + } + $i++; + } + + if ($i > 0) { + // Schedule next iteration after 5 minutes + $this->jobList->scheduleAfter(self::class, $this->time->getTime() + $this->getJobInterval(), [ 'storage_id' => $storageId, 'root_id' => $rootId, 'override_root' => $overrideRoot, - 'last_file_id' => $lastFileId + 'last_file_id' => $queueFile->getFileId(), ]); - } - $i++; - } - if ($i > 0) { - // Schedule next iteration after 5 minutes - $this->jobList->scheduleAfter(self::class, $this->time->getTime() + $this->getJobInterval(), [ - 'storage_id' => $storageId, - 'root_id' => $rootId, - 'override_root' => $overrideRoot, - 'last_file_id' => $queueFile->getFileId(), - ]); - - // the last job to set this value will win - $this->appConfig->setValueInt(Application::APP_ID, 'last_indexed_file_id', $queueFile->getFileId()); + // the last job to set this value will win + $this->appConfig->setValueInt(Application::APP_ID, 'last_indexed_file_id', $queueFile->getFileId()); + } + } finally { + $this->diagnosticService->sendJobEnd(static::class, $this->getId()); } - $this->diagnosticService->sendJobEnd(static::class, $this->getId()); } protected function getJobInterval(): int { diff --git a/lib/BackgroundJobs/SubmitContentJob.php b/lib/BackgroundJobs/SubmitContentJob.php index 3dc86491..7f0570f6 100644 --- a/lib/BackgroundJobs/SubmitContentJob.php +++ b/lib/BackgroundJobs/SubmitContentJob.php @@ -14,13 +14,16 @@ use OCA\ContextChat\Db\QueueContentItemMapper; use OCA\ContextChat\Exceptions\RetryIndexException; use OCA\ContextChat\Logger; +use OCA\ContextChat\Service\DiagnosticService; use OCA\ContextChat\Service\LangRopeService; use OCA\ContextChat\Service\ProviderConfigService; use OCA\ContextChat\Type\Source; +use OCP\App\IAppManager; use OCP\AppFramework\Services\IAppConfig; use OCP\AppFramework\Utility\ITimeFactory; use OCP\BackgroundJob\IJobList; use OCP\BackgroundJob\QueuedJob; +use OCP\DB\Exception; class SubmitContentJob extends QueuedJob { private const BATCH_SIZE = 20; @@ -32,6 +35,8 @@ public function __construct( private IJobList $jobList, private Logger $logger, private IAppConfig $appConfig, + private DiagnosticService $diagnosticService, + private IAppManager $appManager, ) { parent::__construct($timeFactory); } @@ -41,63 +46,85 @@ public function __construct( * @return void */ protected function run($argument): void { - $entities = $this->mapper->getFromQueue(static::BATCH_SIZE); - $maxSize = $this->appConfig->getAppValueInt('indexing_max_size', Application::CC_MAX_SIZE); - - if (empty($entities)) { + if (!$this->appManager->isInstalled('app_api')) { + $this->logger->warning('SubmitContentJob is skipped as app_api is disabled'); return; } - $sources = array_map(function (QueueContentItem $item) use ($maxSize) { - $contentSize = mb_strlen($item->getContent(), '8bit'); - if ($contentSize > $maxSize) { - $this->logger->warning('[SubmitContentJob] Content too large to index', [ - 'contentSize' => $contentSize, - 'maxSize' => $maxSize, - 'itemId' => $item->getItemId(), - 'providerId' => $item->getProviderId(), - 'appId' => $item->getAppId(), - ]); - return null; + try { + + $this->diagnosticService->sendJobStart(static::class, $this->getId()); + $this->diagnosticService->sendHeartbeat(static::class, $this->getId()); + + try { + $entities = $this->mapper->getFromQueue(static::BATCH_SIZE); + } catch (Exception $e) { + $this->logger->warning('Error fetching queue content items in SubmitContentJob : ' . $e->getMessage(), ['exception' => $e]); + return; } - $providerKey = ProviderConfigService::getConfigKey($item->getAppId(), $item->getProviderId()); - $sourceId = ProviderConfigService::getSourceId($item->getItemId(), $providerKey); - return new Source( - explode(',', $item->getUsers()), - $sourceId, - $item->getTitle(), - $item->getContent(), - $item->getLastModified()->getTimeStamp(), - $item->getDocumentType(), - $providerKey, - ); - }, $entities); - $sources = array_filter($sources); + $maxSize = $this->appConfig->getAppValueInt('indexing_max_size', Application::CC_MAX_SIZE); - try { - $loadSourcesResult = $this->service->indexSources($sources); - $this->logger->info('[SubmitContentJob] Indexed sources for providers', [ - 'count' => count($loadSourcesResult['loaded_sources']), - 'loaded_sources' => $loadSourcesResult['loaded_sources'], - 'sources_to_retry' => $loadSourcesResult['sources_to_retry'], - ]); - } catch (RetryIndexException $e) { - $this->logger->debug('[SubmitContentJob] At least one source is already being processed from another request, trying again soon', ['exception' => $e]); - // schedule in 5mins - $this->jobList->scheduleAfter(static::class, $this->time->getTime() + 5 * 60); - return; - } + if (empty($entities)) { + return; + } + + $sources = array_map(function (QueueContentItem $item) use ($maxSize) { + $contentSize = mb_strlen($item->getContent(), '8bit'); + if ($contentSize > $maxSize) { + $this->logger->warning('[SubmitContentJob] Content too large to index', [ + 'contentSize' => $contentSize, + 'maxSize' => $maxSize, + 'itemId' => $item->getItemId(), + 'providerId' => $item->getProviderId(), + 'appId' => $item->getAppId(), + ]); + return null; + } + + $providerKey = ProviderConfigService::getConfigKey($item->getAppId(), $item->getProviderId()); + $sourceId = ProviderConfigService::getSourceId($item->getItemId(), $providerKey); + return new Source( + explode(',', $item->getUsers()), + $sourceId, + $item->getTitle(), + $item->getContent(), + $item->getLastModified()->getTimeStamp(), + $item->getDocumentType(), + $providerKey, + ); + }, $entities); + $sources = array_filter($sources); - foreach ($entities as $entity) { - $providerKey = ProviderConfigService::getConfigKey($entity->getAppId(), $entity->getProviderId()); - $sourceId = ProviderConfigService::getSourceId($entity->getItemId(), $providerKey); - if (!in_array($sourceId, $loadSourcesResult['sources_to_retry'])) { - $this->mapper->removeFromQueue($entity); + try { + $loadSourcesResult = $this->service->indexSources($sources); + $this->logger->info('[SubmitContentJob] Indexed sources for providers', [ + 'count' => count($loadSourcesResult['loaded_sources']), + 'loaded_sources' => $loadSourcesResult['loaded_sources'], + 'sources_to_retry' => $loadSourcesResult['sources_to_retry'], + ]); + } catch (RetryIndexException $e) { + $this->logger->debug('[SubmitContentJob] At least one source is already being processed from another request, trying again soon', ['exception' => $e]); + return; } - } - // schedule in 5mins - $this->jobList->scheduleAfter(static::class, $this->time->getTime() + 5 * 60); + foreach ($entities as $entity) { + $providerKey = ProviderConfigService::getConfigKey($entity->getAppId(), $entity->getProviderId()); + $sourceId = ProviderConfigService::getSourceId($entity->getItemId(), $providerKey); + if (!in_array($sourceId, $loadSourcesResult['sources_to_retry'], true)) { + try { + $this->mapper->removeFromQueue($entity); + } catch (Exception $e) { + $this->logger->error('[SubmitContentJob] Failed to remove item from queue', ['exception' => $e]); + } + } + } + } catch (\Throwable $e) { + $this->logger->error('[SubmitContentJob] Error in SubmitContentJob : ' . $e->getMessage(), ['exception' => $e]); + } finally { + $this->diagnosticService->sendJobEnd(static::class, $this->getId()); + // schedule in 5mins + $this->jobList->scheduleAfter(static::class, $this->time->getTime() + 5 * 60); + } } } diff --git a/lib/Db/FsEvent.php b/lib/Db/FsEvent.php index 7cfdb536..ff2bad20 100644 --- a/lib/Db/FsEvent.php +++ b/lib/Db/FsEvent.php @@ -35,10 +35,9 @@ class FsEvent extends Entity { public function __construct() { // add types in constructor - $this->addType('id', Types::BIGINT); $this->addType('type', Types::STRING); $this->addType('userId', Types::STRING); - $this->addType('nodeId', Types::BIGINT); + $this->addType('nodeId', 'integer'); // stable30 does not support Types::BIGINT here } /** diff --git a/lib/Service/ActionScheduler.php b/lib/Service/ActionScheduler.php index 308a7360..c99cafdb 100644 --- a/lib/Service/ActionScheduler.php +++ b/lib/Service/ActionScheduler.php @@ -9,7 +9,6 @@ namespace OCA\ContextChat\Service; -use OCA\ContextChat\BackgroundJobs\ActionJob; use OCA\ContextChat\Db\QueueAction; use OCA\ContextChat\Db\QueueActionMapper; use OCA\ContextChat\Logger; @@ -40,9 +39,7 @@ private function scheduleAction(string $type, string $payload): void { // do not catch DB exceptions $this->actionMapper->insertIntoQueue($item); - if (!$this->jobList->has(ActionJob::class, null)) { - $this->jobList->add(ActionJob::class, null); - } + // no need to schedule job, it's a timed job } /** diff --git a/lib/Service/FsEventScheduler.php b/lib/Service/FsEventScheduler.php index 0b001526..683240d1 100644 --- a/lib/Service/FsEventScheduler.php +++ b/lib/Service/FsEventScheduler.php @@ -46,7 +46,7 @@ private function getOwnerIdForNode(Node $node): string { private function scheduleEvent(FsEventType $type, string $userId, int $nodeId): void { $item = new FsEvent(); $item->setUserId($userId); - $item->setType($type); + $item->setType($type->value); $item->setNodeId($nodeId); // do not catch DB exceptions diff --git a/lib/Service/FsEventService.php b/lib/Service/FsEventService.php index 0863b514..d4c7653d 100644 --- a/lib/Service/FsEventService.php +++ b/lib/Service/FsEventService.php @@ -82,7 +82,7 @@ public function onDelete(Node $node, bool $recurse = true): void { } try { - $fileRef = ProviderConfigService::getSourceId($node->getId()); + $fileRef = ProviderConfigService::getSourceId($file->getId()); $this->actionService->deleteSources($fileRef); } catch (InvalidPathException|NotFoundException $e) { $this->logger->warning($e->getMessage(), ['exception' => $e]); @@ -107,7 +107,7 @@ public function onInsert(Node $node, bool $recurse = true, bool $update = false) if (!$this->allowedMimeType($file)) { continue; } - if (!$this->allowedPath($node)) { + if (!$this->allowedPath($file)) { continue; } diff --git a/lib/Service/LangRopeService.php b/lib/Service/LangRopeService.php index f4d8da55..31851d5f 100644 --- a/lib/Service/LangRopeService.php +++ b/lib/Service/LangRopeService.php @@ -274,7 +274,7 @@ public function updateAccessDeclarative(array $userIds, string $sourceId): void /** * @param Source[] $sources * @return array{loaded_sources: array, sources_to_retry: array} - * @throws RuntimeException + * @throws RuntimeException|RetryIndexException */ public function indexSources(array $sources): array { if (count($sources) === 0) {