code.grnet.gr Git - aquarium/blob - src/main/scala/gr/grnet/aquarium/actor/service/user/UserActor.scala

   1 /*
   2  * Copyright 2011-2012 GRNET S.A. All rights reserved.
   3  *
   4  * Redistribution and use in source and binary forms, with or
   5  * without modification, are permitted provided that the following
   6  * conditions are met:
   7  *
   8  *   1. Redistributions of source code must retain the above
   9  *      copyright notice, this list of conditions and the following
  10  *      disclaimer.
  11  *
  12  *   2. Redistributions in binary form must reproduce the above
  13  *      copyright notice, this list of conditions and the following
  14  *      disclaimer in the documentation and/or other materials
  15  *      provided with the distribution.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  18  * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  19  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  20  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  21  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  22  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  24  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  25  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  26  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  27  * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  28  * POSSIBILITY OF SUCH DAMAGE.
  29  *
  30  * The views and conclusions contained in the software and
  31  * documentation are those of the authors and should not be
  32  * interpreted as representing official policies, either expressed
  33  * or implied, of GRNET S.A.
  34  */
  35
  36 package gr.grnet.aquarium.actor
  37 package service
  38 package user
  39
  40 import gr.grnet.aquarium.util.date.TimeHelpers
  41 import gr.grnet.aquarium.service.event.BalanceEvent
  42 import gr.grnet.aquarium.event.model.im.IMEventModel
  43 import gr.grnet.aquarium.actor.message.config.AquariumPropertiesLoaded
  44 import gr.grnet.aquarium.actor.message.config.InitializeUserActorState
  45 import gr.grnet.aquarium.actor.message.event.ProcessIMEvent
  46 import gr.grnet.aquarium.actor.message.event.ProcessResourceEvent
  47 import gr.grnet.aquarium.util.{LogHelpers, shortClassNameOf}
  48 import gr.grnet.aquarium.AquariumInternalError
  49 import gr.grnet.aquarium.computation.BillingMonthInfo
  50 import gr.grnet.aquarium.charging.state.{WorkingUserState, UserStateModel}
  51 import gr.grnet.aquarium.policy.PolicyDefinedFullPriceTableRef
  52 import gr.grnet.aquarium.event.model.resource.{StdResourceEvent, ResourceEventModel}
  53 import gr.grnet.aquarium.actor.message.GetUserBalanceRequest
  54 import gr.grnet.aquarium.actor.message.GetUserBalanceResponse
  55 import gr.grnet.aquarium.actor.message.GetUserBalanceResponseData
  56 import gr.grnet.aquarium.actor.message.GetUserStateRequest
  57 import gr.grnet.aquarium.actor.message.GetUserStateResponse
  58 import gr.grnet.aquarium.actor.message.GetUserWalletRequest
  59 import gr.grnet.aquarium.actor.message.GetUserWalletResponse
  60 import gr.grnet.aquarium.actor.message.GetUserWalletResponseData
  61 import gr.grnet.aquarium.actor.message.GetUserBillRequest
  62 import gr.grnet.aquarium.actor.message.GetUserBillResponse
  63 import gr.grnet.aquarium.actor.message.GetUserBillResponseData
  64 import gr.grnet.aquarium.charging.state.WorkingAgreementHistory
  65 import gr.grnet.aquarium.policy.StdUserAgreement
  66 import gr.grnet.aquarium.charging.state.UserStateBootstrap
  67 import gr.grnet.aquarium.charging.bill.BillEntry
  68
  69 /**
  70  *
  71  * @author Christos KK Loverdos <loverdos@gmail.com>
  72  */
  73
  74 class UserActor extends ReflectiveRoleableActor {
  75   private[this] var _userID: String = "<?>"
  76   private[this] var _workingUserState: WorkingUserState = _
  77   private[this] var _userCreationIMEvent: IMEventModel = _
  78   private[this] val _workingAgreementHistory: WorkingAgreementHistory = new WorkingAgreementHistory
  79   private[this] var _latestIMEventID: String = ""
  80   private[this] var _latestResourceEventID: String = ""
  81   private[this] var _userStateBootstrap: UserStateBootstrap = _
  82
  83   def unsafeUserID = {
  84     if(!haveUserID) {
  85       throw new AquariumInternalError("%s not initialized")
  86     }
  87
  88     this._userID
  89   }
  90
  91   override def postStop() {
  92     DEBUG("I am finally stopped (in postStop())")
  93     aquarium.akkaService.notifyUserActorPostStop(this)
  94   }
  95
  96   private[this] def shutmedown(): Unit = {
  97     if(haveUserID) {
  98       aquarium.akkaService.invalidateUserActor(this)
  99     }
 100   }
 101
 102   override protected def onThrowable(t: Throwable, message: AnyRef) = {
 103     LogHelpers.logChainOfCauses(logger, t)
 104     ERROR(t, "Terminating due to: %s(%s)", shortClassNameOf(t), t.getMessage)
 105
 106     shutmedown()
 107   }
 108
 109   def role = UserActorRole
 110
 111   private[this] def chargingService = aquarium.chargingService
 112
 113   private[this] def stdUserStateStoreFunc = (userState: UserStateModel) ⇒ {
 114     aquarium.userStateStore.insertUserState(userState)
 115   }
 116
 117   @inline private[this] def haveUserID = {
 118     this._userID ne null
 119   }
 120
 121   @inline private[this] def haveUserCreationIMEvent = {
 122     this._userCreationIMEvent ne null
 123   }
 124
 125   def onAquariumPropertiesLoaded(event: AquariumPropertiesLoaded): Unit = {
 126   }
 127
 128   @inline private[this] def haveAgreements = {
 129     this._workingAgreementHistory.size > 0
 130   }
 131
 132   @inline private[this] def haveWorkingUserState = {
 133     this._workingUserState ne null
 134   }
 135
 136   @inline private[this] def haveUserStateBootstrap = {
 137     this._userStateBootstrap ne null
 138   }
 139
 140   private[this] def updateAgreementHistoryFrom(imEvent: IMEventModel): Unit = {
 141     if(imEvent.isCreateUser) {
 142       if(haveUserCreationIMEvent) {
 143         throw new AquariumInternalError(
 144           "Got user creation event (id=%s) but I already have one (id=%s)",
 145             this._userCreationIMEvent.id,
 146             imEvent.id
 147         )
 148       }
 149
 150       this._userCreationIMEvent = imEvent
 151     }
 152
 153     val effectiveFromMillis = imEvent.occurredMillis
 154     val role = imEvent.role
 155     // calling unsafe just for the side-effect
 156     assert(null ne aquarium.unsafePriceTableForRoleAt(role, effectiveFromMillis))
 157
 158     val newAgreement = StdUserAgreement(
 159       imEvent.id,
 160       Some(imEvent.id),
 161       effectiveFromMillis,
 162       Long.MaxValue,
 163       role,
 164       PolicyDefinedFullPriceTableRef()
 165     )
 166
 167     this._workingAgreementHistory += newAgreement
 168   }
 169
 170   private[this] def updateLatestIMEventIDFrom(imEvent: IMEventModel): Unit = {
 171     this._latestIMEventID = imEvent.id
 172   }
 173
 174   private[this] def updateLatestResourceEventIDFrom(rcEvent: ResourceEventModel): Unit = {
 175     this._latestResourceEventID = rcEvent.id
 176   }
 177
 178   /**
 179    * Creates the initial state that is related to IMEvents.
 180    */
 181   private[this] def initializeStateOfIMEvents(): Unit = {
 182     // NOTE: this._userID is already set up by onInitializeUserActorState()
 183     aquarium.imEventStore.foreachIMEventInOccurrenceOrder(this._userID) { imEvent ⇒
 184       DEBUG("Replaying %s", imEvent)
 185
 186       updateAgreementHistoryFrom(imEvent)
 187       updateLatestIMEventIDFrom(imEvent)
 188     }
 189
 190     if(haveAgreements) {
 191       DEBUG("Initial agreement history %s", this._workingAgreementHistory.toJsonString)
 192       logSeparator()
 193     }
 194   }
 195
 196   /**
 197    * Resource events are processed only if the user has been created and has agreements.
 198    * Otherwise nothing can be computed.
 199    */
 200   private[this] def shouldProcessResourceEvents: Boolean = {
 201     haveUserCreationIMEvent && haveAgreements && haveUserStateBootstrap
 202   }
 203
 204   private[this] def loadWorkingUserStateAndUpdateAgreementHistory(): Unit = {
 205     assert(this.haveAgreements, "this.haveAgreements")
 206     assert(this.haveUserCreationIMEvent, "this.haveUserCreationIMEvent")
 207
 208     val userCreationMillis = this._userCreationIMEvent.occurredMillis
 209     val userCreationRole = this._userCreationIMEvent.role // initial role
 210     val userCreationIMEventID = this._userCreationIMEvent.id
 211
 212     if(!haveUserStateBootstrap) {
 213       this._userStateBootstrap = UserStateBootstrap(
 214         this._userID,
 215         userCreationMillis,
 216         aquarium.initialUserAgreement(userCreationRole, userCreationMillis, Some(userCreationIMEventID)),
 217         aquarium.initialUserBalance(userCreationRole, userCreationMillis)
 218       )
 219     }
 220
 221     val now = TimeHelpers.nowMillis()
 222     this._workingUserState = chargingService.replayMonthChargingUpTo(
 223       BillingMonthInfo.fromMillis(now),
 224       now,
 225       this._userStateBootstrap,
 226       aquarium.currentResourceTypesMap,
 227       aquarium.userStateStore.insertUserState
 228     )
 229
 230     // Final touch: Update agreement history in the working user state.
 231     // The assumption is that all agreement changes go via IMEvents, so the
 232     // state this._workingAgreementHistory is always the authoritative source.
 233     if(haveWorkingUserState) {
 234       this._workingUserState.workingAgreementHistory.setFrom(this._workingAgreementHistory)
 235       DEBUG("Computed working user state %s", this._workingUserState.toJsonString)
 236     }
 237   }
 238
 239   private[this] def initializeStateOfResourceEvents(event: InitializeUserActorState): Unit = {
 240     if(!this.haveAgreements) {
 241       DEBUG("Cannot initializeResourceEventsState() from %s. There are no agreements", event)
 242       return
 243     }
 244
 245     if(!this.haveUserCreationIMEvent) {
 246       DEBUG("Cannot initializeResourceEventsState() from %s. I never got a CREATE IMEvent", event)
 247       return
 248     }
 249
 250     // We will also need this functionality when receiving IMEvents, so we place it in a method
 251     loadWorkingUserStateAndUpdateAgreementHistory()
 252
 253     if(haveWorkingUserState) {
 254       DEBUG("Initial working user state %s", this._workingUserState.toJsonString)
 255       logSeparator()
 256     }
 257   }
 258
 259   def onInitializeUserActorState(event: InitializeUserActorState): Unit = {
 260     this._userID = event.userID
 261     DEBUG("Got %s", event)
 262
 263     initializeStateOfIMEvents()
 264     initializeStateOfResourceEvents(event)
 265   }
 266
 267   /**
 268    * Process [[gr.grnet.aquarium.event.model.im.IMEventModel]]s.
 269    * When this method is called, we assume that all proper checks have been made and it
 270    * is OK to proceed with the event processing.
 271    */
 272   def onProcessIMEvent(processEvent: ProcessIMEvent): Unit = {
 273     val imEvent = processEvent.imEvent
 274     val hadUserCreationIMEvent = haveUserCreationIMEvent
 275
 276     if(!haveAgreements) {
 277       // This IMEvent has arrived after any ResourceEvents
 278       INFO("Arrived after any ResourceEvent: %s", imEvent.toDebugString)
 279       initializeStateOfIMEvents()
 280     }
 281     else {
 282       if(this._latestIMEventID == imEvent.id) {
 283         // This happens when the actor is brought to life, then immediately initialized, and then
 284         // sent the first IM event. But from the initialization procedure, this IM event will have
 285         // already been loaded from DB!
 286         INFO("Ignoring first %s", imEvent.toDebugString)
 287         logSeparator()
 288
 289         //this._latestIMEventID = imEvent.id
 290         return
 291       }
 292       if(imEvent.isAddCredits)  {
 293         if(!hadUserCreationIMEvent && haveUserCreationIMEvent)
 294         loadWorkingUserStateAndUpdateAgreementHistory()
 295         onHandleAddCreditsEvent(imEvent)
 296
 297       } else {
 298       updateAgreementHistoryFrom(imEvent)
 299       updateLatestIMEventIDFrom(imEvent)
 300       }
 301     }
 302
 303     // Must also update user state if we know when in history the life of a user begins
 304     if(!hadUserCreationIMEvent && haveUserCreationIMEvent) {
 305       INFO("Processing user state, since we had a CREATE IMEvent")
 306       loadWorkingUserStateAndUpdateAgreementHistory()
 307     }
 308
 309     logSeparator()
 310   }
 311
 312   /* Convert astakos message for adding credits
 313     to a regular RESOURCE message */
 314   def onHandleAddCreditsEvent(imEvent : IMEventModel) = {
 315     val credits = imEvent.details(IMEventModel.DetailsNames.credits).toInt.toDouble
 316     val event = new StdResourceEvent(
 317       imEvent.id,
 318       imEvent.occurredMillis,
 319       imEvent.receivedMillis,
 320       imEvent.userID,
 321       imEvent.clientID,
 322       imEvent.eventType,
 323       imEvent.eventType,
 324       credits,
 325       imEvent.eventVersion,
 326       imEvent.details
 327     )
 328     //Console.err.println("Event: " + event)
 329     //Console.err.println("Total credits before: " + _workingUserState.totalCredits)
 330     onProcessResourceEvent(new ProcessResourceEvent(event))
 331     //Console.err.println("Total credits after: " + _workingUserState.totalCredits)
 332     //Console.err.println("OK.")
 333   }
 334
 335   def onProcessResourceEvent(event: ProcessResourceEvent): Unit = {
 336     val rcEvent = event.rcEvent
 337
 338     if(!shouldProcessResourceEvents) {
 339       // This means the user has not been created (at least, as far as Aquarium is concerned).
 340       // So, we do not process any resource event
 341       DEBUG("Not processing %s", rcEvent.toJsonString)
 342       logSeparator()
 343
 344       return
 345     }
 346
 347     // Since the latest resource event per resource is recorded in the user state,
 348     // we do not need to query the store. Just query the in-memory state.
 349     // Note: This is a similar situation with the first IMEvent received right after the user
 350     //       actor is created.
 351     if(this._latestResourceEventID == rcEvent.id) {
 352       INFO("Ignoring first %s", rcEvent.toDebugString)
 353       logSeparator()
 354
 355       return
 356     }
 357
 358     val now = TimeHelpers.nowMillis()
 359     // TODO: Review this and its usage in user state.
 360     // TODO: The assumption is that the resource set increases all the time,
 361     // TODO: so the current map contains everything ever known (assuming we do not run backwards in time).
 362     val currentResourcesMap = aquarium.currentResourceTypesMap
 363
 364     val nowBillingMonthInfo = BillingMonthInfo.fromMillis(now)
 365     val nowYear = nowBillingMonthInfo.year
 366     val nowMonth = nowBillingMonthInfo.month
 367
 368     val eventOccurredMillis = rcEvent.occurredMillis
 369     val eventBillingMonthInfo = BillingMonthInfo.fromMillis(eventOccurredMillis)
 370     val eventYear = eventBillingMonthInfo.year
 371     val eventMonth = eventBillingMonthInfo.month
 372
 373     def computeBatch(): Unit = {
 374       DEBUG("Going for out of sync charging")
 375       this._workingUserState = chargingService.replayMonthChargingUpTo(
 376         nowBillingMonthInfo,
 377         // Take into account that the event may be out-of-sync.
 378         // TODO: Should we use this._latestResourceEventOccurredMillis instead of now?
 379         now max eventOccurredMillis,
 380         this._userStateBootstrap,
 381         currentResourcesMap,
 382         stdUserStateStoreFunc
 383       )
 384
 385       updateLatestResourceEventIDFrom(rcEvent)
 386     }
 387
 388     def computeRealtime(): Unit = {
 389       DEBUG("Going for in sync charging")
 390       chargingService.processResourceEvent(
 391         rcEvent,
 392         this._workingUserState,
 393         nowBillingMonthInfo,
 394         true
 395       )
 396
 397       updateLatestResourceEventIDFrom(rcEvent)
 398     }
 399
 400     val oldTotalCredits = this._workingUserState.totalCredits
 401     // FIXME check these
 402     if(nowYear != eventYear || nowMonth != eventMonth) {
 403       DEBUG(
 404         "nowYear(%s) != eventYear(%s) || nowMonth(%s) != eventMonth(%s)",
 405         nowYear, eventYear,
 406         nowMonth, eventMonth
 407       )
 408       computeBatch()
 409     }
 410     else if(this._workingUserState.latestResourceEventOccurredMillis < rcEvent.occurredMillis) {
 411       DEBUG("this._workingUserState.latestResourceEventOccurredMillis < rcEvent.occurredMillis")
 412       DEBUG(
 413         "%s < %s",
 414         TimeHelpers.toYYYYMMDDHHMMSSSSS(this._workingUserState.latestResourceEventOccurredMillis),
 415         TimeHelpers.toYYYYMMDDHHMMSSSSS(rcEvent.occurredMillis)
 416       )
 417       computeRealtime()
 418     }
 419     else {
 420       computeBatch()
 421     }
 422     val newTotalCredits = this._workingUserState.totalCredits
 423     if(oldTotalCredits * newTotalCredits < 0)
 424       aquarium.eventBus ! new BalanceEvent(this._workingUserState.userID,
 425                                            newTotalCredits>=0)
 426     DEBUG("Updated %s", this._workingUserState)
 427     logSeparator()
 428   }
 429
 430   def onGetUserBillRequest(event: GetUserBillRequest): Unit = {
 431     try{
 432       val timeslot = event.timeslot
 433       val state= if(haveWorkingUserState) Some(this._workingUserState) else None
 434       val billEntry = BillEntry.fromWorkingUserState(timeslot,this._userID,state)
 435       val billData = GetUserBillResponseData(this._userID,billEntry)
 436       sender ! GetUserBillResponse(Right(billData))
 437     } catch {
 438       case e:Exception =>
 439        e.printStackTrace()
 440        sender ! GetUserBillResponse(Left("Internal Server Error [AQU-BILL-0001]"), 500)
 441     }
 442   }
 443
 444   def onGetUserBalanceRequest(event: GetUserBalanceRequest): Unit = {
 445     val userID = event.userID
 446
 447     (haveUserCreationIMEvent, haveWorkingUserState) match {
 448       case (true, true) ⇒
 449         // (User CREATEd, with balance state)
 450         sender ! GetUserBalanceResponse(Right(GetUserBalanceResponseData(this._userID, this._workingUserState.totalCredits)))
 451
 452       case (true, false) ⇒
 453         // (User CREATEd, no balance state)
 454         // Return the default initial balance
 455         sender ! GetUserBalanceResponse(
 456           Right(
 457             GetUserBalanceResponseData(
 458               this._userID,
 459               aquarium.initialUserBalance(this._userCreationIMEvent.role, this._userCreationIMEvent.occurredMillis)
 460         )))
 461
 462       case (false, true) ⇒
 463         // (Not CREATEd, with balance state)
 464         // Clearly this is internal error
 465         sender ! GetUserBalanceResponse(Left("Internal Server Error [AQU-BAL-0001]"), 500)
 466
 467       case (false, false) ⇒
 468         // (Not CREATEd, no balance state)
 469         // The user is completely unknown
 470         sender ! GetUserBalanceResponse(Left("Unknown user %s [AQU-BAL-0004]".format(userID)), 404/*Not found*/)
 471     }
 472   }
 473
 474   def onGetUserStateRequest(event: GetUserStateRequest): Unit = {
 475     haveWorkingUserState match {
 476       case true ⇒
 477         sender ! GetUserStateResponse(Right(this._workingUserState))
 478
 479       case false ⇒
 480         sender ! GetUserStateResponse(Left("No state for user %s [AQU-STA-0006]".format(event.userID)), 404)
 481     }
 482   }
 483
 484   def onGetUserWalletRequest(event: GetUserWalletRequest): Unit = {
 485     haveWorkingUserState match {
 486       case true ⇒
 487         DEBUG("haveWorkingUserState: %s", event)
 488         sender ! GetUserWalletResponse(
 489           Right(
 490             GetUserWalletResponseData(
 491               this._userID,
 492               this._workingUserState.totalCredits,
 493               this._workingUserState.walletEntries.toList
 494         )))
 495
 496       case false ⇒
 497         DEBUG("!haveWorkingUserState: %s", event)
 498         haveUserCreationIMEvent match {
 499           case true ⇒
 500             DEBUG("haveUserCreationIMEvent: %s", event)
 501             sender ! GetUserWalletResponse(
 502               Right(
 503                 GetUserWalletResponseData(
 504                   this._userID,
 505                   aquarium.initialUserBalance(this._userCreationIMEvent.role, this._userCreationIMEvent.occurredMillis),
 506                   Nil
 507             )))
 508
 509           case false ⇒
 510             DEBUG("!haveUserCreationIMEvent: %s", event)
 511             sender ! GetUserWalletResponse(Left("No wallet for user %s [AQU-WAL-00 8]".format(event.userID)), 404)
 512         }
 513     }
 514   }
 515
 516   private[this] def D_userID = {
 517     this._userID
 518   }
 519
 520   private[this] def DEBUG(fmt: String, args: Any*) =
 521     logger.debug("[%s] - %s".format(D_userID, fmt.format(args: _*)))
 522
 523   private[this] def INFO(fmt: String, args: Any*) =
 524     logger.info("[%s] - %s".format(D_userID, fmt.format(args: _*)))
 525
 526   private[this] def WARN(fmt: String, args: Any*) =
 527     logger.warn("[%s] - %s".format(D_userID, fmt.format(args: _*)))
 528
 529   private[this] def ERROR(fmt: String, args: Any*) =
 530     logger.error("[%s] - %s".format(D_userID, fmt.format(args: _*)))
 531
 532   private[this] def ERROR(t: Throwable, fmt: String, args: Any*) =
 533     logger.error("[%s] - %s".format(D_userID, fmt.format(args: _*)), t)
 534 }