/*
 * 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.bookkeeper.stream.storage.impl.metadata;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.bookkeeper.stream.protocol.ProtocolConstants.MIN_DATA_STREAM_ID;
import static org.apache.bookkeeper.stream.protocol.util.ProtoUtils.validateNamespaceName;
import static org.apache.bookkeeper.stream.protocol.util.ProtoUtils.validateStreamName;

import com.google.protobuf.InvalidProtocolBufferException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.api.kv.op.CompareResult;
import org.apache.bookkeeper.api.kv.op.RangeOp;
import org.apache.bookkeeper.api.kv.op.TxnOp;
import org.apache.bookkeeper.api.kv.options.Options;
import org.apache.bookkeeper.api.kv.result.KeyValue;
import org.apache.bookkeeper.common.concurrent.FutureUtils;
import org.apache.bookkeeper.common.util.Bytes;
import org.apache.bookkeeper.statelib.api.mvcc.MVCCAsyncStore;
import org.apache.bookkeeper.stream.proto.NamespaceMetadata;
import org.apache.bookkeeper.stream.proto.NamespaceProperties;
import org.apache.bookkeeper.stream.proto.StreamConfiguration;
import org.apache.bookkeeper.stream.proto.StreamName;
import org.apache.bookkeeper.stream.proto.StreamProperties;
import org.apache.bookkeeper.stream.proto.storage.CreateNamespaceRequest;
import org.apache.bookkeeper.stream.proto.storage.CreateNamespaceResponse;
import org.apache.bookkeeper.stream.proto.storage.CreateStreamRequest;
import org.apache.bookkeeper.stream.proto.storage.CreateStreamResponse;
import org.apache.bookkeeper.stream.proto.storage.DeleteNamespaceRequest;
import org.apache.bookkeeper.stream.proto.storage.DeleteNamespaceResponse;
import org.apache.bookkeeper.stream.proto.storage.DeleteStreamRequest;
import org.apache.bookkeeper.stream.proto.storage.DeleteStreamResponse;
import org.apache.bookkeeper.stream.proto.storage.GetNamespaceRequest;
import org.apache.bookkeeper.stream.proto.storage.GetNamespaceResponse;
import org.apache.bookkeeper.stream.proto.storage.GetStreamRequest;
import org.apache.bookkeeper.stream.proto.storage.GetStreamRequest.IdCase;
import org.apache.bookkeeper.stream.proto.storage.GetStreamResponse;
import org.apache.bookkeeper.stream.proto.storage.StatusCode;
import org.apache.bookkeeper.stream.protocol.util.StorageContainerPlacementPolicy;
import org.apache.bookkeeper.stream.storage.api.metadata.RootRangeStore;

/**
 * A statestore backed implementation of {@link RootRangeStore}.
 */
@Slf4j
public class RootRangeStoreImpl
    implements RootRangeStore {

    private static final byte SYSTEM_TAG = (byte) 0xff;
    private static final byte NS_NAME_TAG = (byte) 0x01;
    private static final byte NS_ID_TAG = (byte) 0x02;

    // separator used for separating streams within a same namespace
    private static final byte NS_STREAM_NAME_SEP = (byte) 0x03;
    private static final byte NS_STREAM_ID_SEP = (byte) 0x04;
    private static final byte STREAM_ID_TAG = (byte) 0x05;
    private static final byte NS_END_SEP = (byte) 0xff;

    static final byte[] NS_ID_KEY = new byte[]{SYSTEM_TAG, 'n', 's', 'i', 'd'};
    static final byte[] STREAM_ID_KEY = new byte[]{SYSTEM_TAG, 's', 't', 'r', 'e', 'a', 'm', 'i', 'd'};

    /**
     * ns name key: [NS_NAME_TAG][ns_name].
     */
    static byte[] getNamespaceNameKey(String nsName) {
        byte[] nsNameBytes = nsName.getBytes(UTF_8);
        byte[] nsNameKey = new byte[nsNameBytes.length + 1];
        System.arraycopy(nsNameBytes, 0, nsNameKey, 1, nsNameBytes.length);
        nsNameKey[0] = NS_NAME_TAG;
        return nsNameKey;
    }

    /**
     * ns id key: [NS_ID_TAG][ns_id].
     */
    static byte[] getNamespaceIdKey(long nsId) {
        byte[] nsIdBytes = new byte[Long.BYTES + 1];
        nsIdBytes[0] = NS_ID_TAG;
        Bytes.toBytes(nsId, nsIdBytes, 1);
        return nsIdBytes;
    }

    /**
     * ns id end key: [NS_ID_TAG][ns_id][NS_END_SEP].
     */
    static byte[] getNamespaceIdEndKey(long nsId) {
        byte[] nsIdBytes = new byte[Long.BYTES + 2];
        nsIdBytes[0] = NS_ID_TAG;
        Bytes.toBytes(nsId, nsIdBytes, 1);
        nsIdBytes[Long.BYTES + 1] = NS_END_SEP;
        return nsIdBytes;
    }

    /**
     * stream name key: [NS_ID_TAG][ns_id][STREAM_NAME_SEP][stream_name].
     */
    static byte[] getStreamNameKey(long nsId, String streamName) {
        byte[] streamNameBytes = streamName.getBytes(UTF_8);
        byte[] streamNameKey = new byte[streamNameBytes.length + Long.BYTES + 2];
        streamNameKey[0] = NS_ID_TAG;
        Bytes.toBytes(nsId, streamNameKey, 1);
        streamNameKey[Long.BYTES + 1] = NS_STREAM_NAME_SEP;
        System.arraycopy(streamNameBytes, 0, streamNameKey, Long.BYTES + 2, streamNameBytes.length);
        return streamNameKey;
    }

    /**
     * stream name id: [NS_ID_TAG][ns_id][STREAM_ID_SEP][stream_id].
     */
    static byte[] getStreamIdKey(long nsId, long streamId) {
        byte[] streamIdBytes = new byte[2 * Long.BYTES + 2];
        streamIdBytes[0] = NS_ID_TAG;
        Bytes.toBytes(nsId, streamIdBytes, 1);
        streamIdBytes[Long.BYTES + 1] = NS_STREAM_ID_SEP;
        Bytes.toBytes(streamId, streamIdBytes, Long.BYTES + 2);
        return streamIdBytes;
    }

    /**
     * stream id: [STREAM_ID_TAG][stream_id].
     */
    static byte[] getStreamIdKey(long streamId) {
        byte[] streamIdBytes = new byte[Long.BYTES + 1];
        streamIdBytes[0] = STREAM_ID_TAG;
        Bytes.toBytes(streamId, streamIdBytes, 1);
        return streamIdBytes;
    }

    private final MVCCAsyncStore<byte[], byte[]> store;
    private final StorageContainerPlacementPolicy placementPolicy;
    private final ScheduledExecutorService executor;

    public RootRangeStoreImpl(MVCCAsyncStore<byte[], byte[]> store,
                              StorageContainerPlacementPolicy placementPolicy,
                              ScheduledExecutorService executor) {
        this.store = store;
        this.placementPolicy = placementPolicy;
        this.executor = executor;
    }

    //
    // Namespaces API
    //

    CompletableFuture<KeyValue<byte[], byte[]>> getValue(byte[] key) {
        RangeOp<byte[], byte[]> op = store.getOpFactory().newRange(
            key,
            Options.get());
        return store.range(op).thenApplyAsync(kvs -> {
            try {
                if (kvs.count() <= 0) {
                    return null;
                } else {
                    return kvs.getKvsAndClear().get(0);
                }
            } finally {
                kvs.close();
            }
        }).whenComplete((kv, cause) -> op.close());
    }

    @Override
    public CompletableFuture<CreateNamespaceResponse> createNamespace(CreateNamespaceRequest request) {
        if (log.isTraceEnabled()) {
            log.trace("Received CreateNamespace request : {}", request);
        }

        return CreateNamespaceProcessor.of().process(
            this,
            request,
            executor);
    }

    StatusCode verifyCreateNamespaceRequest(CreateNamespaceRequest request) {
        String colName = request.getName();
        StatusCode code = StatusCode.SUCCESS;
        if (!validateNamespaceName(colName)) {
            log.error("Failed to create namespace due to invalid namespace name {}", colName);
            code = StatusCode.INVALID_NAMESPACE_NAME;
        }
        return code;
    }

    CompletableFuture<CreateNamespaceResponse> doProcessCreateNamespaceRequest(CreateNamespaceRequest request) {

        return getValue(NS_ID_KEY)
            .thenCompose(nsIdKv -> {
                long currentNsId, currentNsIdRev;
                if (null == nsIdKv) {
                    currentNsId = -1L;
                    currentNsIdRev = -1L;
                } else {
                    currentNsId = Bytes.toLong(nsIdKv.value(), 0);
                    currentNsIdRev = nsIdKv.modifiedRevision();
                    nsIdKv.close();
                }

                return executeCreateNamespaceTxn(currentNsId, currentNsIdRev, request);
            });
    }

    private CompletableFuture<CreateNamespaceResponse> executeCreateNamespaceTxn(long currentNsId,
                                                                                 long currentNsIdRev,
                                                                                 CreateNamespaceRequest request) {
        long namespaceId = currentNsId + 1;
        String nsName = request.getName();

        NamespaceMetadata metadata = NamespaceMetadata.newBuilder()
            .setProps(NamespaceProperties.newBuilder()
                .setNamespaceId(namespaceId)
                .setNamespaceName(nsName)
                .setDefaultStreamConf(request.getNsConf().getDefaultStreamConf()))
            .build();

        byte[] nsNameKey = getNamespaceNameKey(nsName);
        byte[] nsNameVal = Bytes.toBytes(namespaceId);
        byte[] nsIdKey = getNamespaceIdKey(namespaceId);
        byte[] nsIdVal = metadata.toByteArray();

        TxnOp<byte[], byte[]> txn = store.newTxn()
            .If(
                store.newCompareValue(CompareResult.EQUAL, nsNameKey, null),
                currentNsIdRev < 0
                    ? store.newCompareValue(CompareResult.EQUAL, NS_ID_KEY, null) :
                    store.newCompareModRevision(CompareResult.EQUAL, NS_ID_KEY, currentNsIdRev)
            )
            .Then(
                store.newPut(nsNameKey, nsNameVal),
                store.newPut(nsIdKey, nsIdVal),
                store.newPut(NS_ID_KEY, Bytes.toBytes(namespaceId)))
            .build();
        return store.txn(txn)
            .thenApply(txnResult -> {
                try {
                    CreateNamespaceResponse.Builder respBuilder = CreateNamespaceResponse.newBuilder();
                    if (txnResult.isSuccess()) {
                        respBuilder.setCode(StatusCode.SUCCESS);
                        respBuilder.setNsProps(metadata.getProps());
                    } else {
                        // TODO: differentiate the error code
                        respBuilder.setCode(StatusCode.INTERNAL_SERVER_ERROR);
                    }
                    return respBuilder.build();
                } finally {
                    txnResult.close();
                }
            })
            .whenComplete((resp, cause) -> txn.close());
    }

    @Override
    public CompletableFuture<DeleteNamespaceResponse> deleteNamespace(DeleteNamespaceRequest request) {
        return DeleteNamespaceProcessor.of().process(
            this,
            request,
            executor);
    }

    StatusCode verifyDeleteNamespaceRequest(DeleteNamespaceRequest request) {
        String colName = request.getName();
        StatusCode code = StatusCode.SUCCESS;
        if (!validateNamespaceName(colName)) {
            log.error("Failed to delete namespace due to invalid namespace name {}", colName);
            code = StatusCode.INVALID_NAMESPACE_NAME;
        }
        return code;
    }

    CompletableFuture<DeleteNamespaceResponse> doProcessDeleteNamespaceRequest(DeleteNamespaceRequest request) {

        String nsName = request.getName();

        return getNamespace(nsName)
            .thenCompose(nsMetadata -> deleteNamespace(nsName, nsMetadata));
    }

    CompletableFuture<DeleteNamespaceResponse> deleteNamespace(String nsName, NamespaceMetadata nsMetadata) {
        if (null == nsMetadata) {
            return FutureUtils.value(
                DeleteNamespaceResponse.newBuilder()
                    .setCode(StatusCode.NAMESPACE_NOT_FOUND)
                    .build());
        }

        byte[] nsNameKey = getNamespaceNameKey(nsName);
        byte[] nsIdKey = getNamespaceIdKey(nsMetadata.getProps().getNamespaceId());
        byte[] nsIdEndKey = getNamespaceIdEndKey(nsMetadata.getProps().getNamespaceId());


        TxnOp<byte[], byte[]> txnOp = store.newTxn()
            .If(
                store.newCompareValue(CompareResult.NOT_EQUAL, nsNameKey, null),
                store.newCompareValue(CompareResult.NOT_EQUAL, nsIdKey, null)
            )
            .Then(
                store.newDelete(nsNameKey),
                store.newDeleteRange(nsIdKey, nsIdEndKey)
            )
            .build();

        return store.txn(txnOp).thenApply(txnResult -> {
            try {
                DeleteNamespaceResponse.Builder respBuilder = DeleteNamespaceResponse.newBuilder();
                if (txnResult.isSuccess()) {
                    respBuilder.setCode(StatusCode.SUCCESS);
                } else {
                    // TODO: differentiate the error code
                    respBuilder.setCode(StatusCode.INTERNAL_SERVER_ERROR);
                }
                return respBuilder.build();
            } finally {
                txnResult.close();
            }
        }).whenComplete((resp, cause) -> txnOp.close());
    }

    @Override
    public CompletableFuture<GetNamespaceResponse> getNamespace(GetNamespaceRequest request) {
        String nsName = request.getName();
        if (!validateNamespaceName(nsName)) {
            log.error("Failed to get namespace due to invalid namespace name {}", nsName);
            return FutureUtils.value(
                GetNamespaceResponse.newBuilder()
                    .setCode(StatusCode.INVALID_NAMESPACE_NAME)
                    .build());
        } else {
            return getNamespace(request.getName()).thenApply(nsMetadata -> {
                GetNamespaceResponse.Builder nsRespBuilder = GetNamespaceResponse.newBuilder();
                if (null == nsMetadata) {
                    nsRespBuilder.setCode(StatusCode.NAMESPACE_NOT_FOUND);
                } else {
                    nsRespBuilder.setCode(StatusCode.SUCCESS);
                    nsRespBuilder.setNsProps(nsMetadata.getProps());
                }
                return nsRespBuilder.build();
            });
        }
    }

    private CompletableFuture<NamespaceMetadata> getNamespace(long nsId) {
        byte[] nsIdKey = getNamespaceIdKey(nsId);
        return store.get(nsIdKey).thenCompose(nsIdVal -> {
            try {
                return FutureUtils.value(null != nsIdVal ? NamespaceMetadata.parseFrom(nsIdVal) : null);
            } catch (InvalidProtocolBufferException e) {
                return FutureUtils.exception(e);
            }
        });
    }

    private CompletableFuture<NamespaceMetadata> getNamespace(String nsName) {
        byte[] nsNameKey = getNamespaceNameKey(nsName);
        return store.get(nsNameKey)
            .thenCompose(nsIdBytes -> {
                if (null == nsIdBytes) {
                    return FutureUtils.value(null);
                } else {
                    long nsId = Bytes.toLong(nsIdBytes, 0);
                    return getNamespace(nsId);
                }
            });
    }

    //
    // Stream API
    //

    @Override
    public CompletableFuture<CreateStreamResponse> createStream(CreateStreamRequest request) {
        String streamName = request.getName();
        String nsName = request.getNsName();

        StatusCode code = verifyStreamRequest(nsName, streamName);
        if (StatusCode.SUCCESS != code) {
            return FutureUtils.value(CreateStreamResponse.newBuilder().setCode(code).build());
        }

        return createStream(nsName, streamName, request.getStreamConf());
    }

    StatusCode verifyStreamRequest(String nsName, String streamName) {
        StatusCode code = StatusCode.SUCCESS;
        if (!validateNamespaceName(nsName)) {
            log.error("Invalid namespace name {}", nsName);
            code = StatusCode.INVALID_NAMESPACE_NAME;
        } else if (!validateStreamName(streamName)) {
            log.error("Invalid stream name {}", streamName);
            code = StatusCode.INVALID_STREAM_NAME;
        }
        return code;
    }

    private CompletableFuture<CreateStreamResponse> createStream(String nsName,
                                                                 String streamName,
                                                                 StreamConfiguration streamConf) {
        return getNamespace(nsName)
            .thenCompose(nsMetadata -> createStream(
                nsMetadata,
                streamName,
                streamConf
            ))
            .exceptionally(cause ->
                CreateStreamResponse.newBuilder().setCode(StatusCode.INTERNAL_SERVER_ERROR).build());
    }

    private CompletableFuture<CreateStreamResponse> createStream(NamespaceMetadata nsMetadata,
                                                                 String streamName,
                                                                 StreamConfiguration streamConf) {
        if (null == nsMetadata) {
            return FutureUtils.value(CreateStreamResponse.newBuilder().setCode(StatusCode.NAMESPACE_NOT_FOUND).build());
        }

        return getValue(STREAM_ID_KEY)
            .thenCompose(streamIdKv -> {

                long currentStreamId = -1L;
                long currentStreamIdRev = -1L;
                if (null != streamIdKv) {
                    currentStreamId = Bytes.toLong(streamIdKv.value(), 0);
                    currentStreamIdRev = streamIdKv.modifiedRevision();
                    streamIdKv.close();
                }

                return executeCreateStreamTxn(
                    nsMetadata.getProps().getNamespaceId(),
                    streamName,
                    streamConf,
                    currentStreamId,
                    currentStreamIdRev);
            });

    }

    private static void logStreamOp(String op, long nsId, long streamId, String streamName, StatusCode result) {
        log.info("stream {} namespace_id={} stream_id={} stream_name={} result={}",
                op, nsId, streamId, streamName, result);
    }

    private CompletableFuture<CreateStreamResponse> executeCreateStreamTxn(long nsId,
                                                                           String streamName,
                                                                           StreamConfiguration streamConf,
                                                                           long currentStreamId,
                                                                           long currentStreamIdRev) {
        long streamId;
        if (currentStreamId < 0) {
            streamId = MIN_DATA_STREAM_ID;
        } else {
            streamId = currentStreamId + 1;
        }

        long scId = placementPolicy.placeStreamRange(streamId, 0L);

        StreamProperties streamProps = StreamProperties.newBuilder()
            .setStreamId(streamId)
            .setStreamName(streamName)
            .setStorageContainerId(scId)
            .setStreamConf(streamConf)
            .build();

        byte[] nsIdKey = getNamespaceIdKey(nsId);
        byte[] streamNameKey = getStreamNameKey(nsId, streamName);
        byte[] streamNameVal = Bytes.toBytes(streamId);
        byte[] streamIdKey = getStreamIdKey(nsId, streamId);
        byte[] streamIdVal = streamProps.toByteArray();
        byte[] streamReverseIndexKey = getStreamIdKey(streamId);
        byte[] streamReverseIndexValue = Bytes.toBytes(nsId);

        TxnOp<byte[], byte[]> txn = store.newTxn()
            .If(
                store.newCompareValue(CompareResult.NOT_EQUAL, nsIdKey, null),
                currentStreamIdRev < 0
                    ? store.newCompareValue(CompareResult.EQUAL, STREAM_ID_KEY, null) :
                    store.newCompareModRevision(CompareResult.EQUAL, STREAM_ID_KEY, currentStreamIdRev),
                store.newCompareValue(CompareResult.EQUAL, streamNameKey, null)
            )
            .Then(
                store.newPut(streamNameKey, streamNameVal),
                store.newPut(streamIdKey, streamIdVal),
                store.newPut(streamReverseIndexKey, streamReverseIndexValue),
                store.newPut(STREAM_ID_KEY, Bytes.toBytes(streamId))
            )
            .build();

        return store.txn(txn)
            .thenApply(txnResult -> {
                try {
                    CreateStreamResponse.Builder respBuilder = CreateStreamResponse.newBuilder();
                    if (txnResult.isSuccess()) {
                        respBuilder.setCode(StatusCode.SUCCESS);
                        respBuilder.setStreamProps(streamProps);
                    } else {
                        // TODO: differentiate the error codes
                        respBuilder.setCode(StatusCode.INTERNAL_SERVER_ERROR);
                    }
                    logStreamOp("create", nsId, streamId, streamName, respBuilder.getCode());
                    return respBuilder.build();
                } finally {
                    txnResult.close();
                    txn.close();
                }
            })
            .exceptionally(cause -> {
                txn.close();
                return CreateStreamResponse.newBuilder().setCode(StatusCode.INTERNAL_SERVER_ERROR).build();
            });
    }

    @Override
    public CompletableFuture<DeleteStreamResponse> deleteStream(DeleteStreamRequest request) {
        String streamName = request.getName();
        String nsName = request.getNsName();

        StatusCode code = verifyStreamRequest(nsName, streamName);
        if (StatusCode.SUCCESS != code) {
            return FutureUtils.value(DeleteStreamResponse.newBuilder().setCode(code).build());
        }

        return deleteStream(nsName, streamName);
    }

    private CompletableFuture<DeleteStreamResponse> deleteStream(String nsName,
                                                                 String streamName) {
        return getNamespace(nsName)
            .thenCompose(nsMetadata -> deleteStream(nsMetadata, streamName));
    }

    private CompletableFuture<DeleteStreamResponse> deleteStream(NamespaceMetadata nsMetadata,
                                                                 String streamName) {

        if (null == nsMetadata) {
            return FutureUtils.value(DeleteStreamResponse.newBuilder().setCode(StatusCode.NAMESPACE_NOT_FOUND).build());
        }

        long nsId = nsMetadata.getProps().getNamespaceId();
        byte[] streamNameKey = getStreamNameKey(nsId, streamName);

        return store.get(streamNameKey).thenCompose(streamIdBytes -> {
            if (null == streamIdBytes) {
                return FutureUtils.value(
                    DeleteStreamResponse.newBuilder()
                        .setCode(StatusCode.STREAM_NOT_FOUND)
                        .build());
            }

            long streamId = Bytes.toLong(streamIdBytes, 0);
            return deleteStream(
                nsId,
                streamId,
                streamName);
        });
    }

    private CompletableFuture<DeleteStreamResponse> deleteStream(long nsId,
                                                                 long streamId,
                                                                 String streamName) {
        byte[] nsIdKey = getNamespaceIdKey(nsId);
        byte[] streamNameKey = getStreamNameKey(nsId, streamName);
        byte[] streamIdKey = getStreamIdKey(nsId, streamId);
        byte[] streamReverseIndexKey = getStreamIdKey(streamId);

        TxnOp<byte[], byte[]> txnOp = store.newTxn()
            .If(
                store.newCompareValue(CompareResult.NOT_EQUAL, nsIdKey, null),
                store.newCompareValue(CompareResult.NOT_EQUAL, streamNameKey, null),
                store.newCompareValue(CompareResult.NOT_EQUAL, streamIdKey, null),
                store.newCompareValue(CompareResult.NOT_EQUAL, streamReverseIndexKey, null)
            )
            .Then(
                store.newDelete(streamIdKey),
                store.newDelete(streamNameKey),
                store.newDelete(streamReverseIndexKey)
            )
            .build();

        return store.txn(txnOp).thenApply(txnResult -> {
            try {
                DeleteStreamResponse.Builder respBuilder = DeleteStreamResponse.newBuilder();
                if (txnResult.isSuccess()) {
                    respBuilder.setCode(StatusCode.SUCCESS);
                } else {
                    respBuilder.setCode(StatusCode.INTERNAL_SERVER_ERROR);
                }
                logStreamOp("delete", nsId, streamId, streamName, respBuilder.getCode());
                return respBuilder.build();
            } finally {
                txnResult.close();
            }
        }).whenComplete((resp, cause) -> txnOp.close());
    }

    private CompletableFuture<GetStreamResponse> streamPropertiesToResponse(
        CompletableFuture<StreamProperties> propsFuture
    ) {
        GetStreamResponse.Builder respBuilder = GetStreamResponse.newBuilder();
        return propsFuture.thenCompose(streamProps -> {
            if (null == streamProps) {
                return FutureUtils.value(respBuilder.setCode(StatusCode.STREAM_NOT_FOUND).build());
            } else {
                return FutureUtils.value(respBuilder
                    .setCode(StatusCode.SUCCESS)
                    .setStreamProps(streamProps)
                    .build());
            }
        }).exceptionally(cause ->
            respBuilder
                .setCode(StatusCode.INTERNAL_SERVER_ERROR)
                .build()
        );
    }

    @Override
    public CompletableFuture<GetStreamResponse> getStream(GetStreamRequest request) {
        if (IdCase.STREAM_ID == request.getIdCase()) {
            return streamPropertiesToResponse(
                getStreamProps(request.getStreamId()));
        } else if (IdCase.STREAM_NAME == request.getIdCase()) {
            return getStreamProps(request.getStreamName());
        } else {
            return FutureUtils.value(GetStreamResponse.newBuilder()
                .setCode(StatusCode.ILLEGAL_OP)
                .build());
        }
    }

    CompletableFuture<StreamProperties> getStreamProps(long streamId) {
        byte[] streamReverseIndexKey = getStreamIdKey(streamId);

        return store.get(streamReverseIndexKey).thenCompose(nsIdBytes -> {
            if (null == nsIdBytes) {
                return FutureUtils.value(null);
            }

            long nsId = Bytes.toLong(nsIdBytes, 0);
            return getStreamProps(nsId, streamId);
        });
    }

    CompletableFuture<GetStreamResponse> getStreamProps(StreamName streamName) {
        StatusCode code = verifyStreamRequest(
                streamName.getNamespaceName(),
                streamName.getStreamName());
        if (StatusCode.SUCCESS != code) {
            return FutureUtils.value(GetStreamResponse.newBuilder()
                .setCode(code).build());
        }

        byte[] nsNameKey = getNamespaceNameKey(streamName.getNamespaceName());


        return store.get(nsNameKey)
            .thenCompose(nsIdBytes -> {
                if (null == nsIdBytes) {
                    return FutureUtils.value(GetStreamResponse.newBuilder()
                        .setCode(StatusCode.NAMESPACE_NOT_FOUND).build());
                }
                long nsId = Bytes.toLong(nsIdBytes, 0);
                return streamPropertiesToResponse(
                    getStreamProps(nsId, streamName.getStreamName()));
            });
    }

    CompletableFuture<StreamProperties> getStreamProps(long nsId, String streamName) {
        byte[] streamNameKey = getStreamNameKey(nsId, streamName);

        return store.get(streamNameKey).thenCompose(streamIdBytes -> {
            if (null == streamIdBytes) {
                return FutureUtils.value(null);
            }

            long streamId = Bytes.toLong(streamIdBytes, 0);
            return getStreamProps(nsId, streamId);
        });
    }

    CompletableFuture<StreamProperties> getStreamProps(long nsId, long streamId) {
        byte[] streamIdKey = getStreamIdKey(nsId, streamId);

        return store.get(streamIdKey).thenCompose(streamPropBytes -> {
            if (null == streamPropBytes) {
                return FutureUtils.value(null);
            } else {
                try {
                    return FutureUtils.value(StreamProperties.parseFrom(streamPropBytes));
                } catch (InvalidProtocolBufferException e) {
                    return FutureUtils.exception(e);
                }
            }
        });
    }

}
