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

import org.apache.commons.lang.StringUtils;
import org.apache.falcon.ExtensionHandler;
import org.apache.falcon.FalconException;
import org.apache.falcon.LifeCycle;
import org.apache.falcon.client.AbstractFalconClient;
import org.apache.falcon.client.FalconCLIException;
import org.apache.falcon.entity.EntityUtil;
import org.apache.falcon.entity.store.ConfigurationStore;
import org.apache.falcon.entity.v0.Entity;
import org.apache.falcon.entity.v0.EntityType;
import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.entity.v0.feed.Feed;
import org.apache.falcon.entity.v0.process.Cluster;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.entity.v0.process.Validity;
import org.apache.falcon.extensions.store.ExtensionStore;
import org.apache.falcon.resource.APIResult;
import org.apache.falcon.resource.EntityList;
import org.apache.falcon.resource.EntitySummaryResult;
import org.apache.falcon.resource.FeedInstanceResult;
import org.apache.falcon.resource.FeedLookupResult;
import org.apache.falcon.resource.InstanceDependencyResult;
import org.apache.falcon.resource.InstancesResult;
import org.apache.falcon.resource.InstancesSummaryResult;
import org.apache.falcon.resource.LineageGraphResult;
import org.apache.falcon.resource.SchedulableEntityInstanceResult;
import org.apache.falcon.resource.TriageResult;
import org.apache.falcon.resource.admin.AdminResource;
import org.apache.falcon.util.DateUtil;
import org.apache.falcon.workflow.WorkflowEngineFactory;
import org.apache.falcon.workflow.engine.AbstractWorkflowEngine;
import org.apache.hadoop.security.authorize.AuthorizationException;
import org.json.simple.JSONValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.util.TreeMap;
import java.util.ArrayList;
import java.util.Properties;
import java.util.List;
import java.util.Map;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.TimeZone;
import java.util.SortedMap;

/**
 * Client for Falcon Unit.
 */
public class FalconUnitClient extends AbstractFalconClient {

    private static final Logger LOG = LoggerFactory.getLogger(FalconUnitClient.class);
    protected static final int XML_DEBUG_LEN = 10 * 1024;

    private static final String DEFAULT_ORDER_BY = "status";
    private static final String DEFAULT_SORTED_ORDER = "asc";

    protected ConfigurationStore configStore;
    private AbstractWorkflowEngine workflowEngine;
    private LocalSchedulableEntityManager localSchedulableEntityManager;
    private LocalInstanceManager localInstanceManager;
    private LocalExtensionManager localExtensionManager;


    public FalconUnitClient() throws FalconException {
        configStore = ConfigurationStore.get();
        workflowEngine = WorkflowEngineFactory.getWorkflowEngine();
        localSchedulableEntityManager = new LocalSchedulableEntityManager();
        localInstanceManager = new LocalInstanceManager();
        localExtensionManager = new LocalExtensionManager();
    }

    public ConfigurationStore getConfigStore() {
        return this.configStore;
    }


    /**
     * Submit a new entity. Entities can be of type feed, process or data end
     * points. Entity definitions are validated structurally against schema and
     * subsequently for other rules before they are admitted into the system
     *
     * @param type     entity type
     * @param filePath path for the definition of entity
     * @return boolean
     */
    @Override
    public APIResult submit(String type, String filePath, String doAsUser) {

        try {
            return localSchedulableEntityManager.submit(type, filePath, doAsUser);
        } catch (FalconException | IOException e) {
            throw new FalconCLIException("FAILED", e);
        }
    }

    /**
     * Schedules submitted entity.
     *
     * @param entityType entity Type
     * @param entityName entity name
     * @param cluster    cluster on which it has to be scheduled
     * @return
     * @throws FalconException
     */
    @Override
    public APIResult schedule(EntityType entityType, String entityName, String cluster,
                              Boolean skipDryRun, String doAsUser, String properties) {
        try {
            return localSchedulableEntityManager.schedule(entityType, entityName, skipDryRun, properties);
        } catch (FalconException | AuthorizationException e) {
            throw new FalconCLIException(e);
        }
    }

    @Override
    public APIResult delete(EntityType entityType, String entityName, String colo) {
        return localSchedulableEntityManager.delete(entityType, entityName, colo);
    }

    @Override
    public APIResult validate(String entityType, String filePath, Boolean skipDryRun,
                              String doAsUser) {
        try {
            return localSchedulableEntityManager.validate(entityType, filePath, skipDryRun, doAsUser);
        } catch (FalconException e) {
            throw new FalconCLIException(e);
        }
    }

    @Override
    public APIResult update(String entityType, String entityName, String filePath,
                            Boolean skipDryRun, String doAsUser) {
        try {
            return localSchedulableEntityManager.update(entityType, entityName, filePath,
                    skipDryRun, "local", doAsUser);
        } catch (FalconException e) {
            throw new FalconCLIException(e);
        }
    }

    @Override
    public Entity getDefinition(String entityType, String entityName, String doAsUser) {
        String entity = localSchedulableEntityManager.getEntityDefinition(entityType, entityName);
        return Entity.fromString(EntityType.getEnum(entityType), entity);
    }

    //SUSPEND CHECKSTYLE CHECK ParameterNumberCheck
    @Override
    public InstancesResult getStatusOfInstances(String type, String entity, String start, String end, String colo,
                                                List<LifeCycle> lifeCycles, String filterBy, String orderBy,
                                                String sortOrder, Integer offset, Integer numResults, String doAsUser,
                                                Boolean allAttempts) {
        if (orderBy == null) {
            orderBy = DEFAULT_ORDER_BY;
        }
        if (sortOrder == null) {
            sortOrder = DEFAULT_SORTED_ORDER;
        }
        if (offset == null) {
            offset = 0;
        }
        if (numResults == null) {
            numResults = 1;
        }
        return localInstanceManager.getStatusOfInstances(type, entity, start, end, colo, lifeCycles, filterBy, orderBy,
                sortOrder, offset, numResults, allAttempts);

    }

    /**
     * Schedules an submitted process entity immediately.
     *
     * @param entityName   entity name
     * @param startTime    start time for process while scheduling
     * @param numInstances numInstances of process to be scheduled
     * @param cluster      cluster on which process to be scheduled
     * @return boolean
     */
    public APIResult schedule(EntityType entityType, String entityName, String startTime, int numInstances,
                              String cluster, Boolean skipDryRun, String properties) {
        try {
            FalconUnitHelper.checkSchedulableEntity(entityType.toString());
            Entity entity = EntityUtil.getEntity(entityType, entityName);
            boolean clusterPresent = checkAndUpdateCluster(entity, entityType, cluster);
            if (!clusterPresent) {
                LOG.warn("Cluster is not registered with this entity " + entityName);
                return new APIResult(APIResult.Status.FAILED, entity + "Cluster is not registered with this entity "
                        + entityName);
            }
            if (StringUtils.isNotEmpty(startTime) && entityType == EntityType.PROCESS) {
                updateStartAndEndTime((Process) entity, startTime, numInstances, cluster);
            }
            workflowEngine.schedule(entity, skipDryRun, EntityUtil.getPropertyMap(properties));
            LOG.info(entityName + " is scheduled successfully");
            return new APIResult(APIResult.Status.SUCCEEDED, entity + "(" + "PROCESS" + ") scheduled successfully");
        } catch (FalconException e) {
            throw new FalconCLIException("FAILED", e);
        }
    }

    /**
     * Instance status for a given nominalTime.
     *
     * @param entityType  entity type
     * @param entityName  entity name
     * @param nominalTime nominal time of process
     * @return InstancesResult.WorkflowStatus
     */
    public InstancesResult.WorkflowStatus getInstanceStatus(String entityType, String entityName,
                                                            String nominalTime) throws Exception {
        Date startTime = SchemaHelper.parseDateUTC(nominalTime);
        Date endTimeDate = DateUtil.getNextMinute(startTime);
        String endTime = DateUtil.getDateFormatFromTime(endTimeDate.getTime());
        InstancesResult instancesResult = getStatusOfInstances(entityType, entityName, nominalTime, endTime, null,
                null, null, null, null, null, null, null, null);
        if (instancesResult.getInstances() != null && instancesResult.getInstances().length > 0
                && instancesResult.getInstances()[0] != null) {
            LOG.info("Instance status is " + instancesResult.getInstances()[0].getStatus());
            return instancesResult.getInstances()[0].getStatus();
        }
        return null;
    }

    @Override
    public APIResult suspend(EntityType entityType, String entityName, String colo, String doAsUser) {
        return localSchedulableEntityManager.suspend(entityType.name(), entityName, colo);
    }

    @Override
    public APIResult resume(EntityType entityType, String entityName, String colo, String doAsUser) {
        return localSchedulableEntityManager.resume(entityType.name(), entityName, colo);
    }

    @Override
    public APIResult getStatus(EntityType entityType, String entityName, String colo, String doAsUser,
                               boolean showScheduler) {
        return localSchedulableEntityManager.getStatus(entityType.name(), entityName, colo, showScheduler);
    }

    @Override
    public APIResult submitAndSchedule(String entityType, String filePath, Boolean skipDryRun, String doAsUser,
                                       String properties) {
        try {
            return localSchedulableEntityManager.submitAndSchedule(entityType, filePath, skipDryRun, doAsUser,
                    properties);
        } catch (FalconException | IOException e) {
            throw new FalconCLIException(e);
        }
    }

    @Override
    public APIResult registerExtension(String extensionName, String packagePath, String description) {
        return localExtensionManager.registerExtensionMetadata(extensionName, packagePath, description);
    }

    @Override
    public APIResult unregisterExtension(String extensionName) {
        return localExtensionManager.unRegisterExtension(extensionName);
    }

    @Override
    public APIResult submitExtensionJob(String extensionName, String jobName, String configPath, String doAsUser) {

        InputStream configStream = getServletInputStream(configPath);
        try {
            SortedMap<EntityType, List<Entity>> entityMap = getEntityTypeListMap(extensionName, jobName, configStream);
            return localExtensionManager.submitExtensionJob(extensionName, jobName, configStream, entityMap);
        } catch (FalconException | IOException e) {
            throw new FalconCLIException("Failed in submitting extension job " + jobName);
        }
    }

    private SortedMap<EntityType, List<Entity>> getEntityTypeListMap(String extensionName, String jobName,
                                                                     InputStream configStream) {
        List<Entity> entities = getEntities(extensionName, jobName, configStream);
        List<Entity> feeds = new ArrayList<>();
        List<Entity> processes = new ArrayList<>();
        for (Entity entity : entities) {
            if (EntityType.FEED.equals(entity.getEntityType())) {
                feeds.add(entity);
            } else if (EntityType.PROCESS.equals(entity.getEntityType())) {
                processes.add(entity);
            }
        }
        SortedMap<EntityType, List<Entity>> entityMap = new TreeMap<>();
        entityMap.put(EntityType.PROCESS, processes);
        entityMap.put(EntityType.FEED, feeds);
        return entityMap;
    }

    private List<Entity> getEntities(String extensionName, String jobName, InputStream configStream) {
        String packagePath = ExtensionStore.getMetaStore().getDetail(extensionName).getLocation();
        List<Entity> entities;
        try {
            entities = ExtensionHandler.loadAndPrepare(extensionName, jobName, configStream,
                    packagePath);
        } catch (FalconException | IOException e) {
            throw new FalconCLIException("Failed in generating entities" + jobName);
        }
        return entities;
    }

    @Override
    public APIResult submitAndScheduleExtensionJob(String extensionName, String jobName, String configPath,
                                                   String doAsUser) {
        InputStream configStream = getServletInputStream(configPath);
        try {
            SortedMap<EntityType, List<Entity>> entityMap = getEntityTypeListMap(extensionName, jobName, configStream);
            return localExtensionManager.submitAndSchedulableExtensionJob(extensionName, jobName, configStream,
                    entityMap);
        } catch (FalconException | IOException e) {
            throw new FalconCLIException("Failed in submitting extension job " + jobName);
        }
    }

    @Override
    public APIResult updateExtensionJob(String jobName, String configPath, String doAsUser) {
        InputStream configStream = getServletInputStream(configPath);
        try {
            String extensionName = ExtensionStore.getMetaStore().getExtensionJobDetails(jobName).getExtensionName();
            SortedMap<EntityType, List<Entity>> entityMap = getEntityTypeListMap(extensionName, jobName, configStream);
            return localExtensionManager.updateExtensionJob(extensionName, jobName, configStream,
                    entityMap);
        } catch (FalconException | IOException e) {
            throw new FalconCLIException("Failed in updating the extension job:" + jobName);
        }
    }

    @Override
    public APIResult deleteExtensionJob(String jobName, String doAsUser) {
        try {
            return localExtensionManager.deleteExtensionJob(jobName);
        } catch (FalconException | IOException e) {
            throw new FalconCLIException("Failed to delete the extension job:" + jobName);
        }
    }


    @Override
    public APIResult getExtensionJobDetails(final String jobName) {
        return localExtensionManager.getExtensionJobDetails(jobName);
    }

    @Override
    public APIResult getExtensionDetail(String extensionName) {
        return localExtensionManager.getExtensionDetails(extensionName);
    }

    @Override
    public APIResult enumerateExtensions() {
        return localExtensionManager.getExtensions();
    }

    @Override
    public EntityList getEntityList(String entityType, String fields, String nameSubsequence, String tagKeywords,
                                    String filterBy, String filterTags, String orderBy, String sortOrder,
                                    Integer offset, Integer numResults, String doAsUser) {
        return localSchedulableEntityManager.getEntityList(fields, nameSubsequence, tagKeywords, entityType, filterTags,
                filterBy, orderBy, sortOrder, offset, numResults, doAsUser);
    }

    @Override
    public EntitySummaryResult getEntitySummary(String entityType, String cluster, String start, String end,
                                                String fields, String filterBy, String filterTags, String orderBy,
                                                String sortOrder, Integer offset, Integer numResults,
                                                Integer numInstances, String doAsUser) {
        return localSchedulableEntityManager.getEntitySummary(entityType, cluster, start, end, fields, filterBy,
                filterTags, orderBy, sortOrder, offset, numResults, numInstances, doAsUser);
    }

    @Override
    public APIResult touch(String entityType, String entityName, String colo, Boolean skipDryRun,
                           String doAsUser) {
        return localSchedulableEntityManager.touch(entityType, entityName, colo, skipDryRun);
    }

    public InstancesResult killInstances(String type, String entity, String start, String end, String colo,
                                         String clusters, String sourceClusters, List<LifeCycle> lifeCycles,
                                         String doAsUser) throws UnsupportedEncodingException {
        Properties props = getProperties(clusters, sourceClusters);
        return localInstanceManager.killInstance(props, type, entity, start, end, colo, lifeCycles);
    }

    public InstancesResult suspendInstances(String type, String entity, String start, String end, String colo,
                                            String clusters, String sourceClusters, List<LifeCycle> lifeCycles,
                                            String doAsUser) throws UnsupportedEncodingException {
        Properties props = getProperties(clusters, sourceClusters);
        return localInstanceManager.suspendInstance(props, type, entity, start, end, colo, lifeCycles);
    }

    public InstancesResult resumeInstances(String type, String entity, String start, String end, String colo,
                                           String clusters, String sourceClusters, List<LifeCycle> lifeCycles,
                                           String doAsUser) throws UnsupportedEncodingException {
        Properties props = getProperties(clusters, sourceClusters);
        return localInstanceManager.resumeInstance(props, type, entity, start, end, colo, lifeCycles);
    }

    public InstancesResult rerunInstances(String type, String entity, String start, String end, String filePath,
                                          String colo, String clusters, String sourceClusters,
                                          List<LifeCycle> lifeCycles, Boolean isForced, String doAsUser) throws
            IOException {
        Properties props = getProperties(clusters, sourceClusters);
        return localInstanceManager.reRunInstance(type, entity, start, end, props, colo, lifeCycles, isForced);
    }

    public InstancesSummaryResult getSummaryOfInstances(String type, String entity, String start, String end,
                                                        String colo, List<LifeCycle> lifeCycles, String filterBy,
                                                        String orderBy, String sortOrder, String doAsUser) {
        if (StringUtils.isBlank(orderBy)) {
            orderBy = DEFAULT_ORDER_BY;
        }
        if (StringUtils.isBlank(sortOrder)) {
            sortOrder = DEFAULT_SORTED_ORDER;
        }
        return localInstanceManager.getSummary(type, entity, start, end, colo, lifeCycles, filterBy, orderBy,
                sortOrder);
    }

    public FeedInstanceResult getFeedListing(String type, String entity, String start, String end, String colo,
                                             String doAsUser) {
        return localInstanceManager.getListing(type, entity, start, end, colo);
    }

    public InstancesResult getLogsOfInstances(String type, String entity, String start, String end, String colo,
                                              String runId, List<LifeCycle> lifeCycles, String filterBy,
                                              String orderBy, String sortOrder, Integer offset, Integer numResults,
                                              String doAsUser) {
        return localInstanceManager.getLogs(type, entity, start, end, colo, runId, lifeCycles, filterBy, orderBy,
                sortOrder, offset, numResults);
    }

    public InstancesResult getParamsOfInstance(String type, String entity, String start, String colo,
                                               List<LifeCycle> lifeCycles, String doAsUser) throws
            UnsupportedEncodingException {
        return localInstanceManager.getInstanceParams(type, entity, start, colo, lifeCycles);
    }
    //RESUME CHECKSTYLE CHECK ParameterNumberCheck

    public InstanceDependencyResult getInstanceDependencies(String entityType, String entityName, String instanceTime,
                                                            String colo) {
        return localInstanceManager.getInstanceDependencies(entityType, entityName, instanceTime, colo);
    }

    @Override
    public String getVersion(String doAsUser) {
        AdminResource resource = new AdminResource();
        AdminResource.PropertyList propertyList = resource.getVersion();
        Map<String, String> version = new LinkedHashMap<>();
        List<String> list = new ArrayList<>();
        for (AdminResource.Property property : propertyList.properties) {
            Map<String, String> map = new LinkedHashMap<>();
            map.put("key", property.key);
            map.put("value", property.value);
            list.add(JSONValue.toJSONString(map));
        }
        version.put("properties", list.toString());
        return version.toString();
    }

    @Override
    public SchedulableEntityInstanceResult getFeedSlaMissPendingAlerts(String entityType, String entityName,
                                                                       String start, String end, String colo) {
        return null;
    }

    @Override
    public FeedLookupResult reverseLookUp(String entityType, String path, String doAs) {
        return null;
    }

    @Override
    public EntityList getDependency(String entityType, String entityName, String doAs) {
        return null;
    }

    @Override
    public TriageResult triage(String name, String entityName, String start, String colo) {
        return null;
    }
    // SUSPEND CHECKSTYLE CHECK ParameterNumberCheck
    @Override
    public InstancesResult getRunningInstances(String type, String entity, String colo, List<LifeCycle> lifeCycles,
                                               String filterBy, String orderBy, String sortOrder,
                                               Integer offset, Integer numResults, String doAsUser) {
        return null;
    }
    // RESUME CHECKSTYLE CHECK ParameterNumberCheck
    @Override
    public FeedInstanceResult getFeedInstanceListing(String type, String entity, String start, String end,
                                                     String colo, String doAsUser) {
        return null;
    }

    @Override
    public int getStatus(String doAsUser) {
        return 200;
    }

    @Override
    public String getThreadDump(String doAs) {
        return "";
    }

    @Override
    public LineageGraphResult getEntityLineageGraph(String pipeline, String doAs) {
        return null;
    }

    @Override
    public String getDimensionList(String dimensionType, String cluster, String doAs) {
        return null;
    }

    @Override
    public String getReplicationMetricsDimensionList(String schedEntityType, String schedEntityName,
                                                     Integer numResults, String doAs) {
        return null;
    }

    @Override
    public String getDimensionRelations(String dimensionType, String dimensionName, String doAs) {
        return null;
    }

    @Override
    public String getVertex(String id, String doAs) {
        return null;
    }

    @Override
    public String getVertices(String key, String value, String doAs) {
        return null;
    }

    @Override
    public String getVertexEdges(String id, String direction, String doAs) {
        return null;
    }

    @Override
    public String getEdge(String id, String doAs) {
        return null;
    }

    private boolean checkAndUpdateCluster(Entity entity, EntityType entityType, String cluster) {
        if (entityType == EntityType.FEED) {
            return checkAndUpdateFeedClusters(entity, cluster);
        } else if (entityType == EntityType.PROCESS) {
            return checkAndUpdateProcessClusters(entity, cluster);
        } else {
            throw new IllegalArgumentException("entity type {} is not supported " + entityType);
        }
    }

    private boolean checkAndUpdateProcessClusters(Entity entity, String cluster) {
        Process processEntity = (Process) entity;
        List<Cluster> clusters = processEntity.getClusters().getClusters();
        List<Cluster> newClusters = new ArrayList<>();
        if (clusters != null) {
            for (Cluster processCluster : clusters) {
                if (processCluster.getName().equalsIgnoreCase(cluster)) {
                    newClusters.add(processCluster);
                }
            }
        }
        if (newClusters.isEmpty()) {
            LOG.warn("Cluster is not registered with this entity " + entity.getName());
            return false;
        }
        processEntity.getClusters().getClusters().removeAll(clusters);
        processEntity.getClusters().getClusters().addAll(newClusters);
        return true;
    }

    private boolean checkAndUpdateFeedClusters(Entity entity, String cluster) {
        Feed feedEntity = (Feed) entity;
        List<org.apache.falcon.entity.v0.feed.Cluster> clusters = feedEntity.getClusters().getClusters();
        List<org.apache.falcon.entity.v0.feed.Cluster> newClusters = new ArrayList<>();
        if (clusters != null) {
            for (org.apache.falcon.entity.v0.feed.Cluster feedClusters : clusters) {
                if (feedClusters.getName().equalsIgnoreCase(cluster)) {
                    newClusters.add(feedClusters);
                }
            }
        }
        if (newClusters.isEmpty()) {
            LOG.warn("Cluster is not registered with this entity " + entity.getName());
            return false;
        }
        feedEntity.getClusters().getClusters().removeAll(clusters);
        feedEntity.getClusters().getClusters().addAll(newClusters);
        return true;
    }

    private void updateStartAndEndTime(Process processEntity, String startTimeStr, int numInstances, String cluster) {
        List<Cluster> clusters = processEntity.getClusters().getClusters();
        if (clusters != null) {
            for (Cluster processCluster : clusters) {
                if (processCluster.getName().equalsIgnoreCase(cluster)) {
                    Validity validity = new Validity();
                    Date startTime = SchemaHelper.parseDateUTC(startTimeStr);
                    validity.setStart(startTime);
                    Date endTime = EntityUtil.getNextInstanceTime(startTime, processEntity.getFrequency(),
                            TimeZone.getTimeZone("UTC"), numInstances);
                    validity.setEnd(endTime);
                    processCluster.setValidity(validity);
                }
            }
        }
    }

    private Properties getProperties(String clusters, String sourceClusters) {
        Properties props = new Properties();
        if (StringUtils.isNotEmpty(clusters)) {
            props.setProperty(FALCON_INSTANCE_ACTION_CLUSTERS, clusters);
        }
        if (StringUtils.isNotEmpty(sourceClusters)) {
            props.setProperty(FALCON_INSTANCE_SOURCE_CLUSTERS, sourceClusters);
        }
        return props;
    }
}