/*
 * 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;

import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;

import com.google.common.collect.Maps;
import org.eclipse.milo.opcua.stack.client.transport.UaTransport;
import org.eclipse.milo.opcua.stack.client.transport.http.OpcHttpTransport;
import org.eclipse.milo.opcua.stack.client.transport.tcp.OpcTcpTransport;
import org.eclipse.milo.opcua.stack.client.transport.websocket.OpcWebSocketTransport;
import org.eclipse.milo.opcua.stack.core.NamespaceTable;
import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.UaServiceFaultException;
import org.eclipse.milo.opcua.stack.core.channel.EncodingLimits;
import org.eclipse.milo.opcua.stack.core.serialization.SerializationContext;
import org.eclipse.milo.opcua.stack.core.serialization.UaRequestMessage;
import org.eclipse.milo.opcua.stack.core.serialization.UaResponseMessage;
import org.eclipse.milo.opcua.stack.core.transport.TransportProfile;
import org.eclipse.milo.opcua.stack.core.types.DataTypeManager;
import org.eclipse.milo.opcua.stack.core.types.DefaultDataTypeManager;
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.builtin.unsigned.UInteger;
import org.eclipse.milo.opcua.stack.core.types.structured.RequestHeader;
import org.eclipse.milo.opcua.stack.core.types.structured.ResponseHeader;
import org.eclipse.milo.opcua.stack.core.types.structured.ServiceFault;
import org.eclipse.milo.opcua.stack.core.util.ExecutionQueue;
import org.eclipse.milo.opcua.stack.core.util.LongSequence;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

public class UaStackClient {

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

    private final LongSequence requestHandles = new LongSequence(0, UInteger.MAX_VALUE);

    private final Map<UInteger, CompletableFuture<UaResponseMessage>> pending = Maps.newConcurrentMap();

    private final NamespaceTable namespaceTable = new NamespaceTable();

    private final DataTypeManager staticDataTypeManager =
        DefaultDataTypeManager.createAndInitialize(namespaceTable);

    private final DataTypeManager dynamicDataTypeManager =
        DefaultDataTypeManager.createAndInitialize(namespaceTable);

    private final SerializationContext staticSerializationContext;
    private final SerializationContext dynamicSerializationContext;

    private final UaTransport transport;
    private final ExecutionQueue deliveryQueue;

    private final UaStackClientConfig config;

    public UaStackClient(
        UaStackClientConfig config,
        Function<UaStackClient, UaTransport> transportFactory
    ) {

        this.config = config;

        deliveryQueue = new ExecutionQueue(config.getExecutor());

        staticSerializationContext = new SerializationContext() {
            @Override
            public EncodingLimits getEncodingLimits() {
                return config.getEncodingLimits();
            }

            @Override
            public NamespaceTable getNamespaceTable() {
                return namespaceTable;
            }

            @Override
            public DataTypeManager getDataTypeManager() {
                return staticDataTypeManager;
            }
        };

        dynamicSerializationContext = new SerializationContext() {
            @Override
            public EncodingLimits getEncodingLimits() {
                return config.getEncodingLimits();
            }

            @Override
            public NamespaceTable getNamespaceTable() {
                return namespaceTable;
            }

            @Override
            public DataTypeManager getDataTypeManager() {
                return dynamicDataTypeManager;
            }
        };

        transport = transportFactory.apply(this);
    }

    /**
     * @return the {@link UaStackClientConfig} this client was created with.
     */
    public UaStackClientConfig getConfig() {
        return config;
    }

    /**
     * Connect this {@link UaStackClient} to the server in the configured endpoint.
     * <p>
     * Depending on the underlying transport this may or may not actually make a connection attempt.
     *
     * @return this {@link UaStackClient}.
     */
    public CompletableFuture<UaStackClient> connect() {
        return transport.connect()
            .thenApply(t -> UaStackClient.this);
    }

    /**
     * Disconnect this {@link UaStackClient} from the server in the configured endpoint.
     * <p>
     * Depending on the underlying transport this may or may not actually make a disconnect attempt.
     *
     * @return this {@link UaStackClient}.
     */
    public CompletableFuture<UaStackClient> disconnect() {
        return transport.disconnect()
            .whenComplete((u, ex) -> {
                UaException disconnect = new UaException(
                    StatusCodes.Bad_Disconnect, "client disconnect");

                pending.forEach((h, cf) -> cf.completeExceptionally(disconnect));
                pending.clear();
            })
            .thenApply(t -> UaStackClient.this);
    }

    /**
     * Get the {@link UaTransport} used by this client.
     *
     * @return the {@link UaTransport} used by this client.
     */
    public UaTransport getTransport() {
        return transport;
    }

    /**
     * Get the client {@link NamespaceTable}.
     *
     * @return the client {@link NamespaceTable}.
     */
    public NamespaceTable getNamespaceTable() {
        return namespaceTable;
    }

    /**
     * Get the client's "static" {@link DataTypeManager}.
     * <p>
     * This {@link DataTypeManager} is for static codecs that serialize classes that exist at
     * compile time, e.g. structures from namespace 0 and or code-generated structures.
     *
     * @return the client's static {@link DataTypeManager}.
     */
    public DataTypeManager getStaticDataTypeManager() {
        return staticDataTypeManager;
    }

    /**
     * Get the client's "dynamic" {@link DataTypeManager}.
     * <p>
     * This {@link DataTypeManager} is for dynamic codecs that were created by reading the server's
     * DataType Dictionary at runtime and serializes generic representations of structures used by
     * instances of a BsdParser implementation.
     *
     * @return the client's dynamic {@link DataTypeManager}.
     */
    public DataTypeManager getDynamicDataTypeManager() {
        return dynamicDataTypeManager;
    }

    /**
     * Get a "static" {@link SerializationContext} instance.
     * <p>
     * This {@link SerializationContext} instance returns the client's static {@link DataTypeManager}.
     *
     * @return a "static" {@link SerializationContext} instance.
     * @see #getStaticDataTypeManager()
     */
    public SerializationContext getStaticSerializationContext() {
        return staticSerializationContext;
    }

    /**
     * Get a "dynamic" {@link SerializationContext}.
     * <p>
     * This {@link SerializationContext} instance returns the client's dynamic {@link DataTypeManager}.
     *
     * @return a "dynamic" {@link SerializationContext}.
     * @see #getDynamicDataTypeManager()
     */
    public SerializationContext getDynamicSerializationContext() {
        return dynamicSerializationContext;
    }

    /**
     * Create a new {@link RequestHeader} with a null authentication token.
     * <p>
     * A unique request handle will be automatically assigned to the header.
     *
     * @return a new {@link RequestHeader} with a null authentication token.
     */
    public RequestHeader newRequestHeader() {
        return newRequestHeader(NodeId.NULL_VALUE);
    }

    /**
     * Create a new {@link RequestHeader} with {@code authToken}.
     * <p>
     * A unique request handle will be automatically assigned to the header.
     *
     * @param authToken the authentication token to create the header with.
     * @return a new {@link RequestHeader} created with {@code authToken}.
     */
    public RequestHeader newRequestHeader(NodeId authToken) {
        return newRequestHeader(authToken, config.getRequestTimeout());
    }

    /**
     * Create a new {@link RequestHeader} with {@code authToken} and {@code requestTimeout}.
     * <p>
     * A unique request handle will be automatically assigned to the header.
     *
     * @param authToken      the authentication token to create the header with.
     * @param requestTimeout the timeout hint to create the header with.f
     * @return a new {@link RequestHeader} created with {@code authToken} and {@code requestTimeout}.
     */
    public RequestHeader newRequestHeader(NodeId authToken, UInteger requestTimeout) {
        return new RequestHeader(
            authToken,
            DateTime.now(),
            uint(requestHandles.getAndIncrement()),
            uint(0),
            null,
            requestTimeout,
            null
        );
    }

    /**
     * Send a {@link UaRequestMessage} to the connected server.
     * <p>
     * The {@link RequestHeader} of {@code request} must have a unique request handle. Use the
     * {@code newRequestHeader} helper functions to create headers for {@link UaRequestMessage}s.
     *
     * @param request the {@link UaRequestMessage} to send.
     * @return a {@link CompletableFuture} containing the eventual {@link UaResponseMessage} from the server.
     * @see #newRequestHeader()
     * @see #newRequestHeader(NodeId)
     * @see #newRequestHeader(NodeId, UInteger)
     */
    public CompletableFuture<UaResponseMessage> sendRequest(UaRequestMessage request) {
        RequestHeader requestHeader = request.getRequestHeader();
        UInteger requestHandle = requestHeader.getRequestHandle();

        final CompletableFuture<UaResponseMessage> future = new CompletableFuture<>();
        pending.put(requestHandle, future);

        transport.sendRequest(request).whenComplete((response, ex) -> {
            pending.remove(requestHandle);

            deliverResponse(request, response, ex, future);
        });

        return future;
    }

    /**
     * Complete {@code future} with {@code response} on the {@code deliveryQueue}.
     * <p>
     * This is done for two reasons:
     * 1. the transport future is completed on its serialization queue thread, which we want to get off of ASAP.
     * 2. the futures need to be completed serially, in the order received from the server.
     *
     * @param request  the original {@link UaRequestMessage}.
     * @param response the {@link UaResponseMessage}.
     * @param future   the {@link CompletableFuture} awaiting completion.
     */
    private void deliverResponse(
        UaRequestMessage request,
        @Nullable UaResponseMessage response,
        @Nullable Throwable failure,
        CompletableFuture<UaResponseMessage> future
    ) {

        deliveryQueue.submit(() -> {
            if (response != null) {
                ResponseHeader header = response.getResponseHeader();
                UInteger requestHandle = header.getRequestHandle();

                if (header.getServiceResult().isGood()) {
                    future.complete(response);
                } else {
                    ServiceFault serviceFault;

                    if (response instanceof ServiceFault) {
                        serviceFault = (ServiceFault) response;
                    } else {
                        serviceFault = new ServiceFault(header);
                    }

                    if (logger.isDebugEnabled()) {
                        logger.debug(
                            "Received ServiceFault request={} requestHandle={}, result={}",
                            request.getClass().getSimpleName(),
                            requestHandle,
                            header.getServiceResult()
                        );
                    }

                    future.completeExceptionally(new UaServiceFaultException(serviceFault));
                }
            } else {
                assert failure != null;

                if (logger.isDebugEnabled()) {
                    logger.debug(
                        "sendRequest() failed, request={}, requestHandle={}",
                        request.getClass().getSimpleName(),
                        request.getRequestHeader().getRequestHandle(),
                        failure
                    );
                }

                future.completeExceptionally(failure);
            }
        });
    }

    /**
     * Create a {@link UaStackClient} with {@code config}.
     * <p>
     * The {@link UaTransport} instance to create will be inferred from the transport profile URI in the configured
     * endpoint.
     * <p>
     * Supported transports:
     * <ul>
     * <li>TCP + UA Binary</li>
     * <li>HTTP(s) + UA Binary</li>
     * </ul>
     * <p>
     * Experimentally supported:
     * <ul>
     * <li>WebSocket + UA Binary</li>
     * </ul>
     * <p>
     * Not supported:
     * <ul>
     * <li>HTTP(s) + UA JSON</li>
     * <li>HTTP(s) + UA XML</li>
     * <li>WebSocket + UA JSON</li>
     * </ul>
     *
     * @param config the {@link UaStackClientConfig}.
     * @return a {@link UaStackClient} created with {@code config}.
     * @throws UaException if the transport is unsupported.
     */
    public static UaStackClient create(UaStackClientConfig config) throws UaException {
        String transportProfileUri =
            config.getEndpoint().getTransportProfileUri();

        TransportProfile transportProfile =
            TransportProfile.fromUri(transportProfileUri);

        Function<UaStackClient, UaTransport> transportFactory;

        switch (transportProfile) {
            case TCP_UASC_UABINARY:
                transportFactory = OpcTcpTransport::new;
                break;

            case HTTPS_UABINARY:
                transportFactory = OpcHttpTransport::new;
                break;

            case WSS_UASC_UABINARY:
                transportFactory = OpcWebSocketTransport::new;
                break;

            case HTTPS_UAXML:
            case HTTPS_UAJSON:
            case WSS_UAJSON:
            default:
                throw new UaException(
                    StatusCodes.Bad_InternalError,
                    "unsupported transport: " + transportProfileUri);
        }

        return new UaStackClient(config, transportFactory);
    }

}
