/*
 * 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.cassandra.service.accord.interop;

import java.io.IOException;

import javax.annotation.Nullable;

import accord.local.Commands;
import accord.local.Node;
import accord.local.SafeCommand;
import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
import accord.messages.Commit;
import accord.messages.MessageType;
import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Participants;
import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.topology.Topologies;

import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.service.accord.AccordMessageSink.AccordMessageType;
import org.apache.cassandra.service.accord.serializers.CommandSerializers;
import org.apache.cassandra.service.accord.serializers.CommandSerializers.ExecuteAtSerializer;
import org.apache.cassandra.service.accord.serializers.CommitSerializers;
import org.apache.cassandra.service.accord.serializers.DepsSerializers;
import org.apache.cassandra.service.accord.serializers.IVersionedSerializer;
import org.apache.cassandra.service.accord.serializers.KeySerializers;
import org.apache.cassandra.service.accord.serializers.Version;

import static accord.messages.Commit.WithDeps.HasDeps;
import static accord.messages.Commit.WithDeps.NoDeps;
import static accord.messages.Commit.WithTxn.HasTxn;
import static accord.messages.Commit.WithTxn.NoTxn;
import static accord.primitives.SaveStatus.PreApplied;
import static accord.primitives.SaveStatus.ReadyToExecute;

public class AccordInteropStableThenRead extends AccordInteropRead
{
    // TODO (desired): duplicates a lot of StableThenReadSerializer
    public static final IVersionedSerializer<AccordInteropStableThenRead> requestSerializer = new IVersionedSerializer<>()
    {
        @Override
        public void serialize(AccordInteropStableThenRead read, DataOutputPlus out, Version version) throws IOException
        {
            CommandSerializers.txnId.serialize(read.txnId, out);
            KeySerializers.participants.serialize(read.scope, out);
            CommitSerializers.kind.serialize(read.kind, out);
            out.writeUnsignedVInt(read.minEpoch);
            ExecuteAtSerializer.serialize(read.txnId, read.executeAt, out);
            if (read.kind.withTxn != NoTxn)
                CommandSerializers.nullablePartialTxn.serialize(read.partialTxn, out, version);
            if (read.kind.withDeps == HasDeps)
                DepsSerializers.partialDeps.serialize(read.partialDeps, out);
            if (read.kind.withTxn == HasTxn)
                KeySerializers.fullRoute.serialize(read.route, out);
            ReadCommand.serializer.serialize(read.command, out, version.messageVersion());
        }

        @Override
        public AccordInteropStableThenRead deserialize(DataInputPlus in, Version version) throws IOException
        {
            TxnId txnId = CommandSerializers.txnId.deserialize(in);
            Participants<?> scope = KeySerializers.participants.deserialize(in);
            Commit.Kind kind = CommitSerializers.kind.deserialize(in);
            long minEpoch = in.readUnsignedVInt();
            Timestamp executeAt = ExecuteAtSerializer.deserialize(txnId, in);
            PartialTxn partialTxn = kind.withTxn == NoTxn ? null : CommandSerializers.nullablePartialTxn.deserialize(in, version);
            PartialDeps partialDeps = kind.withDeps == NoDeps ? null : DepsSerializers.partialDeps.deserialize(in);
            FullRoute < ?> route = kind.withTxn == HasTxn ? KeySerializers.fullRoute.deserialize(in) : null;
            ReadCommand command = ReadCommand.serializer.deserialize(in, version.messageVersion());
            return new AccordInteropStableThenRead(txnId, scope, kind, minEpoch, executeAt, partialTxn, partialDeps, route, command);
        }

        @Override
        public long serializedSize(AccordInteropStableThenRead read, Version version)
        {
            return CommandSerializers.txnId.serializedSize(read.txnId)
                   + KeySerializers.participants.serializedSize(read.scope)
                   + CommitSerializers.kind.serializedSize(read.kind)
                   + TypeSizes.sizeofUnsignedVInt(read.minEpoch)
                   + ExecuteAtSerializer.serializedSize(read.txnId, read.executeAt)
                   + (read.kind.withTxn == NoTxn ? 0 : CommandSerializers.nullablePartialTxn.serializedSize(read.partialTxn, version))
                   + (read.kind.withDeps != HasDeps ? 0 : DepsSerializers.partialDeps.serializedSize(read.partialDeps))
                   + (read.kind.withTxn != HasTxn ? 0 : KeySerializers.fullRoute.serializedSize(read.route))
                   + ReadCommand.serializer.serializedSize(read.command, version.messageVersion());
        }
    };

    // TODO (required): why is this safe to execute at PreApplied? Document.
    private static final ExecuteOn EXECUTE_ON = new ExecuteOn(ReadyToExecute, PreApplied);

    public final long minEpoch;
    public final Commit.Kind kind;
    public final Timestamp executeAt;
    public final @Nullable PartialTxn partialTxn;
    public final @Nullable PartialDeps partialDeps;
    public final @Nullable FullRoute<?> route;

    public AccordInteropStableThenRead(Node.Id to, Topologies topologies, TxnId txnId, Commit.Kind kind, Timestamp executeAt, Txn txn, Deps deps, FullRoute<?> route, ReadCommand command)
    {
        super(to, topologies, txnId, route, executeAt.epoch(), command);
        this.kind = kind;
        this.minEpoch = topologies.oldestEpoch();
        this.executeAt = executeAt;
        this.partialTxn = kind.withTxn.select(txn, route, topologies, txnId, to);
        this.partialDeps = kind.withDeps.select(deps, route);
        this.route = kind.withTxn.select(route);
    }

    public AccordInteropStableThenRead(TxnId txnId, Participants<?> scope, Commit.Kind kind, long minEpoch, Timestamp executeAt, @Nullable PartialTxn partialTxn, @Nullable PartialDeps partialDeps, @Nullable FullRoute<?> route, ReadCommand command)
    {
        super(txnId, scope, executeAt.epoch(), command);
        this.minEpoch = minEpoch;
        this.kind = kind;
        this.executeAt = executeAt;
        this.partialTxn = partialTxn;
        this.partialDeps = partialDeps;
        this.route = route;
    }

    @Override
    public CommitOrReadNack applyInternal(SafeCommandStore safeStore)
    {
        Route<?> route = this.route == null ? (Route)scope : this.route;
        StoreParticipants participants = StoreParticipants.execute(safeStore, route, txnId, minEpoch(), executeAtEpoch);
        SafeCommand safeCommand = safeStore.get(txnId, participants);
        Commands.commit(safeStore, safeCommand, participants, kind.saveStatus, Ballot.ZERO, txnId, route, partialTxn, executeAt, partialDeps, kind);
        return super.applyInternal(safeStore, safeCommand, participants);
    }

    @Override
    public ReadType kind()
    {
        return ReadType.stableThenRead;
    }

    @Override
    protected ExecuteOn executeOn()
    {
        return EXECUTE_ON;
    }

    @Override
    public MessageType type()
    {
        return AccordMessageType.INTEROP_STABLE_THEN_READ_REQ;
    }

    @Override
    public String toString()
    {
        return "AccordInteropStableThenRead{" +
               "txnId=" + txnId +
               "command=" + command +
               '}';
    }
}
