package com.atlassian.bitbucket.scm.cache.internal;

import com.atlassian.bitbucket.permission.Permission;
import com.atlassian.bitbucket.permission.PermissionValidationService;
import com.atlassian.bitbucket.repository.Repository;
import com.atlassian.bitbucket.scm.cache.ScmCacheService;
import com.atlassian.util.contentcache.CacheStatistics;
import com.atlassian.util.contentcache.ContentCache;
import com.atlassian.util.contentcache.ContentCacheManager;
import com.atlassian.util.contentcache.ContentCacheStatistics;
import com.google.common.base.Predicate;

import javax.annotation.Nonnull;

import static com.google.common.base.Preconditions.checkNotNull;

/**
 * Implementation of {@link ScmCacheService} that applies permission checks before delegating down to the cache.
 */
public class DefaultScmCacheService implements ScmCacheService {

    private static final String NOTIFICATION_PREFIX = "cache-clear:";
    private static final int NOTIFICATION_PREFIX_LENGTH = NOTIFICATION_PREFIX.length();
    private static final String NOTIFICATION_CLEAR_ALL = NOTIFICATION_PREFIX + "all";

    private final ContentCacheManager cacheManager;
    private final PermissionValidationService permissionValidationService;
    private final ClusterNotificationService notificationService;

    public DefaultScmCacheService(ContentCacheManager cacheManager,
                                  ClusterNotificationService notificationService,
                                  PermissionValidationService permissionValidationService) {
        this.cacheManager = cacheManager;
        this.notificationService = notificationService;
        this.permissionValidationService = permissionValidationService;

        notificationService.subscribe(new Predicate<String>() {
            @Override
            public boolean apply(String triggerId) {
                return triggerId.startsWith(NOTIFICATION_PREFIX);
            }
        }, new ClearCacheListener());
    }

    @Override
    public void clear() {
        permissionValidationService.validateForGlobal(Permission.ADMIN);
        cacheManager.clear();

        notificationService.send(NOTIFICATION_CLEAR_ALL);
    }

    @Override
    public void clear(@Nonnull Repository repository) {
        permissionValidationService.validateForRepository(checkNotNull(repository, "repository"), Permission.REPO_ADMIN);
        String region = getCacheRegion(repository);
        cacheManager.getCache(region).clear();

        notificationService.send(NOTIFICATION_PREFIX + region);
    }

    @Override
    public void clear(@Nonnull Repository repository, @Nonnull String cacheKey) {
        permissionValidationService.validateForRepository(checkNotNull(repository, "repository"), Permission.REPO_ADMIN);
        String region = getCacheRegion(repository);
        cacheManager.getCache(region).remove(cacheKey);

        notificationService.send(NOTIFICATION_PREFIX + region + "/" + cacheKey);
    }

    @Nonnull
    @Override
    public ContentCacheManager getCacheManager() {
        return cacheManager;
    }

    @Nonnull
    @Override
    public CacheStatistics getStatistics() {
        permissionValidationService.validateForGlobal(Permission.ADMIN);
        return cacheManager.getStatistics();
    }

    @Override
    public ContentCacheStatistics getStatistics(@Nonnull Repository repository) {
        permissionValidationService.validateForRepository(checkNotNull(repository, "repository"), Permission.REPO_ADMIN);
        return getCache(repository).getStatistics();
    }

    private String getCacheRegion(Repository repository) {
        return CacheUtils.getCacheRegionForRepository(repository);
    }

    private ContentCache getCache(Repository repository) {
        return cacheManager.getCache(getCacheRegion(repository));
    }

    private class ClearCacheListener implements NotificationListener {
        @Override
        public void onNotification(@Nonnull String notification) {
            if (NOTIFICATION_CLEAR_ALL.equals(notification)) {
                cacheManager.clear();
            } else {
                String regionAndKey = notification.substring(NOTIFICATION_PREFIX_LENGTH);
                int index = regionAndKey.indexOf('/');
                if (index == -1) {
                    // only a region is specified
                    cacheManager.getCache(regionAndKey).clear();
                } else {
                    String region = regionAndKey.substring(0, index);
                    String cacheKey = regionAndKey.substring(index + 1);
                    cacheManager.getCache(region).remove(cacheKey);
                }
            }
        }
    }
}
