%%%-------------------------------------------------------------------
  @author neerajsharma
   ( C ) 2018 ,  
%%% @doc
%%%
%%% @end
%%% %CopyrightBegin%
%%%
  Copyright   <  > 2017 .
  All Rights Reserved .
%%%
  Licensed under the Apache License , Version 2.0 ( the " License " ) ;
%%% you may not use this file except in compliance with the License.
%%% You may obtain a copy of the License at
%%%
%%%     -2.0
%%%
%%% Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an " AS IS " BASIS ,
%%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%%% See the License for the specific language governing permissions and
%%% limitations under the License.
%%%
%%% %CopyrightEnd%
%%%-------------------------------------------------------------------
-module(egraph_detail_model).
%% -behaviour(egraph_callback).
-export([init/0, init/2, terminate/1, % CRUD
         validate/2, create/3, read/2, update/3, delete/2]).
-export([create/4, update/4]).
-export([read_resource/1]).
-export([read_resources/1]).
-export([search_resource/5]).
-export([read_all_resource/3]).
-export([reindex_key/2]).
-export_type([egraph_k/0]).

-include("egraph_constants.hrl").

-type egraph_k() :: map().
-type state() :: term().

-define(LAGER_ATTRS, [{type, model}]).

%%%===================================================================
%%% API
%%%===================================================================

%%%===================================================================
%%% Callbacks
%%%===================================================================

%% @doc Initialize the state that the handler will carry for
%% a specific request throughout its progression. The state
%% is then passed on to each subsequent call to this model.
-spec init() -> state().
init() ->
    nostate.

init(_, QsProplist) ->
    [{proplist, QsProplist}].

%% @doc At the end of a request, the state is passed back in
%% to allow for clean up.
-spec terminate(state()) -> term().
terminate(_State) ->
    ok.

%% @doc Return, via a boolean value, whether the user-submitted
%% data structure is considered to be valid by this model's standard.
-spec validate(egraph_k() | term(), state()) -> {boolean(), state()}.
validate(V, State) ->
    {is_map(V) orelse is_list(V) orelse is_boolean(V), State}.

%% @doc Create a new entry. If the id is `undefined', the user
%% has not submitted an id under which to store the resource:
%% the id needs to be generated by the model, and (if successful),
  returned via ` { true ,    .
%% Otherwise, a given id will be passed, and a simple `true' or
%% `false' value may be returned to confirm the results.
%%
%% The created resource is validated before this function is called.
-spec create(egraph_callback:id() | undefined, egraph_k(), state()) ->
        {false | true | {true, egraph_callback:id()}, state()}.
create(undefined, V, State) ->
    create_or_update_info(V, State);
create(Key, V, State) ->
    Info2 = V#{<<"key_data">> => Key},
    create_or_update_info(Info2, State).

  @doc Create a new entry along with an expiry of some seconds .
-spec create(egraph_callback:id() | undefined, egraph_k(),
             [{binary(), binary()}], state()) ->
    {false | true | {true, egraph_callback:id()}, state()}.
create(Key, V, _QsProplist, State) ->
    create(Key, V, State).

  @doc Read a given entry from the store based on its Key .
-spec read(egraph_callback:id(), state()) ->
        { {ok, egraph_k()} |
          {function, Fun :: function()},
          {error, not_found}, state()}.
read(undefined, State) ->
    %% return everthing you have
    {{function, fun read_all_resource/3}, State};
read(Key, State) ->
    QsProplists = proplists:get_value(proplist, State, []),
    KeyType = proplists:get_value(<<"keytype">>, QsProplists, undefined),
    RawKey = case KeyType of
                 <<"rawhex">> ->
                     egraph_util:hex_binary_to_bin(Key);
                 <<"rawint">> ->
                     egraph_shard_util:convert_integer_to_xxhash_bin(
                       egraph_util:convert_to_integer(Key));
                 _ ->
                     egraph_util:generate_xxhash_binary(
                       egraph_util:convert_to_binary(Key))
             end,
    case read_resource(RawKey) of
        {ok, Vals} ->
            {{ok, Vals}, State};
        R ->
            {R, State}
    end.

%% @doc Update an existing resource.
%%
%% The modified resource is validated before this function is called.
-spec update(egraph_callback:id(), egraph_k(), state()) -> {boolean(), state()}.
update(Key, V, State) ->
    Info2 = V#{<<"key_data">> => Key},
    create_or_update_info(Info2, State).

  @doc Update an existing resource with some expiry seconds .
-spec update(egraph_callback:id(), egraph_k(), integer(), state()) ->
    {boolean(), state()}.
update(Key, V, _QsProplist, State) ->
    update(Key, V, State).

%% @doc Delete an existing resource.
-spec delete(egraph_callback:id(), state()) -> {boolean(), state()}.
delete(Key, State) ->
    QsProplists = proplists:get_value(proplist, State, []),
    KeyType = proplists:get_value(<<"keytype">>, QsProplists, undefined),
    RawKey = case KeyType of
                 <<"rawint">> ->
                     egraph_shard_util:convert_integer_to_xxhash_bin(
                       egraph_util:convert_to_integer(Key));
                 <<"rawhex">> ->
                     egraph_util:hex_binary_to_bin(Key);
                 _ ->
                     egraph_util:generate_xxhash_binary(
                       egraph_util:convert_to_binary(Key))
             end,
    {delete_resource(RawKey), State}.

%%%===================================================================
  Internal
%%%===================================================================

create_or_update_info(Info, State) ->
    #{ <<"key_data">> := Key,
       <<"details">> := Details,
       <<"indexes">> := Indexes } = Info,
    %% lowercase_indexes are optional
    true = is_map(Details),
    {GenericIndexes, LowercaseIndexes} = extract_indexes(Indexes),
    true = is_list(GenericIndexes),
    true = is_list(LowercaseIndexes),

    PreferredCompressionId = maps:get(<<"compression_id">>, Info, undefined),
  actually GenericIndexes is as follows :
    %% GenericIndexes = [ [<<"a">>, <<"b">>], [<<"a">>, <<"c">>, <<"d">>], ... ]
    %%
    %% Note that the above is a list of json paths.

    RawKey = egraph_util:generate_xxhash_binary(
               egraph_util:convert_to_binary(Key)),
    TimeoutMsec = egraph_config_util:mysql_rw_timeout_msec(detail),
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(key, RawKey, BaseTableName),
    UpdatedDetails = Details#{?EGRAPH_DETAILS_SPECIAL_KEY => egraph_util:convert_to_binary(Key)},
    SerializedDetails = egraph_compression_util:serialize_data(
                          UpdatedDetails, PreferredCompressionId),
    NewIndexes = #{<<"indexes">> => GenericIndexes,
                   <<"lowercase_indexes">> => LowercaseIndexes},
    SerializedIndexes = egraph_compression_util:serialize_data(
                          NewIndexes,
                          PreferredCompressionId),
    HexKey = egraph_util:bin_to_hex_binary(RawKey),
    ReturnLoc = iolist_to_binary(
                  [HexKey,
                   <<"?keytype=rawhex">>]),
    case read_resource(RawKey, create) of
        {error, not_found} ->
            case sql_insert_record(TableName, RawKey,
                                   SerializedDetails,
                                   SerializedIndexes,
                                   UpdatedDetails,
                                   GenericIndexes,
                                   LowercaseIndexes,
                                   TimeoutMsec) of
                true ->
                    {{true, ReturnLoc}, State};
                false ->
                    {false, State}
            end;
        {ok, [DbInfo]} ->
            DbStoredDetailsInfo = maps:get(<<"details">>, DbInfo),
            DbStoredIndexesInfo = maps:get(<<"indexes">>, DbInfo, null),
            lager:debug("DbStoredDetailsInfo = ~p", [DbStoredDetailsInfo]),
            DbStoredKey = maps:get(?EGRAPH_DETAILS_SPECIAL_KEY,
                                   DbStoredDetailsInfo),
            lager:debug("[compare] ~p == ~p", [DbStoredKey, egraph_util:convert_to_binary(Key)]),
  TODO read details and check for collision
            case DbStoredKey == egraph_util:convert_to_binary(Key) of
                true ->
                    %% the index already exists
                    OldVersion = maps:get(<<"version">>, DbInfo),
                    %% TODO: Check for failures while updating info

                    DetailsHashHexBin = egraph_util:bin_to_hex_binary(
                                          egraph_util:generate_xxhash_binary(
                                            egraph_util:convert_to_binary(
                                              SerializedDetails))),
                    DbDetailsHash = maps:get(<<"details_hash">>, DbInfo, undefined),
                    AnyChange = (DetailsHashHexBin =/= DbDetailsHash) orelse (NewIndexes =/= DbStoredIndexesInfo),
                    case AnyChange of
                        true ->
                            true = sql_update_record(TableName,
                                              OldVersion,
                                              RawKey,
                                              DbStoredDetailsInfo,
                                              DbStoredIndexesInfo,
                                              SerializedDetails,
                                              SerializedIndexes,
                                              GenericIndexes,
                                              LowercaseIndexes,
                                              UpdatedDetails,
                                              TimeoutMsec);
                        %% TODO when only indexes changed then optimize
                        %% update and only write indexes column
                        _ ->
  Lets save some write 
                            ok
                    end,
                    %% TODO: retrieve ?EGRAPH_DETAILS_SPECIAL_INDEX from
                    %%       DbStoredDetailsInfo and fix indexes (delete old and create new)
                    %%       as required.
                    {{true, ReturnLoc}, State};
                false ->
                    %% TODO: How did this happen?
                    lager:error("Collission in hash detected between DbStoredKey = ~p and Key = ~p",
                                [DbStoredKey, egraph_util:convert_to_binary(Key)]),
                    {false, State}
            end
    end.

%% TODO: Find the cluster nodes which must have this data and delete from there.
delete_resource(RawKey) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(key, RawKey, BaseTableName),
    Q = iolist_to_binary([<<"DELETE FROM ">>,
                          TableName,
                          <<" WHERE source=?">>]),
    %% TODO: should we check for collisions?
    Params = [RawKey],
    TimeoutMsec = egraph_config_util:mysql_rw_timeout_msec(detail),
    PoolName = egraph_config_util:mysql_rw_pool(detail),
    case egraph_sql_util:mysql_write_query(
           PoolName,
           Q, Params, TimeoutMsec) of
        ok ->
            true;
        _ ->
            false
    end.

%% TODO: Find the cluster nodes which must have this data and pull from there.
-spec read_resource(binary(), create) -> {ok, [map()]} | {error, term()}.
read_resource(RawKey, create) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                 key, RawKey, BaseTableName),
    RawKeyHexString = egraph_util:bin_to_hex_binary(RawKey),
    Q = iolist_to_binary([<<"SELECT * FROM ">>,
                          TableName,
                          <<" WHERE source=X'">>,
                          RawKeyHexString,
                          <<"'">>]),
  Params = [ RawKey ] ,
    read_generic_resource(Q, undefined, create, BaseTableName, TableName).

%% TODO: Find the cluster nodes which must have this data and pull from there.
-spec read_resource(binary()) -> {ok, [map()]} | {error, term()}.
read_resource(RawKey) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                 key, RawKey, BaseTableName),
    RawKeyHexString = egraph_util:bin_to_hex_binary(RawKey),
    Q = iolist_to_binary([<<"SELECT * FROM ">>,
                          TableName,
                          <<" WHERE source=X'">>,
                          RawKeyHexString,
                          <<"'">>]),
  Params = [ RawKey ] ,
    read_generic_resource(Q, undefined).

%% TODO: Find the cluster nodes which must have this data and pull from there.
%% The assumption is that all the RawKeys are in the same shard
  ( which is same as the first one ) .
-spec read_resources([binary()]) -> [map()].
read_resources(RawKeys) ->
  TODO move to include
	lists:foldl(fun(L, AccIn) ->
                        read_resources_internal(L, AccIn)
                end, [], egraph_util:segments(MaxLen, RawKeys)).

read_resources_internal([H | Rest] = RawKeys, AccIn) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
  derive table name from first one while the
    %% rest of them should be in same shard (as per the contract).
    TableName = egraph_shard_util:sharded_tablename(
                 key, H, BaseTableName),
    Qs = [<<"?">>] ++ [<<",?">> || _ <- Rest],
    Q = iolist_to_binary([<<"SELECT * FROM ">>,
                          TableName,
                          <<" WHERE source in (">>,
                          Qs,
                          <<")">>]),
    Params = RawKeys,
  TODO crashing in case of failure
    case read_generic_resource(Q, Params) of
        {ok, R} ->
            R ++ AccIn;
        _ ->
            AccIn
    end.

-spec search_resource(ShardKey :: integer(),
                      previous | next,
                      Datetime :: calander:datetime(),
                      LimitDatetime :: calander:datetime(),
                      SoftLimit :: non_neg_integer()) ->
    {ok, [map()]} | {error, term()}.
search_resource(ShardKey, previous, Datetime, LimitDatetime, SoftLimit) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                  shard_id, ShardKey, BaseTableName),
    case count_next_updated_datetime(
           previous, TableName, Datetime, SoftLimit) of
        {ok, {MinDatetime, MaxDatetime}} ->
            Q = iolist_to_binary([<<"SELECT * FROM ">>,
                                  TableName,
                                  <<" WHERE updated_datetime >= ? and updated_datetime <= ? order by updated_datetime desc">>]),
            Params = [erlang:max(MinDatetime, LimitDatetime), MaxDatetime],
            read_generic_resource(Q, Params);
        E ->
            E
    end;
search_resource(ShardKey, next, Datetime, LimitDatetime, SoftLimit) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                  shard_id, ShardKey, BaseTableName),
    case count_next_updated_datetime(
           next, TableName, Datetime, SoftLimit) of
        {ok, {MinDatetime, MaxDatetime}} ->
            Q = iolist_to_binary([<<"SELECT * FROM ">>,
                                  TableName,
                                  <<" WHERE updated_datetime >= ? and updated_datetime <= ? order by updated_datetime asc">>]),
            Params = [MinDatetime, erlang:min(MaxDatetime, LimitDatetime)],
            read_generic_resource(Q, Params);
        E ->
            E
    end.

-spec read_all_resource(ShardKey :: integer(),
                        Limit :: integer(),
                        Offset :: integer()) ->
    {ok, [map()], NewOffset :: integer()} | {error, term()}.
read_all_resource(ShardKey, Limit, Offset) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                  shard_id, ShardKey, BaseTableName),
    Q = iolist_to_binary([<<"SELECT * FROM ">>,
                          TableName,
                          <<" LIMIT ? OFFSET ?">>]),
    Params = [Limit, Offset],
    case read_generic_resource(Q, Params) of
        {ok, R} ->
            {ok, R, Offset + length(R)};
        E ->
            E
    end.

read_generic_resource(Query, Params, read, BaseTableName, TableName) ->
    ConvertToMap = true,
    TimeoutMsec = egraph_config_util:mysql_ro_timeout_msec(detail),
    IsRetry = false,
    IsReadOnly = true,
    ReadPools = egraph_config_util:mysql_ro_pools(detail),
    PoolName = lists:nth(1, ReadPools),
    case egraph_sql_util:run_sql_read_query_for_shard(
           PoolName,
           BaseTableName,
           ReadPools,
           TableName,
           Query, Params, TimeoutMsec, IsRetry, IsReadOnly,
           ConvertToMap) of
        {ok, Maps} ->
            Maps2 = lists:foldl(fun transform_result/2, [], Maps),
            {ok, Maps2};
        Error ->
            Error
    end;
read_generic_resource(Query, Params, create, BaseTableName, TableName) ->
    ConvertToMap = true,
    TimeoutMsec = egraph_config_util:mysql_rw_timeout_msec(detail),
    IsRetry = false,
    IsReadOnly = false,
    PoolName = egraph_config_util:mysql_rw_pool(detail),
    ReadPools = [PoolName],
    case egraph_sql_util:run_sql_read_query_for_shard(
           PoolName,
           BaseTableName,
           ReadPools,
           TableName,
           Query, Params, TimeoutMsec, IsRetry, IsReadOnly,
           ConvertToMap) of
        {ok, Maps} ->
            Maps2 = lists:foldl(fun transform_result/2, [], Maps),
            {ok, Maps2};
        Error ->
            Error
    end.

read_generic_resource(Query, Params, false = ConvertToMap) ->
    TimeoutMsec = egraph_config_util:mysql_ro_timeout_msec(detail),
    ReadPools = egraph_config_util:mysql_ro_pools(detail),
    case egraph_sql_util:mysql_query(
           ReadPools, Query, Params, TimeoutMsec, ConvertToMap) of
        {ok, {_ColumnNames, MultipleRowValues}} ->
            {ok, MultipleRowValues};
        Error ->
            Error
    end;
read_generic_resource(Query, Params, true) ->
    read_generic_resource(Query, Params).

read_generic_resource(Query, Params) ->
    ConvertToMap = true,
    TimeoutMsec = egraph_config_util:mysql_ro_timeout_msec(detail),
    ReadPools = egraph_config_util:mysql_ro_pools(detail),
    case egraph_sql_util:mysql_query(
           ReadPools, Query, Params, TimeoutMsec, ConvertToMap) of
        {ok, Maps} ->
            Maps2 = lists:foldl(fun transform_result/2, [], Maps),
            {ok, Maps2};
        Error ->
            Error
    end.

transform_result(E, AccIn) ->
    E2 = case maps:get(<<"source">>, E, undefined) of
             undefined ->
                 E;
             Source ->
                 %% egraph_shard_util:convert_xxhash_bin_to_integer(Source)
                 E#{<<"source">> =>
                    egraph_util:bin_to_hex_binary(Source)}
         end,
    E3 = case maps:get(<<"details_hash">>, E2, undefined) of
             undefined ->
                 E2;
             DetailsHash ->
                 E2#{<<"details_hash">> =>
                     egraph_util:bin_to_hex_binary(DetailsHash)}
         end,
    E4 = case maps:get(<<"updated_datetime">>, E3, undefined) of
             undefined ->
                 E3;
             UpdatedDateTime ->
                 E3#{<<"updated_datetime">> =>
                     qdate:to_string(<<"Y-m-d H:i:s">>, UpdatedDateTime)}
         end,
    E5 = case maps:get(<<"details">>, E4, undefined) of
             undefined ->
                 E4;
             Details ->
                 E4#{<<"details">> =>
                     egraph_compression_util:deserialize_data(Details)}
         end,
    E6 = case maps:get(<<"indexes">>, E5, null) of
             null ->
                 E5;
             Indexes ->
                 E5#{<<"indexes">> =>
                     egraph_compression_util:deserialize_data(Indexes)}
         end,
    [E6 | AccIn].

sql_insert_record(TableName, RawKey, SerializedDetails, SerializedIndexes, UpdatedDetails, GenericIndexes, LowercaseIndexes, TimeoutMsec) ->
    DefaultVersion = 0,
    DetailsHash = egraph_util:generate_xxhash_binary(
                    egraph_util:convert_to_binary(SerializedDetails)),
    UpdatedDateTime = qdate:to_date(erlang:system_time(second)),
    Q = iolist_to_binary([<<"INSERT INTO ">>,
                          TableName,
                          <<" (source, version, details_hash, details, updated_datetime, indexes) VALUES(?, ?, ?, ?, ?, ?)">>]),
    Params = [RawKey, DefaultVersion, DetailsHash, SerializedDetails,
              UpdatedDateTime, SerializedIndexes],
    %% TODO: find out the cluster nodes which must persist this data
    %%       and save it there.

  Create indexes first because they must exist before adding the
    %% data. Think about a scenario where data is inserted into the
    %% database but index creation fails (due to network outage or
    %% software bug). Now if the client retries with the same content
    %% then its alright, but what if the content changed. In which case
    %% when index must be changed as part of that then the deleting of
    %% old indexes will not work since there were none to start with.
  Worse , when no index were created for the first time then
  the second time instead of insert the update method will be
    %% called and no new indexes would be created.
    GenericIndexResults = create_generic_index(GenericIndexes, UpdatedDetails, RawKey, []),
    LowerIndexResults = create_lowercase_index(LowercaseIndexes, UpdatedDetails, RawKey, []),
    %% extract all errors
    case LowerIndexResults ++ GenericIndexResults of
        [] ->
  TODO : open a single transaction instead of multiple , since
            %% this can fail.
            %%
            %% TODO: make an RPC call to save index in each of the cluster
            %%       nodes although the data may live in only subset of them.
            PoolName = egraph_config_util:mysql_rw_pool(detail),
            case egraph_sql_util:mysql_write_query(
                   PoolName, Q, Params, TimeoutMsec) of
                ok ->
                    lager:debug("UpdatedDetails = ~p", [UpdatedDetails]),
                    true;
                ErrorInfo ->
                    %% TODO: rollback the original entry, but
                    lager:error("RawKey = ~p, Cannot rollback although index creation succeeded, ErrorInfo = ~p",
                                [RawKey, ErrorInfo]),
                    false
            end;
        _ ->
            false
    end.

sql_update_record(TableName, OldVersion, RawKey, DbStoredDetailsInfo, DbStoredIndexesInfo,
                  SerializedDetails, SerializedIndexes, GenericIndexes, LowercaseIndexes,
                  UpdatedDetails, TimeoutMsec) ->
    {OldGenIndexes, OldLowerIndexes} = case DbStoredIndexesInfo of
                                           null ->
                                               {maps:get(?EGRAPH_DETAILS_SPECIAL_INDEX,
                                                         DbStoredDetailsInfo),
                                                []};
                                           _ ->
                                               {maps:get(<<"indexes">>,
                                                         DbStoredIndexesInfo),
                                                maps:get(<<"lowercase_indexes">>,
                                                         DbStoredIndexesInfo)}
                                       end,
    Version = OldVersion + 1,
    DetailsHash = egraph_util:generate_xxhash_binary(
                    egraph_util:convert_to_binary(SerializedDetails)),
    UpdatedDateTime = qdate:to_date(erlang:system_time(second)),

    GenericIndexResults = change_indexes(
                            generic, RawKey, OldGenIndexes, GenericIndexes,
                            DbStoredDetailsInfo, UpdatedDetails),
    LowercaseIndexResults = change_indexes(
                              lowercase, RawKey, OldLowerIndexes, LowercaseIndexes,
                              DbStoredDetailsInfo, UpdatedDetails),

    IndexResults = GenericIndexResults ++ LowercaseIndexResults,
    case IndexResults of
        [] ->
            Q = iolist_to_binary([<<"UPDATE ">>,
                                  TableName,
                                  <<" SET version=?, details_hash=?, details=?, updated_datetime=?, indexes=? WHERE source=? and version=?">>]),
            Params = [Version, DetailsHash, SerializedDetails,
                      UpdatedDateTime, SerializedIndexes, RawKey, OldVersion],
            %% TODO: need to check whether update indeed happened or not because
            %% the where clause may not match.
            PoolName = egraph_config_util:mysql_rw_pool(detail),
            case egraph_sql_util:mysql_write_query(
                   PoolName, Q, Params, TimeoutMsec) of
                ok -> true;
                DbUpdateError ->
                    lager:error("RawKey = ~p, Cannot rollback although update failed, DbUpdateError = ~p",
                                [RawKey, DbUpdateError]),
                    false
            end;
        _ ->
            %% TODO: rollback the original entry, but
            lager:error("RawKey = ~p, Cannot rollback although index creation partially failed, IndexResults = ~p",
                        [RawKey, IndexResults]),
            false
    end.

delete_index([], _Type, _Details, _RawKey, AccIn) ->
    AccIn;
delete_index([JsonPath | Rest], Type, Details, RawKey, AccIn) ->
    IndexName = case Type of
                    generic -> lists:last(JsonPath);
                    lowercase -> iolist_to_binary(
                                   [lists:last(JsonPath),
                                    ?EGRAPH_INDEX_SPECIAL_SUFFIX_LOWERCASE])
                end,
    Key = nested:get(JsonPath, Details),
    KeyType = egraph_shard_util:key_datatype(Key),
    case egraph_index_model:delete_resource(Key, KeyType, RawKey, IndexName) of
        true ->
            delete_index(Rest, Type, Details, RawKey, AccIn);
        false ->
            %% retry at least once
            case egraph_index_model:delete_resource(Key, KeyType, RawKey,IndexName) of
                true ->
                    delete_index(Rest, Type, Details, RawKey, AccIn);
                false ->
                    delete_index(Rest, Type, Details, RawKey, [{Key, KeyType, IndexName} | AccIn])
            end
    end.

create_db_index(generic, Indexes, UpdatedDetails, RawKey, AccIn) ->
    create_generic_index(Indexes, UpdatedDetails, RawKey, AccIn);
create_db_index(lowercase, Indexes, UpdatedDetails, RawKey, AccIn) ->
    create_lowercase_index(Indexes, UpdatedDetails, RawKey, AccIn).

create_generic_index(Indexes, UpdatedDetails, RawKey, AccIn) ->
    F = fun(E) -> E end,
    NameFun = fun(L) -> lists:last(L) end,
    create_index(Indexes, UpdatedDetails, RawKey, AccIn, F, NameFun).

%% transform the indexname for the lowercase index by using
%% the last part of json path and then applying a suffix
create_lowercase_index(Indexes, UpdatedDetails, RawKey, AccIn) ->
    F = fun(E) -> string:lowercase(E) end,
    NameFun = fun(L) -> iolist_to_binary(
                          [lists:last(L),
                           ?EGRAPH_INDEX_SPECIAL_SUFFIX_LOWERCASE])
              end,
    create_index(Indexes, UpdatedDetails, RawKey, AccIn, F, NameFun).

create_index([], _UpdatedDetails, _RawKey, AccIn,
             _TransformFun, _IndexNameFun) ->
    AccIn;
create_index([JsonPath | Rest], UpdatedDetails, RawKey, AccIn,
             TransformFun, IndexNameFun) ->
    IndexName = IndexNameFun(JsonPath),
    HexId = egraph_util:bin_to_hex_binary(RawKey),
    Key = TransformFun(nested:get(JsonPath, UpdatedDetails)),
    KeyType = egraph_shard_util:key_datatype(Key),
    IndexInfo = #{<<"key_data">> => Key,
                  <<"id">> => HexId,
                  <<"index_name">> => IndexName,
                  <<"key_type">> => KeyType},
    lager:debug("IndexInfo = ~p", [IndexInfo]),
    case egraph_index_model:create_or_update_info(IndexInfo, []) of
        {false, _} ->
            %% retry at least once
            case egraph_index_model:create_or_update_info(IndexInfo, []) of
                {false, _} ->
                    create_index(Rest, UpdatedDetails, RawKey, [IndexInfo | AccIn],
                                 TransformFun, IndexNameFun);
                _ ->
                    create_index(Rest, UpdatedDetails, RawKey, AccIn,
                                 TransformFun, IndexNameFun)
            end;
        _ ->
            create_index(Rest, UpdatedDetails, RawKey, AccIn,
                         TransformFun, IndexNameFun)
    end.

reindex_key(Details, Indexes) ->
    Key = maps:get(
            ?EGRAPH_DETAILS_SPECIAL_KEY,
            Details),
    RawKey = egraph_util:generate_xxhash_binary(
               egraph_util:convert_to_binary(Key)),
    case Indexes of
        null ->
            GenericIndexes = maps:get(
                               ?EGRAPH_DETAILS_SPECIAL_INDEX,
                               Details,
                               []),
            lager:debug("Reindex Key = ~p", [Key]),
            create_db_index(generic, GenericIndexes, Details, RawKey, []);
        _ ->
            LowercaseIndexes = maps:get(<<"lowercase_indexes">>, Indexes),
            GenericIndexes = maps:get(<<"indexes">>, Indexes),
            lager:debug("Reindex Key = ~p", [Key]),
            A = create_db_index(generic, GenericIndexes, Details, RawKey, []),
            create_db_index(lowercase, LowercaseIndexes, Details, RawKey, A)
    end.

count_next_updated_datetime(previous, TableName, Datetime, SoftLimit) ->
    Q = iolist_to_binary([<<"SELECT updated_datetime FROM ">>,
                          TableName,
                          <<" WHERE updated_datetime <= ? order by updated_datetime desc LIMIT ?">>]),
    Params = [Datetime, SoftLimit],
    case read_generic_resource(Q, Params, false) of
        {ok, MultipleRowValues} ->
            DateTimes = lists:flatten(MultipleRowValues),
            {ok, {lists:last(DateTimes),
                  lists:nth(1, DateTimes)}};
        E ->
            E
    end;
count_next_updated_datetime(next, TableName, Datetime, SoftLimit) ->
    Q = iolist_to_binary([<<"SELECT updated_datetime FROM ">>,
                          TableName,
                          <<" WHERE updated_datetime >= ? order by updated_datetime asc LIMIT ?">>]),
    Params = [Datetime, SoftLimit],
    case read_generic_resource(Q, Params, false) of
        {ok, MultipleRowValues} ->
            DateTimes = lists:flatten(MultipleRowValues),
            {ok, {lists:nth(1, DateTimes),
                  lists:last(DateTimes)}};
        E ->
            E
    end.

%% extract in a backward compatible way
extract_indexes(Indexes) when is_list(Indexes) ->
    {Indexes, []};
extract_indexes(Indexes) when is_map(Indexes) ->
    {maps:get(<<"indexes">>, Indexes),
     maps:get(<<"lowercase_indexes">>, Indexes, [])}.

%%
%% reorder the sequence of addition to ensure that any
%% inconsistency between index and data can be handled
%% in a non-distructive way.
%%
  1 . First delete indexes
  2 . Add new indexes
  3 . Update existing index if the value has changed
  4 . Add the data once all the above is successful .
%%
%% IMPORTANT: Since there are no transactions, so it is possible
%% that there are indexes around which do not point to any data.
%%
change_indexes(Type, RawKey, OldIndexes, Indexes, DbStoredDetailsInfo, UpdatedDetails) ->
    ChangedIndexes = lists:filter(fun(JsonPath) ->
                                          try
                                              OldValue = nested:get(JsonPath, DbStoredDetailsInfo),
                                              NewValue = nested:get(JsonPath, UpdatedDetails),
                                              OldValue =/= NewValue
                                          catch
                                              _:_ ->
                                                  false
                                          end
                                  end, Indexes),
    DeletedIndexes = lists:filter(fun(JsonPath) ->
                        not lists:member(JsonPath, Indexes)
                                  end, OldIndexes),
    DeletedIndexResults = delete_index(
                            ChangedIndexes ++ DeletedIndexes,
                            Type,
                            DbStoredDetailsInfo,
                            RawKey,
                            []),
    case DeletedIndexResults of
        [] ->
            AddedIndexes = lists:filter(fun(JsonPath) ->
                              not lists:member(JsonPath, OldIndexes)
                                        end, Indexes),
            %% first add new index and then delete old ones
            AddedGenIndexResults = create_db_index(
                                     Type,
                                     ChangedIndexes ++ AddedIndexes,
                                     UpdatedDetails,
                                     RawKey,
                                     []),
            AddedGenIndexResults;
        _ ->
            %% TODO: rollback the original entry, but
            lager:error("RawKey = ~p, Cannot rollback although index deletion failed, Type = ~p, DeletedIndexResults = ~p",
                        [RawKey, Type, DeletedIndexResults]),
            DeletedIndexResults
    end.



-------------------------------------------------------------------
 @doc

 @end
 %CopyrightBegin%


 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

     -2.0

 Unless required by applicable law or agreed to in writing, software
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.

 %CopyrightEnd%
-------------------------------------------------------------------
 -behaviour(egraph_callback).
 CRUD
===================================================================
 API
===================================================================
===================================================================
 Callbacks
===================================================================
 @doc Initialize the state that the handler will carry for
 a specific request throughout its progression. The state
 is then passed on to each subsequent call to this model.
 @doc At the end of a request, the state is passed back in
 to allow for clean up.
 @doc Return, via a boolean value, whether the user-submitted
 data structure is considered to be valid by this model's standard.
 @doc Create a new entry. If the id is `undefined', the user
 has not submitted an id under which to store the resource:
 the id needs to be generated by the model, and (if successful),
 Otherwise, a given id will be passed, and a simple `true' or
 `false' value may be returned to confirm the results.

 The created resource is validated before this function is called.
 return everthing you have
 @doc Update an existing resource.

 The modified resource is validated before this function is called.
 @doc Delete an existing resource.
===================================================================
===================================================================
 lowercase_indexes are optional
 GenericIndexes = [ [<<"a">>, <<"b">>], [<<"a">>, <<"c">>, <<"d">>], ... ]

 Note that the above is a list of json paths.
 the index already exists
 TODO: Check for failures while updating info
 TODO when only indexes changed then optimize
 update and only write indexes column
 TODO: retrieve ?EGRAPH_DETAILS_SPECIAL_INDEX from
       DbStoredDetailsInfo and fix indexes (delete old and create new)
       as required.
 TODO: How did this happen?
 TODO: Find the cluster nodes which must have this data and delete from there.
 TODO: should we check for collisions?
 TODO: Find the cluster nodes which must have this data and pull from there.
 TODO: Find the cluster nodes which must have this data and pull from there.
 TODO: Find the cluster nodes which must have this data and pull from there.
 The assumption is that all the RawKeys are in the same shard
 rest of them should be in same shard (as per the contract).
 egraph_shard_util:convert_xxhash_bin_to_integer(Source)
 TODO: find out the cluster nodes which must persist this data
       and save it there.
 data. Think about a scenario where data is inserted into the
 database but index creation fails (due to network outage or
 software bug). Now if the client retries with the same content
 then its alright, but what if the content changed. In which case
 when index must be changed as part of that then the deleting of
 old indexes will not work since there were none to start with.
 called and no new indexes would be created.
 extract all errors
 this can fail.

 TODO: make an RPC call to save index in each of the cluster
       nodes although the data may live in only subset of them.
 TODO: rollback the original entry, but
 TODO: need to check whether update indeed happened or not because
 the where clause may not match.
 TODO: rollback the original entry, but
 retry at least once
 transform the indexname for the lowercase index by using
 the last part of json path and then applying a suffix
 retry at least once
 extract in a backward compatible way

 reorder the sequence of addition to ensure that any
 inconsistency between index and data can be handled
 in a non-distructive way.


 IMPORTANT: Since there are no transactions, so it is possible
 that there are indexes around which do not point to any data.

 first add new index and then delete old ones
 TODO: rollback the original entry, but
  @author neerajsharma
   ( C ) 2018 ,  
  Copyright   <  > 2017 .
  All Rights Reserved .
  Licensed under the Apache License , Version 2.0 ( the " License " ) ;
  distributed under the License is distributed on an " AS IS " BASIS ,
-module(egraph_detail_model).
         validate/2, create/3, read/2, update/3, delete/2]).
-export([create/4, update/4]).
-export([read_resource/1]).
-export([read_resources/1]).
-export([search_resource/5]).
-export([read_all_resource/3]).
-export([reindex_key/2]).
-export_type([egraph_k/0]).

-include("egraph_constants.hrl").

-type egraph_k() :: map().
-type state() :: term().

-define(LAGER_ATTRS, [{type, model}]).



-spec init() -> state().
init() ->
    nostate.

init(_, QsProplist) ->
    [{proplist, QsProplist}].

-spec terminate(state()) -> term().
terminate(_State) ->
    ok.

-spec validate(egraph_k() | term(), state()) -> {boolean(), state()}.
validate(V, State) ->
    {is_map(V) orelse is_list(V) orelse is_boolean(V), State}.

  returned via ` { true ,    .
-spec create(egraph_callback:id() | undefined, egraph_k(), state()) ->
        {false | true | {true, egraph_callback:id()}, state()}.
create(undefined, V, State) ->
    create_or_update_info(V, State);
create(Key, V, State) ->
    Info2 = V#{<<"key_data">> => Key},
    create_or_update_info(Info2, State).

  @doc Create a new entry along with an expiry of some seconds .
-spec create(egraph_callback:id() | undefined, egraph_k(),
             [{binary(), binary()}], state()) ->
    {false | true | {true, egraph_callback:id()}, state()}.
create(Key, V, _QsProplist, State) ->
    create(Key, V, State).

  @doc Read a given entry from the store based on its Key .
-spec read(egraph_callback:id(), state()) ->
        { {ok, egraph_k()} |
          {function, Fun :: function()},
          {error, not_found}, state()}.
read(undefined, State) ->
    {{function, fun read_all_resource/3}, State};
read(Key, State) ->
    QsProplists = proplists:get_value(proplist, State, []),
    KeyType = proplists:get_value(<<"keytype">>, QsProplists, undefined),
    RawKey = case KeyType of
                 <<"rawhex">> ->
                     egraph_util:hex_binary_to_bin(Key);
                 <<"rawint">> ->
                     egraph_shard_util:convert_integer_to_xxhash_bin(
                       egraph_util:convert_to_integer(Key));
                 _ ->
                     egraph_util:generate_xxhash_binary(
                       egraph_util:convert_to_binary(Key))
             end,
    case read_resource(RawKey) of
        {ok, Vals} ->
            {{ok, Vals}, State};
        R ->
            {R, State}
    end.

-spec update(egraph_callback:id(), egraph_k(), state()) -> {boolean(), state()}.
update(Key, V, State) ->
    Info2 = V#{<<"key_data">> => Key},
    create_or_update_info(Info2, State).

  @doc Update an existing resource with some expiry seconds .
-spec update(egraph_callback:id(), egraph_k(), integer(), state()) ->
    {boolean(), state()}.
update(Key, V, _QsProplist, State) ->
    update(Key, V, State).

-spec delete(egraph_callback:id(), state()) -> {boolean(), state()}.
delete(Key, State) ->
    QsProplists = proplists:get_value(proplist, State, []),
    KeyType = proplists:get_value(<<"keytype">>, QsProplists, undefined),
    RawKey = case KeyType of
                 <<"rawint">> ->
                     egraph_shard_util:convert_integer_to_xxhash_bin(
                       egraph_util:convert_to_integer(Key));
                 <<"rawhex">> ->
                     egraph_util:hex_binary_to_bin(Key);
                 _ ->
                     egraph_util:generate_xxhash_binary(
                       egraph_util:convert_to_binary(Key))
             end,
    {delete_resource(RawKey), State}.

  Internal

create_or_update_info(Info, State) ->
    #{ <<"key_data">> := Key,
       <<"details">> := Details,
       <<"indexes">> := Indexes } = Info,
    true = is_map(Details),
    {GenericIndexes, LowercaseIndexes} = extract_indexes(Indexes),
    true = is_list(GenericIndexes),
    true = is_list(LowercaseIndexes),

    PreferredCompressionId = maps:get(<<"compression_id">>, Info, undefined),
  actually GenericIndexes is as follows :

    RawKey = egraph_util:generate_xxhash_binary(
               egraph_util:convert_to_binary(Key)),
    TimeoutMsec = egraph_config_util:mysql_rw_timeout_msec(detail),
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(key, RawKey, BaseTableName),
    UpdatedDetails = Details#{?EGRAPH_DETAILS_SPECIAL_KEY => egraph_util:convert_to_binary(Key)},
    SerializedDetails = egraph_compression_util:serialize_data(
                          UpdatedDetails, PreferredCompressionId),
    NewIndexes = #{<<"indexes">> => GenericIndexes,
                   <<"lowercase_indexes">> => LowercaseIndexes},
    SerializedIndexes = egraph_compression_util:serialize_data(
                          NewIndexes,
                          PreferredCompressionId),
    HexKey = egraph_util:bin_to_hex_binary(RawKey),
    ReturnLoc = iolist_to_binary(
                  [HexKey,
                   <<"?keytype=rawhex">>]),
    case read_resource(RawKey, create) of
        {error, not_found} ->
            case sql_insert_record(TableName, RawKey,
                                   SerializedDetails,
                                   SerializedIndexes,
                                   UpdatedDetails,
                                   GenericIndexes,
                                   LowercaseIndexes,
                                   TimeoutMsec) of
                true ->
                    {{true, ReturnLoc}, State};
                false ->
                    {false, State}
            end;
        {ok, [DbInfo]} ->
            DbStoredDetailsInfo = maps:get(<<"details">>, DbInfo),
            DbStoredIndexesInfo = maps:get(<<"indexes">>, DbInfo, null),
            lager:debug("DbStoredDetailsInfo = ~p", [DbStoredDetailsInfo]),
            DbStoredKey = maps:get(?EGRAPH_DETAILS_SPECIAL_KEY,
                                   DbStoredDetailsInfo),
            lager:debug("[compare] ~p == ~p", [DbStoredKey, egraph_util:convert_to_binary(Key)]),
  TODO read details and check for collision
            case DbStoredKey == egraph_util:convert_to_binary(Key) of
                true ->
                    OldVersion = maps:get(<<"version">>, DbInfo),

                    DetailsHashHexBin = egraph_util:bin_to_hex_binary(
                                          egraph_util:generate_xxhash_binary(
                                            egraph_util:convert_to_binary(
                                              SerializedDetails))),
                    DbDetailsHash = maps:get(<<"details_hash">>, DbInfo, undefined),
                    AnyChange = (DetailsHashHexBin =/= DbDetailsHash) orelse (NewIndexes =/= DbStoredIndexesInfo),
                    case AnyChange of
                        true ->
                            true = sql_update_record(TableName,
                                              OldVersion,
                                              RawKey,
                                              DbStoredDetailsInfo,
                                              DbStoredIndexesInfo,
                                              SerializedDetails,
                                              SerializedIndexes,
                                              GenericIndexes,
                                              LowercaseIndexes,
                                              UpdatedDetails,
                                              TimeoutMsec);
                        _ ->
  Lets save some write 
                            ok
                    end,
                    {{true, ReturnLoc}, State};
                false ->
                    lager:error("Collission in hash detected between DbStoredKey = ~p and Key = ~p",
                                [DbStoredKey, egraph_util:convert_to_binary(Key)]),
                    {false, State}
            end
    end.

delete_resource(RawKey) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(key, RawKey, BaseTableName),
    Q = iolist_to_binary([<<"DELETE FROM ">>,
                          TableName,
                          <<" WHERE source=?">>]),
    Params = [RawKey],
    TimeoutMsec = egraph_config_util:mysql_rw_timeout_msec(detail),
    PoolName = egraph_config_util:mysql_rw_pool(detail),
    case egraph_sql_util:mysql_write_query(
           PoolName,
           Q, Params, TimeoutMsec) of
        ok ->
            true;
        _ ->
            false
    end.

-spec read_resource(binary(), create) -> {ok, [map()]} | {error, term()}.
read_resource(RawKey, create) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                 key, RawKey, BaseTableName),
    RawKeyHexString = egraph_util:bin_to_hex_binary(RawKey),
    Q = iolist_to_binary([<<"SELECT * FROM ">>,
                          TableName,
                          <<" WHERE source=X'">>,
                          RawKeyHexString,
                          <<"'">>]),
  Params = [ RawKey ] ,
    read_generic_resource(Q, undefined, create, BaseTableName, TableName).

-spec read_resource(binary()) -> {ok, [map()]} | {error, term()}.
read_resource(RawKey) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                 key, RawKey, BaseTableName),
    RawKeyHexString = egraph_util:bin_to_hex_binary(RawKey),
    Q = iolist_to_binary([<<"SELECT * FROM ">>,
                          TableName,
                          <<" WHERE source=X'">>,
                          RawKeyHexString,
                          <<"'">>]),
  Params = [ RawKey ] ,
    read_generic_resource(Q, undefined).

  ( which is same as the first one ) .
-spec read_resources([binary()]) -> [map()].
read_resources(RawKeys) ->
  TODO move to include
	lists:foldl(fun(L, AccIn) ->
                        read_resources_internal(L, AccIn)
                end, [], egraph_util:segments(MaxLen, RawKeys)).

read_resources_internal([H | Rest] = RawKeys, AccIn) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
  derive table name from first one while the
    TableName = egraph_shard_util:sharded_tablename(
                 key, H, BaseTableName),
    Qs = [<<"?">>] ++ [<<",?">> || _ <- Rest],
    Q = iolist_to_binary([<<"SELECT * FROM ">>,
                          TableName,
                          <<" WHERE source in (">>,
                          Qs,
                          <<")">>]),
    Params = RawKeys,
  TODO crashing in case of failure
    case read_generic_resource(Q, Params) of
        {ok, R} ->
            R ++ AccIn;
        _ ->
            AccIn
    end.

-spec search_resource(ShardKey :: integer(),
                      previous | next,
                      Datetime :: calander:datetime(),
                      LimitDatetime :: calander:datetime(),
                      SoftLimit :: non_neg_integer()) ->
    {ok, [map()]} | {error, term()}.
search_resource(ShardKey, previous, Datetime, LimitDatetime, SoftLimit) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                  shard_id, ShardKey, BaseTableName),
    case count_next_updated_datetime(
           previous, TableName, Datetime, SoftLimit) of
        {ok, {MinDatetime, MaxDatetime}} ->
            Q = iolist_to_binary([<<"SELECT * FROM ">>,
                                  TableName,
                                  <<" WHERE updated_datetime >= ? and updated_datetime <= ? order by updated_datetime desc">>]),
            Params = [erlang:max(MinDatetime, LimitDatetime), MaxDatetime],
            read_generic_resource(Q, Params);
        E ->
            E
    end;
search_resource(ShardKey, next, Datetime, LimitDatetime, SoftLimit) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                  shard_id, ShardKey, BaseTableName),
    case count_next_updated_datetime(
           next, TableName, Datetime, SoftLimit) of
        {ok, {MinDatetime, MaxDatetime}} ->
            Q = iolist_to_binary([<<"SELECT * FROM ">>,
                                  TableName,
                                  <<" WHERE updated_datetime >= ? and updated_datetime <= ? order by updated_datetime asc">>]),
            Params = [MinDatetime, erlang:min(MaxDatetime, LimitDatetime)],
            read_generic_resource(Q, Params);
        E ->
            E
    end.

-spec read_all_resource(ShardKey :: integer(),
                        Limit :: integer(),
                        Offset :: integer()) ->
    {ok, [map()], NewOffset :: integer()} | {error, term()}.
read_all_resource(ShardKey, Limit, Offset) ->
    BaseTableName = ?EGRAPH_TABLE_DETAILS_BASE,
    TableName = egraph_shard_util:sharded_tablename(
                  shard_id, ShardKey, BaseTableName),
    Q = iolist_to_binary([<<"SELECT * FROM ">>,
                          TableName,
                          <<" LIMIT ? OFFSET ?">>]),
    Params = [Limit, Offset],
    case read_generic_resource(Q, Params) of
        {ok, R} ->
            {ok, R, Offset + length(R)};
        E ->
            E
    end.

read_generic_resource(Query, Params, read, BaseTableName, TableName) ->
    ConvertToMap = true,
    TimeoutMsec = egraph_config_util:mysql_ro_timeout_msec(detail),
    IsRetry = false,
    IsReadOnly = true,
    ReadPools = egraph_config_util:mysql_ro_pools(detail),
    PoolName = lists:nth(1, ReadPools),
    case egraph_sql_util:run_sql_read_query_for_shard(
           PoolName,
           BaseTableName,
           ReadPools,
           TableName,
           Query, Params, TimeoutMsec, IsRetry, IsReadOnly,
           ConvertToMap) of
        {ok, Maps} ->
            Maps2 = lists:foldl(fun transform_result/2, [], Maps),
            {ok, Maps2};
        Error ->
            Error
    end;
read_generic_resource(Query, Params, create, BaseTableName, TableName) ->
    ConvertToMap = true,
    TimeoutMsec = egraph_config_util:mysql_rw_timeout_msec(detail),
    IsRetry = false,
    IsReadOnly = false,
    PoolName = egraph_config_util:mysql_rw_pool(detail),
    ReadPools = [PoolName],
    case egraph_sql_util:run_sql_read_query_for_shard(
           PoolName,
           BaseTableName,
           ReadPools,
           TableName,
           Query, Params, TimeoutMsec, IsRetry, IsReadOnly,
           ConvertToMap) of
        {ok, Maps} ->
            Maps2 = lists:foldl(fun transform_result/2, [], Maps),
            {ok, Maps2};
        Error ->
            Error
    end.

read_generic_resource(Query, Params, false = ConvertToMap) ->
    TimeoutMsec = egraph_config_util:mysql_ro_timeout_msec(detail),
    ReadPools = egraph_config_util:mysql_ro_pools(detail),
    case egraph_sql_util:mysql_query(
           ReadPools, Query, Params, TimeoutMsec, ConvertToMap) of
        {ok, {_ColumnNames, MultipleRowValues}} ->
            {ok, MultipleRowValues};
        Error ->
            Error
    end;
read_generic_resource(Query, Params, true) ->
    read_generic_resource(Query, Params).

read_generic_resource(Query, Params) ->
    ConvertToMap = true,
    TimeoutMsec = egraph_config_util:mysql_ro_timeout_msec(detail),
    ReadPools = egraph_config_util:mysql_ro_pools(detail),
    case egraph_sql_util:mysql_query(
           ReadPools, Query, Params, TimeoutMsec, ConvertToMap) of
        {ok, Maps} ->
            Maps2 = lists:foldl(fun transform_result/2, [], Maps),
            {ok, Maps2};
        Error ->
            Error
    end.

transform_result(E, AccIn) ->
    E2 = case maps:get(<<"source">>, E, undefined) of
             undefined ->
                 E;
             Source ->
                 E#{<<"source">> =>
                    egraph_util:bin_to_hex_binary(Source)}
         end,
    E3 = case maps:get(<<"details_hash">>, E2, undefined) of
             undefined ->
                 E2;
             DetailsHash ->
                 E2#{<<"details_hash">> =>
                     egraph_util:bin_to_hex_binary(DetailsHash)}
         end,
    E4 = case maps:get(<<"updated_datetime">>, E3, undefined) of
             undefined ->
                 E3;
             UpdatedDateTime ->
                 E3#{<<"updated_datetime">> =>
                     qdate:to_string(<<"Y-m-d H:i:s">>, UpdatedDateTime)}
         end,
    E5 = case maps:get(<<"details">>, E4, undefined) of
             undefined ->
                 E4;
             Details ->
                 E4#{<<"details">> =>
                     egraph_compression_util:deserialize_data(Details)}
         end,
    E6 = case maps:get(<<"indexes">>, E5, null) of
             null ->
                 E5;
             Indexes ->
                 E5#{<<"indexes">> =>
                     egraph_compression_util:deserialize_data(Indexes)}
         end,
    [E6 | AccIn].

sql_insert_record(TableName, RawKey, SerializedDetails, SerializedIndexes, UpdatedDetails, GenericIndexes, LowercaseIndexes, TimeoutMsec) ->
    DefaultVersion = 0,
    DetailsHash = egraph_util:generate_xxhash_binary(
                    egraph_util:convert_to_binary(SerializedDetails)),
    UpdatedDateTime = qdate:to_date(erlang:system_time(second)),
    Q = iolist_to_binary([<<"INSERT INTO ">>,
                          TableName,
                          <<" (source, version, details_hash, details, updated_datetime, indexes) VALUES(?, ?, ?, ?, ?, ?)">>]),
    Params = [RawKey, DefaultVersion, DetailsHash, SerializedDetails,
              UpdatedDateTime, SerializedIndexes],

  Create indexes first because they must exist before adding the
  Worse , when no index were created for the first time then
  the second time instead of insert the update method will be
    GenericIndexResults = create_generic_index(GenericIndexes, UpdatedDetails, RawKey, []),
    LowerIndexResults = create_lowercase_index(LowercaseIndexes, UpdatedDetails, RawKey, []),
    case LowerIndexResults ++ GenericIndexResults of
        [] ->
  TODO : open a single transaction instead of multiple , since
            PoolName = egraph_config_util:mysql_rw_pool(detail),
            case egraph_sql_util:mysql_write_query(
                   PoolName, Q, Params, TimeoutMsec) of
                ok ->
                    lager:debug("UpdatedDetails = ~p", [UpdatedDetails]),
                    true;
                ErrorInfo ->
                    lager:error("RawKey = ~p, Cannot rollback although index creation succeeded, ErrorInfo = ~p",
                                [RawKey, ErrorInfo]),
                    false
            end;
        _ ->
            false
    end.

sql_update_record(TableName, OldVersion, RawKey, DbStoredDetailsInfo, DbStoredIndexesInfo,
                  SerializedDetails, SerializedIndexes, GenericIndexes, LowercaseIndexes,
                  UpdatedDetails, TimeoutMsec) ->
    {OldGenIndexes, OldLowerIndexes} = case DbStoredIndexesInfo of
                                           null ->
                                               {maps:get(?EGRAPH_DETAILS_SPECIAL_INDEX,
                                                         DbStoredDetailsInfo),
                                                []};
                                           _ ->
                                               {maps:get(<<"indexes">>,
                                                         DbStoredIndexesInfo),
                                                maps:get(<<"lowercase_indexes">>,
                                                         DbStoredIndexesInfo)}
                                       end,
    Version = OldVersion + 1,
    DetailsHash = egraph_util:generate_xxhash_binary(
                    egraph_util:convert_to_binary(SerializedDetails)),
    UpdatedDateTime = qdate:to_date(erlang:system_time(second)),

    GenericIndexResults = change_indexes(
                            generic, RawKey, OldGenIndexes, GenericIndexes,
                            DbStoredDetailsInfo, UpdatedDetails),
    LowercaseIndexResults = change_indexes(
                              lowercase, RawKey, OldLowerIndexes, LowercaseIndexes,
                              DbStoredDetailsInfo, UpdatedDetails),

    IndexResults = GenericIndexResults ++ LowercaseIndexResults,
    case IndexResults of
        [] ->
            Q = iolist_to_binary([<<"UPDATE ">>,
                                  TableName,
                                  <<" SET version=?, details_hash=?, details=?, updated_datetime=?, indexes=? WHERE source=? and version=?">>]),
            Params = [Version, DetailsHash, SerializedDetails,
                      UpdatedDateTime, SerializedIndexes, RawKey, OldVersion],
            PoolName = egraph_config_util:mysql_rw_pool(detail),
            case egraph_sql_util:mysql_write_query(
                   PoolName, Q, Params, TimeoutMsec) of
                ok -> true;
                DbUpdateError ->
                    lager:error("RawKey = ~p, Cannot rollback although update failed, DbUpdateError = ~p",
                                [RawKey, DbUpdateError]),
                    false
            end;
        _ ->
            lager:error("RawKey = ~p, Cannot rollback although index creation partially failed, IndexResults = ~p",
                        [RawKey, IndexResults]),
            false
    end.

delete_index([], _Type, _Details, _RawKey, AccIn) ->
    AccIn;
delete_index([JsonPath | Rest], Type, Details, RawKey, AccIn) ->
    IndexName = case Type of
                    generic -> lists:last(JsonPath);
                    lowercase -> iolist_to_binary(
                                   [lists:last(JsonPath),
                                    ?EGRAPH_INDEX_SPECIAL_SUFFIX_LOWERCASE])
                end,
    Key = nested:get(JsonPath, Details),
    KeyType = egraph_shard_util:key_datatype(Key),
    case egraph_index_model:delete_resource(Key, KeyType, RawKey, IndexName) of
        true ->
            delete_index(Rest, Type, Details, RawKey, AccIn);
        false ->
            case egraph_index_model:delete_resource(Key, KeyType, RawKey,IndexName) of
                true ->
                    delete_index(Rest, Type, Details, RawKey, AccIn);
                false ->
                    delete_index(Rest, Type, Details, RawKey, [{Key, KeyType, IndexName} | AccIn])
            end
    end.

create_db_index(generic, Indexes, UpdatedDetails, RawKey, AccIn) ->
    create_generic_index(Indexes, UpdatedDetails, RawKey, AccIn);
create_db_index(lowercase, Indexes, UpdatedDetails, RawKey, AccIn) ->
    create_lowercase_index(Indexes, UpdatedDetails, RawKey, AccIn).

create_generic_index(Indexes, UpdatedDetails, RawKey, AccIn) ->
    F = fun(E) -> E end,
    NameFun = fun(L) -> lists:last(L) end,
    create_index(Indexes, UpdatedDetails, RawKey, AccIn, F, NameFun).

create_lowercase_index(Indexes, UpdatedDetails, RawKey, AccIn) ->
    F = fun(E) -> string:lowercase(E) end,
    NameFun = fun(L) -> iolist_to_binary(
                          [lists:last(L),
                           ?EGRAPH_INDEX_SPECIAL_SUFFIX_LOWERCASE])
              end,
    create_index(Indexes, UpdatedDetails, RawKey, AccIn, F, NameFun).

create_index([], _UpdatedDetails, _RawKey, AccIn,
             _TransformFun, _IndexNameFun) ->
    AccIn;
create_index([JsonPath | Rest], UpdatedDetails, RawKey, AccIn,
             TransformFun, IndexNameFun) ->
    IndexName = IndexNameFun(JsonPath),
    HexId = egraph_util:bin_to_hex_binary(RawKey),
    Key = TransformFun(nested:get(JsonPath, UpdatedDetails)),
    KeyType = egraph_shard_util:key_datatype(Key),
    IndexInfo = #{<<"key_data">> => Key,
                  <<"id">> => HexId,
                  <<"index_name">> => IndexName,
                  <<"key_type">> => KeyType},
    lager:debug("IndexInfo = ~p", [IndexInfo]),
    case egraph_index_model:create_or_update_info(IndexInfo, []) of
        {false, _} ->
            case egraph_index_model:create_or_update_info(IndexInfo, []) of
                {false, _} ->
                    create_index(Rest, UpdatedDetails, RawKey, [IndexInfo | AccIn],
                                 TransformFun, IndexNameFun);
                _ ->
                    create_index(Rest, UpdatedDetails, RawKey, AccIn,
                                 TransformFun, IndexNameFun)
            end;
        _ ->
            create_index(Rest, UpdatedDetails, RawKey, AccIn,
                         TransformFun, IndexNameFun)
    end.

reindex_key(Details, Indexes) ->
    Key = maps:get(
            ?EGRAPH_DETAILS_SPECIAL_KEY,
            Details),
    RawKey = egraph_util:generate_xxhash_binary(
               egraph_util:convert_to_binary(Key)),
    case Indexes of
        null ->
            GenericIndexes = maps:get(
                               ?EGRAPH_DETAILS_SPECIAL_INDEX,
                               Details,
                               []),
            lager:debug("Reindex Key = ~p", [Key]),
            create_db_index(generic, GenericIndexes, Details, RawKey, []);
        _ ->
            LowercaseIndexes = maps:get(<<"lowercase_indexes">>, Indexes),
            GenericIndexes = maps:get(<<"indexes">>, Indexes),
            lager:debug("Reindex Key = ~p", [Key]),
            A = create_db_index(generic, GenericIndexes, Details, RawKey, []),
            create_db_index(lowercase, LowercaseIndexes, Details, RawKey, A)
    end.

count_next_updated_datetime(previous, TableName, Datetime, SoftLimit) ->
    Q = iolist_to_binary([<<"SELECT updated_datetime FROM ">>,
                          TableName,
                          <<" WHERE updated_datetime <= ? order by updated_datetime desc LIMIT ?">>]),
    Params = [Datetime, SoftLimit],
    case read_generic_resource(Q, Params, false) of
        {ok, MultipleRowValues} ->
            DateTimes = lists:flatten(MultipleRowValues),
            {ok, {lists:last(DateTimes),
                  lists:nth(1, DateTimes)}};
        E ->
            E
    end;
count_next_updated_datetime(next, TableName, Datetime, SoftLimit) ->
    Q = iolist_to_binary([<<"SELECT updated_datetime FROM ">>,
                          TableName,
                          <<" WHERE updated_datetime >= ? order by updated_datetime asc LIMIT ?">>]),
    Params = [Datetime, SoftLimit],
    case read_generic_resource(Q, Params, false) of
        {ok, MultipleRowValues} ->
            DateTimes = lists:flatten(MultipleRowValues),
            {ok, {lists:nth(1, DateTimes),
                  lists:last(DateTimes)}};
        E ->
            E
    end.

extract_indexes(Indexes) when is_list(Indexes) ->
    {Indexes, []};
extract_indexes(Indexes) when is_map(Indexes) ->
    {maps:get(<<"indexes">>, Indexes),
     maps:get(<<"lowercase_indexes">>, Indexes, [])}.

  1 . First delete indexes
  2 . Add new indexes
  3 . Update existing index if the value has changed
  4 . Add the data once all the above is successful .
change_indexes(Type, RawKey, OldIndexes, Indexes, DbStoredDetailsInfo, UpdatedDetails) ->
    ChangedIndexes = lists:filter(fun(JsonPath) ->
                                          try
                                              OldValue = nested:get(JsonPath, DbStoredDetailsInfo),
                                              NewValue = nested:get(JsonPath, UpdatedDetails),
                                              OldValue =/= NewValue
                                          catch
                                              _:_ ->
                                                  false
                                          end
                                  end, Indexes),
    DeletedIndexes = lists:filter(fun(JsonPath) ->
                        not lists:member(JsonPath, Indexes)
                                  end, OldIndexes),
    DeletedIndexResults = delete_index(
                            ChangedIndexes ++ DeletedIndexes,
                            Type,
                            DbStoredDetailsInfo,
                            RawKey,
                            []),
    case DeletedIndexResults of
        [] ->
            AddedIndexes = lists:filter(fun(JsonPath) ->
                              not lists:member(JsonPath, OldIndexes)
                                        end, Indexes),
            AddedGenIndexResults = create_db_index(
                                     Type,
                                     ChangedIndexes ++ AddedIndexes,
                                     UpdatedDetails,
                                     RawKey,
                                     []),
            AddedGenIndexResults;
        _ ->
            lager:error("RawKey = ~p, Cannot rollback although index deletion failed, Type = ~p, DeletedIndexResults = ~p",
                        [RawKey, Type, DeletedIndexResults]),
            DeletedIndexResults
    end.


