package io.activej.crdt.wal;

import io.activej.crdt.CrdtData;
import io.activej.crdt.function.CrdtFunction;
import io.activej.crdt.primitives.GSet;
import io.activej.crdt.storage.ICrdtStorage;
import io.activej.crdt.storage.local.FileSystemCrdtStorage;
import io.activej.crdt.util.CrdtDataBinarySerializer;
import io.activej.csp.file.ChannelFileWriter;
import io.activej.csp.process.frame.ChannelFrameEncoder;
import io.activej.csp.supplier.ChannelSuppliers;
import io.activej.datastream.csp.ChannelSerializer;
import io.activej.datastream.supplier.StreamSuppliers;
import io.activej.fs.FileSystem;
import io.activej.promise.Promises;
import io.activej.reactor.Reactor;
import io.activej.test.rules.ByteBufRule;
import io.activej.test.rules.EventloopRule;
import io.activej.test.time.TestCurrentTimeProvider;
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.nio.channels.FileChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.stream.Stream;

import static io.activej.common.Utils.first;
import static io.activej.crdt.wal.FileWriteAheadLog.EXT_FINAL;
import static io.activej.crdt.wal.FileWriteAheadLog.FRAME_FORMAT;
import static io.activej.promise.TestUtils.await;
import static io.activej.reactor.Reactor.getCurrentReactor;
import static io.activej.serializer.BinarySerializers.INT_SERIALIZER;
import static io.activej.serializer.BinarySerializers.LONG_SERIALIZER;
import static java.nio.file.StandardOpenOption.WRITE;
import static java.util.stream.Collectors.toSet;
import static org.junit.Assert.*;

public class FileWriteAheadLogTest {

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

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

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

	private final CrdtDataBinarySerializer<Long, GSet<Integer>> serializer = new CrdtDataBinarySerializer<>(LONG_SERIALIZER, new GSet.Serializer<>(INT_SERIALIZER));
	private final CrdtFunction<GSet<Integer>> function = new CrdtFunction<>() {
		@Override
		public GSet<Integer> merge(GSet<Integer> first, long firstTimestamp, GSet<Integer> second, long secondTimestamp) {
			return first.merge(second);
		}

		@Override
		public GSet<Integer> extract(GSet<Integer> state, long timestamp) {
			return state;
		}
	};

	private WalUploader<Long, GSet<Integer>> uploader;
	private FileWriteAheadLog<Long, GSet<Integer>> wal;
	private ICrdtStorage<Long, GSet<Integer>> storage;
	private Executor executor;
	private Path path;

	@Before
	public void setUp() throws Exception {
		executor = Executors.newSingleThreadExecutor();
		Reactor reactor = getCurrentReactor();
		path = temporaryFolder.newFolder().toPath();
		FileSystem storageFS = FileSystem.create(reactor, executor, temporaryFolder.newFolder().toPath());
		await(storageFS.start());
		storage = FileSystemCrdtStorage.create(reactor, storageFS, serializer, function);
		uploader = WalUploader.create(reactor, executor, path, function, serializer, storage);
		wal = FileWriteAheadLog.builder(reactor, executor, path, serializer, uploader)
			.withCurrentTimeProvider(reactor)
			.build();
	}

	@Test
	public void singleFlushSequential() {
		wal = FileWriteAheadLog.builder(getCurrentReactor(), executor, path, serializer, uploader)
			.withCurrentTimeProvider(TestCurrentTimeProvider.ofTimeSequence(100, 10))
			.build();
		await(wal.start());
		List<CrdtData<Long, GSet<Integer>>> expected = List.of(
			new CrdtData<>(1L, 140, GSet.of(1, 2, 3, 6, 9, 10, 11)),
			new CrdtData<>(2L, 130, GSet.of(-12, 0, 2, 3, 100, 200))
		);
		await(wal.put(1L, GSet.of(1, 2, 3)));
		await(wal.put(2L, GSet.of(-12, 0, 200)));
		await(wal.put(1L, GSet.of(1, 6)));
		await(wal.put(2L, GSet.of(2, 3, 100)));
		await(wal.put(1L, GSet.of(9, 10, 11)));

		await(wal.flush());

		List<CrdtData<Long, GSet<Integer>>> actual = await(await(storage.download()).toList());
		assertEquals(expected, actual);
	}

	@Test
	public void singleFlushConsecutive() {
		await(wal.start());
		long now = getCurrentReactor().currentTimeMillis();
		List<CrdtData<Long, GSet<Integer>>> expected = List.of(
			new CrdtData<>(1L, now, GSet.of(1, 2, 3, 6, 9, 10, 11)),
			new CrdtData<>(2L, now, GSet.of(-12, 0, 2, 3, 100, 200))
		);
		await(
			wal.put(1L, GSet.of(1, 2, 3)),
			wal.put(2L, GSet.of(-12, 0, 200)),
			wal.put(1L, GSet.of(1, 6)),
			wal.put(2L, GSet.of(2, 3, 100)),
			wal.put(1L, GSet.of(9, 10, 11)),
			wal.flush()
		);

		List<CrdtData<Long, GSet<Integer>>> actual = await(await(storage.download()).toList());
		assertEquals(expected, actual);
	}

	@Test
	public void multipleFlushesSequential() {
		wal = FileWriteAheadLog.builder(getCurrentReactor(), executor, path, serializer, uploader)
			.withCurrentTimeProvider(TestCurrentTimeProvider.ofTimeSequence(100, 10))
			.build();
		await(wal.start());
		List<CrdtData<Long, GSet<Integer>>> expectedAfterFlush1 = List.of(
			new CrdtData<>(1L, 120, GSet.of(1, 2, 3, 6)),
			new CrdtData<>(2L, 110, GSet.of(-12, 0, 200))
		);
		List<CrdtData<Long, GSet<Integer>>> expectedAfterFlush2 = List.of(
			new CrdtData<>(1L, 140, GSet.of(1, 2, 3, 6, 9, 10, 11)),
			new CrdtData<>(2L, 130, GSet.of(-12, 0, 2, 3, 100, 200))
		);
		await(wal.put(1L, GSet.of(1, 2, 3)));
		await(wal.put(2L, GSet.of(-12, 0, 200)));
		await(wal.put(1L, GSet.of(1, 6)));
		await(wal.flush());

		List<CrdtData<Long, GSet<Integer>>> actualAfterFlush1 = await(await(storage.download()).toList());
		assertEquals(expectedAfterFlush1, actualAfterFlush1);

		await(wal.put(2L, GSet.of(2, 3, 100)));
		await(wal.put(1L, GSet.of(9, 10, 11)));
		await(wal.flush());

		List<CrdtData<Long, GSet<Integer>>> actualAfterFlush2 = await(await(storage.download()).toList());
		assertEquals(expectedAfterFlush2, actualAfterFlush2);
	}

	@Test
	public void multipleFlushesConsecutive() {
		await(wal.start());
		long now = getCurrentReactor().currentTimeMillis();
		List<CrdtData<Long, GSet<Integer>>> expected = List.of(
			new CrdtData<>(1L, now, GSet.of(1, 2, 3, 6, 9, 10, 11)),
			new CrdtData<>(2L, now, GSet.of(-12, 0, 2, 3, 100, 200))
		);
		await(
			wal.put(1L, GSet.of(1, 2, 3)),
			wal.put(2L, GSet.of(-12, 0, 200)),
			wal.put(1L, GSet.of(1, 6)),
			wal.flush(),
			wal.put(2L, GSet.of(2, 3, 100)),
			wal.put(1L, GSet.of(9, 10, 11)),
			wal.flush()
		);

		List<CrdtData<Long, GSet<Integer>>> actual = await(await(storage.download()).toList());
		assertEquals(expected, actual);
	}

	@Test
	public void startupWithRemainingWALFiles() throws IOException {
		long now = getCurrentReactor().currentTimeMillis();
		List<CrdtData<Long, GSet<Integer>>> expected = List.of(
			new CrdtData<>(1L, now, GSet.of(1, 2, 3, 6, 9, 10, 11)),
			new CrdtData<>(2L, now, GSet.of(-12, 0, 2, 3, 100, 200))
		);
		craftWALFile(
			new CrdtData<>(1L, now, GSet.of(1, 2, 3)),
			new CrdtData<>(2L, now, GSet.of(-12, 0, 200)),
			new CrdtData<>(1L, now, GSet.of(1, 6))
		);
		craftWALFile(
			new CrdtData<>(2L, now, GSet.of(2, 3, 100)),
			new CrdtData<>(1L, now, GSet.of(9, 10, 11))
		);

		Set<Path> remainingWalFiles;
		try (Stream<Path> list = Files.list(path)) {
			remainingWalFiles = list.collect(toSet());
			assertEquals(2, remainingWalFiles.size());
			assertTrue(await(await(storage.download()).toList()).isEmpty());
		}

		await(wal.start());
		try (Stream<Path> list = Files.list(path)) {
			Set<Path> walFilesAfterStart = list.collect(toSet());
			assertEquals(1, walFilesAfterStart.size());
			assertFalse(remainingWalFiles.contains(first(walFilesAfterStart)));
		}

		List<CrdtData<Long, GSet<Integer>>> actual = await(await(storage.download()).toList());
		assertEquals(expected, actual);
	}

	@Test
	public void startupWithMalformedWALFiles() throws IOException {
		long now = getCurrentReactor().currentTimeMillis();
		List<CrdtData<Long, GSet<Integer>>> expected = List.of(
			new CrdtData<>(1L, now, GSet.of(1, 2, 3, 6, 9, 10, 11)),
			new CrdtData<>(2L, now, GSet.of(-124, -12, 0, 2, 3, 4, 45, 100, 200))
		);
		Path walFile1 = craftWALFile(
			new CrdtData<>(1L, now, GSet.of(1, 2, 3)),
			new CrdtData<>(2L, now, GSet.of(-12, 0, 200)),
			new CrdtData<>(1L, now, GSet.of(1, 6)),
			new CrdtData<>(1L, now, GSet.of(23, -53))
		);
		Path walFile2 = craftWALFile(
			new CrdtData<>(2L, now, GSet.of(2, 3, 100)),
			new CrdtData<>(1L, now, GSet.of(9, 10, 11)),
			new CrdtData<>(2L, now, GSet.of(-124, 4, 45)),
			new CrdtData<>(1L, now, GSet.of(20988, -300, 75)),
			new CrdtData<>(2L, now, GSet.of(100000, -100000))
		);
		truncateFile(walFile1, 0.75);
		truncateFile(walFile2, 0.75);

		Set<Path> remainingWalFiles;
		try (Stream<Path> list = Files.list(path)) {
			remainingWalFiles = list.collect(toSet());
			assertEquals(2, remainingWalFiles.size());
			assertTrue(await(await(storage.download()).toList()).isEmpty());
		}

		await(wal.start());
		try (Stream<Path> list = Files.list(path)) {
			Set<Path> walFilesAfterStart = list.collect(toSet());
			assertEquals(1, walFilesAfterStart.size());
			assertFalse(remainingWalFiles.contains(first(walFilesAfterStart)));
		}

		List<CrdtData<Long, GSet<Integer>>> actual = await(await(storage.download()).toList());
		assertEquals(expected, actual);
	}

	@Test
	public void startupWithEmptyWALFiles() throws IOException {
		long now = getCurrentReactor().currentTimeMillis();
		List<CrdtData<Long, GSet<Integer>>> expected = List.of(
			new CrdtData<>(1L, now, GSet.of(-53, 1, 2, 3, 6, 23)),
			new CrdtData<>(2L, now, GSet.of(-12, 0, 12, 100, 200))
		);
		craftWALFile(
			new CrdtData<>(1L, now, GSet.of(1, 2, 3)),
			new CrdtData<>(2L, now, GSet.of(-12, 0, 200)),
			new CrdtData<>(1L, now, GSet.of(1, 6)),
			new CrdtData<>(1L, now, GSet.of(23, -53)),
			new CrdtData<>(2L, now, GSet.of(-12, 12, 100))
		);
		craftWALFile();
		Path walFile3 = craftWALFile();
		truncateFile(walFile3, 0);

		Set<Path> remainingWalFiles;
		try (Stream<Path> list = Files.list(path)) {
			remainingWalFiles = list.collect(toSet());
			assertEquals(3, remainingWalFiles.size());
			assertTrue(await(await(storage.download()).toList()).isEmpty());
		}

		await(wal.start());
		try (Stream<Path> list = Files.list(path)) {
			Set<Path> walFilesAfterStart = list.collect(toSet());
			assertEquals(1, walFilesAfterStart.size());
			assertFalse(remainingWalFiles.contains(first(walFilesAfterStart)));
		}

		List<CrdtData<Long, GSet<Integer>>> actual = await(await(storage.download()).toList());
		assertEquals(expected, actual);
	}

	@SafeVarargs
	private Path craftWALFile(CrdtData<Long, GSet<Integer>>... mockData) {
		Path file = path.resolve(UUID.randomUUID() + EXT_FINAL);
		await(StreamSuppliers.ofChannelSupplier(ChannelSuppliers.ofValues(mockData)
				.mapAsync(data -> Promises.delay(Duration.ofMillis(1), data)))
			.transformWith(ChannelSerializer.builder(serializer)
				.withAutoFlushInterval(Duration.ZERO)
				.build())
			.transformWith(ChannelFrameEncoder.create(FRAME_FORMAT))
			.streamTo(ChannelFileWriter.open(executor, file)));
		return file;
	}

	private void truncateFile(Path filename, double percent) throws IOException {
		assert percent >= 0 && percent <= 1;
		long size = Files.size(filename);
		try (FileChannel outChan = FileChannel.open(filename, WRITE)) {
			outChan.truncate((long) (size * percent));
		}
	}
}
