package com.timgroup.eventstore.merging;

import com.timgroup.eventstore.api.*;

import java.nio.charset.StandardCharsets;
import java.util.Optional;
import java.util.stream.Stream;

import static com.timgroup.eventstore.api.NewEvent.newEvent;
import static com.timgroup.eventstore.api.StreamId.streamId;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.singleton;

public final class EventShovel {
    private final EventReader reader;
    private final PositionCodec readerPositionCodec;
    private final EventSource output;

    public EventShovel(EventReader reader, PositionCodec readerPositionCodec, EventSource output) {
        this.reader = reader;
        this.readerPositionCodec = readerPositionCodec;
        this.output = output;
    }

    public void shovelAllNewlyAvailableEvents() {
        try (Stream<ResolvedEvent> backwardsWrittenOutputEvents = output.readAll().readAllBackwards()) {
            Optional<ResolvedEvent> maybeLastWrittenEvent = backwardsWrittenOutputEvents.findFirst();

            Position currentPosition = maybeLastWrittenEvent.map(evt -> readerPositionCodec.deserializePosition(new String(evt.eventRecord().metadata(), UTF_8))).orElse(reader.emptyStorePosition());

            reader.readAllForwards(currentPosition).forEach(evt -> {
                byte[] metadata = readerPositionCodec.serializePosition(evt.position()).getBytes(UTF_8);
                output.writeStream().write(
                        evt.eventRecord().streamId(),
                        singleton(newEvent(evt.eventRecord().eventType(), evt.eventRecord().data(), metadata))
                );
            });
        }
    }
}

@@ -1109,74 +1109,76 @@
 vent
+%0A                    
 .map(
+r
 e
-vt
  -%3E 
-readerPositionCodec.deserializePosition(new String(evt
+extractShovelPositionFromMetadata(re
 .eve
@@ -1202,18 +1202,31 @@
 ta()
-, UTF_8)))
+))%0A                    
 .orE
@@ -1367,37 +1367,32 @@
 a = 
+c
 rea
-derPositionCodec.serialize
+teMetadataWithShovel
 Posi
@@ -1414,24 +1414,38 @@
 on()
+, evt.eventRecord(
 ).
-getBytes(UTF_8
+metadata()
 );%0A 
@@ -1702,11 +1702,950 @@
  %7D%0A    %7D
+%0A%0A    /* David Ellis 03/02/2017 - Please do this properly */%0A    // metadata transfer from upstream -- need resolve how we combine our shovel position record with the upstream metadata%0A    //    from upstream we have completely unknown black box of metadata as a byte%5B%5D%0A    //    strategy: assume it is json -- prove by decoding%0A    //        if json decode fails, throw away upstream metadata and just write ours downstream%0A    //        if json decode passes, add our shovel_position field to it, and re-encode%0A%0A    private Position extractShovelPositionFromMetadata(byte%5B%5D metadata) %7B%0A        return readerPositionCodec.deserializePosition(new String(metadata, UTF_8));%0A    %7D%0A%0A    private byte%5B%5D createMetadataWithShovelPosition(Position shovelPosition, byte%5B%5D upstreamMetadata) %7B%0A        return readerPositionCodec.serializePosition(shovelPosition).getBytes(UTF_8);%0A    %7D%0A%0A    //TODO:%0A    // batched writing%0A    // optimistic locking%0A%0A
 %0A%7D%0A

prepare for metadata madness