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.webflow
21 
22 import akka.http.scaladsl.Http
23 import akka.http.scaladsl.model._
24 import akka.http.scaladsl.model.headers._
25 import akka.http.scaladsl.unmarshalling.PredefinedFromEntityUnmarshallers.stringUnmarshaller
26 import akka.http.scaladsl.unmarshalling.Unmarshal
27 import akka.stream.scaladsl.{Flow, Keep, Sink, Source, SourceQueueWithComplete}
28 import akka.stream.{ActorAttributes, OverflowStrategy, QueueOfferResult}
29 import cats.implicits._
30 import de.heikoseeberger.akkahttpcirce.ErrorAccumulatingCirceSupport
31 import enumeratum.{Enum, EnumEntry}
32 import eu.timepit.refined.api.Refined
33 import eu.timepit.refined.numeric._
34 import grizzled.slf4j.Logging
35 import io.circe.generic.semiauto.deriveDecoder
36 import io.circe.{Decoder, Printer}
37 import org.make.api.technical.ActorSystemComponent
38 import org.make.api.technical.webflow.WebflowClient.UpToOneHundred
39 import org.make.api.technical.webflow.WebflowError.{WebflowErrorName, WebflowErrorResponse}
40 import org.make.api.technical.webflow.WebflowPost.{WebflowImageRef, WebflowPostWithCountry}
41 
42 import java.time.ZonedDateTime
43 import scala.concurrent.ExecutionContext.Implicits.global
44 import scala.concurrent.duration.DurationInt
45 import scala.concurrent.{Future, Promise}
46 import scala.util.{Failure, Success, Try}
47 
48 trait WebflowClient {
49   def getAllPosts(limit: UpToOneHundred, offset: Int): Future[Seq[WebflowPostWithCountry]]
50 }
51 
52 object WebflowClient {
53   type UpToOneHundredRefinement = Interval.Closed[0, 100]
54   type UpToOneHundred = Int Refined UpToOneHundredRefinement
55 }
56 
57 trait WebflowClientComponent {
58   def webflowClient: WebflowClient
59 }
60 
61 trait DefaultWebflowClientComponent extends WebflowClientComponent with ErrorAccumulatingCirceSupport with Logging {
62   self: WebflowConfigurationComponent with ActorSystemComponent =>
63 
64   override lazy val webflowClient: WebflowClient = new DefaultWebflowClient
65 
66   class DefaultWebflowClient extends WebflowClient {
67 
68     lazy val printer: Printer = Printer.noSpaces.copy(dropNullValues = false)
69     val httpPort: Int = 443
70 
71     lazy val httpFlow: Flow[
72       (HttpRequest, Promise[HttpResponse]),
73       (Try[HttpResponse], Promise[HttpResponse]),
74       Http.HostConnectionPool
75     ] =
76       Http(actorSystem).cachedHostConnectionPoolHttps[Promise[HttpResponse]](
77         host = webflowConfiguration.apiUrl.getHost,
78         port = httpPort
79       )
80 
81     private lazy val bufferSize = webflowConfiguration.httpBufferSize
82 
83     lazy val queue: SourceQueueWithComplete[(HttpRequest, Promise[HttpResponse])] = Source
84       .queue[(HttpRequest, Promise[HttpResponse])](bufferSize = bufferSize, OverflowStrategy.backpressure)
85       .throttle(webflowConfiguration.rateLimitPerMinute, 1.minute)
86       .via(httpFlow)
87       .withAttributes(ActorAttributes.dispatcher("make-api.webflow.dispatcher"))
88       .toMat(Sink.foreach {
89         case (Success(resp), p) => p.success(resp)
90         case (Failure(e), p)    => p.failure(e)
91       })(Keep.left)
92       .run()
93 
94     private lazy val authorization: HttpHeader = Authorization(OAuth2BearerToken(webflowConfiguration.token))
95 
96     private val postsCollectionId = webflowConfiguration.collectionsIds
97 
98     private val defaultHeaders = Seq(authorization)
99 
100     override def getAllPosts(limit: UpToOneHundred, offset: Int): Future[Seq[WebflowPostWithCountry]] = {
101       postsCollectionId.toSeq.flatTraverse {
102         case (country, collectionId) =>
103           getCollection(collectionId).flatMap { collection =>
104             getPosts(limit, offset, collectionId, country, collection.slug)
105           }
106       }
107     }
108 
109     private def getCollection(collectionId: String): Future[WebflowCollection] = {
110       val request = HttpRequest(
111         method = HttpMethods.GET,
112         uri = Uri(s"${webflowConfiguration.apiUrl.toString}/collections/$collectionId"),
113         headers = defaultHeaders
114       )
115       doHttpCall(request).flatMap {
116         case HttpResponse(code, headers, entity, _) if code.isFailure() =>
117           handleErrors(code, headers, entity, "getCollection")
118         case HttpResponse(code, _, entity, _) if code.isSuccess() =>
119           Unmarshal(entity.withoutSizeLimit).to[WebflowCollection]
120         case HttpResponse(code, _, entity, _) =>
121           Unmarshal(entity).to[String].flatMap { response =>
122             Future.failed(
123               new IllegalArgumentException(
124                 s"Unexpected response from Webflow with status code $code: $response. Caused by request ${request.toString}"
125               )
126             )
127           }
128       }
129     }
130 
131     private def getPosts(
132       limit: UpToOneHundred,
133       offset: Int,
134       collectionId: String,
135       country: String,
136       collectionSlug: String
137     ): Future[Seq[WebflowPostWithCountry]] = {
138       val paramsQuery = s"limit=${limit.value}&offset=$offset"
139       val request = HttpRequest(
140         method = HttpMethods.GET,
141         uri = Uri(s"${webflowConfiguration.apiUrl.toString}/collections/$collectionId/items?$paramsQuery"),
142         headers = defaultHeaders
143       )
144       logger.debug(s"WebflowRequest: ${request.toString.replace('\n', ' ')}")
145       doHttpCall(request).flatMap {
146         case HttpResponse(code, headers, entity, _) if code.isFailure() =>
147           handleErrors(code, headers, entity, "getItemsFromCollection [posts]")
148         case HttpResponse(code, _, entity, _) if code.isSuccess() =>
149           Unmarshal(entity.withoutSizeLimit)
150             .to[WebflowItems]
151             .map(_.items.map(_.toWebflowPostWithCountry(country, collectionSlug)))
152         case HttpResponse(code, _, entity, _) =>
153           Unmarshal(entity).to[String].flatMap { response =>
154             Future.failed(
155               new IllegalArgumentException(
156                 s"Unexpected response from Webflow with status code $code: $response. Caused by request ${request.toString}"
157               )
158             )
159           }
160       }
161     }
162 
163     private def handleErrors[T](
164       code: StatusCode,
165       headers: Seq[HttpHeader],
166       entity: ResponseEntity,
167       action: String
168     ): Future[T] = {
169       Unmarshal(entity).to[WebflowErrorResponse].flatMap { errorResponse =>
170         errorResponse.code match {
171           case WebflowErrorName.RateLimit =>
172             val maxLimit: String = headers.find(_.is("x-ratelimit-limit")).map(_.value()).getOrElse("unknown")
173             val remaining: String = headers.find(_.is("x-ratelimit-remaining")).map(_.value()).getOrElse("unknown")
174             logger.error(
175               s"Webflow error $code ${errorResponse.code}: ${errorResponse.message}. Max limit per minute: $maxLimit. Remaining: $remaining"
176             )
177             Future.failed(WebflowClientException.RateLimitException(maxLimit, errorResponse.toString))
178           case _ =>
179             logger.error(
180               s"Webflow error $code ${errorResponse.code}: ${errorResponse.message}. Here is some details: ${errorResponse.details
181                 .mkString(", ")}"
182             )
183             Future.failed(WebflowClientException.RequestException(action, code, errorResponse.toString))
184         }
185       }
186     }
187 
188     private def doHttpCall(request: HttpRequest): Future[HttpResponse] = {
189       val promise = Promise[HttpResponse]()
190       queue.offer((request, promise)).flatMap {
191         case QueueOfferResult.Enqueued    => promise.future
192         case QueueOfferResult.Dropped     => Future.failed(WebflowClientException.QueueException.QueueOverflowed)
193         case QueueOfferResult.Failure(ex) => Future.failed(ex)
194         case QueueOfferResult.QueueClosed => Future.failed(WebflowClientException.QueueException.QueueClosed)
195       }
196     }
197   }
198 }
199 
200 object WebflowError {
201 
202   sealed trait WebflowErrorName extends EnumEntry
203 
204   object WebflowErrorName extends Enum[WebflowErrorName] {
205     case object SyntaxError extends WebflowErrorName
206 
207     case object InvalidAPIVersion extends WebflowErrorName
208 
209     case object UnsupportedVersion extends WebflowErrorName
210 
211     case object NotImplemented extends WebflowErrorName
212 
213     case object ValidationError extends WebflowErrorName
214 
215     case object Conflict extends WebflowErrorName
216 
217     case object Unauthorized extends WebflowErrorName
218 
219     case object NotFound extends WebflowErrorName
220 
221     case object RateLimit extends WebflowErrorName
222 
223     case object ServerError extends WebflowErrorName
224 
225     case object UnknownError extends WebflowErrorName
226 
227     override val values: IndexedSeq[WebflowErrorName] = findValues
228     implicit val decoder: Decoder[WebflowErrorName] =
229       Decoder[String].map(withNameInsensitiveOption(_).getOrElse(UnknownError))
230   }
231 
232   final case class WebflowErrorResponse(
233     message: String,
234     code: WebflowErrorName,
235     externalReference: Option[String],
236     details: Seq[String]
237   )
238 
239   object WebflowErrorResponse {
240     implicit val decoder: Decoder[WebflowErrorResponse] = deriveDecoder[WebflowErrorResponse]
241   }
242 }
243 
244 sealed abstract class WebflowClientException(val message: String) extends Exception(message)
245 
246 object WebflowClientException {
247 
248   sealed abstract class QueueException(message: String) extends WebflowClientException(message)
249 
250   object QueueException {
251     case object QueueOverflowed extends QueueException("Queue overflowed. Try again later.")
252 
253     case object QueueClosed
254         extends QueueException("Queue was closed (pool shut down) while running the request. Try again later.")
255   }
256 
257   final case class RequestException(action: String, code: StatusCode, response: String)
258       extends WebflowClientException(s"$action failed with status $code: $response")
259 
260   final case class RateLimitException(maxLimit: String, response: String)
261       extends WebflowClientException(s"Rate limit reached ($maxLimit per minute): $response")
262 }
263 
264 final case class WebflowCollection(slug: String)
265 
266 object WebflowCollection {
267   implicit def decoder: Decoder[WebflowCollection] = deriveDecoder[WebflowCollection]
268 }
269 
270 final case class WebflowItems(items: Seq[WebflowPost], pagination: WebflowPagination)
271 
272 object WebflowItems {
273   implicit def createDecoder: Decoder[WebflowItems] =
274     Decoder.forProduct2("items", "pagination")(WebflowItems.apply)
275 }
276 
277 final case class WebflowPagination(limit: Int, offset: Int, total: Int)
278 
279 object WebflowPagination {
280   implicit val decoder: Decoder[WebflowPagination] = deriveDecoder[WebflowPagination]
281 }
282 
283 final case class WebflowPostData(
284   name: String,
285   slug: String,
286   displayHome: Option[Boolean],
287   postDate: Option[ZonedDateTime],
288   thumbnailImage: Option[WebflowImageRef],
289   summary: Option[String]
290 )
291 
292 object WebflowPostData {
293   implicit val decoder: Decoder[WebflowPostData] = Decoder.forProduct6(
294     "name",
295     "slug",
296     "afficher-cet-article-sur-le-site-de-make-org",
297     "post-date",
298     "thumbnail-image",
299     "post-summary"
300   )(WebflowPostData.apply)
301 }
302 
303 final case class WebflowPost(
304   id: String,
305   lastUpdated: ZonedDateTime,
306   createdOn: ZonedDateTime,
307   fieldData: WebflowPostData,
308   cmsLocaleId: Option[String],
309   isArchived: Boolean,
310   isDraft: Boolean
311 ) {
312   def toWebflowPostWithCountry(country: String, collectionSlug: String): WebflowPostWithCountry =
313     WebflowPostWithCountry(
314       id,
315       isArchived,
316       isDraft,
317       fieldData.name,
318       fieldData.slug,
319       fieldData.displayHome,
320       fieldData.postDate,
321       fieldData.thumbnailImage,
322       fieldData.summary,
323       country,
324       collectionSlug
325     )
326 }
327 
328 object WebflowPost {
329   implicit val decoder: Decoder[WebflowPost] = deriveDecoder[WebflowPost]
330 
331   final case class WebflowImageRef(url: String, alt: Option[String])
332 
333   object WebflowImageRef {
334     implicit val webflowImageRefDecoder: Decoder[WebflowImageRef] = deriveDecoder[WebflowImageRef]
335   }
336 
337   final case class WebflowPostWithCountry(
338     id: String,
339     archived: Boolean,
340     draft: Boolean,
341     name: String,
342     slug: String,
343     displayHome: Option[Boolean],
344     postDate: Option[ZonedDateTime],
345     thumbnailImage: Option[WebflowImageRef],
346     summary: Option[String],
347     country: String,
348     collectionSlug: String
349   )
350 }