Polymorphic support? Newtonsoft support?

[jhoward321]

I'm trying to implement an event sourcing event store using this library and came across a few things I was hoping someone might help me answer. I would like to have a stream of events for a particular interface and have the ability to correctly deserialize those objects to their correct implementation. My plan was to use a custom system.text.json converter, to add a type field on serialization which would be used on deserialization to deserialize to the correct derived type. Basically the same concept as TypeNameHandling.All in newtonsoft. Which brings me to my first question - is it possible to use newtonsoft instead of system.text.json?

I have defined a stream that has all the permutations of possible columns, including the ClassName field. I think using the json converter I can get it to deserialize properly, but the issue I'm having is actually on save. I have a list of IDomainEvents which I save to the stream with this code:

foreach (var domainEvent in aggregateRoot.Events)
{
    _ksqlDbContext.Add(domainEvent, _insertProperties);
}

await _ksqlDbContext.SaveChangesAsync(cancellationToken);

However when it gets persisted, only the columns on the IDomainEvent interface are persisted. It never goes through my json converter until it deserializes. Is there a way to write the actual instance instead of just the IDomainEvent fields? I could do it through a kafka producer, but I was hoping to do it through this library.

Example stream schema:

CREATE OR REPLACE STREAM IF NOT EXISTS Link_events(
    ClassType VARCHAR,
    Name VARCHAR, 
    SendSeparate BOOLEAN, 
    Version BIGINT, 
    AggregateRootId VARCHAR KEY, 
    ExtraField VARCHAR,
    OccurredAt TIMESTAMP) 
WITH (KAFKA_TOPIC='Link_events',VALUE_FORMAT='Json',PARTITIONS='3',REPLICAS='1');

Example inheritance structure:

public interface IDomainEvent
{
    public long Version { get; }

    public Guid AggregateRootId { get; }
    public DateTimeOffset OccurredAt { get; }
}

public abstract record BaseDomainEvent<T>: IDomainEvent where T : IAggregateRoot
{
    /// <summary>
    /// for deserialization
    /// </summary>
    public BaseDomainEvent() { }

    protected BaseDomainEvent(T aggregateRoot)
    {
        if(aggregateRoot is null)
            throw new ArgumentNullException(nameof(aggregateRoot));

        this.Version = aggregateRoot.Version;
        this.AggregateRootId = aggregateRoot.Id;
        this.OccurredAt = DateTimeOffset.UtcNow;
    }

    [JsonInclude]
    public long Version { get; set; }

    [Key]
    [JsonInclude]
    public Guid AggregateRootId { get; set; }
    [JsonInclude]
    public DateTimeOffset OccurredAt { get; set; }
}

public record LinkCreated : BaseDomainEvent<Link>
    {
        [JsonConstructor]
        public LinkCreated(){}

        public LinkCreated(Link link, string name, bool sendSeparate) : base(link)
        {
            if (link is null)
                throw new ArgumentNullException(nameof(link));
            Name = name;
            SendSeparate = sendSeparate;
        }
        [JsonInclude]
        public string Name { get; set; }

        [JsonInclude]
        public bool SendSeparate { get; set; }
    }

    public record NameUpdated : BaseDomainEvent<Link>
    {
        [JsonConstructor]
        public NameUpdated(){}

        public NameUpdated(Link link, string name) : base(link)
        {
            if (link is null)
                throw new ArgumentNullException(nameof(link));

            if (string.IsNullOrEmpty(name))
                throw new ArgumentNullException(nameof(name));

            Name = name;
        }

        [JsonInclude]
        public string Name { get; set; }

        [JsonInclude]
        public string ExtraField = "this should be serialized";
    }

[tomasfabian]

Hi @jhoward321,
ksqlDb.RestApi.Client internally uses System.Text.Json and Newtonsoft is not supported.

In your case it shouldn't matter because the insertion of records ( _ksqlDbContext.Add) uses reflection and JSON deserialization is not used here.

I spotted a possible issue. In your case you are using a record field (named ExtraField) so I added support for them too. The library extracted values only from properties up to this time.

By default the insert statement is generated from public instance props and fields.
In order to exclude a property you can define the IgnoreByInsertsAttribute:

public class Movie
{
  [ksqlDB.RestApi.Client.KSql.RestApi.Statements.Annotations.IgnoreByInserts]
  public int IgnoredProperty { get; set; }
}

You can try to explore the generated insert statement as described here.

I published the package with the above mentioned improvement:

Install-Package ksqlDb.RestApi.Client -Version 2.4.0-rc.2

Let me know if it helped you to progress with your PoC. In case that it didn't help I would like to ask you to extend your example with the underlying class for aggregateRoot.Events and an example instance.
Output from the IKSqlDbRestApiClient.ToInsertStatement and an expected INSERT INTO statement could be also helpful in order to solve your issue.

Thank you very much!

Regards Tomas.

[jhoward321]

Thanks for your reply @tomasfabian. So here's an example generated insert statement:

INSERT INTO Link_events (Version, AggregateRootId, OccurredAt) VALUES (0, '51a589e2-3a89-42ab-ae51-f93a1827df85', '2022-09-26T17:43:57.962+00:00');

AggregateRoot.Events has the following type: public IReadOnlyCollection<IDomainEvent> Events.

The insert statement above is what's generated for a LinkCreated event, which should also be inserting a name value and a send separate value. Since I am accessing a collection of IDomainEvents, only the fields present on the IDomainEvent interface are persisted. I have an option of mapping to a flattened DTO with all of the fields, but then I can't easily take advantage of generic typing. Basically I just want to be able to customize how my message gets serialized. I want to serialize everything on the instance - not just what's on the interface.

Since the custom json converter works for deserializing, I suppose I could insert directly to the underlying topic so I could have full control over the serialization.

Essentially what I want is polymorphic writes - I have figured out the reads. So serialize what's on the instance not just dealing with the common interface. I understand this might not be a common use case, but it would be very useful for our usecase.

[jhoward321]

I can see exactly where it's coming from. In CreateInsert here it iterates through the members it gets from here.

The members function is doing typeof(T) which in this instance is IDomainEvent. Ideally for my usecase the type would be entity.GetType(). Maybe there could be an option in InsertProperties which uses instance type instead of typeof(T) which will always be the interface rather than instance type.

[tomasfabian]

Thank you @jhoward321 for the details. Now I understand what was the missing part in my understanding. Could you please review this commit, please?

The generated Insert statement in the provided unit test is the following:

INSERT INTO MyUpdate (Field, Field2, ExtraField) VALUES ('Value', 'Value2', 'Test value');

This should cover your use case where you are providing an interface.

[jhoward321]

That definitely looks like it will work for my usecase. Wouldn't that be a breaking change though? I think it probably makes sense to have a flag in InsertProperties to enable that behavior. Otherwise it might break existing workflows for other users

[jhoward321]

I'm going to try the rc3 right now

[tomasfabian]

Good point regarding the breaking change. Thx for mentioning it.

Is InsertProperties.UseEntityType a good name for the new setting? Or can you come up with a better name, please?

[jhoward321]

I think InsertProperties.UseInstanceType might be better personally but I think either is fine.

[tomasfabian]

Here is the commit log from our new iteration.

In case that you will confirm me that the rc-3 works properly I will publish you a release. It will contain also the InsertProperties.UseInstanceType config.

[jhoward321]

RC3 is working and publishing all of the fields I expect. Thanks a ton for your help!

[tomasfabian]

You are welcome @jhoward321 and thank you for the provided improvements, too!

I released version 2.4.0. Please mark the questions #32 and #33 as answered if everything is ok.

Happy coding ;)

Regards Tomas