How to make Scheduler with Azure Queue using .net

I created Two methods one for schedule and log in db and second method is related to reschedule message.

1.

Schedule and Reschedule Message

public async Task ScheduleAndLogEmailAsync( NotificationRequest notificationRequest, ScheduleType type, DateTime triggerTime, DateTime? auctionTime, CancellationToken cancellationToken = default) { var MessageId = Guid.NewGuid(); var sender = _serviceBusClient.CreateSender(_serviceBusConfig.EmailNotificationQueue); notificationRequest.MessageId = MessageId.ToString(); var messageBody = JsonConvert.SerializeObject(notificationRequest); var serviceBusMessage = new ServiceBusMessage(messageBody); // Schedule the message and get its sequence number long sequenceNumber = 0; try { sequenceNumber = await sender.ScheduleMessageAsync(serviceBusMessage, triggerTime, cancellationToken); } catch (Exception ex) { //TODO console error log here; var e = ex; } // Log the scheduled message in DB using var scope = _serviceScopeFactory.CreateScope(); var dbContext = scope.ServiceProvider.GetRequiredService<DbContext>(); var scheduledMessage = new ScheduleMessage { TypeId = type, MessageId = MessageId, AzureSequenceQueueId = sequenceNumber.ToString(), TriggerTime = triggerTime, AuctionTime = auctionTime, IsProcessed = false, NotificationRequestJson = messageBody }; dbContext.ScheduleMessages.Add(scheduledMessage); await dbContext.SaveChangesAsync(cancellationToken); } public async Task ReschedulePendingMessagesAsync(ScheduleType scheduleType, TimeSpan newOffset, CancellationToken cancellationToken = default) { using var scope = _serviceScopeFactory.CreateScope(); var dbContext = scope.ServiceProvider.GetRequiredService<DbContext>(); var sender = _serviceBusClient.CreateSender(_serviceBusConfig.EmailNotificationQueue); var now = DateTime.UtcNow; var pendingMessages = await dbContext.ScheduleMessages .Where(x => x.TypeId == scheduleType && !x.IsProcessed && x.Time.HasValue && x.TriggerTime > now) .ToListAsync(cancellationToken); foreach (var message in pendingMessages) { var newTriggerTime = message.Time.Value - newOffset; // Skip if new trigger time is in the past if (newTriggerTime <= DateTime.UtcNow) continue; // Cancel old scheduled message if (long.TryParse(message.AzureSequenceQueueId, out long oldSequenceId)) { try { await sender.CancelScheduledMessageAsync(oldSequenceId, cancellationToken); } catch (Exception ex) { // If already executed or cannot be cancelled, continue silently Console.WriteLine($"[Reschedule] Failed to cancel scheduled message with SequenceId: {oldSequenceId}. Reason: {ex.Message}"); } } // Reschedule with new trigger time var serviceBusMessage = new ServiceBusMessage(message.NotificationRequestJson); var newSequenceId = await sender.ScheduleMessageAsync(serviceBusMessage, newTriggerTime, cancellationToken); // Update DB message.AzureSequenceQueueId = newSequenceId.ToString(); message.TriggerTime = newTriggerTime; } await dbContext.SaveChangesAsync(cancellationToken); }

2. While Process=>

Code Snippet

private async Task ProcessMessageAsync(ProcessMessageEventArgs args) { var sequenceId = args.Message.SequenceNumber.ToString(); if (!string.IsNullOrEmpty(notificationRequest.MessageId)) { // Convert string to Guid Guid messageIdGuid; if (Guid.TryParse(notificationRequest.MessageId, out messageIdGuid)) { var scheduledRecord = await dbContext.ScheduleMessages .FirstOrDefaultAsync(m => m.MessageId == messageIdGuid); if (scheduledRecord != null) { scheduledRecord.AzureSequenceQueueId = sequenceId; scheduledRecord.IsProcessed = true; dbContext.ScheduleMessages.Update(scheduledRecord); await dbContext.SaveChangesAsync(); } } else { Console.WriteLine("Invalid MessageId format."); } } } <br>