checkpoint: simplify batchers, fix action type, handle errors better

Author: davoclavo

assets/svelte/sinks/meilisearch/MeilisearchSinkForm.svelte

@@ -72,8 +72,11 @@
               Return <code>action: :function</code> from your routing function
             </li>
             <li>
-              Specify a filter expression and JavaScript-like function to update
-              specific fields
+              Specify a filter expression and a <a
+                class="underline font-medium"
+                href="https://rhai.rs/book/engine/hello-world.html"
+                target="_blank">Rhai</a> (a JavaScript-like language) function to update
+              specific fields.
             </li>
             <li>
               This is useful for updating nested arrays or specific document

docs/reference/sinks/meilisearch.mdx

@@ -97,7 +97,7 @@ The `function` action allows you to update documents using Meilisearch's functio
 
 When using `action: "function"`, you must also provide:
 - `filter` - A Meilisearch filter expression to select documents to update
-- `function` - A Rhai function expression to apply to the matched documents
+- `function` - A function expression using [Rhai (a JavaScript-like language)](https://docs.meilisearch.com/reference/api/documents#edit-documents-with-rhai) to apply to the matched documents
 - `context` (optional) - Additional context data for the function
 
 Example routing function with function updates:

lib/sequin/runtime/meilisearch_pipeline.ex

@@ -23,16 +23,6 @@ defmodule Sequin.Runtime.MeilisearchPipeline do
         concurrency: concurrency,
         batch_size: consumer.sink.batch_size || 100,
         batch_timeout: 5
-      ],
-      delete: [
-        concurrency: concurrency,
-        batch_size: consumer.sink.batch_size || 100,
-        batch_timeout: 5
-      ],
-      function: [
-        concurrency: concurrency,
-        batch_size: consumer.sink.batch_size || 100,
-        batch_timeout: 5
       ]
     ]
   end
@@ -42,17 +32,17 @@ defmodule Sequin.Runtime.MeilisearchPipeline do
     routing_info = Routing.route_message(context.consumer, message.data)
     %Routing.Consumers.Meilisearch{action: action, index_name: index_name} = routing_info
 
-    batcher =
-      case action do
-        :index -> :default
-        :delete -> :delete
-        :function -> :function
+    action =
+      if is_atom(action) do
+        # Backwards compatibility with existing routing functions with atom actions
+        to_string(action)
+      else
+        action
       end
 
     message =
       message
-      |> Broadway.Message.put_batcher(batcher)
-      |> Broadway.Message.put_batch_key(index_name)
+      |> Broadway.Message.put_batch_key({action, index_name})
       |> Broadway.Message.update_data(fn data ->
         Map.put(data, :routing_info, routing_info)
       end)
@@ -61,14 +51,12 @@ defmodule Sequin.Runtime.MeilisearchPipeline do
   end
 
   @impl SinkPipeline
-  def handle_batch(:default, messages, batch_info, context) do
+  def handle_batch(:default, messages, %{batch_key: {"index", index_name}}, context) do
     %{
       consumer: %SinkConsumer{sink: sink, transform: transform} = consumer,
       test_pid: test_pid
     } = context
 
-    index_name = batch_info.batch_key
-
     setup_allowances(test_pid)
 
     records =
@@ -97,14 +85,12 @@ defmodule Sequin.Runtime.MeilisearchPipeline do
   end
 
   @impl SinkPipeline
-  def handle_batch(:delete, messages, batch_info, context) do
+  def handle_batch(:default, messages, %{batch_key: {"delete", index_name}}, context) do
     %{
       consumer: %SinkConsumer{sink: sink} = consumer,
       test_pid: test_pid
     } = context
 
-    index_name = batch_info.batch_key
-
     setup_allowances(test_pid)
 
     document_ids =
@@ -131,14 +117,12 @@ defmodule Sequin.Runtime.MeilisearchPipeline do
   end
 
   @impl SinkPipeline
-  def handle_batch(:function, messages, batch_info, context) do
+  def handle_batch(:default, messages, %{batch_key: {"function", index_name}}, context) do
     %{
       consumer: %SinkConsumer{sink: sink} = consumer,
       test_pid: test_pid
     } = context
 
-    index_name = batch_info.batch_key
-
     setup_allowances(test_pid)
 
     # Process each message individually

lib/sequin/runtime/routing/consumers/meilisearch.ex

@@ -40,10 +40,10 @@ defmodule Sequin.Runtime.Routing.Consumers.Meilisearch do
   def route(action, _record, _changes, _metadata) do
     meilisearch_action =
       case action do
-        "insert" -> :index
-        "update" -> :index
-        "delete" -> :delete
-        "read" -> :index
+        "insert" -> "index"
+        "update" -> "index"
+        "delete" -> "delete"
+        "read" -> "index"
       end
 
     %{action: meilisearch_action}

lib/sequin/sinks/meilisearch/client.ex

@@ -20,7 +20,7 @@ defmodule Sequin.Sinks.Meilisearch.Client do
       req = base_request(sink)
 
       case Req.get(req, url: "/tasks/#{task_id}") do
-        {:ok, %{body: body}} ->
+        {:ok, %{status: status, body: body}} when status in 200..299 ->
           case body do
             %{"status" => status} when status in ["enqueued", "processing"] ->
               timeout = Sequin.Time.exponential_backoff(200, retries, 10_000)
@@ -42,6 +42,16 @@ defmodule Sequin.Sinks.Meilisearch.Client do
               {:ok}
           end
 
+        {:ok, %{status: status, body: body}} ->
+          message = extract_error_message(body) || "Request failed with status #{status}"
+
+          {:error,
+           Error.service(
+             service: :meilisearch,
+             message: message,
+             details: %{status: status, body: body}
+           )}
+
         {:error, reason} ->
           {:error, Error.service(service: :meilisearch, message: "Unknown error", details: reason)}
       end
@@ -64,9 +74,19 @@ defmodule Sequin.Sinks.Meilisearch.Client do
       )
 
     case Req.put(req) do
-      {:ok, %{body: body}} ->
+      {:ok, %{status: status, body: body}} when status in 200..299 ->
         verify_task_by_id(sink, body["taskUid"], 0)
 
+      {:ok, %{status: status, body: body}} ->
+        message = extract_error_message(body) || "Request failed with status #{status}"
+
+        {:error,
+         Error.service(
+           service: :meilisearch,
+           message: message,
+           details: %{status: status, body: body}
+         )}
+
       {:error, %Req.TransportError{} = error} ->
         {:error,
          Error.service(
@@ -93,9 +113,19 @@ defmodule Sequin.Sinks.Meilisearch.Client do
       )
 
     case Req.post(req) do
-      {:ok, %{body: body}} ->
+      {:ok, %{status: status, body: body}} when status in 200..299 ->
         verify_task_by_id(sink, body["taskUid"], 0)
 
+      {:ok, %{status: status, body: body}} ->
+        message = extract_error_message(body) || "Request failed with status #{status}"
+
+        {:error,
+         Error.service(
+           service: :meilisearch,
+           message: message,
+           details: %{status: status, body: body}
+         )}
+
       {:error, %Req.TransportError{} = error} ->
         {:error,
          Error.service(
@@ -129,9 +159,19 @@ defmodule Sequin.Sinks.Meilisearch.Client do
       )
 
     case Req.post(req) do
-      {:ok, %{body: body}} ->
+      {:ok, %{status: status, body: body}} when status in 200..299 ->
         verify_task_by_id(sink, body["taskUid"], 0)
 
+      {:ok, %{status: status, body: body}} ->
+        message = extract_error_message(body) || "Request failed with status #{status}"
+
+        {:error,
+         Error.service(
+           service: :meilisearch,
+           message: message,
+           details: %{status: status, body: body}
+         )}
+
       {:error, %Req.TransportError{} = error} ->
         {:error,
          Error.service(
@@ -170,9 +210,17 @@ defmodule Sequin.Sinks.Meilisearch.Client do
     req = base_request(sink)
 
     case Req.get(req, url: "/health") do
-      {:ok, %{status: status}} when status == 200 ->
+      {:ok, %{status: status}} when status in 200..299 ->
         :ok
 
+      {:ok, %{status: status, body: body}} ->
+        {:error,
+         Error.service(
+           service: :meilisearch,
+           message: "Health check failed with status #{status}",
+           details: %{status: status, body: body}
+         )}
+
       {:error, reason} ->
         {:error,
          Error.service(