/**
 * 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.servlet;

import java.io.IOException;
import java.io.Writer;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.oozie.DagEngine;
import org.apache.oozie.DagEngineException;
import org.apache.oozie.ErrorCode;
import org.apache.oozie.WorkflowActionBean;
import org.apache.oozie.WorkflowJobBean;
import org.apache.oozie.WorkflowsInfo;
import org.apache.oozie.client.WorkflowJob;
import org.apache.oozie.client.rest.JMSConnectionInfoBean;
import org.apache.oozie.client.rest.RestConstants;
import org.apache.oozie.service.DagEngineService;
import org.apache.oozie.util.XmlUtils;

public class MockDagEngineService extends DagEngineService {
    public static final String JOB_ID = "job-";
    public static final String JOB_ID_END = "-W";
    public static final String ACTION_ID = "action-";
    public static final String EXT_ID = "ext-";
    public static final String WORKFLOW_APP = "<workflow-app/>";
    public static final String CONFIGURATION = "<configuration/>";
    public static final String GROUP = "group";
    public static final String USER = "user";

    public static final String LOG = "log";

    public static Configuration submittedConf;
    public static String did = null;
    public static String user = null;
    public static Properties properties;
    public static List<WorkflowJob> workflows;
    public static List<Boolean> started;
    public static final int INIT_WF_COUNT = 4;

    static {
        reset();
    }

    public static void reset() {
        user = null;
        did = null;
        properties = null;
        workflows = new ArrayList<WorkflowJob>();
        started = new ArrayList<Boolean>();
        for (int i = 0; i < INIT_WF_COUNT; i++) {
            workflows.add(createDummyWorkflow(i));
            started.add(false);
        }
    }

    @Override
    public DagEngine getSystemDagEngine() {
        return new MockDagEngine();
    }

    @Override
    public DagEngine getDagEngine(String user) {
        return new MockDagEngine(user);
    }

    private static class MockDagEngine extends DagEngine {

        public MockDagEngine() {
        }

        public MockDagEngine(String user) {
            super(user);
        }

        @Override
        public String submitJob(Configuration conf, boolean startJob) throws DagEngineException {
            did = "submit";
            submittedConf = conf;
            int idx = workflows.size();
            WorkflowJob job = createDummyWorkflow(idx, XmlUtils.prettyPrint(conf).toString());
            workflows.add(job);
            started.add(startJob);
            MockDagEngineService.user = getUser();
            return job.getId();
        }

        @Override
        public String submitHttpJob(Configuration conf, String jobType) throws DagEngineException {
            if (jobType.equals("pig")) {
                did = "submitPig";
            }
            else if (jobType.equals("mapreduce")) {
                did = "submitMR";
            }
            else if (jobType.equals("sqoop")) {
                did = "submitSqoop";
            }
            int idx = workflows.size();
            WorkflowJob job = createDummyWorkflow(idx, XmlUtils.prettyPrint(conf).toString());
            workflows.add(job);
            started.add(true);
            return job.getId();
        }

        @Override
        public void start(String jobId) throws DagEngineException {
            did = RestConstants.JOB_ACTION_START;
            int idx = validateWorkflowIdx(jobId);
            started.set(idx, true);
        }

        @Override
        public void resume(String jobId) throws DagEngineException {
            did = RestConstants.JOB_ACTION_RESUME;
            int idx = validateWorkflowIdx(jobId);
            started.set(idx, true);
        }

        @Override
        public void suspend(String jobId) throws DagEngineException {
            did = RestConstants.JOB_ACTION_SUSPEND;
            int idx = validateWorkflowIdx(jobId);
            started.set(idx, false);
        }

        @Override
        public void kill(String jobId) throws DagEngineException {
            did = RestConstants.JOB_ACTION_KILL;
            int idx = validateWorkflowIdx(jobId);
            started.set(idx, false);
        }

        @Override
        public WorkflowsInfo suspendJobs(String filter, int start, int len) throws DagEngineException {
            did = RestConstants.JOBS;
            return new WorkflowsInfo(new ArrayList<WorkflowJobBean>(), 0, 0, 0);
        }

        @Override
        public WorkflowsInfo resumeJobs(String filter, int start, int len) throws DagEngineException {
            did = RestConstants.JOBS;
            return new WorkflowsInfo(new ArrayList<WorkflowJobBean>(), 0, 0, 0);
        }

        @Override
        public WorkflowsInfo killJobs(String filter, int start, int len) throws DagEngineException {
            did = RestConstants.JOBS;
            return new WorkflowsInfo(new ArrayList<WorkflowJobBean>(), 0, 0, 0);
        }

        @Override
        public void reRun(String jobId, Configuration conf) throws DagEngineException {
            did = RestConstants.JOB_ACTION_RERUN;
            int idx = validateWorkflowIdx(jobId);
            started.set(idx, true);
        }

        @Override
        public void processCallback(String actionId, String externalStatus, Properties actionData)
                throws DagEngineException {
            if (actionId.equals("ok") && externalStatus.equals("ok")) {
                properties = actionData;
                return;
            }
            throw new DagEngineException(ErrorCode.ETEST, actionId);
        }

        @Override
        public WorkflowJob getJob(String jobId) throws DagEngineException {
            did = RestConstants.JOB_SHOW_INFO;
            int idx = validateWorkflowIdx(jobId);
            return workflows.get(idx);
        }

        @Override
        public WorkflowJob getJob(String jobId, int start, int length) throws DagEngineException {
            return getJob(jobId);
        }

        @Override
        public String getJobStatus(String jobId) throws DagEngineException {
            did = RestConstants.JOB_SHOW_STATUS;
            int idx = validateWorkflowIdx(jobId);
            return workflows.get(idx).getStatus().toString();
        }

        @Override
        public String getDefinition(String jobId) throws DagEngineException {
            did = RestConstants.JOB_SHOW_DEFINITION;
            int idx = validateWorkflowIdx(jobId);
            //FIXME:
            return WORKFLOW_APP;
        }

        @Override
        public void streamLog(String jobId, Writer writer, Map<String, String[]> params) throws IOException, DagEngineException {
            did = RestConstants.JOB_SHOW_LOG;
            validateWorkflowIdx(jobId);
            writer.write(LOG);
        }

        @Override
        public WorkflowsInfo getJobs(String filter, int start, int len) throws DagEngineException {
            parseFilter(filter);
            did = RestConstants.JOBS_FILTER_PARAM;
            return new WorkflowsInfo((List) workflows, start, len, workflows.size());
        }

        @Override
        public String getJobIdForExternalId(String externalId) throws DagEngineException {
            did = RestConstants.JOBS_EXTERNAL_ID_PARAM;
            return (externalId.equals("external-valid")) ? "id-valid" : null;
        }

        public List<Map<String, String>> getWorkflowActionRetries(String actionId) throws DagEngineException {
            did = RestConstants.JOB_SHOW_ACTION_RETRIES_PARAM;
            return new ArrayList<Map<String, String>>();
        }

        private int validateWorkflowIdx(String jobId) throws DagEngineException {
            int idx = -1;
            try {
                if (jobId.endsWith(JOB_ID_END)) {
                    jobId = jobId.replace(JOB_ID, "");
                    jobId = jobId.replace(JOB_ID_END, "");
                }
                else {
                    jobId = jobId.replace(JOB_ID, "");
                }
                idx = Integer.parseInt(jobId);
            }
            catch (Exception e) {
                throw new DagEngineException(ErrorCode.ETEST, jobId);
            }

            if (idx >= workflows.size()) {
                throw new DagEngineException(ErrorCode.ETEST, jobId);
            }

            return idx;
        }
    }

    private static JMSConnectionInfoBean createDummyJMSConnectionInfo() {
        JMSConnectionInfoBean jmsBean = new JMSConnectionInfoBean();
        Properties jmsProps = new Properties();
        jmsProps.setProperty("k1", "v1");
        jmsProps.setProperty("k2", "v2");
        jmsBean.setJNDIProperties(jmsProps);
        jmsBean.setTopicPrefix("topicPrefix");
        return jmsBean;
    }

    private static WorkflowJob createDummyWorkflow(int idx) {
        WorkflowJobBean workflow = new WorkflowJobBean();
        workflow.setId(JOB_ID + idx + JOB_ID_END);
        workflow.setAppPath("hdfs://blah/blah/" + idx + "-blah");
        workflow.setStatus((idx % 2) == 0 ? WorkflowJob.Status.RUNNING : WorkflowJob.Status.SUCCEEDED);
        workflow.setRun(idx);
        workflow.setCreatedTime(new Date());
        workflow.setStartTime(new Date());
        workflow.setEndTime((idx % 2) == 0 ? null : (new Date()));
        workflow.setConf(CONFIGURATION);
        workflow.setAppName("workflow-" + idx);
        workflow.setGroup(GROUP);
        workflow.setUser(USER);

        List<WorkflowActionBean> actions = new ArrayList<WorkflowActionBean>();
        for (int i = 0; i < idx; i++) {
            actions.add(createDummyAction(i));
        }

        workflow.setActions(actions);
        return workflow;
    }

    private static WorkflowJob createDummyWorkflow(int idx, String conf) {
        WorkflowJobBean workflow = new WorkflowJobBean();
        workflow.setId(JOB_ID + idx + JOB_ID_END);
        workflow.setAppPath("hdfs://blah/blah/" + idx + "-blah");
        workflow.setStatus((idx % 2) == 0 ? WorkflowJob.Status.RUNNING : WorkflowJob.Status.SUCCEEDED);
        workflow.setRun(idx);
        workflow.setCreatedTime(new Date());
        workflow.setStartTime(new Date());
        workflow.setEndTime((idx % 2) == 0 ? null : (new Date()));
        workflow.setConf(conf);
        workflow.setAppName("workflow-" + idx);
        workflow.setGroup(GROUP);
        workflow.setUser(USER);

        List<WorkflowActionBean> actions = new ArrayList<WorkflowActionBean>();
        for (int i = 0; i < idx; i++) {
            actions.add(createDummyAction(i));
        }

        workflow.setActions(actions);
        return workflow;
    }

    private static WorkflowActionBean createDummyAction(int idx) {
        WorkflowActionBean action = new WorkflowActionBean();
        int mod = idx % 5;
        action.setId(ACTION_ID + idx);
        action.setExternalId(EXT_ID + idx);
        action.setConsoleUrl("http://blah:blah/blah/" + idx);
        action.setConf("<configuration/>");
        action.setData(null);
        action.setStartTime(new Date());
        action.setEndTime(new Date());
        action.setErrorInfo(null, null);
        action.setExternalStatus((idx % 2) == 0 ? "RUNNING" : "OK");
        action.setName(ACTION_ID + idx);
        action.setRetries(idx);
        action.setTrackerUri("http://trackerhost:blah/blah/" + idx);
        action.setTransition("OK");
        switch (mod) {
            case 0: {
                action.setType("hadoop");
                break;
            }
            case 1: {
                action.setType("fs");
                break;
            }
            case 2: {
                action.setType("pig");
                break;
            }
            case 3: {
                action.setType("ssh");
                break;
            }
            case 4: {
                action.setType("decision");
                break;
            }
        }

        return action;
    }
}