package org.mockserver.netty;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import org.apache.commons.text.StringEscapeUtils;
import org.mockserver.configuration.Configuration;
import org.mockserver.dashboard.DashboardHandler;
import org.mockserver.lifecycle.LifeCycle;
import org.mockserver.log.model.LogEntry;
import org.mockserver.logging.MockServerLogger;
import org.mockserver.metrics.Metrics;
import org.mockserver.metrics.MetricsHandler;
import org.mockserver.mock.HttpState;
import org.mockserver.mock.action.http.HttpActionHandler;
import org.mockserver.model.HttpRequest;
import org.mockserver.model.HttpResponse;
import org.mockserver.model.MediaType;
import org.mockserver.model.PortBinding;
import org.mockserver.netty.proxy.connect.HttpConnectHandler;
import org.mockserver.netty.responsewriter.NettyResponseWriter;
import org.mockserver.responsewriter.ResponseWriter;
import org.mockserver.scheduler.Scheduler;
import org.mockserver.serialization.Base64Converter;
import org.mockserver.serialization.ConfigurationSerializer;
import org.mockserver.serialization.ObjectMapperFactory;
import org.mockserver.serialization.PortBindingSerializer;
import org.mockserver.serialization.model.ConfigurationDTO;
import org.slf4j.event.Level;

import java.net.BindException;
import java.nio.charset.StandardCharsets;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import static io.netty.handler.codec.http.HttpHeaderNames.*;
import static io.netty.handler.codec.http.HttpResponseStatus.*;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.mockserver.exception.ExceptionHandling.closeOnFlush;
import static org.mockserver.exception.ExceptionHandling.connectionClosedException;
import static org.mockserver.log.model.LogEntry.LogMessageType.AUTHENTICATION_FAILED;
import static org.mockserver.metrics.Metrics.Name.REQUESTS_RECEIVED_COUNT;
import static org.mockserver.mock.HttpState.PATH_PREFIX;
import static org.mockserver.model.HttpResponse.response;
import static org.mockserver.model.PortBinding.portBinding;
import static org.mockserver.netty.unification.PortUnificationHandler.enableSslUpstreamAndDownstream;
import static org.mockserver.netty.unification.PortUnificationHandler.isSslEnabledUpstream;

/**
 * @author jamesdbloom
 */
@ChannelHandler.Sharable
@SuppressWarnings("FieldMayBeFinal")
public class HttpRequestHandler extends SimpleChannelInboundHandler<HttpRequest> {

    public static final AttributeKey<Boolean> PROXYING = AttributeKey.valueOf("PROXYING");
    public static final AttributeKey<Set<String>> LOCAL_HOST_HEADERS = AttributeKey.valueOf("LOCAL_HOST_HEADERS");
    private static final Base64Converter BASE_64_CONVERTER = new Base64Converter();
    private final Configuration configuration;
    private LifeCycle server;
    private HttpState httpState;
    private Metrics metrics;
    private MockServerLogger mockServerLogger;
    private PortBindingSerializer portBindingSerializer;
    private HttpActionHandler httpActionHandler;
    private DashboardHandler dashboardHandler;
    private MetricsHandler metricsHandler;
    private OpenAPISpecHandler openAPISpecHandler;
    private ConfigurationSerializer configurationSerializer;

    public HttpRequestHandler(Configuration configuration, LifeCycle server, HttpState httpState, HttpActionHandler httpActionHandler) {
        super(false);
        this.configuration = configuration;
        this.server = server;
        this.httpState = httpState;
        this.metrics = new Metrics(configuration);
        this.mockServerLogger = httpState.getMockServerLogger();
        this.portBindingSerializer = new PortBindingSerializer(mockServerLogger);
        this.httpActionHandler = httpActionHandler;
        this.dashboardHandler = new DashboardHandler();
        this.metricsHandler = new MetricsHandler(configuration);
        this.openAPISpecHandler = new OpenAPISpecHandler();
        this.configurationSerializer = new ConfigurationSerializer(mockServerLogger);
    }

    private static boolean isProxyingRequest(ChannelHandlerContext ctx) {
        if (ctx != null && ctx.channel() != null && ctx.channel().attr(PROXYING).get() != null) {
            return ctx.channel().attr(PROXYING).get();
        }
        return false;
    }

    public static void setProxyingRequest(ChannelHandlerContext ctx, Boolean value) {
        if (ctx != null && ctx.channel() != null) {
            ctx.channel().attr(PROXYING).set(value);
        }
    }

    private static Set<String> getLocalAddresses(ChannelHandlerContext ctx) {
        if (ctx != null &&
            ctx.channel().attr(LOCAL_HOST_HEADERS) != null &&
            ctx.channel().attr(LOCAL_HOST_HEADERS).get() != null) {
            return ctx.channel().attr(LOCAL_HOST_HEADERS).get();
        }
        return new HashSet<>();
    }

    @Override
    protected void channelRead0(final ChannelHandlerContext ctx, final HttpRequest request) {

        if (configuration.metricsEnabled()) {
            metrics.increment(REQUESTS_RECEIVED_COUNT);
        }

        ResponseWriter responseWriter = new NettyResponseWriter(configuration, mockServerLogger, ctx, httpState.getScheduler());
        try {
            configuration.addSubjectAlternativeName(request.getFirstHeader(HOST.toString()));

            if (!httpState.handle(request, responseWriter, false)) {

                if (request.matches("PUT", PATH_PREFIX + "/status", "/status") ||
                    isNotBlank(configuration.livenessHttpGetPath()) && request.matches("GET", configuration.livenessHttpGetPath())) {

                    responseWriter.writeResponse(request, OK, portBindingSerializer.serialize(portBinding(server.getLocalPorts())), "application/json");

                } else if (request.matches("PUT", PATH_PREFIX + "/bind", "/bind")) {

                    PortBinding requestedPortBindings = portBindingSerializer.deserialize(request.getBodyAsString());
                    if (requestedPortBindings != null) {
                        try {
                            List<Integer> actualPortBindings = server.bindServerPorts(requestedPortBindings.getPorts());
                            responseWriter.writeResponse(request, OK, portBindingSerializer.serialize(portBinding(actualPortBindings)), "application/json");
                        } catch (RuntimeException e) {
                            if (e.getCause() instanceof BindException) {
                                responseWriter.writeResponse(request, BAD_REQUEST, e.getMessage() + " port already in use", MediaType.create("text", "plain").toString());
                            } else {
                                throw e;
                            }
                        }
                    }

                } else if (request.matches("PUT", PATH_PREFIX + "/stop", "/stop")) {

                    ctx.writeAndFlush(response().withStatusCode(OK.code()));
                    new Scheduler.SchedulerThreadFactory("MockServer Stop").newThread(() -> server.stop()).start();

                } else if (request.matches("GET", PATH_PREFIX + "/configuration", "/configuration")
                    || request.matches("PUT", PATH_PREFIX + "/configuration", "/configuration")) {

                    if (httpState.getControlPlaneAuthenticationHandler() != null) {
                        try {
                            if (!httpState.getControlPlaneAuthenticationHandler().controlPlaneRequestAuthenticated(request)) {
                                responseWriter.writeResponse(request, UNAUTHORIZED, "Unauthorized for control plane", MediaType.create("text", "plain").toString());
                                return;
                            }
                        } catch (org.mockserver.authentication.AuthenticationException e) {
                            responseWriter.writeResponse(request, UNAUTHORIZED, "Unauthorized for control plane", MediaType.create("text", "plain").toString());
                            return;
                        }
                    }
                    if (request.getMethod().getValue().equals("GET")) {
                        responseWriter.writeResponse(request, OK, configurationSerializer.serialize(configuration), "application/json");
                    } else {
                        try {
                            ConfigurationDTO configurationDTO = ObjectMapperFactory.createObjectMapper().readValue(request.getBodyAsString(), ConfigurationDTO.class);
                            synchronized (configuration) {
                                configurationDTO.applyTo(configuration);
                            }
                            if (mockServerLogger.isEnabledForInstance(Level.INFO)) {
                                mockServerLogger.logEvent(
                                    new LogEntry()
                                        .setLogLevel(Level.INFO)
                                        .setCorrelationId(request.getLogCorrelationId())
                                        .setHttpRequest(request)
                                        .setMessageFormat("configuration updated via API")
                                );
                            }
                            responseWriter.writeResponse(request, OK, configurationSerializer.serialize(configuration), "application/json");
                        } catch (IllegalArgumentException e) {
                            responseWriter.writeResponse(request, BAD_REQUEST, e.getMessage(), MediaType.create("text", "plain").toString());
                        } catch (Exception e) {
                            responseWriter.writeResponse(request, BAD_REQUEST, "Invalid configuration JSON", MediaType.create("text", "plain").toString());
                        }
                    }

                } else if (request.getMethod().getValue().equals("GET") && request.getPath().getValue().startsWith(PATH_PREFIX + "/dashboard")) {

                    dashboardHandler.renderDashboard(ctx, request);

                } else if (request.getMethod().getValue().equals("GET") && request.getPath().getValue().equals(PATH_PREFIX + "/openapi.yaml")) {

                    if (httpState.getControlPlaneAuthenticationHandler() != null) {
                        try {
                            if (!httpState.getControlPlaneAuthenticationHandler().controlPlaneRequestAuthenticated(request)) {
                                responseWriter.writeResponse(request, UNAUTHORIZED, "Unauthorized for control plane", MediaType.create("text", "plain").toString());
                                return;
                            }
                        } catch (org.mockserver.authentication.AuthenticationException e) {
                            responseWriter.writeResponse(request, UNAUTHORIZED, "Unauthorized for control plane", MediaType.create("text", "plain").toString());
                            return;
                        }
                    }
                    openAPISpecHandler.renderOpenAPISpec(ctx, request);

                } else if (request.getMethod().getValue().equals("GET") && request.getPath().getValue().matches(PATH_PREFIX + "/metrics")) {

                    // Always reserve this control-plane path (like /dashboard and /openapi.yaml
                    // above). MetricsHandler serves the metrics when enabled and a CORS-decorated
                    // 404 when disabled, so a cross-origin dashboard reads the disabled state
                    // cleanly instead of the request falling through to mock matching.
                    metricsHandler.renderMetrics(ctx, request);

                } else if (request.getMethod().getValue().equals("CONNECT")) {

                    String username = configuration.proxyAuthenticationUsername();
                    String password = configuration.proxyAuthenticationPassword();
                    if (isNotBlank(username) && isNotBlank(password) &&
                        !request.containsHeader(PROXY_AUTHORIZATION.toString(), "Basic " + BASE_64_CONVERTER.bytesToBase64String((username + ':' + password).getBytes(StandardCharsets.UTF_8), StandardCharsets.US_ASCII))) {
                        HttpResponse response = response()
                            .withStatusCode(PROXY_AUTHENTICATION_REQUIRED.code())
                            .withHeader(PROXY_AUTHENTICATE.toString(), "Basic realm=\"" + StringEscapeUtils.escapeJava(configuration.proxyAuthenticationRealm()) + "\", charset=\"UTF-8\"");
                        ctx.writeAndFlush(response);
                        mockServerLogger.logEvent(
                            new LogEntry()
                                .setType(AUTHENTICATION_FAILED)
                                .setLogLevel(Level.INFO)
                                .setCorrelationId(request.getLogCorrelationId())
                                .setHttpRequest(request)
                                .setHttpResponse(response)
                                .setExpectation(request, response)
                                .setMessageFormat("proxy authentication failed so returning response:{}for forwarded request:{}")
                                .setArguments(response, request)
                        );
                    } else {
                        setProxyingRequest(ctx, Boolean.TRUE);
                        // assume SSL for CONNECT request
                        enableSslUpstreamAndDownstream(ctx.channel());
                        String[] hostParts = HttpRequest.splitHostPort(request.getPath().getValue());
                        String connectHost = hostParts[0];
                        int port = hostParts.length > 1 ? Integer.parseInt(hostParts[1]) : isSslEnabledUpstream(ctx.channel()) ? 443 : 80;
                        if (isNotBlank(connectHost)) {
                            server.getScheduler().submit(() -> configuration.addSubjectAlternativeName(connectHost));
                        }
                        ctx.pipeline().addLast(new HttpConnectHandler(configuration, server, mockServerLogger, connectHost, port));
                        ctx.pipeline().remove(this);
                        ctx.fireChannelRead(request);
                    }

                } else {

                    if (configuration.tlsMutualAuthenticationRequired() && !isSslEnabledUpstream(ctx.channel())) {
                        HttpResponse upgradeResponse = response()
                            .withStatusCode(426)
                            .withReasonPhrase("Upgrade Required")
                            .withHeader("Upgrade", "TLS/1.2, HTTP/1.1")
                            .withHeader("Connection", "Upgrade");
                        if (mockServerLogger.isEnabledForInstance(Level.INFO)) {
                            mockServerLogger.logEvent(
                                new LogEntry()
                                    .setLogLevel(Level.INFO)
                                    .setHttpRequest(request)
                                    .setMessageFormat("no tls for data plane request:{}returning response:{}")
                                    .setArguments(request, upgradeResponse)
                            );
                        }
                        responseWriter.writeResponse(request, upgradeResponse, false);
                    } else {
                        try {
                            httpActionHandler.processAction(request, responseWriter, ctx, getLocalAddresses(ctx), isProxyingRequest(ctx), false);
                        } catch (Throwable throwable) {
                            mockServerLogger.logEvent(
                                new LogEntry()
                                    .setLogLevel(Level.ERROR)
                                    .setHttpRequest(request)
                                    .setMessageFormat("exception processing request:{}error:{}")
                                    .setArguments(request, throwable.getMessage())
                                    .setThrowable(throwable)
                            );
                        }
                    }

                }
            }
        } catch (IllegalArgumentException iae) {
            mockServerLogger.logEvent(
                new LogEntry()
                    .setLogLevel(Level.ERROR)
                    .setHttpRequest(request)
                    .setMessageFormat("exception processing request:{}error:{}")
                    .setArguments(request, iae.getMessage())
            );
            // send request without API CORS headers
            responseWriter.writeResponse(request, BAD_REQUEST, iae.getMessage(), MediaType.create("text", "plain").toString());
        } catch (Exception ex) {
            mockServerLogger.logEvent(
                new LogEntry()
                    .setLogLevel(Level.ERROR)
                    .setHttpRequest(request)
                    .setMessageFormat("exception processing " + request)
                    .setThrowable(ex)
            );
            responseWriter.writeResponse(request, response().withStatusCode(BAD_REQUEST.code()).withBody(ex.getMessage()), true);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        if (connectionClosedException(cause)) {
            mockServerLogger.logEvent(
                new LogEntry()
                    .setLogLevel(Level.ERROR)
                    .setMessageFormat("exception caught by " + server.getClass() + " handler -> closing pipeline " + ctx.channel())
                    .setThrowable(cause)
            );
        }
        closeOnFlush(ctx.channel());
    }
}
