fix: Add create Channel endpoint (#743)

lib/realtime/api/channel.ex

@@ -17,9 +17,9 @@ defmodule Realtime.Api.Channel do
   def changeset(channel, attrs) do
     channel
     |> cast(attrs, [:name, :inserted_at, :updated_at])
+    |> validate_required([:name])
     |> put_timestamp(:updated_at)
     |> maybe_put_timestamp(:inserted_at)
-    |> validate_required([:name])
   end
 
   defp put_timestamp(changeset, field) do

lib/realtime/channels.ex

@@ -31,11 +31,8 @@ defmodule Realtime.Channels do
   @spec create_channel(map(), DBConnection.conn()) :: {:ok, Channel.t()} | {:error, any()}
   def create_channel(attrs, conn) do
     channel = Channel.changeset(%Channel{}, attrs)
-    {query, args} = Repo.insert_query_from_changeset(channel)
 
-    conn
-    |> Postgrex.query(query, args)
-    |> Repo.result_to_single_struct(Channel)
+    Repo.insert(conn, channel, Channel)
   end
 
   @doc """

lib/realtime/repo.ex

@@ -38,13 +38,26 @@ defmodule Realtime.Repo do
     |> result_to_single_struct(result_struct)
   end
 
+  @doc """
+  Inserts a given changeset into the database and converts the result into a given struct
+  """
+  @spec insert(DBConnection.conn(), Ecto.Changeset.t(), module()) ::
+          {:ok, struct()} | {:error, any()} | Ecto.Changeset.t()
+  def insert(conn, changeset, result_struct) do
+    with {:ok, {query, args}} <- insert_query_from_changeset(changeset) do
+      conn
+      |> Postgrex.query(query, args)
+      |> result_to_single_struct(result_struct)
+    end
+  end
+
   @doc """
   Converts a Postgrex.Result into a given struct
   """
   @spec result_to_single_struct({:ok, Postgrex.Result.t()} | {:error, any()}, module()) ::
           {:ok, struct()} | {:ok, nil} | {:error, any()}
   def result_to_single_struct({:ok, %Postgrex.Result{rows: [row], columns: columns}}, struct) do
-    {:ok, Realtime.Repo.load(struct, Enum.zip(columns, row))}
+    {:ok, load(struct, Enum.zip(columns, row))}
   end
 
   def result_to_single_struct({:ok, %Postgrex.Result{rows: []}}, _),
@@ -69,7 +82,10 @@ defmodule Realtime.Repo do
   @doc """
   Creates an insert query from a given changeset
   """
-  @spec insert_query_from_changeset(Ecto.Changeset.t()) :: {String.t(), [any()]}
+  @spec insert_query_from_changeset(Ecto.Changeset.t()) ::
+          {:ok, {String.t(), [any()]}} | {:error, Ecto.Changeset.t()}
+  def insert_query_from_changeset(%{valid?: false} = changeset), do: {:error, changeset}
+
   def insert_query_from_changeset(changeset) do
     schema = changeset.data.__struct__
     source = schema.__schema__(:source)
@@ -93,7 +109,7 @@ defmodule Realtime.Repo do
       |> Enum.map(fn {_, index} -> "$#{index}" end)
       |> Enum.join(",")
 
-    {"INSERT INTO #{table} #{header} VALUES (#{arg_index}) RETURNING *", rows}
+    {:ok, {"INSERT INTO #{table} #{header} VALUES (#{arg_index}) RETURNING *", rows}}
   end
 
   defp run_all_query(conn, query) do

lib/realtime_web/controllers/channels_controller.ex

@@ -4,6 +4,7 @@ defmodule RealtimeWeb.ChannelsController do
 
   alias Realtime.Channels
   alias Realtime.Tenants.Connect
+  alias RealtimeWeb.OpenApiSchemas.ChannelParams
   alias RealtimeWeb.OpenApiSchemas.ChannelResponse
   alias RealtimeWeb.OpenApiSchemas.ChannelResponseList
   alias RealtimeWeb.OpenApiSchemas.NotFoundResponse
@@ -68,4 +69,32 @@ defmodule RealtimeWeb.ChannelsController do
       error -> error
     end
   end
+
+  operation(:create,
+    summary: "Create user channel",
+    parameters: [
+      token: [
+        in: :header,
+        name: "Authorization",
+        schema: %OpenApiSpex.Schema{type: :string},
+        required: true,
+        example:
+          "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpYXQiOjE2ODAxNjIxNTR9.U9orU6YYqXAtpF8uAiw6MS553tm4XxRzxOhz2IwDhpY"
+      ]
+    ],
+    request_body: ChannelParams.params(),
+    responses: %{
+      201 => ChannelResponse.response(),
+      404 => NotFoundResponse.response()
+    }
+  )
+
+  def create(%{assigns: %{tenant: tenant}} = conn, params) do
+    with {:ok, db_conn} <- Connect.lookup_or_start_connection(tenant.external_id),
+         {:ok, channel} <- Channels.create_channel(params, db_conn) do
+      conn
+      |> put_status(:created)
+      |> json(channel)
+    end
+  end
 end

lib/realtime_web/open_api_schemas.ex

@@ -5,6 +5,24 @@ defmodule RealtimeWeb.OpenApiSchemas do
 
   alias OpenApiSpex.Schema
 
+  defmodule ChannelParams do
+    @moduledoc false
+    require OpenApiSpex
+
+    OpenApiSpex.schema(%{
+      type: :object,
+      properties: %{
+        name: %Schema{
+          type: :string,
+          description: "Channel Name",
+          example: "channel-1"
+        }
+      }
+    })
+
+    def params(), do: {"Channel Params", "application/json", __MODULE__}
+  end
+
   defmodule TenantBatchParams do
     @moduledoc false
     require OpenApiSpex

mix.exs

@@ -4,7 +4,7 @@ defmodule Realtime.MixProject do
   def project do
     [
       app: :realtime,
-      version: "2.25.36",
+      version: "2.25.37",
       elixir: "~> 1.14.0",
       elixirc_paths: elixirc_paths(Mix.env()),
       start_permanent: Mix.env() == :prod,

test/realtime/channels_test.exs

@@ -44,6 +44,13 @@ defmodule Realtime.ChannelsTest do
       name = random_string()
       assert {:ok, %Channel{name: ^name}} = Channels.create_channel(%{name: name}, conn)
     end
+
+    test "no channel name has error changeset", %{conn: conn} do
+      assert {:error, %Ecto.Changeset{valid?: false, errors: errors}} =
+               Channels.create_channel(%{}, conn)
+
+      assert ^errors = [name: {"can't be blank", [validation: :required]}]
+    end
   end
 
   describe "get_channel_by_name/2" do

test/realtime/repo_test.exs

@@ -1,7 +1,18 @@
 defmodule Realtime.RepoTest do
   use Realtime.DataCase, async: false
-  alias Realtime.Repo
+
+  import Ecto.Query
+
   alias Realtime.Api.Channel
+  alias Realtime.Repo
+  alias Realtime.Tenants.Connect
+
+  setup do
+    tenant = tenant_fixture()
+    {:ok, conn} = Connect.lookup_or_start_connection(tenant.external_id)
+    truncate_table(conn, "realtime.channels")
+    %{conn: conn, tenant: tenant}
+  end
 
   describe "with_dynamic_repo/2" do
     test "starts a repo with the given config and kills it in the end of the command" do
@@ -114,36 +125,55 @@ defmodule Realtime.RepoTest do
     end
   end
 
-  defp db_config() do
-    tenant = tenant_fixture()
+  describe "all/3" do
+    test "fetches multiple entries and loads a given struct", %{conn: conn, tenant: tenant} do
+      channel_1 = channel_fixture(tenant)
+      channel_2 = channel_fixture(tenant)
 
-    %{
-      "db_host" => db_host,
-      "db_name" => db_name,
-      "db_password" => db_password,
-      "db_port" => db_port,
-      "db_user" => db_user
-    } = args = tenant.extensions |> hd() |> then(& &1.settings)
+      assert {:ok, [^channel_1, ^channel_2]} = Repo.all(conn, Channel, Channel)
+    end
+  end
 
-    {host, port, name, user, pass} =
-      Realtime.Helpers.decrypt_creds(
-        db_host,
-        db_port,
-        db_name,
-        db_user,
-        db_password
-      )
+  describe "one/3" do
+    test "fetches one entry and loads a given struct", %{conn: conn, tenant: tenant} do
+      channel_1 = channel_fixture(tenant)
+      _channel_2 = channel_fixture(tenant)
+      query = from c in Channel, where: c.id == ^channel_1.id
+      assert {:ok, ^channel_1} = Repo.one(conn, query, Channel)
+    end
 
-    ssl_enforced = Realtime.Helpers.default_ssl_param(args)
+    test "raises exception on multiple results", %{conn: conn, tenant: tenant} do
+      _channel_1 = channel_fixture(tenant)
+      _channel_2 = channel_fixture(tenant)
 
-    [
-      hostname: host,
-      port: port,
-      database: name,
-      password: pass,
-      username: user,
-      ssl_enforced: ssl_enforced
-    ]
+      assert_raise RuntimeError, "expected at most one result but got 2 in result", fn ->
+        Repo.one(conn, Channel, Channel)
+      end
+    end
+
+    test "if not found, returns nil", %{conn: conn} do
+      query = from c in Channel, where: c.name == "potato"
+      assert {:ok, nil} = Repo.one(conn, query, Channel)
+    end
+  end
+
+  describe "insert/3" do
+    test "inserts a new entry with a given changeset and returns struct", %{conn: conn} do
+      changeset = Channel.changeset(%Channel{}, %{name: "foo"})
+      assert {:ok, %Channel{}} = Repo.insert(conn, changeset, Channel)
+    end
+
+    test "returns changeset if changeset is invalid", %{conn: conn} do
+      changeset = Channel.changeset(%Channel{}, %{})
+      res = Repo.insert(conn, changeset, Channel)
+      assert {:error, %Ecto.Changeset{valid?: false}} = res
+    end
+
+    test "returns an error on Postgrex error", %{conn: conn} do
+      changeset = Channel.changeset(%Channel{}, %{name: "foo"})
+      assert {:ok, _} = Repo.insert(conn, changeset, Channel)
+      assert {:error, _} = Repo.insert(conn, changeset, Channel)
+    end
   end
 
   describe "result_to_single_struct/2" do
@@ -234,10 +264,43 @@ defmodule Realtime.RepoTest do
         {"INSERT INTO \"realtime\".\"channels\" (\"updated_at\",\"name\",\"inserted_at\") VALUES ($1,$2,$3) RETURNING *",
          [updated_at, "foo", inserted_at]}
 
-      assert Repo.insert_query_from_changeset(changeset) == expected
+      {:ok, res} = Repo.insert_query_from_changeset(changeset)
+      assert res == expected
     end
   end
 
+  defp db_config() do
+    tenant = tenant_fixture()
+
+    %{
+      "db_host" => db_host,
+      "db_name" => db_name,
+      "db_password" => db_password,
+      "db_port" => db_port,
+      "db_user" => db_user
+    } = args = tenant.extensions |> hd() |> then(& &1.settings)
+
+    {host, port, name, user, pass} =
+      Realtime.Helpers.decrypt_creds(
+        db_host,
+        db_port,
+        db_name,
+        db_user,
+        db_password
+      )
+
+    ssl_enforced = Realtime.Helpers.default_ssl_param(args)
+
+    [
+      hostname: host,
+      port: port,
+      database: name,
+      password: pass,
+      username: user,
+      ssl_enforced: ssl_enforced
+    ]
+  end
+
   defp metadata do
     %Ecto.Schema.Metadata{
       prefix: Channel.__schema__(:prefix),

test/realtime_web/controllers/channels_controller_test.exs

@@ -70,4 +70,18 @@ defmodule RealtimeWeb.ChannelsControllerTest do
       assert json_response(conn, 404) == %{"message" => "Not found"}
     end
   end
+
+  describe "create" do
+    test "creates a channel", %{conn: conn} do
+      name = random_string()
+      conn = post(conn, ~p"/api/channels", %{name: name})
+      res = json_response(conn, 201)
+      assert name == res["name"]
+    end
+
+    test "422 if params are invalid", %{conn: conn} do
+      conn = post(conn, ~p"/api/channels", %{})
+      assert json_response(conn, 422) == %{"errors" => %{"name" => ["can't be blank"]}}
+    end
+  end
 end