/**
 * 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.ByteArrayOutputStream;
import java.io.IOException;
import java.io.StringReader;
import java.util.Date;

import com.google.common.base.Charsets;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.CoordinatorJobBean;
import org.apache.oozie.ErrorCode;
import org.apache.oozie.client.CoordinatorJob;
import org.apache.oozie.client.OozieClient;
import org.apache.oozie.command.CommandException;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor;
import org.apache.oozie.executor.jpa.JPAExecutorException;
import org.apache.oozie.executor.jpa.CoordJobQueryExecutor.CoordJobQuery;
import org.apache.oozie.service.JPAService;
import org.apache.oozie.service.Services;
import org.apache.oozie.util.ConfigUtils;
import org.apache.oozie.util.LogUtils;
import org.apache.oozie.util.XConfiguration;
import org.apache.oozie.util.XmlUtils;
import org.eclipse.jgit.diff.DiffFormatter;
import org.eclipse.jgit.diff.EditList;
import org.eclipse.jgit.diff.HistogramDiff;
import org.eclipse.jgit.diff.RawText;
import org.eclipse.jgit.diff.RawTextComparator;
import org.jdom.Element;

/**
 * This class provides the functionalities to update coordinator job XML and properties. It uses CoordSubmitXCommand
 * functionality to validate XML and resolve all the variables or properties using job configurations.
 */
public class CoordUpdateXCommand extends CoordSubmitXCommand {

    private final String jobId;
    private boolean showDiff = true;
    private boolean isConfChange = false;

    //This properties are set in coord jobs by bundle. An update command should not overide it.
    final static String[] bundleConfList = new String[] { OozieClient.BUNDLE_ID };

    StringBuffer diff = new StringBuffer();
    CoordinatorJobBean oldCoordJob = null;

    public CoordUpdateXCommand(boolean dryrun, Configuration conf, String jobId) {
        super(dryrun, conf);
        this.jobId = jobId;
        isConfChange = conf.size() != 0;
    }

    public CoordUpdateXCommand(boolean dryrun, Configuration conf, String jobId, boolean showDiff) {
        super(dryrun, conf);
        this.jobId = jobId;
        this.showDiff = showDiff;
        isConfChange = conf.size() != 0;
    }

    @Override
    protected String storeToDB(String xmlElement, Element eJob, CoordinatorJobBean coordJob) throws CommandException {
        check(oldCoordJob, coordJob);

        ConfigUtils.checkAndSetDisallowedProperties(conf,
                this.oldCoordJob.getUser(),
                new CommandException(ErrorCode.E1003,
                        String.format("%s=%s", OozieClient.USER_NAME, conf.get(OozieClient.USER_NAME))),
                true);

        computeDiff(eJob);
        oldCoordJob.setAppPath(conf.get(OozieClient.COORDINATOR_APP_PATH));
        if (isConfChange) {
            oldCoordJob.setConf(XmlUtils.prettyPrint(conf).toString());
        }
        oldCoordJob.setMatThrottling(coordJob.getMatThrottling());
        oldCoordJob.setOrigJobXml(xmlElement);
        oldCoordJob.setConcurrency(coordJob.getConcurrency());
        oldCoordJob.setExecution(coordJob.getExecution());
        oldCoordJob.setTimeout(coordJob.getTimeout());
        oldCoordJob.setJobXml(XmlUtils.prettyPrint(eJob).toString());

        if (!dryrun) {
            oldCoordJob.setLastModifiedTime(new Date());
            // Should log the changes, this should be useful for debugging.
            LOG.info("Coord update changes : " + diff.toString());
            try {
                CoordJobQueryExecutor.getInstance().executeUpdate(CoordJobQuery.UPDATE_COORD_JOB, oldCoordJob);
            }
            catch (JPAExecutorException jpaee) {
                throw new CommandException(jpaee);
            }
        }
        return jobId;
    }

    @Override
    protected void loadState() throws CommandException {
        super.loadState();
        jpaService = Services.get().get(JPAService.class);
        if (jpaService == null) {
            throw new CommandException(ErrorCode.E0610);
        }
        coordJob = new CoordinatorJobBean();
        try {
            oldCoordJob = CoordJobQueryExecutor.getInstance().get(CoordJobQuery.GET_COORD_JOB, jobId);
        }
        catch (JPAExecutorException e) {
            throw new CommandException(e);
        }

        LogUtils.setLogInfo(oldCoordJob);
        if (!isConfChange || StringUtils.isEmpty(conf.get(OozieClient.COORDINATOR_APP_PATH))) {
            try {
                XConfiguration jobConf = new XConfiguration(new StringReader(oldCoordJob.getConf()));

                if (!isConfChange) {
                    conf = jobConf;
                }
                else {
                    for (String bundleConfKey : bundleConfList) {
                        if (jobConf.get(bundleConfKey) != null) {
                            conf.set(bundleConfKey, jobConf.get(bundleConfKey));
                        }
                    }
                    if (StringUtils.isEmpty(conf.get(OozieClient.COORDINATOR_APP_PATH))) {
                        conf.set(OozieClient.COORDINATOR_APP_PATH, jobConf.get(OozieClient.COORDINATOR_APP_PATH));
                    }
                }
            }
            catch (Exception e) {
                throw new CommandException(ErrorCode.E1023, e.getMessage(), e);
            }
        }
        coordJob.setConf(XmlUtils.prettyPrint(conf).toString());
        setJob(coordJob);
        LogUtils.setLogInfo(coordJob);
    }

    @Override
    protected void verifyPrecondition() throws CommandException {
        if (coordJob.getStatus() == CoordinatorJob.Status.SUCCEEDED
                || coordJob.getStatus() == CoordinatorJob.Status.DONEWITHERROR) {
            LOG.info("Can't update coord job. Job has finished processing");
            throw new CommandException(ErrorCode.E1023, "Can't update coord job. Job has finished processing");
        }
    }

    /**
     * Gets the difference of job definition and properties.
     *
     * @param eJob the e job
     * @return the diff
     */
    private void computeDiff(Element eJob) {
        try {
            diff.append("**********Job definition changes**********").append(System.getProperty("line.separator"));
            diff.append(getDiffinGitFormat(oldCoordJob.getJobXml(), XmlUtils.prettyPrint(eJob).toString()));
            diff.append("******************************************").append(System.getProperty("line.separator"));
            diff.append("**********Job conf changes****************").append(System.getProperty("line.separator"));
            if (isConfChange) {
                diff.append(getDiffinGitFormat(oldCoordJob.getConf(), XmlUtils.prettyPrint(conf).toString()));
            }
            else {
                diff.append("No conf update requested").append(System.getProperty("line.separator"));
            }
            diff.append("******************************************").append(System.getProperty("line.separator"));
        }
        catch (IOException e) {
            diff.append("Error computing diff. Error " + e.getMessage());
            LOG.warn("Error computing diff.", e);
        }
    }

    /**
     * Get the differences in git format.
     *
     * @param string1 the string1
     * @param string2 the string2
     * @return the diff
     * @throws IOException Signals that an I/O exception has occurred.
     */
    private String getDiffinGitFormat(String string1, String string2) throws IOException {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        RawText rt1 = new RawText(string1.getBytes(Charsets.UTF_8));
        RawText rt2 = new RawText(string2.getBytes(Charsets.UTF_8));
        EditList diffList = new EditList();
        diffList.addAll(new HistogramDiff().diff(RawTextComparator.DEFAULT, rt1, rt2));
        new DiffFormatter(out).format(diffList, rt1, rt2);
        return out.toString(Charsets.UTF_8.name());
    }

    @Override
    protected String submit() throws CommandException {
        LOG.info("STARTED Coordinator update");
        submitJob();
        LOG.info("ENDED Coordinator update");
        if (showDiff) {
            return diff.toString();
        }
        else {
            return "";
        }
    }

    /**
     * Check. Frequency can't be changed. EndTime can't be changed. StartTime can't be changed. AppName can't be changed
     * Timeunit can't be changed. Timezone can't be changed
     *
     * @param oldCoord the old coord
     * @param newCoord the new coord
     * @throws CommandException the command exception
     */
    public void check(CoordinatorJobBean oldCoord, CoordinatorJobBean newCoord) throws CommandException {
        if (!oldCoord.getFrequency().equals(newCoord.getFrequency())) {
            throw new CommandException(ErrorCode.E1023, "Frequency can't be changed. Old frequency = "
                    + oldCoord.getFrequency() + " new frequency = " + newCoord.getFrequency());
        }

        if (!oldCoord.getEndTime().equals(newCoord.getEndTime())) {
            throw new CommandException(ErrorCode.E1023, "End time can't be changed. Old end time = "
                    + oldCoord.getEndTime() + " new end time = " + newCoord.getEndTime());
        }

        if (!oldCoord.getStartTime().equals(newCoord.getStartTime())) {
            throw new CommandException(ErrorCode.E1023, "Start time can't be changed. Old start time = "
                    + oldCoord.getStartTime() + " new start time = " + newCoord.getStartTime());
        }

        if (!oldCoord.getAppName().equals(newCoord.getAppName())) {
            throw new CommandException(ErrorCode.E1023, "Coord name can't be changed. Old name = "
                    + oldCoord.getAppName() + " new name = " + newCoord.getAppName());
        }

        if (!oldCoord.getTimeUnitStr().equals(newCoord.getTimeUnitStr())) {
            throw new CommandException(ErrorCode.E1023, "Timeunit can't be changed. Old Timeunit = "
                    + oldCoord.getTimeUnitStr() + " new Timeunit = " + newCoord.getTimeUnitStr());
        }

        if (!oldCoord.getTimeZone().equals(newCoord.getTimeZone())) {
            throw new CommandException(ErrorCode.E1023, "TimeZone can't be changed. Old timeZone = "
                    + oldCoord.getTimeZone() + " new timeZone = " + newCoord.getTimeZone());
        }

    }

    @Override
    protected void queueMaterializeTransitionXCommand(String jobId) {
    }

    @Override
    public void notifyParent() throws CommandException {
    }

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

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

    @Override
    public void transitToNext() {
    }

    @Override
    public String getKey() {
        return getName() + "_" + jobId;
    }

    @Override
    public String getDryRun(CoordinatorJobBean job) throws Exception{
        return super.getDryRun(oldCoordJob);
    }
}