/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 */

package org.apache.oozie.command.coord;

import java.io.IOException;
import java.io.StringReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;

import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.CoordinatorActionBean;
import org.apache.oozie.CoordinatorActionInfo;
import org.apache.oozie.CoordinatorJobBean;
import org.apache.oozie.ErrorCode;
import org.apache.oozie.SLAEventBean;
import org.apache.oozie.XException;
import org.apache.oozie.client.CoordinatorAction;
import org.apache.oozie.client.CoordinatorJob;
import org.apache.oozie.client.Job;
import org.apache.oozie.client.SLAEvent.SlaAppType;
import org.apache.oozie.client.rest.RestConstants;
import org.apache.oozie.command.CommandException;
import org.apache.oozie.command.PreconditionException;
import org.apache.oozie.command.RerunTransitionXCommand;
import org.apache.oozie.command.bundle.BundleStatusUpdateXCommand;
import org.apache.oozie.coord.CoordELFunctions;
import org.apache.oozie.coord.CoordUtils;
import org.apache.oozie.dependency.URIHandler;
import org.apache.oozie.dependency.URIHandler.Context;
import org.apache.oozie.dependency.URIHandlerException;
import org.apache.oozie.executor.jpa.BatchQueryExecutor;
import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
import org.apache.oozie.executor.jpa.CoordActionQueryExecutor.CoordActionQuery;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.service.EventHandlerService;
import org.apache.oozie.service.Services;
import org.apache.oozie.service.URIHandlerService;
import org.apache.oozie.sla.SLAOperations;
import org.apache.oozie.sla.service.SLAService;
import org.apache.oozie.util.InstrumentUtils;
import org.apache.oozie.util.LogUtils;
import org.apache.oozie.util.ParamChecker;
import org.apache.oozie.util.StatusUtils;
import org.apache.oozie.util.XConfiguration;
import org.apache.oozie.util.XLog;
import org.apache.oozie.util.XmlUtils;
import org.apache.oozie.util.db.SLADbOperations;
import org.jdom.Element;
import org.jdom.JDOMException;

/**
 * Rerun coordinator actions by a list of dates or ids. User can specify if refresh or noCleanup.
 * <p>
 * The "rerunType" can be set as {@link RestConstants#JOB_COORD_SCOPE_DATE} or {@link RestConstants#JOB_COORD_SCOPE_ACTION}.
 * <p>
 * The "refresh" is used to indicate if user wants to refresh an action's input and output events.
 * <p>
 * The "noCleanup" is used to indicate if user wants to cleanup output events for given rerun actions
 */
@SuppressWarnings("deprecation")
public class CoordRerunXCommand extends RerunTransitionXCommand<CoordinatorActionInfo> {

    public static final String RERUN_CONF = "rerunConf";
    private String rerunType;
    private String scope;
    private boolean refresh;
    private boolean noCleanup;
    private CoordinatorJobBean coordJob = null;
    protected boolean prevPending;
    private boolean failed;
    private Configuration actionRunConf;

    /**
     * The constructor for class {@link CoordRerunXCommand}
     *
     * @param jobId the job id
     * @param rerunType rerun type {@link RestConstants#JOB_COORD_SCOPE_DATE} or {@link RestConstants#JOB_COORD_SCOPE_ACTION}
     * @param scope the rerun scope for given rerunType separated by ","
     * @param refresh true if user wants to refresh input/output dataset urls
     * @param noCleanup false if user wants to cleanup output events for given rerun actions
     * @param failed true if user wants to rerun only failed nodes
     * @param actionRunConf configuration values for actions
     */
    public CoordRerunXCommand(String jobId, String rerunType, String scope, boolean refresh, boolean noCleanup,
                              boolean failed, Configuration actionRunConf) {
        super("coord_rerun", "coord_rerun", 1);
        this.jobId = ParamChecker.notEmpty(jobId, "jobId");
        this.rerunType = ParamChecker.notEmpty(rerunType, "rerunType");
        this.scope = ParamChecker.notEmpty(scope, "scope");
        this.refresh = refresh;
        this.noCleanup = noCleanup;
        this.failed = failed;
        this.actionRunConf = actionRunConf;
    }

    /**
     * Check if all given actions are eligible to rerun.
     *
     * @param coordActions list of CoordinatorActionBean
     * @return true if all actions are eligible to rerun
     */
    private static boolean checkAllActionsRunnable(List<CoordinatorActionBean> coordActions) {
        Objects.requireNonNull(coordActions, "Coord actions to be rerun cannot be null");
        boolean ret = false;
        for (CoordinatorActionBean coordAction : coordActions) {
            ret = true;
            if (!coordAction.isTerminalStatus()) {
                ret = false;
                break;
            }
        }
        return ret;
    }

    /**
     * Cleanup output-events directories
     *
     * @param eAction coordinator action xml
     */
    @SuppressWarnings("unchecked")
    private void cleanupOutputEvents(Element eAction, Configuration coordJobConf, Map<String, Context> uriHandlerContextMap)
            throws CommandException {
        Element outputList = eAction.getChild("output-events", eAction.getNamespace());
        if (outputList != null) {

            for (Element data : (List<Element>) outputList.getChildren("data-out", eAction.getNamespace())) {
                String nocleanup = data.getAttributeValue("nocleanup");
                if (data.getChild("uris", data.getNamespace()) != null
                        && (nocleanup == null || !nocleanup.equals("true"))) {
                    String uris = data.getChild("uris", data.getNamespace()).getTextTrim();
                    if (uris != null) {
                        String[] uriArr = uris.split(CoordELFunctions.INSTANCE_SEPARATOR);
                        try {
                            for (String uriStr : uriArr) {
                                URI uri = new URI(uriStr);
                                URIHandler handler = Services.get().get(URIHandlerService.class).getURIHandler(uri);
                                String schemeWithAuthority = uri.getScheme() + "://" + uri.getAuthority();
                                if (!uriHandlerContextMap.containsKey(schemeWithAuthority)) {
                                    Context context = handler.getContext(uri, coordJobConf, coordJob.getUser(), false);
                                    uriHandlerContextMap.put(schemeWithAuthority, context);
                                }
                                handler.delete(uri, uriHandlerContextMap.get(schemeWithAuthority));
                                LOG.info("Cleanup the output data " + uri.toString());
                            }
                        }
                        catch (URISyntaxException e) {
                            throw new CommandException(ErrorCode.E0907, e.getMessage());
                        }
                        catch (URIHandlerException e) {
                            throw new CommandException(ErrorCode.E0907, e.getMessage());
                        }
                    }
                }
            }

        }
        else {
            LOG.info("No output-events defined in coordinator xml. Therefore nothing to cleanup");
        }
    }

    /**
     * Refresh an action's input and ouput events.
     *
     * @param coordJob coordinator job bean
     * @param coordAction coordinator action bean
     * @throws Exception thrown if failed to materialize coordinator action
     */
    private void refreshAction(CoordinatorJobBean coordJob, CoordinatorActionBean coordAction) throws Exception {
        Configuration jobConf = null;
        try {
            jobConf = new XConfiguration(new StringReader(coordJob.getConf()));
        }
        catch (IOException ioe) {
            LOG.warn("Configuration parse error. read from DB :" + coordJob.getConf(), ioe);
            throw new CommandException(ErrorCode.E1005, ioe.getMessage(), ioe);
        }
        String jobXml = coordJob.getJobXml();
        Element eJob = XmlUtils.parseXml(jobXml);
        Date actualTime = new Date();
        String actionXml = CoordCommandUtils.materializeOneInstance(jobId, dryrun, (Element) eJob.clone(), coordAction
                .getNominalTime(), actualTime, coordAction.getActionNumber(), jobConf, coordAction);
        LOG.debug("Refresh Action actionId=" + coordAction.getId() + ", actionXml="
                + XmlUtils.prettyPrint(actionXml).toString());
        coordAction.setActionXml(actionXml);
    }

    /**
     * Update an action into database table
     *
     * @param coordJob coordinator job bean
     * @param coordAction coordinator action bean
     * @throws Exception thrown failed to update coordinator action bean or unable to write sla registration event
     */
    private void updateAction(CoordinatorJobBean coordJob, CoordinatorActionBean coordAction)
            throws Exception {
        LOG.debug("updateAction for actionId=" + coordAction.getId());
        if (coordAction.getStatus() == CoordinatorAction.Status.TIMEDOUT) {
            LOG.debug("Updating created time for TIMEDOUT action id =" + coordAction.getId());
            coordAction.setCreatedTime(new Date());
        }
        coordAction.setStatus(CoordinatorAction.Status.WAITING);
        if(!failed) {
            coordAction.setExternalId(null);
        }
        coordAction.setExternalStatus(null);
        coordAction.setRerunTime(new Date());
        coordAction.setLastModifiedTime(new Date());
        coordAction.setErrorCode("");
        coordAction.setErrorMessage("");

        // Pushing the configuration which passed through rerun.
        if(actionRunConf != null && actionRunConf.size() > 0) {
            Configuration createdConf = null;
            if(coordAction.getCreatedConf() != null ) {
                createdConf = new XConfiguration(new StringReader(coordAction.getCreatedConf()));
            } else {
                createdConf = new Configuration();
            }
            createdConf.set(RERUN_CONF, XmlUtils.prettyPrint(actionRunConf).toString());
            coordAction.setCreatedConf(XmlUtils.prettyPrint(createdConf).toString());
        }
        updateList.add(new UpdateEntry<CoordActionQuery>(CoordActionQuery.UPDATE_COORD_ACTION_RERUN, coordAction));
        writeActionRegistration(coordAction.getActionXml(), coordAction, coordJob.getUser(), coordJob.getGroup());
    }

    /**
     * Create SLA RegistrationEvent
     *
     * @param actionXml action xml
     * @param actionBean coordinator action bean
     * @param user user name
     * @param group group name
     * @throws Exception thrown if unable to write sla registration event
     */
    private void writeActionRegistration(String actionXml, CoordinatorActionBean actionBean, String user, String group)
            throws Exception {
        Element eAction = XmlUtils.parseXml(actionXml);
        Element eSla = eAction.getChild("action", eAction.getNamespace()).getChild("info", eAction.getNamespace("sla"));
        SLAEventBean slaEvent = SLADbOperations.createSlaRegistrationEvent(eSla, actionBean.getId(),
                SlaAppType.COORDINATOR_ACTION, user, group, LOG);
        if(slaEvent != null) {
            insertList.add(slaEvent);
        }
    }

    @Override
    public String getEntityKey() {
        return jobId;
    }

    @Override
    protected boolean isLockRequired() {
        return true;
    }

    @Override
    protected void loadState() throws CommandException {
        try {
            coordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, jobId);
            prevPending = coordJob.isPending();
        }
        catch (JPAExecutorException je) {
            throw new CommandException(je);
        }
        LogUtils.setLogInfo(coordJob);
    }

    @Override
    protected void verifyPrecondition() throws CommandException, PreconditionException {
        BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, coordJob.getStatus());

        // no actions have been created for PREP job
        if (coordJob.getStatus() == CoordinatorJob.Status.PREP || coordJob.getStatus() == CoordinatorJob.Status.IGNORED) {
            LOG.info("CoordRerunXCommand is not able to run, job status=" + coordJob.getStatus() + ", jobid=" + jobId);
            // Call the parent so the pending flag is reset and state transition
            // of bundle can happen
            if (coordJob.getBundleId() != null) {
                bundleStatusUpdate.call();
            }
            if (coordJob.getStatus() == CoordinatorJob.Status.PREP) {
                throw new CommandException(ErrorCode.E1018,
                        "coordinator job is PREP so no actions are materialized to rerun!");
            }
            else {
                throw new CommandException(ErrorCode.E1018,
                        "coordinator job is IGNORED, please change it to RUNNING before rerunning actions");
            }
        }
    }

    @Override
    protected void eagerLoadState() throws CommandException {
        try {
            coordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, jobId);
        }
        catch (JPAExecutorException e) {
            throw new CommandException(e);
        }
    }

    @Override
    protected void eagerVerifyPrecondition() throws CommandException, PreconditionException {
        verifyPrecondition();
    }

    @Override
    public void rerunChildren() throws CommandException {
        boolean isError = false;
        try {
            CoordinatorActionInfo coordInfo = null;
            InstrumentUtils.incrJobCounter(getName(), 1, getInstrumentation());
            List<CoordinatorActionBean> coordActions = CoordUtils.getCoordActions(rerunType, jobId, scope, false);
            if (checkAllActionsRunnable(coordActions)) {
                Map<String, Context> uriHandlerContextMap = new HashMap<String, Context>();
                Configuration coordJobConf = null;
                try {
                    coordJobConf = new XConfiguration(new StringReader(coordJob.getConf()));
                }
                catch (IOException e) {
                    throw new CommandException(ErrorCode.E0907, "failed to read coord job conf to clean up output data");
                }
                try {
                    for (CoordinatorActionBean coordAction : coordActions) {
                        String actionXml = coordAction.getActionXml();
                        // Cleanup activity should not run when failed option has been provided
                        if (!noCleanup && !failed) {
                            Element eAction = XmlUtils.parseXml(actionXml);
                            cleanupOutputEvents(eAction, coordJobConf, uriHandlerContextMap);
                        }
                        if (refresh) {
                            refreshAction(coordJob, coordAction);
                        }
                        updateAction(coordJob, coordAction);
                        if (SLAService.isEnabled()) {
                            SLAOperations.updateRegistrationEvent(coordAction.getId());
                        }
                        queue(new CoordActionNotificationXCommand(coordAction), 100);
                        queue(new CoordActionInputCheckXCommand(coordAction.getId(), coordAction.getJobId()), 100);
                        if (coordAction.getPushMissingDependencies() != null) {
                            queue(new CoordPushDependencyCheckXCommand(coordAction.getId(), true), 100);
                        }
                    }
                }
                finally {
                    Iterator<Entry<String, Context>> itr = uriHandlerContextMap.entrySet().iterator();
                    while (itr.hasNext()) {
                        Entry<String, Context> entry = itr.next();
                        entry.getValue().destroy();
                        itr.remove();
                    }
                }
            }
            else {
                isError = true;
                throw new CommandException(ErrorCode.E1018, "part or all actions are not eligible to rerun!");
            }
            coordInfo = new CoordinatorActionInfo(coordActions);

            ret = coordInfo;
        }
        catch (XException xex) {
            isError = true;
            throw new CommandException(xex);
        }
        catch (JDOMException jex) {
            isError = true;
            throw new CommandException(ErrorCode.E0700, jex.getMessage(), jex);
        }
        catch (Exception ex) {
            isError = true;
            throw new CommandException(ErrorCode.E1018, ex.getMessage(), ex);
        }
        finally{
            if(isError){
                transitToPrevious();
            }
        }
    }

    @Override
    public Job getJob() {
        return coordJob;
    }

    @Override
    public void notifyParent() throws CommandException {
        //update bundle action
        if (getPrevStatus() != null && coordJob.getBundleId() != null) {
            BundleStatusUpdateXCommand bundleStatusUpdate = new BundleStatusUpdateXCommand(coordJob, getPrevStatus());
            bundleStatusUpdate.call();
        }
    }

    @Override
    public void updateJob() {
        if (getPrevStatus()!= null){
            Job.Status coordJobStatus = getPrevStatus();
            if(coordJobStatus.equals(Job.Status.PAUSED) || coordJobStatus.equals(Job.Status.PAUSEDWITHERROR)) {
                coordJob.setStatus(coordJobStatus);
            }
            if (prevPending) {
                coordJob.setPending();
            } else {
                coordJob.resetPending();
            }
        }
        updateList.add(new UpdateEntry<CoordJobQuery>(CoordJobQuery.UPDATE_COORD_JOB_STATUS_PENDING, coordJob));
    }

    @Override
    public void performWrites() throws CommandException {
        try {
            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, updateList, null);
            if (EventHandlerService.isEnabled()) {
                generateEvents(coordJob, null);
            }
        }
        catch (JPAExecutorException e) {
            throw new CommandException(e);
        }
    }

    @Override
    public XLog getLog() {
        return LOG;
    }

    @Override
    public final void transitToNext() {
        prevStatus = coordJob.getStatus();
        if (prevStatus == CoordinatorJob.Status.SUCCEEDED || prevStatus == CoordinatorJob.Status.PAUSED
                || prevStatus == CoordinatorJob.Status.SUSPENDED || prevStatus == CoordinatorJob.Status.RUNNING) {
            coordJob.setStatus(Job.Status.RUNNING);
        }
        else {
            // Check for backward compatibility for Oozie versions (3.2 and before)
            // when RUNNINGWITHERROR, SUSPENDEDWITHERROR and
            // PAUSEDWITHERROR is not supported
            coordJob.setStatus(StatusUtils.getStatusIfBackwardSupportTrue(CoordinatorJob.Status.RUNNINGWITHERROR));
        }
        // used for backward support of coordinator 0.1 schema
        coordJob.setStatus(StatusUtils.getStatusForCoordRerun(coordJob, prevStatus));
        coordJob.setPending();
    }

    private final void transitToPrevious() throws CommandException {
        coordJob.setStatus(getPrevStatus());
        if (!prevPending) {
            coordJob.resetPending();
        }
        else {
            coordJob.setPending();
        }
    }
}