package com.atlassian.bitbucket.mesh.network;

import com.google.common.collect.ImmutableMap;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.google.common.base.Preconditions.checkArgument;
import static java.util.Objects.requireNonNull;

class Topology {

    private final MeshNode localNode;
    private final Map<String, MeshNode> nodes;
    private final Map<String, Partition> partitions;

    private Topology(Builder builder) {
        nodes = ImmutableMap.copyOf(builder.nodes);
        partitions = ImmutableMap.copyOf(builder.partitions);

        List<MeshNode> localNodes = nodes.values().stream()
                .filter(MeshNode::isLocal)
                .collect(Collectors.toList());

        //Verify that exactly one local node is defined
        checkArgument(!localNodes.isEmpty(), "Local node was not defined");
        checkArgument(localNodes.size() == 1, "More than one local node was defined: %s", localNodes);
        localNode = localNodes.get(0);
    }

    @Override
    public String toString() {
        //Only display the first 10 partitions because it's not practical to log massive topologies
        int count = partitions.size();
        return partitions.values().stream()
                .map(partition -> partition.getId() + ": " + partition.getReplicas().stream()
                        .map(replica -> replica.getNode().getName())
                        .collect(Collectors.joining(", ", "[", "]")))
                .limit(10)
                .collect(Collectors.joining(", ", "{", count > 10 ? ", ..." + (count - 10) + " more}" : "}"));
    }

    @Nullable
    MeshNode findNodeById(@Nonnull String nodeId) {
        return nodes.get(requireNonNull(nodeId, "nodeId"));
    }

    @Nonnull
    MeshNode getLocalNode() {
        return localNode;
    }

    @Nonnull
    Collection<MeshNode> getNodes() {
        return nodes.values();
    }

    @Nonnull
    Map<String, MeshNode> getNodesMap() {
        return nodes;
    }

    @Nullable
    Partition getPartition(@Nonnull String id) {
        return partitions.get(requireNonNull(id, "id"));
    }

    @Nonnull
    Collection<Partition> getPartitions() {
        return partitions.values();
    }

    static class Builder {

        private final Map<String, MeshNode> nodes;
        private final Map<String, Partition> partitions;

        Builder() {
            nodes = new HashMap<>();
            partitions = new HashMap<>();
        }

        @Nonnull
        Topology build() {
            return new Topology(this);
        }

        @Nonnull
        Builder nodes(@Nullable MeshNode value, @Nullable MeshNode... moreValues) {
            addNode(value);

            if (moreValues != null) {
                for (MeshNode node : moreValues) {
                    addNode(node);
                }
            }
            return this;
        }

        @Nonnull
        Builder nodes(@Nullable Iterable<MeshNode> values) {
            if (values != null) {
                for (MeshNode node : values) {
                    addNode(node);
                }
            }
            return this;
        }

        @Nonnull
        Builder partition(@Nullable Partition value) {
            if (value != null) {
                partitions.put(value.getId(), value);
                //Make sure all nodes referenced in the partition replicas are registered as nodes
                value.getReplicas().stream()
                        .map(PartitionReplica::getNode)
                        .forEach(node -> nodes.computeIfAbsent(node.getId(), id -> node));
            }
            return this;
        }

        @Nonnull
        Builder partitions(@Nullable Iterable<Partition> values) {
            if (values != null) {
                values.forEach(this::partition);
            }
            return this;
        }

        private void addNode(MeshNode node) {
            if (node != null) {
                nodes.put(node.getId(), node);
            }
        }
    }
}
