package com.beecavegames.persistence.aws;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import javax.inject.Inject;
import javax.inject.Named;

import org.apache.commons.configuration.Configuration;

import lombok.extern.slf4j.Slf4j;

import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3Client;
import com.amazonaws.services.s3.model.AmazonS3Exception;
import com.amazonaws.services.s3.model.ListObjectsRequest;
import com.amazonaws.services.s3.model.ObjectListing;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.PutObjectRequest;
import com.amazonaws.services.s3.model.PutObjectResult;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import com.amazonaws.services.s3.model.StorageClass;
import com.beecavegames.concurrency.LabeledCallable;
import com.beecavegames.persistence.AbstractStore;
import com.beecavegames.persistence.StoreException;
import com.beecavegames.persistence.TableScan;
import com.beecavegames.util.BCGObjectMapper;
import com.google.common.base.Function;

@Slf4j
public class S3Store extends AbstractStore {
	
	AmazonS3 service;
	private String bucketName;
	private ExecutorService executor;
	private BCGObjectMapper mapper;
	
	@Inject
	public S3Store(Configuration config, @Named("s3Bucket") String bucketName, AWSCredentials creds, ExecutorService executor) {
		super(config);
		service = new AmazonS3Client(creds);
		mapper = new BCGObjectMapper();
		this.bucketName=bucketName;
		this.executor = executor;
	}
	
	@Override
	/**
	 * In S3Store, this is interpreted as "find me all keys that are lexicographically between
	 * the range keys".  The primaryKey is treated as a "must contain" string
	 */
	public <T extends Serializable> Collection<T> scan(String table, Class<T> type,
			T primaryKey, Comparable<?> rangeFrom, Comparable<?> rangeTo) {
		String from = rangeFrom.toString();
		String to = rangeTo.toString();
		ListObjectsRequest req = new ListObjectsRequest()
			.withBucketName(bucketName).withMarker(getKey(table, from, null))
			.withPrefix(getKey(table, commonPrefix(from,to), null));

		Collection<String> keysToFetch = new ArrayList<>();
		S3ObjectSummary last=null;
		while (last == null || last.getKey().compareTo(to)<0) {
			ObjectListing result = service.listObjects(req);
			for (S3ObjectSummary item: result.getObjectSummaries()) {
				if (item.getKey().compareTo(from) >= 0 && item.getKey().compareTo(to)<0 && item.getKey().contains(getId(primaryKey).toString())) {
					keysToFetch.add(item.getKey());
				}
				last=item;
			}
			req.setMarker(last.getKey());
		}
		if (!keysToFetch.isEmpty()) {
			return new ArrayList<>(loadAll(type, keysToFetch).values());
		} else {
			return Collections.emptyList();
		}
	}

	private static String commonPrefix(String s1, String s2) {
		int ml = Math.min(s1.length(), s2.length());
		for (int i=0; i<ml; i++) {
			if (s1.charAt(i) != s2.charAt(i)) {
				return s1.substring(0, i);
			}
		}
		return s1.substring(0, ml);
	}

	@Override
	public <T extends Serializable> Future<Collection<T>> scanAsync(String table, 
			Class<T> type, T primaryKey, Comparable<?> rangeFrom,
			Comparable<?> rangeTo) {
		throw new UnsupportedOperationException();
	}

	@Override
	public <T extends Serializable> Future<Boolean> saveAsync(final T entity, final int flags) {
		return executor.submit(new LabeledCallable<Boolean>("s3SaveAsync") {

			@Override
			public Boolean call() throws Exception {
				return save(entity, flags);
			}			
		});
	}

	@Override
	public <T extends Serializable> Future<Boolean> saveAllAsync(
			final Iterable<T> entities, int flags) {
		return executor.submit(new LabeledCallable<Boolean>("s3SaveAllAsync") {

			@Override
			public Boolean call() throws Exception {
				List<Future<Boolean>> futures = new ArrayList<>();
				for (T entity: entities) {
					futures.add(saveAsync(entity));
				}
				boolean result = true;
				for (Future<Boolean> f: futures) {
					result &= f.get();
				}
				return result;
			}
			
		});
	}

	@Override
	public <T extends Serializable> Future<T> loadAsync(final String table,
			final Class<T> type, final Object key, final int flags) {
		return executor.submit(new LabeledCallable<T>("s3LoadAsync") {

			@Override
			public T call() throws Exception {
				return load(table, type, key, flags);
			}			
		});
	}

	@Override
	public <K,T extends Serializable> Map<K, Future<T>> loadAllAsync(
			String table, Class<T> type, Iterable<K> keys, int flags) {
		Map<K, Future<T>> rv = new HashMap<>();
		for (K key: keys) {
			rv.put(key, loadAsync(table, type, key, flags));
		}
		return rv;
	}
	
	@Override
	public <T extends Serializable> boolean save(T entity, int flags) {
		ObjectMetadata m=new ObjectMetadata();
		m.setContentType("application/json");
		m.setContentEncoding("gzip");
		try {
			ByteArrayOutputStream out = new ByteArrayOutputStream();
			GZIPOutputStream dout = new GZIPOutputStream(out);
			mapper.writeValue(dout, entity);
			m.setContentLength(out.size());
			
			PutObjectRequest req = new PutObjectRequest(bucketName, getKey(entity).toString(), 
					new ByteArrayInputStream(out.toByteArray()), m);
			if ((flags & REDUCED_REDUNDANCY) != 0) 
				req.setStorageClass(StorageClass.ReducedRedundancy);
			PutObjectResult rv = service.putObject(req);
			return rv != null;
		} catch (IOException e) {
			log.warn("Exception in save", e);
		} catch (AmazonS3Exception ce) {
			log.warn("Error saving S3 object", ce);
		}
		return false;
	}

	@Override
	public <T extends Serializable> boolean saveAll(Iterable<T> entities) {
		try {
			return saveAllAsync(entities).get();
		} catch (InterruptedException | ExecutionException e) {
			log.warn("Exception in saveAll", e);
		}
		return false;
	}

	@Override
	public <T extends Serializable> T load(String table, Class<T> type, Object key, int flags) {
		try {
			S3Object obj = service.getObject(bucketName, getKey(table, key, null));
			return mapper.readValue(new GZIPInputStream(obj.getObjectContent()), type);
		} catch (IOException e) {
			throw new StoreException("Error pulling from S3" , e);
		} catch (AmazonS3Exception ce) {
		}
		return null;
	}

	@Override
	public <K,T extends Serializable> Map<K, T> loadAll(String table, Class<T> type,
			Iterable<K> keys, int flags) {
		Map<K, Future<T>> result = loadAllAsync(table, type, keys, flags);
		Map<K, T> rv = new HashMap<>(result.size());
		for (Entry<K, Future<T>> entry: result.entrySet()) {
			try {
				rv.put(entry.getKey(), entry.getValue().get());
			} catch (InterruptedException | ExecutionException e) {
				log.warn("Exception in loadAll", e);
			}
		}
		return rv;
	}

	@Override
	public void delete(String table, Class<? extends Serializable> type, Object key) {
		service.deleteObject(bucketName, getKey(table, key, null));
	}

	@Override
	public void shutdown() {
		log.info("Shutting down executor service");
		executor.shutdown();
	}
	
	public Object getKey(Serializable entity) {
		return getKeyForEntity(entity);
	}

	public static Object getKeyForEntity(Serializable entity) {
		return getKey(getTable(entity), getId(entity), getRangeValue(entity));
	}
	
	protected static String getKey(String table, Object key, Object rangeKey) {
		StringBuilder b = new StringBuilder(table);
		b.append('/');
		b.append(key);
		if (rangeKey != null) {
			b.append('/');
			b.append(rangeKey);
		}
		return b.toString();
	}

	@Override
	public <T extends Serializable> TableScan<T> startScan(String table, Class<T> class1,
			int entriesPerPage) {
		throw new UnsupportedOperationException();
	}
	
	@Override
	public boolean exists(String table, Class<? extends Serializable> type, Object key) {
		try {
			return service.getObjectMetadata(bucketName, getKey(table, key, null)) != null;
		} catch (AmazonS3Exception s3e) {
			if (s3e.getStatusCode() != 404) {
				log.warn("Error in S3 exists", s3e);
			}
			return false;
		}
	}

	@Override
	public boolean exists(String table, Class<? extends Serializable> type, Object hashKey, Comparable<?> rangeKey) {
		throw new UnsupportedOperationException();
	}

	@Override
	public void delete(String table, Class<? extends Serializable> type, Object hashKey, Comparable<?> rangeKey) {
		throw new UnsupportedOperationException();
	}

	@Override
	public <T extends Serializable> T loadRanged(String table, Class<T> type, Object key, Comparable<?> rangeKey,
			int flags) {
		throw new UnsupportedOperationException();
	}

	@Override
	public int getFlagMask() {
		return REDUCED_REDUNDANCY;
	}

	@Override
	public int scanForExpiredKeys(Class<? extends Serializable> type,
			String typeName,
			Function<Object, Void> callback, int limit) {
		throw new UnsupportedOperationException();
	}

	@Override
	public <T extends Serializable> boolean save(String table, T entity, int flags) {
		throw new UnsupportedOperationException();
	}
}
