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.internal.RequestContext;
   import com.atlassian.vcache.internal.core.ExternalCacheKeyGenerator;
   import com.atlassian.vcache.internal.core.TransactionControlManager;
   import com.atlassian.vcache.internal.core.VCacheCoreUtils;
  import com.atlassian.vcache.internal.core.metrics.MetricsRecorder;
  import com.atlassian.vcache.internal.core.service.AbstractTransactionalExternalCache;
  import com.atlassian.vcache.internal.core.service.VersionedExternalCacheRequestContext;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import redis.clients.jedis.Jedis;
  import redis.clients.jedis.Pipeline;
  import redis.clients.jedis.Response;
  
  import java.time.Duration;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.function.Supplier;
  import java.util.stream.Collectors;
  
  import static com.atlassian.vcache.internal.core.VCacheCoreUtils.marshall;
  import static com.atlassian.vcache.internal.core.VCacheCoreUtils.unmarshall;
  import static java.util.Objects.requireNonNull;
  
  /**
   * Redis implementation of {@link com.atlassian.vcache.TransactionalExternalCache}.
   *
   * @param <V> the value type
   * @since 1.0.0
   */
  class RedisTransactionalExternalCache<V>
          extends AbstractTransactionalExternalCache<V> {
      private static final Logger log = LoggerFactory.getLogger(RedisTransactionalExternalCache.class);
  
      private final Supplier<Jedis> clientSupplier;
      private final ExternalCacheKeyGenerator keyGenerator;
      private final MarshallingPair<V> valueMarshalling;
      private final int defaultTtl;
      private final TransactionControlManager transactionControlManager;
  
      RedisTransactionalExternalCache(
              Supplier<Jedis> clientSupplier,
              Supplier<RequestContext> contextSupplier,
              ExternalCacheKeyGenerator keyGenerator,
              String name,
              MarshallingPair<V> valueMarshalling,
              ExternalCacheSettings settings,
              TransactionControlManager transactionControlManager,
              MetricsRecorder metricsRecorder,
              Duration lockTimeout) {
          super(name, contextSupplier, metricsRecorder, lockTimeout);
          this.clientSupplier = requireNonNull(clientSupplier);
          this.keyGenerator = requireNonNull(keyGenerator);
          this.valueMarshalling = requireNonNull(valueMarshalling);
          this.defaultTtl = VCacheCoreUtils.roundUpToSeconds(settings.getDefaultTtl().get());
          this.transactionControlManager = requireNonNull(transactionControlManager);
      }
  
      @Override
      public void transactionSync() {
          log.trace("Cache {}: synchronising operations", name);
          final VersionedExternalCacheRequestContext<V> cacheContext = ensureCacheContext();
  
          if (cacheContext.hasRemoveAll()) {
              cacheContext.updateCacheVersion(
                      RedisUtils.incrementCacheVersion(clientSupplier, cacheContext.externalCacheVersionKey()));
          }
  
          performKeyedOperations(cacheContext);
          cacheContext.forgetAll();
      }
  
      private void performKeyedOperations(VersionedExternalCacheRequestContext<V> cacheContext) {
          // The approach used to perform the operations is:
          // - schedule all the operations
          // - check the result of all the operations and log a warning is necessary
          // - if any operation failed, wipe the cache
  
          // Right then, lets submit all the operations.
          final List<Response<Long>> deleteResponses;
          final List<Response<String>> putResponses;
          try (Jedis client = clientSupplier.get()) {
              final Pipeline pipeline = client.pipelined();
  
              deleteResponses = cacheContext.getKeyedOperations().stream()
                      .filter(e -> e.getValue().isRemove())
                      .map(e -> pipeline.del(cacheContext.externalEntryKeyFor(e.getKey()).getBytes()))
                      .collect(Collectors.toList());
  
              putResponses = cacheContext.getKeyedOperations().stream()
                      .filter(e -> e.getValue().isPut())
                      .map(e ->
                              RedisUtils.pipelinePutOperationForPolicy(
                                     pipeline,
                                     cacheContext.externalEntryKeyFor(e.getKey()),
                                     e.getValue().getPolicy(),
                                     marshall(e.getValue().getValue(), valueMarshalling),
                                     defaultTtl))
                     .collect(Collectors.toList());
 
             pipeline.sync();
         }
 
         // Check if unable to perform the removes (deletes)
         final long numFailedRemoves = deleteResponses.stream()
                 .filter(r -> r.get() == 0)
                 .count();
         if (numFailedRemoves > 0) {
             log.debug("Cache {}: Failed to perform {} of {} transactional removes, clearing the cache.",
                     name, numFailedRemoves, deleteResponses.size());
             cacheContext.updateCacheVersion(
                     RedisUtils.incrementCacheVersion(clientSupplier, cacheContext.externalCacheVersionKey()));
         } else {
             // Check for failed puts, which is terrible
             final long numFailedPuts = putResponses.stream()
                     .filter(r -> !RedisUtils.OK.equals(r.get()))
                     .count();
             if (numFailedPuts > 0) {
                 log.debug("Cache {}: Failed to perform {} of {} transactional puts, clearing the cache.",
                         name, numFailedPuts, putResponses.size());
                 cacheContext.updateCacheVersion(
                         RedisUtils.incrementCacheVersion(clientSupplier, cacheContext.externalCacheVersionKey()));
             }
         }
     }
 
     @Override
     protected VersionedExternalCacheRequestContext<V> ensureCacheContext() {
         final RequestContext requestContext = contextSupplier.get();
 
         transactionControlManager.registerTransactionalExternalCache(requestContext, name, this);
 
         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);
             final VersionedExternalCacheRequestContext<V> newCacheContext = new VersionedExternalCacheRequestContext<>(
                     keyGenerator, name, requestContext::partitionIdentifier, lockTimeout);
             newCacheContext.updateCacheVersion(
                     RedisUtils.obtainCacheVersion(clientSupplier, newCacheContext.externalCacheVersionKey(),
                             defaultTtl + 1));
             return newCacheContext;
         });
     }
 
     @Override
     protected Logger getLogger() {
         return log;
     }
 
     @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);
     }
 }