sql_bridge

-module(sql_bridge_mysql_otp).
-behaviour(sql_bridge_adapter).
-include("compat.hrl").

-export([start/0,
		 connect/5,
		 query/4,
		 schema_db_column/0,
		 encode/1,
		 start_transaction/1,
		 commit_transaction/1,
		 rollback_transaction/1,
		 with_transaction/2,
         wrap_field/1
		]).

%% 
-export([maybe_replace_tokens/2]).


start() ->
	application:start(poolboy),
	ok.

connect(DB, User, Pass, Host, Port) when is_atom(DB) ->
	WorkerArgs = [
		{database, atom_to_list(DB)},
		{host, Host},
		{user, User},
		{password, Pass},
		{port, Port}
	],
	sql_bridge_utils:start_poolboy_pool(DB, WorkerArgs, mysql),
	ok.

wrap_field(V) ->
    "`" ++ sql_bridge_utils:to_string(V) ++ "`".

start_transaction(DB) ->
    case sql_bridge_utils:trans_depth(DB) of
        0 ->
            query(none, DB, "BEGIN", []);
        _ ->
            ok
    end,
    sql_bridge_utils:trans_deeper(DB),
    sql_bridge_utils:checkout_pool(DB),
    ok.

rollback_transaction(DB) ->
    sql_bridge_utils:trans_shallower(DB),
	query(none, DB, "ROLLBACK", []),
	sql_bridge_utils:checkin_pool(DB),
	ok.

commit_transaction(DB) ->
    case sql_bridge_utils:trans_depth(DB) of
        1 ->
            Res = query(none, DB, "COMMIT", []),
            sql_bridge_utils:clear_trans_depth(DB),
            Res;
        X when X > 1 ->
            sql_bridge_utils:trans_shallower(DB),
            ok %% mysql does not supported nested transactions, so we'll only commit on the last depth
    end,
	sql_bridge_utils:checkin_pool(DB),
	ok.

with_transaction(DB, Fun) ->
	start_transaction(DB),
	try Fun() of
		Res ->
			commit_transaction(DB),
			Res
	catch
		Error:Class:ST ->
            ErrMsg = [{Error, Class}, ST],
            error_logger:info_msg("Errored Query: ~p",[ErrMsg]),
			rollback_transaction(DB),
			{error, ErrMsg}
	end.

query(Type, DB, Q, ParamList) ->
	try query_catched(Type, DB, Q, ParamList)
	catch
		exit:{noproc, _} ->
            {error, no_pool};
        exit:{{{badmatch,{error,closed}}, _}, _} ->
            {error, disconnected};
        exit:{{{case_clause,{error,closed}}, _}, _} ->
            {error, disconnected};
        E:T ->
            error_logger:error_msg("Unhandled Error: ~p:~p", [E,T]),
            throw(unhandled_error)
	end.

query_catched(Type, DB, Q, ParamList) ->
	{Q2, ParamList2} = maybe_replace_tokens(Q, ParamList),
	ToRun = fun(Worker) ->
		Res = mysql_query(Worker, Q2, ParamList2),
        case Res of
            {error, Reason} ->
                error_logger:warning_msg("Error in Query.~nError: ~p~nQuery: ~s",[Reason, Q]);
            _ ->
                ok
        end,
		case Type of
			insert -> mysql:insert_id(Worker);
			update -> mysql:affected_rows(Worker);
			_ -> Res
		end
	end,
	case sql_bridge_utils:with_poolboy_pool(DB, ToRun) of
		{error, Reason} -> {error, Reason};
		Result ->
            {ok, format_result(Type, Result)}
	end.
	
mysql_query(Worker, Q, []) ->
	mysql:query(Worker, Q);
mysql_query(Worker, Q, ParamList) ->
    ParamList2 = pre_encode(ParamList),
	mysql:query(Worker, Q, ParamList2).

pre_encode(List) ->
    lists:map(fun(true) -> 1;
                 (false) -> 0;
                 (undefined) -> null;
                 (X) -> X
              end, List).

maybe_replace_tokens(Q, ParamList) ->
	case sql_bridge_utils:replacement_token() of
		mysql -> {Q, ParamList};
		postgres -> sql_bridge_utils:token_postgres_to_mysql(Q, ParamList)
	end.

format_result(none, _) ->
	ok;
format_result(UID, Count) when UID=:=update;
									 UID=:=insert;
									 UID=:=delete ->
	Count;
format_result(_, ok) ->
	ok;
format_result(tuple, {ok, _Columns, Rows}) ->
	format_tuples(Rows);
format_result(list, {ok, _Columns, Rows}) ->
	format_lists(Rows);
format_result(proplist, {ok, Columns, Rows}) ->
	format_proplists(Columns, Rows);
format_result(dict, {ok, Columns, Rows}) ->
	format_dicts(Columns, Rows);
format_result(map, {ok, Columns, Rows}) ->
	format_maps(Columns, Rows).

format_tuples(Rows) ->
	case sql_bridge_utils:stringify_binaries() of
		true ->
			[list_to_tuple(format_list(Row)) || Row <- Rows];
		false ->
			[list_to_tuple(Row) || Row <- Rows]
	end.

format_lists(Rows) ->
	case sql_bridge_utils:stringify_binaries() of
		true ->
			[format_list(Row) || Row <- Rows];
		false ->
			Rows
	end.

format_list(Row) when is_list(Row) ->
	[normalize_value(V) || V <- Row].

normalize_value(V) when is_tuple(V) ->
    sql_bridge_utils:format_datetime(V);
normalize_value(V) ->
	sql_bridge_stringify:maybe_string(V).

format_proplists(Columns, Rows) ->
	ColNames = extract_colnames(Columns),
	[make_proplist(ColNames, Row) || Row <- Rows].

format_dicts(Columns, Rows) ->
	ColNames = extract_colnames(Columns),
	[make_dict(ColNames, Row) || Row <- Rows].

make_dict(Cols, Row) when is_list(Row) ->
	make_dict(Cols, Row, dict:new()).

make_dict([], [], Dict) ->
	Dict;
make_dict([Col|Cols], [Val|Vals], Dict) ->
	Val2 = normalize_value(Val),
	NewDict = dict:store(Col, Val2, Dict),
	make_dict(Cols, Vals, NewDict).
	
extract_colnames(Columns) ->
	[list_to_atom(binary_to_list(CN)) || CN <- Columns].

make_proplist(Columns, Row) when is_tuple(Row) ->
	make_proplist(Columns, tuple_to_list(Row));
make_proplist([Col|Cols], [Val|Vals]) ->
	Val2 = normalize_value(Val),
	[{Col, Val2} | make_proplist(Cols, Vals)];
make_proplist([], []) ->
	[].

-ifdef(has_maps).
format_maps(Columns, Rows) ->
	ColNames = extract_colnames(Columns),
	[make_map(ColNames, Row) || Row <- Rows].

make_map(Cols, Row) ->
	make_map(Cols, Row, maps:new()).

make_map([], [], Map) ->
	Map;
make_map([Col|Cols],[Val|Vals], Map) ->
	Val2 = normalize_value(Val),
	NewMap = maps:put(Col, Val2, Map),
	make_map(Cols, Vals, NewMap).

-else.
format_maps(_,_) ->
	throw(maps_not_supported).
-endif.

schema_db_column() ->
	"table_schema".

encode(Val) ->
	sql_bridge_utils:with_poolboy_pool(sql_bridge:db(), fun(Worker) ->
		mysql:encode(Worker, Val)
	end).