package com.continuuity.data2.transaction.inmemory;

import com.continuuity.api.common.Bytes;
import com.continuuity.data2.transaction.Transaction;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 *
 */
// todo: synchronize all
// todo: optimize heavily
public class InMemoryTransactionOracle {
  private static List<Long> excludedList;

  // todo: clean it up
  // todo: use moving array instead
  // tx id (write pointer) -> changes made by this tx
  private static Map<Long, Set<byte[]>> committedChangeSets;
  // not committed yet
  private static Map<Long, Set<byte[]>> committingChangeSets;

  private static long readPointer;

  private static long nextWritePointer;

  // todo: do not use static fields, use proper singleton or smth else
  static {
    reset();
  }

  // public for unit-tests
  public static synchronized void reset() {
    excludedList = Lists.newArrayList();
    committedChangeSets = Maps.newHashMap();
    committingChangeSets = Maps.newHashMap();
    readPointer = 0;
    nextWritePointer = 1;
  }

  public static synchronized Transaction start() {
    Transaction tx = new Transaction(readPointer, nextWritePointer, getExcludedListAsArray(excludedList));
    excludedList.add(nextWritePointer);
    Collections.sort(excludedList);
    nextWritePointer++;
    return tx;
  }

  public static synchronized boolean canCommit(Transaction tx, Collection<byte[]> changeIds) {
    if (hasConflicts(tx, changeIds)) {
      return false;
    }

    Set<byte[]> set = Sets.newTreeSet(Bytes.BYTES_COMPARATOR);
    set.addAll(changeIds);
    committingChangeSets.put(tx.getWritePointer(), set);

    return true;
  }

  public static synchronized boolean commit(Transaction tx) {
    // todo: these should be atomic
    // NOTE: whether we succeed or not we don't need to keep changes in committing state: same tx cannot be attempted to
    //       commit twice
    Set<byte[]> changeSet = committingChangeSets.remove(tx.getWritePointer());

    if (changeSet != null) {
      // double-checking if there are conflicts: someone may have committed since canCommit check
      if (hasConflicts(tx, changeSet)) {
        return false;
      }

      committedChangeSets.put(tx.getWritePointer(), changeSet);
    }
    makeVisible(tx);
    return true;
  }

  public static synchronized boolean abort(Transaction tx) {
    committingChangeSets.remove(tx.getWritePointer());
    // makes tx visible (assumes that all operations were rolled back)
    makeVisible(tx);
    return true;
  }

  private static boolean hasConflicts(Transaction tx, Collection<byte[]> changeIds) {
    if (changeIds.size() == 0) {
      return false;
    }

    // Go thru all tx committed after given tx was started and check if any of them has change
    // conflicting with the given
    return hasConflicts(tx, changeIds, committedChangeSets);

    // NOTE: we could try to optimize for some use-cases and also check those being committed for conflicts to
    //       avoid later the cost of rollback. This is very complex, but the cost of rollback is so high that we
    //       can go a bit crazy optimizing stuff around it...
  }

  private static boolean hasConflicts(Transaction tx, Collection<byte[]> changeIds, Map<Long, Set<byte[]>> changeSets) {
    for (Map.Entry<Long, Set<byte[]>> changeSet : changeSets.entrySet()) {
      // basically not visible but committed means "tx committed after given tx was started"
      if (!visible(tx, changeSet.getKey())) {
        if (containsAny(changeSet.getValue(), changeIds)) {
          return true;
        }
      }
    }
    return false;
  }

  private static boolean containsAny(Set<byte[]> set, Collection<byte[]> toSearch) {
    for (byte[] item : toSearch) {
      if (set.contains(item)) {
        return true;
      }
    }

    return false;
  }

  // todo: move to Tx?
  private static boolean visible(Transaction tx, long pointer) {
    if (pointer > tx.getReadPointer()) {
      return false;
    }

    // todo: optimize heavily
    // we rely on array of excludes to be sorted
    return Arrays.binarySearch(tx.getExcludedList(), pointer) < 0;
  }

  private static void makeVisible(Transaction tx) {
    excludedList.remove(new Long(tx.getWritePointer()));

    // moving read pointer
    moveReadPointerIfNeeded(tx.getWritePointer());
  }

  private static void moveReadPointerIfNeeded(long committedWritePointer) {
    if (committedWritePointer > readPointer) {
      readPointer = committedWritePointer;
    }
  }

  private static long[] getExcludedListAsArray(List<Long> excludedList) {
    // todo: optimize (cache, etc. etc.)
    long[] result = new long[excludedList.size()];
    for (int i = 0; i < result.length; i++) {
      result[i] = excludedList.get(i);
    }

    return result;
  }

}

@@ -258,16 +258,157 @@
 ct.Sets;
+%0Aimport it.unimi.dsi.fastutil.longs.LongArrayList;%0Aimport it.unimi.dsi.fastutil.longs.LongList;%0Aimport it.unimi.dsi.fastutil.longs.LongLists;
 %0A%0Aimport
@@ -656,16 +656,58 @@
 racle %7B%0A
+  private static LongList excludedList;%0A//
   privat
@@ -1245,24 +1245,26 @@
 d reset() %7B%0A
+//
     excluded
@@ -1288,24 +1288,64 @@
 rrayList();%0A
+    excludedList = new LongArrayList();%0A
     committe
@@ -4980,25 +4980,23 @@
 sArray(L
-ist%3CLong%3E
+ongList
  exclude
@@ -5096,16 +5096,59 @@
 ize()%5D;%0A
+    return excludedList.toArray(result);%0A//
     for 
@@ -5181,24 +5181,26 @@
 gth; i++) %7B%0A
+//
       result
@@ -5226,23 +5226,29 @@
 get(i);%0A
+//
     %7D%0A
-%0A
+//%0A//
     retu

Use LongList from fastutil