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 akka.actor.typed.Behavior
23 import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
24 import grizzled.slf4j.Logging
25 import org.make.api.technical.ActorProtocol
26 import org.make.api.technical.crm.BasicCrmResponse.ManageContactsWithCsvResponse
27 import org.make.api.technical.crm.CrmSynchroCsvMonitor.Protocol.Command
28 
29 import scala.concurrent.duration.{DurationInt, FiniteDuration}
30 import scala.concurrent.{ExecutionContext, Future, Promise}
31 import scala.util.{Failure, Success}
32 
33 object CrmSynchroCsvMonitor extends Logging {
34 
35   @SuppressWarnings(Array("org.wartremover.warts.Recursion"))
36   def apply(crmClient: CrmClient, promise: Promise[Unit], tickInterval: FiniteDuration)(
37     jobIds: Seq[Long]
38   )(implicit executionContext: ExecutionContext): Behavior[Protocol.Command] = {
39 
40     def running(jobIds: Seq[Long], pendingCalls: Seq[Long]): Behavior[Protocol.Command] = {
41       Behaviors.setup { implicit context =>
42         Behaviors.receiveMessage {
43           case Command.Tick           => handleTick(jobIds, pendingCalls)
44           case Command.QuotaAvailable => Behaviors.same
45           case Command.CrmCallSucceeded(jobId, response) =>
46             handleSuccessfulResponse(jobId, response)(jobIds, pendingCalls)(running)
47           case Command.CrmCallFailed(jobId, e) =>
48             handleErrorResponse(jobId, e)(jobIds, pendingCalls)(running)
49         }
50       }
51     }
52 
53     def blocked(jobIds: Seq[Long], pendingCalls: Seq[Long]): Behavior[Protocol.Command] = {
54       Behaviors.withTimers { timers =>
55         timers.startSingleTimer(Command.QuotaAvailable, 1.hour)
56         Behaviors.receiveMessage {
57           case Command.Tick => Behaviors.same
58           case Command.QuotaAvailable =>
59             logger.info("Job checker becomes available again")
60             running(jobIds, pendingCalls)
61           case Command.CrmCallSucceeded(jobId, response) =>
62             handleSuccessfulResponse(jobId, response)(jobIds, pendingCalls)(blocked)
63           case Command.CrmCallFailed(jobId, e) =>
64             handleErrorResponse(jobId, e)(jobIds, pendingCalls)(blocked)
65         }
66       }
67     }
68 
69     def handleTick(jobIds: Seq[Long], pendingCalls: Seq[Long])(
70       implicit context: ActorContext[Command]
71     ): Behavior[Command] = {
72       (jobIds, pendingCalls) match {
73         case (Seq(), Seq()) =>
74           promise.success {}
75           Behaviors.stopped
76         case (Seq(), _) =>
77           Behaviors.same
78         case (toMonitor +: tail, _) =>
79           val futureCall: Future[ManageContactsWithCsvResponse] = crmClient.monitorCsvImport(toMonitor)
80           context.pipeToSelf(futureCall) {
81             case Success(response) => Command.CrmCallSucceeded(toMonitor, response)
82             case Failure(e)        => Command.CrmCallFailed(toMonitor, e)
83           }
84           running(tail, pendingCalls :+ toMonitor)
85       }
86 
87     }
88 
89     def handleErrorResponse(jobId: Long, error: Throwable)(jobIds: Seq[Long], pendingCalls: Seq[Long])(
90       next: (Seq[Long], Seq[Long]) => Behavior[Command]
91     ): Behavior[Command] =
92       error match {
93         case QuotaExceeded(_, message) =>
94           logger.warn(s"Job checker is becoming blocked for an hour, received message is $message")
95           blocked(jobIds :+ jobId, pendingCalls.filterNot(_ == jobId))
96         case other =>
97           logger.error(s"Error when checking status for job $jobId", other)
98           next(jobIds :+ jobId, pendingCalls.filterNot(_ == jobId))
99       }
100 
101     def handleSuccessfulResponse(jobId: Long, response: BasicCrmResponse.ManageContactsWithCsvResponse)(
102       jobIds: Seq[Long],
103       pendingCalls: Seq[Long]
104     )(next: (Seq[Long], Seq[Long]) => Behavior[Command]): Behavior[Command] =
105       if (response.data.forall(response => response.status == "Completed" || response.status == "Abort")) {
106         logger.info(s"Job $jobId completed successfully: $response")
107         if (response.data.forall(response => response.errorCount > 0)) {
108           logger.error(
109             s"Job $jobId has errors: $response, " +
110               s"file should be at https://api.mailjet.com/v3/DATA/Batchjob/$jobId/CSVError/text:csv"
111           )
112         }
113         next(jobIds, pendingCalls.filterNot(_ == jobId))
114       } else {
115         next(jobIds :+ jobId, pendingCalls.filterNot(_ == jobId))
116       }
117 
118     Behaviors.withTimers { timers =>
119       timers.startTimerWithFixedDelay(Command.Tick, tickInterval)
120       running(jobIds, Seq.empty)
121     }
122   }
123 
124   sealed abstract class Protocol extends ActorProtocol
125 
126   object Protocol {
127 
128     sealed abstract class Command extends Protocol
129 
130     object Command {
131       case object Tick extends Command
132       case object QuotaAvailable extends Command
133       final case class CrmCallSucceeded(jobId: Long, response: ManageContactsWithCsvResponse) extends Command
134       final case class CrmCallFailed(jobId: Long, error: Throwable) extends Command
135     }
136   }
137 
138 }