Aquarium

/*

 * Copyright 2011 GRNET S.A. All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or

 * without modification, are permitted provided that the following

 * conditions are met:

 *

 *   1. Redistributions of source code must retain the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer.

 *

 *   2. Redistributions in binary form must reproduce the above

 *      copyright notice, this list of conditions and the following

 *      disclaimer in the documentation and/or other materials

 *      provided with the distribution.

 *

 * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

 * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

 * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

 * POSSIBILITY OF SUCH DAMAGE.

 *

 * The views and conclusions contained in the software and

 * documentation are those of the authors and should not be

 * interpreted as representing official policies, either expressed

 * or implied, of GRNET S.A.

 */

package gr.grnet.aquarium.user.actor

import gr.grnet.aquarium.actor._

import gr.grnet.aquarium.Configurator

import gr.grnet.aquarium.processor.actor._

import com.ckkloverdos.maybe.{Failed, NoVal, Just, Maybe}

import gr.grnet.aquarium.logic.accounting.{AccountingException, Policy, Accounting}

import gr.grnet.aquarium.user._

import gr.grnet.aquarium.logic.events.{UserEvent, WalletEntry, ResourceEvent}

import gr.grnet.aquarium.logic.accounting.dsl.{DSLPolicy, DSLResource, DSLSimpleResource, DSLComplexResource}

import java.util.Date

import gr.grnet.aquarium.util.{DateUtils, TimeHelpers, Loggable}

/**

 *

 * @author Christos KK Loverdos <loverdos@gmail.com>

 */

class UserActor extends AquariumActor

  with Loggable with Accounting with DateUtils {

  @volatile

  private[this] var _userId: String = _

  @volatile

  private[this] var _userState: UserState = _

  @volatile

  private[this] var _timestampTheshold: Long = _

  def role = UserActorRole

  private[this] def _configurator: Configurator = Configurator.MasterConfigurator

  private[this] def processResourceEvent(resourceEvent: ResourceEvent, checkForOlderEvents: Boolean): Unit = {

    if(checkForOlderEvents) {

      DEBUG("Checking for events older than %s" format resourceEvent)

      processOlderResourceEvents(resourceEvent)

    }

    justProcessTheResourceEvent(resourceEvent, "ACTUAL")

  }

  /**

   * Given an "onoff" event, we try to locate all unprocessed resource events that precede this one.

   */

  def findOlderResourceEventsForOnOff(rcEvent: ResourceEvent, policy: DSLPolicy): List[ResourceEvent] = {

    Nil

  }

  def findOlderResourceEventsForOther(rcEvent: ResourceEvent, policy: DSLPolicy): List[ResourceEvent] = {

    Nil

  }

  /**

   * Find resource events that precede the given one and are unprocessed.

   */

  private[this] def findOlderResourceEvents(rcEvent: ResourceEvent, policy: DSLPolicy): List[ResourceEvent] = {

    if(rcEvent.isOnOffEvent(policy)) {

      findOlderResourceEventsForOnOff(rcEvent, policy)

    } else {

      findOlderResourceEventsForOther(rcEvent, policy)

    }

  }

  /**

   * Terminology:

   *

   * - Credits

   *   The analog of money. Credits are the `universal money` within Aquarium.

   *

   * - Resource

   *   A billable/chargeable entity. We generally need credits to use a resource. When a resource is used,

   *   then consume credits. Examples of resources are the `download bandwidth` and, respectively,

   *   the 'upload bandwidth', the `disk space` and the `VM time` to name a few.

   *

   * - User

   *   An owner of resources and credits.

   *

   * - Resource event

   *   An event that is generated by a system, which is responsible for the resource and describes a

   *   state change for the resource. In particular, a resource event records the time when that state

   *   change occurred (`occurredMillis` attribute) and the changed value (`value` attribute).

   *

   * - Resource event store

   *   A datatabase, in the general sense, where resource events are stored.

   *

   * - Cost policy

   *   A cost policy refers to a resource and it is the policy used in order to charge credits for

   *   resource usage.

   *

   * - "On/Off" cost policy

   *   A particular cost policy for resource events that come in "on"/"off" pairs.

   *   In this case the respective resource can only be (from the accounting/billing perspective) in two

   *   states, namely "on" and "off". The respective resource events record the resource changes between

   *   these two states. It is not the responsibility of Aquarium to guarantee correct interleaving of

   *   "on"/"off" resource events.

   *

   * - "On/Off" resource event

   *   A resource event for which the associated resource has an "On/Off" cost policy.

   *

   * - "On" resource event

   *   An "on/off" event which actually records the "on" state for the particular resource it refers to.

   *

   * - "Off" resource event

   *   An "on/off" event which actually records the "off" state for the particular resource it refers to.

   *

   * - Wallet entry

   *   The entity that describes an accounting event and its corresponding credit value.

   *   For example, downloading files uses the download bandwidth and this costs credits. The exact

   *   cost, which is determined using the corresponding cost policy of the event, is recorded

   *   in one or more wallet entries.

   *

   * - Finalized wallet entry

   *   It is a wallet entry whose recorded credits (its credit value) can be taken into account for

   *   billing the respective user (the resource owner).

   *   When we say that a wallet entry is in a finalized state, we mean that it is a finalized wallet entry, as

   *   described above.

   *

   * - Pending or Non-finalized wallet entry

   *   It is a wallet entry whose recorded credits (probably having a zero value) cannot be taken into

   *   account for billing the respective resource owner. Pending wallet entries are introduced to handle

   *   the "on/off" resource events.

   *   When we say that a wallet entry is in/having a pending state, we mean that it is in a pending or

   *   non-finalized wallet entry, as described above.

   *

   * - Wallet store

   *   A database, in the general sense, where wallet entries are stored.

   *

   * - User Bill

   *   A, usually, periodic statement of how many credits the user has consumed.

   *   It may contain detailed analysis that relates consumed credits to resources.

   *

   * - Billable or Chargeable wallet entry

   *   A wallet entry that can participate in the creation of the user Bill.

   *   A wallet entry is billable if and only if it is finalized.

   *

   * - Processed resource event

   *   A resource event which has generated wallet entries.

   *   We may also designate a processed resource event as `partially processed` in order to distinguish it

   *   from a `fully processed` resource event.

   *

   * - Fully processed resource event

   *   A resource event which has no pending wallet entries or has no pending wallet entries with corresponding new

   *   and finalized wallet entries (the meaning of the `corresponding` wallet entries will become

   *   obvious in action A2 below).

   *

   *

   * When an event is processed, there are three possible actions as an outcome, namely A1, A2, A3:

   *

   * - A1. Wallet entries are created. This happens *always*. Each resource event may create more than one

   *       wallet entries. The credit value associated with each wallet entry may be zero credits but we

   *       always generate wallet entries even those with zero credit values.

   *

   *       We say that the event that causes the generation of wallet entries is the `generator` event.

   *

   *       The semantics of wallet entry creation depend on whether the original event is an "on/off" event

   *       or not.

   *

   *       - If the event is an "on/off" event, then

   *         - If it is an "on" event, then one and only one new wallet entry is created having a pending state.

   *           The credit value of this wallet entry is zero and its `occurredMillis` attribute is the same

   *           as the `occurredMillis` attribute of the generator event.

   *

   *           So what we actually do in this case is to record the "on" state with a non-billable wallet entry.

   *           ==> TODO: How is tis related to the (later) expecting "off" state and corresponding (to the "off"

   *               TODO: state) wallet entries?

   *

   *         - If it is an "off" event, then we need to find the corresponding "on" event.

   *           As long as we have the corresponding "on" event, then based on the relative timespan between

   *           the two "on/off" events and the corresponding resource policy we compute a series of *new*

   *           and *finalized* wallet entries. Each new wallet entry has an `occurredMillis` attribute equal to the

   *           `occurredMillis` attribute of the original, pending, wallet entry. We then say that the new

   *           wallet entries _correspond_ to the original, pending wallet entry.

   *           ==> TODO: verify that the semantics of `occurredMillis` for the new wallet entries, as defined

   *               TODO: above, are correct.

   *

   *           ==> TODO: What do we do with the old, pending wallet entry? Do we keep it or do we delete it?

   *

   *           The new wallet entries are generated according to Algorithm W2.

   *

   *       - If the event is not an "on/off" event then, by design, it can directly generate one or more

   *         finalized wallet entries. These wallet entries are generated by Algorithm W1.

   *

   * - A2. As a byproduct of the above, the user credits total is updated.

   *

   *       For this, we just use the chargeable wallet entries created from action A1. Non-chargeable wallet

   *       entries are ignored completely.

   *

   * - A3. Relevant resource state (some value) is updated. For example, the total bandwidth up value is

   *       increased by the amount.

   *

   *       Since a resource event records the changed value for the resource, we can easily update the current

   *       value for the resource.

   *

   * Under ideal circumstances, whenever a resource event arrives, we immediately process it and, so, the steps

   * described in the above actions are what it only takes to get to the new state.

   *

   * But it may so happen that the above event processing procedure may be interrupted. For example, an Aquarium

   * component or external dependency may fail. In such cases, when a resource event arrives we cannot safely assume

   * that it is the most recent unprocessed event. It is then that we have to take action. And since, in general

   * and from the perspective of a newly arriving resource event we are not certain if a failure has happened the

   * following assumption is valid:

   *

   * - For each arriving resource event, it is possible that there exist events that arrived previously but which

   *   have not been partially or fully processed.

   *

   * So, on arrival of a new event, we need to search our event and wallet stores to find those unprocessed events

   * and process them in succession, as if they are newly arrived.

   *

   * In effect and in the most general case, we never process one event at a time but more than one resource

   * events. Their processing order is the ascending order of their `occurredMillis` attribute.

   * ==> TODO: Verify that this semantics of the processing order is correct.

   *

   *

   * We will need the following algorithms:

   *

   * Algorithm W1: Given an non "On/Off" resource event, generate its wallet entries.

   *

   * Algorithm W2: Given an "Off" resource event and the corresponding "On" resource event, generate

   *               the corresponding wallet entries.

   *

   * Algorithm OnOff: Given an "Off" resource event, find its corresponding "On" resource event.

   *

   * Algorithm F: Given a newly arrived resource event, find the exact list of all unprocessed events up to the new

   *              one.

   * Algorithm P: Process a resource event as if it is the most recent unprocessed one.

   *

   * The implementations are as follows:

   *

   * ============

   * Algorithm W1

   * ============

   * - Input

   *   - A non-"on/off" resource event

   *

   * - Output

   *   - The respective wallet entries

   *

   * - Implementation

   *   TODO: This is done in Accounting.chargeEvents

   *

   * ============

   * Algorithm W2

   * ============

   * - Input

   *   - An "off" resource event

   *   - The corresponding "on" resource event

   *

   * - Output

   *   - The respective wallet entries

   *

   * - Implementation

   *   TODO: Trivial

   *

   * ===============

   * Algorithm OnOff

   * ===============

   * - Input

   *

   * ===========

   * Algorithm F

   * ===========

   * Input:  A resource event (e).

   * Output: A list (l) of all unprocessed resource events up to (e).

   *

   *

   *

   */

  private[this] def thisIsJustForTheDoc: Unit = {

  }

  /**

   * Find and process older resource events.

   *

   * Older resource events are found based on the latest credit calculation, that is the latest

   * wallet entry. If there are resource events past that wallet entry, then we deduce that no wallet entries

   * have been calculated for these resource events and start from there.

   */

  private[this] def processOlderResourceEvents(resourceEvent: ResourceEvent): Unit = {

    assert(_userId == resourceEvent.userId)

    val rceId = resourceEvent.id

    val userId = resourceEvent.userId

    val resourceEventStore = _configurator.resourceEventStore

    val walletEntriesStore = _configurator.walletStore

    // 1. Find latest wallet entry

    val latestWalletEntriesM = walletEntriesStore.findLatestUserWalletEntries(userId)

    latestWalletEntriesM match {

      case Just(latestWalletEntries) ⇒

        // The time on which we base the selection of the older events

        val selectionTime = latestWalletEntries.head.occurredMillis

        // 2. Now find all resource events past the time of the latest wallet entry.

        //    These events have not been processed, except probably those ones

        //    that have the same `occurredMillis` with `selectionTime`

        val oldRCEvents = resourceEventStore.findResourceEventsByUserIdAfterTimestamp(userId, selectionTime)

        // 3. Filter out those resource events for which no wallet entry has actually been

        //    produced.

        val rcEventsToProcess = for {

          oldRCEvent        <- oldRCEvents

          oldRCEventId      = oldRCEvent.id

          latestWalletEntry <- latestWalletEntries if(!latestWalletEntry.fromResourceEvent(oldRCEventId) && rceId != oldRCEventId)

        } yield {

          oldRCEvent

        }

        DEBUG("Found %s events older than %s".format(rcEventsToProcess.size, resourceEvent))

        for {

          rcEventToProcess <- rcEventsToProcess

        } {

          justProcessTheResourceEvent(rcEventToProcess, "OLDER")

        }

      case NoVal ⇒

        DEBUG("No events to process older than %s".format(resourceEvent))

      case Failed(e, m) ⇒

        ERROR("[%s][%s] %s".format(e.getClass.getName, m, e.getMessage))

    }

  }

  private[this] def _storeWalletEntries(walletEntries: List[WalletEntry], allowNonFinalized: Boolean): Unit = {

    val walletEntriesStore = _configurator.walletStore

    for(walletEntry <- walletEntries) {

      val allow = walletEntry.finalized || allowNonFinalized

      if(allow) {

        walletEntriesStore.storeWalletEntry(walletEntry)

      }

    }

  }

  private[this] def _calcNewCreditSum(walletEntries: List[WalletEntry]): Double = {

    val newCredits = for {

      walletEntry <- walletEntries if(walletEntry.finalized)

    } yield walletEntry.value.toDouble

    newCredits.sum

  }

  /**

   * Process the resource event as if nothing else matters. Just do it.

   */

  private[this] def justProcessTheResourceEvent(ev: ResourceEvent, logLabel: String): Unit = {

    val start = System.currentTimeMillis

    DEBUG("Processing [%s] %s".format(logLabel, ev))

    // Initially, the user state (regarding resources) is empty.

    // So we have to compensate for both a totally empty resource state

    // and the update with new values.

    // 1. Find the resource definition

    val policy = Policy.policy

    policy.findResource(ev.resource) match {

      case Some(resource) ⇒

        // 2. Get the instance id and value for the resource

        val instanceIdM = resource match {

          // 2.1 If it is complex, from the details map, get the field which holds the instanceId

          case DSLComplexResource(name, unit, costPolicy, descriminatorField) ⇒

            ev.details.get(descriminatorField) match {

              case Some(instanceId) ⇒

                Just(instanceId)

              case None ⇒

                // We should have some value under this key here....

                Failed(throw new AccountingException("")) //TODO: See what to do here

            }

          // 2.2 If it is simple, ...

          case DSLSimpleResource(name, unit, costPolicy) ⇒

            // ... by convention, the instanceId of a simple resource is just "1"

            // @see [[gr.grnet.aquarium.user.OwnedResourcesSnapshot]]

            Just("1")

        }

        // 3. Did we get a valid instanceId?

        instanceIdM match {

          // 3.1 Yes, time to get/update the current state

          case Just(instanceId) ⇒

            val oldOwnedResources = _userState.ownedResources

            // A. First state diff: the modified resource value

            val StateChangeMillis = TimeHelpers.nowMillis

            val (newOwnedResources, oldRCInstanceOpt, newRCInstance) = oldOwnedResources.

              addOrUpdateResourceSnapshot(resource.name, instanceId, ev.value, ev.occurredMillis)

            val previousRCUpdateTime = oldRCInstanceOpt.map(_.snapshotTime).getOrElse(newRCInstance.snapshotTime)

            // Calculate the wallet entries generated from this resource event

            _userState.maybeDSLAgreement match {

              case Just(agreement) ⇒

                val walletEntriesM = chargeEvent(ev, agreement, ev.value,

                  new Date(previousRCUpdateTime),

                  findRelatedEntries(resource, ev.getInstanceId(policy)))

                walletEntriesM match {

                  case Just(walletEntries) ⇒

                    _storeWalletEntries(walletEntries, true)

                    // B. Second state diff: the new credits

                    val newCreditsSum = _calcNewCreditSum(walletEntries)

                    val oldCredits    = _userState.safeCredits.data

                    val newCredits = CreditSnapshot(oldCredits + newCreditsSum, StateChangeMillis)

                    // Finally, the userState change

                    DEBUG("Credits   = %s".format(this._userId, newCredits))

                    DEBUG("Resources = %s".format(this._userId, newOwnedResources))

                    this._userState = this._userState.copy(

                      credits = newCredits,

                      ownedResources = newOwnedResources

                    )

                  case NoVal ⇒

                    DEBUG("No wallet entries generated for %s".format(ev))

                  case failed @ Failed(e, m) ⇒

                    failed

                }

              case NoVal ⇒

                Failed(new UserDataSnapshotException("No agreement snapshot found for user %s".format(this._userId)))

              case failed @ Failed(e, m) ⇒

                failed

            }

          // 3.2 No, no luck, this is an error

          case NoVal ⇒

            Failed(new UserDataSnapshotException("No instanceId for resource %s of user %s".format(resource, this._userId)))

          case failed @ Failed(e, m) ⇒

            failed

        }

      // No resource definition found, this is an error

      case None ⇒ // Policy.policy.findResource(ev.resource)

        Failed(new UserDataSnapshotException("No resource %s found for user %s".format(ev.resource, this._userId)))

    }

    DEBUG("Finished %s time: %d ms".format(ev.id, System.currentTimeMillis - start))

  }

  private[this] def processCreateUser(event: UserEvent): Unit = {

    val userId = event.userId

    DEBUG("Creating user from state %s", event)

    val usersDB = _configurator.storeProvider.userStateStore

    usersDB.findUserStateByUserId(userId) match {

      case Just(userState) ⇒

        WARN("User already created, state = %s".format(userState))

      case failed @ Failed(e, m) ⇒

        ERROR("[%s][%s] %s", e.getClass.getName, e.getMessage, m)

      case NoVal ⇒

        // OK. Create a default UserState and store it

        val now = TimeHelpers.nowMillis

        val agreementOpt = Policy.policy.findAgreement("default")

        if(agreementOpt.isEmpty) {

          ERROR("No default agreement found. Cannot initialize user state")

        } else {

          this._userState = UserState(

            userId,

            ActiveSuspendedSnapshot(event.isStateActive, now),

            CreditSnapshot(0, now),

            AgreementSnapshot(agreementOpt.get.name, now),

            RolesSnapshot(event.roles, now),

            PaymentOrdersSnapshot(Nil, now),

            OwnedGroupsSnapshot(Nil, now),

            GroupMembershipsSnapshot(Nil, now),

            OwnedResourcesSnapshot(List(), now)

          )

          saveUserState

          DEBUG("Created and stored %s", this._userState)

        }

    }

  }

  private[this] def findRelatedEntries(res: DSLResource, instid: String): List[WalletEntry] = {

    val walletDB = _configurator.storeProvider.walletEntryStore

    walletDB.findPreviousEntry(_userId, res.name, instid, Some(false))

  }

  private[this] def processModifyUser(event: UserEvent): Unit = {

    val now = TimeHelpers.nowMillis

    val newActive = ActiveSuspendedSnapshot(event.isStateActive, now)

    DEBUG("New active status = %s".format(newActive))

    this._userState = this._userState.copy( active = newActive )

  }

  /**

   * Use the provided [[gr.grnet.aquarium.logic.events.UserEvent]] to change any user state.

   */

  private[this] def processUserEvent(event: UserEvent): Unit = {

    if(event.isCreateUser) {

      processCreateUser(event)

    } else if(event.isModifyUser) {

      processModifyUser(event)

    }

  }

  /**

   * Try to load from the DB the latest known info (snapshot data) for the given user.

   */

  private[this] def findUserState(userId: String): Maybe[UserState] = {

    val usersDB = _configurator.storeProvider.userStateStore

    usersDB.findUserStateByUserId(userId)

  }

  /**

   * Tries to makes sure that the internal user state exists.

   *

   * May contact the [[gr.grnet.aquarium.store.UserStateStore]] for that.

   *

   */

  private[this] def ensureUserState(): Unit = {

    if(null eq this._userState) {

      findUserState(this._userId) match {

        case Just(userState) ⇒

          DEBUG("Loaded user state %s from DB", userState)

          //TODO: May be out of sync with the event store, rebuild it here

          this._userState = userState

          rebuildState(this._userState.oldestSnapshotTime)

        case Failed(e, m) ⇒

          ERROR("While loading user state from DB: [%s][%s] %s", e.getClass.getName, e.getMessage, m)

        case NoVal ⇒

          //TODO: Rebuild actor state here.

          rebuildState(0)

          WARN("Request for unknown (to Aquarium) user")

      }

    }

  }

  /**

   * Replay the event log for all events that affect the user state, starting

   * from the provided time instant.

   */

  def rebuildState(from: Long): Unit =

    rebuildState(from, oneYearAhead(new Date(), new Date(Long.MaxValue)).getTime)

  /**

   * Replay the event log for all events that affect the user state.

   */

  def rebuildState(from: Long, to: Long): Unit = {

    val start = System.currentTimeMillis()

    if (_userState == null)

      createBlankState

    //Rebuild state from user events

    val usersDB = _configurator.storeProvider.userEventStore

    val userEvents = usersDB.findUserEventsByUserId(_userId)

    val numUserEvents = userEvents.size

    _userState = replayUserEvents(_userState, userEvents, from, to)

    //Rebuild state from resource events

    val eventsDB = _configurator.storeProvider.resourceEventStore

    val resourceEvents = eventsDB.findResourceEventsByUserIdAfterTimestamp(_userId, from)

    val numResourceEvents = resourceEvents.size

    _userState = replayResourceEvents(_userState, resourceEvents, from, to)

    //Rebuild state from wallet entries

    val wallet = _configurator.storeProvider.walletEntryStore

    val walletEnties = wallet.findUserWalletEntriesFromTo(_userId, new Date(from), new Date(to))

    val numWalletEntries = walletEnties.size

    _userState = replayResourceEvents(_userState, resourceEvents, from, to)

    INFO(("Rebuilt state from %d events (%d user events, " +

      "%d resource events, %d wallet entries) in %d msec").format(

      numUserEvents + numResourceEvents + numWalletEntries,

      numUserEvents, numResourceEvents, numWalletEntries,

      System.currentTimeMillis() - start))

  }

  /**

   * Create an empty state for a user

   */

  def createBlankState = {

    val now = 0

    val agreement = Policy.policy.findAgreement("default")

    this._userState = UserState(

      _userId,

      ActiveSuspendedSnapshot(false, now),

      CreditSnapshot(0, now),

      AgreementSnapshot(agreement.get.name, now),

      RolesSnapshot(List(), now),

      PaymentOrdersSnapshot(Nil, now),

      OwnedGroupsSnapshot(Nil, now),

      GroupMembershipsSnapshot(Nil, now),

      OwnedResourcesSnapshot(List(), now)

    )

  }

  /**

   * Replay user events on the provided user state

   */

  def replayUserEvents(initState: UserState, events: List[UserEvent],

                       from: Long, to: Long): UserState = {

    var act = initState.active

    var rol = initState.roles

    events

      .filter(e => e.occurredMillis >= from && e.occurredMillis < to)

      .foreach {

        e =>

          act = act.copy(

            data = e.isStateActive, snapshotTime = e.occurredMillis)

          // TODO: Implement the following

          //_userState.agreement = _userState.agreement.copy(

          //  data = e.newAgreement, e.occurredMillis)

          rol = rol.copy(data = e.roles,

            snapshotTime = e.occurredMillis)

    }

    initState.copy(active = act, roles = rol)

  }

  /**

   * Replay resource events on the provided user state

   */

  def replayResourceEvents(initState: UserState, events: List[ResourceEvent],

                           from: Long, to: Long): UserState = {

    var res = initState.ownedResources

    events

      .filter(e => e.occurredMillis >= from && e.occurredMillis < to)

      .foreach {

        e =>

          val name = Policy.policy.findResource(e.resource) match {

            case Some(x) => x.name

            case None => ""

          }

          val instanceId = e.getInstanceId(Policy.policy)

          res = res.addOrUpdateResourceSnapshot(name,

            instanceId, e.value, e.occurredMillis)._1

    }

    initState.copy(ownedResources = res)

  }

  /**

   * Replay wallet entries on the provided user state

   */

  def replayWalletEntries(initState: UserState, events: List[WalletEntry],

                          from: Long, to: Long): UserState = {

    var cred = initState.credits

    events

      .filter(e => e.occurredMillis >= from && e.occurredMillis < to)

      .foreach {

        w => cred = cred.copy(data = w.value, snapshotTime = w.occurredMillis)

    }

    initState.copy(credits = cred)

  }

  /**

   * Persist current user state

   */

  private[this] def saveUserState(): Unit = {

    _configurator.storeProvider.userStateStore.deleteUserState(this._userId)

    _configurator.storeProvider.userStateStore.storeUserState(this._userState) match {

      case Just(record) => record

      case NoVal => ERROR("Unknown error saving state")

      case Failed(e, a) =>

        ERROR("Saving state failed: %s error was: %s".format(a,e));

    }

  }

  protected def receive: Receive = {

    case m @ AquariumPropertiesLoaded(props) ⇒

      this._timestampTheshold = props.getLong(Configurator.Keys.user_state_timestamp_threshold).getOr(10000)

      INFO("Setup my timestampTheshold = %s", this._timestampTheshold)

    case m @ UserActorInitWithUserId(userId) ⇒

      this._userId = userId

      DEBUG("Actor starting, loading state")

      ensureUserState()

    case m @ ProcessResourceEvent(resourceEvent) ⇒

      if(resourceEvent.userId != this._userId) {

        ERROR("Received %s but my userId = %s".format(m, this._userId))

      } else {

        ensureUserState()

        processResourceEvent(resourceEvent, true)

      }

    case m @ ProcessUserEvent(userEvent) ⇒

      if(userEvent.userId != this._userId) {

        ERROR("Received %s but my userId = %s".format(m, this._userId))

      } else {

        ensureUserState()

        processUserEvent(userEvent)

      }

    case m @ RequestUserBalance(userId, timestamp) ⇒

      if(this._userId != userId) {

        ERROR("Received %s but my userId = %s".format(m, this._userId))

        // TODO: throw an exception here

      } else {

        // This is the big party.

        // Get the user state, if it exists and make sure it is not stale.

        ensureUserState()

        // Do we have a user state?

        if(_userState ne null) {

          // Yep, we do. See what there is inside it.

          val credits = _userState.credits

          val creditsTimestamp = credits.snapshotTime

          // Check if data is stale

          if(creditsTimestamp + _timestampTheshold > timestamp) {

            // No, it's OK

            self reply UserResponseGetBalance(userId, credits.data)

          } else {

            // Yep, data is stale and must recompute balance

            // FIXME: implement

            ERROR("FIXME: Should have computed a new value for %s".format(credits))

            self reply UserResponseGetBalance(userId, credits.data)

          }

        } else {

          // No user state exists. This is an error.

          val errMsg = "Could not load user state for %s".format(m)

          ERROR(errMsg)

          self reply ResponseUserBalance(userId, 0, Some(errMsg))

        }

      }

    case m @ UserRequestGetState(userId, timestamp) ⇒

      if(this._userId != userId) {

        ERROR("Received %s but my userId = %s".format(m, this._userId))

        // TODO: throw an exception here

      } else {

        // FIXME: implement

        ERROR("FIXME: Should have properly computed the user state")

        ensureUserState()

        self reply UserResponseGetState(userId, this._userState)

      }

  }

  override def postStop {

    DEBUG("Stopping, saving state")

    saveUserState

  }

  private[this] def DEBUG(fmt: String, args: Any*) =

    logger.debug("UserActor[%s]: %s".format(_userId, fmt.format(args:_*)))

  private[this] def INFO(fmt: String, args: Any*) =

      logger.info("UserActor[%s]: %s".format(_userId, fmt.format(args:_*)))

  private[this] def WARN(fmt: String, args: Any*) =

    logger.warn("UserActor[%s]: %s".format(_userId, fmt.format(args:_*)))

  private[this] def ERROR(fmt: String, args: Any*) =

    logger.error("UserActor[%s]: %s".format(_userId, fmt.format(args:_*)))

}