package com.atlassian.cache.memory;
   
   import java.util.Collection;
   import java.util.Objects;
   import java.util.SortedMap;
   import java.util.concurrent.ConcurrentHashMap;
   import java.util.concurrent.ConcurrentMap;
   import java.util.concurrent.ExecutionException;
   import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReadWriteLock;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  import javax.annotation.Nonnull;
  import javax.annotation.Nullable;
  
  import com.atlassian.cache.Cache;
  import com.atlassian.cache.CacheEntryListener;
  import com.atlassian.cache.CacheException;
  import com.atlassian.cache.CacheLoader;
  import com.atlassian.cache.CacheSettings;
  import com.atlassian.cache.CacheStatisticsKey;
  import com.atlassian.cache.impl.CacheEntryListenerSupport;
  import com.atlassian.cache.impl.DefaultCacheEntryListenerSupport;
  import com.atlassian.cache.impl.OneShotLatch;
  import com.atlassian.instrumentation.DefaultInstrumentRegistry;
  import com.atlassian.instrumentation.SimpleTimer;
  import com.atlassian.instrumentation.caches.CacheCollector;
  import com.atlassian.instrumentation.caches.CacheKeys;
  
  import com.google.common.base.Throwables;
  import com.google.common.cache.RemovalListener;
  import com.google.common.cache.RemovalNotification;
  import com.google.common.collect.ImmutableSortedMap;
  import com.google.common.util.concurrent.UncheckedExecutionException;
  
  import static com.atlassian.cache.memory.DelegatingCacheStatistics.toStatistics;
  import static java.util.Objects.requireNonNull;
  
  /**
   * A Cache that delegates Concurrent Map.
   *
   * @since 2.0
   */
  class DelegatingCache<K, V> extends ManagedCacheSupport implements Cache<K, V>
  {
      private final com.google.common.cache.Cache<K, V> internalCache;
      private final CacheEntryListenerSupport<K, V> listenerSupport;
      private final CacheCollector collector;
      private final CacheLoader<K, V> theLoader;
  
      private final ConcurrentMap<K, OneShotLatch> barriers = new ConcurrentHashMap<>(16);
      private final Lock loadLock;
      private final Lock removeAllLock;
      {
          // This needs to be fair to ensure that removeAll does not starve for a busy cache
          final ReadWriteLock loadVsRemoveAllLock = new ReentrantReadWriteLock(true);
          loadLock = loadVsRemoveAllLock.readLock();
          removeAllLock = loadVsRemoveAllLock.writeLock();
      }
  
      private DelegatingCache(final com.google.common.cache.Cache<K, V> internalCache,
                              String name,
                              CacheSettings settings,
                              @Nullable CacheLoader<K, V> theLoader)
      {
          super(name, settings);
          this.internalCache = internalCache;
          this.listenerSupport = new DefaultCacheEntryListenerSupport<>();
          this.theLoader = theLoader;
          this.collector = new DefaultInstrumentRegistry().pullCacheCollector(name, internalCache::size);
  
          // Honor the CacheSettings value.
          if (settings.getStatisticsEnabled() != null && settings.getStatisticsEnabled())
          {
              collector.setEnabled(true);
          }
      }
  
      static <K, V> DelegatingCache<K, V> create(final com.google.common.cache.Cache<K, V> internalCache,
                                                 String name,
                                                 CacheSettings settings,
                                                 CacheLoader<K, V> cacheLoader)
      {
          return new DelegatingCache<>(internalCache, name, settings, cacheLoader);
      }
  
      @Override
      public CacheCollector getCacheCollector()
      {
          return collector;
      }
  
      @Override
      public boolean containsKey(@Nonnull K key)
      {
          return null != internalCache.getIfPresent(key);
      }
  
      @Nonnull
      @Override
     public Collection<K> getKeys()
     {
         try
         {
             return internalCache.asMap().keySet();
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public void put(@Nonnull final K key, @Nonnull final V value)
     {
         try
         {
             V oldValue = internalCache.asMap().put(key, value);
             if (isStatisticsEnabled())
             {
                 collector.put();
             }
             if (oldValue == null)
             {
                 // Here we care only for the case when oldValue was null, e.g. missing as in the other
                 // cases DelegatingRemovalListener will be called with REPLACED notification
                 listenerSupport.notifyAdd(key, value);
             }
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public V get(@Nonnull final K key)
     {
         rejectNullKey(key);
 
         if (theLoader == null)
         {
             V value = internalCache.getIfPresent(key);
             if (isStatisticsEnabled())
             {
                 if (value == null)
                 {
                     collector.miss();
                 }
                 else
                 {
                     collector.hit();
                 }
             }
             return value;
         }
         else
         {
             return get(key, () -> theLoader.load(key));
         }
     }
 
     @Nonnull
     @Override
     public V get(@Nonnull final K key, @Nonnull final com.atlassian.cache.Supplier<? extends V> valueLoader)
     {
         rejectNullKey(key);
 
         // Using the following array as a mechanism for the valueLoader to flag whether it has been invoked.
         // When the valueLoader is invoked, it will record that there was a miss.
         final boolean[] missed = new boolean[1];
         try
         {
             return internalCache.get(key, () -> {
                 missed[0] = true;
                 acquireLockFor(key);
                 loadLock.lock();
 
                 final SimpleTimer timer = isStatisticsEnabled() ? new SimpleTimer(CacheKeys.LOAD_TIME.getName()) : null;
                 if (timer != null)
                 {
                     timer.start();
                 }
 
                 try
                 {
                     return Objects.requireNonNull(valueLoader.get());
                 }
                 finally
                 {
                     if (timer != null)
                     {
                         timer.end();
                         collector.put();
                         collector.getSplits().add(timer);
                     }
                 }
             });
         }
         catch (ExecutionException e)
         {
             throw new CacheException("Unknown failure", e);
         }
         catch (UncheckedExecutionException e)
         {
             Throwable cause = e.getCause();
             Throwables.propagateIfInstanceOf(cause, CacheException.class);
             throw new CacheException(cause);
         }
         finally
         {
             if (missed[0])
             {
                 loadLock.unlock();
                 releaseLockFor(key);
             }
 
             if (isStatisticsEnabled())
             {
                 if (missed[0])
                 {
                     collector.miss();
                 }
                 else
                 {
                     collector.hit();
                 }
             }
         }
     }
 
     @Override
     public void remove(@Nonnull final K key)
     {
         acquireLockFor(key);
         try
         {
             internalCache.invalidate(key);
             if (isStatisticsEnabled())
             {
                 collector.remove();
             }
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
         finally
         {
             releaseLockFor(key);
         }
     }
 
     @Override
     public void removeAll()
     {
         removeAllLock.lock();
         try
         {
             internalCache.invalidateAll();
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
         finally
         {
             removeAllLock.unlock();
         }
     }
 
     @Override
     public V putIfAbsent(@Nonnull K key, @Nonnull V value)
     {
         try
         {
             V oldValue = internalCache.asMap().putIfAbsent(key, value);
             if (oldValue == null)
             {
                 // Here we care only for the case when oldValue was null, e.g. missing as in the other
                 // cases DelegatingRemovalListener will be called with REPLACED notification
                 listenerSupport.notifyAdd(key, value);
             }
             else
             {
                 if (isStatisticsEnabled())
                 {
                     collector.put();
                 }
             }
             return oldValue;
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Override
     public boolean remove(@Nonnull K key, @Nonnull V value)
     {
         try
         {
             return internalCache.asMap().remove(key, value);
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
         finally
         {
             if (isStatisticsEnabled())
             {
                 collector.remove();
             }
         }
     }
 
     @Override
     public boolean replace(@Nonnull K key, @Nonnull V oldValue, @Nonnull V newValue)
     {
         try
         {
             return internalCache.asMap().replace(key, oldValue, newValue);
         }
         catch (Exception e)
         {
             throw new CacheException(e);
         }
     }
 
     @Nonnull
     @Override
     public SortedMap<CacheStatisticsKey, com.atlassian.util.concurrent.Supplier<Long>> getStatistics()
     {
         if (isStatisticsEnabled())
         {
             return toStatistics(collector);
         }
         else
         {
             return ImmutableSortedMap.of();
         }
     }
 
     @Override
     public void clear()
     {
         removeAll();
     }
 
     @Override
     public void setStatistics(final boolean enabled)
     {
         collector.setEnabled(enabled);
     }
 
     @Override
     public boolean isStatisticsEnabled()
     {
         // Either we have it enabled from UI or the config of the cache has it set from start up.
         return collector.isEnabled();
     }
 
     @Override
     public boolean equals(@Nullable final Object other)
     {
         if (other instanceof DelegatingCache)
         {
             DelegatingCache<?, ?> otherDelegatingCache = (DelegatingCache<?, ?>) other;
             if (internalCache.equals(otherDelegatingCache.internalCache))
             {
                 return true;
             }
         }
         return false;
     }
 
     @Override
     public int hashCode()
     {
         return 3 + internalCache.hashCode();
     }
 
     @Override
     public void addListener(@Nonnull CacheEntryListener<K, V> listener, boolean includeValues)
     {
         listenerSupport.add(listener, includeValues);
     }
 
     @Override
     public void removeListener(@Nonnull CacheEntryListener<K, V> listener)
     {
         listenerSupport.remove(listener);
     }
 
     protected static class DelegatingRemovalListener<K, V> implements RemovalListener<K, V>
     {
         private DelegatingCache<K, V> cache;
 
         protected void onSupply(K key, V value)
         {
             cache.listenerSupport.notifyAdd(key, value);
         }
 
         @Override
         public void onRemoval(@Nonnull RemovalNotification<K, V> notification)
         {
             switch (notification.getCause())
             {
                 case COLLECTED:
                 case EXPIRED:
                     cache.listenerSupport.notifyEvict(notification.getKey(), notification.getValue());
                     break;
                 case EXPLICIT:
                     cache.listenerSupport.notifyRemove(notification.getKey(), notification.getValue());
                     break;
                 case REPLACED:
                     K key = requireNonNull(notification.getKey());
                     cache.listenerSupport.notifyUpdate(key, cache.internalCache.getIfPresent(key),
                             notification.getValue());
                     break;
             }
         }
 
         public void setCache(DelegatingCache<K, V> cache)
         {
             this.cache = cache;
         }
     }
 
     void rejectNullKey(K key)
     {
         if (key == null)
         {
             throw new CacheException(new NullPointerException("Null keys are not supported"));
         }
     }
 
     private OneShotLatch acquireLockFor(@Nonnull K key)
     {
         final OneShotLatch barrier = new OneShotLatch();
         while (true)
         {
             final OneShotLatch existing = barriers.putIfAbsent(key, barrier);
 
             // successfully raised a new barrier
             if (existing == null)
             {
                 return barrier;
             }
 
             // There is an existing barrier that happens-before us; wait for it to be released.
             // There is no need to attempt a remove or replace to evict it from the barriers map;
             // The thread that owned it would have taken care of that before releasing it.
             existing.await();
         }
     }
 
     private void releaseLockFor(K key)
     {
         final OneShotLatch barrier = barriers.get(key);
         if (barrier != null && barrier.isHeldByCurrentThread())
         {
             barriers.remove(key);
             barrier.release();
         }
     }
 }