/*
 * 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.apache.jackrabbit.oak.plugins.document;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.collect.Lists.newArrayList;
import static org.apache.jackrabbit.oak.commons.IOUtils.closeQuietly;
import static org.apache.jackrabbit.oak.commons.PropertiesUtil.toLong;
import static org.apache.jackrabbit.oak.plugins.document.DocumentNodeStoreBuilder.DEFAULT_MEMORY_CACHE_SIZE;
import static org.apache.jackrabbit.oak.plugins.document.NodeDocument.MODIFIED_IN_SECS_RESOLUTION;
import static org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentNodeStoreBuilder.newMongoDocumentNodeStoreBuilder;
import static org.apache.jackrabbit.oak.plugins.document.rdb.RDBDocumentNodeStoreBuilder.newRDBDocumentNodeStoreBuilder;
import static org.apache.jackrabbit.oak.spi.blob.osgi.SplitBlobStoreService.ONLY_STANDALONE_TARGET;
import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.registerMBean;
import static org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardUtils.scheduleWithFixedDelay;

import java.io.Closeable;
import java.io.IOException;
import java.text.ParseException;
import java.util.Arrays;
import java.util.Collections;
import java.util.Dictionary;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

import javax.sql.DataSource;

import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.base.Strings;
import com.google.common.io.Closer;
import com.google.common.util.concurrent.UncheckedExecutionException;
import com.mongodb.MongoClientURI;

import org.apache.commons.io.FilenameUtils;
import org.apache.jackrabbit.commons.SimpleValueFactory;
import org.apache.jackrabbit.oak.api.Descriptors;
import org.apache.jackrabbit.oak.api.jmx.CacheStatsMBean;
import org.apache.jackrabbit.oak.api.jmx.CheckpointMBean;
import org.apache.jackrabbit.oak.api.jmx.PersistentCacheStatsMBean;
import org.apache.jackrabbit.oak.cache.CacheStats;
import org.apache.jackrabbit.oak.plugins.document.VersionGarbageCollector.VersionGCStats;
import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentNodeStoreBuilder;
import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore;
import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStoreMetrics;
import org.apache.jackrabbit.oak.plugins.document.rdb.RDBDocumentNodeStoreBuilder;
import org.apache.jackrabbit.oak.plugins.document.util.Utils;
import org.apache.jackrabbit.oak.spi.commit.ObserverTracker;
import org.apache.jackrabbit.oak.osgi.OsgiWhiteboard;
import org.apache.jackrabbit.oak.plugins.blob.BlobGC;
import org.apache.jackrabbit.oak.plugins.blob.BlobGCMBean;
import org.apache.jackrabbit.oak.plugins.blob.BlobGarbageCollector;
import org.apache.jackrabbit.oak.plugins.blob.BlobStoreStats;
import org.apache.jackrabbit.oak.plugins.blob.BlobTrackingStore;
import org.apache.jackrabbit.oak.plugins.blob.SharedDataStore;
import org.apache.jackrabbit.oak.plugins.blob.datastore.BlobIdTracker;
import org.apache.jackrabbit.oak.plugins.blob.datastore.SharedDataStoreUtils;
import org.apache.jackrabbit.oak.plugins.document.persistentCache.PersistentCacheStats;
import org.apache.jackrabbit.oak.plugins.document.util.MongoConnection;
import org.apache.jackrabbit.oak.plugins.document.util.SystemPropertySupplier;
import org.apache.jackrabbit.oak.spi.cluster.ClusterRepositoryInfo;
import org.apache.jackrabbit.oak.spi.blob.BlobStore;
import org.apache.jackrabbit.oak.spi.blob.BlobStoreWrapper;
import org.apache.jackrabbit.oak.spi.blob.GarbageCollectableBlobStore;
import org.apache.jackrabbit.oak.spi.blob.stats.BlobStoreStatsMBean;
import org.apache.jackrabbit.oak.spi.commit.BackgroundObserverMBean;
import org.apache.jackrabbit.oak.spi.gc.DelegatingGCMonitor;
import org.apache.jackrabbit.oak.spi.gc.GCMonitor;
import org.apache.jackrabbit.oak.spi.gc.GCMonitorTracker;
import org.apache.jackrabbit.oak.spi.gc.LoggingGCMonitor;
import org.apache.jackrabbit.oak.spi.state.Clusterable;
import org.apache.jackrabbit.oak.spi.state.NodeStore;
import org.apache.jackrabbit.oak.spi.state.NodeStoreProvider;
import org.apache.jackrabbit.oak.spi.state.RevisionGC;
import org.apache.jackrabbit.oak.spi.state.RevisionGCMBean;
import org.apache.jackrabbit.oak.spi.whiteboard.AbstractServiceTracker;
import org.apache.jackrabbit.oak.spi.whiteboard.Registration;
import org.apache.jackrabbit.oak.spi.whiteboard.Whiteboard;
import org.apache.jackrabbit.oak.spi.whiteboard.WhiteboardExecutor;
import org.apache.jackrabbit.oak.stats.StatisticsProvider;
import org.apache.jackrabbit.oak.spi.descriptors.GenericDescriptors;
import org.jetbrains.annotations.NotNull;
import org.osgi.framework.Bundle;
import org.osgi.framework.BundleException;
import org.osgi.framework.Constants;
import org.osgi.framework.ServiceRegistration;
import org.osgi.service.cm.ConfigurationAdmin;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.component.annotations.Activate;
import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.ConfigurationPolicy;
import org.osgi.service.component.annotations.Deactivate;
import org.osgi.service.component.annotations.Reference;
import org.osgi.service.component.annotations.ReferenceCardinality;
import org.osgi.service.component.annotations.ReferencePolicy;
import org.quartz.CronExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The OSGi service to start/stop a DocumentNodeStore instance.
 */
@Component(
        configurationPolicy = ConfigurationPolicy.REQUIRE,
        configurationPid = {Configuration.PID})
public class DocumentNodeStoreService {

    private static final long MB = 1024 * 1024;
    static final String DEFAULT_URI = "mongodb://localhost:27017/oak";
    static final int DEFAULT_CACHE = (int) (DEFAULT_MEMORY_CACHE_SIZE / MB);
    static final int DEFAULT_BLOB_CACHE_SIZE = 16;
    static final String DEFAULT_DB = "oak";
    static final boolean DEFAULT_SO_KEEP_ALIVE = true;
    static final String DEFAULT_PERSISTENT_CACHE = "cache";
    static final String DEFAULT_JOURNAL_CACHE = "diff-cache";
    static final boolean DEFAULT_CUSTOM_BLOB_STORE = false;
    public static final String CONTINUOUS_RGC_EXPR = "*/5 * * * * ?";
    public static final String CLASSIC_RGC_EXPR = "0 0 2 * * ?";
    public static final long DEFAULT_RGC_TIME_LIMIT_SECS = 3*60*60; // default is 3 hours
    public static final double DEFAULT_RGC_DELAY_FACTOR = 0;
    private static final String DESCRIPTION = "oak.nodestore.description";
    static final long DEFAULT_JOURNAL_GC_INTERVAL_MILLIS = 5*60*1000; // default is 5min
    static final long DEFAULT_JOURNAL_GC_MAX_AGE_MILLIS = 24*60*60*1000; // default is 24hours
    static final boolean DEFAULT_PREFETCH_EXTERNAL_CHANGES = false;
    private static final String DEFAULT_PROP_HOME = "./repository";
    static final long DEFAULT_MAX_REPLICATION_LAG = 6 * 60 * 60;
    static final boolean DEFAULT_BUNDLING_DISABLED = false;
    static final String DEFAULT_VER_GC_EXPRESSION = "";

    /**
     * Revisions older than this time would be garbage collected
     */
    static final long DEFAULT_VER_GC_MAX_AGE = 24 * 60 * 60; //TimeUnit.DAYS.toSeconds(1);


    /**
     * Blob modified before this time duration would be considered for Blob GC
     */
    static final long DEFAULT_BLOB_GC_MAX_AGE = 24 * 60 * 60;

    /**
     * Default interval for taking snapshots of locally tracked blob ids.
     */
    static final long DEFAULT_BLOB_SNAPSHOT_INTERVAL = 12 * 60 * 60;

    // property name constants - values can come from framework properties or OSGi config
    public static final String CUSTOM_BLOB_STORE = "customBlobStore";
    public static final String PROP_REV_RECOVERY_INTERVAL = "lastRevRecoveryJobIntervalInSecs";
    public static final String PROP_DS_TYPE = "documentStoreType";

    private enum DocumentStoreType {
        MONGO, RDB;

        static DocumentStoreType fromString(String type) {
            if (type == null) {
                return MONGO;
            }
            return valueOf(type.toUpperCase(Locale.ROOT));
        }
    }

    private final Logger log = LoggerFactory.getLogger(this.getClass());

    private ServiceRegistration nodeStoreReg;
    private Closer closer;
    private WhiteboardExecutor executor;

    private volatile BlobStore blobStore;

    private volatile DataSource dataSource;

    private volatile DataSource blobDataSource;

    private volatile DocumentNodeStateCache nodeStateCache;

    private DocumentNodeStore nodeStore;
    private ObserverTracker observerTracker;
    private JournalPropertyHandlerFactory journalPropertyHandlerFactory = new JournalPropertyHandlerFactory();
    private ComponentContext context;
    private Whiteboard whiteboard;
    private long deactivationTimestamp = 0;

    @Reference
    private StatisticsProvider statisticsProvider;

    @Reference
    private ConfigurationAdmin configurationAdmin;

    @Reference(service = Preset.class)
    private Preset preset;

    private boolean customBlobStore;

    private ServiceRegistration blobStoreReg;

    private BlobStore defaultBlobStore;

    private Configuration config;

    @Activate
    protected void activate(ComponentContext context, Configuration config) throws Exception {
        closer = Closer.create();
        this.config = DocumentNodeStoreServiceConfiguration.create(context,
                configurationAdmin, preset.configuration, config);
        this.context = context;
        whiteboard = new OsgiWhiteboard(context.getBundleContext());
        executor = new WhiteboardExecutor();
        executor.start(whiteboard);
        customBlobStore = this.config.customBlobStore();
        documentStoreType = DocumentStoreType.fromString(this.config.documentStoreType());

        registerNodeStoreIfPossible();
    }

    private void registerNodeStoreIfPossible() throws IOException {
        // disallow attempts to restart (OAK-3420)
        if (deactivationTimestamp != 0) {
            log.info("DocumentNodeStore was already unregistered ({}ms ago)", System.currentTimeMillis() - deactivationTimestamp);
        } else if (context == null) {
            log.info("Component still not activated. Ignoring the initialization call");
        } else if (customBlobStore && blobStore == null) {
            log.info("Custom BlobStore use enabled. DocumentNodeStoreService would be initialized when "
                    + "BlobStore would be available");
        } else if (documentStoreType == DocumentStoreType.RDB && (dataSource == null || blobDataSource == null)) {
            log.info("DataSource use enabled. DocumentNodeStoreService would be initialized when "
                    + "DataSource would be available (currently available: nodes: {}, blobs: {})", dataSource, blobDataSource);
        } else {
            registerNodeStore();
        }
    }

    private void registerNodeStore() throws IOException {
        DocumentNodeStoreBuilder<?> mkBuilder;
        if (documentStoreType == DocumentStoreType.RDB) {
            RDBDocumentNodeStoreBuilder builder = newRDBDocumentNodeStoreBuilder();
            configureBuilder(builder);
            checkNotNull(dataSource, "DataStore type set [%s] but DataSource reference not initialized", PROP_DS_TYPE);
            if (!customBlobStore) {
                checkNotNull(blobDataSource, "DataStore type set [%s] but BlobDataSource reference not initialized", PROP_DS_TYPE);
                builder.setRDBConnection(dataSource, blobDataSource);
                log.info("Connected to datasources {} {}", dataSource, blobDataSource);
            } else {
                if (blobDataSource != null && blobDataSource != dataSource) {
                    log.info("Ignoring blobDataSource {} as custom blob store takes precedence.", blobDataSource);
                }
                builder.setRDBConnection(dataSource);
                log.info("Connected to datasource {}", dataSource);
            }
            mkBuilder = builder;
        } else {
            String uri = config.mongouri();
            String db = config.db();
            boolean soKeepAlive = config.socketKeepAlive();

            MongoClientURI mongoURI = new MongoClientURI(uri);
            String persistentCache = resolvePath(config.persistentCache(), DEFAULT_PERSISTENT_CACHE);
            String journalCache = resolvePath(config.journalCache(), DEFAULT_JOURNAL_CACHE);

            if (log.isInfoEnabled()) {
                // Take care around not logging the uri directly as it
                // might contain passwords
                log.info("Starting DocumentNodeStore with host={}, db={}, cache size (MB)={}, persistentCache={}, " +
                                "journalCache={}, blobCacheSize (MB)={}, maxReplicationLagInSecs={}",
                        mongoURI.getHosts(), db, config.cache(), persistentCache,
                        journalCache, config.blobCacheSize(), config.maxReplicationLagInSecs());
                log.info("Mongo Connection details {}", MongoConnection.toString(mongoURI.getOptions()));
            }

            MongoDocumentNodeStoreBuilder builder = newMongoDocumentNodeStoreBuilder();
            configureBuilder(builder);
            builder.setMaxReplicationLag(config.maxReplicationLagInSecs(), TimeUnit.SECONDS);
            builder.setSocketKeepAlive(soKeepAlive);
            builder.setMongoDB(uri, db, config.blobCacheSize());
            mkBuilder = builder;

            log.info("Connected to database '{}'", db);
        }

        if (!customBlobStore){
            defaultBlobStore = mkBuilder.getBlobStore();
            log.info("Registering the BlobStore with ServiceRegistry");
            blobStoreReg = context.getBundleContext().registerService(BlobStore.class.getName(),
                    defaultBlobStore , null);
        }

        //Set wrapping blob store after setting the DB
        if (isWrappingCustomBlobStore()) {
            ((BlobStoreWrapper) blobStore).setBlobStore(mkBuilder.getBlobStore());
            mkBuilder.setBlobStore(blobStore);
        }

        // attach GCMonitor
        final GCMonitorTracker gcMonitor = new GCMonitorTracker();
        gcMonitor.start(whiteboard);
        closer.register(asCloseable(gcMonitor));
        Logger vgcLogger = LoggerFactory.getLogger(VersionGarbageCollector.class);
        GCMonitor loggingGCMonitor;
        if (isContinuousRevisionGC()) {
            // log less chatty with continuous RevisionGC
            loggingGCMonitor = new QuietGCMonitor(vgcLogger);
        } else {
            loggingGCMonitor = new LoggingGCMonitor(vgcLogger);
        }
        mkBuilder.setGCMonitor(new DelegatingGCMonitor(
                newArrayList(gcMonitor, loggingGCMonitor)));

        nodeStore = mkBuilder.build();

        // ensure a clusterId is initialized 
        // and expose it as 'oak.clusterid' repository descriptor
        GenericDescriptors clusterIdDesc = new GenericDescriptors();
        clusterIdDesc.put(ClusterRepositoryInfo.OAK_CLUSTERID_REPOSITORY_DESCRIPTOR_KEY, 
                new SimpleValueFactory().createValue(
                        ClusterRepositoryInfo.getOrCreateId(nodeStore)), true, false);
        whiteboard.register(Descriptors.class, clusterIdDesc, Collections.emptyMap());
        
        // If a shared data store register the repo id in the data store
        if (SharedDataStoreUtils.isShared(blobStore)) {
            String repoId = null;
            try {
                repoId = ClusterRepositoryInfo.getOrCreateId(nodeStore);
                ((SharedDataStore) blobStore).setRepositoryId(repoId);
            } catch (Exception e) {
                throw new IOException("Could not register a unique repositoryId", e);
            }

            if (blobStore instanceof BlobTrackingStore) {
                BlobTrackingStore trackingStore = (BlobTrackingStore) blobStore;
                if (trackingStore.getTracker() != null) {
                    trackingStore.getTracker().close();
                }
                ((BlobTrackingStore) blobStore).addTracker(
                    BlobIdTracker.build(getRepositoryHome(), repoId,
                            config.blobTrackSnapshotIntervalInSecs(),
                            (SharedDataStore) blobStore));
            }
        }


        registerJMXBeans(nodeStore, mkBuilder);
        registerLastRevRecoveryJob(nodeStore);
        registerJournalGC(nodeStore);
        registerVersionGCJob(nodeStore);
        registerDocumentStoreMetrics(mkBuilder.getDocumentStore());

        if (!isNodeStoreProvider()) {
            observerTracker = new ObserverTracker(nodeStore);
            observerTracker.start(context.getBundleContext());
        }
        journalPropertyHandlerFactory.start(whiteboard);

        DocumentStore ds = nodeStore.getDocumentStore();

        // OAK-2682: time difference detection applied at startup with a default
        // max time diff of 2000 millis (2sec)
        final long maxDiff = SystemPropertySupplier.create("oak.documentMK.maxServerTimeDiffMillis", 2000L).loggingTo(log).get();
        try {
            if (maxDiff>=0) {
                final long timeDiff = ds.determineServerTimeDifferenceMillis();
                log.info("registerNodeStore: server time difference: {}ms (max allowed: {}ms)", timeDiff, maxDiff);
                if (Math.abs(timeDiff) > maxDiff) {
                    throw new AssertionError("Server clock seems off (" + timeDiff + "ms) by more than configured amount ("
                            + maxDiff + "ms)");
                }
            }
        } catch (RuntimeException e) { // no checked exception
            // in case of a RuntimeException, just log but continue
            log.warn("registerNodeStore: got RuntimeException while trying to determine time difference to server: " + e, e);
        }

        String[] serviceClasses;
        if (isNodeStoreProvider()) {
            registerNodeStoreProvider(nodeStore);
            serviceClasses = new String[]{
                    DocumentNodeStore.class.getName(),
                    Clusterable.class.getName()
            };
        } else {
            serviceClasses = new String[]{
                    NodeStore.class.getName(),
                    DocumentNodeStore.class.getName(),
                    Clusterable.class.getName()
            };
        }

        Dictionary<String, Object> props = new Hashtable<String, Object>();
        props.put(Constants.SERVICE_PID, DocumentNodeStore.class.getName());
        props.put(DESCRIPTION, getMetadata(ds));
        // OAK-2844: in order to allow DocumentDiscoveryLiteService to directly
        // require a service DocumentNodeStore (instead of having to do an 'instanceof')
        // the registration is now done for both NodeStore and DocumentNodeStore here.
        nodeStoreReg = context.getBundleContext().registerService(
            serviceClasses,
            nodeStore, props);
    }

    private void configureBuilder(DocumentNodeStoreBuilder<?> builder) {
        String persistentCache = resolvePath(config.persistentCache(), DEFAULT_PERSISTENT_CACHE);
        String journalCache = resolvePath(config.journalCache(), DEFAULT_JOURNAL_CACHE);
        builder.setStatisticsProvider(statisticsProvider).
                setExecutor(executor).
                memoryCacheSize(config.cache() * MB).
                memoryCacheDistribution(
                        config.nodeCachePercentage(),
                        config.prevDocCachePercentage(),
                        config.childrenCachePercentage(),
                        config.diffCachePercentage()).
                setCacheSegmentCount(config.cacheSegmentCount()).
                setCacheStackMoveDistance(config.cacheStackMoveDistance()).
                setBundlingDisabled(config.bundlingDisabled()).
                setJournalPropertyHandlerFactory(journalPropertyHandlerFactory).
                setLeaseCheckMode(ClusterNodeInfo.DEFAULT_LEASE_CHECK_DISABLED ? LeaseCheckMode.DISABLED : LeaseCheckMode.valueOf(config.leaseCheckMode())).
                setLeaseFailureHandler(new LeaseFailureHandler() {

                    @Override
                    public void handleLeaseFailure() {
                        Bundle bundle = context.getBundleContext().getBundle();
                        String bundleName = bundle.getSymbolicName();
                        try {
                            // plan A: try stopping oak-store-document
                            log.error("handleLeaseFailure: stopping {}...", bundleName);
                            bundle.stop(Bundle.STOP_TRANSIENT);
                            log.error("handleLeaseFailure: stopped {}.", bundleName);
                            // plan A worked, perfect!
                        } catch (BundleException e) {
                            log.error("handleLeaseFailure: exception while stopping " + bundleName + ": " + e, e);
                            // plan B: stop only DocumentNodeStoreService (to stop the background threads)
                            log.error("handleLeaseFailure: stopping DocumentNodeStoreService...");
                            context.disableComponent(DocumentNodeStoreService.class.getName());
                            log.error("handleLeaseFailure: stopped DocumentNodeStoreService");
                            // plan B succeeded.
                        }
                    }
                }).
                setPrefetchExternalChanges(config.prefetchExternalChanges()).
                setUpdateLimit(config.updateLimit()).
                setJournalGCMaxAge(config.journalGCMaxAge()).
                setNodeCachePathPredicate(createCachePredicate());

        if (!Strings.isNullOrEmpty(persistentCache)) {
            builder.setPersistentCache(persistentCache);
        }
        if (!Strings.isNullOrEmpty(journalCache)) {
            builder.setJournalCache(journalCache);
        }

        //Set blobstore before setting the document store
        if (customBlobStore && !isWrappingCustomBlobStore()) {
            checkNotNull(blobStore, "Use of custom BlobStore enabled via  [%s] but blobStore reference not " +
                    "initialized", CUSTOM_BLOB_STORE);
            builder.setBlobStore(blobStore);
        }
    }

    private boolean isWrappingCustomBlobStore() {
        return customBlobStore && blobStore instanceof BlobStoreWrapper;
    }

    private Predicate<Path> createCachePredicate() {
        if (config.persistentCacheIncludes().length == 0) {
            return Predicates.alwaysTrue();
        }
        if (Arrays.equals(config.persistentCacheIncludes(), new String[]{"/"})) {
            return Predicates.alwaysTrue();
        }

        Set<Path> paths = new HashSet<>();
        for (String p : config.persistentCacheIncludes()) {
            p = p != null ? Strings.emptyToNull(p.trim()) : null;
            if (p != null) {
                paths.add(Path.fromString(p));
            }
        }
        log.info("Configuring persistent cache to only cache nodes under paths {}", paths);
        return input -> {
            if (input != null) {
                for (Path p : paths) {
                    if (p.isAncestorOf(input)) {
                        return true;
                    }
                }
            }
            return false;
        };
    }

    private boolean isNodeStoreProvider() {
        return !Strings.isNullOrEmpty(config.role());
    }

    private boolean isContinuousRevisionGC() {
        String expr = getVersionGCExpression();
        String[] elements = expr.split("\\s");
        // simple heuristic to determine if revision GC runs 'frequently'
        return elements.length >= 6 && elements[1].equals("*");
    }

    private String getVersionGCExpression() {
        String defaultExpr = CONTINUOUS_RGC_EXPR;
        String expr = config.versionGCExpression();
        if (Strings.isNullOrEmpty(expr)) {
            expr = defaultExpr;
        }
        // validate expression
        try {
            if (!expr.isEmpty()) {
                new CronExpression(expr);
            }
        } catch (ParseException e) {
            log.warn("Invalid cron expression, falling back to default '" + defaultExpr + "'", e);
            expr = defaultExpr;
        }
        return expr;
    }

    private void registerNodeStoreProvider(final NodeStore ns) {
        Dictionary<String, Object> props = new Hashtable<>();
        props.put(NodeStoreProvider.ROLE, config.role());
        nodeStoreReg = context.getBundleContext().registerService(
                NodeStoreProvider.class.getName(),
                (NodeStoreProvider) () -> ns,
                props);
        log.info("Registered NodeStoreProvider backed by DocumentNodeStore");
    }

    @Deactivate
    protected void deactivate() {
        if (observerTracker != null) {
            observerTracker.stop();
        }

        if (journalPropertyHandlerFactory != null){
            journalPropertyHandlerFactory.stop();
        }

        unregisterNodeStore();
    }

    @Reference(name = "blobStore",
            cardinality = ReferenceCardinality.OPTIONAL,
            policy = ReferencePolicy.DYNAMIC,
            target = ONLY_STANDALONE_TARGET
    )
    @SuppressWarnings("UnusedDeclaration")
    protected void bindBlobStore(BlobStore blobStore) throws IOException {
        if (defaultBlobStore == blobStore){
            return;
        }
        log.info("Initializing DocumentNodeStore with BlobStore [{}]", blobStore);
        this.blobStore = blobStore;
        registerNodeStoreIfPossible();
    }

    @SuppressWarnings("UnusedDeclaration")
    protected void unbindBlobStore(BlobStore blobStore) {
        if (defaultBlobStore == blobStore){
            return;
        }
        this.blobStore = null;
        unregisterNodeStore();
    }

    @Reference(name = "dataSource",
            cardinality = ReferenceCardinality.OPTIONAL,
            policy = ReferencePolicy.DYNAMIC,
            target = "(datasource.name=oak)"
    )
    @SuppressWarnings("UnusedDeclaration")
    protected void bindDataSource(DataSource dataSource) throws IOException {
        if (this.dataSource != null) {
            log.info("Ignoring bindDataSource [{}] because dataSource [{}] is already bound", dataSource, this.dataSource);
        } else {
            log.info("Initializing DocumentNodeStore with dataSource [{}]", dataSource);
            this.dataSource = dataSource;
            registerNodeStoreIfPossible();
        }
    }

    @SuppressWarnings("UnusedDeclaration")
    protected void unbindDataSource(DataSource dataSource) {
        if (this.dataSource != dataSource) {
            log.info("Ignoring unbindDataSource [{}] because dataSource is bound to [{}]", dataSource, this.dataSource);
        } else {
            log.info("Unregistering DocumentNodeStore because dataSource [{}] was unbound", dataSource);
            this.dataSource = null;
            unregisterNodeStore();
        }
    }


    private DocumentStoreType documentStoreType;

    @Reference(name = "blobDataSource",
            cardinality = ReferenceCardinality.OPTIONAL,
            policy = ReferencePolicy.DYNAMIC,
            target = "(datasource.name=oak)"
    )
    @SuppressWarnings("UnusedDeclaration")
    protected void bindBlobDataSource(DataSource dataSource) throws IOException {
        if (this.blobDataSource != null) {
            log.info("Ignoring bindBlobDataSource [{}] because blobDataSource [{}] is already bound", dataSource,
                    this.blobDataSource);
        } else {
            log.info("Initializing DocumentNodeStore with blobDataSource [{}]", dataSource);
            this.blobDataSource = dataSource;
            registerNodeStoreIfPossible();
        }
    }

    @SuppressWarnings("UnusedDeclaration")
    protected void unbindBlobDataSource(DataSource dataSource) {
        if (this.blobDataSource != dataSource) {
            log.info("Ignoring unbindBlobDataSource [{}] because dataSource is bound to [{}]", dataSource, this.blobDataSource);
        } else {
            log.info("Unregistering DocumentNodeStore because blobDataSource [{}] was unbound", dataSource);
            this.blobDataSource = null;
            unregisterNodeStore();
        }
    }

    @Reference(name = "nodeStateCache",
            cardinality = ReferenceCardinality.OPTIONAL,
            policy = ReferencePolicy.DYNAMIC
    )
    @SuppressWarnings("UnusedDeclaration")
    protected void bindNodeStateCache(DocumentNodeStateCache nodeStateCache) throws IOException {
       if (nodeStore != null){
           log.info("Registered DocumentNodeStateCache [{}] with DocumentNodeStore", nodeStateCache);
           nodeStore.setNodeStateCache(nodeStateCache);
       }
    }

    @SuppressWarnings("UnusedDeclaration")
    protected void unbindNodeStateCache(DocumentNodeStateCache nodeStateCache) {
        if (nodeStore != null){
            nodeStore.setNodeStateCache(DocumentNodeStateCache.NOOP);
        }
    }

    private void unregisterNodeStore() {
        deactivationTimestamp = System.currentTimeMillis();

        closeQuietly(closer);

        if (nodeStoreReg != null) {
            nodeStoreReg.unregister();
            nodeStoreReg = null;
        }

        //If we exposed our BlobStore then unregister it *after*
        //NodeStore service. This ensures that if any other component
        //like SecondaryStoreCache depends on this then it remains active
        //untill DocumentNodeStore get deactivated
        if (blobStoreReg != null){
            blobStoreReg.unregister();
            blobStoreReg = null;
        }

        if (nodeStore != null) {
            nodeStore.dispose();
            nodeStore = null;
        }

        if (executor != null) {
            executor.stop();
            executor = null;
        }
    }

    private void registerJMXBeans(final DocumentNodeStore store, DocumentNodeStoreBuilder<?> mkBuilder) throws
            IOException {
        registerCacheStatsMBean(store.getNodeCacheStats());
        registerCacheStatsMBean(store.getNodeChildrenCacheStats());
        for (CacheStats cs : store.getDiffCacheStats()) {
            registerCacheStatsMBean(cs);
        }
        DocumentStore ds = store.getDocumentStore();
        if (ds.getCacheStats() != null) {
            for (CacheStats cacheStats : ds.getCacheStats()) {
                registerCacheStatsMBean(cacheStats);
            }
        }

        if (!isNodeStoreProvider()) {
            addRegistration(
                    registerMBean(whiteboard,
                            CheckpointMBean.class,
                            new DocumentCheckpointMBean(store),
                            CheckpointMBean.TYPE,
                            "Document node store checkpoint management")
            );
        }

        addRegistration(
                registerMBean(whiteboard,
                        DocumentNodeStoreMBean.class,
                        store.getMBean(),
                        DocumentNodeStoreMBean.TYPE,
                        "Document node store management")
        );

        if (mkBuilder.getBlobStoreCacheStats() != null) {
            registerCacheStatsMBean(mkBuilder.getBlobStoreCacheStats());
        }

        if (mkBuilder.getDocumentStoreStatsCollector() instanceof DocumentStoreStatsMBean) {
            addRegistration(
                    registerMBean(whiteboard,
                            DocumentStoreStatsMBean.class,
                            (DocumentStoreStatsMBean) mkBuilder.getDocumentStoreStatsCollector(),
                            DocumentStoreStatsMBean.TYPE,
                            "DocumentStore Statistics")
            );
        }

        // register persistent cache stats
        Map<String, PersistentCacheStats> persistenceCacheStats = mkBuilder.getPersistenceCacheStats();
        for (PersistentCacheStats pcs: persistenceCacheStats.values()) {
            addRegistration(
                    registerMBean(whiteboard,
                            PersistentCacheStatsMBean.class,
                            pcs,
                            PersistentCacheStatsMBean.TYPE,
                            pcs.getName())
            );
        }

        final long versionGcMaxAgeInSecs = config.versionGcMaxAgeInSecs();
        final long blobGcMaxAgeInSecs = config.blobGcMaxAgeInSecs();

        if (store.getBlobStore() instanceof GarbageCollectableBlobStore) {
            BlobGarbageCollector gc = store.createBlobGarbageCollector(blobGcMaxAgeInSecs, 
                                                        ClusterRepositoryInfo.getOrCreateId(nodeStore),
                                                        whiteboard, statisticsProvider);
            addRegistration(registerMBean(whiteboard, BlobGCMBean.class, new BlobGC(gc, executor),
                    BlobGCMBean.TYPE, "Document node store blob garbage collection"));
        }

        Runnable startGC = new RevisionGCJob(store, versionGcMaxAgeInSecs, 0, DEFAULT_RGC_DELAY_FACTOR);
        Runnable cancelGC = () -> store.getVersionGarbageCollector().cancel();
        Supplier<String> status = () -> store.getVersionGarbageCollector().getStatus();
        RevisionGC revisionGC = new RevisionGC(startGC, cancelGC, status, executor);
        addRegistration(registerMBean(whiteboard, RevisionGCMBean.class, revisionGC,
                RevisionGCMBean.TYPE, "Document node store revision garbage collection"));

        addRegistration(registerMBean(whiteboard, RevisionGCStatsMBean.class,
                store.getVersionGarbageCollector().getRevisionGCStats(),
                RevisionGCStatsMBean.TYPE,
                "Document node store revision garbage collection statistics"));

        BlobStoreStats blobStoreStats = mkBuilder.getBlobStoreStats();
        if (!customBlobStore && blobStoreStats != null) {
            addRegistration(registerMBean(whiteboard,
                    BlobStoreStatsMBean.class,
                    blobStoreStats,
                    BlobStoreStatsMBean.TYPE,
                    ds.getClass().getSimpleName()));
        }

        if (!mkBuilder.isBundlingDisabled()){
            addRegistration(registerMBean(whiteboard,
                    BackgroundObserverMBean.class,
                    store.getBundlingConfigHandler().getMBean(),
                    BackgroundObserverMBean.TYPE,
                    "BundlingConfigObserver"));
        }
    }

    private void registerCacheStatsMBean(CacheStats cacheStats) {
        addRegistration(registerMBean(whiteboard, CacheStatsMBean.class,
                cacheStats, CacheStatsMBean.TYPE, cacheStats.getName()));
    }

    private void registerLastRevRecoveryJob(final DocumentNodeStore nodeStore) {
        long leaseTime = toLong(context.getProperties().get(PROP_REV_RECOVERY_INTERVAL),
                ClusterNodeInfo.DEFAULT_LEASE_UPDATE_INTERVAL_MILLIS);
        addRegistration(scheduleWithFixedDelay(whiteboard,
                new LastRevRecoveryJob(nodeStore), TimeUnit.MILLISECONDS.toSeconds(leaseTime),
                false/*runOnSingleClusterNode*/, true /*use dedicated pool*/));
    }

    private void registerJournalGC(final DocumentNodeStore nodeStore) {
        addRegistration(scheduleWithFixedDelay(whiteboard,
                new JournalGCJob(nodeStore),
                jobPropertiesFor(JournalGCJob.class),
                TimeUnit.MILLISECONDS.toSeconds(config.journalGCInterval()),
                true/*runOnSingleClusterNode*/, true /*use dedicated pool*/));
    }

    private void registerVersionGCJob(final DocumentNodeStore nodeStore) {
        String expr = getVersionGCExpression();
        if (expr.isEmpty()) {
            return;
        }
        Map<String, Object> props = jobPropertiesFor(RevisionGCJob.class);
        props.put("scheduler.expression", expr);
        long versionGcMaxAgeInSecs = config.versionGcMaxAgeInSecs();
        long versionGCTimeLimitInSecs = config.versionGCTimeLimitInSecs();
        double versionGCDelayFactor = config.versionGCDelayFactor();
        addRegistration(scheduleWithFixedDelay(whiteboard,
                new RevisionGCJob(nodeStore, versionGcMaxAgeInSecs,
                        versionGCTimeLimitInSecs, versionGCDelayFactor),
                props, MODIFIED_IN_SECS_RESOLUTION, true, true));
    }

    private void registerDocumentStoreMetrics(DocumentStore store) {
        if (store instanceof MongoDocumentStore) {
            addRegistration(scheduleWithFixedDelay(whiteboard,
                    new MongoDocumentStoreMetrics((MongoDocumentStore) store, statisticsProvider),
                    jobPropertiesFor(MongoDocumentStoreMetrics.class),
                    TimeUnit.MINUTES.toSeconds(1), false, true));
        }
    }

    private String resolvePath(String value, String defaultValue) {
        String path = value;
        if (Strings.isNullOrEmpty(value)) {
            path = defaultValue;
        }
        if ("-".equals(path)) {
            // disable this path configuration
            return "";
        }
        // resolve as relative to repository.home
        return FilenameUtils.concat(getRepositoryHome(), path);
    }

    private String getRepositoryHome() {
        String repoHome = config.repository_home();
        if (Strings.isNullOrEmpty(repoHome)) {
            repoHome = DEFAULT_PROP_HOME;
        }
        return repoHome;
    }

    private static String[] getMetadata(DocumentStore ds) {
        Map<String, String> meta = new HashMap<String, String>(ds.getMetadata());
        meta.put("nodeStoreType", "document");
        String[] result = new String[meta.size()];
        int i = 0;
        for (Map.Entry<String, String> e : meta.entrySet()) {
            result[i++] = e.getKey() + "=" + e.getValue();
        }
        return result;
    }

    private void addRegistration(@NotNull Registration reg) {
        closer.register(asCloseable(reg));
    }

    private static Closeable asCloseable(@NotNull final Registration reg) {
        checkNotNull(reg);
        return new Closeable() {
            @Override
            public void close() throws IOException {
                reg.unregister();
            }
        };
    }

    private static Closeable asCloseable(@NotNull final AbstractServiceTracker t) {
        checkNotNull(t);
        return new Closeable() {
            @Override
            public void close() throws IOException {
                t.stop();
            }
        };
    }

    static final class RevisionGCJob implements Runnable, Supplier<String> {

        // log as VersionGarbageCollector
        private static final Logger LOGGER = LoggerFactory.getLogger(VersionGarbageCollector.class);

        // log every hour
        private static final long LOG_INTERVAL = TimeUnit.HOURS.toMillis(1);

        private final DocumentNodeStore nodeStore;
        private final long versionGCMaxAgeInSecs;
        private final long versionGCTimeLimitInSecs;
        private final double versionGCDelayFactor;
        private volatile Object lastResult = "";
        private long lastLogTime;
        private VersionGCStats stats;

        RevisionGCJob(DocumentNodeStore ns,
                      long versionGcMaxAgeInSecs,
                      long versionGCTimeLimitInSecs,
                      double versionGCDelayFactor) {
            this.nodeStore = ns;
            this.versionGCMaxAgeInSecs = versionGcMaxAgeInSecs;
            this.versionGCTimeLimitInSecs = versionGCTimeLimitInSecs;
            this.versionGCDelayFactor = versionGCDelayFactor;
            resetStats();
        }

        @Override
        public void run() {
            VersionGarbageCollector gc = nodeStore.getVersionGarbageCollector();
            gc.setOptions(gc.getOptions().withMaxDuration(TimeUnit.SECONDS, versionGCTimeLimitInSecs).withDelayFactor(versionGCDelayFactor));
            try {
                VersionGCStats s = gc.gc(versionGCMaxAgeInSecs, TimeUnit.SECONDS);
                stats.addRun(s);
                lastResult = s.toString();
            } catch (Exception e) {
                lastResult = e;
                LOGGER.warn("Error occurred while executing the Version Garbage Collector", e);
            } finally {
                maybeLogStats();
            }
        }

        /**
         * Returns the result of the last revision GC run. This method throws
         * an {@link UncheckedExecutionException} if the last run failed with an
         * exception.
         *
         * @return result of the last revision GC run.
         */
        @Override
        public String get() throws UncheckedExecutionException {
            if (lastResult instanceof Exception) {
                throw new UncheckedExecutionException((Exception) lastResult);
            }
            return String.valueOf(lastResult);
        }

        private void resetStats() {
            lastLogTime = nodeStore.getClock().getTime();
            stats = new VersionGCStats();
        }

        private void maybeLogStats() {
            if (nodeStore.getClock().getTime() > lastLogTime + LOG_INTERVAL) {
                LOGGER.info("Garbage collector stats since {}: {}",
                        Utils.timestampToString(lastLogTime), stats);
                resetStats();
            }
        }
    }

    private static final class JournalGCJob implements Runnable {

        private final DocumentNodeStore nodeStore;

        JournalGCJob(DocumentNodeStore ns) {
            this.nodeStore = ns;
        }

        @Override
        public void run() {
            nodeStore.getJournalGarbageCollector().gc();
        }
    }

    private static final class LastRevRecoveryJob implements Runnable {

        private final DocumentNodeStore nodeStore;

        LastRevRecoveryJob(DocumentNodeStore ns) {
            this.nodeStore = ns;
        }

        @Override
        public void run() {
            nodeStore.getLastRevRecoveryAgent().performRecoveryIfNeeded();
        }
    }

    private static Map<String, Object> jobPropertiesFor(Class clazz) {
        Map<String, Object> props = new HashMap<>();
        props.put("scheduler.name", clazz.getName());
        return props;
    }

    @Component(
            service = Preset.class,
            configurationPid = Configuration.PRESET_PID)
    public static class Preset {

        Configuration configuration;

        @Activate
        void activate(Configuration configuration) {
            this.configuration = configuration;
        }
    }
}