Add utilities module and functions for reindexing

lib/dpul_collections/repo.ex

@@ -2,4 +2,13 @@ defmodule DpulCollections.Repo do
   use Ecto.Repo,
     otp_app: :dpul_collections,
     adapter: Ecto.Adapters.Postgres
+
+  if Mix.env() in [:dev, :test] do
+    @spec truncate(Ecto.Schema.t()) :: :ok
+    def truncate(schema) do
+      table_name = schema.__schema__(:source)
+      query("TRUNCATE #{table_name}", [])
+      :ok
+    end
+  end
 end

lib/dpul_collections/utilities.ex

@@ -0,0 +1,36 @@
+defmodule DpulCollections.Utilities do
+  alias DpulCollections.IndexingPipeline
+  alias DpulCollections.IndexingPipeline.Figgy
+
+  def reindex_solr(cache_version) do
+    indexing_processor_marker =
+      IndexingPipeline.get_processor_marker!("figgy_indexer", cache_version)
+
+    IndexingPipeline.delete_processor_marker(indexing_processor_marker)
+    # When the consumer is stopped, it supervisor restarts it immediately.
+    # This resets the state and last queried marker.
+    GenServer.stop(Figgy.IndexingConsumer)
+  end
+
+  def reindex_transformation_cache(cache_version) do
+    transformation_processor_marker =
+      IndexingPipeline.get_processor_marker!("figgy_transformer", cache_version)
+
+    IndexingPipeline.delete_processor_marker(transformation_processor_marker)
+    GenServer.stop(Figgy.TransformationConsumer)
+  end
+
+  def reindex_hydration_cache(cache_version) do
+    hydration_processor_marker =
+      IndexingPipeline.get_processor_marker!("hydrator", cache_version)
+
+    IndexingPipeline.delete_processor_marker(hydration_processor_marker)
+    GenServer.stop(Figgy.HydrationConsumer)
+  end
+
+  def reindex_all(cache_version) do
+    reindex_hydration_cache(cache_version)
+    reindex_transformation_cache(cache_version)
+    reindex_solr(cache_version)
+  end
+end

test/dpul_collections/indexing_pipeline/integration/full_integration_test.exs

@@ -3,8 +3,7 @@ defmodule DpulCollections.IndexingPipeline.FiggyFullIntegrationTest do
 
   alias DpulCollections.Repo
   alias DpulCollections.IndexingPipeline.Figgy
-  alias DpulCollections.IndexingPipeline
-  alias DpulCollections.Solr
+  alias DpulCollections.{IndexingPipeline, Solr, Utilities}
 
   setup do
     Solr.delete_all()
@@ -28,10 +27,14 @@ defmodule DpulCollections.IndexingPipeline.FiggyFullIntegrationTest do
   end
 
   test "a full hydrator and transformer run" do
-    # Start the figgy producer
-    {:ok, indexer} = Figgy.IndexingConsumer.start_link(cache_version: 1, batch_size: 50)
-    {:ok, transformer} = Figgy.TransformationConsumer.start_link(cache_version: 1, batch_size: 50)
-    {:ok, hydrator} = Figgy.HydrationConsumer.start_link(cache_version: 1, batch_size: 50)
+    # Start the figgy pipeline in a way that mimics how it is started in test and prod
+    children = [
+      {Figgy.IndexingConsumer, cache_version: 1, batch_size: 50},
+      {Figgy.TransformationConsumer, cache_version: 1, batch_size: 50},
+      {Figgy.HydrationConsumer, cache_version: 1, batch_size: 50}
+    ]
+
+    Supervisor.start_link(children, strategy: :one_for_one, name: DpulCollections.TestSupervisor)
 
     task =
       Task.async(fn -> wait_for_index_completion() end)
@@ -57,8 +60,19 @@ defmodule DpulCollections.IndexingPipeline.FiggyFullIntegrationTest do
     assert transformer_processor_marker.cache_version == 1
     assert indexing_processor_marker.cache_version == 1
 
-    hydrator |> Broadway.stop(:normal)
-    transformer |> Broadway.stop(:normal)
-    indexer |> Broadway.stop(:normal)
+    # test reindexing
+    Repo.truncate(Figgy.TransformationCacheEntry)
+    Repo.truncate(Figgy.HydrationCacheEntry)
+    Solr.delete_all()
+    assert Solr.document_count() == 0
+    Utilities.reindex_all(1)
+
+    task =
+      Task.async(fn -> wait_for_index_completion() end)
+
+    Task.await(task, 15000)
+    assert Solr.document_count() == transformation_cache_entry_count
+
+    Supervisor.stop(DpulCollections.TestSupervisor, :normal)
   end
 end