Add invalidateCacheTags to post_import queueworker (#4125)

modules/datastore/src/Plugin/QueueWorker/PostImportResourceProcessor.php

@@ -12,6 +12,7 @@
 use Drupal\datastore\Service\ResourceProcessorCollector;
 use Drupal\metastore\ResourceMapper;
 use Drupal\datastore\PostImportResult;
+use Drupal\metastore\Reference\ReferenceLookup;
 use Drupal\datastore\Service\PostImport;
 use Drupal\metastore\DataDictionary\DataDictionaryDiscoveryInterface;
 
@@ -66,6 +67,13 @@ class PostImportResourceProcessor extends QueueWorkerBase implements ContainerFa
    */
   protected $dataDictionaryDiscovery;
 
+  /**
+   * Reference lookup service.
+   *
+   * @var \Drupal\metastore\Reference\ReferenceLookup
+   */
+  protected $referenceLookup;
+
   /**
    * Build queue worker.
    *
@@ -87,6 +95,8 @@ class PostImportResourceProcessor extends QueueWorkerBase implements ContainerFa
    *   The post import service.
    * @param \Drupal\metastore\DataDictionary\DataDictionaryDiscoveryInterface $data_dictionary_discovery
    *   The data-dictionary discovery service.
+   * @param \Drupal\metastore\Reference\ReferenceLookup $referenceLookup
+   *   The reference lookup service.
    */
   public function __construct(
     array $configuration,
@@ -97,7 +107,8 @@ public function __construct(
     ResourceMapper $resource_mapper,
     ResourceProcessorCollector $processor_collector,
     PostImport $post_import,
-    DataDictionaryDiscoveryInterface $data_dictionary_discovery
+    DataDictionaryDiscoveryInterface $data_dictionary_discovery,
+    ReferenceLookup $referenceLookup
   ) {
     parent::__construct($configuration, $plugin_id, $plugin_definition);
     $this->logger = $logger_factory->get('datastore');
@@ -110,6 +121,7 @@ public function __construct(
     // duration of the time the queue is being processed.
     $timeout = (int) $plugin_definition['cron']['lease_time'];
     $alter_table_query_builder->setConnectionTimeout($timeout);
+    $this->referenceLookup = $referenceLookup;
   }
 
   /**
@@ -126,6 +138,7 @@ public static function create(ContainerInterface $container, array $configuratio
       $container->get('dkan.datastore.service.resource_processor_collector'),
       $container->get('dkan.datastore.service.post_import'),
       $container->get('dkan.metastore.data_dictionary_discovery'),
+      $container->get('dkan.metastore.reference_lookup'),
     );
   }
 
@@ -134,6 +147,13 @@ public static function create(ContainerInterface $container, array $configuratio
    */
   public function processItem($data) {
     $postImportResult = $this->postImportProcessItem($data);
+    if ($postImportResult->getPostImportStatus() === 'done') {
+      $this->invalidateCacheTags(DataResource::buildUniqueIdentifier(
+        $data->getIdentifier(),
+        $data->getVersion(),
+        DataResource::DEFAULT_SOURCE_PERSPECTIVE
+      ));
+    }
     // Store the results of the PostImportResult object.
     $postImportResult->storeResult();
   }
@@ -168,6 +188,7 @@ public function postImportProcessItem(DataResource $resource): PostImportResult
       else {
         array_map(fn ($processor) => $processor->process($resource), $processors);
         $postImportResult = $this->createPostImportResult('done', NULL, $resource);
+        $this->logger->notice('Post import job for resource @id completed.', ['@id' => (string) $resource->getIdentifier()]);
       }
     }
     catch (\Exception $e) {
@@ -178,6 +199,16 @@ public function postImportProcessItem(DataResource $resource): PostImportResult
     return $postImportResult;
   }
 
+  /**
+   * Invalidate all appropriate cache tags for this resource.
+   *
+   * @param mixed $resourceId
+   *   A resource ID.
+   */
+  protected function invalidateCacheTags($resourceId) {
+    $this->referenceLookup->invalidateReferencerCacheTags('distribution', $resourceId, 'downloadURL');
+  }
+
   /**
    * Create the PostImportResult object.
    *

modules/datastore/tests/src/Unit/Plugin/QueueWorker/PostImportResourceProcessorTest.php

@@ -19,7 +19,7 @@
 use Drupal\metastore\ResourceMapper;
 use Drupal\datastore\Service\PostImport;
 use Drupal\metastore\MetastoreService;
-
+use Drupal\metastore\Reference\ReferenceLookup;
 use MockChain\Chain;
 use MockChain\Options;
 use PHPUnit\Framework\TestCase;
@@ -74,7 +74,7 @@ public function testPostImportProcessItem() {
 
     // Ensure resources were processed.
     $notices = $container_chain->getStoredInput('notice');
-    $this->assertEmpty($notices);
+    $this->assertContains('Post import job for resource @id completed.', $notices);
     // Ensure no exceptions were thrown.
     $errors = $container_chain->getStoredInput('error');
     $this->assertEmpty($errors);
@@ -237,6 +237,7 @@ protected function getContainerChain() {
       ->add('dkan.metastore.resource_mapper', ResourceMapper::class)
       ->add('dkan.datastore.service.resource_processor_collector', ResourceProcessorCollector::class)
       ->add('dkan.datastore.service.post_import', PostImport::class)
+      ->add('dkan.metastore.reference_lookup', ReferenceLookup::class)
       ->index(0);
 
     $json = '{"identifier":"foo","title":"bar","data":{"fields":[]}}';
@@ -245,7 +246,7 @@ protected function getContainerChain() {
       ->add(Container::class, 'get', $options)
       ->add(LoggerChannelFactoryInterface::class, 'get', LoggerChannelInterface::class)
       ->add(LoggerChannelInterface::class, 'error', NULL, 'error')
-      ->add(LoggerChannelInterface::class, 'notice', NULL, 'notice')
+      ->add(LoggerChannelInterface::class, 'notice', '', 'notice')
       ->add(MetastoreService::class, 'get', new RootedJsonData($json))
       ->add(AlterTableQueryBuilderInterface::class, 'setConnectionTimeout', AlterTableQueryBuilderInterface::class)
       ->add(AlterTableQueryBuilderInterface::class, 'getQuery', AlterTableQueryInterface::class)

modules/datastore/tests/src/Unit/Service/ResourceProcessor/DictionaryEnforcerTest.php

@@ -19,7 +19,7 @@
 use Drupal\datastore\Service\PostImport;
 use Drupal\metastore\ResourceMapper;
 use Drupal\metastore\MetastoreService;
-
+use Drupal\metastore\Reference\ReferenceLookup;
 use MockChain\Chain;
 use MockChain\Options;
 use PHPUnit\Framework\TestCase;
@@ -191,6 +191,7 @@ protected function getContainerChain(int $resource_version) {
       ->add('dkan.datastore.service.resource_processor_collector', ResourceProcessorCollector::class)
       ->add('dkan.datastore.service.resource_processor.dictionary_enforcer', DictionaryEnforcer::class)
       ->add('dkan.datastore.service.post_import', PostImport::class)
+      ->add('dkan.metastore.reference_lookup', ReferenceLookup::class)
       ->index(0);
 
     $json = '{"identifier":"foo","title":"bar","data":{"fields":[]}}';