🛠️ Fix meilisearch task monitoring to unzip responses (#2064)

Author: davoclavo

lib/sequin/sinks/meilisearch/client.ex

@@ -33,7 +33,7 @@ defmodule Sequin.Sinks.Meilisearch.Client do
             case response_or_exception do
               %Req.Response{status: 200, body: encoded_body} ->
                 # NOTE: Req does not automatically decode on retry functions
-                case Jason.decode(encoded_body) do
+                case encoded_body |> :zlib.gunzip() |> Jason.decode() do
                   {:ok, %{"status" => status}} when status in ["enqueued", "processing"] ->
                     true
 

test/sequin/meilisearch_client_test.exs

@@ -5,6 +5,14 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
   alias Sequin.Factory.SinkFactory
   alias Sequin.Sinks.Meilisearch.Client
 
+  defp send_gzipped_response(conn, status_code, response_data) do
+    gzipped_body = response_data |> Jason.encode!() |> :zlib.gzip()
+
+    conn
+    |> Plug.Conn.put_resp_header("content-encoding", "gzip")
+    |> Plug.Conn.send_resp(status_code, gzipped_body)
+  end
+
   @sink %MeilisearchSink{
     type: :meilisearch,
     endpoint_url: "http://127.0.0.1:7700",
@@ -51,9 +59,8 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
         assert conn.method == "GET"
         assert conn.request_path == "/tasks/1"
 
-        Req.Test.json(conn, %{
-          "status" => "success"
-        })
+        response_data = %{"status" => "succeeded"}
+        send_gzipped_response(conn, 200, response_data)
       end)
 
       assert :ok = Client.import_documents(@sink, "test", records)
@@ -77,9 +84,8 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
         assert conn.method == "GET"
         assert conn.request_path == "/tasks/1"
 
-        Req.Test.json(conn, %{
-          "status" => "success"
-        })
+        response_data = %{"status" => "succeeded"}
+        send_gzipped_response(conn, 200, response_data)
       end)
 
       ids = Enum.map(records, & &1["id"])
@@ -158,7 +164,8 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
         assert conn.method == "GET"
         assert conn.request_path == "/tasks/123"
 
-        Req.Test.json(conn, %{"status" => "succeeded"})
+        response_data = %{"status" => "succeeded"}
+        send_gzipped_response(conn, 200, response_data)
       end)
 
       assert :ok =
@@ -198,7 +205,8 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
         assert conn.method == "GET"
         assert conn.request_path == "/tasks/456"
 
-        Req.Test.json(conn, %{"status" => "succeeded"})
+        response_data = %{"status" => "succeeded"}
+        send_gzipped_response(conn, 200, response_data)
       end)
 
       assert :ok =
@@ -222,13 +230,15 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
         assert conn.method == "GET"
         assert conn.request_path == "/tasks/789"
 
-        Req.Test.json(conn, %{
+        response_data = %{
           "status" => "failed",
           "error" => %{
             "message" => "Invalid filter expression",
             "code" => "invalid_search_filter"
           }
-        })
+        }
+
+        send_gzipped_response(conn, 200, response_data)
       end)
 
       assert {:error, error} =
@@ -268,19 +278,22 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
         :counters.add(call_count, 1, 1)
         send(test_pid, {:task_check, count})
 
-        cond do
-          count <= 2 ->
-            # First two checks show task is still processing
-            Req.Test.json(conn, %{"status" => "processing", "taskUid" => 123})
+        response_data =
+          cond do
+            count <= 2 ->
+              # First two checks show task is still processing
+              %{"status" => "processing", "taskUid" => 123}
 
-          count == 3 ->
-            # Third check shows task succeeded
-            Req.Test.json(conn, %{"status" => "succeeded", "taskUid" => 123})
+            count == 3 ->
+              # Third check shows task succeeded
+              %{"status" => "succeeded", "taskUid" => 123}
+
+            true ->
+              # Should not get here
+              %{"status" => "succeeded", "taskUid" => 123}
+          end
 
-          true ->
-            # Should not get here
-            Req.Test.json(conn, %{"status" => "succeeded", "taskUid" => 123})
-        end
+        send_gzipped_response(conn, 200, response_data)
       end)
 
       # Should succeed after multiple task status checks
@@ -315,7 +328,9 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
         send(test_pid, {:task_check, count})
 
         # Always return processing status
-        Req.Test.json(conn, %{"status" => "processing", "taskUid" => 456})
+        response_data = %{"status" => "processing", "taskUid" => 456}
+
+        send_gzipped_response(conn, 200, response_data)
       end)
 
       # Should fail after exhausting retries
@@ -326,7 +341,7 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
 
       # Verify we made all 6 attempts (1 initial + 5 retries)
       for i <- 0..5 do
-        assert_receive {:task_check, ^i}, 15_000
+        assert_receive {:task_check, ^i}, 3_000
       end
 
       refute_receive {:task_check, 6}, 100
@@ -348,14 +363,16 @@ defmodule Sequin.Sinks.Meilisearch.ClientTest do
         assert conn.method == "GET"
         assert conn.request_path == "/tasks/789"
 
-        Req.Test.json(conn, %{
+        response_data = %{
           "status" => "failed",
           "taskUid" => 789,
           "error" => %{
             "code" => "invalid_document_id",
             "message" => "Document ID is invalid"
           }
-        })
+        }
+
+        send_gzipped_response(conn, 200, response_data)
       end)
 
       # Should return error with details

test/sequin/meilisearch_pipeline_test.exs

@@ -9,6 +9,14 @@ defmodule Sequin.Runtime.MeilisearchPipelineTest do
   alias Sequin.Runtime.SinkPipeline
   alias Sequin.Sinks.Meilisearch.Client
 
+  defp send_gzipped_response(conn, status_code, response_data) do
+    gzipped_body = response_data |> Jason.encode!() |> :zlib.gzip()
+
+    conn
+    |> Plug.Conn.put_resp_header("content-encoding", "gzip")
+    |> Plug.Conn.send_resp(status_code, gzipped_body)
+  end
+
   describe "meilisearch pipeline" do
     setup do
       account = AccountsFactory.insert_account!()
@@ -123,7 +131,9 @@ defmodule Sequin.Runtime.MeilisearchPipelineTest do
 
           "GET" ->
             assert conn.request_path =~ "/tasks/"
-            Req.Test.json(conn, %{"status" => "succeeded"})
+
+            response_data = %{"status" => "succeeded"}
+            send_gzipped_response(conn, 200, response_data)
         end
       end)
 
@@ -176,13 +186,15 @@ defmodule Sequin.Runtime.MeilisearchPipelineTest do
         assert conn.method == "GET"
         assert conn.request_path == "/tasks/#{task_uid}"
 
-        Req.Test.json(conn, %{
+        response_data = %{
           "status" => "failed",
           "error" => %{
             "message" => "Invalid function expression",
             "code" => "invalid_document_function"
           }
-        })
+        }
+
+        send_gzipped_response(conn, 200, response_data)
       end)
 
       start_pipeline!(consumer)
@@ -274,9 +286,11 @@ defmodule Sequin.Runtime.MeilisearchPipelineTest do
       assert conn.method == "GET"
       assert conn.request_path == "/tasks/#{task_uid}"
 
-      Req.Test.json(conn, %{
-        "status" => "success"
-      })
+      response_data = %{
+        "status" => "succeeded"
+      }
+
+      send_gzipped_response(conn, 200, response_data)
     end)
 
     start_pipeline!(consumer)