/*
 * Copyright 2016 Patrick Moor <patrick@moor.ws>
 *
 * 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 ws.moor.gletscher.blocks;

import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.hash.HashCode;
import com.google.common.hash.Hashing;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import ws.moor.gletscher.cloud.CloudFileStorage;
import ws.moor.gletscher.util.Signer;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class BlockStore {

  private final CloudFileStorage cloudFileStorage;
  private final Signer signer;
  private final Map<Signature, ListenableFuture<PersistedBlock>> pendingWrites = new HashMap<>();

  private final Object lock = new Object();

  public BlockStore(CloudFileStorage cloudFileStorage, Signer signer) {
    this.cloudFileStorage = cloudFileStorage;
    this.signer = signer;
  }

  public ListenableFuture<PersistedBlock> store(byte[] block, boolean cache) {
    HashCode md5 = Hashing.md5().hashBytes(block);

    int length = block.length;
    Signature signature = signer.computeSignature(block);
    PersistedBlock persisted = new PersistedBlock(signature, length);
    String fileName = toFileName(persisted);

    final SettableFuture<PersistedBlock> future;
    synchronized (lock) {
      if (Futures.getUnchecked(cloudFileStorage.exists(fileName))) {
        return Futures.immediateFuture(persisted);
      } else if (pendingWrites.containsKey(signature)) {
        return pendingWrites.get(signature);
      } else {
        future = SettableFuture.create();
        pendingWrites.put(signature, future);
      }
    }

    ListenableFuture<?> storeFuture = cloudFileStorage.store(
        fileName, block, md5, ImmutableMap.of(), new CloudFileStorage.StoreOptions(cache));
    Futures.addCallback(storeFuture, new FutureCallback<Object>() {
      @Override public void onSuccess(Object result) {
        synchronized (lock) {
          pendingWrites.remove(signature);
        }
        future.set(persisted);
      }

      @Override public void onFailure(Throwable t) {
        synchronized (lock) {
          pendingWrites.remove(signature);
        }
        if (t instanceof CloudFileStorage.FileAlreadyExistsException) {
          future.set(persisted);
        } else {
          future.setException(t);
        }
      }
    });
    return future;
  }

  public ListenableFuture<byte[]> retrieve(PersistedBlock block) {
    return cloudFileStorage.get(toFileName(block));
  }

  private static String toFileName(PersistedBlock block) {
    Signature signature = block.getSignature();
    return String.format("blocks/%02x/%02x/%s:%d",
        signature.getFirstByte(), signature.getSecondByte(), signature.toString(), block.getOriginalLength());
  }

  public Set<PersistedBlock> listAllBlocks() {
    final Set<PersistedBlock> all = new TreeSet<>();

    ExecutorService executor = Executors.newFixedThreadPool(8);
    for (int i = 0; i < 16; i++) {
      final String prefix = String.format("blocks/%x", i);
      executor.execute(() -> {
        Set<PersistedBlock> set = new TreeSet<>();
        Iterator<CloudFileStorage.FileHeader> it = cloudFileStorage.listFiles(prefix, Integer.MAX_VALUE);
        while (it.hasNext()) {
          CloudFileStorage.FileHeader header = it.next();
          set.add(parseFileName(header.name));
        }
        synchronized (all) {
          all.addAll(set);
        }
      });
    }
    MoreExecutors.shutdownAndAwaitTermination(executor, 1, TimeUnit.HOURS);
    return all;
  }

  private PersistedBlock parseFileName(String name) {
    Preconditions.checkArgument(name.startsWith("blocks/"));
    int colon = name.lastIndexOf(":");
    int originalLength = Integer.valueOf(name.substring(colon + 1));
    int slash = name.lastIndexOf("/");
    String hex = name.substring(slash + 1, colon);
    Signature signature = Signature.fromString(hex);
    return new PersistedBlock(signature, originalLength);
  }
}

@@ -816,65 +816,8 @@
 ng;%0A
-import com.google.common.util.concurrent.FutureCallback;%0A
 impo
@@ -981,65 +981,8 @@
 rs;%0A
-import com.google.common.util.concurrent.SettableFuture;%0A
 impo
@@ -1069,34 +1069,8 @@
 r;%0A%0A
-import java.util.HashMap;%0A
 impo
@@ -1096,30 +1096,8 @@
 or;%0A
-import java.util.Map;%0A
 impo
@@ -1375,106 +1375,8 @@
 ner;
-%0A  private final Map%3CSignature, ListenableFuture%3CPersistedBlock%3E%3E pendingWrites = new HashMap%3C%3E();
 %0A%0A  
@@ -1907,423 +1907,8 @@
 );%0A%0A
-    final SettableFuture%3CPersistedBlock%3E future;%0A    synchronized (lock) %7B%0A      if (Futures.getUnchecked(cloudFileStorage.exists(fileName))) %7B%0A        return Futures.immediateFuture(persisted);%0A      %7D else if (pendingWrites.containsKey(signature)) %7B%0A        return pendingWrites.get(signature);%0A      %7D else %7B%0A        future = SettableFuture.create();%0A        pendingWrites.put(signature, future);%0A      %7D%0A    %7D%0A%0A
     
@@ -1927,22 +1927,17 @@
 ture%3C?%3E 
-storeF
+f
 uture = 
@@ -2060,585 +2060,219 @@
     
-Futures.addCallback(storeFuture, new FutureCallback%3CObject%3E() %7B%0A      @Override public void onSuccess(Object result) %7B%0A        synchronized (lock) %7B%0A          pendingWrites.remove(signature);%0A        %7D%0A        future.set(persisted);%0A      %7D%0A%0A      @Override public void onFailure(Throwable t) %7B%0A        synchronized (lock) %7B%0A          pendingWrites.remove(signature);%0A        %7D%0A        if (t instanceof CloudFileStorage.FileAlreadyExistsException) %7B%0A          future.set(persisted);%0A        %7D else %7B%0A          future.setException(t);%0A        %7D%0A      %7D%0A    %7D);%0A    return future
+ListenableFuture%3CPersistedBlock%3E transformed = Futures.transform(future, (unused) -%3E persisted);%0A    return Futures.catching(transformed, CloudFileStorage.FileAlreadyExistsException.class, (unused) -%3E persisted)
 ;%0A  

fix deadlock in BlockStore by getting rid of pending writes map