package com.atlassian.cache.vcache;
   
   import java.io.Serializable;
   import java.time.Duration;
   import javax.annotation.Nonnull;
   import javax.annotation.Nullable;
   
   import com.atlassian.cache.Cache;
   import com.atlassian.cache.CacheLoader;
  import com.atlassian.cache.CacheSettings;
  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.StrongSupplier;
  import com.atlassian.cache.impl.WeakSupplier;
  import com.atlassian.vcache.DirectExternalCache;
  import com.atlassian.vcache.ExternalCacheSettings;
  import com.atlassian.vcache.ExternalCacheSettingsBuilder;
  import com.atlassian.vcache.JvmCache;
  import com.atlassian.vcache.JvmCacheSettings;
  import com.atlassian.vcache.JvmCacheSettingsBuilder;
  import com.atlassian.vcache.StableReadExternalCache;
  import com.atlassian.vcache.VCacheFactory;
  import com.atlassian.vcache.marshallers.MarshallerFactory;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * VCache implementation of the {@link com.atlassian.cache.CacheManager} contract.
   *
   * @since 2.9.0
   */
  public class VCacheCacheManager extends AbstractCacheManager
  {
      private static final String PREFIX = "atlassian-cache.";
      private static final String PREFIX_CACHE = PREFIX + "Cache.";
      private static final String PREFIX_HYBRID_LOCAL_CACHE = PREFIX + "Hybrid.Local.";
      private static final String PREFIX_HYBRID_GLOBAL_CACHE = PREFIX + "Hybrid.Global.";
      private static final String PREFIX_CACHE_REFERENCE = PREFIX + "CacheReference.";
  
      private final VCacheFactory vCacheFactory;
  
      public VCacheCacheManager(VCacheFactory vCacheFactory,
                                CacheSettingsDefaultsProvider cacheSettingsDefaultsProvider)
      {
          super(cacheSettingsDefaultsProvider);
          this.vCacheFactory = requireNonNull(vCacheFactory);
      }
  
      @Override
      protected <K, V> ManagedCache createComputingCache(final String name, final CacheSettings settings, @Nullable final CacheLoader<K, V> loader)
      {
          // 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((java.util.function.Supplier<ManagedCache>) () -> {
              if (!caches.containsKey(name) || loader != null)
              {
                  caches.put(name, new WeakSupplier<>((ManagedCache) doCreateCache(name, loader, settings)));
              }
              return caches.get(name).get();
          });
      }
  
      @Override
      protected ManagedCache createSimpleCache(@Nonnull final String name, @Nonnull final CacheSettings settings)
      {
          ManagedCache existing = getManagedCache(name);
          if (existing != null)
          {
              return existing;
          }
  
          return cacheCreationLocks.apply(name).withLock((java.util.function.Supplier<ManagedCache>) () -> {
              if (!caches.containsKey(name))
              {
                  caches.put(name, new StrongSupplier<>((ManagedCache) doCreateCache(name, null, settings)));
              }
              return caches.get(name).get();
          });
      }
  
  
      @Nonnull
      @Override
      public <V> CachedReference<V> getCachedReference(@Nonnull final String name, @Nonnull final Supplier<V> supplier,
                                                       @Nonnull final CacheSettings settings)
      {
          return cacheCreationLocks.apply(name).withLock((java.util.function.Supplier<CachedReference<V>>) () -> {
              caches.put(name, new WeakSupplier<>((ManagedCache) doCreateCachedReference(name, supplier, settings)));
              //noinspection unchecked
              return (CachedReference<V>) caches.get(name).get();
          });
      }
  
      private <K, V> Cache<K, V> createLocalCache(String name, @Nullable CacheLoader<K, V> loader, CacheSettings settings)
      {
          final String mapName = PREFIX_CACHE + name;
         final JvmCacheSettings vSettings = buildJvmSettings(settings);
         final JvmCache<K, V> vcache = vCacheFactory.getJvmCache(mapName, vSettings);
 
         return new JvmCacheDelegate<>(name, vcache, loader, settings);
     }
 
     private Cache<String, Serializable> createDistributedCache(String name, @Nullable CacheLoader<String, Serializable> loader, CacheSettings settings)
     {
         final ExternalCacheSettings vSettings = buildExternalCacheSettings(settings);
 
         final String mapName = PREFIX_CACHE + name;
         final DirectExternalCache<Serializable> vcache =
                 vCacheFactory.getDirectExternalCache(
                         mapName, MarshallerFactory.serializableMarshaller(Serializable.class), vSettings);
 
         //noinspection unchecked
         return new ExternalCacheDelegate(name, vcache, loader, settings);
     }
 
     private <K, V> Cache createHybridCache(String name, @Nullable CacheLoader<K, V> loader, CacheSettings settings)
     {
         final JvmCacheSettings localSettings = buildJvmSettings(settings);
         final ExternalCacheSettings externalSettings = buildExternalCacheSettings(settings);
 
         final JvmCache<K, Versioned<V>> localVersioned =
                 vCacheFactory.getJvmCache(PREFIX_HYBRID_LOCAL_CACHE + name, localSettings);
         final StableReadExternalCache<String> globalVersions =
                 vCacheFactory.getStableReadExternalCache(
                         PREFIX_HYBRID_GLOBAL_CACHE + name, MarshallerFactory.stringMarshaller(), externalSettings);
 
         return new HybridCache<>(name, localVersioned, globalVersions, loader, settings);
     }
 
     private <K, V> Cache<K, V> doCreateCache(String name, @Nullable CacheLoader<K, V> loader, CacheSettings settings)
     {
         if (settings.getLocal(false))
         {
             return createLocalCache(name, loader, settings);
         }
 
         if (settings.getReplicateViaCopy(true))
         {
             // This is just horrendous. No way to confirm if the keys are strings, and the values are instances
             // of Serializable.
             //noinspection unchecked
             return (Cache<K, V>) createDistributedCache(name, (CacheLoader<String, Serializable>) loader, settings);
         }
 
         //noinspection unchecked
         return (Cache<K, V>) createHybridCache(name, loader, settings);
     }
 
     private <V> CachedReference<V> createLocalCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         final String mapName = PREFIX_CACHE_REFERENCE + name;
         final JvmCacheSettings vSettings = buildJvmSettings(settings);
         final JvmCache<String, V> vCache = vCacheFactory.getJvmCache(mapName, vSettings);
 
         return new JvmCachedReference<>(name, vCache, supplier, settings);
     }
 
     private <V> CachedReference<V> createDistributedCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         final String mapName = PREFIX_CACHE_REFERENCE + name;
         final ExternalCacheSettings vSettings = buildExternalCacheSettings(settings);
 
         final StableReadExternalCache<Serializable> vCache =
                 vCacheFactory.getStableReadExternalCache(
                         mapName,
                         MarshallerFactory.serializableMarshaller(Serializable.class),
                         vSettings);
 
         return new ExternalCachedReference<>(name, vCache, supplier, settings);
     }
 
     private <V> CachedReference<V> createHybridCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         final JvmCacheSettings localSettings = buildJvmSettings(settings);
         final ExternalCacheSettings externalSettings = buildExternalCacheSettings(settings);
 
         final JvmCache<String, Versioned<V>> localVersioned =
                 vCacheFactory.getJvmCache(PREFIX_HYBRID_LOCAL_CACHE + name, localSettings);
         final StableReadExternalCache<String> globalVersions =
                 vCacheFactory.getStableReadExternalCache(
                         PREFIX_HYBRID_GLOBAL_CACHE + name, MarshallerFactory.stringMarshaller(), externalSettings);
 
         return new HybridCachedReference<>(name, localVersioned, globalVersions, supplier, settings);
     }
 
     private <V> CachedReference<V> doCreateCachedReference(String name, Supplier<V> supplier, CacheSettings settings)
     {
         if (settings.getLocal(false))
         {
             return createLocalCachedReference(name, supplier, settings);
         }
 
         if (settings.getReplicateViaCopy(true))
         {
             return createDistributedCachedReference(name, supplier, settings);
         }
 
         return createHybridCachedReference(name, supplier, settings);
     }
 
     private JvmCacheSettings buildJvmSettings(CacheSettings settings)
     {
         final JvmCacheSettingsBuilder bob = new JvmCacheSettingsBuilder();
         if (settings.getExpireAfterWrite() != null)
         {
             bob.defaultTtl(Duration.ofMillis(settings.getExpireAfterWrite()));
         }
         if (settings.getExpireAfterAccess() != null)
         {
             bob.defaultTtl(Duration.ofMillis(settings.getExpireAfterAccess()));
         }
         if (settings.getMaxEntries() != null)
         {
             bob.maxEntries(settings.getMaxEntries());
         }
 
         return bob.build();
     }
 
     private ExternalCacheSettings buildExternalCacheSettings(CacheSettings settings)
     {
         final ExternalCacheSettingsBuilder bob = new ExternalCacheSettingsBuilder();
         if (settings.getExpireAfterWrite() != null)
         {
             bob.defaultTtl(Duration.ofMillis(settings.getExpireAfterWrite()));
         }
         if (settings.getExpireAfterAccess() != null)
         {
             bob.defaultTtl(Duration.ofMillis(settings.getExpireAfterAccess()));
         }
         if (settings.getMaxEntries() != null)
         {
             bob.entryCountHint(settings.getMaxEntries());
         }
 
         return bob.build();
     }
 }