diff --git a/src/Polecat.Tests/Events/binary_event_serialization_tests.cs b/src/Polecat.Tests/Events/binary_event_serialization_tests.cs new file mode 100644 index 00000000..65976c6f --- /dev/null +++ b/src/Polecat.Tests/Events/binary_event_serialization_tests.cs @@ -0,0 +1,421 @@ +using System.Text; +using System.Text.Json; +using JasperFx.Events; +using JasperFx.Events.Projections; +using Polecat.Events; +using Polecat.Linq; +using Polecat.Projections; +using Polecat.Tests.Harness; +using Shouldly; + +namespace Polecat.Tests.Events; + +#region sample_polecat_binary_event_serializer + +/// +/// A deliberately non-JSON binary format, so a test can prove a row really did travel through the +/// binary path: the bytes are a length-prefixed UTF-8 blob that no JSON parser would accept, and a +/// four-byte magic header lets the assertions recognize it on the wire. +/// +public sealed class TestBinaryEventSerializer : IEventBinarySerializer +{ + public static readonly byte[] Magic = "PCB1"u8.ToArray(); + + /// How many times Serialize/Deserialize were called — proves the path was taken. + public int SerializeCount; + + public int DeserializeCount; + + public byte[] Serialize(Type type, object data) + { + Interlocked.Increment(ref SerializeCount); + var payload = JsonSerializer.SerializeToUtf8Bytes(data, type); + var buffer = new byte[Magic.Length + payload.Length]; + Magic.CopyTo(buffer, 0); + payload.CopyTo(buffer, Magic.Length); + return buffer; + } + + public object Deserialize(Type type, byte[] data) + { + Interlocked.Increment(ref DeserializeCount); + if (data.Length < Magic.Length || !data.AsSpan(0, Magic.Length).SequenceEqual(Magic)) + { + throw new InvalidOperationException("Not a payload written by this serializer."); + } + + return JsonSerializer.Deserialize(data.AsSpan(Magic.Length), type)!; + } +} + +#endregion + +public record BinaryPayloadRecorded(string Name, int Amount); + +public record JsonPayloadRecorded(string Name, int Amount); + +[BinaryEvent] +public record AttributeMarkedRecorded(string Name); + +public class BinaryLedger +{ + public Guid Id { get; set; } + public int Total { get; set; } + public List Names { get; set; } = new(); +} + +public partial class BinaryLedgerProjection : SingleStreamProjection +{ + public void Apply(BinaryLedger ledger, BinaryPayloadRecorded e) + { + ledger.Total += e.Amount; + ledger.Names.Add(e.Name); + } + + public void Apply(BinaryLedger ledger, JsonPayloadRecorded e) + { + ledger.Total += e.Amount; + ledger.Names.Add(e.Name); + } +} + +/// +/// polecat#388: pluggable binary event serialization at parity with Marten's +/// IEventBinarySerializer (marten#4515). The design point under test throughout is the +/// coexistence one: an additive nullable bdata column with bdata IS NULL as +/// the per-row discriminator, so JSON and binary events live in the same pc_events table +/// and the feature is switchable on an existing store with no data migration. +/// +public class binary_event_serialization_tests : OneOffConfigurationsContext +{ + private readonly TestBinaryEventSerializer _serializer = new(); + + private async Task ConfigureAndApply(Action configure) + { + ConfigureStore(configure); + await theDatabase.ApplyAllConfiguredChangesToDatabaseAsync(); + } + + private Task ConfigureBinaryStore(Action? extra = null) => ConfigureAndApply(opts => + { + opts.Events.UseBinarySerializer(_serializer); + extra?.Invoke(opts); + }); + + private async Task<(string data, byte[]? bdata)> ReadRawRowAsync(long sequence) + { + await using var conn = await OpenConnectionAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = $"SELECT CONVERT(nvarchar(max), data), bdata FROM {theStore.Options.EventGraph.EventsTableName} WHERE seq_id = @seq"; + var p = cmd.CreateParameter(); + p.ParameterName = "@seq"; + p.Value = sequence; + cmd.Parameters.Add(p); + + await using var reader = await cmd.ExecuteReaderAsync(TestContext.Current.CancellationToken); + (await reader.ReadAsync(TestContext.Current.CancellationToken)).ShouldBeTrue(); + var data = reader.GetString(0); + var bdata = reader.IsDBNull(1) ? null : (byte[])reader.GetValue(1); + return (data, bdata); + } + + [Fact] + public async Task pc_events_carries_a_nullable_bdata_column() + { + await ConfigureAndApply(_ => { }); + + await using var conn = await OpenConnectionAsync(); + await using var cmd = conn.CreateCommand(); + cmd.CommandText = """ + SELECT c.is_nullable, TYPE_NAME(c.system_type_id), c.max_length + FROM sys.columns c + WHERE c.object_id = OBJECT_ID(@table) AND c.name = 'bdata' + """; + var p = cmd.CreateParameter(); + p.ParameterName = "@table"; + p.Value = theStore.Options.EventGraph.EventsTableName; + cmd.Parameters.Add(p); + + await using var reader = await cmd.ExecuteReaderAsync(TestContext.Current.CancellationToken); + (await reader.ReadAsync(TestContext.Current.CancellationToken)) + .ShouldBeTrue("pc_events should carry the bdata column even with no serializer configured"); + reader.GetBoolean(0).ShouldBeTrue("bdata must be nullable — NULL is the JSON-row discriminator"); + reader.GetString(1).ShouldBe("varbinary"); + reader.GetInt16(2).ShouldBe((short)-1); // varbinary(max) + } + + [Fact] + public async Task an_unconfigured_store_writes_json_and_leaves_bdata_null() + { + await ConfigureAndApply(_ => { }); + + await using var session = theStore.LightweightSession(); + var streamId = session.Events.StartStream(new JsonPayloadRecorded("plain", 3)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + + var events = await session.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + var (data, bdata) = await ReadRawRowAsync(events[0].Sequence); + + bdata.ShouldBeNull(); + data.ShouldContain("plain"); + } + + [Fact] + public async Task a_binary_event_round_trips_through_bdata_and_leaves_data_as_a_placeholder() + { + await ConfigureBinaryStore(); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream(new BinaryPayloadRecorded("binary", 7)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + _serializer.SerializeCount.ShouldBe(1); + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events.Count.ShouldBe(1); + events[0].Data.ShouldBeOfType().ShouldBe(new BinaryPayloadRecorded("binary", 7)); + _serializer.DeserializeCount.ShouldBeGreaterThan(0); + + // On the wire: the payload is in bdata, and `data` holds only the placeholder — so the + // saving the feature exists for is real, not just a round trip that happens to work. + var (data, bdata) = await ReadRawRowAsync(events[0].Sequence); + bdata.ShouldNotBeNull(); + bdata!.AsSpan(0, TestBinaryEventSerializer.Magic.Length).SequenceEqual(TestBinaryEventSerializer.Magic) + .ShouldBeTrue(); + Encoding.UTF8.GetString(bdata).ShouldContain("binary"); + data.ShouldBe("{}"); + data.ShouldNotContain("binary"); + } + + [Fact] + public async Task json_and_binary_events_coexist_in_one_stream() + { + // The coexistence property is the whole reason this design is low-risk to adopt: only the + // opted-in type changes format, and per-ROW dispatch means one stream can hold both. + await ConfigureBinaryStore(); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream( + new JsonPayloadRecorded("first", 1), + new BinaryPayloadRecorded("second", 2), + new JsonPayloadRecorded("third", 3)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events.Count.ShouldBe(3); + events.Select(x => x.Data).ShouldBe([ + new JsonPayloadRecorded("first", 1), + new BinaryPayloadRecorded("second", 2), + new JsonPayloadRecorded("third", 3) + ]); + + (await ReadRawRowAsync(events[0].Sequence)).bdata.ShouldBeNull(); + (await ReadRawRowAsync(events[1].Sequence)).bdata.ShouldNotBeNull(); + (await ReadRawRowAsync(events[2].Sequence)).bdata.ShouldBeNull(); + } + + [Fact] + public async Task rows_written_before_the_serializer_was_configured_still_read_as_json() + { + // The "switch it on for an existing store with no migration" claim, exercised directly: + // append as JSON, then reconfigure the same schema with the serializer registered and read + // the pre-existing row back. + await ConfigureAndApply(_ => { }); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream(new BinaryPayloadRecorded("legacy-json", 5)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + // Reconfigure WITHOUT dropping the schema — same tables, now with binary registered. + ConfigureStore(opts => opts.Events.UseBinarySerializer(_serializer)); + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events[0].Data.ShouldBe(new BinaryPayloadRecorded("legacy-json", 5)); + _serializer.DeserializeCount.ShouldBe(0, "a bdata IS NULL row must not go through the binary path"); + + // And the next append for that type does use binary — the two formats sit side by side. + await using (var session = theStore.LightweightSession()) + { + session.Events.Append(streamId, new BinaryPayloadRecorded("now-binary", 6)); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query2 = theStore.QuerySession(); + var after = await query2.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + after.Count.ShouldBe(2); + after.Select(x => ((BinaryPayloadRecorded)x.Data).Name).ShouldBe(["legacy-json", "now-binary"]); + (await ReadRawRowAsync(after[0].Sequence)).bdata.ShouldBeNull(); + (await ReadRawRowAsync(after[1].Sequence)).bdata.ShouldNotBeNull(); + } + + [Fact] + public async Task the_binary_event_attribute_resolves_against_the_default_serializer() + { + await ConfigureAndApply(opts => opts.Events.DefaultBinarySerializer = _serializer); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream( + new AttributeMarkedRecorded("attributed"), + new JsonPayloadRecorded("not-attributed", 1)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events[0].Data.ShouldBe(new AttributeMarkedRecorded("attributed")); + events[1].Data.ShouldBe(new JsonPayloadRecorded("not-attributed", 1)); + + (await ReadRawRowAsync(events[0].Sequence)).bdata.ShouldNotBeNull(); + (await ReadRawRowAsync(events[1].Sequence)).bdata + .ShouldBeNull("[BinaryEvent] is per type — an unmarked type stays on the JSON path"); + } + + [Fact] + public async Task an_explicit_registration_wins_over_the_attribute_and_the_default() + { + var explicitSerializer = new TestBinaryEventSerializer(); + await ConfigureAndApply(opts => + { + opts.Events.DefaultBinarySerializer = _serializer; + opts.Events.UseBinarySerializer(explicitSerializer); + }); + + await using var session = theStore.LightweightSession(); + session.Events.StartStream(new AttributeMarkedRecorded("explicit")); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + + explicitSerializer.SerializeCount.ShouldBe(1); + _serializer.SerializeCount.ShouldBe(0); + } + + [Fact] + public async Task a_marked_type_with_no_serializer_configured_throws_rather_than_silently_writing_json() + { + // A silent fallback would leave a store whose write amplification does not match its + // configuration — which is exactly the problem the feature exists to fix. + await ConfigureAndApply(_ => { }); + + await using var session = theStore.LightweightSession(); + session.Events.StartStream(new AttributeMarkedRecorded("unconfigured")); + + var ex = await Should.ThrowAsync( + async () => await session.SaveChangesAsync(TestContext.Current.CancellationToken)); + ex.Message.ShouldContain("[BinaryEvent]"); + ex.Message.ShouldContain("DefaultBinarySerializer"); + } + + [Fact] + public async Task binary_events_flow_through_an_inline_projection() + { + await ConfigureBinaryStore(opts => + opts.Projections.Add(ProjectionLifecycle.Inline)); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream( + new BinaryPayloadRecorded("b", 10), + new JsonPayloadRecorded("j", 5)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query = theStore.QuerySession(); + var ledger = await query.LoadAsync(streamId, TestContext.Current.CancellationToken); + ledger.ShouldNotBeNull(); + ledger!.Total.ShouldBe(15); + ledger.Names.ShouldBe(["b", "j"]); + } + + [Fact] + public async Task binary_events_flow_through_an_async_projection_via_the_daemon() + { + // The daemon's event loader is a separate reader from FetchStreamAsync, with its own SELECT + // and its own deserialization — so it needs its own coverage for the bdata dispatch. + await ConfigureBinaryStore(opts => + opts.Projections.Add(ProjectionLifecycle.Async)); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream( + new BinaryPayloadRecorded("b1", 4), + new BinaryPayloadRecorded("b2", 6), + new JsonPayloadRecorded("j1", 1)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + using var daemon = await theStore.BuildProjectionDaemonAsync(); + await daemon.StartAllAsync(); + await daemon.WaitForNonStaleData(TimeSpan.FromSeconds(30)); + + await using var query = theStore.QuerySession(); + var ledger = await query.LoadAsync(streamId, TestContext.Current.CancellationToken); + ledger.ShouldNotBeNull(); + ledger!.Total.ShouldBe(11); + ledger.Names.ShouldBe(["b1", "b2", "j1"]); + } + + [Fact] + public async Task binary_events_hydrate_through_the_event_linq_provider() + { + // A third distinct reader (EventListHandler) — same dispatch, separately covered. + await ConfigureBinaryStore(); + + await using (var session = theStore.LightweightSession()) + { + session.Events.StartStream(new BinaryPayloadRecorded("via-linq", 42)); + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await using var query = theStore.QuerySession(); + var events = await query.Events.QueryAllRawEvents() + .Where(e => e.EventTypeName == "binary_payload_recorded") + .ToListAsync(TestContext.Current.CancellationToken); + + events.Count.ShouldBe(1); + events[0].Data.ShouldBe(new BinaryPayloadRecorded("via-linq", 42)); + } + + [Fact] + public async Task masking_a_binary_event_rewrites_bdata_not_just_the_json_column() + { + // If masking only rewrote `data`, the original payload would stay readable in bdata — for a + // GDPR masking operation that is the entire point missed. + await ConfigureBinaryStore(); + theStore.Events.AddMaskingRuleForProtectedInformation( + e => e with { Name = "****" }); + + Guid streamId; + await using (var session = theStore.LightweightSession()) + { + streamId = session.Events.StartStream(new BinaryPayloadRecorded("secret", 9)).Id; + await session.SaveChangesAsync(TestContext.Current.CancellationToken); + } + + await theStore.Advanced.ApplyEventDataMasking(x => x.IncludeStream(streamId), + TestContext.Current.CancellationToken); + + await using var query = theStore.QuerySession(); + var events = await query.Events.FetchStreamAsync(streamId, token: TestContext.Current.CancellationToken); + events[0].Data.ShouldBeOfType().Name.ShouldBe("****"); + + var (data, bdata) = await ReadRawRowAsync(events[0].Sequence); + bdata.ShouldNotBeNull("the masked row must stay binary"); + Encoding.UTF8.GetString(bdata!).ShouldNotContain("secret"); + data.ShouldBe("{}"); + } +} diff --git a/src/Polecat/Events/BinaryEventAttribute.cs b/src/Polecat/Events/BinaryEventAttribute.cs new file mode 100644 index 00000000..68a5cf74 --- /dev/null +++ b/src/Polecat/Events/BinaryEventAttribute.cs @@ -0,0 +1,28 @@ +namespace Polecat.Events; + +/// +/// Marks an event type as binary-serialized: its pc_events.data column holds the +/// '{}' placeholder and the real payload lives in bdata, written and read by an +/// . Mirrors Marten's [BinaryEvent] +/// (); tracked as +/// polecat#388. +/// +/// +/// +/// The serializer for an attribute-marked type is the store-wide +/// opts.Events.DefaultBinarySerializer. An explicit +/// opts.Events.UseBinarySerializer<TEvent>(serializer) 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. +/// +/// +/// 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 bdata = NULL +/// and keep reading through the JSON path. +/// +/// +[AttributeUsage(AttributeTargets.Class | AttributeTargets.Struct, Inherited = false)] +public sealed class BinaryEventAttribute : Attribute +{ +} diff --git a/src/Polecat/Events/Daemon/PolecatEventLoader.cs b/src/Polecat/Events/Daemon/PolecatEventLoader.cs index e7c35afe..1def9708 100644 --- a/src/Polecat/Events/Daemon/PolecatEventLoader.cs +++ b/src/Polecat/Events/Daemon/PolecatEventLoader.cs @@ -80,7 +80,7 @@ private async Task 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(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) { diff --git a/src/Polecat/Events/EventGraph.cs b/src/Polecat/Events/EventGraph.cs index 4361befc..f8d3478c 100644 --- a/src/Polecat/Events/EventGraph.cs +++ b/src/Polecat/Events/EventGraph.cs @@ -440,6 +440,118 @@ public ITagTypeRegistration RegisterTagType(string tableSuffix) where TTag public IReadOnlyList TagTypes => _tagTypes; + // ---- #388: pluggable binary event serialization ------------------------------------------ + // + // Explicit per-type registrations from UseBinarySerializer(...). 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 _binarySerializerByType = new(); + private readonly ConcurrentDictionary _binarySerializerResolution = new(); + + /// + /// Store-wide fallback used for event types marked with + /// that have no explicit per-type registration. Null by + /// default, which leaves every event type on the JSON path. + /// + public IEventBinarySerializer? DefaultBinarySerializer + { + get => _defaultBinarySerializer; + set + { + _defaultBinarySerializer = value; + _binarySerializerResolution.Clear(); + } + } + + private IEventBinarySerializer? _defaultBinarySerializer; + + /// + /// Opt into binary serialization (#388): its payload is written + /// to the bdata column instead of data, and read back through the same + /// serializer. Wins over + . + /// + public EventGraph UseBinarySerializer(IEventBinarySerializer serializer) where TEvent : notnull + { + ArgumentNullException.ThrowIfNull(serializer); + _binarySerializerByType[typeof(TEvent)] = serializer; + _binarySerializerResolution.Clear(); + return this; + } + + /// + /// The governing , or null + /// when that type stays on the JSON path. Explicit registration beats + /// + . + /// + /// + /// The type carries 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. + /// + 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); + + /// + /// The bdata 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 + /// . + /// + 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); + } + + /// + /// What goes in the data column of a binary event's row. An empty JSON object rather + /// than NULL, because data is NOT NULL and typed json on SQL Server 2025 — a + /// row still has to hold something the engine will parse. + /// + internal const string JsonPlaceholderForBinaryEvent = "{}"; + + /// + /// The per-row read counterpart of : + /// being non-null is the on-row discriminator, so JSON rows written before the feature was + /// switched on keep deserializing through unchanged. + /// + 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); + } + /// /// All currently registered event types. /// diff --git a/src/Polecat/Events/EventOperations.cs b/src/Polecat/Events/EventOperations.cs index 63a14485..07089002 100644 --- a/src/Polecat/Events/EventOperations.cs +++ b/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(long sequence, T eventBody) where T : class @@ -527,9 +533,14 @@ public Guid CompletelyReplaceEvent(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> 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> 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(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> 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(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); diff --git a/src/Polecat/Events/IEventBinarySerializer.cs b/src/Polecat/Events/IEventBinarySerializer.cs new file mode 100644 index 00000000..09565aab --- /dev/null +++ b/src/Polecat/Events/IEventBinarySerializer.cs @@ -0,0 +1,42 @@ +namespace Polecat.Events; + +/// +/// Pluggable binary serializer for event data — Polecat's counterpart of Marten's +/// IEventBinarySerializer (, +/// shipped in Marten 9.20.2), at parity so a store-agnostic consumer can wire either flavor. +/// Tracked as polecat#388. +/// +/// +/// +/// Binary serialization is enabled per event type, not store-wide. A store +/// can have JSON events and binary events mixed in the same pc_events table; a row's +/// format is determined by whether its bdata column is NULL (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. +/// +/// +/// Opt in by marking an event type with (resolved against +/// the store-wide opts.Events.DefaultBinarySerializer) or by registering it explicitly +/// with opts.Events.UseBinarySerializer<TEvent>(serializer). An explicit per-type +/// registration wins over the attribute. +/// +/// +/// Implementations must be thread-safe: one instance serves every session in the store. +/// +/// +public interface IEventBinarySerializer +{ + /// + /// Serialize an event data instance to bytes. + /// + /// The runtime CLR type of the event data. + /// The event data to serialize. + byte[] Serialize(Type type, object data); + + /// + /// Deserialize bytes back into an event data instance. + /// + /// The target CLR type to deserialize into. + /// The bytes previously produced by . + object Deserialize(Type type, byte[] data); +} diff --git a/src/Polecat/Events/Internal/PcEventsRowReader.cs b/src/Polecat/Events/Internal/PcEventsRowReader.cs index b14d5fbd..c1649784 100644 --- a/src/Polecat/Events/Internal/PcEventsRowReader.cs +++ b/src/Polecat/Events/Internal/PcEventsRowReader.cs @@ -61,10 +61,16 @@ namespace Polecat.Events.Internal; internal static class PcEventsRowReader { /// - /// Mandatory columns, always projected. Ordinals 0–9. + /// Mandatory columns, always projected. Ordinals 0–10. /// + /// + /// #388 pins bdata at ordinal 10, after the previously-locked 0–9 block and before the + /// optional metadata columns, so the per-row JSON-vs-binary dispatch reads a stable ordinal + /// without disturbing any of the existing ones. The optional metadata slots shift from 10 to 11 + /// — see . + /// internal const string CoreSelectColumns = - "seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, is_archived"; + "seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, is_archived, bdata"; /// /// Compose plus any optional metadata @@ -93,7 +99,7 @@ internal static string ComposeSelectColumns(EventStoreOptions options) internal static string ComposeSelectColumnsWithAlias(EventStoreOptions options, string alias) { var sb = new StringBuilder( - $"{alias}.seq_id, {alias}.id, {alias}.stream_id, {alias}.version, {alias}.data, {alias}.type, {alias}.timestamp, {alias}.tenant_id, {alias}.dotnet_type, {alias}.is_archived"); + $"{alias}.seq_id, {alias}.id, {alias}.stream_id, {alias}.version, {alias}.data, {alias}.type, {alias}.timestamp, {alias}.tenant_id, {alias}.dotnet_type, {alias}.is_archived, {alias}.bdata"); if (options.EnableCorrelationId) sb.Append($", {alias}.correlation_id"); if (options.EnableCausationId) sb.Append($", {alias}.causation_id"); if (options.EnableHeaders) sb.Append($", {alias}.headers"); @@ -162,6 +168,10 @@ internal static string ComposeSelectColumnsWithAlias(EventStoreOptions options, var tenantId = reader.IsDBNull(7) ? ctx.DefaultTenantId : reader.GetString(7); var dotNetTypeName = reader.IsDBNull(8) ? null : reader.GetString(8); var isArchived = reader.GetBoolean(9); + // #388: non-null bdata is the per-row discriminator. Reading it here (rather than inside the + // optional-metadata slot loop) keeps the dispatch on a fixed ordinal and off the JSON path's + // hot loop — a store with no binary events pays one IsDBNull per row. + var bdata = reader.IsDBNull(10) ? null : reader.GetFieldValue(10); var resolvedType = ctx.EventGraph.ResolveEventType(dotNetTypeName); if (resolvedType == null) return null; @@ -171,7 +181,7 @@ internal static string ComposeSelectColumnsWithAlias(EventStoreOptions options, // this collapses N dictionary lookups into 1 per distinct type. var mapping = cache.LookupOrAdd(ctx.EventGraph, resolvedType); - var data = ctx.Serializer.FromJson(resolvedType, json); + var data = ctx.EventGraph.DeserializeEventData(resolvedType, json, bdata, ctx.Serializer); var @event = mapping.Wrap(data); @event.Id = eventId; @@ -253,6 +263,10 @@ internal static EventRecord ReadEventRecord( metadata = headerDoc.RootElement.Clone(); } + // #388: a binary event's `data` column holds only the '{}' placeholder, so the explorer's + // JsonElement view of a binary row is that empty object. The explorer is a diagnostic surface + // that deliberately does not deserialize (no ISerializer, no IEventBinarySerializer), so it + // reports what is in the JSON column rather than pretending to decode the bytes. using var doc = JsonDocument.Parse(rawData); var data = doc.RootElement.Clone(); @@ -294,7 +308,8 @@ internal readonly record struct MetadataSlots(int CorrelationIdx, int CausationI public static MetadataSlots Compute(EventStoreOptions options) { - var ordinal = 10; + // #388: ordinal 10 is bdata; the optional metadata block starts at 11. + var ordinal = 11; var correlation = options.EnableCorrelationId ? ordinal++ : Disabled; var causation = options.EnableCausationId ? ordinal++ : Disabled; var headers = options.EnableHeaders ? ordinal++ : Disabled; diff --git a/src/Polecat/Events/Linq/EventLinqQueryProvider.cs b/src/Polecat/Events/Linq/EventLinqQueryProvider.cs index b68eb90c..7c9cc101 100644 --- a/src/Polecat/Events/Linq/EventLinqQueryProvider.cs +++ b/src/Polecat/Events/Linq/EventLinqQueryProvider.cs @@ -143,7 +143,7 @@ or SingleValueMode.Any or SingleValueMode.Sum or SingleValueMode.Min // Full IEvent result set. #256: append the opt-in metadata columns (when enabled) so // hydrated events carry correlation/causation/user_name — EventListHandler reads them // at the trailing ordinals in this same enable order. - var columns = "seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, is_archived"; + var columns = Internal.PcEventsRowReader.CoreSelectColumns; var eventOptions = _events.EventOptions; if (eventOptions.EnableCorrelationId) columns += ", correlation_id"; if (eventOptions.EnableCausationId) columns += ", causation_id"; diff --git a/src/Polecat/Events/Linq/EventListHandler.cs b/src/Polecat/Events/Linq/EventListHandler.cs index f67676f6..faf68804 100644 --- a/src/Polecat/Events/Linq/EventListHandler.cs +++ b/src/Polecat/Events/Linq/EventListHandler.cs @@ -8,7 +8,8 @@ namespace Polecat.Events.Linq; /// /// Reads IEvent objects from a multi-column result set on pc_events. -/// Column layout: seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, is_archived +/// Column layout: seq_id, id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type, +/// is_archived, bdata (#388), then the optional metadata columns. /// [UnconditionalSuppressMessage("Trimming", "IL2026:RequiresUnreferencedCode", Justification = "Class-level: hydrates IEvent via EventGraph.Wrap (which routes through ISerializer.FromJson). Event types are preserved by EventGraph registration on the caller side per the AOT publishing guide.")] @@ -42,11 +43,13 @@ public async Task> HandleAsync(DbDataReader reader, Cancel var tenantId = sqlReader.GetString(7); var dotNetTypeName = sqlReader.IsDBNull(8) ? null : sqlReader.GetString(8); var isArchived = sqlReader.GetBoolean(9); + // #388: non-null bdata means the payload is binary, not in the JSON `data` column. + var bdata = sqlReader.IsDBNull(10) ? null : sqlReader.GetFieldValue(10); var resolvedType = _events.ResolveEventType(dotNetTypeName); if (resolvedType == null) continue; - var data = _serializer.FromJson(resolvedType, json); + var data = _events.DeserializeEventData(resolvedType, json, bdata, _serializer); var mapping = _events.EventMappingFor(resolvedType); var @event = mapping.Wrap(data); @@ -71,7 +74,7 @@ public async Task> HandleAsync(DbDataReader reader, Cancel // #256: read the opt-in metadata columns appended to the full-event SELECT (when enabled), // in the same enable order the query provider emitted them. var options = _events.EventOptions; - var ordinal = 10; + var ordinal = 11; // #388: ordinal 10 is bdata if (options.EnableCorrelationId) { @event.CorrelationId = sqlReader.IsDBNull(ordinal) ? null : sqlReader.GetString(ordinal); diff --git a/src/Polecat/Events/Protected/OverwriteEventOperation.cs b/src/Polecat/Events/Protected/OverwriteEventOperation.cs index 7018b7c2..3b5046e7 100644 --- a/src/Polecat/Events/Protected/OverwriteEventOperation.cs +++ b/src/Polecat/Events/Protected/OverwriteEventOperation.cs @@ -11,13 +11,16 @@ internal class OverwriteEventOperation : Polecat.Internal.IStorageOperation private readonly EventGraph _events; private readonly IEvent _event; private readonly string _serializedData; + private readonly byte[]? _serializedBdata; private readonly string? _serializedHeaders; - public OverwriteEventOperation(EventGraph events, IEvent @event, string serializedData, string? serializedHeaders) + public OverwriteEventOperation(EventGraph events, IEvent @event, string serializedData, + byte[]? serializedBdata, string? serializedHeaders) { _events = events; _event = @event; _serializedData = serializedData; + _serializedBdata = serializedBdata; _serializedHeaders = serializedHeaders; } @@ -26,24 +29,32 @@ public OverwriteEventOperation(EventGraph events, IEvent @event, string serializ public void ConfigureCommand(ICommandBuilder builder) { - if (_serializedHeaders != null) + builder.Append($"UPDATE {_events.EventsTableName} SET data = "); + builder.AppendParameter(_serializedData); + + // #388: a binary event's payload lives in bdata, so masking has to rewrite THAT — rewriting + // only `data` would leave the original, unmasked payload readable in bdata, which for a GDPR + // masking operation is the whole point missed. The row keeps its own format either way: + // binary events stay binary (data holds the '{}' placeholder), JSON events stay JSON. + builder.Append(", bdata = "); + if (_serializedBdata is null) { - builder.Append($"UPDATE {_events.EventsTableName} SET data = "); - builder.AppendParameter(_serializedData); - builder.Append(", headers = "); - builder.AppendParameter(_serializedHeaders); - builder.Append(" WHERE seq_id = "); - builder.AppendParameter(_event.Sequence); - builder.Append(";"); + builder.Append("NULL"); } else { - builder.Append($"UPDATE {_events.EventsTableName} SET data = "); - builder.AppendParameter(_serializedData); - builder.Append(" WHERE seq_id = "); - builder.AppendParameter(_event.Sequence); - builder.Append(";"); + builder.AppendParameter(_serializedBdata); } + + if (_serializedHeaders != null) + { + builder.Append(", headers = "); + builder.AppendParameter(_serializedHeaders); + } + + builder.Append(" WHERE seq_id = "); + builder.AppendParameter(_event.Sequence); + builder.Append(";"); } public Task PostprocessAsync(DbDataReader reader, IList exceptions, CancellationToken token) diff --git a/src/Polecat/Events/Protected/ReplaceEventOperation.cs b/src/Polecat/Events/Protected/ReplaceEventOperation.cs index 7d06f128..d6f5cdf2 100644 --- a/src/Polecat/Events/Protected/ReplaceEventOperation.cs +++ b/src/Polecat/Events/Protected/ReplaceEventOperation.cs @@ -11,16 +11,18 @@ internal class ReplaceEventOperation : Polecat.Internal.IStorageOperation private readonly EventGraph _events; private readonly long _sequence; private readonly string _serializedData; + private readonly byte[]? _serializedBdata; private readonly string _eventTypeName; private readonly string _dotNetTypeName; private readonly Guid _newId; public ReplaceEventOperation(EventGraph events, long sequence, string serializedData, - string eventTypeName, string dotNetTypeName) + byte[]? serializedBdata, string eventTypeName, string dotNetTypeName) { _events = events; _sequence = sequence; _serializedData = serializedData; + _serializedBdata = serializedBdata; _eventTypeName = eventTypeName; _dotNetTypeName = dotNetTypeName; _newId = Guid.NewGuid(); @@ -34,6 +36,20 @@ public void ConfigureCommand(ICommandBuilder builder) { builder.Append($"UPDATE {_events.EventsTableName} SET data = "); builder.AppendParameter(_serializedData); + + // #388: the replacement body's own event type decides the row's format, so bdata is set (or + // cleared) in the same statement — a replacement that switched a row from binary to JSON + // without clearing bdata would keep reading the OLD payload. + builder.Append(", bdata = "); + if (_serializedBdata is null) + { + builder.Append("NULL"); + } + else + { + builder.AppendParameter(_serializedBdata); + } + builder.Append(", timestamp = SYSDATETIMEOFFSET(), type = "); builder.AppendParameter(_eventTypeName); builder.Append(", dotnet_type = "); diff --git a/src/Polecat/Events/Protected/StreamCompacting.cs b/src/Polecat/Events/Protected/StreamCompacting.cs index 37096457..cb8cbaba 100644 --- a/src/Polecat/Events/Protected/StreamCompacting.cs +++ b/src/Polecat/Events/Protected/StreamCompacting.cs @@ -70,11 +70,21 @@ await archiver.MaybeArchiveAsync(session, request, events, request.CancellationT var compacted = new Compacted(aggregate!, request.StreamId ?? Guid.Empty, request.StreamKey ?? string.Empty); - var serializedData = session.Serializer.ToJson(compacted); - var mapping = session.Options.EventGraph.EventMappingFor(typeof(Compacted)); + var graph = session.Options.EventGraph; + + // #388: the Compacted snapshot follows the same data/bdata rule as any other event — + // binary if Compacted itself is opted in, JSON otherwise. The event being REPLACED may + // have been the other format; ReplaceEventOperation writes both columns so the row ends up + // consistent either way. + var binary = graph.ResolveBinarySerializerFor(typeof(Compacted)); + var serializedBdata = binary?.Serialize(typeof(Compacted), compacted); + var serializedData = serializedBdata is null + ? session.Serializer.ToJson(compacted) + : EventGraph.JsonPlaceholderForBinaryEvent; + var mapping = graph.EventMappingFor(typeof(Compacted)); var replaceOp = new ReplaceEventOperation( - session.Options.EventGraph, request.Sequence, serializedData, + graph, request.Sequence, serializedData, serializedBdata, mapping.EventTypeName, mapping.DotNetTypeName); session.WorkTracker.Add(replaceOp); diff --git a/src/Polecat/Events/Schema/EventsTable.cs b/src/Polecat/Events/Schema/EventsTable.cs index 5bd47463..c02b69a8 100644 --- a/src/Polecat/Events/Schema/EventsTable.cs +++ b/src/Polecat/Events/Schema/EventsTable.cs @@ -44,6 +44,17 @@ public EventsTable(EventGraph events) // Event data — SQL Server 2025 native JSON type by default AddColumn("data", events.JsonColumnType).NotNull(); + // #388: binary event payload, the SQL Server counterpart of Marten's bytea `bdata` + // (marten#4515). Nullable and unconditional on purpose: `bdata IS NULL` is the per-ROW + // discriminator between a JSON event and a binary one, so JSON and binary events coexist in + // this table per event type. That is what makes the feature switchable on an existing store + // with no migration of existing event data — rows written before the column existed simply + // have bdata = NULL and keep reading through the JSON path — and just as safely switchable + // back off. Adding it always (rather than only when a serializer is configured) keeps the + // read projection one fixed shape; Weasel's additive delta adds the column on the next + // schema apply. + AddColumn("bdata", "varbinary(max)").AllowNulls(); + // Event type name for deserialization AddColumn("type", "varchar(500)").NotNull(); diff --git a/src/Polecat/Events/Storage/PolecatQuickAppendEventsOperation.cs b/src/Polecat/Events/Storage/PolecatQuickAppendEventsOperation.cs index eb2ecbb0..5500baa6 100644 --- a/src/Polecat/Events/Storage/PolecatQuickAppendEventsOperation.cs +++ b/src/Polecat/Events/Storage/PolecatQuickAppendEventsOperation.cs @@ -135,7 +135,7 @@ public void ConfigureCommand(Weasel.Core.ICommandBuilder builder, IStorageSessio builder.Append(", "); } - builder.Append("id, stream_id, version, data, type, timestamp, tenant_id, dotnet_type"); + builder.Append("id, stream_id, version, data, bdata, type, timestamp, tenant_id, dotnet_type"); if (options.EnableCorrelationId) builder.Append(", correlation_id"); if (options.EnableCausationId) builder.Append(", causation_id"); if (options.EnableHeaders) builder.Append(", headers"); @@ -164,8 +164,18 @@ public void ConfigureCommand(Weasel.Core.ICommandBuilder builder, IStorageSessio builder.Append(", "); Bind(builder, @event.Version, StorageColumnType.Long); + // #388: a binary event writes the '{}' placeholder to `data` and its payload to `bdata`; + // a JSON event writes the payload to `data` and NULL to `bdata`. Exactly one of the two + // is populated per row, and `bdata IS NULL` is what the read path dispatches on. + var bdata = _graph.SerializeEventBdata(@event); + + builder.Append(", "); + Bind(builder, + bdata is null ? session.Serializer.ToJson(@event.Data) : EventGraph.JsonPlaceholderForBinaryEvent, + StorageColumnType.Json); + builder.Append(", "); - Bind(builder, session.Serializer.ToJson(@event.Data), StorageColumnType.Json); + Bind(builder, (object?)bdata ?? DBNull.Value, StorageColumnType.Binary); builder.Append(", "); Bind(builder, @event.EventTypeName, StorageColumnType.String); diff --git a/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs b/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs index 21d1de1d..4fce484e 100644 --- a/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs +++ b/src/Polecat/Events/Storage/SqlServerEventStoreDialect.cs @@ -50,11 +50,14 @@ public QuickEventStorageDescriptor BuildQuickDescriptor(EventRegistry registry, quickAppendEventsSql: $"insert into {graph.EventsTableName} ", insertStreamSql: $"insert into {graph.StreamsTableName} (...) values (...)", updateStreamVersionSql: $"update {graph.StreamsTableName} set version = ... where ...", - // Polecat is STJ/JSON-only — no binary event serialization. Data is always JSON; bdata is - // always null. Serialize through the session serializer at write time (see the append op); - // the descriptor closure below covers any shared-op path that reaches for it. - serializeEventData: e => serializer.ToJson(e.Data), - serializeEventBdata: _ => null) + // #388: event data is JSON unless the event's type is opted into binary serialization, in + // which case `data` carries the '{}' placeholder and the payload goes to `bdata`. Exactly + // one of the two closures produces a payload for any given event. These cover any shared-op + // path that reaches for them; the SQL Server append operation binds both directly. + serializeEventData: e => graph.SerializeEventBdata(e) is null + ? serializer.ToJson(e.Data) + : EventGraph.JsonPlaceholderForBinaryEvent, + serializeEventBdata: graph.SerializeEventBdata) { IsGuidStreamIdentity = isGuid, Dialect = dialect, diff --git a/src/Polecat/Storage/SqlServerStorageDialect.cs b/src/Polecat/Storage/SqlServerStorageDialect.cs index 7284f75c..949a74d8 100644 --- a/src/Polecat/Storage/SqlServerStorageDialect.cs +++ b/src/Polecat/Storage/SqlServerStorageDialect.cs @@ -95,6 +95,8 @@ public void SetParameterType(System.Data.Common.DbParameter parameter, StorageCo // Polecat binds JSON as string values throughout; SQL Server 2025's native JSON // column type (and nvarchar-backed JSON) accepts nvarchar input. StorageColumnType.Json => SqlDbType.NVarChar, + // #388: the binary event payload column (pc_events.bdata) is varbinary(max). + StorageColumnType.Binary => SqlDbType.VarBinary, _ => throw new ArgumentOutOfRangeException(nameof(type)) }; diff --git a/src/Polecat/StoreOptions.cs b/src/Polecat/StoreOptions.cs index d78d31c0..5ecb44c9 100644 --- a/src/Polecat/StoreOptions.cs +++ b/src/Polecat/StoreOptions.cs @@ -490,6 +490,32 @@ public void AddEventTypes(IEnumerable eventTypes) foreach (var eventType in eventTypes) EventGraph!.AddEventType(eventType); } + /// + /// #388: store-wide fallback for event types + /// marked with that have no explicit per-type + /// registration via . Null by default — every event type + /// stays on the JSON path. Mirrors Marten's opts.Events.DefaultBinarySerializer. + /// + public Polecat.Events.IEventBinarySerializer? DefaultBinarySerializer + { + get => EventGraph!.DefaultBinarySerializer; + set => EventGraph!.DefaultBinarySerializer = value; + } + + /// + /// #388: opt into binary serialization — its payload is written to + /// the pc_events.bdata column instead of data, and read back through the same + /// serializer. Per event type rather than store-wide, so JSON and binary rows coexist in one + /// table and the feature can be switched on (or back off) for an existing store with no data + /// migration. Mirrors Marten's opts.Events.UseBinarySerializer<TEvent>(serializer). + /// + public EventStoreOptions UseBinarySerializer(Polecat.Events.IEventBinarySerializer serializer) + where TEvent : notnull + { + EventGraph!.UseBinarySerializer(serializer); + return this; + } + /// /// Register a tag type for Dynamic Consistency Boundary (DCB) support. /// Creates a tag table with an auto-generated suffix.