package ft;

import com.google.common.hash.Hashing;

public class Document {
  public final String uri;
  public final String content;

  public Document(String uri, String content) {
    this.uri = uri;
    this.content = content;
  }

  public String hash() {
    //use short prefix of hash for example
    return Hashing.sha1().hashString(content).toString().substring(0, 7);
  }
}

package ft;

import org.apache.fluo.api.client.Loader;
import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.api.data.Column;

public class DocLoader implements Loader {

  private final Document doc;

  public static final Column HASH_COL = new Column("uri", "hash");
  public static final Column REF_COUNT_COL = new Column("doc", "refc");
  public static final Column REF_STATUS_COL = new Column("doc", "refs");
  public static final Column CONTENT_COL = new Column("doc", "content");

  public DocLoader(Document doc) {
    this.doc = doc;
  }

  @Override
  public void load(TransactionBase tx, Context context) throws Exception {
    String newHash = doc.hash();
    String oldHash = tx.gets("u:" + doc.uri, HASH_COL);

    // TODO check if uri already has the same content hash.  If so, then nothing to do.

    // TODO set the new hash associated with the URI

    if (oldHash != null) {
      // TODO decrement the reference count at row "d:"+oldHash
      // TODO set REF_STATUS_COL to "unreferenced" when the reference count goes from 1 to 0. Do
      // this for row "d:"+oldHash
    }

    // TODO increment the reference count for the newHash content.
    // TODO add the new content when the reference count does not exists
    // TODO set REF_STATUS_COL to "referenced" when the reference count for the new content goes
    // from 0 to 1.  Do this for row "d:"+newHash
  }
}

  // some test data
  private static Document[] docs1 = new Document[] {
      new Document("http://news.com/a23",
          "Jebediah orbits Mun for 35 days.  No power, forgot solar panels."),
      new Document("http://news.com/a24",
          "Bill plans to rescue Jebediah after taking tourist to Minimus.")};

  private static Document[] docs2 = new Document[] {new Document("http://oldnews.com/a23",
      "Jebediah orbits Mun for 35 days.  No power, forgot solar panels.")};

  private static Document[] docs3 = new Document[] {
      new Document("http://news.com/a23",
          "Jebediah orbits Mun for 38 days.  No power, forgot solar panels."),
      new Document("http://news.com/a24",
          "Crisis at KSC.  Tourist stuck at Minimus.  Bill forgot solar panels.")};

  /**
   * Utility method for loading documents and printing out Fluo table after load completes.
   */
  private static void loadAndPrint(MiniFluo mini, FluoClient client, Document[] docs) {

    try (LoaderExecutor loaderExecutor = client.newLoaderExecutor()) {
      for (Document document : docs) {
        loaderExecutor.execute(new DocLoader(document));
      }
    } // this will close loaderExecutor and wait for all load transactions to complete

    //This line is not needed in this step of the exercise.  However the next step will need this
    //line.
    mini.waitForObservers();

    FluoITHelper.printFluoTable(client);
    System.out.println();
  }

  private static void exercise(MiniFluo mini, FluoClient client) {
    loadAndPrint(mini, client, docs1);
    loadAndPrint(mini, client, docs2);
    loadAndPrint(mini, client, docs3);
  }

package ft;

import java.util.*;

import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.api.data.*;
import org.apache.fluo.api.observer.StringObserver;

public class ContentObserver implements StringObserver {

  public static final Column PROCESSED_COL = new Column("doc", "processed");
  public static final Column WORD_COUNT = new Column("word","docCount");

  /**
   * Utility method to tokenize the content of a document into unique words.
   */
  private List<String> tokenize(String content) {
    return Arrays.asList(content.split("[\\W]+"));
  }

  /**
   *  Adds the passed to delta to the values for each word.
   */
  private void adjustCounts(TransactionBase tx, int delta, List<String> words) {
     Set<String> uniqueWords = new HashSet<String>(words);
    // TODO make a single call to get all of the current word counts.  Could use
    //tx.gets(Collection<RowColumn>)

    // TODO for each word, add delta to the current value and set the new value
  }


  @Override
  public void process(TransactionBase tx, String row, Column col) throws Exception {

    Map<Column, String> colVals =
        tx.gets(row, DocLoader.CONTENT_COL, DocLoader.REF_STATUS_COL, PROCESSED_COL);

    String content = colVals.get(DocLoader.CONTENT_COL);
    String status = colVals.get(DocLoader.REF_STATUS_COL);
    String processed = colVals.getOrDefault(PROCESSED_COL, "false");

    // TODO if status is referenced and not already processed then adjustCounts by +1 and set
    // PROCESSED_COL to true

    // TODO if status is unreferenced then delete all columns for content
    // TODO if status is unreferenced and document was processed, then adjust counts by -1
  }
}

  public static class TourObserverProvider implements ObserverProvider {
    @Override
    public void provide(Registry obsRegistry, Context ctx) {
      obsRegistry.forColumn(DocLoader.REF_STATUS_COL, NotificationType.STRONG)
          .useObserver(new ContentObserver());;
    }
  }

  private static void preInit(FluoConfiguration fluoConfig) {
    fluoConfig.setObserverProvider(TourObserverProvider.class);
  }

package ft;

import org.apache.fluo.api.client.TransactionBase;
import org.apache.fluo.recipes.core.combine.ChangeObserver;

import static ft.ContentObserver.WORD_COUNT;

public class WordCountChangeObserver implements ChangeObserver<String, Long> {

  @Override
  public void process(TransactionBase tx, Iterable<Change<String, Long>> changes) {
    // This print shows per bucket processing.
    System.out.println("== begin processing word count changes ==");

    for (Change<String, Long> change : changes) {

      long oldCount = change.getOldValue().orElse(0l);  //previous count for a word
      long newCount = change.getNewValue().orElse(0l);  //new count for a word

      if (change.getOldValue().isPresent()) {
        // delete old count for word from inverted index
        tx.delete(String.format("ic:%06d:%s", oldCount, change.getKey()), WORD_COUNT);
      }

      if (change.getNewValue().isPresent()) {
        // insert new count for word into inverted index
        tx.set(String.format("ic:%06d:%s", newCount, change.getKey()), WORD_COUNT, "");
      }

      // change.getKey() is a word
      System.out.printf("  update %s %d -> %d\n", change.getKey(), oldCount, newCount);
    }

    System.out.println("== end processing word count changes ==");
  }
}

  public static class TourObserverProvider implements ObserverProvider {
    @Override
    public void provide(Registry obsRegistry, Context ctx) {
      CombineQueue<String, Long> wordCQ = CombineQueue.getInstance("wc", ctx.getAppConfiguration());

      // Pass the word count combine queue to the ContentObserver so it can queue updates 
      // when a documents word counts change.
      obsRegistry.forColumn(DocLoader.REF_STATUS_COL, NotificationType.STRONG)
          .useObserver(new ContentObserver(wordCQ));

      // Register observer to process queued updates, the observer will call the two functions.
      // SummingCombiner is provided by Fluo Recipes.  It sums all updates queued for a key.
      wordCQ.registerObserver(obsRegistry, new SummingCombiner<>(), new WordCountChangeObserver());
    }
  }

  private static void preInit(FluoConfiguration fluoConfig) {
    fluoConfig.setObserverProvider(TourObserverProvider.class);
    CombineQueue.configure("wc").keyType(String.class).valueType(Long.class).buckets(3)
        .save(fluoConfig);
  }

  private CombineQueue<String, Long> wordCQ;

  public ContentObserver(CombineQueue<String, Long> wordCQ) {
    this.wordCQ = wordCQ;
  }

  private void adjustCounts(TransactionBase tx, int delta, List<String> words) {
    HashMap<String, Long> wcUpdates = new HashMap<>();
    words.forEach(word -> wcUpdates.put(word, (long) delta));
    wordCQ.addAll(tx, wcUpdates);
  }

package ft;

import org.apache.fluo.api.config.FluoConfiguration;

/**
 * Generates application config.
 */
public class GenConfig {
  public static void main(String[] args) {
    FluoConfiguration conf = new FluoConfiguration();
    Main.preInit(conf);
    conf.save(System.out);
  }
}

package ft;

import java.nio.charset.StandardCharsets;
import java.nio.file.*;
import javax.inject.Inject;
import org.apache.fluo.api.client.*;
import org.apache.fluo.api.config.FluoConfiguration;

/**
 * Loads one or more document passed in on the command line.
 */
public class Load {
  // when run with fluo exec command, the applications configuration will be injected
  @Inject
  private static FluoConfiguration fluoConfig;

  public static void main(String[] args) throws Exception {
    try (FluoClient client = FluoFactory.newClient(fluoConfig);
        LoaderExecutor loaderExecutor = client.newLoaderExecutor()) 
    {
      for (String filename : args) {
        Path path = Paths.get(filename);
        byte[] encoded = Files.readAllBytes(path);
        String docContent = new String(encoded, StandardCharsets.UTF_8);
        String uri = path.toAbsolutePath().normalize().toUri().toString();
        Document doc = new Document(uri, docContent);
        loaderExecutor.execute(new DocLoader(doc));
      }
    }
  }
}