package com.atlassian.vcache.internal.redis;
   
   import com.atlassian.marshalling.api.MarshallingPair;
   import com.atlassian.vcache.CasIdentifier;
   import com.atlassian.vcache.ExternalCacheException;
   import com.atlassian.vcache.PutPolicy;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
   import redis.clients.jedis.Jedis;
  import redis.clients.jedis.Pipeline;
  import redis.clients.jedis.Response;
  import redis.clients.jedis.exceptions.JedisException;
  
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Set;
  import java.util.function.Function;
  import java.util.function.Supplier;
  import java.util.stream.Collectors;
  
  import static com.atlassian.vcache.ExternalCacheException.Reason.UNCLASSIFIED_FAILURE;
  import static com.atlassian.vcache.internal.core.VCacheCoreUtils.unmarshall;
  
  /**
   * Common utility methods that are specific for the Redis implementation.
   *
   * @since 1.0.0
   */
  class RedisUtils {
      static final String OK = "OK";
      static final byte[] NX = "NX".getBytes();
      static final byte[] EX = "EX".getBytes();
      static final byte[] XX = "EX".getBytes();
  
      private static final Logger log = LoggerFactory.getLogger(RedisUtils.class);
  
      private static final String LUA_OBTAIN_CACHE_VERSION =
              "local cur = redis.call(\"get\",KEYS[1]) " +
                      "if cur then " +
                      "    return cur " +
                      "else " +
                      "    redis.call(\"setex\",KEYS[1],ARGV[2],\"666\") " +
                      "    return \"666\" " +
                      "end";
  
      static ExternalCacheException mapException(Exception ex) {
          if (ex instanceof JedisException) {
              return new ExternalCacheException(ExternalCacheException.Reason.NETWORK_FAILURE, ex);
          }
          return new ExternalCacheException(UNCLASSIFIED_FAILURE, ex);
      }
  
      static <V> Map<String, Optional<V>> directGetBulk(
              Set<String> externalKeys, Supplier<Jedis> clientSupplier, MarshallingPair<V> valueMarshalling) {
          // Returns map of keys that contain values, so need to handle the missing ones
          try (Jedis client = clientSupplier.get()) {
              // Implementation note, I have not used the mput() method here as the implementation is
              // non-trivial, and it's not clear that there is any performance benefit to be had.
              // Using Pipeline ensures a single request/response to Redis.
              final Pipeline pipeline = client.pipelined();
              final Map<String, Response<byte[]>> externalKeyToResponseMap = externalKeys.stream()
                      .collect(Collectors.toMap(
                              k -> k,
                              k -> pipeline.get(k.getBytes())
                      ));
              pipeline.sync();
  
              return externalKeyToResponseMap.entrySet().stream().collect(Collectors.toMap(
                      Map.Entry::getKey,
                      e -> unmarshall(e.getValue().get(), valueMarshalling)
              ));
          }
      }
  
      static boolean putOperationForPolicy(
              PutPolicy policy, Supplier<Jedis> clientSupplier, String externalKey, int defaultTtl, byte[] valueBytes) {
          try (Jedis client = clientSupplier.get()) {
              final Pipeline pipeline = client.pipelined();
              final Response<String> result =
                      pipelinePutOperationForPolicy(pipeline, externalKey, policy, valueBytes, defaultTtl);
              pipeline.sync();
              return OK.equals(result.get());
          }
      }
  
      static byte[] safeExtractValue(CasIdentifier casId) {
          if (casId instanceof RedisCasIdentifier) {
              return ((RedisCasIdentifier) casId).getValue();
          }
  
          log.warn("Passed an unknown CasIdentifier instance of class {}.", casId.getClass().getName());
          throw new ExternalCacheException(UNCLASSIFIED_FAILURE);
      }
  
      private static long obtainCacheVersion(Supplier<Jedis> clientSupplier, String externalCacheVersionKey, long ttlSecs) {
          try (Jedis client = clientSupplier.get()) {
              final List<String> keys = Collections.singletonList(externalCacheVersionKey);
             final List<String> args = Collections.singletonList(Long.toString(ttlSecs));
             final String resultString = (String) client.eval(LUA_OBTAIN_CACHE_VERSION, keys, args);
             return Long.valueOf(resultString);
         }
     }
 
     private static long incrementCacheVersion(Supplier<Jedis> clientSupplier, String externalCacheVersionKey) {
         try (Jedis client = clientSupplier.get()) {
             return client.incr(externalCacheVersionKey);
         }
     }
 
     static Function<String, Long> cacheVersionIncrementer(Supplier<Jedis> clientSupplier) {
         return externalCacheVersionKey -> incrementCacheVersion(clientSupplier, externalCacheVersionKey);
     }
 
     static Function<String, Long> cacheVersionSupplier(Supplier<Jedis> clientSupplier, long ttlSecs) {
         return externalCacheVersionKey -> obtainCacheVersion(clientSupplier, externalCacheVersionKey, ttlSecs);
     }
 
     public static Response<String> pipelinePutOperationForPolicy(Pipeline pipeline,
                                                                  String externalKey,
                                                                  PutPolicy policy,
                                                                  byte[] valueBytes,
                                                                  int defaultTtl) {
         if (policy == PutPolicy.ADD_ONLY) {
             return pipeline.set(externalKey.getBytes(), valueBytes, NX, EX, defaultTtl);
         } else if (policy == PutPolicy.PUT_ALWAYS) {
             return pipeline.setex(externalKey.getBytes(), defaultTtl, valueBytes);
         } else if (policy == PutPolicy.REPLACE_ONLY) {
             return pipeline.set(externalKey.getBytes(), valueBytes, XX, EX, defaultTtl);
         } else {
             throw new IllegalArgumentException("Unknown put policy: " + policy);
         }
     }
 }