fix: adds shutdown endpoint (#1693)

Author: filipecabaco

lib/realtime_web/controllers/fallback_controller.ex

@@ -11,10 +11,12 @@ defmodule RealtimeWeb.FallbackController do
   import RealtimeWeb.ErrorHelpers
 
   def call(conn, {:error, :not_found}) do
+    log_error("TenantNotFound", "Tenant not found")
+
     conn
     |> put_status(:not_found)
     |> put_view(RealtimeWeb.ErrorView)
-    |> render("error.json", message: "Not found")
+    |> render("error.json", message: "not found")
   end
 
   def call(conn, {:error, %Ecto.Changeset{} = changeset}) do

lib/realtime_web/controllers/tenant_controller.ex

@@ -28,7 +28,7 @@ defmodule RealtimeWeb.TenantController do
 
   action_fallback(RealtimeWeb.FallbackController)
 
-  plug :set_observability_attributes when action in [:show, :edit, :update, :delete, :reload, :health]
+  plug :set_observability_attributes when action in [:show, :edit, :update, :delete, :reload, :shutdown, :health]
 
   operation(:index,
     summary: "List tenants",
@@ -77,13 +77,8 @@ defmodule RealtimeWeb.TenantController do
     tenant = Api.get_tenant_by_external_id(id)
 
     case tenant do
-      %Tenant{} = tenant ->
-        render(conn, "show.json", tenant: tenant)
-
-      nil ->
-        conn
-        |> put_status(404)
-        |> render("not_found.json", tenant: nil)
+      %Tenant{} = tenant -> render(conn, "show.json", tenant: tenant)
+      nil -> {:error, :not_found}
     end
   end
 
@@ -201,7 +196,6 @@ defmodule RealtimeWeb.TenantController do
       send_resp(conn, 204, "")
     else
       nil ->
-        log_error("TenantNotFound", "Tenant not found")
         send_resp(conn, 204, "")
 
       err ->
@@ -233,11 +227,7 @@ defmodule RealtimeWeb.TenantController do
   def reload(conn, %{"tenant_id" => tenant_id}) do
     case Api.get_tenant_by_external_id(tenant_id, use_replica?: false) do
       nil ->
-        log_error("TenantNotFound", "Tenant not found")
-
-        conn
-        |> put_status(404)
-        |> render("not_found.json", tenant: nil)
+        {:error, :not_found}
 
       tenant ->
         PostgresCdc.stop_all(tenant, @stop_timeout)
@@ -247,6 +237,37 @@ defmodule RealtimeWeb.TenantController do
     end
   end
 
+  operation(:shutdown,
+    summary: "Shutdowns the Connect module for a tenant",
+    parameters: [
+      token: [
+        in: :header,
+        name: "Authorization",
+        schema: %OpenApiSpex.Schema{type: :string},
+        required: true,
+        example:
+          "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE2ODAxNjIxNTR9.U9orU6YYqXAtpF8uAiw6MS553tm4XxRzxOhz2IwDhpY"
+      ],
+      tenant_id: [in: :path, description: "Tenant ID", type: :string]
+    ],
+    responses: %{
+      204 => EmptyResponse.response(),
+      403 => EmptyResponse.response(),
+      404 => NotFoundResponse.response()
+    }
+  )
+
+  def shutdown(conn, %{"tenant_id" => tenant_id}) do
+    case Api.get_tenant_by_external_id(tenant_id, use_replica?: false) do
+      nil ->
+        {:error, :not_found}
+
+      tenant ->
+        Connect.shutdown(tenant.external_id)
+        send_resp(conn, 204, "")
+    end
+  end
+
   operation(:health,
     summary: "Tenant health",
     parameters: [
@@ -269,18 +290,9 @@ defmodule RealtimeWeb.TenantController do
 
   def health(conn, %{"tenant_id" => tenant_id}) do
     case Tenants.health_check(tenant_id) do
-      {:ok, response} ->
-        json(conn, %{data: response})
-
-      {:error, %{healthy: false} = response} ->
-        json(conn, %{data: response})
-
-      {:error, :tenant_not_found} ->
-        log_error("TenantNotFound", "Tenant not found")
-
-        conn
-        |> put_status(404)
-        |> render("not_found.json", tenant: nil)
+      {:ok, response} -> json(conn, %{data: response})
+      {:error, %{healthy: false} = response} -> json(conn, %{data: response})
+      {:error, :tenant_not_found} -> {:error, :not_found}
     end
   end
 

lib/realtime_web/router.ex

@@ -90,6 +90,7 @@ defmodule RealtimeWeb.Router do
 
     resources("/tenants", TenantController, param: "tenant_id", except: [:edit, :new])
     post("/tenants/:tenant_id/reload", TenantController, :reload)
+    post("/tenants/:tenant_id/shutdown", TenantController, :shutdown)
     get("/tenants/:tenant_id/health", TenantController, :health)
   end
 

test/realtime_web/controllers/tenant_controller_test.exs

@@ -49,7 +49,7 @@ defmodule RealtimeWeb.TenantControllerTest do
     test "returns not found on non existing tenant", %{conn: conn} do
       conn = get(conn, ~p"/api/tenants/no")
       response = json_response(conn, 404)
-      assert response == %{"error" => "not found"}
+      assert response == %{"message" => "not found"}
     end
 
     test "sets appropriate observability metadata", %{conn: conn, tenant: tenant} do
@@ -331,6 +331,59 @@ defmodule RealtimeWeb.TenantControllerTest do
     end
   end
 
+  describe "shutdown Connect module for tenant" do
+    setup [:with_tenant]
+
+    test "shuts down Connect process when tenant exists", %{conn: conn, tenant: %{external_id: external_id}} do
+      Phoenix.PubSub.subscribe(Realtime.PubSub, "realtime:operations:" <> external_id)
+
+      {:ok, connect_pid} = Connect.lookup_or_start_connection(external_id)
+      Process.monitor(connect_pid)
+
+      assert Process.alive?(connect_pid)
+
+      %{status: status} = post(conn, ~p"/api/tenants/#{external_id}/shutdown")
+
+      assert status == 204
+      assert_receive {:DOWN, _, :process, ^connect_pid, _}
+      refute Process.alive?(connect_pid)
+    end
+
+    test "returns 204 when tenant exists but Connect is not running", %{conn: conn, tenant: %{external_id: external_id}} do
+      %{status: status} = post(conn, ~p"/api/tenants/#{external_id}/shutdown")
+      assert status == 204
+    end
+
+    test "returns 404 when tenant does not exist", %{conn: conn} do
+      %{status: status} = post(conn, ~p"/api/tenants/nope/shutdown")
+      assert status == 404
+    end
+
+    test "returns 403 when jwt is invalid", %{conn: conn, tenant: tenant} do
+      conn = put_req_header(conn, "authorization", "Bearer potato")
+      conn = post(conn, ~p"/api/tenants/#{tenant.external_id}/shutdown")
+      assert response(conn, 403) == ""
+    end
+
+    test "sets appropriate observability metadata", %{conn: conn, tenant: tenant} do
+      external_id = tenant.external_id
+
+      Tracer.with_span "test" do
+        Task.async(fn ->
+          post(conn, ~p"/api/tenants/#{tenant.external_id}/shutdown")
+
+          assert Logger.metadata()[:external_id] == external_id
+          assert Logger.metadata()[:project] == external_id
+        end)
+        |> Task.await()
+      end
+
+      assert_receive {:span, span(name: "POST /api/tenants/:tenant_id/shutdown", attributes: attributes)}
+
+      assert attributes(map: %{external_id: ^external_id}) = attributes
+    end
+  end
+
   describe "health check tenant" do
     setup [:with_tenant]