%%% @hidden %%% @doc MySql store implementation. %%% %%% Copyright 2012 Inaka <hello@inaka.net> %%% %%% 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 %%% %%% http://www.apache.org/licenses/LICENSE-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. %%% @end %%% @copyright Inaka %%% -module(sumo_store_mysql). -author("Marcelo Gornstein "). -github("https://github.com/inaka"). -license("Apache License 2.0"). -include_lib("emysql/include/emysql.hrl"). -behavior(sumo_store). %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% Exports. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% Public API. -export([init/1]). -export([create_schema/2]). -export([persist/2]). -export([delete_by/3, delete_all/2]). -export([prepare/3, execute/2, execute/3]). -export([just_execute/2, just_execute/3, get_docs/3, get_docs/4]). -export([find_all/2, find_all/5, find_by/3, find_by/5, find_by/6]). %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% Types. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -record(state, {pool :: atom() | pid()}). -type state() :: #state{}. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% External API. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% -spec init(term()) -> {ok, term()}. init(Options) -> % The storage backend key in the options specifies the name of the process % which creates and initializes the storage backend. Backend = proplists:get_value(storage_backend, Options), Pool = sumo_backend_mysql:get_pool(Backend), {ok, #state{pool=Pool}}. -spec persist(sumo_internal:doc(), state()) -> sumo_store:result(sumo_internal:doc(), state()). persist(Doc, State) -> % Set the real id, replacing undefined by 0 so it is autogenerated DocName = sumo_internal:doc_name(Doc), IdField = sumo_internal:id_field_name(DocName), NewId = case sumo_internal:get_field(IdField, Doc) of undefined -> 0; Id -> Id end, NewDoc = sleep(sumo_internal:set_field(IdField, NewId, Doc)), % Needed because the queries will carry different number of arguments. Statement = case NewId of 0 -> insert; NewId -> update end, Fields = sumo_internal:doc_fields(NewDoc), NPFields = maps:remove(IdField, Fields), % Non-primary fields. FieldNames = maps:keys(Fields), NPFieldNames = maps:keys(NPFields), Fun = fun() -> [ColumnDqls, ColumnSqls] = lists:foldl( fun(Name, [Dqls, Sqls]) -> Dql = [escape(Name)], Sql = "?", [[Dql | Dqls], [Sql | Sqls]] end, [[], []], FieldNames ), NPColumnDqls = lists:foldl( fun(Name, Dqls) -> Dql = [escape(Name)], [Dql | Dqls] end, [], NPFieldNames ), TableName = escape(DocName), ColumnsText = string:join(ColumnDqls, ","), InsertValueSlots = string:join(ColumnSqls, ","), OnDuplicateColumns = [[ColumnName, "=?"] || ColumnName <- NPColumnDqls], OnDuplicateSlots = string:join(OnDuplicateColumns, ","), [ "INSERT INTO ", TableName , " (", ColumnsText, ")" , " VALUES (", InsertValueSlots, ")" , " ON DUPLICATE KEY UPDATE " , OnDuplicateSlots ] end, StatementName = prepare(DocName, Statement, Fun), ColumnValues = lists:reverse([maps:get(K, Fields) || K <- maps:keys(Fields)]), NPColumnValues = lists:reverse([maps:get(K, Fields) || K <- maps:keys(NPFields)]), StatementValues = lists:append(ColumnValues, NPColumnValues), case execute(StatementName, StatementValues, State) of #ok_packet{insert_id = InsertId} -> % XXX TODO darle una vuelta mas de rosca % para el manejo general de cuando te devuelve el primary key % considerar el caso cuando la primary key (campo id) no es integer % tenes que poner unique index en lugar de primary key % la mejor solucion es que el PK siempre sea un integer, como hace mongo LastId = case InsertId of 0 -> NewId; I -> I end, IdField = sumo_internal:id_field_name(DocName), {ok, sumo_internal:set_field(IdField, LastId, Doc), State}; Error -> evaluate_execute_result(Error, State) end. -spec delete_by(sumo:schema_name(), sumo:conditions(), state()) -> sumo_store:result(sumo_store:affected_rows(), state()). delete_by(DocName, Conditions, State) -> {Values, CleanConditions} = sumo_sql_builder:values_conditions(Conditions), Clauses = sumo_sql_builder:where_clause(CleanConditions), HashClause = hash(Clauses), PreStatementName = list_to_atom("delete_by_" ++ HashClause), StatementFun = fun() -> [ "DELETE FROM ", escape(DocName), " WHERE ", lists:flatten(Clauses) ] end, StatementName = prepare(DocName, PreStatementName, StatementFun), Values = [V || {_K, V} <- Conditions], case execute(StatementName, Values, State) of #ok_packet{affected_rows = NumRows} -> {ok, NumRows, State}; Error -> evaluate_execute_result(Error, State) end. -spec delete_all(sumo:schema_name(), state()) -> sumo_store:result(sumo_store:affected_rows(), state()). delete_all(DocName, State) -> StatementName = prepare(DocName, delete_all, fun() -> ["DELETE FROM ", escape(DocName)] end), case execute(StatementName, State) of #ok_packet{affected_rows = NumRows} -> {ok, NumRows, State}; Error -> evaluate_execute_result(Error, State) end. -spec find_all(sumo:schema_name(), state()) -> sumo_store:result([sumo_internal:doc()], state()). find_all(DocName, State) -> find_all(DocName, [], 0, 0, State). -spec find_all(sumo:schema_name(), term(), non_neg_integer(), non_neg_integer(), state()) -> sumo_store:result([sumo_internal:doc()], state()). find_all(DocName, SortFields, Limit, Offset, State) -> find_by(DocName, [], SortFields, Limit, Offset, State). -spec find_by(sumo:schema_name(), sumo:conditions(), state()) -> sumo_store:result([sumo_internal:doc()], state()). find_by(DocName, Conditions, State) -> find_by(DocName, Conditions, [], 0, 0, State). -spec find_by(sumo:schema_name(), sumo:conditions(), non_neg_integer(), non_neg_integer(), state()) -> sumo_store:result([sumo_internal:doc()], state()). find_by(DocName, Conditions, Limit, Offset, State) -> find_by(DocName, Conditions, [], Limit, Offset, State). %% XXX We should have a DSL here, to allow querying in a known language %% to be translated by each driver into its own. -spec find_by(sumo:schema_name(), sumo:conditions(), term(), non_neg_integer(), non_neg_integer(), state()) -> sumo_store:result([sumo_internal:doc()], state()). find_by(DocName, Conditions, SortFields, Limit, Offset, State) -> {Values, CleanConditions} = sumo_sql_builder:values_conditions(Conditions), Clauses = sumo_sql_builder:where_clause(CleanConditions), PreStatementName0 = hash(Clauses), PreStatementName1 = case Limit of 0 -> PreStatementName0; Limit -> PreStatementName0 ++ "_limit" end, {PreStatementName2, OrderByClause} = case SortFields of [] -> {PreStatementName1, []}; _ -> OrderByClause0 = sumo_sql_builder:order_by_clause(SortFields), { PreStatementName1 ++ "_" ++ hash(OrderByClause0), OrderByClause0 } end, WhereClause = case Conditions of [] -> ""; _ -> [" WHERE ", lists:flatten(Clauses)] end, PreName = list_to_atom("find_by_" ++ PreStatementName2), Fun = fun() -> % Select * is not good.. Sql1 = [ "SELECT * FROM ", escape(DocName), WhereClause, OrderByClause ], Sql2 = case Limit of 0 -> Sql1; _ -> [Sql1|[" LIMIT ?, ?"]] end, Sql2 end, StatementName = prepare(DocName, PreName, Fun), ExecArgs = case Limit of 0 -> Values; Limit -> lists:flatten([Values | [Offset, Limit]]) end, case execute(StatementName, ExecArgs, State) of #result_packet{} = Result -> {ok, build_docs(DocName, Result), State}; Error -> evaluate_execute_result(Error, State) end. %% XXX: Refactor: %% Requires {length, X} to be the first field attribute in order to form the %% correct query. :P %% If no indexes are defined, will put an extra comma :P %% Maybe it would be better to just use ALTER statements instead of trying to %% create the schema on the 1st pass. Also, ALTER statements might be better %% for when we have migrations. -spec create_schema(sumo:schema(), state()) -> sumo_store:result(state()). create_schema(Schema, State) -> Name = sumo_internal:schema_name(Schema), Fields = sumo_internal:schema_fields(Schema), FieldsDql = lists:map(fun create_column/1, Fields), Indexes = lists:filter( fun(T) -> length(T) > 0 end, lists:map(fun create_index/1, Fields) ), Dql = [ "CREATE TABLE IF NOT EXISTS ", escape(Name), " (", string:join(FieldsDql, ", "), ", ", string:join(Indexes, ", "), ") ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8" ], case execute(Dql, State) of #ok_packet{} -> {ok, State}; Error -> evaluate_execute_result(Error, State) end. create_column(Field) -> create_column( sumo_internal:field_name(Field), sumo_internal:field_type(Field), sumo_internal:field_attrs(Field)). create_column(Name, integer, Attrs) -> [escape(Name), " INT(11) ", create_column_options(Attrs)]; create_column(Name, float, Attrs) -> [escape(Name), " FLOAT ", create_column_options(Attrs)]; create_column(Name, binary, Attrs) -> [escape(Name), " BLOB ", create_column_options(Attrs)]; create_column(Name, string, Attrs) -> [escape(Name), " VARCHAR ", create_column_options(Attrs)]; create_column(Name, date, Attrs) -> [escape(Name), " DATE ", create_column_options(Attrs)]; create_column(Name, datetime, Attrs) -> [escape(Name), " DATETIME ", create_column_options(Attrs)]; create_column(Name, boolean, Attrs) -> [escape(Name), " BOOLEAN ", create_column_options(Attrs)]; create_column(Name, custom, Attrs) -> case lists:keyfind(type, 1, Attrs) of {type, text} -> [escape(Name), " TEXT ", create_column_options(Attrs)]; _ -> create_column(Name, binary, Attrs) end. create_column_options(Attrs) -> lists:filter(fun(T) -> is_list(T) end, lists:map( fun(Option) -> create_column_option(Option) end, Attrs )). create_column_option(auto_increment) -> ["AUTO_INCREMENT "]; create_column_option(not_null) -> [" NOT NULL "]; create_column_option({length, X}) -> ["(", integer_to_list(X), ") "]; create_column_option(_Option) -> none. create_index(Field) -> Name = sumo_internal:field_name(Field), Attrs = sumo_internal:field_attrs(Field), lists:filter(fun(T) -> is_list(T) end, lists:map( fun(Attr) -> create_index(Name, Attr) end, Attrs )). create_index(Name, id) -> ["PRIMARY KEY(", escape(Name), ")"]; create_index(Name, unique) -> List = atom_to_list(Name), ["UNIQUE KEY ", escape(List), " (", escape(List), ")"]; create_index(Name, index) -> List = atom_to_list(Name), ["KEY ", escape(List), " (", escape(List), ")"]; create_index(_, _) -> none. -spec prepare(sumo:schema_name(), atom(), fun()) -> atom(). prepare(DocName, PreName, Fun) when is_atom(PreName), is_function(Fun) -> Name = statement_name(DocName, PreName), case emysql_statements:fetch(Name) of undefined -> Query = iolist_to_binary(Fun()), log("Preparing query: ~p: ~p", [Name, Query]), ok = emysql:prepare(Name, Query); Q -> log("Using already prepared query: ~p: ~p", [Name, Q]) end, Name. %% @doc Call prepare/3 first, to get a well formed statement name. -spec just_execute(atom() | list(), state()) -> {ok, {raw, ok}, state()} | {error, binary(), state()}. just_execute(Query, State) -> case execute(Query, State) of #ok_packet{} -> {ok, {raw, ok}, State}; Error -> evaluate_execute_result(Error, State) end. -spec just_execute(atom(), list(), state()) -> {ok, {raw, ok}, state()} | {error, binary(), state()}. just_execute(Name, Args, State) -> case execute(Name, Args, State) of #ok_packet{} -> {ok, {raw, ok}, State}; Error -> evaluate_execute_result(Error, State) end. %% @doc Call prepare/3 first, to get a well formed statement name. -spec get_docs(atom(), atom() | list(), state()) -> {ok, {docs, [sumo_internal:doc()]}, state()} | {error, binary(), state()}. get_docs(DocName, Query, State) -> case execute(Query, State) of #result_packet{} = Result -> {ok, {docs, build_docs(DocName, Result)}, State}; Error -> evaluate_execute_result(Error, State) end. -spec get_docs(atom(), atom(), list(), state()) -> {ok, {docs, [sumo_internal:doc()]}, state()} | {error, binary(), state()}. get_docs(DocName, Name, Args, State) -> case execute(Name, Args, State) of #result_packet{} = Result -> {ok, {docs, build_docs(DocName, Result)}, State}; Error -> evaluate_execute_result(Error, State) end. build_docs(DocName, #result_packet{rows = Rows, field_list = Fields}) -> FieldNames = [binary_to_atom(Field#field.name, utf8) || Field <- Fields], [wakeup(build_doc(sumo_internal:new_doc(DocName), FieldNames, Row)) || Row <- Rows]. build_doc(Doc, [], []) -> Doc; build_doc(Doc, [FieldName|FieldNames], [{date, Value}|Values]) -> build_doc(sumo_internal:set_field(FieldName, Value, Doc), FieldNames, Values); build_doc(Doc, [FieldName|FieldNames], [{datetime, Value}|Values]) -> build_doc(sumo_internal:set_field(FieldName, Value, Doc), FieldNames, Values); build_doc(Doc, [FieldName|FieldNames], [Value|Values]) -> build_doc(sumo_internal:set_field(FieldName, Value, Doc), FieldNames, Values). %% @doc Call prepare/3 first, to get a well formed statement name. -spec execute(atom(), list(), state()) -> term(). execute(Name, Args, #state{pool=Pool}) when is_atom(Name), is_list(Args) -> {Time, Value} = timer:tc( emysql, execute, [Pool, Name, Args] ), log("Executed Query: ~s -> ~p (~pms)", [Name, Args, Time/1000]), Value. -spec execute(atom() | list(), state()) -> term(). execute(Name, State) when is_atom(Name) -> execute(Name, [], State); execute(PreQuery, #state{pool=Pool}) when is_list(PreQuery)-> Query = iolist_to_binary(PreQuery), {Time, Value} = timer:tc( emysql, execute, [Pool, Query] ), log("Executed Query: ~s (~pms)", [Query, Time/1000]), Value. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% Private API. %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% @doc We can extend this to wrap around emysql records, so they don't end up %% leaking details in all the store. evaluate_execute_result(#error_packet{status = Status, msg = Msg}, State) -> {error, <>, State}. escape(Name) when is_atom(Name) -> ["`", atom_to_list(Name), "`"]; escape(String) -> ["`", String, "`"]. statement_name(DocName, StatementName) -> list_to_atom(string:join( [atom_to_list(DocName), atom_to_list(StatementName), "stmt"], "_" )). log(Msg, Args) -> case application:get_env(sumo_db, log_queries) of {ok, true} -> _ = lager:debug(Msg, Args), ok; _ -> ok end. -spec hash(iodata()) -> string(). hash(Clause) -> Bin = crypto:hash(md5, Clause), List = binary_to_list(Bin), Fun = fun(Num) -> string:right(integer_to_list(Num, 16), 2, $0) end, lists:flatmap(Fun, List). %% @private sleep(Doc) -> sumo_utils:doc_transform(fun sleep_fun/4, Doc). sleep_fun(_, _, undefined, _) -> null; sleep_fun(boolean, _, false, _) -> 0; sleep_fun(boolean, _, true, _) -> 1; sleep_fun(custom, _, FieldValue, Attrs) -> case lists:keyfind(type, 1, Attrs) of {type, text} -> FieldValue; _ -> term_to_binary(FieldValue) end; sleep_fun(_, _, FieldValue, _) -> FieldValue. %% @private wakeup(Doc) -> sumo_utils:doc_transform(fun wakeup_fun/4, Doc). %% Matches `text' type fields that were saved with `undefined' value and %% avoids being processed by the next clause that will return it as a %% binary (`<<"undefined">>') instead of atom as expected. wakeup_fun(_, _, undefined, _) -> undefined; wakeup_fun(float, _, 0, _) -> 0.0; wakeup_fun(boolean, _, 0, _) -> false; wakeup_fun(boolean, _, 1, _) -> true; wakeup_fun(string, _, FieldValue, _) -> sumo_utils:to_bin(FieldValue); wakeup_fun(custom, FieldName, FieldValue, Attrs) -> case lists:keyfind(type, 1, Attrs) of {type, text} -> wakeup_fun(string, FieldName, FieldValue, Attrs); _ -> binary_to_term(FieldValue) end; wakeup_fun(_, _, FieldValue, _) -> FieldValue.