/*
 * Copyright 2019, TeamDev. All rights reserved.
 *
 * Redistribution and use in source and/or binary forms, with or without
 * modification, must retain the above copyright notice and the following
 * disclaimer.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

package io.spine.server.storage.jdbc.delivery;

import com.google.common.collect.ImmutableList;
import com.google.protobuf.Descriptors.Descriptor;
import io.spine.server.delivery.InboxMessage;
import io.spine.server.delivery.InboxMessageId;
import io.spine.server.storage.jdbc.DataSourceWrapper;
import io.spine.server.storage.jdbc.Type;
import io.spine.server.storage.jdbc.TypeMapping;
import io.spine.server.storage.jdbc.message.MessageTable;
import io.spine.server.storage.jdbc.query.IdColumn;
import io.spine.string.Stringifiers;

import javax.annotation.Nullable;
import java.util.Iterator;

import static io.spine.server.storage.jdbc.Type.BOOLEAN;
import static io.spine.server.storage.jdbc.Type.INT;
import static io.spine.server.storage.jdbc.Type.LONG;
import static io.spine.server.storage.jdbc.Type.STRING;
import static io.spine.server.storage.jdbc.Type.STRING_255;

/**
 * A table in the DB responsible for storing the {@link io.spine.server.delivery.Inbox Inbox} data.
 */
final class InboxTable extends MessageTable<InboxMessageId, InboxMessage> {

    private static final String NAME = "inbox";

    InboxTable(DataSourceWrapper dataSource, TypeMapping typeMapping) {
        super(NAME, IdColumn.of(Column.ID, InboxMessageId.class), dataSource, typeMapping);
    }

    @Override
    protected Descriptor messageDescriptor() {
        return InboxMessage.getDescriptor();
    }

    @Override
    protected Iterable<Column> messageSpecificColumns() {
        return ImmutableList.copyOf(Column.values());
    }

    Iterator<InboxMessage> readAll(long shardIndex) {
        SelectInboxMessagesByShardIndex query = composeSelectByShardIndexQuery(shardIndex);
        Iterator<InboxMessage> iterator = query.execute();
        return iterator;
    }

    private SelectInboxMessagesByShardIndex composeSelectByShardIndexQuery(long shardIndex) {
        SelectInboxMessagesByShardIndex.Builder builder =
                SelectInboxMessagesByShardIndex.newBuilder();
        SelectInboxMessagesByShardIndex query = builder.setTableName(name())
                                                       .setDataSource(dataSource())
                                                       .setShardIndex(shardIndex)
                                                       .build();
        return query;
    }

    enum Column implements MessageTable.Column<InboxMessage> {

        ID(InboxMessage::getId),

        SIGNAL_ID(STRING_255, m -> m.getSignalId()
                                    .getValue()),

        INBOX_ID(STRING_255, m -> Stringifiers.toString(m.getInboxId())),

        SHARD_INDEX(LONG, m -> m.getShardIndex()
                                .getIndex()),

        OF_TOTAL_SHARDS(LONG, m -> m.getShardIndex()
                                    .getOfTotal()),

        IS_EVENT(BOOLEAN, InboxMessage::hasEvent),

        IS_COMMAND(BOOLEAN, InboxMessage::hasCommand),

        LABEL(STRING, m -> m.getLabel()
                            .toString()),

        STATUS(STRING, m -> m.getStatus()
                             .toString()),

        WHEN_RECEIVED(LONG, m -> m.getWhenReceived()
                                  .getSeconds()),

        WHEN_RECEIVED_NANOS(INT, m -> m.getWhenReceived()
                                       .getNanos());

        @Nullable
        private final Type type;
        private final Getter<InboxMessage> getter;

        Column(Type type, Getter<InboxMessage> getter) {
            this.type = type;
            this.getter = getter;
        }

        Column(Getter<InboxMessage> getter) {
            this.type = null;
            this.getter = getter;
        }

        @Nullable
        @Override
        public Type type() {
            return type;
        }

        @Override
        public boolean isPrimaryKey() {
            return this == ID;
        }

        @Override
        public boolean isNullable() {
            return false;
        }

        @Override
        public Getter<InboxMessage> getter() {
            return getter;
        }
    }
}

@@ -1544,32 +1544,57 @@
 rs;%0A
-%0A
 import 
-javax.annotation
+org.checkerframework.checker.nullness.qual
 .Nul
@@ -1596,24 +1596,25 @@
 l.Nullable;%0A
+%0A
 import java.
@@ -4319,34 +4319,16 @@
 os());%0A%0A
-        @Nullable%0A
         
@@ -4341,16 +4341,26 @@
 e final 
+@Nullable 
 Type typ
@@ -4671,26 +4671,8 @@
  %7D%0A%0A
-        @Nullable%0A
     
@@ -4700,16 +4700,26 @@
  public 
+@Nullable 
 Type typ

Fix wrong `@Nullable` annotation used