Changing event data before passing to aggregate's Apply method

[zkbpkp]

I'm trying to implement GDPR-compliant personal data storage using Marten. To do so I want to encrypt personal data before saving it to event store and decrypt it before passing to aggregate's Apply method. I wasn't sure if there is some out of the box way to do something like this so I just copied AggregationProjection and added an additional parameter which is a function that changes event data. The result looks like this:

InterceptingProjection.cs

public class InterceptingProjection<T> : DocumentProjection<T>, IDocumentProjection, IProjection
    where T : class, new()
{
    private IAggregationFinder<T> Finder { get; }
    private IAggregator<T> Aggregator { get; }
    private Func<object, object> Interceptor { get; }

    public InterceptingProjection(IAggregationFinder<T> finder, IAggregator<T> aggregator, Func<object, object> interceptor)
    {
        Finder = finder;
        Aggregator = aggregator;
        Interceptor = interceptor;
    }

    public void Apply(IDocumentSession session, EventPage page)
    {
        MatchingStreams(page).Each(stream =>
        {
            var state = Finder.Find(stream, session);
            Update(state, stream);
            session.Store(state);
        });
    }

    private void Update(T state, EventStream stream)
    {
        stream.Events.Select(e => (IEvent) new Event<object>(Interceptor(e.Data))
        {
            StreamId = e.StreamId,
            StreamKey = e.StreamKey,
            Id = e.Id,
            Version = e.Version,
            Sequence = e.Sequence,
            Timestamp = e.Timestamp,
            TenantId = e.TenantId,
        }).Each(x => x.Apply(state, Aggregator));
    }

    public async Task ApplyAsync(
        IDocumentSession session,
        EventPage page,
        CancellationToken token)
    {
        var matchingStreams = MatchingStreams(page);
        await Finder.FetchAllAggregates(session, matchingStreams, token).ConfigureAwait(false);
        var eventStreamArray = matchingStreams;
        foreach (var stream in eventStreamArray)
        {
            var state = await Finder.FindAsync(stream, session, token).ConfigureAwait(false) ?? new T();
            Update(state, stream);
            session.Store(state);
        }
    }

    public Type[] Consumes => Aggregator.EventTypes;

    public EventStream[] MatchingStreams(EventPage streams) =>
        streams.Streams
            .Where(Aggregator.AppliesTo)
            .ToArray();

    public AsyncOptions AsyncOptions { get; } = new AsyncOptions();
}

The problem is that adding an inline aggregation projection to a projection collection relies on some private API that I cannot access from my extension method:

ProjectionCollectionExtensions.cs

public static class ProjectionCollectionExtensions
{
    public static InterceptingProjection<T> AggregateStreamsWithInterceptor<T>(this ProjectionCollection collection,
        Func<object, object> interceptor) where T : class, new()
    {
        // ERROR: Cannot access private field '_options' here:
        IAggregator<T> aggregator = collection._options.Events.AggregateFor<T>();
        var projection = new InterceptingProjection<T>(new AggregateFinder<T>(), aggregator, interceptor);
        collection.Add(projection);
        return projection;
    }
}

And I don't know if there's any other way to get a proper IAggregator<T> instance from my extension method.

[jeremydmiller]

@KolesnichenkoDS You could achieve this through an IDocumentSessionListener. You'd get a chance to look at the events and potentially modify them just prior to the transaction being processed

[zkbpkp]

@jeremydmiller thanks for the quick response!

I'm not sure if that solves my issue. Please correct me if I'm wrong, but it seems to me that using IDocumentSessionListener I can intercept before the events are written to the db, not after they are read from the db / before they are applied to an aggregate. So it seems that well maybe using that I can actually encrypt an event in BeforeSaveChanges and not sure if an event will be applied to an aggregate before or after that, but anyway if I try to rebuild the inline projection then I still have these events that are applied to an aggregate in an encrypted form.

UPD: Just to clarify, I don't want to automatically encrypt events, I want to pass an already encrypted event to Marten and automatically decrypt them before calling Apply on event, so I don't need to include all this encryption stuff in my domain layer.

[Burwin]

+1

I have a similar need, but I'm using Inline Projections (currently).

• So I could use IDocumentSessionListener to encrypt any events in BeforeSaveChanges, which is good.
• It looks like BeforeSaveChanges is called after applyProjections in DocumentSession, which is good in that my inline-projected aggregate will correctly apply unencrypted events
• Additionally, I can further encrypt any documents, as necessary, in BeforeSaveChanges, which can handle the recently inline-projected aggregate.

However, as the author mentions, I can't find any hook where I can reload the event stream and rebuild the aggregate from unencrypted events... at least not anything very "built-in". I'm still fairly new to Marten, so I could be missing something.

Is this something that could/should be done via the json serializer during deserialization? That feels smelly to me, but maybe that works.

[Burwin]

@KolesnichenkoDS

Shameless plug (I hope this isn't bad etiquette), but I threw together a library for encrypting/decrypting at the field/property level during JSON serialization/deserialization: https://github.com/Burwin/JsonCryption

I'm not sure it solves all your issues, particularly because it sounds like you aren't using Marten to encrypt your events. I don't know how your project is set up, but if it is possible to leave all encryption/decryption to your JsonSerializer within Marten, then this might help. I haven't tested it with anything other than Inline Projections:

• instantiate an event-sourced aggregate
• persist via Marten (with inline projections for the aggregate configured)
• verify fields and properties that should be encrypted are encrypted in the data store, for projected aggregates and events
• query aggregate from Marten
• verify fields and properties of aggregate are decrypted correctly

As you can see, I haven't tested the events themselves, but since they're encrypted/decrypted on serialization/deserialization, I'd expect they wouldn't be an issue (famous last words...).