feat: pluggable binary event serialization (IEventBinarySerializer) (#388)

src/Polecat/Events/BinaryEventAttribute.cs

@@ -0,0 +1,28 @@
+namespace Polecat.Events;
+
+/// <summary>
+///     Marks an event type as binary-serialized: its <c>pc_events.data</c> column holds the
+///     <c>'{}'</c> placeholder and the real payload lives in <c>bdata</c>, written and read by an
+///     <see cref="IEventBinarySerializer" />. Mirrors Marten's <c>[BinaryEvent]</c>
+///     (<see href="https://github.com/JasperFx/marten/issues/4515" />); tracked as
+///     <see href="https://github.com/JasperFx/polecat/issues/388">polecat#388</see>.
+/// </summary>
+/// <remarks>
+///     <para>
+///         The serializer for an attribute-marked type is the store-wide
+///         <c>opts.Events.DefaultBinarySerializer</c>. An explicit
+///         <c>opts.Events.UseBinarySerializer&lt;TEvent&gt;(serializer)</c> registration takes
+///         precedence. If a type is attribute-marked but neither is configured, the store throws when
+///         it first resolves that event type rather than silently writing JSON — a silent fallback
+///         would produce a store whose write amplification quietly does not match its configuration.
+///     </para>
+///     <para>
+///         JSON and binary events coexist per event type in the same table, so applying this to a
+///         single event type is a safe in-place change: existing JSON rows keep <c>bdata = NULL</c>
+///         and keep reading through the JSON path.
+///     </para>
+/// </remarks>
+[AttributeUsage(AttributeTargets.Class | AttributeTargets.Struct, Inherited = false)]
+public sealed class BinaryEventAttribute : Attribute
+{
+}

src/Polecat/Events/Daemon/PolecatEventLoader.cs

@@ -80,7 +80,7 @@ private async Task<EventPage> LoadInternalAsync(EventRequest request, Cancellati
         await using var cmd = conn.CreateCommand();
         cmd.CommandText = $"""
             SELECT TOP(@batchSize) seq_id, id, stream_id, version, data, type, timestamp,
-                tenant_id, dotnet_type, is_archived
+                tenant_id, dotnet_type, is_archived, bdata
             FROM {_events.EventsTableName}
             WHERE seq_id > @floor AND seq_id <= @ceiling AND is_archived = 0{tenantPredicate}
             ORDER BY seq_id;
@@ -106,6 +106,8 @@ SELECT TOP(@batchSize) seq_id, id, stream_id, version, data, type, timestamp,
             var tenantId = reader.GetString(7);
             var dotNetTypeName = reader.IsDBNull(8) ? null : reader.GetString(8);
             var isArchived = reader.GetBoolean(9);
+            // #388: non-null bdata means this row's payload is binary, not JSON.
+            var bdata = reader.IsDBNull(10) ? null : reader.GetFieldValue<byte[]>(10);
 
             // Apply event type allow-list filter (skip events not in the subscription's filter)
             if (_allowedDotNetTypes != null && dotNetTypeName != null &&
@@ -133,7 +135,7 @@ SELECT TOP(@batchSize) seq_id, id, stream_id, version, data, type, timestamp,
             object data;
             try
             {
-                data = _options.Serializer.FromJson(resolvedType, json);
+                data = _events.DeserializeEventData(resolvedType, json, bdata, _options.Serializer);
             }
             catch (Exception ex)
             {

src/Polecat/Events/EventGraph.cs

@@ -440,6 +440,118 @@ public ITagTypeRegistration RegisterTagType<TTag>(string tableSuffix) where TTag
 
     public IReadOnlyList<ITagTypeRegistration> TagTypes => _tagTypes;
 
+    // ---- #388: pluggable binary event serialization ------------------------------------------
+    //
+    // Explicit per-type registrations from UseBinarySerializer<TEvent>(...). Types marked with
+    // [BinaryEvent] but not registered explicitly fall back to DefaultBinarySerializer; that
+    // resolution is cached in _binarySerializerResolution so the attribute probe (and the throw for a
+    // misconfigured type) happens once per event type rather than once per row.
+    private readonly ConcurrentDictionary<Type, IEventBinarySerializer> _binarySerializerByType = new();
+    private readonly ConcurrentDictionary<Type, IEventBinarySerializer?> _binarySerializerResolution = new();
+
+    /// <summary>
+    ///     Store-wide fallback <see cref="IEventBinarySerializer" /> used for event types marked with
+    ///     <see cref="BinaryEventAttribute" /> that have no explicit per-type registration. Null by
+    ///     default, which leaves every event type on the JSON path.
+    /// </summary>
+    public IEventBinarySerializer? DefaultBinarySerializer
+    {
+        get => _defaultBinarySerializer;
+        set
+        {
+            _defaultBinarySerializer = value;
+            _binarySerializerResolution.Clear();
+        }
+    }
+
+    private IEventBinarySerializer? _defaultBinarySerializer;
+
+    /// <summary>
+    ///     Opt <typeparamref name="TEvent" /> into binary serialization (#388): its payload is written
+    ///     to the <c>bdata</c> column instead of <c>data</c>, and read back through the same
+    ///     serializer. Wins over <see cref="BinaryEventAttribute" /> + <see cref="DefaultBinarySerializer" />.
+    /// </summary>
+    public EventGraph UseBinarySerializer<TEvent>(IEventBinarySerializer serializer) where TEvent : notnull
+    {
+        ArgumentNullException.ThrowIfNull(serializer);
+        _binarySerializerByType[typeof(TEvent)] = serializer;
+        _binarySerializerResolution.Clear();
+        return this;
+    }
+
+    /// <summary>
+    ///     The <see cref="IEventBinarySerializer" /> governing <paramref name="eventType" />, or null
+    ///     when that type stays on the JSON path. Explicit registration beats
+    ///     <see cref="BinaryEventAttribute" /> + <see cref="DefaultBinarySerializer" />.
+    /// </summary>
+    /// <exception cref="InvalidOperationException">
+    ///     The type carries <see cref="BinaryEventAttribute" /> but no serializer is configured for it.
+    ///     Deliberately a throw rather than a silent fall back to JSON: a store that quietly ignored
+    ///     the attribute would have write-amplification characteristics that do not match its
+    ///     configuration, which is the whole reason the feature exists.
+    /// </exception>
+    internal IEventBinarySerializer? ResolveBinarySerializerFor(Type eventType)
+        => _binarySerializerResolution.GetOrAdd(eventType, static (type, graph) =>
+        {
+            if (graph._binarySerializerByType.TryGetValue(type, out var explicitSerializer))
+            {
+                return explicitSerializer;
+            }
+
+            if (!type.IsDefined(typeof(BinaryEventAttribute), inherit: false))
+            {
+                return null;
+            }
+
+            return graph._defaultBinarySerializer ?? throw new InvalidOperationException(
+                $"Event type '{type.FullName}' is marked with [BinaryEvent] but no IEventBinarySerializer "
+                + $"is registered. Either call opts.Events.UseBinarySerializer<{type.Name}>(...) explicitly, "
+                + "or set opts.Events.DefaultBinarySerializer to a store-wide fallback.");
+        }, this);
+
+    /// <summary>
+    ///     The <c>bdata</c> bytes for an event on the write path, or null when the event's type stays
+    ///     on the JSON path. The two are mutually exclusive per row — see
+    ///     <see cref="JsonPlaceholderForBinaryEvent" />.
+    /// </summary>
+    internal byte[]? SerializeEventBdata(IEvent @event)
+    {
+        // Deliberately NOT short-circuited on UsesBinaryEventSerialization: a type marked
+        // [BinaryEvent] in a store with no serializer configured has to throw here, and skipping the
+        // resolve for "performance" would turn that into a silent write of JSON. The resolve is a
+        // ConcurrentDictionary hit cached per event type, so a JSON-only store pays one lookup per
+        // appended event and nothing more.
+        var eventType = @event.EventType ?? @event.Data.GetType();
+        var serializer = ResolveBinarySerializerFor(eventType);
+        return serializer?.Serialize(eventType, @event.Data);
+    }
+
+    /// <summary>
+    ///     What goes in the <c>data</c> column of a binary event's row. An empty JSON object rather
+    ///     than NULL, because <c>data</c> is NOT NULL and typed <c>json</c> on SQL Server 2025 — a
+    ///     row still has to hold something the engine will parse.
+    /// </summary>
+    internal const string JsonPlaceholderForBinaryEvent = "{}";
+
+    /// <summary>
+    ///     The per-row read counterpart of <see cref="SerializeEventBdata" />: <paramref name="bdata" />
+    ///     being non-null is the on-row discriminator, so JSON rows written before the feature was
+    ///     switched on keep deserializing through <paramref name="serializer" /> unchanged.
+    /// </summary>
+    internal object DeserializeEventData(Type resolvedType, string json, byte[]? bdata, ISerializer serializer)
+    {
+        if (bdata is null) return serializer.FromJson(resolvedType, json);
+
+        var binary = ResolveBinarySerializerFor(resolvedType)
+                     ?? throw new InvalidOperationException(
+                         $"A pc_events row for '{resolvedType.FullName}' has a non-null bdata column but no "
+                         + "IEventBinarySerializer is registered for that type. Configure it with "
+                         + $"opts.Events.UseBinarySerializer<{resolvedType.Name}>(...) or set "
+                         + "opts.Events.DefaultBinarySerializer — the event cannot be read without it.");
+
+        return binary.Deserialize(resolvedType, bdata);
+    }
+
     /// <summary>
     ///     All currently registered event types.
     /// </summary>

src/Polecat/Events/EventOperations.cs

@@ -512,11 +512,17 @@ public void TombstoneStream(string streamKey)
 
     public void OverwriteEvent(IEvent @event)
     {
-        var serializedData = _sessionBase.Serializer.ToJson(@event.Data);
+        // #388: same data/bdata split as the append path — a binary event's masked payload has to go
+        // back through its IEventBinarySerializer, not into the JSON column.
+        var serializedBdata = _events.SerializeEventBdata(@event);
+        var serializedData = serializedBdata is null
+            ? _sessionBase.Serializer.ToJson(@event.Data)
+            : EventGraph.JsonPlaceholderForBinaryEvent;
         var serializedHeaders = @event.Headers != null
             ? _sessionBase.Serializer.ToJson(@event.Headers)
             : null;
-        _workTracker.Add(new Protected.OverwriteEventOperation(_events, @event, serializedData, serializedHeaders));
+        _workTracker.Add(new Protected.OverwriteEventOperation(
+            _events, @event, serializedData, serializedBdata, serializedHeaders));
     }
 
     public Guid CompletelyReplaceEvent<T>(long sequence, T eventBody) where T : class
@@ -527,9 +533,14 @@ public Guid CompletelyReplaceEvent<T>(long sequence, T eventBody) where T : clas
         _events.AddEventType(typeof(T));
         var mapping = _events.EventMappingFor(typeof(T));
 
-        var serializedData = _sessionBase.Serializer.ToJson(eventBody);
+        // #388: the REPLACEMENT body's type decides the row's format, not the replaced row's.
+        var binary = _events.ResolveBinarySerializerFor(typeof(T));
+        var serializedBdata = binary?.Serialize(typeof(T), eventBody);
+        var serializedData = serializedBdata is null
+            ? _sessionBase.Serializer.ToJson(eventBody)
+            : EventGraph.JsonPlaceholderForBinaryEvent;
         var op = new Protected.ReplaceEventOperation(
-            _events, sequence, serializedData, mapping.EventTypeName, mapping.DotNetTypeName);
+            _events, sequence, serializedData, serializedBdata, mapping.EventTypeName, mapping.DotNetTypeName);
 
         _workTracker.Add(op);
         return op.Id;
@@ -868,10 +879,10 @@ public async Task<IReadOnlyList<IEvent>> QueryByTagsAsync(EventTagQuery query,
         var eventOptions = _events.EventOptions;
 
         // Build SELECT columns matching the event reader format
-        var selectColumns = "e.seq_id, e.id, e.stream_id, e.version, e.data, e.type, e.timestamp, e.tenant_id, e.dotnet_type, e.is_archived";
-        if (eventOptions.EnableCorrelationId) selectColumns += ", e.correlation_id";
-        if (eventOptions.EnableCausationId) selectColumns += ", e.causation_id";
-        if (eventOptions.EnableHeaders) selectColumns += ", e.headers";
+        // #388: composed by PcEventsRowReader rather than spelled out here — this projection has to
+        // stay in lockstep with the canonical one (which now carries bdata at ordinal 10), and two
+        // hand-written copies of the same column list is exactly how that stops being true.
+        var selectColumns = Internal.PcEventsRowReader.ComposeSelectColumnsWithAlias(eventOptions, "e");
 
         var sb = new StringBuilder();
         sb.Append($"SELECT {selectColumns} FROM [{schema}].[pc_events] e");
@@ -949,11 +960,12 @@ public async Task<IReadOnlyList<IEvent>> QueryByTagsAsync(EventTagQuery query,
             var tenantId = reader.GetString(7);
             var dotNetTypeName = reader.IsDBNull(8) ? null : reader.GetString(8);
             var isArchived = reader.GetBoolean(9);
+            var bdata = reader.IsDBNull(10) ? null : reader.GetFieldValue<byte[]>(10); // #388
 
             var resolvedType = _events.ResolveEventType(dotNetTypeName);
             if (resolvedType == null) continue;
 
-            var data = _sessionBase.Serializer.FromJson(resolvedType, json);
+            var data = _events.DeserializeEventData(resolvedType, json, bdata, _sessionBase.Serializer);
             var mapping = _events.EventMappingFor(resolvedType);
             var @event = mapping.Wrap(data);
 
@@ -978,7 +990,7 @@ public async Task<IReadOnlyList<IEvent>> QueryByTagsAsync(EventTagQuery query,
                 @event.StreamKey = rawStreamId.ToString();
             }
 
-            var metaIndex = 10;
+            var metaIndex = 11; // #388: ordinal 10 is bdata
             if (eventOptions.EnableCorrelationId)
             {
                 @event.CorrelationId = reader.IsDBNull(metaIndex) ? null : reader.GetString(metaIndex);
@@ -1075,10 +1087,10 @@ internal static void WriteTagQuerySql(ICommandBuilder builder, EventGraph eventG
         var schema = eventGraph.DatabaseSchemaName;
         var eventOptions = eventGraph.EventOptions;
 
-        var selectColumns = "e.seq_id, e.id, e.stream_id, e.version, e.data, e.type, e.timestamp, e.tenant_id, e.dotnet_type, e.is_archived";
-        if (eventOptions.EnableCorrelationId) selectColumns += ", e.correlation_id";
-        if (eventOptions.EnableCausationId) selectColumns += ", e.causation_id";
-        if (eventOptions.EnableHeaders) selectColumns += ", e.headers";
+        // #388: composed by PcEventsRowReader rather than spelled out here — this projection has to
+        // stay in lockstep with the canonical one (which now carries bdata at ordinal 10), and two
+        // hand-written copies of the same column list is exactly how that stops being true.
+        var selectColumns = Internal.PcEventsRowReader.ComposeSelectColumnsWithAlias(eventOptions, "e");
 
         builder.Append($"SELECT {selectColumns} FROM [{schema}].[pc_events] e");
 
@@ -1148,11 +1160,12 @@ internal static void WriteTagQuerySql(ICommandBuilder builder, EventGraph eventG
         var tenantId = reader.GetString(7);
         var dotNetTypeName = reader.IsDBNull(8) ? null : reader.GetString(8);
         var isArchived = reader.GetBoolean(9);
+        var bdata = reader.IsDBNull(10) ? null : reader.GetFieldValue<byte[]>(10); // #388
 
         var resolvedType = eventGraph.ResolveEventType(dotNetTypeName);
         if (resolvedType == null) return null;
 
-        var data = serializer.FromJson(resolvedType, json);
+        var data = eventGraph.DeserializeEventData(resolvedType, json, bdata, serializer);
         var mapping = eventGraph.EventMappingFor(resolvedType);
         var @event = mapping.Wrap(data);
 
@@ -1165,7 +1178,7 @@ internal static void WriteTagQuerySql(ICommandBuilder builder, EventGraph eventG
         @event.DotNetTypeName = dotNetTypeName!;
         @event.IsArchived = isArchived;
 
-        var metaIndex = 10;
+        var metaIndex = 11; // #388: ordinal 10 is bdata
         if (eventOptions.EnableCorrelationId)
         {
             @event.CorrelationId = reader.IsDBNull(metaIndex) ? null : reader.GetString(metaIndex);

src/Polecat/Events/IEventBinarySerializer.cs

@@ -0,0 +1,42 @@
+namespace Polecat.Events;
+
+/// <summary>
+///     Pluggable binary serializer for event data — Polecat's counterpart of Marten's
+///     <c>IEventBinarySerializer</c> (<see href="https://github.com/JasperFx/marten/issues/4515" />,
+///     shipped in Marten 9.20.2), at parity so a store-agnostic consumer can wire either flavor.
+///     Tracked as <see href="https://github.com/JasperFx/polecat/issues/388">polecat#388</see>.
+/// </summary>
+/// <remarks>
+///     <para>
+///         Binary serialization is enabled <strong>per event type</strong>, not store-wide. A store
+///         can have JSON events and binary events mixed in the same <c>pc_events</c> table; a row's
+///         format is determined by whether its <c>bdata</c> column is <c>NULL</c> (JSON) or not
+///         (binary). That is what makes the feature safe to switch on for an existing store with no
+///         migration of existing event data — and just as safe to switch back off.
+///     </para>
+///     <para>
+///         Opt in by marking an event type with <see cref="BinaryEventAttribute" /> (resolved against
+///         the store-wide <c>opts.Events.DefaultBinarySerializer</c>) or by registering it explicitly
+///         with <c>opts.Events.UseBinarySerializer&lt;TEvent&gt;(serializer)</c>. An explicit per-type
+///         registration wins over the attribute.
+///     </para>
+///     <para>
+///         Implementations must be thread-safe: one instance serves every session in the store.
+///     </para>
+/// </remarks>
+public interface IEventBinarySerializer
+{
+    /// <summary>
+    ///     Serialize an event data instance to bytes.
+    /// </summary>
+    /// <param name="type">The runtime CLR type of the event data.</param>
+    /// <param name="data">The event data to serialize.</param>
+    byte[] Serialize(Type type, object data);
+
+    /// <summary>
+    ///     Deserialize bytes back into an event data instance.
+    /// </summary>
+    /// <param name="type">The target CLR type to deserialize into.</param>
+    /// <param name="data">The bytes previously produced by <see cref="Serialize" />.</param>
+    object Deserialize(Type type, byte[] data);
+}