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

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.junit.Assert;
import org.junit.Test;

import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
import net.bytebuddy.implementation.MethodDelegation;
import net.bytebuddy.implementation.bind.annotation.SuperCall;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.api.TokenSupplier;
import org.apache.cassandra.distributed.shared.NetworkTopology;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.locator.ReplicaPlan;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.service.PendingRangeCalculatorService;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.reads.repair.BlockingReadRepair;
import org.apache.cassandra.service.reads.repair.ReadRepairStrategy;
import org.apache.cassandra.utils.concurrent.SimpleCondition;

import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.QUORUM;
import static org.apache.cassandra.distributed.shared.AssertUtils.assertEquals;
import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
import static org.apache.cassandra.distributed.shared.AssertUtils.row;
import static org.apache.cassandra.net.Verb.READ_REPAIR_REQ;
import static org.apache.cassandra.net.Verb.READ_REPAIR_RSP;
import static org.apache.cassandra.net.Verb.READ_REQ;
import static org.junit.Assert.fail;

public class ReadRepairTest extends TestBaseImpl
{
    /**
     * Tests basic behaviour of read repair with {@code BLOCKING} read repair strategy.
     */
    @Test
    public void testBlockingReadRepair() throws Throwable
    {
        testReadRepair(ReadRepairStrategy.BLOCKING);
    }
    /**
     *
     * Tests basic behaviour of read repair with {@code NONE} read repair strategy.
     */
    @Test
    public void testNoneReadRepair() throws Throwable
    {
        testReadRepair(ReadRepairStrategy.NONE);
    }

    private void testReadRepair(ReadRepairStrategy strategy) throws Throwable
    {
        try (Cluster cluster = init(Cluster.create(3)))
        {
            cluster.schemaChange(withKeyspace("CREATE TABLE %s.t (k int, c int, v int, PRIMARY KEY (k, c)) " +
                                              String.format("WITH read_repair='%s'", strategy)));

            Object[] row = row(1, 1, 1);
            String insertQuery = withKeyspace("INSERT INTO %s.t (k, c, v) VALUES (?, ?, ?)");
            String selectQuery = withKeyspace("SELECT * FROM %s.t WHERE k=1");

            // insert data in two nodes, simulating a quorum write that has missed one node
            cluster.get(1).executeInternal(insertQuery, row);
            cluster.get(2).executeInternal(insertQuery, row);

            // verify that the third node doesn't have the row
            assertRows(cluster.get(3).executeInternal(selectQuery));

            // read with CL=QUORUM to trigger read repair
            assertRows(cluster.coordinator(3).execute(selectQuery, QUORUM), row);

            // verify whether the coordinator has the repaired row depending on the read repair strategy
            if (strategy == ReadRepairStrategy.NONE)
                assertRows(cluster.get(3).executeInternal(selectQuery));
            else
                assertRows(cluster.get(3).executeInternal(selectQuery), row);
        }
    }

    @Test
    public void readRepairTimeoutTest() throws Throwable
    {
        final long reducedReadTimeout = 3000L;
        try (Cluster cluster = init(builder().withNodes(3).start()))
        {
            cluster.forEach(i -> i.runOnInstance(() -> DatabaseDescriptor.setReadRpcTimeout(reducedReadTimeout)));
            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH read_repair='blocking'");
            cluster.get(1).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)");
            cluster.get(2).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)");
            assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"));
            cluster.verbs(READ_REPAIR_RSP).to(1).drop();
            final long start = System.currentTimeMillis();
            try
            {
                cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.ALL);
                fail("Read timeout expected but it did not occur");
            }
            catch (Exception ex)
            {
                // the containing exception class was loaded by another class loader. Comparing the message as a workaround to assert the exception
                Assert.assertTrue(ex.getClass().toString().contains("ReadTimeoutException"));
                long actualTimeTaken = System.currentTimeMillis() - start;
                long magicDelayAmount = 100L; // it might not be the best way to check if the time taken is around the timeout value.
                // Due to the delays, the actual time taken from client perspective is slighly more than the timeout value
                Assert.assertTrue(actualTimeTaken > reducedReadTimeout);
                // But it should not exceed too much
                Assert.assertTrue(actualTimeTaken < reducedReadTimeout + magicDelayAmount);
                assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"),
                           row(1, 1, 1)); // the partition happened when the repaired node sending back ack. The mutation should be in fact applied.
            }
        }
    }

    @Test
    public void failingReadRepairTest() throws Throwable
    {
        try (Cluster cluster = init(builder().withNodes(3).start()))
        {
            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH read_repair='blocking'");

            for (int i = 1 ; i <= 2 ; ++i)
                cluster.get(i).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1)");

            assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"));

            cluster.filters().verbs(READ_REPAIR_REQ.id).to(3).drop();
            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1",
                                                      ConsistencyLevel.QUORUM),
                       row(1, 1, 1));

            // Data was not repaired
            assertRows(cluster.get(3).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1"));
        }
    }

    @Test
    public void movingTokenReadRepairTest() throws Throwable
    {
        try (Cluster cluster = init(Cluster.create(4), 3))
        {
            List<Token> tokens = cluster.tokens();

            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck)) WITH read_repair='blocking'");

            int i = 0;
            while (true)
            {
                Token t = Murmur3Partitioner.instance.getToken(Int32Type.instance.decompose(i));
                // the list of tokens uses zero-based numbering, whereas the cluster nodes use one-based numbering
                if (t.compareTo(tokens.get(2 - 1)) < 0 && t.compareTo(tokens.get(1 - 1)) > 0)
                    break;
                ++i;
            }

            // write only to #4
            cluster.get(4).executeInternal("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, 1, 1)", i);
            // mark #2 as leaving in #4
            cluster.get(4).acceptsOnInstance((InetSocketAddress endpoint) -> {
                StorageService.instance.getTokenMetadata().addLeavingEndpoint(InetAddressAndPort.getByAddressOverrideDefaults(endpoint.getAddress(), endpoint.getPort()));
                PendingRangeCalculatorService.instance.update();
                PendingRangeCalculatorService.instance.blockUntilFinished();
            }).accept(cluster.get(2).broadcastAddress());

            // prevent #4 from reading or writing to #3, so our QUORUM must contain #2 and #4
            // since #1 is taking over the range, this means any read-repair must make it to #1 as well
            // (as a speculative repair in this case, as we prefer to send repair mutations to the initial
            // set of read replicas, which are 2 and 3 here).
            cluster.filters().verbs(READ_REQ.id).from(4).to(3).drop();
            cluster.filters().verbs(READ_REPAIR_REQ.id).from(4).to(3).drop();
            assertRows(cluster.coordinator(4).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?",
                                                      ConsistencyLevel.QUORUM, i),
                       row(i, 1, 1));

            // verify that #1 receives the write
            assertRows(cluster.get(1).executeInternal("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?", i),
                       row(i, 1, 1));
        }
    }

    /**
     * Test that there is no read repair with RF=1, and that altering it to RF>1 doesn't trigger any repair by
     * itself but following queries will use read repair accordingly with the new RF.
     */
    @Test
    public void alterRFAndRunReadRepair() throws Throwable
    {
        try (Cluster cluster = builder().withNodes(2).start())
        {
            cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = " +
                                              "{'class': 'SimpleStrategy', 'replication_factor': 1}"));
            cluster.schemaChange(withKeyspace("CREATE TABLE %s.t (k int PRIMARY KEY, a int, b int)" +
                                              " WITH read_repair='blocking'"));

            // insert a row that will only get to one node due to the RF=1
            Object[] row = row(1, 1, 1);
            cluster.get(1).executeInternal(withKeyspace("INSERT INTO %s.t (k, a, b) VALUES (?, ?, ?)"), row);

            // flush to ensure reads come from sstables
            cluster.get(1).flush(KEYSPACE);

            // at RF=1 it shouldn't matter which node we query, as the data should always come from the only replica
            String query = withKeyspace("SELECT * FROM %s.t WHERE k = 1");
            for (int i = 1; i <= cluster.size(); i++)
                assertRows(cluster.coordinator(i).execute(query, ALL), row);

            // at RF=1 the prevoius queries shouldn't have triggered read repair
            assertRows(cluster.get(1).executeInternal(query), row);
            assertRows(cluster.get(2).executeInternal(query));

            // alter RF
            System.setProperty(Config.PROPERTY_PREFIX + "allow_alter_rf_during_range_movement", "true");
            cluster.schemaChange(withKeyspace("ALTER KEYSPACE %s WITH replication = " +
                                              "{'class': 'SimpleStrategy', 'replication_factor': 2}"));

            // altering the RF shouldn't have triggered any read repair
            assertRows(cluster.get(1).executeInternal(query), row);
            assertRows(cluster.get(2).executeInternal(query));

            // query again at CL=ALL, this time the data should be repaired
            assertRows(cluster.coordinator(2).execute(query, ALL), row);
            assertRows(cluster.get(1).executeInternal(query), row);
            assertRows(cluster.get(2).executeInternal(query), row);
        }
    }

    @Test
    public void testRangeSliceQueryWithTombstonesInMemory() throws Throwable
    {
        testRangeSliceQueryWithTombstones(false);
    }

    @Test
    public void testRangeSliceQueryWithTombstonesOnDisk() throws Throwable
    {
        testRangeSliceQueryWithTombstones(true);
    }

    /**
     * Verify that range queries with CL>ONE don't do unnecessary read-repairs when there are tombstones.
     * <p>
     * See CASSANDRA-8989 and CASSANDRA-9502.
     * <p>
     * Migrated from Python dtest read_repair_test.py:TestReadRepair.test_range_slice_query_with_tombstones()
     */
    private void testRangeSliceQueryWithTombstones(boolean flush) throws Throwable
    {
        try (Cluster cluster = init(Cluster.create(2)))
        {
            cluster.schemaChange(withKeyspace("CREATE TABLE %s.t (k int, c int, v int, PRIMARY KEY(k, c))"));

            ICoordinator coordinator = cluster.coordinator(1);

            // insert some rows in all nodes
            String insertQuery = withKeyspace("INSERT INTO %s.t (k, c, v) VALUES (?, ?, ?)");
            for (int k = 0; k < 10; k++)
            {
                for (int c = 0; c < 10; c++)
                    coordinator.execute(insertQuery, ALL, k, c, k * c);
            }

            // delete a subset of the inserted partitions, plus some others that don't exist
            String deletePartitionQuery = withKeyspace("DELETE FROM %s.t WHERE k = ?");
            for (int k = 5; k < 15; k++)
            {
                coordinator.execute(deletePartitionQuery, ALL, k);
            }

            // delete some of the rows of some of the partitions, including deleted and not deleted partitions
            String deleteRowQuery = withKeyspace("DELETE FROM %s.t WHERE k = ? AND c = ?");
            for (int k = 2; k < 7; k++)
            {
                for (int c = 0; c < 5; c++)
                    coordinator.execute(deleteRowQuery, ALL, k, c);
            }

            // delete some of the rows of some not-existent partitions, including deleted and never-written partitions
            for (int k = 12; k < 17; k++)
            {
                for (int c = 0; c < 5; c++)
                    coordinator.execute(deleteRowQuery, ALL, k, c);
            }

            // flush all the nodes if specified
            if (flush)
            {
                for (int n = 1; n <= cluster.size(); n++)
                    cluster.get(n).flush(KEYSPACE);
            }

            // run a bunch of queries verifying that they don't trigger read repair
            coordinator.execute(withKeyspace("SELECT * FROM %s.t LIMIT 100"), QUORUM);
            for (int k = 0; k < 15; k++)
            {
                coordinator.execute(withKeyspace("SELECT * FROM %s.t WHERE k=?"), QUORUM, k);
                for (int c = 0; c < 10; c++)
                {
                    coordinator.execute(withKeyspace("SELECT * FROM %s.t WHERE k=? AND c=?"), QUORUM, k, c);
                    coordinator.execute(withKeyspace("SELECT * FROM %s.t WHERE k=? AND c>?"), QUORUM, k, c);
                    coordinator.execute(withKeyspace("SELECT * FROM %s.t WHERE k=? AND c<?"), QUORUM, k, c);
                }
            }
            long requests = ReadRepairTester.readRepairRequestsCount(cluster.get(1), "t");
            assertEquals("No read repair requests were expected, found " + requests, 0, requests);
        }
    }

    @Test
    public void readRepairRTRangeMovementTest() throws Throwable
    {
        ExecutorService es = Executors.newFixedThreadPool(1);
        String key = "test1";
        try (Cluster cluster = init(Cluster.build()
                                           .withConfig(config -> config.with(Feature.GOSSIP, Feature.NETWORK)
                                                                       .set("read_request_timeout_in_ms", Integer.MAX_VALUE))
                                           .withTokenSupplier(TokenSupplier.evenlyDistributedTokens(4))
                                           .withNodeIdTopology(NetworkTopology.singleDcNetworkTopology(4, "dc0", "rack0"))
                                           .withNodes(3)
                                           .start()))
        {
            cluster.schemaChange("CREATE TABLE distributed_test_keyspace.tbl (\n" +
                                 "    key text,\n" +
                                 "    column1 int,\n" +
                                 "    PRIMARY KEY (key, column1)\n" +
                                 ") WITH CLUSTERING ORDER BY (column1 ASC)");

            cluster.forEach(i -> i.runOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").disableAutoCompaction()));

            for (int i = 1; i <= 2; i++)
            {
                cluster.get(i).executeInternal("DELETE FROM distributed_test_keyspace.tbl USING TIMESTAMP 50 WHERE key=?;", key);
                cluster.get(i).executeInternal("DELETE FROM distributed_test_keyspace.tbl USING TIMESTAMP 80 WHERE key=? and column1 >= ? and column1 < ?;", key, 10, 100);
                cluster.get(i).executeInternal("DELETE FROM distributed_test_keyspace.tbl USING TIMESTAMP 70 WHERE key=? and column1 = ?;", key, 30);
                cluster.get(i).flush(KEYSPACE);
            }
            cluster.get(3).executeInternal("DELETE FROM distributed_test_keyspace.tbl USING TIMESTAMP 100 WHERE key=?;", key);
            cluster.get(3).flush(KEYSPACE);

            // pause the read until we have bootstrapped a new node below
            SimpleCondition continueRead = new SimpleCondition();
            SimpleCondition readStarted = new SimpleCondition();
            cluster.filters().outbound().from(3).to(1,2).verbs(Verb.READ_REQ.id).messagesMatching((i, i1, iMessage) -> {
                try
                {
                    readStarted.signalAll();
                    continueRead.await();
                }
                catch (InterruptedException e)
                {
                    throw new RuntimeException(e);
                }
                return false;
            }).drop();
            Future<Object[][]> read = es.submit(() -> cluster.coordinator(3)
                                                          .execute("SELECT * FROM distributed_test_keyspace.tbl WHERE key=? and column1 >= ? and column1 <= ?",
                                                                   ConsistencyLevel.ALL, key, 20, 40));
            readStarted.await();
            IInstanceConfig config = cluster.newInstanceConfig();
            config.set("auto_bootstrap", true);
            cluster.bootstrap(config).startup();
            continueRead.signalAll();
            read.get();
        }
        finally
        {
            es.shutdown();
        }
    }

    /**
     * Range queries before CASSANDRA-11427 will trigger read repairs for puregable tombstones on hosts that already
     * compacted given tombstones. This will result in constant transfer and compaction actions sourced by few nodes
     * seeding purgeable tombstones and triggered e.g. by periodical jobs scanning data range wise.
     * <p>
     * See CASSANDRA-11427.
     * <p>
     * Migrated from Python dtest read_repair_test.py:TestReadRepair.test_gcable_tombstone_resurrection_on_range_slice_query()
     */
    @Test
    public void testGCableTombstoneResurrectionOnRangeSliceQuery() throws Throwable
    {
        try (Cluster cluster = init(Cluster.create(2)))
        {
            cluster.schemaChange(withKeyspace("CREATE TABLE %s.t (k int, c int, PRIMARY KEY(k, c)) " +
                                              "WITH gc_grace_seconds=0 AND compaction = " +
                                              "{'class': 'SizeTieredCompactionStrategy', 'enabled': 'false'}"));

            ICoordinator coordinator = cluster.coordinator(1);

            // insert some data
            coordinator.execute(withKeyspace("INSERT INTO %s.t(k, c) VALUES (0, 0)"), ALL);
            coordinator.execute(withKeyspace("INSERT INTO %s.t(k, c) VALUES (1, 1)"), ALL);

            // create partition tombstones in all nodes for both existent and not existent partitions
            coordinator.execute(withKeyspace("DELETE FROM %s.t WHERE k=0"), ALL); // exists
            coordinator.execute(withKeyspace("DELETE FROM %s.t WHERE k=2"), ALL); // doesn't exist

            // create row tombstones in all nodes for both existent and not existent rows
            coordinator.execute(withKeyspace("DELETE FROM %s.t WHERE k=1 AND c=1"), ALL); // exists
            coordinator.execute(withKeyspace("DELETE FROM %s.t WHERE k=3 AND c=1"), ALL); // doesn't exist

            // flush single sstable with tombstones
            cluster.get(1).flush(KEYSPACE);
            cluster.get(2).flush(KEYSPACE);

            // purge tombstones from node2 with compaction (gc_grace_seconds=0)
            cluster.get(2).forceCompact(KEYSPACE, "t");

            // run an unrestricted range query verifying that it doesn't trigger read repair
            coordinator.execute(withKeyspace("SELECT * FROM %s.t"), ALL);
            long requests = ReadRepairTester.readRepairRequestsCount(cluster.get(1), "t");
            assertEquals("No read repair requests were expected, found " + requests, 0, requests);
        }
    }

    @Test
    public void partitionDeletionRTTimestampTieTest() throws Throwable
    {
        try (Cluster cluster = init(builder()
                                    .withNodes(3)
                                    .withInstanceInitializer(RRHelper::install)
                                    .start()))
        {
            cluster.schemaChange(withKeyspace("CREATE TABLE distributed_test_keyspace.tbl0 (pk bigint,ck bigint,value bigint, PRIMARY KEY (pk, ck)) WITH  CLUSTERING ORDER BY (ck ASC) AND read_repair='blocking';"));
            long pk = 0L;
            cluster.coordinator(1).execute("INSERT INTO distributed_test_keyspace.tbl0 (pk, ck, value) VALUES (?,?,?) USING TIMESTAMP 1", ConsistencyLevel.ALL, pk, 1L, 1L);
            cluster.coordinator(1).execute("DELETE FROM distributed_test_keyspace.tbl0 USING TIMESTAMP 2 WHERE pk=? AND ck>?;", ConsistencyLevel.ALL, pk, 2L);
            cluster.get(3).executeInternal("DELETE FROM distributed_test_keyspace.tbl0 USING TIMESTAMP 2 WHERE pk=?;", pk);
            assertRows(cluster.coordinator(1).execute("SELECT * FROM distributed_test_keyspace.tbl0 WHERE pk=? AND ck>=? AND ck<?;",
                                                      ConsistencyLevel.ALL, pk, 1L, 3L));
        }
    }

    public static class RRHelper
    {
        static void install(ClassLoader cl, int nodeNumber)
        {
            // Only on coordinating node
            if (nodeNumber == 1)
            {
                new ByteBuddy().rebase(BlockingReadRepair.class)
                               .method(named("repairPartition"))
                               .intercept(MethodDelegation.to(RRHelper.class))
                               .make()
                               .load(cl, ClassLoadingStrategy.Default.INJECTION);
            }
        }

        // This verifies new behaviour in 4.0 that was introduced in CASSANDRA-15369, but did not work
        // on timestamp tie of RT and partition deletion: we should not generate RT bounds in such case,
        // since monotonicity is already ensured by the partition deletion, and RT is unnecessary there.
        // For details, see CASSANDRA-16453.
        public static Object repairPartition(DecoratedKey partitionKey, Map<Replica, Mutation> mutations, ReplicaPlan.ForTokenWrite writePlan, @SuperCall Callable<Void> r) throws Exception
        {
            Assert.assertEquals(2, mutations.size());
            for (Mutation value : mutations.values())
            {
                for (PartitionUpdate update : value.getPartitionUpdates())
                {
                    Assert.assertFalse(update.hasRows());
                    Assert.assertFalse(update.partitionLevelDeletion().isLive());
                }
            }
            return r.call();
        }
    }
}