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.ExternalCacheException.Reason.TRANSACTION_FAILURE;
  import static com.atlassian.vcache.VCacheUtils.fold;
  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, (n, ex) -> {});
          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();
  
          fold(perform(() -> {
                      if (cacheContext.hasRemoveAll()) {
                          cacheContext.updateCacheVersion(RedisUtils.cacheVersionIncrementer(clientSupplier));
                      }
  
                      performKeyedOperations(cacheContext);
  
                      return null;
                  }),
                  v -> null,
                  err -> {
                      log.warn("Cache {}: an operation failed during transaction sync ({}). Clearing cache to remove stale entries.", name, err.getMessage());
                      try {
                          cacheContext.updateCacheVersion(RedisUtils.cacheVersionIncrementer(clientSupplier));
                      } catch (RuntimeException e) {
                          log.error("Cache {}: failed to clear the cache: {}", name, e.getMessage());
                      }
  
                      return null;
                  });
  
          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.warn("Cache {}: Failed to perform {} of {} transactional removes, clearing the cache.",
                     name, numFailedRemoves, deleteResponses.size());
             throw new ExternalCacheException(TRANSACTION_FAILURE);
         } 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.warn("Cache {}: Failed to perform {} of {} transactional puts, clearing the cache.",
                         name, numFailedPuts, putResponses.size());
                 throw new ExternalCacheException(TRANSACTION_FAILURE);
             }
         }
     }
 
     @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);
             return new VersionedExternalCacheRequestContext<>(
                     keyGenerator, name, requestContext::partitionIdentifier,
                     RedisUtils.cacheVersionSupplier(clientSupplier, defaultTtl + 1),
                     lockTimeout);
         });
     }
 
     @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);
     }
 }