1 /*
2  *  Make.org Core API
3  *  Copyright (C) 2021 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.eventstream.EventStream
23 import akka.actor.typed.scaladsl.ActorContext
24 import akka.actor.typed.ActorRef
25 import akka.persistence.typed.scaladsl.{Effect, ReplyEffect}
26 import akka.util.Timeout
27 import grizzled.slf4j.Logging
28 import org.make.api.extensions.MakeSettings
29 import org.make.api.sessionhistory.SessionHistoryActor._
30 import org.make.api.sessionhistory.SessionHistoryResponse.Error.{ExpiredSession, LockAlreadyAcquired}
31 import org.make.api.sessionhistory.SessionHistoryResponse.{Envelope, LockAcquired, LockReleased}
32 import org.make.api.technical.ActorSystemHelper._
33 import org.make.api.technical.BetterLoggingActors.BetterLoggingTypedActorRef
34 import org.make.api.userhistory._
35 import org.make.core.history.HistoryActions._
36 import org.make.core.proposal.{ProposalId, QualificationKey}
37 import org.make.core.session.SessionId
38 import org.make.core.technical.IdGenerator
39 import org.make.core.user.UserId
40 import org.make.core.{DateHelper, MakeSerializable, RequestContext, SprayJsonFormatters}
41 import spray.json.DefaultJsonProtocol._
42 import spray.json.{DefaultJsonProtocol, RootJsonFormat}
43 
44 import scala.concurrent.ExecutionContext.Implicits.global
45 import java.time.ZonedDateTime
46 import scala.util.{Failure, Success}
47 
48 final case class SessionHistory(events: List[SessionHistoryEvent[_]]) extends MakeSerializable
49 
50 object SessionHistory {
51   implicit val persister: RootJsonFormat[SessionHistory] = DefaultJsonProtocol.jsonFormat1(SessionHistory.apply)
52 }
53 
54 sealed trait State extends MakeSerializable with Logging {
55   def lastEventDate: Option[ZonedDateTime]
56 
57   def handleCommand(
58     userHistoryCoordinator: ActorRef[UserHistoryCommand],
59     lockHandler: LockHandler,
60     settings: MakeSettings,
61     idGenerator: IdGenerator
62   )(
63     implicit context: ActorContext[SessionHistoryCommand],
64     timeout: Timeout
65   ): SessionHistoryCommand => Effect[SessionHistoryEvent[_], State]
66 
67   def handleEvent(settings: MakeSettings): SessionHistoryEvent[_] => State
68 
69   def retrieveOrExpireSession(
70     command: GetCurrentSession,
71     settings: MakeSettings,
72     lastEventDate: Option[ZonedDateTime]
73   ): ReplyEffect[SessionHistoryEvent[_], State] = {
74     val isSessionExpired: Boolean =
75       lastEventDate.exists(_.isBefore(DateHelper.now().minusNanos(settings.SessionCookie.lifetime.toNanos)))
76 
77     if (isSessionExpired) {
78       Effect
79         .persist(
80           SessionExpired(
81             sessionId = command.sessionId,
82             requestContext = RequestContext.empty,
83             action =
84               SessionAction(date = DateHelper.now(), actionType = "sessionExpired", arguments = command.newSessionId)
85           )
86         )
87         .thenReply(command.replyTo)(_ => Envelope(command.newSessionId))
88     } else {
89       Effect
90         .persist[SessionHistoryEvent[_], State](
91           SaveLastEventDate(
92             command.sessionId,
93             RequestContext.empty,
94             action = SessionAction(
95               date = DateHelper.now(),
96               actionType = "saveLastEventDate",
97               arguments = Some(DateHelper.now())
98             )
99           )
100         )
101         .thenReply(command.replyTo)(_ => Envelope(command.sessionId))
102     }
103   }
104 
105   def acquireLockForVotesIfPossible(
106     command: LockProposalForVote,
107     lockHandler: LockHandler
108   ): ReplyEffect[SessionHistoryEvent[_], State] = {
109     lockHandler.getOrExpire(command.proposalId) match {
110       case None =>
111         lockHandler.add(command.proposalId -> Vote)
112         Effect.reply(command.replyTo)(LockAcquired)
113       case Some(_) =>
114         Effect.reply(command.replyTo)(
115           LockAlreadyAcquired(
116             s"Trying to vote on a locked proposal ${command.proposalId.value} for session ${command.sessionId.value}"
117           )
118         )
119     }
120   }
121 
122   def acquireLockForQualificationIfPossible(
123     command: LockProposalForQualification,
124     lockHandler: LockHandler
125   ): ReplyEffect[SessionHistoryEvent[_], State] = {
126     lockHandler.getOrExpire(command.proposalId) match {
127       case None =>
128         lockHandler.add(command.proposalId -> ChangeQualifications(Seq(command.key)))
129         Effect.reply(command.replyTo)(LockAcquired)
130       case Some(Vote) =>
131         Effect.reply(command.replyTo)(
132           LockAlreadyAcquired(
133             s"Trying to qualify on a locked proposal ${command.proposalId.value} for session ${command.sessionId.value} (vote)"
134           )
135         )
136       case Some(ChangeQualifications(keys)) if keys.contains(command.key) =>
137         Effect.reply(command.replyTo)(
138           LockAlreadyAcquired(
139             s"Trying to qualify on a locked proposal ${command.proposalId.value} for session ${command.sessionId.value} (same qualification)"
140           )
141         )
142       case Some(ChangeQualifications(keys)) =>
143         lockHandler.add(command.proposalId -> ChangeQualifications(keys ++ Seq(command.key)))
144         Effect.reply(command.replyTo)(LockAcquired)
145     }
146   }
147 
148   def releaseProposalLock(
149     command: ReleaseProposalForVote,
150     lockHandler: LockHandler
151   ): ReplyEffect[SessionHistoryEvent[_], State] = {
152     lockHandler.release(command.proposalId)
153     Effect.reply(command.replyTo)(LockReleased)
154   }
155 
156   def releaseLockForQualification(
157     command: ReleaseProposalForQualification,
158     lockHandler: LockHandler
159   ): ReplyEffect[SessionHistoryEvent[_], State] = {
160     lockHandler.getOrExpire(command.proposalId) match {
161       case Some(ChangeQualifications(keys)) =>
162         val remainingLocks = keys.filter(_ != command.key)
163         if (remainingLocks.isEmpty) {
164           lockHandler.release(command.proposalId)
165         } else {
166           lockHandler.add(command.proposalId -> ChangeQualifications(remainingLocks))
167         }
168       case _ =>
169     }
170     Effect.reply(command.replyTo)(LockReleased)
171   }
172 
173   def onUserHistoryFailure[T](command: UserHistoryFailure[T]): ReplyEffect[SessionHistoryEvent[_], State] = {
174     logger.warn(s"Error while requesting user history: ${command.exception}")
175     Effect.reply(command.replyTo)(Envelope(command.response))
176   }
177 
178   def stopSessionHistoryActor(
179     command: StopSessionActor,
180     lastEventDate: Option[ZonedDateTime]
181   ): ReplyEffect[SessionHistoryEvent[_], State] = {
182     Effect
183       .persist[SessionHistoryEvent[_], State](
184         SaveLastEventDate(
185           command.id,
186           RequestContext.empty,
187           action = SessionAction(date = DateHelper.now(), actionType = "saveLastEventDate", arguments = lastEventDate)
188         )
189       )
190       .thenStop()
191       .thenReply(command.replyTo)(_ => Envelope(Ack))
192   }
193 }
194 
195 final case class Active(sessionHistory: SessionHistory, lastEventDate: Option[ZonedDateTime])
196     extends State
197     with Logging {
198   override def handleCommand(
199     userHistoryCoordinator: ActorRef[UserHistoryCommand],
200     lockHandler: LockHandler,
201     settings: MakeSettings,
202     idGenerator: IdGenerator
203   )(
204     implicit context: ActorContext[SessionHistoryCommand],
205     timeout: Timeout
206   ): SessionHistoryCommand => Effect[SessionHistoryEvent[_], State] = {
207     case command: GetCurrentSession                    => retrieveOrExpireSession(command, settings, lastEventDate)
208     case EventEnvelope(_, event, replyTo)              => Effect.persist(event).thenReply(replyTo)(_ => Envelope(Ack))
209     case command: SessionVoteValuesCommand             => onRetrieveVoteValues(sessionHistory, command)
210     case command: SessionVotedProposalsPaginateCommand => onRetrieveVotedProposals(sessionHistory, command)
211     case command: UserConnected                        => transformSession(sessionHistory, command, userHistoryCoordinator)
212     case command: LockProposalForVote                  => acquireLockForVotesIfPossible(command, lockHandler)
213     case command: LockProposalForQualification         => acquireLockForQualificationIfPossible(command, lockHandler)
214     case command: ReleaseProposalForVote               => releaseProposalLock(command, lockHandler)
215     case command: ReleaseProposalForQualification      => releaseLockForQualification(command, lockHandler)
216     // Unhandled
217     case _: SessionClosed         => Effect.unhandled
218     case _: UserHistorySuccess[_] => Effect.unhandled
219     case _: UserHistoryFailure[_] => Effect.unhandled
220     // Test
221     case GetState(_, replyTo)      => Effect.reply(replyTo)(Envelope(Active(sessionHistory, lastEventDate)))
222     case command: StopSessionActor => stopSessionHistoryActor(command, lastEventDate)
223   }
224 
225   override def handleEvent(settings: MakeSettings): SessionHistoryEvent[_] => State = {
226     case event: SaveLastEventDate =>
227       Active(
228         sessionHistory.copy(events = (event :: sessionHistory.events).take(settings.maxUserHistoryEvents)),
229         lastEventDate = event.action.arguments
230       )
231     case event: SessionTransforming =>
232       Transforming(
233         sessionHistory.copy(events = (event :: sessionHistory.events).take(settings.maxUserHistoryEvents)),
234         event.requestContext,
235         lastEventDate = Some(event.action.date)
236       )
237     case event: SessionExpired => Expired(event.action.arguments, lastEventDate)
238     case event: SessionHistoryEvent[_] =>
239       Active(
240         sessionHistory.copy(events = (event :: sessionHistory.events).take(settings.maxUserHistoryEvents)),
241         lastEventDate = Some(event.action.date)
242       )
243   }
244 
245   def onRetrieveVoteValues(
246     state: SessionHistory,
247     command: SessionVoteValuesCommand
248   ): ReplyEffect[SessionHistoryEvent[_], State] = {
249     val voteRelatedActions: Seq[VoteRelatedAction] = actions(state, command.proposalIds)
250     val voteAndQualifications: Map[ProposalId, VoteAndQualifications] = voteByProposalId(voteRelatedActions).map {
251       case (proposalId, voteAction) =>
252         proposalId -> VoteAndQualifications(
253           voteAction.key,
254           qualifications(voteRelatedActions).getOrElse(proposalId, Map.empty),
255           voteAction.date,
256           voteAction.trust
257         )
258     }
259     Effect.reply(command.replyTo)(Envelope(voteAndQualifications))
260   }
261 
262   @SuppressWarnings(Array("org.wartremover.warts.IterableOps"))
263   def voteByProposalId(actions: Seq[VoteRelatedAction]): Map[ProposalId, VoteAction] = {
264     actions.collect {
265       case voteAction: GenericVoteAction => voteAction
266     }.groupBy(_.proposalId)
267       .map {
268         case (proposalId, voteActions) =>
269           proposalId -> voteActions.maxBy(_.date.toString)
270       }
271       .collect {
272         case (k, value: VoteAction) => k -> value
273       }
274   }
275 
276   def onRetrieveVotedProposals(
277     state: SessionHistory,
278     command: SessionVotedProposalsPaginateCommand
279   ): ReplyEffect[SessionHistoryEvent[_], State] = {
280     val votedProposals: Seq[ProposalId] = voteByProposalId(voteActions(state)).toSeq.sortBy {
281       case (_, votesAndQualifications) => votesAndQualifications.date
282     }.collect {
283       case (proposalId, _) if command.proposalsIds.forall(_.contains(proposalId)) => proposalId
284     }.slice(command.offset.extractInt, command.offset.extractInt + command.limit.extractInt)
285 
286     Effect.reply(command.replyTo)(Envelope(votedProposals))
287   }
288 
289   def transformSession(
290     state: SessionHistory,
291     command: UserConnected,
292     userHistoryCoordinator: ActorRef[UserHistoryCommand]
293   )(
294     implicit context: ActorContext[SessionHistoryCommand],
295     timeout: Timeout
296   ): ReplyEffect[SessionHistoryEvent[_], State] = {
297     logger.info(
298       s"Transforming session ${command.id} to user ${command.userId.value} with events ${state.events.map(_.toString).mkString(", ")}"
299     )
300 
301     val events: Seq[LoggableHistoryEvent[_]] = state.events.collect {
302       case event: LoggableHistoryEvent[_] => event
303     }.sortBy(_.action.date.toString)
304 
305     context.pipeToSelf(
306       userHistoryCoordinator ?? (InjectSessionEvents(
307         command.userId,
308         events.map(_.toUserHistoryEvent(command.userId)),
309         _
310       ))
311     ) {
312       case Success(_)         => SessionClosed(command.id, command.userId, command.replyTo)
313       case Failure(exception) => UserHistoryFailure[LogResult](command.id, Failed, exception, command.replyTo)
314     }
315 
316     Effect
317       .persist(
318         SessionTransforming(
319           command.id,
320           command.requestContext,
321           SessionAction(DateHelper.now(), "transformingSession", command.userId)
322         )
323       )
324       .thenNoReply()
325   }
326 
327   private def actions(state: SessionHistory, proposalIds: Seq[ProposalId]): Seq[VoteRelatedAction] =
328     state.events.flatMap {
329       case LogSessionVoteEvent(_, _, SessionAction(date, _, SessionVote(proposalId, voteKey, trust)))
330           if proposalIds.contains(proposalId) =>
331         Some(VoteAction(proposalId, date, voteKey, trust))
332       case LogSessionUnvoteEvent(_, _, SessionAction(date, _, SessionUnvote(proposalId, voteKey, trust)))
333           if proposalIds.contains(proposalId) =>
334         Some(UnvoteAction(proposalId, date, voteKey, trust))
335       case LogSessionQualificationEvent(
336           _,
337           _,
338           SessionAction(date, _, SessionQualification(proposalId, qualificationKey, trust))
339           ) if proposalIds.contains(proposalId) =>
340         Some(QualificationAction(proposalId, date, qualificationKey, trust))
341       case LogSessionUnqualificationEvent(
342           _,
343           _,
344           SessionAction(date, _, SessionUnqualification(proposalId, qualificationKey, trust))
345           ) if proposalIds.contains(proposalId) =>
346         Some(UnqualificationAction(proposalId, date, qualificationKey, trust))
347       case _ => None
348     }
349 
350   @SuppressWarnings(Array("org.wartremover.warts.IterableOps"))
351   private def qualifications(actions: Seq[VoteRelatedAction]): Map[ProposalId, Map[QualificationKey, VoteTrust]] = {
352     actions.collect {
353       case qualificationAction: GenericQualificationAction => qualificationAction
354     }.groupBy { qualificationAction =>
355       qualificationAction.proposalId -> qualificationAction.key
356     }.map {
357       case (groupKey, qualificationAction) => groupKey -> qualificationAction.maxBy(_.date.toString)
358     }.collect {
359       case (k, v: QualificationAction) => k -> v
360     }.toList.map {
361       case ((proposalId, key), action) => proposalId -> (key -> action.trust)
362     }.groupBy {
363       case (proposalId, _) => proposalId
364     }.map {
365       case (proposalId, qualificationList) =>
366         proposalId -> qualificationList.map {
367           case (_, key) => key
368         }.toMap
369     }
370   }
371 
372   private def voteActions(state: SessionHistory): Seq[VoteRelatedAction] = state.events.flatMap {
373     case LogSessionVoteEvent(_, _, SessionAction(date, _, SessionVote(proposalId, voteKey, trust))) =>
374       Some(VoteAction(proposalId, date, voteKey, trust))
375     case LogSessionUnvoteEvent(_, _, SessionAction(date, _, SessionUnvote(proposalId, voteKey, trust))) =>
376       Some(UnvoteAction(proposalId, date, voteKey, trust))
377     case LogSessionQualificationEvent(
378         _,
379         _,
380         SessionAction(date, _, SessionQualification(proposalId, qualificationKey, trust))
381         ) =>
382       Some(QualificationAction(proposalId, date, qualificationKey, trust))
383     case LogSessionUnqualificationEvent(
384         _,
385         _,
386         SessionAction(date, _, SessionUnqualification(proposalId, qualificationKey, trust))
387         ) =>
388       Some(UnqualificationAction(proposalId, date, qualificationKey, trust))
389     case _ => None
390   }
391 }
392 
393 object Active extends SprayJsonFormatters {
394   implicit val jsonFormat: RootJsonFormat[Active] = DefaultJsonProtocol.jsonFormat2(Active.apply)
395 }
396 
397 final case class Transforming(
398   sessionHistory: SessionHistory,
399   requestContext: RequestContext,
400   lastEventDate: Option[ZonedDateTime]
401 ) extends State {
402   override def handleCommand(
403     userHistoryCoordinator: ActorRef[UserHistoryCommand],
404     lockHandler: LockHandler,
405     settings: MakeSettings,
406     idGenerator: IdGenerator
407   )(
408     implicit context: ActorContext[SessionHistoryCommand],
409     timeout: Timeout
410   ): SessionHistoryCommand => Effect[SessionHistoryEvent[_], State] = {
411     case _: EventEnvelope[_]                      => Effect.stash()
412     case command: LockProposalForVote             => acquireLockForVotesIfPossible(command, lockHandler)
413     case command: LockProposalForQualification    => acquireLockForQualificationIfPossible(command, lockHandler)
414     case command: ReleaseProposalForVote          => releaseProposalLock(command, lockHandler)
415     case command: ReleaseProposalForQualification => releaseLockForQualification(command, lockHandler)
416     // Internal
417     case command: SessionClosed         => onSessionClosed(command, requestContext, idGenerator)
418     case command: UserHistoryFailure[_] => onUserHistoryFailure(command)
419     // Unhandled
420     case _: GetCurrentSession                    => Effect.stash()
421     case _: SessionVoteValuesCommand             => Effect.stash()
422     case _: SessionVotedProposalsPaginateCommand => Effect.stash()
423     case _: UserConnected                        => Effect.unhandled
424     case _: UserHistorySuccess[_]                => Effect.unhandled
425     // Test
426     case GetState(_, replyTo) =>
427       Effect.reply(replyTo)(Envelope(Transforming(sessionHistory, requestContext, lastEventDate)))
428     case command: StopSessionActor => stopSessionHistoryActor(command, lastEventDate)
429   }
430 
431   override def handleEvent(settings: MakeSettings): SessionHistoryEvent[_] => State = {
432     case transformed: SessionTransformed =>
433       Closed(transformed.action.arguments, lastEventDate = Some(transformed.action.date))
434     case event: SaveLastEventDate =>
435       Transforming(
436         sessionHistory.copy(events = (event :: sessionHistory.events).take(settings.maxUserHistoryEvents)),
437         requestContext,
438         lastEventDate = event.action.arguments
439       )
440     case event: SessionHistoryEvent[_] =>
441       Transforming(
442         sessionHistory.copy(events = (event :: sessionHistory.events).take(settings.maxUserHistoryEvents)),
443         requestContext,
444         lastEventDate = Some(event.action.date)
445       )
446   }
447 
448   def onSessionClosed(command: SessionClosed, requestContext: RequestContext, idGenerator: IdGenerator)(
449     implicit context: ActorContext[SessionHistoryCommand]
450   ): ReplyEffect[SessionHistoryEvent[_], State] = {
451     Effect
452       .persist(
453         SessionTransformed(
454           sessionId = command.id,
455           requestContext = RequestContext.empty,
456           action = SessionAction(date = DateHelper.now(), actionType = "transformSession", arguments = command.userId)
457         )
458       )
459       .thenRun { _: State =>
460         context.system.eventStream ! EventStream.Publish(
461           UserConnectedEvent(
462             connectedUserId = Some(command.userId),
463             eventDate = DateHelper.now(),
464             userId = command.userId,
465             requestContext = requestContext,
466             eventId = Some(idGenerator.nextEventId())
467           )
468         )
469       }
470       .thenReply(command.replyTo)(_ => Envelope(Ack))
471       .thenUnstashAll()
472   }
473 }
474 
475 object Transforming extends SprayJsonFormatters {
476   implicit val jsonFormat: RootJsonFormat[Transforming] = DefaultJsonProtocol.jsonFormat3(Transforming.apply)
477 }
478 
479 final case class Closed(userId: UserId, lastEventDate: Option[ZonedDateTime]) extends State with Logging {
480   override def handleCommand(
481     userHistoryCoordinator: ActorRef[UserHistoryCommand],
482     lockHandler: LockHandler,
483     settings: MakeSettings,
484     idGenerator: IdGenerator
485   )(
486     implicit context: ActorContext[SessionHistoryCommand],
487     timeout: Timeout
488   ): SessionHistoryCommand => Effect[SessionHistoryEvent[_], State] = {
489     case command: GetCurrentSession => retrieveOrExpireSession(command, settings, lastEventDate)
490     case command @ EventEnvelope(_, _: LoggableHistoryEvent[_], _) =>
491       transferEventWhilstClosed(command, userId, userHistoryCoordinator)
492     case command: SessionVoteValuesCommand => onRetrieveUserVoteValues(command, userId, userHistoryCoordinator)
493     case command: SessionVotedProposalsPaginateCommand =>
494       onRetrieveUserVotedProposals(command, userId, userHistoryCoordinator)
495     case command: UserConnected                   => transformClosedSession(command, userId)
496     case command: LockProposalForVote             => acquireLockForVotesIfPossible(command, lockHandler)
497     case command: LockProposalForQualification    => acquireLockForQualificationIfPossible(command, lockHandler)
498     case command: ReleaseProposalForVote          => releaseProposalLock(command, lockHandler)
499     case command: ReleaseProposalForQualification => releaseLockForQualification(command, lockHandler)
500     // Internal
501     case UserHistorySuccess(_, response, replyTo) => Effect.reply(replyTo)(Envelope(response))
502     case command: UserHistoryFailure[_]           => onUserHistoryFailure(command)
503     // Unhandled
504     case _: SessionClosed => Effect.unhandled
505     // Test
506     case GetState(_, replyTo)      => Effect.reply(replyTo)(Envelope(Closed(userId, lastEventDate)))
507     case command: StopSessionActor => stopSessionHistoryActor(command, lastEventDate)
508   }
509 
510   override def handleEvent(settings: MakeSettings): SessionHistoryEvent[_] => State = {
511     case transformed: SessionTransformed =>
512       Closed(transformed.action.arguments, lastEventDate = Some(transformed.action.date))
513     case event: SaveLastEventDate =>
514       Closed(userId, lastEventDate = event.action.arguments)
515     case event: SessionExpired =>
516       Expired(event.action.arguments, lastEventDate)
517     case event: SessionHistoryEvent[_] =>
518       Closed(userId, lastEventDate = Some(event.action.date))
519   }
520 
521   def onRetrieveUserVoteValues(
522     command: SessionVoteValuesCommand,
523     userId: UserId,
524     userHistoryCoordinator: ActorRef[UserHistoryCommand]
525   )(
526     implicit context: ActorContext[SessionHistoryCommand],
527     timeout: Timeout
528   ): ReplyEffect[SessionHistoryEvent[_], State] = {
529     context.pipeToSelf(userHistoryCoordinator ?? (RequestVoteValues(userId, command.proposalIds, _))) {
530       case Success(value) =>
531         UserHistorySuccess[Map[ProposalId, VoteAndQualifications]](command.id, value.value, command.replyTo)
532       case Failure(exception) =>
533         UserHistoryFailure[Map[ProposalId, VoteAndQualifications]](command.id, Map.empty, exception, command.replyTo)
534     }
535     Effect.noReply
536   }
537 
538   def onRetrieveUserVotedProposals(
539     command: SessionVotedProposalsPaginateCommand,
540     userId: UserId,
541     userHistoryCoordinator: ActorRef[UserHistoryCommand]
542   )(
543     implicit context: ActorContext[SessionHistoryCommand],
544     timeout: Timeout
545   ): ReplyEffect[SessionHistoryEvent[_], State] = {
546     context.pipeToSelf(
547       userHistoryCoordinator ?? (RequestUserVotedProposalsPaginate(
548         userId = userId,
549         filterVotes = None,
550         filterQualifications = None,
551         proposalsIds = command.proposalsIds,
552         limit = command.limit,
553         offset = command.offset,
554         _
555       ))
556     ) {
557       case Success(value) => UserHistorySuccess[Seq[ProposalId]](command.id, value.value, command.replyTo)
558       case Failure(exception) =>
559         UserHistoryFailure[Seq[ProposalId]](command.id, Seq.empty, exception, command.replyTo)
560     }
561     Effect.noReply
562   }
563 
564   def transformClosedSession(command: UserConnected, userId: UserId): ReplyEffect[SessionHistoryEvent[_], State] = {
565     if (command.userId != userId) {
566       logger.warn(s"Session ${command.id} has moved from user ${userId.value} to user ${command.userId.value}")
567       Effect.persist(
568         SessionTransformed(
569           command.id,
570           command.requestContext,
571           SessionAction(DateHelper.now(), "sessionTransformed", command.userId)
572         )
573       )
574     }
575     Effect.reply(command.replyTo)(Envelope(Ack))
576   }
577 
578   def transferEventWhilstClosed(
579     command: EventEnvelope[LoggableHistoryEvent[_]],
580     userId: UserId,
581     userHistoryCoordinator: ActorRef[UserHistoryCommand]
582   )(
583     implicit context: ActorContext[SessionHistoryCommand],
584     timeout: Timeout
585   ): ReplyEffect[SessionHistoryEvent[_], State] = {
586     context.pipeToSelf(userHistoryCoordinator ?? { replyTo: ActorRef[UserHistoryResponse[Ack.type]] =>
587       UserHistoryTransactionalEnvelope(userId, command.event.toUserHistoryEvent(userId), replyTo)
588     }) {
589       case Success(UserHistoryResponse(log)) => UserHistorySuccess[LogResult](command.id, log, command.replyTo)
590       case Failure(exception)                => UserHistoryFailure[LogResult](command.id, Failed, exception, command.replyTo)
591     }
592     Effect.noReply
593   }
594 }
595 
596 object Closed extends SprayJsonFormatters {
597   implicit val jsonFormat: RootJsonFormat[Closed] = DefaultJsonProtocol.jsonFormat2(Closed.apply)
598 }
599 
600 final case class Expired(newSessionId: SessionId, lastEventDate: Option[ZonedDateTime]) extends State {
601   override def handleCommand(
602     userHistoryCoordinator: ActorRef[UserHistoryCommand],
603     lockHandler: LockHandler,
604     settings: MakeSettings,
605     idGenerator: IdGenerator
606   )(
607     implicit context: ActorContext[SessionHistoryCommand],
608     timeout: Timeout
609   ): SessionHistoryCommand => Effect[SessionHistoryEvent[_], State] = {
610     case command: GetCurrentSession           => retrieveOrExpireSession(command, settings, lastEventDate)
611     case command: SessionExpiredCommand[_, _] => Effect.reply(command.replyTo)(ExpiredSession(newSessionId))
612     // Unhandled
613     case _: SessionClosed         => Effect.unhandled
614     case _: UserHistorySuccess[_] => Effect.unhandled
615     case _: UserHistoryFailure[_] => Effect.unhandled
616     // Test
617     case GetState(_, replyTo)      => Effect.reply(replyTo)(Envelope(Expired(newSessionId, lastEventDate)))
618     case command: StopSessionActor => stopSessionHistoryActor(command, lastEventDate)
619   }
620 
621   override def handleEvent(settings: MakeSettings): SessionHistoryEvent[_] => State = { _: SessionHistoryEvent[_] =>
622     Expired(newSessionId, lastEventDate)
623   }
624 }
625 
626 object Expired extends SprayJsonFormatters {
627   implicit val jsonFormat: RootJsonFormat[Expired] = DefaultJsonProtocol.jsonFormat2(Expired.apply)
628 }