/*
 * 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.locator;

import java.util.Set;
import java.util.function.Predicate;

import com.google.common.annotations.VisibleForTesting;

import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.reads.ReadCoordinator;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ownership.DataPlacement;
import org.apache.cassandra.utils.FBUtilities;

/**
 * The relevant replicas for an operation over a given range or token.
 *
 * @param <E>
 */
public abstract class ReplicaLayout<E extends Endpoints<E>>
{
    private final E natural;
    // the snapshot of the replication strategy that corresponds to the replica layout
    private final AbstractReplicationStrategy replicationStrategy;

    ReplicaLayout(AbstractReplicationStrategy replicationStrategy, E natural)
    {
        this.replicationStrategy = replicationStrategy;
        this.natural = natural;
    }

    /**
     * The 'natural' owners of the ring position(s), as implied by the current ring layout.
     * This excludes any pending owners, i.e. those that are in the process of taking ownership of a range, but
     * have not yet finished obtaining their view of the range.
     */
    public final E natural()
    {
        return natural;
    }

    public final AbstractReplicationStrategy replicationStrategy()
    {
        return replicationStrategy;
    }

    /**
     * All relevant owners of the ring position(s) for this operation, as implied by the current ring layout.
     * For writes, this will include pending owners, and for reads it will be equivalent to natural()
     */
    public E all()
    {
        return natural;
    }

    public String toString()
    {
        return "ReplicaLayout [ natural: " + natural + " ]";
    }

    public static class ForTokenRead extends ReplicaLayout<EndpointsForToken> implements ForToken
    {
        public ForTokenRead(AbstractReplicationStrategy replicationStrategy, EndpointsForToken natural)
        {
            super(replicationStrategy, natural);
        }

        @Override
        public Token token()
        {
            return natural().token();
        }

        public ReplicaLayout.ForTokenRead filter(Predicate<Replica> filter)
        {
            EndpointsForToken filtered = natural().filter(filter);
            // AbstractReplicaCollection.filter returns itself if all elements match the filter
            if (filtered == natural()) return this;
            return new ReplicaLayout.ForTokenRead(replicationStrategy(), filtered);
        }
    }

    public static class ForRangeRead extends ReplicaLayout<EndpointsForRange> implements ForRange
    {
        final AbstractBounds<PartitionPosition> range;

        public ForRangeRead(AbstractReplicationStrategy replicationStrategy, AbstractBounds<PartitionPosition> range, EndpointsForRange natural)
        {
            super(replicationStrategy, natural);
            this.range = range;
        }

        @Override
        public AbstractBounds<PartitionPosition> range()
        {
            return range;
        }

        public ReplicaLayout.ForRangeRead filter(Predicate<Replica> filter)
        {
            EndpointsForRange filtered = natural().filter(filter);
            // AbstractReplicaCollection.filter returns itself if all elements match the filter
            if (filtered == natural()) return this;
            return new ReplicaLayout.ForRangeRead(replicationStrategy(), range(), filtered);
        }
    }

    public static class ForWrite<E extends Endpoints<E>> extends ReplicaLayout<E>
    {
        final E all;
        final E pending;

        ForWrite(AbstractReplicationStrategy replicationStrategy, E natural, E pending, E all)
        {
            super(replicationStrategy, natural);
            assert pending != null && !haveWriteConflicts(natural, pending);
            if (all == null)
                all = Endpoints.concat(natural, pending);
            this.all = all;
            this.pending = pending;
        }

        public final E all()
        {
            return all;
        }

        public final E pending()
        {
            return pending;
        }

        public String toString()
        {
            return "ReplicaLayout [ natural: " + natural() + ", pending: " + pending + " ]";
        }
    }

    public static class ForTokenWrite extends ForWrite<EndpointsForToken> implements ForToken
    {
        public ForTokenWrite(AbstractReplicationStrategy replicationStrategy, EndpointsForToken natural, EndpointsForToken pending)
        {
            this(replicationStrategy, natural, pending, null);
        }

        public ForTokenWrite(AbstractReplicationStrategy replicationStrategy, EndpointsForToken natural, EndpointsForToken pending, EndpointsForToken all)
        {
            super(replicationStrategy, natural, pending, all);
        }

        @Override
        public Token token() { return natural().token(); }

        public ForTokenWrite filter(Predicate<Replica> filter)
        {
            EndpointsForToken filtered = all().filter(filter);
            // AbstractReplicaCollection.filter returns itself if all elements match the filter
            if (filtered == all()) return this;
            if (pending().isEmpty()) return new ForTokenWrite(replicationStrategy(), filtered, pending(), filtered);
            // unique by endpoint, so can for efficiency filter only on endpoint
            return new ForTokenWrite(
                    replicationStrategy(),
                    natural().keep(filtered.endpoints()),
                    pending().keep(filtered.endpoints()),
                    filtered);
        }
    }

    public interface ForRange
    {
        public AbstractBounds<PartitionPosition> range();
    }

    public interface ForToken
    {
        public Token token();
    }

    /**
     * Gets the 'natural' and 'pending' replicas that own a given token, with no filtering or processing.
     *
     * Since a write is intended for all nodes (except, unless necessary, transient replicas), this method's
     * only responsibility is to fetch the 'natural' and 'pending' replicas, then resolve any conflicts
     * {@link ReplicaLayout#haveWriteConflicts(Endpoints, Endpoints)}
     */
    public static ReplicaLayout.ForTokenWrite forTokenWriteLiveAndDown(Keyspace ks, Token token)
    {
        return forTokenWriteLiveAndDown(ks.getMetadata(), token);
    }

    public static ReplicaLayout.ForTokenWrite forTokenWriteLiveAndDown(KeyspaceMetadata ks, Token token)
    {
        ClusterMetadata metadata = ClusterMetadata.current();
        return forTokenWriteLiveAndDown(metadata, ks, token);
    }

    // TODO: cleanup/remove Keyspace overloads
    public static ReplicaLayout.ForTokenWrite forTokenWriteLiveAndDown(ClusterMetadata metadata, Keyspace ks, Token token)
    {
        return forTokenWriteLiveAndDown(metadata, ks.getMetadata(), token);
    }

    public static ReplicaLayout.ForTokenWrite forTokenWriteLiveAndDown(ClusterMetadata metadata, KeyspaceMetadata ks, Token token)
    {
        AbstractReplicationStrategy replicationStrategy = ks.replicationStrategy;
        EndpointsForToken natural;
        EndpointsForToken pending;
        if (ks.params.replication.isLocal())
        {
            natural = forLocalStrategyToken(metadata, replicationStrategy, token);
            pending = EndpointsForToken.empty(token);
        }
        else
        {
            // todo deduplicate so that "pending" contains "read - write",
            // which is a hack until we revisit how consistency level handles pending
            DataPlacement dataPlacement = metadata.placement(ks.params.replication);
            natural = forNonLocalStrategyTokenRead(dataPlacement, token);
            // perf optimization to avoid double endpoints search and filtering for a typical case
            // DataPlacement constructor does a deduplication of reads/writes, so we can use cheap == comparision here
            if (dataPlacement.reads == dataPlacement.writes)
                pending = EndpointsForToken.empty(token);
            else
                pending = forNonLocalStrategyTokenWrite(dataPlacement, token).without(natural.endpoints());
        }
        return forTokenWrite(replicationStrategy, natural, pending);
    }

    public static ReplicaLayout.ForTokenWrite forTokenWrite(AbstractReplicationStrategy replicationStrategy, EndpointsForToken natural, EndpointsForToken pending)
    {
        if (haveWriteConflicts(natural, pending))
        {
            natural = resolveWriteConflictsInNatural(natural, pending);
            pending = resolveWriteConflictsInPending(natural, pending);
        }
        return new ReplicaLayout.ForTokenWrite(replicationStrategy, natural, pending);
    }

    /**
     * Detect if we have any endpoint in both pending and full; this can occur either due to races (there is no isolation)
     * or because an endpoint is transitioning between full and transient replication status.
     *
     * We essentially always prefer the full version for writes, because this is stricter.
     *
     * For transient->full transitions:
     *
     *   Since we always write to any pending transient replica, effectively upgrading it to full for the transition duration,
     *   it might at first seem to be OK to continue treating the conflict replica as its 'natural' transient form,
     *   as there is always a quorum of nodes receiving the write.  However, ring ownership changes are not atomic or
     *   consistent across the cluster, and it is possible for writers to see different ring states.
     *
     *   Furthermore, an operator would expect that the full node has received all writes, with no extra need for repair
     *   (as the normal contract dictates) when it completes its transition.
     *
     *   While we cannot completely eliminate risks due to ring inconsistencies, this approach is the most conservative
     *   available to us today to mitigate, and (we think) the easiest to reason about.
     *
     * For full->transient transitions:
     *
     *   In this case, things are dicier, because in theory we can trigger this change instantly.  All we need to do is
     *   drop some data, surely?
     *
     *   Ring movements can put us in a pickle; any other node could believe us to be full when we have become transient,
     *   and perform a full data request to us that we believe ourselves capable of answering, but that we are not.
     *   If the ring is inconsistent, it's even feasible that a transient request would be made to the node that is losing
     *   its transient status, that also does not know it has yet done so, resulting in all involved nodes being unaware
     *   of the data inconsistency.
     *
     *   This happens because ring ownership changes are implied by a single node; not all owning nodes get a say in when
     *   the transition takes effect.  As such, a node can hold an incorrect belief about its own ownership ranges.
     *
     *   This race condition is somewhat inherent in present day Cassandra, and there's actually a limit to what we can do about it.
     *   It is a little more dangerous with transient replication, however, because we can completely answer a request without
     *   ever touching a digest, meaning we are less likely to attempt to repair any inconsistency.
     *
     *   We aren't guaranteed to contact any different nodes for the data requests, of course, though we at least have a chance.
     *
     * Note: If we have any pending transient->full movement, we need to move the full replica to our 'natural' bucket
     * to avoid corrupting our count.  This is fine for writes, all we're doing is ensuring we always write to the node,
     * instead of selectively.
     *
     * @param natural
     * @param pending
     * @param <E>
     * @return
     */
    static <E extends Endpoints<E>> boolean haveWriteConflicts(E natural, E pending)
    {
        if (pending.isEmpty())
            return false;
        Set<InetAddressAndPort> naturalEndpoints = natural.endpoints();
        for (InetAddressAndPort pendingEndpoint : pending.endpoints())
        {
            if (naturalEndpoints.contains(pendingEndpoint))
                return true;
        }
        return false;
    }

    /**
     * MUST APPLY FIRST
     * See {@link ReplicaLayout#haveWriteConflicts}
     * @return a 'natural' replica collection, that has had its conflicts with pending repaired
     */
    @VisibleForTesting
    static EndpointsForToken resolveWriteConflictsInNatural(EndpointsForToken natural, EndpointsForToken pending)
    {
        EndpointsForToken.Builder resolved = natural.newBuilder(natural.size());
        for (Replica replica : natural)
        {
            // always prefer the full natural replica, if there is a conflict
            if (replica.isTransient())
            {
                Replica conflict = pending.byEndpoint().get(replica.endpoint());
                if (conflict != null)
                {
                    // it should not be possible to have conflicts of the same replication type for the same range
                    assert conflict.isFull();
                    // If we have any pending transient->full movement, we need to move the full replica to our 'natural' bucket
                    // to avoid corrupting our count
                    resolved.add(conflict);
                    continue;
                }
            }
            resolved.add(replica);
        }
        return resolved.build();
    }

    /**
     * MUST APPLY SECOND
     * See {@link ReplicaLayout#haveWriteConflicts}
     * @return a 'pending' replica collection, that has had its conflicts with natural repaired
     */
    @VisibleForTesting
    static EndpointsForToken resolveWriteConflictsInPending(EndpointsForToken natural, EndpointsForToken pending)
    {
        return pending.without(natural.endpoints());
    }

    /**
     * @return the read layout for a token - this includes natural replicas, i.e. those that are not pending.
     * They are reverse sorted by the badness score of the configured snitch
     */
    static ReplicaLayout.ForTokenRead forTokenReadSorted(ClusterMetadata metadata, Keyspace keyspace, AbstractReplicationStrategy replicationStrategy, TableId tableId, Token token, ReadCoordinator coordinator)
    {
        EndpointsForToken replicas = keyspace.getMetadata().params.replication.isLocal()
                                     ? forLocalStrategyToken(metadata, replicationStrategy, token)
                                     : coordinator.forNonLocalStrategyTokenRead(metadata, keyspace.getMetadata(), tableId, token);

        replicas = DatabaseDescriptor.getNodeProximity().sortedByProximity(FBUtilities.getBroadcastAddressAndPort(), replicas);

        return new ReplicaLayout.ForTokenRead(replicationStrategy, replicas);
    }

    /**
     * TODO: we should really double check that the provided range does not overlap multiple token ring regions
     * @return the read layout for a range - these are reverse sorted by the badness score of the configured snitch
     */
    static ReplicaLayout.ForRangeRead forRangeReadSorted(ClusterMetadata metadata, Keyspace keyspace, AbstractReplicationStrategy replicationStrategy, AbstractBounds<PartitionPosition> range)
    {
        EndpointsForRange replicas = keyspace.getMetadata().params.replication.isLocal()
                                     ? forLocalStrategyRange(metadata, replicationStrategy, range)
                                     : forNonLocalStategyRangeRead(metadata, keyspace.getMetadata(), range);

        replicas = DatabaseDescriptor.getNodeProximity().sortedByProximity(FBUtilities.getBroadcastAddressAndPort(), replicas);
        return new ReplicaLayout.ForRangeRead(replicationStrategy, range, replicas);
    }

    static EndpointsForRange forNonLocalStategyRangeRead(ClusterMetadata metadata, KeyspaceMetadata keyspace, AbstractBounds<PartitionPosition> range)
    {
        return metadata.placement(keyspace.params.replication).reads.forRange(range.right.getToken()).get();
    }

    public static EndpointsForToken forNonLocalStrategyTokenRead(ClusterMetadata metadata, KeyspaceMetadata keyspace, Token token)
    {
        return forNonLocalStrategyTokenRead(metadata.placement(keyspace.params.replication), token);
    }

    public static EndpointsForToken forNonLocalStrategyTokenRead(DataPlacement dataPlacement, Token token)
    {
        return dataPlacement.reads.forToken(token).get();
    }

    private static EndpointsForToken forNonLocalStrategyTokenWrite(DataPlacement dataPlacement, Token token)
    {
        return dataPlacement.writes.forToken(token).get();
    }

    static EndpointsForRange forLocalStrategyRange(ClusterMetadata metadata, AbstractReplicationStrategy replicationStrategy, AbstractBounds<PartitionPosition> range)
    {
        return replicationStrategy.calculateNaturalReplicas(range.right.getToken(), metadata);
    }

    static EndpointsForToken forLocalStrategyToken(ClusterMetadata metadata, AbstractReplicationStrategy replicationStrategy, Token t)
    {
        if (!(t instanceof LocalPartitioner.LocalToken))
            return replicationStrategy.calculateNaturalReplicas(t, metadata).forToken(t);

        // local tokens use a different partitioner than the global one... so update the ranges
        return EndpointsForToken.of(t, new Replica(FBUtilities.getBroadcastAddressAndPort(), new Range<>(t, t), true));
    }
}
