/*_############################################################################
  _## 
  _##  SNMP4J-UNIX-TRANSPORT - UnixSocketStreamTransportMapping.java  
  _## 
  _##  Copyright (C) 2021-2026  Frank Fock (SNMP4J.org)
  _##  
  _##  Licensed under the Apache License, Version 2.0 (the "License");
  _##  you may not use this file except in compliance with the License.
  _##  You may obtain a copy of the License at
  _##  
  _##      http://www.apache.org/licenses/LICENSE-2.0
  _##  
  _##  Unless required by applicable law or agreed to in writing, software
  _##  distributed under the License is distributed on an "AS IS" BASIS,
  _##  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  _##  See the License for the specific language governing permissions and
  _##  limitations under the License.
  _##  
  _##########################################################################*/
package org.snmp4j.transport.unix;

import org.snmp4j.SNMP4JSettings;
import org.snmp4j.TransportStateReference;
import org.snmp4j.log.LogAdapter;
import org.snmp4j.log.LogFactory;
import org.snmp4j.security.SecurityLevel;
import org.snmp4j.smi.Address;
import org.snmp4j.transport.*;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.*;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Optional;
import java.util.Set;

/**
 * A transport mapping implementation for Unix domain socket communication that leverages
 * stream-oriented semantics. This class is built upon the AbstractConnectionOrientedTransportMapping,
 * providing enhanced support for Unix domain socket addresses.
 * <p>
 * The class allows for sending and receiving messages, setting socket options, and managing
 * Unix domain socket communication-specific behaviors. It supports asynchronous message
 * handling and can handle multiple concurrent Unix domain socket-based connections.
 */
public class UnixSocketStreamTransportMapping extends AbstractConnectionOrientedTransportMapping<UnixDomainAddress,
        UnixSocketStreamTransportMapping.SocketEntry> {

    private static final LogAdapter logger = LogFactory.getLogger(UnixSocketStreamTransportMapping.class);

    /**
     * The Unix domain socket address to which this transport mapping is bound.
     */
    protected UnixDomainAddress listenAddress;
    /**
     * The server thread responsible for handling incoming connections.
     */
    protected ServerThread serverThread;

    /**
     * The socket path that has been bound by this transport mapping and that therefore has to be removed
     * when this transport mapping is closed, or {@code null} if this transport mapping has not created a
     * socket file. Only a socket file that has been created by this transport mapping is ever removed, so
     * that the socket of another SNMP entity that uses the same path is left alone.
     *
     * @since 1.2.2
     */
    protected volatile Path boundSocketPath;

    /**
     * Constructs a UnixSocketStreamTransportMapping with the specified Unix domain socket address and
     * server mode enabled, that is, incoming connections on that address are accepted.
     *
     * @param listenAddress the Unix domain socket address to which this transport mapping is bound.
     */
    public UnixSocketStreamTransportMapping(UnixDomainAddress listenAddress) {
        this(listenAddress, true);
    }

    /**
     * Constructs a UnixSocketStreamTransportMapping with the specified Unix domain socket address and
     * enables server mode on request.
     * <p>
     * In client mode ({@code serverEnabled} is {@code false}) no socket is bound to listen for incoming
     * connections. Unlike an IP based transport mapping, which can bind an ephemeral port for its outgoing
     * connections, a Unix domain socket path is the address itself: listening on it in client mode would
     * occupy the path of the transport mapping that serves incoming connections on it and fail with
     * "Address already in use". The <em>outgoing</em> socket of a client mode transport mapping is bound to
     * the given address instead, so that the peer can tell this SNMP entity apart from other entities that
     * connect to it - an unbound Unix domain socket has no address at all, thus all peers that connect
     * without binding their socket look alike to the entity that accepts their connections. See
     * {@link ServerThread#bindOutgoingSocket(SocketChannel)}.
     *
     * @param listenAddress the Unix domain socket address to which this transport mapping is bound.
     * @param serverEnabled if {@code true} the server mode is enabled and incoming new connections are
     *                      accepted. Use {@code false} to allow outgoing messages and their responses only
     *                      (client mode).
     * @since 1.2.2
     */
    public UnixSocketStreamTransportMapping(UnixDomainAddress listenAddress, boolean serverEnabled) {
        this.listenAddress = listenAddress;
        this.serverEnabled = serverEnabled;
    }

    /**
     * Constructs a UnixSocketStreamTransportMapping without specifying a Unix domain socket address.
     * This constructor is useful when the address will be set later.
     */
    public UnixSocketStreamTransportMapping() {
        this(null, false);
    }

    @Override
    protected <T> void setSocketOption(SocketOption<T> socketOption, T value) throws IOException {
        serverThread.getServerSocketChannel().setOption(socketOption, value);
    }

    @Override
    public MessageLengthDecoder getMessageLengthDecoder() {
        return null;
    }

    @Override
    public void setMessageLengthDecoder(MessageLengthDecoder messageLengthDecoder) {
    }

    @Override
    public Class<? extends Address> getSupportedAddressClass() {
        return UnixDomainAddress.class;
    }

    @Override
    public Set<Class<? extends Address>> getSupportedAddressClasses() {
        return super.getSupportedAddressClasses();
    }

    @Override
    public UnixDomainAddress getListenAddress() {
        return listenAddress;
    }

    @Override
    public void sendMessage(UnixDomainAddress address, byte[] message, TransportStateReference tmStateReference,
                            long timeoutMillis, int maxRetries) throws IOException {
        if (isTransportAudit() && !fireTransportAuditEvent(new TransportAuditEvent(this,
                TransportAuditEvent.TransportOperation.SEND, address, getListenAddress(), message))) {
            if (logger.isDebugEnabled()) {
                logger.debug("Transport audit denied sending message to " + address);
            }
            return;
        }
        if (getListenWorkerTask() == null || serverThread == null) {
//            if (isOpenSocketOnSending()) {
                listen();
//            }
//            else {
//                handleDroppedMessageToSend(address, message, tmStateReference, timeoutMillis, maxRetries);
//            }
        }
        if (serverThread != null) {
            if ((!suspendedAddresses.isEmpty()) && suspendedAddresses.contains(address)) {
                handleDroppedMessageToSend(address, message, tmStateReference, timeoutMillis, maxRetries);
            }
            else {
                serverThread.sendMessage(address, message, tmStateReference, sockets);
            }
        }
    }

    @Override
    public void close() throws IOException {
        if (isTransportAudit()) {
            fireTransportAuditEvent(new TransportAuditEvent(this,
                    TransportAuditEvent.TransportOperation.CLOSE, null, getListenAddress(), null));
        }
        super.close();
        removeBoundSocketFile();
    }

    /**
     * Removes the socket file that this transport mapping has created, if any. A socket file that has not
     * been created by this transport mapping is never removed, because it may belong to another SNMP entity
     * that is listening on the same path.
     *
     * @throws IOException
     *    if the socket file exists but could not be removed.
     * @since 1.2.2
     */
    protected void removeBoundSocketFile() throws IOException {
        Path bound = boundSocketPath;
        if (bound != null) {
            boundSocketPath = null;
            Files.deleteIfExists(bound);
            if (logger.isDebugEnabled()) {
                logger.debug("Removed Unix domain socket file " + bound);
            }
        }
    }

    @Override
    public boolean close(UnixDomainAddress remoteAddress) throws IOException {
        AbstractSocketEntry<UnixDomainAddress> remoteSocketEntry = sockets.remove(remoteAddress);
        if (remoteSocketEntry == null) {
            return false;
        }
        remoteSocketEntry.getSocketChannel().close();
        return true;
    }

    @Override
    public void listen() throws IOException {
        if (getListenWorkerTask() != null) {
            throw new SocketException("Port already listening");
        }
        if (isTransportAudit() && !fireTransportAuditEvent(new TransportAuditEvent(this,
                TransportAuditEvent.TransportOperation.LISTEN, null, getListenAddress(), null))) {
            if (logger.isDebugEnabled()) {
                logger.debug("Transport audit denied listening on " + getListenAddress());
            }
            return;
        }
        serverThread = new ServerThread(listenAddress);
        if (logger.isInfoEnabled()) {
            if (serverThread.getServerSocketChannel() == null) {
                // Client mode or no listen address: outgoing connections only, nothing has been bound.
                logger.info("Unix socket transport mapping started for outgoing connections only, " +
                        "listen address " + getListenAddress() + " has not been bound");
            }
            else {
                logger.info("Unix socket address " + getListenAddress() + " bound successfully");
            }
        }
        listenWorkerTask = SNMP4JSettings.getThreadFactory().createWorkerThread(
                "UnixSocketStreamTransportMapping_" + getListenAddress(), serverThread, true);
        if (getConnectionTimeout() > 0) {
            // run as daemon
            socketCleaner = SNMP4JSettings.getTimerFactory().createTimer();
        }
        getListenWorkerTask().run();
    }

    @Override
    public boolean isListening() {
        return serverThread != null && !serverThread.isStop();
    }

    @Override
    public void wakeupServerSelector() {
        serverThread.getSelector().wakeup();
    }

    private static void postBind(UnixDomainAddress address) {
        address.getPath().toFile().deleteOnExit();
    }

    public static class SocketEntry extends AbstractSocketEntry<UnixDomainAddress> {

        public SocketEntry(UnixDomainAddress address, SocketChannel socket) {
            super(address, socket);
            // with Unix Domain sockets there is no handshaking, thus we are finished if connected
            setHandshakeFinished(true);
        }

        @Override
        public String toString() {
            return getSocketChannel().toString();
        }

        @Override
        public void closeSession() {
            try {
                socketChannel.close();
            } catch (IOException e) {
                throw new UncheckedIOException(e);
            }
        }

        @Override
        public Object getSessionID() {
            return socketChannel;
        }
    }

    /**
     * Read the next network packet for the application.
     *
     * @param socketChannel
     *         the socket channel to read the data from.
     * @return a byte buffer with the application data of the packet.
     * @throws IOException
     *         if there occurs an IO exception during preparation.
     * @since 3.7
     */
    protected Optional<ByteBuffer> readNextInPacket(SocketChannel socketChannel)
            throws IOException {
        ByteBuffer buf = ByteBuffer.allocate(getMaxInboundMessageSize());
        int bytesRead = socketChannel.read(buf);
        if (bytesRead < 0) {
            return Optional.empty();
        }
        byte[] bytes = new byte[bytesRead];
        buf.flip();
        buf.get(bytes);
        return Optional.of(ByteBuffer.wrap(bytes));
    }

    protected class ServerThread extends AbstractTransportServerThread<UnixDomainAddress, SocketEntry> {

        private ServerThread(UnixDomainAddress unixDomainAddress) throws IOException {
            super(UnixSocketStreamTransportMapping.this, unixDomainAddress);
            // The selector has already been opened by the super constructor.
            initListener(unixDomainAddress, selector);
        }

        public Throwable getLastError() {
            return lastError;
        }

        public boolean isStop() {
            return stop;
        }

        public void setStop(boolean stop) {
            this.stop = stop;
        }

        protected ServerSocketChannel getServerSocketChannel() {
            return ssc;
        }

        protected boolean initListener(UnixDomainAddress address, Selector selector) throws IOException {
            if (!isServerEnabled()) {
                // Client mode: this transport mapping only opens outgoing connections, thus its socket path
                // must not be bound. Binding it anyway would occupy the path of another transport mapping
                // that serves incoming connections on it - unlike an IP based transport, which can fall back
                // to an ephemeral port, a Unix domain socket path is the address itself, so the bind fails
                // with "Address already in use" as soon as both exist. The socket path is bound to the
                // outgoing socket instead, see bindOutgoingSocket(SocketChannel).
                if (logger.isDebugEnabled()) {
                    logger.debug("Server is disabled, not listening on Unix socket address " + address);
                }
                return false;
            }
            if (address != null) {
                /*
                if (transportAuditListeners != null) {
                    if (!fireTransportAuditEvent(
                            new TransportAuditEvent(this, TransportAuditEvent.TransportOperation.LISTEN,
                                    null, address, null))) {
                        return false;
                    }
                }
                */
                ProtocolFamily family = address.getFamily();
                SocketAddress socketAddress = address.getSocketAddress();
                if (socketAddress != null) {
                    ssc = ServerSocketChannel.open(family);
                    ssc.bind(socketAddress);
                    boundSocketPath = address.getPath();
                    ssc.configureBlocking(false);
                    postBind(address);
                    ssc.register(selector, SelectionKey.OP_ACCEPT, null);
                    logger.info("Unix Domain Server listening on " + ssc.getLocalAddress());
                    return true;
                }
            }
            return false;
        }

        @Override
        public void run() {
            doServer(sockets);
        }

        @Override
        protected boolean readMessage(SelectionKey sk, SocketChannel socketChannel,
                                      UnixDomainAddress incomingAddress, SocketEntry socketEntry) throws IOException {
            TransportStateReference tmStateReference =
                    new TransportStateReference(UnixSocketStreamTransportMapping.this,
                            listenAddress, null,
                            SecurityLevel.undefined, SecurityLevel.undefined,
                            false, socketChannel);
            var byteBufferRead = readNextInPacket(socketChannel);
            byteBufferRead.ifPresent(message ->
                    fireProcessMessage(new UnixDomainAddress((java.net.UnixDomainSocketAddress)
                            incomingAddress.getSocketAddress()), message, tmStateReference));
            if (byteBufferRead.isEmpty()) {
                String remote = socketChannel.getRemoteAddress().toString();
                if (logger.isInfoEnabled()) {
                    logger.info("UnixDomainServer: closing connection from: " + remote + " with ID "+ socketEntry);
                }
                socketChannel.close();
                return false;
            }
            return true;
        }

        @Override
        protected void processQueues() {
            // nothing to do
        }

        @Override
        public SocketEntry createSocketEntry(UnixDomainAddress address, SocketChannel socketChannel,
                                             boolean useClientMode, TransportStateReference tmStateReference) {
            if (!useClientMode && ((address == null) || !address.isValid())) {
                // The peer did not bind its socket before connecting, so it has no address at all. All such
                // peers share the same (empty) address and therefore the same entry in the socket map of
                // this transport mapping, which means that only the connection that arrived last can be
                // addressed. See bindOutgoingSocket(SocketChannel) for the client side of this.
                logger.warn("Accepted a connection from a Unix domain socket that is not bound to a socket " +
                        "path, thus this peer cannot be told apart from other peers that connect without " +
                        "binding their socket: " + socketChannel);
            }
            return new SocketEntry(address, socketChannel);
        }

        @Override
        protected SocketChannel openSocketChannel(ProtocolFamily family) throws IOException {
            SocketChannel socketChannel = SocketChannel.open(family);
            try {
                bindOutgoingSocket(socketChannel);
            } catch (IOException iox) {
                socketChannel.close();
                throw iox;
            }
            return socketChannel;
        }

        /**
         * Binds an outgoing socket to the socket path of this transport mapping, so that the peer this
         * socket connects to can tell this SNMP entity apart from other entities that connect to it. A Unix
         * domain socket that has not been bound has no address at all, thus every peer that connects without
         * binding its socket is reported with the same empty address by the accepting entity - which then
         * cannot address more than one of them.
         * <p>
         * The socket is bound in client mode only: in server mode the socket path is already bound by the
         * server socket that accepts incoming connections and a path can be bound once only. If the path
         * cannot be bound - for example because another process is using it - the connection is opened with
         * an unbound socket, because being unidentifiable to the peer is better than not being able to
         * communicate with it at all.
         *
         * @param socketChannel
         *    the outgoing socket to bind, must not be connected yet.
         * @throws IOException
         *    if a socket file created by this transport mapping for an earlier connection could not be
         *    removed.
         * @since 1.2.2
         */
        protected void bindOutgoingSocket(SocketChannel socketChannel) throws IOException {
            UnixDomainAddress localAddress = listenAddress;
            if (isServerEnabled() || (localAddress == null) || !localAddress.isValid()) {
                return;
            }
            Path socketPath = localAddress.getPath();
            if (socketPath.equals(boundSocketPath)) {
                // The socket file of an earlier connection of this transport mapping is still there: closing
                // a Unix domain socket does not remove its socket file, so it has to be removed here to be
                // able to bind the path again, for example when a connection is re-established.
                boundSocketPath = null;
                Files.deleteIfExists(socketPath);
            }
            try {
                socketChannel.bind(localAddress.getSocketAddress());
                boundSocketPath = socketPath;
                postBind(localAddress);
                if (logger.isDebugEnabled()) {
                    logger.debug("Bound outgoing Unix domain socket to " + socketChannel.getLocalAddress());
                }
            } catch (IOException iox) {
                logger.warn("Could not bind outgoing Unix domain socket to '" + socketPath + "', connecting " +
                        "with an unbound socket that the peer cannot identify: " + iox.getMessage());
            }
        }

        @Override
        public SocketEntry removeSocketEntry(UnixDomainAddress incomingAddress) {
            return sockets.remove(incomingAddress);
        }

        @Override
        protected UnixDomainAddress createIncomingAddress(SocketChannel socketChannel) throws IOException {
            return new UnixDomainAddress((UnixDomainSocketAddress) socketChannel.getRemoteAddress());
        }

        @Override
        public void shutdown() {
            shutdown = true;
            if (logger.isDebugEnabled()) {
                logger.debug("UnixDomainServer: shutting down");
            }
        }

        @Override
        public boolean join(long timeout) throws InterruptedException {
            shutdown();
            return false;
        }
    }
}
