package com.timgroup.eventstore.writerutils;

import com.timgroup.eventstore.api.EventStreamReader;
import com.timgroup.eventstore.api.EventStreamWriter;
import com.timgroup.eventstore.api.NewEvent;
import com.timgroup.eventstore.api.NoSuchStreamException;
import com.timgroup.eventstore.api.ResolvedEvent;
import com.timgroup.eventstore.api.StreamId;
import com.timgroup.eventstore.api.WrongExpectedVersionException;

import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.stream.Stream;

import static com.timgroup.eventstore.api.EventStreamReader.EmptyStreamEventNumber;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;

public final class IdempotentEventStreamWriter implements EventStreamWriter {

    public static class IncompatibleNewEventException extends RuntimeException {
        public final ResolvedEvent currentEvent;
        public final NewEvent newEvent;

        public IncompatibleNewEventException(String message, ResolvedEvent currentEvent, NewEvent newEvent) {
            super("at position " + currentEvent.position() + ": " + message);
            this.newEvent = newEvent;
            this.currentEvent = currentEvent;
        }
    }

    public interface IsCompatible {
        void throwIfIncompatible(ResolvedEvent currentEvent, NewEvent newEvent) throws IncompatibleNewEventException;
    }

    private final EventStreamWriter underlying;
    private final EventStreamReader reader;
    private final IsCompatible isCompatible;

    private IdempotentEventStreamWriter(EventStreamWriter underlying, EventStreamReader reader, IsCompatible isCompatible) {
        this.underlying = requireNonNull(underlying);
        this.reader = requireNonNull(reader);
        this.isCompatible = requireNonNull(isCompatible);
    }

    @Override
    public void write(StreamId streamId, Collection<NewEvent> events) throws IncompatibleNewEventException {
        write(streamId, events, EmptyStreamEventNumber);
    }

    @Override
    public void write(StreamId streamId, Collection<NewEvent> events, long expectedVersion) throws IncompatibleNewEventException {
        Iterator<NewEvent> newEventsIt = events.iterator();
        long newExpectedVersion = expectedVersion;
        try (Stream<ResolvedEvent> resolvedEventStream = reader.readStreamForwards(streamId, expectedVersion)){
            Iterator<ResolvedEvent> currentEventsIt = resolvedEventStream.iterator();

            while (newEventsIt.hasNext() && currentEventsIt.hasNext()) {
                newExpectedVersion += 1;
                NewEvent newEvent = newEventsIt.next();
                ResolvedEvent currentEvent = currentEventsIt.next();
                isCompatible.throwIfIncompatible(currentEvent, newEvent);
            }
        } catch (NoSuchStreamException e) {
            // Ignore
        }

        if (newEventsIt.hasNext()) {
            final Collection<NewEvent> newEventsToSave = new LinkedList<>();
            newEventsIt.forEachRemaining(newEventsToSave::add);
            try {
                underlying.write(streamId, newEventsToSave, newExpectedVersion);
            } catch (WrongExpectedVersionException e) {
                write(streamId, newEventsToSave, newExpectedVersion);
            }
        }
    }

    @Override
    public String toString() {
        return "IdempotentEventStreamWriter{" +
                "underlying=" + underlying +
                ", reader=" + reader +
                ", isCompatible=" + isCompatible +
                '}';
    }

    public static EventStreamWriter idempotent(EventStreamWriter underlying, EventStreamReader reader) {
        return idempotent(underlying, reader, BASIC_COMPATIBILITY_CHECK);
    }

    public static EventStreamWriter idempotentWithMetadataCheck(EventStreamWriter underlying, EventStreamReader reader) {
        return idempotent(underlying, reader, (a, b) -> {
            BASIC_COMPATIBILITY_CHECK.throwIfIncompatible(a, b);
            METADATA_COMPATIBILITY_CHECK.throwIfIncompatible(a, b);
        });
    }

    public static EventStreamWriter idempotent(EventStreamWriter underlying, EventStreamReader reader, IsCompatible isCompatible) {
        return new IdempotentEventStreamWriter(underlying, reader, isCompatible);
    }

    @SuppressWarnings("WeakerAccess")
    public static final IsCompatible BASIC_COMPATIBILITY_CHECK = (a, b) -> {
        if (!a.eventRecord().eventType().equals(b.type())) {
            throw new IncompatibleNewEventException("Event types don't match -- old: " + a.eventRecord().eventType() + ", new: " + b.type(), a, b);
        }
        if (!Arrays.equals(a.eventRecord().data(), b.data())) {
            throw new IncompatibleNewEventException("Event bodies don't match -- old: " + new String(a.eventRecord().data(), UTF_8) + ", new: " + new String(b.data(), UTF_8), a, b);
        }
    };

    @SuppressWarnings("WeakerAccess")
    public static final IsCompatible METADATA_COMPATIBILITY_CHECK = (a, b) -> {
        if (!Arrays.equals(a.eventRecord().metadata(), b.metadata())) {
            throw new IncompatibleNewEventException("Event metadata doesn't match -- old: " + new String(a.eventRecord().metadata(), UTF_8) + ", new: " + new String(b.metadata(), UTF_8), a, b);
        }
    };
}

@@ -4578,267 +4578,662 @@
 ion(
-%22Event types don't match -- old: %22 + a.eventRecord().eventType() + %22, new: %22 + b.type(), a, b);%0A        %7D%0A        if (!Arrays.equals(a.eventRecord().data(), b.data())) %7B%0A            throw new IncompatibleNewEventException(%22Event bodies don't match -- old: %22 +
+%0A                String.format(%0A                    %22Event types don't match -- old: %25s, new: %25s (bodies old: %25s, new %25s)%22,%0A                    a.eventRecord().eventType(),%0A                    b.type(),%0A                    new String(a.eventRecord().data(), UTF_8),%0A                    new String(b.data(), UTF_8)%0A                ),%0A                a,%0A                b%0A            );%0A        %7D%0A        if (!Arrays.equals(a.eventRecord().data(), b.data())) %7B%0A            throw new IncompatibleNewEventException(%0A                String.format(%0A                    %22Event bodies don't match -- old: %25s, new: %25s (types old: %25s, new %25s)%22,%0A                   
  new
@@ -5266,38 +5266,45 @@
 ta(), UTF_8)
- + %22, new: %22 +
+,%0A                   
  new String(
@@ -5312,37 +5312,179 @@
 .data(), UTF_8),
- a, b
+%0A                    a.eventRecord().eventType(),%0A                    b.type()%0A                ),%0A                a,%0A                b%0A            
 );%0A        %7D%0A   

add types and bodies info when idempotent write fails