/****************************************************************
 * Licensed to the Apache Software Foundation (ASF) under one   *
 * or more contributor license agreements.  See the NOTICE file *
 * distributed with this work for additional information        *
 * regarding copyright ownership.  The ASF licenses this file   *
 * to you 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 org.apache.james.jmap.routes

import java.io.InputStream
import java.nio.charset.StandardCharsets
import java.util.stream
import java.util.stream.Stream

import eu.timepit.refined.api.Refined
import eu.timepit.refined.auto._
import eu.timepit.refined.numeric.NonNegative
import eu.timepit.refined.refineV
import io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE
import io.netty.handler.codec.http.HttpResponseStatus.{BAD_REQUEST, CREATED, FORBIDDEN, INTERNAL_SERVER_ERROR, UNAUTHORIZED}
import io.netty.handler.codec.http.{HttpMethod, HttpResponseStatus}
import javax.inject.{Inject, Named}
import org.apache.commons.fileupload.util.LimitedInputStream
import org.apache.james.jmap.HttpConstants.JSON_CONTENT_TYPE
import org.apache.james.jmap.core.Id.Id
import org.apache.james.jmap.core.{AccountId, Id, JmapRfc8621Configuration, ProblemDetails}
import org.apache.james.jmap.exceptions.UnauthorizedException
import org.apache.james.jmap.http.Authenticator
import org.apache.james.jmap.http.rfc8621.InjectionKeys
import org.apache.james.jmap.json.{ResponseSerializer, UploadSerializer}
import org.apache.james.jmap.mail.BlobId
import org.apache.james.jmap.mail.Email.Size
import org.apache.james.jmap.routes.UploadRoutes.{LOGGER, sanitizeSize}
import org.apache.james.jmap.{Endpoint, JMAPRoute, JMAPRoutes}
import org.apache.james.mailbox.model.{AttachmentMetadata, ContentType}
import org.apache.james.mailbox.{AttachmentManager, MailboxSession}
import org.apache.james.util.ReactorUtils
import org.slf4j.{Logger, LoggerFactory}
import reactor.core.publisher.Mono
import reactor.core.scala.publisher.SMono
import reactor.core.scheduler.Schedulers
import reactor.netty.http.server.{HttpServerRequest, HttpServerResponse}

case class TooBigUploadException() extends RuntimeException

object UploadRoutes {
  val LOGGER: Logger = LoggerFactory.getLogger(classOf[DownloadRoutes])

  type Size = Long Refined NonNegative
  val Zero: Size = 0L

  def sanitizeSize(value: Long): Size = {
    val size: Either[String, Size] = refineV[NonNegative](value)
    size.fold(e => {
      LOGGER.error(s"Encountered an invalid upload files size: $e")
      Zero
    },
      refinedValue => refinedValue)
  }
}

case class UploadResponse(accountId: AccountId,
                          blobId: BlobId,
                          `type`: ContentType,
                          size: Size)

class UploadRoutes @Inject()(@Named(InjectionKeys.RFC_8621) val authenticator: Authenticator,
                             val configuration: JmapRfc8621Configuration,
                             val attachmentManager: AttachmentManager,
                             val serializer: UploadSerializer) extends JMAPRoutes {

  class CancelledUploadException extends RuntimeException
  class MaxFileSizeUploadException extends RuntimeException

  private val accountIdParam: String = "accountId"
  private val uploadURI = s"/upload/{$accountIdParam}/"

  override def routes(): stream.Stream[JMAPRoute] = Stream.of(
    JMAPRoute.builder
      .endpoint(new Endpoint(HttpMethod.POST, uploadURI))
      .action(this.post)
      .corsHeaders,
    JMAPRoute.builder
      .endpoint(new Endpoint(HttpMethod.OPTIONS, uploadURI))
      .action(JMAPRoutes.CORS_CONTROL)
      .noCorsHeaders)

  def post(request: HttpServerRequest, response: HttpServerResponse): Mono[Void] = {
    request.requestHeaders.get(CONTENT_TYPE) match {
      case contentType => SMono.fromPublisher(
        authenticator.authenticate(request))
        .flatMap(session => post(request, response, ContentType.of(contentType), session))
        .onErrorResume {
          case e: UnauthorizedException =>
            LOGGER.warn("Unauthorized", e)
            respondDetails(response,
              ProblemDetails(status = UNAUTHORIZED, detail = e.getMessage),
              UNAUTHORIZED)
          case _: TooBigUploadException =>
            respondDetails(response,
              ProblemDetails(status = BAD_REQUEST, detail = "Attempt to upload exceed max size"),
              BAD_REQUEST)
          case _: ForbiddenException =>
            respondDetails(response,
              ProblemDetails(status = FORBIDDEN, detail = "Upload to other accounts is forbidden"),
              FORBIDDEN)
          case e =>
            LOGGER.error("Unexpected error upon uploads", e)
            respondDetails(response,
              ProblemDetails(status = INTERNAL_SERVER_ERROR, detail = e.getMessage),
              INTERNAL_SERVER_ERROR)
        }
        .asJava()
        .subscribeOn(Schedulers.elastic())
        .`then`()
      case _ => response.status(BAD_REQUEST).send
    }
  }

  def post(request: HttpServerRequest, response: HttpServerResponse, contentType: ContentType, session: MailboxSession): SMono[Void] = {
    Id.validate(request.param(accountIdParam)) match {
      case Right(id: Id) =>
        val targetAccountId: AccountId = AccountId(id)
        AccountId.from(session.getUser).map(accountId => accountId.equals(targetAccountId))
          .fold[SMono[Void]](
            e => SMono.raiseError(e),
            value => if (value) {
              SMono.fromCallable(() => ReactorUtils.toInputStream(request.receive.asByteBuffer))
              .flatMap(content => handle(targetAccountId, contentType, content, session, response))
            } else {
              SMono.raiseError(ForbiddenException())
            })

      case Left(throwable: Throwable) => SMono.raiseError(throwable)
    }
  }

  def handle(accountId: AccountId, contentType: ContentType, content: InputStream, mailboxSession: MailboxSession, response: HttpServerResponse): SMono[Void] = {
    val maxSize: Long = configuration.maxUploadSize.value.value

    SMono.fromCallable(() => new LimitedInputStream(content, maxSize) {
      override def raiseError(max: Long, count: Long): Unit = if (count > max) {
        throw TooBigUploadException()
      }})
      .flatMap(uploadContent(accountId, contentType, _, mailboxSession))
      .flatMap(uploadResponse => SMono.fromPublisher(response
              .header(CONTENT_TYPE, uploadResponse.`type`.asString())
              .status(CREATED)
              .sendString(SMono.just(serializer.serialize(uploadResponse).toString()))))
  }

  def uploadContent(accountId: AccountId, contentType: ContentType, inputStream: InputStream, session: MailboxSession): SMono[UploadResponse] =
    SMono
      .fromPublisher(attachmentManager.storeAttachment(contentType, inputStream, session))
      .map(fromAttachment(_, accountId))

  private def fromAttachment(attachmentMetadata: AttachmentMetadata, accountId: AccountId): UploadResponse =
    UploadResponse(
        blobId = BlobId.of(attachmentMetadata.getAttachmentId.getId).get,
        `type` = ContentType.of(attachmentMetadata.getType.asString),
        size = sanitizeSize(attachmentMetadata.getSize),
        accountId = accountId)

  private def respondDetails(httpServerResponse: HttpServerResponse, details: ProblemDetails, statusCode: HttpResponseStatus = BAD_REQUEST): SMono[Void] =
    SMono.fromPublisher(httpServerResponse.status(statusCode)
      .header(CONTENT_TYPE, JSON_CONTENT_TYPE)
      .sendString(SMono.fromCallable(() => ResponseSerializer.serialize(details).toString), StandardCharsets.UTF_8)
      .`then`)
}
