diff --git a/proto.lock b/proto.lock index dc67c74ab5..44878fffc3 100644 --- a/proto.lock +++ b/proto.lock @@ -4543,6 +4543,16 @@ "type": "bool" } ], + "maps": [ + { + "key_type": "string", + "field": { + "id": 12, + "name": "annotations", + "type": "string" + } + } + ], "messages": [ { "name": "Transient", @@ -4614,6 +4624,16 @@ "name": "no_emit_options", "type": "event_store.client.Empty" } + ], + "maps": [ + { + "key_type": "string", + "field": { + "id": 5, + "name": "annotations", + "type": "string" + } + } ] } ] diff --git a/src/EventStore.ClusterNode/Components/Services/ProjectionBrowserService.cs b/src/EventStore.ClusterNode/Components/Services/ProjectionBrowserService.cs index 7f18d27ee3..10704dd56a 100644 --- a/src/EventStore.ClusterNode/Components/Services/ProjectionBrowserService.cs +++ b/src/EventStore.ClusterNode/Components/Services/ProjectionBrowserService.cs @@ -2,8 +2,10 @@ using System.Collections.Generic; using System.Linq; using System.Security.Claims; +using System.Text.Json; using System.Threading; using System.Threading.Tasks; +using EventStore.Common.Utils; using EventStore.Core.Bus; using EventStore.Core.Messaging; using EventStore.Core.Services.Transport.Grpc; @@ -185,7 +187,8 @@ public async Task Create( checkpointsEnabled, emitEnabled, trackEmittedStreams, - enableRunAs: true), + enableRunAs: true, + definitionMetadata: DefinitionMetadataFor("create")), cancellationToken); } @@ -217,7 +220,8 @@ public async Task UpdateQuery( name, new ProjectionManagementMessage.RunAs(CurrentUser), request.Query, - request.EmitEnabled), + request.EmitEnabled, + DefinitionMetadataFor("update")), cancellationToken); } @@ -605,6 +609,17 @@ private async Task ReadStatistics( private ClaimsPrincipal CurrentUser => httpContextAccessor.HttpContext?.User ?? new ClaimsPrincipal(new ClaimsIdentity()); + private static byte[] DefinitionMetadataFor(string operation) + { + var annotations = new Dictionary + { + ["trogondb.com/tool"] = "embedded-ui", + ["trogondb.com/tool-version"] = VersionInfo.Version, + ["trogondb.com/operation"] = operation + }; + return JsonSerializer.SerializeToUtf8Bytes(annotations); + } + private Task HasAccess(Operation operation, CancellationToken cancellationToken) => authorizationProvider.CheckAccessAsync(CurrentUser, operation, cancellationToken).AsTask(); diff --git a/src/EventStore.Core.Tests/Helpers/TestFixtureWithExistingEvents.cs b/src/EventStore.Core.Tests/Helpers/TestFixtureWithExistingEvents.cs index d6a228d6ee..2544afaf57 100644 --- a/src/EventStore.Core.Tests/Helpers/TestFixtureWithExistingEvents.cs +++ b/src/EventStore.Core.Tests/Helpers/TestFixtureWithExistingEvents.cs @@ -556,7 +556,9 @@ record = new EventRecord( eventNumber, tfPosition, correlationId, e.EventId, tfPosition, 0, streamId, ExpectedVersion.Any, _timeProvider.UtcNow, - PrepareFlags.SingleWrite | (e.IsJson ? PrepareFlags.IsJson : PrepareFlags.None), + PrepareFlags.SingleWrite + | (e.IsJson ? PrepareFlags.IsJson : PrepareFlags.None) + | (e.IsPropertyMetadata ? PrepareFlags.IsPropertyMetadata : PrepareFlags.None), e.EventType, e.Data, e.Metadata) }); //NOTE: DO NOT MAKE ARRAY foreach (var eventRecord in eventRecords) @@ -576,7 +578,7 @@ record = public void Handle(StorageMessage.EventCommitted msg) { - var metadata = Json.ParseJson(msg.Event.Metadata); + var metadata = Json.ParseJson(msg.Event.CustomMetadata); if (metadata != null) { EventTimeStamps.Add(metadata.Added.ToUniversalTime()); diff --git a/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_posting_a_persistent_projection_and_writes_succeed.cs b/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_posting_a_persistent_projection_and_writes_succeed.cs index 776f35b524..d2c12a5dd9 100644 --- a/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_posting_a_persistent_projection_and_writes_succeed.cs +++ b/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_posting_a_persistent_projection_and_writes_succeed.cs @@ -15,6 +15,8 @@ namespace EventStore.Projections.Core.Tests.Services.projections_manager; public class when_posting_a_persistent_projection_and_writes_succeed : TestFixtureWithProjectionCoreAndManagementServices { + private static readonly byte[] DefinitionMetadata = { 1, 2, 3 }; + protected override void Given() { NoStream("$projections-test-projection-order"); @@ -34,7 +36,8 @@ protected override IEnumerable When() new ProjectionManagementMessage.Command.Post( _bus, ProjectionMode.Continuous, _projectionName, ProjectionManagementMessage.RunAs.System, "JS", @"fromAll().when({$any:function(s,e){return s;}});", - enabled: true, checkpointsEnabled: true, emitEnabled: true, trackEmittedStreams: true); + enabled: true, checkpointsEnabled: true, emitEnabled: true, trackEmittedStreams: true, + definitionMetadata: DefinitionMetadata); } [Test, Category("v8")] @@ -57,6 +60,17 @@ public void a_projection_updated_event_is_written() v => v.Events[0].EventType == ProjectionEventTypes.ProjectionUpdated)); } + [Test, Category("v8")] + public void projection_definition_metadata_is_written() + { + var writtenEvent = _consumer.HandledMessages.OfType() + .Single(v => v.EventStreamId == "$projections-test-projection") + .Events[0]; + + Assert.IsTrue(writtenEvent.IsPropertyMetadata); + CollectionAssert.AreEqual(DefinitionMetadata, writtenEvent.Metadata); + } + [Test, Category("v8")] public void a_projection_updated_message_is_published() { diff --git a/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_updating_a_persistent_projection_query_text.cs b/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_updating_a_persistent_projection_query_text.cs index 18fac433c2..338a39595e 100644 --- a/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_updating_a_persistent_projection_query_text.cs +++ b/src/EventStore.Projections.Core.Tests/Services/projections_manager/when_updating_a_persistent_projection_query_text.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Linq; +using EventStore.Core.Messages; using EventStore.Core.Messaging; using EventStore.Core.Tests; using EventStore.Projections.Core.Messages; @@ -13,6 +14,8 @@ namespace EventStore.Projections.Core.Tests.Services.projections_manager; [TestFixture(typeof(LogFormat.V2), typeof(string))] public class when_updating_a_persistent_projection_query_text : TestFixtureWithProjectionCoreAndManagementServices { + private static readonly byte[] UpdatedDefinitionMetadata = { 4, 5, 6 }; + protected override void Given() { NoStream("$projections-test-projection"); @@ -41,7 +44,7 @@ protected override IEnumerable When() yield return (new ProjectionManagementMessage.Command.UpdateQuery( _bus, _projectionName, ProjectionManagementMessage.RunAs.System, - _newProjectionSource, emitEnabled: null)); + _newProjectionSource, emitEnabled: null, definitionMetadata: UpdatedDefinitionMetadata)); } [Test, Category("v8")] @@ -113,4 +116,16 @@ public void correct_query_has_been_prepared() Assert.IsNotNull(lastPrepared); Assert.IsFalse(lastPrepared.SourceDefinition.AllEvents); } + + [Test, Category("v8")] + public void projection_definition_metadata_is_written() + { + var writtenEvent = _consumer.HandledMessages.OfType() + .Where(v => v.EventStreamId == "$projections-test-projection") + .Last() + .Events[0]; + + Assert.IsTrue(writtenEvent.IsPropertyMetadata); + CollectionAssert.AreEqual(UpdatedDefinitionMetadata, writtenEvent.Metadata); + } } diff --git a/src/EventStore.Projections.Core/Messages/ProjectionManagementMessage.cs b/src/EventStore.Projections.Core/Messages/ProjectionManagementMessage.cs index b76f4097b4..844f369e70 100644 --- a/src/EventStore.Projections.Core/Messages/ProjectionManagementMessage.cs +++ b/src/EventStore.Projections.Core/Messages/ProjectionManagementMessage.cs @@ -57,11 +57,12 @@ public class ProjectionPost public bool EmitEnabled { get; } public bool EnableRunAs { get; } public bool TrackEmittedStreams { get; } + public byte[] DefinitionMetadata { get; } public ProjectionPost( ProjectionMode mode, RunAs runAs, string name, string handlerType, string query, bool enabled, bool checkpointsEnabled, bool emitEnabled, bool enableRunAs, - bool trackEmittedStreams) + bool trackEmittedStreams, byte[] definitionMetadata = null) { Mode = mode; RunAs = runAs; @@ -73,6 +74,7 @@ public ProjectionPost( EmitEnabled = emitEnabled; EnableRunAs = enableRunAs; TrackEmittedStreams = trackEmittedStreams; + DefinitionMetadata = definitionMetadata ?? Empty.ByteArray; } } } @@ -89,11 +91,12 @@ public partial class Post : ControlMessage private readonly bool _emitEnabled; private readonly bool _enableRunAs; private readonly bool _trackEmittedStreams; + private readonly byte[] _definitionMetadata; public Post( IEnvelope envelope, ProjectionMode mode, string name, RunAs runAs, string handlerType, string query, bool enabled, bool checkpointsEnabled, bool emitEnabled, bool trackEmittedStreams, - bool enableRunAs = false) + bool enableRunAs = false, byte[] definitionMetadata = null) : base(envelope, runAs) { _name = name; @@ -105,12 +108,13 @@ public Post( _emitEnabled = emitEnabled; _trackEmittedStreams = trackEmittedStreams; _enableRunAs = enableRunAs; + _definitionMetadata = definitionMetadata ?? Empty.ByteArray; } public Post( IEnvelope envelope, ProjectionMode mode, string name, RunAs runAs, Type handlerType, string query, bool enabled, bool checkpointsEnabled, bool emitEnabled, bool trackEmittedStreams, - bool enableRunAs = false) + bool enableRunAs = false, byte[] definitionMetadata = null) : base(envelope, runAs) { _name = name; @@ -122,6 +126,7 @@ public Post( _emitEnabled = emitEnabled; _trackEmittedStreams = trackEmittedStreams; _enableRunAs = enableRunAs; + _definitionMetadata = definitionMetadata ?? Empty.ByteArray; } // shortcut for posting ad-hoc JS queries @@ -136,6 +141,7 @@ public Post(IEnvelope envelope, RunAs runAs, string query, bool enabled) _checkpointsEnabled = false; _emitEnabled = false; _trackEmittedStreams = false; + _definitionMetadata = Empty.ByteArray; } public ProjectionMode Mode @@ -182,6 +188,11 @@ public bool TrackEmittedStreams { get { return _trackEmittedStreams; } } + + public byte[] DefinitionMetadata + { + get { return _definitionMetadata; } + } } [DerivedMessage(ProjectionMessage.Management)] @@ -271,14 +282,17 @@ public partial class UpdateQuery : ControlMessage private readonly string _name; private readonly string _query; private readonly bool? _emitEnabled; + private readonly byte[] _definitionMetadata; public UpdateQuery( - IEnvelope envelope, string name, RunAs runAs, string query, bool? emitEnabled) + IEnvelope envelope, string name, RunAs runAs, string query, bool? emitEnabled, + byte[] definitionMetadata = null) : base(envelope, runAs) { _name = name; _query = query; _emitEnabled = emitEnabled; + _definitionMetadata = definitionMetadata ?? Empty.ByteArray; } public string Query @@ -295,6 +309,11 @@ public bool? EmitEnabled { get { return _emitEnabled; } } + + public byte[] DefinitionMetadata + { + get { return _definitionMetadata; } + } } [DerivedMessage(ProjectionMessage.Management)] diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs index 97658491dc..d1d1411726 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs @@ -74,7 +74,8 @@ public override async Task Create(CreateReq request, ServerCallConte var envelope = new CallbackEnvelope(OnMessage); _publisher.Publish(new ProjectionManagementMessage.Command.Post(envelope, projectionMode, name, runAs, - handlerType, options.Query, enabled, checkpointsEnabled, emitEnabled, trackEmittedStreams, true)); + handlerType, options.Query, enabled, checkpointsEnabled, emitEnabled, trackEmittedStreams, true, + ToMetadata(options.Annotations))); await createdSource.Task; diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Update.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Update.cs index be3379d76e..8cbaf2f0b2 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Update.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Update.cs @@ -38,7 +38,7 @@ public override async Task Update(UpdateReq request, ServerCallConte var envelope = new CallbackEnvelope(OnMessage); _publisher.Publish( new ProjectionManagementMessage.Command.UpdateQuery(envelope, name, runAs, query, - emitEnabled)); + emitEnabled, ToMetadata(options.Annotations))); await updatedSource.Task; diff --git a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.cs b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.cs index 6930d48ed2..83117ccf5e 100644 --- a/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.cs +++ b/src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.cs @@ -7,6 +7,8 @@ using EventStore.Core.Services.Transport.Grpc; using EventStore.Plugins.Authorization; using EventStore.Projections.Core.Messages; +using Google.Protobuf; +using Google.Protobuf.Collections; using Google.Protobuf.WellKnownTypes; using Grpc.Core; @@ -103,6 +105,19 @@ private static Struct GetProtoStruct(JsonElement element) private static string ToJson(T value) where T : class => value is null ? "{}" : value.ToJson(); + internal static byte[] ToMetadata(MapField annotations) + { + if (annotations is null || annotations.Count == 0) + { + return EventStore.Common.Utils.Empty.ByteArray; + } + + var metadata = annotations.ToDictionary( + annotation => annotation.Key, + annotation => annotation.Value); + return JsonSerializer.SerializeToUtf8Bytes(metadata); + } + private static Value GetProtoValue(string value, bool isJson) { if (string.IsNullOrEmpty(value)) diff --git a/src/EventStore.Projections.Core/Services/Management/ManagedProjection.cs b/src/EventStore.Projections.Core/Services/Management/ManagedProjection.cs index 62a1cd74e1..40a8abb7c8 100644 --- a/src/EventStore.Projections.Core/Services/Management/ManagedProjection.cs +++ b/src/EventStore.Projections.Core/Services/Management/ManagedProjection.cs @@ -133,6 +133,8 @@ private readonly private ManagedProjectionStateBase _stateHandler; private IEnvelope _lastReplyEnvelope; + private byte[] _pendingDefinitionMetadata = Empty.ByteArray; + private long _pendingDefinitionMetadataVersion; private bool _writing; private IODispatcher _ioDispatcher; private ProjectionConfig _projectionConfig; @@ -439,6 +441,7 @@ public void Handle(ProjectionManagementMessage.Command.UpdateQuery message) _lastAccessed = _timeProvider.UtcNow; Prepared = false; + SetPendingDefinitionMetadata(message.DefinitionMetadata); UpdateQuery(message); UpdateProjectionVersion(); SetLastReplyEnvelope(message.Envelope); @@ -571,17 +574,19 @@ public void Handle(ProjectionManagementMessage.Command.UpdateConfig message) PersistedProjectionState.MaxWriteBatchLength = message.MaxWriteBatchLength; PersistedProjectionState.MaxAllowedWritesInFlight = message.MaxAllowedWritesInFlight; PersistedProjectionState.ProjectionExecutionTimeout = message.ProjectionExecutionTimeout; + SetPendingDefinitionMetadata(Empty.ByteArray); UpdateProjectionVersion(); _pendingWritePersistedState = true; if (!TryCreatePersistedStateEvent(Guid.NewGuid(), PersistedProjectionState, - ProjectionNamesBuilder.ProjectionsStreamPrefix + _name, out var persistedStateEvent, out var recordSize)) + ProjectionNamesBuilder.ProjectionsStreamPrefix + _name, out var persistedStateEvent, out var recordSize, + out var definitionMetadataVersion)) { FailPersistedStateWrite(recordSize, message.Envelope); return; } - WritePersistedState(persistedStateEvent); + WritePersistedState(persistedStateEvent, definitionMetadataVersion); message.Envelope.ReplyWith(new ProjectionManagementMessage.Updated(message.Name)); } @@ -725,9 +730,10 @@ private bool IsExpiredProjection() _lastAccessed.Add(_projectionsQueryExpiry) < _timeProvider.UtcNow && _persistedStateLoaded; } - public void InitializeNew(PersistedState persistedState, IEnvelope replyEnvelope) + public void InitializeNew(PersistedState persistedState, IEnvelope replyEnvelope, byte[] definitionMetadata = null) { LoadPersistedState(persistedState); + SetPendingDefinitionMetadata(definitionMetadata); UpdateProjectionVersion(); _pendingWritePersistedState = true; SetLastReplyEnvelope(replyEnvelope); @@ -853,14 +859,15 @@ internal void WriteStartOrLoadStopped() if (_pendingWritePersistedState) { if (!TryCreatePersistedStateEvent(Guid.NewGuid(), PersistedProjectionState, - ProjectionNamesBuilder.ProjectionsStreamPrefix + _name, out var persistedStateEvent, out var recordSize)) + ProjectionNamesBuilder.ProjectionsStreamPrefix + _name, out var persistedStateEvent, out var recordSize, + out var definitionMetadataVersion)) { FailPersistedStateWrite(recordSize, _lastReplyEnvelope); _lastReplyEnvelope = null; return; } - WritePersistedState(persistedStateEvent); + WritePersistedState(persistedStateEvent, definitionMetadataVersion); } else { @@ -874,10 +881,12 @@ internal void StartCompleted() } private bool TryCreatePersistedStateEvent(Guid correlationId, PersistedState persistedState, string eventStreamId, - out ClientMessage.WriteEvents persistedStateEvent, out int recordSize) + out ClientMessage.WriteEvents persistedStateEvent, out int recordSize, out long definitionMetadataVersion) { var data = persistedState.ToJsonBytes(); - recordSize = Event.SizeOnDisk(ProjectionEventTypes.ProjectionUpdated, data, Empty.ByteArray); + var metadata = _pendingDefinitionMetadata; + definitionMetadataVersion = _pendingDefinitionMetadataVersion; + recordSize = Event.SizeOnDisk(ProjectionEventTypes.ProjectionUpdated, data, metadata); if (recordSize > TFConsts.EffectiveMaxLogRecordSize) { persistedStateEvent = null; @@ -887,14 +896,29 @@ private bool TryCreatePersistedStateEvent(Guid correlationId, PersistedState per persistedStateEvent = new ClientMessage.WriteEvents( correlationId, correlationId, _writeDispatcher.Envelope, true, eventStreamId, ExpectedVersion.Any, new Event(Guid.NewGuid(), ProjectionEventTypes.ProjectionUpdated, true, data, - Empty.ByteArray), + metadata.Length > 0, metadata), SystemAccounts.System); return true; } + private void SetPendingDefinitionMetadata(byte[] definitionMetadata) + { + _pendingDefinitionMetadata = definitionMetadata ?? Empty.ByteArray; + _pendingDefinitionMetadataVersion++; + } + + private void ClearPendingDefinitionMetadata(long definitionMetadataVersion) + { + if (_pendingDefinitionMetadataVersion == definitionMetadataVersion) + { + SetPendingDefinitionMetadata(Empty.ByteArray); + } + } + private void FailPersistedStateWrite(int recordSize, IEnvelope replyEnvelope) { _pendingWritePersistedState = false; + SetPendingDefinitionMetadata(Empty.ByteArray); var reason = $"Projection '{_name}' definition record size ({recordSize} bytes) exceeds the maximum of {TFConsts.EffectiveMaxLogRecordSize} bytes."; if (Created) @@ -910,13 +934,14 @@ private void FailPersistedStateWrite(int recordSize, IEnvelope replyEnvelope) replyEnvelope?.ReplyWith(new ProjectionManagementMessage.RecordTooLarge(reason)); } - private void WritePersistedState(ClientMessage.WriteEvents persistedStateEvent) + private void WritePersistedState(ClientMessage.WriteEvents persistedStateEvent, long definitionMetadataVersion) { if (Mode == ProjectionMode.Transient) { //TODO: move to common completion procedure _lastWrittenVersion = PersistedProjectionState.Version ?? -1; _pendingWritePersistedState = false; + ClearPendingDefinitionMetadata(definitionMetadataVersion); StartOrLoadStopped(); return; } @@ -924,11 +949,12 @@ private void WritePersistedState(ClientMessage.WriteEvents persistedStateEvent) _writing = true; _writeDispatcher.Publish( persistedStateEvent, - m => WritePersistedStateCompleted(m, persistedStateEvent, persistedStateEvent.EventStreamId)); + m => WritePersistedStateCompleted(m, persistedStateEvent, persistedStateEvent.EventStreamId, + definitionMetadataVersion)); } private void WritePersistedStateCompleted(ClientMessage.WriteEventsCompleted message, - ClientMessage.WriteEvents eventToRetry, string eventStreamId) + ClientMessage.WriteEvents eventToRetry, string eventStreamId, long definitionMetadataVersion) { if (!_writing) { @@ -939,6 +965,7 @@ private void WritePersistedStateCompleted(ClientMessage.WriteEventsCompleted mes { _logger.Information("'{projection}' projection source has been written", _name); _pendingWritePersistedState = false; + ClearPendingDefinitionMetadata(definitionMetadataVersion); var writtenEventNumber = message.FirstEventNumber; if (writtenEventNumber != (PersistedProjectionState.Version ?? writtenEventNumber)) { @@ -961,7 +988,7 @@ private void WritePersistedStateCompleted(ClientMessage.WriteEventsCompleted mes OperationResult.WrongExpectedVersion) { _logger.Information("Retrying write projection source for {projection}", _name); - WritePersistedState(eventToRetry); + WritePersistedState(eventToRetry, definitionMetadataVersion); } else { @@ -1170,6 +1197,7 @@ private void Abort() public void Fault(string reason) { _logger.Error("The '{projection}' projection faulted due to '{e}'", _name, reason); + SetPendingDefinitionMetadata(Empty.ByteArray); SetState(ManagedProjectionState.Faulted); _faultedReason = reason; } diff --git a/src/EventStore.Projections.Core/Services/Management/ProjectionManager.cs b/src/EventStore.Projections.Core/Services/Management/ProjectionManager.cs index 4b71c5a802..42840d7eda 100644 --- a/src/EventStore.Projections.Core/Services/Management/ProjectionManager.cs +++ b/src/EventStore.Projections.Core/Services/Management/ProjectionManager.cs @@ -1484,6 +1484,7 @@ public class NewProjectionInitializer private readonly ProjectionManagementMessage.RunAs _runAs; private readonly IEnvelope _replyEnvelope; private readonly string _name; + private readonly byte[] _definitionMetadata; public NewProjectionInitializer( long projectionId, @@ -1497,7 +1498,8 @@ public NewProjectionInitializer( bool enableRunAs, bool trackEmittedStreams, ProjectionManagementMessage.RunAs runAs, - IEnvelope replyEnvelope) + IEnvelope replyEnvelope, + byte[] definitionMetadata = null) { if (projectionMode >= ProjectionMode.Continuous && !checkpointsEnabled) { @@ -1521,6 +1523,7 @@ public NewProjectionInitializer( _runAs = runAs; _replyEnvelope = replyEnvelope; _name = name; + _definitionMetadata = definitionMetadata ?? Empty.ByteArray; } public void CreateAndInitializeNewProjection( @@ -1553,7 +1556,8 @@ public void CreateAndInitializeNewProjection( ProjectionSubsystemVersion = ProjectionsSubsystem.VERSION, ProjectionExecutionTimeout = null }, - _replyEnvelope); + _replyEnvelope, + _definitionMetadata); } } @@ -1614,12 +1618,13 @@ public class PendingProjection public bool EnableRunAs { get; } public bool TrackEmittedStreams { get; } public long ProjectionId { get; } + public byte[] DefinitionMetadata { get; } public PendingProjection( long projectionId, ProjectionMode mode, SerializedRunAs runAs, string name, string handlerType, string query, bool enabled, bool checkpointsEnabled, bool emitEnabled, bool enableRunAs, - bool trackEmittedStreams) + bool trackEmittedStreams, byte[] definitionMetadata) { ProjectionId = projectionId; Mode = mode; @@ -1632,20 +1637,23 @@ public PendingProjection( EmitEnabled = emitEnabled; EnableRunAs = enableRunAs; TrackEmittedStreams = trackEmittedStreams; + DefinitionMetadata = definitionMetadata ?? Empty.ByteArray; } public PendingProjection(long projectionId, ProjectionManagementMessage.Command.PostBatch.ProjectionPost projection) : this(projectionId, projection.Mode, projection.RunAs, projection.Name, projection.HandlerType, projection.Query, projection.Enabled, projection.CheckpointsEnabled, - projection.EmitEnabled, projection.EnableRunAs, projection.TrackEmittedStreams) + projection.EmitEnabled, projection.EnableRunAs, projection.TrackEmittedStreams, + projection.DefinitionMetadata) { } public PendingProjection(long projectionId, ProjectionManagementMessage.Command.Post projection) : this(projectionId, projection.Mode, projection.RunAs, projection.Name, projection.HandlerType, projection.Query, projection.Enabled, projection.CheckpointsEnabled, - projection.EmitEnabled, projection.EnableRunAs, projection.TrackEmittedStreams) + projection.EmitEnabled, projection.EnableRunAs, projection.TrackEmittedStreams, + projection.DefinitionMetadata) { } @@ -1663,7 +1671,8 @@ public NewProjectionInitializer CreateInitializer(IEnvelope replyEnvelope) EnableRunAs, TrackEmittedStreams, RunAs, - replyEnvelope); + replyEnvelope, + DefinitionMetadata); } } diff --git a/src/Protos/Grpc/projections.proto b/src/Protos/Grpc/projections.proto index b2ad59fca0..4b06cb0a17 100644 --- a/src/Protos/Grpc/projections.proto +++ b/src/Protos/Grpc/projections.proto @@ -42,6 +42,7 @@ message CreateReq { bool checkpoints_enabled = 9; bool emit_enabled = 10; bool track_emitted_streams = 11; + map annotations = 12; message Transient { string name = 1; @@ -67,6 +68,7 @@ message UpdateReq { bool emit_enabled = 3; event_store.client.Empty no_emit_options = 4; } + map annotations = 5; } }