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