code.grnet.gr Git - aquarium/blob - src/main/scala/gr/grnet/aquarium/user/UserStateComputations.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.user
  37
  38
  39 import scala.collection.mutable
  40 import com.ckkloverdos.maybe.{Failed, NoVal, Just, Maybe}
  41 import gr.grnet.aquarium.util.{ContextualLogger, Loggable, justForSure, failedForSure}
  42 import gr.grnet.aquarium.util.date.{TimeHelpers, MutableDateCalc}
  43 import gr.grnet.aquarium.logic.accounting.dsl.{DSLAgreement, DSLResourcesMap}
  44 import gr.grnet.aquarium.store.{StoreProvider, PolicyStore}
  45 import gr.grnet.aquarium.logic.accounting.Accounting
  46 import gr.grnet.aquarium.logic.accounting.algorithm.CostPolicyAlgorithmCompiler
  47 import gr.grnet.aquarium.AquariumException
  48 import gr.grnet.aquarium.event.{NewWalletEntry, ResourceEvent}
  49
  50 /**
  51  *
  52  * @author Christos KK Loverdos <loverdos@gmail.com>
  53  */
  54 class UserStateComputations extends Loggable {
  55   def createInitialUserState(userId: String,
  56                              userCreationMillis: Long,
  57                              isActive: Boolean,
  58                              credits: Double,
  59                              roleNames: List[String] = List(),
  60                              agreementName: String = DSLAgreement.DefaultAgreementName) = {
  61     val now = userCreationMillis
  62
  63     UserState(
  64       userId,
  65       userCreationMillis,
  66       0L,
  67       false,
  68       null,
  69       ImplicitlyIssuedResourceEventsSnapshot(List(), now),
  70       Nil,
  71       Nil,
  72       LatestResourceEventsSnapshot(List(), now),
  73       0L,
  74       0L,
  75       ActiveStateSnapshot(isActive, now),
  76       CreditSnapshot(credits, now),
  77       AgreementSnapshot(List(Agreement(agreementName, userCreationMillis)), now),
  78       RolesSnapshot(roleNames, now),
  79       OwnedResourcesSnapshot(Nil, now),
  80       Nil,
  81       UserStateChangeReasonCodes.InitialCalculationCode,
  82       InitialUserStateCalculation
  83     )
  84   }
  85
  86   def createInitialUserStateFrom(us: UserState): UserState = {
  87     createInitialUserState(
  88       us.userID,
  89       us.userCreationMillis,
  90       us.activeStateSnapshot.isActive,
  91       us.creditsSnapshot.creditAmount,
  92       us.rolesSnapshot.roles,
  93       us.agreementsSnapshot.agreementsByTimeslot.valuesIterator.toList.last
  94     )
  95   }
  96
  97   def findUserStateAtEndOfBillingMonth(userId: String,
  98                                        billingMonthInfo: BillingMonthInfo,
  99                                        storeProvider: StoreProvider,
 100                                        currentUserState: UserState,
 101                                        defaultResourcesMap: DSLResourcesMap,
 102                                        accounting: Accounting,
 103                                        algorithmCompiler: CostPolicyAlgorithmCompiler,
 104                                        calculationReason: UserStateChangeReason,
 105                                        contextualLogger: Maybe[ContextualLogger] = NoVal): Maybe[UserState] = {
 106
 107     val clog = ContextualLogger.fromOther(
 108       contextualLogger,
 109       logger,
 110       "findUserStateAtEndOfBillingMonth(%s)", billingMonthInfo)
 111     clog.begin()
 112
 113     def doCompute: Maybe[UserState] = {
 114       doFullMonthlyBilling(
 115         userId,
 116         billingMonthInfo,
 117         storeProvider,
 118         currentUserState,
 119         defaultResourcesMap,
 120         accounting,
 121         algorithmCompiler,
 122         calculationReason,
 123         Just(clog))
 124     }
 125
 126     val userStateStore = storeProvider.userStateStore
 127     val resourceEventStore = storeProvider.resourceEventStore
 128
 129     val userCreationMillis = currentUserState.userCreationMillis
 130     val userCreationDateCalc = new MutableDateCalc(userCreationMillis)
 131     val billingMonthStartMillis = billingMonthInfo.startMillis
 132     val billingMonthStopMillis  = billingMonthInfo.stopMillis
 133
 134     if(billingMonthStopMillis < userCreationMillis) {
 135       // If the user did not exist for this billing month, piece of cake
 136       clog.debug("User did not exist before %s", userCreationDateCalc)
 137
 138       // NOTE: Reason here will be: InitialUserStateCalculation
 139       val initialUserState0 = createInitialUserStateFrom(currentUserState)
 140       val initialUserStateM = userStateStore.insertUserState2(initialUserState0)
 141
 142       clog.debug("Returning ZERO state [_idM=%s] %s".format(initialUserStateM.map(_._id), initialUserStateM))
 143       clog.end()
 144
 145       initialUserStateM
 146     } else {
 147       // Ask DB cache for the latest known user state for this billing period
 148       val latestUserStateM = Maybe { userStateStore.findLatestUserStateForEndOfBillingMonth(
 149         userId,
 150         billingMonthInfo.year,
 151         billingMonthInfo.month) match {
 152
 153         case Some(latestUserState) ⇒
 154           latestUserState
 155         case None ⇒
 156           null
 157       }}
 158
 159       latestUserStateM match {
 160         case NoVal ⇒
 161           // Not found, must compute
 162           clog.debug("No user state found from cache, will have to (re)compute")
 163           val result = doCompute
 164           clog.end()
 165           result
 166
 167         case failed @ Failed(e) ⇒
 168           clog.warn("Failure while quering cache for user state: %s", failed)
 169           clog.end()
 170           failed
 171
 172         case Just(latestUserState) ⇒
 173           // Found a "latest" user state but need to see if it is indeed the true and one latest.
 174           // For this reason, we must count the events again.
 175          val latestStateOOSEventsCounter = latestUserState.billingPeriodOutOfSyncResourceEventsCounter
 176          val actualOOSEventsCounterM = resourceEventStore.countOutOfSyncEventsForBillingPeriod(
 177            userId,
 178            billingMonthStartMillis,
 179            billingMonthStopMillis)
 180
 181          actualOOSEventsCounterM match {
 182            case NoVal ⇒
 183              val errMsg = "No counter computed for out of sync events. Should at least be zero."
 184              clog.warn(errMsg)
 185              val result = Failed(new AquariumException(errMsg))
 186              clog.end()
 187              result
 188
 189            case failed @ Failed(_) ⇒
 190              clog.warn("Failure while querying for out of sync events: %s", failed)
 191              clog.end()
 192              failed
 193
 194            case Just(actualOOSEventsCounter) ⇒
 195              val counterDiff = actualOOSEventsCounter - latestStateOOSEventsCounter
 196              counterDiff match {
 197                // ZERO, we are OK!
 198                case 0 ⇒
 199                  // NOTE: Keep the caller's calculation reason
 200                  Just(latestUserState.copyForChangeReason(calculationReason))
 201
 202                // We had more, so must recompute
 203                case n if n > 0 ⇒
 204                  clog.debug(
 205                    "Found %s out of sync events (%s more), will have to (re)compute user state", actualOOSEventsCounter, n)
 206                  val result = doCompute
 207                  clog.end()
 208                  result
 209
 210                // We had less????
 211                case n if n < 0 ⇒
 212                  val errMsg = "Found %s out of sync events (%s less). DB must be inconsistent".format(actualOOSEventsCounter, n)
 213                  clog.warn(errMsg)
 214                  val result = Failed(new AquariumException(errMsg))
 215                  clog.end()
 216                  result
 217              }
 218          }
 219       }
 220     }
 221   }
 222
 223   //+ Utility methods
 224   def rcDebugInfo(rcEvent: ResourceEvent) = {
 225     rcEvent.toDebugString(false)
 226   }
 227   //- Utility methods
 228
 229   def processResourceEvent(startingUserState: UserState,
 230                            userStateWorker: UserStateWorker,
 231                            currentResourceEvent: ResourceEvent,
 232                            policyStore: PolicyStore,
 233                            stateChangeReason: UserStateChangeReason,
 234                            billingMonthInfo: BillingMonthInfo,
 235                            walletEntriesBuffer: mutable.Buffer[NewWalletEntry],
 236                            algorithmCompiler: CostPolicyAlgorithmCompiler,
 237                            clogM: Maybe[ContextualLogger] = NoVal): UserState = {
 238
 239     val clog = ContextualLogger.fromOther(clogM, logger, "walletEntriesForResourceEvent(%s)", currentResourceEvent.id)
 240
 241     var _workingUserState = startingUserState
 242
 243     val theResource = currentResourceEvent.safeResource
 244     val theInstanceId = currentResourceEvent.safeInstanceId
 245     val theValue = currentResourceEvent.value
 246
 247     val accounting = userStateWorker.accounting
 248     val resourcesMap = userStateWorker.resourcesMap
 249
 250     val currentResourceEventDebugInfo = rcDebugInfo(currentResourceEvent)
 251     clog.begin(currentResourceEventDebugInfo)
 252
 253     userStateWorker.debugTheMaps(clog)(rcDebugInfo)
 254
 255     // Ignore the event if it is not billable (but still record it in the "previous" stuff).
 256     // But to make this decision, first we need the resource definition (and its cost policy).
 257     val dslResourceOpt = resourcesMap.findResource(theResource)
 258     dslResourceOpt match {
 259       // We have a resource (and thus a cost policy)
 260       case Some(dslResource) ⇒
 261         val costPolicy = dslResource.costPolicy
 262         clog.debug("Cost policy %s for %s", costPolicy, dslResource)
 263         val isBillable = costPolicy.isBillableEventBasedOnValue(theValue)
 264         if(!isBillable) {
 265           // The resource event is not billable
 266           clog.debug("Ignoring not billable event %s", currentResourceEventDebugInfo)
 267         } else {
 268           // The resource event is billable
 269           // Find the previous event.
 270           // This is (potentially) needed to calculate new credit amount and new resource instance amount
 271           val previousResourceEventM = userStateWorker.findAndRemovePreviousResourceEvent(theResource, theInstanceId)
 272           clog.debug("PreviousM %s", previousResourceEventM.map(rcDebugInfo(_)))
 273
 274           val havePreviousResourceEvent = previousResourceEventM.isJust
 275           val needPreviousResourceEvent = costPolicy.needsPreviousEventForCreditAndAmountCalculation
 276           if(needPreviousResourceEvent && !havePreviousResourceEvent) {
 277             // This must be the first resource event of its kind, ever.
 278             // TODO: We should normally check the DB to verify the claim (?)
 279             clog.info("Ignoring first event of its kind %s", currentResourceEventDebugInfo)
 280             userStateWorker.updateIgnored(currentResourceEvent)
 281           } else {
 282             val defaultInitialAmount = costPolicy.getResourceInstanceInitialAmount
 283             val oldAmount = _workingUserState.getResourceInstanceAmount(theResource, theInstanceId, defaultInitialAmount)
 284             val oldCredits = _workingUserState.creditsSnapshot.creditAmount
 285
 286             // A. Compute new resource instance accumulating amount
 287             val newAmount = costPolicy.computeNewAccumulatingAmount(oldAmount, theValue)
 288
 289             clog.debug("theValue = %s, oldAmount = %s, newAmount = %s, oldCredits = %s", theValue, oldAmount, newAmount, oldCredits)
 290
 291             // B. Compute new wallet entries
 292             clog.debug("agreementsSnapshot = %s", _workingUserState.agreementsSnapshot)
 293             val alltimeAgreements = _workingUserState.agreementsSnapshot.agreementsByTimeslot
 294
 295             //              clog.debug("Computing full chargeslots")
 296             val fullChargeslotsM = accounting.computeFullChargeslots(
 297               previousResourceEventM,
 298               currentResourceEvent,
 299               oldCredits,
 300               oldAmount,
 301               newAmount,
 302               dslResource,
 303               resourcesMap,
 304               alltimeAgreements,
 305               algorithmCompiler,
 306               policyStore,
 307               Just(clog)
 308             )
 309
 310             // We have the chargeslots, let's associate them with the current event
 311             fullChargeslotsM match {
 312               case Just((referenceTimeslot, fullChargeslots)) ⇒
 313                 if(fullChargeslots.length == 0) {
 314                   // At least one chargeslot is required.
 315                   throw new AquariumException("No chargeslots computed for resource event %s".format(currentResourceEvent.id))
 316                 }
 317                 clog.debugSeq("fullChargeslots", fullChargeslots, 0)
 318
 319                 // C. Compute new credit amount (based on the charge slots)
 320                 val newCreditsDiff = fullChargeslots.map(_.computedCredits.get).sum
 321                 val newCredits = oldCredits - newCreditsDiff
 322
 323                 if(stateChangeReason.shouldStoreCalculatedWalletEntries) {
 324                   val newWalletEntry = NewWalletEntry(
 325                     userStateWorker.userId,
 326                     newCreditsDiff,
 327                     oldCredits,
 328                     newCredits,
 329                     TimeHelpers.nowMillis(),
 330                     referenceTimeslot,
 331                     billingMonthInfo.year,
 332                     billingMonthInfo.month,
 333                     if(havePreviousResourceEvent)
 334                       List(currentResourceEvent, justForSure(previousResourceEventM).get)
 335                     else
 336                       List(currentResourceEvent),
 337                     fullChargeslots,
 338                     dslResource,
 339                     currentResourceEvent.isSynthetic
 340                   )
 341                   clog.debug("New %s", newWalletEntry)
 342
 343                   walletEntriesBuffer += newWalletEntry
 344                 } else {
 345                   clog.debug("newCreditsDiff = %s, newCredits = %s", newCreditsDiff, newCredits)
 346                 }
 347
 348                 _workingUserState = _workingUserState.copy(
 349                   creditsSnapshot = CreditSnapshot(newCredits, TimeHelpers.nowMillis()),
 350                   stateChangeCounter = _workingUserState.stateChangeCounter + 1,
 351                   totalEventsProcessedCounter = _workingUserState.totalEventsProcessedCounter + 1
 352                 )
 353
 354               case NoVal ⇒
 355                 // At least one chargeslot is required.
 356                 throw new AquariumException("No chargeslots computed")
 357
 358               case failed@Failed(e) ⇒
 359                 throw new AquariumException(e, "Error computing chargeslots")
 360             }
 361           }
 362         }
 363
 364         // After processing, all events billable or not update the previous state
 365         userStateWorker.updatePrevious(currentResourceEvent)
 366
 367         _workingUserState = _workingUserState.copy(
 368           latestResourceEventsSnapshot = userStateWorker.previousResourceEvents.toImmutableSnapshot(TimeHelpers.nowMillis())
 369         )
 370
 371       // We do not have a resource (and thus, no cost policy)
 372       case None ⇒
 373         // Now, this is a matter of politics: what do we do if no policy was found?
 374         clog.warn("Unknown resource for %s", currentResourceEventDebugInfo)
 375     } // dslResourceOpt match
 376
 377     clog.end(currentResourceEventDebugInfo)
 378
 379     _workingUserState
 380   }
 381
 382   def processResourceEvents(resourceEvents: Traversable[ResourceEvent],
 383                             startingUserState: UserState,
 384                             userStateWorker: UserStateWorker,
 385                             policyStore: PolicyStore,
 386                             stateChangeReason: UserStateChangeReason,
 387                             billingMonthInfo: BillingMonthInfo,
 388                             walletEntriesBuffer: mutable.Buffer[NewWalletEntry],
 389                             algorithmCompiler: CostPolicyAlgorithmCompiler,
 390                             clogM: Maybe[ContextualLogger] = NoVal): UserState = {
 391
 392     var _workingUserState = startingUserState
 393
 394     for(currentResourceEvent <- resourceEvents) {
 395
 396       _workingUserState = processResourceEvent(
 397         _workingUserState,
 398         userStateWorker,
 399         currentResourceEvent,
 400         policyStore,
 401         stateChangeReason,
 402         billingMonthInfo,
 403         walletEntriesBuffer,
 404         algorithmCompiler,
 405         clogM
 406       )
 407     }
 408
 409     _workingUserState
 410   }
 411
 412
 413   def doFullMonthlyBilling(userId: String,
 414                            billingMonthInfo: BillingMonthInfo,
 415                            storeProvider: StoreProvider,
 416                            currentUserState: UserState,
 417                            defaultResourcesMap: DSLResourcesMap,
 418                            accounting: Accounting,
 419                            algorithmCompiler: CostPolicyAlgorithmCompiler,
 420                            calculationReason: UserStateChangeReason = NoSpecificChangeReason,
 421                            contextualLogger: Maybe[ContextualLogger] = NoVal): Maybe[UserState] = Maybe {
 422
 423
 424     val clog = ContextualLogger.fromOther(
 425       contextualLogger,
 426       logger,
 427       "doFullMonthlyBilling(%s)", billingMonthInfo)
 428     clog.begin()
 429
 430     val clogJ = Just(clog)
 431
 432     val previousBillingMonthUserStateM = findUserStateAtEndOfBillingMonth(
 433       userId,
 434       billingMonthInfo.previousMonth,
 435       storeProvider,
 436       currentUserState,
 437       defaultResourcesMap,
 438       accounting,
 439       algorithmCompiler,
 440       calculationReason.forPreviousBillingMonth,
 441       clogJ
 442     )
 443
 444     if(previousBillingMonthUserStateM.isNoVal) {
 445       throw new AquariumException("Could not calculate initial user state for billing %s".format(billingMonthInfo))
 446     }
 447     if(previousBillingMonthUserStateM.isFailed) {
 448       throw failedForSure(previousBillingMonthUserStateM).exception
 449     }
 450
 451     val startingUserState = justForSure(previousBillingMonthUserStateM).get
 452
 453     val userStateStore = storeProvider.userStateStore
 454     val resourceEventStore = storeProvider.resourceEventStore
 455     val policyStore = storeProvider.policyStore
 456
 457     val billingMonthStartMillis = billingMonthInfo.startMillis
 458     val billingMonthEndMillis = billingMonthInfo.stopMillis
 459
 460     // Keep the working (current) user state. This will get updated as we proceed with billing for the month
 461     // specified in the parameters.
 462     // NOTE: The calculation reason is not the one we get from the previous user state but the one our caller specifies
 463     var _workingUserState = startingUserState.copyForChangeReason(calculationReason)
 464
 465     val userStateWorker = UserStateWorker.fromUserState(_workingUserState, accounting, defaultResourcesMap)
 466
 467     userStateWorker.debugTheMaps(clog)(rcDebugInfo)
 468
 469     // First, find and process the actual resource events from DB
 470     val allResourceEventsForMonth = resourceEventStore.findAllRelevantResourceEventsForBillingPeriod(
 471       userId,
 472       billingMonthStartMillis,
 473       billingMonthEndMillis)
 474
 475     val newWalletEntries = scala.collection.mutable.ListBuffer[NewWalletEntry]()
 476
 477     _workingUserState = processResourceEvents(
 478       allResourceEventsForMonth,
 479       _workingUserState,
 480       userStateWorker,
 481       policyStore,
 482       calculationReason,
 483       billingMonthInfo,
 484       newWalletEntries,
 485       algorithmCompiler,
 486       clogJ
 487     )
 488
 489     // Second, for the remaining events which must contribute an implicit OFF, we collect those OFFs
 490     // ... in order to generate an implicit ON later
 491     val (specialEvents, theirImplicitEnds) = userStateWorker.
 492       findAndRemoveGeneratorsOfImplicitEndEvents(billingMonthEndMillis)
 493     if(specialEvents.lengthCompare(1) >= 0 || theirImplicitEnds.lengthCompare(1) >= 0) {
 494       clog.debug("")
 495       clog.debug("Process implicitly issued events")
 496       clog.debugSeq("specialEvents", specialEvents, 0)
 497       clog.debugSeq("theirImplicitEnds", theirImplicitEnds, 0)
 498     }
 499
 500     // Now, the previous and implicitly started must be our base for the following computation, so we create an
 501     // appropriate worker
 502     val specialUserStateWorker = UserStateWorker(
 503       userStateWorker.userId,
 504       LatestResourceEventsWorker.fromList(specialEvents),
 505       ImplicitlyIssuedResourceEventsWorker.Empty,
 506       IgnoredFirstResourceEventsWorker.Empty,
 507       userStateWorker.accounting,
 508       userStateWorker.resourcesMap
 509     )
 510
 511     _workingUserState = processResourceEvents(
 512       theirImplicitEnds,
 513       _workingUserState,
 514       specialUserStateWorker,
 515       policyStore,
 516       calculationReason,
 517       billingMonthInfo,
 518       newWalletEntries,
 519       algorithmCompiler,
 520       clogJ
 521     )
 522
 523     val lastUpdateTime = TimeHelpers.nowMillis()
 524
 525     _workingUserState = _workingUserState.copy(
 526       implicitlyIssuedSnapshot = userStateWorker.implicitlyIssuedStartEvents.toImmutableSnapshot(lastUpdateTime),
 527       latestResourceEventsSnapshot = userStateWorker.previousResourceEvents.toImmutableSnapshot(lastUpdateTime),
 528       stateChangeCounter = _workingUserState.stateChangeCounter + 1,
 529       parentUserStateId = startingUserState.idOpt,
 530       newWalletEntries = newWalletEntries.toList
 531     )
 532
 533     clog.debug("calculationReason = %s", calculationReason)
 534
 535     if(calculationReason.shouldStoreUserState) {
 536       val storedUserStateM = userStateStore.insertUserState2(_workingUserState)
 537       storedUserStateM match {
 538         case Just(storedUserState) ⇒
 539           clog.info("Saved [_id=%s] %s", storedUserState._id, storedUserState)
 540           _workingUserState = storedUserState
 541         case NoVal ⇒
 542           clog.warn("Could not store %s", _workingUserState)
 543         case failed @ Failed(e) ⇒
 544           clog.error(e, "Could not store %s", _workingUserState)
 545       }
 546     }
 547
 548     clog.debug("RETURN %s", _workingUserState)
 549     clog.end()
 550     _workingUserState
 551   }
 552 }
 553
 554 /**
 555  * A helper object holding intermediate state/results during resource event processing.
 556  *
 557  * @param previousResourceEvents
 558  *          This is a collection of all the latest resource events.
 559  *          We want these in order to correlate incoming resource events with their previous (in `occurredMillis` time)
 560  *          ones. Will be updated on processing the next resource event.
 561  *
 562  * @param implicitlyIssuedStartEvents
 563  *          The implicitly issued resource events at the beginning of the billing period.
 564  *
 565  * @param ignoredFirstResourceEvents
 566  *          The resource events that were first (and unused) of their kind.
 567  *
 568  * @author Christos KK Loverdos <loverdos@gmail.com>
 569  */
 570 case class UserStateWorker(userId: String,
 571                            previousResourceEvents: LatestResourceEventsWorker,
 572                            implicitlyIssuedStartEvents: ImplicitlyIssuedResourceEventsWorker,
 573                            ignoredFirstResourceEvents: IgnoredFirstResourceEventsWorker,
 574                            accounting: Accounting,
 575                            resourcesMap: DSLResourcesMap) {
 576
 577   /**
 578    * Finds the previous resource event by checking two possible sources: a) The implicitly terminated resource
 579    * events and b) the explicit previous resource events. If the event is found, it is removed from the
 580    * respective source.
 581    *
 582    * If the event is not found, then this must be for a new resource instance.
 583    * (and probably then some `zero` resource event must be implied as the previous one)
 584    *
 585    * @param resource
 586    * @param instanceId
 587    * @return
 588    */
 589   def findAndRemovePreviousResourceEvent(resource: String, instanceId: String): Maybe[ResourceEvent] = {
 590     // implicitly issued events are checked first
 591     implicitlyIssuedStartEvents.findAndRemoveResourceEvent(resource, instanceId) match {
 592       case just @ Just(_) ⇒
 593         just
 594       case NoVal ⇒
 595         // explicit previous resource events are checked second
 596         previousResourceEvents.findAndRemoveResourceEvent(resource, instanceId) match {
 597           case just @ Just(_) ⇒
 598             just
 599           case noValOrFailed ⇒
 600             noValOrFailed
 601         }
 602       case failed ⇒
 603         failed
 604     }
 605   }
 606
 607   def updateIgnored(resourceEvent: ResourceEvent): Unit = {
 608     ignoredFirstResourceEvents.updateResourceEvent(resourceEvent)
 609   }
 610
 611   def updatePrevious(resourceEvent: ResourceEvent): Unit = {
 612     previousResourceEvents.updateResourceEvent(resourceEvent)
 613   }
 614
 615   def debugTheMaps(clog: ContextualLogger)(rcDebugInfo: ResourceEvent ⇒ String): Unit = {
 616     if(previousResourceEvents.size > 0) {
 617       val map = previousResourceEvents.latestEventsMap.map { case (k, v) => (k, rcDebugInfo(v)) }
 618       clog.debugMap("previousResourceEvents", map, 0)
 619     }
 620     if(implicitlyIssuedStartEvents.size > 0) {
 621       val map = implicitlyIssuedStartEvents.implicitlyIssuedEventsMap.map { case (k, v) => (k, rcDebugInfo(v)) }
 622       clog.debugMap("implicitlyTerminatedResourceEvents", map, 0)
 623     }
 624     if(ignoredFirstResourceEvents.size > 0) {
 625       val map = ignoredFirstResourceEvents.ignoredFirstEventsMap.map { case (k, v) => (k, rcDebugInfo(v)) }
 626       clog.debugMap("ignoredFirstResourceEvents", map, 0)
 627     }
 628   }
 629
 630 //  private[this]
 631 //  def allPreviousAndAllImplicitlyStarted: List[ResourceEvent] = {
 632 //    val buffer: FullMutableResourceTypeMap = scala.collection.mutable.Map[FullResourceType, ResourceEvent]()
 633 //
 634 //    buffer ++= implicitlyIssuedStartEvents.implicitlyIssuedEventsMap
 635 //    buffer ++= previousResourceEvents.latestEventsMap
 636 //
 637 //    buffer.valuesIterator.toList
 638 //  }
 639
 640   /**
 641    * Find those events from `implicitlyIssuedStartEvents` and `previousResourceEvents` that will generate implicit
 642    * end events along with those implicitly issued events. Before returning, remove the events that generated the
 643    * implicit ends from the internal state of this instance.
 644    *
 645    * @see [[gr.grnet.aquarium.logic.accounting.dsl.DSLCostPolicy]]
 646    */
 647   def findAndRemoveGeneratorsOfImplicitEndEvents(newOccuredMillis: Long
 648                                                 ): (List[ResourceEvent], List[ResourceEvent]) = {
 649     val buffer = mutable.ListBuffer[(ResourceEvent, ResourceEvent)]()
 650     val checkSet = mutable.Set[ResourceEvent]()
 651
 652     def doItFor(map: ResourceEvent.FullMutableResourceTypeMap): Unit = {
 653       val resourceEvents = map.valuesIterator
 654       for {
 655         resourceEvent <- resourceEvents
 656         dslResource   <- resourcesMap.findResource(resourceEvent.safeResource)
 657         costPolicy    =  dslResource.costPolicy
 658       } {
 659         if(costPolicy.supportsImplicitEvents) {
 660           if(costPolicy.mustConstructImplicitEndEventFor(resourceEvent)) {
 661             val implicitEnd = costPolicy.constructImplicitEndEventFor(resourceEvent, newOccuredMillis)
 662
 663             if(!checkSet.contains(resourceEvent)) {
 664               checkSet.add(resourceEvent)
 665               buffer append ((resourceEvent, implicitEnd))
 666             }
 667
 668             // remove it anyway
 669             map.remove((resourceEvent.safeResource, resourceEvent.safeInstanceId))
 670           }
 671         }
 672       }
 673     }
 674
 675     doItFor(previousResourceEvents.latestEventsMap)                // we give priority for previous
 676     doItFor(implicitlyIssuedStartEvents.implicitlyIssuedEventsMap) // ... over implicitly issued...
 677
 678     (buffer.view.map(_._1).toList, buffer.view.map(_._2).toList)
 679   }
 680 }
 681
 682 object UserStateWorker {
 683   def fromUserState(userState: UserState, accounting: Accounting, resourcesMap: DSLResourcesMap): UserStateWorker = {
 684     UserStateWorker(
 685       userState.userID,
 686       userState.latestResourceEventsSnapshot.toMutableWorker,
 687       userState.implicitlyIssuedSnapshot.toMutableWorker,
 688       IgnoredFirstResourceEventsWorker.Empty,
 689       accounting,
 690       resourcesMap
 691     )
 692   }
 693 }