dinerl

-module(dinerl).

-define(DINERL_DATA, dinerl_data).
-define(ARGS_KEY, args).
-define(NONE, <<"NONE">>).
-define(ALL_OLD, <<"ALL_OLD">>).
-define(UPDATED_OLD, <<"UPDATED_OLD">>).
-define(ALL_NEW, <<"ALL_NEW">>).
-define(UPDATED_NEW, <<"UPDATED_NEW">>).

-include("dinerl_types.hrl").

-export([setup/3, setup/1, setup/0, api/1, api/2, api/3, api/4]).
-export([create_table/4, create_table/5, delete_table/1, delete_table/2]).
-export([describe_table/1, describe_table/2, update_table/3, update_table/4]).
-export([list_tables/0, list_tables/1, list_tables/2, put_item/3, put_item/4]).
-export([delete_item/3, delete_item/4]).
-export([get_item/3, get_item/4, get_item/5]).
-export([get_items/1, get_items/2, get_items/3, get_items/4, get_items/5]).
-export([update_item/3, update_item/4]).
-export([update_item_with_expression/3]).
-export([update_item_with_expression/4]).
-export([update_item_with_expression/5]).
-export([update_item_with_expression/6]).
-export([update_item_with_expression/7]).
-export([query_item/3, query_item/4, query_item/5]).
-export([query/2, query/3, query/4]).
-export([update_data/1]).
-export([batch_write_item/3]).

-spec setup(access_key_id(), secret_access_key(), zone()) -> {ok, clientarguments()}.
setup(AccessKeyId, SecretAccessKey, Zone) ->
    application:set_env(erliam, aws_access_key, AccessKeyId),
    application:set_env(erliam, aws_secret_key, SecretAccessKey),
    setup_(Zone).

-spec setup(zone()) -> {ok, clientarguments()}.
setup(Zone) ->
    setup_(Zone).

-spec setup() -> {ok, clientarguments()}.
setup() ->
    {ok, Zone} = imds:zone(),
    setup_(Zone).

-spec setup_(string()) -> {ok, clientarguments()}.
setup_(Zone) ->
    ets:new(?DINERL_DATA, [named_table, public]),
    R = update_data(Zone),
    timer:apply_interval(1000, ?MODULE, update_data, [Zone]),
    R.

-spec api(method()) -> result().
api(Name) ->
    api(Name, {struct, []}).

-spec api(method(), any()) -> result().
api(Name, Body) ->
    api(Name, Body, undefined).

-spec api(method(), any(), undefined | integer()) -> result().
api(Name, Body, Timeout) ->
    api(Name, Body, Timeout, undefined).

-spec api(method(), any(), undefined | integer(), undefined | zone()) -> result().
api(Name, Body, Timeout, Region) ->
    try ets:lookup_element(?DINERL_DATA, ?ARGS_KEY, 2) of
        {Credentials, Zone, Date} ->
            TargetRegion =
                case Region of
                    undefined ->
                        Zone;
                    _ ->
                        Region
                end,
            dinerl_client:api(Credentials, TargetRegion, Date, Name, Body, Timeout)
    catch
        _:{badarg, _} ->
            {error, missing_credentials, ""}
    end.

-spec create_table(string() | binary(), keyschema(), integer(), integer()) -> jsonf().
create_table(Name, Key, ReadsPerSecond, WritesPerSecond) ->
    create_table(Name, Key, ReadsPerSecond, WritesPerSecond, undefined).

-spec create_table(string() | binary(),
                   keyschema(),
                   integer(),
                   integer(),
                   undefined | integer()) ->
                      jsonf().
create_table(Name, Key, ReadsPerSecond, WritesPerSecond, Timeout) ->
    api(create_table,
        [{<<"TableName">>, Name},
         {<<"KeySchema">>, Key},
         {<<"ProvisionedThroughput">>,
          [{<<"ReadsPerSecond">>, ReadsPerSecond}, {<<"WritesPerSecond">>, WritesPerSecond}]}],
        Timeout).

delete_table(Name) ->
    describe_table(Name, undefined).

delete_table(Name, Timeout) ->
    api(delete_table, [{<<"TableName">>, Name}], Timeout).

describe_table(Name) ->
    describe_table(Name, undefined).

describe_table(Name, Timeout) ->
    api(describe_table, [{<<"TableName">>, Name}], Timeout).

update_table(Name, ReadsPerSecond, WritesPerSecond) ->
    update_table(Name, ReadsPerSecond, WritesPerSecond, undefined).

update_table(Name, ReadsPerSecond, WritesPerSecond, Timeout) ->
    api(update_table,
        [{<<"TableName">>, Name},
         {<<"ProvisionedThroughput">>,
          [{<<"ReadsPerSecond">>, ReadsPerSecond}, {<<"WritesPerSecond">>, WritesPerSecond}]}],
        Timeout).

list_tables() ->
    list_tables([]).

list_tables(List) ->
    list_tables(List, undefined).

list_tables(List, Timeout) ->
    list_tables(List, [], Timeout).

list_tables([], [], Timeout) ->
    list_tables([], {}, Timeout);
list_tables([], Body, Timeout) ->
    api(list_tables, Body, Timeout);
list_tables([{start_name, Name} | Rest], Acc, Timeout) ->
    list_tables(Rest, [{<<"ExclusiveStartTableName">>, Name} | Acc], Timeout);
list_tables([{limit, N} | Rest], Acc, Timeout) ->
    list_tables(Rest, [{<<"Limit">>, N} | Acc], Timeout).

put_item(Table, Attributes, Options) ->
    put_item(Table, Attributes, Options, undefined).

put_item(Table, Attributes, Options, Timeout) ->
    put_item(Table, Attributes, Options, [], Timeout).

put_item(Table, Attributes, [], PartialBody, Timeout) ->
    api(put_item,
        [{<<"TableName">>, Table}, {<<"Item">>, Attributes} | PartialBody],
        Timeout);
put_item(T, A, [{return, all_old} | Rest], Acc, Timeout) ->
    put_item(T, A, Rest, [{<<"ReturnValues">>, ?ALL_OLD} | Acc], Timeout);
put_item(T, A, [{return, none} | Rest], Acc, Timeout) ->
    put_item(T, A, Rest, [{<<"ReturnValues">>, ?NONE} | Acc], Timeout);
put_item(T, A, [{expected, V} | Rest], Acc, Timeout) ->
    put_item(T, A, Rest, [{<<"Expected">>, attr_updates(V, [])} | Acc], Timeout).

delete_item(Table, Key, Options) ->
    delete_item(Table, Key, Options, undefined).

delete_item(Table, Key, Options, Timeout) ->
    delete_item(Table, Key, Options, [], Timeout).

delete_item(Table, Key, [], PartialBody, Timeout) ->
    api(delete_item, [{<<"TableName">>, Table}, {<<"Key">>, Key} | PartialBody], Timeout);
delete_item(T, K, [{return, all_old} | Rest], Acc, Timeout) ->
    delete_item(T, K, Rest, [{<<"ReturnValues">>, ?ALL_OLD} | Acc], Timeout);
delete_item(T, K, [{return, none} | Rest], Acc, Timeout) ->
    delete_item(T, K, Rest, [{<<"ReturnValues">>, ?NONE} | Acc], Timeout);
delete_item(T, K, [{expected, V} | Rest], Acc, Timeout) ->
    delete_item(T, K, Rest, [{<<"Expected">>, attr_updates(V, [])} | Acc], Timeout).

get_item(Table, Key, Options) ->
    get_item(Table, Key, Options, [], undefined, undefined).

get_item(Table, Key, Options, Timeout) ->
    get_item(Table, Key, Options, [], Timeout, undefined).

get_item(Table, Key, Options, Timeout, Region) ->
    get_item(Table, Key, Options, [], Timeout, Region).

get_item(T, K, [], Acc, Timeout, Region) ->
    api(get_item, [{<<"TableName">>, T}, {<<"Key">>, K} | Acc], Timeout, Region);
get_item(T, K, [{consistent, V} | Rest], Acc, Timeout, Region) ->
    get_item(T, K, Rest, [{<<"ConsistentRead">>, V} | Acc], Timeout, Region);
get_item(T, K, [{attrs, V} | Rest], Acc, Timeout, Region) ->
    get_item(T, K, Rest, [{<<"AttributesToGet">>, V} | Acc], Timeout, Region).

get_items(Table, Keys, Options) ->
    do_get_items([{Table, Keys, Options}], [], undefined, undefined).

get_items(Table, Keys, Options, Timeout) ->
    do_get_items([{Table, Keys, Options}], [], Timeout, undefined).

get_items(Table, Keys, Options, Timeout, Region) ->
    do_get_items([{Table, Keys, Options}], [], Timeout, Region).

get_items(MultiTableQuery) ->
    do_get_items(MultiTableQuery, [], undefined, undefined).

get_items(MultiTableQuery, Timeout) ->
    do_get_items(MultiTableQuery, [], Timeout, undefined).

do_get_items([], Acc, Timeout, Region) ->
    api(batch_get_item, [{<<"RequestItems">>, Acc}], Timeout, Region);
do_get_items([{Table, Keys, Options} | Rest], Acc, Timeout, Region) ->
    Attrs = proplists:get_value(attrs, Options, []),
    do_get_items(Rest, [get_body_request(Table, Keys, Attrs) | Acc], Timeout, Region).

get_body_request(Table, Keys, []) ->
    {Table, [{<<"Keys">>, Keys}]};
get_body_request(Table, Keys, Attrs) ->
    {Table, [{<<"Keys">>, Keys}, {<<"AttributesToGet">>, Attrs}]}.

update_item_with_expression(TableName, Key, UpdateExpression) ->
    update_item_with_expression(TableName,
                                Key,
                                UpdateExpression,
                                undefined,
                                <<"NONE">>,
                                [],
                                undefined).

update_item_with_expression(TableName,
                            Key,
                            UpdateExpression,
                            ExpressionAttributeValues) ->
    update_item_with_expression(TableName,
                                Key,
                                UpdateExpression,
                                ExpressionAttributeValues,
                                <<"NONE">>,
                                [],
                                undefined).

update_item_with_expression(TableName,
                            Key,
                            UpdateExpression,
                            ExpressionAttributeValues,
                            ReturnValues) ->
    update_item_with_expression(TableName,
                                Key,
                                UpdateExpression,
                                ExpressionAttributeValues,
                                ReturnValues,
                                [],
                                undefined).

update_item_with_expression(TableName,
                            Key,
                            UpdateExpression,
                            ExpressionAttributeValues,
                            ReturnValues,
                            Acc) ->
    update_item_with_expression(TableName,
                                Key,
                                UpdateExpression,
                                ExpressionAttributeValues,
                                ReturnValues,
                                Acc,
                                undefined).

update_item_with_expression(TableName,
                            Key,
                            UpdateExpression,
                            ExpressionAttributeValues,
                            ReturnValues,
                            Acc,
                            Timeout) ->
    MandatoryParams =
        [{<<"TableName">>, TableName},
         {<<"Key">>, Key},
         {<<"UpdateExpression">>, UpdateExpression}],
    OptionalParams =
        [{<<"ExpressionAttributeValues">>, ExpressionAttributeValues},
         {<<"ReturnValues">>, ReturnValues}],
    DefinedOptionalParams =
        lists:filter(fun ({_, undefined}) ->
                             false;
                         (_) ->
                             true
                     end,
                     OptionalParams),

    api(update_item, MandatoryParams ++ DefinedOptionalParams ++ Acc, Timeout).

update_item(Table, Key, Options) ->
    update_item(Table, Key, Options, undefined).

update_item(Table, Key, Options, Timeout) ->
    update_item(Table, Key, Options, [], Timeout).

update_item(T, K, [], Acc, Timeout) ->
    api(update_item, [{<<"TableName">>, T}, {<<"Key">>, K} | Acc], Timeout);
update_item(T, K, [{update, AttributeUpdates} | Rest], Acc, Timeout) ->
    update_item(T,
                K,
                Rest,
                [{<<"AttributeUpdates">>, attr_updates(AttributeUpdates, [])} | Acc],
                Timeout);
update_item(T, K, [{expected, V} | Rest], Acc, Timeout) ->
    update_item(T, K, Rest, [{<<"Expected">>, attr_updates(V, [])} | Acc], Timeout);
update_item(T, K, [{return, none} | Rest], Acc, Timeout) ->
    update_item(T, K, Rest, [{<<"ReturnValues">>, ?NONE} | Acc], Timeout);
update_item(T, K, [{return, all_old} | Rest], Acc, Timeout) ->
    update_item(T, K, Rest, [{<<"ReturnValues">>, ?ALL_OLD} | Acc], Timeout);
update_item(T, K, [{return, updated_old} | Rest], Acc, Timeout) ->
    update_item(T, K, Rest, [{<<"ReturnValues">>, ?UPDATED_OLD} | Acc], Timeout);
update_item(T, K, [{return, all_new} | Rest], Acc, Timeout) ->
    update_item(T, K, Rest, [{<<"ReturnValues">>, ?ALL_NEW} | Acc], Timeout);
update_item(T, K, [{return, updated_new} | Rest], Acc, Timeout) ->
    update_item(T, K, Rest, [{<<"ReturnValues">>, ?UPDATED_NEW} | Acc], Timeout).

%% query_item options:
%% Uses amazon api version:
%% http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/API_Query_v20111205.html
%% limit: int, max number of results
%% count: bool, only return the total count
%% scan_index_forward: bool, set to false to reverse the sort order
%% consistent: bool, make a consistent read, default false
%% exclusive_start_key: output from LastEvaluatedKey when limit(size or limit param) is reached
%% attrs: array( binary ), [<<"a">>,<<"b">>], only return these attributes
%% range_condition: {array( attributes), operation }
%% eg, dinerl:query_item(<<"table">>,[{<<"S">>, <<"hash_value">>}],
%%   [{range_condition, { [[{<<"S">>, <<"range_value">>}]]  ,<<"EQ">> }}]).
query_item(Table, Key, Options) ->
    query_item(Table, Key, Options, [], undefined, undefined).

query_item(Table, Key, Options, Timeout) ->
    query_item(Table, Key, Options, [], Timeout, undefined).

query_item(Table, Key, Options, Timeout, Region) ->
    query_item(Table, Key, Options, [], Timeout, Region).

query_item(T, K, List, Acc, Timeout, Region) ->
    NewParameters =
        [{<<"TableName">>, T}, {<<"HashKeyValue">>, K} | convert_query_parameters(List, Acc)],
    api(query_item_20111205, NewParameters, Timeout, Region).

%% Uses new API
query(Table, Options) ->
    query(Table, Options, undefined, undefined).

query(Table, Options, Timeout) ->
    query(Table, Options, Timeout, undefined).

query(Table, Options, Timeout, Region) ->
    NewParameters = [{<<"TableName">>, Table} | convert_query_parameters(Options, [])],
    api(query_item_20120810, NewParameters, Timeout, Region).

convert_query_parameters([], Acc) ->
    Acc;
convert_query_parameters([{limit, V} | Rest], Acc) ->
    convert_query_parameters(Rest, [{<<"Limit">>, V} | Acc]);
convert_query_parameters([{count, V} | Rest], Acc) ->
    convert_query_parameters(Rest, [{<<"Count">>, V} | Acc]);
convert_query_parameters([{index_name, IndexName} | Rest], Acc) ->
    convert_query_parameters(Rest, [{<<"IndexName">>, IndexName} | Acc]);
convert_query_parameters([{scan_index_forward, V} | Rest], Acc) ->
    convert_query_parameters(Rest, [{<<"ScanIndexForward">>, V} | Acc]);
convert_query_parameters([{consistent, V} | Rest], Acc) ->
    convert_query_parameters(Rest, [{<<"ConsistentRead">>, V} | Acc]);
convert_query_parameters([{exclusive_start_key, V} | Rest], Acc) ->
    convert_query_parameters(Rest, [{<<"ExclusiveStartKey">>, V} | Acc]);
convert_query_parameters([{range_condition, {V, Op}} | Rest], Acc) ->
    convert_query_parameters(Rest,
                             [{<<"RangeKeyCondition">>,
                               [{<<"AttributeValueList">>, V}, {<<"ComparisonOperator">>, Op}]}
                              | Acc]);
convert_query_parameters([{attrs, V} | Rest], Acc) ->
    convert_query_parameters(Rest, [{<<"AttributesToGet">>, V} | Acc]);
convert_query_parameters([{project_expression, ProjectionExpression} | Rest], Acc) ->
    convert_query_parameters(Rest,
                             [{<<"ProjectionExpression">>, ProjectionExpression} | Acc]);
convert_query_parameters([{key_condition_expression, KeyConditionExpression} | Rest],
                         Acc) ->
    convert_query_parameters(Rest,
                             [{<<"KeyConditionExpression">>, KeyConditionExpression} | Acc]);
convert_query_parameters([{condition_expression, KeyConditionExpression} | Rest], Acc) ->
    convert_query_parameters(Rest,
                             [{<<"ConditionExpression">>, KeyConditionExpression} | Acc]);
convert_query_parameters([{filter_expression, KeyConditionExpression} | Rest], Acc) ->
    convert_query_parameters(Rest, [{<<"FilterExpression">>, KeyConditionExpression} | Acc]);
convert_query_parameters([{expression_attribute_values, ExpressionAttributeValues}
                          | Rest],
                         Acc) ->
    convert_query_parameters(Rest,
                             [{<<"ExpressionAttributeValues">>, ExpressionAttributeValues} | Acc]);
convert_query_parameters([{expression_attribute_names, ExpressionAttributeNames} | Rest],
                         Acc) ->
    convert_query_parameters(Rest,
                             [{<<"ExpressionAttributeNames">>, ExpressionAttributeNames} | Acc]).

batch_write_item(TableName, PutItems, DeleteKeys) ->
    api(batch_write_item,
        [{<<"RequestItems">>,
          [{TableName,
            lists:map(fun make_batch_put/1, PutItems)
            ++ lists:map(fun make_batch_delete/1, DeleteKeys)}]}]).

make_batch_put(Item) ->
    [{<<"PutRequest">>, [{<<"Item">>, Item}]}].

make_batch_delete(Key) ->
    [{<<"DeleteRequest">>, [{<<"Key">>, Key}]}].

%% Internal
%%
%% Every second it updates the Date part of the arguments and copies the latest cached
%% credentials from erliam.
-spec update_data(zone()) -> {ok, clientarguments()}.
update_data(Zone) ->
    NewDate = awsv4:isonow(),
    NewArgs = {erliam:credentials(), Zone, NewDate},
    ets:insert(?DINERL_DATA, {?ARGS_KEY, NewArgs}),
    {ok, NewArgs}.

expected([], Acc) ->
    Acc;
expected([{Option, Value} | Rest], Acc) ->
    expected(Rest, [value_and_action({Option, Value}) | Acc]).

attr_updates([], Acc) ->
    Acc;
attr_updates([{AttrName, Opts} | Rest], Acc) ->
    attr_updates(Rest, [{AttrName, expected(Opts, [])} | Acc]).

value_and_action({value, V}) ->
    {<<"Value">>, V};
value_and_action({action, put}) ->
    {<<"Action">>, <<"PUT">>};
value_and_action({action, add}) ->
    {<<"Action">>, <<"ADD">>};
value_and_action({action, delete}) ->
    {<<"Action">>, <<"DELETE">>};
value_and_action({exists, V}) ->
    {<<"Exists">>, V}.