Is modelling of such collaboration possible using fun-cqrs?

We must remove ForwardingActorSubscriber and make ProjectionActor a ActorSubscriber.

We must be able to configure different passivation strategies for the AggregateActor via configuration.

This was previously possible when declaring aggregate, but since the last refactoring it is not possible anymore because the user facing api is now akka agnostic.

Currently the InMemoryBackend extends Backend[Identity]
Because of that you can not simply replace the AkkaBackend with the InMemoryBackend for testing. You need to write a complicated wrapper that translates the InMemoryAggregateRef[A] into a AggregateRef[A, Future] + quite some other unnecessary boilerplate

It would be nice if I could instantiate a InMemoryBackend[T[_]] by providing a a:A => T[A]

First thanks very much for Fun.CQRS.
I am building my first CQRS based system and am using this Library. It would be great if there was a place to discuss it. I have questions that might not want to be issues. For example I am curious in the Lottery example if you had a requirement that each Game had a unique name how you would add this in?

If you have no plans can I suggest Gitter.

Hi, this is pretty cool, and I just wondered if there were any plans to carry on developing it into a releasable library some time in future?

Hi,

thank you for this great library which makes CQRS/ES development really enjoyable.

I would like to know if you have any plan to upgrade akka-stream-experimental dependency to the latest version (2.0.3) ?

The old akka-stream-experimental (1.0) is not compatible with akka-persistence journal implementations like akka-persistence-jdbc and throws a NoSuchMethodError on ProjectionActor.recoveryCompleted.

Note that the first "non-experimental" version will be included in Akka 2.4.2 (currently in RC)

Regards

A command in Fun.CQRS must extends the DomainCommand trait

trait DomainCommand {
  val id: CommandId = CommandId()
}

And an event must extends DomainEvent:

trait DomainEvent {
  def id: EventId
  def commandId: CommandId
}

The reason why they have an EntityId and a CommandId is to support joining Aggregates and Projections. Which itself is a dubious feature.

This design increases the complexity when defining an Aggregate and is not obvious to developers why they must deal with it.

This should be a option feature. Enabled by means of Facets. For instance:

trait DomainCommand

case class CommandId(value: UUID = UUID.randomUUID())

trait CommandIdFacet {
  def commandId: CommandId
}

trait DomainEvent

case class EventId(value: UUID = UUID.randomUUID())

trait EventIdFacet {
  def eventId: EventId
}

and then...

sealed trait LotteryCommand extends ProtocolCommand 
                                    with CommandIdFacet {
    val commandId = CommandId()
 }

This is a regression bug introduced when fixing #60

We have invokers for command handlers (command side) that can be interpreted depending on the chosen backend (akka, in-memory, etc).

The next step is to refactor the Projection to use the same technique. This will open the door for other backend implementation. For instance, a Slick backend could benefit of invokers for DBIO for both side (command and query side).

InactivityTimeoutPassivationStrategySupport will never work properly.

Aggregates cannot 'kill' themselves based on a timeout because the AggregateManager can't be aware of it and start buffering Commands that arriving while Aggregate is being killed.

As a result, Commands arriving while an Aggregate is killing himself are simply lost.

Play Json extensions should whether be removed or move to an apart project.

As of now, we are not able to move to scala 2.12 because there is no play-json (yet) for 2.12.

This extension is not necessary for Fun.CQRS at all, so should not follow the same release cycle

The current implementation works only with total Functions and it requires a ClassTag.

This approach feels very unnatural to many developers as in general we should always be able to pass a PF where a Function is expected.

For instance, the following code won't compile because of the required ClassTag

 handleCommand {
     case AddName(name) if name.contains("a") => NameWasAdded(...)
     case AddName(name)  => NameWithoutAWasAdded(...)
}

Developers get a compilation error and that's totally unexpected.

This issue depends on #75 because first we need better type inference to be able to mix PartialFunction with InvokerDirectives

Clustering support will fix issue related with timeout and passivation.

A couple of days a go we realized that messages to actors that are being passivated could get timeouts.

I had a discussion on the Lagom gitter channel to understand how they solved it. I thought that they were having the same issue, but they are not.

Below is the copy-n-pase of the Gitter disucssion:

Renato Cavalcanti @rcavalcanti Sep 07 00:16
Hi guys, I have some thoughts and questions about how PersistentEntity passivation works in Lagom.
I understand that messages are buffered and delivered later if they arrive while a PersistentEntity is being passivated. However, what happens if many other messages are arriving for other entities on the same shard region?
The ShardRegion must be notified of its termination before re-instantiate it (which may take some time) and deliver all message in the buffer.
If in the mean time the mailbox of the ShardRegion is saturated with other messages, the Termination message will only be handled once all the other messages are delivered (messages for other actors). That said we may get some 'ask timeout' for the buffered message.
My questions are: does this situation can occur in Lagom? And if so, don't you think is something that should be revised?
Notice that here I'm considering that the 'ask' pattern is being used (which is the most common case in CQRS apps).

James Roper @jroper Sep 08 06:28
I would have thought that that buffer would not prevent other entities from receiving messages
rather the buffer would be per entity

James Roper @jroper Sep 08 06:36
I've just had a look at the source code for cluster sharding, and that's exactly what happens. it's not handled in the ShardRegion actor, it's handled in the Shard actor. When an entity in a shard is passivated, a stop message is sent to that entity, and a buffer is created for any subsequent messages for that entity, that buffer is then put in a map of entity IDs to message buffers. when the shard gets notified that the entity has stopped, if there's any messages in the entity buffer, it restarts the entity and drains the buffer. this has no impact on any other entities running in the shard, or in the shard region.

Renato Cavalcanti @rcavalcanti Sep 08 10:06
Ah that's great. I should have looked deep in the code. Thanks.
I'm facing this problem in Fun.CQRS, but it's not cluster enabled yet. My solution was similar, I was planning to add a Proxy between my AggregateManager and the AggregateActor.
But actually, the actual fix is to add support to clustering

As currently implemented the ProjectionActor will restart always from event 0 which of course is not the desired behavior.

This can be implemented as a trait that defines some basic interface to save and retrieve the offset.

A default implementation can be based on PersistentActor, thus the state of ProjectionActor with PersistentActor is the offset. However, the goal is to have a flexible enough implementation that allow us to define other ways of saving the offset.

When AggregateActor saves a snapshot, it saves it as a Tuple2 with the Actor state and an Aggregate Option. However, when using an alternative serialization format (such as Json with Stamina lib), this means you need to define the serialization not for any concrete class, but for Tuple2. This causes a problem when there is more than one aggregate as it's not possible to differentiate one from the other.

Hope this makes sense! Do we really need to persist the state of the actor with a snapshot? Isn't it always Available after processing a SnapshotOffer?

Issue originaly reported by @vpavkin in Gitter.
...
@rcavalcanti Hello again!
I've been playing around with FunCQRS, it is awesome! I have it working along with https://github.com/dnvriend/akka-persistence-jdbc plugin, and found a small issue I would like to hear your opinion on.
I'm sure there's some idea behind this implementation, so this is more of a "why it's the way it is" question than a bug-report 😄
Code I talk about is this (https://github.com/strongtyped/fun-cqrs/blob/develop/modules/akka/src/main/scala/io/funcqrs/akka/ProjectionActor.scala#L70) :
sourceProvider.source(lastProcessedOffset.map(_ + 1).getOrElse(0)).runWith(actorSink)
The map here treats the offset sequence as 1-based, but the getOrElse clause as zero-based.
akka-persistence-jdbc plugin is inconsistent with that: it has 1-based offset. It works ok with 0, but when you restart the projection, it starts from the N + 1 and misses event N (here I talk about events that were persisted while the projection was off). Some adjustments needed to resolve this inconsistency.
Generally, the offset for eventsByTag is defined by each plugin and it's hard to expect an abstraction to support all the cases. But at least I would expect the abstraction to be consistent between the "first start" and "restart" cases.
Again, I might be missing something here.
Thanks for your help! 😄
...

When I create two back ends in the same actor system as shown here

I get the following exception

After some research for #61, it seems to be possible to remove the Actions method variants handleCommandAsync, tryToHandleCommand and manyEvents.

I was searching for ways to make Actions extensible so users could choose other return types for command handlers.

The new design is inspired by the magnet pattern and previous work from @kiequoo in Fun.CQRS.

Basically we can define actions methods as such:

// simplified signature for concisiness
def handleCommand[C, E, F[_]](cmdHandler: C => F[E])
                             (implicit ivk: InvokerDirective[F]): Actions[Aggregate]

// with InvokerDirective defined as
trait InvokerDirective[-F[_]] {
  def newInvoker[C, E](cmdHandler: (C) => F[E]): CommandHandlerInvoker[C, E]
}

handleCommand method can accept a function from C => F[E] as long as an InvokerDirective is in scope for F[_]. The directive plays the role of a CommandHandlerInvoker factory.

For general usage, the implicit will always be in scope. We can have implicits for Future, Try and Identity in the InvokerDirective companion object.

Extensions can create their own directives and bringing them in scope whenever they need.

Notice that #61 is not yet finished. There other pain points to solve before we can have extensions points in Fun.CQRS. However, the above improvement can be included even if we fail to create a Cats extension.

Projection names are useful for persisting the projection pointer and eventually for join on projections results.

This make them obligatory, however we can fallback to sensible defaults (eg: class.getSimpleName) in order to reduce configuration overhead.

I believe that this merge, which tries to make sure events don't get applied to the state more than once, is partly flawed.

Indeed, the in-memory state will be the correct one, but as far as I can tell, events still get persisted to the persistent backend, and upon replay, will be used to calculate the new state.

In order to avoid this, you need to make sure that your internal Success message is idempotent, which can e.g. be done if you add a version (which could be the total event counter) to your AggregateLike and compare that version number before persisting the events.

We are working on a new DSL to configure behaviors.

The main feature of this new DSL is that it makes it possible to bind the outcome of a Command Validation to an Event handler.

Up to now, we have the following syntax:

behaviorFor[Lottery]
      .whenCreating {
        // creational command and event
        command { cmd: CreateLottery => LotteryCreated(cmd.name, metadata(id, cmd)) }
          .action { evt => Lottery(name = evt.name, id = id) }


      } whenUpdating { lottery =>

        // Select a winner when run!
        command { cmd: Run.type =>
          lottery.selectParticipant().map { winner => WinnerSelected(winner, metadata(id, cmd)) }
        } action { evt =>
          lottery.copy(winner = Option(evt.winner))
        }
      } 
}

CQRSContext.watchEvent will only work if events are processed by a single Projection

If two or more ProjectionActors can consume the same event, than we can never know which one have already processed the events.

Therefore we must be explicit about which projections we want to watch

Eventually, if possible, we could also watch more than one projection at a time.

PersistedOffsetCustom.saveCurrentOffset returns Unit, but should return Future[Unit].

Custom implementations must be able to run db operations async without blocking the ProjectionActor.

Currently we have two kinds of AggregateRefs: AggregateRef itself and AggregateActorRef. However, they don't share the same trait. One could expect that AggregateActorRef extends AggregateRef, but it's not the case.

The reason for that is tell and ask methods (! and ?) have different signatures when using the AkkaBackend. For instance,

In AggregateRef (used by InMemoryBackend)

// F can be Identity, Try or Future
def !(cmd: Command): Unit
def ?(cmd: Command): F[Events]

In AggregateActorRef (used by AkkaBackend)

def !(cmd: Command)(implicit sender: ActorRef = Actor.noSender): Unit
def ?(cmd: Command)(implicit timeout: Timeout, sender: ActorRef = Actor.noSender): Future[Events]

Moreover, the Backend trait doesn't have a aggregateRef method. This method is defined on each Backend implementation and return their specific AggregateRef.

When I was designing the Backends, I played with the idea of having a default timeout for AggregateActorRef. At the end I discarded the idea, but I would like to reconsider it.

This unification proposal will have the following consequences:

1. No ActorRef will be sent anymore, only the timeout. This may sound a bad idea, but it aligns with the more general principle that whatever we do in Fun.CQRS is must be akka agnostic.
2. If timeouts are configured with a default, we should be able use another value and eventually have a method where we could ask for an AggregateRef configured programmatically with another timeout.
   (btw, this approach was adopted by Lagom)
3. the join method is now only available on the AggregateActorRef. We must decide if it's something that we want to add to AggregateRef or if we should keep it specific to the AkkaBackend.

There is a very specific situation where events produced by a single command won't be fully applied.

To illustrate the situation, we could extend the Lottery sample with a create command that adds participant while creating a Lottery.

case class CreateLotteryWithParticipants(name:String, participants: List[String])

This command is supposed to create a new lottery and add each participant sent with the command. If we choose to generate different events (recommended), we end up with the following list of events:

// command send
CreateLottery("demo", List("John", "Paul")

// events emitted
LotteryCreated("demo")
ParticipantAdded("John")
ParticipantAdded("Paul")

It's not possible to add participants while in Initialized state, which is the case when handling create events. We can't add participants in this phase because we need a reference to the lottery aggregate.

The solution is evaluate the behavior PF before each event. If we have a list of Events that we must apply to the aggregate, we must foldLeft on it, but on each pass we must call the behavior PF pick the right actions and apply the event using the actions coming out of the Behavior PF.

When watching events to be applied on Projections, we should be able to identify, in case of timeouts, that the events were effectively applied on the Write Model, but not yet on the Read Model.

In the occurrence of a timeout, we can't know, in the current implementation, if the timeout is on the Read Model or simply because the command failed on the Write Model.

It is currently using EventEnvelope with an offset of type Long whereas in the latest Akka libs, that was deprecated in favor of EventEnvelope2 with an offset of type Offset

In reference to the issue with when handling a command the handle event code gets called twice.

The following details why it occurs:

This happens because after applying a command we apply all the events to the aggregate before sending it to persist but we discard the aggregate state after persisting, we re-apply them without discarding this is done to ensure that no event is persisted if they can’t be handle later
however, we could do it in a more intelligent way we could apply it only once and keep two versions of the aggregate: before and after if we succeed in persisting the events on the store, we drop the old version, we keep the updated one and make the aggregate available to receive the next command.

This was a bad idea. Because of trait initialization order we were always initializing an ActorSystem and eventually overriding it.

We could solve it by making it lazy, but there is little added value to have one initialized by Fun.CQRS itself. Better to let the users pass their on ActorSystem.

If you have a list of events, command monitor only returns when they've all been processed.
Make it possible to return after first (n?) events are processed

Current Behavior force us to always work with a Future.

In most of the case, we don't need a Future, but because of the current design even the tests have to deal with Futures.

If we can abstract over the return type (using Free Monads or a variation of it) we could let it be picked or defined by the chosen backend.

For instance, a in-memory test backend could us Id monad or a Try while the Akka backend a Future.

AggregateRef and AggregateActorRef have a exists() method to check if the aggregate is initialized.

This method is confusing because it doesn't follow the convention in the scala world.

exists should be renamed to isDefined or isInitialized.

and we should have a new method with a predicate.
def exists(pred: A => Boolean): F[Boolean]
Where A is an AggregateLike and F the Functor defined by the Backend.

This bug can happens if a Behavior defines process a command but doesn't define a handler for the related event.

As a result, the event will be persisted, but not applied.

It'll be hard to enforce it at compile time, but we should be able to check it if the Behavior will be able to handle the event before we send it to the event store.

behaviorFor is broken since we made BehaviorDsl builds immutable.

Ideally we should remove the need of writing:
val dsl = new BehaviorDsl[Foo]
import dsl.api._

Hi,

I'm trying to understand all the work behind the AkkaBackend and I'm stuck trying to figure out how ViewBoundedAggregateActorRef can be used.
Could you provide some example on this topic ?

Thanks !

This idea have being around for quite some time. Fun.CQRS make extensive usage of type projections.

Type projections are not ideal because:

• it offers poor type inference in some cases
• it will probably be removed in Scala 3.0 because of unsoundness (read this)

We can easily replace it with path dependent types which will improve a lot the quality of code and reduce its complexity. And at the same time make it much more robust in terms of type inference.

The way to achieve this is explained during my talk at ScalaX 2016.
https://skillsmatter.com/skillscasts/9112-method-reification-and-type-safety-in-a-cqrs-world
(video coming soon)

Slides and code
https://github.com/strongtyped/reifications-in-cqrs

The API will become much lighter as well:

1. No need for ProtocolLike, AggregateLike, DomainCommand and DomainEvent traits.
2. when declaring Actions, we won't need to repeat the type parameter, like in

 actions[Lottery].handleCommand { ... }

This a breaking change, but the impact will be minimal. Existing application will probably need to do some migration.

I noticed something a bit strange when using PersistedOffsetDb. It seems the first event has an offset of 1, second of 2 etc. So once we have processed e.g. two events, it does saveCurrentOffset with a value of 2. If we then shutdown the app and restart, it loads in the persisted offset, which is 2, and starts loading the stream with that offset, which causes it to process the last event with an offset of 2 again. Not really a problem as long as the projection is idempotent put it feels slightly wrong to replay the event again when there was no error.

Not sure of the best fix, though. Or whether it is only LevelDB that works this way (meaning LeveldbReadJournal.eventsByTag(tag, 0) and LeveldbReadJournal.eventsByTag(tag, 1) actually return exactly the same events)

Fun.CQRS works currently only with Akka and the dependency to Akka is explicit.

We would like to introduce the notion of a backend. So we could have different backend implementations: Akka, in-memory (for test), etc.

We'll need to refactor the EventsSourceProvider as it depends on akka-streams api. Instead, it should provide a reactive-stream Publisher

Currently we use LevelDB everywhere. This is suboptimal because it requires some tricks sbt / classpath to make it work during test.

As a result, we don't have yet tests for the actors. If we move to a in-memory plugin we can write decent tests for the actors

Make sure it works with command monitor
If you disallow empty lists, can you make it clear in the api?

If we have a very low Snapshot threshold, snapshot can be saved out of order therefore AggregateActor must wait for the Snapshot confirmation.

Moreover, old Snapshot must be deleted to free up storage.

We will need to adapt:

1. AggregateActor to expect more than one event at creation time
2. ProjectionMonitor return types: we support ProjectionResults for create (1 event) and update (n events). This can be simplified to only one kind of ProjectionResult
3. BehaviorDsl magnets will need to be adapted as well. Eventually, unifying them.
4. BindingDsl also needs to be adapted (let's put it on hold for this as it's not clear yet how it'll evolve).

If applyCommand never succeeds or fails, your AggregateActor is pretty much stuck in a waiting state, unable to handle any new commands.

At the same time, if the applyCommand completes after the ask timeout you have, your ask may fail, but the command is/was still processed.

We solved this by wrapping or applyCommand method in an ExpiringFuture, which will fail with a TimeoutException once its given timeout passed. Maybe something that is useful for you guys as well?

The goal is to have an extension module that we can use Validated or Xor as return time for command handlers.

Things that need to be added

• It should be possible to configure a Backend to use another Interpreter
• The Cats extension must provide its own Backend for testing.
• We should be able to reuse existing Interpreters, for instance: the AsyncInterpreter used by the Akka Backend. If we want Cats support, we need to augment the AsyncInterpreter and pass the augmented version to the Akka Backend

Currently the error message is:

java.util.NoSuchElementException: null [info] at scala.collection.concurrent.TrieMap.apply(TrieMap.scala:837) ~[scala-library-2.11.8.jar:na] [info] at io.funcqrs.akka.backend.AkkaBackend$class.aggregateRef(AkkaBackend.scala:32) ~[fun-cqrs-akka_2.11-0.4.6.jar:0.4.6]