<commit_before>%%%-------------------------------------------------------------------
%%% @author <NAME>
%%% @copyright (C) 2017 ACK CYFRONET AGH
%%% This software is released under the MIT license
%%% cited in 'LICENSE.txt'.
%%% @end
%%%-------------------------------------------------------------------
%%% @doc Driver that coordinates access to memory stores.
%%% @end
%%%-------------------------------------------------------------------
-module(memory_store_driver).
-author("<NAME>").
-behaviour(tp_behaviour).

-include("global_definitions.hrl").
-include("modules/tp/tp.hrl").
-include("modules/datastore/memory_store_driver.hrl").
-include("modules/datastore/datastore_common.hrl").
-include("modules/datastore/datastore_models_def.hrl").
-include("modules/datastore/datastore_common_internal.hrl").
-include_lib("ctool/include/logging.hrl").

% TODO - add option - enable mnesia transaction log and/or replication
% of data to other nodes
% TODO - add setting of cache level - from simple ets to dumping to
% couch immediately
% TODO - check size of key and doc during modify operation

%% API
-export([modify/2, init/1, terminate/1, commit/2, merge_changes/2,
  commit_backoff/1]).
%% Helper functions
-export([driver_to_level/1, resolve_conflict/3]).

% Types
-type state() :: #state{}.
-type value_doc() :: datastore:document() | undefined | not_found.
-type value_link() :: list().
-type message() :: {atom(), list()}.
-type resolved_conflict() :: {Document :: datastore:document(),
  Bucket :: datastore:bucket(), ToDel :: false | datastore:document()}.
-type change() :: ok | to_save | {to_save, resolved_conflict()} |
  {[datastore:ext_key()], [{datastore:ext_key(), resolved_conflict()}]}.

-export_type([value_doc/0, value_link/0, message/0]).

%%%===================================================================
%%% API functions
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Handles all operation executed at memory store.
%% @end
%%--------------------------------------------------------------------
-spec modify(Messages :: [message()], State :: state()) ->
  {Answers :: list(), {true, change()} | false, NewState :: state()}.
modify([{clear, _}], #state{link_proc = true, key = Key, driver = Driver,
  model_config = ModelConfig} = State) ->
  Ans = memory_store_driver_links:clear(Driver, ModelConfig, Key),
  {[Ans], false, State};
modify([{clear, _}], #state{driver = Driver, model_config = ModelConfig,
  key = Key} = State) ->
  Ans = memory_store_driver_docs:clear(Driver, ModelConfig, Key),
  {[Ans], false, State};
modify(Messages, #state{link_proc = LP, current_value = CurrentValue,
  driver = Driver, flush_driver = FD,
  model_config = ModelConfig, key = Key} = State) ->
  FilteredMessages = lists:filter(fun
    ({clear, _}) -> false;
    (_) -> true
  end, Messages),
  {A, NV, Changes} = try
    case LP of
      true ->
        memory_store_driver_links:handle_link_messages(FilteredMessages,
          CurrentValue, Driver, FD, ModelConfig, Key);
      _ ->
        memory_store_driver_docs:handle_messages(FilteredMessages,
          CurrentValue, Driver, FD, ModelConfig, Key)
    end
  catch
    throw:{Reason, ThrowNV, ThrowChanges} ->
      ?error_stacktrace("Modify error for key ~p: ~p", [Key, Reason]),
      ThrowA = lists:map(fun(_) -> Reason end, FilteredMessages),
      {ThrowA, ThrowNV, ThrowChanges}
  end,

  FinalChanges = case {FD, Changes} of
    {undefined, _} -> false;
    {_, {[], []}} -> false;
    {_, ok} -> false;
    _ -> {true, Changes}
  end,
  case length(FilteredMessages) =:= length(Messages) of
    true ->
      {A, FinalChanges, State#state{current_value = NV}};
    _ ->
      {ReversedAns, []} = lists:foldl(fun
        ({clear, _}, {Acc, AnsList}) ->
          {[ok | Acc], AnsList};
        (_, {Acc, [Ans | AnsList]}) ->
          {[Ans | Acc], AnsList}
      end, {[], A}, Messages),
      {lists:reverse(ReversedAns), FinalChanges,
        State#state{current_value = NV}}
  end.

%%--------------------------------------------------------------------
%% @doc
%% Initializes state of memory driver.
%% @end
%%--------------------------------------------------------------------
-spec init(Args :: list()) -> {ok, tp:init()}.
% TODO - add arg for save and create etc operations that get is not needed.
init([Driver, MC, Key, FD, LinkProc]) ->
  init([Driver, MC, Key, FD, LinkProc, caches_controller:get_idle_timeout()]);
init([Driver, MC, Key, FD, true = LinkProc, IdleT]) ->
  {ok, #tp_init{data = #state{driver = Driver, model_config = MC, key = Key,
    link_proc = LinkProc, flush_driver = FD, current_value = []},
    idle_timeout = IdleT, min_commit_delay = get_flush_min_interval(FD),
    max_commit_delay = get_flush_max_interval(FD)}};
init([Driver, MC, Key, FD, LinkProc, IdleT]) ->
  {ok, #tp_init{data = #state{driver = Driver, model_config = MC, key = Key,
    link_proc = LinkProc, flush_driver = FD}, idle_timeout = IdleT,
    min_commit_delay = get_flush_min_interval(FD),
    max_commit_delay = get_flush_max_interval(FD)}}.

%%--------------------------------------------------------------------
%% @doc
%% Checks if memory store driver can be stopped.
%% @end
%%--------------------------------------------------------------------
-spec terminate(State :: state()) -> ok | {error, term()}.
terminate(#state{flush_driver = undefined}) ->
  ok;
terminate(#state{link_proc = true, key = Key, driver = Driver,
  model_config = ModelConfig}) ->
  case application:get_env(?CLUSTER_WORKER_APP_NAME,
    tp_proc_terminate_clear_memory) of
    {ok, true} ->
      memory_store_driver_links:clear(Driver, ModelConfig, Key);
    _ ->
      ok
  end,
  ok;
terminate(#state{key = Key, driver = Driver, model_config = ModelConfig}) ->
  case application:get_env(?CLUSTER_WORKER_APP_NAME,
    tp_proc_terminate_clear_memory) of
    {ok, true} ->
      memory_store_driver_docs:clear(Driver, ModelConfig, Key);
    _ ->
      ok
  end,
  ok.

%%--------------------------------------------------------------------
%% @doc
%% Saves changes on persistent storage if possible.
%% @end
%%--------------------------------------------------------------------
-spec commit(Modified :: change(), State :: state()) -> true | {false, change()}.
commit({ModifiedKeys, []}, #state{model_config = MC, current_value = CV,
  flush_driver = Driver, link_proc = true}) ->
  ModifiedList = lists:map(fun(K) ->
    {K, proplists:get_value(K, CV)}
  end, ModifiedKeys),

  NotSaved = dump_docs(MC, Driver, ModifiedList),

  case NotSaved of
    [] ->
      true;
    _ ->
      {false, NotSaved}
  end;
commit({ModifiedKeys, ResolvedChanges}, #state{model_config = MC, current_value = CV,
  flush_driver = Driver, link_proc = true} = State) ->
  NotSaved = dump_docs(MC, Driver, ResolvedChanges),

  RevsToDel = lists:foldl(fun({K, {_Document, _Bucket, ToDel}}, Acc) ->
    case lists:member(K, NotSaved) of
      true ->
        Acc;
      _ ->
        [{K, {delete_doc_asynch, ToDel}} | Acc]
    end
  end, [], ResolvedChanges),
  NotSaved2 = dump_docs(MC, Driver, RevsToDel),

  case {NotSaved, NotSaved2} of
    {[], []} ->
      ModifiedKeys2 = lists:foldl(fun(K, Acc) ->
        V = proplists:get_value(K, CV),
        {ToCheck, _, _} = proplists:get_value(K, ResolvedChanges),
        case check_resolved_doc(V, ToCheck) of
          true ->
            Acc;
          _ ->
            [K | Acc]
        end
      end, [], ModifiedKeys),

      commit({ModifiedKeys2, []}, State);
    _ ->
      ResolvedChanges2 = lists:foldl(fun(K, Acc) ->
        [proplists:get_value(K, ResolvedChanges) | Acc]
      end, [], NotSaved),

      ResolvedChanges2_2 = lists:foldl(fun(K, Acc) ->
        [proplists:get_value(K, RevsToDel) | Acc]
      end, [], NotSaved2),
      % TODO - delete old rev even if new revision appears before successful del
      {false, {ModifiedKeys, ResolvedChanges2 ++ ResolvedChanges2_2}}
  end;
commit(ok, _State)->
  true;
commit(to_save, #state{model_config = MC, key = Key, current_value = CV,
  flush_driver = Driver})->
  Ans = case CV of
    not_found -><commit_msg>Hotfix - minor update after merge<commit_after>%%%-------------------------------------------------------------------
%%% @author <NAME>
%%% @copyright (C) 2017 ACK CYFRONET AGH
%%% This software is released under the MIT license
%%% cited in 'LICENSE.txt'.
%%% @end
%%%-------------------------------------------------------------------
%%% @doc Driver that coordinates access to memory stores.
%%% @end
%%%-------------------------------------------------------------------
-module(memory_store_driver).
-author("<NAME>").
-behaviour(tp_behaviour).

-include("global_definitions.hrl").
-include("modules/tp/tp.hrl").
-include("modules/datastore/memory_store_driver.hrl").
-include("modules/datastore/datastore_common.hrl").
-include("modules/datastore/datastore_models_def.hrl").
-include("modules/datastore/datastore_common_internal.hrl").
-include_lib("ctool/include/logging.hrl").

% TODO - add option - enable mnesia transaction log and/or replication
% of data to other nodes
% TODO - add setting of cache level - from simple ets to dumping to
% couch immediately
% TODO - check size of key and doc during modify operation

%% API
-export([modify/2, init/1, terminate/1, commit/2, merge_changes/2,
  commit_backoff/1]).
%% Helper functions
-export([driver_to_level/1, resolve_conflict/3]).

% Types
-type state() :: #state{}.
-type value_doc() :: datastore:document() | undefined | not_found.
-type value_link() :: list().
-type message() :: {atom(), list()}.
-type resolved_conflict() :: {Document :: datastore:document(),
  Bucket :: datastore:bucket(), ToDel :: false | datastore:document()}.
-type change() :: ok | to_save | {to_save, resolved_conflict()} |
  {[datastore:ext_key()], [{datastore:ext_key(), resolved_conflict()}]}.

-export_type([value_doc/0, value_link/0, message/0]).

%%%===================================================================
%%% API functions
%%%===================================================================

%%--------------------------------------------------------------------
%% @doc
%% Handles all operation executed at memory store.
%% @end
%%--------------------------------------------------------------------
-spec modify(Messages :: [message()], State :: state()) ->
  {Answers :: list(), {true, change()} | false, NewState :: state()}.
modify([{clear, _}], #state{link_proc = true, key = Key, driver = Driver,
  model_config = ModelConfig} = State) ->
  Ans = memory_store_driver_links:clear(Driver, ModelConfig, Key),
  {[Ans], false, State};
modify([{clear, _}], #state{driver = Driver, model_config = ModelConfig,
  key = Key} = State) ->
  Ans = memory_store_driver_docs:clear(Driver, ModelConfig, Key),
  {[Ans], false, State};
modify(Messages, #state{link_proc = LP, current_value = CurrentValue,
  driver = Driver, flush_driver = FD,
  model_config = ModelConfig, key = Key} = State) ->
  FilteredMessages = lists:filter(fun
    ({clear, _}) -> false;
    (_) -> true
  end, Messages),
  {A, NV, Changes} = try
    case LP of
      true ->
        memory_store_driver_links:handle_link_messages(FilteredMessages,
          CurrentValue, Driver, FD, ModelConfig, Key);
      _ ->
        memory_store_driver_docs:handle_messages(FilteredMessages,
          CurrentValue, Driver, FD, ModelConfig, Key)
    end
  catch
    throw:{Reason, ThrowNV, ThrowChanges} ->
      ?error_stacktrace("Modify error for key ~p: ~p", [Key, Reason]),
      ThrowA = lists:map(fun(_) -> Reason end, FilteredMessages),
      {ThrowA, ThrowNV, ThrowChanges}
  end,

  FinalChanges = case {FD, Changes} of
    {undefined, _} -> false;
    {_, {[], []}} -> false;
    {_, ok} -> false;
    _ -> {true, Changes}
  end,
  case length(FilteredMessages) =:= length(Messages) of
    true ->
      {A, FinalChanges, State#state{current_value = NV}};
    _ ->
      {ReversedAns, []} = lists:foldl(fun
        ({clear, _}, {Acc, AnsList}) ->
          {[ok | Acc], AnsList};
        (_, {Acc, [Ans | AnsList]}) ->
          {[Ans | Acc], AnsList}
      end, {[], A}, Messages),
      {lists:reverse(ReversedAns), FinalChanges,
        State#state{current_value = NV}}
  end.

%%--------------------------------------------------------------------
%% @doc
%% Initializes state of memory driver.
%% @end
%%--------------------------------------------------------------------
-spec init(Args :: list()) -> {ok, tp:init()}.
% TODO - add arg for save and create etc operations that get is not needed.
init([Driver, MC, Key, FD, LinkProc]) ->
  init([Driver, MC, Key, FD, LinkProc, caches_controller:get_idle_timeout()]);
init([Driver, MC, Key, FD, true = LinkProc, IdleT]) ->
  {ok, #tp_init{data = #state{driver = Driver, model_config = MC, key = Key,
    link_proc = LinkProc, flush_driver = FD, current_value = []},
    idle_timeout = IdleT, min_commit_delay = get_flush_min_interval(FD),
    max_commit_delay = get_flush_max_interval(FD)}};
init([Driver, MC, Key, FD, LinkProc, IdleT]) ->
  {ok, #tp_init{data = #state{driver = Driver, model_config = MC, key = Key,
    link_proc = LinkProc, flush_driver = FD}, idle_timeout = IdleT,
    min_commit_delay = get_flush_min_interval(FD),
    max_commit_delay = get_flush_max_interval(FD)}}.

%%--------------------------------------------------------------------
%% @doc
%% Checks if memory store driver can be stopped.
%% @end
%%--------------------------------------------------------------------
-spec terminate(State :: state()) -> ok | {error, term()}.
terminate(#state{flush_driver = undefined}) ->
  ok;
terminate(#state{link_proc = true, key = Key, driver = Driver,
  model_config = ModelConfig}) ->
  case application:get_env(?CLUSTER_WORKER_APP_NAME,
    tp_proc_terminate_clear_memory) of
    {ok, true} ->
      memory_store_driver_links:clear(Driver, ModelConfig, Key);
    _ ->
      ok
  end,
  ok;
terminate(#state{key = Key, driver = Driver, model_config = ModelConfig}) ->
  case application:get_env(?CLUSTER_WORKER_APP_NAME,
    tp_proc_terminate_clear_memory) of
    {ok, true} ->
      memory_store_driver_docs:clear(Driver, ModelConfig, Key);
    _ ->
      ok
  end,
  ok.

%%--------------------------------------------------------------------
%% @doc
%% Saves changes on persistent storage if possible.
%% @end
%%--------------------------------------------------------------------
-spec commit(Modified :: change(), State :: state()) -> true | {false, change()}.
commit({ModifiedKeys, []}, #state{model_config = MC, current_value = CV,
  flush_driver = Driver, link_proc = true}) ->
  ModifiedList = lists:map(fun(K) ->
    {K, proplists:get_value(K, CV)}
  end, ModifiedKeys),

  NotSaved = dump_docs(MC, Driver, ModifiedList),

  case NotSaved of
    [] ->
      true;
    _ ->
      {false, NotSaved}
  end;
commit({ModifiedKeys, ResolvedChanges}, #state{model_config = MC, current_value = CV,
  flush_driver = Driver, link_proc = true} = State) ->
  NotSaved = dump_docs(MC, Driver, ResolvedChanges),

  RevsToDel = lists:foldl(fun({K, {_Document, _Bucket, ToDel}}, Acc) ->
    case lists:member(K, NotSaved) of
      true ->
        Acc;
      _ ->
        [{K, {delete_doc_asynch, ToDel}} | Acc]
    end
  end, [], ResolvedChanges),
  NotSaved2 = dump_docs(MC, Driver, RevsToDel),

  case {NotSaved, NotSaved2} of
    {[], []} ->
      ModifiedKeys2 = lists:foldl(fun(K, Acc) ->
        V = proplists:get_value(K, CV),
        {ToCheck, _, _} = proplists:get_value(K, ResolvedChanges),
        case proplists:get_value(K, ResolvedChanges) of
          {ToCheck, _, _} ->
            case check_resolved_doc(V, ToCheck) of
              true ->
                Acc;
              _ ->
                [K | Acc]
            end;
          _ ->
            [K | Acc]
        end
      end, [], ModifiedKeys),

      commit({ModifiedKeys2, []}, State);
    _ ->
      ResolvedChanges2 = lists:foldl(fun(K, Acc) ->
        [proplists:get_value(K, ResolvedChanges) | Acc]
      end, [], NotSaved),

      ResolvedChanges2_2 = lists:foldl(fun(K, Acc) ->
        [proplists:get_value(K, RevsToDel) | Acc]
      end, [], NotSaved2),
      % TODO - delete old rev even if new revision appears before successful del
      {false, {ModifiedKeys, ResolvedChanges2 ++ ResolvedChanges2_2}}
  end;
commit(ok, _State)->
  true;
commit(to_save, #state{model_config = MC, key = Key, current_value = CV,
  flush_driver = Driver})->
  Ans = case CV of
    not_found ->
{'fraction_non_alphanumeric': 0.247990329404654, 'fraction_numerical': 0.0031429434874584465, 'mean_word_length': 2.522674047264211, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 107, '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': '4930425', 'n_tokens_mistral': 5142, 'n_tokens_neox': 5029, 'n_words': 1192}