feat(projections): preserve definition write provenance

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"
+                    }
+                  }
                 ]
               }
             ]

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<ProjectionCommandResult> Create(
 				checkpointsEnabled,
 				emitEnabled,
 				trackEmittedStreams,
-				enableRunAs: true),
+				enableRunAs: true,
+				definitionMetadata: DefinitionMetadataFor("create")),
 			cancellationToken);
 	}
 
@@ -217,7 +220,8 @@ public async Task<ProjectionCommandResult> UpdateQuery(
 				name,
 				new ProjectionManagementMessage.RunAs(CurrentUser),
 				request.Query,
-				request.EmitEnabled),
+				request.EmitEnabled,
+				DefinitionMetadataFor("update")),
 			cancellationToken);
 	}
 
@@ -605,6 +609,17 @@ private async Task<ProjectionStatisticsRead> ReadStatistics(
 	private ClaimsPrincipal CurrentUser =>
 		httpContextAccessor.HttpContext?.User ?? new ClaimsPrincipal(new ClaimsIdentity());
 
+	private static byte[] DefinitionMetadataFor(string operation)
+	{
+		var annotations = new Dictionary<string, string>
+		{
+			["trogondb.com/tool"] = "embedded-ui",
+			["trogondb.com/tool-version"] = VersionInfo.Version,
+			["trogondb.com/operation"] = operation
+		};
+		return JsonSerializer.SerializeToUtf8Bytes(annotations);
+	}
+
 	private Task<bool> HasAccess(Operation operation, CancellationToken cancellationToken) =>
 		authorizationProvider.CheckAccessAsync(CurrentUser, operation, cancellationToken).AsTask();
 

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<ParkedMessageMetadata>(msg.Event.Metadata);
+		var metadata = Json.ParseJson<ParkedMessageMetadata>(msg.Event.CustomMetadata);
 		if (metadata != null)
 		{
 			EventTimeStamps.Add(metadata.Added.ToUniversalTime());

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<TLogFormat, TStreamId> : TestFixtureWithProjectionCoreAndManagementServices<TLogFormat, TStreamId>
 {
+	private static readonly byte[] DefinitionMetadata = { 1, 2, 3 };
+
 	protected override void Given()
 	{
 		NoStream("$projections-test-projection-order");
@@ -34,7 +36,8 @@ protected override IEnumerable<WhenStep> 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<ClientMessage.WriteEvents>()
+			.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()
 	{

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<TLogFormat, TStreamId> : TestFixtureWithProjectionCoreAndManagementServices<TLogFormat, TStreamId>
 {
+	private static readonly byte[] UpdatedDefinitionMetadata = { 4, 5, 6 };
+
 	protected override void Given()
 	{
 		NoStream("$projections-test-projection");
@@ -41,7 +44,7 @@ protected override IEnumerable<WhenStep> 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<ClientMessage.WriteEvents>()
+			.Where(v => v.EventStreamId == "$projections-test-projection")
+			.Last()
+			.Events[0];
+
+		Assert.IsTrue(writtenEvent.IsPropertyMetadata);
+		CollectionAssert.AreEqual(UpdatedDefinitionMetadata, writtenEvent.Metadata);
+	}
 }

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)]

src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Create.cs

@@ -74,7 +74,8 @@ public override async Task<CreateResp> 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;
 

src/EventStore.Projections.Core/Services/Grpc/ProjectionManagement.Update.cs

@@ -38,7 +38,7 @@ public override async Task<UpdateResp> 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;
 

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>(T value) where T : class =>
 			value is null ? "{}" : value.ToJson();
 
+		internal static byte[] ToMetadata(MapField<string, string> 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))