package com.atlassian.vcache.internal.core.service;
   
   import com.atlassian.vcache.PutPolicy;
   import com.atlassian.vcache.StableReadExternalCache;
   import com.atlassian.vcache.internal.ExternalCacheExceptionListener;
   import com.atlassian.vcache.internal.MetricLabel;
   import com.atlassian.vcache.internal.core.metrics.CacheType;
   import com.atlassian.vcache.internal.core.metrics.MetricsRecorder;
   import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.time.Duration;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.CompletionStage;
  import java.util.concurrent.ExecutionException;
  import java.util.function.Function;
  import java.util.function.Supplier;
  import java.util.stream.Collectors;
  import java.util.stream.StreamSupport;
  
  import static com.atlassian.vcache.VCacheUtils.unsafeJoin;
  import static com.atlassian.vcache.internal.core.VCacheCoreUtils.isEmpty;
  import static java.util.Objects.requireNonNull;
  
  /**
   * Provides operations common to {@link com.atlassian.vcache.StableReadExternalCache} instances.
   * <p>
   * Locking is provided to synchronise the local and remote caches. There are two locking strategies - a cache wide
   * read-write lock and a per-key lock. The cache wide lock provides exclusion for multi-key operations such as
   * removeAll and getBulk. The key lock provides locking for single key get and put whilst honouring the cache lock.
   *
   * @param <V> the value type
   * @since 1.0.0
   */
  public abstract class AbstractStableReadExternalCache<V>
          extends AbstractExternalCache<V>
          implements StableReadExternalCache<V> {
  
      private static final Logger log = LoggerFactory.getLogger(AbstractStableReadExternalCache.class);
      protected final MetricsRecorder metricsRecorder;
  
      protected AbstractStableReadExternalCache(
              String name,
              MetricsRecorder metricsRecorder,
              Duration lockTimeout,
              ExternalCacheExceptionListener externalCacheExceptionListener) {
          super(name, lockTimeout, externalCacheExceptionListener);
          this.metricsRecorder = requireNonNull(metricsRecorder);
      }
  
      protected abstract boolean internalPut(String internalKey, V value, PutPolicy policy);
  
      protected abstract void internalRemoveAll();
  
      protected abstract void internalRemove(Iterable<String> keys);
  
      /**
       * Handles the creation of an entry, if required.
       *
       * @param internalKey    the internal key for the entry.
       * @param candidateValue the candidate value to add, if required
       * @return the value associated with the key.
       */
      protected abstract V handleCreation(String internalKey, V candidateValue)
              throws ExecutionException, InterruptedException;
  
      /**
       * Performs a direct get operation against the external cache using the supplied external key.
       */
      protected abstract Optional<V> directGet(String externalKey);
  
      /**
       * Performs a direct bulk get operation against the external cache using the supplied external keys.
       */
      protected abstract Map<String, Optional<V>> directGetBulk(Set<String> externalKeys);
  
      @Override
      public final CompletionStage<Optional<V>> get(String internalKey) {
          return perform(() -> {
              final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
  
              return cacheContext.getGlobalLock().withLock(() -> internalGetWithoutLock(internalKey, cacheContext));
          });
      }
  
      @Override
      public final CompletionStage<V> get(String internalKey, Supplier<V> supplier) {
          return perform(() -> {
              final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
  
              // Check for existing value. Do it under a lock so that we can forget the recorded value (a side-effect
              // of calling get().
              final Optional<V> existingValue = cacheContext.getGlobalLock().withLock(() -> {
                  final Optional<V> value = internalGetWithoutLock(internalKey, cacheContext);
  
                 if (value.isPresent()) {
                     return value;
                 }
 
                 // No value existed, so now forget we did the lookup, so we can see if other threads do it.
                 cacheContext.forgetValue(internalKey);
                 return Optional.empty();
             });
 
             return existingValue.orElseGet(() -> {
                 // Calculate a candidate value, not holding a lock
                 final V candidateValue = requireNonNull(supplier.get());
 
                 // Now record the candidate value
                 return cacheContext.getGlobalLock().withLock(() -> {
                     // Check for a value being added by another thread in the current request context.
                     final Optional<Optional<V>> doubleCheck = cacheContext.getValueRecorded(internalKey);
                     if (doubleCheck.isPresent() && doubleCheck.get().isPresent()) {
                         return doubleCheck.get().get();
                     }
 
                     // Now attempt to add remotely
                     try {
                         final V finalValue = handleCreation(internalKey, candidateValue);
                         cacheContext.recordValue(internalKey, Optional.of(finalValue));
                         return finalValue;
                     } catch (final Exception e) {
                         // We have a supplier created value and memcached is not working:- This mitigates flogging the supplier
                         cacheContext.recordValue(internalKey, Optional.of(candidateValue));
                         return candidateValue;
                     }
                 });
             });
         });
     }
 
     private Optional<V> internalGetWithoutLock(String internalKey, AbstractExternalCacheRequestContext<V> cacheContext) {
         // Introduced so we could avoid duplicate locking and retrieval of AbstractExternalCacheRequestContext.
         // Provides performance boost for get with supplier operation.
 
         // Check if we have recorded a value already
         final Optional<Optional<V>> recordedValue = cacheContext.getValueRecorded(internalKey);
 
         return recordedValue.orElseGet(() -> {
             // Now check externally
             final String externalKey = cacheContext.externalEntryKeyFor(internalKey);
             metricsRecorder.record(name, CacheType.EXTERNAL, MetricLabel.NUMBER_OF_REMOTE_GET, 1);
             final Optional<V> externalValue = directGet(externalKey);
             cacheContext.recordValue(internalKey, externalValue);
 
             return externalValue;
         });
     }
 
 
     @Override
     public final CompletionStage<Map<String, Optional<V>>> getBulk(Iterable<String> internalKeys) {
         return perform(() -> {
             if (isEmpty(internalKeys)) {
                 return new HashMap<>();
             }
 
             final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
             return cacheContext.getGlobalLock().withLock(() -> {
                 // Get the recorded values first
                 final Map<String, Optional<V>> grandResult = checkValuesRecorded(internalKeys);
 
                 // Calculate the externalKeys for the entries that are missing
                 final Set<String> missingExternalKeys = StreamSupport.stream(internalKeys.spliterator(), false)
                         .filter(k -> !grandResult.containsKey(k))
                         .map(cacheContext::externalEntryKeyFor)
                         .collect(Collectors.toSet());
 
                 if (missingExternalKeys.isEmpty()) {
                     getLogger().trace("Cache {}: getBulk(): have all the requested entries cached", name);
                     return grandResult;
                 }
                 getLogger().trace("Cache {}: getBulk(): not cached {} requested entries", name, missingExternalKeys.size());
 
                 // Get the missing values.
                 metricsRecorder.record(name, CacheType.EXTERNAL, MetricLabel.NUMBER_OF_REMOTE_GET, 1);
                 final Map<String, Optional<V>> candidateValues = directGetBulk(missingExternalKeys);
 
                 return candidateValues.entrySet().stream().collect(
                         () -> grandResult,
                         (m, e) -> {
                             final Optional<V> result = e.getValue();
                             cacheContext.recordValue(cacheContext.internalEntryKeyFor(e.getKey()), result);
                             m.put(cacheContext.internalEntryKeyFor(e.getKey()), result);
                         },
                         Map::putAll
                 );
             });
         });
     }
 
     @Override
     public final CompletionStage<Map<String, V>> getBulk(
             Function<Set<String>, Map<String, V>> factory, Iterable<String> internalKeys) {
         return perform(() -> {
             if (isEmpty(internalKeys)) {
                 return new HashMap<>();
             }
 
             final Map<String, V> grandResult = new HashMap<>();
             final Set<String> missingInternalKeys = new HashSet<>();
             final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
             // Use the getBulk() to get the known state for each key. Do this logic under lock as the cacheContext
             // is being updated.
             cacheContext.getGlobalLock().withLock(() -> {
                 final Map<String, Optional<V>> knownState = unsafeJoin(getBulk(internalKeys));
                 knownState.forEach((key, value) -> {
                     if (value.isPresent()) {
                         grandResult.put(key, value.get());
                     } else {
                         missingInternalKeys.add(key);
                         cacheContext.forgetValue(key); // to allow future cache operations
                     }
                 });
             });
 
             // Bail out if we have all the entries requested
             if (missingInternalKeys.isEmpty()) {
                 return grandResult;
             }
 
             // Create the candidate missing values NOT under a lock
             final Map<String, V> candidateValues = factory.apply(missingInternalKeys);
             FactoryUtils.verifyFactoryResult(candidateValues, missingInternalKeys);
 
             // Now under lock and key, merge the candidate values in to the results and the external cache.
             cacheContext.getGlobalLock().withLock(() ->
                     candidateValues.entrySet().forEach(entry -> {
                         metricsRecorder.record(name, CacheType.EXTERNAL, MetricLabel.NUMBER_OF_REMOTE_GET, 1);
                         V finalValue;
                         try {
                             final boolean added = unsafeJoin(put(entry.getKey(), entry.getValue(), PutPolicy.ADD_ONLY));
 
                             if (added) {
                                 finalValue = entry.getValue();
                             } else {
                                 log.trace("Was unable to store the candidate value, so needing to retrieve what's there now");
                                 finalValue = unsafeJoin(get(entry.getKey(), entry::getValue));
                             }
                         } catch (final Exception ignore) {
                             // We could not save the value in Memcached so just put it into the local cache.
                             finalValue = entry.getValue();
                         }
                         grandResult.put(entry.getKey(), finalValue);
                         cacheContext.recordValue(entry.getKey(), Optional.of(finalValue));
                     }));
 
             return grandResult;
         });
     }
 
     @Override
     public final CompletionStage<Boolean> put(final String internalKey, final V value, final PutPolicy policy) {
         return perform(() -> {
             final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
 
             final boolean successful = cacheContext.getGlobalLock().withLock(() -> internalPut(internalKey, value, policy));
             if (successful) {
                 cacheContext.recordValue(internalKey, Optional.of(value));
             } else {
                 cacheContext.forgetValue(internalKey);
             }
             return successful;
         });
     }
 
     @Override
     public final CompletionStage<Void> remove(final Iterable<String> keys) {
         return perform(() -> {
             ensureCacheContext().getGlobalLock().withLock(() -> internalRemove(keys));
             return null;
         });
     }
 
     @Override
     public final CompletionStage<Void> removeAll() {
         return perform(() -> {
             final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
             cacheContext.getGlobalLock().withLock(() -> {
                 internalRemoveAll();
                 cacheContext.forgetAllValues();
             });
             return null;
         });
     }
 
     private Map<String, Optional<V>> checkValuesRecorded(Iterable<String> internalKeys) {
         final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
         //noinspection OptionalGetWithoutIsPresent
         return StreamSupport.stream(internalKeys.spliterator(), false)
                 .filter(k -> cacheContext.getValueRecorded(k).isPresent())
                 .distinct()
                 .collect(Collectors.toMap(
                         k -> k,
                         k -> cacheContext.getValueRecorded(k).get())
                 );
     }
 }