/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.underfs.gcs.v2;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.PositionReader;
import alluxio.conf.PropertyKey;
import alluxio.exception.runtime.UnimplementedRuntimeException;
import alluxio.retry.RetryPolicy;
import alluxio.underfs.ChecksumType;
import alluxio.underfs.ObjectUnderFileSystem;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.UnderFileSystemConfiguration;
import alluxio.underfs.options.OpenOptions;
import alluxio.util.ModeUtils;
import alluxio.util.UnderFileSystemUtils;
import alluxio.util.io.PathUtils;

import com.google.api.client.util.Base64;
import com.google.api.gax.paging.Page;
import com.google.api.gax.retrying.RetrySettings;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.storage.Blob;
import com.google.cloud.storage.BlobId;
import com.google.cloud.storage.BlobInfo;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageException;
import com.google.cloud.storage.StorageOptions;
import com.google.common.collect.Lists;
import org.apache.commons.codec.digest.DigestUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.threeten.bp.Duration;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import javax.annotation.concurrent.ThreadSafe;

/**
 * GCS FS {@link UnderFileSystem} implementation based on the Google cloud storage library.
 */
@ThreadSafe
public class GCSV2UnderFileSystem extends ObjectUnderFileSystem {
  // TODO(lu) StorageException has isRetryable() method, can help handle retry
  private static final Logger LOG = LoggerFactory.getLogger(GCSV2UnderFileSystem.class);

  /** Static hash for a directory's empty contents. */
  private static final String DIR_HASH = Base64.encodeBase64String(DigestUtils.md5(new byte[0]));

  /** Google cloud storage client. */
  private final Storage mStorageClient;

  /** Bucket name of user's configured Alluxio bucket. */
  private final String mBucketName;

  /**
   * Constructs a new instance of {@link GCSV2UnderFileSystem}.
   *
   * @param uri the {@link AlluxioURI} for this UFS
   * @param conf the configuration for this UFS
   * @return the created {@link GCSV2UnderFileSystem} instance
   */
  public static GCSV2UnderFileSystem createInstance(AlluxioURI uri,
      UnderFileSystemConfiguration conf) throws IOException {
    String bucketName = UnderFileSystemUtils.getBucketName(uri);
    GoogleCredentials credentials;
    if (conf.isSet(PropertyKey.GCS_CREDENTIAL_PATH)) {
      String credsPath = conf.getString(PropertyKey.GCS_CREDENTIAL_PATH);
      credentials = GoogleCredentials
          .fromStream(new FileInputStream(credsPath))
          .createScoped(Lists.newArrayList("https://www.googleapis.com/auth/cloud-platform"));
      LOG.info("Created GCSV2UnderFileSystem with credentials in {}", credsPath);
    } else {
      // The environment variable GOOGLE_APPLICATION_CREDENTIALS is set
      // or the application is running in Google App engine or compute engine
      credentials = GoogleCredentials.getApplicationDefault();
      LOG.info("Created GCSV2UnderFileSystem with default Google application credentials");
    }
    Storage storage = StorageOptions.newBuilder().setRetrySettings(
        RetrySettings.newBuilder()
            .setInitialRetryDelay(
                Duration.ofMillis(conf.getMs(PropertyKey.UNDERFS_GCS_RETRY_INITIAL_DELAY_MS)))
            .setMaxRetryDelay(
                Duration.ofMillis(conf.getMs(PropertyKey.UNDERFS_GCS_RETRY_MAX_DELAY_MS)))
            .setRetryDelayMultiplier(
                conf.getInt(PropertyKey.UNDERFS_GCS_RETRY_DELAY_MULTIPLIER))
            .setMaxAttempts(conf.getInt(PropertyKey.UNDERFS_GCS_RETRY_MAX))
            .setTotalTimeout(
                Duration.ofMillis(conf.getMs(PropertyKey.UNDERFS_GCS_RETRY_TOTAL_DURATION_MS)))
            .setJittered(conf.getBoolean(PropertyKey.UNDERFS_GCS_RETRY_JITTER))
            .build())
        .setCredentials(credentials).build().getService();
    return new GCSV2UnderFileSystem(uri, storage, bucketName, conf);
  }

  /**
   * Constructor for {@link GCSV2UnderFileSystem}.
   *
   * @param uri the {@link AlluxioURI} for this UFS
   * @param storageClient the Google cloud storage client
   * @param bucketName bucket name of user's configured Alluxio bucket
   * @param conf configuration for this UFS
   */
  protected GCSV2UnderFileSystem(AlluxioURI uri, Storage storageClient,
      String bucketName, UnderFileSystemConfiguration conf) {
    super(uri, conf);
    mStorageClient = storageClient;
    mBucketName = bucketName;
  }

  @Override
  public String getUnderFSType() {
    return "gcs";
  }

  @Override
  public PositionReader openPositionRead(String path, long fileLength) {
    throw new UnimplementedRuntimeException("Position read is not implemented");
  }

  // Setting GCS owner via Alluxio is not supported yet. This is a no-op.
  @Override
  public void setOwner(String path, String user, String group) {}

  // Setting GCS mode via Alluxio is not supported yet. This is a no-op.
  @Override
  public void setMode(String path, short mode) {}

  @Override
  protected boolean copyObject(String src, String dst) {
    LOG.debug("Copying {} to {}", src, dst);
    Storage.CopyRequest request = Storage.CopyRequest.newBuilder()
        .setSource(BlobId.of(mBucketName, src))
        .setTarget(BlobId.of(mBucketName, dst))
        .build();
    try {
      Blob blob = mStorageClient.copy(request).getResult();
      if (blob != null) {
        return true;
      }
    } catch (StorageException e) {
      LOG.error("Failed to copy file {} to {}", src, dst, e);
    }
    return false;
  }

  @Override
  public boolean createEmptyObject(String key) {
    try {
      BlobId blobId = BlobId.of(mBucketName, key);
      BlobInfo blobInfo = BlobInfo.newBuilder(blobId).build();
      Blob blob = mStorageClient.create(blobInfo);
      if (blob == null) {
        LOG.error("Failed to create object {}", key);
        return false;
      }
    } catch (StorageException e) {
      LOG.error("Failed to create object: {}", key, e);
      return false;
    }
    return true;
  }

  @Override
  protected OutputStream createObject(String key) throws IOException {
    return new GCSV2OutputStream(mBucketName, key, mStorageClient);
  }

  @Override
  protected boolean deleteObject(String key) {
    // TODO(lu) Google cloud API supports bunch delete
    BlobId blobId = BlobId.of(mBucketName, key);
    try {
      if (!mStorageClient.delete(blobId)) {
        LOG.error("Failed to delete object {}", key);
        return false;
      }
    } catch (StorageException e) {
      LOG.error("Failed to delete {}", key, e);
      return false;
    }
    return true;
  }

  @Override
  protected String getFolderSuffix() {
    return mUfsConf.getString(PropertyKey.UNDERFS_GCS_DIRECTORY_SUFFIX);
  }

  @Override
  protected ObjectListingChunk getObjectListingChunk(String key, boolean recursive)
      throws IOException {
    key = PathUtils.normalizePath(key, PATH_SEPARATOR);
    // In case key is root (empty string) do not normalize prefix
    key = key.equals(PATH_SEPARATOR) ? "" : key;
    Page<Blob> blobPage;
    try {
      if (recursive) {
        blobPage = mStorageClient.list(mBucketName, Storage.BlobListOption.prefix(key));
      } else {
        blobPage = mStorageClient.list(mBucketName, Storage.BlobListOption.prefix(key),
            Storage.BlobListOption.currentDirectory());
      }
    } catch (StorageException e) {
      LOG.error("Failed to get object listing result of {}", key, e);
      throw new IOException(e);
    }
    if (blobPage != null && blobPage.getValues().iterator().hasNext()) {
      return new GCSObjectListingChunk(blobPage);
    }
    return null;
  }

  /**
   * Wrapper over GCS.
   */
  private final class GCSObjectListingChunk implements ObjectListingChunk {
    final Page<Blob> mBlobPage;

    /**
     * Creates an instance of {@link GCSObjectListingChunk}.
     *
     * @param blobPage blob page
     */
    GCSObjectListingChunk(Page<Blob> blobPage) {
      mBlobPage = blobPage;
    }

    @Override
    public ObjectStatus[] getObjectStatuses() {
      Iterator<Blob> blobs = mBlobPage.getValues().iterator();
      List<Blob> blobList = new ArrayList<>();
      while (blobs.hasNext()) {
        blobList.add(blobs.next());
      }
      ObjectStatus[] res = new ObjectStatus[blobList.size()];
      for (int i = 0; i < res.length; i++) {
        Blob blob = blobList.get(i);
        res[i] = getBlobStatus(blob);
      }
      return res;
    }

    @Override
    public String[] getCommonPrefixes() {
      return new String[0];
    }

    @Override
    public ObjectListingChunk getNextChunk() {
      if (mBlobPage.hasNextPage()) {
        return new GCSObjectListingChunk(mBlobPage.getNextPage());
      }
      return null;
    }
  }

  @Override
  protected ObjectStatus getObjectStatus(String key) throws IOException {
    try {
      BlobId info = BlobId.of(mBucketName, key);
      Blob blob = mStorageClient.get(info);
      if (blob == null) {
        // file not found, possible for exists calls
        return null;
      }
      return getBlobStatus(blob);
    } catch (StorageException e) {
      if (e.getCode() == 404) { // file not found, possible for exists calls
        return null;
      }
      throw new IOException(String
          .format("Failed to get object status of %s, %s", key, mBucketName), e);
    }
  }

  @Override
  protected ObjectPermissions getPermissions() {
    // TODO(lu) inherit acl
    return new ObjectPermissions("", "",
        ModeUtils.getUMask(mUfsConf.getString(PropertyKey.UNDERFS_GCS_DEFAULT_MODE)).toShort());
  }

  @Override
  protected String getRootKey() {
    return Constants.HEADER_GCS + mBucketName;
  }

  @Override
  protected InputStream openObject(String key, OpenOptions options,
      RetryPolicy retryPolicy) throws IOException {
    return new GCSV2InputStream(mBucketName, key, mStorageClient, options.getOffset());
  }

  /**
   * Gets the blob status.
   *
   * @param blob the blob to get status from
   * @return the blob status
   */
  private ObjectStatus getBlobStatus(Blob blob) {
    Long time = blob.getUpdateTime() != null ? blob.getUpdateTime()
        : blob.getCreateTime() != null ? blob.getCreateTime() : null;
    String checkSum;
    if (mUfsConf.getEnum(PropertyKey.UNDERFS_CHECKSUM_TYPE, ChecksumType.class)
                .equals(ChecksumType.MD5)) {
      checkSum = blob.getMd5() == null ? DIR_HASH : blob.getMd5();
    }
    else {
      checkSum = blob.getCrc32c() == null ? DIR_HASH : blob.getCrc32c();
    }
    return new ObjectStatus(blob.getName(), checkSum, blob.getSize(), time);
  }
}
