Feature/nebulex-backend (WIP)

.tool-versions

@@ -0,0 +1,2 @@
+elixir 1.13.4
+erlang 24.3.4.2

lib/phoenix_live_session.ex

@@ -85,6 +85,7 @@ defmodule PhoenixLiveSession do
   @default_table :phoenix_live_sessions
   @default_lifetime 48 * 60 * 60_000
   @default_clean_interval 60_000
+  @default_strategy PhoenixLiveSession.Strategy.ETS
   @max_tries 100
 
   #
@@ -98,56 +99,45 @@ defmodule PhoenixLiveSession do
 
   def get(_conn, sid, opts) do
     table = Keyword.fetch!(opts, :table)
-
+    cache = Keyword.fetch!(opts, :strategy)
     maybe_clean(opts)
 
-    case :ets.lookup(table, sid) do
-      [{^sid, data, _expires_at}] ->
-        :ets.update_element(table, sid, {3, expires_at(opts)})
+    case cache.get(table, sid, opts) do
+      {:ok, data, _expires_at} ->
         {sid, put_meta(data, sid, opts)}
 
-      [] ->
+      nil ->
         {nil, %{}}
     end
   end
 
-  defp clean(table, opts) do
-    lifetime = Keyword.fetch!(opts, :lifetime)
-    now = DateTime.utc_now()
-
-    cutoff =
-      now
-      |> DateTime.add(-1 * lifetime, :millisecond)
-      |> DateTime.to_unix()
-
-    :ets.select_delete(table, [{{:_, :_, :"$1"}, [{:<, :"$1", cutoff}], [true]}])
-    :ets.insert(table, {"last_clean", nil, DateTime.to_unix(now)})
-  end
-
   def put(_conn, nil, data, opts) do
     put_new(data, opts)
   end
 
   def put(_conn, sid, data, opts) do
     table = Keyword.fetch!(opts, :table)
-    :ets.insert(table, {sid, data, expires_at(opts)})
+    cache = Keyword.fetch!(opts, :strategy)
+    cache.put(table, sid, data, expires_at(opts), opts)
     broadcast_update(sid, data, opts)
     sid
   end
 
   def delete(_conn, sid, opts) do
     table = Keyword.fetch!(opts, :table)
+    cache = Keyword.fetch!(opts, :strategy)
     broadcast_update(sid, %{}, opts)
-    :ets.delete(table, sid)
+    cache.delete(table, sid, opts)
     :ok
   end
 
   defp put_new(data, opts, counter \\ 0)
        when counter < @max_tries do
     table = Keyword.fetch!(opts, :table)
+    cache = Keyword.fetch!(opts, :strategy)
     sid = Base.encode64(:crypto.strong_rand_bytes(96))
 
-    if :ets.insert_new(table, {sid, data, expires_at(opts)}) do
+    if cache.put_new(table, sid, data, expires_at(opts), opts) do
       broadcast_update(sid, data, opts)
       sid
     else
@@ -157,16 +147,17 @@ defmodule PhoenixLiveSession do
 
   defp put_in(sid, key, value, opts) do
     table = Keyword.fetch!(opts, :table)
+    cache = Keyword.fetch!(opts, :strategy)
 
-    case :ets.lookup(table, sid) do
-      [{^sid, data, _expires_at}] ->
+    case cache.get(table, sid, opts) do
+      {:ok, data, _expires_at} ->
         updated_data = Map.put(data, key, value)
-        :ets.update_element(table, sid, {2, updated_data})
-        :ets.update_element(table, sid, {3, expires_at(opts)})
+        # Nebulex only has a :ets.put_in equivalent if using the Local caching strategy
+        cache.put(table, sid, updated_data, expires_at(opts), opts)
         broadcast_update(sid, updated_data, opts)
         sid
 
-      [] ->
+      nil ->
         put(nil, sid, %{key => value}, opts)
     end
   end
@@ -176,6 +167,7 @@ defmodule PhoenixLiveSession do
     |> Keyword.put_new(:table, @default_table)
     |> Keyword.put_new(:lifetime, @default_lifetime)
     |> Keyword.put_new(:clean_interval, @default_clean_interval)
+    |> Keyword.put_new(:strategy, @default_strategy)
   end
 
   defp put_meta(data, sid, opts) do
@@ -186,17 +178,18 @@ defmodule PhoenixLiveSession do
 
   defp maybe_clean(opts) do
     table = Keyword.fetch!(opts, :table)
+    cache = Keyword.fetch!(opts, :strategy)
     clean_interval = Keyword.fetch!(opts, :clean_interval)
     latest_possible_clean = DateTime.utc_now() |> DateTime.add(-1 * clean_interval, :millisecond)
 
-    case :ets.lookup(table, "last_clean") do
-      [{"last_clean", _, last_clean}] ->
+    case cache.get(table, "last_clean", opts) do
+      {:ok, _data, last_clean} ->
         if latest_possible_clean > last_clean do
-          clean(table, opts)
+          cache.clean_expired(table, opts)
         end
 
-      [] ->
-        clean(table, opts)
+      nil ->
+        cache.clean_expired(table, opts)
     end
   end
 

lib/strategy/ets.ex

@@ -0,0 +1,61 @@
+defmodule PhoenixLiveSession.Strategy.ETS do
+  # Application.get_env(:routific, :api_key)
+  @moduledoc """
+  Session store that uses ETS
+
+  ## Options
+    * `:pub_sub` - Required.- Module for handling PubSub (e.g. `MyApp.PubSub`).
+    * `:table` - ETS table name. Defaults to `:phoenix_live_sessions`
+    * `:lifetime` - Lifetime (in ms) of sessions before they are cleared.
+       Reads and writes refresh session lifetime. Defaults to two days.
+    * `clean_interval` - Interval (in ms) after which expired PhoenixLiveSession
+      are cleared. Defaulst to 60 seconds.
+
+  ## Caveats
+  Since sessions are stored in memory, they will be lost when restarting
+  your server and are not shared between servers in multi-node setups.
+  """
+
+  alias PhoenixLiveSession.Strategy
+  @behaviour Strategy
+
+  @impl Strategy
+  def get(table, sid, _opts \\ []) do
+    case :ets.lookup(table, sid) do
+      [{^sid, data, expires_at}] ->
+        {:ok, data, expires_at}
+
+      [] ->
+        nil
+    end
+  end
+
+  @impl Strategy
+  def put(table, sid, data, expires_at, _opts \\ []) do
+    :ets.insert(table, {sid, data, expires_at})
+  end
+
+  @impl Strategy
+  def put_new(table, sid, data, expires_at, _opts \\ []) do
+    :ets.insert_new(table, {sid, data, expires_at})
+  end
+
+  @impl Strategy
+  def delete(table, sid, _opts \\ []) do
+    :ets.delete(table, sid)
+  end
+
+  @impl Strategy
+  def clean_expired(table, opts) do
+    lifetime = Keyword.fetch!(opts, :lifetime)
+    now = DateTime.utc_now()
+
+    cutoff =
+      now
+      |> DateTime.add(-1 * lifetime, :millisecond)
+      |> DateTime.to_unix()
+
+    :ets.select_delete(table, [{{:_, :_, :"$1"}, [{:<, :"$1", cutoff}], [true]}])
+    :ets.insert(table, {"last_clean", nil, DateTime.to_unix(now)})
+  end
+end

lib/strategy/nebulex.ex

@@ -0,0 +1,46 @@
+defmodule PhoenixLiveSession.Strategy.Nebulex do
+  @moduledoc """
+  Session store that uses Nebulex as a distributed storage.
+  """
+  alias PhoenixLiveSession.Strategy
+  @behaviour Strategy
+
+  @impl Strategy
+  def get(table, sid, opts) do
+    cache = Keyword.fetch!(opts, :nebulex_cache)
+
+    with {data, expires_at} <- cache.get({table, sid}) do
+      {:ok, data, expires_at}
+    else
+      true ->
+        cache.delete({table, sid})
+        nil
+
+      nil ->
+        nil
+    end
+  end
+
+  @impl Strategy
+  def clean_expired(_table, _opts) do
+    :ok
+  end
+
+  @impl Strategy
+  def put(table, sid, data, expires_at, opts) do
+    cache = Keyword.fetch!(opts, :nebulex_cache)
+    cache.put({table, sid}, {data, expires_at})
+  end
+
+  @impl Strategy
+  def put_new(table, sid, data, expires_at, opts) do
+    cache = Keyword.fetch!(opts, :nebulex_cache)
+    cache.put_new({table, sid}, {data, expires_at})
+  end
+
+  @impl Strategy
+  def delete(table, sid, opts) do
+    cache = Keyword.fetch!(opts, :nebulex_cache)
+    cache.delete({table, sid})
+  end
+end

lib/strategy/strategy.ex

@@ -0,0 +1,10 @@
+defmodule PhoenixLiveSession.Strategy do
+  @callback get(table :: atom, sid :: any, opts :: any) ::
+              {:ok, data :: term, expires_at :: integer} | nil
+  @callback put(table :: atom, sid :: any, data :: term, expires_at :: integer, opts :: any) ::
+              term
+  @callback put_new(table :: atom, sid :: any, data :: term, expires_at :: integer, opts :: any) ::
+              boolean
+  @callback delete(table :: atom, sid :: any, opts :: any) :: term
+  @callback clean_expired(table :: atom, opts :: term) :: term
+end

mix.exs

@@ -28,7 +28,7 @@ defmodule PhoenixLiveSession.MixProject do
     %{
       description: "In-memory live sessions for LiveViews and Phoenix controllers.",
       licenses: ["Apache-2.0"],
-      links: %{"GitHub" => "https://github.com/pentacent/phoenix_live_session"},
+      links: %{"GitHub" => "https://github.com/pentacent/phoenix_live_session"}
     }
   end
 
@@ -45,7 +45,8 @@ defmodule PhoenixLiveSession.MixProject do
       {:plug, "~> 1.10"},
       {:phoenix_pubsub, "~> 2.0"},
       {:phoenix_live_view, "~> 0.5"},
-      {:ex_doc, ">= 0.0.0", only: :dev, runtime: false}
+      {:ex_doc, ">= 0.0.0", only: :dev, runtime: false},
+      {:nebulex, "~> 2.4"}
     ]
   end
 end

mix.lock

@@ -4,6 +4,7 @@
   "makeup": {:hex, :makeup, "1.0.5", "d5a830bc42c9800ce07dd97fa94669dfb93d3bf5fcf6ea7a0c67b2e0e4a7f26c", [:mix], [{:nimble_parsec, "~> 0.5 or ~> 1.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cfa158c02d3f5c0c665d0af11512fed3fba0144cf1aadee0f2ce17747fba2ca9"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.15.0", "98312c9f0d3730fde4049985a1105da5155bfe5c11e47bdc7406d88e01e4219b", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.1", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "75ffa34ab1056b7e24844c90bfc62aaf6f3a37a15faa76b07bc5eba27e4a8b4a"},
   "mime": {:hex, :mime, "1.6.0", "dabde576a497cef4bbdd60aceee8160e02a6c89250d6c0b29e56c0dfb00db3d2", [:mix], [], "hexpm", "31a1a8613f8321143dde1dafc36006a17d28d02bdfecb9e95a880fa7aabd19a7"},
+  "nebulex": {:hex, :nebulex, "2.4.0", "6c4a3bcd27cc1abb98ef60031ef033388e4ce8fcff31f88d60254d9c427a51f6", [:mix], [{:decorator, "~> 1.4", [hex: :decorator, repo: "hexpm", optional: true]}, {:shards, "~> 1.0", [hex: :shards, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "c3645d36ee9a3622bd0ae99fd32c5ee4bf2d5fb2b00ca0578e06a56b7a0f8545"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.1.0", "3a6fca1550363552e54c216debb6a9e95bd8d32348938e13de5eda962c0d7f89", [:mix], [], "hexpm", "08eb32d66b706e913ff748f11694b17981c0b04a33ef470e33e11b3d3ac8f54b"},
   "phoenix": {:hex, :phoenix, "1.5.9", "a6368d36cfd59d917b37c44386e01315bc89f7609a10a45a22f47c007edf2597", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_html, "~> 2.13 or ~> 3.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:plug, "~> 1.10", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 1.0 or ~> 2.2", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.1.2 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7e4bce20a67c012f1fbb0af90e5da49fa7bf0d34e3a067795703b74aef75427d"},
   "phoenix_html": {:hex, :phoenix_html, "2.14.3", "51f720d0d543e4e157ff06b65de38e13303d5778a7919bcc696599e5934271b8", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "efd697a7fff35a13eeeb6b43db884705cba353a1a41d127d118fda5f90c8e80f"},

test/phoenix_live_session_test.exs → test/strategy/ets/session_test.exs

@@ -8,7 +8,7 @@
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
 
-defmodule PhoenixLiveSessionTest do
+defmodule PhoenixLiveSession.Strategy.ETSTest do
   use ExUnit.Case, async: true
   alias PhoenixLiveSession, as: LiveSession
 

test/strategy/nebulex/session_test.exs

@@ -0,0 +1,97 @@
+# Copyright 2013 Plataformatec.
+# Copyright 2020 Pentacent.
+#
+# You may find a copy of the original file here:
+# https://github.com/elixir-plug/plug/blob/ed0541110749531b1fd89cd2fac102ccef4041dc/test/plug/session/ets_test.exs
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+defmodule PhoenixLiveSession.Strategy.NebulexTest do
+  use ExUnit.Case, async: false
+  alias PhoenixLiveSession, as: LiveSession
+  alias PhoenixLiveSession.Strategy.Nebulex
+
+  @pub_sub PhoenixLiveSessionTestPubSub
+
+  setup_all do
+    Supervisor.start_link([{Phoenix.PubSub, name: @pub_sub}], strategy: :one_for_one)
+
+    :ok
+  end
+
+  @tag :skip
+  test "subscribe to and modify session" do
+    opts = LiveSession.init(pub_sub: @pub_sub, store: Nebulex)
+    LiveSession.put(%{}, "sid", %{}, opts)
+    {"sid", session} = LiveSession.get(%{}, "sid", opts)
+
+    socket =
+      %Phoenix.LiveView.Socket{}
+      |> Map.put(:connected?, true)
+      |> Map.put(:transport_pid, "fake-pid")
+      |> LiveSession.maybe_subscribe(session)
+
+    LiveSession.put_session(socket, "foo", "bar")
+
+    assert {:messages, [message]} = Process.info(self(), :messages)
+    assert {:live_session_updated, %{"foo" => "bar"}} = message
+
+    LiveSession.put_session(socket, "fizz", "buzz")
+
+    assert {:messages, [_, message]} = Process.info(self(), :messages)
+    assert {:live_session_updated, %{"foo" => "bar", "fizz" => "buzz"}} = message
+  end
+
+  @tag :skip
+  test "put and get session" do
+    opts = LiveSession.init(pub_sub: @pub_sub, store: Nebulex)
+
+    assert "sid-foo" = LiveSession.put(%{}, "sid-foo", %{foo: :bar}, opts)
+    assert "sid-bar" = LiveSession.put(%{}, "sid-bar", %{bar: :foo}, opts)
+
+    assert {"sid-foo", %{foo: :bar}} = LiveSession.get(%{}, "sid-foo", opts)
+    assert {"sid-bar", %{bar: :foo}} = LiveSession.get(%{}, "sid-bar", opts)
+
+    assert {nil, %{}} = LiveSession.get(%{}, "sid-unknown", opts)
+  end
+
+  @tag :skip
+  test "delete session" do
+    opts = LiveSession.init(pub_sub: @pub_sub, store: Nebulex)
+
+    LiveSession.put(%{}, "sid-foo", %{foo: :bar}, opts)
+    LiveSession.put(%{}, "sid-bar", %{bar: :foo}, opts)
+
+    LiveSession.delete(%{}, "sid-foo", opts)
+
+    assert {nil, %{}} = LiveSession.get(%{}, "sid-foo", opts)
+    assert {"sid-bar", %{bar: :foo}} = LiveSession.get(%{}, "sid-bar", opts)
+  end
+
+  @tag :skip
+  test "generate new sid" do
+    opts = LiveSession.init(pub_sub: @pub_sub, store: Nebulex)
+    sid = LiveSession.put(%{}, nil, %{}, opts)
+    assert byte_size(sid) == 128
+  end
+
+  @tag :skip
+  test "invalidate sid if unknown" do
+    opts = LiveSession.init(pub_sub: @pub_sub, store: Nebulex)
+    assert {nil, %{}} = LiveSession.get(%{}, "sid-unknown", opts)
+  end
+
+  @tag :skip
+  test "put session data without subscribing" do
+    opts = LiveSession.init(pub_sub: @pub_sub, store: Nebulex)
+    sid = LiveSession.put(nil, nil, %{foo: :bar}, opts)
+    {_, session} = LiveSession.get(nil, sid, opts)
+
+    {_, updated_session} = LiveSession.put_session(session, "fizz", :buzz)
+
+    assert %{"fizz" => :buzz} = updated_session
+    assert %{"fizz" => :buzz} = elem(LiveSession.get(nil, sid, opts), 1)
+  end
+end