-module(roar_ffi).

-export([on_distributed/3, subscribe_distributed/3, publish_distributed/3]).

on_distributed(Scope, Topic, Listener) ->
    ScopeAtom = binary_to_atom(Scope, utf8),
    ensure_scope_started(ScopeAtom),
    ListenerPid = spawn(fun() -> distributed_listener_loop(Listener) end),
    pg:join(ScopeAtom, Topic, ListenerPid),

    fun() ->
       pg:leave(ScopeAtom, Topic, ListenerPid),
       ListenerPid ! stop
    end.

distributed_listener_loop(Listener) ->
    receive
        {roar_distributed, Msg} ->
            Listener(Msg),
            distributed_listener_loop(Listener);
        stop ->
            ok
    end.

subscribe_distributed(Scope, Topic, LocalWhisper) ->
    ScopeAtom = binary_to_atom(Scope, utf8),
    ensure_scope_started(ScopeAtom),
    BridgePid = spawn(fun() -> distribution_bridge_loop(LocalWhisper, Topic) end),
    pg:join(ScopeAtom, Topic, BridgePid),
    nil.

distribution_bridge_loop(LocalWhisper, Topic) ->
    receive
        {roar_distributed, Msg} ->
            whisper_ffi:publish(LocalWhisper, Topic, Msg),
            distribution_bridge_loop(LocalWhisper, Topic);
        stop ->
            ok
    end.

publish_distributed(Scope, Topic, Msg) ->
    ScopeAtom = binary_to_atom(Scope, utf8),
    ensure_scope_started(ScopeAtom),
    Members = pg:get_members(ScopeAtom, Topic),
    LocalNode = node(),
    lists:foreach(fun(Pid) ->
                     PidNode = node(Pid),
                     case PidNode =:= LocalNode of
                         true -> ok;
                         false -> Pid ! {roar_distributed, Msg}
                     end
                  end,
                  Members).

ensure_scope_started(Scope) ->
    case pg:start_link(Scope) of
        {ok, _Pid} ->
            ok;
        {error, {already_started, _Pid}} ->
            ok;
        {error, Reason} ->
            error(Reason)
    end.