package com.atlassian.vcache.internal.core.service;
   
   import com.atlassian.marshalling.api.MarshallingException;
   import com.atlassian.vcache.ExternalCacheException;
   import com.atlassian.vcache.VCacheUtils;
   import com.atlassian.vcache.internal.MetricLabel;
   import com.atlassian.vcache.internal.core.metrics.CacheType;
   import com.atlassian.vcache.internal.core.metrics.MetricsRecorder;
   
  import java.util.HashMap;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.CompletableFuture;
  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.ExternalCacheException.Reason.MARSHALLER_FAILURE;
  import static com.atlassian.vcache.ExternalCacheException.Reason.UNCLASSIFIED_FAILURE;
  import static com.atlassian.vcache.internal.core.VCacheCoreUtils.failed;
  import static com.atlassian.vcache.internal.core.VCacheCoreUtils.isEmpty;
  import static java.util.Objects.requireNonNull;
  
  /**
   * Provides operations common to both {@link com.atlassian.vcache.StableReadExternalCache} and
   * {@link com.atlassian.vcache.TransactionalExternalCache} instances.
   *
   * @param <V> the value type
   * @since 1.0.0
   */
  public abstract class AbstractNonDirectExternalCache<V>
          extends AbstractExternalCache<V> {
  
      protected final MetricsRecorder metricsRecorder;
  
      protected AbstractNonDirectExternalCache(String name, MetricsRecorder metricsRecorder) {
          super(name);
          this.metricsRecorder = requireNonNull(metricsRecorder);
      }
  
      /**
       * Handles the creation of an entry, if required.
       *
       * @param internalKey the internal key for the entry.
       * @param supplier    called to create the value, if required
       * @return the value associated with the key.
       */
      protected abstract V handleCreation(String internalKey, Supplier<V> supplier)
              throws ExecutionException, InterruptedException;
  
      /**
       * Handles the creation of a number of entries
       *
       * @param factory      called to create the required values.
       * @param externalKeys the external keys for the entries.
       * @return the values associated with the supplied keys (mapped on internal key)
       */
      protected abstract Map<String, V> handleCreation(
              Function<Set<String>, Map<String, V>> factory, Set<String> externalKeys)
              throws ExecutionException, InterruptedException;
  
      /**
       * Checks for values recorded for specified internal keys.
       */
      protected abstract Map<String, Optional<V>> checkValuesRecorded(Iterable<String> internalKeys);
  
      /**
       * 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 CompletionStage<Optional<V>> get(String internalKey) {
          final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
  
          final String externalKey = cacheContext.externalEntryKeyFor(internalKey);
  
          final CompletionStage<Optional<V>> value = cacheContext.computeValue(internalKey,
                  (key, oldValue) -> {
  
                      if (oldValue != null && !oldValue.toCompletableFuture().isCompletedExceptionally()) {
                          return oldValue;
                      }
                      // If a transactional cache has fired removeAll, then we always consider the remote to be empty.
                      // This is only true fo transactional caches.
                      if (cacheContext.hasRemoveAll()) {
                          return CompletableFuture.completedFuture(Optional.empty());
                      }
  
                      // We go to the remote cache. If get get an exception, then we return an Optional.empty but put
                     // the exceptional value into the map so that we will try again.
                     return CompletableFuture.<Optional<V>>completedFuture(Optional.empty())
                             .thenApply((ignore) -> {
                                 metricsRecorder.record(name, CacheType.EXTERNAL, MetricLabel.NUMBER_OF_REMOTE_GET, 1);
                                 return directGet(externalKey);
                             });
                 });
 
         // Handles the case where the direct failed - we return empty but leave the exceptional future in the mapso we
         // can retry next time (if there is one).
         return CompletableFuture.completedFuture(VCacheUtils.fold(value, v -> v, e -> Optional.empty()));
     }
 
     @Override
     public CompletionStage<V> get(String internalKey, Supplier<V> supplier) {
         final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
 
         return cacheContext.computeValue(internalKey,
                 (key, oldValue) -> {
                     if (oldValue != null && !oldValue.toCompletableFuture().isCompletedExceptionally()) {
                         return oldValue.thenCompose(value -> {
                             if (value.isPresent()) {
                                 return CompletableFuture.completedFuture(value);
                             } else {
                                 return getNewValue(internalKey, supplier);
                             }
                         });
                     } else {
                         getLogger().trace("Cache {}, creating candidate for key {}", name, internalKey);
 
                         final String externalKey = cacheContext.externalEntryKeyFor(internalKey);
                         metricsRecorder.record(name, CacheType.EXTERNAL, MetricLabel.NUMBER_OF_REMOTE_GET, 1);
 
                         // If a transactional cache has fired removeAll, then we always consider the remote to be empty.
                         // This is only true fo transactional caches.
                         if (!cacheContext.hasRemoveAll()) {
                             final Optional<V> result = directGet(externalKey);
                             if (result.isPresent()) {
                                 // A valid value exists in the external cache
                                 return CompletableFuture.completedFuture(result);
                             }
                         }
                         getLogger().trace("Cache {}, creating candidate for key {}", name, internalKey);
                         return getNewValue(internalKey, supplier);
                     }
                 })
                 // Strip the Optional that is needed for the cacheContext here without completing the future.
                 .thenCompose(value -> CompletableFuture.completedFuture(value.get()));
     }
 
     @Override
     public CompletionStage<Map<String, Optional<V>>> getBulk(Iterable<String> internalKeys) {
         return perform(() -> {
             if (isEmpty(internalKeys)) {
                 return new HashMap<>();
             }
 
             // Get the recorded values first
             final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
             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 CompletionStage<Map<String, V>> getBulk(
             Function<Set<String>, Map<String, V>> factory, Iterable<String> internalKeys) {
         return perform(() -> {
             if (isEmpty(internalKeys)) {
                 return new HashMap<>();
             }
 
             // Get the recorded values first and collect the ones that have values.
             final AbstractExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
             final Map<String, V> grandResult = checkValuesRecorded(internalKeys).entrySet().stream()
                     .filter(e -> e.getValue().isPresent())
                     .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().get()));
 
             // Calculate the candidate externalKeys for the entries that are missing
             final Set<String> candidateMissingExternalKeys = StreamSupport.stream(internalKeys.spliterator(), false)
                     .filter(k -> !grandResult.containsKey(k))
                     .map(cacheContext::externalEntryKeyFor)
                     .collect(Collectors.toSet());
 
             // Bail out if we have all the entries requested
             if (candidateMissingExternalKeys.isEmpty()) {
                 getLogger().trace("Cache {}: getBulk(Function): had all the requested entries cached", name);
                 return grandResult;
             }
             getLogger().trace("Cache {}: getBulk(Function): checking external cache for {} keys",
                     name, candidateMissingExternalKeys.size());
 
             final Map<String, V> missingValues = handleCreation(factory, candidateMissingExternalKeys);
             cacheContext.recordValues(missingValues);
             grandResult.putAll(missingValues);
 
             return grandResult;
         });
     }
 
     private CompletionStage<Optional<V>> getNewValue(String internalKey, Supplier<V> supplier) {
         try {
             return CompletableFuture.completedFuture(Optional.of(handleCreation(internalKey, supplier)));
         } catch (MarshallingException ex) {
             return failed(new CompletableFuture<Optional<V>>(), new ExternalCacheException(MARSHALLER_FAILURE, ex));
         } catch (ExecutionException | InterruptedException ex) {
             return failed(new CompletableFuture<Optional<V>>(), new ExternalCacheException(UNCLASSIFIED_FAILURE, ex));
         } catch (ExternalCacheException ece) {
             return failed(new CompletableFuture<Optional<V>>(), ece);
         } catch (Exception ex) {
             return failed(new CompletableFuture<Optional<V>>(), mapException(ex));
         }
     }
 }