package com.twitter.finagle

import com.twitter.finagle.client.{DefaultPool, StackClient, StdStackClient, Transporter}
import com.twitter.finagle.mysql._
import com.twitter.finagle.mysql.param._
import com.twitter.finagle.mysql.transport.Packet
import com.twitter.finagle.param.{
  Monitor => ParamMonitor,
  ExceptionStatsHandler => _,
  ResponseClassifier => _,
  Tracer => _,
  _
}
import com.twitter.finagle.service.{ResponseClassifier, RetryBudget}
import com.twitter.finagle.stats.{ExceptionStatsHandler, NullStatsReceiver, StatsReceiver}
import com.twitter.finagle.tracing.Tracer
import com.twitter.finagle.transport.{Transport, TransportContext}
import com.twitter.util.{Duration, FuturePool, Monitor}
import java.net.SocketAddress
import java.util.concurrent.ExecutorService

/**
 * Supplements a [[com.twitter.finagle.Client]] with convenient
 * builder methods for constructing a mysql client.
 */
trait MysqlRichClient { self: com.twitter.finagle.Client[Request, Result] =>

  /**
   * Whether the client supports unsigned integer fields
   */
  protected val supportUnsigned: Boolean

  def richClientStatsReceiver: StatsReceiver = NullStatsReceiver

  /**
   * Creates a new `RichClient` connected to the logical
   * destination described by `dest` with the assigned
   * `label`. The `label` is used to scope client stats.
   */
  def newRichClient(dest: Name, label: String): mysql.Client with mysql.Transactions =
    mysql.Client(newClient(dest, label), richClientStatsReceiver, supportUnsigned)

  /**
   * Creates a new `RichClient` connected to the logical
   * destination described by `dest`.
   *
   * @param dest the location to connect to, e.g. "host:port". See the
   *             [[https://twitter.github.io/finagle/guide/Names.html user guide]]
   *             for details on destination names.
   */
  def newRichClient(dest: String): mysql.Client with mysql.Transactions =
    mysql.Client(newClient(dest), richClientStatsReceiver, supportUnsigned)
}

/**
 * @example {{{
 * val client = Mysql.client
 *   .withCredentials("<username>", "<password>")
 *   .withDatabase("<db>")
 *   .newRichClient("inet!localhost:3306")
 * }}}
 */
object Mysql extends com.twitter.finagle.Client[Request, Result] with MysqlRichClient {

  protected val supportUnsigned: Boolean = UnsignedColumns.param.default.supported

  object Client {

    private val params: Stack.Params = StackClient.defaultParams +
      ProtocolLibrary("mysql") +
      DefaultPool.Param(
        low = 0,
        high = 1,
        bufferSize = 0,
        idleTime = Duration.Top,
        maxWaiters = Int.MaxValue
      ) +
      ParamMonitor(ServerErrorMonitor(Seq()))

    private val stack: Stack[ServiceFactory[Request, Result]] = StackClient.newStack
      .prepend(RollbackFactory.module)
      .replace(StackClient.Role.protoTracing, MysqlTracingFilter.module)
      .insertAfter(DefaultPool.Role, PoisonConnection.module)
      .insertAfter(StackClient.Role.prepConn, ConnectionInitSql.module)
  }

  /**
   * Implements a mysql client in terms of a
   * [[com.twitter.finagle.client.StackClient]]. The client inherits a wealth
   * of features from finagle including connection pooling and load
   * balancing.
   *
   * Additionally, this class provides methods via [[MysqlRichClient]] for constructing
   * a client which exposes an API that has use case specific methods, for example
   * [[mysql.Client.read]], [[mysql.Client.modify]], and [[mysql.Client.prepare]].
   * This is an easier experience for most users.
   *
   * @example
   * {{{
   * import com.twitter.finagle.Mysql
   * import com.twitter.finagle.mysql.Client
   * import com.twitter.util.Future
   *
   * val client: Client = Mysql.client
   *   .withCredentials("username", "password")
   *   .withDatabase("database")
   *   .newRichClient("host:port")
   * val names: Future[Seq[String]] =
   *   client.select("SELECT name FROM employee") { row =>
   *     row.stringOrNull("name")
   *   }
   * }}}
   */
  case class Client(
    stack: Stack[ServiceFactory[Request, Result]] = Client.stack,
    params: Stack.Params = Client.params)
      extends StdStackClient[Request, Result, Client]
      with WithSessionPool[Client]
      with WithDefaultLoadBalancer[Client]
      with MysqlRichClient {

    protected val supportUnsigned: Boolean = params[UnsignedColumns].supported

    protected def copy1(
      stack: Stack[ServiceFactory[Request, Result]] = this.stack,
      params: Stack.Params = this.params
    ): Client = copy(stack, params)

    protected type In = Packet
    protected type Out = Packet
    protected type Context = TransportContext

    protected def newTransporter(addr: SocketAddress): Transporter[In, Out, Context] =
      new MysqlTransporter(addr, params)

    protected def newDispatcher(
      transport: Transport[In, Out] { type Context <: Client.this.Context }
    ): Service[Request, Result] =
      mysql.ClientDispatcher(transport, params)

    /**
     * The maximum number of concurrent prepare statements.
     */
    def withMaxConcurrentPrepareStatements(num: Int): Client =
      configured(MaxConcurrentPrepareStatements(num))

    /**
     * The credentials to use when authenticating a new session.
     *
     * @param p if `null`, no password is used.
     */
    def withCredentials(u: String, p: String): Client =
      configured(Credentials(Option(u), Option(p)))

    /**
     * Database to use when this client establishes a new session.
     */
    def withDatabase(db: String): Client =
      configured(Database(Option(db)))

    /**
     * The default character set used when establishing a new session.
     */
    def withCharset(charset: Short): Client =
      configured(Charset(charset))

    /**
     * Don't set the CLIENT_FOUND_ROWS flag when establishing a new
     * session. This will make "INSERT ... ON DUPLICATE KEY UPDATE"
     * statements return the "correct" update count.
     *
     * See https://dev.mysql.com/doc/refman/5.7/en/information-functions.html#function_row-count
     */
    def withAffectedRows(): Client =
      configured(FoundRows(false))

    /**
     * Installs a module on the client which issues a ROLLBACK statement when a service is put
     * back into the pool. This exists to ensure that a "clean" connection is always returned
     * from the connection pool. For example, it prevents situations where an unfinished
     * transaction has been written to the wire, the service has been released back into
     * the pool, the same service is again checked out of the pool, and a statement
     * that causes an implicit commit is issued.
     *
     * The additional work incurred for the rollback may result in less throughput from the
     * connection pool and, as such, may require configuring the pool (via `withSessionPool`)
     * to offer more available connections connections.
     *
     * @see [[com.twitter.finagle.mysql.RollbackFactory]]
     * @see https://dev.mysql.com/doc/en/implicit-commit.html
     * @note this module is installed by default.
     */
    def withRollback: Client =
      if (stack.contains(RollbackFactory.Role)) {
        this
      } else {
        withStack(stack.prepend(RollbackFactory.module))
      }

    /**
     * Removes the module on the client which issues a ROLLBACK statement each time a
     * service is put back into the pool. This ''may'' result in better performance
     * at the risk of receiving a connection from the pool with uncommitted state.
     *
     * Instead of disabling this feature, consider configuring the connection pool
     * for the client (via `withSessionPool`) to offer more available connections.
     *
     * @see [[com.twitter.finagle.mysql.RollbackFactory]]
     * @see https://dev.mysql.com/doc/en/implicit-commit.html
     * @note the rollback module is installed by default.
     */
    def withNoRollback: Client =
      withStack(stack.remove(RollbackFactory.Role))

    /**
     * The connection init request to use when establishing a new session.
     */
    def withConnectionInitRequest(request: Request): Client =
      configured(ConnectionInitRequest(Some(request)))

    // Java-friendly forwarders
    // See https://issues.scala-lang.org/browse/SI-8905
    override val withSessionPool: SessionPoolingParams[Client] =
      new SessionPoolingParams(this)
    override val withLoadBalancer: DefaultLoadBalancingParams[Client] =
      new DefaultLoadBalancingParams(this)
    override val withTransport: ClientTransportParams[Client] =
      new ClientTransportParams(this)
    override val withSession: ClientSessionParams[Client] =
      new ClientSessionParams(this)
    override val withSessionQualifier: SessionQualificationParams[Client] =
      new SessionQualificationParams(this)
    override val withAdmissionControl: ClientAdmissionControlParams[Client] =
      new ClientAdmissionControlParams(this)

    override def withLabel(label: String): Client = super.withLabel(label)
    override def withStatsReceiver(statsReceiver: StatsReceiver): Client =
      super.withStatsReceiver(statsReceiver)
    override def withMonitor(monitor: Monitor): Client = super.withMonitor(monitor)
    override def withTracer(tracer: Tracer): Client = super.withTracer(tracer)
    override def withExceptionStatsHandler(exceptionStatsHandler: ExceptionStatsHandler): Client =
      super.withExceptionStatsHandler(exceptionStatsHandler)
    override def withRequestTimeout(timeout: Duration): Client =
      super.withRequestTimeout(timeout)
    override def withResponseClassifier(responseClassifier: ResponseClassifier): Client =
      super.withResponseClassifier(responseClassifier)
    override def withRetryBudget(budget: RetryBudget): Client = super.withRetryBudget(budget)
    override def withRetryBackoff(backoff: Stream[Duration]): Client =
      super.withRetryBackoff(backoff)

    override def withStack(stack: Stack[ServiceFactory[Request, Result]]): Client =
      super.withStack(stack)
    override def withStack(
      fn: Stack[ServiceFactory[Request, Result]] => Stack[ServiceFactory[Request, Result]]
    ): Client =
      super.withStack(fn)
    override def withExecutionOffloaded(executor: ExecutorService): Client =
      super.withExecutionOffloaded(executor)
    override def withExecutionOffloaded(pool: FuturePool): Client =
      super.withExecutionOffloaded(pool)
    override def configured[P](psp: (P, Stack.Param[P])): Client = super.configured(psp)
    override def filtered(filter: Filter[Request, Result, Request, Result]): Client =
      super.filtered(filter)

    override def richClientStatsReceiver: StatsReceiver = params[Stats].statsReceiver
  }

  def client: Mysql.Client = Client()

  def newClient(dest: Name, label: String): ServiceFactory[Request, Result] =
    client.newClient(dest, label)

  def newService(dest: Name, label: String): Service[Request, Result] =
    client.newService(dest, label)
}
