Handle files index via async job (#9)

* #54 feat: handle files index via async job

* chore: add annotations

src/Index/CollectionHandler.php

@@ -10,9 +10,11 @@
 
 use BEdita\Chatlas\Client\ChatlasClient;
 use BEdita\Core\Filesystem\FilesystemRegistry;
+use BEdita\Core\Model\Entity\AsyncJob;
 use BEdita\Core\Model\Entity\ObjectEntity;
 use Cake\Http\Client\FormData;
 use Cake\Log\LogTrait;
+use Cake\ORM\Locator\LocatorAwareTrait;
 use Cake\Utility\Hash;
 use Laminas\Diactoros\UploadedFile;
 
@@ -21,6 +23,7 @@
  */
 class CollectionHandler
 {
+    use LocatorAwareTrait;
     use LogTrait;
 
     /**
@@ -152,7 +155,7 @@ protected function addDocument(ObjectEntity $collection, ObjectEntity $entity):
             return;
         }
         if ($entity->get('type') === 'files') {
-            $this->uploadDocument($collection, $entity);
+            $this->uploadDocumentJob($collection, $entity);
 
             return;
         }
@@ -175,7 +178,7 @@ protected function addDocument(ObjectEntity $collection, ObjectEntity $entity):
      * @param \BEdita\Core\Model\Entity\ObjectEntity $entity Document entity
      * @return void
      */
-    protected function uploadDocument(ObjectEntity $collection, ObjectEntity $entity): void
+    public function uploadDocument(ObjectEntity $collection, ObjectEntity $entity): void
     {
         $form = new FormData();
         if (empty($entity->get('streams'))) {
@@ -208,6 +211,30 @@ protected function uploadDocument(ObjectEntity $collection, ObjectEntity $entity
             $form
         );
         $entity->set('index_updated', date('c'));
+        $entity->set('index_status', 'done');
+        $entity->getTable()->saveOrFail($entity, ['_skipAfterSave' => true]);
+    }
+
+    /**
+     * Create async job to upload file to index
+     *
+     * @param \BEdita\Core\Model\Entity\ObjectEntity $collection Collection entity
+     * @param \BEdita\Core\Model\Entity\ObjectEntity $entity File entity
+     * @return void
+     */
+    protected function uploadDocumentJob(ObjectEntity $collection, ObjectEntity $entity): void
+    {
+        $asyncJob = new AsyncJob([
+            'service' => 'BEdita/Chatlas.IndexFile',
+            'max_attempts' => 3,
+            'priority' => 5,
+            'payload' => [
+                'collection_id' => $collection->id,
+                'file_id' => $entity->id,
+            ],
+        ]);
+        $this->fetchTable('AsyncJobs')->saveOrFail($asyncJob);
+        $entity->set('index_status', 'processing');
         $entity->getTable()->saveOrFail($entity, ['_skipAfterSave' => true]);
     }
 

src/Job/Service/IndexFileService.php

@@ -0,0 +1,52 @@
+<?php
+declare(strict_types=1);
+
+/**
+ * Chatlas BEdita plugin
+ *
+ * Copyright 2023 Atlas Srl
+ */
+namespace BEdita\Chatlas\Job\Service;
+
+use BEdita\Chatlas\Index\CollectionHandler;
+use BEdita\Core\Job\JobService;
+use Cake\Log\LogTrait;
+use Cake\ORM\Locator\LocatorAwareTrait;
+use Throwable;
+
+class IndexFileService implements JobService
+{
+    use LocatorAwareTrait;
+    use LogTrait;
+
+    /**
+     * Run an async job using $payload input data and optional $options.
+     *
+     * It can return:
+     * - a boolean i.e. `true` on success, `false` on failure
+     * - an array with keys:
+     *   - 'success' (required) => `true` on success, `false` on failure
+     *   - 'messages' (optional) => array of messages
+     *
+     * @param array $payload Input data for running this job.
+     * @param array $options Options for running this job.
+     * @return bool
+     */
+    public function run(array $payload, array $options = []): bool
+    {
+        $handler = new CollectionHandler();
+        try {
+            /** @var \BEdita\Core\Model\Entity\ObjectEntity $collection */
+            $collection = $this->fetchTable('Collections')->get($payload['collection_id']);
+            /** @var \BEdita\Core\Model\Entity\ObjectEntity $file */
+            $file = $this->fetchTable('Files')->get($payload['file_id']);
+            $handler->uploadDocument($collection, $file);
+
+            return true;
+        } catch (Throwable $th) {
+            $this->log(sprintf('IndexFile async job error - %s', $th->getMessage()), 'error');
+        }
+
+        return false;
+    }
+}

tests/TestCase/Index/CollectionHandlerTest.php

@@ -6,6 +6,7 @@
 use BEdita\Chatlas\Index\CollectionHandler;
 use BEdita\Chatlas\Test\TestMockTrait;
 use BEdita\Core\Filesystem\FilesystemRegistry;
+use BEdita\Core\Model\Entity\AsyncJob;
 use BEdita\Core\Model\Entity\ObjectEntity;
 use BEdita\Core\Model\Entity\Stream;
 use Cake\TestSuite\TestCase;
@@ -69,14 +70,14 @@ public function testRemoveCollection(): void
     }
 
     /**
-     * Test `uploadDocument()` method.
+     * Test `uploadDocumentJob()` method.
      *
      * @return void
-     * @covers ::uploadDocument()
+     * @covers ::uploadDocumentJob()
      * @covers ::updateDocument()
      * @covers ::addDocument()
      */
-    public function testUploadEmptyStream(): void
+    public function testUploadDocumentJob(): void
     {
         $entity = $this->mockEntity('files', [
             'index_updated' => null,
@@ -85,8 +86,29 @@ public function testUploadEmptyStream(): void
         $entity->setNew(false);
         $handler = new CollectionHandler();
         $collection = new ObjectEntity();
+        $this->mockTable('AsyncJobs', new AsyncJob());
         $handler->updateDocument($collection, $entity);
         static::assertNull($entity->get('index_updated'));
+        static::assertEquals('processing', $entity->get('index_status'));
+    }
+
+    /**
+     * Test `uploadDocument()` method.
+     *
+     * @return void
+     * @covers ::uploadDocument()
+     */
+    public function testUploadEmptyStream(): void
+    {
+        $entity = $this->mockEntity('files', [
+            'index_updated' => null,
+            'status' => 'on',
+        ]);
+        $entity->setNew(false);
+        $handler = new CollectionHandler();
+        $collection = new ObjectEntity();
+        $handler->uploadDocument($collection, $entity);
+        static::assertNull($entity->get('index_updated'));
     }
 
     /**
@@ -118,7 +140,7 @@ public function testUploadStream(): void
 
         $handler = new CollectionHandler();
         $collection = new ObjectEntity();
-        $handler->updateDocument($collection, $entity);
+        $handler->uploadDocument($collection, $entity);
         static::assertNotEmpty($entity->get('index_updated'));
     }
 

tests/TestCase/Job/Service/IndexFileServiceTest.php

@@ -0,0 +1,73 @@
+<?php
+declare(strict_types=1);
+
+namespace BEdita\Chatlas\Test\TestCase\Job\Service;
+
+use BEdita\Chatlas\Job\Service\IndexFileService;
+use BEdita\Chatlas\Test\TestMockTrait;
+use BEdita\Core\Filesystem\FilesystemRegistry;
+use BEdita\Core\Model\Entity\ObjectEntity;
+use BEdita\Core\Model\Entity\Stream;
+use Cake\TestSuite\TestCase;
+
+/**
+ * @coversDefaultClass \BEdita\Chatlas\Job\Service\IndexFileServiceTest
+ */
+class IndexFileServiceTest extends TestCase
+{
+    use TestMockTrait;
+
+    /**
+     * Test `run()` method.
+     *
+     * @return void
+     * @covers ::run()
+     */
+    public function testRun(): void
+    {
+        FilesystemRegistry::dropAll();
+        FilesystemRegistry::setConfig([
+            'default' => [
+                'className' => 'BEdita/Core.Local',
+                'path' => './tests' . DS . 'uploads',
+            ],
+        ]);
+
+        $stream = new Stream();
+        $stream->set('uri', 'default://test.txt', ['guard' => false]);
+        $stream->set('file_size', 1, ['guard' => false]);
+        $file = $this->mockEntity('files', [
+            'index_updated' => null,
+            'status' => 'on',
+            'streams' => [$stream],
+        ]);
+        $this->mockClientResponse();
+        $this->mockTable('Collections', new ObjectEntity());
+        $this->mockTable('Files', $file);
+
+        $service = new IndexFileService();
+        $success = $service->run([
+            'collection_id' => 1,
+            'file_id' => 2,
+        ]);
+        static::assertTrue($success);
+        static::assertNotEmpty($file->get('index_updated'));
+        static::assertEquals('done', $file->get('index_status'));
+    }
+
+    /**
+     * Test `run()` method with failure.
+     *
+     * @return void
+     * @covers ::run()
+     */
+    public function testFailure(): void
+    {
+        $service = new IndexFileService();
+        $success = $service->run([
+            'collection_id' => 1,
+            'file_id' => 2,
+        ]);
+        static::assertFalse($success);
+    }
+}

tests/TestMockTrait.php

@@ -8,7 +8,9 @@
 use Cake\Core\Configure;
 use Cake\Http\Client\Adapter\Stream;
 use Cake\Http\Client\Response;
+use Cake\ORM\Entity;
 use Cake\ORM\Table;
+use Cake\ORM\TableRegistry;
 
 trait TestMockTrait
 {
@@ -78,4 +80,26 @@ protected function mockEntity(?string $type = null, array $data = [], array $ass
 
         return $mockEntity;
     }
+
+    /**
+     * Create mock table
+     *
+     * @param string $alias Table alias
+     * @param Entity $entity Entity
+     * @return void
+     */
+    protected function mockTable(string $alias, Entity $entity): void
+    {
+        $mockTable = $this->getMockBuilder(Table::class)
+            ->onlyMethods(['saveOrFail', 'get'])
+            ->getMock();
+        $mockTable->method('saveOrFail')
+            ->willReturn($entity);
+        $mockTable->method('get')
+            ->willReturn($entity);
+
+        $locator = TableRegistry::getTableLocator();
+        $locator->remove($alias);
+        $locator->set($alias, $mockTable);
+    }
 }