package com.atlassian.bitbucket.mesh.throttle;

import com.atlassian.bitbucket.mesh.platform.OperatingSystemService;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.text.DecimalFormat;
import java.time.Clock;
import java.util.OptionalDouble;
import java.util.OptionalLong;
import java.util.concurrent.TimeUnit;

import static com.atlassian.bitbucket.mesh.throttle.ThrottlingConstants.LOGGER_PREFIX_THROTTLING_PROFILE;
import static com.atlassian.bitbucket.mesh.throttle.ThrottlingOutputUtils.newBriefNumberFormat;
import static com.atlassian.bitbucket.mesh.throttle.ThrottlingOutputUtils.toHumanDuration;
import static java.util.concurrent.TimeUnit.HOURS;
import static java.util.concurrent.TimeUnit.SECONDS;

/**
 * Strategy for the {@link ThrottlingStrategy#ADAPTIVE adaptive throttling}
 */
class AdaptiveResourceThrottleStrategy implements ResourceThrottleStrategy {

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

    private static final long FALLBACK_WARN_INTERVAL = HOURS.toMillis(1);

    private final int acquireTimeoutSeconds;
    private final Clock clock;
    private final double cpuHistoricalWeighting;
    private final int intervalSeconds;
    private final int maxTickets;
    private final int minTickets;
    private final int fallbackTickets;
    private final OperatingSystemService operatingSystemService;
    private final ThrottledResource resource;
    private final double targetCpuLoad;
    private final int fallbackThresholdSeconds;
    private final Logger profileLogger;
    private final double typicalCpuPerTicket;

    private volatile State state;

    AdaptiveResourceThrottleStrategy(ThrottledResource resource, Clock clock,
                                     OperatingSystemService operatingSystemService, int intervalSeconds, int minTickets,
                                     int maxTickets, int fallbackTickets, double targetCpuLoad,
                                     double typicalCpuPerTicket, double cpuHistoricalWeighting,
                                     int acquireTimeoutSeconds, int fallbackThresholdSeconds) {

        this.typicalCpuPerTicket = typicalCpuPerTicket;
        this.acquireTimeoutSeconds = acquireTimeoutSeconds;
        this.clock = clock;
        this.cpuHistoricalWeighting = cpuHistoricalWeighting;
        this.intervalSeconds = intervalSeconds;
        this.maxTickets = maxTickets;
        this.minTickets = minTickets;
        this.fallbackTickets = fallbackTickets;
        this.operatingSystemService = operatingSystemService;
        this.resource = resource;
        this.targetCpuLoad = targetCpuLoad;
        this.fallbackThresholdSeconds = fallbackThresholdSeconds;

        // a dedicated logger for profiling of throttling changes. OFF by default. it should be directed to a separate
        // file to not pollute the application log
        profileLogger = LoggerFactory.getLogger(LOGGER_PREFIX_THROTTLING_PROFILE + resource);

        state = new StartingState(fallbackTickets, clock.millis());
    }

    @Nonnull
    @Override
    public SemaphoreTicketBucket create() {
        // assume the fallback position to start with where have no evidence the CPU load is available.
        // subsequent throttling intervals will adjust the tickets up or down provided a CPU load value is available
        return new SemaphoreTicketBucket(resource, fallbackTickets, acquireTimeoutSeconds, SECONDS);
    }

    @Nonnull
    @Override
    public OptionalLong getUpdateInterval(@Nonnull TimeUnit timeUnit) {
        return OptionalLong.of(SECONDS.convert(intervalSeconds, timeUnit));
    }

    @Override
    public String toString() {
        return "[" +
                "state=" + state + ", " +
                "config=[" +
                "ticket-range=[" + minTickets + ".." + maxTickets + "], " +
                "target-cpu=" + targetCpuLoad + ", " +
                "fallback-tickets=" + fallbackTickets + ", " +
                "fallback-timeout=" + fallbackThresholdSeconds + "s, " +
                "acquire-timeout=" + acquireTimeoutSeconds + "s" +
                "]" +
                "]";
    }

    @Override
    public void update(@Nonnull SemaphoreTicketBucket bucket) {
        long now = clock.millis();

        State nextState = null;
        try {
            bucket.setAcquireTimeoutMillis(SECONDS.toMillis(acquireTimeoutSeconds));

            TicketSummary ticketSummary = bucket.summarize();
            OptionalDouble optCpuLoad = operatingSystemService.getCpuLoad();

            int bucketTickets = ticketSummary.getTotal();
            int previousTickets = (int) Math.round(state.tickets);
            if (updateTicketsIfDifferent(bucket, bucketTickets, previousTickets)) {
                log.warn("[{}] Throttling ticket limit ({}) and the tickets on the bucket ({}) have diverged since " +
                        "the last interval", resource, previousTickets, bucketTickets);
            }

            nextState = state.transition(now, optCpuLoad, ticketSummary);
        } catch (RuntimeException e) {
            log.error("[{}] Error encountered during adaptive throttling update. Setting tickets to {} as a fallback " +
                    "measure", resource, fallbackTickets, e);
        }

        if (nextState == null) {
            updateTicketsIfDifferent(bucket, fallbackTickets, (int) Math.round(state.tickets));
            return;
        }

        updateTicketsIfDifferent(bucket, (int) Math.round(nextState.tickets), (int) Math.round(state.tickets));

        state = nextState;
    }

    private boolean updateTicketsIfDifferent(@Nonnull SemaphoreTicketBucket bucket, int newValue, int oldValue) {
        if (oldValue == newValue) {
            return false;
        }

        bucket.setPermits(newValue);
        return true;
    }

    @VisibleForTesting
    int getAcquireTimeoutSeconds() {
        return acquireTimeoutSeconds;
    }

    @VisibleForTesting
    double getCpuHistoricalWeighting() {
        return cpuHistoricalWeighting;
    }

    @VisibleForTesting
    int getMaxTickets() {
        return maxTickets;
    }

    @VisibleForTesting
    int getMinTickets() {
        return minTickets;
    }

    @VisibleForTesting
    int getFallbackTickets() {
        return fallbackTickets;
    }

    @VisibleForTesting
    int getFallbackThresholdSeconds() {
        return fallbackThresholdSeconds;
    }

    @VisibleForTesting
    double getTargetCpuLoad() {
        return targetCpuLoad;
    }

    @VisibleForTesting
    int getTotalTickets() {
        return (int) Math.round(state.tickets);
    }

    @VisibleForTesting
    double getTypicalCpuPerTicket() {
        return typicalCpuPerTicket;
    }

    @VisibleForTesting
    boolean isAdapting() {
        return state instanceof AdaptiveState || state instanceof StartingState;
    }

    @VisibleForTesting
    boolean isDisabled() {
        return state instanceof FixedState;
    }

    @VisibleForTesting
    boolean isPaused() {
        return state instanceof PausedState;
    }

    private void logCpuAvailable() {
        profileLogger.trace("[{}] The system CPU load is available again", resource);
        log.info("[{}] The system CPU load is available again. Adaptive throttling will now resume", resource);
    }

    private void logMissingCpuLoad(long duration) {
        profileLogger.trace("[{}] The system CPU load is still not available. Setting a fixed ticket limit of {}",
                resource, fallbackTickets);
        log.warn("[{}] The system CPU load has not been available for {}. " +
                        "A fixed ticket limit of {} is in place until the reading is next available",
                resource, toHumanDuration(duration), fallbackTickets);
    }

    private void logUpdate(AdaptiveState nextState, double cpuLoad, double targetTickets, TicketSummary ticketSummary) {
        if (profileLogger.isTraceEnabled()) {
            DecimalFormat briefly = newBriefNumberFormat();

            long sinceLastUpdate = state.sinceUpdate(nextState.timestamp);

            int usedTickets = ticketSummary.getUsed();
            int totalTickets = ticketSummary.getTotal();
            int ticketQueueLength = ticketSummary.getQueuedRequests();
            OptionalDouble previousSmoothedCpu = state.smoothedCpuLoad;
            OptionalDouble nextSmoothedCpu = nextState.smoothedCpuLoad;

            profileLogger.trace("[{}] Ticket: {}->{}, Smoothed CPU: {}->{}, Inputs: " +
                            "[" +
                            "cpu={}, " +
                            "used-tickets={}/{}, ticket-queue-length={}, " +
                            "target-tickets={}, since-last-update={}" +
                            "]",
                    resource, state.tickets, nextState.tickets,
                    previousSmoothedCpu.isPresent() ? briefly.format(previousSmoothedCpu.getAsDouble()) : "unknown",
                    briefly.format(nextSmoothedCpu.getAsDouble()), cpuLoad, usedTickets, totalTickets,
                    ticketQueueLength, briefly.format(targetTickets), toHumanDuration(sinceLastUpdate));
        }
    }

    /**
     * Base class for the various states adaptive throttling can be in
     */
    abstract class State {

        final double tickets;
        final long timestamp;
        final OptionalDouble smoothedCpuLoad;

        public State(OptionalDouble smoothedCpuLoad, double tickets, long timestamp) {
            this.smoothedCpuLoad = smoothedCpuLoad;
            this.tickets = tickets;
            this.timestamp = timestamp;
        }

        public abstract State transition(long now, OptionalDouble cpuLoad, TicketSummary ticketSummary);

        /**
         * @return the duration in millis from this state to the supplied parameter as a UTC timestamp
         */
        public long sinceUpdate(long when) {
            return when - timestamp;
        }

        @Override
        public String toString() {
            return MoreObjects.toStringHelper(this)
                    .add("tickets", tickets)
                    .add("timestamp", timestamp)
                    .add("smoothedCpuLoad", smoothedCpuLoad.isPresent() ? smoothedCpuLoad.getAsDouble() : "unknown")
                    .toString();
        }

        /**
         * @return a smoothed CPU value based of the {@link #smoothedCpuLoad current value} and the supplied value
         * and the {@link #cpuHistoricalWeighting historical weighting}. If an existing smoothed value is not present
         */
        OptionalDouble smooth(double cpuLoad) {
            // if there was no previous smoothing value, the new value is its initial value
            if (!smoothedCpuLoad.isPresent()) {
                return OptionalDouble.of(cpuLoad);
            }

            return OptionalDouble.of(((1d - cpuHistoricalWeighting) * cpuLoad)
                    + (cpuHistoricalWeighting * smoothedCpuLoad.getAsDouble()));
        }

        State transitionToAdaptive(long now, double cpuLoad, TicketSummary ticketSummary) {
            int curUsedTickets = ticketSummary.getUsed();

            double smoothedCpuLoad = smooth(cpuLoad).getAsDouble();
            double targetTickets = curUsedTickets + ((targetCpuLoad - smoothedCpuLoad) / typicalCpuPerTicket);
            int tickets = Math.min(maxTickets, Math.max(minTickets, (int) Math.round(targetTickets)));

            AdaptiveState update = new AdaptiveState(smoothedCpuLoad, tickets, now);

            logUpdate(update, cpuLoad, targetTickets, ticketSummary);

            return update;
        }

        /**
         * @param now the current time
         * @param deadline the deadline by which this state should have been transitioned from. If after {@code now}
         *                 then the deadline has passed and the state must transition to
         *                 {@link FixedState fixed throttling}
         * @param timestamp the timestamp for the fixed state - normally the same as {@code now} but may be different
         *                  if transitioning from fixed to fixed throttling but with an altered deadline (the deadline
         *                  is used for checking if CPU load has become available but also for periodically logging a
         *                  reminder that it is still unavailable
         * @return the new or existing state depenbding on whether the deadline has passed
         */
        State transitionToFixedIfAfter(long now, long deadline, long timestamp) {
            if (now < deadline) {
                return this;
            }

            logMissingCpuLoad(sinceUpdate(now));

            return new FixedState(now + FALLBACK_WARN_INTERVAL, fallbackTickets, timestamp);
        }
    }

    /**
     * The initial state before the first change to adapt to the CPU load and ticket count. If CPU load is unavailable
     * on the first update, adaptive throttling is {@link PausedState paused} until it is available otherwise
     * {@link AdaptiveState adaptive changes} are made.
     */
    class StartingState extends State {

        public StartingState(double tickets, long timestamp) {
            super(OptionalDouble.empty(), tickets, timestamp);
        }

        @Override
        public State transition(long now, OptionalDouble cpuLoad, TicketSummary ticketSummary) {
            if (!cpuLoad.isPresent()) {
                return new PausedState(now + SECONDS.toMillis(fallbackThresholdSeconds), tickets, now);
            }

            return transitionToAdaptive(now, cpuLoad.getAsDouble(), ticketSummary);
        }
    }

    /**
     * When adaptive throttling is paused temporarily because of missing CPU load data. If the CPU load data is
     * persistently unavailable, throttling reverts to {@link FixedState fixed}. If the CPU load data
     * becomes available again, throttling switches back to {@link AdaptiveState adaptive}.
     */
    class PausedState extends State {

        private final long fallbackAfter;

        public PausedState(long fallbackAfter, double tickets, long timestamp) {
            super(OptionalDouble.empty(), tickets, timestamp);
            this.fallbackAfter = fallbackAfter;
        }

        @Override
        public State transition(long now, OptionalDouble cpuLoad, TicketSummary ticketSummary) {
            if (!cpuLoad.isPresent()) {
                return transitionToFixedIfAfter(now, fallbackAfter, now);
            }

            profileLogger.trace("[{}] The system CPU load is available again", resource);
            log.info("[{}] The system CPU load is available again. Adaptive throttling will now resume", resource);

            return transitionToAdaptive(now, cpuLoad.getAsDouble(), ticketSummary);
        }

        @Override
        public String toString() {
            return MoreObjects.toStringHelper(this)
                    .add("fallbackAfter", fallbackAfter)
                    .add("tickets", tickets)
                    .add("timestamp", timestamp)
                    .add("smoothedCpuLoad", smoothedCpuLoad.isPresent() ? smoothedCpuLoad.getAsDouble() : "unknown")
                    .toString();
        }
    }

    /**
     * When throttling can't adapt to changing CPU load and used tickets because CPU load is unavailable or a previous
     * adaptive update fail with an exception
     */
    class FixedState extends State {

        private final long nextWarning;

        public FixedState(long nextWarning, double tickets, long timestamp) {
            super(OptionalDouble.empty(), tickets, timestamp);
            this.nextWarning = nextWarning;
        }

        @Override
        public State transition(long now, OptionalDouble cpuLoad, TicketSummary ticketSummary) {
            if (!cpuLoad.isPresent()) {
                // keep the existing timestamp - we're only only updating the nextWarning deadline
                return transitionToFixedIfAfter(now, nextWarning, timestamp);
            }

            logCpuAvailable();

            return transitionToAdaptive(now, cpuLoad.getAsDouble(), ticketSummary);
        }

        @Override
        public String toString() {
            return MoreObjects.toStringHelper(this)
                    .add("nextWarning", nextWarning)
                    .add("tickets", tickets)
                    .add("timestamp", timestamp)
                    .add("smoothedCpuLoad", smoothedCpuLoad.isPresent() ? smoothedCpuLoad.getAsDouble() : "unknown")
                    .toString();
        }
    }

    /**
     * When adaptive throttling is operating normally and responding to changes in used tickets and CPU load
     */
    class AdaptiveState extends State {

        private AdaptiveState(double smoothedCpuLoad, double tickets, long timestamp) {
            super(OptionalDouble.of(smoothedCpuLoad), tickets, timestamp);
        }

        @Override
        public State transition(long now, OptionalDouble cpuLoad, TicketSummary summary) {
            if (!cpuLoad.isPresent()) {
                return new PausedState(now + SECONDS.toMillis(fallbackThresholdSeconds), tickets, now);
            }

            return transitionToAdaptive(now, cpuLoad.getAsDouble(), summary);
        }
    }
}
