package org.kuali.common.devops.archive.s3;
   
   import static com.amazonaws.services.s3.Headers.CONTENT_TYPE;
   import static com.google.common.collect.Lists.newArrayList;
   import static com.google.common.collect.Maps.newHashMap;
   import static com.google.common.collect.Sets.difference;
   import static java.lang.String.format;
   import static org.kuali.common.core.io.Files.md5;
   import static org.kuali.common.core.io.Paths.fromFile;
  import static org.kuali.common.util.log.Loggers.newLogger;
  
  import java.io.File;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.Callable;
  
  import org.kuali.common.aws.s3.S3Service;
  import org.kuali.common.aws.s3.model.ObjectMetadata;
  import org.kuali.common.aws.s3.model.PutFileRequest;
  import org.kuali.common.aws.s3.model.PutObjectResult;
  import org.kuali.common.core.build.ValidatingBuilder;
  import org.kuali.common.core.validate.annotation.IdiotProofImmutable;
  import org.kuali.common.devops.archive.ArchivedFileMetadata;
  import org.kuali.common.devops.archive.sas.MetadataProvider;
  import org.slf4j.Logger;
  
  import com.google.common.base.Optional;
  import com.google.common.collect.ImmutableList;
  
  @IdiotProofImmutable
  public final class PutS3FileCallable implements Callable<List<PutS3FileCallableResult>> {
  
  	private static final Logger logger = newLogger();
  
  	private static final String EMPTY_MD5_HASH = "68b329da9893e34099c7d8ad5cb9c940";
  
  	private final String bucket;
  	private final S3Service s3;
  	private final ImmutableList<KeyFile> pairs;
  	private final MetadataProvider provider;
  
  	@Override
  	public List<PutS3FileCallableResult> call() {
  		List<PutS3FileCallableResult> list = newArrayList();
  		for (KeyFile pair : pairs) {
  			boolean skip = isSkip(pair);
  			if (!skip) {
  				PutFileRequest request = getPutFileRequest(pair, bucket);
  				logger.debug(format("create -> %s", pair.getKey()));
  				PutObjectResult result = s3.putFile(request);
  				PutS3FileCallableResult element = PutS3FileCallableResult.builder().withPair(pair).withResult(result).build();
  				list.add(element);
  			} else {
  				logger.debug(format("skip   -> %s", pair.getKey()));
  			}
  		}
  		return ImmutableList.copyOf(list);
  	}
  
  	private boolean isSkip(KeyFile pair) {
  
  		// Check to see if there is metadata for this key
  		Optional<ObjectMetadata> optional = s3.getOptionalMetadata(bucket, pair.getKey());
  
  		// if there is no metadata there is no archived file
  		if (!optional.isPresent()) {
  			return false;
  		}
  
  		// extract the metadata
  		ObjectMetadata meta = optional.get();
  
  		// make sure all of the required metadata keys are present
  		Set<String> missingKeys = difference(MetadataConverter.KEYS, meta.getUserMetadata().keySet());
  
  		if (!missingKeys.isEmpty()) {
  			return false;
  		}
  
  		// If the file we are uploading is a directory and all of the metadata is there, we are good to go
  		if (pair.getFile().isDirectory()) {
  			return true;
  		}
  
  		// compare the metadata S3 is storing for this key with metadata calculated from the file on the file system
  		ArchivedFileMetadata afd = MetadataConverter.INSTANCE.reverse().convert(meta.getUserMetadata());
  
  		// if the sizes are different we must re-upload
  		if (afd.getSize() != pair.getFile().length()) {
  			return false;
  		}
  
  		// Both last modified and size are identical, it's ok to skip uploading this file
  		if (afd.getLastModified() == pair.getFile().lastModified()) {
  			return true;
  		}
  
  		// length is the same, but last modified is different
 		// don't skip unless the md5 checksums match
 		return md5(pair.getFile()).equals(afd.getMd5());
 	}
 
 	private PutFileRequest getPutFileRequest(KeyFile pair, String bucket) {
 		File file = pair.getFile();
 		long size = file.isDirectory() ? 0L : file.length();
 		long lastModified = file.lastModified();
 		String md5 = file.isDirectory() ? EMPTY_MD5_HASH : md5(file);
 		ArchivedFileMetadata meta = ArchivedFileMetadata.builder().withLastModified(lastModified).withMd5(md5).withSize(size).build();
 		Map<String, String> userMetadata = newHashMap(MetadataConverter.INSTANCE.convert(meta));
 		userMetadata.putAll(provider.getUserMetadata(fromFile(pair.getFile())));
 		Map<String, Object> raw = newHashMap();
 		// TODO Provide this somewhere else, maybe make a MetadataProviderProvider, that chains together one or more providers
 		if (pair.getFile().getName().equals("log")) {
 			raw.put(CONTENT_TYPE, "text/plain");
 		}
 		if (pair.getFile().getName().endsWith(".log")) {
 			raw.put(CONTENT_TYPE, "text/plain");
 		}
 		if (pair.getFile().getName().endsWith(".out")) {
 			raw.put(CONTENT_TYPE, "text/plain");
 		}
 		ObjectMetadata omd = ObjectMetadata.builder().withUserMetadata(userMetadata).withRawMetadata(raw).build();
 		return PutFileRequest.builder().withFile(file).withBucket(bucket).withKey(pair.getKey()).withMetadata(omd).build();
 	}
 
 	private PutS3FileCallable(Builder builder) {
 		this.s3 = builder.s3;
 		this.pairs = ImmutableList.copyOf(builder.pairs);
 		this.bucket = builder.bucket;
 		this.provider = builder.provider;
 	}
 
 	public static Builder builder() {
 		return new Builder();
 	}
 
 	public static class Builder extends ValidatingBuilder<PutS3FileCallable> {
 
 		private String bucket;
 		private S3Service s3;
 		private List<KeyFile> pairs;
 		private MetadataProvider provider = EmptyMetadataProvider.INSTANCE;
 
 		public Builder withS3(S3Service s3) {
 			this.s3 = s3;
 			return this;
 		}
 
 		public Builder withProvider(MetadataProvider provider) {
 			this.provider = provider;
 			return this;
 		}
 
 		public Builder withPairs(List<KeyFile> pairs) {
 			this.pairs = pairs;
 			return this;
 		}
 
 		public Builder withBucket(String bucket) {
 			this.bucket = bucket;
 			return this;
 		}
 
 		@Override
 		public PutS3FileCallable build() {
 			return validate(new PutS3FileCallable(this));
 		}
 	}
 
 	public S3Service getS3() {
 		return s3;
 	}
 
 	public List<KeyFile> getPairs() {
 		return pairs;
 	}
 
 	public String getBucket() {
 		return bucket;
 	}
 
 	public MetadataProvider getProvider() {
 		return provider;
 	}
 
 }