feat(store): add core key-value and list operations

lib/excache/store/ops.ex

@@ -0,0 +1,210 @@
+defmodule Excache.Store.Ops do
+  @moduledoc false
+
+  # Every operation runs inside the owning shard process, so ETS access is
+  # serialized per shard and read-modify-write sequences are atomic.
+  # Entry layout: {key, kind, value, expires_at} where kind is one of
+  # :string | :list | :hash | :set | :zset and expires_at is epoch ms or nil.
+
+  alias Excache.SkipList
+  alias Excache.Store.Ops.Helpers
+
+  import Helpers,
+    only: [
+      fetch: 3,
+      collection: 6,
+      with_kind: 6,
+      pop_list: 3,
+      rank_window: 3,
+      slice: 3,
+      at: 2,
+      expired?: 2,
+      expiry_from_opts: 2,
+      remaining: 2,
+      compile_pattern: 1
+    ]
+
+  def apply_op(table, now, {:stream, op}),
+    do: Excache.Store.Ops.Streams.apply_op(table, now, op)
+
+  def apply_op(table, now, {:binary, op}),
+    do: Excache.Store.Ops.Binary.apply_op(table, now, op)
+
+  def apply_op(table, now, {:extra, op}),
+    do: Excache.Store.Ops.Extra.apply_op(table, now, op)
+
+  def apply_op(table, now, {:peek, key}), do: fetch(table, key, now)
+
+  def apply_op(table, now, {:get, key}) do
+    case fetch(table, key, now) do
+      :miss -> nil
+      {:string, value, _ea} -> {:ok, value}
+      _ -> {:error, :wrongtype}
+    end
+  end
+
+  def apply_op(table, now, {:set, key, value, opts}) do
+    existing = fetch(table, key, now)
+    exists = existing != :miss
+
+    cond do
+      Keyword.get(opts, :nx, false) and exists ->
+        nil
+
+      Keyword.get(opts, :xx, false) and not exists ->
+        nil
+
+      true ->
+        :ets.insert(table, {key, :string, value, expiry_from_opts(opts, now)})
+        :ok
+    end
+  end
+
+  def apply_op(table, now, {:del, key}) do
+    case fetch(table, key, now) do
+      :miss ->
+        0
+
+      _ ->
+        :ets.delete(table, key)
+        1
+    end
+  end
+
+  def apply_op(table, now, {:exists, key}) do
+    case fetch(table, key, now) do
+      :miss -> 0
+      _ -> 1
+    end
+  end
+
+  def apply_op(table, now, {:type, key}) do
+    case fetch(table, key, now) do
+      :miss -> "none"
+      {kind, _value, _ea} -> Atom.to_string(kind)
+    end
+  end
+
+  def apply_op(table, now, {:incrby, key, delta}) do
+    case fetch(table, key, now) do
+      :miss ->
+        :ets.insert(table, {key, :string, Integer.to_string(delta), nil})
+        {:ok, delta}
+
+      {:string, value, ea} ->
+        case Integer.parse(value) do
+          {current, ""} ->
+            updated = current + delta
+            :ets.insert(table, {key, :string, Integer.to_string(updated), ea})
+            {:ok, updated}
+
+          _ ->
+            {:error, :not_integer}
+        end
+
+      _ ->
+        {:error, :wrongtype}
+    end
+  end
+
+  def apply_op(table, now, {:append, key, value}) do
+    case fetch(table, key, now) do
+      :miss ->
+        :ets.insert(table, {key, :string, value, nil})
+        {:ok, byte_size(value)}
+
+      {:string, current, ea} ->
+        updated = current <> value
+        :ets.insert(table, {key, :string, updated, ea})
+        {:ok, byte_size(updated)}
+
+      _ ->
+        {:error, :wrongtype}
+    end
+  end
+
+  def apply_op(table, now, {:getset, key, value}) do
+    previous =
+      case fetch(table, key, now) do
+        :miss -> {:ok, nil}
+        {:string, current, _ea} -> {:ok, current}
+        _ -> {:error, :wrongtype}
+      end
+
+    case previous do
+      {:ok, prior} ->
+        :ets.insert(table, {key, :string, value, nil})
+        {:ok, prior}
+
+      error ->
+        error
+    end
+  end
+
+  def apply_op(table, now, {:expire, key, ms}) do
+    case fetch(table, key, now) do
+      :miss ->
+        0
+
+      {kind, value, _ea} ->
+        :ets.insert(table, {key, kind, value, now + ms})
+        1
+    end
+  end
+
+  def apply_op(table, now, {:ttl, key, unit}) do
+    case fetch(table, key, now) do
+      :miss -> -2
+      {_kind, _value, nil} -> -1
+      {_kind, _value, ea} -> remaining(ea - now, unit)
+    end
+  end
+
+  def apply_op(table, now, {:persist, key}) do
+    case fetch(table, key, now) do
+      :miss ->
+        0
+
+      {_kind, _value, nil} ->
+        0
+
+      {kind, value, _ea} ->
+        :ets.insert(table, {key, kind, value, nil})
+        1
+    end
+  end
+
+  def apply_op(table, now, {:lpush, key, values}) do
+    collection(table, key, now, :list, [], fn current ->
+      updated = Enum.reverse(values) ++ current
+      {updated, {:ok, length(updated)}}
+    end)
+  end
+
+  def apply_op(table, now, {:rpush, key, values}) do
+    collection(table, key, now, :list, [], fn current ->
+      updated = current ++ values
+      {updated, {:ok, length(updated)}}
+    end)
+  end
+
+  def apply_op(table, now, {:lpop, key, count}) do
+    collection(table, key, now, :list, [], fn current -> pop_list(current, :left, count) end)
+  end
+
+  def apply_op(table, now, {:rpop, key, count}) do
+    collection(table, key, now, :list, [], fn current -> pop_list(current, :right, count) end)
+  end
+
+  def apply_op(table, now, {:lrange, key, start, stop}) do
+    with_kind(table, key, now, :list, [], fn list -> {:ok, slice(list, start, stop)} end)
+  end
+
+  def apply_op(table, now, {:llen, key}) do
+    with_kind(table, key, now, :list, [], fn list -> {:ok, length(list)} end)
+  end
+
+  def apply_op(table, now, {:lindex, key, index}) do
+    with_kind(table, key, now, :list, [], fn list -> at(list, index) end)
+  end
+