diff --git a/lib/excache/command/streams.ex b/lib/excache/command/streams.ex
new file mode 100644
index 0000000..66f2543
--- /dev/null
+++ b/lib/excache/command/streams.ex
@@ -0,0 +1,375 @@
+defmodule Excache.Command.Streams do
+  @moduledoc false
+
+  alias Excache.{Store, Stream}
+  alias Excache.Command.Reply
+
+  def dispatch("XADD", args, state), do: xadd(args, state)
+  def dispatch("XLEN", [key], state), do: reply({:integer, orZero(Store.xlen(key))}, state)
+  def dispatch("XRANGE", args, state), do: xrange(args, state, :forward)
+  def dispatch("XREVRANGE", args, state), do: xrange(args, state, :reverse)
+  def dispatch("XDEL", [key | [_ | _] = ids], state), do: xdel(key, ids, state)
+  def dispatch("XTRIM", args, state), do: xtrim(args, state)
+  def dispatch("XSETID", args, state), do: xsetid(args, state)
+  def dispatch("XREAD", args, state), do: xread(args, state)
+  def dispatch("XINFO", args, state), do: xinfo(args, state)
+  def dispatch("XGROUP", args, state), do: xgroup(args, state)
+  def dispatch("XREADGROUP", args, state), do: xreadgroup(args, state)
+  def dispatch("XACK", [key, group | [_ | _] = ids], state), do: xack(key, group, ids, state)
+  def dispatch("XPENDING", args, state), do: xpending(args, state)
+  def dispatch("XCLAIM", args, state), do: xclaim(args, state)
+  def dispatch("XAUTOCLAIM", args, state), do: xautoclaim(args, state)
+  def dispatch(_name, _args, _state), do: :unhandled
+
+  def rewriteAddId(args, id) do
+    case parseAdd(args) do
+      {:ok, key, _nomkstream, trim, _idToken, fields} ->
+        [key] ++ trimTokens(trim) ++ [id] ++ fields
+
+      :error ->
+        args
+    end
+  end
+
+  defp xadd(args, state) do
+    case parseAdd(args) do
+      {:ok, key, nomkstream, trim, idToken, fields} ->
+        case Stream.parseAddId(idToken, now()) do
+          {:ok, idSpec} ->
+            case Store.xadd(key, idSpec, fields, trim, nomkstream) do
+              {:ok, nil} -> reply({:bulk, nil}, state)
+              {:ok, id} -> reply({:bulk, id}, state)
+              {:error, reason} -> reply(streamError(reason), state)
+            end
+
+          :error ->
+            reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
+        end
+
+      :error ->
+        reply(Reply.syntax(), state)
+    end
+  end
+
+  defp xrange([key, startRaw, endRaw], state, direction),
+    do: xrangeWith(key, startRaw, endRaw, :all, state, direction)
+
+  defp xrange([key, startRaw, endRaw, count | rest], state, direction) do
+    case {String.upcase(count), rest} do
+      {"COUNT", [countRaw]} ->
+        case Integer.parse(countRaw) do
+          {value, ""} when value >= 0 ->
+            xrangeWith(key, startRaw, endRaw, value, state, direction)
+
+          _ ->
+            reply(Reply.not_integer(), state)
+        end
+
+      _ ->
+        reply(Reply.syntax(), state)
+    end
+  end
+
+  defp xrange(_args, state, _direction), do: reply(Reply.syntax(), state)
+
+  defp xrangeWith(key, startRaw, endRaw, count, state, direction) do
+    with {:ok, startBound} <- Stream.parseRangeStart(startRaw),
+         {:ok, endBound} <- Stream.parseRangeEnd(endRaw) do
+      result =
+        case direction do
+          :forward -> Store.xrange(key, startBound, endBound, count)
+          :reverse -> Store.xrevrange(key, endBound, startBound, count)
+        end
+
+      case result do
+        {:ok, entries} -> reply({:array, Enum.map(entries, &entryReply/1)}, state)
+        {:error, reason} -> reply(streamError(reason), state)
+      end
+    else
+      :error ->
+        reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
+    end
+  end
+
+  defp xdel(key, idsRaw, state) do
+    case parseIds(idsRaw) do
+      {:ok, ids} ->
+        case Store.xdel(key, ids) do
+          {:ok, removed} -> reply({:integer, removed}, state)
+          {:error, reason} -> reply(streamError(reason), state)
+        end
+
+      :error ->
+        reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
+    end
+  end
+
+  defp xtrim([key | rest], state) do
+    case parseTrim(rest) do
+      {:ok, {trim, _leftover}} when trim != nil ->
+        case Store.xtrim(key, trim) do
+          {:ok, removed} -> reply({:integer, removed}, state)
+          {:error, reason} -> reply(streamError(reason), state)
+        end
+
+      _ ->
+        reply(Reply.syntax(), state)
+    end
+  end
+
+  defp xtrim(_args, state), do: reply(Reply.syntax(), state)
+
+  defp xsetid([key, idRaw | _opts], state) do
+    case parseThresholdId(idRaw) do
+      {:ok, id} ->
+        case Store.xsetid(key, id) do
+          {:ok, "OK"} -> reply({:simple, "OK"}, state)
+          {:error, :nokey} -> reply({:error, "ERR no such key"}, state)
+          {:error, :setid_smaller} -> reply({:error, setidSmallerMessage()}, state)
+          {:error, reason} -> reply(streamError(reason), state)
+        end
+
+      :error ->
+        reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
+    end
+  end
+
+  defp xsetid(_args, state), do: reply(Reply.syntax(), state)
+
+  defp xread(args, state) do
+    case parseRead(args) do
+      {:ok, count, pairs} ->
+        collected =
+          Enum.reduce(pairs, [], fn {key, idRaw}, acc ->
+            afterId = resolveReadId(key, idRaw)
+
+            case Store.xread(key, afterId, count) do
+              {:ok, nil} -> acc
+              {:ok, entries} -> [streamEntriesReply(key, entries) | acc]
+              {:error, _} -> acc
+            end
+          end)
+          |> Enum.reverse()
+
+        case collected do
+          [] -> reply({:array, nil}, state)
+          results -> reply({:array, results}, state)
+        end
+
+      :error ->
+        reply(Reply.syntax(), state)
+    end
+  end
+
+  defp xreadgroup(args, state) do
+    case parseReadGroup(args) do
+      {:ok, group, consumer, count, noack, pairs} ->
+        collected =
+          Enum.reduce_while(pairs, {:ok, []}, fn {key, idRaw}, {:ok, acc} ->
+            mode = if idRaw == ">", do: {:new}, else: {:history, parseThresholdBang(idRaw)}
+
+            case Store.xreadgroup(key, group, consumer, mode, count, noack) do
+              {:ok, []} -> {:cont, {:ok, acc}}
+              {:ok, entries} -> {:cont, {:ok, [streamEntriesReply(key, entries) | acc]}}
+              {:error, :nogroup} -> {:halt, {:error, nogroup(key, group)}}
+              {:error, reason} -> {:halt, {:error, streamError(reason)}}
+            end
+          end)
+
+        case collected do
+          {:ok, []} -> reply({:array, nil}, state)
+          {:ok, results} -> reply({:array, Enum.reverse(results)}, state)
+          {:error, term} -> reply(term, state)
+        end
+
+      :error ->
+        reply(Reply.syntax(), state)
+    end
+  end
+
+  defp xinfo(["STREAM", key | _], state) do
+    case Store.xinfoStream(key) do
+      {:ok, info} -> reply({:array, infoReply(info)}, state)
+      {:error, :nokey} -> reply({:error, "ERR no such key"}, state)
+      {:error, reason} -> reply(streamError(reason), state)
+    end
+  end
+
+  defp xinfo(_args, state), do: reply(Reply.syntax(), state)
+
+  defp xgroup(["CREATE", key, group, startRaw | opts], state) do
+    mkstream = Enum.any?(opts, &(String.upcase(&1) == "MKSTREAM"))
+
+    case parseGroupStart(startRaw) do
+      {:ok, startSpec} ->
+        case Store.xgroupCreate(key, group, startSpec, mkstream) do
+          {:ok, "OK"} -> reply({:simple, "OK"}, state)
+          {:error, :busygroup} -> reply({:error, "BUSYGROUP Consumer Group name already exists"}, state)
+          {:error, :nokey} -> reply({:error, noKeyGroupMessage()}, state)
+          {:error, reason} -> reply(streamError(reason), state)
+        end
+
+      :error ->
+        reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
+    end
+  end
+
+  defp xgroup(["DESTROY", key, group], state) do
+    case Store.xgroupDestroy(key, group) do
+      {:ok, removed} -> reply({:integer, removed}, state)
+      {:error, reason} -> reply(streamError(reason), state)
+    end
+  end
+
+  defp xgroup(["CREATECONSUMER", key, group, consumer], state) do
+    case Store.xgroupCreateConsumer(key, group, consumer) do
+      {:ok, added} -> reply({:integer, added}, state)
+      {:error, :nogroup} -> reply(nogroup(key, group), state)
+      {:error, reason} -> reply(streamError(reason), state)
+    end
+  end
+
+  defp xgroup(["DELCONSUMER", key, group, consumer], state) do
+    case Store.xgroupDelConsumer(key, group, consumer) do
+      {:ok, removed} -> reply({:integer, removed}, state)
+      {:error, :nogroup} -> reply(nogroup(key, group), state)
+      {:error, reason} -> reply(streamError(reason), state)
+    end
+  end
+
+  defp xgroup(["SETID", key, group, idRaw | _opts], state) do
+    case parseGroupStart(idRaw) do
+      {:ok, startSpec} ->
+        id =
+          case startSpec do
+            :last -> lastIdOf(key)
+            {:id, value} -> value
+          end
+
+        case Store.xgroupSetId(key, group, id) do
+          {:ok, "OK"} -> reply({:simple, "OK"}, state)
+          {:error, :nogroup} -> reply(nogroup(key, group), state)
+          {:error, reason} -> reply(streamError(reason), state)
+        end
+
+      :error ->
+        reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
+    end
+  end
+
+  defp xgroup(_args, state), do: reply(Reply.syntax(), state)
+
+  defp xack(key, group, idsRaw, state) do
+    case parseIds(idsRaw) do
+      {:ok, ids} ->
+        case Store.xack(key, group, ids) do
+          {:ok, count} -> reply({:integer, count}, state)
+          {:error, reason} -> reply(streamError(reason), state)
+        end
+
+      :error ->
+        reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
+    end
+  end
+
+  defp xpending([key, group], state) do
+    case Store.xpendingSummary(key, group) do
+      {:ok, summary} -> reply(pendingSummaryReply(summary), state)
+      {:error, :nogroup} -> reply(nogroup(key, group), state)
+      {:error, reason} -> reply(streamError(reason), state)
+    end
+  end
+
+  defp xpending([key, group | rest], state) do
+    case parsePendingRange(rest) do
+      {:ok, startRaw, endRaw, count, consumer, minIdle} ->
+        with {:ok, startId} <- pendingBound(startRaw, {0, 0}),
+             {:ok, endId} <- pendingBound(endRaw, {maxMs(), maxSeq()}) do
+          case Store.xpendingRange(key, group, startId, endId, count, consumer, minIdle) do
+            {:ok, rows} -> reply({:array, Enum.map(rows, &pendingRowReply/1)}, state)
+            {:error, :nogroup} -> reply(nogroup(key, group), state)
+            {:error, reason} -> reply(streamError(reason), state)
+          end
+        else
+          :error ->
+            reply({:error, "ERR Invalid stream ID specified as stream command argument"}, state)
+        end
+
+      :error ->
+        reply(Reply.syntax(), state)
+    end
+  end
+
+  defp xpending(_args, state), do: reply(Reply.syntax(), state)
+
+  defp xclaim([key, group, consumer, minIdleRaw | rest], state) do
+    with {minIdle, ""} <- Integer.parse(minIdleRaw),
+         {:ok, ids, opts} <- parseClaim(rest) do
+      case Store.xclaim(key, group, consumer, minIdle, ids, opts) do
+        {:ok, results} -> reply(claimReply(results, opts.justId), state)
+        {:error, :nogroup} -> reply(nogroup(key, group), state)
+        {:error, reason} -> reply(streamError(reason), state)
+      end
+    else
+      _ -> reply(Reply.syntax(), state)
+    end
+  end
+
+  defp xclaim(_args, state), do: reply(Reply.syntax(), state)
+
+  defp xautoclaim([key, group, consumer, minIdleRaw, startRaw | rest], state) do
+    with {minIdle, ""} <- Integer.parse(minIdleRaw),
+         {:ok, startId} <- Stream.parseRangeStart(startRaw),
+         {:ok, count, justId} <- parseAutoclaim(rest) do
+      {_kind, startTuple} = startId
+
+      case Store.xautoclaim(key, group, consumer, minIdle, startTuple, count) do
+        {:ok, {cursor, claimed, deleted}} ->
+          reply(autoclaimReply(cursor, claimed, deleted, justId), state)
+
+        {:error, :nogroup} ->
+          reply(nogroup(key, group), state)
+
+        {:error, reason} ->
+          reply(streamError(reason), state)
+      end
+    else
+      _ -> reply(Reply.syntax(), state)
+    end
+  end
+
+  defp xautoclaim(_args, state), do: reply(Reply.syntax(), state)
+
+  defp parseAdd([key | rest]), do: parseAddOptions(rest, key, false, nil)
+  defp parseAdd(_args), do: :error
+
+  defp parseAddOptions([token | rest], key, nomkstream, trim) do
+    case String.upcase(token) do
+      "NOMKSTREAM" ->
+        parseAddOptions(rest, key, true, trim)
+
+      "MAXLEN" ->
+        case parseTrimBody(rest, :maxlen) do
+          {:ok, parsed, remaining} -> parseAddOptions(remaining, key, nomkstream, parsed)
+          :error -> :error
+        end
+
+      "MINID" ->
+        case parseTrimBody(rest, :minid) do
+          {:ok, parsed, remaining} -> parseAddOptions(remaining, key, nomkstream, parsed)
+          :error -> :error
+        end
+
+      _ ->
+        case rest do
+          [_ | _] = fields ->
+            if rem(length(fields), 2) == 0, do: {:ok, key, nomkstream, trim, token, fields}, else: :error
+
+          [] ->
+            :error
+        end
+    end
+  end
+
+  defp parseAddOptions([], _key, _nomkstream, _trim), do: :error
+