content: /*
   * Copyright (C) 2020 ActiveJ LLC.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   * http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package io.activej.crdt.primitives;
  
  import io.activej.common.time.CurrentTimeProvider;
  import io.activej.serializer.BinaryInput;
  import io.activej.serializer.BinaryOutput;
  import io.activej.serializer.BinarySerializer;
  import org.jetbrains.annotations.Nullable;
  
  import java.util.*;
  import java.util.Map.Entry;
  import java.util.stream.Collectors;
  import java.util.stream.Stream;
  
  import static java.lang.Math.max;
  import static java.util.stream.Collectors.toMap;
  
  @SuppressWarnings("NullableProblems")
  public final class LWWSet<E> implements Set<E>, CrdtType<LWWSet<E>> {
  	private final Map<E, Timestamps> set;
  
  	CurrentTimeProvider now = CurrentTimeProvider.ofSystem();
  
  	private LWWSet(Map<E, Timestamps> set) {
  		this.set = set;
  	}
  
  	public LWWSet() {
  		this(new HashMap<>());
  	}
  
  	@SafeVarargs
  	public static <T> LWWSet<T> of(T... items) {
  		LWWSet<T> set = new LWWSet<>();
  		set.addAll(List.of(items));
  		return set;
  	}
  
  	@Override
  	public LWWSet<E> merge(LWWSet<E> other) {
  		Map<E, Timestamps> newSet = new HashMap<>(set);
  		for (Entry<E, Timestamps> entry : other.set.entrySet()) {
  			newSet.merge(entry.getKey(), entry.getValue(), (ts1, ts2) -> new Timestamps(max(ts1.added, ts2.added), max(ts1.removed, ts2.removed)));
  		}
  		return new LWWSet<>(newSet);
  	}
  
  	@Override
  	public @Nullable LWWSet<E> extract(long timestamp) {
  		Map<E, Timestamps> newSet = set.entrySet().stream()
  			.filter(entry -> entry.getValue().added > timestamp || entry.getValue().removed > timestamp)
  			.collect(toMap(Entry::getKey, Entry::getValue));
  		if (newSet.isEmpty()) {
  			return null;
  		}
  		return new LWWSet<>(newSet);
  	}
  
  	@Override
  	public Stream<E> stream() {
  		return set.entrySet().stream().filter(e -> e.getValue().exists()).map(Entry::getKey);
  	}
  
  	@Override
  	public int size() {
  		//noinspection ReplaceInefficientStreamCount ikwiad
  		return (int) stream().count();
  	}
  
  	@Override
  	public boolean isEmpty() {
  		return size() == 0;
  	}
  
  	@Override
  	public boolean contains(Object o) {
  		//noinspection SuspiciousMethodCalls
  		Timestamps timestamps = set.get(o);
  		return timestamps != null && timestamps.added >= timestamps.removed;
  	}
  
  	@Override
  	public Iterator<E> iterator() {
  		return stream().iterator();
  	}
  
  	@Override
  	public Object[] toArray() {
  		//noinspection SimplifyStreamApiCallChains
  		return stream().toArray();
  	}
  
  	@SuppressWarnings("SimplifyStreamApiCallChains")
  	@Override
  	public <T> T[] toArray(T[] a) {
  		return stream().toArray($ -> a);
  	}
  
  	@Override
  	public boolean add(E t) {
  		Timestamps timestamps = set.get(t);
  		if (timestamps == null) {
  			set.put(t, new Timestamps(now.currentTimeMillis(), 0));
  			return true;
  		}
  		boolean notExisted = !timestamps.exists();
  		timestamps.added = now.currentTimeMillis();
  		return notExisted;
  	}
  
  	@Override
  	@SuppressWarnings("unchecked")
  	public boolean remove(Object o) {
  		//noinspection SuspiciousMethodCalls
  		Timestamps timestamps = set.get(o);
  		if (timestamps == null) {
  			set.put((E) o, new Timestamps(0, now.currentTimeMillis()));
  			return false;
  		}
  		boolean existed = timestamps.exists();
  		timestamps.removed = now.currentTimeMillis();
  		return existed;
  	}
  
  	@Override
  	public boolean containsAll(Collection<?> c) {
  		return stream().allMatch(c::contains);
  	}
  
  	@Override
  	public boolean addAll(Collection<? extends E> c) {
  		boolean added = false;
  		for (E e : c) {
  			added |= add(e);
  		}
  		return added;
  	}
  
  	@Override
  	public boolean retainAll(Collection<?> c) {
  		boolean removed = false;
  		for (E item : this) {
  			if (!c.contains(item)) {
  				remove(item);
  				removed = true;
  			}
  		}
  		return removed;
  	}
  
  	@Override
  	public boolean removeAll(Collection<?> c) {
  		boolean removed = false;
  		for (Object o : c) {
  			removed |= remove(o);
  		}
  		return removed;
  	}
  
  	@Override
  	public void clear() {
  		forEach(this::remove);
  	}
  
  	@Override
  	public String toString() {
  		return set.entrySet()
  			.stream()
  			.filter(e -> e.getValue().exists())
  			.map(e -> Objects.toString(e.getKey()))
  			.collect(Collectors.joining(", ", "[", "]"));
  	}
  
  	public static final class Timestamps {
  		long added, removed;
  
  		Timestamps(long added, long removed) {
  			this.added = added;
  			this.removed = removed;
  		}
  
  		boolean exists() {
  			return added >= removed;
  		}
  	}
  
  	public static class Serializer<T> implements BinarySerializer<LWWSet<T>> {
  		private final BinarySerializer<T> valueSerializer;
  
  		public Serializer(BinarySerializer<T> valueSerializer) {
  			this.valueSerializer = valueSerializer;
  		}
  
  		@Override
  		public void encode(BinaryOutput out, LWWSet<T> item) {
  			out.writeVarInt(item.set.size());
  			for (Entry<T, Timestamps> entry : item.set.entrySet()) {
  				valueSerializer.encode(out, entry.getKey());
  				Timestamps timestamps = entry.getValue();
  				out.writeLong(timestamps.added);
  				out.writeLong(timestamps.removed);
  			}
  		}
  
  		@Override
  		public LWWSet<T> decode(BinaryInput in) {
  			int size = in.readVarInt();
  			Map<T, Timestamps> set = new HashMap<>(size);
  			for (int i = 0; i < size; i++) {
  				set.put(valueSerializer.decode(in), new Timestamps(in.readLong(), in.readLong()));
  			}
  			return new LWWSet<>(set);
  		}
  	}
  }
  