/*
 * 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.kafka.server.common;

import org.apache.kafka.common.feature.SupportedVersionRange;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static org.apache.kafka.server.common.UnitTestFeatureVersion.FV0.UT_FV0_0;

/**
 * This is enum for the various features implemented for Kafka clusters.
 * KIP-584: Versioning Scheme for Features introduced the idea of various features, but only added one feature -- MetadataVersion.
 * KIP-1022: Formatting and Updating Features allowed for more features to be added. In order to set and update features,
 * they need to be specified via the StorageTool or FeatureCommand tools.
 * <br>
 * Having a unified enum for the features that will use a shared type in the API used to set and update them
 * makes it easier to process these features.
 */
public enum Feature {

    /**
     * Features defined. If a feature is included in this list, and marked to be used in production they will also be specified when
     * formatting a cluster via the StorageTool. MetadataVersion is handled separately, so it is not included here.
     *
     * See {@link TestFeatureVersion} as an example. See {@link FeatureVersion} when implementing a new feature.
     */
    KRAFT_VERSION(KRaftVersion.FEATURE_NAME, KRaftVersion.values(), KRaftVersion.LATEST_PRODUCTION),
    TRANSACTION_VERSION(TransactionVersion.FEATURE_NAME, TransactionVersion.values(), TransactionVersion.LATEST_PRODUCTION),
    GROUP_VERSION(GroupVersion.FEATURE_NAME, GroupVersion.values(), GroupVersion.LATEST_PRODUCTION),
    ELIGIBLE_LEADER_REPLICAS_VERSION(EligibleLeaderReplicasVersion.FEATURE_NAME, EligibleLeaderReplicasVersion.values(), EligibleLeaderReplicasVersion.LATEST_PRODUCTION),
    SHARE_VERSION(ShareVersion.FEATURE_NAME, ShareVersion.values(), ShareVersion.LATEST_PRODUCTION),
    STREAMS_VERSION(StreamsVersion.FEATURE_NAME, StreamsVersion.values(), StreamsVersion.LATEST_PRODUCTION),

    /**
     * Features defined only for unit tests and are not used in production.
     */
    TEST_VERSION(TestFeatureVersion.FEATURE_NAME, TestFeatureVersion.values(), TestFeatureVersion.LATEST_PRODUCTION),
    UNIT_TEST_VERSION_0(UnitTestFeatureVersion.FV0.FEATURE_NAME, new FeatureVersion[]{UT_FV0_0}, UnitTestFeatureVersion.FV0.LATEST_PRODUCTION),
    UNIT_TEST_VERSION_1(UnitTestFeatureVersion.FV1.FEATURE_NAME, UnitTestFeatureVersion.FV1.values(), UnitTestFeatureVersion.FV1.LATEST_PRODUCTION),
    UNIT_TEST_VERSION_2(UnitTestFeatureVersion.FV2.FEATURE_NAME, UnitTestFeatureVersion.FV2.values(), UnitTestFeatureVersion.FV2.LATEST_PRODUCTION),
    UNIT_TEST_VERSION_3(UnitTestFeatureVersion.FV3.FEATURE_NAME, UnitTestFeatureVersion.FV3.values(), UnitTestFeatureVersion.FV3.LATEST_PRODUCTION),
    UNIT_TEST_VERSION_4(UnitTestFeatureVersion.FV4.FEATURE_NAME, UnitTestFeatureVersion.FV4.values(), UnitTestFeatureVersion.FV4.LATEST_PRODUCTION),
    UNIT_TEST_VERSION_5(UnitTestFeatureVersion.FV5.FEATURE_NAME, UnitTestFeatureVersion.FV5.values(), UnitTestFeatureVersion.FV5.LATEST_PRODUCTION),
    UNIT_TEST_VERSION_6(UnitTestFeatureVersion.FV6.FEATURE_NAME, UnitTestFeatureVersion.FV6.values(), UnitTestFeatureVersion.FV6.LATEST_PRODUCTION),
    UNIT_TEST_VERSION_7(UnitTestFeatureVersion.FV7.FEATURE_NAME, UnitTestFeatureVersion.FV7.values(), UnitTestFeatureVersion.FV7.LATEST_PRODUCTION);

    public static final Feature[] FEATURES;

    // The list of features that are not unit test features.
    public static final List<Feature> TEST_AND_PRODUCTION_FEATURES;

    public static final List<Feature> PRODUCTION_FEATURES;

    public static final List<String> PRODUCTION_FEATURE_NAMES;
    private final String name;
    private final FeatureVersion[] featureVersions;

    // The latest production version of the feature, owned and updated by the feature owner
    // in the respective feature definition. The value should not be smaller than the default
    // value calculated with {@link #defaultValue(MetadataVersion)}.
    public final FeatureVersion latestProduction;

    Feature(String name,
            FeatureVersion[] featureVersions,
            FeatureVersion latestProduction) {
        this.name = name;
        this.featureVersions = featureVersions;
        this.latestProduction = latestProduction;
    }

    static {
        Feature[] enumValues = Feature.values();
        FEATURES = Arrays.copyOf(enumValues, enumValues.length);

        TEST_AND_PRODUCTION_FEATURES = Arrays.stream(FEATURES).filter(feature ->
            !feature.name.startsWith("unit." + TestFeatureVersion.FEATURE_NAME)
        ).collect(Collectors.toList());

        PRODUCTION_FEATURES = Arrays.stream(FEATURES).filter(feature ->
            !feature.name.equals(TEST_VERSION.featureName()) &&
            !feature.name.startsWith("unit." + TestFeatureVersion.FEATURE_NAME)
        ).collect(Collectors.toList());
        PRODUCTION_FEATURE_NAMES = PRODUCTION_FEATURES.stream().map(feature ->
                feature.name).collect(Collectors.toList());

        validateDefaultValueAndLatestProductionValue(TEST_VERSION);
        for (Feature feature : PRODUCTION_FEATURES) {
            validateDefaultValueAndLatestProductionValue(feature);
        }
    }

    public String featureName() {
        return name;
    }

    public FeatureVersion[] featureVersions() {
        return featureVersions;
    }

    public short latestProduction() {
        return latestProduction.featureLevel();
    }

    public short minimumProduction() {
        return featureVersions[0].featureLevel();
    }

    public short latestTesting() {
        return featureVersions[featureVersions.length - 1].featureLevel();
    }

    public SupportedVersionRange supportedVersionRange() {
        return new SupportedVersionRange(
            minimumProduction(),
            latestTesting()
        );
    }

    /**
     * Creates a FeatureVersion from a level.
     *
     * @param level                        the level of the feature
     * @param allowUnstableFeatureVersions whether unstable versions can be used
     * @return the FeatureVersionUtils.FeatureVersion for the feature the enum is based on.
     * @throws IllegalArgumentException    if the feature is not known.
     */
    public FeatureVersion fromFeatureLevel(short level,
                                           boolean allowUnstableFeatureVersions) {
        return Arrays.stream(featureVersions).filter(featureVersion ->
            featureVersion.featureLevel() == level && (allowUnstableFeatureVersions || level <= latestProduction())).findFirst().orElseThrow(
                () -> new IllegalArgumentException("No feature:" + featureName() + " with feature level " + level));
    }

    /**
     * A method to validate the feature can be set. If a given feature relies on another feature, the dependencies should be
     * captured in {@link FeatureVersion#dependencies()}
     * <p>
     * For example, say feature X level x relies on feature Y level y:
     * if feature X >= x then throw an error if feature Y < y.
     *
     * @param feature                   the feature we are validating
     * @param features                  the feature versions we have (or want to set)
     * @throws IllegalArgumentException if the feature is not valid
     */
    public static void validateVersion(FeatureVersion feature, Map<String, Short> features) {
        Short metadataVersion = features.get(MetadataVersion.FEATURE_NAME);

        if (feature.featureLevel() >= 1 && (metadataVersion == null || metadataVersion < MetadataVersion.MINIMUM_VERSION.featureLevel()))
            throw new IllegalArgumentException(feature.featureName() + " could not be set to " + feature.featureLevel() +
                    " because it depends on metadata.version=" + MetadataVersion.MINIMUM_VERSION.featureLevel() + " (" + MetadataVersion.MINIMUM_VERSION + ")");

        for (Map.Entry<String, Short> dependency: feature.dependencies().entrySet()) {
            Short featureLevel = features.get(dependency.getKey());

            if (featureLevel == null || featureLevel < dependency.getValue()) {
                throw new IllegalArgumentException(feature.featureName() + " could not be set to " + feature.featureLevel() +
                        " because it depends on " + dependency.getKey() + " level " + dependency.getValue());
            }
        }
    }

    /**
     * A method to return the default (latest production) version of a feature based on the metadata version provided.
     *
     * Every time a new feature is added, it should create a mapping from metadata version to feature version
     * with {@link FeatureVersion#bootstrapMetadataVersion()}. The feature version should be marked as production ready
     * before the metadata version is made production ready.
     *
     * @param metadataVersion the metadata version we want to use to set the default.
     * @return the default version given the feature and provided metadata version
     */
    public FeatureVersion defaultVersion(MetadataVersion metadataVersion) {
        FeatureVersion version = featureVersions[0];
        for (Iterator<FeatureVersion> it = Arrays.stream(featureVersions).iterator(); it.hasNext(); ) {
            FeatureVersion feature = it.next();
            if (feature.bootstrapMetadataVersion().isLessThan(metadataVersion) || feature.bootstrapMetadataVersion().equals(metadataVersion))
                version = feature;
            else
                return version;
        }
        return version;
    }

    public short defaultLevel(MetadataVersion metadataVersion) {
        return defaultVersion(metadataVersion).featureLevel();
    }

    public static Feature featureFromName(String featureName) {
        for (Feature feature : FEATURES) {
            if (feature.name.equals(featureName))
                return feature;
        }
        throw new IllegalArgumentException("Feature " + featureName + " not found.");
    }

    public boolean isProductionReady(short featureVersion) {
        return featureVersion <= latestProduction();
    }

    public boolean hasFeatureVersion(FeatureVersion featureVersion) {
        for (FeatureVersion v : featureVersions()) {
            if (v == featureVersion) {
                return true;
            }
        }
        return false;
    }

    /**
     * The method ensures that the following statements are met:
     * 1. The latest production value is one of the feature values.
     * 2. The latest production value >= the default value.
     * 3. The dependencies of the latest production value <= their latest production values.
     * 4. The dependencies of all default values <= their default values.
     * 5. If the latest production depends on MetadataVersion, the value should be <= MetadataVersion.LATEST_PRODUCTION.
     * 6. If any default value depends on MetadataVersion, the value should be <= the default value bootstrap MV.
     *
     * Suppose we have feature X as the feature being validated.
     * Invalid examples:
     *     - The feature X has default version = XV_10 (dependency = {}), latest production = XV_5 (dependency = {})
     *       (Violating rule 2. The latest production value XV_5 is smaller than the default value)
     *     - The feature X has latest production = XV_11 (dependency = {Y: YV_4})
     *       The feature Y has latest production = YV_3 (dependency = {})
     *       (Violating rule 3. For latest production XV_11, Y's latest production YV_3 is smaller than the dependency value YV_4)
     *     - The feature X has default version = XV_10 (dependency = {Y: YV_4})
     *       The feature Y has default version = YV_3 (dependency = {})
     *       (Violating rule 4. For default version XV_10, Y's default value YV_3 is smaller than the dependency value YV_4)
     *     - The feature X has latest production = XV_11 (dependency = {MetadataVersion: IBP_4_0_IV1}), MetadataVersion.LATEST_PRODUCTION is IBP_4_0_IV0
     *       (Violating rule 5. The dependency MV IBP_4_0_IV1 is behind MV latest production IBP_4_0_IV0)
     *     - The feature X has default version = XV_10 (dependency = {MetadataVersion: IBP_4_0_IV1}) and bootstrap MV = IBP_4_0_IV0
     *       (Violating rule 6. When MV latest production is IBP_4_0_IV0, feature X will be set to XV_10 by default whereas it depends on MV IBP_4_0_IV1)
     * Valid examples:
     *     - The feature X has default version = XV_10 (dependency = {}), latest production = XV_10 (dependency = {})
     *     - The feature X has default version = XV_10 (dependency = {Y: YV_3}), latest production = XV_11 (dependency = {Y: YV_4})
     *       The feature Y has default version = YV_3 (dependency = {}), latest production = YV_4 (dependency = {})
     *     - The feature X has default version = XV_10 (dependency = {MetadataVersion: IBP_4_0_IV0}), bootstrap MV = IBP_4_0_IV0,
     *                       latest production = XV_11 (dependency = {MetadataVersion: IBP_4_0_IV1}), MV latest production = IBP_4_0_IV1
     *
     * @param feature the feature to validate.
     * @return true if the feature is valid, false otherwise.
     * @throws IllegalArgumentException if the feature violates any of the rules thus is not valid.
     */
    public static void validateDefaultValueAndLatestProductionValue(
        Feature feature
    ) throws IllegalArgumentException {
        FeatureVersion defaultVersion = feature.defaultVersion(MetadataVersion.LATEST_PRODUCTION);
        FeatureVersion latestProduction = feature.latestProduction;

        if (!feature.hasFeatureVersion(latestProduction)) {
            throw new IllegalArgumentException(String.format("Feature %s has latest production version %s " +
                    "which is not one of its feature versions.", feature.name(), latestProduction));
        }

        if (latestProduction.featureLevel() < defaultVersion.featureLevel()) {
            throw new IllegalArgumentException(String.format("Feature %s has latest production value %s " +
                    "smaller than its default version %s with latest production MV.",
                feature.name(), latestProduction, defaultVersion));
        }

        for (Map.Entry<String, Short> dependency: latestProduction.dependencies().entrySet()) {
            String dependencyFeatureName = dependency.getKey();
            if (!dependencyFeatureName.equals(MetadataVersion.FEATURE_NAME)) {
                Feature dependencyFeature = featureFromName(dependencyFeatureName);
                if (!dependencyFeature.isProductionReady(dependency.getValue())) {
                    throw new IllegalArgumentException(String.format("Feature %s has latest production FeatureVersion %s " +
                            "with dependency %s that is not production ready. (%s latest production: %s)",
                        feature.name(), latestProduction, dependencyFeature.fromFeatureLevel(dependency.getValue(), true),
                        dependencyFeature, dependencyFeature.latestProduction));
                }
            } else {
                if (dependency.getValue() > MetadataVersion.LATEST_PRODUCTION.featureLevel()) {
                    throw new IllegalArgumentException(String.format("Feature %s has latest production FeatureVersion %s " +
                            "with MV dependency %s that is not production ready. (MV latest production: %s)",
                        feature.name(), latestProduction, MetadataVersion.fromFeatureLevel(dependency.getValue()),
                        MetadataVersion.LATEST_PRODUCTION));
                }
            }
        }

        for (MetadataVersion metadataVersion: MetadataVersion.values()) {
            defaultVersion = feature.defaultVersion(metadataVersion);
            for (Map.Entry<String, Short> dependency: defaultVersion.dependencies().entrySet()) {
                String dependencyFeatureName = dependency.getKey();
                if (!dependencyFeatureName.equals(MetadataVersion.FEATURE_NAME)) {
                    Feature dependencyFeature = featureFromName(dependencyFeatureName);
                    if (dependency.getValue() > dependencyFeature.defaultLevel(metadataVersion)) {
                        throw new IllegalArgumentException(String.format("Feature %s has default FeatureVersion %s " +
                                "when MV=%s with dependency %s that is behind its default version %s.",
                            feature.name(), defaultVersion, metadataVersion,
                            dependencyFeature.fromFeatureLevel(dependency.getValue(), true),
                            dependencyFeature.defaultVersion(metadataVersion)));
                    }
                } else {
                    if (dependency.getValue() > defaultVersion.bootstrapMetadataVersion().featureLevel()) {
                        throw new IllegalArgumentException(String.format("Feature %s has default FeatureVersion %s " +
                                "when MV=%s with MV dependency %s that is behind its bootstrap MV %s.",
                            feature.name(), defaultVersion, metadataVersion,
                            MetadataVersion.fromFeatureLevel(dependency.getValue()),
                            defaultVersion.bootstrapMetadataVersion()));
                    }
                }
            }
        }
    }
}
