/**
 * Copyright (C) FuseSource, Inc.
 * http://fusesource.com
 *
 * Licensed 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.fusesource.fabric.internal;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.zookeeper.KeeperException;
import org.fusesource.fabric.api.Container;
import org.fusesource.fabric.api.FabricException;
import org.fusesource.fabric.api.Profile;
import org.fusesource.fabric.service.FabricServiceImpl;
import org.fusesource.fabric.zookeeper.ZkPath;
import org.fusesource.fabric.zookeeper.utils.ZooKeeperUtils;
import org.linkedin.zookeeper.client.IZKClient;

public class ProfileImpl implements Profile {

    public final static String AGENT_PID = "org.fusesource.fabric.agent";
;

    private final String id;
    private final String version;
    private final FabricServiceImpl service;

    public ProfileImpl(String id, String version, FabricServiceImpl service) {
        this.id = id;
        this.version = version;
        this.service = service;
    }

    public String getId() {
        return id;
    }

    public String getVersion() {
        return version;
    }

    public FabricServiceImpl getService() {
        return service;
    }

    //In some cases we need to sort profiles by Id.
    @Override
    public int compareTo(Profile profile) {
        return id.compareTo(profile.getId());
    }

    public enum ConfigListType {

        BUNDLES("bundle"),
        FABS("fab"),
        FEATURES("feature"),
        REPOSITORIES("repository"),
        OVERRIDES("override");

        private String value;

        private ConfigListType(String value) {
            this.value = value;
        }
        public String toString() {
            return value;
        }
    }

    public List<String> getBundles() {
        return getContainerConfigList(this, ConfigListType.BUNDLES);
    }

    public List<String> getFabs() {
        return getContainerConfigList(this, ConfigListType.FABS);
    }

    public List<String> getFeatures() {
        return getContainerConfigList(this, ConfigListType.FEATURES);
    }

    public List<String> getRepositories() {
        return getContainerConfigList(this, ConfigListType.REPOSITORIES);
    }

    @Override
    public List<String> getOverrides() {
        return getContainerConfigList(this, ConfigListType.OVERRIDES);
    }

    @Override
    public void setBundles(List<String> values) {
        setContainerConfigList(this, values, ConfigListType.BUNDLES);
    }

    @Override
    public void setFabs(List<String> values) {
        setContainerConfigList(this, values, ConfigListType.FABS);
    }

    @Override
    public void setFeatures(List<String> values) {
        setContainerConfigList(this, values, ConfigListType.FEATURES);
    }

    @Override
    public void setRepositories(List<String> values) {
        setContainerConfigList(this, values, ConfigListType.REPOSITORIES);
    }

    @Override
    public void setOverrides(List<String> values) {
        setContainerConfigList(this, values, ConfigListType.OVERRIDES);
    }

    public static List<String> getContainerConfigList(Profile p, ConfigListType type) {
        try {
            Properties containerProps = getContainerProperties(p);
            ArrayList<String> rc = new ArrayList<String>();
            for ( Map.Entry<Object, Object> e : containerProps.entrySet() ) {
                if ( ((String)e.getKey()).startsWith(type + ".") ) {
                    rc.add((String)e.getValue());
                }
            }
            return rc;

        } catch (Exception e) {
            throw new FabricException(e);
        }
    }

    public static void setContainerConfigList(Profile p, List<String> values, ConfigListType type) {
        Map<String,Map<String, String>> config = p.getConfigurations();
        String prefix = type + ".";
        Map<String, String> map = config.get(AGENT_PID);
        if (map == null) {
            map = new HashMap<String, String>();
            config.put(AGENT_PID, map);
        } else {
            List<String> keys = new ArrayList<String>(map.keySet());
            for (String key : keys) {
                if (key.startsWith(prefix)) {
                    map.remove(key);
                }
            }
        }
        for (String value : values) {
            map.put(prefix + value, value);
        }
        p.setConfigurations(config);
    }

    public static Properties getContainerProperties(Profile p) throws IOException {
        byte[] b = p.getFileConfigurations().get(AGENT_PID + ".properties");
        if (b != null) {
            return toProperties(b);
        } else {
            return new Properties();
        }
    }

    public Profile[] getParents() {
        try {
            String node = ZkPath.CONFIG_VERSIONS_PROFILE.getPath(version, id);
            String str = service.getZooKeeper().getStringData(node);
            if (str == null || str.isEmpty()) {
                return new Profile[0];
            }
            str = str.trim();
            List<Profile> profiles = new ArrayList<Profile>();
            for (String p : str.split(" ")) {
                profiles.add(new ProfileImpl(p, version, service));
            }
            return profiles.toArray(new Profile[profiles.size()]);
        } catch (Exception e) {
            throw new FabricException(e);
        }
    }

    public void setParents(Profile[] parents) {
        try {
            String str = "";
            for (Profile parent : parents) {
                if (!version.equals(parent.getVersion())) {
                    throw new IllegalArgumentException("Version mismatch setting parent profile " + parent + " with version "
                            + parent.getVersion() + " expected version " + version);
                }
                if (!str.isEmpty()) {
                    str += " ";
                }
                str += parent.getId();
            }
            service.getZooKeeper().setData( ZkPath.CONFIG_VERSIONS_PROFILE.getPath(version, id), str );
        } catch (Exception e) {
            throw new FabricException(e);
        }
    }

    public Container[] getAssociatedContainers() {
        try {
            ArrayList<Container> rc = new ArrayList<Container>();
            Container[] containers = service.getContainers();
            for (Container container : containers) {
                if (!container.getVersion().getName().equals(getVersion())) {
                    continue;
                }
                for (Profile p : container.getProfiles()) {
                    if (this.equals(p)) {
                        rc.add(container);
                        break;
                    }
                }
            }
            return rc.toArray(new Container[0]);
        } catch (Exception e) {
            throw new FabricException(e);
        }
    }

    public boolean isOverlay() {
        return false;
    }

    public Profile getOverlay() {
        return new ProfileOverlayImpl(this);
    }

    @Override
    public Map<String, byte[]> getFileConfigurations() {
        try {
            Map<String, byte[]> configurations = new HashMap<String, byte[]>();
            String path = ZkPath.CONFIG_VERSIONS_PROFILE.getPath(version, id);
            List<String> pids = service.getZooKeeper().getChildren(path);
            for (String pid : pids) {
                configurations.put(pid, getFileConfiguration(pid));
            }
            return configurations;
        } catch (Exception e) {
            throw new FabricException(e);
        }
    }

    public byte[] getFileConfiguration(String pid) throws InterruptedException, KeeperException {
        IZKClient zooKeeper = service.getZooKeeper();
        String path = ZkPath.CONFIG_VERSIONS_PROFILE.getPath(version, id) + "/" + pid;
        if (zooKeeper.exists(path) == null) {
            return null;
        }
        if (zooKeeper.getData(path) == null) {
            List<String> children = zooKeeper.getChildren(path);
            StringBuffer buf = new StringBuffer();
            for (String child : children) {
                String value = zooKeeper.getStringData(path + "/" + child);
                buf.append(String.format("%s = %s\n", child, value));
            }
            return buf.toString().getBytes();
        } else {
            return zooKeeper.getData(path);
        }
    }

    @Override
    public void setFileConfigurations(Map<String, byte[]> configurations) {
        try {
            IZKClient zooKeeper = service.getZooKeeper();
            Map<String, byte[]> oldCfgs = getFileConfigurations();
            // Store new configs
            String path = ZkPath.CONFIG_VERSIONS_PROFILE.getPath(version, id);
            for (String pid : configurations.keySet()) {
                oldCfgs.remove(pid);
                byte[] newCfg = configurations.get(pid);
                String configPath =  path + "/" + pid;
                if (zooKeeper.exists(configPath) != null && zooKeeper.getChildren(configPath).size() > 0) {
                    List<String> kids = zooKeeper.getChildren(configPath);
                    ArrayList<String> saved = new ArrayList<String>();
                    // old format, we assume that the byte stream is in
                    // a .properties format
                    for (String line : new String(newCfg).split("\n")) {
                        if (line.startsWith("#") || line.length() == 0) {
                            continue;
                        }
                        String name_value[] = line.split("=", 2);
                        if (name_value.length < 2) {
                            continue;
                        }
                        String newPath = configPath + "/" + name_value[0].trim();
                        ZooKeeperUtils.set(zooKeeper, newPath, name_value[1].trim());
                        saved.add(name_value[0].trim());
                    }
                    for ( String kid : kids ) {
                        if (!saved.contains(kid)) {
                            zooKeeper.deleteWithChildren(configPath + "/" + kid);
                        }
                    }
                } else {
                    ZooKeeperUtils.set(zooKeeper, configPath, newCfg);
                }
            }
            for (String pid : oldCfgs.keySet()) {
                zooKeeper.deleteWithChildren(path + "/" + pid);
            }
        } catch (Exception e) {
            throw new FabricException(e);
        }
    }

    public Map<String, Map<String, String>> getConfigurations() {
        try {
            Map<String, Map<String, String>> configurations = new HashMap<String, Map<String, String>>();
            Map<String, byte[]> configs = getFileConfigurations();
            for (Map.Entry<String, byte[]> entry: configs.entrySet()){
                if(entry.getKey().endsWith(".properties")) {
                    String pid = stripSuffix(entry.getKey(), ".properties");
                    configurations.put(pid, getConfiguration(pid));
                }
            }
            return configurations;
        } catch (Exception e) {
            throw new FabricException(e);
        }
    }

    @Override
    public Map<String, String> getContainerConfiguration() {
        Map<String, String> map = getConfigurations().get(AGENT_PID);
        if (map == null) {
            map = new HashMap<String, String>();
        }
        return map;
    }

    private Map<String, String> getConfiguration(String pid) throws InterruptedException, KeeperException, IOException {
        IZKClient zooKeeper = service.getZooKeeper();
        String path = ZkPath.CONFIG_VERSIONS_PROFILE.getPath(version, id) + "/" + pid +".properties";
        if (zooKeeper.exists(path) == null) {
            return null;
        }
        byte[] data = zooKeeper.getData(path);
        return toMap(toProperties(data));
    }

    static public byte[] toBytes(Properties source) throws IOException {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        source.store(baos, null);
        return baos.toByteArray();
    }

    static public Properties toProperties(byte[] source) throws IOException {
        Properties rc = new Properties();
        rc.load(new ByteArrayInputStream(source));
        return rc;
    }

    static public Map<String, String> toMap(Properties source) {
        Map<String, String> rc = new HashMap<String, String>();
        for (Map.Entry<Object, Object> entry: source.entrySet()){
            rc.put((String) entry.getKey(), (String) entry.getValue());
        }
        return rc;
    }

    static public Properties toProperties(Map<String, String> source) {
        Properties rc = new Properties();
        for (Map.Entry<String, String> entry: source.entrySet()){
            rc.put(entry.getKey(), entry.getValue());
        }
        return rc;
    }

    static public String stripSuffix(String value, String suffix) throws IOException {
        if(value.endsWith(suffix)) {
            return value.substring(0, value.length() -suffix.length());
        } else {
            return value;
        }
    }


    public void setConfigurations(Map<String, Map<String, String>> configurations) {
        try {
            IZKClient zooKeeper = service.getZooKeeper();
            Map<String, Map<String, String>> oldCfgs = getConfigurations();
            // Store new configs
            String path = ZkPath.CONFIG_VERSIONS_PROFILE.getPath(version, id);
            for (String pid : configurations.keySet()) {
                oldCfgs.remove(pid);
                byte[] data = toBytes(toProperties(configurations.get(pid)));
                String p =  path + "/" + pid + ".properties";
                ZooKeeperUtils.set(zooKeeper, p, data);
            }
            for (String key : oldCfgs.keySet()) {
                zooKeeper.deleteWithChildren(path + "/" + key +".properties");
            }
        } catch (Exception e) {
            throw new FabricException(e);
        }
    }

    public void delete() {
        service.deleteProfile(this);
    }

    public boolean configurationEquals(Profile other) {
         Profile[] parents = getParents();
         Profile[] otherParents = other.getParents();
         Arrays.sort(parents);
         Arrays.sort(otherParents);
         if (!getConfigurations().equals(other.getConfigurations())) {
             return false;
         }
         if (parents.length != otherParents.length) {
             return false;
         }

         for (int i = 0; i < parents.length; i++) {
             if (!parents[i].configurationEquals(otherParents[i])) {
                 return false;
             }
         }
         return true;
    }

    @Override
    public String toString() {
        return "ProfileImpl[" +
                "id='" + id + '\'' +
                ", version='" + version + '\'' +
                ']';
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        ProfileImpl profile = (ProfileImpl) o;
        if (!id.equals(profile.id)) return false;
        if (!version.equals(profile.version)) return false;
        return true;
    }

    @Override
    public int hashCode() {
        int result = id.hashCode();
        result = 31 * result + version.hashCode();
        return result;
    }

    @Override
    public boolean isAbstract() {
        return id == null || id.equals("mq-base") || id.startsWith("fabric-ensemble-0000");
    }
}
