package com.github.msemys.esjc;

import com.github.msemys.esjc.operation.StreamDeletedException;
import com.github.msemys.esjc.operation.StreamNotFoundException;
import org.junit.Test;

import java.util.List;
import java.util.stream.Stream;

import static com.github.msemys.esjc.matcher.RecordedEventListMatcher.containsInOrder;
import static java.util.stream.Collectors.toList;
import static java.util.stream.IntStream.range;
import static org.hamcrest.CoreMatchers.instanceOf;
import static org.junit.Assert.*;

public class ITStreamEventsForward extends AbstractEventStoreTest {

    public ITStreamEventsForward(EventStore eventstore) {
        super(eventstore);
    }

    @Test(expected = IllegalArgumentException.class)
    public void failsToProcessWhenBatchSizeIsZero() {
        eventstore.streamEventsForward(generateStreamName(), 0, 0, false);
    }

    @Test(expected = IllegalArgumentException.class)
    public void failsToProcessWhenBatchSizeOutOfRange() {
        eventstore.streamEventsForward("foo", StreamPosition.START, Integer.MAX_VALUE, false);
    }

    @Test
    public void failsToProcessNonExistingStream() {
        final String stream = generateStreamName();

        Stream<ResolvedEvent> eventStream = eventstore.streamEventsForward(stream, 0, 5, false);

        try {
            eventStream.findFirst();
            fail("should fail with 'StreamNotFoundException'");
        } catch (Exception e) {
            assertThat(e, instanceOf(StreamNotFoundException.class));
            assertEquals(stream, ((StreamNotFoundException) e).stream);
        }
    }

    @Test
    public void failsToProcessDeletedStream() {
        final String stream = generateStreamName();

        eventstore.deleteStream(stream, ExpectedVersion.NO_STREAM, true).join();

        Stream<ResolvedEvent> eventStream = eventstore.streamEventsForward(stream, 0, 5, false);

        try {
            eventStream.findFirst();
            fail("should fail with 'StreamDeletedException'");
        } catch (Exception e) {
            assertThat(e, instanceOf(StreamDeletedException.class));
            assertEquals(stream, ((StreamDeletedException) e).stream);
        }
    }

    @Test
    public void readsStreamEventsToEnd() {
        final String stream = generateStreamName();

        List<EventData> events = newTestEvents(10);
        eventstore.appendToStream(stream, ExpectedVersion.NO_STREAM, events).join();

        List<ResolvedEvent> result = eventstore.streamEventsForward(stream, 8, 5, false).collect(toList());

        assertEquals(2, result.size());
        assertThat(recordedEventsFrom(result), containsInOrder(events.stream().skip(8).collect(toList())));
    }

    @Test
    public void readsStreamEventsFromStartToEndWithSmallBatchSize() {
        final String stream = generateStreamName();

        eventstore.appendToStream(stream, ExpectedVersion.NO_STREAM, newTestEvents(10)).join();

        assertEquals(10, eventstore.streamEventsForward(stream, StreamPosition.START, 2, false).count());
    }

    @Test
    public void readsStreamEventsFromStartToEndWithLargeBatchSize() {
        final String stream = generateStreamName();

        eventstore.appendToStream(stream, ExpectedVersion.NO_STREAM, newTestEvents(10)).join();

        assertEquals(10, eventstore.streamEventsForward(stream, StreamPosition.START, 20, false).count());
    }

    @Test
    public void readsStreamEventsFromStartToEndWithMaxBatchSize() {
        final String stream = generateStreamName();

        eventstore.appendToStream(stream, ExpectedVersion.NO_STREAM, newTestEvents(10)).join();

        assertEquals(10, eventstore.streamEventsForward(stream, StreamPosition.START, 4096, false).count());
    }

    @Test
    public void readsEventsInSameOrderAsWritten() {
        final String stream = generateStreamName();

        List<EventData> events = newTestEvents(10);
        eventstore.appendToStream(stream, ExpectedVersion.NO_STREAM, events).join();

        List<ResolvedEvent> result = eventstore.streamEventsForward(stream, StreamPosition.START, 3, false).collect(toList());

        assertEquals(10, result.size());
        assertThat(recordedEventsFrom(result), containsInOrder(events));
    }

    @Test
    public void parallelReadsEventsInSameOrderAsWritten() {
        final String stream = generateStreamName();

        List<EventData> events = newTestEvents(100);
        eventstore.appendToStream(stream, ExpectedVersion.NO_STREAM, events).join();

        List<ResolvedEvent> result = eventstore.streamEventsForward(stream, StreamPosition.START, 6, false)
            .parallel()
            .collect(toList());

        assertEquals(100, result.size());
        assertThat(recordedEventsFrom(result), containsInOrder(events));
    }

    @Test
    public void aggregatesEvents() {
        final String stream = generateStreamName();

        List<EventData> events = range(0, 101)
            .mapToObj(i -> EventData.newBuilder()
                .type(i % 2 == 0 ? "even" : "odd")
                .data(String.valueOf(i))
                .build())
            .collect(toList());

        eventstore.appendToStream(stream, ExpectedVersion.NO_STREAM, events).join();

        assertEquals(51, eventstore.streamEventsForward(stream, StreamPosition.START, 11, false)
            .filter(e -> e.event.eventType.equals("even"))
            .count());

        assertEquals(50, eventstore.streamEventsForward(stream, StreamPosition.START, 11, false)
            .filter(e -> e.event.eventType.equals("odd"))
            .count());
    }

    @Test
    public void parallelAggregatesEvents() {
        final String stream = generateStreamName();

        List<EventData> events = range(0, 101)
            .mapToObj(i -> EventData.newBuilder()
                .type(i % 2 == 0 ? "even" : "odd")
                .data(String.valueOf(i))
                .build())
            .collect(toList());

        eventstore.appendToStream(stream, ExpectedVersion.NO_STREAM, events).join();

        assertEquals(51, eventstore.streamEventsForward(stream, StreamPosition.START, 11, false)
            .parallel()
            .filter(e -> e.event.eventType.equals("even"))
            .count());

        assertEquals(50, eventstore.streamEventsForward(stream, StreamPosition.START, 11, false)
            .parallel()
            .filter(e -> e.event.eventType.equals("odd"))
            .count());
    }

}
