package com.atlassian.cache.hazelcast;
   
   import java.io.Serializable;
   import java.util.Map;
   
   import javax.annotation.Nonnull;
   import javax.annotation.Nullable;
   import javax.annotation.PostConstruct;
   import javax.annotation.PreDestroy;
  
  import com.atlassian.cache.Cache;
  import com.atlassian.cache.CacheFactory;
  import com.atlassian.cache.CacheLoader;
  import com.atlassian.cache.CacheSettings;
  import com.atlassian.cache.CacheSettingsBuilder;
  import com.atlassian.cache.CacheSettingsDefaultsProvider;
  import com.atlassian.cache.CachedReference;
  import com.atlassian.cache.ManagedCache;
  import com.atlassian.cache.Supplier;
  import com.atlassian.cache.impl.AbstractCacheManager;
  import com.atlassian.cache.impl.ReferenceKey;
  import com.atlassian.cache.impl.StrongSupplier;
  import com.atlassian.cache.impl.WeakSupplier;
  import com.atlassian.hazelcast.serialization.OsgiSafe;
  
  import com.hazelcast.config.Config;
  import com.hazelcast.config.ConfigurationException;
  import com.hazelcast.config.MapConfig;
  import com.hazelcast.core.HazelcastInstance;
  import com.hazelcast.core.IMap;
  import com.hazelcast.core.ITopic;
  import com.hazelcast.core.MembershipAdapter;
  import com.hazelcast.core.MembershipEvent;
  import com.hazelcast.map.impl.MapContainer;
  import com.hazelcast.map.impl.MapService;
  import com.hazelcast.map.impl.MapServiceContext;
  import com.hazelcast.map.impl.proxy.MapProxyImpl;
  import com.hazelcast.map.listener.EntryAddedListener;
  import com.hazelcast.map.listener.EntryUpdatedListener;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static com.google.common.base.Preconditions.checkNotNull;
  
  /**
   * Hazelcast implementation of the {@link com.atlassian.cache.CacheManager} contract
   */
  public class HazelcastCacheManager extends AbstractCacheManager
  {
      private static final Logger log = LoggerFactory.getLogger(HazelcastCacheManager.class);
  
      protected static final String PREFIX = "atlassian-cache.";
      protected static final String PREFIX_CACHE = PREFIX + "Cache.";
      protected static final String PREFIX_CACHE_REFERENCE = PREFIX + "CacheReference.";
      protected static final String SETTINGS_MAP_NAME = PREFIX + "settings";
  
      private final HazelcastInstance hazelcast;
      private final CacheFactory localCacheFactory;
      private final IMap<String, CacheSettings> mapSettings;
      private final String mapSettingsUpdatedListenerId;
      private final String mapSettingsAddedListenerId;
      private final String membershipListenerId;
  
      private MapServiceContext mapServiceContext;
  
      public HazelcastCacheManager(HazelcastInstance hazelcast, CacheFactory localCacheFactory,
              CacheSettingsDefaultsProvider cacheSettingsDefaultsProvider)
      {
          super(cacheSettingsDefaultsProvider);
  
          this.hazelcast = hazelcast;
          this.localCacheFactory = localCacheFactory;
          this.mapSettings = hazelcast.getMap(SETTINGS_MAP_NAME);
  
          // add listeners to be notified of remote changes to the map config so the local config can be updated too
          this.mapSettingsUpdatedListenerId = mapSettings.addEntryListener(
                  (EntryUpdatedListener<String, CacheSettings>) event -> reconfigureMap(event.getKey(), event.getValue()),
                  true);
          this.mapSettingsAddedListenerId = mapSettings.addEntryListener(
                  (EntryAddedListener<String, CacheSettings>) event -> configureMap(event.getKey(), event.getValue()),
                  true);
  
          // add a listener for join events to ensure that the local cache configs are updated
          this.membershipListenerId = hazelcast.getCluster().addMembershipListener(new MembershipAdapter()
          {
              @Override
              public void memberAdded(MembershipEvent membershipEvent)
              {
                  maybeUpdateMapContainers();
              }
          });
      }
  
      public HazelcastInstance getHazelcastInstance()
      {
          return hazelcast;
      }
  
     /**
      * Initializes the manager. This method must be called after the bean has been created.
      */
     @PostConstruct
     public void init()
     {
         maybeUpdateMapContainers();
     }
 
     @Override
     protected <K, V> ManagedCache createComputingCache(@Nonnull final String name, @Nonnull final CacheSettings settings, final CacheLoader<K, V> loader)
     {
         checkSettingsAreCompatible(name, settings);
 
         // when a loader is provided, always create a new ManagedCache to ensure the correct loader is being used.
         // if the cache already existed, the backing values will be reused
         return cacheCreationLocks.apply(name).withLock(new java.util.function.Supplier<ManagedCache>()
         {
             @Override
             public ManagedCache get()
             {
                 if (!caches.containsKey(name) || loader != null)
                 {
                     caches.put(name, new WeakSupplier<ManagedCache>((ManagedCache) doCreateCache(name, loader, settings)));
                 }
                 return caches.get(name).get();
             }
         });
     }
 
     @Override
     protected ManagedCache createSimpleCache(@Nonnull final String name, @Nonnull final CacheSettings settings)
     {
         checkSettingsAreCompatible(name, settings);
 
         ManagedCache existing = getManagedCache(name);
         if (existing != null)
         {
             return existing;
         }
 
         return cacheCreationLocks.apply(name).withLock(new java.util.function.Supplier<ManagedCache>()
         {
             @Override
             public ManagedCache get()
             {
                 if (!caches.containsKey(name))
                 {
                     caches.put(name, new StrongSupplier<ManagedCache>((ManagedCache) doCreateCache(name, null, settings)));
                 }
                 return caches.get(name).get();
             }
         });
     }
 
     /**
      * De-registers listeners. This method must be called when the bean is no longer required.
      */
     @PreDestroy
     public void destroy()
     {
         mapSettings.removeEntryListener(mapSettingsAddedListenerId);
         mapSettings.removeEntryListener(mapSettingsUpdatedListenerId);
         hazelcast.getCluster().removeMembershipListener(membershipListenerId);
     }
 
     @Nonnull
     @Override
     public <V> CachedReference<V> getCachedReference(@Nonnull final String name, @Nonnull final Supplier<V> supplier,
             @Nonnull final CacheSettings settings)
     {
         checkSettingsAreCompatible(name, settings);
 
         return cacheCreationLocks.apply(name).withLock(new java.util.function.Supplier<CachedReference<V>>()
         {
             @Override
             public CachedReference<V> get()
             {
                 caches.put(name, new WeakSupplier<ManagedCache>((ManagedCache) doCreateCachedReference(name, supplier, settings)));
                 //noinspection unchecked
                 return (CachedReference<V>) caches.get(name).get();
             }
         });
     }
 
     protected void checkSettingsAreCompatible(String name, CacheSettings settings)
     {
     }
 
     protected MapConfig configureMap(String mapName, CacheSettings settings)
     {
         Config config = hazelcast.getConfig();
         MapConfig mapConfig = config.findMapConfig(mapName);
         if (!mapConfig.getName().equals(mapName))
         {
             mapConfig = convertAndStoreMapConfig(mapName, settings, config, mapConfig);
 
             // store the config in the settings map to trigger other nodes to also configure the IMap
             mapSettings.putIfAbsent(mapName, asSerializable(settings));
         }
         else
         {
             log.debug("Using existing cache configuration for cache {}", mapName);
             mapSettings.computeIfAbsent(mapName, (key)-> asSerializable(settings));
         }
         updateMapContainer(mapName, mapConfig);
 
         return mapConfig.getAsReadOnly();
     }
 
     protected MapConfig reconfigureMap(String mapName, CacheSettings newSettings)
     {
         Config config = hazelcast.getConfig();
 
         MapConfig baseConfig = config.getMapConfig(mapName);
 
         return convertAndStoreMapConfig(mapName, newSettings, config, baseConfig);
     }
 
     /**
      * Update the cache settings of an existing cache. Called by {@link com.atlassian.cache.ManagedCache} methods
      *
      * @param mapName The name of the map to update
      * @param newSettings CacheSettings object representing the new settings
      * @return true if the update succeeded, false otherwise
      */
     public boolean updateCacheSettings(@Nonnull String mapName, @Nonnull CacheSettings newSettings)
     {
         try {
             //Do nothing if the map doesn't exist
             MapConfig mapConfig = hazelcast.getConfig().findMapConfig(mapName);
             if (!mapConfig.getName().equals(mapName)) {
                 return false;
             }
 
             MapConfig newConfig = reconfigureMap(mapName, newSettings);
 
             boolean result = updateMapContainer(mapName, newConfig);
             if (result) {
                 // store the config in the settings map to trigger other nodes to also configure the IMap
                 mapSettings.put(mapName, asSerializable(newSettings));
             }
 
             return result;
         }catch (ConfigurationException exception) {
             if(log.isDebugEnabled()) {
                 log.debug("Updating configuration of {} failed: ", mapName, exception);
             } else {
                 log.warn("Updating cache settings on {} is not possible", mapName);
             }
             return false;
         }
     }
 
     private MapConfig convertAndStoreMapConfig(String mapName, CacheSettings newSettings, Config config, MapConfig baseConfig)
     {
         MapConfig newConfig = new MapConfig(baseConfig);
         newConfig.setName(mapName);
         newConfig.setStatisticsEnabled(true);
         new HazelcastMapConfigConfigurator().configureMapConfig(newSettings, newConfig, hazelcast.getPartitionService().getPartitions().size());
         MapConfig oldConfig = config.findMapConfig(mapName);
         if(oldConfig ==null || !oldConfig.getName().equals(mapName)) {
             config.addMapConfig(newConfig);
         }
         return newConfig;
     }
 
     protected <K, V> Cache<K, V> createAsyncHybridCache(String name, CacheLoader<K, V> loader, CacheSettings settings)
     {
         final String topicName = PREFIX_CACHE + name;
         final ITopic<K> topic = hazelcast.getTopic(topicName);
 
         return new HazelcastAsyncHybridCache<K, V>(name, localCacheFactory, topic, loader, this, settings);
     }
 
     protected <V> CachedReference<V> createAsyncHybridCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         final String topicName = PREFIX_CACHE_REFERENCE + name;
         final ITopic<ReferenceKey> topic = hazelcast.getTopic(topicName);
 
         return new HazelcastAsyncHybridCachedReference<V>(name, localCacheFactory, topic, supplier, this, settings);
     }
 
     protected <K, V> Cache<K, V> createDistributedCache(String name, CacheLoader<K, V> loader, CacheSettings settings)
     {
         final String mapName = PREFIX_CACHE + name;
         configureMap(mapName, settings);
         final IMap<K, OsgiSafe<V>> map = hazelcast.getMap(mapName);
         final CacheVersion cacheVersion = new CacheVersion(hazelcast, mapName);
 
         return new HazelcastCache<K, V>(name, map, loader, cacheVersion, this);
     }
 
     protected <V> CachedReference<V> createDistributedCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         // override the settings to ensure the reference is flushable and the max is set to 1000. A low value for
         // maxEntries would trigger continuous cache invalidations because of the way map eviction works in Hazelcast.
         final CacheSettings overriddenSettings = checkNotNull(settings, "settings").override(
                 new CacheSettingsBuilder().flushable().maxEntries(1000).build());
 
         final String mapName = PREFIX_CACHE_REFERENCE + name;
         configureMap(mapName, overriddenSettings);
         final IMap<String, OsgiSafe<V>> map = hazelcast.getMap(mapName);
 
         return new HazelcastCachedReference<V>(name, map, supplier, this);
     }
 
     protected <K, V> Cache<K, V> createHybridCache(String name, CacheLoader<K, V> loader, CacheSettings settings)
     {
         final String mapName = PREFIX_CACHE + name;
         configureMap(mapName, settings);
         final IMap<K, Long> map = hazelcast.getMap(mapName);
 
         return new HazelcastHybridCache<K, V>(name, localCacheFactory, map, loader, this);
     }
 
     protected <V> CachedReference<V> createHybridCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         final String mapName = PREFIX_CACHE_REFERENCE + name;
         configureMap(mapName, settings);
         final IMap<ReferenceKey, Long> map = hazelcast.getMap(mapName);
 
         return new HazelcastHybridCachedReference<V>(name, localCacheFactory, map, supplier, this);
     }
 
     private CacheSettings asSerializable(CacheSettings settings)
     {
         if (settings instanceof Serializable)
         {
             return settings;
         }
 
         // DefaultCacheSettings implements Serializable, make a copy
         return new CacheSettingsBuilder(settings).build();
     }
 
     private <K, V> Cache<K, V> doCreateCache(String name, CacheLoader<K, V> loader, CacheSettings settings)
     {
         if (settings.getLocal(false))
         {
             return localCacheFactory.getCache(name, loader, settings);
         }
 
         if (settings.getReplicateViaCopy(true))
         {
             return createDistributedCache(name, loader, settings);
         }
 
         if (settings.getReplicateAsynchronously(true))
         {
             return createAsyncHybridCache(name, loader, settings);
         }
         else
         {
             return createHybridCache(name, loader, settings);
         }
     }
 
     private <V> CachedReference<V> doCreateCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         if (settings.getLocal(false))
         {
             return localCacheFactory.getCachedReference(name, supplier, settings);
         }
 
         // remote cached reference
         if (settings.getReplicateViaCopy(true))
         {
             return createDistributedCachedReference(name, supplier, settings);
         }
 
         if (settings.getReplicateAsynchronously(true))
         {
             return createAsyncHybridCachedReference(name, supplier, settings);
         }
         else
         {
             return createHybridCachedReference(name, supplier, settings);
         }
     }
 
     private MapContainer getMapContainer(@Nonnull String name)
     {
         if (mapServiceContext == null)
         {
             MapProxyImpl<String, CacheSettings> proxy = hazelcast.getDistributedObject(MapService.SERVICE_NAME, SETTINGS_MAP_NAME);
             mapServiceContext = proxy.getService().getMapServiceContext();
         }
 
         return mapServiceContext.getMapContainer(name);
     }
 
     private boolean updateMapContainer(String mapName, MapConfig config)
     {
         final MapContainer container = getMapContainer(mapName);
         if (container == null)
         {
             return false;
         }
 
         container.setMapConfig(config);
         container.initEvictor();
 
         return true;
     }
 
     private void maybeUpdateMapContainers()
     {
         for (Map.Entry<String, CacheSettings> entry : mapSettings.entrySet())
         {
             configureMap(entry.getKey(), entry.getValue());
         }
     }
 
     /**
      * Get the current MapConfig for the given IMap
      *
      * @param mapName the name of the map
      * @return The current mapConfig for that map if it exists, null otherwise
      */
     @Nullable
     MapConfig getMapConfig(@Nonnull String mapName)
     {
         MapContainer mapContainer = getMapContainer(mapName);
         MapConfig mapConfig = mapContainer.getMapConfig();
         if (!mapConfig.getName().equals(mapName))
         {
             return null;
         }
         else
         {
             return mapConfig;
         }
     }
 
     /**
      * Gets the original CacheSettings object used to configure the given map, before it was transformed into a
      * Hazelcast MapConfig
      *
      * @param mapName the name of the map
      * @return The original CacheSettings object, or null if the map doesn't exist
      */
     @Nullable
     CacheSettings getCacheSettings(@Nonnull String mapName)
     {
         return mapSettings.get(mapName);
     }
 }