package com.timgroup.eventstore.merging;

import com.timgroup.eventstore.api.ResolvedEvent;

import javax.annotation.Nonnull;
import java.time.Duration;
import java.time.Instant;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Objects.requireNonNull;

public interface MergingStrategy<T extends Comparable<T>> {

    @Nonnull T toComparable(ResolvedEvent event);

    @Nonnull default Duration delay() { return Duration.ZERO; }

    @Nonnull default MergingStrategy<T> withDelay(Duration delay) {
        return new DelayedMergingStrategy<>(delay, this);
    }

    final class StreamIndexMergingStrategy implements MergingStrategy<Integer> {
        @Nonnull
        @Override
        public Integer toComparable(ResolvedEvent event) {
            return 0;
        }
    }

    final class EffectiveTimestampMergingStrategy implements MergingStrategy<Instant> {
        private static final Pattern EFFECTIVE_TIMESTAMP_PATTERN = Pattern.compile("\"effective_timestamp\"\\s*:\\s*\"([^\"]+)\"");

        private final boolean fallBackToTimestamp;

        public EffectiveTimestampMergingStrategy() {
            this(false);
        }

        public EffectiveTimestampMergingStrategy(boolean fallBackToTimestamp) {
            this.fallBackToTimestamp = fallBackToTimestamp;
        }

        @Nonnull
        @Override
        public Instant toComparable(ResolvedEvent event) {
            return effectiveTimestampFrom(event);
        }

        private Instant effectiveTimestampFrom(ResolvedEvent event) {
            String metadata = new String(event.eventRecord().metadata(), UTF_8);
            if (!metadata.isEmpty()) {
                Matcher matcher = EFFECTIVE_TIMESTAMP_PATTERN.matcher(metadata);
                if (matcher.find()) {
                    return Instant.parse(matcher.group(1));
                }
            }
            if (fallBackToTimestamp) {
                return event.eventRecord().timestamp();
            }
            throw new IllegalStateException("no timestamp in metadata of " + event);
        }
    }

    final class DelayedMergingStrategy<T extends Comparable<T>> implements MergingStrategy<T> {
        private final Duration delay;
        private final MergingStrategy<T> delegate;

        private DelayedMergingStrategy(Duration delay, MergingStrategy<T> delegate) {
            this.delay = requireNonNull(delay);
            this.delegate = requireNonNull(delegate);
        }

        @Nonnull
        @Override
        public T toComparable(ResolvedEvent event) {
            return delegate.toComparable(event);
        }

        @Nonnull
        @Override
        public Duration delay() {
            return this.delay;
        }
    }
}

@@ -172,16 +172,90 @@
 nstant;%0A
+import java.util.function.BiFunction;%0Aimport java.util.function.Function;%0A
 import j
@@ -717,32 +717,405 @@
 , this);%0A    %7D%0A%0A
+    default %3CU extends Comparable%3CU%3E%3E MergingStrategy%3CU%3E andThen(Function%3C? super T, ? extends U%3E operator) %7B%0A        return re -%3E operator.apply(toComparable(re));%0A    %7D%0A%0A    default %3CU extends Comparable%3CU%3E%3E MergingStrategy%3CU%3E andThen(BiFunction%3C? super ResolvedEvent, ? super T, ? extends U%3E modifier) %7B%0A        return re -%3E modifier.apply(re, toComparable(re));%0A    %7D%0A%0A
     final class 

support modifying merging strategy by appending a modifier