diff --git a/src/Marten/EventStorage/ClosedShapeEventDocumentStorage.cs b/src/Marten/EventStorage/ClosedShapeEventDocumentStorage.cs index e645d46d38..6e3beeb6a5 100644 --- a/src/Marten/EventStorage/ClosedShapeEventDocumentStorage.cs +++ b/src/Marten/EventStorage/ClosedShapeEventDocumentStorage.cs @@ -110,6 +110,16 @@ public override Weasel.Storage.IStorageOperation QuickAppendEvents(StreamAction : StringStorage.QuickAppendEvents(stream); } + // #4968: route stream archive through the shared Weasel seam. Marten only ever archives (never + // un-archives through this path), so pass archived: true; the dialect's factory builds the Postgres + // ArchiveStreamOperation. + public override Weasel.Storage.IStorageOperation ArchiveStream(object streamId, string tenantId) + { + return Events.StreamIdentity == StreamIdentity.AsGuid + ? GuidStorage.ArchiveStream(streamId, tenantId, true) + : StringStorage.ArchiveStream(streamId, tenantId, true); + } + public override Weasel.Storage.IStorageOperation InsertStream(StreamAction stream) { return Events.StreamIdentity == StreamIdentity.AsGuid diff --git a/src/Marten/EventStorage/Dialects/PostgresEventStoreDialect.cs b/src/Marten/EventStorage/Dialects/PostgresEventStoreDialect.cs index bfce25ac5b..931faea0ab 100644 --- a/src/Marten/EventStorage/Dialects/PostgresEventStoreDialect.cs +++ b/src/Marten/EventStorage/Dialects/PostgresEventStoreDialect.cs @@ -12,12 +12,14 @@ using Marten.Events.CodeGeneration; using Marten.Events.Schema; using Marten.Exceptions; +using Marten.Events.Operations; using Marten.Services; using Marten.Storage; using Marten.Storage.Metadata; using Npgsql; using NpgsqlTypes; using Weasel.Postgresql; +using Weasel.Storage; namespace Marten.EventStorage.Dialects; @@ -716,4 +718,34 @@ private static string BuildUpdateStreamVersionSql(EventGraph graph) => /// private static string BuildAssertStreamVersionSql(EventGraph graph) => $"select version from {graph.DatabaseSchemaName}.{StreamsTable.TableName} where id = "; + + /// + /// #4968: vend the shared Weasel.Storage.Events auxiliary-operation seam. Marten routes the stream + /// ARCHIVE operation through it, single-sourcing the invocation contract with Polecat. Tombstone and + /// progression stay Marten-local and are intentionally left null: + /// + /// Tombstone — Marten's ESTABLISHES the tombstone + /// bookkeeping stream (keyed only by tenant, no stream id) rather than hard-deleting a stream, so it + /// does not match the seam's TombstoneStream(streamId, tenantId) hard-delete contract. + /// Progression — Marten's + /// guards its update with the sequence FLOOR (where last_seq_id = floor), which the seam's + /// UpdateProgress(name, sequence, upsert) factory signature cannot carry; routing it would drop + /// the optimistic-concurrency guard. + /// + /// + public EventAuxiliaryOperations BuildAuxiliaryOperations(EventRegistry registry) + { + var graph = (EventGraph)registry; + return new EventAuxiliaryOperations( + ArchiveStream: (streamId, tenantId, archived) => + { + if (!archived) + { + throw new NotSupportedException( + "Marten does not support un-archiving a stream; only archive is issued through the event-store seam."); + } + + return new ArchiveStreamOperation(graph, streamId) { TenantId = tenantId }; + }); + } } diff --git a/src/Marten/Events/EventDocumentStorage.cs b/src/Marten/Events/EventDocumentStorage.cs index 3a435803f3..9e375a2a3f 100644 --- a/src/Marten/Events/EventDocumentStorage.cs +++ b/src/Marten/Events/EventDocumentStorage.cs @@ -253,6 +253,7 @@ public abstract Weasel.Storage.IStorageOperation AppendEvent(EventGraph events, public abstract IQueryHandler QueryForStream(StreamAction stream); public abstract Weasel.Storage.IStorageOperation UpdateStreamVersion(StreamAction stream); public abstract Weasel.Storage.IStorageOperation AssertStreamVersion(StreamAction stream); + public abstract Weasel.Storage.IStorageOperation ArchiveStream(object streamId, string tenantId); public string StreamStateSelectSql => Marten.EventStorage.StreamStateSql.Build(Events); diff --git a/src/Marten/Events/EventStore.Archiving.cs b/src/Marten/Events/EventStore.Archiving.cs index ca04a3979b..3986d0146f 100644 --- a/src/Marten/Events/EventStore.Archiving.cs +++ b/src/Marten/Events/EventStore.Archiving.cs @@ -8,13 +8,13 @@ internal partial class EventStore { public void ArchiveStream(Guid streamId) { - var op = new ArchiveStreamOperation(_store.Events, streamId){TenantId = _session.TenantId}; + var op = _session.EventStorage().ArchiveStream(streamId, _session.TenantId); _session.QueueOperation(op); } public void ArchiveStream(string streamKey) { - var op = new ArchiveStreamOperation(_store.Events, streamKey){TenantId = _session.TenantId}; + var op = _session.EventStorage().ArchiveStream(streamKey, _session.TenantId); _session.QueueOperation(op); } } diff --git a/src/Marten/Events/IEventStorage.cs b/src/Marten/Events/IEventStorage.cs index cf6f48c063..4439e981b9 100644 --- a/src/Marten/Events/IEventStorage.cs +++ b/src/Marten/Events/IEventStorage.cs @@ -76,6 +76,12 @@ public interface IEventStorage: ISelector, ISelector, IDocu Weasel.Storage.IStorageOperation QuickAppendEvents(StreamAction stream); + /// + /// #4968: create a storage operation to archive (soft-delete) a single stream and its events, + /// routed through the shared Weasel.Storage event-store auxiliary-operation seam. + /// + Weasel.Storage.IStorageOperation ArchiveStream(object streamId, string tenantId); + Weasel.Storage.IStorageOperation QuickAppendEventWithVersion(StreamAction stream, IEvent e); } diff --git a/src/Marten/Internal/Sessions/DocumentSessionBase.ProjectionStorage.cs b/src/Marten/Internal/Sessions/DocumentSessionBase.ProjectionStorage.cs index d10cabacbb..fbb50ed7d2 100644 --- a/src/Marten/Internal/Sessions/DocumentSessionBase.ProjectionStorage.cs +++ b/src/Marten/Internal/Sessions/DocumentSessionBase.ProjectionStorage.cs @@ -200,51 +200,55 @@ public void StoreProjection(TDoc aggregate, IEvent lastEvent, AggregationScope s public void ArchiveStream(TId sliceId, string tenantId) { - var op = archiveOperationBuilderFor()(sliceId); - op.TenantId = tenantId; + // #4968: unwrap to the raw stream id and build the archive operation through the shared Weasel + // seam (session.EventStorage().ArchiveStream) rather than instantiating ArchiveStreamOperation here. + var streamId = streamIdUnwrapperFor()(sliceId); + var op = _session.EventStorage().ArchiveStream(streamId, tenantId); _session.QueueOperation(op); } - private static ImHashMap _archiveBuilders = ImHashMap.Empty; + private static ImHashMap _archiveStreamIdUnwrappers = ImHashMap.Empty; - private Func archiveOperationBuilderFor() + // #4968: resolve the raw stream id (Guid/string) from TId, unwrapping a strong-typed id. The archive + // operation itself is now built through the dialect's seam factory, so this only handles id extraction. + private Func streamIdUnwrapperFor() { - if (_archiveBuilders.TryFind(typeof(TId), out var raw)) + if (_archiveStreamIdUnwrappers.TryFind(typeof(TId), out var raw)) { - return (Func)raw; + return (Func)raw; } - Func builder = null; + Func unwrapper; if (((IMartenSession)_session).Options.Events.StreamIdentity == StreamIdentity.AsGuid) { if (typeof(TId) == typeof(Guid)) { - builder = id => new ArchiveStreamOperation(((IMartenSession)_session).Options.EventGraph, id); + unwrapper = id => id; } else { var valueType = ValueTypeInfo.ForType(typeof(TId)); var unWrapper = valueType.UnWrapper(); - builder = id => new ArchiveStreamOperation(((IMartenSession)_session).Options.EventGraph, unWrapper(id)); + unwrapper = id => unWrapper(id); } } else { if (typeof(TId) == typeof(string)) { - builder = id => new ArchiveStreamOperation(((IMartenSession)_session).Options.EventGraph, id); + unwrapper = id => id; } else { var valueType = ValueTypeInfo.ForType(typeof(TId)); var unWrapper = valueType.UnWrapper(); - builder = id => new ArchiveStreamOperation(((IMartenSession)_session).Options.EventGraph, unWrapper(id)); + unwrapper = id => unWrapper(id); } } - _archiveBuilders = _archiveBuilders.AddOrUpdate(typeof(TId), builder); - return builder; + _archiveStreamIdUnwrappers = _archiveStreamIdUnwrappers.AddOrUpdate(typeof(TId), unwrapper); + return unwrapper; } //TODO fix in IProjectionStorage