package com.atlassian.cache.vcache;
   
   import java.util.Collection;
   import java.util.Objects;
   import java.util.Optional;
   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.Supplier;
  import com.atlassian.vcache.JvmCache;
  import com.atlassian.vcache.PutPolicy;
  import com.atlassian.vcache.StableReadExternalCache;
  
  import com.google.common.base.Throwables;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static com.atlassian.cache.vcache.Utils.asString;
  import static com.atlassian.vcache.VCacheUtils.join;
  import static java.util.Objects.requireNonNull;
  
  class HybridCache<K, V>
          extends ManagedCacheSupport
          implements Cache<K, V>
  {
      private static final Logger log = LoggerFactory.getLogger(HybridCache.class);
  
      private final JvmCache<K, Versioned<V>> localVersioned;
      private final StableReadExternalCache<String> globalVersions;
      private final CacheLoader<K, V> cacheLoader;
  
      HybridCache(String name,
                  JvmCache<K, Versioned<V>> localVersioned,
                  StableReadExternalCache<String> globalVersions,
                  @Nullable CacheLoader<K, V> cacheLoader,
                  CacheSettings settings)
      {
          super(name, settings);
          this.localVersioned = requireNonNull(localVersioned);
          this.globalVersions = requireNonNull(globalVersions);
          this.cacheLoader = cacheLoader;
      }
  
      @Override
      public void clear()
      {
          globalVersions.removeAll();
          localVersioned.removeAll();
      }
  
      @Override
      public boolean isLocal()
      {
          return false;
      }
  
      @Override
      public boolean containsKey(@Nonnull K key)
      {
          return localVersioned.get(key).isPresent();
      }
  
      @Nonnull
      @Override
      public Collection<K> getKeys()
      {
          return localVersioned.getKeys();
      }
  
      @Nullable
      @Override
      public V get(@Nonnull K key)
      {
          if (cacheLoader != null)
          {
              return get(key, () -> cacheLoader.load(key));
          }
  
          final Optional<Versioned<V>> versionedOptional = localVersioned.get(key);
  
          if (versionedOptional.isPresent())
          {
              final Optional<String> globalVersion = join(globalVersions.get(asString(key)));
  
              if (globalVersion.isPresent()
                      && globalVersion.get().equals(versionedOptional.get().getVersion()))
              {
                  // versions match, cache is up to date
                  return versionedOptional.get().getValue();
              }
  
              // Value in localVersioned is outdated, clear it.
              //noinspection unchecked
              localVersioned.remove(key);
         }
 
         return null; // No match
     }
 
     @Nonnull
     @Override
     public V get(@Nonnull K key, @Nonnull Supplier<? extends V> valueSupplier)
     {
         final Optional<Versioned<V>> versionedOptional = localVersioned.get(key);
 
         if (versionedOptional.isPresent())
         {
             final Optional<String> globalVersion = join(globalVersions.get(asString(key)));
 
             if (globalVersion.isPresent()
                     && globalVersion.get().equals(versionedOptional.get().getVersion()))
             {
                 // versions match, cache is up to date
                 return versionedOptional.get().getValue();
             }
 
             // Value in localVersioned is outdated, clear it.
             //noinspection unchecked
             localVersioned.remove(key);
         }
 
         return localVersioned.get(key, () -> loadAndVersion(key, valueSupplier)).getValue();
     }
 
     @Override
     public void put(@Nonnull K key, @Nonnull V value)
     {
         final String newVersion = newVersion(key);
         localVersioned.put(key, new Versioned<>(newVersion, value));
     }
 
     /**
      * {@inheritDoc}
      * <p>
      * This implementation has a weak spot: When there is an existing value, this method must return it. However, if
      * that value is stored in a remote node, it cannot be returned. <tt>null</tt> is returned instead, which signals
      * success. In order to not break any contracts, this implementation will invalidate the value on all nodes in this
      * situation.
      */
     @Nullable
     @Override
     public V putIfAbsent(@Nonnull K key, @Nonnull V value)
     {
         final String newVersion = Utils.uniqueId();
         final Versioned<V> versioned = new Versioned<>(newVersion, value);
         final Optional<Versioned<V>> oldValue = localVersioned.putIfAbsent(key, versioned);
 
         if (oldValue.isPresent())
         {
             // There is already an existing value, so just return it.
             return oldValue.get().getValue();
         }
 
         // Entry was missing, so need to do a PUT_ALWAYS on the version number, to invalidate the value on all
         // other nodes (see Javadoc comment above)
         if (!join(globalVersions.put(asString(key), newVersion, PutPolicy.PUT_ALWAYS)))
         {
             log.info("Unable to add a version for key '{}' in cache with name '{}'", key, localVersioned.getName());
         }
 
         return null;
     }
 
     @Override
     public void remove(@Nonnull K key)
     {
         // add a new version to trigger a cluster-wide invalidation
         newVersion(key);
         //noinspection unchecked
         localVersioned.remove(key);
     }
 
     @Override
     public boolean remove(@Nonnull K key, @Nonnull V value)
     {
         final Optional<Versioned<V>> localValue = localVersioned.get(key);
 
         if (!localValue.isPresent() || !Objects.equals(value, localValue.get().getValue()))
         {
             return false;
         }
 
         remove(key);
         return true;
     }
 
     @Override
     public void removeAll()
     {
         if (isFlushable())
         {
             globalVersions.removeAll();
             localVersioned.removeAll();
         }
     }
 
     @Override
     public boolean replace(@Nonnull K key, @Nonnull V oldValue, @Nonnull V newValue)
     {
         final Optional<Versioned<V>> localValue = localVersioned.get(key);
 
         if (!localValue.isPresent() || !Objects.equals(oldValue, localValue.get().getValue()))
         {
             return false;
         }
 
         final String newVersion = newVersion(key);
         if (localVersioned.replaceIf(key, localValue.get(), new Versioned<>(newVersion, newValue)))
         {
             join(globalVersions.put(asString(key), newVersion, PutPolicy.PUT_ALWAYS));
             return true;
         }
 
         return false;
     }
 
     @Override
     public void addListener(@Nonnull CacheEntryListener<K, V> listener, boolean includeValues)
     {
         throw new UnsupportedOperationException("Unsupported when using the VCache implementation");
     }
 
     @Override
     public void removeListener(@Nonnull CacheEntryListener<K, V> listener)
     {
         throw new UnsupportedOperationException("Unsupported when using the VCache implementation");
     }
 
     @Nonnull
     private String getVersion(K key)
     {
         // get the existing value, or create a new one
         return join(globalVersions.get(asString(key), Utils::uniqueId));
     }
 
     @Nonnull
     private Versioned<V> loadAndVersion(final K key, Supplier<? extends V> supplier)
     {
         try
         {
             final String version = getVersion(key);
 
             final V value = supplier.get();
             if (value == null)
             {
                 throw new CacheException("The generated value for cache '" + getName() + "' was null for key '" +
                         key + "'. Null values are not supported.");
             }
 
             log.debug("Generated value '{}' for key '{}' in cache with name '{}'", value, key, localVersioned.getName());
             return new Versioned<>(version, value);
         }
         catch (RuntimeException e)
         {
             Throwables.propagateIfInstanceOf(e, CacheException.class);
             throw new CacheException(
                     "Error generating a value for key '" + key + "' in cache '" + localVersioned.getName() + "'", e);
         }
     }
 
     @Nonnull
     private String newVersion(K key)
     {
         final String result = Utils.uniqueId();
         join(globalVersions.put(asString(key), result, PutPolicy.PUT_ALWAYS));
         return result;
     }
 }