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

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Objects;
import com.google.common.collect.ImmutableMap;

import org.slf4j.LoggerFactory;

import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.InheritingClass;
import org.apache.cassandra.config.ParameterizedClass;
import org.apache.cassandra.db.memtable.Memtable;
import org.apache.cassandra.db.memtable.SkipListMemtableFactory;
import org.apache.cassandra.exceptions.ConfigurationException;

/**
 * Memtable types and options are specified with these parameters. Memtable classes must either contain a static
 * {@code FACTORY} field (if they take no arguments other than class), or implement a
 * {@code factory(Map<String, String>)} method.
 *
 * The latter should consume any further options (using {@code map.remove}).
 *
 * See Memtable_API.md for further details on the configuration and usage of memtable implementations.
  */
public final class MemtableParams
{
    private final Memtable.Factory factory;
    private final String configurationKey;

    private MemtableParams(Memtable.Factory factory, String configurationKey)
    {
        this.configurationKey = configurationKey;
        this.factory = factory;
    }

    public String configurationKey()
    {
        return configurationKey;
    }

    public Memtable.Factory factory()
    {
        return factory;
    }

    @Override
    public String toString()
    {
        return configurationKey;
    }

    @Override
    public boolean equals(Object o)
    {
        if (this == o)
            return true;

        if (!(o instanceof MemtableParams))
            return false;

        MemtableParams c = (MemtableParams) o;

        return Objects.equal(configurationKey, c.configurationKey);
    }

    @Override
    public int hashCode()
    {
        return configurationKey.hashCode();
    }

    private static final String DEFAULT_CONFIGURATION_KEY = "default";
    private static final Memtable.Factory DEFAULT_MEMTABLE_FACTORY = SkipListMemtableFactory.INSTANCE;
    private static final ParameterizedClass DEFAULT_CONFIGURATION = SkipListMemtableFactory.CONFIGURATION;
    private static final Map<String, ParameterizedClass>
        CONFIGURATION_DEFINITIONS = expandDefinitions(DatabaseDescriptor.getMemtableConfigurations());
    private static final Map<String, MemtableParams> CONFIGURATIONS = new HashMap<>();
    public static final MemtableParams DEFAULT = get(null);

    public static Set<String> knownDefinitions()
    {
        return CONFIGURATION_DEFINITIONS.keySet();
    }

    public static MemtableParams get(String key)
    {
        if (key == null)
            key = DEFAULT_CONFIGURATION_KEY;

        synchronized (CONFIGURATIONS)
        {
            return CONFIGURATIONS.computeIfAbsent(key, MemtableParams::parseConfiguration);
        }
    }

    public static MemtableParams getWithFallback(String key)
    {
        try
        {
            return get(key);
        }
        catch (ConfigurationException e)
        {
            LoggerFactory.getLogger(MemtableParams.class).error("Invalid memtable configuration \"" + key + "\" in schema. " +
                                                                "Falling back to default to avoid schema mismatch.\n" +
                                                                "Please ensure the correct definition is given in cassandra.yaml.",
                                                                e);
            return new MemtableParams(DEFAULT.factory(), key);
        }
    }

    @VisibleForTesting
    static Map<String, ParameterizedClass> expandDefinitions(Map<String, InheritingClass> memtableConfigurations)
    {
        if (memtableConfigurations == null)
            return ImmutableMap.of(DEFAULT_CONFIGURATION_KEY, DEFAULT_CONFIGURATION);

        LinkedHashMap<String, ParameterizedClass> configs = new LinkedHashMap<>(memtableConfigurations.size() + 1);

        // If default is not overridden, add an entry first so that other configurations can inherit from it.
        // If it is, process it in its point of definition, so that the default can inherit from another configuration.
        if (!memtableConfigurations.containsKey(DEFAULT_CONFIGURATION_KEY))
            configs.put(DEFAULT_CONFIGURATION_KEY, DEFAULT_CONFIGURATION);

        Map<String, InheritingClass> inheritingClasses = new LinkedHashMap<>();

        for (Map.Entry<String, InheritingClass> entry : memtableConfigurations.entrySet())
        {
            if (entry.getValue().inherits != null)
            {
                if (entry.getKey().equals(entry.getValue().inherits))
                    throw new ConfigurationException(String.format("Configuration entry %s can not inherit itself.", entry.getKey()));

                if (memtableConfigurations.get(entry.getValue().inherits) == null && !entry.getValue().inherits.equals(DEFAULT_CONFIGURATION_KEY))
                    throw new ConfigurationException(String.format("Configuration entry %s inherits non-existing entry %s.",
                                                                   entry.getKey(), entry.getValue().inherits));

                inheritingClasses.put(entry.getKey(), entry.getValue());
            }
            else
                configs.put(entry.getKey(), entry.getValue().resolve(configs));
        }

        for (Map.Entry<String, InheritingClass> inheritingEntry : inheritingClasses.entrySet())
        {
            String inherits = inheritingEntry.getValue().inherits;
            while (inherits != null)
            {
                InheritingClass nextInheritance = inheritingClasses.get(inherits);
                if (nextInheritance == null)
                    inherits = null;
                else
                    inherits = nextInheritance.inherits;

                if (inherits != null && inherits.equals(inheritingEntry.getKey()))
                    throw new ConfigurationException(String.format("Detected loop when processing key %s", inheritingEntry.getKey()));
            }
        }

        while (!inheritingClasses.isEmpty())
        {
            Set<String> forRemoval = new HashSet<>();
            for (Map.Entry<String, InheritingClass> inheritingEntry : inheritingClasses.entrySet())
            {
                if (configs.get(inheritingEntry.getValue().inherits) != null)
                {
                    configs.put(inheritingEntry.getKey(), inheritingEntry.getValue().resolve(configs));
                    forRemoval.add(inheritingEntry.getKey());
                }
            }

            assert !forRemoval.isEmpty();

            for (String toRemove : forRemoval)
                inheritingClasses.remove(toRemove);
        }

        return ImmutableMap.copyOf(configs);
    }

    private static MemtableParams parseConfiguration(String configurationKey)
    {
        ParameterizedClass definition = CONFIGURATION_DEFINITIONS.get(configurationKey);

        if (definition == null)
            throw new ConfigurationException("Memtable configuration \"" + configurationKey + "\" not found.");
        return new MemtableParams(getMemtableFactory(definition), configurationKey);
    }


    private static Memtable.Factory getMemtableFactory(ParameterizedClass options)
    {
        // Special-case this so that we don't initialize memtable class for tests that need to delay that.
        if (options == DEFAULT_CONFIGURATION)
            return DEFAULT_MEMTABLE_FACTORY;

        String className = options.class_name;
        if (className == null || className.isEmpty())
            throw new ConfigurationException("The 'class_name' option must be specified.");

        className = className.contains(".") ? className : "org.apache.cassandra.db.memtable." + className;
        try
        {
            Memtable.Factory factory;
            Class<?> clazz = Class.forName(className);
            final Map<String, String> parametersCopy = options.parameters != null
                                                       ? new HashMap<>(options.parameters)
                                                       : new HashMap<>();
            try
            {
                Method factoryMethod = clazz.getDeclaredMethod("factory", Map.class);
                factory = (Memtable.Factory) factoryMethod.invoke(null, parametersCopy);
            }
            catch (NoSuchMethodException e)
            {
                // continue with FACTORY field
                Field factoryField = clazz.getDeclaredField("FACTORY");
                factory = (Memtable.Factory) factoryField.get(null);
            }
            if (!parametersCopy.isEmpty())
                throw new ConfigurationException("Memtable class " + className + " does not accept any futher parameters, but " +
                                                 parametersCopy + " were given.");
            return factory;
        }
        catch (NoSuchFieldException | ClassNotFoundException | IllegalAccessException | InvocationTargetException | ClassCastException e)
        {
            if (e.getCause() instanceof ConfigurationException)
                throw (ConfigurationException) e.getCause();
            throw new ConfigurationException("Could not create memtable factory for class " + options, e);
        }
    }
}
