ecron

-module(ecron_job).
-behaviour(gen_server).

-export([predict_datetime/3]).

-export([activate/1, deactivate/1]).
-export([statistic/1]).
-export([start_link/1, stop/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).

-define(MAX_SIZE, 20).

-define(AlreadyEnded, already_ended).
-define(Waiting, waiting).
-define(Running, running).
-define(Deactivate, deactivate).
-define(Activate, activate).

-define(Countdown_to_activate, countdown_to_activate).
-define(Stop, stop).
-define(Statistic, statistic).
-define(Instant, instant).

start_link(Ecron) -> gen_server:start_link(?MODULE, [Ecron], []).
stop(Pid) -> gen_server:call(Pid, ?Stop).
activate(Pid) -> gen_server:call(Pid, ?Activate).
deactivate(Pid) -> gen_server:call(Pid, ?Deactivate).
statistic(Pid) -> gen_server:call(Pid, ?Statistic).

init([Ecron]) ->
    #{start_time := StartTime, name := Name} = Ecron,
    TimeZone = application:get_env(ecron, time_zone, local),
    MaxTimeout = application:get_env(ecron, adjusting_time_second, 7 * 24 * 3600),
    case ets:insert_new(ecron, {Name, self()}) of
        true ->
            State = #{
                worker => undefined, ecron => Ecron, adjusting_time_second => MaxTimeout,
                status => ?Deactivate, results => [],
                run_microsecond => [], time_zone => TimeZone,
                ok => 0, failed => 0},
            erlang:process_flag(trap_exit, true),
            {ok, maybe_activate(StartTime, State)};
        false ->
            {stop, {shutdown, already_exist}}
    end.

handle_call(?Activate, _From, State = #{worker := Worker}) ->
    erlang:is_pid(Worker) andalso erlang:exit(Worker, ?Deactivate),
    {reply, ok, do_activate(State#{status =>?Deactivate, worker => undefined})};

handle_call(?Deactivate, _From, State = #{worker := Worker}) ->
    erlang:is_pid(Worker) andalso erlang:exit(Worker, ?Deactivate),
    {reply, ok, State#{status => ?Deactivate, worker => undefined}};

handle_call(?Statistic, _From, State = #{ecron := Ecron, time_zone := TimeZone}) ->
    Next = predict_datetime(Ecron, {TimeZone, current_datetime(TimeZone)}, 10),
    #{start_time := StartTime, end_time := EndTime} = Ecron,
    {reply,
        {ok, State#{
            next => Next,
            ecron =>
            Ecron#{start_time => to_rfc3339(TimeZone, StartTime),
                end_time => to_rfc3339(TimeZone, EndTime)}}
        },
        State};

handle_call(?Stop, _From, State) ->
    {stop, normal, ok, State};
handle_call(_Msg, _From, State) ->
    {reply, error, State}.

handle_cast({update_status, Status, Worker}, State = #{status := OldStatus})
    when OldStatus =/= ?Deactivate andalso OldStatus =/= ?AlreadyEnded ->
    {noreply, State#{status => Status, worker => Worker}};
handle_cast({update_run_statistic, Status, Start, End, {OkInc, FailedInc, Result}},
    State = #{status := OldStatus})
    when OldStatus =/= ?Deactivate andalso OldStatus =/= ?AlreadyEnded ->
    Cost = erlang:convert_time_unit(End - Start, native, microsecond),
    #{ok := OkCount, failed := FailedCount, results := Results, run_microsecond := Costs} = State,
    {noreply, State#{
        status => Status,
        ok => OkCount + OkInc,
        failed => FailedCount + FailedInc,
        run_microsecond => lists:sublist([Cost | Costs], ?MAX_SIZE),
        results => lists:sublist([Result | Results], ?MAX_SIZE)}
    };
handle_cast(_Msg, State) ->
    {noreply, State}.

handle_info(?Countdown_to_activate, State = #{ecron := #{start_time := StartTime}}) ->
    {noreply, maybe_activate(StartTime, State)};
handle_info(?Activate, State = #{status := ?Deactivate}) ->
    {noreply, do_activate(State)};
handle_info({'EXIT', _OldWorker, ?Deactivate}, State) ->
    {noreply, State};
handle_info({'EXIT', _OldWorker, normal}, State) ->
    {noreply, State};
handle_info(Info, State) ->
    logger:error("~p unknow message:~p~n ~p~n", [self(), Info, State]),
    {noreply, State}.

terminate(_Reason, #{ecron := #{name := JobName}, worker := Worker}) ->
    is_pid(Worker) andalso erlang:exit(Worker, killed),
    ets:delete(ecron, JobName),
    ok.

predict_datetime(#{type := every, crontab := Ms, start_time := Start, end_time := End},
    Now, Num) ->
    predict_every_datetime(Start, End, trunc(Ms / 1000), Now, Num, []);
predict_datetime(#{type := cron, crontab := Job, start_time := Start, end_time := End},
    Now, Num) ->
    predict_cron_datetime(Start, End, Job, Now, Num, []);
predict_datetime(Type, Job, Num) ->
    TimeZone = application:get_env(ecron, time_zone, local),
    Now = current_datetime(TimeZone),
    predict_datetime(#{type => Type, crontab => Job, start_time => unlimited, end_time => unlimited},
        {TimeZone, Now}, Num).

%%%===================================================================

%%% Internal functions

%%%===================================================================

do_activate(State = #{ecron := Ecron, status := ?Deactivate,
    time_zone := TimeZone, adjusting_time_second := MaxTimeout}) ->
    #{type := Type, mfa := MFA, crontab := Crontab, end_time := EndTime} = Ecron,
    EndSec = case EndTime of unlimited -> unlimited; _ -> calendar:datetime_to_gregorian_seconds(EndTime) end,
    Self = self(),
    WorkerFun =
        case Type of
            every -> fun() -> schedule_every(Crontab, MFA, EndSec, Self, TimeZone) end;
            cron ->
                fun() ->
                    schedule_cron(current_datetime(TimeZone),
                        Crontab, MFA, EndSec, Self, MaxTimeout, TimeZone) end
        end,
    State#{worker => spawn_link(WorkerFun), status => ?Waiting}.

schedule_every(Ms, MFA, EndSec, ParentPid, TimeZone) ->
    case unlimited =:= EndSec orelse
        calendar:datetime_to_gregorian_seconds(current_datetime(TimeZone)) < EndSec
    of
        true ->
            gen_server:cast(ParentPid, {update_status, ?Running, self()}),
            Start = erlang:monotonic_time(),
            Result = safe_apply_job(MFA),
            gen_server:cast(ParentPid, {update_run_statistic, ?Waiting, Start, erlang:monotonic_time(), Result}),
            timer:sleep(Ms),
            schedule_every(Ms, MFA, EndSec, ParentPid, TimeZone);
        false ->
            gen_server:cast(ParentPid, {update_status, ?AlreadyEnded, undefined})
    end.

schedule_cron(PrevDateTime, DateSpec, MFA, EndSec, ParentPid, MaxTimeout, TimeZone) ->
    CurrentDateTime = current_datetime(TimeZone),
    case unlimited =:= EndSec orelse
        calendar:datetime_to_gregorian_seconds(CurrentDateTime) < EndSec
    of
        true ->
            NextValidDateTime = next_schedule_datetime(DateSpec, CurrentDateTime, PrevDateTime),
            SleepSec = second_diff(CurrentDateTime, NextValidDateTime),
            gen_server:cast(ParentPid, {update_status, ?Waiting, self()}),
            case SleepSec > MaxTimeout of
                true ->
                    timer:sleep(MaxTimeout * 1000),
                    schedule_cron(CurrentDateTime, DateSpec, MFA, EndSec, ParentPid, MaxTimeout, TimeZone);
                false ->
                    timer:sleep(SleepSec * 1000),
                    gen_server:cast(ParentPid, {update_status, ?Running, self()}),
                    Start = erlang:monotonic_time(),
                    Res = safe_apply_job(MFA),
                    gen_server:cast(ParentPid,
                        {update_run_statistic, ?Waiting, Start, erlang:monotonic_time(), Res}),
                    schedule_cron(CurrentDateTime, DateSpec, MFA, EndSec, ParentPid, MaxTimeout, TimeZone)
            end;
        false ->
            gen_server:cast(ParentPid, {update_status, ?AlreadyEnded, undefined})
    end.

safe_apply_job(Exec) ->
    try
        case Exec of
            {M, F, A} ->
                {1, 0, apply(M, F, A)};
            {F, A} ->
                {1, 0, apply(F, A)}
        end
    catch
        Error:Reason:Stacktrace ->
            Format = "Job ~p in process ~p with result:~n~p",
            Message = lists:flatten(io_lib:format(
                Format,
                [Error, self(), {Reason, Stacktrace}])),
            logger:error(Message),
            {0, 1, {Error, Reason}}
    end.

second_diff(CurrentDateTime, NextDateTime) ->
    CurrentSeconds = calendar:datetime_to_gregorian_seconds(CurrentDateTime),
    NextSeconds = calendar:datetime_to_gregorian_seconds(NextDateTime),
    NextSeconds - CurrentSeconds.

next_schedule_datetime(DateSpec, NowDateTime, PreDateTime) ->
    Incr = case NowDateTime =:= PreDateTime of true -> 1; false -> 0 end,
    FSeconds = calendar:datetime_to_gregorian_seconds(NowDateTime) + Incr,
    ForwardDateTime = calendar:gregorian_seconds_to_datetime(FSeconds),
    DefaultMin = #{second => 0, minute => 0, hour => 0, day_of_month => 1, month => 1, day_of_week => 0},
    Min = spec_min(maps:to_list(DateSpec), DefaultMin),
    next_schedule_datetime_1(DateSpec, Min, ForwardDateTime).

next_schedule_datetime_1(DateSpec, Min, DateTime) ->
    #{
        second := SecondSpec, minute := MinuteSpec, hour := HourSpec,
        day_of_month := DayOfMonthSpec, month := MonthSpec,
        day_of_week := DayOfWeekSpec} = DateSpec,
    
    {{Year, Month, Day}, {Hour, Minute, Second}} = DateTime,
    case valid_datetime(MonthSpec, Month) of
        false ->
            IncrMonth = nearest(month, Month, 12, DateSpec),
            forward_month(DateTime, IncrMonth, Min, DateSpec);
        true ->
            DayOfWeek = case calendar:day_of_the_week(Year, Month, Day) of 7 -> 0; DOW -> DOW end,
            DOMValid = valid_datetime(DayOfMonthSpec, Day),
            DOWValid = valid_datetime(DayOfWeekSpec, DayOfWeek),
            case (DOMValid andalso DOWValid) orelse
                ((DayOfMonthSpec =/= '*') andalso
                    (DayOfWeekSpec =/= '*') andalso
                    (DOMValid orelse DOWValid))
            of
                false ->
                    LastDay = calendar:last_day_of_the_month(Year, Month),
                    IncrDay = nearest(day_of_month, Day, LastDay, DateSpec),
                    IncrWeek = nearest(day_of_week, DayOfWeek, 6, DateSpec),
                    forward_day(DateTime, min(IncrDay, IncrWeek), Min, DateSpec);
                true ->
                    case valid_datetime(HourSpec, Hour) of
                        false ->
                            IncrHour = nearest(hour, Hour, 23, DateSpec),
                            forward_hour(DateTime, IncrHour, Min, DateSpec);
                        true ->
                            case valid_datetime(MinuteSpec, Minute) of
                                false ->
                                    IncrMinute = nearest(minute, Minute, 59, DateSpec),
                                    forward_minute(DateTime, IncrMinute, Min, DateSpec);
                                true ->
                                    case valid_datetime(SecondSpec, Second) of
                                        false ->
                                            IncrSec = nearest(second, Second, 59, DateSpec),
                                            forward_second(DateTime, IncrSec, Min, DateSpec);
                                        true -> DateTime
                                    end
                            end
                    end
            end
    end.

forward_second(DateTime, Incr, Min, Spec) when Incr > 0 ->
    {{Year, Month, Day}, {Hour, Minute, Second}} = DateTime,
    case Second + Incr of
        NewSecond when NewSecond > 59 ->
            IncrMinute = nearest(minute, Minute, 59, Spec),
            forward_minute(DateTime, IncrMinute, Min, Spec);
        NewSecond -> {{Year, Month, Day}, {Hour, Minute, NewSecond}}
    end.

forward_minute(DateTime, Incr, Min, Spec) when Incr > 0 ->
    {{Year, Month, Day}, {Hour, Minute, _Second}} = DateTime,
    #{second := SecondM} = Min,
    case Minute + Incr of
        NewMinute when NewMinute > 59 ->
            IncrHour = nearest(hour, Hour, 23, Spec),
            forward_hour(DateTime, IncrHour, Min, Spec);
        NewMinute -> {{Year, Month, Day}, {Hour, NewMinute, SecondM}}
    end.

forward_hour(DateTime, Incr, Min, Spec) when Incr > 0 ->
    {{Year, Month, Day}, {Hour, _Minute, _Second}} = DateTime,
    #{minute := MinuteM, second := SecondM} = Min,
    case Hour + Incr of
        NewHour when NewHour > 23 ->
            LastDay = calendar:last_day_of_the_month(Year, Month),
            IncrDay = nearest(day_of_month, Day, LastDay, Spec),
            DayOfWeek = case calendar:day_of_the_week(Year, Month, Day) of 7 -> 0; DOW -> DOW end,
            IncrWeek = nearest(day_of_week, DayOfWeek, 6, Spec),
            forward_day(DateTime, min(IncrDay, IncrWeek), Min, Spec);
        NewHour -> {{Year, Month, Day}, {NewHour, MinuteM, SecondM}}
    end.

forward_day(DateTime, Incr, Min, Spec) when Incr >= 0 ->
    {{Year, Month, Day}, {_Hour, _Minute, _Second}} = DateTime,
    #{hour := HourM, minute := MinuteM, second := SecondM} = Min,
    LastDay = calendar:last_day_of_the_month(Year, Month),
    case Day + Incr of
        NewDay when NewDay > LastDay ->
            IncrMonth = nearest(month, Month, 12, Spec),
            forward_month(DateTime, IncrMonth, Min, Spec);
        NewDay ->
            NewDateTime = {{Year, Month, NewDay}, {HourM, MinuteM, SecondM}},
            #{day_of_week := DayOfWeekSpec, day_of_month := DayOfMonthSpec} = Spec,
            case check_valid_day(Year, Month, NewDay, DayOfMonthSpec, DayOfWeekSpec) of
                true -> NewDateTime;
                false ->
                    IncrDay = nearest(day_of_month, NewDay, LastDay, Spec),
                    DayOfWeek = case calendar:day_of_the_week(Year, Month, NewDay) of 7 -> 0; DOW -> DOW end,
                    IncrWeek = nearest(day_of_week, DayOfWeek, 6, Spec),
                    forward_day(NewDateTime, min(IncrDay, IncrWeek), Min, Spec)
            end
    end.

check_valid_day(Year, Month, NDay, DayOfMonthSpec, DayOfWeekSpec) ->
    DayOfWeek = case calendar:day_of_the_week(Year, Month, NDay) of 7 -> 0; DOW -> DOW end,
    DOMValid = valid_datetime(DayOfMonthSpec, NDay),
    DOWValid = valid_datetime(DayOfWeekSpec, DayOfWeek),
    (DOMValid andalso DOWValid) orelse
        ((DayOfMonthSpec =/= '*') andalso
            (DayOfWeekSpec =/= '*') andalso
            (DOMValid orelse DOWValid)).


forward_month(DateTime, Incr, Min, Spec) when Incr > 0 ->
    {{Year, Month, _Day}, {_Hour, _Minute, _Second}} = DateTime,
    #{month := MonthM, hour := HourM, minute := MinuteM, second := SecondM} = Min,
    NewDateTime =
        {{NYear, NMonth, NDay}, {_NHour, _NMinute, _NSecond}} =
        case Month + Incr of
            NewMonth when NewMonth > 12 -> {{Year + 1, MonthM, 1}, {HourM, MinuteM, SecondM}};
            NewMonth -> {{Year, NewMonth, 1}, {HourM, MinuteM, SecondM}}
        end,
    #{day_of_week := DayOfWeekSpec, day_of_month := DayOfMonthSpec} = Spec,
    case check_valid_day(NYear, NMonth, NDay, DayOfMonthSpec, DayOfWeekSpec) of
        true -> NewDateTime;
        false ->
            LastDay = calendar:last_day_of_the_month(NYear, NMonth),
            IncrDay = nearest(day_of_month, NDay, LastDay, Spec),
            DayOfWeek = case calendar:day_of_the_week(NYear, NMonth, NDay) of 7 -> 0; DOW -> DOW end,
            IncrWeek = nearest(day_of_week, DayOfWeek, 6, Spec),
            forward_day(NewDateTime, min(IncrDay, IncrWeek), Min, Spec)
    end.

nearest(Type, Current, Max, Spec) ->
    Values = maps:get(Type, Spec),
    case nearest_1(Values, Values, Max, Current + 1) of
        {forward, Next} -> Max - Current + Next + 1;
        {done, Next} -> Next - Current
    end.

nearest_1('*', _, MaxLimit, Next) ->
    case Next > MaxLimit of
        true -> {forward, 1};
        false -> {done, Next}
    end;
nearest_1([], [{Min, _} | _], _Max, _Next) -> {forward, Min};
nearest_1([], [Min | _], _Max, _Next) -> {forward, Min};
nearest_1([{Min, Max} | Rest], Spec, MaxLimit, Next) ->
    if
        Next > Max -> nearest_1(Rest, Spec, MaxLimit, Next);
        Next =< Min -> {done, Min};
        true -> {done, Next}
    end;
nearest_1([Expect | Rest], Spec, MaxLimit, Next) ->
    if
        Next > Expect -> nearest_1(Rest, Spec, MaxLimit, Next);
        true -> {done, Expect}
    end.

valid_datetime('*', _Value) -> true;
valid_datetime([], _Value) -> false;
valid_datetime([Value | _T], Value) -> true;
valid_datetime([{Lower, Upper} | _], Value) when Lower =< Value andalso Value =< Upper -> true;
valid_datetime([_ | T], Value) -> valid_datetime(T, Value).

maybe_activate(StartTime, State = #{adjusting_time_second := MaxTimeout, time_zone := TimeZone}) ->
    case StartTime of
        unlimited -> do_activate(State);
        _ ->
            case second_diff(current_datetime(TimeZone), StartTime) of
                Sec when Sec > MaxTimeout ->
                    erlang:send_after(MaxTimeout * 1000, self(), ?Countdown_to_activate),
                    State;
                Sec when Sec > 0 ->
                    erlang:send_after(Sec * 1000, self(), ?Activate),
                    State;
                _ -> do_activate(State)
            end
    end.

predict_cron_datetime(_Start, _End, _Job, _Now, 0, Acc) -> lists:reverse(Acc);
predict_cron_datetime(Start, End, Job, {TimeZone, Now}, Num, Acc) ->
    Next = next_schedule_datetime(Job, Now, Now),
    case in_range(Next, Start, End) of
        ?AlreadyEnded -> lists:reverse(Acc);
        ?Deactivate -> predict_cron_datetime(Start, End, Job, {TimeZone, Start}, Num, Acc);
        ?Running ->
            predict_cron_datetime(Start, End, Job, {TimeZone, Next},
                Num - 1, [to_rfc3339(TimeZone, Next) | Acc])
    end.

predict_every_datetime(_Start, _End, _Second, _Now, 0, Acc) -> lists:reverse(Acc);
predict_every_datetime(Start, End, Second, {TimeZone, Now}, Num, Acc) ->
    FSeconds = calendar:datetime_to_gregorian_seconds(Now) + Second,
    Next = calendar:gregorian_seconds_to_datetime(FSeconds),
    case in_range(Next, Start, End) of
        ?AlreadyEnded -> lists:reverse(Acc);
        ?Deactivate -> predict_every_datetime(Start, End, Second, {TimeZone, Start}, Num, Acc);
        ?Running ->
            predict_every_datetime(Start, End, Second, {TimeZone, Next}, Num - 1,
                [to_rfc3339(TimeZone, Next) | Acc])
    end.

in_range(_Current, unlimited, unlimited) -> ?Running;
in_range(Current, unlimited, End) ->
    case second_diff(End, Current) > 0 of
        true -> ?AlreadyEnded;
        false -> ?Running
    end;
in_range(Current, Start, unlimited) ->
    case second_diff(Start, Current) < 0 of
        true -> ?Deactivate;
        false -> ?Running
    end;
in_range(Current, Start, End) ->
    case second_diff(End, Current) > 0 of
        true -> ?AlreadyEnded;
        false ->
            case second_diff(Start, Current) < 0 of
                true -> ?Deactivate;
                false -> ?Running
            end
    end.

current_datetime(local) -> calendar:local_time();
current_datetime(utc) -> calendar:universal_time().

-define(SECONDS_FROM_0_TO_1970, 719528 * 86400).

to_rfc3339(_TimeZone, unlimited) -> unlimited;
to_rfc3339(TimeZone, Time) ->
    UniversalTime =
        case TimeZone of
            utc -> Time;
            local -> [T1 | _] = calendar:local_time_to_universal_time_dst(Time), T1
        end,
    SystemTime = calendar:datetime_to_gregorian_seconds(UniversalTime) - ?SECONDS_FROM_0_TO_1970,
    calendar:system_time_to_rfc3339(SystemTime).

spec_min([], Acc) -> Acc;
spec_min([{Key, Value} | Rest], Acc) ->
    NewAcc =
        case Value of
            '*' -> Acc;
            [{Min, _} | _] -> Acc#{Key => Min};
            [Min | _] -> Acc#{Key => Min}
        end,
    spec_min(Rest, NewAcc).

%% For PropEr Test

-ifdef(TEST).
-compile(export_all).
-endif.