surreal

%%% @private
-module(surreal_prv_websocket).
-behaviour(websocket_client).

-export([start_link/2, send_message/2, message_broker/0]).
-export([
    init/1, onconnect/2, websocket_info/3, websocket_handle/3, ondisconnect/2, websocket_terminate/3
]).

start_link(Url, NotifyPid) ->
    websocket_client:start_link(Url, ?MODULE, [
        {pid, spawn(fun message_broker/0)}, {notify, NotifyPid}
    ]).

send_message(Conn, #{<<"id">> := Id} = Msg) ->
    {Pid, MonitorReference} = spawn_monitor(fun() ->
        receive
            Value ->
                exit({ok, Value})
        end
    end),

    ets:insert(surreal_pool, {Id, Pid}),

    Encoded = jiffy:encode(Msg),
    websocket_client:cast(Conn, {text, Encoded}),

    receive
        {'DOWN', MonitorReference, process, Pid, {ok, Result}} ->
            ets:delete(surreal_pool, Id),
            Result
    end.

message_broker() ->
    receive
        #{<<"id">> := Id} = Data ->
            case ets:lookup(surreal_pool, Id) of
                [{Id, Pid} | _Other] ->
                    Pid ! Data;
                _ ->
                    noop
            end;
        _ ->
            noop
    end,

    message_broker().

%%% -----------------------
%%% Handlers starting here.
%%% -----------------------

init([{pid, State}, {notify, NotifyPid}]) ->
    case ets:whereis(surreal_pool) of
        undefined -> ets:new(surreal_pool, [public, named_table, {decentralized_counters, true}]);
        _ -> noop
    end,

    {reconnect, {State, NotifyPid}}.

onconnect(_WSReq, State) ->
    {ok, State}.

websocket_info({close}, _ConnState, State) ->
    {close, <<>>, State}.

websocket_handle({ping, <<>>}, _ConnState, {State, NotifyPid}) ->
    % Send message to pid to know the connection is established.
    NotifyPid ! ok,
    {ok, State};
websocket_handle({ping, <<>>}, _ConnState, State) ->
    {ok, State};
websocket_handle({text, Msg}, _ConnState, State) ->
    State ! jiffy:decode(Msg, [return_maps]),
    {ok, State};
websocket_handle(_Message, _ConnState, State) ->
    {ok, State}.

ondisconnect({remote, closed}, State) ->
    {reconnect, State};
ondisconnect(Reason, State) ->
    {close, Reason, State}.

websocket_terminate(_Reason, _ConnState, _State) ->
    ok.