path: root/src/mem3/src/mem3_reshard.erl

% Licensed under the Apache License, Version 2.0 (the "License"); you may not
% use this file except in compliance with the License. You may obtain a copy of
% the License at
%
%   http://www.apache.org/licenses/LICENSE-2.0
%
% Unless required by applicable law or agreed to in writing, software
% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
% License for the specific language governing permissions and limitations under
% the License.

-module(mem3_reshard).


-behaviour(gen_server).


-export([
    start_link/0,

    start/0,
    stop/1,

    start_split_job/1,
    stop_job/2,
    resume_job/1,
    remove_job/1,

    get_state/0,
    jobs/0,
    job/1,
    is_disabled/0,

    report/2,
    checkpoint/2,

    now_sec/0,
    update_history/4,
    shard_from_name/1,
    reset_state/0
]).

-export([
    init/1,
    terminate/2,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    code_change/3
]).


-include("mem3_reshard.hrl").


-define(JOB_ID_VERSION, 1).
-define(JOB_STATE_VERSION, 1).
-define(DEFAULT_MAX_JOBS, 48).
-define(DEFAULT_MAX_HISTORY, 20).
-define(JOB_PREFIX, <<"reshard-job-">>).
-define(STATE_PREFIX, <<"reshard-state-">>).


%% Public API

-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
start_link() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).


-spec start() -> ok | {error, any()}.
start() ->
    case is_disabled() of
        true -> {error, resharding_disabled};
        false -> gen_server:call(?MODULE, start, infinity)
    end.


-spec stop(binary()) -> ok | {error, any()}.
stop(Reason) ->
    case is_disabled() of
        true -> {error, resharding_disabled};
        false -> gen_server:call(?MODULE, {stop, Reason}, infinity)
    end.


-spec start_split_job(#shard{} | binary()) -> {ok, binary()} | {error, term()}.
start_split_job(#shard{} = Shard) ->
    start_split_job(Shard, 2);

start_split_job(ShardName) when is_binary(ShardName) ->
    start_split_job(shard_from_name(ShardName), 2).


-spec start_split_job(#shard{}, split()) -> {ok, binary()} | {error, any()}.
start_split_job(#shard{} = Source, Split) ->
    case is_disabled() of
        true -> {error, resharding_disabled};
        false -> validate_and_start_job(Source, Split)
    end.


-spec stop_job(binary(), binary()) -> ok | {error, any()}.
stop_job(JobId, Reason) when is_binary(JobId), is_binary(Reason) ->
    case is_disabled() of
        true -> {error, resharding_disabled};
        false -> gen_server:call(?MODULE, {stop_job, JobId, Reason}, infinity)
    end.


-spec resume_job(binary()) -> ok | {error, any()}.
resume_job(JobId) when is_binary(JobId) ->
    case is_disabled() of
        true -> {error, resharding_disabled};
        false -> gen_server:call(?MODULE, {resume_job, JobId}, infinity)
    end.


-spec remove_job(binary()) -> ok | {error, any()}.
remove_job(JobId) when is_binary(JobId) ->
    case is_disabled() of
        true -> {error, resharding_disabled};
        false -> gen_server:call(?MODULE, {remove_job, JobId}, infinity)
    end.


-spec get_state() -> {[_ | _]}.
get_state() ->
    gen_server:call(?MODULE, get_state, infinity).


-spec jobs() -> [[tuple()]].
jobs() ->
    ets:foldl(fun(Job, Acc) ->
        Opts = [iso8601],
        Props = mem3_reshard_store:job_to_ejson_props(Job, Opts),
        [{Props} | Acc]
    end, [], ?MODULE).


-spec job(job_id()) -> {ok, {[_ | _]}} | {error, not_found}.
job(JobId) ->
    case job_by_id(JobId) of
        #job{} = Job ->
            Opts = [iso8601],
            Props = mem3_reshard_store:job_to_ejson_props(Job, Opts),
            {ok, {Props}};
        not_found ->
            {error, not_found}
    end.


% Return true resharding is disabled in the application level settings
-spec is_disabled() -> boolean().
is_disabled() ->
    case application:get_env(mem3, reshard_disabled) of
        {ok, "true"} -> true;
        {ok, true} -> true;
        _ -> false
    end.


% State reporting callbacks. Used by mem3_reshard_job module.
-spec report(pid(), #job{}) -> ok.
report(Server, #job{} = Job) when is_pid(Server) ->
    gen_server:cast(Server, {report, Job}).


-spec checkpoint(pid(), #job{}) -> ok.
checkpoint(Server, #job{} = Job) ->
    couch_log:notice("~p checkpointing ~p ~p", [?MODULE, Server, jobfmt(Job)]),
    gen_server:cast(Server, {checkpoint, Job}).


% Utility functions used from other mem3_reshard modules

-spec now_sec() -> non_neg_integer().
now_sec() ->
    {Mega, Sec, _Micro} = os:timestamp(),
    Mega * 1000000 + Sec.


-spec update_history(atom(), binary() | null, time_sec(), list()) -> list().
update_history(State, State, Ts, History) ->
    % State is the same as detail. Make the detail null to avoid duplication
    update_history(State, null, Ts, History);

update_history(State, Detail, Ts, History) ->
    % Reverse, so we can process the last event as the head using
    % head matches, then after append and trimming, reserse again
    Rev = lists:reverse(History),
    UpdatedRev = update_history_rev(State, Detail, Ts, Rev),
    TrimmedRev = lists:sublist(UpdatedRev, max_history()),
    lists:reverse(TrimmedRev).


-spec shard_from_name(binary()) -> #shard{}.
shard_from_name(<<"shards/", _:8/binary, "-", _:8/binary, "/",
        Rest/binary>> = Shard) ->
    Range = mem3:range(Shard),
    [DbName, Suffix] = binary:split(Rest, <<".">>),
    build_shard(Range, DbName, Suffix).


% For debugging only

-spec reset_state() -> ok.
reset_state() ->
    gen_server:call(?MODULE, reset_state, infinity).


% Gen server functions

init(_) ->
    % Advertise resharding API feature only if it is not disabled
    % reshard feature not implemented on FDB
    % case is_disabled() of
    %     true -> ok;
    %     false -> config:enable_feature('reshard')
    % end,
    couch_log:notice("~p start init()", [?MODULE]),
    EtsOpts = [named_table, {keypos, #job.id}, {read_concurrency, true}],
    ?MODULE = ets:new(?MODULE, EtsOpts),
    ManagerPid = self(),
    State = #state{
        state = running,
        state_info = [],
        update_time = now_sec(),
        node = node(),
        db_monitor = spawn_link(fun() -> db_monitor(ManagerPid) end)
    },
    State1 = mem3_reshard_store:init(State, ?JOB_PREFIX, state_id()),
    State2 = mem3_reshard_store:load_state(State1, running),
    State3 = maybe_disable(State2),
    gen_server:cast(self(), reload_jobs),
    {ok, State3}.


terminate(Reason, State) ->
    couch_log:notice("~p terminate ~p ~p", [?MODULE, Reason, statefmt(State)]),
    catch unlink(State#state.db_monitor),
    catch exit(State#state.db_monitor, kill),
    lists:foreach(fun(Job) -> kill_job_int(Job) end, running_jobs()).


handle_call(start, _From, #state{state = stopped} = State) ->
    State1 = State#state{
        state = running,
        update_time = now_sec(),
        state_info = info_delete(reason, State#state.state_info)
    },
    ok = mem3_reshard_store:store_state(State1),
    State2 = maybe_disable(State1),
    State3 = reload_jobs(State2),
    {reply, ok, State3};

handle_call(start, _From, State) ->
    {reply, ok, State};

handle_call({stop, Reason}, _From, #state{state = running} = State) ->
    State1 = State#state{
        state = stopped,
        update_time = now_sec(),
        state_info = info_update(reason, Reason, State#state.state_info)
    },
    ok = mem3_reshard_store:store_state(State1),
    lists:foreach(fun(Job) -> temporarily_stop_job(Job) end, running_jobs()),
    {reply, ok, State1};

handle_call({stop, _}, _From, State) ->
    {reply, ok, State};

handle_call({start_job, #job{id = Id, source = Source} = Job}, _From, State) ->
    couch_log:notice("~p start_job call ~p", [?MODULE, jobfmt(Job)]),
    Total = ets:info(?MODULE, size),
    SourceOk = mem3_reshard_validate:source(Source),
    case {job_by_id(Id), Total + 1 =<  get_max_jobs(), SourceOk} of
        {not_found, true, ok} ->
            handle_start_job(Job, State);
        {#job{}, _, _} ->
            {reply, {error, job_already_exists}, State};
        {_, false, _} ->
            {reply, {error, max_jobs_exceeded}, State};
        {_, _, {error, _} = SourceError} ->
            {reply, SourceError, State}
    end;

handle_call({resume_job, _}, _From, #state{state = stopped} = State) ->
    case couch_util:get_value(reason, State#state.state_info) of
        undefined ->
            {reply, {error, stopped}, State};
        Reason ->
            {reply, {error, {stopped, Reason}}, State}
    end;

handle_call({resume_job, Id}, _From, State) ->
    couch_log:notice("~p resume_job call ~p", [?MODULE, Id]),
    case job_by_id(Id) of
        #job{job_state = stopped} = Job ->
            case start_job_int(Job, State) of
                ok ->
                    {reply, ok, State};
                {error, Error} ->
                    {reply, {error, Error}, State}
            end;
        #job{} ->
            {reply, ok, State};
        not_found ->
            {reply, {error, not_found}, State}
    end;

handle_call({stop_job, Id, Reason}, _From, State) ->
    couch_log:notice("~p stop_job Id:~p Reason:~p", [?MODULE, Id, Reason]),
    case job_by_id(Id) of
        #job{job_state = JSt} = Job when JSt =:= running orelse JSt =:= new
                orelse JSt =:= stopped ->
            ok = stop_job_int(Job, stopped, Reason, State),
            {reply, ok, State};
        #job{} ->
            {reply, ok, State};
        not_found ->
            {reply, {error, not_found}, State}
    end;

handle_call({remove_job, Id}, _From, State) ->
    {reply, remove_job_int(Id, State), State};

handle_call(get_state, _From, #state{state = GlobalState} = State) ->
    StateProps = mem3_reshard_store:state_to_ejson_props(State),
    Stats0 =  #{running => 0, completed => 0, failed => 0, stopped => 0},
    StateStats = ets:foldl(fun(#job{job_state = JS}, Acc) ->
        % When jobs are disabled globally their state is not checkpointed as
        % "stopped", but it stays as "running". But when returning the state we
        % don't want to mislead and indicate that there are "N running jobs"
        % when the global state is "stopped".
        JS1 = case GlobalState =:= stopped andalso JS =:= running of
            true -> stopped;
            false -> JS
        end,
        Acc#{JS1 => maps:get(JS1, Acc, 0) + 1}
    end, Stats0, ?MODULE),
    Total = ets:info(?MODULE, size),
    StateStats1 = maps:to_list(StateStats) ++ [{total, Total}],
    Result = {lists:sort(StateProps ++ StateStats1)},
    {reply, Result, State};

handle_call(reset_state, _From, State) ->
    {reply, ok, reset_state(State)};

handle_call(Call, From, State) ->
    couch_log:error("~p unknown call ~p from: ~p", [?MODULE, Call, From]),
    {noreply, State}.


handle_cast({db_deleted, DbName}, State) ->
    % Remove only completed jobs. Other running states would `fail` but
    % job result would stick around so users can inspect them.
    JobIds = jobs_by_db_and_state(DbName, completed),
    [remove_job_int(JobId, State) || JobId <- JobIds],
    {noreply, State};

handle_cast({report, Job}, State) ->
    report_int(Job),
    {noreply, State};

handle_cast({checkpoint, Job}, State) ->
    {noreply, checkpoint_int(Job, State)};

handle_cast(reload_jobs, State) ->
    couch_log:notice("~p starting reloading jobs", [?MODULE]),
    State1 = reload_jobs(State),
    couch_log:notice("~p finished reloading jobs", [?MODULE]),
    {noreply, State1};

handle_cast(Cast, State) ->
    couch_log:error("~p unexpected cast ~p", [?MODULE, Cast]),
    {noreply, State}.


handle_info({'DOWN', _Ref, process, Pid, Info}, State) ->
    case job_by_pid(Pid) of
        {ok, Job} ->
            couch_log:notice("~p job ~s exit ~p", [?MODULE, Job#job.id, Info]),
            ok = handle_job_exit(Job, Info, State);
        {error, not_found} ->
            couch_log:error("~p job not found: ~p ~p", [?MODULE, Pid, Info])
    end,
    {noreply, State};

handle_info(Info, State) ->
    couch_log:error("~p unexpected info ~p", [?MODULE, Info]),
    {noreply, State}.


code_change(_OldVsn, State, _Extra) ->
    {ok, State}.


%% Private API

validate_and_start_job(#shard{} = Source, Split) ->
    case mem3_reshard_validate:start_args(Source, Split) of
        ok ->
            Target = target_shards(Source, Split),
            case mem3_reshard_validate:targets(Source, Target) of
                ok ->
                    TStamp = now_sec(),
                    Job = #job{
                        type = split,
                        job_state = new,
                        split_state = new,
                        start_time = TStamp,
                        update_time = TStamp,
                        node = node(),
                        source = Source,
                        target = Target
                    },
                    Job1 = Job#job{id = job_id(Job)},
                    Job2 = update_job_history(Job1),
                    gen_server:call(?MODULE, {start_job, Job2}, infinity);
                {error, Error} ->
                    {error, Error}
            end;
        {error, Error} ->
            {error, Error}
    end.


handle_start_job(#job{} = Job, #state{state = running} = State) ->
    case start_job_int(Job, State) of
        ok ->
            {reply, {ok, Job#job.id}, State};
        {error, Error} ->
            {reply, {error, Error}, State}
    end;

handle_start_job(#job{} = Job, #state{state = stopped} = State) ->
    ok = mem3_reshard_store:store_job(State, Job),
    % Since resharding is stopped on this node, the job is temporarily marked
    % as stopped in the ets table so as not to return a "running" result which
    % would look odd.
    temporarily_stop_job(Job),
    {reply, {ok, Job#job.id}, State}.


% Insert job in the ets table as a temporarily stopped job. This would happen
% when a job is reloaded or added when node-wide resharding is stopped.
-spec temporarily_stop_job(#job{}) -> #job{}.
temporarily_stop_job(Job) ->
    Job1 = kill_job_int(Job),
    OldInfo = Job1#job.state_info,
    Reason = <<"Shard splitting disabled">>,
    Job2 = Job1#job{
        job_state = stopped,
        update_time = now_sec(),
        start_time = 0,
        state_info = info_update(reason, Reason, OldInfo),
        pid = undefined,
        ref = undefined
    },
    Job3 = update_job_history(Job2),
    true = ets:insert(?MODULE, Job3),
    Job3.


-spec reload_jobs(#state{}) -> #state{}.
reload_jobs(State) ->
    Jobs = mem3_reshard_store:get_jobs(State),
    lists:foldl(fun reload_job/2, State, Jobs).


% This is a case when main application is stopped but a job is reloaded that
% was checkpointed in running state. Set that state to stopped to avoid the API
% results looking odd.
-spec reload_job(#job{}, #state{}) -> #state{}.
reload_job(#job{job_state = JS} = Job, #state{state = stopped} = State)
        when JS =:= running orelse JS =:= new ->
    temporarily_stop_job(Job),
    State;

% This is a case when a job process should be spawend
reload_job(#job{job_state = JS} = Job, #state{state = running} = State)
        when JS =:= running orelse JS =:= new ->
    case start_job_int(Job, State) of
        ok ->
            State;
        {error, Error} ->
            Msg = "~p could not resume ~s error: ~p",
            couch_log:error(Msg, [?MODULE, jobfmt(Job), Error]),
            State
    end;

% If job is disabled individually (stopped by the user), is completed or failed
% then simply load it into the ets table
reload_job(#job{job_state = JS} = Job, #state{} = State)
        when JS =:= failed orelse JS =:= completed orelse JS =:= stopped ->
    true = ets:insert(?MODULE, Job),
    State.


-spec get_max_jobs() -> integer().
get_max_jobs() ->
    config:get_integer("reshard", "max_jobs", ?DEFAULT_MAX_JOBS).


-spec start_job_int(#job{}, #state{}) -> ok | {error, term()}.
start_job_int(Job, State) ->
    case spawn_job(Job) of
        {ok, #job{} = Job1} ->
            Job2 = update_job_history(Job1),
            ok = mem3_reshard_store:store_job(State, Job2),
            true = ets:insert(?MODULE, Job2),
            ok;
        {error, Error} ->
            {error, Error}
    end.


-spec spawn_job(#job{}) -> {ok, pid()} | {error, term()}.
spawn_job(#job{} = Job0) ->
    Job = Job0#job{
        job_state = running,
        start_time = 0,
        update_time = now_sec(),
        state_info = info_delete(reason, Job0#job.state_info),
        manager = self(),
        workers = [],
        retries = 0
    },
    case mem3_reshard_job_sup:start_child(Job) of
        {ok, Pid} ->
            Ref = monitor(process, Pid),
            {ok, Job#job{pid = Pid, ref = Ref}};
        {error, Reason} ->
            {error, Reason}
    end.


-spec stop_job_int(#job{}, job_state(), term(), #state{}) -> ok.
stop_job_int(#job{} = Job, JobState, Reason, State) ->
    couch_log:info("~p stop_job_int ~p newstate: ~p reason:~p", [?MODULE,
        jobfmt(Job), JobState, Reason]),
    Job1 = kill_job_int(Job),
    Job2 = Job1#job{
        job_state = JobState,
        update_time = now_sec(),
        state_info = [{reason, Reason}]
    },
    ok = mem3_reshard_store:store_job(State, Job2),
    true = ets:insert(?MODULE, Job2),
    couch_log:info("~p stop_job_int stopped ~p", [?MODULE, jobfmt(Job2)]),
    ok.


-spec kill_job_int(#job{}) -> #job{}.
kill_job_int(#job{pid = undefined} = Job) ->
    Job;

kill_job_int(#job{pid = Pid, ref = Ref} = Job) ->
    couch_log:info("~p kill_job_int ~p", [?MODULE, jobfmt(Job)]),
    demonitor(Ref, [flush]),
    case erlang:is_process_alive(Pid) of
        true ->
            ok = mem3_reshard_job_sup:terminate_child(Pid);
        false ->
            ok
    end,
    Job1 = Job#job{pid = undefined, ref = undefined},
    true = ets:insert(?MODULE, Job1),
    Job1.


-spec handle_job_exit(#job{}, term(), #state{}) -> ok.
handle_job_exit(#job{split_state = completed} = Job, normal, State) ->
    couch_log:notice("~p completed job ~s exited", [?MODULE, Job#job.id]),
    Job1 = Job#job{
        pid = undefined,
        ref = undefined,
        job_state = completed,
        update_time = now_sec(),
        state_info = []
    },
    Job2 = update_job_history(Job1),
    ok = mem3_reshard_store:store_job(State, Job2),
    true = ets:insert(?MODULE, Job2),
    ok;

handle_job_exit(#job{job_state = running} = Job, normal, _State) ->
    couch_log:notice("~p running job ~s stopped", [?MODULE, Job#job.id]),
    OldInfo = Job#job.state_info,
    Job1 = Job#job{
        pid = undefined,
        ref = undefined,
        job_state = stopped,
        update_time = now_sec(),
        state_info = info_update(reason, <<"Job stopped">>, OldInfo)
    },
    true = ets:insert(?MODULE, update_job_history(Job1)),
    ok;

handle_job_exit(#job{job_state = running} = Job, shutdown, _State) ->
    couch_log:notice("~p job ~s shutdown", [?MODULE, Job#job.id]),
    OldInfo = Job#job.state_info,
    Job1 = Job#job{
        pid = undefined,
        ref = undefined,
        job_state = stopped,
        update_time = now_sec(),
        state_info = info_update(reason, <<"Job shutdown">>, OldInfo)
    },
    true = ets:insert(?MODULE, update_job_history(Job1)),
    ok;

handle_job_exit(#job{job_state = running} = Job, {shutdown, Msg}, _State) ->
    couch_log:notice("~p job ~s shutdown ~p", [?MODULE, Job#job.id, Msg]),
    OldInfo = Job#job.state_info,
    Job1 = Job#job{
        pid = undefined,
        ref = undefined,
        job_state = stopped,
        update_time = now_sec(),
        state_info = info_update(reason, <<"Job shutdown">>, OldInfo)
    },
    true = ets:insert(?MODULE, update_job_history(Job1)),
    ok;

handle_job_exit(#job{} = Job, Error, State) ->
    couch_log:notice("~p job ~s failed ~p", [?MODULE, Job#job.id, Error]),
    OldInfo = Job#job.state_info,
    Job1 = Job#job{
        pid = undefined,
        ref = undefined,
        job_state = failed,
        update_time = now_sec(),
        state_info = info_update(reason, Error, OldInfo)
    },
    Job2 = update_job_history(Job1),
    ok = mem3_reshard_store:store_job(State, Job2),
    true = ets:insert(?MODULE, Job2),
    ok.


-spec job_by_id(job_id()) -> #job{} | not_found.
job_by_id(Id) ->
    case ets:lookup(?MODULE, Id) of
        [] ->
            not_found;
        [#job{} = Job] ->
            Job
    end.


-spec job_by_pid(pid()) -> {ok, #job{}} | {error, not_found}.
job_by_pid(Pid) when is_pid(Pid) ->
    case ets:match_object(?MODULE, #job{pid=Pid, _='_'}) of
        [] ->
            {error, not_found};
        [#job{} = Job] ->
            {ok, Job}
    end.


-spec state_id() -> binary().
state_id() ->
    Ver = iolist_to_binary(io_lib:format("~3..0B", [?JOB_STATE_VERSION])),
    <<?STATE_PREFIX/binary, Ver/binary>>.


-spec job_id(#job{}) -> binary().
job_id(#job{source = #shard{name = SourceName}}) ->
    HashInput = [SourceName, atom_to_binary(node(), utf8)],
    IdHashList = couch_util:to_hex(crypto:hash(sha256, HashInput)),
    IdHash = iolist_to_binary(IdHashList),
    Prefix = iolist_to_binary(io_lib:format("~3..0B", [?JOB_ID_VERSION])),
    <<Prefix/binary, "-", IdHash/binary>>.


-spec target_shards(#shard{}, split()) -> [#shard{}].
target_shards(#shard{name = Name, range = [B, E], dbname = DbName}, Split) when
        is_integer(Split), Split >= 2, (E - B + 1) >= Split ->
    Ranges = target_ranges([B, E], Split),
    <<"shards/", _:8/binary, "-", _:8/binary, "/", DbAndSuffix/binary>> = Name,
    [DbName, Suffix] = binary:split(DbAndSuffix, <<".">>),
    [build_shard(R, DbName, Suffix) || R <- Ranges].


-spec target_ranges([range_pos()], split()) -> [[range_pos()]].
target_ranges([Begin, End], Split) when (End - Begin + 1) >= Split,
        Split >=2 ->
    Len = End - Begin + 1, % + 1 since intervals are inclusive
    NewLen = Len div Split,
    Rem = Len rem Split,
    Ranges = [[I, I + NewLen - 1] || I <- lists:seq(Begin, End - Rem, NewLen)],
    % Adjust last end to always match the original end to ensure we always
    % cover the whole range. In case when remainder is larger this will make
    % the last range larger. Improve the algorithm later to re-distribute
    % the remainder equally amonst the chunks.
    {BeforeLast, [[BeginLast, _]]} = lists:split(Split - 1, Ranges),
    BeforeLast ++ [[BeginLast, End]].


-spec build_shard([non_neg_integer()], binary(), binary()) -> #shard{}.
build_shard(Range, DbName, Suffix) ->
    Shard = #shard{dbname = DbName, range = Range, node = node()},
    mem3_util:name_shard(Shard, <<".", Suffix/binary>>).


-spec running_jobs() -> [#job{}].
running_jobs() ->
    Pat = #job{job_state = running, _ = '_'},
    ets:match_object(?MODULE, Pat).


-spec info_update(atom(), any(), [tuple()]) -> [tuple()].
info_update(Key, Val, StateInfo) ->
    lists:keystore(Key, 1, StateInfo, {Key, Val}).


-spec info_delete(atom(), [tuple()]) -> [tuple()].
info_delete(Key, StateInfo) ->
    lists:keydelete(Key, 1, StateInfo).


-spec checkpoint_int(#job{}, #state{}) -> #state{}.
checkpoint_int(#job{} = Job, State) ->
    couch_log:debug("~p checkpoint ~s", [?MODULE, jobfmt(Job)]),
    case report_int(Job) of
        ok ->
            ok = mem3_reshard_store:store_job(State, Job),
            ok = mem3_reshard_job:checkpoint_done(Job),
            State;
        not_found ->
            couch_log:error("~p checkpoint couldn't find ~p", [?MODULE, Job]),
            State
    end.


-spec report_int(#job{}) -> ok | not_found.
report_int(Job) ->
    case ets:lookup(?MODULE, Job#job.id) of
        [#job{ref = Ref, pid = CurPid}] ->
            case Job#job.pid =:= CurPid of
                true ->
                    couch_log:debug("~p reported ~s", [?MODULE, jobfmt(Job)]),
                    % Carry over the reference from ets as the #job{} coming
                    % from the job process won't have it's own monitor ref.
                    true = ets:insert(?MODULE, Job#job{ref = Ref}),
                    ok;
                false ->
                    LogMsg = "~p ignoring old job report ~p curr pid:~p",
                    couch_log:warning(LogMsg, [?MODULE, jobfmt(Job), CurPid]),
                    not_found
            end;
        _ ->
            couch_log:error("~p reporting : couldn't find ~p", [?MODULE, Job]),
            not_found
    end.


-spec remove_job_int(#job{}, #state{}) -> ok | {error, not_found}.
remove_job_int(Id, State) ->
    couch_log:notice("~p call remove_job Id:~p", [?MODULE, Id]),
    case job_by_id(Id) of
        #job{} = Job ->
            kill_job_int(Job),
            ok = mem3_reshard_store:delete_job(State, Id),
            ets:delete(?MODULE, Job#job.id),
            ok;
        not_found ->
            {error, not_found}
    end.


% This function is for testing and debugging only
-spec reset_state(#state{}) -> #state{}.
reset_state(#state{} = State) ->
    couch_log:warning("~p resetting state", [?MODULE]),
    ok = mem3_reshard_store:delete_state(State),
    couch_log:warning("~p killing all running jobs", [?MODULE]),
    [kill_job_int(Job) || Job <- running_jobs()],
    ets:delete_all_objects(?MODULE),
    couch_log:warning("~p resetting all job states", [?MODULE]),
    Jobs = mem3_reshard_store:get_jobs(State),
    lists:foldl(fun(#job{id = Id}, StateAcc) ->
        couch_log:warning("~p resetting job state ~p", [?MODULE, Id]),
        ok = mem3_reshard_store:delete_job(StateAcc, Id),
        StateAcc
    end, State, Jobs),
    couch_log:warning("~p resetting state done", [?MODULE]),
    State#state{
        state = running,
        state_info = [],
        update_time = now_sec()
    }.


-spec update_job_history(#job{}) -> #job{}.
update_job_history(#job{job_state = St, update_time = Ts} = Job) ->
    Hist = Job#job.history,
    Reason = case couch_util:get_value(reason, Job#job.state_info) of
        undefined -> null;
        Val -> couch_util:to_binary(Val)
    end,
    Job#job{history = update_history(St, Reason, Ts, Hist)}.


update_history_rev(State, null, Ts, [{_, State, Detail} | Rest]) ->
    % Just updated the detail, state stays the same, no new entry added
    [{Ts, State, Detail} | Rest];

update_history_rev(State, Detail, Ts, [{_, State, Detail} | Rest]) ->
    % State and detail were same as last event, just update the timestamp
    [{Ts, State, Detail} | Rest];

update_history_rev(State, Detail, Ts, [{_, State, Detail} | Rest]) ->
    % State and detail were same as last event, just update the timestamp
    [{Ts, State, Detail} | Rest];

update_history_rev(State, Detail, Ts, History) ->
    [{Ts, State, Detail} | History].


-spec max_history() -> non_neg_integer().
max_history() ->
    config:get_integer("reshard", "max_history", ?DEFAULT_MAX_HISTORY).


-spec maybe_disable(#state{}) -> #state{}.
maybe_disable(#state{} = State) ->
    case is_disabled() of
        true ->
            Reason = <<"Resharding disabled by application level config">>,
            SInfo = State#state.state_info,
            State#state{
                state = stopped,
                state_info = info_update(reason, Reason, SInfo)
            };
        false ->
            State
    end.


-spec jobs_by_db_and_state(binary(), split_state() | '_') -> [job_id()].
jobs_by_db_and_state(Db, State) ->
    DbName = mem3:dbname(Db),
    Pat = #job{
        id = '$1',
        source =#shard{dbname = DbName, _ = '_'},
        job_state = State,
        _ = '_'
    },
    [JobId || [JobId] <- ets:match(?MODULE, Pat)].


-spec db_exists(binary()) -> boolean().
db_exists(Name) ->
    try
        mem3:shards(mem3:dbname(Name)),
        true
    catch
        error:database_does_not_exist ->
            false
    end.


-spec db_monitor(pid()) -> no_return().
db_monitor(Server) ->
    couch_log:notice("~p db monitor ~p starting", [?MODULE, self()]),
    EvtRef = erlang:monitor(process, couch_event_server),
    couch_event:register_all(self()),
    db_monitor_loop(Server, EvtRef).


-spec db_monitor_loop(pid(), reference()) -> no_return().
db_monitor_loop(Server, EvtRef) ->
    receive
        {'$couch_event', DbName, deleted} ->
            case db_exists(DbName) of
                true ->
                    % Could be source shard being deleted during splitting
                    ok;
                false ->
                    case length(jobs_by_db_and_state(DbName, '_')) > 0 of
                        true ->
                            % Notify only if there are jobs with that db
                            gen_server:cast(Server, {db_deleted, DbName});
                        false ->
                            ok
                    end
            end,
            db_monitor_loop(Server, EvtRef);
        {'$couch_event', _, _} ->
            db_monitor_loop(Server, EvtRef);
        {'DOWN', EvtRef, _, _, Info} ->
            couch_log:error("~p db monitor listener died ~p", [?MODULE, Info]),
            exit({db_monitor_died, Info});
        Msg ->
            couch_log:error("~p db monitor unexpected msg ~p", [?MODULE, Msg]),
            db_monitor_loop(Server, EvtRef)
    end.


-spec statefmt(#state{} | term()) -> string().
statefmt(#state{state = StateName}) ->
    Total = ets:info(?MODULE, size),
    Active = mem3_reshard_job_sup:count_children(),
    Msg = "#state{~s total:~B active:~B}",
    Fmt = io_lib:format(Msg, [StateName, Total, Active]),
    lists:flatten(Fmt);

statefmt(State) ->
    Fmt = io_lib:format("<Unknown split state:~p>", [State]),
    lists:flatten(Fmt).


-spec jobfmt(#job{}) -> string().
jobfmt(#job{} = Job) ->
    mem3_reshard_job:jobfmt(Job).