1 /*
2  *  Make.org Core API
3  *  Copyright (C) 2018 Make.org
4  *
5  * This program is free software: you can redistribute it and/or modify
6  *  it under the terms of the GNU Affero General Public License as
7  *  published by the Free Software Foundation, either version 3 of the
8  *  License, or (at your option) any later version.
9  *
10  *  This program is distributed in the hope that it will be useful,
11  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
12  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  *  GNU Affero General Public License for more details.
14  *
15  *  You should have received a copy of the GNU Affero General Public License
16  *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
17  *
18  */
19 
20 package org.make.api.technical.crm
21 
22 import java.net.URI
23 import java.nio.file.Path
24 
25 import cats.implicits._
26 import akka.http.scaladsl.Http
27 import akka.http.scaladsl.model._
28 import akka.http.scaladsl.model.headers.{Authorization, BasicHttpCredentials}
29 import akka.http.scaladsl.unmarshalling.PredefinedFromEntityUnmarshallers.stringUnmarshaller
30 import akka.http.scaladsl.unmarshalling.Unmarshal
31 import akka.stream.scaladsl.{FileIO, Flow, Keep, Sink, Source, SourceQueueWithComplete}
32 import akka.stream.{ActorAttributes, OverflowStrategy, QueueOfferResult}
33 import grizzled.slf4j.Logging
34 import de.heikoseeberger.akkahttpcirce.ErrorAccumulatingCirceSupport
35 import io.circe.syntax._
36 import io.circe.Printer
37 import org.make.api.extensions.MailJetConfigurationComponent
38 import org.make.api.technical.ActorSystemComponent
39 import org.make.api.technical.crm.BasicCrmResponse._
40 import org.make.api.technical.crm.CrmClient.{Account, Marketing, Transactional}
41 import org.make.core.Order
42 import org.make.core.technical.Pagination
43 
44 import scala.concurrent.{ExecutionContext, Future, Promise}
45 import scala.util.{Failure, Success, Try}
46 
47 trait DefaultCrmClientComponent extends CrmClientComponent with ErrorAccumulatingCirceSupport with Logging {
48   self: MailJetConfigurationComponent with ActorSystemComponent =>
49 
50   override lazy val crmClient: CrmClient = new DefaultCrmClient
51 
52   class DefaultCrmClient extends CrmClient {
53 
54     private val stringifyJson = Printer.noSpaces.copy(dropNullValues = false).print(_)
55     private lazy val url = new URI(mailJetConfiguration.url)
56     private val httpsPort: Int = 443
57     type Ctx = Promise[HttpResponse]
58 
59     lazy val httpFlow: Flow[(HttpRequest, Ctx), (Try[HttpResponse], Ctx), Http.HostConnectionPool] =
60       Http(actorSystem).cachedHostConnectionPoolHttps[Ctx](host = url.getHost, port = httpsPort)
61 
62     private lazy val bufferSize = mailJetConfiguration.httpBufferSize
63 
64     lazy val queue: SourceQueueWithComplete[(HttpRequest, Promise[HttpResponse])] = Source
65       .queue[(HttpRequest, Promise[HttpResponse])](bufferSize = bufferSize, OverflowStrategy.backpressure)
66       .via(httpFlow)
67       .withAttributes(ActorAttributes.dispatcher("make-api.mail-jet.dispatcher"))
68       .toMat(Sink.foreach {
69         case (Success(resp), p) => p.success(resp)
70         case (Failure(e), p)    => p.failure(e)
71       })(Keep.left)
72       .run()
73 
74     private lazy val campaignApiAuthorization = Authorization(
75       BasicHttpCredentials(mailJetConfiguration.campaignApiKey, mailJetConfiguration.campaignSecretKey)
76     )
77 
78     private lazy val transactionalApiAuthorization = Authorization(
79       BasicHttpCredentials(mailJetConfiguration.apiKey, mailJetConfiguration.secretKey)
80     )
81 
82     private val choseAccount: Account => Authorization = {
83       case Marketing     => campaignApiAuthorization
84       case Transactional => transactionalApiAuthorization
85     }
86 
87     import CrmClientException.RequestException._
88 
89     override def manageContactList(manageContactList: ManageManyContacts, account: Account)(
90       implicit executionContext: ExecutionContext
91     ): Future[ManageManyContactsResponse] = {
92       val request =
93         HttpRequest(
94           method = HttpMethods.POST,
95           uri = Uri(s"${mailJetConfiguration.url}/v3/REST/contact/managemanycontacts"),
96           headers = Seq(choseAccount(account)),
97           entity = HttpEntity(ContentTypes.`application/json`, stringifyJson(manageContactList.asJson))
98         )
99       runWithRetries(request, "manageContactList", ManageContactListException).flatMap(
100         Unmarshal(_).to[ManageManyContactsResponse]
101       )
102     }
103 
104     override def sendCsv(listId: String, csv: Path, account: Account)(
105       implicit executionContext: ExecutionContext
106     ): Future[SendCsvResponse] = {
107       val request = HttpRequest(
108         method = HttpMethods.POST,
109         uri = Uri(s"${mailJetConfiguration.url}/v3/DATA/contactslist/$listId/CSVData/application:octet-stream"),
110         headers = Seq(choseAccount(account)),
111         entity = HttpEntity(contentType = ContentTypes.`application/octet-stream`, data = FileIO.fromPath(csv))
112       )
113       runWithRetries(request, "sendCsv", SendCsvException).flatMap(Unmarshal(_).to[SendCsvResponse])
114     }
115 
116     override def manageContactListWithCsv(csvImport: CsvImport, account: Account)(
117       implicit executionContext: ExecutionContext
118     ): Future[ManageContactsWithCsvResponse] = {
119       val request =
120         HttpRequest(
121           method = HttpMethods.POST,
122           uri = Uri(s"${mailJetConfiguration.url}/v3/REST/csvimport"),
123           headers = Seq(choseAccount(account)),
124           entity = HttpEntity(ContentTypes.`application/json`, stringifyJson(csvImport.asJson))
125         )
126       runWithRetries(request, "manageContactListWithCsv", CsvImportException)
127         .flatMap(Unmarshal(_).to[ManageContactsWithCsvResponse])
128     }
129 
130     override def monitorCsvImport(jobId: Long, account: Account)(
131       implicit executionContext: ExecutionContext
132     ): Future[ManageContactsWithCsvResponse] = {
133       val request =
134         HttpRequest(
135           method = HttpMethods.GET,
136           uri = Uri(s"${mailJetConfiguration.url}/v3/REST/csvimport/$jobId"),
137           headers = Seq(choseAccount(account))
138         )
139       runWithRetries(request, "monitorCsvImport", MonitorCsvImportException)
140         .flatMap(Unmarshal(_).to[ManageContactsWithCsvResponse])
141     }
142 
143     override def sendEmail(message: SendMessages, account: Account)(
144       implicit executionContext: ExecutionContext
145     ): Future[SendEmailResponse] = {
146       val messagesWithErrorHandling = message.copy(messages = message.messages.map(
147         _.copy(templateErrorReporting = Some(
148           TemplateErrorReporting(
149             mailJetConfiguration.errorReportingRecipient,
150             Some(mailJetConfiguration.errorReportingRecipientName)
151           )
152         )
153         )
154       )
155       )
156 
157       val request = HttpRequest(
158         method = HttpMethods.POST,
159         uri = Uri(s"${mailJetConfiguration.url}/v3.1/send"),
160         headers = Seq(choseAccount(account)),
161         entity = HttpEntity(ContentTypes.`application/json`, stringifyJson(messagesWithErrorHandling.asJson))
162       )
163       runWithRetries(request, "sendEmail", SendEmailException).flatMap(Unmarshal(_).to[SendEmailResponse])
164     }
165 
166     override def getUsersInformationMailFromList(
167       listId: Option[String] = None,
168       sort: Option[String] = None,
169       order: Option[Order] = None,
170       countOnly: Option[Boolean] = None,
171       limit: Pagination.Limit,
172       offset: Pagination.Offset = Pagination.Offset(0),
173       account: Account
174     )(implicit executionContext: ExecutionContext): Future[ContactsResponse] = {
175       val sortQuery: Option[String] = (sort, order) match {
176         case (Some(s), Some(o)) => Some(s"$s+$o")
177         case (Some(s), _)       => Some(s)
178         case _                  => None
179       }
180       val params: Map[String, Option[String]] = Map(
181         "ContactsList" -> listId,
182         "Sort" -> sortQuery,
183         "Limit" -> Some(limit.toString),
184         "Offset" -> Some(offset.toString),
185         "countOnly" -> countOnly.map(bool => if (bool) "1" else "0")
186       )
187       val paramsQuery = params.collect { case (k, Some(v)) => s"$k=$v" }.mkString("&")
188       val request = HttpRequest(
189         method = HttpMethods.GET,
190         uri = Uri(s"${mailJetConfiguration.url}/v3/REST/contact?$paramsQuery"),
191         headers = Seq(choseAccount(account))
192       )
193       runWithRetries(request, "getUsersInformationMailFromList", GetUsersInformationMailFromListException)
194         .flatMap(Unmarshal(_).to[ContactsResponse])
195     }
196 
197     override def getContactsProperties(offset: Int, limit: Pagination.Limit, account: Account)(
198       implicit executionContext: ExecutionContext
199     ): Future[GetMailjetContactProperties] = {
200       val paramsQuery = s"Limit=${limit.value.value}&Offset=$offset"
201       val request = HttpRequest(
202         method = HttpMethods.GET,
203         uri = Uri(s"${mailJetConfiguration.url}/v3/REST/contactdata?$paramsQuery"),
204         headers = Seq(choseAccount(account))
205       )
206       runWithRetries(request, "getContactsProperties", GetContactsPropertiesException)
207         .flatMap(Unmarshal(_).to[GetMailjetContactProperties])
208     }
209 
210     private def getContactByMailOrContactId(identifier: String, account: Account)(
211       implicit executionContext: ExecutionContext
212     ): Future[ContactsResponse] = {
213       val request = HttpRequest(
214         method = HttpMethods.GET,
215         uri = Uri(s"${mailJetConfiguration.url}/v3/REST/contact/$identifier"),
216         headers = Seq(choseAccount(account))
217       )
218       runWithRetries(request, "getContactByMailOrContactId", GetUsersMailFromListException)
219         .flatMap(Unmarshal(_).to[ContactsResponse])
220     }
221 
222     override def deleteContactById(contactId: String, account: Account)(
223       implicit executionContext: ExecutionContext
224     ): Future[Unit] = {
225       val request = HttpRequest(
226         method = HttpMethods.DELETE,
227         uri = Uri(s"${mailJetConfiguration.url}/v4/contacts/$contactId"),
228         headers = Seq(choseAccount(account))
229       )
230       runWithRetries(request, "deleteContactById", DeleteUserException).map(_.discardBytes()).void.recoverWith {
231         case error: DeleteUserException if error.code == StatusCodes.NotFound =>
232           logger.error(s"deleteContactById failed with status ${error.code}: ${error.response}")
233           Future.unit
234       }
235     }
236 
237     override def deleteContactByEmail(email: String, account: Account)(
238       implicit executionContext: ExecutionContext
239     ): Future[Unit] = {
240       getContactByMailOrContactId(email, account).flatMap {
241         case BasicCrmResponse(_, _, head :: _) =>
242           val id = head.id.toString
243           deleteContactById(id, account)
244         case _ => Future.unit
245       }.recoverWith {
246         case e: CrmClientException =>
247           logger.error(e.message)
248           Future.unit
249         case e => Future.failed(e)
250       }
251     }
252 
253     private def runWithRetries(
254       request: HttpRequest,
255       id: String,
256       exceptionFn: (StatusCode, String) => CrmClientException.RequestException,
257       maxRetries: Int = 5
258     )(implicit ec: ExecutionContext): Future[ResponseEntity] = {
259 
260       @SuppressWarnings(Array("org.wartremover.warts.Recursion"))
261       def loop(retries: Int): Future[HttpResponse] = {
262         val promise = Promise[HttpResponse]()
263         queue.offer((request, promise)).flatMap {
264           case QueueOfferResult.Enqueued =>
265             promise.future.flatMap {
266               case HttpResponse(StatusCodes.InternalServerError, _, entity, _) if retries > 0 =>
267                 Unmarshal(entity).to[String].flatMap { response =>
268                   logger.warn(s"Error '$response' from Mailjet, $retries retries left")
269                   loop(retries - 1)
270                 }
271               case other => Future.successful(other)
272             }.recoverWith {
273               case error =>
274                 if (retries > 0) {
275                   logger.error(s"Mailjet request $request failed with:", error)
276                   loop(retries - 1)
277                 } else {
278                   Future.failed(error)
279                 }
280             }
281           case QueueOfferResult.Dropped     => Future.failed(CrmClientException.QueueException.QueueOverflowed)
282           case QueueOfferResult.Failure(ex) => Future.failed(ex)
283           case QueueOfferResult.QueueClosed => Future.failed(CrmClientException.QueueException.QueueClosed)
284         }
285       }
286       loop(maxRetries).flatMap({
287         case HttpResponse(code, _, entity, _) if code.isSuccess => Future.successful(entity)
288         case HttpResponse(code, _, entity, _) =>
289           Unmarshal(entity)
290             .to[String]
291             .flatMap(response => {
292               val exception: Exception = code match {
293                 case StatusCodes.TooManyRequests => QuotaExceeded(id, response)
294                 case code                        => exceptionFn(code, response)
295               }
296               Future.failed(exception)
297             })
298       })
299     }
300   }
301 }
302 
303 final case class QuotaExceeded(method: String, message: String) extends Exception(message)
304 
305 sealed abstract class CrmClientException(val message: String) extends Exception(message) with Product with Serializable
306 
307 object CrmClientException {
308 
309   sealed abstract class QueueException(message: String) extends CrmClientException(message)
310 
311   sealed abstract class RequestException(action: String, code: StatusCode, response: String)
312       extends CrmClientException(s"$action failed with status $code: $response")
313 
314   object QueueException {
315 
316     case object QueueOverflowed extends QueueException("Queue overflowed. Try again later.")
317 
318     case object QueueClosed
319         extends QueueException("Queue was closed (pool shut down) while running the request. Try again later.")
320 
321   }
322 
323   object RequestException {
324 
325     final case class ManageContactListException(code: StatusCode, response: String)
326         extends RequestException("manageContactList", code, response)
327 
328     final case class SendCsvException(code: StatusCode, response: String)
329         extends RequestException("send csv", code, response)
330 
331     final case class CsvImportException(code: StatusCode, response: String)
332         extends RequestException("csv import", code, response)
333 
334     final case class MonitorCsvImportException(code: StatusCode, response: String)
335         extends RequestException("monitor csv import", code, response)
336 
337     final case class SendEmailException(code: StatusCode, response: String)
338         extends RequestException("send email", code, response)
339 
340     final case class GetUsersInformationMailFromListException(code: StatusCode, response: String)
341         extends RequestException("getUsersInformationMailFromList", code, response)
342 
343     final case class GetContactsPropertiesException(code: StatusCode, response: String)
344         extends RequestException("getContactsProperties", code, response)
345 
346     final case class ManageContactListJobDetailsException(code: StatusCode, response: String)
347         extends RequestException("manageContactListJobDetails", code, response)
348 
349     final case class GetUsersMailFromListException(code: StatusCode, response: String)
350         extends RequestException("getUsersMailFromList", code, response)
351 
352     final case class DeleteUserException(code: StatusCode, response: String)
353         extends RequestException("deleteContactById", code, response)
354   }
355 }