package com.atlassian.vcache.internal.redis;
   
   import com.atlassian.marshalling.api.MarshallingPair;
   import com.atlassian.vcache.ExternalCacheException;
   import com.atlassian.vcache.ExternalCacheSettings;
   import com.atlassian.vcache.PutPolicy;
   import com.atlassian.vcache.internal.MetricLabel;
   import com.atlassian.vcache.internal.RequestContext;
   import com.atlassian.vcache.internal.core.ExternalCacheKeyGenerator;
  import com.atlassian.vcache.internal.core.VCacheCoreUtils;
  import com.atlassian.vcache.internal.core.metrics.CacheType;
  import com.atlassian.vcache.internal.core.metrics.MetricsRecorder;
  import com.atlassian.vcache.internal.core.service.AbstractStableReadExternalCache;
  import com.atlassian.vcache.internal.core.service.VersionedExternalCacheRequestContext;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import redis.clients.jedis.Jedis;
  
  import java.time.Duration;
  import java.util.List;
  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.internal.core.VCacheCoreUtils.isEmpty;
  import static com.atlassian.vcache.internal.core.VCacheCoreUtils.unmarshall;
  import static java.util.Objects.requireNonNull;
  
  /**
   * Redis implementation of {@link com.atlassian.vcache.StableReadExternalCache}.
   *
   * @param <V> the value type
   * @since 1.0.0
   */
  class RedisStableReadExternalCache<V>
          extends AbstractStableReadExternalCache<V> {
      private static final Logger log = LoggerFactory.getLogger(RedisStableReadExternalCache.class);
  
      private final Supplier<Jedis> clientSupplier;
      private final Supplier<RequestContext> contextSupplier;
      private final ExternalCacheKeyGenerator keyGenerator;
      private final MarshallingPair<V> valueMarshalling;
      private final int defaultTtl;
  
      RedisStableReadExternalCache(
              Supplier<Jedis> clientSupplier,
              Supplier<RequestContext> contextSupplier,
              ExternalCacheKeyGenerator keyGenerator,
              String name,
              MarshallingPair<V> valueMarshalling,
              ExternalCacheSettings settings,
              MetricsRecorder metricsRecorder,
              Duration lockTimeout) {
          super(name, metricsRecorder, lockTimeout, (n, ex) -> {});
          this.clientSupplier = requireNonNull(clientSupplier);
          this.contextSupplier = requireNonNull(contextSupplier);
          this.keyGenerator = requireNonNull(keyGenerator);
          this.valueMarshalling = requireNonNull(valueMarshalling);
          //noinspection OptionalGetWithoutIsPresent
          this.defaultTtl = VCacheCoreUtils.roundUpToSeconds(settings.getDefaultTtl().get());
      }
  
      @Override
      public boolean internalPut(String internalKey, V value, PutPolicy policy) {
          final String externalKey = ensureCacheContext().externalEntryKeyFor(internalKey);
          final byte[] valueBytes = valueMarshalling.getMarshaller().marshallToBytes(value);
          return RedisUtils.putOperationForPolicy(
                  policy, clientSupplier, externalKey, defaultTtl, valueBytes);
      }
  
      @Override
      protected void internalRemove(Iterable<String> internalKeys) {
          if (isEmpty(internalKeys)) {
              return;
          }
  
          // Lodge all the requests for delete
          try (Jedis client = clientSupplier.get()) {
              final VersionedExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
              final List<byte[]> externalKeysList = StreamSupport.stream(internalKeys.spliterator(), false)
                      .map(cacheContext::externalEntryKeyFor)
                      .map(String::getBytes)
                      .collect(Collectors.toList());
              final byte[][] externalKeysAsBytes = externalKeysList.toArray(new byte[externalKeysList.size()][]);
  
              final long numDeleted = client.del(externalKeysAsBytes);
              if (numDeleted != externalKeysAsBytes.length) {
                  log.info("Cache {}: only able to delete {} of {} keys", name, numDeleted, externalKeysAsBytes.length);
              }
  
              // Forget the values removed
              StreamSupport.stream(internalKeys.spliterator(), false)
                      .forEach(k -> cacheContext.recordValue(k, Optional.empty()));
         }
     }
 
     @Override
     protected void internalRemoveAll() {
         ensureCacheContext().updateCacheVersion(RedisUtils.cacheVersionIncrementer(clientSupplier));
     }
 
     @Override
     protected Logger getLogger() {
         return log;
     }
 
     @Override
     protected VersionedExternalCacheRequestContext<V> ensureCacheContext() {
         final RequestContext requestContext = contextSupplier.get();
 
         return requestContext.computeIfAbsent(this, () -> {
             // Need to build a new context, which involves getting the current cache version, or setting it if it does
             // not exist.
             log.trace("Cache {}: Setting up a new context", name);
             return new VersionedExternalCacheRequestContext<>(
                     keyGenerator, name, requestContext::partitionIdentifier,
                     RedisUtils.cacheVersionSupplier(clientSupplier, defaultTtl + 1),
                     lockTimeout);
         });
     }
 
     @Override
     protected V handleCreation(String internalKey, V candidateValue)
             throws ExecutionException, InterruptedException {
         final VersionedExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
         final byte[] candidateValueBytes = valueMarshalling.getMarshaller().marshallToBytes(candidateValue);
         final String externalKey = cacheContext.externalEntryKeyFor(internalKey);
 
         // Loop until either able to add the candidate value, or retrieve one that has been added by another thread
         try (Jedis client = clientSupplier.get()) {
             for (; ; ) {
                 metricsRecorder.record(name, CacheType.EXTERNAL, MetricLabel.NUMBER_OF_REMOTE_GET, 1);
                 final long addOp = client.setnx(externalKey.getBytes(), candidateValueBytes);
                 if (addOp == 1) {
                     // I break here, rather than just return, due to battling with the compiler. Unless written
                     // this way, the compiler will not allow the lambda structure.
                     break;
                 }
 
                 getLogger().info("Cache {}, unable to add candidate for key {}, retrieve what was added", name, internalKey);
                 metricsRecorder.record(name, CacheType.EXTERNAL, MetricLabel.NUMBER_OF_REMOTE_GET, 1);
                 final Optional<V> otherAddedValue = unmarshall(client.get(externalKey.getBytes()), valueMarshalling);
                 if (otherAddedValue.isPresent()) {
                     return otherAddedValue.get();
                 }
 
                 getLogger().info("Cache {}, unable to retrieve recently added candidate for key {}, looping", name, internalKey);
             }
         }
         return candidateValue;
     }
 
     @Override
     public CompletionStage<Map<String, V>> getBulk(Function<Set<String>, Map<String, V>> factory, String... keys) {
         return super.getBulk(factory, keys);
     }
 
     @Override
     protected final ExternalCacheException mapException(Exception ex) {
         return RedisUtils.mapException(ex);
     }
 
     @Override
     protected final Optional<V> directGet(String externalKey) {
         try (Jedis client = clientSupplier.get()) {
             return unmarshall(client.get(externalKey.getBytes()), valueMarshalling);
         }
     }
 
     @Override
     protected final Map<String, Optional<V>> directGetBulk(Set<String> externalKeys) {
         return RedisUtils.directGetBulk(externalKeys, clientSupplier, valueMarshalling);
     }
 }