Initial commit

.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

.gitignore

@@ -0,0 +1,30 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where third-party dependencies like ExDoc output generated docs.
+/doc/
+
+# Ignore .fetch files in case you like to edit your project deps locally.
+/.fetch
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+srh-*.tar
+
+# Temporary files, for example, from tests.
+/tmp/
+
+.idea/
+
+*.iml

README.md

@@ -0,0 +1,11 @@
+# Serverless Redis HTTP
+A redis connection pooler for serverless applications. This allows your serverless functions to talk to Redis via HTTP,
+while also not having to worry about the Redis max connection limits.
+
+Totally open source so you still own the data - run it with your own Redis server.
+
+I will also offer some kinda SaaS version when it's a bit more done ;) - but still BYOR (bring your own Redis)!
+
+Don't use this yet - it's still in development.
+
+Also, it's totally compatible with the Upstash Redis Client SDKs.

lib/srh.ex

@@ -0,0 +1,21 @@
+defmodule Srh do
+  use Application
+
+  def start(_type, _args) do
+    children = [
+      {GenRegistry, worker_module: Srh.Redis.Client},
+      {
+        Plug.Cowboy,
+        scheme: :http,
+        plug: Srh.Http.BaseRouter,
+        options: [
+          port: 8080
+        ]
+      }
+    ]
+
+    opts = [strategy: :one_for_one, name: Srh.Supervisor]
+
+    Supervisor.start_link(children, opts)
+  end
+end

lib/srh/auth/token_resolver.ex

@@ -0,0 +1,17 @@
+defmodule Srh.Auth.TokenResolver do
+  def resolve(token) do
+    {
+      :ok,
+      Jason.decode!(
+        # This is done to replicate what will eventually be API endpoints, so they keys are not atoms
+        Jason.encode!(
+          %{
+            srh_id: "1000",
+            connection_string: "redis://localhost:6379",
+            max_connections: 10
+          }
+        )
+      )
+    }
+  end
+end

lib/srh/http/base_router.ex

@@ -0,0 +1,43 @@
+defmodule Srh.Http.BaseRouter do
+  use Plug.Router
+  alias Srh.Http.RequestValidator
+  alias Srh.Http.CommandHandler
+
+  plug :match
+  plug Plug.Parsers, parsers: [:json], pass: ["application/json"], json_decoder: Jason
+  plug :dispatch
+
+  get "/" do
+    handle_response({:ok, "Welcome to Serverless Redis HTTP!"}, conn)
+  end
+
+  post "/" do
+    case conn
+         |> get_req_header("authorization")
+         |> RequestValidator.validate_bearer_header()
+      do
+      {:ok, token} ->
+        CommandHandler.handle_command(conn, token)
+      {:error, _} ->
+        {:malformed_data, "Missing/Invalid authorization header"}
+    end
+    |> handle_response(conn)
+  end
+
+  match _ do
+    send_resp(conn, 404, "Endpoint not found")
+  end
+
+  defp handle_response(response, conn) do
+    %{code: code, message: message} =
+      case response do
+        {:ok, data} -> %{code: 200, message: Jason.encode!(data)}
+        {:not_found, message} -> %{code: 404, message: message}
+        {:malformed_data, message} -> %{code: 400, message: message}
+        {:server_error, _} -> %{code: 500, message: "An error occurred internally"}
+      end
+
+    conn
+    |> send_resp(code, message)
+  end
+end

lib/srh/http/command_handler.ex

@@ -0,0 +1,35 @@
+defmodule Srh.Http.CommandHandler do
+  alias Srh.Http.RequestValidator
+  alias Srh.Auth.TokenResolver
+  alias Srh.Redis.Client
+
+  def handle_command(conn, token) do
+    case RequestValidator.validate_redis_body(conn.body_params) do
+      {:ok, command_array} ->
+        IO.inspect(command_array)
+        do_handle_command(command_array, token)
+      {:error, error_message} ->
+        {:malformed_data, error_message}
+    end
+  end
+
+  defp do_handle_command(command_array, token) do
+    case TokenResolver.resolve(token) do
+      {:ok, connection_info} ->
+        dispatch_command(command_array, connection_info)
+      {:error, _} -> {:error, "Invalid token"}
+    end
+  end
+
+  defp dispatch_command(command_array, %{"srh_id" => srh_id, "max_connections" => max_connections} = connection_info)
+       when is_number(max_connections) do
+    case GenRegistry.lookup_or_start(Client, srh_id, [max_connections, connection_info]) do
+      {:ok, pid} ->
+        # Run the command
+        {:ok, res} = Client.redis_command(pid, command_array)
+        {:ok, %{result: res}}
+      {:error, msg} ->
+        {:server_error, msg}
+    end
+  end
+end

lib/srh/http/request_validator.ex

@@ -0,0 +1,24 @@
+defmodule Srh.Http.RequestValidator do
+  def validate_redis_body(%{"_json" => command_array}) when is_list(command_array), do: {:ok, command_array}
+
+  def validate_redis_body(payload),
+      do: {:error, "Invalid command array. Expected a string array at root of the command and its arguments."}
+
+  def validate_bearer_header(header_value_array) when is_list(header_value_array) do
+    do_validate_bearer_header(header_value_array)
+  end
+
+  # any amount of items left
+  defp do_validate_bearer_header([first_item | rest]) do
+    case first_item
+         |> String.split(" ") do
+      ["Bearer", token] ->
+        {:ok, token}
+      _ ->
+        do_validate_bearer_header(rest)
+    end
+  end
+  
+  # no items left
+  defp do_validate_bearer_header([]), do: {:error, :not_found}
+end

lib/srh/redis/client.ex

@@ -0,0 +1,80 @@
+defmodule Srh.Redis.Client do
+  use GenServer
+  alias Srh.Redis.ClientRegistry
+  alias Srh.Redis.ClientWorker
+
+  @idle_death_time 1000 * 3
+
+  def start_link(max_connections, connection_info) do
+    GenServer.start_link(__MODULE__, {max_connections, connection_info}, [])
+  end
+
+  def init({max_connections, connection_info}) do
+    IO.puts("Client starting alive! Srh_id=#{Map.get(connection_info, "srh_id", "not found")}")
+
+    Process.send(self(), :create_registry, [])
+
+    {
+      :ok,
+      %{
+        registry_pid: nil,
+        idle_death_ref: nil,
+        max_connections: max_connections,
+        connection_info: connection_info
+      }
+    }
+  end
+
+  def redis_command(client, command_array)  do
+    GenServer.call(client, {:redis_command, command_array})
+  end
+
+  def handle_call({:redis_command, command_array}, _from, %{registry_pid: registry_pid} = state)
+      when is_pid(registry_pid) do
+    {:ok, worker} = ClientRegistry.find_worker(registry_pid)
+    Process.send(self(), :reset_idle_death, [])
+    {:reply, ClientWorker.redis_command(worker, command_array), state}
+  end
+
+  def handle_call(_msg, _from, state) do
+    {:reply, :ok, state}
+  end
+
+  def handle_cast(_msg, state) do
+    {:noreply, state}
+  end
+
+  def handle_info(:idle_death, state) do
+    IO.puts("Client dying! No requests for period. Srh_id=#{Map.get(state.connection_info, "srh_id", "not found")}")
+    ClientRegistry.destroy_workers(state.registry_pid)
+
+    {:stop, :normal, state}
+  end
+
+  def handle_info(:reset_idle_death, state) do
+    if state.idle_death_ref != nil do
+      Process.cancel_timer(state.idle_death_ref)
+    end
+
+    {
+      :noreply,
+      %{state | idle_death_ref: Process.send_after(self(), :idle_death, @idle_death_time)}
+    }
+  end
+
+  def handle_info(:create_registry, state) do
+    {:ok, pid} = ClientRegistry.start_link()
+
+    # Spin up three workers
+    for _ <- 0..2 do
+      {:ok, worker} = ClientWorker.start_link(state.connection_info)
+      ClientRegistry.add_worker(pid, worker)
+    end
+
+    {:noreply, %{state | registry_pid: pid}}
+  end
+
+  def handle_info(_msg, state) do
+    {:noreply, state}
+  end
+end

lib/srh/redis/client_registry.ex

@@ -0,0 +1,87 @@
+defmodule Srh.Redis.ClientRegistry do
+  use GenServer
+
+  def start_link() do
+    GenServer.start_link(__MODULE__, %{}, [])
+  end
+
+  def init(_opts) do
+    {
+      :ok,
+      %{
+        worker_pids: [],
+        last_worker_index: 0
+      }
+    }
+  end
+
+  def find_worker(registry) do
+    GenServer.call(registry, {:find_worker})
+  end
+
+  def add_worker(registry, pid) do
+    GenServer.cast(registry, {:add_worker, pid})
+  end
+
+  def destroy_workers(registry) do
+    GenServer.cast(registry, {:destroy_workers})
+  end
+
+  def handle_call({:find_worker}, _from, state) do
+    case length(state.worker_pids) do
+      0 ->
+        {:reply, {:error, :none_available}, state}
+
+      len ->
+        target = state.last_worker_index + 1
+
+        corrected_target = case target >= len do
+          true -> 0
+          false -> target
+        end
+
+        IO.puts("Worker target: #{corrected_target}, pool size: #{len}")
+
+        {:reply, {:ok, Enum.at(state.worker_pids, corrected_target)}, %{state | last_worker_index: corrected_target}}
+    end
+  end
+
+  def handle_call(_msg, _from, state) do
+    {:reply, :ok, state}
+  end
+
+  def handle_cast({:add_worker, pid}, state) do
+    Process.monitor(pid)
+
+    {
+      :noreply,
+      %{
+        state
+      |
+        worker_pids:
+          [pid | state.worker_pids]
+          |> Enum.uniq()
+      }
+    }
+  end
+
+  def handle_cast({:destroy_workers}, state) do
+    for worker_pid <- state.worker_pids do
+      Process.exit(worker_pid, :normal)
+    end
+
+    {:noreply, %{state | worker_pids: [], last_worker_index: 0}}
+  end
+
+  def handle_cast(_msg, state) do
+    {:noreply, state}
+  end
+
+  def handle_info({:DOWN, pid, :normal, _ref}, state) do
+    {:noreply, %{state | worker_pids: List.delete(state.worker_pids, pid)}}
+  end
+
+  def handle_info(_msg, state) do
+    {:noreply, state}
+  end
+end

lib/srh/redis/client_worker.ex

@@ -0,0 +1,68 @@
+defmodule Srh.Redis.ClientWorker do
+  use GenServer
+
+  def start_link(connection_info) do
+    GenServer.start_link(__MODULE__, connection_info, [])
+  end
+
+  def init(connection_info) do
+    IO.puts("Client worker reporting for duty! Srh_id=#{Map.get(connection_info, "srh_id", "not found")}")
+
+    Process.send(self(), :create_connection, [])
+
+    {
+      :ok,
+      %{
+        connection_info: connection_info,
+        redix_pid: nil
+      }
+    }
+  end
+
+  def redis_command(worker, command_array) do
+    GenServer.call(worker, {:redis_command, command_array})
+  end
+
+  def handle_call({:redis_command, command_array}, _from, %{redix_pid: redix_pid} = state)
+      when is_pid(redix_pid) do
+    case Redix.command(redix_pid, command_array) do
+      {:ok, res} ->
+        IO.puts("Worker PID:")
+        IO.inspect(self())
+        IO.puts("Worker Redis response:")
+        IO.inspect(res)
+        {:reply, {:ok, res}, state}
+
+      {:error, res} ->
+        {:reply, {:error, res}, state}
+    end
+  end
+
+  def handle_call(msg, _from, state) do
+    IO.inspect(msg)
+    {:reply, :ok, state}
+  end
+
+  def handle_cast(_msg, state) do
+    {:noreply, state}
+  end
+
+  # TODO: Handle host / port connections
+  def handle_info(
+        :create_connection,
+        %{
+          connection_info: %{
+            "connection_string" => connection_string
+          }
+        } = state
+      )
+      when is_binary(connection_string) do
+    {:ok, pid} = Redix.start_link(connection_string)
+    IO.puts("Redis up and running for worker Srh_id=#{Map.get(state.connection_info, "srh_id", "not found")}")
+    {:noreply, %{state | redix_pid: pid}}
+  end
+
+  def handle_info(_msg, state) do
+    {:noreply, state}
+  end
+end

mix.exs

@@ -0,0 +1,34 @@
+defmodule Srh.MixProject do
+  use Mix.Project
+
+  def project do
+    [
+      app: :srh,
+      version: "0.1.0",
+      elixir: "~> 1.13",
+      start_permanent: Mix.env() == :prod,
+      deps: deps()
+    ]
+  end
+
+  # Run "mix help compile.app" to learn about applications.
+  def application do
+    [
+      extra_applications: [:logger],
+      mod: {Srh, []}
+    ]
+  end
+
+  # Run "mix help deps" to learn about dependencies.
+  defp deps do
+    [
+      {:redix, "~> 1.1"},
+      {:castore, ">= 0.0.0"},
+      {:plug, "~> 1.13"},
+      {:cowboy, "~> 2.9"},
+      {:plug_cowboy, "~> 2.5"},
+      {:jason, "~> 1.3"},
+      {:gen_registry, "~> 1.1"}
+    ]
+  end
+end

mix.lock

@@ -0,0 +1,15 @@
+%{
+  "castore": {:hex, :castore, "0.1.17", "ba672681de4e51ed8ec1f74ed624d104c0db72742ea1a5e74edbc770c815182f", [:mix], [], "hexpm", "d9844227ed52d26e7519224525cb6868650c272d4a3d327ce3ca5570c12163f9"},
+  "cowboy": {:hex, :cowboy, "2.9.0", "865dd8b6607e14cf03282e10e934023a1bd8be6f6bacf921a7e2a96d800cd452", [:make, :rebar3], [{:cowlib, "2.11.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "2c729f934b4e1aa149aff882f57c6372c15399a20d54f65c8d67bef583021bde"},
+  "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.4.0", "f239f68b588efa7707abce16a84d0d2acf3a0f50571f8bb7f56a15865aae820c", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7d98bac1ee4565d31b62d59f8823dfd8356a169e7fcbb83831b8a5397404c9de"},
+  "cowlib": {:hex, :cowlib, "2.11.0", "0b9ff9c346629256c42ebe1eeb769a83c6cb771a6ee5960bd110ab0b9b872063", [:make, :rebar3], [], "hexpm", "2b3e9da0b21c4565751a6d4901c20d1b4cc25cbb7fd50d91d2ab6dd287bc86a9"},
+  "gen_registry": {:hex, :gen_registry, "1.1.0", "30d483ae11dee615e500783fcec3f493de77fc80d346ba785947731207f790b7", [:mix], [], "hexpm", "c3db77e15c62599d01dc6b65aec6a3412a99b00d7785ef4a065c0c043b5a4478"},
+  "jason": {:hex, :jason, "1.3.0", "fa6b82a934feb176263ad2df0dbd91bf633d4a46ebfdffea0c8ae82953714946", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "53fc1f51255390e0ec7e50f9cb41e751c260d065dcba2bf0d08dc51a4002c2ac"},
+  "mime": {:hex, :mime, "2.0.2", "0b9e1a4c840eafb68d820b0e2158ef5c49385d17fb36855ac6e7e087d4b1dcc5", [:mix], [], "hexpm", "e6a3f76b4c277739e36c2e21a2c640778ba4c3846189d5ab19f97f126df5f9b7"},
+  "plug": {:hex, :plug, "1.13.6", "187beb6b67c6cec50503e940f0434ea4692b19384d47e5fdfd701e93cadb4cc2", [: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", "02b9c6b9955bce92c829f31d6284bf53c591ca63c4fb9ff81dfd0418667a34ff"},
+  "plug_cowboy": {:hex, :plug_cowboy, "2.5.2", "62894ccd601cf9597e2c23911ff12798a8a18d237e9739f58a6b04e4988899fe", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "ea6e87f774c8608d60c8d34022a7d073bd7680a0a013f049fc62bf35efea1044"},
+  "plug_crypto": {:hex, :plug_crypto, "1.2.2", "05654514ac717ff3a1843204b424477d9e60c143406aa94daf2274fdd280794d", [:mix], [], "hexpm", "87631c7ad914a5a445f0a3809f99b079113ae4ed4b867348dd9eec288cecb6db"},
+  "ranch": {:hex, :ranch, "1.8.0", "8c7a100a139fd57f17327b6413e4167ac559fbc04ca7448e9be9057311597a1d", [:make, :rebar3], [], "hexpm", "49fbcfd3682fab1f5d109351b61257676da1a2fdbe295904176d5e521a2ddfe5"},
+  "redix": {:hex, :redix, "1.1.5", "6fc460d66a5c2287e83e6d73dddc8d527ff59cb4d4f298b41e03a4db8c3b2bd5", [:mix], [{:castore, "~> 0.1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "679afdd4c14502fe9c11387ff1cdcb33065a1cf511097da1eee407f17c7a418b"},
+  "telemetry": {:hex, :telemetry, "1.1.0", "a589817034a27eab11144ad24d5c0f9fab1f58173274b1e9bae7074af9cbee51", [:rebar3], [], "hexpm", "b727b2a1f75614774cff2d7565b64d0dfa5bd52ba517f16543e6fc7efcc0df48"},
+}