package io.crazy88.beatrix.boot.akka;

import static io.crazy88.beatrix.boot.akka.AkkaConfigPropertySource.PROPERTY_SOURCE_NAME;
import static java.util.Collections.emptyMap;
import static java.util.Collections.unmodifiableList;
import static java.util.Collections.unmodifiableMap;
import static java.util.Comparator.comparing;
import static java.util.stream.Collectors.toMap;
import static java.util.stream.StreamSupport.stream;
import static org.springframework.core.env.StandardEnvironment.SYSTEM_ENVIRONMENT_PROPERTY_SOURCE_NAME;
import static org.springframework.core.env.StandardEnvironment.SYSTEM_PROPERTIES_PROPERTY_SOURCE_NAME;
import static org.springframework.util.CollectionUtils.isEmpty;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Stream;

import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.EnumerablePropertySource;
import org.springframework.core.env.PropertySource;

class AkkaConfigPropertySourceAdapter {

    private static final Pattern INDEXED_PROPERTY_PATTERN = Pattern.compile("^(?<path>.*)\\[(?<index>\\d+)\\]$");

    static Map<String, Object> getConfigMap(ConfigurableEnvironment environment) {
        return convertIndexedProperties(getPropertyMap(environment));
    }

    @SuppressWarnings("unchecked")
    private static Map<String, Object> convertIndexedProperties(Map<String, String> propertyMap) {

        if (isEmpty(propertyMap)) {
            return emptyMap();
        }

        Map<String, Object> converted = new LinkedHashMap<>();

        for (Map.Entry<String, String> entry: propertyMap.entrySet()) {
            Matcher m = INDEXED_PROPERTY_PATTERN.matcher(entry.getKey());
            String path = m.matches() ? m.group("path") : entry.getKey();
            Object existing = converted.get(path);

            // Make sure properties are either list-based or normal
            if ((existing != null) && ((m.matches() && !(existing instanceof List)) || (!m.matches() && !(existing instanceof String)))) {
                throw new IllegalArgumentException("Invalid property hierarchy - property must be either a regular or a list-based property (path:" + path + ", existing: " + existing + ", key:" + entry.getKey() + ", value:" + entry.getValue());
            }

            // Convert indexed properties to lists
            if (m.matches()) {

                int index = Integer.parseInt(m.group("index"));
                ArrayList<String> list = (ArrayList<String>) converted.getOrDefault(path, new ArrayList<>());

                // Extend the list, if necessary
                list.ensureCapacity(index + 1);
                while (list.size() <= index) {
                    list.add(null);
                }

                list.set(index, entry.getValue());
                converted.put(path, list);

            } else {
                // Copy normal properties
                converted.put(entry.getKey(), entry.getValue());
            }
        }

        // Make sure any contained lists are immutable
        for (Map.Entry<String, Object> entry : converted.entrySet()) {
            if (entry.getValue() instanceof List) {
                entry.setValue(unmodifiableList((List<String>) entry.getValue()));
            }
        }

        return unmodifiableMap(converted);
    }

    private static Map<String, String> getPropertyMap(ConfigurableEnvironment environment) {
        return stream(environment.getPropertySources().spliterator(), false)
                .filter(AkkaConfigPropertySourceAdapter::isEligiblePropertySource)
                .sorted(comparing(source -> source.getName().length()))
                .map(EnumerablePropertySource.class::cast)
                .flatMap(source -> Stream.of(source.getPropertyNames()))
                .filter(AkkaConfigPropertySourceAdapter::isEligibleProperty)
                .map(name -> ImmutablePair.of(name, environment.getProperty(name)))
                .collect(toMap(Pair::getKey, Pair::getValue, (a, b) -> b));
    }

    private static boolean isEligiblePropertySource(PropertySource source) {
        String name = source.getName();
        return (source instanceof EnumerablePropertySource)
                && !(name.equalsIgnoreCase(SYSTEM_ENVIRONMENT_PROPERTY_SOURCE_NAME)
                        || name.equalsIgnoreCase(SYSTEM_PROPERTIES_PROPERTY_SOURCE_NAME)
                        || name.equalsIgnoreCase(PROPERTY_SOURCE_NAME));
    }

    private static boolean isEligibleProperty(String name) {
        return name.startsWith("akka") || name.startsWith("simple-akka-downing.");
    }
}