package pl.allegro.tech.hermes.infrastructure.logback;

import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.turbo.TurboFilter;
import ch.qos.logback.core.spi.FilterReply;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Marker;
import org.slf4j.MarkerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * <p>This class is an implementation of {@link TurboFilter} interface that allows messages of configured loggers
 * to be aggregated and logged only once with specified time interval.</p>
 *
 * <p>Logged message contains a suffix "[occurrences=N]" with a number of logged events.
 * The logging part is a little bit tricky, as the <code>TurboFilter</code> interface does not support a way of logging out of the box,
 * it just tells whether a logged message should be passed further or not.
 * In order to logging be possible from the <code>TurboFilter</code> a small trick has to be applied, the logged message is enriched
 * with a custom {@link Marker} which is checked by the filter itself, so it won't filter it's own messages
 * and we manage to avoid recursion.</p>
 *
 * <p>An instance of <code>AggregatingTurboFilter</code> starts it's own scheduled executor service with a single thread
 * that logs the messages asynchronously.</p>
 *
 * <p>Example logback configuration:
 * <pre>{@code
 *  <configuration>
 *      ...
 *      <turboFilter class="pl.allegro.tech.hermes.infrastructure.logback.AggregatingTurboFilter">
 *          <reportingIntervalMillis>10000</reportingIntervalMillis>
 *          <aggregatedLogger>org.apache.zookeeper.ClientCnxn</aggregatedLogger>
 *          <aggregatedLogger>pl.allegro.tech.hermes.common.message.wrapper.MessageContentWrapper</aggregatedLogger>
 *          ...
 *      </turboFilter>
 *      ...
 * </configuration>
 * }</pre>
 */
public class AggregatingTurboFilter extends TurboFilter {

    static final Marker MARKER = MarkerFactory.getMarker("AggregatingTurboFilterMarker");

    private ScheduledExecutorService executorService;
    private final List<String> aggregatedLogger = new ArrayList<>();
    private long reportingIntervalMillis = 10_000;

    private final Map<Logger, LoggerAggregates> logAggregates = new ConcurrentHashMap<>();

    private static final LongAdder filterClassCounter = new LongAdder();

    @Override
    public void start() {
        super.start();
        if (!aggregatedLogger.isEmpty()) {
            ThreadFactory threadFactory = new ThreadFactoryBuilder()
                    .setNameFormat("aggregating-filter-" + filterClassCounter.longValue() + "-thread-%d")
                    .build();
            filterClassCounter.increment();
            executorService = Executors.newSingleThreadScheduledExecutor(threadFactory);
            executorService.scheduleAtFixedRate(this::report, 0L, getReportingIntervalMillis(), TimeUnit.MILLISECONDS);
        }
    }

    @Override
    public void stop() {
        executorService.shutdownNow();
        super.stop();
    }

    void report() {
        logAggregates.forEach((logger, loggerAggregates) ->
                loggerAggregates.aggregates.keySet().forEach(entry -> {
                    loggerAggregates.aggregates.computeIfPresent(entry, (key, summary) -> {
                        logger.log(key.marker, Logger.FQCN, key.level,
                                key.message + " [occurrences=" + summary.logsCount + "]",
                                key.params, summary.lastException);
                        return null;
                    });
                }));
    }

    @Override
    public FilterReply decide(Marker marker, Logger logger, Level level, String message, Object[] params, Throwable ex) {
        if (isAggregatedLog(marker)) { // prevent recursion for aggregated logger events
            return FilterReply.NEUTRAL;
        }
        if (!aggregatedLogger.contains(logger.getName())) {
            return FilterReply.NEUTRAL;
        }

        if (ex == null) {
            Optional<Throwable> throwable = extractLastParamThrowable(params);
            if (throwable.isPresent()) {
                ex = throwable.get();
                params = Arrays.copyOfRange(params, 0, params.length - 1);
            }
        }
        Throwable exception = ex;

        LoggingEventKey loggingEventKey = new LoggingEventKey(message, params, level, getEnrichedMarker(marker));
        logAggregates.computeIfAbsent(logger, l -> new LoggerAggregates())
                .aggregates.merge(loggingEventKey, new AggregateSummary(ex),
                (currentAggregate, emptyAggregate) -> AggregateSummary.incrementCount(currentAggregate, exception));

        return FilterReply.DENY;
    }

    private boolean isAggregatedLog(Marker marker) {
        return marker != null && (marker.equals(MARKER) || marker.contains(MARKER));
    }

    private Optional<Throwable> extractLastParamThrowable(Object[] params) {
        return Optional.ofNullable(params)
                .map(Arrays::stream)
                .flatMap(a -> a.skip(params.length - 1)
                        .findFirst()
                        .filter(o -> o instanceof Throwable)
                        .map(Throwable.class::cast));
    }

    private Marker getEnrichedMarker(Marker marker) {
        if (marker == null) {
            return MARKER;
        }
        marker.add(MARKER);
        return marker;
    }

    public void addAggregatedLogger(String logger) {
        this.aggregatedLogger.add(logger);
    }

    public long getReportingIntervalMillis() {
        return reportingIntervalMillis;
    }

    public void setReportingIntervalMillis(long reportingIntervalMillis) {
        this.reportingIntervalMillis = reportingIntervalMillis;
    }

    private static class LoggerAggregates {

        private final Map<LoggingEventKey, AggregateSummary> aggregates = new ConcurrentHashMap<>();
    }

    private static class AggregateSummary {

        private final int logsCount;
        private final Throwable lastException;

        private AggregateSummary(Throwable lastException) {
            this(1, lastException);
        }

        private AggregateSummary(int logsCount, Throwable lastException) {
            this.logsCount = logsCount;
            this.lastException = lastException;
        }

        private static AggregateSummary incrementCount(AggregateSummary currentAggregate, Throwable lastException) {
            return new AggregateSummary(currentAggregate.logsCount + 1,
                    Optional.ofNullable(lastException).orElse(currentAggregate.lastException));
        }
    }

    private static class LoggingEventKey {

        private final String message;
        private final int level;
        private final Marker marker;
        private final Object[] params;

        LoggingEventKey(String message, Object[] params, Level level, Marker marker) {
            this.message = message;
            this.params = params;
            this.level = Level.toLocationAwareLoggerInteger(level);
            this.marker = marker;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            LoggingEventKey that = (LoggingEventKey) o;
            return level == that.level
                    && Objects.equals(message, that.message)
                    && Objects.equals(marker, that.marker)
                    && Arrays.equals(params, that.params);
        }

        @Override
        public int hashCode() {
            int result = Objects.hash(message, level, marker);
            result = 31 * result + Arrays.hashCode(params);
            return result;
        }
    }
}


