package com.atlassian.stash.internal.cluster;

import com.google.common.collect.Maps;
import com.hazelcast.nio.ObjectDataInput;
import com.hazelcast.nio.ObjectDataOutput;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

import static com.atlassian.stash.internal.cluster.ClusterJoinCheckAction.PASSIVATE_ANY_NODE;

/**
 * {@link ClusterJoinCheck join check} that verifies that the other node meets the {@link ClusterJoinRequirement requirements}
 * for forming a cluster. If any requirement doesn't match or a requirement is missing on either of the nodes, the
 * check will fail and signal that one of the nodes needs to be passivated.
 */
@Component
public class RequirementsClusterJoinCheck implements ClusterJoinCheck {

    private final Map<String, ClusterJoinRequirement> requirements;

    @Autowired
    public RequirementsClusterJoinCheck(ClusterJoinRequirement... requirements) {
        this.requirements = new HashMap<>(requirements.length);
        for (ClusterJoinRequirement requirement : requirements) {
            this.requirements.put(requirement.getName(), requirement);
        }
    }

    @Nonnull
    @Override
    public ClusterJoinCheckResult accept(@Nonnull ClusterJoinRequest request) throws IOException {
        ClusterJoinCheckResult.Builder resultBuilder = new ClusterJoinCheckResult.Builder();

        Map<String, ClusterJoinRequirement> requirements = Maps.newHashMap(this.requirements);

        try {
            // read the advertised requirements
            Map<String, String> remoteRequirements = readMap(request.in());
            for (Map.Entry<String, String> entry : remoteRequirements.entrySet()) {
                String name = entry.getKey();
                String expected = entry.getValue();

                ClusterJoinRequirement requirement = requirements.remove(name);
                if (requirement == null) {
                    resultBuilder.passivate(PASSIVATE_ANY_NODE, "Connecting node requires property '" + name + "' to be '" + expected +
                            "' but the property is not defined");

                } else {
                    String actual = requirement.getValue();
                    if (!Objects.equals(expected, actual)) {
                        resultBuilder.passivate(PASSIVATE_ANY_NODE, "Required property '" + name + "' should be '" + expected + "' but is '" + actual + "'");
                    }
                }
            }
            for (String missing : requirements.keySet()) {
                resultBuilder.passivate(PASSIVATE_ANY_NODE, "Connecting node did not provide required property '" + missing + "'");
            }
        } catch (NoClassDefFoundError e) {
            // can happen if a requirement from the other node has a value of an unknown type
            resultBuilder.passivate(PASSIVATE_ANY_NODE, "Could not validate whether the node meets the requirements for connecting to the " +
                    "cluster: '" + e.getMessage() + "'");
        }

        return resultBuilder.build();
    }

    @Nonnull
    @Override
    public ClusterJoinCheckResult connect(@Nonnull ClusterJoinRequest request) throws IOException {
        Map<String, String> expectedValues = new HashMap<>(requirements.size());
        for (ClusterJoinRequirement requirement : requirements.values()) {
            expectedValues.put(requirement.getName(), requirement.getValue());
        }

        writeMap(request.out(), expectedValues);

        // the remote node validates the requirements
        return ClusterJoinCheckResult.OK;
    }

    @Nonnull
    @Override
    public String getName() {
        return getClass().getName();
    }

    @Override
    public int getOrder() {
        return 10;
    }

    @Nonnull
    @Override
    public ClusterJoinCheckResult onUnknown(@Nonnull ClusterJoinRequest request) {
        // other node does not have the RequirementsClusterJoinCheck
        ClusterJoinCheckResult.Builder resultBuilder = new ClusterJoinCheckResult.Builder();
        for (String missing : requirements.keySet()) {
            resultBuilder.passivate(PASSIVATE_ANY_NODE, "Node did not provide required property '" + missing + "'");
        }

        return resultBuilder.build();
    }

    /**
     * Reads a map with string keys and values.
     * <p>
     * Note that the key and value are both restricted to {@code String}. This is intended to avoid a potential
     * vulnerability in using {@code in.readObject()}, because this method is used to deserialize <i>untrusted</i>
     * data--data received over a socket from an unauthenticated cluster node during startup.
     *
     * @param in the stream to read the map data from
     * @return the read map
     * @throws IOException if map data cannot be read from the provided stream
     */
    private static Map<String, String> readMap(ObjectDataInput in) throws IOException {
        int length = in.readInt();

        Map<String, String> map = new HashMap<>(length, 1.0f);
        for (int i = 0; i < length; i++) {
            map.put(in.readUTF(), in.readUTF());
        }

        return map;
    }

    /**
     * Writes a map with string keys and string values.
     *
     * @param out the stream to write the map to
     * @param map the map to write
     * @throws IOException if map data cannot be written to the provided stream
     */
    private static void writeMap(ObjectDataOutput out, Map<String, String> map) throws IOException {
        out.writeInt(map.size());
        for (Map.Entry<String, String> entry : map.entrySet()) {
            out.writeUTF(entry.getKey());

            //Note: The _contract_ for ObjectDataOutput.writeUTF states that null values will trigger
            //      NullPointerExceptions. However, Hazelcast's implementation of writeUTF does _not_
            //      honor that contract and instead internally uses a marker boolean to flag nulls
            out.writeUTF(entry.getValue());
        }
    }
}
