package phrasecount;

import static phrasecount.Constants.DOC_CONTENT_COL;
import static phrasecount.Constants.DOC_REF_COUNT_COL;
import static phrasecount.Constants.INDEX_STATUS_COL;
import static phrasecount.Constants.STAT_DOC_COUNT_COL;
import static phrasecount.Constants.STAT_SUM_COL;
import static phrasecount.Constants.TYPEL;

import java.util.Map.Entry;
import java.util.Set;

import org.apache.accumulo.core.data.ByteSequence;

import accismus.api.Column;
import accismus.api.Observer;
import accismus.api.Transaction;
import accismus.api.types.TypedTransaction;

public class DocumentIndexer implements Observer {

  private static enum IndexStatus {
    INDEXED, UNINDEXED
  }

  public void process(Transaction tx, ByteSequence row, Column col) throws Exception {

    TypedTransaction ttx = TYPEL.transaction(tx);

    // check if the trigger row exist, if not an exception will thrown and the framework will check for collisions
    ttx.get(row, col).equals("");

    IndexStatus status = getStatus(ttx, row);
    int refCount = ttx.get().row(row).col(DOC_REF_COUNT_COL).toInteger();

    if (status == IndexStatus.UNINDEXED && refCount > 0) {
      indexDocument(ttx, row, 1);
      ttx.set().row(row).col(INDEX_STATUS_COL).val(IndexStatus.INDEXED.name());
    } else if (status == IndexStatus.INDEXED && refCount == 0) {
      indexDocument(ttx, row, -1);
      deleteDocument(ttx, row);
    }

    tx.delete(row, col);

  }

  private void deleteDocument(TypedTransaction tx, ByteSequence row) {
    // TODO it would probably be useful to have a deleteRow method on Transaction... this method could start off w/ a simple implementation and later be
    // optimized... or could have a delete range option

    // TODO this is brittle, this code assumes it knows all possible columns
    tx.delete(row, DOC_CONTENT_COL);
    tx.delete(row, DOC_REF_COUNT_COL);
    tx.delete(row, INDEX_STATUS_COL);

  }

  private void indexDocument(TypedTransaction ttx, ByteSequence row, int multiplier) throws Exception {
    String content = ttx.get().row(row).col(Constants.DOC_CONTENT_COL).toString();

    // this makes the assumption that the implementation of getPhrases is invariant. This is probably a bad assumption. A possible way to make this more robust
    // is to store the output of getPhrases when indexing and use the stored output when unindexing. Alternatively, could store the version of Document used for
    // indexing.
    Set<Entry<String,Integer>> phrases = new Document(null, content).getPhrases().entrySet();

    for (Entry<String,Integer> entry : phrases) {
      String phrase = entry.getKey();
      String phraseRow = "phrase:" + phrase;
      Integer sum = ttx.get().row(phraseRow).col(STAT_SUM_COL).toInteger(0);
      Integer docCount = ttx.get().row(phraseRow).col(STAT_DOC_COUNT_COL).toInteger(0);

      int newSum = sum + multiplier * entry.getValue();
      int newDocCount = docCount + multiplier * 1;

      if (newSum > 0)
        ttx.set().row(phraseRow).col(STAT_SUM_COL).val(newSum);
      else
        ttx.delete().row(phraseRow).col(STAT_SUM_COL);

      if (newDocCount > 0)
        ttx.set().row(phraseRow).col(STAT_DOC_COUNT_COL).val(newDocCount);
      else
        ttx.delete().row(phraseRow).col(STAT_DOC_COUNT_COL);
    }

  }

  private IndexStatus getStatus(TypedTransaction tx, ByteSequence row) throws Exception {
    String status = tx.get().row(row).col(INDEX_STATUS_COL).toString();

    if (status == null)
      return IndexStatus.UNINDEXED;

    return IndexStatus.valueOf(status);
  }

}

@@ -348,17 +348,39 @@
 til.
-Map.Entry
+ArrayList;%0Aimport java.util.Map
 ;%0Aim
@@ -394,19 +394,25 @@
 va.util.
-Set
+Map.Entry
 ;%0A%0Aimpor
@@ -580,27 +580,114 @@
 es.Typed
-Transaction
+Snapshot.Value;%0Aimport accismus.api.types.TypedTransaction;%0A%0Aimport com.google.common.collect.Sets
 ;%0A%0Apubli
@@ -2562,17 +2562,11 @@
     
-Set%3CEntry
+Map
 %3CStr
@@ -2577,17 +2577,16 @@
 Integer%3E
-%3E
  phrases
@@ -2632,18 +2632,318 @@
 es()
-.entrySet(
+;%0A%0A    ArrayList%3CString%3E rows = new ArrayList%3CString%3E(phrases.size());%0A%0A    for (String phrase : phrases.keySet()) %7B%0A      String phraseRow = %22phrase:%22 + phrase;%0A      rows.add(phraseRow);%0A    %7D%0A%0A    Map%3CString,Map%3CColumn,Value%3E%3E storedPhrases = ttx.getd(rows, Sets.newHashSet(STAT_SUM_COL, STAT_DOC_COUNT_COL)
 );%0A%0A
@@ -2988,16 +2988,27 @@
  phrases
+.entrySet()
 ) %7B%0A    
@@ -3090,56 +3090,171 @@
 se;%0A
+%0A
       
-I
+i
 nt
-eger
  sum = 
-ttx.get().row(phraseRow).col
+0;%0A      int docCount = 0;%0A%0A      Map%3CColumn,Value%3E columns = storedPhrases.get(phraseRow);%0A      if (columns != null) %7B%0A        sum = columns.get
 (STA
@@ -3274,17 +3274,16 @@
 Integer(
-0
 );%0A     
@@ -3283,23 +3283,17 @@
 ;%0A      
-Integer
+ 
  docCoun
@@ -3300,36 +3300,19 @@
 t = 
-ttx.get().row(phraseRow).col
+columns.get
 (STA
@@ -3342,11 +3342,18 @@
 ger(
-0
 );
+%0A      %7D
 %0A%0A  

use batch get for phrases