Split processing of docs uploads and search index updating

Author: ericmj

lib/hexdocs.ex

@@ -1,32 +1,32 @@
 defmodule Hexdocs do
   @key_regex ~r"docs/(.*)-(.*).tar.gz$"
 
-  def process_object(key) do
-    key
-    |> build_message()
-    |> send_message()
+  def process_object(type, key) when type in [:upload, :search] do
+    publish_message(type, key)
   end
 
-  def process_all_objects() do
+  def process_all_objects(type) when type in [:upload, :search] do
     (Hexdocs.Store.list(:repo_bucket, "docs/") ++ Hexdocs.Store.list(:repo_bucket, "repos/"))
     |> Enum.shuffle()
-    |> batched_send()
-  end
-
-  defp build_message(key) do
-    %{
-      "Records" => [%{"eventName" => "ObjectCreated:Put", "s3" => %{"object" => %{"key" => key}}}]
-    }
+    |> batched_send(type)
   end
 
-  def batched_send(keys) do
+  def batched_send(keys, type) when type in [:upload, :search] do
     keys
     |> Stream.filter(&Regex.match?(@key_regex, &1))
-    |> Stream.map(&build_message/1)
+    |> Stream.map(&build_message(type, &1))
     |> Task.async_stream(&send_message/1, max_concurrency: 10, ordered: false, timeout: 60_000)
     |> Stream.run()
   end
 
+  defp build_message(:upload, key) do
+    %{"hexdocs:upload" => key}
+  end
+
+  defp build_message(:search, key) do
+    %{"hexdocs:search" => key}
+  end
+
   defp send_message(map) do
     queue = Application.fetch_env!(:hexdocs, :queue_id)
     message = Jason.encode!(map)
@@ -41,4 +41,22 @@ defmodule Hexdocs do
     |> Task.async_stream(&send_message/1, max_concurrency: 10, ordered: false, timeout: 60_000)
     |> Stream.run()
   end
+
+  def publish_upload_message(key) do
+    publish_message(:upload, key)
+  end
+
+  def publish_search_message(key) do
+    publish_message(:search, key)
+  end
+
+  def publish_message(:upload, key) do
+    %{"hexdocs:upload" => key}
+    |> send_message()
+  end
+
+  def publish_message(:search, key) do
+    %{"hexdocs:search" => key}
+    |> send_message()
+  end
 end

lib/hexdocs/bucket.ex

@@ -305,10 +305,6 @@ defmodule Hexdocs.Bucket do
     |> purge()
   end
 
-  defp purge_hexdocs_cache("hexpm", package, _versions, :unversioned) do
-    purge([docspage_unversioned_cdn_key("hexpm", package)])
-  end
-
   defp purge_hexdocs_cache(_repository, _package, _version, _publish_unversioned?) do
     :ok
   end

lib/hexdocs/hex_repo.ex

@@ -1,8 +1,9 @@
 defmodule Hexdocs.HexRepo do
   @callback get_names() :: {:ok, [binary()]} | {:error, term()}
 
-  @module Application.fetch_env!(:hexdocs, :hex_repo_impl)
+  @module Application.compile_env(:hexdocs, :hex_repo_impl)
 
+  @compile {:no_warn_undefined, [{@module, :get_names, 0}]}
   defdelegate get_names(), to: @module
 end
 

lib/hexdocs/queue.ex

@@ -77,76 +77,136 @@ defmodule Hexdocs.Queue do
     message
   end
 
-  @impl true
-  def handle_batch(_batcher, messages, _batch_info, _context) do
-    messages
+  def handle_message(%{data: %{"hexdocs:upload" => key}} = message) do
+    process_docs(key, :upload)
+    message
   end
 
-  defp handle_record(%{"eventName" => "ObjectCreated:" <> _, "s3" => s3}) do
+  def handle_message(%{data: %{"hexdocs:search" => key}} = message) do
+    process_docs(key, :search)
+    message
+  end
+
+  defp process_docs(key, type) do
     start = System.os_time(:millisecond)
-    key = s3["object"]["key"]
-    Logger.info("OBJECT CREATED #{key}")
+    event_name = if type == :upload, do: "hexdocs:upload", else: "hexdocs:search"
+    log_prefix = if type == :upload, do: "UPLOAD", else: "SEARCH INDEX"
+
+    Sentry.Context.set_extra_context(%{queue_event: event_name})
+    Logger.info("#{log_prefix} #{key}")
 
     case key_components(key) do
       {:ok, repository, package, version} ->
         Sentry.Context.set_extra_context(%{
-          queue_event: "ObjectCreated",
+          queue_event: event_name,
           repository: repository,
           package: package,
           version: version
         })
 
         body = Hexdocs.Store.get(:repo_bucket, key)
 
-        {version, all_versions} =
-          if package in @special_package_names do
-            version =
-              case Version.parse(version) do
-                {:ok, version} ->
-                  version
-
-                # main or MAJOR.MINOR
-                :error ->
-                  version
-              end
-
-            all_versions = Hexdocs.SourceRepo.versions!(Map.fetch!(@special_packages, package))
-            {version, all_versions}
-          else
-            version = Version.parse!(version)
-            all_versions = all_versions(repository, package)
-            {version, all_versions}
+        case type do
+          :upload ->
+            process_upload(key, repository, package, version, body, start)
+
+          :search ->
+            process_search(key, package, version, body, start)
+        end
+
+      :error ->
+        Logger.info("#{key}: skip")
+    end
+  end
+
+  defp process_upload(key, repository, package, version, body, start) do
+    {version, all_versions} =
+      if package in @special_package_names do
+        version =
+          case Version.parse(version) do
+            {:ok, version} ->
+              version
+
+            # main or MAJOR.MINOR
+            :error ->
+              version
           end
 
-        case Hexdocs.Tar.unpack(body, repository: repository, package: package, version: version) do
-          {:ok, files} ->
-            files = rewrite_files(files)
+        all_versions = Hexdocs.SourceRepo.versions!(Map.fetch!(@special_packages, package))
+        {version, all_versions}
+      else
+        version = Version.parse!(version)
+        all_versions = all_versions(repository, package)
+        {version, all_versions}
+      end
+
+    case Hexdocs.Tar.unpack(body, repository: repository, package: package, version: version) do
+      {:ok, files} ->
+        files = rewrite_files(files)
+
+        Hexdocs.Bucket.upload(
+          repository,
+          package,
+          version,
+          all_versions,
+          files
+        )
+
+        if Hexdocs.Utils.latest_version?(package, version, all_versions) do
+          update_index_sitemap(repository, key)
+          update_package_sitemap(repository, key, package, files)
+          update_package_names_csv(repository)
+        end
 
-            Hexdocs.Bucket.upload(
-              repository,
-              package,
-              version,
-              all_versions,
-              files
-            )
+        elapsed = System.os_time(:millisecond) - start
+        Logger.info("FINISHED UPLOADING DOCS #{key} #{elapsed}ms")
 
-            if Hexdocs.Utils.latest_version?(package, version, all_versions) do
-              update_index_sitemap(repository, key)
-              update_package_sitemap(repository, key, package, files)
-              update_package_names_csv(repository)
-            end
+      {:error, reason} ->
+        Logger.error("Failed unpack #{repository}/#{package} #{version}: #{reason}")
+    end
+  end
 
-            if repository == "hexpm" do
-              update_search_index(key, package, version, files)
-            end
+  defp process_search(key, package, version, body, start) do
+    version = Version.parse!(version)
 
-            elapsed = System.os_time(:millisecond) - start
-            Logger.info("FINISHED UPLOADING AND INDEXING DOCS #{key} #{elapsed}ms")
+    case Hexdocs.Tar.unpack(body, package: package, version: version) do
+      {:ok, files} ->
+        update_search_index(key, package, version, files)
 
-          {:error, reason} ->
-            Logger.error("Failed unpack #{repository}/#{package} #{version}: #{reason}")
+        elapsed = System.os_time(:millisecond) - start
+        Logger.info("FINISHED INDEXING DOCS #{key} #{elapsed}ms")
+
+      {:error, reason} ->
+        Logger.error("Failed unpack #{package} #{version}: #{reason}")
+    end
+  end
+
+  @impl true
+  def handle_batch(_batcher, messages, _batch_info, _context) do
+    messages
+  end
+
+  defp handle_record(%{"eventName" => "ObjectCreated:" <> _, "s3" => s3}) do
+    key = s3["object"]["key"]
+    Logger.info("OBJECT CREATED #{key}")
+
+    case key_components(key) do
+      {:ok, repository, _package, _version} ->
+        Sentry.Context.set_extra_context(%{
+          queue_event: "ObjectCreated",
+          repository: repository
+        })
+
+        # Publish upload message
+        publish_message(%{"hexdocs:upload" => key})
+
+        # Publish search message for hexpm repository
+        if repository == "hexpm" do
+          publish_message(%{"hexdocs:search" => key})
         end
 
+        Logger.info("PUBLISHED MESSAGES FOR #{key}")
+
       :error ->
         :skip
     end
@@ -281,6 +341,14 @@ defmodule Hexdocs.Queue do
     end
   end
 
+  defp publish_message(map) do
+    queue = Application.fetch_env!(:hexdocs, :queue_id)
+    message = Jason.encode!(map)
+
+    ExAws.SQS.send_message(queue, message)
+    |> ExAws.request!()
+  end
+
   @doc false
   def paths_for_sitemaps() do
     key_regex = ~r"docs/(.*)-(.*).tar.gz$"

lib/hexdocs/search/search.ex

@@ -42,8 +42,7 @@ defmodule Hexdocs.Search do
           json
 
         _ when is_binary(search_data_js) ->
-          Logger.error("Unexpected search_data format for #{package} #{version}")
-          nil
+          raise "Unexpected search_data format for #{package} #{version}"
 
         nil ->
           nil
@@ -55,12 +54,7 @@ defmodule Hexdocs.Search do
           :json.decode(search_data_json)
         catch
           _kind, reason ->
-            Logger.error(
-              "Failed to decode search data json for #{package} #{version}: " <>
-                inspect(reason)
-            )
-
-            nil
+            raise "Failed to decode search data json for #{package} #{version}: #{inspect(reason)}"
         end
       end
 
@@ -73,11 +67,7 @@ defmodule Hexdocs.Search do
         nil
 
       _ ->
-        Logger.error(
-          "Failed to extract search items and proglang from search data for #{package} #{version}"
-        )
-
-        nil
+        raise "Failed to extract search items and proglang from search data for #{package} #{version}"
     end
   end
 

lib/hexdocs/search/typesense.ex

@@ -33,17 +33,15 @@ defmodule Hexdocs.Search.Typesense do
               :ok
 
             %{"success" => false, "error" => error, "document" => document} ->
-              Logger.error(
-                "Failed to index search item for #{package} #{version} for document #{inspect(document)}: #{inspect(error)}"
-              )
+              raise "Failed to index search item for #{package} #{version} for document #{inspect(document)}: #{inspect(error)}"
           end
         end)
 
       {:ok, status, _resp_headers, _body} ->
-        Logger.error("Failed to index search items for #{package} #{version}: status=#{status}")
+        raise "Failed to index search items for #{package} #{version}: status=#{status}"
 
       {:error, reason} ->
-        Logger.error("Failed to index search items #{package} #{version}: #{inspect(reason)}")
+        raise "Failed to index search items #{package} #{version}: #{inspect(reason)}"
     end
   end
 
@@ -60,12 +58,10 @@ defmodule Hexdocs.Search.Typesense do
         :ok
 
       {:ok, status, _resp_headers, _body} ->
-        Logger.error("Failed to delete search items for #{package} #{version}: status=#{status}")
+        raise "Failed to delete search items for #{package} #{version}: status=#{status}"
 
       {:error, reason} ->
-        Logger.error(
-          "Failed to delete search items for #{package} #{version}: #{inspect(reason)}"
-        )
+        raise "Failed to delete search items for #{package} #{version}: #{inspect(reason)}"
     end
   end
 

mix.exs

@@ -46,7 +46,7 @@ defmodule Hexdocs.MixProject do
       {:jason, "~> 1.1"},
       {:logster, "~> 1.0"},
       {:plug_cowboy, "~> 2.0"},
-      {:sentry, "~> 10.8"},
+      {:sentry, "~> 11.0"},
       {:ssl_verify_fun, "~> 1.1", manager: :rebar3, override: true},
       {:sweet_xml, "~> 0.7.0"},
       {:hex_core, "~> 0.11.0"},