package com.timgroup.eventstore.mysql;

import com.timgroup.filefeed.reading.ReadableFeedStorage;

import java.time.Instant;
import java.util.Comparator;
import java.util.Optional;

import static com.timgroup.filefeed.reading.StorageLocation.TimGroupEventStoreFeedStore;
import static org.joda.time.Instant.ofEpochMilli;

public class FileFeedCacheMaxPositionFetcher implements MaxPositionFetcher {


    private final ReadableFeedStorage readableFeedStorage;
    private final ArchiveKeyFormat archiveKeyFormat;

    public FileFeedCacheMaxPositionFetcher(ReadableFeedStorage readableFeedStorage, ArchiveKeyFormat archiveKeyFormat) {
        this.readableFeedStorage = readableFeedStorage;
        this.archiveKeyFormat = archiveKeyFormat;
    }

    public Optional<BasicMysqlEventStorePosition> maxPosition() {
        return readableFeedStorage.list(TimGroupEventStoreFeedStore, archiveKeyFormat.eventStorePrefix())
                .stream()
                .reduce((r1, r2) -> r2)
                .map(archiveKeyFormat::positionValueFrom);
    }

    @Override
    public Optional<BasicMysqlEventStorePosition> maxPositionBefore(Instant cutOff) {
        return readableFeedStorage.list(TimGroupEventStoreFeedStore, archiveKeyFormat.eventStorePrefix())
                .stream()
                .sorted(Comparator.reverseOrder())
                .filter(file -> readableFeedStorage
                        .getArrivalTime(TimGroupEventStoreFeedStore, file)
                        .map(fileArrivalTime -> !fileArrivalTime.isAfter(ofEpochMilli(cutOff.toEpochMilli())))
                        .orElse(false))
                .reduce((r1, r2) -> r1)
                .map(archiveKeyFormat::positionValueFrom);
    }
}

@@ -960,29 +960,40 @@
    .
-reduce((r1, r2) -%3E r2
+sorted(Comparator.reverseOrder()
 )%0A  
@@ -1039,32 +1039,61 @@
 sitionValueFrom)
+%0A                .findFirst()
 ;%0A    %7D%0A%0A    @Ov
@@ -1650,48 +1650,8 @@
 e))%0A
-                .reduce((r1, r2) -%3E r1)%0A
     
@@ -1703,16 +1703,45 @@
 lueFrom)
+%0A                .findFirst()
 ;%0A    %7D%0A

clean up