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.sessionhistory
21 
22 import akka.actor.typed.ActorRef
23 import akka.stream.scaladsl.{Sink, Source}
24 import akka.util.Timeout
25 import org.make.api.extensions.MakeSettingsComponent
26 import org.make.api.sessionhistory.SessionHistoryResponse.{Error, LockResponse}
27 import org.make.api.sessionhistory.SessionHistoryResponse.Error.ExpiredSession
28 import org.make.api.technical.BetterLoggingActors.BetterLoggingTypedActorRef
29 import org.make.api.technical.{ActorSystemComponent, StreamUtils, TimeSettings}
30 import org.make.core.technical.Pagination
31 import org.make.core.history.HistoryActions.VoteAndQualifications
32 import org.make.core.proposal.{ProposalId, QualificationKey}
33 import org.make.core.session.SessionId
34 import org.make.core.user.UserId
35 import org.make.core.RequestContext
36 
37 import scala.concurrent.ExecutionContext.global
38 import scala.concurrent.{ExecutionContext, Future}
39 import cats.implicits._
40 
41 trait SessionHistoryCoordinatorComponent {
42   def sessionHistoryCoordinator: ActorRef[SessionHistoryCommand]
43 }
44 
45 trait SessionHistoryCoordinatorService {
46   def getCurrentSessionId(sessionId: SessionId, newSessionId: SessionId): Future[SessionId]
47   def logTransactionalHistory[T <: LoggableHistoryEvent[_]](command: T): Future[Unit]
48   def convertSession(sessionId: SessionId, userId: UserId, requestContext: RequestContext): Future[Unit]
49   def retrieveVoteAndQualifications(
50     sessionId: SessionId,
51     proposalIds: Seq[ProposalId]
52   ): Future[Map[ProposalId, VoteAndQualifications]]
53   def retrieveVotedProposals(
54     sessionId: SessionId,
55     proposalsIds: Option[Seq[ProposalId]] = None
56   ): Future[Seq[ProposalId]]
57   def lockSessionForVote(sessionId: SessionId, proposalId: ProposalId): Future[Unit]
58   def lockSessionForQualification(sessionId: SessionId, proposalId: ProposalId, key: QualificationKey): Future[Unit]
59   def unlockSessionForVote(sessionId: SessionId, proposalId: ProposalId): Future[Unit]
60   def unlockSessionForQualification(sessionId: SessionId, proposalId: ProposalId, key: QualificationKey): Future[Unit]
61 }
62 
63 trait SessionHistoryCoordinatorServiceComponent {
64   def sessionHistoryCoordinatorService: SessionHistoryCoordinatorService
65 }
66 
67 final case class ConcurrentModification(message: String) extends Exception(message)
68 
69 trait DefaultSessionHistoryCoordinatorServiceComponent extends SessionHistoryCoordinatorServiceComponent {
70   self: SessionHistoryCoordinatorComponent with ActorSystemComponent with MakeSettingsComponent =>
71 
72   type Receiver[T] = ActorRef[SessionHistoryResponse[Error.Expired, T]]
73   type LockReceiver = ActorRef[SessionHistoryResponse[Error.LockError, LockResponse]]
74 
75   override lazy val sessionHistoryCoordinatorService: SessionHistoryCoordinatorService =
76     new DefaultSessionHistoryCoordinatorService
77 
78   class DefaultSessionHistoryCoordinatorService extends SessionHistoryCoordinatorService {
79 
80     implicit val ctx: ExecutionContext = global
81     private val proposalsPerPage: Int = makeSettings.maxHistoryProposalsPerPage
82     implicit val timeout: Timeout = TimeSettings.defaultTimeout
83 
84     override def getCurrentSessionId(sessionId: SessionId, newSessionId: SessionId): Future[SessionId] = {
85       (sessionHistoryCoordinator ?? (GetCurrentSession(sessionId, newSessionId, _))).map(_.value)
86     }
87 
88     override def logTransactionalHistory[T <: LoggableHistoryEvent[_]](command: T): Future[Unit] = {
89       def log(id: SessionId)(replyTo: Receiver[LogResult]): EventEnvelope[T] = EventEnvelope(id, command, replyTo)
90       askExpired(sessionHistoryCoordinator, log)(command.sessionId).flatMap(shapeResponse).void
91     }
92 
93     override def retrieveVoteAndQualifications(
94       sessionId: SessionId,
95       proposalIds: Seq[ProposalId]
96     ): Future[Map[ProposalId, VoteAndQualifications]] = {
97       def votesValues(id: SessionId)(replyTo: Receiver[Map[ProposalId, VoteAndQualifications]]) =
98         SessionVoteValuesCommand(id, proposalIds, replyTo)
99       askExpired(sessionHistoryCoordinator, votesValues)(sessionId).flatMap(shapeResponse)
100     }
101 
102     override def convertSession(sessionId: SessionId, userId: UserId, requestContext: RequestContext): Future[Unit] = {
103       def userConnected(id: SessionId)(replyTo: Receiver[LogResult]) =
104         UserConnected(id, userId, requestContext, replyTo)
105       askExpired(sessionHistoryCoordinator, userConnected)(sessionId).flatMap(shapeResponse).void
106     }
107 
108     private def retrieveVotedProposalsPage(
109       request: RequestSessionVotedProposals,
110       offset: Int
111     ): Future[Seq[ProposalId]] = {
112       def requestPaginate(proposalsIds: Option[Seq[ProposalId]]) =
113         RequestSessionVotedProposalsPaginate(
114           sessionId = request.sessionId,
115           proposalsIds = proposalsIds,
116           limit = Pagination.Limit(proposalsPerPage),
117           offset = Pagination.Offset(offset)
118         )
119       request.proposalsIds match {
120         case Some(proposalsIds) if proposalsIds.sizeIs > proposalsPerPage =>
121           Source(proposalsIds)
122             .sliding(proposalsPerPage, proposalsPerPage)
123             .mapAsync(5) { someProposalsIds =>
124               doRequestVotedProposalsPage(requestPaginate(Some(someProposalsIds)))
125             }
126             .mapConcat(identity)
127             .runWith(Sink.seq)
128         case _ =>
129           doRequestVotedProposalsPage(requestPaginate(request.proposalsIds))
130       }
131     }
132 
133     private def doRequestVotedProposalsPage(request: RequestSessionVotedProposalsPaginate): Future[Seq[ProposalId]] = {
134       def voted(id: SessionId)(replyTo: Receiver[Seq[ProposalId]]) =
135         SessionVotedProposalsPaginateCommand(id, request.proposalsIds, request.limit, request.offset, replyTo)
136       askExpired(sessionHistoryCoordinator, voted)(request.sessionId).flatMap(shapeResponse)
137     }
138 
139     override def retrieveVotedProposals(
140       sessionId: SessionId,
141       proposalsIds: Option[Seq[ProposalId]] = None
142     ): Future[Seq[ProposalId]] = {
143       StreamUtils
144         .asyncPageToPageSource(retrieveVotedProposalsPage(RequestSessionVotedProposals(sessionId, proposalsIds), _))
145         .mapConcat(identity)
146         .runWith(Sink.seq)
147     }
148 
149     override def lockSessionForVote(sessionId: SessionId, proposalId: ProposalId): Future[Unit] = {
150       def lock(id: SessionId)(replyTo: LockReceiver) = LockProposalForVote(id, proposalId, replyTo)
151       askExpired(sessionHistoryCoordinator, lock)(sessionId).flatMap(shapeResponse).void
152     }
153 
154     override def lockSessionForQualification(
155       sessionId: SessionId,
156       proposalId: ProposalId,
157       key: QualificationKey
158     ): Future[Unit] = {
159       def lock(id: SessionId)(replyTo: LockReceiver) = LockProposalForQualification(id, proposalId, key, replyTo)
160       askExpired(sessionHistoryCoordinator, lock)(sessionId).flatMap(shapeResponse).void
161     }
162 
163     override def unlockSessionForVote(sessionId: SessionId, proposalId: ProposalId): Future[Unit] = {
164       def release(id: SessionId)(replyTo: LockReceiver) = ReleaseProposalForVote(id, proposalId, replyTo)
165       askExpired(sessionHistoryCoordinator, release)(sessionId).flatMap(shapeResponse).void
166     }
167 
168     override def unlockSessionForQualification(
169       sessionId: SessionId,
170       proposalId: ProposalId,
171       key: QualificationKey
172     ): Future[Unit] = {
173       def release(id: SessionId)(replyTo: LockReceiver) = ReleaseProposalForQualification(id, proposalId, key, replyTo)
174       askExpired(sessionHistoryCoordinator, release)(sessionId).flatMap(shapeResponse).void
175     }
176 
177     private def shapeResponse[E, T](response: SessionHistoryResponse[E, T]): Future[T] = response match {
178       case SessionHistoryResponse.Envelope(value)                    => Future.successful(value)
179       case l: SessionHistoryResponse.LockResponse                    => Future.successful(l)
180       case SessionHistoryResponse.Error.LockAlreadyAcquired(message) => Future.failed(ConcurrentModification(message))
181       case error: SessionHistoryResponse.Error[_] =>
182         Future.failed(
183           new IllegalStateException(
184             s"Unknown response from session history actor: ${error.toString} of class ${error.getClass.getName}}"
185           )
186         )
187     }
188 
189     @SuppressWarnings(Array("org.wartremover.warts.Recursion"))
190     private def askExpired[Req, Res](ref: ActorRef[Req], replyTo: SessionId => ActorRef[Res] => Req)(
191       id: SessionId
192     ): Future[Res] = {
193       (ref ?? replyTo(id)).flatMap {
194         case ExpiredSession(newSessionId) => askExpired(ref, replyTo)(newSessionId)
195         case other                        => Future.successful(other)
196       }
197     }
198 
199   }
200 }
201 
202 final case class RequestSessionVotedProposals(sessionId: SessionId, proposalsIds: Option[Seq[ProposalId]] = None)
203 final case class RequestSessionVotedProposalsPaginate(
204   sessionId: SessionId,
205   proposalsIds: Option[Seq[ProposalId]] = None,
206   limit: Pagination.Limit,
207   offset: Pagination.Offset
208 )