/*
 * 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.ignite.internal.metastorage;

import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import org.apache.ignite.configuration.schemas.runner.NodeConfiguration;
import org.apache.ignite.internal.configuration.ConfigurationManager;
import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.metastorage.client.CompactedException;
import org.apache.ignite.internal.metastorage.client.Condition;
import org.apache.ignite.internal.metastorage.client.Entry;
import org.apache.ignite.internal.metastorage.client.MetaStorageService;
import org.apache.ignite.internal.metastorage.client.MetaStorageServiceImpl;
import org.apache.ignite.internal.metastorage.client.Operation;
import org.apache.ignite.internal.metastorage.client.OperationTimeoutException;
import org.apache.ignite.internal.metastorage.client.WatchListener;
import org.apache.ignite.internal.metastorage.server.KeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageListener;
import org.apache.ignite.internal.metastorage.watch.AggregatedWatch;
import org.apache.ignite.internal.metastorage.watch.KeyCriterion;
import org.apache.ignite.internal.metastorage.watch.WatchAggregator;
import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.util.Cursor;
import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.vault.VaultManager;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteLogger;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.lang.NodeStoppingException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.network.TopologyEventHandler;
import org.apache.ignite.raft.client.service.RaftGroupService;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.internal.util.ByteUtils.bytesToLong;
import static org.apache.ignite.internal.util.ByteUtils.longToBytes;

/**
 * MetaStorage manager is responsible for:
 * <ul>
 *     <li>Handling cluster init message.</li>
 *     <li>Managing meta storage lifecycle including instantiation meta storage raft group.</li>
 *     <li>Providing corresponding meta storage service proxy interface</li>
 * </ul>
 */
// TODO: IGNITE-14586 Remove @SuppressWarnings when implementation provided.
@SuppressWarnings("unused")
public class MetaStorageManager implements IgniteComponent {
    /** Logger. */
    private static final IgniteLogger LOG = IgniteLogger.forClass(MetaStorageManager.class);

    /** Meta storage raft group name. */
    private static final String METASTORAGE_RAFT_GROUP_NAME = "metastorage_raft_group";

    /**
     * Special key for the vault where the applied revision for {@link MetaStorageManager#storeEntries}
     * operation is stored. This mechanism is needed for committing processed watches to {@link VaultManager}.
     */
    public static final ByteArray APPLIED_REV = ByteArray.fromString("applied_revision");

    /** Vault manager in order to commit processed watches with corresponding applied revision. */
    private final VaultManager vaultMgr;

    /** Configuration manager that handles local configuration. */
    private final ConfigurationManager locCfgMgr;

    /** Cluster network service that is used in order to handle cluster init message. */
    private final ClusterService clusterNetSvc;

    /** Raft manager that is used for metastorage raft group handling. */
    private final Loza raftMgr;

    /** Meta storage service. */
    private volatile CompletableFuture<MetaStorageService> metaStorageSvcFut;

    /** Raft group service. */
    private volatile CompletableFuture<RaftGroupService> raftGroupServiceFut;

    /**
     * Aggregator of multiple watches to deploy them as one batch.
     *
     * @see WatchAggregator
     */
    private final WatchAggregator watchAggregator = new WatchAggregator();

    /**
     * Future which will be completed with {@link IgniteUuid},
     * when aggregated watch will be successfully deployed.
     * Can be resolved to {@link Optional#empty()} if no watch deployed at the moment.
     */
    private CompletableFuture<Optional<IgniteUuid>> deployFut = new CompletableFuture<>();

    /**
     * If true - all new watches will be deployed immediately.
     *
     * If false - all new watches will be aggregated to one batch
     * for further deploy by {@link MetaStorageManager#deployWatches()}
     */
    private boolean deployed;

    /** Flag indicates if meta storage nodes were set on start */
    private boolean metaStorageNodesOnStart;

    /** Actual storage for the Metastorage. */
    private final KeyValueStorage storage;

    /** Busy lock for stop synchronisation. */
    private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();

    /**
     * The constructor.
     *
     * @param vaultMgr Vault manager.
     * @param locCfgMgr Local configuration manager.
     * @param clusterNetSvc Cluster network service.
     * @param raftMgr Raft manager.
     * @param storage Storage. This component owns this resource and will manage its lifecycle.
     */
    public MetaStorageManager(
        VaultManager vaultMgr,
        ConfigurationManager locCfgMgr,
        ClusterService clusterNetSvc,
        Loza raftMgr,
        KeyValueStorage storage
    ) {
        this.vaultMgr = vaultMgr;
        this.locCfgMgr = locCfgMgr;
        this.clusterNetSvc = clusterNetSvc;
        this.raftMgr = raftMgr;
        this.storage = storage;
    }

    /** {@inheritDoc} */
    @Override public void start() {
        String[] metastorageNodes = this.locCfgMgr.configurationRegistry().getConfiguration(NodeConfiguration.KEY)
            .metastorageNodes().value();

        Predicate<ClusterNode> metaStorageNodesContainsLocPred =
            clusterNode -> Arrays.asList(metastorageNodes).contains(clusterNode.name());

        if (metastorageNodes.length > 0) {
            metaStorageNodesOnStart = true;

            List<ClusterNode> metaStorageMembers = clusterNetSvc.topologyService().allMembers().stream()
                .filter(metaStorageNodesContainsLocPred)
                .collect(Collectors.toList());

            storage.start();

            raftGroupServiceFut = raftMgr.prepareRaftGroup(
                METASTORAGE_RAFT_GROUP_NAME,
                metaStorageMembers,
                () -> new MetaStorageListener(storage)
            );

            this.metaStorageSvcFut = raftGroupServiceFut.thenApply(service ->
                new MetaStorageServiceImpl(service, clusterNetSvc.topologyService().localMember().id())
            );

            if (hasMetastorageLocally(locCfgMgr)) {
                clusterNetSvc.topologyService().addEventHandler(new TopologyEventHandler() {
                    @Override public void onAppeared(ClusterNode member) {
                        // No-op.
                    }

                    @Override public void onDisappeared(ClusterNode member) {
                        metaStorageSvcFut.thenCompose(svc -> svc.closeCursors(member.id()));
                    }
                });
            }
        }
        else
            this.metaStorageSvcFut = new CompletableFuture<>();

        // TODO: IGNITE-14088: Uncomment and use real serializer factory
//        Arrays.stream(MetaStorageMessageTypes.values()).forEach(
//            msgTypeInstance -> net.registerMessageMapper(
//                msgTypeInstance.msgType(),
//                new DefaultMessageMapperProvider()
//            )
//        );

        // TODO: IGNITE-14414 Cluster initialization flow. Here we should complete metaStorageServiceFuture.
//        clusterNetSvc.messagingService().addMessageHandler((message, senderAddr, correlationId) -> {});
    }

    /** {@inheritDoc} */
    @Override public void stop() {
        busyLock.block();

        Optional<IgniteUuid> watchId;

        try {
            // If deployed future is not done, that means that stop was called in the middle of
            // IgniteImpl.start, before deployWatches, or before init phase.
            // It is correct to check completeness of the future because the method calls are guarded by busy lock.
            // TODO: add busy lock for init method https://issues.apache.org/jira/browse/IGNITE-14414
            if (deployFut.isDone()) {
                watchId = deployFut.get();

                try {
                    if (watchId.isPresent())
                        metaStorageSvcFut.get().stopWatch(watchId.get());
                }
                catch (InterruptedException | ExecutionException e) {
                    LOG.error("Failed to get meta storage service.");

                    throw new IgniteInternalException(e);
                }
            }
        } catch (InterruptedException | ExecutionException e) {
            LOG.error("Failed to get watch.");

            throw new IgniteInternalException(e);
        }

        try {
            if (raftGroupServiceFut != null) {
                raftGroupServiceFut.get().shutdown();

                raftMgr.stopRaftGroup(METASTORAGE_RAFT_GROUP_NAME, metastorageNodes());
            }
        }
        catch (InterruptedException | ExecutionException e) {
            LOG.error("Failed to get meta storage raft group service.");

            throw new IgniteInternalException(e);
        }

        try {
            storage.close();
        }
        catch (Exception e) {
            throw new IgniteInternalException("Exception when stopping the storage", e);
        }
    }

    /**
     * Deploy all registered watches.
     */
    public synchronized void deployWatches() throws NodeStoppingException {
        if (!busyLock.enterBusy())
            throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

        try {
            var watch = watchAggregator.watch(
                appliedRevision() + 1,
                this::storeEntries
            );

            if (watch.isEmpty())
                deployFut.complete(Optional.empty());
            else {
                CompletableFuture<Void> fut =
                    dispatchAppropriateMetaStorageWatch(watch.get()).thenAccept(id -> deployFut.complete(Optional.of(id)));

                if (metaStorageNodesOnStart)
                    fut.join();
                else {
                    // TODO: need to wait for this future in init phase https://issues.apache.org/jira/browse/IGNITE-14414
                }
            }

            deployed = true;
        } finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * Register watch listener by key.
     *
     * @param key The target key.
     * @param lsnr Listener which will be notified for each update.
     * @return Subscription identifier. Could be used in {@link #unregisterWatch} method in order to cancel
     * subscription
     */
    public synchronized CompletableFuture<Long> registerWatch(
        @Nullable ByteArray key,
        @NotNull WatchListener lsnr
    ) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return waitForReDeploy(watchAggregator.add(key, lsnr));
        } finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * Register watch listener by key prefix.
     *
     * @param key Prefix to listen.
     * @param lsnr Listener which will be notified for each update.
     * @return Subscription identifier. Could be used in {@link #unregisterWatch} method in order to cancel
     * subscription
     */
    public synchronized CompletableFuture<Long> registerWatchByPrefix(
        @Nullable ByteArray key,
        @NotNull WatchListener lsnr
    ) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return waitForReDeploy(watchAggregator.addPrefix(key, lsnr));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * Register watch listener by collection of keys.
     *
     * @param keys Collection listen.
     * @param lsnr Listener which will be notified for each update.
     * @return Subscription identifier. Could be used in {@link #unregisterWatch} method in order to cancel
     * subscription
     */
    public synchronized CompletableFuture<Long> registerWatch(
        @NotNull Collection<ByteArray> keys,
        @NotNull WatchListener lsnr
    ) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return waitForReDeploy(watchAggregator.add(keys, lsnr));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * Register watch listener by range of keys.
     *
     * @param from Start key of range.
     * @param to End key of range (exclusively).
     * @param lsnr Listener which will be notified for each update.
     * @return future with id of registered watch.
     */
    public synchronized CompletableFuture<Long> registerWatch(
        @NotNull ByteArray from,
        @NotNull ByteArray to,
        @NotNull WatchListener lsnr
    ) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return waitForReDeploy(watchAggregator.add(from, to, lsnr));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * Unregister watch listener by id.
     *
     * @param id of watch to unregister.
     * @return future, which will be completed when unregister finished.
     */
    public synchronized CompletableFuture<Void> unregisterWatch(long id) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            watchAggregator.cancel(id);
            if (deployed)
                return updateWatches().thenAccept(v -> {});
            else
                return deployFut.thenAccept(uuid -> {});
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#get(ByteArray)
     */
    public @NotNull CompletableFuture<Entry> get(@NotNull ByteArray key) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.get(key));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#get(ByteArray, long)
     */
    public @NotNull CompletableFuture<Entry> get(@NotNull ByteArray key, long revUpperBound) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.get(key, revUpperBound));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#getAll(Set)
     */
    public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAll(Set<ByteArray> keys) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.getAll(keys));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#getAll(Set, long)
     */
    public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAll(Set<ByteArray> keys, long revUpperBound) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.getAll(keys, revUpperBound));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#put(ByteArray, byte[])
     */
    public @NotNull CompletableFuture<Void> put(@NotNull ByteArray key, byte[] val) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.put(key, val));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#getAndPut(ByteArray, byte[])
     */
    public @NotNull CompletableFuture<Entry> getAndPut(@NotNull ByteArray key, byte[] val) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.getAndPut(key, val));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#putAll(Map)
     */
    public @NotNull CompletableFuture<Void> putAll(@NotNull Map<ByteArray, byte[]> vals) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.putAll(vals));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#getAndPutAll(Map)
     */
    public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAndPutAll(@NotNull Map<ByteArray, byte[]> vals) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.getAndPutAll(vals));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#remove(ByteArray)
     */
    public @NotNull CompletableFuture<Void> remove(@NotNull ByteArray key) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.remove(key));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#getAndRemove(ByteArray)
     */
    public @NotNull CompletableFuture<Entry> getAndRemove(@NotNull ByteArray key) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.getAndRemove(key));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#removeAll(Set)
     */
    public @NotNull CompletableFuture<Void> removeAll(@NotNull Set<ByteArray> keys) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.removeAll(keys));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#getAndRemoveAll(Set)
     */
    public @NotNull CompletableFuture<Map<ByteArray, Entry>> getAndRemoveAll(@NotNull Set<ByteArray> keys) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.getAndRemoveAll(keys));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * Invoke with single success/failure operation.
     *
     * @see MetaStorageService#invoke(Condition, Operation, Operation)
     */
    public @NotNull CompletableFuture<Boolean> invoke(
        @NotNull Condition cond,
        @NotNull Operation success,
        @NotNull Operation failure
    ) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.invoke(cond, success, failure));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#invoke(Condition, Collection, Collection)
     */
    public @NotNull CompletableFuture<Boolean> invoke(
            @NotNull Condition cond,
            @NotNull Collection<Operation> success,
            @NotNull Collection<Operation> failure
    ) {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(svc -> svc.invoke(cond, success, failure));
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#range(ByteArray, ByteArray, long)
     */
    public @NotNull Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo, long revUpperBound) throws NodeStoppingException {
        if (!busyLock.enterBusy())
            throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

        try {
            return new CursorWrapper<>(
                metaStorageSvcFut.thenApply(svc -> svc.range(keyFrom, keyTo, revUpperBound))
            );
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * Retrieves entries for the given key range in lexicographic order.
     * Entries will be filtered out by the current applied revision as an upper bound.
     * Applied revision is a revision of the last successful vault update.
     *
     * @param keyFrom Start key of range (inclusive). Couldn't be {@code null}.
     * @param keyTo End key of range (exclusive). Could be {@code null}.
     * @return Cursor built upon entries corresponding to the given range and applied revision.
     * @throws OperationTimeoutException If the operation is timed out.
     * @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
     * @see ByteArray
     * @see Entry
     */
    public @NotNull Cursor<Entry> rangeWithAppliedRevision(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo) throws NodeStoppingException {
        if (!busyLock.enterBusy())
            throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

        try {
            return new CursorWrapper<>(
                metaStorageSvcFut.thenApply(svc -> svc.range(keyFrom, keyTo, appliedRevision()))
            );
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#range(ByteArray, ByteArray)
     */
    public @NotNull Cursor<Entry> range(@NotNull ByteArray keyFrom, @Nullable ByteArray keyTo) throws NodeStoppingException {
        if (!busyLock.enterBusy())
            throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

        try {
            return new CursorWrapper<>(
                metaStorageSvcFut.thenApply(svc -> svc.range(keyFrom, keyTo))
            );
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * Retrieves entries for the given key prefix in lexicographic order.
     * Entries will be filtered out by the current applied revision as an upper bound.
     * Applied revision is a revision of the last successful vault update.
     *
     * Prefix query is a synonym of the range query {@code (prefixKey, nextKey(prefixKey))}.
     *
     * @param keyPrefix Prefix of the key to retrieve the entries. Couldn't be {@code null}.
     * @return Cursor built upon entries corresponding to the given range and applied revision.
     * @throws OperationTimeoutException If the operation is timed out.
     * @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
     * @see ByteArray
     * @see Entry
     */
    public @NotNull Cursor<Entry> prefixWithAppliedRevision(@NotNull ByteArray keyPrefix) throws NodeStoppingException {
        if (!busyLock.enterBusy())
            throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

        try {
            var rangeCriterion = KeyCriterion.RangeCriterion.fromPrefixKey(keyPrefix);

            return new CursorWrapper<>(
                metaStorageSvcFut.thenApply(svc -> svc.range(rangeCriterion.from(), rangeCriterion.to(), appliedRevision()))
            );
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * Retrieves entries for the given key prefix in lexicographic order. Short cut for
     * {@link #prefix(ByteArray, long)} where {@code revUpperBound == -1}.
     *
     * @param keyPrefix Prefix of the key to retrieve the entries. Couldn't be {@code null}.
     * @return Cursor built upon entries corresponding to the given range and revision.
     * @throws OperationTimeoutException If the operation is timed out.
     * @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
     * @see ByteArray
     * @see Entry
     */
    public @NotNull Cursor<Entry> prefix(@NotNull ByteArray keyPrefix) throws NodeStoppingException {
        return prefix(keyPrefix, -1);
    }

    /**
     * Retrieves entries for the given key prefix in lexicographic order. Entries will be filtered out by upper bound
     * of given revision number.
     *
     * Prefix query is a synonym of the range query {@code range(prefixKey, nextKey(prefixKey))}.
     *
     * @param keyPrefix Prefix of the key to retrieve the entries. Couldn't be {@code null}.
     * @param revUpperBound  The upper bound for entry revision. {@code -1} means latest revision.
     * @return Cursor built upon entries corresponding to the given range and revision.
     * @throws OperationTimeoutException If the operation is timed out.
     * @throws CompactedException If the desired revisions are removed from the storage due to a compaction.
     * @see ByteArray
     * @see Entry
     */
    public @NotNull Cursor<Entry> prefix(@NotNull ByteArray keyPrefix, long revUpperBound) throws NodeStoppingException {
        if (!busyLock.enterBusy())
            throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

        try {
            var rangeCriterion = KeyCriterion.RangeCriterion.fromPrefixKey(keyPrefix);
            return new CursorWrapper<>(
                metaStorageSvcFut.thenApply(svc -> svc.range(rangeCriterion.from(), rangeCriterion.to(), revUpperBound))
            );
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @see MetaStorageService#compact()
     */
    public @NotNull CompletableFuture<Void> compact() {
        if (!busyLock.enterBusy())
            return CompletableFuture.failedFuture(new NodeStoppingException("Operation has been cancelled (node is stopping)."));

        try {
            return metaStorageSvcFut.thenCompose(MetaStorageService::compact);
        }
        finally {
            busyLock.leaveBusy();
        }
    }

    /**
     * @return Applied revision for {@link VaultManager#putAll} operation.
     */
    private long appliedRevision() {
        byte[] appliedRevision = vaultMgr.get(APPLIED_REV).join().value();

        return appliedRevision == null ? 0L : bytesToLong(appliedRevision);
    }

    /**
     * Stop current batch of consolidated watches and register new one from current {@link WatchAggregator}.
     *
     * @return Ignite UUID of new consolidated watch.
     */
    private CompletableFuture<Optional<IgniteUuid>> updateWatches() {
        long revision = appliedRevision() + 1;

        deployFut = deployFut
            .thenCompose(idOpt ->
                idOpt
                    .map(id -> metaStorageSvcFut.thenCompose(svc -> svc.stopWatch(id)))
                    .orElseGet(() -> CompletableFuture.completedFuture(null))
            )
            .thenCompose(r ->
                watchAggregator.watch(revision, this::storeEntries)
                    .map(watch -> dispatchAppropriateMetaStorageWatch(watch).thenApply(Optional::of))
                    .orElseGet(() -> CompletableFuture.completedFuture(Optional.empty()))
            );

        return deployFut;
    }

    /**
     * Store entries with appropriate associated revision.
     *
     * @param entries to store.
     * @param revision associated revision.
     */
    private void storeEntries(Collection<IgniteBiTuple<ByteArray, byte[]>> entries, long revision) {
        Map<ByteArray, byte[]> batch = IgniteUtils.newHashMap(entries.size() + 1);

        batch.put(APPLIED_REV, longToBytes(revision));

        entries.forEach(e -> batch.put(e.getKey(), e.getValue()));

        byte[] appliedRevisionBytes = vaultMgr.get(APPLIED_REV).join().value();

        long appliedRevision = appliedRevisionBytes == null ? 0L : bytesToLong(appliedRevisionBytes);

        if (revision <= appliedRevision) {
            throw new IgniteInternalException(String.format(
                "Current revision (%d) must be greater than the revision in the Vault (%d)",
                revision, appliedRevision
            ));
        }

        vaultMgr.putAll(batch).join();
    }

    /**
     * @param id of watch to redeploy.
     * @return future, which will be completed after redeploy finished.
     */
    private CompletableFuture<Long> waitForReDeploy(long id) {
        if (deployed)
            return updateWatches().thenApply(uid -> id);
        else
            return deployFut.thenApply(uid -> id);
    }

    /**
     * Checks whether the given node hosts meta storage.
     *
     * @param nodeName Node unique name.
     * @param metastorageMembers Meta storage members names.
     * @return {@code true} if the node has meta storage, {@code false} otherwise.
     */
    public static boolean hasMetastorage(String nodeName, String[] metastorageMembers) {
        boolean isNodeHasMetasorage = false;

        for (String name : metastorageMembers) {
            if (name.equals(nodeName)) {
                isNodeHasMetasorage = true;

                break;
            }
        }

        return isNodeHasMetasorage;
    }

    /**
     * Checks whether the local node hosts meta storage.
     *
     * @param configurationMgr Configuration manager.
     * @return {@code true} if the node has meta storage, {@code false} otherwise.
     */
    public boolean hasMetastorageLocally(ConfigurationManager configurationMgr) {
        String[] metastorageMembers = configurationMgr
            .configurationRegistry()
            .getConfiguration(NodeConfiguration.KEY)
            .metastorageNodes()
            .value();

        return hasMetastorage(vaultMgr.name().join(), metastorageMembers);
    }

    // TODO: IGNITE-14691 Temporally solution that should be removed after implementing reactive watches.
    /** Cursor wrapper. */
    private final class CursorWrapper<T> implements Cursor<T> {
        /** Inner cursor future. */
        private final CompletableFuture<Cursor<T>> innerCursorFut;

        /** Inner iterator future. */
        private final CompletableFuture<Iterator<T>> innerIterFut;

        private final InnerIterator it = new InnerIterator();

        /**
         * @param innerCursorFut Inner cursor future.
         */
        CursorWrapper(
            CompletableFuture<Cursor<T>> innerCursorFut
        ) {
            this.innerCursorFut = innerCursorFut;
            this.innerIterFut = innerCursorFut.thenApply(Iterable::iterator);
        }

            /** {@inheritDoc} */
        @Override public void close() throws Exception {
            if (!busyLock.enterBusy())
                throw new NodeStoppingException("Operation has been cancelled (node is stopping).");

            try {
                innerCursorFut.thenApply(cursor -> {
                    try {
                        cursor.close();

                        return null;
                    }
                    catch (Exception e) {
                        throw new IgniteInternalException(e);
                    }
                }).get();
            }
            finally {
                busyLock.leaveBusy();
            }
        }

        /** {@inheritDoc} */
        @NotNull @Override public Iterator<T> iterator() {
            return it;
        }

        /** {@inheritDoc} */
        @Override public boolean hasNext() {
            return it.hasNext();
        }

        /** {@inheritDoc} */
        @Override public T next() {
            return it.next();
        }

        private class InnerIterator implements Iterator<T> {
            /** {@inheritDoc} */
            @Override public boolean hasNext() {
                if (!busyLock.enterBusy())
                    return false;

                try {
                    try {
                        return innerIterFut.thenApply(Iterator::hasNext).get();
                    }
                    catch (InterruptedException | ExecutionException e) {
                        throw new IgniteInternalException(e);
                    }
                }
                finally {
                    busyLock.leaveBusy();
                }
            }

            /** {@inheritDoc} */
            @Override public T next() {
                if (!busyLock.enterBusy())
                    throw new NoSuchElementException("No such element because node is stopping.");

                try {
                    try {
                        return innerIterFut.thenApply(Iterator::next).get();
                    }
                    catch (InterruptedException | ExecutionException e) {
                        throw new IgniteInternalException(e);
                    }
                }
                finally {
                    busyLock.leaveBusy();
                }
            }
        }
    }

    /**
     * Dispatches appropriate metastorage watch method according to inferred watch criterion.
     *
     * @param aggregatedWatch Aggregated watch.
     * @return Future, which will be completed after new watch registration finished.
     */
    private CompletableFuture<IgniteUuid> dispatchAppropriateMetaStorageWatch(AggregatedWatch aggregatedWatch) {
        if (aggregatedWatch.keyCriterion() instanceof KeyCriterion.CollectionCriterion) {
            var criterion = (KeyCriterion.CollectionCriterion) aggregatedWatch.keyCriterion();

            return metaStorageSvcFut.thenCompose(metaStorageSvc -> metaStorageSvc.watch(
                criterion.keys(),
                aggregatedWatch.revision(),
                aggregatedWatch.listener()));
        }
        else if (aggregatedWatch.keyCriterion() instanceof KeyCriterion.ExactCriterion) {
            var criterion = (KeyCriterion.ExactCriterion) aggregatedWatch.keyCriterion();

            return metaStorageSvcFut.thenCompose(metaStorageSvc -> metaStorageSvc.watch(
                criterion.key(),
                aggregatedWatch.revision(),
                aggregatedWatch.listener()));
        }
        else if (aggregatedWatch.keyCriterion() instanceof KeyCriterion.RangeCriterion) {
            var criterion = (KeyCriterion.RangeCriterion) aggregatedWatch.keyCriterion();

            return metaStorageSvcFut.thenCompose(metaStorageSvc -> metaStorageSvc.watch(
                criterion.from(),
                criterion.to(),
                aggregatedWatch.revision(),
                aggregatedWatch.listener()));
        }
        else
            throw new UnsupportedOperationException("Unsupported type of criterion");
    }

    /**
     * Return metastorage nodes.
     *
     * This code will be deleted after node init phase is developed.
     * https://issues.apache.org/jira/browse/IGNITE-14414
     */
    private List<ClusterNode> metastorageNodes() {
        String[] metastorageNodes = this.locCfgMgr.configurationRegistry().getConfiguration(NodeConfiguration.KEY)
            .metastorageNodes().value();

        Predicate<ClusterNode> metaStorageNodesContainsLocPred =
            clusterNode -> Arrays.asList(metastorageNodes).contains(clusterNode.name());

        List<ClusterNode> metaStorageMembers = clusterNetSvc.topologyService().allMembers().stream()
            .filter(metaStorageNodesContainsLocPred)
            .collect(Collectors.toList());

        return metaStorageMembers;
    }
}