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