package com.atlassian.vcache.internal.core.service;
   
   import java.util.Collections;
   import java.util.HashMap;
   import java.util.Map;
   import java.util.Optional;
   import java.util.Set;
   import java.util.function.Supplier;
   import javax.annotation.Nonnull;
  
  import com.atlassian.vcache.PutPolicy;
  import com.atlassian.vcache.internal.core.ExternalCacheKeyGenerator;
  
  import com.google.common.collect.BiMap;
  import com.google.common.collect.HashBiMap;
  import com.google.common.collect.Maps;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * Represents the request context for an {@link com.atlassian.vcache.ExternalCache}.
   * @param <V> the value type
   */
  public abstract class AbstractExternalCacheRequestContext<V>
  {
      private static final Logger log = LoggerFactory.getLogger(AbstractExternalCacheRequestContext.class);
  
      protected final String name;
      private final ExternalCacheKeyGenerator keyGenerator;
      private final Supplier<String> partitionSupplier;
  
      // Need to synchronise the map to ensure consistency. Easier to use the Guava BiMap to achieve the result.
      private final BiMap<String, String> internalToExternalKeyMap = Maps.synchronizedBiMap(HashBiMap.create());
      // Make the inverse map unmodifiable, to ensure no duplicate processing.
      private final Map<String, String> externalToInternalKeyMap = Collections.unmodifiableMap(
              internalToExternalKeyMap.inverse());
  
      private final Map<String, Optional<V>> internalKeyToValueMap = new HashMap<>();
      private boolean hasRemoveAll;
      private final Map<String, DeferredOperation<V>> keyedOperationMap = new HashMap<>();
  
      protected AbstractExternalCacheRequestContext(ExternalCacheKeyGenerator keyGenerator,
                                                    String name,
                                                    Supplier<String> partitionSupplier)
      {
          this.keyGenerator = requireNonNull(keyGenerator);
          this.name = requireNonNull(name);
          this.partitionSupplier = requireNonNull(partitionSupplier);
      }
  
      protected abstract long cacheVersion();
  
      protected void clearKeyMaps()
      {
          internalToExternalKeyMap.clear();
      }
  
      @Nonnull
      public String externalEntryKeyFor(String internalKey)
      {
          final String cached = internalToExternalKeyMap.get(internalKey);
          if (cached != null)
          {
              return cached;
          }
  
          final String result = keyGenerator.entryKey(
                  partitionSupplier.get(), name, cacheVersion(), internalKey);
  
          // Do a forcePut in case there is a race condition and another thread is also adding the entry
          internalToExternalKeyMap.forcePut(internalKey, result);
          return result;
      }
  
      /**
       * Returns the internal key for a supplied external key. The {@link #externalEntryKeyFor(String)} must have
       * returned the value now being specified, in the current request.
       *
       * @param externalKey the external key previously returned by {@link #externalEntryKeyFor(String)}
       * @return the internal key
       */
      @Nonnull
      public String internalEntryKeyFor(String externalKey)
      {
          return requireNonNull(externalToInternalKeyMap.get(externalKey));
      }
  
      @Nonnull
      public Optional<Optional<V>> getValueRecorded(String internalKey)
      {
          return Optional.ofNullable(internalKeyToValueMap.get(internalKey));
      }
  
      public void recordValue(String internalKey, Optional<V> outcome)
      {
          log.trace("Cache {}, recording value for {}", name, internalKey);
          internalKeyToValueMap.put(internalKey, outcome);
     }
 
     public void recordValues(Map<String, V> knownValues)
     {
         log.trace("Cache {}, recording {} known values", name, knownValues.size());
         knownValues.entrySet().stream().collect(
                 () -> internalKeyToValueMap,
                 (m, e) -> m.put(e.getKey(), Optional.of(e.getValue())),
                 Map::putAll);
     }
 
     public void forgetValue(String internalKey)
     {
         log.trace("Cache {}, forgetting value for {}", name, internalKey);
         internalKeyToValueMap.remove(internalKey);
     }
 
     public void forgetAllValues()
     {
         log.trace("Cache {}, forgetting all values", name);
         internalKeyToValueMap.clear();
     }
 
     public void recordPut(String internalKey, V value, PutPolicy policy)
     {
         recordValue(internalKey, Optional.of(value));
         keyedOperationMap.put(internalKey, DeferredOperation.putOperation(value, policy));
     }
 
     public void recordRemove(Iterable<String> internalKeys)
     {
         for (String internalKey : internalKeys)
         {
             recordValue(internalKey, Optional.empty());
             keyedOperationMap.put(internalKey, DeferredOperation.removeOperation());
         }
     }
 
     public void recordRemoveAll()
     {
         forgetAllValues();
         hasRemoveAll = true;
         keyedOperationMap.clear(); // Forget everything, as starting again
     }
 
     public boolean hasRemoveAll()
     {
         return hasRemoveAll;
     }
 
     public void forgetAll()
     {
         forgetAllValues();
         hasRemoveAll = false;
         keyedOperationMap.clear();
     }
 
     public Set<Map.Entry<String, DeferredOperation<V>>> getKeyedOperations()
     {
         return keyedOperationMap.entrySet();
     }
 
     /**
      * Represents a deferred operation.
      * @param <V> the value type
      */
     public static class DeferredOperation<V>
     {
         private final boolean remove;
         private final Optional<V> value;
         private final Optional<PutPolicy> policy;
 
         private DeferredOperation()
         {
             this.remove = true;
             this.value = Optional.empty();
             this.policy = Optional.empty();
         }
 
         private DeferredOperation(V value, PutPolicy policy)
         {
             this.remove = false;
             this.value = Optional.of(value);
             this.policy = Optional.of(policy);
         }
 
         public boolean isRemove()
         {
             return remove;
         }
 
         public boolean isPut()
         {
             return !remove;
         }
 
         public V getValue()
         {
             return value.get();
         }
 
         public PutPolicy getPolicy()
         {
             return policy.get();
         }
 
         public static <V> DeferredOperation<V> removeOperation()
         {
             return new DeferredOperation<>();
         }
 
         public static <V> DeferredOperation<V> putOperation(V value, PutPolicy policy)
         {
             return new DeferredOperation<>(value, policy);
         }
     }
 }