/*
 * ====================================================================
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of the Apache Software Foundation.  For more
 * information on the Apache Software Foundation, please see
 * <http://www.apache.org/>.
 *
 */
package org.apache.hc.client5.http.impl.cache;

import java.net.URI;
import java.time.Instant;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import org.apache.hc.client5.http.cache.HttpAsyncCacheStorage;
import org.apache.hc.client5.http.cache.HttpCacheCASOperation;
import org.apache.hc.client5.http.cache.HttpCacheEntry;
import org.apache.hc.client5.http.cache.HttpCacheEntryFactory;
import org.apache.hc.client5.http.cache.HttpCacheUpdateException;
import org.apache.hc.client5.http.cache.Resource;
import org.apache.hc.client5.http.cache.ResourceFactory;
import org.apache.hc.client5.http.cache.ResourceIOException;
import org.apache.hc.client5.http.impl.Operations;
import org.apache.hc.client5.http.validator.ETag;
import org.apache.hc.client5.http.validator.ValidatorType;
import org.apache.hc.core5.concurrent.CallbackContribution;
import org.apache.hc.core5.concurrent.Cancellable;
import org.apache.hc.core5.concurrent.ComplexCancellable;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.http.HttpHeaders;
import org.apache.hc.core5.http.HttpHost;
import org.apache.hc.core5.http.HttpRequest;
import org.apache.hc.core5.http.HttpResponse;
import org.apache.hc.core5.http.HttpStatus;
import org.apache.hc.core5.http.Method;
import org.apache.hc.core5.http.message.RequestLine;
import org.apache.hc.core5.http.message.StatusLine;
import org.apache.hc.core5.util.ByteArrayBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class BasicHttpAsyncCache implements HttpAsyncCache {

    private static final Logger LOG = LoggerFactory.getLogger(BasicHttpAsyncCache.class);

    private final ResourceFactory resourceFactory;
    private final HttpCacheEntryFactory cacheEntryFactory;
    private final CacheKeyGenerator cacheKeyGenerator;
    private final HttpAsyncCacheStorage storage;

    public BasicHttpAsyncCache(
            final ResourceFactory resourceFactory,
            final HttpCacheEntryFactory cacheEntryFactory,
            final HttpAsyncCacheStorage storage,
            final CacheKeyGenerator cacheKeyGenerator) {
        this.resourceFactory = resourceFactory;
        this.cacheEntryFactory = cacheEntryFactory;
        this.cacheKeyGenerator = cacheKeyGenerator;
        this.storage = storage;
    }

    public BasicHttpAsyncCache(
            final ResourceFactory resourceFactory,
            final HttpAsyncCacheStorage storage,
            final CacheKeyGenerator cacheKeyGenerator) {
        this(resourceFactory, HttpCacheEntryFactory.INSTANCE, storage, cacheKeyGenerator);
    }

    public BasicHttpAsyncCache(final ResourceFactory resourceFactory, final HttpAsyncCacheStorage storage) {
        this( resourceFactory, storage, CacheKeyGenerator.INSTANCE);
    }

    @Override
    public Cancellable match(final HttpHost host, final HttpRequest request, final FutureCallback<CacheMatch> callback) {
        final String rootKey = cacheKeyGenerator.generateKey(host, request);
        if (LOG.isDebugEnabled()) {
            LOG.debug("Get cache entry: {}", rootKey);
        }
        final ComplexCancellable complexCancellable = new ComplexCancellable();
        complexCancellable.setDependency(storage.getEntry(rootKey, new FutureCallback<HttpCacheEntry>() {

            @Override
            public void completed(final HttpCacheEntry root) {
                if (root != null) {
                    if (root.hasVariants()) {
                        final List<String> variantNames = CacheKeyGenerator.variantNames(root);
                        final String variantKey = cacheKeyGenerator.generateVariantKey(request, variantNames);
                        if (root.getVariants().contains(variantKey)) {
                            final String cacheKey = variantKey + rootKey;
                            if (LOG.isDebugEnabled()) {
                                LOG.debug("Get cache variant entry: {}", cacheKey);
                            }
                            complexCancellable.setDependency(storage.getEntry(
                                    cacheKey,
                                    new FutureCallback<HttpCacheEntry>() {

                                        @Override
                                        public void completed(final HttpCacheEntry entry) {
                                            callback.completed(new CacheMatch(
                                                    entry != null ? new CacheHit(rootKey, cacheKey, entry) : null,
                                                    new CacheHit(rootKey, root)));
                                        }

                                        @Override
                                        public void failed(final Exception ex) {
                                            if (ex instanceof ResourceIOException) {
                                                if (LOG.isWarnEnabled()) {
                                                    LOG.warn("I/O error retrieving cache entry with key {}", cacheKey);
                                                }
                                                callback.completed(null);
                                            } else {
                                                callback.failed(ex);
                                            }
                                        }

                                        @Override
                                        public void cancelled() {
                                            callback.cancelled();
                                        }

                                    }));
                            return;
                        } else {
                            callback.completed(new CacheMatch(null, new CacheHit(rootKey, root)));
                        }
                    } else {
                        callback.completed(new CacheMatch(new CacheHit(rootKey, root), null));
                    }
                } else {
                    callback.completed(null);
                }
            }

            @Override
            public void failed(final Exception ex) {
                if (ex instanceof ResourceIOException) {
                    if (LOG.isWarnEnabled()) {
                        LOG.warn("I/O error retrieving cache entry with key {}", rootKey);
                    }
                    callback.completed(null);
                } else {
                    callback.failed(ex);
                }
            }

            @Override
            public void cancelled() {
                callback.cancelled();
            }

        }));
        return complexCancellable;
    }

    @Override
    public Cancellable getVariants(
            final CacheHit hit, final FutureCallback<Collection<CacheHit>> callback) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Get variant cache entries: {}", hit.rootKey);
        }
        final ComplexCancellable complexCancellable = new ComplexCancellable();
        final HttpCacheEntry root = hit.entry;
        final String rootKey = hit.rootKey;
        if (root != null && root.hasVariants()) {
            final List<String> variantCacheKeys = root.getVariants().stream()
                    .map(e -> e + rootKey)
                    .collect(Collectors.toList());
            complexCancellable.setDependency(storage.getEntries(
                    variantCacheKeys,
                    new FutureCallback<Map<String, HttpCacheEntry>>() {

                        @Override
                        public void completed(final Map<String, HttpCacheEntry> resultMap) {
                            final List<CacheHit> cacheHits = resultMap.entrySet().stream()
                                    .map(e -> new CacheHit(hit.rootKey, e.getKey(), e.getValue()))
                                    .collect(Collectors.toList());
                            callback.completed(cacheHits);
                        }

                        @Override
                        public void failed(final Exception ex) {
                            if (ex instanceof ResourceIOException) {
                                if (LOG.isWarnEnabled()) {
                                    LOG.warn("I/O error retrieving cache entry with keys {}", variantCacheKeys);
                                }
                                callback.completed(Collections.emptyList());
                            } else {
                                callback.failed(ex);
                            }
                        }

                        @Override
                        public void cancelled() {
                            callback.cancelled();
                        }

                    }));
        } else {
            callback.completed(Collections.emptyList());
        }
        return complexCancellable;
    }

    Cancellable storeInternal(final String cacheKey, final HttpCacheEntry entry, final FutureCallback<Boolean> callback) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Store entry in cache: {}", cacheKey);
        }

        return storage.putEntry(cacheKey, entry, new FutureCallback<Boolean>() {

            @Override
            public void completed(final Boolean result) {
                if (callback != null) {
                    callback.completed(result);
                }
            }

            @Override
            public void failed(final Exception ex) {
                if (ex instanceof ResourceIOException) {
                    if (LOG.isWarnEnabled()) {
                        LOG.warn("I/O error storing cache entry with key {}", cacheKey);
                    }
                    if (callback != null) {
                        callback.completed(false);
                    }
                } else {
                    if (callback != null) {
                        callback.failed(ex);
                    }
                }
            }

            @Override
            public void cancelled() {
                if (callback != null) {
                    callback.cancelled();
                }
            }

        });
    }

    Cancellable updateInternal(final String cacheKey, final HttpCacheCASOperation casOperation, final FutureCallback<Boolean> callback) {
        return storage.updateEntry(cacheKey, casOperation, new FutureCallback<Boolean>() {

            @Override
            public void completed(final Boolean result) {
                if (callback != null) {
                    callback.completed(result);
                }
            }

            @Override
            public void failed(final Exception ex) {
                if (ex instanceof HttpCacheUpdateException) {
                    if (LOG.isWarnEnabled()) {
                        LOG.warn("Cannot update cache entry with key {}", cacheKey);
                    }
                    if (callback != null) {
                        callback.completed(false);
                    }
                } else if (ex instanceof ResourceIOException) {
                    if (LOG.isWarnEnabled()) {
                        LOG.warn("I/O error updating cache entry with key {}", cacheKey);
                    }
                    if (callback != null) {
                        callback.completed(false);
                    }
                } else {
                    if (callback != null) {
                        callback.failed(ex);
                    }
                }
            }

            @Override
            public void cancelled() {
                if (callback != null) {
                    callback.cancelled();
                }
            }

        });
    }

    private void removeInternal(final String cacheKey) {
        storage.removeEntry(cacheKey, new FutureCallback<Boolean>() {

            @Override
            public void completed(final Boolean result) {
            }

            @Override
            public void failed(final Exception ex) {
                if (LOG.isWarnEnabled()) {
                    if (ex instanceof ResourceIOException) {
                        LOG.warn("I/O error removing cache entry with key {}", cacheKey);
                    } else {
                        LOG.warn("Unexpected error removing cache entry with key {}", cacheKey, ex);
                    }
                }
            }

            @Override
            public void cancelled() {
            }

        });
    }

    Cancellable store(
            final String rootKey,
            final String variantKey,
            final HttpCacheEntry entry,
            final FutureCallback<CacheHit> callback) {
        if (variantKey == null) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Store entry in cache: {}", rootKey);
            }
            return storeInternal(rootKey, entry, new CallbackContribution<Boolean>(callback) {

                @Override
                public void completed(final Boolean result) {
                    callback.completed(new CacheHit(rootKey, entry));
                }

            });
        } else {
            final String variantCacheKey = variantKey + rootKey;

            if (LOG.isDebugEnabled()) {
                LOG.debug("Store variant entry in cache: {}", variantCacheKey);
            }

            return storeInternal(variantCacheKey, entry, new CallbackContribution<Boolean>(callback) {

                @Override
                public void completed(final Boolean result) {
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("Update root entry: {}", rootKey);
                    }

                    updateInternal(rootKey,
                            existing -> {
                                final Set<String> variantMap = existing != null ? new HashSet<>(existing.getVariants()) : new HashSet<>();
                                variantMap.add(variantKey);
                                return cacheEntryFactory.createRoot(entry, variantMap);
                            },
                            new CallbackContribution<Boolean>(callback) {

                                @Override
                                public void completed(final Boolean result) {
                                    callback.completed(new CacheHit(rootKey, variantCacheKey, entry));
                                }

                            });
                }

            });
        }
    }

    @Override
    public Cancellable store(
            final HttpHost host,
            final HttpRequest request,
            final HttpResponse originResponse,
            final ByteArrayBuffer content,
            final Instant requestSent,
            final Instant responseReceived,
            final FutureCallback<CacheHit> callback) {
        final String rootKey = cacheKeyGenerator.generateKey(host, request);
        if (LOG.isDebugEnabled()) {
            LOG.debug("Create cache entry: {}", rootKey);
        }
        final Resource resource;
        try {
            final ETag eTag = ETag.get(originResponse);
            resource = content != null ? resourceFactory.generate(
                    rootKey,
                    eTag != null && eTag.getType() == ValidatorType.STRONG ? eTag.getValue() : null,
                    content.array(), 0, content.length()) : null;
        } catch (final ResourceIOException ex) {
            if (LOG.isWarnEnabled()) {
                LOG.warn("I/O error creating cache entry with key {}", rootKey);
            }
            final HttpCacheEntry backup = cacheEntryFactory.create(
                    requestSent,
                    responseReceived,
                    host,
                    request,
                    originResponse,
                    content != null ? HeapResourceFactory.INSTANCE.generate(null, content.array(), 0, content.length()) : null);
            callback.completed(new CacheHit(rootKey, backup));
            return Operations.nonCancellable();
        }
        final HttpCacheEntry entry = cacheEntryFactory.create(requestSent, responseReceived, host, request, originResponse, resource);
        final String variantKey = cacheKeyGenerator.generateVariantKey(request, entry);
        return store(rootKey,variantKey, entry, callback);
    }

    @Override
    public Cancellable update(
            final CacheHit stale,
            final HttpHost host,
            final HttpRequest request,
            final HttpResponse originResponse,
            final Instant requestSent,
            final Instant responseReceived,
            final FutureCallback<CacheHit> callback) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Update cache entry: {}", stale.getEntryKey());
        }
        final HttpCacheEntry updatedEntry = cacheEntryFactory.createUpdated(
                requestSent,
                responseReceived,
                host,
                request,
                originResponse,
                stale.entry);
        final String variantKey = cacheKeyGenerator.generateVariantKey(request, updatedEntry);
        return store(stale.rootKey, variantKey, updatedEntry, callback);
    }

    @Override
    public Cancellable storeFromNegotiated(
            final CacheHit negotiated,
            final HttpHost host,
            final HttpRequest request,
            final HttpResponse originResponse,
            final Instant requestSent,
            final Instant responseReceived,
            final FutureCallback<CacheHit> callback) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Update negotiated cache entry: {}", negotiated.getEntryKey());
        }
        final HttpCacheEntry updatedEntry = cacheEntryFactory.createUpdated(
                requestSent,
                responseReceived,
                host,
                request,
                originResponse,
                negotiated.entry);

        storeInternal(negotiated.getEntryKey(), updatedEntry, null);

        final String rootKey = cacheKeyGenerator.generateKey(host, request);
        final HttpCacheEntry copy = cacheEntryFactory.copy(updatedEntry);
        final String variantKey = cacheKeyGenerator.generateVariantKey(request, copy);
        return store(rootKey, variantKey, copy, callback);
    }

    private void evictAll(final HttpCacheEntry root, final String rootKey) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Evicting root cache entry {}", rootKey);
        }
        removeInternal(rootKey);
        if (root.hasVariants()) {
            for (final String variantKey : root.getVariants()) {
                final String variantEntryKey = variantKey + rootKey;
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Evicting variant cache entry {}", variantEntryKey);
                }
                removeInternal(variantEntryKey);
            }
        }
    }

    private Cancellable evict(final String rootKey) {
        return storage.getEntry(rootKey, new FutureCallback<HttpCacheEntry>() {

            @Override
            public void completed(final HttpCacheEntry root) {
                if (root != null) {
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("Evicting root cache entry {}", rootKey);
                    }
                    evictAll(root, rootKey);
                }
            }

            @Override
            public void failed(final Exception ex) {
            }

            @Override
            public void cancelled() {
            }

        });
    }

    private Cancellable evict(final String rootKey, final HttpResponse response) {
        return storage.getEntry(rootKey, new FutureCallback<HttpCacheEntry>() {

            @Override
            public void completed(final HttpCacheEntry root) {
                if (root != null) {
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("Evicting root cache entry {}", rootKey);
                    }
                    final ETag existingETag = root.getETag();
                    final ETag newETag = ETag.get(response);
                    if (existingETag != null && newETag != null &&
                            !ETag.strongCompare(existingETag, newETag) &&
                            !HttpCacheEntry.isNewer(root, response)) {
                        evictAll(root, rootKey);
                    }
                }
            }

            @Override
            public void failed(final Exception ex) {
            }

            @Override
            public void cancelled() {
            }

        });
    }

    @Override
    public Cancellable evictInvalidatedEntries(
            final HttpHost host, final HttpRequest request, final HttpResponse response, final FutureCallback<Boolean> callback) {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Flush cache entries invalidated by exchange: {}; {} -> {}", host, new RequestLine(request), new StatusLine(response));
        }
        final int status = response.getCode();
        if (status >= HttpStatus.SC_SUCCESS && status < HttpStatus.SC_CLIENT_ERROR &&
                !Method.isSafe(request.getMethod())) {
            final String rootKey = cacheKeyGenerator.generateKey(host, request);
            evict(rootKey);
            final URI requestUri = CacheKeyGenerator.normalize(CacheKeyGenerator.getRequestUri(host, request));
            if (requestUri != null) {
                final URI contentLocation = CacheSupport.getLocationURI(requestUri, response, HttpHeaders.CONTENT_LOCATION);
                if (contentLocation != null && CacheSupport.isSameOrigin(requestUri, contentLocation)) {
                    final String cacheKey = cacheKeyGenerator.generateKey(contentLocation);
                    evict(cacheKey, response);
                }
                final URI location = CacheSupport.getLocationURI(requestUri, response, HttpHeaders.LOCATION);
                if (location != null && CacheSupport.isSameOrigin(requestUri, location)) {
                    final String cacheKey = cacheKeyGenerator.generateKey(location);
                    evict(cacheKey, response);
                }
            }
        }
        callback.completed(Boolean.TRUE);
        return Operations.nonCancellable();
    }

}
