feat(command): add stream argument parsing and reply builders

lib/excache/command/streams.ex

@@ -373,3 +373,400 @@ defmodule Excache.Command.Streams do
 
   defp parseAddOptions([], _key, _nomkstream, _trim), do: :error
 
+  defp parseTrim([]), do: {:ok, {nil, []}}
+
+  defp parseTrim([token | rest]) do
+    case String.upcase(token) do
+      "MAXLEN" ->
+        case parseTrimBody(rest, :maxlen) do
+          {:ok, parsed, remaining} -> {:ok, {parsed, remaining}}
+          :error -> :error
+        end
+
+      "MINID" ->
+        case parseTrimBody(rest, :minid) do
+          {:ok, parsed, remaining} -> {:ok, {parsed, remaining}}
+          :error -> :error
+        end
+
+      _ ->
+        :error
+    end
+  end
+
+  defp parseTrimBody(tokens, kind) do
+    tokens = dropApproxMarker(tokens)
+
+    case tokens do
+      [threshold | rest] ->
+        rest = dropLimit(rest)
+
+        case buildTrim(kind, threshold) do
+          {:ok, parsed} -> {:ok, parsed, rest}
+          :error -> :error
+        end
+
+      [] ->
+        :error
+    end
+  end
+
+  defp dropApproxMarker([marker | rest]) when marker in ["~", "="], do: rest
+  defp dropApproxMarker(tokens), do: tokens
+
+  defp dropLimit(["LIMIT", _value | rest]), do: rest
+  defp dropLimit(["limit", _value | rest]), do: rest
+  defp dropLimit(tokens), do: tokens
+
+  defp buildTrim(:maxlen, threshold) do
+    case Integer.parse(threshold) do
+      {value, ""} when value >= 0 -> {:ok, {:maxlen, value}}
+      _ -> :error
+    end
+  end
+
+  defp buildTrim(:minid, threshold) do
+    case parseThresholdId(threshold) do
+      {:ok, id} -> {:ok, {:minid, id}}
+      :error -> :error
+    end
+  end
+
+  defp trimTokens(nil), do: []
+  defp trimTokens({:maxlen, count}), do: ["MAXLEN", Integer.to_string(count)]
+  defp trimTokens({:minid, {ms, seq}}), do: ["MINID", "#{ms}-#{seq}"]
+
+  defp parseRead(args) do
+    {count, rest} = takeReadCount(args, :all)
+
+    case splitStreams(rest) do
+      {:ok, keys, ids} when length(keys) == length(ids) and keys != [] ->
+        {:ok, count, Enum.zip(keys, ids)}
+
+      _ ->
+        :error
+    end
+  end
+
+  defp parseReadGroup(["GROUP", group, consumer | rest]) do
+    {count, rest} = takeReadCount(rest, :all)
+    {noack, rest} = takeNoack(rest)
+
+    case splitStreams(rest) do
+      {:ok, keys, ids} when length(keys) == length(ids) and keys != [] ->
+        {:ok, group, consumer, count, noack, Enum.zip(keys, ids)}
+
+      _ ->
+        :error
+    end
+  end
+
+  defp parseReadGroup(_args), do: :error
+
+  defp takeReadCount(["COUNT", raw | rest], _default) do
+    case Integer.parse(raw) do
+      {value, ""} when value >= 0 -> {value, rest}
+      _ -> {:all, rest}
+    end
+  end
+
+  defp takeReadCount([token, raw | rest], default) when token in ["count"] do
+    takeReadCount(["COUNT", raw | rest], default)
+  end
+
+  defp takeReadCount(rest, default), do: {default, rest}
+
+  defp takeNoack(["NOACK" | rest]), do: {true, rest}
+  defp takeNoack(["noack" | rest]), do: {true, rest}
+  defp takeNoack(rest), do: {false, rest}
+
+  defp splitStreams(tokens) do
+    case Enum.split_while(tokens, &(String.upcase(&1) != "STREAMS")) do
+      {leading, ["STREAMS" | rest]} ->
+        # any leading BLOCK ms is tolerated and ignored here
+        _ = leading
+        half = div(length(rest), 2)
+
+        if rem(length(rest), 2) == 0 and half > 0 do
+          {:ok, Enum.take(rest, half), Enum.drop(rest, half)}
+        else
+          :error
+        end
+
+      _ ->
+        :error
+    end
+  end
+
+  defp resolveReadId(key, "$"), do: lastIdOf(key)
+  defp resolveReadId(_key, "+"), do: {maxMs(), maxSeq()}
+
+  defp resolveReadId(_key, raw) do
+    case parseThresholdId(raw) do
+      {:ok, id} -> id
+      :error -> {0, 0}
+    end
+  end
+
+  defp lastIdOf(key) do
+    case Store.xinfoStream(key) do
+      {:ok, info} -> info.lastGeneratedId
+      _ -> {0, 0}
+    end
+  end
+
+  defp parseGroupStart("$"), do: {:ok, :last}
+  defp parseGroupStart("0"), do: {:ok, {:id, {0, 0}}}
+
+  defp parseGroupStart(raw) do
+    case parseThresholdId(raw) do
+      {:ok, id} -> {:ok, {:id, id}}
+      :error -> :error
+    end
+  end
+
+  defp parsePendingRange(["IDLE", idleRaw | rest]) do
+    case Integer.parse(idleRaw) do
+      {idle, ""} -> withPendingCore(rest, idle)
+      _ -> :error
+    end
+  end
+
+  defp parsePendingRange(rest), do: withPendingCore(rest, nil)
+
+  defp withPendingCore([startRaw, endRaw, countRaw | rest], minIdle) do
+    case Integer.parse(countRaw) do
+      {count, ""} when count >= 0 ->
+        consumer = List.first(rest)
+        {:ok, startRaw, endRaw, count, consumer, minIdle}
+
+      _ ->
+        :error
+    end
+  end
+
+  defp withPendingCore(_rest, _minIdle), do: :error
+
+  defp pendingBound("-", _default), do: {:ok, {0, 0}}
+  defp pendingBound("+", _default), do: {:ok, {maxMs(), maxSeq()}}
+
+  defp pendingBound(raw, _default) do
+    case parseThresholdId(raw) do
+      {:ok, id} -> {:ok, id}
+      :error -> :error
+    end
+  end
+
+  defp parseClaim(tokens), do: parseClaimIds(tokens, [])
+
+  defp parseClaimIds([token | rest], acc) do
+    case parseThresholdId(token) do
+      {:ok, id} ->
+        parseClaimIds(rest, [id | acc])
+
+      :error ->
+        case acc do
+          [] -> :error
+          _ -> parseClaimOpts([token | rest], Enum.reverse(acc), defaultClaimOpts())
+        end
+    end
+  end
+
+  defp parseClaimIds([], acc) when acc != [], do: {:ok, Enum.reverse(acc), defaultClaimOpts()}
+  defp parseClaimIds([], _acc), do: :error
+
+  defp parseClaimOpts([], ids, opts), do: {:ok, ids, opts}
+
+  defp parseClaimOpts([token | rest], ids, opts) do
+    case String.upcase(token) do
+      "IDLE" -> claimIntOpt(rest, ids, opts, :idle)
+      "TIME" -> claimIntOpt(rest, ids, opts, :time)
+      "RETRYCOUNT" -> claimIntOpt(rest, ids, opts, :retryCount)
+      "FORCE" -> parseClaimOpts(rest, ids, %{opts | force: true})
+      "JUSTID" -> parseClaimOpts(rest, ids, %{opts | justId: true})
+      "LASTID" -> parseClaimOpts(tl(rest), ids, opts)
+      _ -> :error
+    end
+  end
+
+  defp claimIntOpt([raw | rest], ids, opts, field) do
+    case Integer.parse(raw) do
+      {value, ""} -> parseClaimOpts(rest, ids, Map.put(opts, field, value))
+      _ -> :error
+    end
+  end
+
+  defp claimIntOpt([], _ids, _opts, _field), do: :error
+
+  defp defaultClaimOpts,
+    do: %{idle: nil, time: nil, retryCount: nil, force: false, justId: false}
+
+  defp parseAutoclaim(["COUNT", raw | rest]) do
+    case Integer.parse(raw) do
+      {value, ""} when value > 0 -> finishAutoclaim(rest, value)
+      _ -> :error
+    end
+  end
+
+  defp parseAutoclaim(rest), do: finishAutoclaim(rest, 100)
+
+  defp finishAutoclaim([], count), do: {:ok, count, false}
+  defp finishAutoclaim(["JUSTID"], count), do: {:ok, count, true}
+  defp finishAutoclaim(["justid"], count), do: {:ok, count, true}
+  defp finishAutoclaim(_rest, _count), do: :error
+
+  defp parseIds(idsRaw) do
+    Enum.reduce_while(idsRaw, {:ok, []}, fn raw, {:ok, acc} ->
+      case parseThresholdId(raw) do
+        {:ok, id} -> {:cont, {:ok, [id | acc]}}
+        :error -> {:halt, :error}
+      end
+    end)
+    |> case do
+      {:ok, ids} -> {:ok, Enum.reverse(ids)}
+      :error -> :error
+    end
+  end
+
+  defp parseThresholdId(raw) do
+    case String.split(raw, "-", parts: 2) do
+      [ms] -> withMs(ms, 0)
+      [ms, seq] -> withMsSeq(ms, seq)
+    end
+  end
+
+  defp parseThresholdBang(raw) do
+    case parseThresholdId(raw) do
+      {:ok, id} -> id
+      :error -> {0, 0}
+    end
+  end
+
+  defp withMs(ms, seq) do
+    case Integer.parse(ms) do
+      {value, ""} when value >= 0 -> {:ok, {value, seq}}
+      _ -> :error
+    end
+  end
+
+  defp withMsSeq(ms, seq) do
+    with {msValue, ""} <- Integer.parse(ms),
+         {seqValue, ""} <- Integer.parse(seq),
+         true <- msValue >= 0 and seqValue >= 0 do
+      {:ok, {msValue, seqValue}}
+    else
+      _ -> :error
+    end
+  end
+
+  defp entryReply({id, fields}) do
+    {:array, [{:bulk, Stream.formatId(id)}, fieldsReply(fields)]}
+  end
+
+  defp fieldsReply(nil), do: {:array, nil}
+  defp fieldsReply(fields), do: {:array, Enum.map(fields, &{:bulk, &1})}
+
+  defp streamEntriesReply(key, entries) do
+    {:array, [{:bulk, key}, {:array, Enum.map(entries, &entryReply/1)}]}
+  end
+
+  defp infoReply(info) do
+    [
+      {:bulk, "length"},
+      {:integer, info.length},
+      {:bulk, "last-generated-id"},
+      {:bulk, Stream.formatId(info.lastGeneratedId)},
+      {:bulk, "max-deleted-entry-id"},
+      {:bulk, Stream.formatId(info.maxDeletedId)},
+      {:bulk, "entries-added"},
+      {:integer, info.entriesAdded},
+      {:bulk, "groups"},
+      {:integer, info.groups},
+      {:bulk, "first-entry"},
+      firstOrNil(info.firstEntry),
+      {:bulk, "last-entry"},
+      firstOrNil(info.lastEntry)
+    ]
+  end
+
+  defp firstOrNil(nil), do: {:array, nil}
+  defp firstOrNil(entry), do: entryReply(entry)
+
+  defp pendingSummaryReply({0, _minId, _maxId, _perConsumer}) do
+    {:array, [{:integer, 0}, {:bulk, nil}, {:bulk, nil}, {:array, nil}]}
+  end
+
+  defp pendingSummaryReply({total, minId, maxId, perConsumer}) do
+    consumers =
+      Enum.map(perConsumer, fn {consumer, count} ->
+        {:array, [{:bulk, consumer}, {:bulk, Integer.to_string(count)}]}
+      end)
+
+    {:array,
+     [
+       {:integer, total},
+       {:bulk, Stream.formatId(minId)},
+       {:bulk, Stream.formatId(maxId)},
+       {:array, consumers}
+     ]}
+  end
+
+  defp pendingRowReply({id, consumer, idle, deliveryCount}) do
+    {:array,
+     [
+       {:bulk, Stream.formatId(id)},
+       {:bulk, consumer},
+       {:integer, idle},
+       {:integer, deliveryCount}
+     ]}
+  end
+
+  defp claimReply(results, true) do
+    {:array, Enum.map(results, fn {id, _fields} -> {:bulk, Stream.formatId(id)} end)}
+  end
+
+  defp claimReply(results, false) do
+    {:array, Enum.map(results, &entryReply/1)}
+  end
+
+  defp autoclaimReply(cursor, claimed, deleted, justId) do
+    entriesReply =
+      if justId do
+        {:array, Enum.map(claimed, fn {id, _fields} -> {:bulk, Stream.formatId(id)} end)}
+      else
+        {:array, Enum.map(claimed, &entryReply/1)}
+      end
+
+    deletedReply = {:array, Enum.map(deleted, fn id -> {:bulk, Stream.formatId(id)} end)}
+    {:array, [{:bulk, Stream.formatId(cursor)}, entriesReply, deletedReply]}
+  end
+
+  defp reply(term, state), do: {[term], state}
+
+  defp orZero({:ok, value}), do: value
+  defp orZero({:error, _}), do: 0
+
+  defp streamError(:wrongtype), do: Reply.wrongtype()
+  defp streamError(:smaller),
+    do: {:error, "ERR The ID specified in XADD is equal or smaller than the target stream top item"}
+
+  defp streamError(:zero), do: {:error, "ERR The ID specified in XADD must be greater than 0-0"}
+  defp streamError(other), do: {:error, "ERR #{other}"}
+
+  defp nogroup(key, group) do
+    {:error,
+     "NOGROUP No such key '#{key}' or consumer group '#{group}' in XREADGROUP with GROUP option"}
+  end
+
+  defp noKeyGroupMessage do
+    "ERR The XGROUP subcommand requires the key to exist. Note that for CREATE you may want to use the MKSTREAM option to create an empty stream automatically."
+  end
+
+  defp setidSmallerMessage do
+    "ERR The ID specified in XSETID is smaller than the target stream top item"
+  end
+
+  defp now, do: System.system_time(:millisecond)
+  defp maxMs, do: 0xFFFFFFFFFFFFFFFF
+  defp maxSeq, do: 0xFFFFFFFFFFFFFFFF
+end