/*
 * Copyright (c) 2019 the Eclipse Milo Authors
 *
 * This program and the accompanying materials are made
 * available under the terms of the Eclipse Public License 2.0
 * which is available at https://www.eclipse.org/legal/epl-2.0/
 *
 * SPDX-License-Identifier: EPL-2.0
 */

package org.eclipse.milo.opcua.stack.server.transport.uasc;

import java.io.IOException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import com.google.common.primitives.Ints;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.AttributeKey;
import org.eclipse.milo.opcua.stack.core.Stack;
import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.channel.ChannelParameters;
import org.eclipse.milo.opcua.stack.core.channel.EncodingLimits;
import org.eclipse.milo.opcua.stack.core.channel.ExceptionHandler;
import org.eclipse.milo.opcua.stack.core.channel.SerializationQueue;
import org.eclipse.milo.opcua.stack.core.channel.headers.HeaderDecoder;
import org.eclipse.milo.opcua.stack.core.channel.messages.AcknowledgeMessage;
import org.eclipse.milo.opcua.stack.core.channel.messages.ErrorMessage;
import org.eclipse.milo.opcua.stack.core.channel.messages.HelloMessage;
import org.eclipse.milo.opcua.stack.core.channel.messages.MessageType;
import org.eclipse.milo.opcua.stack.core.channel.messages.TcpMessageDecoder;
import org.eclipse.milo.opcua.stack.core.channel.messages.TcpMessageEncoder;
import org.eclipse.milo.opcua.stack.core.transport.TransportProfile;
import org.eclipse.milo.opcua.stack.core.util.EndpointUtil;
import org.eclipse.milo.opcua.stack.server.UaStackServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class UascServerHelloHandler extends ByteToMessageDecoder implements HeaderDecoder {

    static final AttributeKey<String> ENDPOINT_URL_KEY = AttributeKey.valueOf("endpoint-url");

    /**
     * Cumulative count of all connection rejections for the lifetime of the server.
     */
    @SuppressWarnings("WeakerAccess")
    public static final AtomicLong CUMULATIVE_DEADLINES_MISSED = new AtomicLong(0L);

    private static final int MAX_HELLO_MESSAGE_SIZE = 8 + 20 + 4 + 4096;

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private volatile boolean receivedHello = false;

    private final UaStackServer stackServer;
    private final TransportProfile transportProfile;

    public UascServerHelloHandler(UaStackServer stackServer, TransportProfile transportProfile) {
        if (transportProfile != TransportProfile.TCP_UASC_UABINARY &&
            transportProfile != TransportProfile.WSS_UASC_UABINARY) {

            throw new IllegalArgumentException("transportProfile: " + transportProfile);
        }

        this.stackServer = stackServer;
        this.transportProfile = transportProfile;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        int helloDeadlineMs = Stack.ConnectionLimits.HELLO_DEADLINE_MS;

        logger.debug("Scheduling Hello deadline for +" + helloDeadlineMs + "ms");

        ctx.executor().schedule(
            () -> {
                if (!receivedHello) {
                    long cumulativeDeadlinesMissed =
                        CUMULATIVE_DEADLINES_MISSED.incrementAndGet();

                    logger.debug("No Hello received after " +
                        helloDeadlineMs + "ms; closing channel. " +
                        "cumulativeDeadlinesMissed=" + cumulativeDeadlinesMissed);

                    ctx.close();
                }
            },
            helloDeadlineMs, TimeUnit.MILLISECONDS
        );

        super.channelActive(ctx);
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
        if (buffer.readableBytes() >= HEADER_LENGTH) {
            int messageLength = getMessageLength(buffer, MAX_HELLO_MESSAGE_SIZE);

            if (buffer.readableBytes() >= messageLength) {
                MessageType messageType = MessageType.fromMediumInt(
                    buffer.getMediumLE(buffer.readerIndex())
                );

                if (messageType == MessageType.Hello) {
                    onHello(ctx, buffer.readSlice(messageLength));
                } else {
                    throw new UaException(
                        StatusCodes.Bad_TcpMessageTypeInvalid,
                        "unexpected MessageType: " + messageType
                    );
                }
            }
        }
    }

    private void onHello(ChannelHandlerContext ctx, ByteBuf buffer) throws UaException {
        logger.debug("[remote={}] Received Hello message.", ctx.channel().remoteAddress());

        receivedHello = true;

        final HelloMessage hello = TcpMessageDecoder.decodeHello(buffer);

        String endpointUrl = hello.getEndpointUrl();

        boolean endpointMatch = endpointUrl != null &&
            stackServer.getEndpointDescriptions()
                .stream()
                .anyMatch(endpoint ->
                    Objects.equals(
                        EndpointUtil.getPath(endpointUrl),
                        EndpointUtil.getPath(endpoint.getEndpointUrl()))
                );

        if (!endpointMatch) {
            throw new UaException(
                StatusCodes.Bad_TcpEndpointUrlInvalid,
                "unrecognized endpoint url: " + endpointUrl);
        }

        ctx.channel().attr(ENDPOINT_URL_KEY).set(endpointUrl);

        long remoteProtocolVersion = hello.getProtocolVersion();
        long remoteReceiveBufferSize = hello.getReceiveBufferSize();
        long remoteSendBufferSize = hello.getSendBufferSize();
        long remoteMaxMessageSize = hello.getMaxMessageSize();
        long remoteMaxChunkCount = hello.getMaxChunkCount();

        if (remoteProtocolVersion < PROTOCOL_VERSION) {
            throw new UaException(StatusCodes.Bad_ProtocolVersionUnsupported,
                "unsupported protocol version: " + remoteProtocolVersion);
        }

        EncodingLimits config = stackServer.getConfig().getEncodingLimits();

        /* Our receive buffer size is determined by the remote send buffer size. */
        long localReceiveBufferSize = Math.min(remoteSendBufferSize, config.getMaxChunkSize());

        /* Our send buffer size is determined by the remote receive buffer size. */
        long localSendBufferSize = Math.min(remoteReceiveBufferSize, config.getMaxChunkSize());

        /* Max chunk count the remote can send us; not influenced by remote configuration. */
        long localMaxChunkCount = config.getMaxChunkCount();

        /* Max message size the remote can send us. Determined by our max chunk count and receive buffer size. */
        long localMaxMessageSize = Math.min(localReceiveBufferSize * localMaxChunkCount, config.getMaxMessageSize());

        ChannelParameters parameters = new ChannelParameters(
            Ints.saturatedCast(localMaxMessageSize),
            Ints.saturatedCast(localReceiveBufferSize),
            Ints.saturatedCast(localSendBufferSize),
            Ints.saturatedCast(localMaxChunkCount),
            Ints.saturatedCast(remoteMaxMessageSize),
            Ints.saturatedCast(remoteReceiveBufferSize),
            Ints.saturatedCast(remoteSendBufferSize),
            Ints.saturatedCast(remoteMaxChunkCount)
        );

        SerializationQueue serializationQueue = new SerializationQueue(
            stackServer.getConfig().getExecutor(),
            parameters,
            stackServer.getSerializationContext()
        );

        ctx.pipeline().addLast(new UascServerAsymmetricHandler(stackServer, transportProfile, serializationQueue));
        ctx.pipeline().remove(this);

        logger.debug("[remote={}] Removed HelloHandler, added AsymmetricHandler.", ctx.channel().remoteAddress());

        AcknowledgeMessage acknowledge = new AcknowledgeMessage(
            PROTOCOL_VERSION,
            localReceiveBufferSize,
            localSendBufferSize,
            localMaxMessageSize,
            localMaxChunkCount
        );

        ByteBuf messageBuffer = TcpMessageEncoder.encode(acknowledge);

        // Using ctx.executor() is necessary to ensure this handler is removed
        // before the message can be written and another response arrives.
        ctx.executor().execute(() -> ctx.writeAndFlush(messageBuffer));

        logger.debug("[remote={}] Sent Acknowledge message.", ctx.channel().remoteAddress());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (cause instanceof IOException) {
            ctx.close();
            logger.debug("[remote={}] IOException caught; channel closed",
                ctx.channel().remoteAddress(), cause);
        } else {
            ErrorMessage errorMessage = ExceptionHandler.sendErrorMessage(ctx, cause);

            if (cause instanceof UaException) {
                logger.debug("[remote={}] UaException caught; sent {}",
                    ctx.channel().remoteAddress(), errorMessage, cause);
            } else {
                logger.error("[remote={}] Exception caught; sent {}",
                    ctx.channel().remoteAddress(), errorMessage, cause);
            }
        }
    }

}
