package com.atlassian.bitbucket.mesh.repository;

import com.atlassian.bitbucket.mesh.git.hook.GitHookService;
import com.atlassian.bitbucket.mesh.io.MoreFiles;
import com.atlassian.bitbucket.mesh.junit.LogListener;
import com.atlassian.bitbucket.mesh.junit.TemporaryDirectory;
import com.atlassian.bitbucket.mesh.partition.Partition;
import com.google.common.collect.ImmutableList;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static com.atlassian.bitbucket.mesh.repository.ManagedDataStore.LAYOUT_VERSION;
import static com.atlassian.bitbucket.mesh.repository.ManagedDataStore.TYPE;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.*;
import static org.junit.Assert.*;

@RunWith(MockitoJUnitRunner.class)
public class ManagedDataStoreTest {

    private static final String HIERARCHY_ID = "abcde1234554321edcba";
    private static final String PARTITION_ID = "0001";
    @Rule
    public final LogListener logListener = new LogListener(ManagedDataStore.class);
    @Rule
    public final TemporaryDirectory tempDir = new TemporaryDirectory();

    @Mock
    private GitHookService hookService;
    private ManagedDataStore store;

    @After
    public void tearDown() {
        if (store != null) {
            store.stop();
        }
    }

    //These attach tests deal with how ManagedDataStore _locates_ the shard to attach a repository to, but
    //they do not test how attachment works. That's done by ManagedShardTest instead

    @Test
    public void testAttachStorageForNewPartition() throws IOException {
        Path storeDir = tempDir.newFolder("store");
        DefaultManagedRepository repository = new DefaultManagedRepository(PARTITION_ID, HIERARCHY_ID, "1");

        store = createDataStore(ImmutableList.of(storeDir));
        store.start();

        StorageStatus status = store.attachStorage(repository);
        assertSame(StorageStatus.NONEXISTENT, status);

        //Verify the repository shows a path on the expected shard
        assertTrue("The repository is not on the expected shard",
                MoreFiles.isWithin(repository.getPath(), tempDir.getRoot()));
    }

    @Test
    public void testAttachStorageForPreviouslyMappedPartition() throws IOException {
        Path first = prepareShard(tempDir.newFolder("first"), PARTITION_ID);
        Path second = prepareShard(tempDir.newFolder("second"), "ffffffff");
        DefaultManagedRepository repository = new DefaultManagedRepository(PARTITION_ID, HIERARCHY_ID, "1");

        store = createDataStore(ImmutableList.of(first, second));
        store.start();

        StorageStatus status = store.attachStorage(repository);
        assertSame(StorageStatus.NONEXISTENT, status);

        //Verify the repository shows a path on the expected shard
        assertTrue("The repository is not on the expected shard",
                MoreFiles.isWithin(repository.getPath(), MoreFiles.resolve(first, "data", PARTITION_ID)));
    }

    @Test
    public void testAttachStorageWithoutShards() {
        DefaultManagedRepository repository = new DefaultManagedRepository(PARTITION_ID, HIERARCHY_ID, "1");

        store = createDataStore(ImmutableList.of());
        store.start();

        StorageStatus status = store.attachStorage(repository);
        assertSame(StorageStatus.NO_STORE, status);
    }

    @Test
    public void testClearMarkDeletePartition() throws IOException {
        Path storeDir = tempDir.newFolder("store");
        Path storeDir2 = tempDir.newFolder("store2");
        Partition partition1 = new Partition("0001");
        Partition partition2 = new Partition("ffffffff");

        store = createDataStore(ImmutableList.of(storeDir, storeDir2));

        store.start();
        DefaultManagedRepository repo1 = new DefaultManagedRepository(partition1.getHex(), "abcde1234554321edcba", "1");
        DefaultManagedRepository repo2 = new DefaultManagedRepository(partition2.getHex(), "123456abcde123455432", "2");
        DefaultManagedRepository repo3 = new DefaultManagedRepository("0005", "123456abcde123455432", "3");
        store.attachStorage(repo1);
        store.attachStorage(repo2);
        store.attachStorage(repo3);

        store.markPartitionDeleted(partition1);
        store.markPartitionDeleted(partition2);
        store.clearPartitionDeleted(partition1);

        Map<Partition, Instant> deletedPartitions = new HashMap<>();
        store.streamDeletedPartitions(deletedPartitions::put);

        assertThat(deletedPartitions, allOf(aMapWithSize(1), hasKey(partition2)));
    }

    @Test
    public void testCreatingSharedIgnoresLostAndFoundDirectory() throws IOException {
        Path storeDir = tempDir.newFolder("store");
        Files.createDirectories(storeDir.resolve("lost+found"));

        store = createDataStore(ImmutableList.of(storeDir));

        assertEquals(1, store.getInfo().getShardInfo().size());
    }

    @Test
    public void testDeletePartitionFilesCallsShard() throws IOException {
        //Doesn't verify that partition files are deleted, that's done in ManagedShardTest
        Path storeDir = tempDir.newFolder("store");
        Path storeDir2 = tempDir.newFolder("store2");
        Partition partition1 = new Partition("0001");
        Partition partition2 = new Partition("ffffffff");
        Partition unattachedPartition = new Partition("0a0a0a0a");

        store = createDataStore(ImmutableList.of(storeDir, storeDir2));

        store.start();
        DefaultManagedRepository repo1 = new DefaultManagedRepository(partition1.getHex(), "abcde1234554321edcba", "1");
        DefaultManagedRepository repo2 = new DefaultManagedRepository(partition2.getHex(), "123456abcde123455432", "2");
        DefaultManagedRepository repo3 = new DefaultManagedRepository("0005", "123456abcde123455432", "3");
        store.attachStorage(repo1);
        store.attachStorage(repo2);
        store.attachStorage(repo3);

        store.markPartitionDeleted(partition1);

        assertTrue(store.deletePartitionFiles(partition1));
        assertTrue(store.deletePartitionFiles(partition2));
        assertFalse(store.deletePartitionFiles(unattachedPartition));
    }

    @Test
    public void testDeletePartitionFilesRemovesPartitionAssociation() throws IOException {
        //Doesn't verify that partition files are deleted, that's done in ManagedShardTest
        Path storeDir = tempDir.newFolder("store");
        Path storeDir2 = tempDir.newFolder("store2");
        Partition partition1 = new Partition("0001");
        Partition partition2 = new Partition("ffffffff");
        Partition unattachedPartition = new Partition("0a0a0a0a");

        store = createDataStore(ImmutableList.of(storeDir, storeDir2));

        store.start();
        DefaultManagedRepository repo1 = new DefaultManagedRepository(partition1.getHex(), "abcde1234554321edcba", "1");
        DefaultManagedRepository repo2 = new DefaultManagedRepository(partition2.getHex(), "123456abcde123455432", "2");
        DefaultManagedRepository repo3 = new DefaultManagedRepository("0005", "123456abcde123455432", "3");
        store.attachStorage(repo1);
        store.attachStorage(repo2);
        store.attachStorage(repo3);

        store.markPartitionDeleted(partition1);

        Iterable<String> partitions = store.getAllPartitions();
        assertThat(partitions, hasItem(equalTo(partition1.getHex())));
        assertThat(partitions, hasItem(equalTo(partition2.getHex())));
        assertThat(partitions, hasItem(equalTo("0005")));

        assertTrue(store.deletePartitionFiles(partition1));
        assertTrue(store.deletePartitionFiles(partition2));
        assertFalse(store.deletePartitionFiles(unattachedPartition));

        partitions = store.getAllPartitions();
        assertThat(partitions, not(hasItem(equalTo(partition1.getHex()))));
        assertThat(partitions, not(hasItem(equalTo(partition2.getHex()))));
        assertThat(partitions, hasItem(equalTo("0005")));
    }

    @Test
    public void testGetAllPartitionsReturnsAllPartitions() throws IOException {
        Path storeDir = tempDir.newFolder("store");
        Path storeDir2 = tempDir.newFolder("store2");

        store = createDataStore(ImmutableList.of(storeDir, storeDir2));

        store.start();
        DefaultManagedRepository repo1 = new DefaultManagedRepository("0001", "abcde1234554321edcba", "1");
        DefaultManagedRepository repo2 = new DefaultManagedRepository("ffffffff", "123456abcde123455432", "2");
        DefaultManagedRepository repo3 = new DefaultManagedRepository("0005", "123456abcde123455432", "3");
        store.attachStorage(repo1);
        store.attachStorage(repo2);
        store.attachStorage(repo3);

        assertThat(store.getAllPartitions(), containsInAnyOrder("0001", "ffffffff", "0005"));
    }

    @Test
    public void testGetInfoMultipleShards() throws IOException {
        Path storeDir = tempDir.newFolder("store");
        Path storeDir2 = tempDir.newFolder("store2");
        store = createDataStore(ImmutableList.of(storeDir, storeDir2));

        store.start();
        DefaultManagedRepository repo1 = new DefaultManagedRepository("0001", "abcde1234554321edcba", "1");
        DefaultManagedRepository repo2 = new DefaultManagedRepository("ffffffff", "123456abcde123455432", "2");
        store.attachStorage(repo1);
        store.attachStorage(repo2);

        DataStoreInfo storeInfo = store.getInfo();

        assertEquals(2, storeInfo.getShardInfo().size());
        assertEquals(2, storeInfo.getShardInfo().stream()
                .map(ShardInfo::getPartitions)
                .mapToInt(List::size)
                .sum());
    }

    @Test
    public void testGetInfoSingleShard() throws IOException {
        Path storeDir = tempDir.newFolder("store");
        store = createDataStore(ImmutableList.of(storeDir));

        store.start();
        DefaultManagedRepository repo1 = new DefaultManagedRepository("0001", "abcde1234554321edcba", "1");
        DefaultManagedRepository repo2 = new DefaultManagedRepository("ffffffff", "123456abcde123455432", "2");
        store.attachStorage(repo1);
        store.attachStorage(repo2);

        DataStoreInfo storeInfo = store.getInfo();

        assertEquals(1, storeInfo.getShardInfo().size());
        //should have no partition listing as there is only one shard
        assertEquals(0, storeInfo.getShardInfo().get(0).getPartitions().size());
    }

    @Test
    public void testGetType() {
        assertEquals(DataStoreType.MANAGED, createDataStore(ImmutableList.of()).getType());
    }

    @Test
    public void testMarkAndStreamDeletedPartitions() throws IOException {
        Path storeDir = tempDir.newFolder("store");
        Path storeDir2 = tempDir.newFolder("store2");
        Partition partition1 = new Partition("0001");
        Partition partition2 = new Partition("ffffffff");

        store = createDataStore(ImmutableList.of(storeDir, storeDir2));

        store.start();
        DefaultManagedRepository repo1 = new DefaultManagedRepository(partition1.getHex(), "abcde1234554321edcba", "1");
        DefaultManagedRepository repo2 = new DefaultManagedRepository(partition2.getHex(), "123456abcde123455432", "2");
        DefaultManagedRepository repo3 = new DefaultManagedRepository("0005", "123456abcde123455432", "3");
        store.attachStorage(repo1);
        store.attachStorage(repo2);
        store.attachStorage(repo3);

        store.markPartitionDeleted(partition1);
        store.markPartitionDeleted(partition2);

        Map<Partition, Instant> deletedPartitions = new HashMap<>();
        store.streamDeletedPartitions(deletedPartitions::put);

        assertThat(deletedPartitions, allOf(aMapWithSize(2), hasKey(partition1), hasKey(partition2)));
    }

    @Test
    public void testStartWithInvalidShard() throws IOException {
        //Write a file in the shard path where a directory is expected
        Path data = MoreFiles.write(tempDir.newFile("data"), "This shouldn't be a file");

        store = createDataStore(ImmutableList.of(data));
        store.start();

        //Trying to open /path/to/file if /path/to is itself a file throws a generic IOException, which
        //DataStoreProperties.load propagates
        logListener.assertWarn("properties could not be read");
    }

    @Test
    public void testStartWithNonEmptyShard() throws IOException {
        //Don't write store.properties, but create some directories so the store isn't empty
        tempDir.newFolder("data", PARTITION_ID);
        MoreFiles.write(tempDir.newFile("home.properties"), "What this do?");

        store = createDataStore(ImmutableList.of(tempDir.getRoot()));
        store.start();

        logListener.assertInfo("type cannot be determined");
    }

    @Test
    public void testStartWithPartitionOnMultipleStores() throws IOException {
        Path first = prepareShard(tempDir.newFolder("first"), PARTITION_ID);
        Path second = prepareShard(tempDir.newFolder("second"), PARTITION_ID);

        store = createDataStore(ImmutableList.of(first, second));
        store.start();

        logListener.assertWarn("Partition 0001 was found on multiple data stores");
    }

    @Test
    public void testStartWithUnsupportedShard() throws IOException {
        //Prepare a store.properties with an unexpected type
        new DataStoreProperties(LAYOUT_VERSION, LAYOUT_VERSION, "legacy").store(tempDir.getRoot());

        store = createDataStore(ImmutableList.of(tempDir.getRoot()));
        store.start();

        logListener.assertWarn("Data store of type 'legacy' is not supported");
    }

    @Test
    public void testStartWithoutShards() {
        store = createDataStore(ImmutableList.of());
        store.start();
    }

    private static Path prepareShard(Path rootDir, String... partitions) throws IOException {
        new DataStoreProperties(LAYOUT_VERSION, LAYOUT_VERSION, TYPE).store(rootDir);

        Path dataDir = MoreFiles.mkdir(rootDir, "data");
        for (String partition : partitions) {
            MoreFiles.mkdir(dataDir, partition);
        }

        return rootDir;
    }

    private ManagedDataStore createDataStore(List<Path> shardDirs) {
        return new ManagedDataStore(LAYOUT_VERSION, hookService, shardDirs);
    }
}