package service
package user
-import java.util.Date
-import com.ckkloverdos.maybe.{Failed, NoVal, Just}
-
import gr.grnet.aquarium.actor._
-import gr.grnet.aquarium.Configurator
-import gr.grnet.aquarium.user._
-import gr.grnet.aquarium.util.Loggable
+import akka.config.Supervision.Temporary
+import gr.grnet.aquarium.actor.message.event.{ProcessResourceEvent, ProcessIMEvent}
+import gr.grnet.aquarium.actor.message.config.{InitializeUserState, ActorProviderConfigured, AquariumPropertiesLoaded}
import gr.grnet.aquarium.util.date.TimeHelpers
-import gr.grnet.aquarium.logic.accounting.RoleAgreements
-import gr.grnet.aquarium.actor.message.service.router._
-import message.config.{ActorProviderConfigured, AquariumPropertiesLoaded}
-import gr.grnet.aquarium.event.im.IMEventModel
-import gr.grnet.aquarium.event.WalletEntry
-
+import gr.grnet.aquarium.event.model.im.IMEventModel
+import gr.grnet.aquarium.actor.message.{GetUserStateResponse, GetUserBalanceResponseData, GetUserBalanceResponse, GetUserStateRequest, GetUserBalanceRequest}
+import gr.grnet.aquarium.util.{LogHelpers, shortClassNameOf, shortNameOfClass, shortNameOfType}
+import gr.grnet.aquarium.computation.reason.{RealtimeBillingCalculation, InitialUserActorSetup, UserStateChangeReason, IMEventArrival, InitialUserStateSetup}
+import gr.grnet.aquarium.{AquariumInternalError, Aquarium}
+import gr.grnet.aquarium.computation.state.parts.IMStateSnapshot
+import gr.grnet.aquarium.computation.BillingMonthInfo
+import gr.grnet.aquarium.computation.state.{UserStateBootstrap, UserState}
+import gr.grnet.aquarium.event.model.resource.ResourceEventModel
/**
*
* @author Christos KK Loverdos <loverdos@gmail.com>
*/
-class UserActor extends ReflectiveAquariumActor {
- @volatile
+class UserActor extends ReflectiveRoleableActor {
+ private[this] var _userID: String = "<?>"
+ private[this] var _imState: IMStateSnapshot = _
private[this] var _userState: UserState = _
- @volatile
- private[this] var _timestampTheshold: Long = _
+ private[this] var _latestResourceEventOccurredMillis = TimeHelpers.nowMillis() // some valid datetime
+
+ self.lifeCycle = Temporary
+
+ private[this] def _shutmedown(): Unit = {
+ if(haveUserState) {
+ UserActorCache.invalidate(_userID)
+ }
+
+ self.stop()
+ }
+
+ override protected def onThrowable(t: Throwable, message: AnyRef) = {
+ LogHelpers.logChainOfCauses(logger, t)
+ ERROR(t, "Terminating due to: %s(%s)", shortClassNameOf(t), t.getMessage)
+
+ _shutmedown()
+ }
def role = UserActorRole
- private[this] def _configurator: Configurator = Configurator.MasterConfigurator
- private[this] def _userId = _userState.userId
+ private[this] def aquarium: Aquarium = Aquarium.Instance
- /**
- * Replay the event log for all events that affect the user state.
- */
- def rebuildState(from: Long, to: Long): Unit = {
- val start = TimeHelpers.nowMillis()
- if(_userState == null)
- createBlankState
-
- //Rebuild state from user events
- val usersDB = _configurator.storeProvider.imEventStore
- val userEvents = usersDB.findIMEventsByUserId(_userId)
- val numUserEvents = userEvents.size
- _userState = replayIMEvents(_userState, userEvents, from, to)
-
- //Rebuild state from resource events
- val eventsDB = _configurator.storeProvider.resourceEventStore
- val resourceEvents = eventsDB.findResourceEventsByUserIdAfterTimestamp(_userId, from)
- val numResourceEvents = resourceEvents.size
- // _userState = replayResourceEvents(_userState, resourceEvents, from, to)
-
- //Rebuild state from wallet entries
- val wallet = _configurator.storeProvider.walletEntryStore
- val walletEnties = wallet.findWalletEntriesAfter(_userId, new Date(from))
- val numWalletEntries = walletEnties.size
- _userState = replayWalletEntries(_userState, walletEnties, from, to)
-
- INFO(("Rebuilt state from %d events (%d user events, " +
- "%d resource events, %d wallet entries) in %d msec").format(
- numUserEvents + numResourceEvents + numWalletEntries,
- numUserEvents, numResourceEvents, numWalletEntries,
- TimeHelpers.nowMillis() - start))
+ private[this] def userStateComputations = aquarium.userStateComputations
+
+ private[this] def stdUserStateStoreFunc = (userState: UserState) ⇒ {
+ aquarium.userStateStore.insertUserState(userState)
}
- /**
- * Create an empty state for a user
- */
- def createBlankState = {
- this._userState = DefaultUserStateComputations.createInitialUserState(this._userId, 0L, true, 0.0)
+ private[this] def _timestampTheshold = {
+ aquarium.props.getLong(Aquarium.Keys.user_state_timestamp_threshold).getOr(1000L * 60 * 5 /* 5 minutes */)
}
- /**
- * Replay user events on the provided user state
- */
- def replayIMEvents(initState: UserState, events: List[IMEventModel],
- from: Long, to: Long): UserState = {
- initState
+ private[this] def haveUserState = {
+ this._userState ne null
}
+ private[this] def haveIMState = {
+ this._imState ne null
+ }
- /**
- * Replay wallet entries on the provided user state
- */
- def replayWalletEntries(initState: UserState, events: List[WalletEntry],
- from: Long, to: Long): UserState = {
- initState
+ def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
+ }
+
+ def onActorProviderConfigured(event: ActorProviderConfigured): Unit = {
+ }
+
+ private[this] def _updateIMStateRoleHistory(imEvent: IMEventModel, roleCheck: Option[String]) = {
+ if(haveIMState) {
+ val (newState,
+ creationTimeChanged,
+ activationTimeChanged,
+ roleChanged) = this._imState.updatedWithEvent(imEvent, roleCheck)
+
+ this._imState = newState
+ (creationTimeChanged, activationTimeChanged, roleChanged)
+ } else {
+ this._imState = IMStateSnapshot.initial(imEvent)
+ (
+ imEvent.isCreateUser,
+ true, // first activation status is a change by default??
+ true // first role is a change by default??
+ )
+ }
}
/**
- * Persist current user state
+ * Creates the IMStateSnapshot and returns the number of updates it made to it.
*/
- private[this] def saveUserState(): Unit = {
- _configurator.storeProvider.userStateStore.storeUserState(this._userState) match {
- case Just(record) => record
- case NoVal => ERROR("Unknown error saving state")
- case Failed(e) =>
- ERROR("Saving state failed: %s".format(e));
+ private[this] def createInitialIMState(): Unit = {
+ val store = aquarium.imEventStore
+
+ var _roleCheck = None: Option[String]
+
+ // this._userID is already set up
+ store.replayIMEventsInOccurrenceOrder(this._userID) { imEvent ⇒
+ DEBUG("Replaying %s", imEvent)
+
+ val (creationTimeChanged, activationTimeChanged, roleChanged) = _updateIMStateRoleHistory(imEvent, _roleCheck)
+ _roleCheck = this._imState.roleHistory.lastRoleName
+
+ DEBUG(
+ "(creationTimeChanged, activationTimeChanged, roleChanged)=(%s, %s, %s) using %s",
+ creationTimeChanged, activationTimeChanged, roleChanged,
+ imEvent
+ )
}
+
+ DEBUG("New %s = %s", shortNameOfType[IMStateSnapshot], this._imState)
}
- def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
- this._timestampTheshold = event.props.getLong(Configurator.Keys.user_state_timestamp_threshold).getOr(10000)
- INFO("Setup my timestampTheshold = %s", this._timestampTheshold)
+ /**
+ * Resource events are processed only if the user has been activated.
+ */
+ private[this] def shouldProcessResourceEvents: Boolean = {
+ haveIMState && this._imState.hasBeenCreated
}
- def onProcessResourceEvent(event: ProcessResourceEvent): Unit = {
- val resourceEvent = event.rcEvent
- if(resourceEvent.userID != this._userId) {
- ERROR("Received %s but my userId = %s".format(event, this._userId))
- } else {
- //ensureUserState()
- // calcWalletEntries()
- //processResourceEvent(resourceEvent, true)
+ private[this] def loadUserStateAndUpdateRoleHistory(): Unit = {
+ val userCreationMillis = this._imState.userCreationMillis.get
+ val initialRole = this._imState.roleHistory.firstRole.get.name
+
+ val userStateBootstrap = UserStateBootstrap(
+ this._userID,
+ userCreationMillis,
+ initialRole,
+ aquarium.initialAgreementForRole(initialRole, userCreationMillis),
+ aquarium.initialBalanceForRole(initialRole, userCreationMillis)
+ )
+
+ val now = TimeHelpers.nowMillis()
+ val userState = userStateComputations.doMonthBillingUpTo(
+ BillingMonthInfo.fromMillis(now),
+ now,
+ userStateBootstrap,
+ aquarium.currentResourcesMap,
+ InitialUserActorSetup(),
+ stdUserStateStoreFunc,
+ None
+ )
+
+ this._userState = userState
+
+ // Final touch: Update role history
+ if(haveIMState && haveUserState) {
+ if(this._userState.roleHistory != this._imState.roleHistory) {
+ this._userState = newUserStateWithUpdatedRoleHistory(InitialUserActorSetup())
+ }
}
}
- private[this] def processCreateUser(event: IMEventModel): Unit = {
- val userId = event.userID
- DEBUG("Creating user from state %s", event)
- val usersDB = _configurator.storeProvider.userStateStore
- usersDB.findUserStateByUserId(userId) match {
- case Just(userState) ⇒
- WARN("User already created, state = %s".format(userState))
- case failed@Failed(e) ⇒
- ERROR("[%s] %s", e.getClass.getName, e.getMessage)
- case NoVal ⇒
- val agreement = RoleAgreements.agreementForRole(event.role)
- DEBUG("User %s assigned agreement %s".format(userId, agreement.name))
-
- this._userState = DefaultUserStateComputations.createInitialUserState(
- userId,
- event.occurredMillis,
- event.isActive, 0.0, List(event.role), agreement.name)
- saveUserState
- DEBUG("Created and stored %s", this._userState)
+ private[this] def createInitialUserState(event: InitializeUserState): Unit = {
+ if(!haveIMState) {
+ // Should have been created from `createIMState()`
+ DEBUG("Cannot create user state from %s, since %s = %s", event, shortNameOfClass(classOf[IMStateSnapshot]), this._imState)
+ return
+ }
+
+ if(!this._imState.hasBeenCreated) {
+ // Cannot set the initial state!
+ DEBUG("Cannot create %s from %s, since user has not been created", shortNameOfType[UserState], event)
+ return
+ }
+
+ // We will also need this functionality when receiving IMEvents,
+ // so we place it in a method
+ loadUserStateAndUpdateRoleHistory()
+
+ if(haveUserState) {
+ DEBUG("%s = %s", shortNameOfType[UserState], this._userState)
}
}
- private[this] def processModifyUser(event: IMEventModel): Unit = {
- val now = TimeHelpers.nowMillis()
- val newActive = ActiveStateSnapshot(event.isStateActive, now)
+ def onInitializeUserState(event: InitializeUserState): Unit = {
+ this._userID = event.userID
+ DEBUG("Got %s", event)
- DEBUG("New active status = %s".format(newActive))
+ createInitialIMState()
+ createInitialUserState(event)
+ }
- this._userState = this._userState.copy(activeStateSnapshot = newActive)
+ /**
+ * Creates a new user state, taking into account the latest role history in IM state snapshot.
+ * Having an IM state snapshot is a prerequisite, otherwise you get an exception; so check before you
+ * call this.
+ */
+ private[this] def newUserStateWithUpdatedRoleHistory(stateChangeReason: UserStateChangeReason): UserState = {
+ // FIXME: Also update agreement
+ this._userState.newWithRoleHistory(this._imState.roleHistory, stateChangeReason)
}
- def onProcessIMEvent(event: ProcessIMEvent): Unit = {
- val userEvent = event.imEvent
- if(userEvent.userID != this._userId) {
- ERROR("Received %s but my userId = %s".format(userEvent, this._userId))
- } else {
- if(userEvent.isCreateUser) {
- processCreateUser(userEvent)
- } else if(userEvent.isModifyUser) {
- processModifyUser(userEvent)
+ /**
+ * Process [[gr.grnet.aquarium.event.model.im.IMEventModel]]s.
+ * When this method is called, we assume that all proper checks have been made and it
+ * is OK to proceed with the event processing.
+ */
+ def onProcessIMEvent(processEvent: ProcessIMEvent): Unit = {
+ val imEvent = processEvent.imEvent
+
+ if(!haveIMState) {
+ // This is an error. Should have been initialized from somewhere ...
+ throw new AquariumInternalError("Got %s while uninitialized".format(processEvent))
+ }
+
+ if(this._imState.latestIMEvent.id == imEvent.id) {
+ // This happens when the actor is brought to life, then immediately initialized, and then
+ // sent the first IM event. But from the initialization procedure, this IM event will have
+ // already been loaded from DB!
+ INFO("Ignoring first %s just after %s birth", imEvent.toDebugString, shortClassNameOf(this))
+ logSeparator()
+
+ return
+ }
+
+ val (creationTimeChanged,
+ activationTimeChanged,
+ roleChanged) = _updateIMStateRoleHistory(imEvent, this._imState.roleHistory.lastRoleName)
+
+ DEBUG(
+ "(creationTimeChanged, activationTimeChanged, roleChanged)=(%s, %s, %s) using %s",
+ creationTimeChanged, activationTimeChanged, roleChanged,
+ imEvent
+ )
+
+ // Must also update user state if we know when in history the life of a user begins
+ if(creationTimeChanged) {
+ if(!haveUserState) {
+ loadUserStateAndUpdateRoleHistory()
+ INFO("Loaded %s due to %s", shortNameOfType[UserState], imEvent)
+ } else {
+ // Just update role history
+ this._userState = newUserStateWithUpdatedRoleHistory(IMEventArrival(imEvent))
+ INFO("Updated %s due to %s", shortNameOfType[UserState], imEvent)
}
}
+
+ DEBUG("New %s = %s", shortNameOfType[IMStateSnapshot], this._imState)
+ logSeparator()
}
- def onRequestUserBalance(event: RequestUserBalance): Unit = {
- val userId = event.userID
- val timestamp = event.timestamp
+ def onProcessResourceEvent(event: ProcessResourceEvent): Unit = {
+ val rcEvent = event.rcEvent
+
+ if(!shouldProcessResourceEvents) {
+ // This means the user has not been activated. So, we do not process any resource event
+ DEBUG("Not processing %s", rcEvent.toJsonString)
+ logSeparator()
- if(TimeHelpers.nowMillis() - _userState.newestSnapshotTime > 60 * 1000) {
- // calcWalletEntries()
+ return
}
- self reply UserResponseGetBalance(userId, _userState.creditsSnapshot.creditAmount)
- }
- def onUserRequestGetState(event: UserRequestGetState): Unit = {
- val userId = event.userID
- if(this._userId != userId) {
- ERROR("Received %s but my userId = %s".format(event, this._userId))
- // TODO: throw an exception here
- } else {
- // FIXME: implement
- ERROR("FIXME: Should have properly computed the user state")
- // ensureUserState()
- self reply UserResponseGetState(userId, this._userState)
+ this._userState.findLatestResourceEventID match {
+ case Some(id) ⇒
+ if(id == rcEvent.id) {
+ INFO("Ignoring first %s just after %s birth", rcEvent.toDebugString, shortClassNameOf(this))
+ logSeparator()
+
+ return
+ }
+
+ case _ ⇒
}
- }
- def onActorProviderConfigured(event: ActorProviderConfigured): Unit = {
+ val now = TimeHelpers.nowMillis()
+ val dt = now - this._latestResourceEventOccurredMillis
+ val belowThreshold = dt <= _timestampTheshold
+
+ if(belowThreshold) {
+ this._latestResourceEventOccurredMillis = event.rcEvent.occurredMillis
+
+ DEBUG("Below threshold (%s ms). Not processing %s", this._timestampTheshold, rcEvent.toJsonString)
+ return
+ }
+
+ val userID = this._userID
+ val userCreationMillis = this._imState.userCreationMillis.get
+ val initialRole = this._imState.roleHistory.firstRoleName.getOrElse(aquarium.defaultInitialUserRole)
+ val initialAgreement = aquarium.initialAgreementForRole(initialRole, userCreationMillis)
+ val initialCredits = aquarium.initialBalanceForRole(initialRole, userCreationMillis)
+ val userStateBootstrap = UserStateBootstrap(
+ userID,
+ userCreationMillis,
+ initialRole,
+ initialAgreement,
+ initialCredits
+ )
+ val billingMonthInfo = BillingMonthInfo.fromMillis(now)
+ val currentResourcesMap = aquarium.currentResourcesMap
+ val calculationReason = RealtimeBillingCalculation(None, now)
+ val eventOccurredMillis = rcEvent.occurredMillis
+
+ DEBUG("Using %s", currentResourcesMap)
+
+ this._userState = aquarium.userStateComputations.doMonthBillingUpTo(
+ billingMonthInfo,
+ now max eventOccurredMillis, // take into account that the event may be out-of-sync
+ userStateBootstrap,
+ currentResourcesMap,
+ calculationReason,
+ stdUserStateStoreFunc
+ )
+
+ this._latestResourceEventOccurredMillis = event.rcEvent.occurredMillis
+
+ DEBUG("New %s = %s", shortClassNameOf(this._userState), this._userState)
}
- override def postStop {
- DEBUG("Actor[%s] stopping, saving state", self.uuid)
- saveUserState
+ def onGetUserBalanceRequest(event: GetUserBalanceRequest): Unit = {
+ val userID = event.userID
+
+ (haveIMState, haveUserState) match {
+ case (true, true) ⇒
+ // (have IMState, have UserState)
+ this._imState.hasBeenActivated match {
+ case true ⇒
+ // (have IMState, activated, have UserState)
+ self reply GetUserBalanceResponse(Right(GetUserBalanceResponseData(userID, this._userState.totalCredits)))
+
+ case false ⇒
+ // (have IMState, not activated, have UserState)
+ // Since we have user state, we should have been activated
+ self reply GetUserBalanceResponse(Left("Internal Server Error [AQU-BAL-0001]"), 500)
+ }
+
+ case (true, false) ⇒
+ // (have IMState, no UserState)
+ this._imState.hasBeenActivated match {
+ case true ⇒
+ // (have IMState, activated, no UserState)
+ // Since we are activated, we should have some state.
+ self reply GetUserBalanceResponse(Left("Internal Server Error [AQU-BAL-0002]"), 500)
+ case false ⇒
+ // (have IMState, not activated, no UserState)
+ // The user is virtually unknown
+ self reply GetUserBalanceResponse(Left("User %s not activated [AQU-BAL-0003]".format(userID)), 404 /*Not found*/)
+ }
+
+ case (false, true) ⇒
+ // (no IMState, have UserState)
+ // A bit ridiculous situation
+ self reply GetUserBalanceResponse(Left("Unknown user %s [AQU-BAL-0004]".format(userID)), 404/*Not found*/)
+
+ case (false, false) ⇒
+ // (no IMState, no UserState)
+ self reply GetUserBalanceResponse(Left("Unknown user %s [AQU-BAL-0005]".format(userID)), 404/*Not found*/)
+ }
}
- override def preRestart(reason: Throwable) {
- ERROR(reason, "preRestart: Actor[%s]", self.uuid)
+ def onGetUserStateRequest(event: GetUserStateRequest): Unit = {
+ haveUserState match {
+ case true ⇒
+ self reply GetUserStateResponse(Right(this._userState))
+
+ case false ⇒
+ self reply GetUserStateResponse(Left("No state for user %s [AQU-STA-0006]".format(event.userID)))
+ }
}
- override def postRestart(reason: Throwable) {
- ERROR(reason, "postRestart: Actor[%s]", self.uuid)
+ private[this] def D_userID = {
+ this._userID
}
private[this] def DEBUG(fmt: String, args: Any*) =
- logger.debug("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)))
+ logger.debug("[%s] - %s".format(D_userID, fmt.format(args: _*)))
private[this] def INFO(fmt: String, args: Any*) =
- logger.info("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)))
+ logger.info("[%s] - %s".format(D_userID, fmt.format(args: _*)))
private[this] def WARN(fmt: String, args: Any*) =
- logger.warn("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)))
+ logger.warn("[%s] - %s".format(D_userID, fmt.format(args: _*)))
private[this] def ERROR(fmt: String, args: Any*) =
- logger.error("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)))
+ logger.error("[%s] - %s".format(D_userID, fmt.format(args: _*)))
private[this] def ERROR(t: Throwable, fmt: String, args: Any*) =
- logger.error("UserActor[%s]: %s".format(_userId, fmt.format(args: _*)), t)
+ logger.error("[%s] - %s".format(D_userID, fmt.format(args: _*)), t)
}
projects / aquarium / blobdiff