-module(dqe_get_fun).
-behaviour(dflow).

-export([init/1, describe/1, start/2, emit/3, done/2]).

-record(state, {
          bucket :: binary(),
          key :: binary(),
          start :: non_neg_integer(),
          count :: non_neg_integer(),
          get_chunk :: pos_integer(),

          dqe_fun :: atom(),
          chunk :: pos_integer(),
          acc = <<>> :: binary(),
          fun_state :: dqe_fun:fun_state()

         }).

init([Fun, FunState, Start, Count, Resolution, Bucket, Key]) ->
    {ok, Chunk} = application:get_env(dqe, get_chunk),
    init([Start, Count, Resolution, Bucket, Key, Chunk]);
init([Start, Count, Resolution, Bucket, KeyL, Get_Chunk]) when is_list(KeyL)->
    Key = dproto:metric_from_list(KeyL),
    init([Start, Count, Resolution, Bucket, Key, Get_Chunk]);
init([Fun, FunState, Start, Count, _Resolution, Bucket, Key, Get_Chunk]) ->
    Chunk = Fun:chunk(FunState),
    {ok, #state{start = Start, count = Count, bucket = Bucket, key = Key,
                get_chunk = Get_Chunk, dqe_fun = Fun, chunk = Chunk,
                fun_state = FunState}, []}.

describe(#state{bucket = Bucket, key = Key}) ->
    [Bucket, "/", Key].

start(run, State = #state{count = 0}) ->
    {done, State};

start(run,
      State = #state{start = Start, count = Count, get_chunk = Get_Chunk,
                     bucket = Bucket, key = Key, acc = Acc}) when
      Count >= Get_Chunk ->
    %% We do a bit of cheating here this allows us to loop.
    State1 = State#state{start = Start + Get_Chunk, count = Count - Get_Chunk},
    case ddb_connection:get(Bucket, Key, Start, Get_Chunk) of
        {error, _Error} ->
            {done, State};
        {ok, <<>>} ->
            dflow:start(self(), run),
            {emit, mmath_bin:realize(mmath_bin:empty(Get_Chunk)), State1};
        {ok, Data} ->
            dflow:start(self(), run),
            {emit, mmath_bin:realize(Data), State1}
    end;

start(run, State = #state{start = Start, count = Count,
                     bucket = Bucket, key = Key}) ->
    case ddb_connection:get(Bucket, Key, Start, Count) of
        {error, _Error} ->
            {done, State};
        {ok, <<>>} ->
            do(State, mmath_bin:empty(Count), done);
            {done, mmath_bin:realize(), State};
        {ok, Data} ->
            do(State, mmath_bin:empty(Data), done)
    end.

do(State, Data, Action) ->
    %%<<Acc/binary, mmath_bin:empty(Count)>>
    {Action, Data, State}.

emit(_Child, _Data, State) ->
    {ok, State}.

done(_, State) ->
    {done, State}.