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.{AbstractBillEntry, 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     DEBUG("Got %s", imEvent.toJsonString)
 316
 317     val credits = imEvent.details(IMEventModel.DetailsNames.credits).toInt.toDouble
 318     val event = new StdResourceEvent(
 319       imEvent.id,
 320       imEvent.occurredMillis,
 321       imEvent.receivedMillis,
 322       imEvent.userID,
 323       imEvent.clientID,
 324       imEvent.eventType,
 325       imEvent.eventType,
 326       credits,
 327       imEvent.eventVersion,
 328       imEvent.details
 329     )
 330     DEBUG("Transformed to %s", event)
 331     DEBUG("Total credits before: %s", _workingUserState.totalCredits)
 332     aquarium.resourceEventStore.insertResourceEvent(event)
 333     onProcessResourceEvent(new ProcessResourceEvent(event))
 334     DEBUG("Total credits after: %s", _workingUserState.totalCredits)
 335     //Console.err.println("OK.")
 336   }
 337
 338   def onProcessResourceEvent(event: ProcessResourceEvent): Unit = {
 339     val rcEvent = event.rcEvent
 340
 341     if(!shouldProcessResourceEvents) {
 342       // This means the user has not been created (at least, as far as Aquarium is concerned).
 343       // So, we do not process any resource event
 344       DEBUG("Not processing %s", rcEvent.toJsonString)
 345       logSeparator()
 346
 347       return
 348     }
 349
 350     // Since the latest resource event per resource is recorded in the user state,
 351     // we do not need to query the store. Just query the in-memory state.
 352     // Note: This is a similar situation with the first IMEvent received right after the user
 353     //       actor is created.
 354     if(this._latestResourceEventID == rcEvent.id) {
 355       INFO("Ignoring first %s", rcEvent.toDebugString)
 356       logSeparator()
 357
 358       return
 359     }
 360
 361     val now = TimeHelpers.nowMillis()
 362     // TODO: Review this and its usage in user state.
 363     // TODO: The assumption is that the resource set increases all the time,
 364     // TODO: so the current map contains everything ever known (assuming we do not run backwards in time).
 365     val currentResourcesMap = aquarium.currentResourceTypesMap
 366
 367     val nowBillingMonthInfo = BillingMonthInfo.fromMillis(now)
 368     val nowYear = nowBillingMonthInfo.year
 369     val nowMonth = nowBillingMonthInfo.month
 370
 371     val eventOccurredMillis = rcEvent.occurredMillis
 372     val eventBillingMonthInfo = BillingMonthInfo.fromMillis(eventOccurredMillis)
 373     val eventYear = eventBillingMonthInfo.year
 374     val eventMonth = eventBillingMonthInfo.month
 375
 376     def computeBatch(): Unit = {
 377       DEBUG("Going for out of sync charging")
 378       this._workingUserState = chargingService.replayMonthChargingUpTo(
 379         nowBillingMonthInfo,
 380         // Take into account that the event may be out-of-sync.
 381         // TODO: Should we use this._latestResourceEventOccurredMillis instead of now?
 382         now max eventOccurredMillis,
 383         this._userStateBootstrap,
 384         currentResourcesMap,
 385         stdUserStateStoreFunc
 386       )
 387
 388       updateLatestResourceEventIDFrom(rcEvent)
 389     }
 390
 391     def computeRealtime(): Unit = {
 392       DEBUG("Going for in sync charging")
 393       chargingService.processResourceEvent(
 394         rcEvent,
 395         this._workingUserState,
 396         nowBillingMonthInfo,
 397         true
 398       )
 399
 400       updateLatestResourceEventIDFrom(rcEvent)
 401     }
 402
 403     val oldTotalCredits = this._workingUserState.totalCredits
 404     // FIXME check these
 405     if(nowYear != eventYear || nowMonth != eventMonth) {
 406       DEBUG(
 407         "nowYear(%s) != eventYear(%s) || nowMonth(%s) != eventMonth(%s)",
 408         nowYear, eventYear,
 409         nowMonth, eventMonth
 410       )
 411       computeBatch()
 412     }
 413     else if(this._workingUserState.latestResourceEventOccurredMillis < rcEvent.occurredMillis) {
 414       DEBUG("this._workingUserState.latestResourceEventOccurredMillis < rcEvent.occurredMillis")
 415       DEBUG(
 416         "%s < %s",
 417         TimeHelpers.toYYYYMMDDHHMMSSSSS(this._workingUserState.latestResourceEventOccurredMillis),
 418         TimeHelpers.toYYYYMMDDHHMMSSSSS(rcEvent.occurredMillis)
 419       )
 420       computeRealtime()
 421     }
 422     else {
 423       computeBatch()
 424     }
 425     val newTotalCredits = this._workingUserState.totalCredits
 426     if(oldTotalCredits * newTotalCredits < 0)
 427       aquarium.eventBus ! new BalanceEvent(this._workingUserState.userID,
 428                                            newTotalCredits>=0)
 429     DEBUG("Updated %s", this._workingUserState)
 430     logSeparator()
 431   }
 432
 433   def onGetUserBillRequest(event: GetUserBillRequest): Unit = {
 434     try{
 435       val timeslot = event.timeslot
 436       val state= if(haveWorkingUserState) Some(this._workingUserState) else None
 437       val billEntry = AbstractBillEntry.fromWorkingUserState(timeslot,this._userID,state)
 438       val billData = GetUserBillResponseData(this._userID,billEntry)
 439       sender ! GetUserBillResponse(Right(billData))
 440     } catch {
 441       case e:Exception =>
 442        e.printStackTrace()
 443        sender ! GetUserBillResponse(Left("Internal Server Error [AQU-BILL-0001]"), 500)
 444     }
 445   }
 446
 447   def onGetUserBalanceRequest(event: GetUserBalanceRequest): Unit = {
 448     val userID = event.userID
 449
 450     (haveUserCreationIMEvent, haveWorkingUserState) match {
 451       case (true, true) ⇒
 452         // (User CREATEd, with balance state)
 453         val realtimeMillis = TimeHelpers.nowMillis()
 454         chargingService.calculateRealtimeWorkingUserState(
 455           this._workingUserState,
 456           BillingMonthInfo.fromMillis(realtimeMillis),
 457           realtimeMillis
 458         )
 459
 460         sender ! GetUserBalanceResponse(Right(GetUserBalanceResponseData(this._userID, this._workingUserState.totalCredits)))
 461
 462       case (true, false) ⇒
 463         // (User CREATEd, no balance state)
 464         // Return the default initial balance
 465         sender ! GetUserBalanceResponse(
 466           Right(
 467             GetUserBalanceResponseData(
 468               this._userID,
 469               aquarium.initialUserBalance(this._userCreationIMEvent.role, this._userCreationIMEvent.occurredMillis)
 470         )))
 471
 472       case (false, true) ⇒
 473         // (Not CREATEd, with balance state)
 474         // Clearly this is internal error
 475         sender ! GetUserBalanceResponse(Left("Internal Server Error [AQU-BAL-0001]"), 500)
 476
 477       case (false, false) ⇒
 478         // (Not CREATEd, no balance state)
 479         // The user is completely unknown
 480         sender ! GetUserBalanceResponse(Left("Unknown user %s [AQU-BAL-0004]".format(userID)), 404/*Not found*/)
 481     }
 482   }
 483
 484   def onGetUserStateRequest(event: GetUserStateRequest): Unit = {
 485     haveWorkingUserState match {
 486       case true ⇒
 487         val realtimeMillis = TimeHelpers.nowMillis()
 488         chargingService.calculateRealtimeWorkingUserState(
 489           this._workingUserState,
 490           BillingMonthInfo.fromMillis(realtimeMillis),
 491           realtimeMillis
 492         )
 493
 494         sender ! GetUserStateResponse(Right(this._workingUserState))
 495
 496       case false ⇒
 497         sender ! GetUserStateResponse(Left("No state for user %s [AQU-STA-0006]".format(event.userID)), 404)
 498     }
 499   }
 500
 501   def onGetUserWalletRequest(event: GetUserWalletRequest): Unit = {
 502     haveWorkingUserState match {
 503       case true ⇒
 504         DEBUG("haveWorkingUserState: %s", event)
 505         val realtimeMillis = TimeHelpers.nowMillis()
 506         chargingService.calculateRealtimeWorkingUserState(
 507           this._workingUserState,
 508           BillingMonthInfo.fromMillis(realtimeMillis),
 509           realtimeMillis
 510         )
 511
 512         sender ! GetUserWalletResponse(
 513           Right(
 514             GetUserWalletResponseData(
 515               this._userID,
 516               this._workingUserState.totalCredits,
 517               this._workingUserState.walletEntries.toList
 518         )))
 519
 520       case false ⇒
 521         DEBUG("!haveWorkingUserState: %s", event)
 522         haveUserCreationIMEvent match {
 523           case true ⇒
 524             DEBUG("haveUserCreationIMEvent: %s", event)
 525             sender ! GetUserWalletResponse(
 526               Right(
 527                 GetUserWalletResponseData(
 528                   this._userID,
 529                   aquarium.initialUserBalance(this._userCreationIMEvent.role, this._userCreationIMEvent.occurredMillis),
 530                   Nil
 531             )))
 532
 533           case false ⇒
 534             DEBUG("!haveUserCreationIMEvent: %s", event)
 535             sender ! GetUserWalletResponse(Left("No wallet for user %s [AQU-WAL-00 8]".format(event.userID)), 404)
 536         }
 537     }
 538   }
 539
 540   private[this] def D_userID = {
 541     this._userID
 542   }
 543
 544   private[this] def DEBUG(fmt: String, args: Any*) =
 545     logger.debug("[%s] - %s".format(D_userID, fmt.format(args: _*)))
 546
 547   private[this] def INFO(fmt: String, args: Any*) =
 548     logger.info("[%s] - %s".format(D_userID, fmt.format(args: _*)))
 549
 550   private[this] def WARN(fmt: String, args: Any*) =
 551     logger.warn("[%s] - %s".format(D_userID, fmt.format(args: _*)))
 552
 553   private[this] def ERROR(fmt: String, args: Any*) =
 554     logger.error("[%s] - %s".format(D_userID, fmt.format(args: _*)))
 555
 556   private[this] def ERROR(t: Throwable, fmt: String, args: Any*) =
 557     logger.error("[%s] - %s".format(D_userID, fmt.format(args: _*)), t)
 558 }