package com.atlassian.bitbucket.internal.mirroring.mirror.client;

import com.atlassian.bitbucket.Product;
import com.atlassian.bitbucket.auth.AuthenticationSystemException;
import com.atlassian.bitbucket.auth.CaptchaRequiredAuthenticationException;
import com.atlassian.bitbucket.i18n.I18nService;
import com.atlassian.bitbucket.i18n.KeyedMessage;
import com.atlassian.bitbucket.internal.mirroring.MirroringRequest;
import com.atlassian.bitbucket.internal.mirroring.mirror.*;
import com.atlassian.bitbucket.internal.mirroring.mirror.auth.DelegatedAuthenticationFailureException;
import com.atlassian.bitbucket.internal.mirroring.mirror.jwt.JwtSignedRequestFactory;
import com.atlassian.bitbucket.internal.mirroring.mirror.nav.MirroringNavBuilder;
import com.atlassian.bitbucket.internal.mirroring.mirror.nav.MirroringUrl;
import com.atlassian.bitbucket.internal.mirroring.mirror.rest.RestSimpleRsaSshKey;
import com.atlassian.bitbucket.internal.mirroring.mirror.rest.server.RestServerExternalRepository;
import com.atlassian.bitbucket.internal.mirroring.rest.RestAnalyticsSettings;
import com.atlassian.bitbucket.internal.mirroring.rest.RestMirroringRequest;
import com.atlassian.bitbucket.internal.mirroring.rest.auth.*;
import com.atlassian.bitbucket.internal.mirroring.ssh.encoding.PublicKeyEncodingHelper;
import com.atlassian.bitbucket.internal.mirroring.user.ApplicationUserWithPermissions;
import com.atlassian.bitbucket.json.JsonRenderer;
import com.atlassian.bitbucket.mirroring.mirror.AnalyticsSettings;
import com.atlassian.bitbucket.mirroring.mirror.UpstreamServer;
import com.atlassian.bitbucket.rest.RestErrorMessage;
import com.atlassian.bitbucket.rest.RestErrors;
import com.atlassian.bitbucket.rest.RestMapEntity;
import com.atlassian.bitbucket.server.ApplicationPropertiesService;
import com.atlassian.bitbucket.util.*;
import com.atlassian.httpclient.api.HttpClient;
import com.atlassian.httpclient.api.Request;
import com.atlassian.httpclient.api.Response;
import com.atlassian.httpclient.api.ResponseTransformation;
import com.atlassian.util.concurrent.Promise;
import com.atlassian.util.concurrent.Promises;
import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import org.apache.commons.lang3.StringUtils;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nonnull;
import javax.validation.ConstraintViolationException;
import javax.validation.Validator;
import java.io.IOException;
import java.security.PublicKey;
import java.util.*;
import java.util.function.BiFunction;
import java.util.stream.Stream;

import static com.atlassian.bitbucket.internal.mirroring.ConstraintViolationUtils.violationToString;
import static com.atlassian.bitbucket.util.MoreStreams.streamIterable;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.net.MediaType.JSON_UTF_8;
import static java.util.Optional.empty;
import static java.util.Optional.of;

public class DefaultServerUpstreamClient extends AbstractUpstreamClient implements ServerUpstreamClient {

    private static final Predicate<RestErrorMessage> ERROR_WITH_CREDENTIALS_CONTEXT =
            input -> input != null && "credentials".equals(input.getContext()) && StringUtils.isNotEmpty(input.getMessage());

    private static final Predicate<RestErrorMessage> ERROR_WITH_CAPTCHA_EXCEPTION =
            input -> input != null
                    && StringUtils.defaultString(input.getExceptionName()).toLowerCase(Locale.US).contains("captcha")
                    && StringUtils.isNotEmpty(input.getMessage());

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

    private final MirroringConfig config;
    private final PublicKeyEncodingHelper publicKeyEncodingHelper;
    private final ApplicationPropertiesService propertiesService;

    public DefaultServerUpstreamClient(MirroringConfig config, HttpClient httpClient, I18nService i18nService,
                                       JsonRenderer jsonRenderer, MirroringNavBuilder mirroringNavBuilder,
                                       PublicKeyEncodingHelper publicKeyEncodingHelper,
                                       ApplicationPropertiesService propertiesService,
                                       JwtSignedRequestFactory requestFactory, UpstreamServer upstream, Validator validator) {
        super(httpClient, i18nService, jsonRenderer, mirroringNavBuilder, requestFactory, upstream, validator);

        this.config = config;
        this.publicKeyEncodingHelper = publicKeyEncodingHelper;
        this.propertiesService = propertiesService;
    }

    @Nonnull
    @Override
    public Promise<ApplicationUserWithPermissions> authenticateForUser(@Nonnull String username, @Nonnull String password) {
        return authenticateForUser(createAuthRequest(checkNotNull(username, "username"), checkNotNull(password, "password")));
    }

    @Nonnull
    @Override
    public Promise<ApplicationUserWithPermissions> authenticateForUser(@Nonnull String username, @Nonnull PublicKey publicKey) {
        return authenticateForUser(createAuthRequest(checkNotNull(username, "username"), checkNotNull(publicKey, "publicKey")));
    }

    @Override
    @Nonnull
    public Promise<AnalyticsSettings> getAnalyticsSettings() {
        MirroringUrl analyticsUrl = mirroringNavBuilder.serverUpstream(upstream).analyticsSettings();

        return newAuthenticatedJsonRequest(analyticsUrl)
                .get()
                .transform(this.<RestAnalyticsSettings>standardResponseBuilder(analyticsUrl)
                        .notFound(this::throwRequestFailed)
                        .ok(response -> parseEntity(response, RestAnalyticsSettings.class)
                                .orElseThrow(requestFailedSupplier(response)))
                        .build())
                .map(this::transformAnalyticsSettings);
    }

    @Nonnull
    @Override
    public Promise<Void> deleteSshKeys() {
        MirroringUrl sshKeysUrl = mirroringNavBuilder.serverUpstream(upstream).sshKeys();

        return newAuthenticatedJsonRequest(sshKeysUrl)
                .delete()
                .transform(this.<Void>standardResponseBuilder(sshKeysUrl)
                        .notFound(ensureErrorResponseIsFromUpstreamAndThen(returnNull()))
                        .ok(returnNull())
                        .build());
    }

    @Nonnull
    public Promise<Boolean> isMirrorInstalled() {
        MirroringUrl mirrorUrl = mirroringNavBuilder.serverUpstream(upstream).mirrors(PageUtils.newRequest(0, 250));

        String serverId = checkNotNull(propertiesService.getServerId(), "serverId");

        // note that this request is not signed with a JWT token - the list of mirror servers is publicly accessible
        return newUnauthenticatedJsonRequest(mirrorUrl.getAbsolute())
                .get()
                .transform(this.<Boolean>standardResponseBuilder(mirrorUrl)
                        .notFound(this::throwRequestFailed)
                        .ok(response -> {
                            Page<RestMinimalMirrorServer> mirrors =
                                    parseEntityPage(response, RestMinimalMirrorServer.class)
                                            .orElseThrow(requestFailedSupplier(response));

                            return streamIterable(mirrors.getValues())
                                    .anyMatch(mirror -> serverId.equals(mirror.getId()));
                        })
                        .build());
    }

    @SuppressWarnings("unchecked")
    @Nonnull
    @Override
    public Stream<ExternalRepository> getRepositories() {
        return streamIterable(new PagedIterable<>(this::getRepoPage, config.getUpstreamRepositoryRequestPageSize()));
    }

    @Nonnull
    @Override
    public Promise<ExternalRepository> getRepository(@Nonnull String repoId) {
        return getRepositoryAs(repoId, RestServerExternalRepository.class);
    }

    @Nonnull
    @Override
    public Request.Builder newScmHttpRequest(@Nonnull String relativePath) {
        return newAuthenticatedRequest(mirroringNavBuilder.upstream(upstream).relativeToBase(relativePath));
    }

    @Nonnull
    @Override
    public Promise<Void> registerAsMirror(@Nonnull MirroringRequest request) {
        checkNotNull(request, "request");

        try {
            ValidationUtils.validate(validator, request);
        } catch (ConstraintViolationException e) {
            return Promises.rejected(
                    new MirrorRegistrationFailedException(i18nService.createKeyedMessage(
                            "bitbucket.mirroring.upstream.registration.invalid",
                            request, violationToString(e), e)));
        }

        String jsonRequest = jsonRenderer.render(new RestMirroringRequest(request), Collections.<String, Object>emptyMap());

        String registrationUrl = upstream.getBaseUrl();
        if (!registrationUrl.endsWith("/requests")) {
            registrationUrl = mirroringNavBuilder.serverUpstream(upstream).mirrorRequests().getAbsolute();
        }

        return newUnauthenticatedJsonRequest(registrationUrl)
                .setEntity(jsonRequest)
                .post()
                .fold(wrapAsOrRethrowIfSame(MirrorRegistrationFailedException.class,
                        "bitbucket.mirroring.upstream.registration.error", MirrorRegistrationFailedException::new),
                        response -> {
                            if (!(response.isOk() || response.isCreated() || response.isNoContent())) {
                                throw new MirrorRegistrationFailedException(
                                        i18nService.createKeyedMessage("bitbucket.mirroring.upstream.registration.failed",
                                                response.getStatusCode(), response.getStatusText()));
                            }
                            return (Void) null;
                        });
    }

    @Nonnull
    @Override
    public Promise<Void> uploadSshKey(@Nonnull String key) {
        checkNotNull(key, "key");
        String jsonKey = jsonRenderer.render(new RestSimpleRsaSshKey(key), Collections.<String, Object>emptyMap());

        return newAuthenticatedJsonRequest(mirroringNavBuilder.serverUpstream(upstream).sshKeys())
                .setEntity(jsonKey)
                .post()
                .fold(wrapAsOrRethrowIfSame(MirrorKeyUploadFailedException.class,
                        "bitbucket.mirroring.upstream.upload.ssh.error", MirrorKeyUploadFailedException::new),
                        response -> {
                            if (log.isTraceEnabled()) {
                                log.trace("The upstream's SSH keys endpoint responded with status code {} and body:\n{}",
                                        response.getStatusCode(), response.getEntity());
                            }
                            if (!response.isCreated()) {
                                log.debug("Failed to add SSH key to the upstream. Status code: {}", response.getStatusCode());
                                throw new MirrorKeyUploadFailedException(
                                        i18nService.createKeyedMessage("bitbucket.mirroring.upstream.upload.ssh.failed",
                                                response.getStatusCode(), response.getStatusText()));
                            }
                            return (Void) null;
                        });
    }

    @Override
    protected void ensureErrorResponseIsFromUpstream(Response response) {
        // if the error response is missing the JSON content type, it is not BBS we are talking to
        // (proxy error etc) so we can't trust that 'not found' means anything definitive
        try {
            if (isContentType(response, JSON_UTF_8.withoutParameters())) {
                return;
            }
        } catch (IllegalArgumentException e) {
            // fall through if content type malformed
        }
        throw throwUnexpectedNotFoundResponse(response);
    }

    private Promise<ApplicationUserWithPermissions> authenticateForUser(RestAuthenticationRequest authRequest) {
        checkNotNull(authRequest, "authRequest");

        try {
            MirroringUrl authenticateUrl = mirroringNavBuilder.serverUpstream(upstream).authenticate();
            if (log.isDebugEnabled()) {
                log.debug("Delegated authentication endpoint: {}", authenticateUrl.getAbsolute());
            }

            String jsonRequest = jsonRenderer.render(authRequest, Collections.<String, Object>emptyMap());

            return newAuthenticatedJsonRequest(authenticateUrl)
                    .setEntity(jsonRequest)
                    .post()
                    .map(this::parseAuthenticationResponse);
        } catch (Exception e) {
            return Promises.rejected(e);
        }
    }

    private RestAuthenticationRequest createAuthRequest(String username, PublicKey publicKey) {
        return new RestAuthenticationRequest(new RestSshCredentials(
                username, new RestPublicKey(publicKeyEncodingHelper.encodeBytesAsBase64(publicKey), publicKey.getAlgorithm())));
    }

    private RestAuthenticationRequest createAuthRequest(String username, String password) {
        return new RestAuthenticationRequest(new RestUsernamePasswordCredentials(username, password));
    }

    private RestErrorMessage findFirstNonEmptyError(RestErrors restErrors) {
        return restErrors == null ? null : Iterables.find(restErrors.getErrors(),
                errorMessage -> StringUtils.isNotEmpty(errorMessage.getMessage()), null);
    }

    private RestErrorMessage findUserCaptchaError(RestErrors restErrors) {
        return restErrors == null ? null : Iterables.find(restErrors.getErrors(), ERROR_WITH_CAPTCHA_EXCEPTION, null);
    }

    private RestErrorMessage findUserCredentialsError(RestErrors restErrors) {
        return restErrors == null ? null : Iterables.find(restErrors.getErrors(), ERROR_WITH_CREDENTIALS_CONTEXT, null);
    }

    @SuppressWarnings("unchecked")
    private Page<ExternalRepository> getRepoPage(@Nonnull PageRequest pageRequest) {
        MirroringUrl reposUrl = mirroringNavBuilder.serverUpstream(upstream).repositories(pageRequest);

        log.debug("Retrieving page [start={}, limit={}] of upstream repositories", pageRequest.getStart(), pageRequest.getLimit());
        Promise<Page<RestServerExternalRepository>> promise = newAuthenticatedJsonRequest(reposUrl)
                .get()
                .transform(toPageOfRepos(reposUrl));
        promise.done(page -> log.debug("Finished retrieving page [start={}, limit={}] of upstream repositories", pageRequest.getStart(), pageRequest.getLimit()));
        promise.fail(error -> log.debug("Failed retrieving page [start={}, limit={}] of upstream repositories", pageRequest.getStart(), pageRequest.getLimit()));
        return (Page) promise.claim();
    }

    private ApplicationUserWithPermissions parseAuthenticationResponse(Response response) {
        RestApplicationUserWithPermissions user;
        log.debug("The delegated authentication endpoint responded with status code {}", response.getStatusCode());

        if (response.isOk()) {
            user = parseEntity(response, RestApplicationUserWithPermissions.class).orElse(null);

            if (user == null) {
                log.warn("The delegated authentication endpoint gave an unrecognised response");
                throw new AuthenticationSystemException(
                        i18nService.createKeyedMessage("bitbucket.mirroring.authentication.delegation.failure.upstream"));
            }
        } else {
            RestErrors restErrors = parseEntity(response, RestErrors.class).orElse(null);

            if (log.isTraceEnabled() && restErrors != null) {
                traceErrorMessages(restErrors);
            }

            if (response.isUnauthorized()) {
                RestErrorMessage userCredentialsError = findUserCredentialsError(restErrors);
                if (userCredentialsError != null) {
                    log.debug("Delegated authentication failed because of invalid credentials", userCredentialsError);
                    throw new DelegatedAuthenticationFailureException(
                            i18nService.createKeyedMessage("bitbucket.mirroring.authentication.delegation.failed", userCredentialsError.getMessage()));
                }

                RestErrorMessage userCaptchaError = findUserCaptchaError(restErrors);
                if (userCaptchaError != null) {
                    log.debug("Delegated authentication failed because the user is CAPTCHA'd upstream");
                    throw new CaptchaRequiredAuthenticationException(
                            i18nService.createKeyedMessage("bitbucket.mirroring.authentication.delegation.failure.captcha.required",
                                    Product.NAME));
                }

                log.info("Delegated authentication failed because the upstream rejected our mirror credentials. " +
                        "Perhaps the mirror has been disabled in the upstream?");
                throw new DelegatedAuthenticationFailureException(
                        i18nService.createKeyedMessage("bitbucket.mirroring.authentication.delegation.failure.misconfiguration"));
            } else {
                log.warn("Delegated authentication failed because the upstream responded with an unexpected status code {}",
                        response.getStatusCode());
                return throwAuthenticationSystemException(
                        findFirstNonEmptyError(restErrors), "bitbucket.mirroring.authentication.delegation.failure.unexpected");
            }
        }
        return user;
    }

    private <T> Optional<Page<T>> parseEntityPage(Response response, final Class<? extends T> type) {
        String entity = response.getEntity();

        if (StringUtils.isBlank(entity)) {
            log.debug("The upstream server responded with code {} but the response entity was empty", response.getStatusCode());
            return empty();
        } else if (log.isTraceEnabled()) {
            log.trace("The upstream server responded with code and entity: {}", response.getStatusCode(), entity);
        }

        try {
            ObjectMapper mapper = new ObjectMapper();
            JsonNode page = mapper.readTree(entity);
            if (page == null || page.isNull()) {
                log.info("The response from the upstream server is not a JSON object");
                return empty();
            }
            JsonNode valuesNode = page.get("values");
            if (valuesNode == null || valuesNode.isNull()) {
                log.info("The response from the upstream server doesn't appear to be a JSON page object");
                return empty();
            }
            List<T> values = new ArrayList<>();
            for (JsonNode node : valuesNode) {
                values.add(ValidationUtils.validate(validator, mapper.readValue(node, type)));
            }
            PageRequest pageRequest = PageUtils.newRequest(page.get("start").asInt(), page.get("limit").asInt());
            return of(PageUtils.createPage(values, page.get("isLastPage").asBoolean(), pageRequest));
        } catch (IOException e) {
            log.warn("Failed to deserialize an instance of {} from the response", type.getName(), e);
        }
        return empty();
    }

    private ApplicationUserWithPermissions throwAuthenticationSystemException(RestErrorMessage firstErrorMessage, String keyBase) {
        if (firstErrorMessage == null || StringUtils.isBlank(firstErrorMessage.getMessage())) {
            throw new AuthenticationSystemException(
                    i18nService.createKeyedMessage(keyBase));
        } else {
            throw new AuthenticationSystemException(
                    i18nService.createKeyedMessage(keyBase + ".with.message", firstErrorMessage.getMessage()));
        }
    }

    private ResponseTransformation<Page<RestServerExternalRepository>> toPageOfRepos(MirroringUrl url) {
        return this.<Page<RestServerExternalRepository>>repositoryEndpointResponseBuilder(url)
                .notFound(this::throwRequestFailed)
                .ok(response -> parseEntityPage(response, RestServerExternalRepository.class).orElseThrow(requestFailedSupplier(response)))
                .build();
    }

    private void traceErrorMessages(RestErrors restErrors) {
        if (log.isTraceEnabled() && restErrors != null) {
            for (RestErrorMessage errorMessage : restErrors.getErrors()) {
                log.trace("Delegated authentication error: {}", errorMessage);
            }
        }
    }

    private AnalyticsSettings transformAnalyticsSettings(@Nonnull RestAnalyticsSettings analyticsSettings) {
        return new SimpleAnalyticsSettings(analyticsSettings.canCollectAnalytics(),
                analyticsSettings.getSupportEntitlementNumber());
    }

    private <T extends RuntimeException, R> Function<Throwable, R> wrapAsOrRethrowIfSame(
            Class<T> type, String messageKey, BiFunction<KeyedMessage, Throwable, T> ctor) {

        return e -> {
            if (type.isAssignableFrom(e.getClass())) {
                throw type.cast(e);
            } else {
                throw ctor.apply(i18nService.createKeyedMessage(messageKey, e), e);
            }
        };
    }

    private static class RestMinimalMirrorServer extends RestMapEntity {
        public String getId() {
            return getStringProperty("id");
        }
    }
}
