<commit_before>-module(imem_adapter).
-author('<NAME> <<EMAIL>>').

-include("dderl.hrl").
-include("gres.hrl").

-include_lib("imem/include/imem_sql.hrl").

-export([ init/0
        , process_cmd/6
        , disconnect/1
        , rows/2
        , get_deps/0
        , process_query/4
        , bind_arg_types/0
        , add_conn_info/2
        , connect_map/1
        ]).

-record(priv, {connections = [], conn_info}).

-define(E2B(__T), gen_adapter:encrypt_to_binary(__T)).
-define(D2T(__B), gen_adapter:decrypt_to_term(__B)).

bind_arg_types() -> imem_datatype:bind_arg_types().

-spec init() -> ok.
init() ->
    dderl_dal:add_adapter(imem, <<"IMEM DB">>),
    SystemViews = [
        { <<"Remote Tables">>
        , <<"select
                to_name(qname),
                size as rows,
                memory,
                nodef(expiry) as expires,
                nodef(tte) as expires_after
            from
                all_tables,
                ddSize
            where
                name = element(2, qname) and size <> to_atom('undefined')
            order by 1 asc">>
        , []},
        { <<"All ddViews">>
        , <<"select
                c.owner,
                v.name
            from
                ddView as v,
                ddCmd as c
            where
                c.id = v.cmd
                and (c.conns = to_list('[]') or is_member(:ddConn.id, c.conns))
                and (c.owner = user or c.owner = to_atom('system'))
                and (is_member(:ddAdapter.id, c.adapters))
            order by
                2 asc,
                1 asc">>
        , local}
    ],
    %% TODO: This should be added on the load of the other adapter but we don't know the id...
    AddViewResult = gen_adapter:add_cmds_views(undefined, system, imem, false, SystemViews),
    case lists:member(need_replace, AddViewResult) of
        true ->
            View = dderl_dal:get_view(undefined, <<"All ddViews">>, imem, system),
            dderl_dal:add_adapter_to_cmd(undefined, View#ddView.cmd, oci);
        _ ->
            [_, ViewId] = AddViewResult,
            View = dderl_dal:get_view(undefined, ViewId),
            dderl_dal:add_adapter_to_cmd(undefined, View#ddView.cmd, oci)
    end.

-spec add_conn_info(undefined | #priv{}, map()) -> #priv{}.
add_conn_info(undefined, ConnInfo) ->
    add_conn_info(#priv{connections = []}, ConnInfo);
add_conn_info(#priv{} = Priv, ConnInfo) when is_map(ConnInfo) ->
    Priv#priv{conn_info = ConnInfo}.

-spec connect_map(#ddConn{}) -> map().
connect_map(#ddConn{adapter = imem} = C) ->
    add_conn_extra(C, #{id => C#ddConn.id,
                        name => C#ddConn.name,
                        adapter => <<"imem">>,
                        owner => dderl_dal:user_name(C#ddConn.owner),
                        schema => atom_to_binary(C#ddConn.schm, utf8)}).

add_conn_extra(#ddConn{access = Access}, Conn)
  when is_map(Access), is_map(Conn) ->
        maps:merge(Conn, maps:remove(owner,maps:remove(<<"owner">>,Access)));
add_conn_extra(#ddConn{access = Access}, Conn) when is_list(Access), is_map(Conn) ->
    case proplists:get_value(type, Access, proplists:get_value(<<"method">>, Access, tcp)) of
        Local when Local == local; Local == <<"local">> ->
            Conn#{method => <<"local">>};
        LocalSec when LocalSec == local_sec; LocalSec == <<"local_sec">> ->
            Conn#{method => <<"local_sec">>,
                           user => proplists:get_value(user, Access, <<>>)};
        Rpc when Rpc == rpc; Rpc == <<"rpc">> ->
            Conn#{method => <<"rpc">>,
                  node => proplists:get_value(node, Access, <<>>),
                  user => proplists:get_value(user, Access, <<>>)};
        Tcp when Tcp == tcp; Tcp == <<"tcp">> ->
            Conn#{method => <<"tcp">>,
                  host => proplists:get_value(ip, Access, <<>>),
                  port => proplists:get_value(port, Access, <<>>),
                  user => proplists:get_value(user, Access, <<>>),
                  secure => proplists:get_value(secure, Access, false)}
    end.

-spec conn_method([{binary(),binary()}]|binary()) -> tcp | rpc | local.
conn_method(<<"rpc">>)          -> rpc;
conn_method(<<"local">>)        -> local;
conn_method(Other)
  when Other == <<"tcp">>;
       Other == '$not_defined'  -> tcp;
conn_method([{K,V}|_] = PropList) when is_binary(K), is_binary(V) ->
    conn_method(proplists:get_value(<<"method">>, PropList, '$not_defined')).

-spec connect_erlimem(tcp | rpc | local | local_sec, {atom(), pid()},
                      binary(), list({binary(),binary()}), map()) ->
    {ok, {atom(),pid()}, any()}.
connect_erlimem(tcp, _Sess, SessionId, Params, ConnInfo) ->
    Opts = case proplists:get_value(<<"secure">>, Params, false) of
               false -> [];
               true -> [ssl]
           end,
    IpAddr = binary_to_list(proplists:get_value(<<"host">>, Params)),
    Port = binary_to_integer(proplists:get_value(<<"port">>, Params)),
    Schema = try binary_to_existing_atom(
                 proplists:get_value(<<"schema">>, Params, '$bad_schema'),
                 utf8) of
               AtomSchema -> AtomSchema
           catch _:_ -> error("Invalid schema for tcp")
           end,
    connect_erlimem_password({tcp, IpAddr, Port, Opts}, Schema, SessionId, ConnInfo, Params);
connect_erlimem(rpc, _Sess, SessionId, Params, ConnInfo) ->
    Node = try binary_to_existing_atom(
                 proplists:get_value(<<"node">>, Params, '$bad_node'),
                 utf8) of
               AtomPort -> AtomPort
           catch _:_ -> error("Invalid node for rpc")
           end,
    Schema = try binary_to_existing_atom(
                 proplists:get_value(<<"schema">>, Params, '$bad_schema'),
                 utf8) of
               AtomSchema -> AtomSchema
           catch _:_ -> error("Invalid schema for rpc")
           end,
    connect_erlimem_password({rpc, Node}, Schema, SessionId, ConnInfo, Params);
connect_erlimem(local, Sess, SessionId, Params, ConnInfo) ->
    case proplists:get_value(<<"secure">>, Params, false) of
        true ->
            connect_erlimem(local_sec, Sess, SessionId, Params, ConnInfo);
        false ->
            case dderl_dal:can_connect_locally(Sess) of
                true ->
                    {ok, ErlImemSess} = erlimem:open(local, imem_meta:schema()),
                    {ok, ErlImemSess, #{node => list_to_binary(imem_meta:node_shard())}};
                _ -> error(<<"Local connection unauthorized">>)
            end
    end;
connect_erlimem(local_sec, ErlImemSess, _SessionId, _Params, _ConnInfo) ->
    {ok, ErlImemSess, #{node => list_to_binary(imem_meta:node_shard())}}.

connect_erlimem_password(Connect, Schema, SessionId, ConnInfo, Params) ->
    {ok, ErlImemSess}
    = case erlimem:open(Connect, Schema) of
          {ok, _} = E -> E;
          Error ->
              error(list_to_binary(io_lib:format("~p", [Error])))
      end,
    case ErlImemSess:auth(dderl,SessionId,{access,ConnInfo}) of
        {ok, [{pwdmd5,_}|_]} ->
            User = proplists:get_value(<<"user">>, Params, <<>>),
            Password = proplists:get_value(<<"password">>, Params, []),
            case ErlImemSess:auth(dderl,SessionId,{pwdmd5,{User,list_to_binary(Password)}}) of
                Ok when Ok == {ok,[]}; Ok == ok ->
                    case ErlImemSess:run_cmd(login,[]) of
                        {error,{{'SecurityException',{?PasswordChangeNeeded,_}},ST}} ->
                            ?Warn("Password expired ~s~n~p", [User, ST]),
                            {ok, ErlImemSess, #{changePass=>User}};
                        _ ->
                            {ok, ErlImemSess, '$no_extra'}<commit_msg>connect tcp treated as an internal connction #444 (#447)<commit_after>-module(imem_adapter).
-author('<NAME> <<EMAIL>>').

-include("dderl.hrl").
-include("gres.hrl").

-include_lib("imem/include/imem_sql.hrl").

-export([ init/0
        , process_cmd/6
        , disconnect/1
        , rows/2
        , get_deps/0
        , process_query/4
        , bind_arg_types/0
        , add_conn_info/2
        , connect_map/1
        ]).

-record(priv, {connections = [], conn_info}).

-define(E2B(__T), gen_adapter:encrypt_to_binary(__T)).
-define(D2T(__B), gen_adapter:decrypt_to_term(__B)).

bind_arg_types() -> imem_datatype:bind_arg_types().

-spec init() -> ok.
init() ->
    dderl_dal:add_adapter(imem, <<"IMEM DB">>),
    SystemViews = [
        { <<"Remote Tables">>
        , <<"select
                to_name(qname),
                size as rows,
                memory,
                nodef(expiry) as expires,
                nodef(tte) as expires_after
            from
                all_tables,
                ddSize
            where
                name = element(2, qname) and size <> to_atom('undefined')
            order by 1 asc">>
        , []},
        { <<"All ddViews">>
        , <<"select
                c.owner,
                v.name
            from
                ddView as v,
                ddCmd as c
            where
                c.id = v.cmd
                and (c.conns = to_list('[]') or is_member(:ddConn.id, c.conns))
                and (c.owner = user or c.owner = to_atom('system'))
                and (is_member(:ddAdapter.id, c.adapters))
            order by
                2 asc,
                1 asc">>
        , local}
    ],
    %% TODO: This should be added on the load of the other adapter but we don't know the id...
    AddViewResult = gen_adapter:add_cmds_views(undefined, system, imem, false, SystemViews),
    case lists:member(need_replace, AddViewResult) of
        true ->
            View = dderl_dal:get_view(undefined, <<"All ddViews">>, imem, system),
            dderl_dal:add_adapter_to_cmd(undefined, View#ddView.cmd, oci);
        _ ->
            [_, ViewId] = AddViewResult,
            View = dderl_dal:get_view(undefined, ViewId),
            dderl_dal:add_adapter_to_cmd(undefined, View#ddView.cmd, oci)
    end.

-spec add_conn_info(undefined | #priv{}, map()) -> #priv{}.
add_conn_info(undefined, ConnInfo) ->
    add_conn_info(#priv{connections = []}, ConnInfo);
add_conn_info(#priv{} = Priv, ConnInfo) when is_map(ConnInfo) ->
    Priv#priv{conn_info = ConnInfo}.

-spec connect_map(#ddConn{}) -> map().
connect_map(#ddConn{adapter = imem} = C) ->
    add_conn_extra(C, #{id => C#ddConn.id,
                        name => C#ddConn.name,
                        adapter => <<"imem">>,
                        owner => dderl_dal:user_name(C#ddConn.owner),
                        schema => atom_to_binary(C#ddConn.schm, utf8)}).

add_conn_extra(#ddConn{access = Access}, Conn)
  when is_map(Access), is_map(Conn) ->
        maps:merge(Conn, maps:remove(owner,maps:remove(<<"owner">>,Access)));
add_conn_extra(#ddConn{access = Access}, Conn) when is_list(Access), is_map(Conn) ->
    case proplists:get_value(type, Access, proplists:get_value(<<"method">>, Access, tcp)) of
        Local when Local == local; Local == <<"local">> ->
            Conn#{method => <<"local">>};
        LocalSec when LocalSec == local_sec; LocalSec == <<"local_sec">> ->
            Conn#{method => <<"local_sec">>,
                           user => proplists:get_value(user, Access, <<>>)};
        Rpc when Rpc == rpc; Rpc == <<"rpc">> ->
            Conn#{method => <<"rpc">>,
                  node => proplists:get_value(node, Access, <<>>),
                  user => proplists:get_value(user, Access, <<>>)};
        Tcp when Tcp == tcp; Tcp == <<"tcp">> ->
            Conn#{method => <<"tcp">>,
                  host => proplists:get_value(ip, Access, <<>>),
                  port => proplists:get_value(port, Access, <<>>),
                  user => proplists:get_value(user, Access, <<>>),
                  secure => proplists:get_value(secure, Access, false)}
    end.

-spec conn_method([{binary(),binary()}]|binary()) -> tcp | rpc | local.
conn_method(<<"rpc">>)          -> rpc;
conn_method(<<"local">>)        -> local;
conn_method(Other)
  when Other == <<"tcp">>;
       Other == '$not_defined'  -> tcp;
conn_method([{K,V}|_] = PropList) when is_binary(K), is_binary(V) ->
    conn_method(proplists:get_value(<<"method">>, PropList, '$not_defined')).

-spec connect_erlimem(tcp | rpc | local | local_sec, {atom(), pid()},
                      binary(), list({binary(),binary()}), map()) ->
    {ok, {atom(),pid()}, any()}.
connect_erlimem(tcp, _Sess, SessionId, Params, ConnInfo) ->
    Opts = case proplists:get_value(<<"secure">>, Params, false) of
               false -> [];
               true -> [ssl]
           end,
    IpAddr = binary_to_list(proplists:get_value(<<"host">>, Params)),
    Port = binary_to_integer(proplists:get_value(<<"port">>, Params)),
    Schema = try binary_to_existing_atom(
                 proplists:get_value(<<"schema">>, Params, '$bad_schema'),
                 utf8) of
               AtomSchema -> AtomSchema
           catch _:_ -> error("Invalid schema for tcp")
           end,
    connect_erlimem_password({tcp, IpAddr, Port, Opts}, Schema, SessionId, ConnInfo, Params);
connect_erlimem(rpc, _Sess, SessionId, Params, ConnInfo) ->
    Node = try binary_to_existing_atom(
                 proplists:get_value(<<"node">>, Params, '$bad_node'),
                 utf8) of
               AtomPort -> AtomPort
           catch _:_ -> error("Invalid node for rpc")
           end,
    Schema = try binary_to_existing_atom(
                 proplists:get_value(<<"schema">>, Params, '$bad_schema'),
                 utf8) of
               AtomSchema -> AtomSchema
           catch _:_ -> error("Invalid schema for rpc")
           end,
    connect_erlimem_password({rpc, Node}, Schema, SessionId, ConnInfo, Params);
connect_erlimem(local, Sess, SessionId, Params, ConnInfo) ->
    case proplists:get_value(<<"secure">>, Params, false) of
        true ->
            connect_erlimem(local_sec, Sess, SessionId, Params, ConnInfo);
        false ->
            case dderl_dal:can_connect_locally(Sess) of
                true ->
                    {ok, ErlImemSess} = erlimem:open(local, imem_meta:schema()),
                    {ok, ErlImemSess, #{node => list_to_binary(imem_meta:node_shard())}};
                _ -> error(<<"Local connection unauthorized">>)
            end
    end;
connect_erlimem(local_sec, ErlImemSess, _SessionId, _Params, _ConnInfo) ->
    {ok, ErlImemSess, #{node => list_to_binary(imem_meta:node_shard())}}.

connect_erlimem_password(Connect, Schema, SessionId, ConnInfo, Params) ->
    {ok, ErlImemSess}
    = case erlimem:open(Connect, Schema) of
          {ok, _} = E -> E;
          Error ->
              error(list_to_binary(io_lib:format("~p", [Error])))
      end,
    case ErlImemSess:auth(dderl,SessionId,{access,ConnInfo#{type => internal}}) of
        {ok, [{pwdmd5,_}|_]} ->
            User = proplists:get_value(<<"user">>, Params, <<>>),
            Password = proplists:get_value(<<"password">>, Params, []),
            case ErlImemSess:auth(dderl,SessionId,{pwdmd5,{User,list_to_binary(Password)}}) of
                Ok when Ok == {ok,[]}; Ok == ok ->
                    case ErlImemSess:run_cmd(login,[]) of
                        {error,{{'SecurityException',{?PasswordChangeNeeded,_}},ST}} ->
                            ?Warn("Password expired ~s~n~p", [User, ST]),
                            {ok, ErlImemSess, #{changePass=>User}};
                        _ ->
                            {ok, ErlImemSess, '$no_extra'}
{'fraction_non_alphanumeric': 0.15788456583354807, 'fraction_numerical': 0.003091986601391394, 'mean_word_length': 2.201691070323778, 'pattern_counts': {'":': 0, '<': 163, '<?xml version=': 0, '>': 292, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '74213', 'n_tokens_mistral': 5148, 'n_tokens_neox': 4982, 'n_words': 960}