package com.timgroup.eventstore.stitching;

import com.timgroup.eventstore.api.EventReader;
import com.timgroup.eventstore.api.EventSource;
import com.timgroup.eventstore.api.Position;
import com.timgroup.eventstore.api.PositionCodec;
import com.timgroup.eventstore.api.ResolvedEvent;

import javax.annotation.CheckReturnValue;
import javax.annotation.Nonnull;
import javax.annotation.ParametersAreNonnullByDefault;
import java.util.stream.Stream;

import static java.util.Objects.requireNonNull;

@ParametersAreNonnullByDefault
public final class BackfillStitchingEventReader implements EventReader {

    private final StitchedPosition emptyStorePosition;
    private final EventReader backfill;
    private final EventReader live;

    public BackfillStitchingEventReader(EventSource backfill, EventSource live, Position liveCutoffStartPosition) {
        this(backfill.readAll(), live.readAll(), liveCutoffStartPosition);
    }

    public BackfillStitchingEventReader(EventReader backfill, EventReader live, Position liveCutoffStartPosition) {
        this.backfill = backfill;
        this.live = requireNonNull(live);
        this.emptyStorePosition = new StitchedPosition(backfill.emptyStorePosition(), liveCutoffStartPosition);
    }

    @Nonnull
    @Override
    public Position emptyStorePosition() {
        return emptyStorePosition;
    }

    @Nonnull
    @Override
    public PositionCodec storePositionCodec() {
        return StitchedPosition.codec(backfill.storePositionCodec(), live.storePositionCodec());
    }

    @Nonnull
    @CheckReturnValue
    @Override
    public Stream<ResolvedEvent> readAllForwards(Position positionExclusive) {
        StitchedPosition stitchedPosition = (StitchedPosition) positionExclusive;
        if (stitchedPosition.isInBackfill(emptyStorePosition)) {
            return BackfillStitchingEventForwardsSpliterator.stitchedStreamFrom(
                    backfill.readAllForwards(stitchedPosition.backfillPosition),
                    live.readAllForwards(stitchedPosition.livePosition),
                    stitchedPosition
            );
        } else {
            return BackfillStitchingEventForwardsSpliterator.stitchedStreamFrom(
                    Stream.empty(),
                    live.readAllForwards(stitchedPosition.livePosition),
                    stitchedPosition
            );
        }
    }

    @Override
    public String toString() {
        return "BackfillStitchingEventReader{" +
                "emptyStorePosition=" + emptyStorePosition +
                ", backfill=" + backfill +
                ", live=" + live +
                '}';
    }
}

@@ -600,63 +600,8 @@
  %7B%0A%0A
-    private final StitchedPosition emptyStorePosition;%0A
     
@@ -671,16 +671,68 @@
 er live;
+%0A    private final Position liveCutoffStartPosition;
 %0A%0A    pu
@@ -1132,77 +1132,31 @@
 his.
-emptyStorePosition = new StitchedPosition(backfill.emptyStore
+liveCutoffStart
 Position
 (), 
@@ -1151,19 +1151,18 @@
 Position
-(),
+ =
  liveCut
@@ -1169,33 +1169,32 @@
 offStartPosition
-)
 ;%0A    %7D%0A%0A    @No
@@ -1216,32 +1216,40 @@
 ride%0A    public 
+Stitched
 Position emptySt
@@ -1279,34 +1279,92 @@
  return 
-emptyStorePosition
+new StitchedPosition(backfill.emptyStorePosition(), liveCutoffStartPosition)
 ;%0A    %7D%0A
@@ -1795,26 +1795,31 @@
 ackfill(
-emptyStore
+liveCutoffStart
 Position
@@ -2501,41 +2501,51 @@
    %22
-emptyStorePosition=%22 + emptyStore
+liveCutoffStartPosition=%22 + liveCutoffStart
 Posi

store liveCutoffStartPosition