/*
 * 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.client.transport.uasc;

import java.net.ConnectException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import com.digitalpetri.netty.fsm.ChannelActions;
import com.digitalpetri.netty.fsm.ChannelFsm;
import com.digitalpetri.netty.fsm.ChannelFsmConfig;
import com.digitalpetri.netty.fsm.ChannelFsmFactory;
import com.digitalpetri.netty.fsm.Event;
import com.digitalpetri.netty.fsm.State;
import com.digitalpetri.strictmachine.FsmContext;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ConnectTimeoutException;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import org.eclipse.milo.opcua.stack.client.UaStackClient;
import org.eclipse.milo.opcua.stack.client.UaStackClientConfig;
import org.eclipse.milo.opcua.stack.client.transport.tcp.OpcClientTcpChannelInitializer;
import org.eclipse.milo.opcua.stack.client.transport.websocket.OpcClientWebSocketChannelInitializer;
import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.transport.TransportProfile;
import org.eclipse.milo.opcua.stack.core.types.builtin.DateTime;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.structured.CloseSecureChannelRequest;
import org.eclipse.milo.opcua.stack.core.types.structured.RequestHeader;
import org.eclipse.milo.opcua.stack.core.util.EndpointUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;

public class ClientChannelFsm {

    private static final String CHANNEL_FSM_LOGGER_NAME = "org.eclipse.milo.opcua.stack.client.ChannelFsm";

    public static ChannelFsm newChannelFsm(UaStackClient client) {
        ChannelFsmConfig fsmConfig = ChannelFsmConfig.newBuilder()
            .setLazy(false) // reconnect immediately
            .setMaxIdleSeconds(0) // keep alive handled by SessionFsm
            .setMaxReconnectDelaySeconds(16)
            .setPersistent(true)
            .setChannelActions(new ClientChannelActions(client))
            .setExecutor(client.getConfig().getExecutor())
            .setScheduler(client.getConfig().getScheduledExecutor())
            .setLoggerName(CHANNEL_FSM_LOGGER_NAME)
            .build();

        ChannelFsmFactory fsmFactory = new ChannelFsmFactory(fsmConfig);

        return fsmFactory.newChannelFsm();
    }

    private static class ClientChannelActions implements ChannelActions {

        private static final Logger LOGGER = LoggerFactory.getLogger(CHANNEL_FSM_LOGGER_NAME);

        private final UaStackClientConfig config;
        private final UaStackClient client;

        ClientChannelActions(UaStackClient client) {
            this.client = client;
            this.config = client.getConfig();
        }

        @Override
        public CompletableFuture<Channel> connect(FsmContext<State, Event> ctx) {
            CompletableFuture<ClientSecureChannel> handshake = new CompletableFuture<>();

            String transportProfileUri = config.getEndpoint().getTransportProfileUri();
            TransportProfile transportProfile = TransportProfile.fromUri(transportProfileUri);

            ChannelInitializer<SocketChannel> initializer;

            if (transportProfile == TransportProfile.TCP_UASC_UABINARY) {
                initializer = new OpcClientTcpChannelInitializer(client, handshake);
            } else {
                initializer = new OpcClientWebSocketChannelInitializer(client, handshake);
            }

            Bootstrap bootstrap = new Bootstrap();

            bootstrap.group(config.getEventLoop())
                .channel(NioSocketChannel.class)
                .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.getConnectTimeout().intValue())
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(initializer);

            try {
                String endpointUrl = config.getEndpoint().getEndpointUrl();

                String host = EndpointUtil.getHost(endpointUrl);
                assert host != null;

                int port = EndpointUtil.getPort(endpointUrl);

                bootstrap.connect(host, port).addListener((ChannelFuture f) -> {
                    if (!f.isSuccess()) {
                        Throwable cause = f.cause();

                        if (cause instanceof ConnectTimeoutException) {
                            handshake.completeExceptionally(
                                new UaException(StatusCodes.Bad_Timeout, f.cause()));
                        } else if (cause instanceof ConnectException) {
                            handshake.completeExceptionally(
                                new UaException(StatusCodes.Bad_ConnectionRejected, f.cause()));
                        } else {
                            handshake.completeExceptionally(cause);
                        }
                    }
                });
            } catch (Throwable e) {
                UaException failure = new UaException(
                    StatusCodes.Bad_TcpEndpointUrlInvalid, e);

                handshake.completeExceptionally(failure);
            }

            return handshake.thenApply(ClientSecureChannel::getChannel);
        }

        @Override
        public CompletableFuture<Void> disconnect(FsmContext<State, Event> ctx, Channel channel) {
            CompletableFuture<Void> disconnectFuture = new CompletableFuture<>();

            final TimerTask onTimeout = t -> channel.close().addListener(
                (ChannelFutureListener) channelFuture ->
                    disconnectFuture.complete(null)
            );

            final Timeout timeout = config.getWheelTimer().newTimeout(
                onTimeout,
                5,
                TimeUnit.SECONDS
            );

            channel.pipeline().addFirst(new ChannelInboundHandlerAdapter() {
                @Override
                public void channelInactive(ChannelHandlerContext channelContext) throws Exception {
                    LOGGER.debug("[{}] channelInactive() disconnect complete", ctx.getInstanceId());
                    timeout.cancel();
                    disconnectFuture.complete(null);
                    super.channelInactive(channelContext);
                }
            });

            RequestHeader requestHeader = new RequestHeader(
                NodeId.NULL_VALUE,
                DateTime.now(),
                uint(0),
                uint(0),
                null,
                uint(0),
                null
            );

            LOGGER.debug("[{}] Sending CloseSecureChannelRequest...", ctx.getInstanceId());

            channel.pipeline().fireUserEventTriggered(
                new CloseSecureChannelRequest(requestHeader)
            );

            return disconnectFuture;
        }

        @Override
        public CompletableFuture<Void> keepAlive(FsmContext<State, Event> ctx, Channel channel) {
            return CompletableFuture.completedFuture(null);
        }

    }
}
