%%%-------------------------------------------------------------------
%%% @author Heinz Nikolaus Gies <heinz@licenser.net>
%%% @copyright (C) 2014, Heinz Nikolaus Gies
%%% @doc
%%%
%%% @end
%%% Created :  4 Jun 2014 by Heinz Nikolaus Gies <heinz@licenser.net>
%%%-------------------------------------------------------------------

%% @doc An mstore instance is per directory, consisting of multiple `mstore'
%% and `idx' files. Data points are stored according to their timestamp, and
%% stores typically retain up to one week's (depends on `file_size') worth of
%% data. One master `mstore' file will be created for all metric paths in the
%% directory, and is used for coverage queries.
%%
%% Index (`idx') files are keyed on and offset position and consist of metric
%% names.  These metrics are used to calculate the relative position of a
%% metric's values in the `mstore' storage file.
%%
%% The format of an `idx' file is as follows, where `M..' are metric paths:
%% +--------+-----------+----------+----+----------+----+-----+----------+----+
%% | Offset | File Size | Size(M1) | M1 | Size(M2) | M2 | ... | Size(Mn) | Mn |
%% +--------+-----------+----------+----+----------+----+-----+----------+----+
%%                      |---- Pos 0 ----|---- Pos 2 ----|     |---- Pos n ----|
%%
%% Metric names are appended to the index the first time that they are
%% encountered.
%%
%% Storage (mstore) files may store values for more than one metric at a time,
%% and is segmented sequentially.  The position for a value within a
%% segment is calculated as follows: P = Base + ((TS - Offset) * Data size),
%% where Base is a function of the metric's position in the index:
%% +-------------+-----+------------------------------------------------+
%% | Pos(M1) = 0 | --> | Base = 0 X File size X Data size = 0           |
%% | Pos(M2) = 1 | --> | Base = 1 X File size X Data size = 4838400     |
%% | ...         |     |                                                |
%% | Pos(Mn) = N | --> | Base = N X File size X Data size = N * 4838400 |
%% +-------------+-----+------------------------------------------------+
%%
%% Only one process should open the mstore for writing at a given time. At any
%% moment, up to two `mstore' files may be open for writing by the mstore.
%% These files represent the current and previously written to files e.g. the
%% current and previous week.

-module(mstore).

-export([
         new/2,
         open/1,
         open/2,
         close/1,
         delete/1,
         delete/2,
         reindex/1,
         get/4,
         bitmap/3,
         open_mfile/1,
         put/4,
         metrics/1,
         count/1,
         file_size/1,
         fold/3, fold/4,
         make_splits/3]).
-export_type([mstore/0]).


-include_lib("mmath/include/mmath.hrl").
-include("mstore.hrl").

-define(SIZE_TYPE, unsigned-integer).

-define(VERSION, 4).
-define(OPTS, [raw, binary]).
-define(MFILE_VER, 1).

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

-record(mstore, {
          size,
          max_files = 2,
          files=[],
          dir,
          metrics = btrie:new()
         }).

-opaque mstore() :: #mstore{}.

-type open_opt() ::
        {max_files, pos_integer()}.

-type new_opt() ::
        {file_size, pos_integer()}.

%%--------------------------------------------------------------------
%% Public API
%%--------------------------------------------------------------------

%%--------------------------------------------------------------------
%% @doc
%% Creates a new set or opens an existing one.
%% @end
%%--------------------------------------------------------------------
-spec new(Dir :: string() | binary(),
          Opts :: [new_opt() | open_opt()]) ->
                 {ok, mstore()} |
                 {error, filesize_missmatch}.

new(Dir, Opts) when (is_list(Dir) orelse is_binary(Dir)) , is_list(Opts) ->
    {file_size, FileSize} = proplists:lookup(file_size, Opts),
    case new_(FileSize, Dir) of
        {ok, M} ->
            {ok, apply_opts(M, Opts)};
        E ->
            E
    end.

new_(FileSize, Dir) when
      is_integer(FileSize), FileSize > 0,
      is_binary(Dir) ->
    new_(FileSize, binary_to_list(Dir));

new_(FileSize, Dir) ->
    IdxFile = filename:join([Dir, "mstore"]),
    case open_mfile(IdxFile) of
        {ok, F, _Metrics} when F =/= FileSize ->
            {error, filesize_missmatch};
        {ok, F,  Metrics} ->
            {ok, #mstore{size=F, dir=Dir, metrics=Metrics}};
        _ ->
            file:make_dir(Dir),
            MStore = #mstore{size=FileSize, dir=Dir},
            ok = file:write_file(IdxFile, index_header(MStore)),
            {ok, MStore}
    end.

%%--------------------------------------------------------------------
%% @doc
%% Opens an existing set.
%% @end
%%--------------------------------------------------------------------
-spec open(Dir :: string()) ->
                  {ok, mstore()} | {error, enoent | not_found | invalid_file}.
open(Dir) ->
    open(Dir, []).

-spec open(Dir :: string(), [open_opt()]) ->
                  {ok, mstore()} | {error, enoent | not_found | invalid_file}.
open(Dir, Opts) ->
    case open_mfile(filename:join([Dir, "mstore"])) of
        {ok, FileSize, Metrics} ->
            M =  #mstore{size=FileSize, dir=Dir,
                         metrics=Metrics},
            {ok, apply_opts(M, Opts)};
        E ->
            E
    end.

%%--------------------------------------------------------------------
%% @doc
%% Closes a metric set.
%% @end
%%--------------------------------------------------------------------
-spec close(mstore()) -> ok.

close(#mstore{files=Files}) ->
    [mfile:close(S) || {_, S} <- Files],
    ok.

%%--------------------------------------------------------------------
%% @doc
%% Recursively deletes a metric set.
%% @end
%%--------------------------------------------------------------------
-spec delete(mstore()) -> ok | {error, atom()}.

delete(MStore = #mstore{dir=Dir}) ->
    close(MStore),
    case file:list_dir(Dir) of
        {ok, Files} ->
            Files1 = [[Dir, $/ | File] || File <- Files],
            [file:delete(F) || F <- Files1],
            file:del_dir(Dir);
        {error, enoent} ->
            ok
    end.

%%--------------------------------------------------------------------
%% @doc
%% Deletes entries before *Before*, not exact since it's rounded to
%% the next file.
%% @end
%%--------------------------------------------------------------------

-spec delete(mstore(), pos_integer()) -> {ok, mstore()}.
delete(MStore = #mstore{size = S, dir = Dir, files = Files}, Before) ->
    Before1 = ((Before div S) - 1) * S,
    Chunks = chunks(Dir, ".mstore"),
    Chunks1 = lists:takewhile(fun(C) ->
                                      C < Before1
                              end, Chunks),
    case Chunks1 of
        [] ->
            {ok, MStore};
        _ ->
            [mfile:close(F) || {_, F} <- Files],
            [begin
                 F = filename:join([Dir , integer_to_list(C)]),
                 file:delete(F ++ ".mstore"),
                 file:delete(F ++ ".bitmap"),
                 file:delete(F ++ ".idx")
             end || C <- Chunks1],
            reindex(MStore#mstore{files = []})
    end.

%%--------------------------------------------------------------------
%% @doc
%% Reads data from a set.
%% @end
%%--------------------------------------------------------------------
-spec get(
        mstore(),
        binary(),
        non_neg_integer(),
        pos_integer()) ->
                 {'error',atom()} |
                 {ok, binary()}.

get(#mstore{size=S, files=FS, dir=Dir},
    Metric, Time, Count) when
      is_binary(Metric),
      is_integer(Time), Time >= 0,
      is_integer(Count), Count > 0 ->
    ?DT_READ_ENTRY(Metric, Time, Count),
    Parts = make_splits(Time, Count, S),
    R = do_get(S, FS, Dir, Metric, Parts, <<>>),
    ?DT_READ_RETURN,
    R.

%%--------------------------------------------------------------------
%% @doc
%% Reads the bitmap for a given time slot. Time must not be allinged
%% @end
%%--------------------------------------------------------------------

-spec bitmap(
        mstore(),
        binary(),
        non_neg_integer()) ->
                 {'error',atom()} |
                 {ok, bitmap:bitmap()}.

bitmap(#mstore{size=S, files=FS, dir=Dir},
    Metric, Time) when
      is_binary(Metric),
      is_integer(Time), Time >= 0 ->
    R = do_get_bitmap(S, FS, Dir, Metric, Time),
    R.

%%--------------------------------------------------------------------
%% @doc
%% Folds over all the metrics in a store. Please be aware that the
%% data for a single metric is not guaranteed to be delivered at a time
%% it is split on non set values and file boundaries!
%% @end
%%--------------------------------------------------------------------

-spec fold(mstore(), FoldFun :: mfile:fold_fun(), AccIn :: any()) ->
                  AccOut :: any().
fold(MStore, FoldFun, Acc) ->
    fold(MStore, FoldFun, infinity, Acc).

%%--------------------------------------------------------------------
%% @doc
%% Same as {@link: fold/3} but data is chunked at a maximum size of
%% *ChunkSize*.
%% @end
%%--------------------------------------------------------------------

-spec fold(mstore(), FoldFun :: mfile:fold_fun(),
           ChunkSize :: pos_integer() | infinity,
           AccIn :: any()) ->
                  AccOut :: any().
fold(#mstore{dir=Dir}, Fun, Chunk, Acc) ->
    serialize_dir(Dir, Fun, Chunk, Acc).

%%--------------------------------------------------------------------
%% @doc
%% Writes data into a mstore.
%% @end
%%--------------------------------------------------------------------
-spec put(mstore(), binary(), non_neg_integer(),
          integer() | [integer()] | binary()) ->
                 mstore().

put(MStore, Metric, Time, [V0 | _] = Values)
  when is_integer(V0) ->
    put(MStore, Metric, Time, mmath_bin:from_list(Values));

put(MStore, Metric, Time, V) when is_integer(V) ->
    put(MStore, Metric, Time, mmath_bin:from_list([V]));

put(MStore = #mstore{size=S, files=CurFiles, metrics=Ms},
    Metric, Time, Value)
  when is_binary(Value),
       is_integer(Time), Time >= 0,
       is_binary(Metric) ->
    ?DT_WRITE_ENTRY(Metric, Time, mmath_bin:length(Value)),
    Count = mmath_bin:length(Value),
    Parts = make_splits(Time, Count, S),
    Parts1 = [{B, trunc(C * ?DATA_SIZE)} || {B, C} <- Parts],
    MStore1 = case btrie:is_key(Metric, Ms) of
                  true ->
                      MStore;
                  false ->
                      MStorex = MStore#mstore{metrics=btrie:store(Metric, Ms)},
                      FileName = filename:join([MStorex#mstore.dir, "mstore"]),
                      ok = file:write_file(FileName,
                                           <<(byte_size(Metric)):16/integer,
                                             Metric/binary>>,
                                           [read, append]),
                      MStorex
              end,
    CurFiles1 = do_put(MStore1, Metric, Parts1, Value, CurFiles),
    ?DT_WRITE_RETURN,
    limit_files(MStore1#mstore{files = CurFiles1}).

%%--------------------------------------------------------------------
%% @doc
%% Returns a set of all metrics in the store.
%% @end
%%--------------------------------------------------------------------

-spec metrics(mstore()) -> btrie:btrie().

metrics(#mstore{metrics=M}) ->
    M.

%%--------------------------------------------------------------------
%% @doc
%% Returns the chunk in the mstore, each metric row in a file is a
%% chunk. So the total is the sum of the metrics in each file.
%% @end
%%--------------------------------------------------------------------

-spec count(mstore()) -> non_neg_integer().
count(#mstore{dir=Dir}) ->
    count(Dir);
count(Dir) when is_list(Dir) ->
    Names = list_mfiles(Dir),
    lists:foldl(fun(N, Cnt) ->
                        Cnt + mfile:count(N)
                end, 0, Names).

%%--------------------------------------------------------------------
%% @doc
%% Returns the File Size of a store
%% @end
%%--------------------------------------------------------------------
-spec file_size(mstore()) -> non_neg_integer().
file_size(#mstore{size = FileSize}) ->
    FileSize.

%%--------------------------------------------------------------------
%% @doc
%% Utility function to split a continous data streams in chunks for a
%% given time range size.
%%
%% * Time - the offset of the first value.
%% * Count - the total number of values.
%% * Size - The size of the chunking.
%% @end
%%--------------------------------------------------------------------
-spec make_splits(Time :: non_neg_integer(),
                  Count :: pos_integer(),
                  Size :: pos_integer()) ->
                         [{StartTime :: non_neg_integer(),
                           ChunkSize :: pos_integer()}].

make_splits(Time, Count, Size) ->
    make_splits(Time, Count, Size, []).

-spec reindex(mstore()) -> {ok, mstore()}.

reindex(MStore = #mstore{dir = Dir}) ->
    IdxFileOld = filename:join([Dir, "mstore"]),
    IdxFileNew = filename:join([Dir, "mstore.new"]),

    Files = list_mfiles(Dir),
    {ok, IO} = file:open(IdxFileNew, [write | ?OPTS]),
    ok = file:write(IO, index_header(MStore)),
    Metrics = lists:foldl(fun (F, Set) ->
                                  reindex_chunk(IO, F, Set)
                          end, btrie:new(), Files),
    ok = file:close(IO),
    NewIndex = << <<(byte_size(Metric)):16/integer, Metric/binary>>
                  || Metric <- btrie:fetch_keys(Metrics) >>,
    ok = file:write_file(IdxFileNew, <<(index_header(MStore))/binary,
                                       NewIndex/binary>>),
    ok = file:delete(IdxFileOld),
    ok = file:rename(IdxFileNew, IdxFileOld),
    {ok, MStore#mstore{metrics = Metrics}}.

%%====================================================================
%% Private functions.
%%====================================================================

apply_opts(MStore, []) ->
    MStore;
apply_opts(MStore, [{max_files, N} | R]) when is_integer(N), N >= 0 ->
    apply_opts(MStore#mstore{max_files = N}, R);
apply_opts(MStore, [_ | R]) ->
    apply_opts(MStore, R).


chunks(Dir, Ext) ->
    FileName = filename:join([Dir, "*" ++ Ext]),
    lists:sort([list_to_integer(filename:rootname(filename:basename(F))) ||
                   F <- filelib:wildcard(FileName)]).

reindex_chunk(IO, File, Set) ->
    mfile:fold_idx(fun({entry, M}, Acc) ->
                           ok = file:write(IO, <<(byte_size(M)):16/integer, M/binary>>),
                           btrie:store(M, Acc);
                      (_, Acc) ->
                           Acc
                   end, Set, File).

index_header(#mstore{size=FileSize}) ->
    <<?VERSION:16/?SIZE_TYPE, FileSize:64/?SIZE_TYPE>>.

make_splits(_Time, 0, _Size, Acc) ->
    lists:reverse(Acc);

make_splits(Time, Count, Size, Acc) ->
    Base = (Time div Size)*Size,
    case Time - Base of
        D when (D + Count) < Size ->
            lists:reverse([{Time, Count} | Acc]);
        D ->
            Inc = Size-D,
            make_splits(Time + Inc, Count - Inc, Size, [{Time, Inc} | Acc])
    end.

open_mfile(F) ->
    Chunk = 4*1024,
    case file:open(F, [read | ?OPTS]) of
        {ok, IO} ->
            case file:read(IO, Chunk) of
                {ok, <<2:16/?SIZE_TYPE, FileSize:64/?SIZE_TYPE, R/binary>>} ->
                    Set = do_fold_idx(IO, Chunk,
                                      fun({entry, M}, Acc) ->
                                              btrie:store(M, Acc)
                                      end, btrie:new(), R),
                    {ok, FileSize, Set};
                {ok, <<3:16/?SIZE_TYPE, FileSize:64/?SIZE_TYPE,
                       _:64/?SIZE_TYPE, R/binary>>} ->
                    Set = do_fold_idx(IO, Chunk,
                                      fun({entry, M}, Acc) ->
                                              btrie:store(M, Acc)
                                      end, btrie:new(), R),
                    {ok, FileSize, Set};
                {ok, <<?VERSION:16/?SIZE_TYPE,
                       FileSize:64/?SIZE_TYPE, R/binary>>} ->
                    Set = do_fold_idx(IO, Chunk,
                                      fun({entry, M}, Acc) ->
                                              btrie:store(M, Acc)
                                      end, btrie:new(), R),
                    {ok, FileSize, Set};
                {ok, _} ->
                    file:close(IO),
                    {error, invalid_file};
                _ ->
                    file:close(IO),
                    {error, invalid_file}
            end;
        E ->
            E
    end.

do_put(_, _, [], <<>>, Files) ->
    Files;

do_put(MStore = #mstore{size=S}, Metric,
       [{Time, Size} | R], InData,
       [{FileBase, F} | FileRest]) when
      ((Time div S)*S) =:= FileBase ->
    <<Data:Size/binary, DataRest/binary>> = InData,
    {ok, F1} = mfile:write(F, Metric, Time, Data),
    do_put(MStore, Metric, R, DataRest, [{FileBase, F1} | FileRest]);

do_put(MStore = #mstore{size=S}, Metric,
       [{Time, Size} | R], InData,
       [First, {FileBase, F}]) when
      ((Time div S)*S) =:= FileBase ->
    <<Data:Size/binary, DataRest/binary>> = InData,
    {ok, F1} = mfile:write(F, Metric, Time, Data),
    do_put(MStore, Metric, R, DataRest, [{FileBase, F1}, First]);

do_put(MStore = #mstore{size=S, dir=D}, Metric,
       [{Time, Size} | R], InData,
       [First, {_Other, F}]) ->
    <<Data:Size/binary, DataRest/binary>> = InData,
    FileBase = (Time div S)*S,
    mfile:close(F),
    Base = [D, $/, integer_to_list(FileBase)],
    {ok, F1} = mfile:open(Base, [{offset, FileBase},
                                 {file_size, S},
                                 {mode, write}]),
    {ok, F2} = mfile:write(F1, Metric, Time, Data),
    do_put(MStore, Metric, R, DataRest, [{FileBase, F2}, First]);

do_put(MStore = #mstore{size=S, dir=D}, Metric,
       [{Time, Size} | R], InData,
       Files) when length(Files) =< 2 ->
    <<Data:Size/binary, DataRest/binary>> = InData,
    FileBase = (Time div S)*S,
    Base = [D, $/, integer_to_list(FileBase)],
    {ok, F1} = mfile:open(Base, [{offset, FileBase},
                                 {file_size, S},
                                 {mode, write}]),
    {ok, F2} = mfile:write(F1, Metric, Time, Data),
    do_put(MStore, Metric, R, DataRest, [{FileBase, F2} | Files]).

limit_files(MStore = #mstore{max_files = MaxFiles,
                             files = CurFiles = [First, {_FileBase, F} | R]})
  when length(CurFiles) > MaxFiles ->
    mfile:close(F),
    limit_files(MStore#mstore{files = [First | R]});
limit_files(MStore = #mstore{max_files = MaxFiles,
                             files = CurFiles = [{_FileBase, F} | R]})
  when length(CurFiles) > MaxFiles ->
    mfile:close(F),
    limit_files(MStore#mstore{files = R});
limit_files(MStore) ->
    MStore.

do_get_bitmap(S, [{FileBase, F} | _],
              _Dir, Metric, Time)
  when ((Time div S) * S) =:= FileBase ->
    case mfile:bitmap(F, Metric) of
        {ok, B, _F1} ->
            {ok, B};
        E ->
            E
    end;
do_get_bitmap(S, [_, {FileBase, F}],
              _Dir, Metric, Time)
  when ((Time div S) * S) =:= FileBase ->
    case mfile:bitmap(F, Metric) of
        {ok, B, _F1} ->
            {ok, B};
        E ->
            E
    end;
do_get_bitmap(S, _Files, Dir, Metric, Time) ->
    Base = ((Time div S) * S),
    File = filename:join([Dir, integer_to_list(Base)]),
    case mfile:open(File) of
        {ok, F} ->
            case mfile:bitmap(F, Metric) of
                {ok, B, F1} ->
                    mfile:close(F1),
                    {ok, B};
                E ->
                    mfile:close(F),
                    E
            end;
        {error, enoent} ->
            {error, not_found}
    end.

do_get(_, _, _, _, [], Acc) ->
    {ok, Acc};

do_get(S, [{FileBase, F} | _] = FS,
       Dir, Metric, [{Time, Count} | R], Acc)
  when ((Time div S)*S) =:= FileBase ->
    case mfile:read(F, Metric, Time, Count) of
        {ok, D} ->
            Acc1 = <<Acc/binary, D/binary>>,
            Acc2 = case mmath_bin:length(D) of
                       L when L < Count ->
                           Missing = Count - L,
                           <<Acc1/binary, (mmath_bin:empty(Missing))/binary>>;
                       _ ->
                           Acc1
                   end,
            do_get(S, FS, Dir, Metric, R, Acc2);
        {error,not_found} ->
            Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
            do_get(S, FS, Dir, Metric, R, Acc1);
        eof ->
            Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
            do_get(S, FS, Dir, Metric, R, Acc1);
        E ->
            E
    end;

do_get(S,
       [_, {FileBase, F}] = FS,
       Dir, Metric, [{Time, Count} | R], Acc)
  when ((Time div S)*S) =:= FileBase ->
    case mfile:read(F, Metric, Time, Count) of
        {ok, D} ->
            Acc1 = <<Acc/binary, D/binary>>,
            Acc2 = case mmath_bin:length(D) of
                       L when L < Count ->
                           Missing = Count - L,
                           <<Acc1/binary, (mmath_bin:empty(Missing))/binary>>;
                       _ ->
                           Acc1
                   end,
            do_get(S, FS, Dir, Metric, R, Acc2);
        {error,not_found} ->
            Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
            do_get(S, FS, Dir, Metric, R, Acc1);
        eof ->
            Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
            do_get(S, FS, Dir, Metric, R, Acc1);
        E ->
            E
    end;

do_get(S, FS, Dir, Metric, [{Time, Count} | R], Acc) ->
    FileBase = (Time div S)*S,
    Base = filename:join([Dir, integer_to_list(FileBase)]),
    {ok, F} = mfile:open(Base, [{offset, FileBase},
                                {file_size, S}]),
    Result = mfile:read(F, Metric, Time, Count),
    case Result of
        {ok, D} ->
            mfile:close(F),
            Acc1 = <<Acc/binary, D/binary>>,
            Acc2 = case mmath_bin:length(D) of
                       L when L < Count ->
                           Missing = Count - L,
                           <<Acc1/binary, (mmath_bin:empty(Missing))/binary>>;
                       _ ->
                           Acc1
                   end,
            do_get(S, FS, Dir, Metric, R, Acc2);
        {error,not_found} ->
            mfile:close(F),
            Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
            do_get(S, FS, Dir, Metric, R, Acc1);
        eof ->
            mfile:close(F),
            Acc1 = <<Acc/binary, (mmath_bin:empty(Count))/binary>>,
            do_get(S, FS, Dir, Metric, R, Acc1);
        E ->
            mfile:close(F),
            E
    end.


serialize_dir(Dir, Fun, Chunk, Acc) ->
    Idxs = list_mfiles(Dir),
    lists:foldl(fun(I, AccIn) ->
                        mfile:fold(I, Fun, Chunk, AccIn)
                end, Acc, Idxs).

do_fold_idx(IO, Chunk, Fun, AccIn, <<_L:16/integer, M:_L/binary, R/binary>>) ->
    Acc = Fun({entry, M}, AccIn),
    do_fold_idx(IO, Chunk, Fun, Acc, R);
do_fold_idx(IO, Chunk, Fun, AccIn, R) ->
    case file:read(IO, Chunk) of
        {ok, Data} ->
            do_fold_idx(IO, Chunk, Fun, AccIn, <<R/binary, Data/binary>>);
        eof ->
            file:close(IO),
            case R of
                <<>> ->
                    AccIn;
                _ ->
                    {error, invalid_file}
            end;
        E ->
            file:close(IO),
            E
    end.

list_mfiles(Dir) ->
    Fs = filelib:wildcard(filename:join([Dir, "*.idx"])),
    [filename:rootname(F, ".idx") || F <- Fs].