jc

%%% ----------------------------------------------------------------------------
%%% @author Jim Rosenblum
%%% @copyright (C) 2011-2015, Jim Rosenblum
%%% @doc Library module that manages the mnesia tables mapping 
%%% {@link map_name(). Map} and {@link key(). Key} to a {@link value(). Value}. 
%%%
%%% IT IS UP TO THE CALLER TO WRAP THESE FUNCTIONS IN THE APPROPRIATE MNESIA 
%%% TRANSACTIONS
%%%
%%% This module interacts with mnesia to persist and access keys in service
%%% of the jc module.
%%%
%%% Up to four additional, indexed fields may be used to support faster 
%%% evict_match and values_match operations. These operations only make sense 
%%% for JSON values and look for Keys whose Values contain a specified value
%%% at a given 'path' within the JSON Value. That is, values_match and 
%%% evict_match select or delete based on a 'path' criteria, such as 
%%% "some.path.in.json.value=2". The additional fields, if used, are populated
%%% during a PUT operation -- the JSON Value is travesed for the target provided
%%% by the path criteria and the targetd value is put in an indxed field for 
%%% fast look up in service of the values_match and evict_match operations.
%%%
%%% {key_to_val, {Key, Map}, Map, Key, Value, i1, i2, i3, i4, ...}
%%% to_index contains a row for each {Map and JSON path to be indexed
%%% {to_index, {bed,{"some","path"}, bed, 6}
%%% the above record indicates that i1 (the 6th field) should be used
%%% to store the value at some.path in Values cached in the Bed map.
%%%
%%% @version {@version}
%%% @end
%%% Created : 16 Oct 2011 by Jim Rosenblum
%%% ----------------------------------------------------------------------------
-module(jc_store).
-compile(nowarn_deprecated_function). % accomidate now() for v < 18

%% Jc_store public API
-export([clear/1,
	 evict/2, 
	 fun_match/4,
	 flush/1,
	 get/2,
	 get_map/1, get_map_since/2, key_set/1,
	 maps/0,
	 put/5]).

%% Meta-data API.
-export([up_nodes/0, stats/1]).

%% Custom-field indexing API.
-export([indexes/0, indexes/1, create_index/2, start_indexing/2, stop_indexing/2]).

%% Callback used by the jc_eviction manager to delete a cached value at TTL.
-export([delete_record_by_ref/1]).


 % Record and type definitions.
-include("../include/records.hrl").   

 % Use QLC for some querying.
-include_lib("stdlib/include/qlc.hrl").


-define(NO_SEQ, -1).
       

%% =============================================================================
%% Meta data API
%% =============================================================================


%% -----------------------------------------------------------------------------
%% Return list of running mnesia nodes.
%%
-spec up_nodes() -> list(node()).

up_nodes()->
    mnesia:system_info(running_db_nodes).


%% -----------------------------------------------------------------------------
%% @doc Return vairous stats information. Currently size and uptime information.
%% 
-spec stats(size | up) -> {size, [{TableNm::atom(), 
				 {records, RecCnt::non_neg_integer()},
				 {bytes, Words::non_neg_integer()}}]} |
			  {uptime, term()} |
			  {error, not_found | badarg}.
stats(size) ->
    Data = 
	[{T, 
	  {records, mnesia:table_info(T, size)}, 
	  {bytes, mnesia:table_info(T, memory) * erlang:system_info(wordsize)}}
	 || T <-mnesia:system_info(tables)],
    {size, Data};

stats(up) ->
    case mnesia:dirty_read(stats, 'jc_store_up_time') of
	[#stats{value=Value}] ->
	    {uptime, Value};
	[] ->
	    {error, not_found}
    end;

stats(_) ->
    {error, badarg}.


%%------------------------------------------------------------------------------
%% @doc Return a sorted list of all maps currently in the cache.
%% 
-spec maps() -> [map_name()].

maps() ->
    QH = qlc:sort(
	   qlc:q([Map || #key_to_value{map=Map} <- 
			     mnesia:table(key_to_value)], {unique, true})
	  ),
    qlc:e(QH).



%% =============================================================================
%% Storage service API used by jc
%% =============================================================================


%% -----------------------------------------------------------------------------
%% @doc Clear all data for the given map.
%%
-spec clear(map_name()) -> ok.

clear(Map) ->
    F = fun() ->
		Items = mnesia:index_read(key_to_value, Map, #key_to_value.map),
		[mnesia:delete_object(Rec) || Rec <- Items],
		mnesia:delete({seq, Map})
	end,
     mnesia:sync_dirty(F),
    ok.


%% -----------------------------------------------------------------------------
%% @doc Evict Map/Key from the cache
%% 
-spec evict(map_name(), key()) ->  ok.

evict(Map, Key) ->
    mnesia:delete({key_to_value, {Key, Map}}),
    ok.


%% -----------------------------------------------------------------------------
%% @doc Delete the entire cache with or without triggering notifications based
%% on the parameter.
%%
-spec flush(silent | loud) ->  ok.

flush(silent) ->
    mnesia:clear_table(key_to_value),
    mnesia:clear_table(seq),
    mnesia:clear_table(auto_index),
    ok;

flush(loud) ->
    [mnesia:delete(key_to_value, Akey, write) || 
	Akey <- mnesia:all_keys(key_to_value)],

    mnesia:clear_table(seq),
    mnesia:clear_table(auto_index),
    ok.


%%------------------------------------------------------------------------------
%% @doc Return the #key_to_value{} which contains the data associated with 
%% {@link map_name(). Key} and {@link key(). Key}.
%% 
-spec get(map_name(), key()) -> {ok, key_to_value() | jc_miss}.

get(Map, Key) ->
    case mnesia:read(key_to_value, {Key, Map}) of
	[Rec] ->
	    {ok, Rec};
	[] ->
	    {ok, jc_miss}
    end.


%% -----------------------------------------------------------------------------
%% @doc Return all the #key_to_value() for a given {@link map_name(). Map}.
%% 
-spec get_map(Map::map_name()) -> {ok, [Results::key_to_value()]}.

get_map(Map) ->
    {ok, mnesia:index_read(key_to_value, Map, #key_to_value.map)}.


%% -----------------------------------------------------------------------------
%% @doc Return all keys in the given cache {@link map_name(). Map}.
%% 
-spec key_set(Map::map_name()) -> {ok, [Keys::key()]}.

key_set(Map) ->
    Q = qlc:q([R#key_to_value.key || R <- mnesia:table(key_to_value),
				     R#key_to_value.map == Map]),
    {ok, qlc:e(Q)}.


%% -----------------------------------------------------------------------------
%% @doc Return all the #key_to_value() for a given {@link map_name(). Map} 
%% whose create_tm is older than now - AgeSecs.
%%
-spec get_map_since(Map::map_name(), AgeSecs::seconds()) -> 
			   [Results::key_to_value()] | 
			   {error, module(), tuple() | bad_object}.

get_map_since(Map, AgeSecs) ->
    Limit = now_to_uepoch() - (AgeSecs * 1000000),
    Q = qlc:q([R || R <- mnesia:table(key_to_value),
		    R#key_to_value.map == Map,
		    R#key_to_value.create_tm < Limit]),
    qlc:e(Q).
		     

%% -----------------------------------------------------------------------------
%% @doc Put the {link @map_name(). Map} and {@link key(). Key} record.
%% Return a tuple with a reference to the record so that an eviction timer can
%% be created by the caller. IF THE PUT EVICTED AN ENTRY, JC_PSUB WILL DETECT
%% THIS, CANCEL THE OLD TIMER AND RAISE THE EVENT TO SUBSCRIBERS.
%% 
-spec put(map_name(), key(), value(), ttl(), rec_ref()) -> 
		 {ok, {put, rec_ref()}}.

put(Map, Key, Value, TTL, Ref) ->
    Time = now_to_uepoch(),
    Record = #key_to_value{jc_key      = {Key, Map},
			   map         = Map,
			   key         = Key, 
			   value       = Value,
			   create_tm   = Time,
			   last_update = Time,
			   ttl_secs    = TTL,
			   ref         = Ref},

    Augmented = add_indexes(Map, Value, Record),
    mnesia:write(Augmented),

    {ok, {put, Ref}}.



% -----------------------------------------------------------------------------
% Populate additional fields, configured for this map, with the value indicated
% by the configured JSON path in the Value JSON.
%
add_indexes(Map, Value, Record) ->
    case mnesia:index_read(to_index, Map, #to_index.map_name) of
	[] -> 
	    % nothing to do
	    Record;
	Indexes ->
	    try decode(Value) of
		JsonStruct ->
		    index(Indexes, JsonStruct, Record)
	    catch
		_:_ -> 
		    Record 
	    end
    end.


% ------------------------------------------------------------------------------
% For each JSON-path to index, get the value from the decoded JSON structure
% at the path, and add that value to the correct field in the record.
%
index([], _JStruct, Record) ->
    Record;
index([#to_index{map_path = {Map, Path}, position = Pos}|Rs], JStruct, Record)->
    try
	case ej_get(Path, JStruct) of
	    undefined   -> index(Rs, JStruct, Record);
	    jc_ej_error -> index(Rs, JStruct, Record);
	    JValue      -> 
		index(Rs, JStruct, setelement(Pos, Record, {Map, JValue}))
	end
    catch
	_:_ -> index(Rs, JStruct, Record)
    end.


%% -----------------------------------------------------------------------------
%% @doc Delete the cache element by its record reference. Used by eviction 
%% manager.
%%
-spec delete_record_by_ref(rec_ref()) ->  ok.

delete_record_by_ref(RecRef) ->
    case mnesia:index_read(key_to_value, RecRef, #key_to_value.ref) of
	[] ->
	    ok;
	Records  ->
	    F = fun(#key_to_value{jc_key=JCK}) ->
			mnesia:delete({key_to_value, JCK})
		end,
	    lists:foreach(F, Records)
    end.


%% -----------------------------------------------------------------------------
%% @doc Select those Map items whose JSON value, at the given path, equals Test 
%% and invoke the supplied function/4 with the Map, Key, Value and 
%% fold accumulator as parameters. 
%%
-spec fun_match(map_name(), Paths::tuple(), Test::atom(), Fun::fun()) -> 
		       [term()].


fun_match(Map, Path, Test, Fun) ->
    case try_index_match(Map, Path, Test, Fun) of
	undefined -> 
	    map_match(Map, Path, Test, Fun);
	Results -> 
	    Results
    end.


% ------------------------------------------------------------------------------
% If an entry in to_index indicates that the path is backed by an indexible
% custom field, use it to select only the records wich meet the criteria.
%
try_index_match(Map, Path, Test, Fun)->
    case index_get(Map, Path, Test) of
	undefined ->
	    lager:debug("~p: no index values found for ~p", 
			[?MODULE, {Map, Path}]),
	    undefined;
	Recs ->
	    lager:debug("~p: using index for ~p", [?MODULE, {Map, Path}]),
	    F = fun(#key_to_value{key = K, value = V}, Acc) ->
			Fun(Map, K, V, Acc)
		end,
	    lists:foldl(F, [], Recs)
    end.



% ------------------------------------------------------------------------------
% Do an index read based on the index-ed field indiated by the to_index record
% for the given Map and Json Path. If using the analyzer, update the statistics.
%
index_get(Map, Path, Test) -> 
    update_stats(Map, Path),
    case mnesia:read(to_index, {Map, Path}) of
	[] -> undefined;
	[#to_index{position = Pos}] ->
	    mnesia:index_read(key_to_value, {Map, Test}, Pos)
    end.


update_stats(Map, Path) ->
    jc_analyzer:update_statistic(Map, Path).


% ------------------------------------------------------------------------------
% No indexes to help so do it the expensive way: for all records in the map,
% decode each JSON value looking for the item indicated by Path and checking for 
% equality with Test.
%
map_match(Map, Path, Test, Fun) ->
    {ok, Recs} = get_map(Map),
    F = fun(#key_to_value{key=K, value=V}, Acc) ->
		case decode(V) of
		    jc_error -> 
			Acc;
		    Obj -> 

			Target = ej_get(Path, Obj),
			case Target of
			    Test ->
				Fun(Map, K, V, Acc);
			    _  -> 
				Acc
			end
		end;
	   (_, Acc) ->
		Acc		
	end,
    
    lists:foldl(F, [], Recs).


decode(Value) ->
    try
	V = case is_binary(Value) of
		true -> Value;
		false -> iolist_to_binary(Value)
	    end,
        jsone:decode(V, [{object_format, proplist}])
    catch 
	_:_-> jc_error
    end.



%% =============================================================================
%% Custom-field Indexing API
%% =============================================================================


% There are CUSTOM_CNT custom fields on the key_to_value record available for
% use as indexed fields to support JSON path-querries of the cach'ed Value.
-define(CUSTOM_FIELD_NAMES, [i1, i2, i3, i4]).
-define(CUSTOM_FIELD_POSITIONS, [#key_to_value.i1, 
				 #key_to_value.i2, 
				 #key_to_value.i3, 
				 #key_to_value.i4]).
-define(CUSTOM_CNT, length(?CUSTOM_FIELD_NAMES)).
-define(FIELD_NAME(I), lists:nth(I-#key_to_value.i1 + 1, ?CUSTOM_FIELD_NAMES)).



%% -----------------------------------------------------------------------------
%% @doc Start using a custom, field for the value found in JSON Values at
%% the path with respect to a given map.
%% Start indexing the bed map with respect to "menu.id" - 
%% start_indexing(bed, "menu.id"}).
%%
-spec start_indexing(map_name(), Path::string() | binary()) -> 
			    ok | {error, no_indexes_available | any()}.

start_indexing(Map, Path) ->
    case make_ej_path(Path) of
	error ->
	    lager:error("~p: could not start indexing ~p.", 
		       [?MODULE, {Map, Path}]);
	BPath ->
	    create_index(Map, BPath)
    end.



%% -----------------------------------------------------------------------------
%% @private used by {@link start_indexing/2. start_indexing/2} and also by
%% jc_analyzer to create indexes on json querries that have been used alot. This
%% function takes tuples as opposed to a string, dot-path specificationfor the
%% search criteria.
%%
-spec create_index(map(), tuple()) -> ok | {error, term()}.

create_index(Map, BPath) ->
    F = fun() ->
		mnesia:delete({to_index, {Map, BPath}}),
		case next_custom_field(Map) of
		    {error, _} = Error->
			Error;
		    {ok, NextPos} ->
			mnesia:write(#to_index{map_path = {Map, BPath},
					       map_name = Map,
					       position = NextPos}),
			clear(Map),
			NextPos
		end
	end,

    case mnesia:transaction(F) of
	{atomic, {error, E}} ->
	    lager:warning("~p: could not start indexing ~p: ~p.",
			  [?MODULE, {Map, BPath}, E]),
	    {error, E};

	{atomic, Pos} ->
	    lager:info("~p: to_index now includes ~p in pos. ~p.", 
		       [?MODULE, {Map, BPath}, Pos]),
            % didn't want the overhead of an indexed column unless
            % we are going to use it.
	    mnesia:add_table_index(key_to_value, ?FIELD_NAME(Pos)),
	    ok;

	{aborted, Reason} ->
	    lager:info("~p: could not start indexing ~p: ~p.", 
		       [?MODULE, {Map, BPath}, Reason]),
	    {error, Reason}
    end.


% ------------------------------------------------------------------------------
% Return the position of the next custom field available for use. Gather all 
% fields being use for a given map, subtrct the list of their positions from the
%  list of positions of all custom fields to see if there is an availabe field.
%
next_custom_field(Map) ->
    QH = qlc:q([R#to_index.position || R <- mnesia:table(to_index),
				       R#to_index.map_name == Map]),
    InUse = qlc:e(QH),
    case (?CUSTOM_FIELD_POSITIONS -- InUse) of
	[] -> 
	    {error, no_indexes_available};
	Available ->
	    {ok, hd(Available)}
    end.


%% -----------------------------------------------------------------------------
%% @doc Stop indexing Map entries with respect to the json path expressed as
%% tuples: {"menu", "id"}, for example. If no one is using a custom field, 
%% stop indexing it.
%%
-spec stop_indexing(map_name(), Path::string() | binary()) -> ok.

stop_indexing(Map, Path) ->
    case make_ej_path(Path) of
	error ->
	    lager:error("~p: could not stop indexing ~p.", 
		       [?MODULE, {Map, Path}]);
	BPath ->
	    lager:info("~p: updating to_index to exclude ~p.", 
		       [?MODULE, {Map, BPath}]),
	    mnesia:transaction(fun() -> mnesia:delete({to_index, {Map, BPath}}) end),
	    remove_unused_index(),
	    ok
    end.

% if no to_index row refers to a column, stop indexing that column.
remove_unused_index()->
    F = fun(Pos) ->
		case mnesia:dirty_index_read(to_index,Pos,#to_index.position) of
		    [] -> 
			mnesia:del_table_index(key_to_value, ?FIELD_NAME(Pos));
		    _ ->
			ok
		end
	end,
    _ = [F(Position) || Position <- ?CUSTOM_FIELD_POSITIONS].
		 

%% -----------------------------------------------------------------------------
%% Return all the JSON-path indexes being used.
%%
-spec indexes() -> [{{map_name, tuple()}, Pos::non_neg_integer()}].

indexes() -> 
    QH = qlc:q([{R#to_index.map_path, R#to_index.position} 
		|| R <- mnesia:table(to_index)]),
    mnesia:async_dirty(fun() -> qlc:e(QH) end ).


%% -----------------------------------------------------------------------------
%% Return all the JSON-path indexes being used for a given map.
%%
-spec indexes(map_name()) -> [{{map_name, tuple()}, Pos::non_neg_integer()}].

indexes(Map) ->
    QH = qlc:q([{R#to_index.map_path, R#to_index.position} 
		|| R <- mnesia:table(to_index),
	           R#to_index.map_name == Map]),
    mnesia:async_dirty(fun() -> qlc:e(QH) end ).



%% =============================================================================
%% Utility Functions
%% =============================================================================


%% -----------------------------------------------------------------------------
%% @doc Utility fn to return the number of micro-seconds since 1/1/1970 
%% -- UNIX-style epoch.
%%

now_to_uepoch() ->
    {A,B,C} = timestamp(),
    ((A * 1000000 + B) * 1000000) + C.


% walk the path of a decoded JSON object using the ej library.
ej_get(Paths, Obj) ->
    try
	jwalk:get(Paths, Obj)
    catch
	_:_ -> jc_ej_error
    end.




% ------------------------------------------------------------------------------
% The Ej library navigates a decoded JSON structure via tuple descibing the 
% path to traverse. Elements are either an object key (string) or an index 
% into an array (number). 
%
-spec make_ej_path(string() | binary()) -> tuple() | error.

make_ej_path(DotString) when is_binary(DotString) -> 
    make_ej_path(binary_to_list(DotString));

make_ej_path(DotString) ->
    try 
	Tokenized = string:tokens(DotString, "."),
	list_to_tuple([to_path_elt(list_to_binary(Elt)) || Elt <- Tokenized])
    catch
	_:_ -> error
    end.

to_path_elt(<<"'", B/binary>>) -> 
    S = size(B) - 1,
    <<E:S/binary, _/binary>> = B,
    E;
to_path_elt(Element) ->    
    try binary_to_integer(Element)
    catch
	_:_ ->
	    Element
    end.

     

% Try to used erlang 18+ timestamp(), support older versions if necessary.
timestamp() ->
    try
	erlang:timestamp()
    catch
	error:undef ->
	    erlang:now()
end.