package com.atlassian.bitbucket.mesh.throttle;

import com.atlassian.bitbucket.mesh.platform.OperatingSystemService;
import com.atlassian.bitbucket.mesh.util.ExpressionUtils;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import io.atlassian.fugue.Pair;
import org.slf4j.Logger;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.core.env.Environment;

import javax.annotation.Nonnull;
import java.time.Clock;
import java.util.*;
import java.util.function.Function;
import java.util.function.Supplier;

import static com.atlassian.bitbucket.mesh.throttle.AdaptiveThrottlingUnavailableException.Reason.*;
import static com.atlassian.bitbucket.mesh.throttle.ThrottlingConstants.*;
import static com.atlassian.bitbucket.mesh.throttle.ThrottlingStrategy.FIXED;
import static com.atlassian.bitbucket.mesh.util.ExpressionUtils.parseExpressionAsDouble;
import static com.atlassian.bitbucket.mesh.util.ExpressionUtils.parseExpressionAsInt;
import static com.atlassian.bitbucket.mesh.util.MoreCollectors.toImmutableMap;
import static com.google.common.base.MoreObjects.firstNonNull;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.lang.Boolean.parseBoolean;
import static java.lang.Double.parseDouble;
import static java.lang.Integer.parseInt;
import static org.apache.commons.lang3.StringUtils.*;
import static org.slf4j.LoggerFactory.getLogger;

/**
 * Loads {@link ResourceThrottleStrategy} for a named ticketed resource.
 */
public class ResourceThrottleStrategyProvider {

    private static final Logger log = getLogger(ResourceThrottleStrategyProvider.class);

    private final Clock clock;
    private final Environment environment;
    private final OperatingSystemService operatingSystemService;

    private Map<ThrottledResource, ResourceThrottleStrategy> resourceStrategies;

    public ResourceThrottleStrategyProvider(Clock clock, Environment environment,
                                            OperatingSystemService operatingSystemService) {
        this.clock = clock;
        this.environment = environment;
        this.operatingSystemService = operatingSystemService;

        resourceStrategies = Collections.emptyMap();
    }

    @Nonnull
    public Map<ThrottledResource, ResourceThrottleStrategy> getAll() {
        return resourceStrategies;
    }

    public void initialize() {
        reload();
    }

    public void reload() {
        int strategyLength = RESOURCE_PROP_SUFFIX_STRATEGY.length();

        resourceStrategies = getThrottleProperties().keySet().stream()
                .map(key -> {
                    int split = key.indexOf('.');
                    if (split == -1) {
                        // found throttle.resource.<resource-name>
                        return key;
                    }
                    if (key.length() == split + strategyLength && key.endsWith(RESOURCE_PROP_SUFFIX_STRATEGY)) {
                        // found throttle.resource.<resource-name>.strategy
                        return key.substring(0, split);
                    }

                    // found an unexpected key with a resource prefix
                    return null;
                })
                .filter(Objects::nonNull)
                .distinct()
                .filter(ResourceThrottleStrategyProvider::isThrottledResource)
                .collect(toImmutableMap(ThrottledResource::forId, this::load));
    }

    @VisibleForTesting
    OptionalLong getJvmMaxMemoryInBytes() {
        long maxMemory = Runtime.getRuntime().maxMemory();
        if (maxMemory == Long.MAX_VALUE) {
            return OptionalLong.empty();
        }
        return OptionalLong.of(maxMemory);
    }

    private static boolean isThrottledResource(String name) {
        for (ThrottledResource resource : ThrottledResource.values()) {
            if (resource.getId().equals(name)) {
                return true;
            }
        }
        return false;
    }

    /**
     * Parses the bucket limit provided by the environment and returns it as an integer.
     *
     * @param resourceName  the name of the resource for which the limit is defined.
     * @param expression    the property value
     * @param fallbackValue the fallback value if the configured value is invalid
     * @return the bucket limit
     */
    private static int parseBucketLimit(String resourceName, String expression, String limitDescription,
                                        Supplier<Integer> fallbackValue) {

        Supplier<Integer> safeFallbackValue = () -> {
            Integer result = fallbackValue.get();
            checkNotNull(result, "[%s] The fallback %s ticket limit cannot be null",
                    resourceName, limitDescription, result);
            checkArgument(result >= 0, "[%s] The fallback %s ticket limit must be positive",
                    resourceName, limitDescription, result);
            return result;
        };

        if (isBlank(expression)) {
            int fallback = safeFallbackValue.get();
            log.warn("[{}] The {} ticket limit property is missing. Defaulting to '{}'",
                    resourceName, limitDescription, fallback);
            return fallback;
        }

        OptionalInt expressionResult = parseExpressionAsInt(expression);
        if (!expressionResult.isPresent()) {
            int fallback = safeFallbackValue.get();
            log.warn("[{}] The {} ticket limit expression '{}' is invalid. Only numbers, " +
                     "+, -, /, *, (, ) and cpu are supported. Defaulting to {}",
                    resourceName, limitDescription, expression, fallback);
            return fallback;
        }

        if (expressionResult.getAsInt() >= 1) {
            log.debug("[{}] Configuration-derived value for {} ticket limit: {}", resourceName, limitDescription,
                    expressionResult.getAsInt());
            return expressionResult.getAsInt();
        }

        log.warn("[{}] The {} ticket limit expression '{}' evaluated to an invalid value ({}). It must evaluate " +
                 "to a positive number. Defaulting to {}", resourceName, limitDescription, expression,
                expressionResult.getAsInt(), 1);
        return 1;
    }

    private int getJvmMaxMemoryOrFail(String resourceName) {
        OptionalLong memory = getJvmMaxMemoryInBytes();

        if (memory.isPresent()) {
            return Ints.saturatedCast(memory.getAsLong() / MEM_1MB);
        }

        log.warn("[{}] The maximum memory required by this JVM cannot be determined by adaptive throttling. " +
                "Selecting fixed throttling instead", resourceName);
        throw new AdaptiveThrottlingUnavailableException(MEMORY_UNAVAILABLE);
    }

    private Map<String, String> getThrottleProperties() {
        String keyPrefix = stripEnd(RESOURCE_PROP_PREFIX, ".");
        return Binder.get(environment)
                .bind(keyPrefix, Bindable.mapOf(String.class, String.class))
                .orElseGet(Collections::emptyMap);
    }

    @Nonnull
    private ResourceThrottleStrategy load(@Nonnull String resourceName) {
        try {
            ThrottlingStrategy strategy = getThrottlingStrategyFor(resourceName);

            if (strategy == ThrottlingStrategy.ADAPTIVE) {
                if (!ThrottledResource.PACK.getId().equals(resourceName)) {
                    log.warn("[{}] Adaptive throttling is not compatible with this resource. Selecting fixed throttling " +
                            "instead", resourceName);
                    throw new AdaptiveThrottlingUnavailableException(ERROR_CONFIGURATION);
                }

                try {
                    // if adaptive throttling can't be used because of insufficient memory etc
                    // AdaptiveThrottlingUnavailableException may also be thrown here
                    return loadAdaptive(resourceName);
                } catch (AdaptiveThrottlingUnavailableException e) {
                    throw e;
                } catch (RuntimeException e) {
                    log.error("[{}]: Error while configuring adaptive throttling. Selecting fixed throttling instead",
                            resourceName, e);
                    throw new AdaptiveThrottlingUnavailableException(ERROR_RUNTIME);
                }
            }
        } catch (AdaptiveThrottlingUnavailableException e) {
            return loadFixedAsFallback(resourceName, e.getFixedTickets(), e.getReason());
        }

        return loadFixed(resourceName);
    }

    private ResourceThrottleStrategy loadAdaptive(String resourceName) {
        int queueTimeoutSeconds = parseTicketAcquireTimeout(resourceName);
        int minTicketsConfigured = getAdaptiveMinTickets(resourceName);
        int maxTicketsConfigured = getAdaptiveMaxTickets(resourceName);
        Pair<Integer, Integer> minMaxTickets = Pair.pair(minTicketsConfigured, maxTicketsConfigured);
        if (isMemoryBackstopEnabled(resourceName)) {
            int maxTicketsSafeForMemory = getMaxTicketsSafeForMemory(resourceName);
            minMaxTickets = chooseValidMinMaxTickets(
                    resourceName, minTicketsConfigured, maxTicketsConfigured, maxTicketsSafeForMemory);
        }
        int noCpuTickets = getFixedTicketLimit(resourceName);
        int intervalSeconds = getAdaptiveIntervalInSeconds(resourceName);
        int noCpuThresholdSeconds = INTERVALS_MAX_WITHOUT_CPU * intervalSeconds;
        double cpuTargetLoad = getCpuTargetLoad(resourceName);
        double cpuPerTypicalTicket = getCpuPerTypicalTicket(resourceName);
        double cpuHistoricalWeighting = getCpuHistoricalWeighting(resourceName);

        log.debug("[{}] Configured adaptive throttling: [range={}..{}, no-cpu-tickets={}, cpu-target={}, " +
                "cpu-per-typical-ticket={}, cpu-historical-weighting={}, interval={}s, queue-timeout={}s, " +
                "no-cpu-threshold={}s]",
                resourceName, minMaxTickets.left(), minMaxTickets.right(), noCpuTickets, cpuTargetLoad,
                cpuPerTypicalTicket, cpuHistoricalWeighting, intervalSeconds, queueTimeoutSeconds,
                noCpuThresholdSeconds);

        return new AdaptiveResourceThrottleStrategy(ThrottledResource.forId(resourceName), clock,
                operatingSystemService, intervalSeconds, minMaxTickets.left(), minMaxTickets.right(),
                noCpuTickets, cpuTargetLoad, cpuPerTypicalTicket, cpuHistoricalWeighting, queueTimeoutSeconds,
                noCpuThresholdSeconds);
    }

    private Pair<Integer, Integer> chooseValidMinMaxTickets(String resourceName, int configMinTickets,
                                                            int configMaxTickets, int memorySafeMaxTickets) {
        int minTickets = configMinTickets;
        int maxTickets = configMaxTickets;

        if (memorySafeMaxTickets < minTickets) {
            log.warn("[{}] This machine's total memory cannot safely support the configured minimum number of " +
                            "tickets ({} configured, {} possible). Selecting fixed throttling instead", resourceName,
                    minTickets, memorySafeMaxTickets);

            throw new AdaptiveThrottlingUnavailableException(MEMORY_INSUFFICIENT);
        }

        if (memorySafeMaxTickets < maxTickets) {
            // only log at INFO level if the adaptive ticket max deviates from the standard ie we can't honour the
            // configurer's request for that many tickets. Note that we can't differentiate between the default and
            // a configured value that is the same as the default
            if (maxTickets != FALLBACK_ADAPTIVE_TICKETS) {
                log.info("[{}] This machine's total available memory cannot safely support the configured maximum of " +
                                "{} tickets. Reducing maximum tickets to {} instead",
                        resourceName, maxTickets, memorySafeMaxTickets);
            } else {
                log.debug("[{}] This machine's total available memory cannot safely support the default maximum of " +
                                "{} tickets. Reducing maximum tickets to {} instead",
                        resourceName, maxTickets, memorySafeMaxTickets);
            }

            maxTickets = memorySafeMaxTickets;
        }

        if (minTickets >= maxTickets) {
            log.warn("[{}] Unable to set minimum tickets to {}. The maximum number of tickets {} is {}. " +
                    "Selecting fixed throttling instead", resourceName, minTickets,
                    maxTickets == configMaxTickets ? "configured" : "permitted for the total memory", maxTickets);

            throw new AdaptiveThrottlingUnavailableException(maxTickets, ERROR_CONFIGURATION);
        }

        log.debug("[{}] Choosing an adaptive ticket range of {}..{}", resourceName, minTickets, maxTickets);

        return Pair.pair(minTickets, maxTickets);
    }

    private int getMaxTicketsSafeForMemory(String resourceName) {
        OptionalLong totalPhysicalMemoryBytes = operatingSystemService.getTotalPhysicalMemory();
        if (!totalPhysicalMemoryBytes.isPresent()) {
            log.warn("[{}] The total memory on this machine cannot be determined. " +
                    "Selecting fixed throttling instead", resourceName);
            throw new AdaptiveThrottlingUnavailableException(MEMORY_UNAVAILABLE);
        }
        log.debug("[{}] The total memory on this machine is {} MB", resourceName,
                totalPhysicalMemoryBytes.getAsLong() / MEM_1MB);

        int typicalMemPerTicket = getTypicalMemPerTicket(resourceName);
        log.debug("[{}] Estimated typical memory per ticket: {} MB", resourceName, typicalMemPerTicket);

        int mainJvmMaxMemory = getWebappMemoryEstimate(resourceName);
        log.debug("[{}] Estimated application webapp requirement: {} MB", resourceName, mainJvmMaxMemory);

        int reservedMemory = getReservedMemory(resourceName);
        log.debug("[{}] Reserved for other applications running on the node: {} MB", resourceName, reservedMemory);

        long nonTicketOperationalMemoryBytes = (mainJvmMaxMemory + reservedMemory) * MEM_1MB;
        if (nonTicketOperationalMemoryBytes > totalPhysicalMemoryBytes.getAsLong()) {
            log.warn("[{}] This machine's total memory cannot support the application memory requirements " +
                    "({} MB required, {} MB total). Selecting fixed throttling instead",
                    resourceName, mainJvmMaxMemory + reservedMemory, totalPhysicalMemoryBytes.getAsLong() / MEM_1MB);

            throw new AdaptiveThrottlingUnavailableException(MEMORY_INSUFFICIENT);
        }

        long ticketOperationalMemoryBytes = totalPhysicalMemoryBytes.getAsLong() - nonTicketOperationalMemoryBytes;
        log.debug("[{}] Estimated combined application webapp and search requirements: {} MB", resourceName,
                nonTicketOperationalMemoryBytes / MEM_1MB);
        log.debug("[{}] Estimated remaining memory available to ticketed operations: {} MB", resourceName,
                ticketOperationalMemoryBytes / MEM_1MB);

        int memorySafeMax = (int) Math.floor(ticketOperationalMemoryBytes / (typicalMemPerTicket * MEM_1MB));
        log.debug("[{}] Estimated safe max tickets: {}", resourceName, memorySafeMax);
        return memorySafeMax;
    }

    private boolean isMemoryBackstopEnabled(String resourceName) {
        boolean fallback = FALLBACK_ADAPTIVE_MEM_BACKSTOP_ENABLED;

        String value = trimToNull(resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_ADAPTIVE_MEM_BACKSTOP_ENABLED));
        if (value == null) {
            return fallback;
        }

        boolean result = parseBoolean(value);
        if (!result) {
            log.debug("[{}] Memory backstop has been disabled (value: '{}'). Adaptive throttling will not keep the " +
                    "ticket limit min/max within safe limits for the machine's total memory'",
                    resourceName, value);
        }
        return result;
    }

    private FixedResourceThrottleStrategy loadFixed(String resourceName) {
        int tickets = getFixedTicketLimit(resourceName);
        int queueTimeoutSeconds = parseTicketAcquireTimeout(resourceName);

        log.debug("[{}] Configured fixed throttling: [tickets={}, queue-timeout={}s]",
                resourceName, tickets, queueTimeoutSeconds);

        return new FixedResourceThrottleStrategy(ThrottledResource.forId(resourceName), queueTimeoutSeconds, tickets);
    }

    private FixedResourceThrottleStrategy loadFixedAsFallback(String resourceName, OptionalInt suggestedFixedTickets,
                                                              AdaptiveThrottlingUnavailableException.Reason fallbackReason) {

        int tickets = suggestedFixedTickets.orElseGet(() -> getFixedTicketLimit(resourceName));
        int queueTimeoutSeconds = parseTicketAcquireTimeout(resourceName);

        log.debug("[{}] Configured fixed throttling as a fallback: " +
                "[tickets={}, queue-timeout={}s, reason={}]", resourceName, tickets, queueTimeoutSeconds, fallbackReason);

        return new FixedResourceThrottleStrategy(ThrottledResource.forId(resourceName), queueTimeoutSeconds, tickets);
    }

    private String resourceProperty(String resourceName, String suffix) {
        return environment.getProperty(RESOURCE_PROP_PREFIX + resourceName + suffix);
    }

    private double getCpuTargetLoad(String resourceName) {
        double fallback = FALLBACK_ADAPTIVE_TARGET_CPU_USAGE;

        String value = resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_ADAPTIVE_CPU_TARGET_USAGE);
        if (value == null) {
            log.debug("[{}] The adaptive throttling target CPU usage is unspecified. Defaulting to '{}%'",
                    resourceName, fallback * 100);
        } else {
            try {
                double targetCpuUsage = parseDouble(value);
                if (targetCpuUsage >= 0.0d && targetCpuUsage <= 1.0d) {
                    return targetCpuUsage;
                }
            } catch (NumberFormatException e) {
                // warning logged below
            }
            log.warn("[{}] Invalid adaptive throttling target CPU load of '{}' specified. This value must be a floating " +
                    "point number between 0.0 and 1.0. Defaulting to '{}' ('{}%')",
                    resourceName, value, fallback, fallback * 100);
        }

        return fallback;
    }

    private double getCpuPerTypicalTicket(String resourceName) {
        double fallback = FALLBACK_ADAPTIVE_CPU_PER_TICKET;

        String expression = resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_ADAPTIVE_CPU_PER_TICKET);
        if (expression == null) {
            log.debug("[{}] The value for the typical CPU usage per ticketed operation is unspecified. " +
                    "Defaulting to '{}' ('{}%')", resourceName, fallback, fallback * 100);
        } else {
            OptionalDouble typicalCpuUsage = parseExpressionAsDouble(expression);
            if (typicalCpuUsage.isPresent() &&
                    typicalCpuUsage.getAsDouble() > 0.0d && typicalCpuUsage.getAsDouble() <= 1.0d) {
                return typicalCpuUsage.getAsDouble();
            }

            log.warn("[{}] Invalid expression '{}' specified for the typical CPU usage per ticketed operation. " +
                    "This must evaluate to a positive value of 1.0 or lower (evaulated as {}). Defaulting to '{}' ('{}%')",
                    resourceName, expression, typicalCpuUsage.isPresent() ? typicalCpuUsage.getAsDouble() : "<error>",
                    fallback, fallback * 100);
        }

        return fallback;
    }

    private double getCpuHistoricalWeighting(String resourceName) {
        double fallback = FALLBACK_ADAPTIVE_CPU_HISTORICAL_WEIGHTING;

        String value = resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_ADAPTIVE_CPU_HISTORICAL_WEIGHTING);
        if (value == null) {
            log.debug("[{}] The adaptive throttling CPU historical weighting is unspecified. Defaulting to '{}'",
                    resourceName, fallback);
        } else {
            try {
                double baseCpuUsage = parseDouble(value);
                if (baseCpuUsage >= 0.0d && baseCpuUsage < 1.0d) {
                    return baseCpuUsage;
                }
            } catch (NumberFormatException e) {
                // warning logged below
            }
            log.warn("[{}] Invalid adaptive throttling CPU historical weighting of '{}' specified. This value must be a " +
                    "floating point number between 0.0 (inclusive) and 1.0 (exclusive). Defaulting to '{}'",
                    resourceName, value, fallback);
        }

        return fallback;
    }

    private int getAdaptiveIntervalInSeconds(String resourceName) {
        int fallback = FALLBACK_ADAPTIVE_INTERVAL_SECONDS;

        String value = resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_ADAPTIVE_INTERVAL);
        if (value == null) {
            log.debug("[{}] The adaptive throttling interval property is unspecified. Defaulting to '{}' seconds",
                    resourceName, fallback);
        } else {
            try {
                int intervalSeconds = parseInt(value);
                if (intervalSeconds > 0) {
                    return intervalSeconds;
                }
            } catch (NumberFormatException e) {
                // warning logged below
            }
            log.warn("[{}] Invalid adaptive throttling interval of '{}' specified. Defaulting to '{}' seconds",
                    resourceName, value, fallback);
        }

        return fallback;
    }

    private int getAdaptiveMaxTickets(String resourceName) {
        String expression = resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_ADAPTIVE_LIMIT_MAX);
        return parseBucketLimit(resourceName, expression, "maximum", this::getDefaultAdaptiveMaxTickets);
    }

    private int getAdaptiveMinTickets(String resourceName) {
        String expression = resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_ADAPTIVE_LIMIT_MIN);
        return parseBucketLimit(resourceName, expression, "minimum", this::getDefaultAdaptiveMinTickets);
    }

    private int getDefaultAdaptiveMaxTickets() {
        return (int) Math.round(FALLBACK_ADAPTIVE_CPU_SCALE_MAX_TICKETS * Runtime.getRuntime().availableProcessors());
    }

    private int getDefaultAdaptiveMinTickets() {
        return (int) Math.round(FALLBACK_ADAPTIVE_CPU_SCALE_MIN_TICKETS * Runtime.getRuntime().availableProcessors());
    }

    private int getFixedTicketLimit(String resourceName) {
        int fallbackValue = firstNonNull(FALLBACK_FIXED_TICKETS.get(ThrottledResource.forId(resourceName)), 1);

        // fixed throttling was chosen because one of:
        // 1. the deprecated property throttle.resource.resource-name=12 was specified
        // 2. throttle.resource.resource-name.strategy=fixed
        // 3. we've fallen back for some reason
        //
        // in all cases it is safe to try to take the fixed limit throttle.resource.resource-name first before
        // throttle.resource.resource-name.fixed.limit
        String expression = firstNonNull(
                // try format: throttle.resource.resource-name=12
                trimToNull(environment.getProperty(RESOURCE_PROP_PREFIX + resourceName)),
                // try format: throttle.resource.resource-name.fixed.limit=12
                firstNonNull(resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_FIXED_LIMIT), ""));

        if (expression.isEmpty()) {
            return fallbackValue;
        }

        return parseBucketLimit(resourceName, expression, "fixed", () -> fallbackValue);
    }

    private int getReservedMemory(String resourceName) {
        return getEstimatedMemoryValueInMb(resourceName, "reserved memory size",
                RESOURCE_PROP_SUFFIX_ADAPTIVE_MEM_RESERVED, ExpressionUtils::parseExpressionAsInt,
                () -> 0);
    }

    /**
     * If "throttle.resource.scm-hosting" is set to a non-empty value then fixed throttling is chosen to allow bridging
     * the fixed throttling configuration in the deprecated format to the strategy selection and configuration.
     * Otherwise base it on "throttle.resource.scm-hosting.strategy".
     */
    private ThrottlingStrategy getThrottlingStrategyFor(String resourceName) {
        String deprecatedFixedLimit = environment.getProperty(RESOURCE_PROP_PREFIX + resourceName);
        // a null value means the property is absent but an empty value means it's present so even then we should
        // switch to the deprecated syntax
        if (deprecatedFixedLimit != null) {
            // only scm-hosting allows adaptive throttling and the old syntax hasn't been deprecated for those yet
            // so let's only log for what makes sense to the customer
            if (resourceName.equals(ThrottledResource.PACK.getId())) {
                String throttlingPropertiesUrl =
                        //BSERVDEV-14170: use this instead once fixed:
                        //helpPathService.getPageUrl("bitbucket.help.config.properties.throttling");
                        "https://confluence.atlassian.com/display/BitbucketServer/Bitbucket+Server+config+properties" +
                                "#BitbucketServerconfigproperties-throttling";
                log.info("A custom configuration for the {} tickets was detected and will be used. Please note that " +
                        "you may get better throughput by configuring adaptive throttling. See {} for more details.",
                        resourceName, throttlingPropertiesUrl);
            }
            return FIXED;
        }

        String value = resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_STRATEGY);

        if (value == null) {
            log.debug("[{}] No throttling strategy specified. Selecting {} throttling", resourceName, FIXED.getId());
            return FIXED;
        }

        return ThrottlingStrategy.fromId(value.toLowerCase(Locale.ROOT)).orElseGet(() -> {
                log.warn("[{}] Invalid throttling strategy: '{}'. Selecting {} throttling",
                        resourceName, value, FIXED.getId());
            throw new AdaptiveThrottlingUnavailableException(ERROR_CONFIGURATION);
        });
    }

    private int getTypicalMemPerTicket(String resourceName) {
        return getEstimatedMemoryValueInMb(resourceName, "estimated memory usage per ticketed operation",
                RESOURCE_PROP_SUFFIX_ADAPTIVE_MEM_PER_TICKET, ExpressionUtils::parseExpressionAsInt,
                () -> FALLBACK_ADAPTIVE_MEM_PER_TICKET);
    }

    private int getWebappMemoryEstimate(String resourceName) {
        Function<String, OptionalInt> toMemory = value -> {
            if (value.trim().equalsIgnoreCase(RESOURCE_PROP_VALUE_ADAPTIVE_MEM_WEBAPP_JVM_MAX)) {
                return OptionalInt.of(getJvmMaxMemoryOrFail(resourceName));
            } else {
                return parseExpressionAsInt(value);
            }
        };

        return getEstimatedMemoryValueInMb(resourceName, "application webapp size",
                RESOURCE_PROP_SUFFIX_ADAPTIVE_MEM_WEBAPP, toMemory,
                () -> getJvmMaxMemoryOrFail(resourceName));
    }

    private int getEstimatedMemoryValueInMb(String resourceName, String description, String property,
                                            Function<String, OptionalInt> converter, Supplier<Integer> fallback) {
        String value = resourceProperty(resourceName, property);
        if (value == null) {
            int fallbackValue = fallback.get();
            log.debug("[{}] The estimated adaptive throttling {} is missing. Defaulting to '{} MB'",
                    resourceName, description, fallbackValue);
            return fallbackValue;
        }

        try {
            OptionalInt memory = converter.apply(value);
            if (memory.isPresent() && memory.getAsInt() >= 0L) {
                return memory.getAsInt();
            }
        } catch (NumberFormatException e) {
            // warning logged below
        }

        int fallbackValue = fallback.get();
        log.warn("[{}] Invalid estimate adaptive throttling {} value of '{}' specified. " +
                "Defaulting to '{}' MB",
                resourceName, description, value, fallbackValue);

        return fallbackValue;
    }

    private int parseTicketAcquireTimeout(String resourceName) {
        int fallback = firstNonNull(FALLBACK_TICKET_TIMEOUT_SECONDS.get(ThrottledResource.forId(resourceName)), 0);

        String value = resourceProperty(resourceName, RESOURCE_PROP_SUFFIX_TIMEOUT);

        if (value == null) {
            log.debug("[{}] Ticket acquire timeout is unspecified. Defaulting to '{}' seconds", resourceName, fallback);
            return fallback;
        }

        try {
            return parseInt(value);
        } catch (NumberFormatException e) {
            log.warn("[{}] Invalid ticket acquire timeout of '{}' seconds. Defaulting to '{}' seconds",
                    resourceName, value, fallback);
        }
        return fallback;
    }
}
