From 686cff74f01d168ff8a818be31b880a7239aa721 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Tue, 25 Aug 2026 15:50:00 +0200 Subject: [PATCH] feat(store): add stream operations --- lib/excache/store/ops/streams.ex | 337 +++++++++++++++++++++++++++++++ 1 file changed, 337 insertions(+) create mode 100644 lib/excache/store/ops/streams.ex diff --git a/lib/excache/store/ops/streams.ex b/lib/excache/store/ops/streams.ex new file mode 100644 index 0000000..c8260fe --- /dev/null +++ b/lib/excache/store/ops/streams.ex @@ -0,0 +1,337 @@ +defmodule Excache.Store.Ops.Streams do + @moduledoc false + + alias Excache.Stream + alias Excache.Stream.Groups + alias Excache.Store.Ops.Helpers + + def apply_op(table, now, {:xadd, key, idSpec, fields, trim, nomkstream}) do + case fetchStream(table, key, now) do + :wrongtype -> + {:error, :wrongtype} + + :miss when nomkstream -> + {:ok, nil} + + existing -> + {stream, ea} = + case existing do + :miss -> {Stream.new(), nil} + {:ok, current, currentEa} -> {current, currentEa} + end + + case Stream.add(stream, idSpec, fields, now) do + {:ok, updated, id} -> + trimmed = applyTrim(updated, trim) + store(table, key, trimmed, ea) + {:ok, Stream.formatId(id)} + + {:error, reason} -> + {:error, reason} + end + end + end + + def apply_op(table, now, {:xlen, key}) do + withStream(table, key, now, 0, fn stream -> {:ok, Stream.length(stream)} end) + end + + def apply_op(table, now, {:xrange, key, startBound, endBound, count}) do + withStream(table, key, now, [], fn stream -> + {:ok, Stream.range(stream, startBound, endBound, count)} + end) + end + + def apply_op(table, now, {:xrevrange, key, endBound, startBound, count}) do + withStream(table, key, now, [], fn stream -> + {:ok, Stream.reverseRange(stream, endBound, startBound, count)} + end) + end + + def apply_op(table, now, {:xread, key, afterId, count}) do + case fetchStream(table, key, now) do + :miss -> + {:ok, nil} + + :wrongtype -> + {:error, :wrongtype} + + {:ok, stream, _ea} -> + case Stream.afterId(stream, afterId, count) do + [] -> {:ok, nil} + entries -> {:ok, entries} + end + end + end + + def apply_op(table, now, {:xdel, key, ids}) do + case fetchStream(table, key, now) do + :miss -> + {:ok, 0} + + :wrongtype -> + {:error, :wrongtype} + + {:ok, stream, ea} -> + {updated, removed} = Stream.delete(stream, ids) + store(table, key, updated, ea) + {:ok, removed} + end + end + + def apply_op(table, now, {:xtrim, key, strategy}) do + case fetchStream(table, key, now) do + :miss -> + {:ok, 0} + + :wrongtype -> + {:error, :wrongtype} + + {:ok, stream, ea} -> + {updated, removed} = trimBy(stream, strategy) + store(table, key, updated, ea) + {:ok, removed} + end + end + + def apply_op(table, now, {:xsetid, key, id}) do + case fetchStream(table, key, now) do + :miss -> + {:error, :nokey} + + :wrongtype -> + {:error, :wrongtype} + + {:ok, stream, ea} -> + cond do + id < Stream.lastEntryId(stream) -> + {:error, :setid_smaller} + + true -> + store(table, key, Stream.setLastId(stream, id), ea) + {:ok, "OK"} + end + end + end + + def apply_op(table, now, {:xinfoStream, key}) do + case fetchStream(table, key, now) do + :miss -> {:error, :nokey} + :wrongtype -> {:error, :wrongtype} + {:ok, stream, _ea} -> {:ok, Stream.info(stream)} + end + end + + def apply_op(table, now, {:xgroupCreate, key, name, startSpec, mkstream}) do + case fetchStream(table, key, now) do + :wrongtype -> + {:error, :wrongtype} + + :miss when mkstream -> + stream = Stream.new() + {:ok, created} = Stream.createGroup(stream, name, resolveStart(stream, startSpec)) + store(table, key, created, nil) + {:ok, "OK"} + + :miss -> + {:error, :nokey} + + {:ok, stream, ea} -> + case Stream.createGroup(stream, name, resolveStart(stream, startSpec)) do + {:ok, created} -> + store(table, key, created, ea) + {:ok, "OK"} + + {:error, :busygroup} -> + {:error, :busygroup} + end + end + end + + def apply_op(table, now, {:xgroupDestroy, key, name}) do + withGroupMutation(table, key, now, fn stream -> + {updated, removed} = Stream.destroyGroup(stream, name) + {updated, {:ok, removed}} + end) + end + + def apply_op(table, now, {:xgroupCreateConsumer, key, name, consumer}) do + withExistingGroup(table, key, now, name, fn stream -> + {updated, added} = Groups.createConsumer(stream, name, consumer, now) + {updated, {:ok, added}} + end) + end + + def apply_op(table, now, {:xgroupDelConsumer, key, name, consumer}) do + withExistingGroup(table, key, now, name, fn stream -> + {updated, removed} = Groups.deleteConsumer(stream, name, consumer) + {updated, {:ok, removed}} + end) + end + + def apply_op(table, now, {:xgroupSetId, key, name, id}) do + withExistingGroup(table, key, now, name, fn stream -> + {Groups.setId(stream, name, id), {:ok, "OK"}} + end) + end + + def apply_op(table, now, {:xreadgroup, key, name, consumer, mode, count, noack}) do + case fetchStream(table, key, now) do + :miss -> + {:error, :nogroup} + + :wrongtype -> + {:error, :wrongtype} + + {:ok, stream, ea} -> + case Stream.fetchGroup(stream, name) do + :error -> + {:error, :nogroup} + + {:ok, _group} -> + {updated, entries} = readGroup(stream, name, consumer, mode, count, noack, now) + store(table, key, updated, ea) + {:ok, entries} + end + end + end + + def apply_op(table, now, {:xack, key, name, ids}) do + case fetchStream(table, key, now) do + {:ok, stream, ea} -> + case Stream.fetchGroup(stream, name) do + {:ok, _group} -> + {updated, count} = Groups.ack(stream, name, ids) + store(table, key, updated, ea) + {:ok, count} + + :error -> + {:ok, 0} + end + + _ -> + {:ok, 0} + end + end + + def apply_op(table, now, {:xpendingSummary, key, name}) do + withGroupRead(table, key, now, name, fn stream -> + {:ok, Groups.pendingSummary(stream, name)} + end) + end + + def apply_op(table, now, {:xpendingRange, key, name, startId, endId, count, consumer, minIdle}) do + withGroupRead(table, key, now, name, fn stream -> + {:ok, Groups.pendingRange(stream, name, startId, endId, count, consumer, minIdle, now)} + end) + end + + def apply_op(table, now, {:xclaim, key, name, consumer, minIdle, ids, opts}) do + withGroupMutationChecked(table, key, now, name, fn stream -> + {updated, results} = Groups.claim(stream, name, consumer, minIdle, ids, opts, now) + {updated, {:ok, results}} + end) + end + + def apply_op(table, now, {:xautoclaim, key, name, consumer, minIdle, startId, count}) do + withGroupMutationChecked(table, key, now, name, fn stream -> + {updated, claimed, deleted, cursor} = + Groups.autoclaim(stream, name, consumer, minIdle, startId, count, now) + + {updated, {:ok, {cursor, claimed, deleted}}} + end) + end + + defp fetchStream(table, key, now) do + case Helpers.fetch(table, key, now) do + :miss -> :miss + {:stream, stream, ea} -> {:ok, stream, ea} + _ -> :wrongtype + end + end + + defp store(table, key, stream, ea) do + :ets.insert(table, {key, :stream, stream, ea}) + end + + defp withStream(table, key, now, default, fun) do + case fetchStream(table, key, now) do + :miss -> {:ok, default} + :wrongtype -> {:error, :wrongtype} + {:ok, stream, _ea} -> fun.(stream) + end + end + + defp withGroupMutation(table, key, now, fun) do + case fetchStream(table, key, now) do + :miss -> + {:ok, 0} + + :wrongtype -> + {:error, :wrongtype} + + {:ok, stream, ea} -> + {updated, reply} = fun.(stream) + store(table, key, updated, ea) + reply + end + end + + defp withExistingGroup(table, key, now, name, fun) do + case fetchStream(table, key, now) do + :miss -> + {:error, :nogroup} + + :wrongtype -> + {:error, :wrongtype} + + {:ok, stream, ea} -> + case Stream.fetchGroup(stream, name) do + :error -> + {:error, :nogroup} + + {:ok, _group} -> + {updated, reply} = fun.(stream) + store(table, key, updated, ea) + reply + end + end + end + + defp withGroupMutationChecked(table, key, now, name, fun) do + withExistingGroup(table, key, now, name, fun) + end + + defp withGroupRead(table, key, now, name, fun) do + case fetchStream(table, key, now) do + :miss -> + {:error, :nogroup} + + :wrongtype -> + {:error, :wrongtype} + + {:ok, stream, _ea} -> + case Stream.fetchGroup(stream, name) do + :error -> {:error, :nogroup} + {:ok, _group} -> fun.(stream) + end + end + end + + defp readGroup(stream, name, consumer, {:new}, count, noack, now) do + Groups.readNew(stream, name, consumer, count, noack, now) + end + + defp readGroup(stream, name, consumer, {:history, startId}, count, _noack, now) do + Groups.readHistory(stream, name, consumer, startId, count, now) + end + + defp resolveStart(stream, :last), do: Stream.lastEntryId(stream) + defp resolveStart(_stream, {:id, id}), do: id + + defp applyTrim(stream, nil), do: stream + defp applyTrim(stream, strategy), do: elem(trimBy(stream, strategy), 0) + + defp trimBy(stream, {:maxlen, count}), do: Stream.trimMaxlen(stream, count) + defp trimBy(stream, {:minid, id}), do: Stream.trimMinId(stream, id) +end