[DB-2019] Add persistent subscriptions support for secondary indexes

src/KurrentDB.Core/ClusterVNode.cs

@@ -1246,6 +1246,28 @@ GossipAdvertiseInfo GetGossipAdvertiseInfo() {
 		perSubscrBus.Subscribe<SubscriptionMessage.PersistentSubscriptionTimerTick>(persistentSubscription);
 		perSubscrBus.Subscribe<SubscriptionMessage.PersistentSubscriptionsRestart>(persistentSubscription);
 
+		// PERSISTENT SUBSCRIPTIONS TO INDEX
+		_mainBus.Subscribe<StorageMessage.SecondaryIndexCommitted>(perSubscrQueue);
+		_mainBus.Subscribe<StorageMessage.SecondaryIndexDeleted>(perSubscrQueue);
+		_mainBus.Subscribe<ClientMessage.CreatePersistentSubscriptionToIndex>(perSubscrQueue);
+		_mainBus.Subscribe<ClientMessage.UpdatePersistentSubscriptionToIndex>(perSubscrQueue);
+		_mainBus.Subscribe<ClientMessage.DeletePersistentSubscriptionToIndex>(perSubscrQueue);
+		_mainBus.Subscribe<ClientMessage.ConnectToPersistentSubscriptionToIndex>(perSubscrQueue);
+
+		var persistentSubscriptionIndex = new PersistentSubscriptionIndexService(
+			perSubscrQueue, psubDispatcher, _mainQueue, consumerStrategyRegistry, secondaryIndexReaders);
+		perSubscrBus.Subscribe<SubscriptionMessage.PersistentSubscriptionIndexEntriesLoaded>(persistentSubscriptionIndex);
+		perSubscrBus.Subscribe<SubscriptionMessage.PersistentSubscriptionIndexEntryChanged>(persistentSubscription);
+		perSubscrBus.Subscribe<ClientMessage.CreatePersistentSubscriptionToIndex>(persistentSubscriptionIndex);
+		perSubscrBus.Subscribe<ClientMessage.UpdatePersistentSubscriptionToIndex>(persistentSubscriptionIndex);
+		perSubscrBus.Subscribe<ClientMessage.DeletePersistentSubscriptionToIndex>(persistentSubscriptionIndex);
+		perSubscrBus.Subscribe<ClientMessage.ConnectToPersistentSubscriptionToIndex>(persistentSubscriptionIndex);
+		perSubscrBus.Subscribe<StorageMessage.SecondaryIndexCommitted>(persistentSubscriptionIndex);
+		perSubscrBus.Subscribe<StorageMessage.SecondaryIndexDeleted>(persistentSubscriptionIndex);
+		perSubscrBus.Subscribe<SubscriptionMessage.PersistentSubscriptionTimerTick>(persistentSubscriptionIndex);
+		perSubscrBus.Subscribe<SystemMessage.BecomeShuttingDown>(persistentSubscriptionIndex);
+		perSubscrBus.Subscribe<SystemMessage.StateChangeMessage>(persistentSubscriptionIndex);
+
 		// STORAGE SCAVENGER
 		var scavengerDispatcher = new IODispatcher(_mainQueue, _mainQueue);
 		_mainBus.Subscribe<ClientMessage.ReadStreamEventsBackwardCompleted>(scavengerDispatcher.BackwardReader);

src/KurrentDB.Core/Messages/ClientMessage.cs

@@ -1521,6 +1521,194 @@ public enum DeletePersistentSubscriptionToAllResult {
 		}
 	}
 
+	// Index persistent subscriptions
+
+	[DerivedMessage(CoreMessage.Client)]
+	public partial class ConnectToPersistentSubscriptionToIndex : ReadRequestMessage {
+		public readonly Guid ConnectionId;
+		public readonly string ConnectionName;
+		public readonly string GroupName;
+		public readonly string IndexName;
+		public readonly int AllowedInFlightMessages;
+		public readonly string From;
+
+		public ConnectToPersistentSubscriptionToIndex(Guid internalCorrId, Guid correlationId, IEnvelope envelope,
+			Guid connectionId, string connectionName, string groupName, string indexName,
+			int allowedInFlightMessages, string from, ClaimsPrincipal user, DateTime? expires = null)
+			: base(internalCorrId, correlationId, envelope, user, expires) {
+			GroupName = Ensure.NotNullOrEmpty(groupName);
+			IndexName = Ensure.NotNullOrEmpty(indexName);
+			ConnectionId = Ensure.NotEmptyGuid(connectionId);
+			ConnectionName = connectionName;
+			AllowedInFlightMessages = Ensure.Nonnegative(allowedInFlightMessages);
+			From = from;
+		}
+	}
+
+	[DerivedMessage(CoreMessage.Client)]
+	public partial class CreatePersistentSubscriptionToIndex : ReadRequestMessage {
+		public readonly TFPos StartFrom;
+		public readonly int MessageTimeoutMilliseconds;
+		public readonly bool RecordStatistics;
+
+		public readonly bool ResolveLinkTos;
+		public readonly int MaxRetryCount;
+		public readonly int BufferSize;
+		public readonly int LiveBufferSize;
+		public readonly int ReadBatchSize;
+
+		public readonly string GroupName;
+		public readonly string IndexName;
+		public readonly int MaxSubscriberCount;
+		public readonly string NamedConsumerStrategy;
+		public readonly int MaxCheckPointCount;
+		public readonly int MinCheckPointCount;
+		public readonly int CheckPointAfterMilliseconds;
+
+		public CreatePersistentSubscriptionToIndex(Guid internalCorrId, Guid correlationId, IEnvelope envelope,
+			string groupName, string indexName, bool resolveLinkTos, TFPos startFrom,
+			int messageTimeoutMilliseconds, bool recordStatistics, int maxRetryCount, int bufferSize,
+			int liveBufferSize, int readBatchSize,
+			int checkPointAfterMilliseconds, int minCheckPointCount, int maxCheckPointCount,
+			int maxSubscriberCount, string namedConsumerStrategy, ClaimsPrincipal user, DateTime? expires = null)
+			: base(internalCorrId, correlationId, envelope, user, expires) {
+			ResolveLinkTos = resolveLinkTos;
+			GroupName = groupName;
+			IndexName = indexName;
+			StartFrom = startFrom;
+			MessageTimeoutMilliseconds = messageTimeoutMilliseconds;
+			RecordStatistics = recordStatistics;
+			MaxRetryCount = maxRetryCount;
+			BufferSize = bufferSize;
+			LiveBufferSize = liveBufferSize;
+			ReadBatchSize = readBatchSize;
+			MaxCheckPointCount = maxCheckPointCount;
+			MinCheckPointCount = minCheckPointCount;
+			CheckPointAfterMilliseconds = checkPointAfterMilliseconds;
+			MaxSubscriberCount = maxSubscriberCount;
+			NamedConsumerStrategy = namedConsumerStrategy;
+		}
+	}
+
+	[DerivedMessage(CoreMessage.Client)]
+	public partial class CreatePersistentSubscriptionToIndexCompleted : ReadResponseMessage {
+		public readonly Guid CorrelationId;
+		public readonly string Reason;
+		public readonly CreatePersistentSubscriptionToIndexResult Result;
+
+		public CreatePersistentSubscriptionToIndexCompleted(Guid correlationId, CreatePersistentSubscriptionToIndexResult result, string reason) {
+			CorrelationId = Ensure.NotEmptyGuid(correlationId);
+			Result = result;
+			Reason = reason;
+		}
+
+		public enum CreatePersistentSubscriptionToIndexResult {
+			Success = 0,
+			AlreadyExists = 1,
+			Fail = 2,
+			AccessDenied = 3
+		}
+	}
+
+	[DerivedMessage(CoreMessage.Client)]
+	public partial class UpdatePersistentSubscriptionToIndex : ReadRequestMessage {
+		public readonly TFPos StartFrom;
+		public readonly int MessageTimeoutMilliseconds;
+		public readonly bool RecordStatistics;
+
+		public readonly bool ResolveLinkTos;
+		public readonly int MaxRetryCount;
+		public readonly int BufferSize;
+		public readonly int LiveBufferSize;
+		public readonly int ReadBatchSize;
+
+		public readonly string GroupName;
+		public readonly string IndexName;
+		public readonly int MaxSubscriberCount;
+
+		public readonly int MaxCheckPointCount;
+		public readonly int MinCheckPointCount;
+		public readonly int CheckPointAfterMilliseconds;
+		public readonly string NamedConsumerStrategy;
+
+		public UpdatePersistentSubscriptionToIndex(Guid internalCorrId, Guid correlationId, IEnvelope envelope,
+			string groupName, string indexName, bool resolveLinkTos, TFPos startFrom,
+			int messageTimeoutMilliseconds, bool recordStatistics, int maxRetryCount, int bufferSize,
+			int liveBufferSize, int readBatchSize, int checkPointAfterMilliseconds, int minCheckPointCount,
+			int maxCheckPointCount, int maxSubscriberCount, string namedConsumerStrategy, ClaimsPrincipal user,
+			DateTime? expires = null)
+			: base(internalCorrId, correlationId, envelope, user, expires) {
+			ResolveLinkTos = resolveLinkTos;
+			GroupName = groupName;
+			IndexName = indexName;
+			StartFrom = startFrom;
+			MessageTimeoutMilliseconds = messageTimeoutMilliseconds;
+			RecordStatistics = recordStatistics;
+			MaxRetryCount = maxRetryCount;
+			BufferSize = bufferSize;
+			LiveBufferSize = liveBufferSize;
+			ReadBatchSize = readBatchSize;
+			MaxCheckPointCount = maxCheckPointCount;
+			MinCheckPointCount = minCheckPointCount;
+			CheckPointAfterMilliseconds = checkPointAfterMilliseconds;
+			MaxSubscriberCount = maxSubscriberCount;
+			NamedConsumerStrategy = namedConsumerStrategy;
+		}
+	}
+
+	[DerivedMessage(CoreMessage.Client)]
+	public partial class UpdatePersistentSubscriptionToIndexCompleted : ReadResponseMessage {
+		public readonly Guid CorrelationId;
+		public readonly string Reason;
+		public readonly UpdatePersistentSubscriptionToIndexResult Result;
+
+		public UpdatePersistentSubscriptionToIndexCompleted(Guid correlationId, UpdatePersistentSubscriptionToIndexResult result, string reason) {
+			CorrelationId = Ensure.NotEmptyGuid(correlationId);
+			Result = result;
+			Reason = reason;
+		}
+
+		public enum UpdatePersistentSubscriptionToIndexResult {
+			Success = 0,
+			DoesNotExist = 1,
+			Fail = 2,
+			AccessDenied = 3
+		}
+	}
+
+	[DerivedMessage(CoreMessage.Client)]
+	public partial class DeletePersistentSubscriptionToIndex : ReadRequestMessage {
+		public readonly string GroupName;
+		public readonly string IndexName;
+
+		public DeletePersistentSubscriptionToIndex(Guid internalCorrId, Guid correlationId, IEnvelope envelope,
+			string indexName, string groupName, ClaimsPrincipal user, DateTime? expires = null)
+			: base(internalCorrId, correlationId, envelope, user, expires) {
+			GroupName = groupName;
+			IndexName = indexName;
+		}
+	}
+
+	[DerivedMessage(CoreMessage.Client)]
+	public partial class DeletePersistentSubscriptionToIndexCompleted : ReadResponseMessage {
+		public readonly Guid CorrelationId;
+		public readonly string Reason;
+		public readonly DeletePersistentSubscriptionToIndexResult Result;
+
+		public DeletePersistentSubscriptionToIndexCompleted(Guid correlationId, DeletePersistentSubscriptionToIndexResult result, string reason) {
+			CorrelationId = Ensure.NotEmptyGuid(correlationId);
+			Result = result;
+			Reason = reason;
+		}
+
+		public enum DeletePersistentSubscriptionToIndexResult {
+			Success = 0,
+			DoesNotExist = 1,
+			Fail = 2,
+			AccessDenied = 3
+		}
+	}
+
 	[DerivedMessage(CoreMessage.Client)]
 	public partial class PersistentSubscriptionAckEvents : ReadRequestMessage {
 		public readonly string SubscriptionId;

src/KurrentDB.Core/Messages/SubscriptionMessage.cs

@@ -2,8 +2,10 @@
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
 using System;
+using System.Collections.Generic;
 using KurrentDB.Core.Messaging;
 using KurrentDB.Core.Services;
+using KurrentDB.Core.Services.PersistentSubscription;
 
 namespace KurrentDB.Core.Messages;
 
@@ -80,4 +82,28 @@ public partial class PersistentSubscriptionsStarted : Message {
 	[DerivedMessage(CoreMessage.Subscription)]
 	public partial class PersistentSubscriptionsStopped : Message {
 	}
+
+	[DerivedMessage(CoreMessage.Subscription)]
+	public partial class PersistentSubscriptionIndexEntriesLoaded : Message {
+		public readonly IReadOnlyList<PersistentSubscriptionEntry> IndexEntries;
+
+		public PersistentSubscriptionIndexEntriesLoaded(IReadOnlyList<PersistentSubscriptionEntry> indexEntries) {
+			IndexEntries = indexEntries;
+		}
+	}
+
+	/// <summary>
+	/// Notifies the main PersistentSubscriptionService that an index subscription entry
+	/// was created or deleted, so it can update its config for forwarding lookups.
+	/// </summary>
+	[DerivedMessage(CoreMessage.Subscription)]
+	public partial class PersistentSubscriptionIndexEntryChanged : Message {
+		public readonly PersistentSubscriptionEntry Entry;
+		public readonly bool IsDelete;
+
+		public PersistentSubscriptionIndexEntryChanged(PersistentSubscriptionEntry entry, bool isDelete) {
+			Entry = entry;
+			IsDelete = isDelete;
+		}
+	}
 }

src/KurrentDB.Core/Services/PersistentSubscription/IPersistentSubscriptionEventSource.cs

@@ -6,10 +6,19 @@
 
 namespace KurrentDB.Core.Services.PersistentSubscription;
 
+public enum EventSourceKind {
+	Stream,
+	All,
+	Index
+}
+
 public interface IPersistentSubscriptionEventSource {
-	bool FromStream { get; }
+	EventSourceKind Kind { get; }
+	bool FromStream => Kind == EventSourceKind.Stream;
+	bool FromAll => Kind == EventSourceKind.All;
+	bool FromIndex => Kind == EventSourceKind.Index;
 	string EventStreamId { get; }
-	bool FromAll { get; }
+	string IndexName { get; }
 	string ToString();
 	IPersistentSubscriptionStreamPosition StreamStartPosition { get; }
 	IPersistentSubscriptionStreamPosition GetStreamPositionFor(ResolvedEvent @event);

src/KurrentDB.Core/Services/PersistentSubscription/PersistentSubscriptionAllStreamEventSource.cs

@@ -8,10 +8,10 @@
 
 namespace KurrentDB.Core.Services.PersistentSubscription;
 
-public class PersistentSubscriptionAllStreamEventSource : IPersistentSubscriptionEventSource {
-	public bool FromStream => false;
+public sealed class PersistentSubscriptionAllStreamEventSource : IPersistentSubscriptionEventSource {
+	public EventSourceKind Kind => EventSourceKind.All;
 	public string EventStreamId => throw new InvalidOperationException();
-	public bool FromAll => true;
+	public string IndexName => throw new InvalidOperationException();
 	public override string ToString() => SystemStreams.AllStream;
 	public IEventFilter EventFilter { get; }
 

src/KurrentDB.Core/Services/PersistentSubscription/PersistentSubscriptionConfig.cs

@@ -59,6 +59,7 @@ public BadConfigDataException(string message, Exception inner) : base(message, i
 public class PersistentSubscriptionEntry {
 	public string Stream;
 	public string Group;
+	public string IndexName;
 	public EventFilter.EventFilterDto Filter;
 	public bool ResolveLinkTos;
 	public bool ExtraStatistics;

src/KurrentDB.Core/Services/PersistentSubscription/PersistentSubscriptionIndexEventSource.cs

@@ -0,0 +1,46 @@
+// Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
+// Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
+
+using System;
+using System.Diagnostics;
+using KurrentDB.Core.Data;
+using KurrentDB.Core.Services.Storage.ReaderIndex;
+
+namespace KurrentDB.Core.Services.PersistentSubscription;
+
+public sealed class PersistentSubscriptionIndexEventSource : IPersistentSubscriptionEventSource {
+	public EventSourceKind Kind => EventSourceKind.Index;
+	public string EventStreamId => throw new InvalidOperationException();
+	public string IndexName { get; }
+	public IEventFilter EventFilter => null;
+
+	public PersistentSubscriptionIndexEventSource(string indexName) {
+		IndexName = indexName ?? throw new ArgumentNullException(nameof(indexName));
+	}
+
+	public override string ToString() => IndexName;
+
+	public IPersistentSubscriptionStreamPosition StreamStartPosition =>
+		new PersistentSubscriptionAllStreamPosition(0L, 0L);
+
+	public IPersistentSubscriptionStreamPosition GetStreamPositionFor(ResolvedEvent @event) {
+		if (@event.OriginalPosition.HasValue) {
+			return new PersistentSubscriptionAllStreamPosition(
+				@event.OriginalPosition.Value.CommitPosition,
+				@event.OriginalPosition.Value.PreparePosition);
+		}
+		throw new InvalidOperationException();
+	}
+
+	public IPersistentSubscriptionStreamPosition GetStreamPositionFor(string checkpoint) {
+		const string C = "C:";
+		const string P = "P:";
+		string[] tokens = checkpoint.Split("/");
+		Debug.Assert(tokens.Length == 2);
+		Debug.Assert(tokens[0].StartsWith(C));
+		Debug.Assert(tokens[1].StartsWith(P));
+		long commitPosition = long.Parse(tokens[0].Substring(C.Length));
+		long preparePosition = long.Parse(tokens[1].Substring(P.Length));
+		return new PersistentSubscriptionAllStreamPosition(commitPosition, preparePosition);
+	}
+}