fix(electric): Rewrite PostgreSQL URI parser (#706)

Main improvements:
* support `postgres://` scheme
* default to using the username as the database name if the latter is
missing from the URI (default behaviour of `psql`)

Fixes VAX-1264, VAX-1265.

.changeset/lovely-tigers-thank.md

@@ -0,0 +1,5 @@
+---
+"@core/electric": patch
+---
+
+[VAX-1264, VAX-1265] Fix some edge cases in the parsing of DATABASE_URL.

components/electric/config/runtime.exs

@@ -120,9 +120,7 @@ if config_env() == :prod do
 
   postgresql_connection =
     System.fetch_env!("DATABASE_URL")
-    |> PostgresqlUri.parse()
-    |> then(&Keyword.put(&1, :host, &1[:hostname]))
-    |> Keyword.delete(:hostname)
+    |> Electric.Utils.parse_postgresql_uri()
     |> Keyword.put_new(:ssl, require_ssl?)
     |> Keyword.put(:ipv6, use_ipv6?)
     |> Keyword.update(:timeout, 5_000, &String.to_integer/1)

components/electric/lib/electric/utils.ex

@@ -208,4 +208,167 @@ defmodule Electric.Utils do
       |> Map.put(:tcp_opts, [:inet6])
     end
   end
+
+  @doc """
+  Parse a PostgreSQL URI into a keyword list.
+
+  ## Examples
+
+      iex> parse_postgresql_uri("postgresql://postgres:[email protected]/app-db")
+      [
+        host: "example.com",
+        port: 5432,
+        database: "app-db",
+        username: "postgres",
+        password: "password",
+      ]
+
+      iex> parse_postgresql_uri("postgresql://[email protected]:81/__shadow")
+      [
+        host: "192.168.111.33",
+        port: 81,
+        database: "__shadow",
+        username: "electric"
+      ]
+
+      iex> parse_postgresql_uri("postgresql://pg@[2001:db8::1234]:4321")
+      [
+        host: "2001:db8::1234",
+        port: 4321,
+        database: "pg",
+        username: "pg"
+      ]
+
+      iex> parse_postgresql_uri("postgresql://user@localhost:5433/")
+      [
+        host: "localhost",
+        port: 5433,
+        database: "user",
+        username: "user"
+      ]
+
+      iex> parse_postgresql_uri("postgresql://user@localhost:5433/mydb?options=-c%20synchronous_commit%3Doff")
+      [
+        host: "localhost",
+        port: 5433,
+        database: "mydb",
+        username: "user"
+      ]
+
+      iex> parse_postgresql_uri("postgresql://electric@localhost/db?replication=database")
+      [
+        host: "localhost",
+        port: 5432,
+        database: "db",
+        username: "electric",
+        replication: "database"
+      ]
+
+      iex> parse_postgresql_uri("postgresql://electric@localhost/db?replication=off")
+      [
+        host: "localhost",
+        port: 5432,
+        database: "db",
+        username: "electric"
+      ]
+
+  For the `sslmode` keyword, any value but "disable" will result in enabling SSL.
+
+      iex> parse_postgresql_uri("postgres://super_user@localhost:7801/postgres?sslmode=yesplease")
+      [
+        host: "localhost",
+        port: 7801,
+        database: "postgres",
+        username: "super_user",
+        ssl: true
+      ]
+  """
+  @spec parse_postgresql_uri(binary) :: keyword
+  def parse_postgresql_uri(uri_str) do
+    %URI{scheme: scheme, host: host, port: port, path: path, userinfo: userinfo, query: query} =
+      URI.parse(uri_str)
+
+    :ok = assert_valid_scheme!(scheme)
+
+    :ok = assert_valid_host!(host)
+    port = port || 5432
+
+    {username, password} = parse_userinfo!(userinfo)
+
+    database = parse_database(path, username)
+
+    query_params =
+      if query do
+        URI.decode_query(query)
+      else
+        %{}
+      end
+
+    [
+      host: host,
+      port: port,
+      database: database,
+      username: username,
+      password: password
+    ]
+    |> add_replication_param(query_params["replication"])
+    |> add_ssl_param(query_params["sslmode"])
+    |> Enum.reject(fn {_key, val} -> is_nil(val) end)
+  end
+
+  defp assert_valid_scheme!(scheme) when scheme in ["postgres", "postgresql"], do: :ok
+
+  defp assert_valid_scheme!(scheme) do
+    raise "Invalid scheme in DATABASE_URL: #{inspect(scheme)}"
+  end
+
+  defp assert_valid_host!(str) do
+    if is_binary(str) and String.trim(str) != "" do
+      :ok
+    else
+      raise "Missing host in DATABASE_URL"
+    end
+  end
+
+  defp parse_userinfo!(str) do
+    try do
+      true = is_binary(str)
+
+      {username, password} =
+        case String.split(str, ":") do
+          [username] -> {username, nil}
+          [username, password] -> {username, password}
+        end
+
+      false = String.trim(username) == ""
+
+      {username, password}
+    rescue
+      _ -> raise "Invalid username or password in DATABASE_URL: #{inspect(str)}"
+    end
+  end
+
+  defp parse_database(nil, username), do: username
+  defp parse_database("/", username), do: username
+  defp parse_database("/" <> dbname, _username), do: dbname
+
+  defp add_replication_param(params, nil), do: params
+
+  defp add_replication_param(params, str) when is_binary(str) do
+    case String.downcase(str) do
+      off when off in ~w[false off no 0] -> params
+      "database" -> params ++ [replication: "database"]
+      other -> raise "Unsupported replication mode in DATABASE_URL: #{inspect(other)}"
+    end
+  end
+
+  defp add_ssl_param(params, nil), do: params
+
+  defp add_ssl_param(params, str) when is_binary(str) do
+    if String.downcase(str) == "disable" or String.trim(str) == "" do
+      params
+    else
+      params ++ [ssl: true]
+    end
+  end
 end

components/electric/mix.exs

@@ -42,7 +42,6 @@ defmodule Electric.MixProject do
       {:backoff, "~> 1.1"},
       {:mox, "~> 1.0.2"},
       {:mock, "~> 0.3.0", only: :test},
-      {:postgresql_uri, "~> 0.1.0"},
       {:ssl_verify_fun, "~> 1.1.7", override: true},
       {:jason, "~> 1.4"},
       {:dialyxir, "~> 1.3", only: [:dev], runtime: false},

components/electric/mix.lock

@@ -42,7 +42,6 @@
   "pg_query_ex": {:git, "https://github.com/electric-sql/pg_query_ex.git", "fee6dc748deb80e3fc3c6f77d5fa77faef91336d", []},
   "plug": {:hex, :plug, "1.14.2", "cff7d4ec45b4ae176a227acd94a7ab536d9b37b942c8e8fa6dfc0fff98ff4d80", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "842fc50187e13cf4ac3b253d47d9474ed6c296a8732752835ce4a86acdf68d13"},
   "plug_crypto": {:hex, :plug_crypto, "1.2.5", "918772575e48e81e455818229bf719d4ab4181fcbf7f85b68a35620f78d89ced", [:mix], [], "hexpm", "26549a1d6345e2172eb1c233866756ae44a9609bd33ee6f99147ab3fd87fd842"},
-  "postgresql_uri": {:hex, :postgresql_uri, "0.1.0", "fab3aa8b3e5fe6c4df6d2e80b89a3e99580404b15dde727606e370b74026060b", [:mix], [], "hexpm", "7db308c2eaab0bf7c2864e6bfdd1ed4496f4370ef2f0b778cbe39019b4e0460c"},
   "postgrex": {:hex, :postgrex, "0.17.2", "a3ec9e3239d9b33f1e5841565c4eb200055c52cc0757a22b63ca2d529bbe764c", [:mix], [{:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:table, "~> 0.1.0", [hex: :table, repo: "hexpm", optional: true]}], "hexpm", "80a918a9e9531d39f7bd70621422f3ebc93c01618c645f2d91306f50041ed90c"},
   "protox": {:hex, :protox, "1.7.2", "ed800ae523fa7b7e68a411645922c8061daacfa57aacff2a38d63aa092304a8c", [:mix], [{:decimal, "~> 1.9 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "02230413186776adac7d05237e17e4d20e69b270ae086fab6f1fffda8094d9b0"},
   "req": {:hex, :req, "0.4.3", "bb4cd1661a234b9c779b984dd137761f7ff705f45d0008ba40c8f420a4307b43", [:mix], [{:brotli, "~> 0.3.1", [hex: :brotli, repo: "hexpm", optional: true]}, {:ezstd, "~> 1.0", [hex: :ezstd, repo: "hexpm", optional: true]}, {:finch, "~> 0.9", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}, {:mime, "~> 1.6 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:nimble_csv, "~> 1.0", [hex: :nimble_csv, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: true]}], "hexpm", "9bc88c84f101cfe884260d3413b72aaad6d94ccedccc1f2bcef8e94bd68c5536"},