/*
    * #%L
    * wcm.io
    * %%
    * Copyright (C) 2014 wcm.io
    * %%
    * Licensed under the Apache License, Version 2.0 (the "License");
    * you may not use this file except in compliance with the License.
    * You may obtain a copy of the License at
   *
   *      http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   * #L%
   */
  package io.wcm.caravan.pipeline.impl.operators;
  
  import java.util.ArrayList;
  import java.util.List;
  import java.util.Map;
  import java.util.SortedSet;
  import java.util.TreeSet;
  import java.util.concurrent.TimeUnit;
  
  import org.apache.commons.lang3.StringUtils;
  import org.apache.commons.lang3.math.NumberUtils;
  import org.apache.http.HttpStatus;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.codahale.metrics.Counter;
  import com.codahale.metrics.MetricRegistry;
  import com.codahale.metrics.Timer;
  import com.fasterxml.jackson.databind.JsonNode;
  import com.fasterxml.jackson.databind.node.ObjectNode;
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.ImmutableMap;
  
  import io.wcm.caravan.commons.metrics.rx.TimerMetricsOperator;
  import io.wcm.caravan.io.http.request.CaravanHttpRequest;
  import io.wcm.caravan.pipeline.JsonPipelineInputException;
  import io.wcm.caravan.pipeline.JsonPipelineOutput;
  import io.wcm.caravan.pipeline.cache.CacheDateUtils;
  import io.wcm.caravan.pipeline.cache.CachePersistencyOptions;
  import io.wcm.caravan.pipeline.cache.CacheStrategy;
  import io.wcm.caravan.pipeline.cache.spi.CacheAdapter;
  import io.wcm.caravan.pipeline.impl.JacksonFunctions;
  import io.wcm.caravan.pipeline.impl.JsonPipelineContextImpl;
  import io.wcm.caravan.pipeline.impl.JsonPipelineOutputImpl;
  import rx.Observable;
  import rx.Observable.Transformer;
  import rx.Observer;
  import rx.Subscriber;
  import rx.exceptions.Exceptions;
  
  /**
   * a rather complicated transformer that implements the pipelines caching capabilities
   */
  public class CachePointTransformer implements Transformer<JsonPipelineOutput, JsonPipelineOutput> {
  
    private static final Logger log = LoggerFactory.getLogger(CachePointTransformer.class);
  
    private JsonPipelineContextImpl context;
    private final List<CaravanHttpRequest> requests;
    private final String descriptor;
    private final CacheStrategy strategy;
    private final String correlationId;
  
    /**
     * @param context a context of the actual JSON pipeline
     * @param requests the outgoing REST request(s) used to obtain the JSON data to be cached
     * @param descriptor the unique id of the pipeline (to build a cache key)
     * @param strategy the CacheStrategy to get storage time and refresh interval
     */
    public CachePointTransformer(JsonPipelineContextImpl context, List<CaravanHttpRequest> requests, String descriptor, CacheStrategy strategy) {
      super();
      this.context = context;
      this.requests = requests;
      this.descriptor = descriptor;
      this.strategy = strategy;
      StringBuffer sb = new StringBuffer();
      for (CaravanHttpRequest request : requests) {
        if (sb.length() == 0) {
          sb.append(request.getCorrelationId());
        }
        else {
          sb.append(",").append(request.getCorrelationId());
        }
      }
      this.correlationId = sb.toString();
    }
  
    private static SortedSet<String> getSourceServiceIds(List<CaravanHttpRequest> requests) {
      SortedSet<String> sourceServiceIds = new TreeSet<String>();
      for (CaravanHttpRequest request : requests) {
       sourceServiceIds.add(request.getServiceId());
     }
     return sourceServiceIds;
   }
 
   private String getSourceServicePrefix() {
     return StringUtils.join(getSourceServiceIds(requests), '+');
   }
 
   @Override
   public Observable<JsonPipelineOutput> call(Observable<JsonPipelineOutput> output) {
 
     // the code within the lambda passed to Observable#create will be executed when subscribe is called on the "cachedSource" observable
     Observable<JsonPipelineOutput> cachedSource = Observable.create((subscriber) -> {
 
       // construct a unique cache key from the pipeline's descriptor
       String sourceServicePrefix = getSourceServicePrefix();
       CacheAdapter cacheAdapter = context.getCacheAdapter();
       final String cacheKey = sourceServicePrefix + ":" + descriptor;
 
       // the caching strategy determines if the storage time should be extended for cache hits(i.e. Time-to-Idle behaviour)
       CachePersistencyOptions options = strategy.getCachePersistencyOptions(requests);
 
       // try to asynchronously(!) fetch the response from the cache
       Observable<String> cachedJsonString = cacheAdapter.get(cacheKey, options);
 
       // create service specific metrics
       MetricRegistry metricRegistry = context.getMetricRegistry();
       Timer timer = metricRegistry.timer(MetricRegistry.name(getClass(), sourceServicePrefix, "latency", "get"));
 
       CacheMetrics cacheMetrics = new CacheMetrics(metricRegistry, sourceServicePrefix);
 
       // CacheResponseObserver will decide what to do when the response is ready (or could not be retrieved from cache)
       cachedJsonString
       .lift(new TimerMetricsOperator<String>(timer))
       .subscribe(new CacheResponseObserver(cacheKey, output, subscriber, cacheMetrics));
     });
 
     return cachedSource;
   }
 
   private final class CacheMetrics {
 
     private final Counter hitsCounter;
     private final Counter missesCounter;
     private final Counter stalesCounter;
     private final Counter fallbacksCounter;
     private final Counter errorsCounter;
 
     private CacheMetrics(MetricRegistry metricRegistry, String metricServicePrefix) {
       hitsCounter = metricRegistry.counter(MetricRegistry.name(CachePointTransformer.class, metricServicePrefix, "hits"));
       missesCounter = metricRegistry.counter(MetricRegistry.name(CachePointTransformer.class, metricServicePrefix, "misses"));
       stalesCounter = metricRegistry.counter(MetricRegistry.name(CachePointTransformer.class, metricServicePrefix, "stales"));
       fallbacksCounter = metricRegistry.counter(MetricRegistry.name(CachePointTransformer.class, metricServicePrefix, "fallbacks"));
       errorsCounter = metricRegistry.counter(MetricRegistry.name(CachePointTransformer.class, metricServicePrefix, "errors"));
     }
   }
 
   /**
    * An observer that is subscribed to the {@link Observable} returned by
    * {@link CacheAdapter#get(String, CachePersistencyOptions)} , and is responsible for
    * <ul>
    * <li>unwrapping the JSON content from the caching envelope if it was successfully retrieved from cache</li>
    * <li>forwarding the unwrapped response to the subscriber given in the constructor</li>
    * <li>fetch the response from the Pipeline's dataSource if it couldn't be retrieved from cache</li>
    * <li>store the fetched responses to couchbase (wrapped in an envelope with metadata</li>
    * </ul>
    */
   public final class CacheResponseObserver implements Observer<String> {
 
     private final String cacheKey;
     private final Observable<JsonPipelineOutput> originalSource;
     private final Subscriber<? super JsonPipelineOutput> subscriber;
 
     private final CacheMetrics cacheMetrics;
 
     private boolean cacheHit;
 
     private CacheResponseObserver(String cacheKey, Observable<JsonPipelineOutput> originalSource, Subscriber<? super JsonPipelineOutput> subscriberToForwardTo,
         CacheMetrics cacheMetrics) {
       this.cacheKey = cacheKey;
       this.originalSource = originalSource;
       this.subscriber = subscriberToForwardTo;
 
       this.cacheMetrics = cacheMetrics;
     }
 
     @Override
     public void onNext(String cachedContent) {
 
       CacheEnvelope cacheEntry = CacheEnvelope.fromEnvelopeString(cachedContent, cacheKey);
       if (cacheEntry == null) {
         log.warn("CACHE ERROR for {} - the cached response could not be parsed,\n{}", this.cacheKey, correlationId);
 
         // increase the error counter for the source serviceID
         cacheMetrics.errorsCounter.inc();
 
         // ignore cache envelopes that can not be parsed
         return;
       }
       cacheHit = true;
 
       int responseAge = cacheEntry.getResponseAge();
       int refreshInterval = strategy.getCachePersistencyOptions(requests).getRefreshInterval();
 
       int expirySeconds = cacheEntry.getExpirySeconds();
 
       int maxAgeFromClient = getClientMaxAge();
 
       // check if the content from cache is fresh enough to serve it
       if (responseAge < refreshInterval && responseAge < maxAgeFromClient && expirySeconds > 0) {
 
         log.debug("CACHE HIT for {},\n{}", this.cacheKey, correlationId);
 
         // increase the hits counter for the source serviceID
         cacheMetrics.hitsCounter.inc();
 
         // the document could be retrieved, so forward it (parsed as a JsonNode) to the actual subscriber to the cachedSource
         serveCachedContent(cacheEntry, refreshInterval);
       }
       else {
         // this means the cached content is outdated - we better fetch the data from the backend
         String reason;
         if (responseAge >= refreshInterval) {
           reason = "it's " + responseAge + " seconds old and the cache strategy has a refresh interval of " + refreshInterval + " seconds.";
         }
         else if (responseAge >= maxAgeFromClient) {
           reason = "it's " + responseAge + " seconds old and the client requested a max-age of " + maxAgeFromClient + " seconds.";
         }
         else {
           reason = "it has expired " + (-expirySeconds) + " seconds ago, according to the original max-age header from the http-response";
         }
 
         log.debug("CACHE STALE - content for {} is available, but {},\n{}", cacheKey, reason, correlationId);
 
         fetchAndStore(new Subscriber<JsonPipelineOutput>() {
 
           @Override
           public void onNext(JsonPipelineOutput fetchedOutput) {
 
             // increase the stales counter for the source serviceID
             cacheMetrics.stalesCounter.inc();
 
             subscriber.onNext(fetchedOutput);
           }
 
           @Override
           public void onCompleted() {
             subscriber.onCompleted();
           }
 
           @Override
           public void onError(Throwable e) {
             Exceptions.throwIfFatal(e);
 
             // if the cached response was a an error as well (e.g. 404), then do not use it as a fallback.
             // instead  just forward the actual exception that occurred
             if (cacheEntry.getStatusCode() >= 400) {
               subscriber.onError(e);
               return;
             }
 
 
             if (e instanceof JsonPipelineInputException && ((JsonPipelineInputException)e).getStatusCode() == 404) {
               log.warn("CACHE FALLBACK - Using stale content from cache as a fallback after failing to fresh content for " + cacheKey + ",\n"
                   + correlationId + "\n" + e.getMessage());
             }
             else {
               log.warn("CACHE FALLBACK - Using stale content from cache as a fallback after failing to fresh content for " + cacheKey + ",\n"
                   + correlationId, e);
             }
 
             // increase the fallbacks counter for the source serviceID
             cacheMetrics.fallbacksCounter.inc();
 
             JsonPipelineOutputImpl pipelineOutput = new JsonPipelineOutputImpl(cacheEntry.getContentNode(), requests);
 
             // when fallback content is served from cache, it should not be cached by the client at all
             subscriber.onNext(pipelineOutput.withMaxAge(0));
             subscriber.onCompleted();
           }
         });
       }
     }
 
     private int getClientMaxAge() {
       int maxAgeFromClient = (int)TimeUnit.DAYS.toSeconds(365);
       for (String cacheControl : requests.get(0).getHeaders().get("Cache-Control")) {
         if (cacheControl.startsWith("max-age")) {
           int maxAge = NumberUtils.toInt(StringUtils.substringAfter(cacheControl, "="), maxAgeFromClient);
           if (maxAge > 0) {
             maxAgeFromClient = maxAge;
           }
         }
       }
       return maxAgeFromClient;
     }
 
     private void serveCachedContent(CacheEnvelope cacheEntry, int refreshInterval) {
 
       if (cacheEntry.getStatusCode() == HttpStatus.SC_NOT_FOUND) {
         // the cache entry is a 404 response that should be thrown as an exception to be handled by the subscriber
         String cachedInfoSuffix = " (Cached from " + cacheEntry.getSources() + " at " + cacheEntry.getGeneratedDate() + ")";
         subscriber.onError(new JsonPipelineInputException(HttpStatus.SC_NOT_FOUND, cacheEntry.getReasonString() + cachedInfoSuffix));
       }
       else {
         // make sure to set the max-age content-header just to the time the cached content will become stale
         int maxAge = refreshInterval - cacheEntry.getResponseAge();
         maxAge = Math.min(maxAge, cacheEntry.getExpirySeconds());
 
         subscriber.onNext(new JsonPipelineOutputImpl(cacheEntry.getContentNode(), requests).withMaxAge(maxAge));
         subscriber.onCompleted();
       }
     }
 
     @Override
     public void onCompleted() {
       if (!cacheHit) {
         // there was no emission, so the response has to be fetched from the service
         log.debug("CACHE MISS for {} fetching response from {} through pipeline,\n{}", cacheKey, getSourceServicePrefix(), correlationId);
 
         // increase the misses counter for the source serviceID
         cacheMetrics.missesCounter.inc();
 
         fetchAndStore(subscriber);
       }
     }
 
     @Override
     public void onError(Throwable e) {
       Exceptions.throwIfFatal(e);
 
       // also fall back to the actual service if the couchbase request failed
       log.warn("Failed to connect to couchbase server, falling back to direct connection to " + getSourceServicePrefix() + ",\n"
           + correlationId, e);
       fetchAndStore(subscriber);
     }
 
     private void fetchAndStore(Subscriber<? super JsonPipelineOutput> backendResponseSubscriber) {
 
       // fetch the output with a new subscription, which will also store the response in the cache when it is retrieved
       originalSource.subscribe(new Observer<JsonPipelineOutput>() {
 
         @Override
         public void onNext(JsonPipelineOutput fetchedModel) {
           CachePersistencyOptions options = strategy.getCachePersistencyOptions(requests);
 
           int contentMaxAge = options.getRefreshInterval();
           if (fetchedModel.getMaxAge() >= 0) {
             contentMaxAge = Math.min(contentMaxAge, fetchedModel.getMaxAge());
           }
 
           log.debug("CACHE PUT - response for {} has been fetched and will be put in the cache, max-age={} sec,\n{}", cacheKey, contentMaxAge,
               correlationId);
 
           CacheEnvelope cacheEntry = CacheEnvelope.from200Response(fetchedModel.getPayload(), contentMaxAge, requests,
               cacheKey, descriptor, context.getProperties());
           context.getCacheAdapter().put(cacheKey, cacheEntry.getEnvelopeString(), options);
 
           // everything else is just forwarding to the subscriber to the cachedSource
           backendResponseSubscriber.onNext(fetchedModel.withMaxAge(contentMaxAge));
         }
 
         @Override
         public void onCompleted() {
           backendResponseSubscriber.onCompleted();
         }
 
         @Override
         public void onError(Throwable e) {
           Exceptions.throwIfFatal(e);
 
           if (e instanceof JsonPipelineInputException) {
             if (((JsonPipelineInputException)e).getStatusCode() == HttpStatus.SC_NOT_FOUND) {
 
               int maxAgeFor404 = 60;
               CachePersistencyOptions options = CachePersistencyOptions.createTransient(maxAgeFor404);
               log.debug("CACHE PUT - 404 response for {} will be stored in the cache, max-age={} sec,\n{}",
                   descriptor, options.getRefreshInterval(), correlationId);
 
               CacheEnvelope cacheEntry = CacheEnvelope.from404Response(e.getMessage(), maxAgeFor404, requests, cacheKey, descriptor, context.getProperties());
               context.getCacheAdapter().put(cacheKey, cacheEntry.getEnvelopeString(), options);
             }
           }
           backendResponseSubscriber.onError(e);
         }
       });
     }
   }
 
   /**
    * Implements generation and parsing of the cache "envelope" document, that wraps the JSON output of the pipeline to
    * be able to store additional metadata in the cache
    */
   public static final class CacheEnvelope {
 
     private static final String CACHE_METADATA_PROPERTY = "metadata";
     private static final String CACHE_CONTENT_PROPERTY = "content";
 
     private final ObjectNode envelopeNode;
     private final ObjectNode metadataNode;
     private final JsonNode contentNode;
 
     private CacheEnvelope(ObjectNode envelopeNode) {
       this.envelopeNode = envelopeNode;
       metadataNode = (ObjectNode)envelopeNode.get(CACHE_METADATA_PROPERTY);
       contentNode = envelopeNode.get(CACHE_CONTENT_PROPERTY);
     }
 
     /**
      * Parse a JSON string that was obtained from the couchbase cache
      * @param jsonString JSON string
      * @param cacheKey Cache key
      * @return the CacheEntry - or null if the json String was not in the expected format
      */
     public static CacheEnvelope fromEnvelopeString(String jsonString, String cacheKey) {
       try {
         ObjectNode envelopeFromCache = JacksonFunctions.stringToObjectNode(jsonString);
         if (!envelopeFromCache.has(CACHE_METADATA_PROPERTY) || !envelopeFromCache.has(CACHE_CONTENT_PROPERTY)) {
           log.warn("Ignoring cached document {}, because it doesn't have the expected metadata/content envelope.", cacheKey);
           return null;
         }
 
         return new CacheEnvelope(envelopeFromCache);
       }
       catch (JsonPipelineInputException e) {
         log.warn("Failed parse cached JSON document from " + cacheKey, e);
         return null;
       }
     }
 
     /**
      * Create a new CacheEnvelope to store in the couchbase cache
      * @param contentNode Content node
      * @param maxAge how many seconds 404 responses in cache should stay valid
      * @param requests Requests
      * @param cacheKey Cache key
      * @param pipelineDescriptor Pipeline descriptor
      * @param contextProperties Context properties
      * @return the new CacheEnvelope instance
      */
     public static CacheEnvelope from200Response(JsonNode contentNode, int maxAge, List<CaravanHttpRequest> requests, String cacheKey,
         String pipelineDescriptor, Map<String, String> contextProperties) {
 
       ObjectNode envelope = createEnvelopeNode(contentNode, HttpStatus.SC_OK, maxAge, requests, cacheKey, pipelineDescriptor, null, contextProperties);
       return new CacheEnvelope(envelope);
     }
 
     /**
      * Create a new CacheEnvelope to store in the couchbase cache
      * @param reason Reason
      * @param maxAge how many seconds 404 responses in cache should stay valid
      * @param requests Requests
      * @param cacheKey Cache key
      * @param pipelineDescriptor Pipeline descriptor
      * @param contextProperties Context properties
      * @return the new CacheEnvelope instance
      */
     public static CacheEnvelope from404Response(String reason, int maxAge, List<CaravanHttpRequest> requests, String cacheKey,
         String pipelineDescriptor, Map<String, String> contextProperties) {
 
       JsonNode contentNode = JacksonFunctions.emptyObject();
       int statusCode = HttpStatus.SC_NOT_FOUND;
 
       ObjectNode envelope = createEnvelopeNode(contentNode, statusCode, maxAge, requests, cacheKey, pipelineDescriptor, reason, contextProperties);
       return new CacheEnvelope(envelope);
     }
 
     static CacheEnvelope fromContentString(String contentJson, int age) {
       ObjectNode envelopeNode = createEnvelopeNode(JacksonFunctions.stringToObjectNode(contentJson), 200, 0, ImmutableList.of(),
           "Cache-Key", "Descriptor", null, ImmutableMap.of());
 
       CacheEnvelope envelope = new CacheEnvelope(envelopeNode);
 
       envelope.getMetadataNode().put("generated", CacheDateUtils.formatRelativeTime(-age));
 
       return envelope;
     }
 
 
     private static ObjectNode createEnvelopeNode(JsonNode contentNode, int statusCode, int maxAge, List<CaravanHttpRequest> requests,
         String cacheKey, String pipelineDescriptor, String reason, Map<String, String> contextProperties) {
 
       ObjectNode envelope = JacksonFunctions.emptyObject();
       ObjectNode metadata = envelope.putObject(CACHE_METADATA_PROPERTY);
 
       metadata.put("cacheKey", cacheKey);
       metadata.set("sources", JacksonFunctions.pojoToNode(getSourceServiceIds(requests)));
       metadata.put("pipeline", pipelineDescriptor);
       metadata.put("generated", CacheDateUtils.formatCurrentTime());
       if (maxAge > 0) {
         metadata.put("expires", CacheDateUtils.formatRelativeTime(maxAge));
       }
       metadata.put("statusCode", statusCode);
 
       List<String> sourcePaths = new ArrayList<String>();
       for (CaravanHttpRequest req : requests) {
         sourcePaths.add(StringUtils.substringBefore(req.getUrl(), "?"));
       }
       metadata.set("sourcePaths", JacksonFunctions.pojoToNode(sourcePaths));
 
 
       if (StringUtils.isNotBlank(reason)) {
         metadata.put("reason", reason);
       }
       metadata.set("contextProperties", JacksonFunctions.pojoToNode(contextProperties));
       envelope.set(CACHE_CONTENT_PROPERTY, contentNode);
 
       return envelope;
     }
 
 
     /**
      * @return the full envelope (as JSON string) to be stored in the cache
      */
     public String getEnvelopeString() {
       return JacksonFunctions.nodeToString(envelopeNode);
     }
 
     JsonNode getContentNode() {
       return contentNode;
     }
 
     ObjectNode getMetadataNode() {
       return metadataNode;
     }
 
     int getStatusCode() {
       return metadataNode.at("/statusCode").asInt(HttpStatus.SC_OK);
     }
 
     String getReasonString() {
       return metadataNode.at("/reason").asText("Not Found");
     }
 
     String getSources() {
       return metadataNode.at("/sources").toString();
     }
 
     int getResponseAge() {
       return CacheDateUtils.getSecondsSince(getGeneratedDate());
     }
 
     String getGeneratedDate() {
       return metadataNode.at("/generated").asText();
     }
 
     int getExpirySeconds() {
       if (!metadataNode.has("expires")) {
         return (int)TimeUnit.DAYS.toSeconds(365);
       }
       String expiryDate = metadataNode.at("/expires").asText();
       return CacheDateUtils.getSecondsUntil(expiryDate);
     }
 
     /**
      * @param newDate Generated date
      */
     public void setGeneratedDate(String newDate) {
       metadataNode.put("generated", newDate);
     }
 
     /**
      * @param newDate Expires date
      */
     public void setExpiresDate(String newDate) {
       metadataNode.put("expires", newDate);
     }
   }
 }