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.userhistory
21 
22 import cats.implicits._
23 import akka.actor.typed.ActorRef
24 import akka.stream.scaladsl.{Sink, Source}
25 import akka.util.Timeout
26 import org.make.api.extensions.MakeSettingsComponent
27 import org.make.api.sessionhistory.Ack
28 import org.make.api.technical.BetterLoggingActors._
29 import org.make.api.technical.{ActorSystemComponent, MakePersistentActor, 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, VoteKey}
33 import org.make.core.user._
34 import org.make.core.{ValidationError, ValidationFailedError}
35 
36 import scala.concurrent.ExecutionContext.Implicits.global
37 import scala.concurrent.Future
38 
39 trait UserHistoryCoordinatorComponent {
40   def userHistoryCoordinator: ActorRef[UserHistoryCommand]
41 }
42 
43 trait UserHistoryCoordinatorService {
44   def logHistory(event: UserHistoryEvent[_]): Unit
45   def logTransactionalHistory(event: TransactionalUserHistoryEvent[_]): Future[Unit]
46   def retrieveVoteAndQualifications(
47     userId: UserId,
48     proposalIds: Seq[ProposalId]
49   ): Future[Map[ProposalId, VoteAndQualifications]]
50   def retrieveVotedProposals(request: RequestUserVotedProposals): Future[Seq[ProposalId]]
51   def delete(userId: UserId): Future[Unit]
52 }
53 
54 trait UserHistoryCoordinatorServiceComponent {
55   def userHistoryCoordinatorService: UserHistoryCoordinatorService
56 }
57 
58 trait DefaultUserHistoryCoordinatorServiceComponent extends UserHistoryCoordinatorServiceComponent {
59   self: UserHistoryCoordinatorComponent with ActorSystemComponent with MakeSettingsComponent =>
60 
61   override lazy val userHistoryCoordinatorService: UserHistoryCoordinatorService =
62     new DefaultUserHistoryCoordinatorService
63 
64   class DefaultUserHistoryCoordinatorService extends UserHistoryCoordinatorService {
65 
66     private val proposalsPerPage: Int = makeSettings.maxHistoryProposalsPerPage
67     implicit val timeout: Timeout = TimeSettings.defaultTimeout
68 
69     override def logHistory(event: UserHistoryEvent[_]): Unit = {
70       userHistoryCoordinator ! UserHistoryEnvelope(event.userId, event)
71     }
72 
73     override def logTransactionalHistory(event: TransactionalUserHistoryEvent[_]): Future[Unit] = {
74       (userHistoryCoordinator ?? { replyTo: ActorRef[UserHistoryResponse[Ack.type]] =>
75         UserHistoryTransactionalEnvelope(event.userId, event, replyTo)
76       }).void
77     }
78 
79     override def retrieveVoteAndQualifications(
80       userId: UserId,
81       proposalIds: Seq[ProposalId]
82     ): Future[Map[ProposalId, VoteAndQualifications]] = {
83       (userHistoryCoordinator ?? (RequestVoteValues(userId, proposalIds, _))).flatMap(shapeResponse)
84     }
85 
86     private def retrieveVotedProposalsPage(request: RequestUserVotedProposals, offset: Int): Future[Seq[ProposalId]] = {
87       def requestPaginate(
88         proposalsIds: Option[Seq[ProposalId]],
89         replyTo: ActorRef[UserHistoryResponse[Seq[ProposalId]]]
90       ) =
91         RequestUserVotedProposalsPaginate(
92           userId = request.userId,
93           filterVotes = request.filterVotes,
94           filterQualifications = request.filterQualifications,
95           proposalsIds = proposalsIds,
96           limit = Pagination.Limit(proposalsPerPage),
97           offset = Pagination.Offset(offset),
98           replyTo
99         )
100       request.proposalsIds match {
101         case Some(proposalsIds) if proposalsIds.sizeIs > proposalsPerPage =>
102           Source(proposalsIds)
103             .sliding(proposalsPerPage, proposalsPerPage)
104             .mapAsync(5) { someProposalsIds =>
105               (userHistoryCoordinator ?? (requestPaginate(Some(someProposalsIds), _)))
106                 .flatMap(shapeResponse)
107             }
108             .mapConcat(identity)
109             .runWith(Sink.seq)
110         case _ =>
111           (userHistoryCoordinator ?? (requestPaginate(request.proposalsIds, _))).flatMap(shapeResponse)
112       }
113     }
114 
115     override def retrieveVotedProposals(request: RequestUserVotedProposals): Future[Seq[ProposalId]] = {
116       StreamUtils
117         .asyncPageToPageSource(retrieveVotedProposalsPage(request, _))
118         .mapConcat(identity)
119         .runWith(Sink.seq)
120     }
121 
122     override def delete(userId: UserId): Future[Unit] =
123       (userHistoryCoordinator ?? (Stop(userId, _))) >>
124         MakePersistentActor.delete(userId, UserHistoryActor.JournalPluginId)
125 
126     private def shapeResponse[T](response: UserHistoryResponse[T]): Future[T] = response match {
127       case UserHistoryResponse(value) => Future.successful(value)
128       case other =>
129         Future.failed(
130           ValidationFailedError(errors = Seq(ValidationError("status", "invalid_state", Some(other.toString))))
131         )
132     }
133   }
134 }
135 
136 final case class RequestUserVotedProposals(
137   userId: UserId,
138   filterVotes: Option[Seq[VoteKey]] = None,
139   filterQualifications: Option[Seq[QualificationKey]] = None,
140   proposalsIds: Option[Seq[ProposalId]] = None
141 )