eredis_streams

-module(eredis_streams_db).

-export(
  [
    init_xgroup/3,
    is_master/4,
    is_failover_master/4,
    stream_fetch_messages/6,
    stream_fetch_pending_messages/4,
    push_message/3,
    xack_message/4,
    xlen/2,
    xclaim_message/6,
    xdel/3
  ]
).

init_xgroup(Pool, Stream, Group) ->
  case eredis_pool:q(Pool, ["XGROUP", "CREATE", Stream, Group, "$", "MKSTREAM"]) of
    {error, <<"BUSYGROUP Consumer Group name already exists">>} -> ok;
    {ok, _} -> ok;
    Err -> Err
  end.


is_master(Pool, Stream, Wid, TimeoutMs) ->
  case eredis_pool:q(Pool, ["GET", master_lock_keyname(Stream)]) of
    {ok, Id} when Id == undefined orelse Id == Wid ->
      eredis_pool:transaction(
        Pool,
        fun
          (Conn) ->
            eredis:q(Conn, ["SETNX", master_lock_keyname(Stream), Wid]),
            eredis:q(Conn, ["EXPIRE", master_lock_keyname(Stream), round(TimeoutMs / 1000)])
        end
      ),
      true;

    _Else -> false
  end.


is_failover_master(Pool, Stream, Wid, TimeoutMs) ->
  case eredis_pool:q(Pool, ["GET", failover_master_lock_keyname(Stream)]) of
    {ok, Id} when Id == undefined orelse Id == Wid ->
      eredis_pool:transaction(
        Pool,
        fun
          (Conn) ->
            eredis:q(Conn, ["SETNX", failover_master_lock_keyname(Stream), Wid]),
            eredis:q(
              Conn,
              ["EXPIRE", failover_master_lock_keyname(Stream), round(TimeoutMs / 1000)]
            )
        end
      ),
      true;

    _Else -> false
  end.


stream_fetch_messages(Pool, Stream, Group, Process, Count, BlockTimeoutMs) ->
  case
  eredis_pool:q(
    Pool,
    [
      "XREADGROUP",
      "GROUP",
      Group,
      Process,
      "BLOCK",
      BlockTimeoutMs,
      "COUNT",
      Count,
      "STREAMS",
      Stream,
      ">"
    ]
  ) of
    {ok, undefined} -> [];

    {ok, [_ | _] = Result} ->
      [{MsgId, to_proplist(Fields)} || [MsgId, Fields] <- messages_from_stream(Result, Stream)];

    Else -> Else
  end.


stream_fetch_pending_messages(Pool, Stream, Group, Count) ->
  case eredis_pool:q(Pool, ["XPENDING", Stream, Group, "-", "+", integer_to_list(Count)]) of
    {ok, Xs} when Xs == undefined orelse Xs == [] -> [];

    {ok, [_ | _] = Result} ->
      [
        #{
          msg_id => MsgId,
          xgroup => Group,
          orig_subscriber => SubscriberName,
          time_ms => binary_to_integer(TimeMs),
          cnt => binary_to_integer(Cnt)
        }
        || [MsgId, SubscriberName, TimeMs, Cnt] <- Result
      ];

    Else -> Else
  end.


push_message(Pool, Stream, Message) ->
  eredis_pool:q(Pool, ["XADD", Stream, "*", "message", Message]).

xack_message(Pool, Stream, Group, MsgId) ->
  case eredis_pool:q(Pool, ["XACK", Stream, Group, MsgId]) of
    {ok, _} -> ok;
    Else -> Else
  end.


xlen(Pool, Stream) ->
  case eredis_pool:q(Pool, ["XLEN", Stream]) of
    {ok, Cnt} -> binary_to_integer(Cnt);
    Err -> Err
  end.


xclaim_message(Pool, Stream, Group, MsgId, NewProcessName, TimeoutMs) ->
  case eredis_pool:q(Pool, ["XCLAIM", Stream, Group, NewProcessName, TimeoutMs, MsgId]) of
    {ok, undefined} -> [];
    {ok, [_ | _] = Result} -> [{XMsgId, to_proplist(Fields)} || [XMsgId, Fields] <- Result];
    Else -> Else
  end.


xdel(Pool, Stream, MsgId) -> eredis_pool:q(Pool, ["XDEL", Stream, MsgId]).

%%%===================================================================
%%% Internal functions
%%%===================================================================

master_lock_keyname(Stream) ->
  iolist_to_binary(["ota:master_lock:", integer_to_list(erlang:phash2(Stream))]).

failover_master_lock_keyname(Stream) ->
  iolist_to_binary(["ota:failover_master_lock:", integer_to_list(erlang:phash2(Stream))]).

messages_from_stream(Result, Stream) ->
  case [Messages || [StreamId, [_ | _] = Messages] <- Result, string:equal(StreamId, Stream)] of
    [_ | _] = Res -> hd(Res);
    _ -> []
  end.


to_proplist([]) -> [];
to_proplist([K, V | T]) -> [{K, V} | to_proplist(T)].