Upgrading the .NET library: from PUSH consumers to PULL consumers

[Alvaro82]

I am migrating a wrapper from the previous .NET library to this new version and I have found a problem with the durable consumers that were already created previously.
With the previous library we used PUSH consumers by default, and in the current one they are not supported so we have to migrate them to PULL consumers and it is not possible to update them.
In case you have the same problem I share with you the method I have prepared to make the migration transparent to my wrapper clients.
Any comments or suggestions are welcome.

// This method checks whether the consumer was already created with the previous version of the library using PUSH consumers.
private async Task<INatsJSConsumer> CreateOrUpdateConsumerAsync(SubscriptionOptions options, CancellationToken cancellationToken = default)
{
	const long MaxAckPendingValue = 1;
	INatsJSContext js = _connection.CreateJetStreamContext();
	if (string.IsNullOrEmpty(options.DurableName))
	{
		return await CreateEphemeralConsumerAsync();
	}

	// It is a durable consumer.
	INatsJSConsumer? existingConsumer = await GetConsumerOrDefaultAsync();
	if (existingConsumer is null)
	{
		return await CreatePullConsumerAsync();
	}

	if (string.IsNullOrEmpty(existingConsumer.Info.Config.DeliverSubject))
	{
		return await UpdatePullConsumerAsync();
	}

	// The existing consumer is a PUSH consumer created with the previous version of the library.
	return await ReplacePushConsumerAsync();

	async Task<INatsJSConsumer> CreateEphemeralConsumerAsync()
	{
		return await js.CreateConsumerAsync(
			options.Stream,
			new ConsumerConfig
			{
				DeliverPolicy = options.DeliverAll ? ConsumerConfigDeliverPolicy.All : ConsumerConfigDeliverPolicy.New,
				AckPolicy = ConsumerConfigAckPolicy.Explicit,
				FilterSubjects = options.Subjects,
				MaxAckPending = MaxAckPendingValue
			},
			cancellationToken);
	}

	async Task<INatsJSConsumer> CreatePullConsumerAsync()
	{
		return await js.CreateConsumerAsync(
			options.Stream,
			new ConsumerConfig(options.DurableName)
			{
				DeliverPolicy = options.DeliverAll ? ConsumerConfigDeliverPolicy.All : ConsumerConfigDeliverPolicy.New,
				AckPolicy = ConsumerConfigAckPolicy.Explicit,
				FilterSubjects = options.Subjects,
				MaxAckPending = MaxAckPendingValue
			},
			cancellationToken);
	}

	async Task<INatsJSConsumer> UpdatePullConsumerAsync()
	{
		// The only parameter that is updated is the FilterSubjects.
		return await js.UpdateConsumerAsync(
			options.Stream,
			new ConsumerConfig(options.DurableName)
			{
				DeliverPolicy = existingConsumer.Info.Config.DeliverPolicy,
				OptStartSeq = existingConsumer.Info.Config.OptStartSeq,
				AckPolicy = existingConsumer.Info.Config.AckPolicy,
				FilterSubjects = options.Subjects,
				MaxAckPending = MaxAckPendingValue
			},
			cancellationToken);
	}

	async Task<INatsJSConsumer> ReplacePushConsumerAsync()
	{
		bool deleted = await js.DeleteConsumerAsync(existingConsumer.Info.StreamName, existingConsumer.Info.Name, cancellationToken);
		if (!deleted)
		{
			throw new InvalidOperationException($"Failed to delete the old PUSH consumer: {existingConsumer.Info.Name}");
		}
		_logger.LogInformation("Deleted the old PUSH consumer: {Name}", existingConsumer.Info.Name);

		// Create a new PULL consumer based on the sequence of the old PUSH consumer.
		return await js.CreateConsumerAsync(
			options.Stream,
			new ConsumerConfig(options.DurableName)
			{
				DeliverPolicy = ConsumerConfigDeliverPolicy.ByStartSequence,
				OptStartSeq = existingConsumer.Info.AckFloor.ConsumerSeq + 1,
				AckPolicy = ConsumerConfigAckPolicy.Explicit,
				FilterSubjects = options.Subjects,
				MaxAckPending = MaxAckPendingValue
			},
			cancellationToken);
	}

	async Task<INatsJSConsumer?> GetConsumerOrDefaultAsync()
	{
		try
		{
			return await js.GetConsumerAsync(options.Stream, options.DurableName, cancellationToken);
		}
		catch (NatsJSApiException)
		{
			// Consumer not found.
			return default;
		}
	}
}