Evject.pm6

class Domain { ... }

class Aggregate {
    has $.id;
    
    trusts Domain;
    method !apply-events($id, @events) {
        $!id = $id;
        self.apply($_) for @events;
    }

    proto method apply($) {*}
}

role EventStore {
    method load($type, $id) { ... }
    method store($type, $id, @new-events, $base-version) { ... }
}

class X::EventStore::Conflict is Exception {
}

class Domain {
    has EventStore $.event-store = die "Must supply an event store";
    has $.attempts = 2;

    method process(Aggregate $type, $id, &call) {
        my $attempts-remaining = $.attempts;
        while $attempts-remaining-- > 0 {
            my $agg = $type.new();
            my @events = $.event-store.load($type, $id);
            $agg!Aggregate::apply-events($id, @events);
            my @new-events = call($agg);
            $.event-store.store($type, $id, @new-events, @events.elems);
            return;
            CATCH { when X::EventStore::Conflict { $attempts-remaining++; } }
        }
    }
}

InMemoryEventStore.pm6

use Evject;
use Monitor;

monitor InMemoryEventStore does EventStore {
    has %!streams;

    method load($type, $id) {
        my @copy = @(%!streams{aggregate-key($type, $id)} // ());
        return @copy;
    }
    
    method store($type, $id, @new-events, $base-version) {
        my $key = aggregate-key($type, $id);
        my @current := (%!streams{$key} //= @[]);
        if @current.elems != $base-version {
            X::EventStore::Conflict.throw;
        }
        push @current, @new-events;
    }
    
    sub aggregate-key($type, $id) {
        $type.^name ~ $id
    }
}