/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License
 * 2.0; you may not use this file except in compliance with the Elastic License
 * 2.0.
 */
package org.elasticsearch.xpack.ccr.action;

import org.elasticsearch.action.ActionListener;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.ShardRoutingState;
import org.elasticsearch.cluster.routing.TestShardRouting;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.Index;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.IndexService;
import org.elasticsearch.index.IndexSettings;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.IndexShardNotStartedException;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.shard.ShardNotFoundException;
import org.elasticsearch.index.translog.Translog;
import org.elasticsearch.test.ESSingleNodeTestCase;
import org.mockito.Mockito;

import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.LongStream;

import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasToString;
import static org.hamcrest.Matchers.instanceOf;

public class ShardChangesActionTests extends ESSingleNodeTestCase {

    @Override
    protected boolean resetNodeAfterTest() {
        return true;
    }

    public void testGetOperations() throws Exception {
        final Settings settings = Settings.builder()
                .put("index.number_of_shards", 1)
                .put("index.number_of_replicas", 0)
                .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
                .build();
        final IndexService indexService = createIndex("index", settings);

        final int numWrites = randomIntBetween(10, 4096);
        for (int i = 0; i < numWrites; i++) {
            client().prepareIndex("index", "doc", Integer.toString(i)).setSource("{}", XContentType.JSON).get();
        }

        // A number of times, get operations within a range that exists:
        int iters = randomIntBetween(8, 32);
        IndexShard indexShard = indexService.getShard(0);
        for (int iter = 0; iter < iters; iter++) {
            int min = randomIntBetween(0, numWrites - 1);
            int max = randomIntBetween(min, numWrites - 1);
            int size = max - min + 1;
            final Translog.Operation[] operations = ShardChangesAction.getOperations(
                    indexShard,
                    indexShard.getLastKnownGlobalCheckpoint(),
                    min,
                    size,
                    indexShard.getHistoryUUID(),
                    new ByteSizeValue(Long.MAX_VALUE, ByteSizeUnit.BYTES));
            final List<Long> seenSeqNos = Arrays.stream(operations).map(Translog.Operation::seqNo).collect(Collectors.toList());
            final List<Long> expectedSeqNos = LongStream.rangeClosed(min, max).boxed().collect(Collectors.toList());
            assertThat(seenSeqNos, equalTo(expectedSeqNos));
        }

        {
            // get operations for a range for which no operations exist
            final IllegalStateException e = expectThrows(
                    IllegalStateException.class,
                    () -> ShardChangesAction.getOperations(
                            indexShard,
                            indexShard.getLastKnownGlobalCheckpoint(),
                            numWrites,
                            numWrites + 1,
                            indexShard.getHistoryUUID(),
                            new ByteSizeValue(Long.MAX_VALUE, ByteSizeUnit.BYTES)));
            final String message = String.format(
                    Locale.ROOT,
                    "not exposing operations from [%d] greater than the global checkpoint [%d]",
                    numWrites,
                    indexShard.getLastKnownGlobalCheckpoint());
            assertThat(e, hasToString(containsString(message)));
        }

        // get operations for a range some operations do not exist:
        Translog.Operation[] operations = ShardChangesAction.getOperations(indexShard, indexShard.getLastKnownGlobalCheckpoint(),
            numWrites  - 10, numWrites + 10, indexShard.getHistoryUUID(), new ByteSizeValue(Long.MAX_VALUE, ByteSizeUnit.BYTES));
        assertThat(operations.length, equalTo(10));

        // Unexpected history UUID:
        Exception e = expectThrows(IllegalStateException.class, () -> ShardChangesAction.getOperations(indexShard,
            indexShard.getLastKnownGlobalCheckpoint(), 0, 10, "different-history-uuid",
            new ByteSizeValue(Long.MAX_VALUE, ByteSizeUnit.BYTES)));
        assertThat(e.getMessage(), equalTo("unexpected history uuid, expected [different-history-uuid], actual [" +
                indexShard.getHistoryUUID() + "]"));

        // invalid range
        {
            final long fromSeqNo = randomLongBetween(Long.MIN_VALUE, -1);
            final int batchSize = randomIntBetween(0, Integer.MAX_VALUE);
            final IllegalArgumentException invalidRangeError = expectThrows(IllegalArgumentException.class,
                () -> ShardChangesAction.getOperations(indexShard, indexShard.getLastKnownGlobalCheckpoint(),
                    fromSeqNo, batchSize, indexShard.getHistoryUUID(), new ByteSizeValue(Long.MAX_VALUE, ByteSizeUnit.BYTES)));
            assertThat(invalidRangeError.getMessage(),
                equalTo("Invalid range; from_seqno [" + fromSeqNo + "], to_seqno [" + (fromSeqNo + batchSize - 1) + "]"));
        }
    }

    public void testGetOperationsWhenShardNotStarted() throws Exception {
        IndexShard indexShard = Mockito.mock(IndexShard.class);

        ShardRouting shardRouting = TestShardRouting.newShardRouting("index", 0, "_node_id", true, ShardRoutingState.INITIALIZING);
        Mockito.when(indexShard.routingEntry()).thenReturn(shardRouting);
        expectThrows(IndexShardNotStartedException.class, () -> ShardChangesAction.getOperations(indexShard,
            indexShard.getLastKnownGlobalCheckpoint(), 0, 1, indexShard.getHistoryUUID(),
            new ByteSizeValue(Long.MAX_VALUE, ByteSizeUnit.BYTES)));
    }

    public void testGetOperationsExceedByteLimit() throws Exception {
        final Settings settings = Settings.builder()
                .put("index.number_of_shards", 1)
                .put("index.number_of_replicas", 0)
                .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
                .build();
        final IndexService indexService = createIndex("index", settings);

        final long numWrites = 32;
        for (int i = 0; i < numWrites; i++) {
            client().prepareIndex("index", "doc", Integer.toString(i)).setSource("{}", XContentType.JSON).get();
        }

        final IndexShard indexShard = indexService.getShard(0);
        final Translog.Operation[] operations = ShardChangesAction.getOperations(indexShard, indexShard.getLastKnownGlobalCheckpoint(),
            0, randomIntBetween(100, 500), indexShard.getHistoryUUID(), new ByteSizeValue(256, ByteSizeUnit.BYTES));
        assertThat(operations.length, equalTo(7));
        assertThat(operations[0].seqNo(), equalTo(0L));
        assertThat(operations[1].seqNo(), equalTo(1L));
        assertThat(operations[2].seqNo(), equalTo(2L));
        assertThat(operations[3].seqNo(), equalTo(3L));
        assertThat(operations[4].seqNo(), equalTo(4L));
        assertThat(operations[5].seqNo(), equalTo(5L));
        assertThat(operations[6].seqNo(), equalTo(6L));
    }

    public void testGetOperationsAlwaysReturnAtLeastOneOp() throws Exception {
        final Settings settings = Settings.builder()
            .put("index.number_of_shards", 1)
            .put("index.number_of_replicas", 0)
            .put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), true)
            .build();
        final IndexService indexService = createIndex("index", settings);

        client().prepareIndex("index", "doc", "0").setSource("{}", XContentType.JSON).get();

        final IndexShard indexShard = indexService.getShard(0);
        final Translog.Operation[] operations =
            ShardChangesAction.getOperations(
                    indexShard, indexShard.getLastKnownGlobalCheckpoint(), 0, 1, indexShard.getHistoryUUID(), ByteSizeValue.ZERO);
        assertThat(operations.length, equalTo(1));
        assertThat(operations[0].seqNo(), equalTo(0L));
    }

    public void testIndexNotFound() throws InterruptedException {
        final CountDownLatch latch = new CountDownLatch(1);
        final AtomicReference<Exception> reference = new AtomicReference<>();
        final ShardChangesAction.TransportAction transportAction = node().injector().getInstance(ShardChangesAction.TransportAction.class);
        transportAction.execute(
                new ShardChangesAction.Request(new ShardId(new Index("non-existent", "uuid"), 0), "uuid"),
                new ActionListener<ShardChangesAction.Response>() {
                    @Override
                    public void onResponse(final ShardChangesAction.Response response) {
                        fail();
                    }

                    @Override
                    public void onFailure(final Exception e) {
                        reference.set(e);
                        latch.countDown();
                    }
        });
        latch.await();
        assertNotNull(reference.get());
        assertThat(reference.get(), instanceOf(IndexNotFoundException.class));
    }

    public void testShardNotFound() throws InterruptedException {
        final int numberOfShards = randomIntBetween(1, 5);
        final IndexService indexService = createIndex("index", Settings.builder().put("index.number_of_shards", numberOfShards).build());
        final CountDownLatch latch = new CountDownLatch(1);
        final AtomicReference<Exception> reference = new AtomicReference<>();
        final ShardChangesAction.TransportAction transportAction = node().injector().getInstance(ShardChangesAction.TransportAction.class);
        transportAction.execute(
                new ShardChangesAction.Request(new ShardId(indexService.getMetadata().getIndex(), numberOfShards), "uuid"),
                new ActionListener<ShardChangesAction.Response>() {
                    @Override
                    public void onResponse(final ShardChangesAction.Response response) {
                        fail();
                    }

                    @Override
                    public void onFailure(final Exception e) {
                        reference.set(e);
                        latch.countDown();
                    }
                });
        latch.await();
        assertNotNull(reference.get());
        assertThat(reference.get(), instanceOf(ShardNotFoundException.class));
    }

}
