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

import com.atlassian.bitbucket.event.cluster.ClusterNodeAddedEvent;
import com.atlassian.bitbucket.permission.Permission;
import com.atlassian.bitbucket.permission.PermissionValidationService;
import com.atlassian.bitbucket.scm.cache.ScmCacheConfig;
import com.atlassian.bitbucket.scm.cache.ScmRequestProtocol;
import com.atlassian.bitbucket.scm.cache.ScmRequestType;
import com.atlassian.bitbucket.server.ApplicationPropertiesService;
import com.atlassian.event.api.EventListener;
import com.atlassian.sal.api.pluginsettings.PluginSettings;
import com.atlassian.sal.api.pluginsettings.PluginSettingsFactory;
import com.google.common.base.Predicates;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import java.util.*;
import java.util.concurrent.TimeUnit;

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

/**
 * A note on cluster coherency. When a configuration change is made on one node (e.g. over REST) a notification is sent
 * to the other cluster members to trigger them to reload their configuration as well. The big advantage of this
 * solution is that all data is local and no remote calls are required for normal operations (e.g.
 * {@link #createExpiryDate(ScmRequestType)}).
 * <p>
 * This approach is very similar to a hybrid cache, but with an important distinction. To ensure complete consistency,
 * the hybrid cache checks the cluster version on each access, which could involve a remote call.
 */
public class DefaultScmCacheConfig implements ScmCacheConfig {

    private static final Logger log = LoggerFactory.getLogger(DefaultScmCacheService.class);
    private static final String DEFAULT_PROTOCOLS = ScmRequestProtocol.HTTP.name() + "," + ScmRequestProtocol.SSH.name();

    private static final int DEFAULT_REFS_MAXCOUNT = 1;
    private static final long DEFAULT_REFS_TTL = TimeUnit.MINUTES.toSeconds(1);
    private static final int DEFAULT_UPLOAD_PACK_MAXCOUNT = 3;
    private static final long DEFAULT_UPLOAD_PACK_TTL = TimeUnit.HOURS.toSeconds(4);

    private static final String KEY_PROTOCOLS = "protocols";
    private static final String SUFFIX_ENABLED = ".enabled";
    private static final String SUFFIX_MAXCOUNT = ".maxCount";
    private static final String SUFFIX_TTL = ".ttl";

    static final String NOTIFICATION_CONFIG_CHANGED = "config.changed";

    private final Map<ScmRequestType, ScmRequestConfig> configByRequestType;
    private final Set<ScmRequestProtocol> enabledProtocols;
    private final PermissionValidationService permissionValidationService;
    private final PluginSettings pluginSettings;
    private final ApplicationPropertiesService propertiesService;
    private final ClusterNotificationService notificationService;

    public DefaultScmCacheConfig(ClusterNotificationService clusterNotificationService,
                                 PermissionValidationService permissionValidationService,
                                 PluginSettingsFactory pluginSettingsFactory,
                                 ApplicationPropertiesService propertiesService) {
        this.permissionValidationService = permissionValidationService;
        this.propertiesService = propertiesService;
        this.notificationService = clusterNotificationService;

        pluginSettings = pluginSettingsFactory.createSettingsForKey("com.atlassian.bitbucket.server.bitbucket-scm-cache");

        configByRequestType = Maps.newConcurrentMap();
        enabledProtocols = Sets.newSetFromMap(Maps.<ScmRequestProtocol, Boolean>newConcurrentMap());
        loadConfig();

        clusterNotificationService.subscribe(Predicates.equalTo(NOTIFICATION_CONFIG_CHANGED), new ConfigReloadListener());
    }

    @Override
    public Date createExpiryDate(@Nonnull ScmRequestType requestType) {
        long ttl = getTimeToLive(requestType);
        return ttl > 0 ? new Date(System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(ttl)) : null;
    }

    @Nonnull
    @Override
    public Set<ScmRequestProtocol> getEnabledProtocols() {
        return ImmutableSet.copyOf(enabledProtocols);
    }

    @Override
    public int getMaxCount(@Nonnull ScmRequestType requestType) {
        return configByRequestType.get(checkNotNull(requestType, "requestType")).maxCount;
    }

    @Override
    public long getTimeToLive(@Nonnull ScmRequestType requestType) {
        return configByRequestType.get(checkNotNull(requestType, "requestType")).timeToLive;
    }

    @Override
    public boolean isEnabled(@Nonnull ScmRequestType requestType) {
        return configByRequestType.get(checkNotNull(requestType, "requestType")).enabled;
    }

    @Override
    public boolean isEnabled(@Nonnull ScmRequestProtocol protocol) {
        return enabledProtocols.contains(checkNotNull(protocol, "protocol"));
    }

    @EventListener
    public void onClusterNodeAdded(ClusterNodeAddedEvent event) {
        // as a precaution in case this is a split brain situation being resolved
        loadConfig();
    }

    @Override
    public void setEnabled(@Nonnull ScmRequestType requestType, boolean enabled) {
        permissionValidationService.validateForGlobal(Permission.ADMIN);
        configByRequestType.get(checkNotNull(requestType, "requestType")).enabled = enabled;

        // persist the updated ttl in plugin settings
        pluginSettings.put(requestType.getName() + SUFFIX_ENABLED, Boolean.toString(enabled));

        notificationService.send(NOTIFICATION_CONFIG_CHANGED);
    }

    @Override
    public void setEnabled(@Nonnull ScmRequestProtocol protocol, boolean enabled) {
        checkNotNull(protocol, "protocol");
        permissionValidationService.validateForGlobal(Permission.ADMIN);
        if ((enabled && enabledProtocols.add(protocol)) || (!enabled && enabledProtocols.remove(protocol))) {
            if (enabledProtocols.isEmpty()) {
                pluginSettings.remove(KEY_PROTOCOLS);
            } else {
                pluginSettings.put(KEY_PROTOCOLS, StringUtils.join(enabledProtocols, ","));
            }
            notificationService.send(NOTIFICATION_CONFIG_CHANGED);
        }
    }

    @Override
    public void setMaxCount(@Nonnull ScmRequestType requestType, int maxCount) {
        permissionValidationService.validateForGlobal(Permission.ADMIN);
        configByRequestType.get(checkNotNull(requestType, "requestType")).maxCount = maxCount;

        // persist the updated maxCount in plugin settings
        pluginSettings.put(requestType.getName() + SUFFIX_MAXCOUNT, Integer.toString(maxCount));
        notificationService.send(NOTIFICATION_CONFIG_CHANGED);
    }

    @Override
    public void setTimeToLive(@Nonnull ScmRequestType requestType, long timeToLiveSeconds) {
        permissionValidationService.validateForGlobal(Permission.ADMIN);
        configByRequestType.get(checkNotNull(requestType, "requestType")).timeToLive = timeToLiveSeconds;

        // persist the updated ttl in plugin settings
        pluginSettings.put(requestType.getName() + SUFFIX_TTL, Long.toString(timeToLiveSeconds));
        notificationService.send(NOTIFICATION_CONFIG_CHANGED);
    }


    private ScmRequestConfig getConfig(ApplicationPropertiesService propertiesService, ScmRequestType requestType,
                                       boolean defaultEnabled, int defaultMaxCount, long defaultTtl) {

        boolean enabled = loadProperty(propertiesService, requestType.getName() + SUFFIX_ENABLED, defaultEnabled);
        int maxCount = loadProperty(propertiesService, requestType.getName() + SUFFIX_MAXCOUNT, defaultMaxCount);
        long ttlMillis = loadProperty(propertiesService, requestType.getName() + SUFFIX_TTL, defaultTtl);

        return new ScmRequestConfig(enabled, maxCount, ttlMillis);
    }

    private Collection<ScmRequestProtocol> getEnabledProtocols(ApplicationPropertiesService propertiesService) {

        Object settings = pluginSettings.get(KEY_PROTOCOLS);
        String protocols = (settings instanceof String) ? (String) settings :
                propertiesService.getPluginProperty(CacheConstants.PROP_PROTOCOLS, DEFAULT_PROTOCOLS);

        if (StringUtils.isEmpty(protocols)) {
            return Collections.emptyList();
        }

        List<ScmRequestProtocol> result = Lists.newArrayListWithCapacity(ScmRequestProtocol.values().length);
        for (String protocol : StringUtils.split(protocols.toUpperCase(Locale.US), ", \t\n\r")) {
            try {
                result.add(ScmRequestProtocol.valueOf(protocol));
            } catch (IllegalArgumentException e) {
                log.warn("Unsupported protocol {}. Caching will not be enabled for this protocol", protocol);
            }
        }
        return result;
    }

    private void loadConfig() {
        configByRequestType.put(ScmRequestType.REFS,
                getConfig(propertiesService, ScmRequestType.REFS, false, DEFAULT_REFS_MAXCOUNT, DEFAULT_REFS_TTL));
        configByRequestType.put(ScmRequestType.UPLOAD_PACK,
                getConfig(propertiesService, ScmRequestType.UPLOAD_PACK, true, DEFAULT_UPLOAD_PACK_MAXCOUNT, DEFAULT_UPLOAD_PACK_TTL));

        Collection<ScmRequestProtocol> protocols = getEnabledProtocols(propertiesService);
        enabledProtocols.clear();
        enabledProtocols.addAll(protocols);
    }

    private boolean loadProperty(ApplicationPropertiesService propertiesService, String name, boolean defaultValue) {
        Object setting = pluginSettings.get(name);
        if (setting instanceof String) {
            return Boolean.valueOf((String) setting);
        }

        return propertiesService.getPluginProperty(CacheConstants.PLUGIN_PROP_PREFIX + name, defaultValue);
    }

    private int loadProperty(ApplicationPropertiesService propertiesService, String name, int defaultValue) {
        Object setting = pluginSettings.get(name);
        if (setting instanceof String) {
            return Integer.parseInt((String) setting);
        }

        return propertiesService.getPluginProperty(CacheConstants.PLUGIN_PROP_PREFIX + name, defaultValue);
    }

    private long loadProperty(ApplicationPropertiesService propertiesService, String name, long defaultValue) {
        Object setting = pluginSettings.get(name);
        if (setting instanceof String) {
            return Long.parseLong((String) setting);
        }

        return propertiesService.getPluginProperty(CacheConstants.PLUGIN_PROP_PREFIX + name, defaultValue);
    }

    private static final class ScmRequestConfig {
        private boolean enabled;
        private int maxCount;
        private long timeToLive;

        private ScmRequestConfig(boolean enabled, int maxCount, long timeToLive) {
            this.enabled = enabled;
            this.maxCount = maxCount;
            this.timeToLive = timeToLive;
        }
    }

    private final class ConfigReloadListener implements NotificationListener {
        @Override
        public void onNotification(@Nonnull String notification) {
            loadConfig();
        }
    }
}
