<commit_before>-module(connection).

-behaviour(gen_server).

-export([start/1, start_link/1]).
-export([stop/1]).
-export([init/1, terminate/2, code_change/3, handle_call/3,
         handle_cast/2, handle_info/2]).

-record(state, {
          user_id,
          user_data,
          publishers,
          subscribers,
          transport
         }).

start_link(From) ->
    Opts = [],
    gen_server:start_link(?MODULE, [From], Opts).

start(From) ->
    Opts = [],
    gen_server:start(?MODULE, [From], Opts).

init([From]) ->
    erlang:monitor(process, From),
    {ok, #state{publishers=dict:new(), subscribers=dict:new(), transport=From}}.

handle_cast({process_message, Message}, State) ->
    StateNew = try jsx:decode(Message) of
                   Msg -> process_message(Msg, State)
               catch _:_ ->
                         State
               end,
    {noreply, StateNew};

handle_cast(_Message, State) ->
    {noreply, State}.

handle_call(_Request, _From, State) ->
    {reply, unknown_request, State}.

handle_info({'DOWN', _Ref, process, _Pid, _}, State) ->
    {stop, shutdown, State};
handle_info({presence_response, Msg}, State = #state{transport = Transport}) ->
    Transport ! {text, Msg},
    {noreply, State};

handle_info({received_message, Msg}, State = #state{transport = Transport}) ->
    Transport ! {text, Msg},
    {noreply, State};

handle_info({just_send, Msg}, State = #state{transport = Transport}) ->
    Transport ! {text, jsx:encode([{<<"info">>, Msg}])},
    {noreply, State};

handle_info({timeout, _Ref, Msg}, State = #state{transport = Transport}) ->
    Transport ! {text, Msg},
    {noreply, State};

handle_info(_Info, State) ->
    {stop, {unhandled_message, _Info}, State}.

stop(Pid) ->
    gen_server:call(Pid, stop, infinity).


terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------

process_message([{<<"joinchannel">>, Channel}], State = #state{publishers=Pubs, subscribers=Subs, user_id=UserId}) ->
    {ok, PublisherPid} = supervisor:start_child(msg_exchange_sup, [Channel, UserId, self()]),
    {ok, SubscriberPid} = supervisor:start_child(msg_queue_sup, [Channel, UserId, self()]),
    NewPubs = case dict:find(Channel, Pubs) of
                  error -> dict:store(Channel, PublisherPid, Pubs);
                  {ok, _} -> Pubs
              end,
    NewSubs = case dict:find(Channel, Subs) of
                  error -> dict:store(Channel, SubsriberPid, Subs);
                  {ok, _} -> Subs
              end,
    State#state{publishers = NewPubs, subscribers = NewSubs};

process_message([{<<"send">>, Message}, {<<"channel">>, Channel}], State = #state{publishers=Pubs, user_id=UserId, user_data=UserData}) ->
    % TODO: make robust
    {ok, PublisherPid} = dict:find(Channel, Pubs),
    gen_server:call(PublisherPid, {publish, jsx:encode([{<<"message">>, Message},
                                                        {<<"publisher">>, Channel}, 
                                                        {<<"user_id">>, UserId},
                                                        {<<"user_data">>, UserData}
                                                       ])}),
    State;

process_message([{<<"presence">>, Channel}], State = #state{publishers=Pubs}) ->
    case dict:find(Channel, Pubs) of
        error -> ok;
        {ok, _} ->
            {UsersPub, UsersSub} = presence_serv:presence(Channel),
            self() ! {presence_response, jsx:encode([{<<"publishers">>, UsersPub},
                                                     {<<"subscribers">>, UsersSub},
                                                     {<<"publisher">>, Channel}
                                                    ])}
    end,
    State;

process_message([{<<"authenticate">>, AssumedUserId},{<<"token">>, Token}], State ) ->
    {ok, AuthenticateUrl} = application:get_env(carotene, authenticate_url),
    case httpc:request(post, {AuthenticateUrl, [], "application/x-www-form-urlencoded", "user_id="++binary_to_list(AssumedUserId)++"&token="++binary_to_list(Token)}, [], []) of
        {ok, {{_Version, 200, _ReasonPhrase}, _Headers, Body}} -> 
            AuthResult = jsx:decode(binary:list_to_bin(Body)),
            {UserID, UserData} = case AuthResult of
                                     % TODO: Can return in different order
                                     [{<<"authenticated">>, true}, {<<"user_data">>, ResUserData}] ->
                                         self() ! {just_send, <<"Authenticated">>},
                                         {AssumedUserId, ResUserData};
                                     _ -> self() ! {just_send, <<"Authentication failed">>},
                                          {undefined, undefined}
                                 end,
            State#state{user_id = UserID, user_data = UserData};
        _ -> self() ! {just_send, <<"Authentication error: Bad response from server">>},
             State
    end;

process_message(_, State) ->
    io:format("unknown message~n"),
    State .<commit_msg>renaming exchanges to publishers and queues to subcsribers for consistency<commit_after>-module(connection).

-behaviour(gen_server).

-export([start/1, start_link/1]).
-export([stop/1]).
-export([init/1, terminate/2, code_change/3, handle_call/3,
         handle_cast/2, handle_info/2]).

-record(state, {
          user_id,
          user_data,
          publishers,
          subscribers,
          transport
         }).

start_link(From) ->
    Opts = [],
    gen_server:start_link(?MODULE, [From], Opts).

start(From) ->
    Opts = [],
    gen_server:start(?MODULE, [From], Opts).

init([From]) ->
    erlang:monitor(process, From),
    {ok, #state{publishers=dict:new(), subscribers=dict:new(), transport=From}}.

handle_cast({process_message, Message}, State) ->
    StateNew = try jsx:decode(Message) of
                   Msg -> 
                       process_message(Msg, State)
               catch _:_ ->
                         State
               end,
    {noreply, StateNew};

handle_cast(_Message, State) ->
    {noreply, State}.

handle_call(_Request, _From, State) ->
    {reply, unknown_request, State}.

handle_info({'DOWN', _Ref, process, _Pid, _}, State) ->
    {stop, shutdown, State};
handle_info({presence_response, Msg}, State = #state{transport = Transport}) ->
    Transport ! {text, Msg},
    {noreply, State};

handle_info({received_message, Msg}, State = #state{transport = Transport}) ->
    Transport ! {text, Msg},
    {noreply, State};

handle_info({just_send, Msg}, State = #state{transport = Transport}) ->
    Transport ! {text, jsx:encode([{<<"info">>, Msg}])},
    {noreply, State};

handle_info({timeout, _Ref, Msg}, State = #state{transport = Transport}) ->
    Transport ! {text, Msg},
    {noreply, State};

handle_info(_Info, State) ->
    {stop, {unhandled_message, _Info}, State}.

stop(Pid) ->
    gen_server:call(Pid, stop, infinity).


terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------

process_message([{<<"joinchannel">>, Channel}], State = #state{publishers=Pubs, subscribers=Subs, user_id=UserId}) ->
    {ok, PublisherPid} = supervisor:start_child(msg_exchange_sup, [Channel, UserId, self()]),
    {ok, SubscriberPid} = supervisor:start_child(msg_queue_sup, [Channel, UserId, self()]),
    NewPubs = case dict:find(Channel, Pubs) of
                  error -> dict:store(Channel, PublisherPid, Pubs);
                  {ok, _} -> Pubs
              end,
    NewSubs = case dict:find(Channel, Subs) of
                  error -> dict:store(Channel, SubscriberPid, Subs);
                  {ok, _} -> Subs
              end,
    State#state{publishers = NewPubs, subscribers = NewSubs};

process_message([{<<"send">>, Message}, {<<"channel">>, Channel}], State = #state{publishers=Pubs, user_id=UserId, user_data=UserData}) ->
    % TODO: make robust
    {ok, PublisherPid} = dict:find(Channel, Pubs),
    gen_server:call(PublisherPid, {publish, jsx:encode([{<<"message">>, Message},
                                                        {<<"channel">>, Channel}, 
                                                        {<<"user_id">>, UserId},
                                                        {<<"user_data">>, UserData}
                                                       ])}),
    State;

process_message([{<<"presence">>, Channel}], State = #state{publishers=Pubs}) ->
    case dict:find(Channel, Pubs) of
        error -> ok;
        {ok, _} ->
            {UsersPub, UsersSub} = presence_serv:presence(Channel),
            self() ! {presence_response, jsx:encode([{<<"publishers">>, UsersPub},
                                                     {<<"subscribers">>, UsersSub},
                                                     {<<"publisher">>, Channel}
                                                    ])}
    end,
    State;

process_message([{<<"authenticate">>, AssumedUserId},{<<"token">>, Token}], State ) ->
    {ok, AuthenticateUrl} = application:get_env(carotene, authenticate_url),
    case httpc:request(post, {AuthenticateUrl, [], "application/x-www-form-urlencoded", "user_id="++binary_to_list(AssumedUserId)++"&token="++binary_to_list(Token)}, [], []) of
        {ok, {{_Version, 200, _ReasonPhrase}, _Headers, Body}} -> 
            AuthResult = jsx:decode(binary:list_to_bin(Body)),
            {UserID, UserData} = case AuthResult of
                                     % TODO: Can return in different order
                                     [{<<"authenticated">>, true}, {<<"user_data">>, ResUserData}] ->
                                         self() ! {just_send, <<"Authenticated">>},
                                         {AssumedUserId, ResUserData};
                                     _ -> self() ! {just_send, <<"Authentication failed">>},
                                          {undefined, undefined}
                                 end,
            State#state{user_id = UserID, user_data = UserData};
        _ -> self() ! {just_send, <<"Authentication error: Bad response from server">>},
             State
    end;

process_message(Msg, State) ->
    io:format("unknown message ~p ~n", [Msg]),
    State .
{'fraction_non_alphanumeric': 0.17895233600755073, 'fraction_numerical': 0.0022652194431335535, 'mean_word_length': 1.9672360683281993, 'pattern_counts': {'":': 0, '<': 79, '<?xml version=': 0, '>': 143, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 50, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6111640', 'n_tokens_mistral': 3090, 'n_tokens_neox': 2990, 'n_words': 598}