-module(odi_graph). -behaviour(gen_server). %% API -export([ begin_transaction/1, create_vertex/3, create_edge/5, update/3, delete/3, query/4, query/5, record_load/3, get_cache/1, commit/2, rollback/1 ]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(SERVER, ?MODULE). -type record_data()::#{string()|atom() => any()}. -record(state, { con :: pid(), commands = [] :: [odi:tx_operation()], command_pos = #{} :: #{odi:rid() => pos_integer()}, classes :: #{string() => record_data()}, global_properties :: #{non_neg_integer() => record_data()}, cache = #{} :: #{odi:rid() => {Version::non_neg_integer(), document, Class::string(), Data::record_data()}} }). %%%=================================================================== %%% API %%%=================================================================== -spec begin_transaction(Con::pid()) -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}. %% @doc %% Starts a transaction begin_transaction(Con) -> gen_server:start_link({local, ?SERVER}, ?MODULE, [Con], []). -spec create_vertex(T::pid(), TempId::neg_integer()|odi:rid(), Record::{Class::string(), Data:: record_data()}) -> ok. create_vertex(T, TempId, Record) -> gen_server:call(T, {create_vertex, TempId, Record}). -spec create_edge(T::pid(), TempId::neg_integer()|odi:rid(), FromId::pos_integer()|odi:rid(), ToId::pos_integer()|odi:rid(), Record::{Class::string(), Data:: record_data()}) -> ok. create_edge(T, TempId, FromId, ToId, Record) -> gen_server:call(T, {create_edge, TempId, FromId, ToId, Record}). -spec update(T::pid(), Rid::odi:rid(), Data:: record_data()) -> ok. update(T, Rid, Data) -> gen_server:call(T, {update, Rid, Data}). -spec delete(T::pid(), Rid::odi:rid(), Version::pos_integer()) -> ok. delete(T, Rid, Version) -> gen_server:call(T, {delete, Rid, Version}). -spec query(T::pid(), Query::string(), Limit::integer(), FetchPlan::string()|default) -> [odi:fetched_record()] | odi:error(). query(T, Query, Limit, FetchPlan) -> gen_server:call(T, {query, Query, Limit, FetchPlan, null}). -spec query(T::pid(), Query::string(), Limit::integer(), FetchPlan::string()|default, Params::#{string()=>any()}) -> [odi:fetched_record()] | odi:error(). query(T, Query, Limit, FetchPlan, Params) -> gen_server:call(T, {query, Query, Limit, FetchPlan, Params}). -spec record_load(T::pid(), Rid::odi:rid(), FetchPlan::string()) -> odi:fetched_record()|null. record_load(T, Rid, FetchPlan) -> gen_server:call(T, {record_load, Rid, FetchPlan}). -spec get_cache(T::pid()) -> [odi:fetched_record()]. get_cache(T) -> gen_server:call(T, {get_cache}). -spec commit(T::pid(), TxId::pos_integer()) -> IdRemaps::#{integer() => odi:rid()} | odi:error(). commit(T, TxId) -> gen_server:call(T, {commit, TxId}). -spec rollback(T::pid()) -> ok. rollback(T) -> gen_server:stop(T, normal, 1000). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %%-------------------------------------------------------------------- %% @private %% @doc %% Initializes the server %% %% @spec init(Args) -> {ok, State} | %% {ok, State, Timeout} | %% ignore | %% {stop, Reason} %% @end %%-------------------------------------------------------------------- -spec(init(Args :: term()) -> {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} | {stop, Reason :: term()} | ignore). init([Con]) -> %% [{true, raw, 0, raw, Config}] = odi:record_load(Con, {0, 0}, "", false), %% io:format("Config: ~s~n", [Config]), [{true, document, _Version, _Class, RawSchemas}] = odi:record_load(Con, {0, 1}, "*:-1 index:0", true), Schema = odi_typed:untypify_record(RawSchemas), Classes = index_classes(Schema), GlobalProperties = odi_typed:index_global_properties(Schema), lager:debug("Classes: ~p", [Classes]), lager:debug("GlobalProperties: ~p", [GlobalProperties]), %% Indexes = odi:record_load(Con, {0, 2}, "*:-1 index:0", true), %% io:format("Indexes: ~p~n", [Indexes]), %% {Sequences, []} = odi:query(Con, "SELECT FROM OSequence", -1, ""), %% io:format("Sequences: ~p~n", [Sequences]), {ok, #state{con=Con, classes=Classes, global_properties=GlobalProperties}}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling call messages %% %% @end %%-------------------------------------------------------------------- -spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()}, State :: #state{}) -> {reply, Reply :: term(), NewState :: #state{}} | {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} | {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} | {stop, Reason :: term(), NewState :: #state{}}). handle_call({create_vertex, TempId, Record}, _From, #state{classes=Classes}=State) -> TranslatedRecord = odi_typed:typify_record(Record, Classes), lager:debug("Translated: ~p", [TranslatedRecord]), State2 = add_create_command(rid(TempId), TranslatedRecord, State), {reply, ok, State2}; handle_call({create_edge, TempId, FromId, ToId, {Class, Data}}, _From, #state{classes=Classes}=State) -> LinkedData = Data#{"out" => rid(FromId), "in" => rid(ToId)}, TranslatedRecord = odi_typed:typify_record({Class, LinkedData}, Classes), lager:debug("Translated edge: ~p", [TranslatedRecord]), Rid = rid(TempId), State2 = add_edge_ref(rid(FromId), Rid, "out_" ++ Class, State), State3 = add_edge_ref(rid(ToId), Rid, "in_" ++ Class, State2), State4 = add_create_command(Rid, TranslatedRecord,State3), {reply, ok, State4}; handle_call({update, Rid, Data}, _From, State) -> {reply, ok, update_impl(rid(Rid), Data, State)}; handle_call({delete, Rid, Version}, _From, State) -> {reply, ok, delete_impl(rid(Rid), Version, State)}; handle_call({query, Query, Limit, FetchPlan, Params}, _From, #state{con=Con}=State) -> handle_query_result(odi:query(Con, Query, Limit, FetchPlan, odi_typed:typify_map(Params)), State); handle_call({record_load , Rid, FetchPlan}, _From, State) -> {Record, State2} = record_load_impl(Rid, FetchPlan, State), {reply, untypify_results(Record), State2}; handle_call({get_cache}, _From, #state{cache=Cache}=State) -> {reply, untypify_results(maps:values(Cache)), State}; handle_call({commit, TxId}, _From, #state{con=Con, commands=Commands}=State) -> lager:debug("Committing ~p", [Commands]), case odi:tx_commit(Con, TxId, true, Commands) of {error, Messages} -> {stop, normal, {error, Messages}, State}; {Ids, _Update, _Changes} -> {stop, normal, get_id_remaps(Ids, #{}), State} end; handle_call(Request, _From, State) -> lager:error("Unknown call: ~p", [Request]), {reply, {error, "Unknown call"}, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling cast messages %% %% @end %%-------------------------------------------------------------------- -spec(handle_cast(Request :: term(), State :: #state{}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). handle_cast(Request, State) -> lager:error("Unknown cast: ~p", [Request]), {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling all non call/cast messages %% %% @spec handle_info(Info, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- -spec(handle_info(Info :: timeout() | term(), State :: #state{}) -> {noreply, NewState :: #state{}} | {noreply, NewState :: #state{}, timeout() | hibernate} | {stop, Reason :: term(), NewState :: #state{}}). handle_info(Info, State) -> lager:error("Unknown info: ~p", [Info]), {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% This function is called by a gen_server when it is about to %% terminate. It should be the opposite of Module:init/1 and do any %% necessary cleaning up. When it returns, the gen_server terminates %% with Reason. The return value is ignored. %% %% @spec terminate(Reason, State) -> void() %% @end %%-------------------------------------------------------------------- -spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()), State :: #state{}) -> term()). terminate(_Reason, _State) -> ok. %%-------------------------------------------------------------------- %% @private %% @doc %% Convert process state when code is changed %% %% @spec code_change(OldVsn, State, Extra) -> {ok, NewState} %% @end %%-------------------------------------------------------------------- -spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{}, Extra :: term()) -> {ok, NewState :: #state{}} | {error, Reason :: term()}). code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== index_classes(Schemas) -> #{"classes" := ClassList} = Schemas, IndexedClasses = odi_typed:index_records(ClassList, "name", #{}), maps:map(fun(_K, V) -> index_embedded_set_field("properties", "name", V) end, IndexedClasses). index_embedded_set_field(FieldName, IndexName, Record) -> #{FieldName := ValueList} = Record, IndexedField = odi_typed:index_records(ValueList, IndexName, #{}), Record#{FieldName => IndexedField}. rid({_ClusterId, _ClusterPosition}=Id) -> Id; rid(_TempPosition) when is_integer(_TempPosition) -> {-1, _TempPosition}. add_edge_ref(VertexId, EdgeId, PropertyName, #state{command_pos=CommandPos}=State) -> lager:debug("Trying to find ~p for updating ~p", [VertexId, PropertyName]), case CommandPos of #{VertexId := VertexCommandPos} -> add_edge_ref_to_transaction(VertexCommandPos, EdgeId, PropertyName, State); _ -> add_edge_ref_to_existing(VertexId, EdgeId, PropertyName, State) end. add_edge_ref_to_transaction(VertexCommandPos, EdgeId, PropertyName, #state{commands=Commands}=State) -> NewCommand = case lists:nth(VertexCommandPos, Commands) of {create, Rid, document, {Class, Data}} -> lager:debug("updating ~p in create ~p", [PropertyName, Rid]), NewData = add_edge_ref_to_data(Data, PropertyName, EdgeId), {create, Rid, document, {Class, NewData}}; {update, Rid, document, Version, true, {Class, Data}} -> lager:debug("updating ~p in update ~p", [PropertyName, Rid]), NewData = add_edge_ref_to_data(Data, PropertyName, EdgeId), {update, Rid, document, Version, true, {Class, NewData}} end, NewCommands = replace_nths(NewCommand, VertexCommandPos, Commands), State#state{commands = NewCommands}. add_edge_ref_to_existing(VertexId, EdgeId, PropertyName, State) -> lager:debug("fetching ~p to update ~p", [VertexId, PropertyName]), {{VertexId, document, Version, Class, Data}, State2} = record_load_impl(VertexId, "", State), NewData = add_edge_ref_to_data(Data, PropertyName, EdgeId), add_update_command(VertexId, Version, {Class, NewData}, State2). add_edge_ref_to_data(Data, PropertyName, EdgeId) -> case Data of #{PropertyName := {linkbag, Links}} -> Data#{PropertyName => {linkbag, Links ++ [EdgeId]}}; _ -> Data#{PropertyName => {linkbag, [EdgeId]}} end. add_create_command(Rid, Record, #state{commands=Commands, command_pos=CommandPos}=State) -> State#state{ commands=Commands ++ [{create, Rid, document, Record}], command_pos=CommandPos#{Rid => length(Commands) + 1} }. add_update_command(Rid, Version, Record, #state{commands=Commands, command_pos=CommandPos}=State) -> State#state{ commands=Commands ++ [{update, Rid, document, Version, true, Record}], command_pos=CommandPos#{Rid => length(Commands) + 1} }. replace_nths(Value, N, List) -> lists:sublist(List, N - 1) ++ [Value] ++ lists:nthtail(N, List). get_id_remaps([], Map) -> Map; get_id_remaps([{{-1, OldId}, NewRid} | Rest], Map) -> get_id_remaps(Rest, Map#{OldId => NewRid}); get_id_remaps([{Rid, Rid} | Rest], Map) -> get_id_remaps(Rest, Map). update_impl(Rid, Data, #state{command_pos=CommandPos}=State) -> lager:debug("Trying to find ~p for updating data", [Rid]), case CommandPos of #{Rid := VertexCommandPos} -> update_in_transaction(VertexCommandPos, Data, State); _ -> update_existing(Rid, Data, State) end. update_in_transaction(VertexCommandPos, UpdateData, #state{commands=Commands}=State) -> NewCommand = case lists:nth(VertexCommandPos, Commands) of {create, Rid, document, {Class, Data}} -> lager:debug("updating ~p in create ~p", [UpdateData, Rid]), NewData = update_data(Class, Data, UpdateData, State), {create, Rid, document, {Class, NewData}}; {update, Rid, document, Version, true, {Class, Data}} -> lager:debug("updating ~p in update ~p", [UpdateData, Rid]), NewData = update_data(Class, Data, UpdateData, State), {update, Rid, document, Version, true, {Class, NewData}} end, NewCommands = replace_nths(NewCommand, VertexCommandPos, Commands), State#state{commands = NewCommands}. update_existing(Rid, UpdateData, State) -> lager:debug("fetching ~p to update ~p", [Rid, UpdateData]), {{Rid, document, Version, Class, Data}, State2} = record_load_impl(Rid, "", State), NewData = update_data(Class, Data, UpdateData, State2), add_update_command(Rid, Version, {Class, NewData}, State2). update_data(Class, Orig, Updates, #state{classes=Classes}) -> {Class, TypifiedUpdates} = odi_typed:typify_record({Class, Updates}, Classes), maps:merge(Orig, TypifiedUpdates). delete_impl(Rid, Version, #state{commands=Commands, command_pos=CommandPos}=State) -> State#state{ commands=Commands ++ [{delete, Rid, document, Version}], command_pos=CommandPos#{Rid => length(Commands) + 1} }. record_load_impl(Rid, FetchPlan, #state{con=Con, cache=Cache}=State) -> case Cache of #{Rid := Record} -> lager:debug("Cache hit for ~p", [Rid]), {Record, State}; _ -> lager:debug("Cache miss for ~p", [Rid]), case odi:record_load(Con, Rid, FetchPlan, true) of [{true, document, Version, Class, Data} | Rest] -> FixedRecord = {Rid, document, Version, Class, Data}, {FixedRecord, cache_records([FixedRecord], cache_records(Rest, State))}; [] -> {null, State} end end. cache_records(Records, #state{cache=Cache}=State) -> State#state{cache=index_records(Records, Cache)}. index_records([], Acc) -> Acc; index_records([Cur | Rest], Acc) -> Rid = element(1, Cur), case Rid of {-1, -1} -> index_records(Rest, Acc); _ -> index_records(Rest, Acc#{Rid => Cur}) end. untypify_results(null) -> null; untypify_results({Rid, document, Version, Class, Data}) -> {Rid, document, Version, Class, odi_typed:untypify_record(Data)}; untypify_results([]) -> []; untypify_results([Cur | Rest]) -> [untypify_results(Cur) | untypify_results(Rest)]. handle_query_result({error, Messages}, State) -> {reply, {error, Messages}, State}; handle_query_result({Results, ForCache}, State) -> {reply, untypify_results(Results), cache_records(Results, cache_records(ForCache, State))}.