/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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.eclipse.aether.named.ipc;

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.ByteChannel;
import java.nio.channels.Channels;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Implementation of the server side.
 * The server instance is bound to a given maven repository.
 *
 * @since 2.0.1
 */
public class IpcServer {
    /**
     * Should the IPC server not fork? (i.e. for testing purposes)
     *
     * @configurationSource {@link System#getProperty(String, String)}
     * @configurationType {@link java.lang.Boolean}
     * @configurationDefaultValue {@link #DEFAULT_NO_FORK}
     */
    public static final String SYSTEM_PROP_NO_FORK = "aether.named.ipc.nofork";

    public static final boolean DEFAULT_NO_FORK = false;

    /**
     * IPC idle timeout in seconds. If there is no IPC request during idle time, it will stop.
     *
     * @configurationSource {@link System#getProperty(String, String)}
     * @configurationType {@link java.lang.Integer}
     * @configurationDefaultValue {@link #DEFAULT_IDLE_TIMEOUT}
     */
    public static final String SYSTEM_PROP_IDLE_TIMEOUT = "aether.named.ipc.idleTimeout";

    public static final int DEFAULT_IDLE_TIMEOUT = 60;

    /**
     * IPC socket family to use.
     *
     * @configurationSource {@link System#getProperty(String, String)}
     * @configurationType {@link java.lang.String}
     * @configurationDefaultValue {@link #DEFAULT_FAMILY}
     */
    public static final String SYSTEM_PROP_FAMILY = "aether.named.ipc.family";

    public static final String DEFAULT_FAMILY = "unix";

    /**
     * Should the IPC server not use native executable?
     *
     * @configurationSource {@link System#getProperty(String, String)}
     * @configurationType {@link java.lang.Boolean}
     * @configurationDefaultValue {@link #DEFAULT_NO_NATIVE}
     */
    public static final String SYSTEM_PROP_NO_NATIVE = "aether.named.ipc.nonative";

    public static final boolean DEFAULT_NO_NATIVE = true;

    /**
     * The name if the IPC server native executable (without file extension like ".exe")
     *
     * @configurationSource {@link System#getProperty(String, String)}
     * @configurationType {@link java.lang.String}
     * @configurationDefaultValue {@link #DEFAULT_NATIVE_NAME}
     */
    public static final String SYSTEM_PROP_NATIVE_NAME = "aether.named.ipc.nativeName";

    public static final String DEFAULT_NATIVE_NAME = "ipc-sync";

    /**
     * Should the IPC server log debug messages? (i.e. for testing purposes)
     *
     * @configurationSource {@link System#getProperty(String, String)}
     * @configurationType {@link java.lang.Boolean}
     * @configurationDefaultValue {@link #DEFAULT_DEBUG}
     */
    public static final String SYSTEM_PROP_DEBUG = "aether.named.ipc.debug";

    public static final boolean DEFAULT_DEBUG = false;

    private final ServerSocketChannel serverSocket;
    private final Map<SocketChannel, Thread> clients = new HashMap<>();
    private final AtomicInteger counter = new AtomicInteger();
    private final Map<String, Lock> locks = new ConcurrentHashMap<>();
    private final Map<String, Context> contexts = new ConcurrentHashMap<>();
    private static final boolean DEBUG =
            Boolean.parseBoolean(System.getProperty(SYSTEM_PROP_DEBUG, Boolean.toString(DEFAULT_DEBUG)));
    private final long idleTimeout;
    private volatile long lastUsed;
    private volatile boolean closing;

    public IpcServer(SocketFamily family) throws IOException {
        serverSocket = family.openServerSocket();
        long timeout = TimeUnit.SECONDS.toNanos(DEFAULT_IDLE_TIMEOUT);
        String str = System.getProperty(SYSTEM_PROP_IDLE_TIMEOUT);
        if (str != null) {
            try {
                TimeUnit unit = TimeUnit.SECONDS;
                if (str.endsWith("ms")) {
                    unit = TimeUnit.MILLISECONDS;
                    str = str.substring(0, str.length() - 2);
                }
                long dur = Long.parseLong(str);
                timeout = unit.toNanos(dur);
            } catch (NumberFormatException e) {
                error("Property " + SYSTEM_PROP_IDLE_TIMEOUT + " specified with invalid value: " + str, e);
            }
        }
        idleTimeout = timeout;
    }

    public static void main(String[] args) throws Exception {
        // When spawning a new process, the child process is create within
        // the same process group.  This means that a few signals are sent
        // to the whole group.  This is the case for SIGINT (Ctrl-C) and
        // SIGTSTP (Ctrl-Z) which are both sent to all the processed in the
        // group when initiated from the controlling terminal.
        // This is only a problem when the client creates the daemon, but
        // without ignoring those signals, a client being interrupted will
        // also interrupt and kill the daemon.
        try {
            sun.misc.Signal.handle(new sun.misc.Signal("INT"), sun.misc.SignalHandler.SIG_IGN);
            if (IpcClient.IS_WINDOWS) {
                sun.misc.Signal.handle(new sun.misc.Signal("TSTP"), sun.misc.SignalHandler.SIG_IGN);
            }
        } catch (Throwable t) {
            error("Unable to ignore INT and TSTP signals", t);
        }

        String family = args[0];
        String tmpAddress = args[1];
        String rand = args[2];

        runServer(SocketFamily.valueOf(family), tmpAddress, rand);
    }

    static IpcServer runServer(SocketFamily family, String tmpAddress, String rand) throws IOException {
        IpcServer server = new IpcServer(family);
        run(server::run, false); // this is one-off
        String address = SocketFamily.toString(server.getLocalAddress());
        SocketAddress socketAddress = SocketFamily.fromString(tmpAddress);
        try (SocketChannel socket = SocketChannel.open(socketAddress)) {
            try (DataOutputStream dos = new DataOutputStream(Channels.newOutputStream(socket))) {
                dos.writeUTF(rand);
                dos.writeUTF(address);
                dos.flush();
            }
        }

        return server;
    }

    private static void debug(String msg, Object... args) {
        if (DEBUG) {
            System.out.printf("[ipc] [debug] " + msg + "\n", args);
        }
    }

    private static void info(String msg, Object... args) {
        System.out.printf("[ipc] [info] " + msg + "\n", args);
    }

    private static void error(String msg, Throwable t) {
        System.out.println("[ipc] [error] " + msg);
        t.printStackTrace(System.out);
    }

    private static void run(Runnable runnable, boolean daemon) {
        Thread thread = new Thread(runnable);
        if (daemon) {
            thread.setDaemon(true);
        }
        thread.start();
    }

    public SocketAddress getLocalAddress() throws IOException {
        return serverSocket.getLocalAddress();
    }

    public void run() {
        try {
            info("IpcServer started at %s", getLocalAddress().toString());
            use();
            run(this::expirationCheck, true);
            while (!closing) {
                SocketChannel socket = this.serverSocket.accept();
                run(() -> client(socket), false);
            }
        } catch (Throwable t) {
            if (!closing) {
                error("Error running sync server loop", t);
            }
        }
    }

    private void client(SocketChannel socket) {
        int c;
        synchronized (clients) {
            clients.put(socket, Thread.currentThread());
            c = clients.size();
        }
        info("New client connected (%d connected)", c);
        use();
        Map<String, Context> clientContexts = new ConcurrentHashMap<>();
        try {
            ByteChannel wrapper = new ByteChannelWrapper(socket);
            DataInputStream input = new DataInputStream(Channels.newInputStream(wrapper));
            DataOutputStream output = new DataOutputStream(Channels.newOutputStream(wrapper));
            while (!closing) {
                int requestId = input.readInt();
                int sz = input.readInt();
                List<String> request = new ArrayList<>(sz);
                for (int i = 0; i < sz; i++) {
                    request.add(input.readUTF());
                }
                if (request.isEmpty()) {
                    throw new IOException("Received invalid request");
                }
                use();
                String contextId;
                Context context;
                String command = request.remove(0);
                switch (command) {
                    case IpcMessages.REQUEST_CONTEXT:
                        if (request.size() != 1) {
                            throw new IOException("Expected one argument for " + command + " but got " + request);
                        }
                        boolean shared = Boolean.parseBoolean(request.remove(0));
                        context = new Context(shared);
                        contexts.put(context.id, context);
                        clientContexts.put(context.id, context);
                        synchronized (output) {
                            debug("Created context %s", context.id);
                            output.writeInt(requestId);
                            output.writeInt(2);
                            output.writeUTF(IpcMessages.RESPONSE_CONTEXT);
                            output.writeUTF(context.id);
                            output.flush();
                        }
                        break;
                    case IpcMessages.REQUEST_ACQUIRE:
                        if (request.size() < 1) {
                            throw new IOException(
                                    "Expected at least one argument for " + command + " but got " + request);
                        }
                        contextId = request.remove(0);
                        context = contexts.get(contextId);
                        if (context == null) {
                            throw new IOException(
                                    "Unknown context: " + contextId + ". Known contexts = " + contexts.keySet());
                        }
                        context.lock(request).thenRun(() -> {
                            try {
                                synchronized (output) {
                                    debug("Locking in context %s", context.id);
                                    output.writeInt(requestId);
                                    output.writeInt(1);
                                    output.writeUTF(IpcMessages.RESPONSE_ACQUIRE);
                                    output.flush();
                                }
                            } catch (IOException e) {
                                try {
                                    socket.close();
                                } catch (IOException ioException) {
                                    e.addSuppressed(ioException);
                                }
                                error("Error writing lock response", e);
                            }
                        });
                        break;
                    case IpcMessages.REQUEST_CLOSE:
                        if (request.size() != 1) {
                            throw new IOException("Expected one argument for " + command + " but got " + request);
                        }
                        contextId = request.remove(0);
                        context = contexts.remove(contextId);
                        clientContexts.remove(contextId);
                        if (context == null) {
                            throw new IOException(
                                    "Unknown context: " + contextId + ". Known contexts = " + contexts.keySet());
                        }
                        context.unlock();
                        synchronized (output) {
                            debug("Closing context %s", context.id);
                            output.writeInt(requestId);
                            output.writeInt(1);
                            output.writeUTF(IpcMessages.RESPONSE_CLOSE);
                            output.flush();
                        }
                        break;
                    case IpcMessages.REQUEST_STOP:
                        if (request.size() != 0) {
                            throw new IOException("Expected zero argument for " + command + " but got " + request);
                        }
                        synchronized (output) {
                            debug("Stopping server");
                            output.writeInt(requestId);
                            output.writeInt(1);
                            output.writeUTF(IpcMessages.RESPONSE_STOP);
                            output.flush();
                        }
                        close();
                        break;
                    default:
                        throw new IOException("Unknown request: " + request.get(0));
                }
            }
        } catch (Throwable t) {
            if (!closing) {
                error("Error processing request", t);
            }
        } finally {
            if (!closing) {
                info("Client disconnecting...");
            }
            clientContexts.values().forEach(context -> {
                contexts.remove(context.id);
                context.unlock();
            });
            try {
                socket.close();
            } catch (IOException ioException) {
                // ignore
            }
            synchronized (clients) {
                clients.remove(socket);
                c = clients.size();
            }
            if (!closing) {
                info("%d clients left", c);
            }
        }
    }

    private void use() {
        lastUsed = System.nanoTime();
    }

    private void expirationCheck() {
        while (true) {
            long current = System.nanoTime();
            long left = (lastUsed + idleTimeout) - current;
            if (left < 0) {
                info("IpcServer expired, closing");
                close();
                break;
            } else {
                try {
                    Thread.sleep(TimeUnit.NANOSECONDS.toMillis(left));
                } catch (InterruptedException e) {
                    info("IpcServer expiration check interrupted, closing");
                    close();
                    break;
                }
            }
        }
    }

    void close() {
        closing = true;
        try {
            serverSocket.close();
        } catch (IOException e) {
            error("Error closing server socket", e);
        }
        clients.forEach((s, t) -> {
            try {
                s.close();
            } catch (IOException e) {
                // ignore
            }
            t.interrupt();
        });
    }

    static class Waiter {
        final Context context;
        final CompletableFuture<Void> future;

        Waiter(Context context, CompletableFuture<Void> future) {
            this.context = context;
            this.future = future;
        }
    }

    static class Lock {

        final String key;

        List<Context> holders;
        List<Waiter> waiters;

        Lock(String key) {
            this.key = key;
        }

        public synchronized CompletableFuture<Void> lock(Context context) {
            if (holders == null) {
                holders = new ArrayList<>();
            }
            if (holders.isEmpty() || holders.get(0).shared && context.shared) {
                holders.add(context);
                return CompletableFuture.completedFuture(null);
            }
            if (waiters == null) {
                waiters = new ArrayList<>();
            }

            CompletableFuture<Void> future = new CompletableFuture<>();
            waiters.add(new Waiter(context, future));
            return future;
        }

        public synchronized void unlock(Context context) {
            if (holders.remove(context)) {
                while (waiters != null
                        && !waiters.isEmpty()
                        && (holders.isEmpty() || holders.get(0).shared && waiters.get(0).context.shared)) {
                    Waiter waiter = waiters.remove(0);
                    holders.add(waiter.context);
                    waiter.future.complete(null);
                }
            } else if (waiters != null) {
                for (Iterator<Waiter> it = waiters.iterator(); it.hasNext(); ) {
                    Waiter waiter = it.next();
                    if (waiter.context == context) {
                        it.remove();
                        waiter.future.cancel(false);
                    }
                }
            }
        }
    }

    class Context {

        final String id;
        final boolean shared;
        final List<String> locks = new CopyOnWriteArrayList<>();

        Context(boolean shared) {
            this.id = String.format("%08x", counter.incrementAndGet());
            this.shared = shared;
        }

        public CompletableFuture<?> lock(List<String> keys) {
            locks.addAll(keys);
            CompletableFuture<?>[] futures = keys.stream()
                    .map(k -> IpcServer.this.locks.computeIfAbsent(k, Lock::new))
                    .map(l -> l.lock(this))
                    .toArray(CompletableFuture[]::new);
            return CompletableFuture.allOf(futures);
        }

        public void unlock() {
            locks.stream()
                    .map(k -> IpcServer.this.locks.computeIfAbsent(k, Lock::new))
                    .forEach(l -> l.unlock(this));
        }
    }
}