/*
 * GRAKN.AI - THE KNOWLEDGE GRAPH
 * Copyright (C) 2018 Grakn Labs Ltd
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */

package ai.grakn.core.server.deduplicator;

import ai.grakn.GraknConfigKey;
import ai.grakn.Keyspace;
import ai.grakn.concept.ConceptId;
import ai.grakn.core.server.GraknConfig;
import ai.grakn.core.server.deduplicator.queue.Attribute;
import ai.grakn.core.server.deduplicator.queue.RocksDbQueue;
import ai.grakn.core.server.factory.EngineGraknTxFactory;
import ai.grakn.kb.internal.EmbeddedGraknTx;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

import static ai.grakn.core.server.deduplicator.AttributeDeduplicator.deduplicate;

/**
 * This class is responsible for de-duplicating attributes. It is done to ensure that every attribute in Grakn stays unique.
 *
 * Marking an attribute for deduplication:
 * When the {@link EmbeddedGraknTx#commit()} is invoked, it will trigger the {@link #markForDeduplication(Keyspace, String, ConceptId)}
 * which inserts the attribute to an internal queue for deduplication.
 *
 * De-duplicating attributes in the de-duplicator daemon:
 * The de-duplicator daemon is an always-on background thread which performs deduplication on incoming attributes.
 * When a new attribute is inserted, it will immediately trigger the deduplicate operation, meaning that duplicates are merged in almost real-time speed.
 * The daemon is started and stopped with {@link #startDeduplicationDaemon()} and {@link #stopDeduplicationDaemon()}
 *
 * Fault tolerance:
 * The de-duplicator daemon is fault-tolerant and will re-process incoming attributes if Grakn crashes in the middle of a deduplication.
 *
 * @author Ganeshwara Herawan Hananda
 */
public class AttributeDeduplicatorDaemon {
    private static Logger LOG = LoggerFactory.getLogger(AttributeDeduplicatorDaemon.class);
    private static final int QUEUE_GET_BATCH_MAX = 1000;
    private static final Path queueDataDirRelative = Paths.get("queue"); // path to the queue storage location, relative to the data directory

    private EngineGraknTxFactory txFactory;
    private RocksDbQueue queue;

    private boolean stopDaemon = false;

    /**
     * Instantiates {@link AttributeDeduplicatorDaemon}
     * @param config a reference to an instance of {@link GraknConfig} which is initialised from a grakn.properties.
     * @param txFactory an {@link EngineGraknTxFactory} instance which provides access to write into the database
     */
    public AttributeDeduplicatorDaemon(GraknConfig config, EngineGraknTxFactory txFactory) {
        Path dataDir = Paths.get(config.getProperty(GraknConfigKey.DATA_DIR));
        Path queueDataDir = dataDir.resolve(queueDataDirRelative);
        this.queue = new RocksDbQueue(queueDataDir);
        this.txFactory = txFactory;
    }

    /**
     * Marks an attribute for deduplication. The attribute will be inserted to an internal queue to be processed by the de-duplicator daemon in real-time.
     * The attribute must have been inserted to the database, prior to calling this method.
     *
     * @param keyspace keyspace of the attribute
     * @param index the value of the attribute
     * @param conceptId the concept id of the attribute
     */
    public void markForDeduplication(Keyspace keyspace, String index, ConceptId conceptId) {
        Attribute attribute = Attribute.create(keyspace, index, conceptId);
        LOG.trace("insert(" + attribute + ")");
        queue.insert(attribute);
    }

    /**
     * Starts a daemon which performs deduplication on incoming attributes in real-time.
     * The thread listens to the {@link RocksDbQueue} queue for incoming attributes and applies
     * the {@link AttributeDeduplicator#deduplicate(EngineGraknTxFactory, KeyspaceIndexPair)} algorithm.
     *
     */
    public CompletableFuture<Void> startDeduplicationDaemon() {
        stopDaemon = false;
        CompletableFuture<Void> daemon = CompletableFuture.supplyAsync(() -> {
            LOG.info("startDeduplicationDaemon() - attribute de-duplicator daemon started.");
            while (!stopDaemon) {
                try {
                    List<Attribute> attributes = queue.read(QUEUE_GET_BATCH_MAX);

                    LOG.trace("starting a new batch to process these new attributes: " + attributes);

                    // group the attributes into a set of unique (keyspace -> value) pair
                    Set<KeyspaceIndexPair> uniqueKeyValuePairs = attributes.stream()
                            .map(attr -> KeyspaceIndexPair.create(attr.keyspace(), attr.index()))
                            .collect(Collectors.toSet());

                    // perform deduplicate for each (keyspace -> value)
                    for (KeyspaceIndexPair keyspaceIndexPair : uniqueKeyValuePairs) {
                        deduplicate(txFactory, keyspaceIndexPair);
                    }

                    LOG.trace("new attributes processed.");

                    queue.ack(attributes);
                }
                catch (InterruptedException | RuntimeException e) {
                    LOG.error("An exception has occurred in the attribute de-duplicator daemon. ", e);
                }
            }
            LOG.info("startDeduplicationDaemon() - attribute de-duplicator daemon stopped");
            return null;
        });

        daemon.exceptionally(e -> {
            LOG.error("An unhandled exception has occurred in the attribute de-duplicator daemon. ", e);
            return null;
        });

        return daemon;
    }

    /**
     * Stops the attribute uniqueness daemon
     */
    public void stopDeduplicationDaemon() {
        LOG.info("stopDeduplicationDaemon() - stopping the attribute de-duplicator daemon...");
        stopDaemon = true;
    }
}


@@ -1143,16 +1143,79 @@
 raknTx;%0A
+import com.google.common.util.concurrent.ThreadFactoryBuilder;%0A
 import o
@@ -1411,16 +1411,100 @@
 Future;%0A
+import java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors;%0A
 import j
@@ -2986,16 +2986,193 @@
 ectory%0A%0A
+    private ExecutorService executorServiceForDaemon = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat(%22attribute-deduplicator-daemon-%25d%22).build());%0A%0A
     priv
@@ -6442,32 +6442,58 @@
  null;%0A        %7D
+, executorServiceForDaemon
 );%0A%0A        daem

Update executor service name used by attribute deduplicator daemon (#4559)