/*
 * 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.jackrabbit.oak.plugins.document.mongo;

import org.apache.commons.math3.util.Precision;
import org.apache.jackrabbit.oak.plugins.document.Throttler;
import org.jetbrains.annotations.NotNull;

import java.util.concurrent.atomic.AtomicReference;

import static java.util.Objects.requireNonNull;
import static org.apache.jackrabbit.oak.plugins.document.Throttler.NO_THROTTLING;

/**
 * Factory to create Mongo Throttlers
 */
public final class MongoThrottlerFactory {

    private MongoThrottlerFactory() {
    }

    /**
     * Returns an instance of @{@link Throttler} which throttles the system exponentially based on remaining threshold defined.
     * @param threshold threshold for throttling
     * @param oplogWindow current oplog window for mongo
     * @param throttlingTime time duration for throttling
     * @return an exponential throttler to throttle system if required
     */
    public static Throttler exponentialThrottler(final int threshold, final AtomicReference<Double> oplogWindow, final long throttlingTime) {
        requireNonNull(oplogWindow);
        return new ExponentialThrottler(threshold, oplogWindow, throttlingTime);
    }

    /**
     * Returns an instance of @{@link Throttler} which throttles the system based on throttling factor set externally
     *
     * @param factor         current oplog window for mongo
     * @param time time duration for throttling
     * @return an external factor throttler to throttle the system if required
     */
    public static Throttler extFactorThrottler(final AtomicReference<Integer> factor, final long time) {
        requireNonNull(factor);
        return new ExtFactorThrottler(factor, time);
    }

    /**
     * A {@link Throttler} which doesn't do any throttling, no matter how much system is under load
     * @return No throttler
     */
    public static Throttler noThrottler() {
        return NO_THROTTLING;
    }

    private static class ExponentialThrottler implements Throttler {

        private final int threshold;
        @NotNull
        private final AtomicReference<Double> oplogWindow;
        private final long throttlingTime;

        public ExponentialThrottler(final int threshold, final @NotNull AtomicReference<Double> oplogWindow, final long throttlingTime) {
            this.threshold = threshold;
            this.oplogWindow = oplogWindow;
            this.throttlingTime = throttlingTime;
        }

        /**
         * The time duration (in Millis) for which we need to throttle the system.
         *
         * @return the throttling time duration (in Millis)
         */
        @Override
        public long throttlingTime() {
            final double threshold = this.threshold;
            final double oplogWindow = this.oplogWindow.get();
            long throttleTime = throttlingTime;

            if (Precision.compareTo(oplogWindow,threshold/8,  0.001) <= 0) {
                throttleTime = throttleTime * 8;
            } else if (Precision.compareTo(oplogWindow,threshold/4, 0.001) <= 0) {
                throttleTime = throttleTime * 4;
            } else if (Precision.compareTo(oplogWindow, threshold/2, 0.001) <= 0) {
                throttleTime = throttleTime * 2;
            } else if (Precision.compareTo(oplogWindow, threshold,0.001) <= 0) {
                throttleTime = throttlingTime;
            } else {
                throttleTime = 0;
            }
            return throttleTime;
        }
    }

    private static class ExtFactorThrottler implements Throttler {

        @NotNull
        private final AtomicReference<Integer> factor;
        private final long time;

        public ExtFactorThrottler(final @NotNull AtomicReference<Integer> factor, final long time) {
            this.factor = factor;
            this.time = time;
        }

        /**
         * The time duration (in Millis) for which we need to throttle the system.
         *
         * @return the throttling time duration (in Millis)
         */
        @Override
        public long throttlingTime() {

            if (factor.get() <= 0) {
                return 0;
            }

            return time * factor.get();
        }
    }
}
