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

import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.codec.binary.Hex;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.falcon.FalconException;
import org.apache.falcon.Tag;
import org.apache.falcon.entity.WorkflowNameBuilder.WorkflowName;
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.Frequency;
import org.apache.falcon.entity.v0.SchemaHelper;
import org.apache.falcon.entity.v0.cluster.Cluster;
import org.apache.falcon.entity.v0.feed.ClusterType;
import org.apache.falcon.entity.v0.feed.Feed;
import org.apache.falcon.entity.v0.process.*;
import org.apache.falcon.entity.v0.process.Process;
import org.apache.falcon.hadoop.HadoopClientFactory;
import org.apache.falcon.util.DeploymentUtil;
import org.apache.falcon.util.RuntimeProperties;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.reflect.Method;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * Helper to get entity object.
 */
public final class EntityUtil {
    public static final Logger LOG = LoggerFactory.getLogger(EntityUtil.class);

    private static final long MINUTE_IN_MS = 60000L;
    private static final long HOUR_IN_MS = 3600000L;
    private static final long DAY_IN_MS = 86400000L;
    private static final long MONTH_IN_MS = 2592000000L;

    public static final String PROCESS_CHECKSUM_FILE = "checksums";
    public static final String PROCESS_USER_DIR = "user";
    public static final String PROCESS_USERLIB_DIR = "userlib";
    public static final String SUCCEEDED_FILE_NAME = "_SUCCESS";

    private EntityUtil() {}

    public static <T extends Entity> T getEntity(EntityType type, String entityName) throws FalconException {
        ConfigurationStore configStore = ConfigurationStore.get();
        T entity = configStore.get(type, entityName);
        if (entity == null) {
            throw new EntityNotRegisteredException(entityName + " (" + type + ") not found");
        }
        return entity;
    }

    public static <T extends Entity> T getEntity(String type, String entityName) throws FalconException {
        EntityType entityType;
        try {
            entityType = EntityType.valueOf(type.toUpperCase());
        } catch (IllegalArgumentException e) {
            throw new FalconException("Invalid entity type: " + type, e);
        }
        return getEntity(entityType, entityName);
    }

    public static TimeZone getTimeZone(String tzId) {
        if (tzId == null) {
            throw new IllegalArgumentException("Invalid TimeZone: Cannot be null.");
        }
        TimeZone tz = TimeZone.getTimeZone(tzId);
        if (!tzId.equals("GMT") && tz.getID().equals("GMT")) {
            throw new IllegalArgumentException("Invalid TimeZone: " + tzId);
        }
        return tz;
    }

    public static Date getEndTime(Entity entity, String cluster) {
        if (entity.getEntityType() == EntityType.PROCESS) {
            return getEndTime((Process) entity, cluster);
        } else {
            return getEndTime((Feed) entity, cluster);
        }
    }

    public static Date parseDateUTC(String dateStr) throws FalconException {
        try {
            return SchemaHelper.parseDateUTC(dateStr);
        } catch (Exception e) {
            throw new FalconException(e);
        }
    }

    public static Date getStartTime(Entity entity, String cluster) {
        if (entity.getEntityType() == EntityType.PROCESS) {
            return getStartTime((Process) entity, cluster);
        } else {
            return getStartTime((Feed) entity, cluster);
        }
    }

    public static Date getEndTime(Process process, String cluster) {
        org.apache.falcon.entity.v0.process.Cluster processCluster = ProcessHelper.getCluster(process, cluster);
        return processCluster.getValidity().getEnd();
    }

    public static Date getStartTime(Process process, String cluster) {
        org.apache.falcon.entity.v0.process.Cluster processCluster = ProcessHelper.getCluster(process, cluster);
        return processCluster.getValidity().getStart();
    }

    public static Date getEndTime(Feed feed, String cluster) {
        org.apache.falcon.entity.v0.feed.Cluster clusterDef = FeedHelper.getCluster(feed, cluster);
        return clusterDef.getValidity().getEnd();
    }

    public static Date getStartTime(Feed feed, String cluster) {
        org.apache.falcon.entity.v0.feed.Cluster clusterDef = FeedHelper.getCluster(feed, cluster);
        return clusterDef.getValidity().getStart();
    }

    public static int getParallel(Entity entity) {
        if (entity.getEntityType() == EntityType.PROCESS) {
            return getParallel((Process) entity);
        } else {
            return getParallel((Feed) entity);
        }
    }

    public static void setStartDate(Entity entity, String cluster, Date startDate) {
        if (entity.getEntityType() == EntityType.PROCESS) {
            setStartDate((Process) entity, cluster, startDate);
        } else {
            setStartDate((Feed) entity, cluster, startDate);
        }
    }

    public static void setEndTime(Entity entity, String cluster, Date endDate) {
        if (entity.getEntityType() == EntityType.PROCESS) {
            setEndTime((Process) entity, cluster, endDate);
        } else {
            setEndTime((Feed) entity, cluster, endDate);
        }
    }

    public static void setParallel(Entity entity, int parallel) {
        if (entity.getEntityType() == EntityType.PROCESS) {
            setParallel((Process) entity, parallel);
        } else {
            setParallel((Feed) entity, parallel);
        }
    }

    public static int getParallel(Process process) {
        return process.getParallel();
    }

    public static void setStartDate(Process process, String cluster, Date startDate) {
        org.apache.falcon.entity.v0.process.Cluster processCluster = ProcessHelper.getCluster(process, cluster);
        processCluster.getValidity().setStart(startDate);
    }

    public static void setParallel(Process process, int parallel) {
        process.setParallel(parallel);
    }

    public static void setEndTime(Process process, String cluster, Date endDate) {
        org.apache.falcon.entity.v0.process.Cluster processCluster = ProcessHelper.getCluster(process, cluster);
        processCluster.getValidity().setEnd(endDate);
    }

    public static int getParallel(Feed feed) {
        return 1;
    }

    public static void setStartDate(Feed feed, String cluster, Date startDate) {
        org.apache.falcon.entity.v0.feed.Cluster clusterDef = FeedHelper.getCluster(feed, cluster);
        clusterDef.getValidity().setStart(startDate);
    }

    public static void setEndTime(Feed feed, String cluster, Date endDate) {
        org.apache.falcon.entity.v0.feed.Cluster clusterDef = FeedHelper.getCluster(feed, cluster);
        clusterDef.getValidity().setStart(endDate);
    }

    public static void setParallel(Feed feed, int parallel) {
    }

    public static Frequency getFrequency(Entity entity) {
        if (entity.getEntityType() == EntityType.PROCESS) {
            return getFrequency((Process) entity);
        } else {
            return getFrequency((Feed) entity);
        }
    }

    public static Frequency getFrequency(Process process) {
        return process.getFrequency();
    }

    public static Frequency getFrequency(Feed feed) {
        return feed.getFrequency();
    }

    public static TimeZone getTimeZone(Entity entity) {
        if (entity.getEntityType() == EntityType.PROCESS) {
            return getTimeZone((Process) entity);
        } else {
            return getTimeZone((Feed) entity);
        }
    }

    public static TimeZone getTimeZone(Process process) {
        return process.getTimezone();
    }

    public static TimeZone getTimeZone(Feed feed) {
        return feed.getTimezone();
    }

    public static Date getNextStartTime(Date startTime, Frequency frequency, TimeZone timezone, Date now) {
        if (startTime.after(now)) {
            return startTime;
        }

        Calendar startCal = Calendar.getInstance(timezone);
        startCal.setTime(startTime);

        int count = 0;
        switch (frequency.getTimeUnit()) {
        case months:
            count = (int) ((now.getTime() - startTime.getTime()) / MONTH_IN_MS);
            break;
        case days:
            count = (int) ((now.getTime() - startTime.getTime()) / DAY_IN_MS);
            break;
        case hours:
            count = (int) ((now.getTime() - startTime.getTime()) / HOUR_IN_MS);
            break;
        case minutes:
            count = (int) ((now.getTime() - startTime.getTime()) / MINUTE_IN_MS);
            break;
        default:
        }

        final int freq = frequency.getFrequencyAsInt();
        if (count > 2) {
            startCal.add(frequency.getTimeUnit().getCalendarUnit(), ((count - 2) / freq) * freq);
        }
        while (startCal.getTime().before(now)) {
            startCal.add(frequency.getTimeUnit().getCalendarUnit(), freq);
        }
        return startCal.getTime();
    }

    public static int getInstanceSequence(Date startTime, Frequency frequency, TimeZone tz, Date instanceTime) {
        if (startTime.after(instanceTime)) {
            return -1;
        }

        Calendar startCal = Calendar.getInstance(tz);
        startCal.setTime(startTime);

        int count = 0;
        switch (frequency.getTimeUnit()) {
        case months:
            count = (int) ((instanceTime.getTime() - startTime.getTime()) / MONTH_IN_MS);
            break;
        case days:
            count = (int) ((instanceTime.getTime() - startTime.getTime()) / DAY_IN_MS);
            break;
        case hours:
            count = (int) ((instanceTime.getTime() - startTime.getTime()) / HOUR_IN_MS);
            break;
        case minutes:
            count = (int) ((instanceTime.getTime() - startTime.getTime()) / MINUTE_IN_MS);
            break;
        default:
        }

        final int freq = frequency.getFrequencyAsInt();
        if (count > 2) {
            startCal.add(frequency.getTimeUnit().getCalendarUnit(), (count / freq) * freq);
            count = (count / freq);
        } else {
            count = 0;
        }
        while (startCal.getTime().before(instanceTime)) {
            startCal.add(frequency.getTimeUnit().getCalendarUnit(), freq);
            count++;
        }
        return count + 1;
    }

    public static String md5(Entity entity) throws FalconException {
        return new String(Hex.encodeHex(DigestUtils.md5(stringOf(entity))));
    }

    public static boolean equals(Entity lhs, Entity rhs) throws FalconException {
        return equals(lhs, rhs, null);
    }

    public static boolean equals(Entity lhs, Entity rhs, String[] filterProps) throws FalconException {
        if (lhs == null && rhs == null) {
            return true;
        }
        if (lhs == null || rhs == null) {
            return false;
        }

        if (lhs.equals(rhs)) {
            String lhsString = stringOf(lhs, filterProps);
            String rhsString = stringOf(rhs, filterProps);
            return lhsString.equals(rhsString);
        } else {
            return false;
        }
    }

    public static String stringOf(Entity entity) throws FalconException {
        return stringOf(entity, null);
    }

    private static String stringOf(Entity entity, String[] filterProps) throws FalconException {
        Map<String, String> map = new HashMap<String, String>();
        mapToProperties(entity, null, map, filterProps);
        List<String> keyList = new ArrayList<String>(map.keySet());
        Collections.sort(keyList);
        StringBuilder builer = new StringBuilder();
        for (String key : keyList) {
            builer.append(key).append('=').append(map.get(key)).append('\n');
        }
        return builer.toString();
    }

    @SuppressWarnings("rawtypes")
    private static void mapToProperties(Object obj, String name, Map<String, String> propMap, String[] filterProps)
        throws FalconException {

        if (obj == null) {
            return;
        }

        if (filterProps != null && name != null) {
            for (String filter : filterProps) {
                if (name.matches(filter.replace(".", "\\.").replace("[", "\\[").replace("]", "\\]"))) {
                    return;
                }
            }
        }

        if (Date.class.isAssignableFrom(obj.getClass())) {
            propMap.put(name, SchemaHelper.formatDateUTC((Date) obj));
        } else if (obj.getClass().getPackage().getName().equals("java.lang")) {
            propMap.put(name, String.valueOf(obj));
        } else if (TimeZone.class.isAssignableFrom(obj.getClass())) {
            propMap.put(name, ((TimeZone) obj).getID());
        } else if (Enum.class.isAssignableFrom(obj.getClass())) {
            propMap.put(name, ((Enum) obj).name());
        } else if (List.class.isAssignableFrom(obj.getClass())) {
            List list = (List) obj;
            for (int index = 0; index < list.size(); index++) {
                mapToProperties(list.get(index), name + "[" + index + "]", propMap, filterProps);
            }
        } else {
            try {
                Method method = obj.getClass().getDeclaredMethod("toString");
                propMap.put(name, (String) method.invoke(obj));
            } catch (NoSuchMethodException e) {
                try {
                    Map map = PropertyUtils.describe(obj);
                    for (Object entry : map.entrySet()) {
                        String key = (String)((Map.Entry)entry).getKey();
                        if (!key.equals("class")) {
                            mapToProperties(map.get(key), name != null ? name + "." + key : key, propMap,
                                    filterProps);
                        }
                    }
                } catch (Exception e1) {
                    throw new FalconException(e1);
                }
            } catch (Exception e) {
                throw new FalconException(e);
            }
        }
    }

    public static WorkflowName getWorkflowName(Tag tag, List<String> suffixes,
                                               Entity entity) {
        WorkflowNameBuilder<Entity> builder = new WorkflowNameBuilder<Entity>(
                entity);
        builder.setTag(tag);
        builder.setSuffixes(suffixes);
        return builder.getWorkflowName();
    }

    public static WorkflowName getWorkflowName(Tag tag, Entity entity) {
        return getWorkflowName(tag, null, entity);
    }

    public static WorkflowName getWorkflowName(Entity entity) {
        return getWorkflowName(null, null, entity);
    }

    public static String getWorkflowNameSuffix(String workflowName,
                                               Entity entity) throws FalconException {
        WorkflowNameBuilder<Entity> builder = new WorkflowNameBuilder<Entity>(
                entity);
        return builder.getWorkflowSuffixes(workflowName).replaceAll("_", "");
    }

    public static Tag getWorkflowNameTag(String workflowName, Entity entity) {
        WorkflowNameBuilder<Entity> builder = new WorkflowNameBuilder<Entity>(
                entity);
        return builder.getWorkflowTag(workflowName);
    }

    public static List<String> getWorkflowNames(Entity entity, String cluster) {
        switch(entity.getEntityType()) {
        case FEED:
            return Arrays.asList(getWorkflowName(Tag.RETENTION, entity).toString(),
                getWorkflowName(Tag.REPLICATION, entity).toString());

        case PROCESS:
            return Arrays.asList(getWorkflowName(Tag.DEFAULT, entity).toString());

        default:
        }
        throw new IllegalArgumentException("Unhandled type: " + entity.getEntityType());
    }

    public static <T extends Entity> T getClusterView(T entity, String clusterName) {
        switch (entity.getEntityType()) {
        case CLUSTER:
            return entity;

        case FEED:
            Feed feed = (Feed) entity.copy();
            org.apache.falcon.entity.v0.feed.Cluster feedCluster = FeedHelper.getCluster(feed, clusterName);
            Iterator<org.apache.falcon.entity.v0.feed.Cluster> itr = feed.getClusters().getClusters().iterator();
            while (itr.hasNext()) {
                org.apache.falcon.entity.v0.feed.Cluster cluster = itr.next();
                //In addition to retaining the required clster, retain the sources clusters if this is the target
                // cluster
                //1. Retain cluster if cluster n
                if (!(cluster.getName().equals(clusterName)
                        || (feedCluster.getType() == ClusterType.TARGET
                        && cluster.getType() == ClusterType.SOURCE))) {
                    itr.remove();
                }
            }
            return (T) feed;

        case PROCESS:
            Process process = (Process) entity.copy();
            Iterator<org.apache.falcon.entity.v0.process.Cluster> procItr =
                process.getClusters().getClusters().iterator();
            while (procItr.hasNext()) {
                org.apache.falcon.entity.v0.process.Cluster cluster = procItr.next();
                if (!cluster.getName().equals(clusterName)) {
                    procItr.remove();
                }
            }
            return (T) process;
        default:
        }
        throw new UnsupportedOperationException("Not supported for entity type " + entity.getEntityType());
    }

    public static Set<String> getClustersDefined(Entity entity) {
        Set<String> clusters = new HashSet<String>();
        switch (entity.getEntityType()) {
        case CLUSTER:
            clusters.add(entity.getName());
            break;

        case FEED:
            Feed feed = (Feed) entity;
            for (org.apache.falcon.entity.v0.feed.Cluster cluster : feed.getClusters().getClusters()) {
                clusters.add(cluster.getName());
            }
            break;

        case PROCESS:
            Process process = (Process) entity;
            for (org.apache.falcon.entity.v0.process.Cluster cluster : process.getClusters().getClusters()) {
                clusters.add(cluster.getName());
            }
            break;
        default:
        }
        return clusters;
    }

    public static Set<String> getClustersDefinedInColos(Entity entity) {
        Set<String> entityClusters = EntityUtil.getClustersDefined(entity);
        if (DeploymentUtil.isEmbeddedMode()) {
            return entityClusters;
        }

        Set<String> myClusters = DeploymentUtil.getCurrentClusters();
        Set<String> applicableClusters = new HashSet<String>();
        for (String cluster : entityClusters) {
            if (myClusters.contains(cluster)) {
                applicableClusters.add(cluster);
            }
        }
        return applicableClusters;
    }

    public static Retry getRetry(Entity entity) throws FalconException {
        switch (entity.getEntityType()) {
        case FEED:
            if (!RuntimeProperties.get()
                    .getProperty("feed.retry.allowed", "true")
                    .equalsIgnoreCase("true")) {
                return null;
            }
            Retry retry = new Retry();
            retry.setAttempts(Integer.parseInt(RuntimeProperties.get()
                    .getProperty("feed.retry.attempts", "3")));
            retry.setDelay(new Frequency(RuntimeProperties.get().getProperty(
                    "feed.retry.frequency", "minutes(5)")));
            retry.setPolicy(PolicyType.fromValue(RuntimeProperties.get()
                    .getProperty("feed.retry.policy", "exp-backoff")));
            return retry;
        case PROCESS:
            Process process = (Process) entity;
            return process.getRetry();
        default:
            throw new FalconException("Cannot create Retry for entity:" + entity.getName());
        }
    }

    //Staging path that stores scheduler configs like oozie coord/bundle xmls, parent workflow xml
    //Each entity update creates a new staging path
    //Base staging path is the base path for all staging dirs
    public static Path getBaseStagingPath(org.apache.falcon.entity.v0.cluster.Cluster cluster, Entity entity) {
        return new Path(ClusterHelper.getLocation(cluster, "staging"),
                "falcon/workflows/" + entity.getEntityType().name().toLowerCase() + "/" + entity.getName());
    }

    //Creates new staging path for entity schedule/update
    //Staging path containd md5 of the cluster view of the entity. This is required to check if update is required
    public static Path getNewStagingPath(org.apache.falcon.entity.v0.cluster.Cluster cluster, Entity entity)
        throws FalconException {
        Entity clusterView = getClusterView(entity, cluster.getName());
        return new Path(getBaseStagingPath(cluster, entity),
            md5(clusterView) + "_" + String.valueOf(System.currentTimeMillis()));
    }

    //Returns all staging paths for the entity
    public static FileStatus[] getAllStagingPaths(org.apache.falcon.entity.v0.cluster.Cluster cluster,
        Entity entity)
        throws FalconException {
        Path basePath = getBaseStagingPath(cluster, entity);
        FileSystem fs = HadoopClientFactory.get().createFileSystem(ClusterHelper.getConfiguration(cluster));
        try {
            FileStatus[] filesArray = fs.listStatus(basePath, new PathFilter() {
                @Override
                public boolean accept(Path path) {
                    if (path.getName().equals("logs")) {
                        return false;
                    }
                    return true;
                }
            });

            return filesArray;

        } catch (FileNotFoundException e) {
            LOG.info("Staging path " + basePath + " doesn't exist, entity is not scheduled");
            //Staging path doesn't exist if entity is not scheduled
        } catch (IOException e) {
            throw new FalconException(e);
        }
        return null;
    }

    public static LateProcess getLateProcess(Entity entity)
        throws FalconException {

        switch (entity.getEntityType()) {
        case FEED:
            if (!RuntimeProperties.get().getProperty("feed.late.allowed", "true").equalsIgnoreCase("true")) {
                return null;
            }

            LateProcess lateProcess = new LateProcess();
            lateProcess.setDelay(new Frequency(RuntimeProperties.get().getProperty("feed.late.frequency", "hours(3)")));
            lateProcess.setPolicy(
                    PolicyType.fromValue(RuntimeProperties.get().getProperty("feed.late.policy", "exp-backoff")));
            LateInput lateInput = new LateInput();
            lateInput.setInput(entity.getName());
            //TODO - Assuming the late workflow is not used
            lateInput.setWorkflowPath("ignore.xml");
            lateProcess.getLateInputs().add(lateInput);
            return lateProcess;
        case PROCESS:
            Process process = (Process) entity;
            return process.getLateProcess();
        default:
            throw new FalconException("Cannot create Late Process for entity:" + entity.getName());
        }
    }

    public static Path getLogPath(org.apache.falcon.entity.v0.cluster.Cluster cluster, Entity entity) {
        return new Path(getBaseStagingPath(cluster, entity), "logs");
    }

    public static String fromUTCtoURIDate(String utc) throws FalconException {
        DateFormat utcFormat = new SimpleDateFormat(
                "yyyy'-'MM'-'dd'T'HH':'mm'Z'");
        Date utcDate;
        try {
            utcDate = utcFormat.parse(utc);
        } catch (ParseException e) {
            throw new FalconException("Unable to parse utc date:", e);
        }
        DateFormat uriFormat = new SimpleDateFormat("yyyy'-'MM'-'dd'-'HH'-'mm");
        return uriFormat.format(utcDate);
    }

    public static boolean responsibleFor(String colo) {
        return DeploymentUtil.isEmbeddedMode() || (!DeploymentUtil.isPrism()
                && colo.equals(DeploymentUtil.getCurrentColo()));
    }

    public static Date getNextStartTime(Entity entity, Cluster cluster, Date effectiveTime) {
        switch(entity.getEntityType()) {
        case FEED:
            Feed feed = (Feed) entity;
            org.apache.falcon.entity.v0.feed.Cluster feedCluster = FeedHelper.getCluster(feed, cluster.getName());
            return getNextStartTime(feedCluster.getValidity().getStart(), feed.getFrequency(), feed.getTimezone(),
                effectiveTime);

        case PROCESS:
            Process process = (Process) entity;
            org.apache.falcon.entity.v0.process.Cluster processCluster = ProcessHelper.getCluster(process,
                cluster.getName());
            return getNextStartTime(processCluster.getValidity().getStart(), process.getFrequency(),
                process.getTimezone(), effectiveTime);

        default:
        }

        throw new IllegalArgumentException("Unhandled type: " + entity.getEntityType());
    }

    public static boolean isTableStorageType(Cluster cluster, Entity entity) throws FalconException {
        return entity.getEntityType() == EntityType.PROCESS
            ? isTableStorageType(cluster, (Process) entity) : isTableStorageType(cluster, (Feed) entity);
    }

    public static boolean isTableStorageType(Cluster cluster, Feed feed) throws FalconException {
        Storage.TYPE storageType = FeedHelper.getStorageType(feed, cluster);
        return Storage.TYPE.TABLE == storageType;
    }

    public static boolean isTableStorageType(Cluster cluster, Process process) throws FalconException {
        Storage.TYPE storageType = ProcessHelper.getStorageType(cluster, process);
        return Storage.TYPE.TABLE == storageType;
    }

}