package io.activej.crdt.storage.local;

import io.activej.crdt.CrdtData;
import io.activej.crdt.CrdtTombstone;
import io.activej.crdt.function.CrdtFunction;
import io.activej.crdt.util.CrdtDataBinarySerializer;
import io.activej.datastream.supplier.StreamSupplier;
import io.activej.datastream.supplier.StreamSuppliers;
import io.activej.fs.FileMetadata;
import io.activej.fs.FileSystem;
import io.activej.reactor.Reactor;
import io.activej.test.rules.ByteBufRule;
import io.activej.test.rules.EventloopRule;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.stream.Stream;

import static io.activej.common.Utils.entriesToLinkedHashMap;
import static io.activej.common.Utils.first;
import static io.activej.crdt.function.CrdtFunction.ignoringTimestamp;
import static io.activej.promise.TestUtils.await;
import static io.activej.reactor.Reactor.getCurrentReactor;
import static io.activej.serializer.BinarySerializers.*;
import static org.junit.Assert.*;

public final class FileSystemCrdtStorageTest {
	private static final CrdtFunction<Set<Integer>> CRDT_FUNCTION = ignoringTimestamp(FileSystemCrdtStorageTest::union);
	private static final CrdtDataBinarySerializer<String, Set<Integer>> SERIALIZER = new CrdtDataBinarySerializer<>(UTF8_SERIALIZER, ofSet(INT_SERIALIZER));

	@ClassRule
	public static final EventloopRule eventloopRule = new EventloopRule();

	@ClassRule
	public static final ByteBufRule byteBufRule = new ByteBufRule();

	@Rule
	public final TemporaryFolder temporaryFolder = new TemporaryFolder();

	private FileSystem fileSystem;
	private FileSystemCrdtStorage<String, Set<Integer>> client;

	@Before
	public void setup() throws IOException {
		Reactor reactor = getCurrentReactor();
		fileSystem = FileSystem.create(reactor, Executors.newCachedThreadPool(), temporaryFolder.newFolder().toPath());
		client = FileSystemCrdtStorage.create(reactor, fileSystem, SERIALIZER, CRDT_FUNCTION);
		await(fileSystem.start());
		await(client.start());
	}

	@Test
	public void testEmptyUpload() {
		await(StreamSuppliers.<CrdtData<String, Set<Integer>>>empty().streamTo(client.upload()));
		assertTrue(await(fileSystem.list("**")).isEmpty());
	}

	@Test
	public void testEmptyRemove() {
		await(StreamSuppliers.<CrdtTombstone<String>>empty().streamTo(client.remove()));
		assertTrue(await(fileSystem.list("**")).isEmpty());
	}

	@Test
	public void testEmptyConsolidation() {
		await(client.consolidate());
		assertTrue(await(fileSystem.list("**")).isEmpty());
	}

	@Test
	public void testConsolidation() {
		long timestamp = getCurrentReactor().currentTimeMillis();

		List<CrdtData<String, Set<Integer>>> expected = List.of(
			new CrdtData<>("12_test_1", timestamp, Set.of(123, 124, 125, 2, 542)),
			new CrdtData<>("12_test_2", timestamp, Set.of(12, 13)),
			new CrdtData<>("1_test_1", timestamp, Set.of(1, 2, 3)),
			new CrdtData<>("1_test_2", timestamp, Set.of(2, 3, 7)),
			new CrdtData<>("1_test_3", timestamp, Set.of(78, 2, 3)),
			new CrdtData<>("2_test_1", timestamp, Set.of(1, 2, 3)),
			new CrdtData<>("2_test_2", timestamp, Set.of(2, 3, 4)),
			new CrdtData<>("2_test_3", timestamp, Set.of(0, 1, 2))
		);

		await(StreamSuppliers.ofStream(Stream.of(
				new CrdtData<>("1_test_1", timestamp, Set.of(1, 2, 3)),
				new CrdtData<>("1_test_2", timestamp, Set.of(2, 3, 7)),
				new CrdtData<>("1_test_3", timestamp, Set.of(78, 2, 3)),
				new CrdtData<>("12_test_1", timestamp, Set.of(123, 124, 125)),
				new CrdtData<>("12_test_2", timestamp, Set.of(12))).sorted())
			.streamTo(client.upload()));
		await(StreamSuppliers.ofStream(Stream.of(
				new CrdtData<>("2_test_1", timestamp, Set.of(1, 2, 3)),
				new CrdtData<>("2_test_2", timestamp, Set.of(2, 3, 4)),
				new CrdtData<>("2_test_3", timestamp, Set.of(0, 1, 2)),
				new CrdtData<>("12_test_1", timestamp, Set.of(123, 542, 125, 2)),
				new CrdtData<>("12_test_2", timestamp, Set.of(12, 13))).sorted())
			.streamTo(client.upload()));

		Map<String, FileMetadata> filesBefore = await(fileSystem.list("**"));
		System.out.println(filesBefore);
		assertEquals(2, filesBefore.size());

		List<CrdtData<String, Set<Integer>>> downloadedBefore = await(client.download().then(StreamSupplier::toList));
		assertEquals(expected, downloadedBefore);

		await(client.consolidate());

		Map<String, FileMetadata> filesAfter = await(fileSystem.list("**"));
		System.out.println(filesAfter);
		assertEquals(1, filesAfter.size());
		assertFalse(filesBefore.containsKey(first(filesAfter.keySet())));

		List<CrdtData<String, Set<Integer>>> downloadedAfter = await(client.download().then(StreamSupplier::toList));
		assertEquals(expected, downloadedAfter);
	}

	@Test
	public void testTombstoneConsolidation() {
		List<CrdtData<String, Set<Integer>>> expected = List.of(
			new CrdtData<>("a", 100, Set.of(1, 2, 3)),
			new CrdtData<>("b", 300, Set.of(5, 6, 7)),
			new CrdtData<>("c", 400, Set.of(78, 2, 3))
		);

		await(StreamSuppliers.ofValues(
			new CrdtData<>("a", 100, Set.of(1, 2, 3)),
			new CrdtData<>("b", 200, Set.of(2, 3, 7))
		).streamTo(client.upload()));
		await(StreamSuppliers.ofValues(
			new CrdtData<>("b", 300, Set.of(5, 6, 7)),
			new CrdtData<>("c", 400, Set.of(78, 2, 3))
		).streamTo(client.upload()));
		await(StreamSuppliers.ofValues(
			new CrdtData<>("c", 100, Set.of(123, 124, 125, 3)),
			new CrdtData<>("d", 500, Set.of(12))
		).streamTo(client.upload()));
		await(StreamSuppliers.ofValues(
			new CrdtData<>("d", 600, Set.of(56, 76)),
			new CrdtData<>("e", 300, Set.of(124))
		).streamTo(client.upload()));

		await(StreamSuppliers.ofStream(Stream.of(new CrdtTombstone<>("a", 50))).streamTo(client.remove()));
		await(StreamSuppliers.ofStream(Stream.of(new CrdtTombstone<>("b", 250))).streamTo(client.remove()));
		await(StreamSuppliers.ofStream(Stream.of(new CrdtTombstone<>("c", 300))).streamTo(client.remove()));
		await(StreamSuppliers.ofStream(Stream.of(new CrdtTombstone<>("d", 600))).streamTo(client.remove()));
		await(StreamSuppliers.ofStream(Stream.of(new CrdtTombstone<>("e", 400))).streamTo(client.remove()));

		Map<String, FileMetadata> filesBefore = await(fileSystem.list("**"));
		System.out.println(filesBefore);
		assertEquals(9, filesBefore.size());

		List<CrdtData<String, Set<Integer>>> downloadedBefore = await(client.download().then(StreamSupplier::toList));
		assertEquals(expected, downloadedBefore);

		await(client.consolidate());

		Map<String, FileMetadata> filesAfter = await(fileSystem.list("**"));
		System.out.println(filesAfter);
		assertEquals(1, filesAfter.size());
		String consolidated = first(filesAfter.keySet());
		assertFalse(filesBefore.containsKey(consolidated));

		List<CrdtData<String, Set<Integer>>> downloadedAfter = await(client.download().then(StreamSupplier::toList));
		assertEquals(expected, downloadedAfter);
	}

	@Test
	public void pickFilesForConsolidation() {
		testPickFilesForConsolidation(
			Set.of("a", "c", "e"),
			Map.of(
				"a", 12,
				"b", 120,
				"c", 53,
				"d", 348,
				"e", 97)
		);
		testPickFilesForConsolidation(
			Set.of("a", "c"),
			Map.of(
				"a", 120,
				"b", 12,
				"c", 530
			));
		testPickFilesForConsolidation(
			Set.of("b", "d"),
			Map.of(
				"a", 120,
				"b", 12,
				"c", 530,
				"d", 43
			));
		testPickFilesForConsolidation(
			Set.of(),
			Map.of(
				"a", 120,
				"b", 12,
				"c", 5,
				"d", 4345
			));
	}

	private static void testPickFilesForConsolidation(Set<String> expected, Map<String, Integer> fileToSizeMap) {
		Map<String, FileMetadata> files = fileToSizeMap.entrySet().stream()
			.collect(entriesToLinkedHashMap(size -> FileMetadata.of(size, 0)));
		Set<String> filesForConsolidation = FileSystemCrdtStorage.pickFilesForConsolidation(files);

		assertEquals(expected, filesForConsolidation);
	}

	private static Set<Integer> union(Set<Integer> first, Set<Integer> second) {
		Set<Integer> res = new HashSet<>(Math.max((int) ((first.size() + second.size()) / .75f) + 1, 16));
		res.addAll(first);
		res.addAll(second);
		return res;
	}
}
