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 akka.actor.typed.scaladsl.{ActorContext, Behaviors}
23 import akka.actor.typed.{ActorRef, Behavior}
24 import akka.persistence.typed.PersistenceId
25 import akka.persistence.typed.scaladsl.{Effect, EventSourcedBehavior, ReplyEffect, RetentionCriteria}
26 import grizzled.slf4j.Logging
27 import org.make.api.sessionhistory.Ack
28 import org.make.api.technical.EffectBuilderHelper._
29 import org.make.api.userhistory.UserHistoryResponse.SessionEventsInjected
30 import org.make.core.technical.Pagination
31 import org.make.core.history.HistoryActions._
32 import org.make.core.proposal.{ProposalId, QualificationKey, VoteKey}
33 
34 import java.time.ZonedDateTime
35 
36 object UserHistoryActor extends Logging {
37 
38   val JournalPluginId: String = "make-api.event-sourcing.users.journal"
39   val SnapshotPluginId: String = "make-api.event-sourcing.users.snapshot"
40   val QueryJournalPluginId: String = "make-api.event-sourcing.users.query"
41 
42   def apply(): Behavior[UserHistoryCommand] = {
43     Behaviors.setup { implicit context =>
44       EventSourcedBehavior
45         .withEnforcedReplies[UserHistoryCommand, UserHistoryEvent[_], UserVotesAndQualifications](
46           persistenceId = PersistenceId.ofUniqueId(context.self.path.name),
47           emptyState = UserVotesAndQualifications(Map.empty),
48           commandHandler(),
49           eventHandler
50         )
51         .withJournalPluginId(JournalPluginId)
52         .withSnapshotPluginId(SnapshotPluginId)
53         .withRetention(RetentionCriteria.snapshotEvery(numberOfEvents = 10, keepNSnapshots = 50))
54     }
55   }
56 
57   def getVotesValues(
58     userHistory: UserVotesAndQualifications,
59     proposalIds: Seq[ProposalId]
60   ): UserHistoryResponse[Map[ProposalId, VoteAndQualifications]] = {
61     UserHistoryResponse(userHistory.votesAndQualifications.filter {
62       case (proposalId, _) => proposalIds.contains(proposalId)
63     })
64   }
65 
66   def getVotedProposals(
67     userHistory: UserVotesAndQualifications,
68     filterVotes: Option[Seq[VoteKey]],
69     filterQualifications: Option[Seq[QualificationKey]],
70     proposalIds: Option[Seq[ProposalId]],
71     limit: Pagination.Limit,
72     offset: Pagination.Offset
73   ): UserHistoryResponse[Seq[ProposalId]] = {
74     val response: Seq[ProposalId] = userHistory.votesAndQualifications.filter {
75       case (proposalId, votesAndQualifications) =>
76         proposalIds.forall(_.contains(proposalId)) &&
77           filterVotes.forall(_.contains(votesAndQualifications.voteKey)) &&
78           filterQualifications.forall { qualifications =>
79             votesAndQualifications.qualificationKeys.exists { case (value, _) => qualifications.contains(value) }
80           } && votesAndQualifications.trust.isTrusted
81     }.toSeq.sortBy { case (_, votesAndQualifications) => votesAndQualifications.date }.map {
82       case (proposalId, _) => proposalId
83     }.slice(offset.extractInt, offset.extractInt + limit.extractInt)
84     UserHistoryResponse(response)
85   }
86 
87   def commandHandler()(implicit context: ActorContext[_]): (
88     UserVotesAndQualifications,
89     UserHistoryCommand
90   ) => ReplyEffect[UserHistoryEvent[_], UserVotesAndQualifications] = {
91     case (_, command: InjectSessionEvents) =>
92       Effect
93         .persist(command.events)
94         .thenReply(command.replyTo)(_ => UserHistoryResponse(SessionEventsInjected))
95     case (state, RequestVoteValues(_, values, replyTo)) => onRetrieveVoteValues(state, values, replyTo)
96     case (
97         state,
98         RequestUserVotedProposalsPaginate(_, filterVotes, filterQualifications, proposalsIds, limit, offset, replyTo)
99         ) =>
100       onRetrieveUserVotedProposals(state, filterVotes, filterQualifications, proposalsIds, limit, offset, replyTo)
101     case (_, UserHistoryTransactionalEnvelope(_, event: TransactionalUserHistoryEvent[_], replyTo)) =>
102       Effect
103         .persist[UserHistoryEvent[_], UserVotesAndQualifications](event)
104         .thenPublish(event)
105         .thenReply(replyTo)(_ => UserHistoryResponse(Ack))
106     case (_, UserHistoryEnvelope(_, event: UserHistoryEvent[_])) =>
107       Effect
108         .persist[UserHistoryEvent[_], UserVotesAndQualifications](event)
109         .thenPublish(event)
110         .thenNoReply()
111     case (_, Stop(_, replyTo)) => Effect.stop().thenReply(replyTo)(_ => UserHistoryResponse(Ack))
112   }
113 
114   def eventHandler: (UserVotesAndQualifications, UserHistoryEvent[_]) => UserVotesAndQualifications = {
115     case (state, LogUserVoteEvent(_, _, UserAction(date, _, UserVote(proposalId, voteKey, voteTrust)))) =>
116       applyVoteEvent(state, proposalId, voteKey, date, voteTrust)
117     case (state, LogUserUnvoteEvent(_, _, UserAction(_, _, UserUnvote(proposalId, voteKey, _)))) =>
118       applyUnvoteEvent(state, proposalId, voteKey)
119     case (
120         state,
121         LogUserQualificationEvent(_, _, UserAction(_, _, UserQualification(proposalId, qualificationKey, voteTrust)))
122         ) =>
123       applyQualificationEvent(state, proposalId, qualificationKey, voteTrust)
124     case (
125         state,
126         LogUserUnqualificationEvent(_, _, UserAction(_, _, UserUnqualification(proposalId, qualificationKey, _)))
127         ) =>
128       applyUnqualificationEvent(state, proposalId, qualificationKey)
129     case (state, _) => state
130   }
131 
132   def onRetrieveVoteValues(
133     state: UserVotesAndQualifications,
134     proposalIds: Seq[ProposalId],
135     replyTo: ActorRef[UserHistoryResponse[Map[ProposalId, VoteAndQualifications]]]
136   ): ReplyEffect[UserHistoryEvent[_], UserVotesAndQualifications] = {
137     Effect.reply(replyTo)(getVotesValues(state, proposalIds))
138   }
139 
140   def onRetrieveUserVotedProposals(
141     state: UserVotesAndQualifications,
142     filterVotes: Option[Seq[VoteKey]],
143     filterQualifications: Option[Seq[QualificationKey]],
144     proposalIds: Option[Seq[ProposalId]],
145     limit: Pagination.Limit,
146     offset: Pagination.Offset,
147     replyTo: ActorRef[UserHistoryResponse[Seq[ProposalId]]]
148   ): ReplyEffect[UserHistoryEvent[_], UserVotesAndQualifications] = {
149     Effect.reply(replyTo)(getVotedProposals(state, filterVotes, filterQualifications, proposalIds, limit, offset))
150   }
151 
152   def applyVoteEvent(
153     state: UserVotesAndQualifications,
154     proposalId: ProposalId,
155     voteKey: VoteKey,
156     voteDate: ZonedDateTime,
157     voteTrust: VoteTrust
158   ): UserVotesAndQualifications = {
159     state.votesAndQualifications.get(proposalId) match {
160       case Some(voteAndQualifications) if voteAndQualifications.voteKey == voteKey => state
161       case _ =>
162         state.copy(votesAndQualifications = state.votesAndQualifications + (proposalId -> VoteAndQualifications(
163           voteKey,
164           Map.empty,
165           voteDate,
166           voteTrust
167         ))
168         )
169     }
170   }
171 
172   def applyQualificationEvent(
173     state: UserVotesAndQualifications,
174     proposalId: ProposalId,
175     qualificationKey: QualificationKey,
176     voteTrust: VoteTrust
177   ): UserVotesAndQualifications = {
178     state.votesAndQualifications.get(proposalId) match {
179       case None => state
180       case Some(voteAndQualifications) =>
181         val newVoteAndQualifications: VoteAndQualifications = voteAndQualifications
182           .copy(qualificationKeys = voteAndQualifications.qualificationKeys + (qualificationKey -> voteTrust))
183         state
184           .copy(votesAndQualifications = state.votesAndQualifications + (proposalId -> newVoteAndQualifications))
185     }
186   }
187 
188   def applyUnqualificationEvent(
189     state: UserVotesAndQualifications,
190     proposalId: ProposalId,
191     qualificationKey: QualificationKey
192   ): UserVotesAndQualifications = {
193     state.votesAndQualifications.get(proposalId) match {
194       case None => state
195       case Some(voteAndQualifications) =>
196         val newVoteAndQualifications =
197           voteAndQualifications
198             .copy(qualificationKeys = voteAndQualifications.qualificationKeys.filter {
199               case (key, _) => key != qualificationKey
200             })
201         state
202           .copy(votesAndQualifications = state.votesAndQualifications + (proposalId -> newVoteAndQualifications))
203     }
204   }
205 
206   def applyUnvoteEvent(
207     state: UserVotesAndQualifications,
208     proposalId: ProposalId,
209     voteKey: VoteKey
210   ): UserVotesAndQualifications = {
211     if (state.votesAndQualifications.get(proposalId).exists(_.voteKey == voteKey)) {
212       state.copy(votesAndQualifications = state.votesAndQualifications - proposalId)
213     } else {
214       state
215     }
216   }
217 }