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,
         primary_key/2,
         is_auto_increment/3,
         %primary_key_with_auto_increment_and_type/2,
         %auto_increment/2,
         field_type/3,
         maybe_replace_tokens/2
        ]).

-export([parse_field_type/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},
        %% This is an option for mysql-otp. It is probably worth adding it as a tweakable option for sql_bridge
        {decode_decimal, number}
    ],
    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 ->
sql_bridge_utils:checkout_pool(DB),
            query(none, DB, "BEGIN", []);
        _ ->
            ok
    end,
    sql_bridge_utils:trans_deeper(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],
            logger:warning("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 ->
            logger:error("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} ->
                logger:warning("Error in Query.~nError: ~p~nQuery: ~s",[Reason, Q]);
            _ ->
                ok
        end,
        %% NOTE: For now, not going to match on the {error,_} tuple just to see
        %% the impact it might have on my libraries before doing this.
        case {Res, Type} of
            %{{error, _}, _} -> Res;
            {_, 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([], []) ->
    [].

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).

schema_db_column() ->
    "table_schema".

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

-spec field_type(DB :: sql_bridge:db(),
                     Table :: sql_bridge:table(),
                     Field :: sql_bridge:field()) -> undefined | sql_bridge:field_type().

field_type(DB, Table, Field) ->
    [T1, T2, T3] = sql_bridge_utils:create_placeholders(3),
    SQL = [<<"select column_type from information_schema.columns ">>,
           <<" where table_schema=">>, T1,
           <<" and table_name=">>, T2,
           <<" and column_name=">>, T3],
    Res = sql_bridge:fffr(SQL, [DB, Table, Field]),
    %io:format("MySQL - Field Type (~p, ~p, ~p): ~p~n", [DB, Table, Field, Res]),
    case Res of
        not_found ->
            undefined;
        Type ->
            parse_field_type(Type)
    end.

parse_field_type(T) when T=="text";
                         T=="mediumtext";
                         T=="longtext";
                         T=="blob";
                         T=="mediumblob";
                         T=="longblob" ->
    {text, 50};
parse_field_type("uuid") ->
    {uuid, undefined};
parse_field_type(FieldType) ->
    RETypes = "(tinyint|smallint|mediumint|int|bigint|varchar|char)",
    RELength = "(?:\\(([0-9]+)\\))?",
    REFollower = "(signed|unsigned)?",
    RE = "^" ++ RETypes ++ "\\s*" ++ RELength ++ "\\s*" ++ REFollower ++ "$",
    case re:run(FieldType, RE, [{capture, all_but_first, binary}]) of
        {match, [Type, Length | _]} when Type == <<"varchar">>;
                                         Type == <<"char">> ->
            {text, binary_to_integer(Length)};
        {match, [Type, _Size]} ->
            Range = int_range({Type, <<"signed">>}),
            {integer, Range};
        {match, [Type, _Size, Unsigned]} ->
            Range = int_range({Type, Unsigned}),
            {integer, Range};
        nomatch ->
            undefined
    end.

int_range(TypeUnsigned) ->
    AType = case TypeUnsigned of
        {<<"tinyint">>, <<"unsigned">>} -> uint1;
        {<<"smallint">>, <<"unsigned">>} -> uint2;
        {<<"mediumint">>, <<"unsigned">>} -> uint3;
        {<<"int">>, <<"unsigned">>} -> uint4;
        {<<"bigint">>, <<"unsigned">>} -> uint8;

        {<<"tinyint">>, _} -> int1;
        {<<"smallint">>, _} -> int2;
        {<<"mediumint">>, _} -> int3;
        {<<"int">>, _} -> int4;
        {<<"bigint">>, _} -> int8
    end,
    sql_bridge_utils:int_ranges(AType).


primary_key(DB, Table) ->
    [T1, T2] = sql_bridge_utils:create_placeholders(2),
    SQL = [<<"select column_name
            from information_schema.columns
            where table_schema=">>, T1,
                <<" and table_name=">>, T2,
                <<" and column_key='PRI'">>],
    case sql_bridge:fffr(SQL, [DB, Table]) of
        not_found ->
            undefined;
        Field ->
            list_to_atom(Field)
    end.

is_auto_increment(DB, Table, Field) ->
    [T1, T2, T3] = sql_bridge_utils:create_placeholders(3),
    SQL = "select column_name
            from information_schema.columns
            where table_schema=" ++ T1 ++ "
                and table_name=" ++ T2 ++ "
                and column_name=" ++ T3 ++ "
                and extra like '%auto_increment%'",
    case sql_bridge:fffr(SQL, [DB, Table, Field]) of
        not_found ->
            false;
        _ ->
            true
    end.