/*
 * Copyright 2014-2019 the original author or authors.
 *
 * 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.
 */

package de.codecentric.boot.admin.server.web.client;

import java.net.URI;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.actuate.endpoint.http.ActuatorMediaType;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.ClientRequest;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.util.UriComponents;
import org.springframework.web.util.UriComponentsBuilder;
import reactor.core.publisher.Mono;

import de.codecentric.boot.admin.server.domain.values.Endpoint;
import de.codecentric.boot.admin.server.web.client.exception.ResolveEndpointException;

import static de.codecentric.boot.admin.server.utils.MediaType.ACTUATOR_V1_MEDIATYPE;
import static java.util.Arrays.asList;
import static java.util.Collections.singletonList;
import static org.springframework.http.MediaType.APPLICATION_JSON;

public final class InstanceExchangeFilterFunctions {

	private static final Logger log = LoggerFactory.getLogger(InstanceExchangeFilterFunctions.class);

	public static final String ATTRIBUTE_ENDPOINT = "endpointId";

	@SuppressWarnings("deprecation") // We need to support Spring Boot 1.x apps...
	private static final List<MediaType> DEFAULT_ACCEPT_MEDIATYPES = asList(
			MediaType.parseMediaType(ActuatorMediaType.V2_JSON), MediaType.parseMediaType(ActuatorMediaType.V1_JSON),
			MediaType.APPLICATION_JSON);

	private static final List<MediaType> DEFAULT_LOGFILE_ACCEPT_MEDIATYPES = singletonList(MediaType.TEXT_PLAIN);

	private InstanceExchangeFilterFunctions() {
	}

	public static InstanceExchangeFilterFunction addHeaders(HttpHeadersProvider httpHeadersProvider) {
		return (instance, request, next) -> {
			request = ClientRequest.from(request)
					.headers((headers) -> headers.addAll(httpHeadersProvider.getHeaders(instance))).build();
			return next.exchange(request);
		};
	}

	public static InstanceExchangeFilterFunction rewriteEndpointUrl() {
		return (instance, request, next) -> {
			if (request.url().isAbsolute()) {
				log.trace("Absolute URL '{}' for instance {} not rewritten", request.url(), instance.getId());
				if (request.url().toString().equals(instance.getRegistration().getManagementUrl())) {
					request = ClientRequest.from(request).attribute(ATTRIBUTE_ENDPOINT, Endpoint.ACTUATOR_INDEX)
							.build();
				}
				return next.exchange(request);
			}

			UriComponents requestUrl = UriComponentsBuilder.fromUri(request.url()).build();
			if (requestUrl.getPathSegments().isEmpty()) {
				return Mono.error(new ResolveEndpointException("No endpoint specified"));
			}

			String endpointId = requestUrl.getPathSegments().get(0);
			Optional<Endpoint> endpoint = instance.getEndpoints().get(endpointId);

			if (!endpoint.isPresent()) {
				return Mono.error(new ResolveEndpointException("Endpoint '" + endpointId + "' not found"));
			}

			URI rewrittenUrl = rewriteUrl(requestUrl, endpoint.get().getUrl());
			log.trace("URL '{}' for Endpoint {} of instance {} rewritten to {}", requestUrl, endpoint.get().getId(),
					instance.getId(), rewrittenUrl);
			request = ClientRequest.from(request).attribute(ATTRIBUTE_ENDPOINT, endpoint.get().getId())
					.url(rewrittenUrl).build();
			return next.exchange(request);
		};
	}

	private static URI rewriteUrl(UriComponents oldUrl, String targetUrl) {
		String[] newPathSegments = oldUrl.getPathSegments().subList(1, oldUrl.getPathSegments().size())
				.toArray(new String[] {});
		return UriComponentsBuilder.fromUriString(targetUrl).pathSegment(newPathSegments).query(oldUrl.getQuery())
				.build(true).toUri();
	}

	public static InstanceExchangeFilterFunction convertLegacyEndpoints(List<LegacyEndpointConverter> converters) {
		return (instance, request, next) -> {
			Mono<ClientResponse> clientResponse = next.exchange(request);

			Optional<Object> endpoint = request.attribute(ATTRIBUTE_ENDPOINT);
			if (!endpoint.isPresent()) {
				return clientResponse;
			}

			for (LegacyEndpointConverter converter : converters) {
				if (converter.canConvert(endpoint.get())) {
					return clientResponse.map((response) -> {
						if (isLegacyResponse(response)) {
							return convertLegacyResponse(converter, response);
						}
						return response;
					});
				}
			}
			return clientResponse;
		};
	}

	private static Boolean isLegacyResponse(ClientResponse response) {
		return response.headers().contentType()
				.map((t) -> ACTUATOR_V1_MEDIATYPE.isCompatibleWith(t) || APPLICATION_JSON.isCompatibleWith(t))
				.orElse(false);
	}

	private static ClientResponse convertLegacyResponse(LegacyEndpointConverter converter, ClientResponse response) {
		return ClientResponse.from(response).headers((headers) -> {
			headers.replace(HttpHeaders.CONTENT_TYPE, singletonList(ActuatorMediaType.V2_JSON));
			headers.remove(HttpHeaders.CONTENT_LENGTH);
		}).body(response.bodyToFlux(DataBuffer.class).transform(converter::convert)).build();
	}

	public static InstanceExchangeFilterFunction setDefaultAcceptHeader() {
		return (instance, request, next) -> {
			if (request.headers().getAccept().isEmpty()) {
				Boolean isRequestForLogfile = request.attribute(ATTRIBUTE_ENDPOINT).map(Endpoint.LOGFILE::equals)
						.orElse(false);
				List<MediaType> acceptedHeaders = isRequestForLogfile ? DEFAULT_LOGFILE_ACCEPT_MEDIATYPES
						: DEFAULT_ACCEPT_MEDIATYPES;
				request = ClientRequest.from(request).headers((headers) -> headers.setAccept(acceptedHeaders)).build();
			}
			return next.exchange(request);
		};
	}

	public static InstanceExchangeFilterFunction retry(int defaultRetries, Map<String, Integer> retriesPerEndpoint) {
		return (instance, request, next) -> {
			int retries = 0;
			if (!request.method().equals(HttpMethod.DELETE) && !request.method().equals(HttpMethod.PATCH)
					&& !request.method().equals(HttpMethod.POST) && !request.method().equals(HttpMethod.PUT)) {
				retries = request.attribute(ATTRIBUTE_ENDPOINT).map(retriesPerEndpoint::get).orElse(defaultRetries);
			}
			return next.exchange(request).retry(retries);
		};
	}

	public static InstanceExchangeFilterFunction timeout(Duration defaultTimeout,
			Map<String, Duration> timeoutPerEndpoint) {
		return (instance, request, next) -> {
			Duration timeout = request.attribute(ATTRIBUTE_ENDPOINT).map(timeoutPerEndpoint::get)
					.orElse(defaultTimeout);
			return next.exchange(request).timeout(timeout);
		};
	}

	// Accept header is broken on /logfile. We need to add "*/*" for old clients
	// see https://github.com/spring-projects/spring-boot/issues/16188
	public static InstanceExchangeFilterFunction logfileAcceptWorkaround() {
		return (instance, request, next) -> {
			if (request.attribute(ATTRIBUTE_ENDPOINT).map(Endpoint.LOGFILE::equals).orElse(false)) {
				List<MediaType> newAcceptHeaders = Stream
						.concat(request.headers().getAccept().stream(), Stream.of(MediaType.ALL))
						.collect(Collectors.toList());
				request = ClientRequest.from(request).headers((h) -> h.setAccept(newAcceptHeaders)).build();
			}
			return next.exchange(request);
		};
	}

}
