ra

%% @hidden
-module(ra_log).

-compile([inline_list_funcs]).

-export([init/1,
         close/1,
         append/2,
         write/2,
         append_sync/2,
         write_sync/2,
         take/3,
         last_index_term/1,
         set_last_index/2,
         handle_event/2,
         last_written/1,
         fetch/2,
         fetch_term/2,
         next_index/1,
         install_snapshot/2,
         read_snapshot/1,
         snapshot_index_term/1,
         update_release_cursor/4,
         %% meta
         read_meta/2,
         read_meta/3,
         write_meta/3,
         write_meta/4,

         can_write/1,
         exists/2,
         overview/1,
         %% config
         write_config/2,
         read_config/1,

         delete_everything/1,
         release_resources/2
        ]).

-include("ra.hrl").

-define(METRICS_CACHE_POS, 2).
-define(METRICS_OPEN_MEM_TBL_POS, 3).
-define(METRICS_CLOSED_MEM_TBL_POS, 4).
-define(METRICS_SEGMENT_POS, 5).

-define(DEFAULT_RESEND_WINDOW_SEC, 20).
-define(SNAPSHOT_INTERVAL, 4096).
-define(LOG_APPEND_TIMEOUT, 5000).

-type ra_meta_key() :: atom().
-type snapshot() :: ra_log_snapshot:state().
-type segment_ref() :: {From :: ra_index(), To :: ra_index(),
                        File :: string()}.
-type event_body() :: {written, {From :: ra_index(),
                                 To :: ra_index(),
                                 ToTerm :: ra_term()}} |
                      {segments, ets:tid(), [segment_ref()]} |
                      {resend_write, ra_index()} |
                      {snapshot_written,
                       ra_idxterm(),
                       File :: file:filename(),
                       Old :: [file:filename()]}.

-type event() :: {ra_log_event, event_body()}.

-record(state,
        {uid :: ra_uid(),
         first_index = -1 :: ra_index(),
         last_index = -1 :: -1 | ra_index(),
         last_term = 0 :: ra_term(),
         last_written_index_term = {0, 0} :: ra_idxterm(),
         segment_refs = [] :: [segment_ref()],
         open_segments = ra_flru:new(5, fun flru_handler/1) :: ra_flru:state(),
         directory :: file:filename(),
         snapshot_state :: maybe({ra_index(), ra_term(),
                                  maybe(file:filename())}),
         snapshot_interval = ?SNAPSHOT_INTERVAL :: non_neg_integer(),
         % if this is set a snapshot write is in progress for the
         % index specified
         snapshot_index_in_progress :: maybe(ra_index()),
         cache = #{} :: #{ra_index() => {ra_term(), log_entry()}},
         wal :: atom(), % registered name
         last_resend_time :: maybe(integer()),
         resend_window_seconds = ?DEFAULT_RESEND_WINDOW_SEC :: integer()
        }).

-type ra_log() :: #state{}.

-type ra_log_init_args() :: #{data_dir => string(),
                              uid := ra_uid(),
                              wal => atom(),
                              snapshot_interval => non_neg_integer(),
                              resend_window => integer(),
                              max_open_segments => non_neg_integer()}.

-export_type([ra_log_init_args/0,
              ra_log/0,
              ra_meta_key/0,
              snapshot/0,
              segment_ref/0,
              event/0,
              event_body/0
             ]).

-spec init(ra_log_init_args()) -> ra_log().
init(#{uid := UId} = Conf) ->
    %% overriding the data_dir is only here for test compatibility
    %% as it needs to match what the segment writer has it makes no real
    %% sense to make it independently configurable
    Dir = case Conf of
              #{data_dir := D} -> D;
              _ ->
                  ra_env:server_data_dir(UId)
          end,
    MaxOpen = maps:get(max_open_segments, Conf, 5),

    ok =  ra_lib:ensure_dir(Dir),

    % initialise metrics for this server
    true = ets:insert(ra_log_metrics, {UId, 0, 0, 0, 0}),
    Wal = maps:get(wal, Conf, ra_log_wal),
    ResendWindow = maps:get(resend_window, Conf, ?DEFAULT_RESEND_WINDOW_SEC),
    SnapInterval = maps:get(snapshot_interval, Conf, ?SNAPSHOT_INTERVAL),

    % create subdir for log id
    % TODO: safely encode UId for use a directory name

    % recover current range and any references to segments
    {{FirstIdx, LastIdx0}, SegRefs} = case recover_range(UId) of
                                          {undefined, SRs} ->
                                              {{-1, -1}, SRs};
                                          R ->  R
                                      end,
    % recove las snapshot file
    SnapshotState =
        case lists:sort(filelib:wildcard(filename:join(Dir, "*.snapshot"))) of
            [File | _] ->
                %% TODO provide function that only reads the index and term
                %% of the snapshot file.
                {ok, {SI, ST}} = ra_log_snapshot:read_indexterm(File),
                {SI, ST, File};
            [] ->
                undefined
        end,
    {SnapIdx, SnapTerm} = case SnapshotState of
                              undefined -> {-1, -1};
                              {I, T, _} -> {I, T}
                          end,
    State000 = #state{directory = Dir,
                      uid = UId,
                      first_index = max(SnapIdx + 1, FirstIdx),
                      last_index = max(SnapIdx, LastIdx0),
                      segment_refs = SegRefs,
                      snapshot_state = SnapshotState,
                      snapshot_interval = SnapInterval,
                      wal = Wal,
                      open_segments = ra_flru:new(MaxOpen, fun flru_handler/1),
                      resend_window_seconds = ResendWindow},

    LastIdx = State000#state.last_index,
    % recover the last term
    {LastTerm0, State00} = case LastIdx of
                               SnapIdx ->
                                   {SnapTerm, State000};
                               -1 ->
                                   {0, State000};
                               LI ->
                                   fetch_term(LI, State000)
                           end,
    LastTerm = ra_lib:default(LastTerm0, -1),
    State0 = State00#state{last_term = LastTerm,
                           last_written_index_term = {LastIdx, LastTerm}},


    % initialized with a default 0 index 0 term dummy value
    % and an empty meta data map
    State = maybe_append_0_0_entry(State0),
    ?INFO("~s: ra_log:init recovered last_index_term ~w~n",
          [State#state.uid, last_index_term(State)]),
    State.

-spec close(ra_log()) -> ok.
close(#state{open_segments = OpenSegs}) ->
    % deliberately ignoring return value
    % close all open segments
    _ = ra_flru:evict_all(OpenSegs),
    ok.

-spec append(Entry :: log_entry(), State :: ra_log()) ->
    ra_log() | no_return().
append(Entry, #state{last_index = LastIdx} = State0)
      when element(1, Entry) =:= LastIdx + 1 ->
    wal_write(State0, Entry);
append({Idx, _, _}, #state{last_index = LastIdx}) ->
    Msg = lists:flatten(io_lib:format("tried writing ~b - expected ~b",
                                      [Idx, LastIdx+1])),
    exit({integrity_error, Msg}).

-spec write(Entries :: [log_entry()], State :: ra_log()) ->
    {ok, ra_log()} |
    {error, {integrity_error, term()} | wal_down}.
write([{FstIdx, _, _} = First | Rest] = Entries,
      #state{last_index = LastIdx,
             snapshot_state = SnapState} = State00)
  when FstIdx =< LastIdx + 1 andalso FstIdx >= 0 ->
    case SnapState of
        {SnapIdx, _, _} when FstIdx =:= SnapIdx + 1 ->
            % it is the next entry after a snapshot
            % we need to tell the wal to truncate as we
            % are not going to receive any entries prior to the snapshot
            State0 = wal_truncate_write(State00, First),
            % write the rest normally
            % TODO: batch api for wal
            write_entries(Rest, State0);
        _ ->
            write_entries(Entries, State00)
    end;
write([], State) ->
    {ok, State};
write([{Idx, _, _} | _], #state{uid = UId, last_index = LastIdx}) ->
    Msg = lists:flatten(io_lib:format("~p: ra_log:write/2 "
                                      "tried writing ~b - expected ~b",
                                      [UId, Idx, LastIdx+1])),
    {error, {integrity_error, Msg}}.

-spec take(ra_index(), non_neg_integer(), ra_log()) ->
    {[log_entry()], ra_log()}.
take(Start, Num, #state{uid = UId, first_index = FirstIdx,
                        last_index = LastIdx} = State)
  when Start >= FirstIdx andalso Start =< LastIdx ->
    % 0. Check that the request isn't outside of first_index and last_index
    % 1. Check the local cache for any unflushed entries, carry reminders
    % 2. Check ra_log_open_mem_tables
    % 3. Check ra_log_closed_mem_tables in turn
    % 4. Check on disk segments in turn
    case cache_take(Start, Num, State) of
        {Entries, MetricOps0, undefined} ->
            ok = update_metrics(UId, MetricOps0),
            {Entries, State};
        {Entries0, MetricOps0, Rem0} ->
            case open_mem_tbl_take(UId, Rem0, MetricOps0, Entries0) of
                {Entries1, MetricOps, undefined} ->
                    ok = update_metrics(UId, MetricOps),
                    {Entries1, State};
                {Entries1, MetricOps1, Rem1} ->
                    case closed_mem_tbl_take(UId, Rem1, MetricOps1, Entries1) of
                        {Entries2, MetricOps, undefined} ->
                            ok = update_metrics(UId, MetricOps),
                            {Entries2, State};
                        {Entries2, MetricOps2, {S, E} = Rem2} ->
                            case catch segment_take(State, Rem2, Entries2) of
                                {Open, undefined, Entries} ->
                                    MOp = {?METRICS_SEGMENT_POS, E - S + 1},
                                    ok = update_metrics(UId,
                                                        [MOp | MetricOps2]),
                                    {Entries, State#state{open_segments = Open}}
                            end
                    end
            end
    end;
take(_Start, _Num, State) ->
    {[], State}.

-spec last_index_term(ra_log()) -> ra_idxterm().
last_index_term(#state{last_index = LastIdx, last_term = LastTerm}) ->
    {LastIdx, LastTerm}.

-spec last_written(ra_log()) -> ra_idxterm().
last_written(#state{last_written_index_term = LWTI}) ->
    LWTI.

%% forces the last index and last written index back to a prior index
-spec set_last_index(ra_index(), ra_log()) ->
    {ok, ra_log()} | {not_found, ra_log()}.
set_last_index(Idx, #state{last_written_index_term = {LWIdx0, _}} = State0) ->
    case fetch_term(Idx, State0) of
        {undefined, State} ->
            {not_found, State};
        {Term, State1} ->
            LWIdx = min(Idx, LWIdx0),
            {LWTerm, State2} = fetch_term(LWIdx, State1),
            %% this should always be found but still assert just in case
            true = LWTerm =/= undefined,
            {ok, State2#state{last_index = Idx,
                              last_term = Term,
                              last_written_index_term = {LWIdx, LWTerm}}}
    end.

-spec handle_event(event_body(), ra_log()) ->
    ra_log().
handle_event({written, {FromIdx, ToIdx, Term}},
             #state{last_written_index_term = {LastWrittenIdx0,
                                               LastWrittenTerm0},
                    snapshot_state = SnapState} = State0)
  when FromIdx =< LastWrittenIdx0 + 1 ->
    % We need to ignore any written events for the same index
    % but in a prior term if we do not we may end up confirming
    % to a leader writes that have not yet
    % been fully flushed
    case fetch_term(ToIdx, State0) of
        {Term, State} ->
            % this case truncation shouldn't be too expensive as the cache
            % only containes the unflushed window of entries typically less than
            % 10ms worth of entries
            truncate_cache(ToIdx,
                           State#state{last_written_index_term = {ToIdx, Term}});
        {undefined, State} when FromIdx =< element(1, SnapState) ->
            % A snapshot happened before the written event came in
            % This can only happen on a leader when consensus is achieved by
            % followers returning appending the entry and the leader committing
            % and processing a snapshot before the written event comes in.
            % ensure last_written_index_term does not go backwards
            LastWrittenIdxTerm = {max(LastWrittenIdx0, ToIdx),
                                  max(LastWrittenTerm0, Term)},
            State#state{last_written_index_term = LastWrittenIdxTerm};
        {_X, State} ->
            ?INFO("~s: written event did not find term ~b for index ~b "
                  "found ~w",
                  [State#state.uid, Term, ToIdx, _X]),
            State
    end;
handle_event({written, {FromIdx, _, _}}, %% ToIdx, Term
             #state{uid = UId,
                    last_written_index_term = {LastWrittenIdx, _}} = State0)
  when FromIdx > LastWrittenIdx + 1 ->
    % leaving a gap is not ok - resend from cache
    Expected = LastWrittenIdx + 1,
    ?WARN("~s: ra_log: written gap detected at ~b expected ~b!",
         [UId, FromIdx, Expected]),
    resend_from(Expected, State0);
handle_event({segments, Tid, NewSegs},
             #state{uid = UId,
                    open_segments = Open0,
                    segment_refs = SegmentRefs} = State0) ->
    % Append new segment refs
    % mem_table cleanup
    % any closed mem tables older than the one just having been flushed should
    % be ok to delete
    ClosedTables = closed_mem_tables(UId),
    Active = lists:takewhile(fun ({_, _, _, _, T}) -> T =/= Tid end,
                             ClosedTables),
    % not fast but we rarely should have more than one or two closed tables
    % at any time
    Obsolete = ClosedTables -- Active,
    % assert at least one table was found
    % commented out as if the wal restarts more than once in short succession
    % we may be processing a segments event referring to open mem tables
    % that the wal replaced during init. the segment file is still valid and
    % we'll probably receive another copy shortly.
    % [_|_] = Obsolete,
    [begin
         true = ets:delete_object(ra_log_closed_mem_tables, ClosedTbl),
         % Then delete the actual ETS table
         true = ets:delete(T)
     end || {_, _, _, _, T} = ClosedTbl  <- Obsolete],
    %% check if any of the segrefs refer to open segment and re-load their
    %% new indexs if so
    Open = lists:foldl(fun ({_, _, F}, Acc0) ->
                               case ra_flru:evict(F, Acc0) of
                                   {_, Acc} ->
                                       %% TODO: evicting here could result in
                                       %% unecessary
                                       %% file handle churn but may not be worth
                                       %% optimising
                                       %% Alternative would be something like:
                                       %% ra_log_segment:reload_index/1
                                       Acc;
                                   error ->
                                       Acc0
                               end
                       end, Open0, SegmentRefs),

    % compact seg ref list so that only the latest range for a segment
    % file has an entry
    State0#state{segment_refs = compact_seg_refs(NewSegs ++ SegmentRefs),
                 open_segments = Open,
                 % re-enable snapshots based on release cursor updates
                 % in case snapshot_written was lost
                 snapshot_index_in_progress = undefined};
handle_event({snapshot_written, {Idx, Term}, File, Old},
             #state{uid = UId, open_segments = OpenSegs0,
                    directory = Dir,
                    segment_refs = SegRefs0} = State0) ->
    % delete any segments outside of first_index
    {SegRefs, OpenSegs, NumObsolete} =
        % all segments created prior to the first reclaimable should be
        % reclaimed even if they have a more up to date end point
        case lists:partition(fun({_, To, _}) when To > Idx -> true;
                                (_) -> false
                             end, SegRefs0) of
            {_, []} ->
                {SegRefs0, OpenSegs0, 0};
            {Active, Obsolete} ->
                % close all relevant active segments
                ObsoleteKeys = [element(3, O) || O <- Obsolete],
                % close any open segments
                OpenSegs1 = lists:foldl(fun (K, OS0) ->
                                                case ra_flru:evict(K, OS0) of
                                                    {_, OS} ->
                                                        OS;
                                                    error ->
                                                        OS0
                                                end
                                        end, OpenSegs0, ObsoleteKeys),
                % [ok = ra_log_segment:close(S)
                %  || S <- maps:values(maps:with(ObsoleteKeys, OpenSegs0))],
                ObsoleteFiles = [filename:join(Dir, O) || O <- ObsoleteKeys],
                ok = ra_log_segment_writer:delete_segments(UId, Idx,
                                                           ObsoleteFiles),
                {Active, OpenSegs1, length(ObsoleteKeys)}
        end,
    ?INFO("~s: Snapshot written at index ~b in term ~b. ~b"
          " obsolete segments~n",
          [UId, Idx, Term, NumObsolete]),
    true = ets:insert(ra_log_snapshot_state, {UId, Idx}),
    %% delete old files
    [file:delete(F) || F <- Old],
    truncate_cache(Idx,
                   State0#state{first_index = Idx + 1,
                                segment_refs = SegRefs,
                                open_segments = OpenSegs,
                                snapshot_index_in_progress = undefined,
                                snapshot_state = {Idx, Term, File}});
handle_event({resend_write, Idx}, State) ->
    % resend missing entries from cache.
    % The assumption is they are available in the cache
    resend_from(Idx, State).

-spec next_index(ra_log()) -> ra_index().
next_index(#state{last_index = LastIdx}) ->
    LastIdx + 1.

-spec fetch(ra_index(), ra_log()) ->
    {maybe(log_entry()), ra_log()}.
fetch(Idx, State0) ->
    case take(Idx, 1, State0) of
        {[], State} ->
            {undefined, State};
        {[Entry], State} ->
            {Entry, State}
    end.

-spec fetch_term(ra_index(), ra_log()) ->
    {maybe(ra_term()), ra_log()}.
fetch_term(Idx, #state{last_index = LastIdx,
                       first_index = FirstIdx} = State0)
  when Idx < FirstIdx orelse Idx > LastIdx ->
    {undefined, State0};
fetch_term(Idx, #state{cache = Cache, uid = UId} = State0) ->
    case Cache of
        #{Idx := {Term, _}} ->
            {Term, State0};
        _ ->
            case ets:lookup(ra_log_open_mem_tables, UId) of
                [{_, From, To, Tid}] when Idx >= From andalso Idx =< To ->
                    Term = ets:lookup_element(Tid, Idx, 2),
                    {Term, State0};
                _ ->
                    case closed_mem_table_term_query(Idx, UId) of
                        undefined ->
                            segment_term_query(Idx, State0);
                        Term ->
                            {Term, State0}
                    end
            end
    end.

-spec install_snapshot(Snapshot :: ra_log:snapshot(),
                       State :: ra_log()) ->
    ra_log().
install_snapshot({Idx, Term, _, _} = Snapshot,
                 #state{directory = Dir} = State) ->
    % syncronous call when follower receives a snapshot
    {ok, File, Old} = ra_log_snapshot_writer:write_snapshot_call(Dir, Snapshot),
    handle_event({snapshot_written, {Idx, Term}, File, Old},
                 State#state{last_index = Idx,
                             last_written_index_term = {Idx, Term}}).

-spec read_snapshot(State :: ra_log()) ->
    maybe(ra_log:snapshot()).
read_snapshot(#state{snapshot_state = undefined}) ->
    undefined;
read_snapshot(#state{snapshot_state = {_, _, File}}) ->
    case ra_log_snapshot:read(File) of
        {ok, Snapshot} ->
            Snapshot;
        {error, enoent} ->
            undefined
    end.

-spec snapshot_index_term(State :: ra_log()) -> maybe(ra_idxterm()).
snapshot_index_term(#state{snapshot_state = {Idx, Term, _}}) ->
    {Idx, Term};
snapshot_index_term(_State) ->
    undefined.

-spec update_release_cursor(Idx :: ra_index(), Cluster :: ra_cluster(),
                            MachineState :: term(), State :: ra_log()) ->
    ra_log().
update_release_cursor(_, _, _, %% Idx, Cluster, MacState
                      #state{snapshot_index_in_progress = SIIP} = State)
  when is_integer(SIIP) ->
    % if a snapshot is in progress don't even evaluate
    % new segments will always set snapshot_index_in_progress = undefined
    % to ensure liveliness in case a snapshot_written message is lost.
    State;
update_release_cursor(Idx, Cluster, MachineState,
                      #state{segment_refs = SegRefs,
                             snapshot_state = SnapState,
                             snapshot_interval = SnapInter} = State0) ->
    ClusterServerIds = maps:keys(Cluster),
    SnapLimit = case SnapState of
                    undefined -> SnapInter;
                    {I, _, _} -> I + SnapInter
                end,
    % The release cursor index is the last entry _not_ contributing
    % to the current state. I.e. the last entry that can be discarded.
    % Check here if any segments can be release.
    case lists:any(fun({_, To, _}) when To =< Idx -> true;
                      (_) -> false end, SegRefs) of
        true ->
            % segments can be cleared up
            % take a snapshot at the release_cursor
            % TODO: here we use the current cluster configuration in
            % the snapshot,
            % _not_ the configuration at the snapshot point.
            % Given cluster changes
            % are applied as they are received I cannot think of any scenarios
            % where this can cause a problem. That said there may
            % well be :dragons: here.
            % The MachineState is a dehydrated version of the state at
            % the release_cursor point.
            case fetch_term(Idx, State0) of
                {undefined, _} ->
                    exit({term_not_found_for_index, Idx});
                {Term, State} ->
                    write_snapshot({Idx, Term, ClusterServerIds, MachineState},
                                   State)
            end;
        false when Idx > SnapLimit ->
            %% periodically take snapshots event if segments cannot be cleared
            %% up
            case fetch_term(Idx, State0) of
                {undefined, State} ->
                    State;
                {Term, State} ->
                    write_snapshot({Idx, Term, ClusterServerIds, MachineState},
                                   State)
            end;
        false ->
            State0
    end.


-spec read_meta(Key :: ra_meta_key(),
                State :: ra_log()) -> maybe(term()).
read_meta(Key, #state{uid = Id}) ->
    ra_log_meta:fetch(Id, Key).

-spec read_meta(Key :: ra_meta_key(), State :: ra_log(),
                Default :: term()) -> term().
read_meta(Key, Log, Default) ->
    ra_lib:default(read_meta(Key, Log), Default).

-spec write_meta(Key :: ra_meta_key(), Value :: term(), State :: ra_log()) ->
    ok.
write_meta(Key, Value, Log) ->
    write_meta(Key, Value, Log, true).

-spec write_meta(Key :: ra_meta_key(), Value :: term(),
                 State :: ra_log(), Sync :: boolean()) -> ok.
write_meta(Key, Value, #state{uid = Id}, true) ->
    ok = ra_log_meta:store_sync(Id, Key, Value);
write_meta(Key, Value, #state{uid = Id}, false) ->
    ok = ra_log_meta:store(Id, Key, Value).

append_sync({Idx, Term, _} = Entry, Log0) ->
    Log = ra_log:append(Entry, Log0),
    await_written_idx(Idx, Term, Log).

write_sync(Entries, Log0) ->
    {Idx, Term, _} = lists:last(Entries),
    case ra_log:write(Entries, Log0) of
        {ok, Log} ->
            await_written_idx(Idx, Term, Log);
        {error, _} = Err ->
            Err
    end.

can_write(#state{wal = Wal}) ->
    undefined =/= whereis(Wal).

-spec exists(ra_idxterm(), ra_log()) ->
    {boolean(), ra_log()}.
exists({Idx, Term}, Log0) ->
    case fetch_term(Idx, Log0) of
        {Term, Log} -> {true, Log};
        {_, Log} -> {false, Log}
    end.

overview(#state{last_index = LastIndex,
                last_written_index_term = LWIT,
                snapshot_index_in_progress = SIIP,
                segment_refs = Segs,
                snapshot_state = SnapshotState,
                open_segments = OpenSegs}) ->
    #{type => ?MODULE,
      last_index => LastIndex,
      last_written_index_term => LWIT,
      num_segments => length(Segs),
      open_segments => ra_flru:size(OpenSegs),
      snapshot_index_in_progress => SIIP,
      snapshot_index => case SnapshotState of
                            undefined -> undefined;
                            _ -> element(1, SnapshotState)
                        end
     }.

write_config(Config, #state{directory = Dir}) ->
    ConfigPath = filename:join(Dir, "config"),
    ok = file:write_file(ConfigPath,
                         list_to_binary(io_lib:format("~p.", [Config]))),
    ok.

read_config(Dir) ->
    ConfigPath = filename:join(Dir, "config"),
    case filelib:is_file(ConfigPath) of
        true ->
            {ok, [C]} =  file:consult(ConfigPath),
            {ok, C};
        false ->
            not_found
    end.

delete_everything(#state{directory = Dir} = Log) ->
    _ = close(Log),
    try ra_lib:recursive_delete(Dir) of
        ok -> ok
    catch
        _:_ = Err ->
            ?WARN("ra_log:delete_everything/1 failed to delete "
                  "directory ~s~n Error: ~p~n", [Dir, Err])
    end,
    ok.

release_resources(MaxOpenSegments,
                  #state{open_segments = OpenSegs} = State) ->
    % close all open segments
    % deliberately ignoring return value
    _ = ra_flru:evict_all(OpenSegs),
    %% open a new segment with the new max open segment value
    State#state{open_segments = ra_flru:new(MaxOpenSegments,
                                            fun flru_handler/1)}.

%%% Local functions

wal_truncate_write(#state{uid = UId, cache = Cache, wal = Wal} = State,
                   {Idx, Term, Data}) ->
    % this is the next write after a snapshot was taken or received
    % we need to indicate to the WAL that this may be a non-contiguous write
    % and that prior entries should be considered stale
    ok = ra_log_wal:truncate_write({UId, self()}, Wal, Idx, Term, Data),
    State#state{last_index = Idx, last_term = Term,
                cache = Cache#{Idx => {Term, Data}}}.

wal_write(#state{uid = UId, cache = Cache, wal = Wal} = State,
          {Idx, Term, Data}) ->
    case ra_log_wal:write({UId, self()}, Wal, Idx, Term, Data) of
        ok ->
            State#state{last_index = Idx, last_term = Term,
                        cache = Cache#{Idx => {Term, Data}}};
        {error, wal_down} ->
            exit(wal_down)
    end.

truncate_cache(Idx, #state{cache = Cache0} = State) ->
    Cache = maps:filter(fun (K, _) when K > Idx -> true;
                            (_, _) -> false
                        end, Cache0),
    State#state{cache = Cache}.

update_metrics(Id, Ops) ->
    _ = ets:update_counter(ra_log_metrics, Id, Ops),
    ok.

open_mem_tbl_take(Id, {Start0, End}, MetricOps, Acc0) ->
    case ets:lookup(ra_log_open_mem_tables, Id) of
        [{_, TStart, TEnd, Tid}] ->
            {Entries, Count, Rem} = mem_tbl_take({Start0, End}, TStart, TEnd,
                                                 Tid, 0, Acc0),
            {Entries, [{?METRICS_OPEN_MEM_TBL_POS, Count} | MetricOps], Rem};
        [] ->
            {Acc0, MetricOps, {Start0, End}}
    end.

closed_mem_tbl_take(Id, {Start0, End}, MetricOps, Acc0) ->
    case closed_mem_tables(Id) of
        [] ->
            {Acc0, MetricOps, {Start0, End}};
        Tables ->
            {Entries, Count, Rem} =
            lists:foldl(fun({_, _, TblSt, TblEnd, Tid}, {Ac, Count, Range}) ->
                                mem_tbl_take(Range, TblSt, TblEnd, Tid, Count, Ac)
                        end, {Acc0, 0, {Start0, End}}, Tables),
            {Entries, [{?METRICS_CLOSED_MEM_TBL_POS, Count} | MetricOps], Rem}
    end.

closed_mem_table_term_query(Idx, Id) ->
    case closed_mem_tables(Id) of
        [] ->
            undefined;
        Tables ->
            closed_mem_table_term_query0(Idx, Tables)
    end.

closed_mem_table_term_query0(_Idx, []) ->
    undefined;
closed_mem_table_term_query0(Idx, [{_, _, From, To, Tid} | _Tail])
  when Idx >= From andalso Idx =< To ->
    ets:lookup_element(Tid, Idx, 2);
closed_mem_table_term_query0(Idx, [_ | Tail]) ->
    closed_mem_table_term_query0(Idx, Tail).

closed_mem_tables(Id) ->
    case ets:lookup(ra_log_closed_mem_tables, Id) of
        [] ->
            [];
        Tables ->
            lists:sort(fun (A, B) ->
                               element(2, A) > element(2, B)
                       end, Tables)
    end.

mem_tbl_take(undefined, _TblStart, _TblEnd, _Tid, Count, Acc0) ->
    {Acc0, Count, undefined};
mem_tbl_take({_Start0, End} = Range, TblStart, _TblEnd, _Tid, Count, Acc0)
  when TblStart > End ->
    % optimisation to bypass request that has no overlap
    {Acc0, Count, Range};
mem_tbl_take({Start0, End}, TblStart, TblEnd, Tid, Count, Acc0)
  when TblEnd >= End ->
    Start = max(TblStart, Start0),
    Entries = lookup_range(Tid, Start, End, Acc0),
    Remainder = case Start =:= Start0 of
                    true ->
                        % the range was fully covered by the mem table
                        undefined;
                    false ->
                        {Start0, Start-1}
                end,
    {Entries, Count + (End - Start + 1), Remainder}.

lookup_range(Tid, Start, Start, Acc) ->
    [Entry] = ets:lookup(Tid, Start),
    [Entry | Acc];
lookup_range(Tid, Start, End, Acc) when End > Start ->
    [Entry] = ets:lookup(Tid, End),
    lookup_range(Tid, Start, End-1, [Entry | Acc]).


segment_take(#state{segment_refs = SegRefs,
                    open_segments = OpenSegs,
                    directory = Dir},
             Range, Entries0) ->
    lists:foldl(
      fun(_, {_, undefined, _} = Acc) ->
              %% we're done reading
              throw(Acc);
         ({From, _, _}, {_, {_, End}, _} = Acc)
           when From > End ->
              Acc;
         ({From, To, Fn}, {Open0, {Start0, End}, E0})
           when To >= End ->
              {Seg, Open} =
                  case ra_flru:fetch(Fn, Open0) of
                      {ok, S, Open1} ->
                          {S, Open1};
                      error ->
                          AbsFn = filename:join(Dir, Fn),
                          case ra_log_segment:open(AbsFn,
                                                   #{mode => read}) of
                              {ok, S} ->
                                  {S, ra_flru:insert(Fn, S, Open0)};
                              {error, Err} ->
                                  exit({ra_log_failed_to_open_segment, Err,
                                        AbsFn})
                          end
                  end,

              % actual start point cannot be prior to first segment
              % index
              Start = max(Start0, From),
              Num = End - Start + 1,
              Entries = ra_log_segment:read_cons(Seg, Start, Num,
                                                 fun binary_to_term/1,
                                                 E0),
              Rem = case Start of
                        Start0 -> undefined;
                        _ ->
                            {Start0, Start-1}
                    end,
              {Open, Rem, Entries}
      end, {OpenSegs, Range, Entries0}, SegRefs).

segment_term_query(Idx, #state{segment_refs = SegRefs,
                               directory = Dir,
                               open_segments = OpenSegs} = State) ->
    {Result, Open} = segment_term_query0(Idx, SegRefs, OpenSegs, Dir),
    {Result, State#state{open_segments = Open}}.

segment_term_query0(Idx, [{From, To, Filename} | _], Open0, Dir)
  when Idx >= From andalso Idx =< To ->
    case ra_flru:fetch(Filename, Open0) of
        {ok, Seg, Open} ->
            Term = ra_log_segment:term_query(Seg, Idx),
            {Term, Open};
        error ->
            AbsFn = filename:join(Dir, Filename),
            {ok, Seg} = ra_log_segment:open(AbsFn, #{mode => read}),
            Term = ra_log_segment:term_query(Seg, Idx),
            {Term, ra_flru:insert(Filename, Seg, Open0)}
    end;
segment_term_query0(Idx, [_ | Tail], Open, Dir) ->
    segment_term_query0(Idx, Tail, Open, Dir);
segment_term_query0(_Idx, [], Open, _) ->
    {undefined, Open}.


cache_take(Start, Num, #state{cache = Cache, last_index = LastIdx}) ->
    Highest = min(LastIdx, Start + Num - 1),
    % cache needs to be queried in reverse to ensure
    % we can bail out when an item is not found
    case cache_take0(Highest, Start, Cache, []) of
        [] ->
            {[], [], {Start, Highest}};
        [Last | _] = Entries when element(1, Last) =:= Start ->
            % there is no remainder - everything was in the cache
            {Entries, [{?METRICS_CACHE_POS, Highest - Start + 1}], undefined};
        [Last | _] = Entries ->
            LastEntryIdx = element(1, Last),
            {Entries, [{?METRICS_CACHE_POS, LastIdx - Start + 1}],
             {Start, LastEntryIdx - 1}}
    end.

cache_take0(Next, Last, _Cache, Acc)
  when Next < Last ->
    Acc;
cache_take0(Next, Last, Cache, Acc) ->
    case Cache of
        #{Next := Entry} ->
            cache_take0(Next-1, Last, Cache,
                        [erlang:insert_element(1, Entry, Next) | Acc]);
        _ ->
            Acc
    end.

maybe_append_0_0_entry(State0 = #state{last_index = -1}) ->
    State = append({0, 0, undefined}, State0),
    receive
        {ra_log_event, {written, {0, 0, 0}}} -> ok
    end,
    State#state{first_index = 0, last_written_index_term = {0, 0}};
maybe_append_0_0_entry(State) ->
    State.

resend_from(Idx, #state{uid = UId} = State0) ->
    try resend_from0(Idx, State0) of
        State -> State
    catch
        exit:wal_down ->
            ?WARN("~s: ra_log: resending from ~b failed with wal_down",
                  [UId, Idx]),
            State0
    end.

resend_from0(Idx, #state{last_index = LastIdx,
                         last_resend_time = undefined,
                         cache = Cache} = State) ->
    ?INFO("~s: ra_log: resending from ~b to ~b",
          [State#state.uid, Idx, LastIdx]),
    lists:foldl(fun (I, Acc) ->
                        X = maps:get(I, Cache),
                        wal_write(Acc, erlang:insert_element(1, X, I))
                end,
                State#state{last_resend_time = erlang:system_time(seconds)},
                % TODO: replace with recursive function
                lists:seq(Idx, LastIdx));
resend_from0(Idx, #state{last_resend_time = LastResend,
                        resend_window_seconds = ResendWindow} = State) ->
    case erlang:system_time(seconds) > LastResend + ResendWindow of
        true ->
            % it has been more than a minute since last resend
            % ok to try again
            resend_from(Idx, State#state{last_resend_time = undefined});
        false ->
            State
    end.


compact_seg_refs(SegRefs) ->
    lists:reverse(
      lists:foldl(fun ({_, _, File} = S, Acc) ->
                          case lists:any(fun({_, _, F}) when F =:= File ->
                                                 true;
                                            (_) -> false
                                         end, Acc) of
                              true -> Acc;
                              false -> [S | Acc]
                          end
                  end, [], SegRefs)).

verify_entries(_, []) ->
    ok;
verify_entries(Idx, [{NextIdx, _, _} | Tail]) when Idx + 1 == NextIdx ->
    verify_entries(NextIdx, Tail);
verify_entries(Idx, Tail) ->
    Msg = io_lib:format("ra_log:verify_entries/2 "
                        "tried writing ~p - expected ~b",
                        [Tail, Idx+1]),
    {error, {integrity_error, lists:flatten(Msg)}}.

write_entries([], State) ->
    {ok, State};
write_entries([{FstIdx, _, _} | Rest] = Entries, State0) ->
    case verify_entries(FstIdx, Rest) of
        ok ->
            try
                % TODO: wal should provide batch api
                State = lists:foldl(fun (Entry, S) ->
                                            wal_write(S, Entry)
                                    end, State0, Entries),
                {ok, State}
            catch
                exit:wal_down ->
                    {error, wal_down}
            end;
        Error ->
            Error
    end.

write_snapshot(Snapshot, #state{directory = Dir} = State) ->
    ok = ra_log_snapshot_writer:write_snapshot(self(), Dir, Snapshot),
    State#state{snapshot_index_in_progress = element(1, Snapshot)}.

flru_handler({_, Seg}) ->
    _ = ra_log_segment:close(Seg),
    ok.

recover_range(UId) ->
    % 0. check open mem_tables (this assumes wal has finished recovering
    % which means it is essential that ra_servers are part of the same
    % supervision tree
    % 1. check closed mem_tables to extend
    OpenRanges = case ets:lookup(ra_log_open_mem_tables, UId) of
                     [] ->
                         [];
                     [{UId, First, Last, _}] ->
                         [{First, Last}]
                 end,
    ClosedRanges = [{F, L} || {_, _, F, L, _} <- closed_mem_tables(UId)],
    % 2. check segments
    SegFiles = ra_log_segment_writer:my_segments(UId),
    SegRefs =
    [begin
         {ok, Seg} = ra_log_segment:open(S, #{mode => read}),
         SegRef = ra_log_segment:segref(Seg),
         ok = ra_log_segment:close(Seg),
         SegRef
     end || S <- lists:reverse(SegFiles)],
    SegRanges = [{F, L} || {F, L, _} <- SegRefs],
    Ranges = OpenRanges ++ ClosedRanges ++ SegRanges,
    {pick_range(Ranges, undefined), SegRefs}.

% picks the current range from a sorted (newest to oldest) list of ranges
pick_range([], Res) ->
    Res;
pick_range([H | Tail], undefined) ->
    pick_range(Tail, H);
pick_range([{Fst, _Lst} | Tail], {CurFst, CurLst}) ->
    pick_range(Tail, {min(Fst, CurFst), CurLst}).


%% TODO: implent synchronous writes using gen_batch_server:call/3
await_written_idx(Idx, Term, Log) ->
    receive
        {ra_log_event, {written, {_, Idx, Term}} = Evt} ->
            handle_event(Evt, Log);
        {ra_log_event, {written, _} = Evt} ->
            await_written_idx(Idx, Term, handle_event(Evt, Log))
    after ?LOG_APPEND_TIMEOUT ->
              throw(ra_log_append_timeout)
    end.


%%%% TESTS

-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").

compact_seg_refs_test() ->
    % {From, To, File}
    Refs = [{10, 100, "2"}, {10, 75, "2"}, {10, 50, "2"}, {1, 9, "1"}],
    [{10, 100, "2"}, {1, 9, "1"}] = compact_seg_refs(Refs),
    ok.

cache_take0_test() ->
    Cache = #{1 => {a}, 2 => {b}, 3 => {c}},
    State = #state{cache = Cache, last_index = 3, first_index = 1},
    % no remainder
    {[{2, b}], _, undefined} = cache_take(2, 1, State),
    {[{2, b}, {3, c}], _,  undefined} = cache_take(2, 2, State),
    {[{1, a}, {2, b}, {3, c}], _, undefined} = cache_take(1, 3, State),
    % small remainder
    {[{3, c}], _, {1, 2}} = cache_take(1, 3, State#state{cache = #{3 => {c}}}),
    {[], _, {1, 3}} = cache_take(1, 3, State#state{cache = #{4 => {d}}}),
    ok.

open_mem_tbl_take_test() ->
    _ = ets:new(ra_log_open_mem_tables, [named_table]),
    Tid = ets:new(test_id, []),
    true = ets:insert(ra_log_open_mem_tables, {test_id, 3, 7, Tid}),
    Entries = [{3, 2, "3"}, {4, 2, "4"},
               {5, 2, "5"}, {6, 2, "6"},
               {7, 2, "7"}],
    % seed the mem table
    [ets:insert(Tid, E) || E <- Entries],

    {Entries, _, undefined} = open_mem_tbl_take(test_id, {3, 7}, [], []),
    EntriesPlus8 = Entries ++ [{8, 2, "8"}],
    {EntriesPlus8, _, {1, 2}} = open_mem_tbl_take(test_id, {1, 7}, [], [{8, 2, "8"}]),
    {[{6, 2, "6"}], _, undefined} = open_mem_tbl_take(test_id, {6, 6}, [], []),
    {[], _, {1, 2}} = open_mem_tbl_take(test_id, {1, 2}, [], []),

    ets:delete(Tid),
    ets:delete(ra_log_open_mem_tables),

    ok.

closed_mem_tbl_take_test() ->
    _ = ets:new(ra_log_closed_mem_tables, [named_table, bag]),
    Tid1 = ets:new(test_id, []),
    Tid2 = ets:new(test_id, []),
    M1 = erlang:unique_integer([monotonic, positive]),
    M2 = erlang:unique_integer([monotonic, positive]),
    true = ets:insert(ra_log_closed_mem_tables, {test_id, M1, 5, 7, Tid1}),
    true = ets:insert(ra_log_closed_mem_tables, {test_id, M2, 8, 10, Tid2}),
    Entries1 = [{5, 2, "5"}, {6, 2, "6"}, {7, 2, "7"}],
    Entries2 = [{8, 2, "8"}, {9, 2, "9"}, {10, 2, "10"}],
    % seed the mem tables
    [ets:insert(Tid1, E) || E <- Entries1],
    [ets:insert(Tid2, E) || E <- Entries2],

    {Entries1, _, undefined} = closed_mem_tbl_take(test_id, {5, 7}, [], []),
    {Entries2, _, undefined} = closed_mem_tbl_take(test_id, {8, 10}, [], []),
    {[{9, 2, "9"}], _, undefined} = closed_mem_tbl_take(test_id, {9, 9}, [], []),
    ok.

pick_range_test() ->
    Ranges1 = [{76, 90}, {50, 75}, {1, 100}],
    {1, 90} = pick_range(Ranges1, undefined),

    Ranges2 = [{76, 110}, {50, 75}, {1, 49}],
    {1, 110} = pick_range(Ranges2, undefined),

    Ranges3 = [{25, 30}, {25, 35}, {1, 50}],
    {1, 30} = pick_range(Ranges3, undefined),
    ok.
-endif.