content: /*
   * Copyright (c) 2013 - 2014, Allen A. George <allen dot george at gmail dot com>
   *
   * All rights reserved.
   *
   * Redistribution and use in source and binary forms, with or without
   * modification, are permitted provided that the following conditions are met:
   *     * Redistributions of source code must retain the above copyright
   *       notice, this list of conditions and the following disclaimer.
   *     * Redistributions in binary form must reproduce the above copyright
   *       notice, this list of conditions and the following disclaimer in the
   *       documentation and/or other materials provided with the distribution.
   *     * Neither the name of libraft nor the
   *       names of its contributors may be used to endorse or promote products
   *       derived from this software without specific prior written permission.
   *
   * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
   * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
   * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
   * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
   * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
   * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
   * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
   * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
   * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
   * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
   */
  
  package io.libraft.kayvee.store;
  
  import com.fasterxml.jackson.core.JsonGenerator;
  import com.fasterxml.jackson.databind.MappingIterator;
  import com.fasterxml.jackson.databind.ObjectMapper;
  import com.fasterxml.jackson.databind.ObjectReader;
  import com.google.common.collect.Lists;
  import com.google.common.collect.Maps;
  import com.google.common.io.Closeables;
  import io.libraft.Committed;
  import io.libraft.kayvee.api.KeyValue;
  
  import javax.annotation.Nullable;
  import java.io.IOException;
  import java.io.InputStream;
  import java.io.OutputStream;
  import java.util.Collection;
  import java.util.List;
  import java.util.Map;
  
  import static com.google.common.base.Preconditions.checkArgument;
  
  /**
   * Represents the local server's view of the replicated key-value store.
   * Methods in the class are called to transform the local server's key-value
   * state whenever a {@link KayVeeCommand} is committed to the cluster.
   * <p/>
   * This state is not persistent, and will have to be refreshed on every
   * restart. As a result, this component is best thought of as a cache that
   * contains the applied cluster state to a given point in time.
   * <p/>
   * This component is thread-safe.
   */
  public class LocalStore {
  
      private Map<String, String> entries = Maps.newHashMap(); // cannot be final because I replace this map completely when I load new entries from a snapshot
  
      private long lastAppliedIndex = 0;
  
      /**
       * Get the log index of the last applied change.
       * The index returned will monotonically increase within the
       * lifetime of the process.
       *
       * @return index >=0 of the last applied change
       */
      synchronized long getLastAppliedIndex() {
          return lastAppliedIndex;
      }
  
      //
      // IMPORTANT: always update lastAppliedIndex first for all the command operations below
      //
  
      /**
       * Apply a {@link Committed.Type#SKIP}. A {@code SKIP}
       * is <strong>not</strong> a {@link KayVeeCommand}, but is instead
       * generated by underlying Raft cluster operations. It does
       * not affect the key-value state, but does update the last applied index.
       *
       * @param index log index >= 0 associated with the {@code SKIP}
       */
      synchronized void skip(long index) {
          incrementLastAppliedIndex(index);
      }
  
      /**
       * Noop operation. This call does not affect the
       * key-value state, but does update the last applied index.
       *
       * @param index log index >= 0 associated with this command
       */
      synchronized void nop(long index) {
          incrementLastAppliedIndex(index);
      }
  
      /**
       * Get the value for a key.
       *
       * @param index log index >= 0 associated with this command
       * @param key non-null (possibly empty) key for which the value should be retrieved
       * @return a {@code KeyValue} instance containing the most up-to-date {@code key=>value} pair for {@code key}
       * @throws KayVeeException if {@code key} does not exist
       */
      synchronized KeyValue get(long index, String key) throws KayVeeException {
          incrementLastAppliedIndex(index);
  
          String value = entries.get(key);
  
          if (value == null) {
              throw new KeyNotFoundException(key);
          }
  
          return new KeyValue(key, value);
      }
  
      /**
       * Get all (key, value) pairs.
       *
       * @param index log index >= 0 associated with this command
       * @return a <strong>copy</strong> of the most up-to-date {@code key=>value} pairs for all keys
       */
      synchronized Collection<KeyValue> getAll(long index) {
          incrementLastAppliedIndex(index);
  
          Collection<KeyValue> copiedEntries = Lists.newArrayListWithCapacity(entries.size());
          for (Map.Entry<String, String> entry : entries.entrySet()) {
              copiedEntries.add(new KeyValue(entry.getKey(), entry.getValue()));
          }
  
          return copiedEntries;
      }
  
      /**
       * Set a key to a value.
       *
       * @param index log index >= 0 associated with this command
       * @param key non-null (possibly empty) key for which the value should be set
       * @param value non-null, non-empty value for this key
       * @return a {@code KeyValue} instance containing the most up-to-date {@code key=>value} pair for {@code key}
       */
      synchronized KeyValue set(long index, String key, String value) {
          incrementLastAppliedIndex(index);
  
          entries.put(key, value);
  
          return new KeyValue(key, entries.get(key));
      }
  
      /**
       * Do a compare-and-set (CAS), aka. test-and-set, operation for key.
       *
       * @param index log index >= 0 associated with this command
       * @param key non-null (possibly empty) key for which the value should be set
       * @param expectedValue existing value associated with {@code key}.
       *                      If {@code expectedValue} is null {@code LocalStore} <strong>should not</strong>
       *                      contain a {@code key=>value} pair for {@code key}
       * @param newValue new value to be associated with {@code key}.
       *                 If {@code newValue} is null the existing {@code key=>value} pair is deleted
       * @return a {@code KeyValue} instance containing the most up-to-date {@code key=>value} pair for {@code key},
       * or {@code null} if this operation removed the {@code key=>value} pair
       * @throws KeyNotFoundException if {@code key} does not exist
       * @throws ValueMismatchException if {@code expectedValue} does not match the <strong>current</strong> value for {@code key}
       * @throws KeyAlreadyExistsException if {@code expectedValue} is null (indicating that
       * a new key-value mapping should be created), but a {@code key=>value} pair already exists
       */
      synchronized @Nullable KeyValue compareAndSet(long index, String key, @Nullable String expectedValue, @Nullable String newValue) throws KeyNotFoundException, ValueMismatchException, KeyAlreadyExistsException {
          if (expectedValue == null && newValue == null) {
              // while I _could_ update the lastAppliedIndex here, calling code should never call us with these arguments
              throw new IllegalArgumentException(String.format("both expectedValue and newValue null for %s (index:%s)", key, index));
          }
  
          incrementLastAppliedIndex(index);
  
          String existingValue = entries.get(key);
  
          // all possibilities
          // -------------------------------------------------------------------
          // | existingValue | expectedValue |    result
          // |     null      |     null      |    assert newValue != null; create key=>newValue
          // |     null      |     !null     |    KeyNotFoundException
          // |     !null     |     null      |    KeyAlreadyExistsException
          // |     !null     |     !null     |    expectedValue != existingValue ? ValueMismatchException : ( newValue != null ? update key=>newValue : delete key)
          //
          // existingValue != null && expectedValue != null (from last row above)
          // --------------------------------------------------------------------
          // |     match     | result
          // |       Y       | newValue != null ? update key=>newValue : delete key)
          // |       N       | ValueMismatchException
  
          if (existingValue == null) {
              if (expectedValue == null) {
                  entries.put(key, newValue);
              } else {
                  throw new KeyNotFoundException(key);
              }
          } else {
              if (expectedValue != null) {
                  if (existingValue.equals(expectedValue)) {
                      if (newValue != null) {
                          entries.put(key, newValue);
                      } else {
                          entries.remove(key);
                      }
                  } else {
                      throw new ValueMismatchException(key, expectedValue, existingValue);
                  }
              } else {
                  throw new KeyAlreadyExistsException(key);
              }
          }
  
          // TODO (AG): I could just look at newValue, but for now I'll look at the map again
          String finalValue = entries.get(key);
          return finalValue != null ? new KeyValue(key, finalValue) : null;
      }
  
      /**
       * Delete the value for a key. This operation is a noop if the key does not exist.
       *
       * @param index log index >= 0 associated with this command
       * @param key non-null (possibly empty) key for which the value should be deleted
       */
      synchronized void delete(long index, String key) {
          incrementLastAppliedIndex(index);
  
          entries.remove(key);
      }
  
      //
      // state {de}serialization
      // see: http://jackson-users.ning.com/forum/topics/appending-pojos-to-a-json-file
      //
  
      // FIXME (AG): what should I do if one of these two methods throws a RuntimeException with getCause() as a JsonProcessingException?
  
      /**
       * Store all {@code key=>value} pairs to the
       * given {@code snapshotOutputStream}. This implementation
       * stores all pairs as a sequence of JSON {@link KeyValue}
       * instances and closes the stream after all pairs are written.
       * <strong>No</strong> internal state is modified by this operation.
       *
       * @param snapshotOutputStream {@code OutputStream} to which all {@code key=>value} pairs are written
       * @return log index >= 0 of the
       * @throws IOException if the {@code key=>value} pairs cannot
       * be written to {@code snapshotOutputStream}. Since no internal state is
       * modified {@code LocalStore} can still be used safely after an exception is thrown
       * @throws RuntimeException if the {@code key=>value} pairs in {@code LocalStore} cannot be converted to JSON
       */
      synchronized long dumpState(OutputStream snapshotOutputStream) throws IOException {
          try {
              ObjectMapper mapper = new ObjectMapper();
              // by default Jackson 'takes ownership' of a closeable passed
              // to it and _closes_ it after writing a value. this is probably
              // OK if we're writing a set of JSON nodes contained in a
              // single root object. however, since we're writing a set of
              // objects without a root object we have to disable this
              // behavior so that all values are written to the underlying stream
              mapper.configure(JsonGenerator.Feature.AUTO_CLOSE_TARGET, false);
              for (Map.Entry<String, String> entry : entries.entrySet()) {
                  mapper.writeValue(snapshotOutputStream, new KeyValue(entry.getKey(), entry.getValue()));
              }
  
              return lastAppliedIndex;
          } finally {
              Closeables.close(snapshotOutputStream, true);
          }
      }
  
      /**
       * Flushes all {@code key=>value} pairs and replaces them with
       * the values read from the given {@code snapshotInputStream}.
       * This implementation expects the {@code InputStream} to contain
       * a sequence of JSON {@link KeyValue} instances.
       *
       * @param lastAppliedIndex log index >= 0 of the last change contained in this snapshot
       * @param snapshotInputStream {@code InputStream} from which all {@code key=> value} pairs should be read
       * @throws IOException if valid {@code key=>value} pairs cannot be read from {@code snapshotInputStream}.
       * Since no internal state is modified until the {@code snapshotInputStream} is completely read and
       * parsed, {@code LocalStore} can still be used safely after an {@code IOException} is thrown
       * @throws RuntimeException if the {@code key=>value} pairs in {@code snapshotInputStream} cannot be converted to JSON
       */
      synchronized void loadState(long lastAppliedIndex, InputStream snapshotInputStream) throws IOException {
          try {
              // read values from the snapshot into a temporary map
              Map<String, String> snapshotEntries = Maps.newHashMap();
  
              ObjectMapper mapper = new ObjectMapper();
              ObjectReader reader = mapper.reader(KeyValue.class);
              MappingIterator<KeyValue> it = reader.readValues(snapshotInputStream);
              while (it.hasNext()) {
                  KeyValue keyValue = it.next();
                  snapshotEntries.put(keyValue.getKey(), keyValue.getValue()); // values are read into this temporary map!
              }
  
              // update the index
              // NOTE: AFAIU, we should never get a snapshot which contains fewer entries than we've applied
              updateLastAppliedIndex(lastAppliedIndex);
  
              // replace the old entries map with the new one generated from the snapshot
              entries = snapshotEntries;
          } finally {
              Closeables.close(snapshotInputStream, true);
          }
      }
  
      /**
       * Increment by one the value of {@link LocalStore#lastAppliedIndex}.
       *
       * @param expectedNewIndex new value of {@code lastAppliedIndex} after the increment
       * @throws IllegalArgumentException if {@code lastAppliedIndex} is
       * not positive or {@code lastAppliedIndex + 1 != expectedNewIndex}
       */
      private void incrementLastAppliedIndex(long expectedNewIndex) {
          checkArgument(expectedNewIndex > 0, "expectedNewIndex must be positive: given:%s", expectedNewIndex);
          checkArgument(expectedNewIndex == lastAppliedIndex + 1, "lastAppliedIndex not increasing monotonically: lastAppliedIndex:%s expectedNewIndex:%s", lastAppliedIndex, expectedNewIndex);
  
          lastAppliedIndex++;
      }
  
      /**
       * Update the value of {@link LocalStore#lastAppliedIndex}.
       * <p/>
       * Unlike {@link LocalStore#incrementLastAppliedIndex(long)}, this method
       * allows the caller to set the new value {@code lastAppliedIndex} to <strong>any</strong>
       * value greater than its current value
       *
       * @param index new value of {@code lastAppliedIndex}
       * @throws IllegalArgumentException if {@code lastAppliedIndex} is not positive or {@code index <= lastAppliedIndex}
       */
      private void updateLastAppliedIndex(long index) {
          checkArgument(index >= 0, "index must be >=0: given:%s", index);
          checkArgument(index >= lastAppliedIndex, "lastAppliedIndex not increasing monotonically: lastAppliedIndex:%s index:%s", lastAppliedIndex, index);
  
          lastAppliedIndex = index;
      }
  
      //
      // the following commands are to be used within unit tests only
      // modifications to the key=>value state _do not_ update lastAppliedIndex and _do not_ perform any verification
      //
  
      /**
       * Set the log index associated with the last change
       * (command or {@code SKIP}) {@code LocalStore} applied.
       * <p/>
       * <strong>This method is package-private for testing
       * reasons only!</strong> It should <strong>never</strong>
       * be called in a non-test context!
       *
       * @param index log index >= 0 for the last change applied
       */
      synchronized void setLastAppliedIndexForUnitTestsOnly(long index) {
          checkArgument(index >= 0, "index:%s", index);
          lastAppliedIndex = index;
      }
  
      /**
       * Set the value for a key.
       * <p/>
       * <strong>This method is package-private for testing
       * reasons only!</strong> It should <strong>never</strong>
       * be called in a non-test context!
       *
       * @param key non-null (possibly empty) key for which the value should be set
       * @param value non-null, non-empty value for this key
       */
      synchronized void setKeyValueForUnitTestsOnly(String key, String value) {
          entries.put(key, value);
      }
  
      /**
       * Get the value for a key.
       * <p/>
       * <strong>This method is package-private for testing
       * reasons only!</strong> It should <strong>never</strong>
       * be called in a non-test context!
       *
       * @param key non-null (possibly empty) key for which the value should be retrieved
       * @return current value associated with this key, or null if the key does not exist
       */
      synchronized @Nullable String getKeyValueForUnitTestsOnly(String key) {
          return entries.get(key);
      }
  
      /**
       * Get all the {@code key=>value} pairs stored in {@code LocalStore}.
       * <p/>
       * <strong>This method is package-private for testing
       * reasons only!</strong> It should <strong>never</strong>
       * be called in a non-test context!
       *
       * @return a set of all {@code key=>value} pairs stored in {@code LocalStore}
       */
      synchronized Collection<KeyValue> getAllForUnitTestsOnly() {
          List<KeyValue> copiedEntries = Lists.newArrayListWithCapacity(entries.size());
  
          for (Map.Entry<String, String> entry : entries.entrySet()) {
              copiedEntries.add(new KeyValue(entry.getKey(), entry.getValue()));
          }
  
          return copiedEntries;
      }
  }
  