package com.atlassian.stash.internal.commit.graph.cache;

import com.atlassian.bitbucket.ServerException;
import com.atlassian.bitbucket.commit.graph.*;
import com.atlassian.bitbucket.concurrent.ConcurrencyService;
import com.atlassian.bitbucket.concurrent.VersionTracker;
import com.atlassian.bitbucket.event.repository.RepositoryRefsChangedEvent;
import com.atlassian.bitbucket.i18n.I18nService;
import com.atlassian.bitbucket.permission.Permission;
import com.atlassian.bitbucket.repository.RefChange;
import com.atlassian.bitbucket.repository.RefChangeType;
import com.atlassian.bitbucket.repository.Repository;
import com.atlassian.bitbucket.repository.RepositoryService;
import com.atlassian.bitbucket.scm.ScmService;
import com.atlassian.bitbucket.server.ApplicationMode;
import com.atlassian.bitbucket.user.SecurityService;
import com.atlassian.bitbucket.util.Timer;
import com.atlassian.bitbucket.util.TimerUtils;
import com.atlassian.bitbucket.util.concurrent.ExecutorUtils;
import com.atlassian.event.api.EventListener;
import com.atlassian.stash.internal.ApplicationConstants;
import com.atlassian.stash.internal.ApplicationSettings;
import com.atlassian.stash.internal.commit.graph.CachedCommitGraphSource;
import com.atlassian.stash.internal.commit.graph.CommitGraphSource;
import com.atlassian.stash.internal.spring.AbstractSmartLifecycle;
import com.atlassian.util.contentcache.*;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Iterables;
import com.hazelcast.core.IExecutorService;
import com.hazelcast.spring.context.SpringAware;
import org.apache.commons.io.output.NullOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import javax.annotation.Nonnull;
import java.io.IOException;
import java.io.OutputStream;
import java.io.Serializable;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import static com.atlassian.stash.internal.commit.graph.cache.CachedCommitGraphOutputStream.CachedCommitGraphOutputStreamIOException;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Predicates.compose;
import static com.google.common.base.Predicates.equalTo;

/**
 * Implementation of the {@link CommitGraphSource} that caches commit graphs on disk.
 * <p>
 * Cache files are stored under {@code STASH_HOME/caches/commit-graph} with one file per repository.
 * <p>
 * Cache entries are created on ref change events for non-fork repositories and direct calls when
 * the cache is missing or invalid (typically from a UI thread).
 * <p>
 * Cache creation is run in a bounded size {@link ExecutorService} with a bounded work queue. The work queue is
 * a {@link SetPriorityBlockingQueue} to allow direct-call initiated creation to preempt background
 * ref-change creation. If a direct call requires a cache entry to be created, this creation is
 * run in the background to ensure it is completed.
 *
 * @since 2.11
 */
public class DefaultCachedCommitGraphSource extends AbstractSmartLifecycle implements CachedCommitGraphSource {

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

    private static final String CACHE_KEY = "commit-graph";
    private static final String PUMPER_KEY = "commit-graph-pumper";
    private static final int MAX_RETRIES = 5;

    private final IExecutorService clusterExecutorService;
    private final ContentCache contentCache;
    private final ContentCacheManager contentCacheManager;
    private final StreamPumper contentCachePump;
    private final Queue<Runnable> delayedJobs;
    private final ExecutorService executorService;
    private final I18nService i18nService;
    private final long maxTraverseWaitTime;
    private final ApplicationMode mode;
    private final VersionTracker<Integer> repoVersionTracker;
    private final ScmService scmService;

    public DefaultCachedCommitGraphSource(ApplicationSettings applicationSettings, IExecutorService clusterExecutorService,
                                          ConcurrencyService concurrencyService, ExecutorService executorService,
                                          I18nService i18nService, long maxTraverseWaitTime,
                                          @Value("#{applicationPropertiesService.mode}") ApplicationMode mode,
                                          long requiredFreeSpace, ScmService scmService) {

        this.clusterExecutorService = clusterExecutorService;
        this.contentCachePump = new BackgroundThreadStreamPumper(PUMPER_KEY);
        this.contentCacheManager  = new FileContentCacheManager
                .Builder(applicationSettings.getCacheDir())
                .minFreeSpaceBytes(requiredFreeSpace)
                .streamPumper(contentCachePump)
                .build();
        this.contentCache = createFileCache();
        this.executorService = executorService;
        this.i18nService = i18nService;
        this.maxTraverseWaitTime = maxTraverseWaitTime;
        this.mode = mode;
        this.repoVersionTracker = concurrencyService.getVersionTracker(getClass().getName() + ".repoVersionTracker");
        this.scmService = scmService;

        delayedJobs = new LinkedList<>();
    }

    @Override
    public int getPhase() {
        // Cache population relies on the SCM plugins being initialized for the repository
        return ApplicationConstants.LIFECYCLE_PHASE_COMMIT_GRAPH_SOURCE;
    }

    @EventListener
    public void onRepositoryRefChanges(RepositoryRefsChangedEvent event) throws Exception {
        if (!isRunning() || isMirrorMode()) {
            // no commit graph mutations in mirror mode or if we're not running
            return;
        }

        Repository repository = event.getRepository();
        boolean hasRefChanges = !event.getRefChanges().isEmpty();
        boolean onlyDeletions = Iterables.all(event.getRefChanges(), compose(equalTo(RefChangeType.DELETE), RefChange::getType));
        boolean isFork = repository.isFork();
        log.debug("Checking RepositoryRefsChangedEvent for ref changes on [{}]; has changes: {}, is fork: {}, only deletions {}",
                repository,
                hasRefChanges,
                isFork,
                onlyDeletions);
        if (hasRefChanges && !onlyDeletions) {
            // cause other nodes in the cluster to be aware that mutation has invalidated its
            // commit graph cache for this repository. For forks this will be sufficient to
            // ensure their cache is subsequently removed by the invalidation job
            repoVersionTracker.increment(repository.getId());

            if (!isFork) {
                // To reduce load during automatic syncs only index non-forks
                clusterExecutorService.executeOnAllMembers(new RebuildCacheTask(repository.getId()));
            }
        }
    }

    @Override
    public void purgeStaleEntries() {
        if (!isMirrorMode()) {
            for (CacheEntryStatistics entryStats : contentCache.getStatistics().getEntries()) {
                String entryKey = entryStats.getKey();
                int[] repoIdAndVersion = toRepoIdAndVersion(entryKey);

                if (repoIdAndVersion == null) {
                    continue;
                }

                int repoId = repoIdAndVersion[0];
                int version = repoIdAndVersion[1];

                int currentRepoVersion = repoVersionTracker.get(repoId);
                if (version != currentRepoVersion) {
                    contentCache.remove(getCacheKey(repoId, version));
                }
            }
        }
    }

    @Override
    public void rebuildFor(Repository repository) {
        if (!isMirrorMode()) {
            AbstractCacheTraversalJob cacheTraversalJob = new RetryingCacheTraversalJob(repository, NullOutputStream.NULL_OUTPUT_STREAM);

            if (!isRunning()) {
                synchronized (delayedJobs) {
                    if (!isRunning()) {
                        delayedJobs.offer(cacheTraversalJob);
                        return;
                    }
                }
            }

            executorService.execute(cacheTraversalJob);
        }
    }

    @Override
    public void start() {
        super.start();
        if (!isMirrorMode()) {
            synchronized (delayedJobs) {
                Iterator<Runnable> it = delayedJobs.iterator();
                while (it.hasNext()) {
                    executorService.execute(it.next());
                    it.remove();
                }
            }
        }
    }

    @Override
    public void stop() {
        contentCache.clear();
        contentCachePump.shutdown();
        ExecutorUtils.shutdown(executorService, log);
        super.stop();
    }

    @Override
    public void traverse(@Nonnull TraversalRequest request, @Nonnull TraversalCallback callback) {
        checkNotNull(request, "request");
        checkNotNull(callback, "callback");

        if (!isRunning() || isMirrorMode()) {
            return;
        }

        Repository repository = request.getRepository();
        CommitGraphContext commitGraphContext = new CommitGraphContext.Builder()
                .exclude(request.getExcludes())
                .include(request.getIncludes())
                .build();
        DelegatingCompletionWaitingTraversalCallback futureCallback =
                new DelegatingCompletionWaitingTraversalCallback(new SubgraphTraversalCallback(callback, commitGraphContext));
        String cacheKey = getCacheKey(repository);
        ContentProvider contentProvider = CachedCommitGraphUtils.getTimedScmContentProvider(repository, scmService, i18nService);

        try (Timer ignored = TimerUtils.start(repository.getId() + ": reading traversal index");
             OutputStream outputStream = new CachedCommitGraphOutputStream(futureCallback)) {
            CacheAccess cacheAccess = contentCache.access(cacheKey, outputStream, contentProvider);
            CacheResult cacheResult = cacheAccess.getResult();
            if (cacheResult == CacheResult.MISS) {
                log.debug("{} during traversal for {}, delegating job to executorService", cacheResult, repository);
                SingleCacheTraversalJob job = new SingleCacheTraversalJob(repository, outputStream, cacheAccess);
                executorService.execute(job);
                try {
                    futureCallback.await(maxTraverseWaitTime, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    throw new ServerException(
                            i18nService.createKeyedMessage("bitbucket.commit.graph.traversal.read.interruptedexception", repository), e);
                }
            } else {
                log.debug("{} during traversal for {}", cacheResult, repository);
                try {
                    cacheAccess.stream();
                } catch (CachedCommitGraphOutputStreamIOException e) {
                    // ignore this as it is an artifact of how the ContentCache deals with closed streams
                }
            }
        } catch (IOException e) {
            throw new ServerException(
                    i18nService.createKeyedMessage("bitbucket.commit.graph.traversal.read.ioexception",
                            repository), e);
        }
    }

    @VisibleForTesting
    protected ContentCache createFileCache() {
        return contentCacheManager.getCache(CACHE_KEY);
    }

    private String getCacheKey(Repository repository) {
        return getCacheKey(repository.getId(), repoVersionTracker.get(repository.getId()));
    }

    private String getCacheKey(int repositoryId, int version) {
        return Integer.toString(version) + "/" + Integer.toString(repositoryId);
    }

    private boolean isMirrorMode() {
        return mode == ApplicationMode.MIRROR;
    }

    private int[] toRepoIdAndVersion(String cacheKey) {
        String[] segments = cacheKey.split("/");
        if (segments.length == 2) {
            try {
                return new int[]{Integer.parseInt(segments[1]), Integer.parseInt(segments[0])};
            } catch (NumberFormatException e) {
                //Ignore, unknown key format
            }
        }

        //An unexpected key that is not in "version/id" format
        return null;
    }

    /**
     * @since 2.11
     */
    static class DelegatingCompletionWaitingTraversalCallback extends TraversalCallback {

        private final TraversalCallback delegate;
        private final CountDownLatch countDownLatch;

        private volatile boolean finished;

        public DelegatingCompletionWaitingTraversalCallback(TraversalCallback delegate) {
            this.delegate = delegate;
            this.countDownLatch = new CountDownLatch(1);
            this.finished = false;
        }

        public void await() throws InterruptedException {
            countDownLatch.await();
        }

        public void await(long timeout, TimeUnit unit) throws InterruptedException {
            countDownLatch.await(timeout, unit);
        }

        public boolean isFinished() {
            return finished;
        }

        @Override
        public void onEnd(@Nonnull TraversalSummary summary) {
            delegate.onEnd(summary);
            this.finished = true;
            countDownLatch.countDown();
        }

        @Override
        public TraversalStatus onNode(@Nonnull CommitGraphNode node) {
            return delegate.onNode(node);
        }

        @Override
        public void onStart(@Nonnull TraversalContext context) {
            delegate.onStart(context);
        }
    }

    /**
     * @since 2.11
     */
    class RetryingCacheTraversalJob extends AbstractCacheTraversalJob {

        private static final int PRIORITY = 1;

        private final OutputStream outputStream;
        private final int maxRetries;

        private int retries;

        public RetryingCacheTraversalJob(Repository repository,
                                         OutputStream outputStream) {
            super(PRIORITY, repository, outputStream);
            this.outputStream = outputStream;
            this.maxRetries = MAX_RETRIES;
            this.retries = 0;
        }

        @Override
        public void run() {
            if (!isRunning()) {
                return;
            }
            try {
                streamCache();
            } catch (Exception e) {
                if (retries < maxRetries) {
                    retries++;
                    executorService.execute(this);
                } else {
                    log.warn("Failed to stream commit graph on {} after {} retries", repository.toString(), maxRetries, e);
                }
            }
        }

        private void streamCache() throws IOException {
            String cacheKey = getCacheKey(repository);
            ContentProvider contentProvider =
                    CachedCommitGraphUtils.getTimedScmContentProvider(repository, scmService, i18nService);
            contentCache.remove(cacheKey);
            CacheAccess cacheAccess = contentCache.access(cacheKey, outputStream, contentProvider);
            if (cacheAccess.getResult() == CacheResult.MISS) {
                // skip hits/bypass when we are just trying to generate a cache entry
                cacheAccess.stream();
            } else {
                cacheAccess.cancel();
            }
        }
    }

    /**
     * @since 2.11
     */
    class SingleCacheTraversalJob extends AbstractCacheTraversalJob {

        private static final int PRIORITY = 2;

        private final CacheAccess cacheAccess;

        public SingleCacheTraversalJob(Repository repository,
                                       OutputStream outputStream,
                                       CacheAccess cacheAccess) {
            super(PRIORITY, repository, outputStream);
            this.cacheAccess = cacheAccess;
        }

        @Override
        public void run() {
            try {
                cacheAccess.stream();
            } catch (CachedCommitGraphOutputStreamIOException e) {
                // ignore this as it is an artifact of how the ContentCache deals with closed streams
            } catch (IOException e) {
                rebuildFor(repository);
                throw new ServerException(
                        i18nService.createKeyedMessage("bitbucket.commit.graph.traversal.read.ioexception",
                                repository), e);
            }
        }
    }

    @SpringAware
    static class RebuildCacheTask implements Runnable, Serializable {

        @Autowired
        transient RepositoryService repositoryService;
        @Autowired
        transient CachedCommitGraphSource cache;
        @Autowired
        transient SecurityService securityservice;

        private final int repoId;

        public RebuildCacheTask(int repoId) {
            this.repoId = repoId;
        }

        @Override
        public void run() {
            Repository repository = securityservice.withPermission(Permission.ADMIN,
                    "Looking up repository to invalidate its commit graph cache").call(() ->
                repositoryService.getById(repoId));

            if (repository != null) {
                cache.rebuildFor(repository);
            }
        }
    }
}
