/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * license agreements; and to You under the Apache License, version 2.0:
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * This file is part of the Apache Pekko project, which was derived from Akka.
 */

/*
 * Copyright (C) since 2016 Lightbend Inc. <https://www.lightbend.com>
 */

package org.apache.pekko.stream.connectors.googlecloud.storage.scaladsl

import org.apache.pekko
import pekko.http.scaladsl.model.ContentType
import pekko.stream.{ Attributes, Materializer }
import pekko.stream.connectors.googlecloud.storage.impl.GCStorageStream
import pekko.stream.connectors.googlecloud.storage.{ Bucket, StorageObject }
import pekko.stream.scaladsl.{ RunnableGraph, Sink, Source }
import pekko.util.ByteString
import pekko.{ Done, NotUsed }

import scala.concurrent.Future

/**
 * Factory of Google Cloud Storage operations.
 */
object GCStorage {

  /**
   * Gets information on a bucket
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/buckets/get
   *
   * @param bucketName the name of the bucket to look up
   * @return a `Future` containing `Bucket` if it exists
   */
  def getBucket(bucketName: String)(implicit mat: Materializer,
      attr: Attributes = Attributes()): Future[Option[Bucket]] =
    GCStorageStream.getBucket(bucketName)

  /**
   * Gets information on a bucket
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/buckets/get
   *
   * @param bucketName the name of the bucket to look up
   * @return a `Source` containing `Bucket` if it exists
   */
  def getBucketSource(bucketName: String): Source[Option[Bucket], NotUsed] =
    GCStorageStream.getBucketSource(bucketName)

  /**
   * Creates a new bucket
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/buckets/insert
   *
   * @param bucketName the name of the bucket
   * @param location the region to put the bucket in
   * @return a `Future` of `Bucket` with created bucket
   */
  def createBucket(bucketName: String, location: String)(implicit mat: Materializer,
      attr: Attributes = Attributes()): Future[Bucket] =
    GCStorageStream.createBucket(bucketName, location)

  /**
   * Creates a new bucket
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/buckets/insert
   *
   * @param bucketName the name of the bucket
   * @param location the region to put the bucket in
   * @return a `Source` of `Bucket` with created bucket
   */
  def createBucketSource(bucketName: String, location: String): Source[Bucket, NotUsed] =
    GCStorageStream.createBucketSource(bucketName, location)

  /**
   * Deletes bucket
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/buckets/delete
   *
   * @param bucketName the name of the bucket
   * @return a `Future` of `Done` on successful deletion
   */
  def deleteBucket(bucketName: String)(implicit mat: Materializer, attr: Attributes = Attributes()): Future[Done] =
    GCStorageStream.deleteBucket(bucketName)

  /**
   * Deletes bucket
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/buckets/delete
   *
   * @param bucketName the name of the bucket
   * @return a `Source` of `Done` on successful deletion
   */
  def deleteBucketSource(bucketName: String): Source[Done, NotUsed] =
    GCStorageStream.deleteBucketSource(bucketName)

  /**
   * Get storage object
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/objects/get
   *
   * @param bucket the name of the bucket
   * @param objectName the name of the object
   * @return a `Source` containing `StorageObject` if it exists
   */
  def getObject(bucket: String, objectName: String): Source[Option[StorageObject], NotUsed] =
    GCStorageStream.getObject(bucket, objectName)

  /**
   * Get storage object
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/objects/get
   *
   * @param bucket the name of the bucket
   * @param objectName the name of the object
   * @param generation the generation of the object
   * @return a `Source` containing `StorageObject` if it exists
   */
  def getObject(bucket: String, objectName: String, generation: Option[Long]): Source[Option[StorageObject], NotUsed] =
    GCStorageStream.getObject(bucket, objectName, generation)

  /**
   * Deletes object in bucket
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/objects/delete
   *
   * @param bucketName the name of the bucket
   * @param objectName the name of the object
   * @return a `Source` of `Boolean` with `true` if object is deleted, `false` if object that we want to deleted doesn't exist
   */
  def deleteObject(bucketName: String, objectName: String): Source[Boolean, NotUsed] =
    GCStorageStream.deleteObjectSource(bucketName, objectName)

  /**
   * Deletes object in bucket
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/objects/delete
   *
   * @param bucketName the name of the bucket
   * @param objectName the name of the object
   * @param generation the generation of the object
   * @return a `Source` of `Boolean` with `true` if object is deleted, `false` if object that we want to deleted doesn't exist
   */
  def deleteObject(bucketName: String, objectName: String, generation: Option[Long]): Source[Boolean, NotUsed] =
    GCStorageStream.deleteObjectSource(bucketName, objectName, generation)

  /**
   * Lists the bucket contents
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/objects/list
   *
   * @param bucket the bucket name
   * @param prefix the bucket prefix
   * @return a `Source` of `StorageObject`
   */
  def listBucket(bucket: String, prefix: Option[String]): Source[StorageObject, NotUsed] =
    GCStorageStream.listBucket(bucket, prefix)

  /**
   * Lists the bucket contents
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/objects/list
   *
   * @param bucket the bucket name
   * @param prefix the bucket prefix
   * @param versions `true` to list both live and archived bucket contents
   * @return a `Source` of `StorageObject`
   */
  def listBucket(bucket: String, prefix: Option[String], versions: Boolean): Source[StorageObject, NotUsed] =
    GCStorageStream.listBucket(bucket, prefix, versions)

  /**
   * Downloads object from bucket.
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/objects/get
   *
   * @param bucket the bucket name
   * @param objectName the bucket prefix
   * @return  The source will emit an empty [[scala.Option Option]] if an object can not be found.
   *         Otherwise [[scala.Option Option]] will contain a source of object's data.
   */
  def download(bucket: String, objectName: String): Source[Option[Source[ByteString, NotUsed]], NotUsed] =
    GCStorageStream.download(bucket, objectName)

  /**
   * Downloads object from bucket.
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/objects/get
   *
   * @param bucket the bucket name
   * @param objectName the bucket prefix
   * @param generation the generation of the object
   * @return  The source will emit an empty [[scala.Option Option]] if an object can not be found.
   *         Otherwise [[scala.Option Option]] will contain a source of object's data.
   */
  def download(bucket: String,
      objectName: String,
      generation: Option[Long]): Source[Option[Source[ByteString, NotUsed]], NotUsed] =
    GCStorageStream.download(bucket, objectName, generation)

  /**
   * Uploads object, use this for small files and `resumableUpload` for big ones
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/how-tos/simple-upload
   *
   * @param bucket the bucket name
   * @param objectName the object name
   * @param data a `Source` of `ByteString`
   * @param contentType  the number of bytes that will be uploaded (required!)
   * @return a `Source` containing the `StorageObject` of the uploaded object
   */
  def simpleUpload(bucket: String,
      objectName: String,
      data: Source[ByteString, ?],
      contentType: ContentType): Source[StorageObject, NotUsed] =
    GCStorageStream.putObject(bucket, objectName, data, contentType)

  /**
   * Uploads object by making multiple requests
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/how-tos/resumable-upload
   *
   * @param bucket the bucket name
   * @param objectName the object name
   * @param contentType `ContentType`
   * @param chunkSize the size of the request sent to google cloud storage in bytes, must be a multiple of 256KB
   * @param metadata custom metadata for the object
   * @return a `Sink` that accepts `ByteString`'s and materializes to a `Future` of `StorageObject`
   */
  def resumableUpload(bucket: String,
      objectName: String,
      contentType: ContentType,
      chunkSize: Int,
      metadata: Map[String, String]): Sink[ByteString, Future[StorageObject]] =
    resumableUpload(bucket, objectName, contentType, chunkSize, Some(metadata))

  /**
   * Uploads object by making multiple requests
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/how-tos/resumable-upload
   *
   * @param bucket the bucket name
   * @param objectName the object name
   * @param contentType `ContentType`
   * @param chunkSize the size of the request sent to google cloud storage in bytes, must be a multiple of 256KB
   * @return a `Sink` that accepts `ByteString`'s and materializes to a `Future` of `StorageObject`
   */
  def resumableUpload(bucket: String,
      objectName: String,
      contentType: ContentType,
      chunkSize: Int): Sink[ByteString, Future[StorageObject]] =
    resumableUpload(bucket, objectName, contentType, chunkSize, metadata = None)

  private def resumableUpload(bucket: String,
      objectName: String,
      contentType: ContentType,
      chunkSize: Int,
      metadata: Option[Map[String, String]]): Sink[ByteString, Future[StorageObject]] = {
    assert(
      (chunkSize >= (256 * 1024)) && (chunkSize % (256 * 1024) == 0),
      "Chunk size must be a multiple of 256KB")
    GCStorageStream.resumableUpload(bucket, objectName, contentType, chunkSize, metadata)
  }

  /**
   * Uploads object by making multiple requests with default chunk size of 5MB
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/how-tos/resumable-upload
   *
   * @param bucket the bucket name
   * @param objectName the object name
   * @param contentType `ContentType`
   * @return a `Sink` that accepts `ByteString`'s and materializes to a `scala.concurrent.Future Future` of `StorageObject`
   */
  def resumableUpload(bucket: String,
      objectName: String,
      contentType: ContentType): Sink[ByteString, Future[StorageObject]] =
    GCStorageStream.resumableUpload(bucket, objectName, contentType)

  /**
   * Rewrites object to wanted destination by making multiple requests.
   *
   * @see https://cloud.google.com/storage/docs/json_api/v1/objects/rewrite
   *
   * @param sourceBucket the source bucket
   * @param sourceObjectName the source object name
   * @param destinationBucket the destination bucket
   * @param destinationObjectName the destination bucket name
   * @return a runnable graph which upon materialization will return a [[scala.concurrent.Future Future ]] containing the `StorageObject` with info about rewritten file
   */
  def rewrite(sourceBucket: String,
      sourceObjectName: String,
      destinationBucket: String,
      destinationObjectName: String): RunnableGraph[Future[StorageObject]] =
    GCStorageStream.rewrite(sourceBucket, sourceObjectName, destinationBucket, destinationObjectName)

  /**
   * Deletes folder and its content.
   *
   * @param bucket the bucket name
   * @param prefix the object prefix
   * @return a `Source` of `Boolean` with all `true` if everything is deleted
   */
  def deleteObjectsByPrefix(bucket: String, prefix: Option[String]): Source[Boolean, NotUsed] =
    GCStorageStream.deleteObjectsByPrefixSource(bucket, prefix)
}
