Add src/hk_messaging_portal.erl

src/hk_messaging_portal.erl

@@ -0,0 +1,149 @@
+%%%-------------------------------------------------------------------
+%%% @doc HK-MSG :: Messaging Portal.
+%%%
+%%% Structured communication between users, agents, browser
+%%% sessions, and system services, per the `<message_schema>':
+%%% every message is a `#message{}' with a stable id, sender,
+%%% recipient, timestamp, conversation id, type, payload, and
+%%% delivery status. This is deliberately *not* a raw pub/sub bus --
+%%% `hk_event_bus' already is one, for system observability. This
+%%% module is the addressed, stored, replayable inbox/outbox: a
+%%% conversation's history survives after every participant has
+%%% disconnected, which is what "delivered" vs. "queued" vs. "read"
+%%% status is for.
+%%%
+%%% Delivery: a recipient (a user's frontend connection, an agent
+%%% process, ...) calls `subscribe/1' with the recipient id it
+%%% answers to; a message sent to a currently-subscribed recipient
+%%% is pushed immediately and marked `delivered', otherwise it stays
+%%% `queued' until the recipient subscribes and calls
+%%% `list_conversation/1' (or a future subscribe replays the
+%%% backlog -- see the `subscribe/1' implementation).
+%%% @end
+%%%-------------------------------------------------------------------
+-module(hk_messaging_portal).
+-behaviour(gen_server).
+
+-include("hyperkitty.hrl").
+
+-export([start_link/0]).
+-export([send/5, subscribe/1, unsubscribe/0, list_conversation/1, mark_read/1, get/1]).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2]).
+
+-define(MSG_TAB, hk_messages_tab).
+-define(CONV_TAB, hk_messages_by_conversation_tab).
+
+start_link() ->
+    hk_sup_util:start_link_retry(
+        fun() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []) end).
+
+-spec send(binary(), binary(), binary(), atom(), map()) -> {ok, binary()}.
+send(Sender, Recipient, ConversationId, MessageType, Payload) ->
+    gen_server:call(?MODULE, {send, Sender, Recipient, ConversationId, MessageType, Payload}).
+
+-spec subscribe(binary()) -> ok.
+subscribe(RecipientId) ->
+    gen_server:call(?MODULE, {subscribe, RecipientId, self()}).
+
+-spec unsubscribe() -> ok.
+unsubscribe() ->
+    gen_server:call(?MODULE, {unsubscribe, self()}).
+
+-spec list_conversation(binary()) -> [map()].
+list_conversation(ConversationId) ->
+    Ids = case ets:lookup(?CONV_TAB, ConversationId) of
+              [{ConversationId, MsgIds}] -> MsgIds;
+              [] -> []
+          end,
+    [hk_schema:to_map(M) || Id <- lists:reverse(Ids), {ok, M} <- [raw_get(Id)]].
+
+-spec mark_read(binary()) -> ok | {error, not_found}.
+mark_read(MessageId) ->
+    gen_server:call(?MODULE, {mark_read, MessageId}).
+
+-spec get(binary()) -> {ok, map()} | {error, not_found}.
+get(MessageId) ->
+    case raw_get(MessageId) of
+        {ok, M} -> {ok, hk_schema:to_map(M)};
+        Error -> Error
+    end.
+
+init([]) ->
+    ets:new(?MSG_TAB, [set, named_table, protected]),
+    ets:new(?CONV_TAB, [set, named_table, protected]),
+    {ok, #{subscribers => #{}, monitors => #{}}}.
+
+handle_call({send, Sender, Recipient, ConversationId, MessageType, Payload}, _From, State) ->
+    MessageId = hk_id:new(<<"msg">>),
+    Now = hk_id:timestamp_ms(),
+    Subscribed = maps:is_key(Recipient, maps:get(subscribers, State)),
+    Status = case Subscribed of true -> delivered; false -> queued end,
+    Message = #message{
+        message_id = MessageId, sender = Sender, recipient = Recipient,
+        timestamp = Now, conversation_id = ConversationId,
+        message_type = MessageType, payload = Payload, status = Status
+    },
+    ets:insert(?MSG_TAB, {MessageId, Message}),
+    append_conversation_index(ConversationId, MessageId),
+    emit(<<"message.sent">>, Message),
+    case Subscribed of
+        true ->
+            Pid = maps:get(Recipient, maps:get(subscribers, State)),
+            Pid ! {hk_message, hk_schema:to_map(Message)},
+            emit(<<"message.received">>, Message);
+        false ->
+            ok
+    end,
+    {reply, {ok, MessageId}, State};
+
+handle_call({subscribe, RecipientId, Pid}, _From, State = #{subscribers := Subs, monitors := Mons}) ->
+    MonRef = erlang:monitor(process, Pid),
+    {reply, ok, State#{subscribers := Subs#{RecipientId => Pid},
+                        monitors := Mons#{MonRef => RecipientId}}};
+
+handle_call({unsubscribe, Pid}, _From, State = #{subscribers := Subs}) ->
+    Subs2 = maps:filter(fun(_K, V) -> V =/= Pid end, Subs),
+    {reply, ok, State#{subscribers := Subs2}};
+
+handle_call({mark_read, MessageId}, _From, State) ->
+    case raw_get(MessageId) of
+        {ok, M} ->
+            ets:insert(?MSG_TAB, {MessageId, M#message{status = read}}),
+            {reply, ok, State};
+        Error ->
+            {reply, Error, State}
+    end.
+
+handle_cast(_Msg, State) ->
+    {noreply, State}.
+
+handle_info({'DOWN', MonRef, process, _Pid, _Reason}, State = #{subscribers := Subs, monitors := Mons}) ->
+    case maps:take(MonRef, Mons) of
+        {RecipientId, Mons2} ->
+            {noreply, State#{subscribers := maps:remove(RecipientId, Subs), monitors := Mons2}};
+        error ->
+            {noreply, State}
+    end;
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+%%% internal
+
+raw_get(MessageId) ->
+    case ets:lookup(?MSG_TAB, MessageId) of
+        [{MessageId, M}] -> {ok, M};
+        [] -> {error, not_found}
+    end.
+
+append_conversation_index(ConversationId, MessageId) ->
+    Existing = case ets:lookup(?CONV_TAB, ConversationId) of
+                   [{ConversationId, Ids}] -> Ids;
+                   [] -> []
+               end,
+    ets:insert(?CONV_TAB, {ConversationId, [MessageId | Existing]}).
+
+emit(Category, #message{message_id = Id, sender = Sender, recipient = Recipient,
+                         conversation_id = ConvId}) ->
+    Event = hk_event:new(Category, {message, Id},
+                          #{sender => Sender, recipient => Recipient, conversation_id => ConvId}),
+    catch hk_event_bus:publish(Event).