diff --git a/internal b/internal index 7b41502b..d18d756c 160000 --- a/internal +++ b/internal @@ -1 +1 @@ -Subproject commit 7b41502b090103c774ca588883faaf6c5ae6db18 +Subproject commit d18d756cd7719e77eaddbefae0df150d854cb401 diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/AnalyzeOnEnrollmentEndEnvelopeSerializationTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/AnalyzeOnEnrollmentEndEnvelopeSerializationTests.cs index a7af09f7..a032d014 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/AnalyzeOnEnrollmentEndEnvelopeSerializationTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/AnalyzeOnEnrollmentEndEnvelopeSerializationTests.cs @@ -6,7 +6,7 @@ namespace AutopilotMonitor.Functions.Tests; /// /// JSON round-trip tests for . The producer /// (AzureQueueAnalyzeOnEnrollmentEndProducer) serializes with Newtonsoft and the -/// queue worker (AnalyzeOnEnrollmentEndQueueWorker) deserializes the same shape — +/// queue worker (AnalyzeOnEnrollmentEndQueueFunction) deserializes the same shape — /// pinning the contract here keeps the queue from silently breaking when fields are added. /// public class AnalyzeOnEnrollmentEndEnvelopeSerializationTests diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/AzureQueueImeMsiArchiveProducerTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/AzureQueueImeMsiArchiveProducerTests.cs index 36d321fb..4212384f 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/AzureQueueImeMsiArchiveProducerTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/AzureQueueImeMsiArchiveProducerTests.cs @@ -61,7 +61,8 @@ public class AzureQueueImeMsiArchiveProducerTests public async Task EnqueueAsync_send_failure_is_swallowed() { var (producer, queue, _) = BuildSut(); - queue.Setup(q => q.SendMessageAsync(It.IsAny(), It.IsAny())) + queue.Setup(q => q.SendMessageAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) .ThrowsAsync(new RequestFailedException(503, "queue busy")); // Fire-and-forget continuation on the ingest path — must never throw. @@ -107,8 +108,9 @@ public class AzureQueueImeMsiArchiveProducerTests queue.Setup(q => q.CreateIfNotExistsAsync( It.IsAny>(), It.IsAny())) .ReturnsAsync((Response?)null); - queue.Setup(q => q.SendMessageAsync(It.IsAny(), It.IsAny())) - .Returns((body, _) => + queue.Setup(q => q.SendMessageAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((body, _, _, _) => { sentBodies.Add(body); var receipt = QueuesModelFactory.SendReceipt( diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/ImeMsiArchiveQueueFunctionTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/ImeMsiArchiveQueueFunctionTests.cs new file mode 100644 index 00000000..b86b77b6 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/ImeMsiArchiveQueueFunctionTests.cs @@ -0,0 +1,180 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Functions.Queue; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Functions.Services.Ime; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models; +using Azure.Storage.Queues.Models; +using Microsoft.Extensions.Caching.Memory; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Newtonsoft.Json; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Contract of the ime-msi-archive queue function (successor of the self-polling worker): +/// merge the archiver's outcome onto the ImeVersionHistory row on EVERY attempt (also +/// failures, so operators see a stuck download), return for terminal outcomes so the host +/// deletes the message, throw for retryable ones so the host's retry → poison ladder applies, +/// re-enqueue with a delay — not drop — while ImeMsiArchivingEnabled is off, and drop +/// version-less envelopes before the archiver runs. +/// +public class ImeMsiArchiveQueueFunctionTests +{ + private const string Version = "1.104.102.0"; + + [Fact] + public async Task Success_merges_row_and_completes() + { + var h = new Harness(); + h.Archiver.Result = new ImeMsiArchiveResult( + true, ImeMsiArchiver.Statuses.Archived, Retryable: false, + $"{Version}/IntuneWindowsAgent.msi", "abc123", 42L, ImeMsiArchiver.FallbackMsiUrls[1]); + + await h.Sut.ProcessAsync(Message(Envelope()), CancellationToken.None); + + h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync( + Version, ImeMsiArchiver.Statuses.Archived, + $"{Version}/IntuneWindowsAgent.msi", "abc123", 42L, ImeMsiArchiver.FallbackMsiUrls[1]), + Times.Once); + h.Producer.Verify(p => p.EnqueueAsync( + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task Retryable_failure_merges_row_and_throws_for_host_retry() + { + var h = new Harness(); + h.Archiver.Result = new ImeMsiArchiveResult( + false, ImeMsiArchiver.Statuses.FailedDownload, Retryable: true, + null, null, null, ImeMsiArchiver.FallbackMsiUrls[1]); + + await Assert.ThrowsAsync( + () => h.Sut.ProcessAsync(Message(Envelope()), CancellationToken.None)); + + // Row always tells the truth — the failure status IS merged before the throw. + h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync( + Version, ImeMsiArchiver.Statuses.FailedDownload, + null, null, null, ImeMsiArchiver.FallbackMsiUrls[1]), + Times.Once); + } + + [Fact] + public async Task Permanent_failure_merges_row_and_completes() + { + var h = new Harness(); + h.Archiver.Result = new ImeMsiArchiveResult( + false, ImeMsiArchiver.Statuses.FailedTooLarge, Retryable: false, + null, null, 999L, ImeMsiArchiver.FallbackMsiUrls[1]); + + await h.Sut.ProcessAsync(Message(Envelope()), CancellationToken.None); + + h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync( + Version, ImeMsiArchiver.Statuses.FailedTooLarge, + null, null, 999L, ImeMsiArchiver.FallbackMsiUrls[1]), + Times.Once); + } + + [Fact] + public async Task Paused_requeues_with_delay_and_skips_archiver() + { + var h = new Harness(archivingEnabled: false); + + await h.Sut.ProcessAsync(Message(Envelope()), CancellationToken.None); + + Assert.Equal(0, h.Archiver.Calls); + h.Producer.Verify(p => p.EnqueueAsync( + It.Is(e => e.Version == Version), + ImeMsiArchiveQueueFunction.PauseRequeueDelay, + It.IsAny()), + Times.Once); + h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + [Fact] + public async Task Envelope_without_version_is_dropped_before_archiver() + { + var h = new Harness(); + + await h.Sut.ProcessAsync(Message(new ImeMsiArchiveEnvelope + { + Version = string.Empty, + EnqueuedAt = DateTime.UtcNow, + }), CancellationToken.None); + + Assert.Equal(0, h.Archiver.Calls); + h.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync( + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + h.Producer.Verify(p => p.EnqueueAsync( + It.IsAny(), It.IsAny(), It.IsAny()), + Times.Never); + } + + // ============================================================ Helpers ==== + + private static ImeMsiArchiveEnvelope Envelope() => new() + { + Version = Version, + MsiDownloadUrl = ImeMsiArchiver.FallbackMsiUrls[1], + MsiMatchedBy = "productVersion", + TenantId = "11111111-1111-1111-1111-111111111111", + SessionId = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + EnqueuedAt = DateTime.UtcNow, + }; + + private static QueueMessage Message(ImeMsiArchiveEnvelope envelope) => QueuesModelFactory.QueueMessage( + messageId: "msg-" + Guid.NewGuid().ToString("N"), + popReceipt: "pop-" + Guid.NewGuid().ToString("N"), + body: new BinaryData(JsonConvert.SerializeObject(envelope)), + dequeueCount: 1); + + /// Scripted archiver — the function only consumes the classified result. + private sealed class StubArchiver : ImeMsiArchiver + { + private int _calls; + + public ImeMsiArchiveResult Result { get; set; } = new( + true, Statuses.Archived, Retryable: false, null, null, null, null); + + public int Calls => Volatile.Read(ref _calls); + + public override Task ArchiveAsync( + ImeMsiArchiveEnvelope envelope, CancellationToken cancellationToken = default) + { + Interlocked.Increment(ref _calls); + return Task.FromResult(Result); + } + } + + private sealed class Harness + { + public StubArchiver Archiver { get; } = new(); + public Mock Repo { get; } = new(); + public Mock Producer { get; } = new(); + public ImeMsiArchiveQueueFunction Sut { get; } + + public Harness(bool archivingEnabled = true) + { + var adminConfig = new Mock( + Mock.Of(), + NullLogger.Instance, + new MemoryCache(new MemoryCacheOptions())); + adminConfig.Setup(a => a.GetConfigurationAsync()) + .ReturnsAsync(new AdminConfiguration { ImeMsiArchivingEnabled = archivingEnabled }); + + Sut = new ImeMsiArchiveQueueFunction( + Archiver, Repo.Object, adminConfig.Object, Producer.Object, + NullLogger.Instance); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/ImeMsiArchiveQueueWorkerTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/ImeMsiArchiveQueueWorkerTests.cs deleted file mode 100644 index 8b20f0d8..00000000 --- a/src/Backend/AutopilotMonitor.Functions.Tests/ImeMsiArchiveQueueWorkerTests.cs +++ /dev/null @@ -1,261 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; -using Azure; -using Azure.Storage.Queues; -using Azure.Storage.Queues.Models; -using AutopilotMonitor.Functions.Services; -using AutopilotMonitor.Functions.Services.Ime; -using AutopilotMonitor.Shared.DataAccess; -using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Caching.Memory; -using Microsoft.Extensions.Logging.Abstractions; -using Moq; -using Newtonsoft.Json; -using Xunit; - -namespace AutopilotMonitor.Functions.Tests; - -/// -/// Worker-level tests for the ime-msi-archive queue consumer (daily review 2026-08-18: -/// previously untested). The worker's contract: merge the archiver's outcome onto the -/// ImeVersionHistory row on EVERY attempt (also failures, so operators see a stuck -/// download), delete the message for terminal outcomes, rethrow for retryable ones -/// (visibility-timeout ladder), pause — not drop — while ImeMsiArchivingEnabled is off, -/// and drop version-less envelopes without invoking the archiver. -/// -public class ImeMsiArchiveQueueWorkerTests -{ - private const string Version = "1.104.102.0"; - - [Fact] - public async Task Worker_success_merges_row_and_deletes_message() - { - var harness = new Harness(); - harness.Archiver.Result = new ImeMsiArchiveResult( - true, ImeMsiArchiver.Statuses.Archived, Retryable: false, - $"{Version}/IntuneWindowsAgent.msi", "abc123", 42L, ImeMsiArchiver.FallbackMsiUrls[1]); - harness.EnqueueMessage(JsonConvert.SerializeObject(Envelope())); - - await harness.RunUntilAsync(() => harness.MainQueueDeleted()); - - harness.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync( - Version, ImeMsiArchiver.Statuses.Archived, - $"{Version}/IntuneWindowsAgent.msi", "abc123", 42L, ImeMsiArchiver.FallbackMsiUrls[1]), - Times.Once); - harness.MainQueue.Verify(q => q.DeleteMessageAsync( - It.IsAny(), It.IsAny(), It.IsAny()), - Times.Once); - } - - [Fact] - public async Task Worker_retryable_failure_merges_row_but_leaves_message() - { - var harness = new Harness(); - harness.Archiver.Result = new ImeMsiArchiveResult( - false, ImeMsiArchiver.Statuses.FailedDownload, Retryable: true, - null, null, null, ImeMsiArchiver.FallbackMsiUrls[1]); - harness.EnqueueMessage(JsonConvert.SerializeObject(Envelope())); - - await harness.RunUntilAsync(() => harness.Archiver.Calls > 0); - - // Row always tells the truth — the failure status IS merged... - harness.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync( - Version, ImeMsiArchiver.Statuses.FailedDownload, - null, null, null, ImeMsiArchiver.FallbackMsiUrls[1]), - Times.AtLeastOnce); - // ...but the message stays for the visibility-timeout retry → poison ladder. - harness.MainQueue.Verify(q => q.DeleteMessageAsync( - It.IsAny(), It.IsAny(), It.IsAny()), - Times.Never); - } - - [Fact] - public async Task Worker_permanent_failure_merges_row_and_completes_message() - { - var harness = new Harness(); - harness.Archiver.Result = new ImeMsiArchiveResult( - false, ImeMsiArchiver.Statuses.FailedTooLarge, Retryable: false, - null, null, 999L, ImeMsiArchiver.FallbackMsiUrls[1]); - harness.EnqueueMessage(JsonConvert.SerializeObject(Envelope())); - - await harness.RunUntilAsync(() => harness.MainQueueDeleted()); - - harness.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync( - Version, ImeMsiArchiver.Statuses.FailedTooLarge, - null, null, 999L, ImeMsiArchiver.FallbackMsiUrls[1]), - Times.Once); - harness.MainQueue.Verify(q => q.DeleteMessageAsync( - It.IsAny(), It.IsAny(), It.IsAny()), - Times.Once); - } - - [Fact] - public async Task Worker_pauses_while_archiving_disabled() - { - var harness = new Harness(archivingEnabled: false); - harness.EnqueueMessage(JsonConvert.SerializeObject(Envelope())); - - await harness.RunForAsync(TimeSpan.FromMilliseconds(500)); - - // Pause, not drop: no receive, no archive attempt — the message parks in the queue. - harness.MainQueue.Verify(q => q.ReceiveMessagesAsync( - It.IsAny(), It.IsAny(), It.IsAny()), - Times.Never); - Assert.Equal(0, harness.Archiver.Calls); - } - - [Fact] - public async Task Worker_drops_envelope_without_version_before_archiver() - { - var harness = new Harness(); - harness.EnqueueMessage(JsonConvert.SerializeObject(new ImeMsiArchiveEnvelope - { - Version = string.Empty, - EnqueuedAt = DateTime.UtcNow, - })); - - await harness.RunUntilAsync(() => harness.MainQueueDeleted()); - - Assert.Equal(0, harness.Archiver.Calls); - harness.Repo.Verify(r => r.UpdateImeVersionArchiveInfoAsync( - It.IsAny(), It.IsAny(), It.IsAny(), - It.IsAny(), It.IsAny(), It.IsAny()), - Times.Never); - } - - // ============================================================ Helpers ==== - - private static ImeMsiArchiveEnvelope Envelope() => new() - { - Version = Version, - MsiDownloadUrl = ImeMsiArchiver.FallbackMsiUrls[1], - MsiMatchedBy = "productVersion", - TenantId = "11111111-1111-1111-1111-111111111111", - SessionId = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", - EnqueuedAt = DateTime.UtcNow, - }; - - /// Scripted archiver — the worker only consumes the classified result. - private sealed class StubArchiver : ImeMsiArchiver - { - private int _calls; - - public ImeMsiArchiveResult Result { get; set; } = new( - true, Statuses.Archived, Retryable: false, null, null, null, null); - - public int Calls => Volatile.Read(ref _calls); - - public override Task ArchiveAsync( - ImeMsiArchiveEnvelope envelope, CancellationToken cancellationToken = default) - { - Interlocked.Increment(ref _calls); - return Task.FromResult(Result); - } - } - - /// Mock-queue poll-loop harness, same shape as SessionDeletionWorkerTests. - private sealed class Harness - { - public Mock MainQueue { get; } - public Mock PoisonQueue { get; } - public StubArchiver Archiver { get; } = new(); - public Mock Repo { get; } = new(); - public ImeMsiArchiveQueueWorker Sut { get; } - - private readonly Queue _pendingMessages = new(); - - public Harness(bool archivingEnabled = true) - { - MainQueue = new Mock(); - PoisonQueue = new Mock(); - - MainQueue.Setup(q => q.CreateIfNotExistsAsync( - It.IsAny>(), It.IsAny())) - .ReturnsAsync((Response?)null); - PoisonQueue.Setup(q => q.CreateIfNotExistsAsync( - It.IsAny>(), It.IsAny())) - .ReturnsAsync((Response?)null); - - MainQueue.Setup(q => q.ReceiveMessagesAsync( - It.IsAny(), It.IsAny(), It.IsAny())) - .Returns((maxMessages, _, _) => - { - var batch = new List(); - lock (_pendingMessages) - { - while (batch.Count < maxMessages && _pendingMessages.Count > 0) - batch.Add(_pendingMessages.Dequeue()); - } - return Task.FromResult(Response.FromValue(batch.ToArray(), new Mock().Object)); - }); - - MainQueue.Setup(q => q.DeleteMessageAsync( - It.IsAny(), It.IsAny(), It.IsAny())) - .ReturnsAsync(new Mock().Object); - - MainQueue.Setup(q => q.UpdateMessageAsync( - It.IsAny(), It.IsAny(), - It.IsAny(), It.IsAny(), It.IsAny())) - .Returns((_, _, _, vis, _) => - { - var updated = QueuesModelFactory.UpdateReceipt( - "pop-extended-" + Guid.NewGuid().ToString("N"), - DateTimeOffset.UtcNow.Add(vis)); - return Task.FromResult(Response.FromValue(updated, new Mock().Object)); - }); - - var adminConfig = new Mock( - Mock.Of(), - NullLogger.Instance, - new MemoryCache(new MemoryCacheOptions())); - adminConfig.Setup(a => a.GetConfigurationAsync()) - .ReturnsAsync(new AdminConfiguration { ImeMsiArchivingEnabled = archivingEnabled }); - - Sut = new ImeMsiArchiveQueueWorker( - MainQueue.Object, PoisonQueue.Object, - Archiver, Repo.Object, adminConfig.Object, - NullLogger.Instance, - pollInterval: TimeSpan.FromMilliseconds(50)); - } - - public void EnqueueMessage(string body) - { - var msg = QueuesModelFactory.QueueMessage( - messageId: "msg-" + Guid.NewGuid().ToString("N"), - popReceipt: "pop-" + Guid.NewGuid().ToString("N"), - body: new BinaryData(body), - dequeueCount: 1); - lock (_pendingMessages) _pendingMessages.Enqueue(msg); - } - - public async Task RunForAsync(TimeSpan duration) - { - using var cts = new CancellationTokenSource(duration); - try { await Sut.StartAsync(cts.Token); } - catch (OperationCanceledException) { } - try { await Task.Delay(duration, cts.Token); } - catch (OperationCanceledException) { } - try { await Sut.StopAsync(CancellationToken.None); } - catch (OperationCanceledException) { } - } - - public async Task RunUntilAsync(Func condition, TimeSpan? timeout = null) - { - var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(10)); - using var cts = new CancellationTokenSource(); - try { await Sut.StartAsync(cts.Token); } - catch (OperationCanceledException) { } - while (!condition() && DateTime.UtcNow < deadline) - await Task.Delay(25); - cts.Cancel(); - try { await Sut.StopAsync(CancellationToken.None); } - catch (OperationCanceledException) { } - } - - public bool MainQueueDeleted() => - MainQueue.Invocations.Any(i => i.Method.Name == nameof(QueueClient.DeleteMessageAsync)); - } -} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/NotificationDispatchHandlerTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/NotificationDispatchHandlerTests.cs new file mode 100644 index 00000000..75846d2e --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/NotificationDispatchHandlerTests.cs @@ -0,0 +1,150 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Functions.Services.Notifications; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models; +using AutopilotMonitor.Shared.Models.Notifications; +using Microsoft.Extensions.Caching.Memory; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Newtonsoft.Json; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// The notification-dispatch consumer resolves channel ids against the tenant's CURRENT +/// configuration — no URL or secret ever rides in the queue — and sends only to channels that +/// are still enabled. Plus the wire contract: the envelope round-trips through Newtonsoft with +/// the alert's facts, sections and severity intact. +/// +public class NotificationDispatchHandlerTests +{ + private const string Tenant = "11111111-1111-1111-1111-111111111111"; + + [Fact] + public async Task Sends_to_enabled_channels_matching_the_envelope_ids_only() + { + var h = new Harness(new[] + { + Channel("teams", enabled: true), + Channel("slack", enabled: true), + Channel("old", enabled: false), + }); + + await h.Sut.HandleAsync(Envelope("teams", "old"), CancellationToken.None); + + var sent = Assert.Single(h.Sent); + Assert.Equal(new[] { "teams" }, sent.Channels.Select(c => c.Id)); + Assert.Equal("enrollment_succeeded", sent.Alert.EventType); + } + + [Fact] + public async Task Unknown_channel_ids_are_skipped_without_sending() + { + var h = new Harness(new[] { Channel("teams", enabled: true) }); + + await h.Sut.HandleAsync(Envelope("deleted-meanwhile"), CancellationToken.None); + + Assert.Empty(h.Sent); + } + + [Fact] + public async Task Missing_tenant_configuration_drops_the_alert() + { + var h = new Harness(channels: null); + + await h.Sut.HandleAsync(Envelope("teams"), CancellationToken.None); + + Assert.Empty(h.Sent); + } + + [Fact] + public void Envelope_round_trips_through_newtonsoft() + { + var envelope = Envelope("teams"); + envelope.Alert.Facts.Add(new NotificationFact { Name = "Device", Value = "PC-0001" }); + envelope.Alert.Severity = NotificationSeverity.Error; + + var restored = JsonConvert.DeserializeObject( + JsonConvert.SerializeObject(envelope))!; + + Assert.Equal(Tenant, restored.TenantId); + Assert.Equal(new[] { "teams" }, restored.ChannelIds); + Assert.Equal("enrollment_succeeded", restored.Alert.EventType); + Assert.Equal(NotificationSeverity.Error, restored.Alert.Severity); + var fact = Assert.Single(restored.Alert.Facts); + Assert.Equal("PC-0001", fact.Value); + } + + // ============================================================ Helpers ==== + + private static NotificationChannel Channel(string id, bool enabled) => new() + { + Id = id, + Name = id, + ProviderType = (int)WebhookProviderType.Slack, + Url = $"https://example.invalid/{id}", + Enabled = enabled, + NotifyOnSuccess = true, + }; + + private static NotificationDispatchEnvelope Envelope(params string[] channelIds) => new() + { + TenantId = Tenant, + SessionId = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", + ChannelIds = channelIds.ToList(), + Alert = new NotificationAlert + { + EventType = "enrollment_succeeded", + Title = "Enrollment Succeeded", + Summary = "PC-0001 enrolled", + Severity = NotificationSeverity.Info, + ThemeColor = "00AA00", + }, + EnqueuedAt = DateTime.UtcNow, + }; + + private sealed class Harness + { + public List<(IReadOnlyList Channels, NotificationAlert Alert)> Sent { get; } = new(); + public NotificationDispatchHandler Sut { get; } + + public Harness(IEnumerable? channels) + { + var configService = new Mock( + Mock.Of(), + NullLogger.Instance, + new MemoryCache(new MemoryCacheOptions())) + { CallBase = false }; + + if (channels is null) + { + configService.Setup(c => c.TryGetConfigurationAsync(Tenant)) + .ReturnsAsync((new TenantConfiguration { TenantId = Tenant }, false)); + } + else + { + var config = new TenantConfiguration + { + TenantId = Tenant, + NotificationChannelsJson = NotificationChannel.SerializeList(channels), + }; + configService.Setup(c => c.TryGetConfigurationAsync(Tenant)) + .ReturnsAsync((config, true)); + } + + var dispatcher = new Mock(null!, null!); + dispatcher.Setup(d => d.SendToChannelsAsync(It.IsAny>(), It.IsAny())) + .Callback, NotificationAlert>((c, a) => Sent.Add((c.ToList(), a))) + .Returns(Task.CompletedTask); + + Sut = new NotificationDispatchHandler( + configService.Object, dispatcher.Object, NullLogger.Instance); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/QueueEnvelopeReaderTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/QueueEnvelopeReaderTests.cs new file mode 100644 index 00000000..3e945131 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/QueueEnvelopeReaderTests.cs @@ -0,0 +1,84 @@ +using System; +using AutopilotMonitor.Functions.Services.Queueing; +using Azure.Storage.Queues.Models; +using Microsoft.Extensions.Logging.Abstractions; +using Newtonsoft.Json; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// The shared decode step of every [QueueTrigger] function: a false means the message is +/// permanently unusable and the function returns so the host deletes it — never a throw, which +/// would burn dequeue attempts on a message that can never succeed. +/// +public class QueueEnvelopeReaderTests +{ + private sealed class Envelope + { + public string TenantId { get; set; } = string.Empty; + public int Count { get; set; } + } + + [Fact] + public void Valid_json_yields_envelope() + { + var msg = Message(JsonConvert.SerializeObject(new Envelope { TenantId = "t1", Count = 7 })); + + var ok = QueueEnvelopeReader.TryRead(msg, NullLogger.Instance, out var envelope); + + Assert.True(ok); + Assert.Equal("t1", envelope.TenantId); + Assert.Equal(7, envelope.Count); + } + + [Fact] + public void Malformed_json_is_rejected_without_throwing() + { + var msg = Message("{ not json"); + + var ok = QueueEnvelopeReader.TryRead(msg, NullLogger.Instance, out var envelope); + + Assert.False(ok); + Assert.Null(envelope); + } + + [Fact] + public void Null_literal_is_rejected() + { + var msg = Message("null"); + + var ok = QueueEnvelopeReader.TryRead(msg, NullLogger.Instance, out _); + + Assert.False(ok); + } + + [Fact] + public void Failed_validation_is_rejected() + { + var msg = Message(JsonConvert.SerializeObject(new Envelope { TenantId = string.Empty })); + + var ok = QueueEnvelopeReader.TryRead( + msg, NullLogger.Instance, out _, validate: e => !string.IsNullOrEmpty(e.TenantId)); + + Assert.False(ok); + } + + [Fact] + public void Validation_passes_when_predicate_holds() + { + var msg = Message(JsonConvert.SerializeObject(new Envelope { TenantId = "t1" })); + + var ok = QueueEnvelopeReader.TryRead( + msg, NullLogger.Instance, out var envelope, validate: e => !string.IsNullOrEmpty(e.TenantId)); + + Assert.True(ok); + Assert.Equal("t1", envelope.TenantId); + } + + private static QueueMessage Message(string body) => QueuesModelFactory.QueueMessage( + messageId: "msg-" + Guid.NewGuid().ToString("N"), + popReceipt: "pop", + body: new BinaryData(body), + dequeueCount: 1); +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/QueuePollingWorkerBatchSizeTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/QueuePollingWorkerBatchSizeTests.cs new file mode 100644 index 00000000..9131fc30 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions.Tests/QueuePollingWorkerBatchSizeTests.cs @@ -0,0 +1,107 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services.Queueing; +using Azure; +using Azure.Storage.Queues; +using Azure.Storage.Queues.Models; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Newtonsoft.Json; +using Xunit; + +namespace AutopilotMonitor.Functions.Tests; + +/// +/// Audit 2026-09-05 F02: the self-managed poll loop processes a batch sequentially and the +/// heartbeat only covers the in-flight message, so every message received beyond the first sat +/// invisible with its clock running and could reappear for another worker before this loop +/// reached it. The base therefore receives ONE message per poll and re-receives immediately +/// after a non-empty result — no worker may inherit a larger batch by accident. +/// +public class QueuePollingWorkerBatchSizeTests +{ + private sealed class Envelope + { + public int Seq { get; set; } + } + + private sealed class ProbeWorker : QueuePollingWorker + { + public List Handled { get; } = new(); + + public ProbeWorker(QueueClient main, QueueClient poison) + : base(main, poison, NullLogger.Instance, pollIntervalOverride: TimeSpan.FromMilliseconds(20)) + { + } + + public int ExposedBatchSize => BatchSize; + + protected override Task HandleAsync(Envelope envelope, CancellationToken ct) + { + lock (Handled) Handled.Add(envelope.Seq); + return Task.CompletedTask; + } + } + + [Fact] + public void Default_batch_size_is_one() + { + var sut = new ProbeWorker(new Mock().Object, new Mock().Object); + Assert.Equal(1, sut.ExposedBatchSize); + } + + [Fact] + public async Task Loop_receives_one_message_at_a_time_and_drains_all_pending() + { + var pending = new Queue(); + for (var i = 1; i <= 3; i++) + { + pending.Enqueue(QueuesModelFactory.QueueMessage( + messageId: "msg-" + i, + popReceipt: "pop-" + i, + body: new BinaryData(JsonConvert.SerializeObject(new Envelope { Seq = i })), + dequeueCount: 1)); + } + + var requestedBatchSizes = new List(); + var main = new Mock(); + var poison = new Mock(); + main.Setup(q => q.CreateIfNotExistsAsync(It.IsAny>(), It.IsAny())) + .ReturnsAsync((Response?)null); + poison.Setup(q => q.CreateIfNotExistsAsync(It.IsAny>(), It.IsAny())) + .ReturnsAsync((Response?)null); + main.Setup(q => q.ReceiveMessagesAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .Returns((max, _, _) => + { + var batch = new List(); + lock (pending) + { + requestedBatchSizes.Add(max); + while (batch.Count < max && pending.Count > 0) + batch.Add(pending.Dequeue()); + } + return Task.FromResult(Response.FromValue(batch.ToArray(), new Mock().Object)); + }); + main.Setup(q => q.DeleteMessageAsync(It.IsAny(), It.IsAny(), It.IsAny())) + .ReturnsAsync(new Mock().Object); + + var sut = new ProbeWorker(main.Object, poison.Object); + using var cts = new CancellationTokenSource(); + await sut.StartAsync(cts.Token); + var deadline = DateTime.UtcNow + TimeSpan.FromSeconds(10); + while (DateTime.UtcNow < deadline) + { + lock (sut.Handled) if (sut.Handled.Count == 3) break; + await Task.Delay(20); + } + cts.Cancel(); + await sut.StopAsync(CancellationToken.None); + + Assert.Equal(new[] { 1, 2, 3 }, sut.Handled); + Assert.NotEmpty(requestedBatchSizes); + Assert.All(requestedBatchSizes, n => Assert.Equal(1, n)); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineFailurePropagationTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineFailurePropagationTests.cs index aea83be1..124d1c08 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineFailurePropagationTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/RuleEngineFailurePropagationTests.cs @@ -10,7 +10,7 @@ namespace AutopilotMonitor.Functions.Tests; /// Pins the queue-path retry contract: must /// propagate storage exceptions to the caller. The previous outer try/catch swallowed all /// errors and returned an empty , which caused -/// AnalyzeOnEnrollmentEndQueueWorker to delete the message as success and never retry — +/// AnalyzeOnEnrollmentEndQueueFunction to delete the message as success and never retry — /// rule results were silently lost on transient Table Storage failures. /// public class RuleEngineFailurePropagationTests diff --git a/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityCorrelateEnvelopeSerializationTests.cs b/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityCorrelateEnvelopeSerializationTests.cs index 49df00a3..4508f735 100644 --- a/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityCorrelateEnvelopeSerializationTests.cs +++ b/src/Backend/AutopilotMonitor.Functions.Tests/VulnerabilityCorrelateEnvelopeSerializationTests.cs @@ -6,7 +6,7 @@ namespace AutopilotMonitor.Functions.Tests; /// /// JSON round-trip tests for . The producer /// (AzureQueueVulnerabilityCorrelateProducer) serializes with Newtonsoft and the -/// queue worker (VulnerabilityCorrelateQueueWorker) deserializes the same shape — +/// queue worker (VulnerabilityCorrelateQueueFunction) deserializes the same shape — /// pinning the contract here keeps the queue from silently breaking when fields are added. /// public class VulnerabilityCorrelateEnvelopeSerializationTests diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/ReportDistressFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/ReportDistressFunction.cs index 0fe77f2c..c4418353 100644 --- a/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/ReportDistressFunction.cs +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Ingest/ReportDistressFunction.cs @@ -37,7 +37,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest private readonly IDistressReportRepository _repository; private readonly TelemetryClient _telemetryClient; private readonly HardwareRejectionThrottleService _hardwareThrottle; - private readonly NotificationChannelDispatcher _channelDispatcher; + private readonly INotificationDispatchProducer _notificationProducer; private readonly IHardwareRejectionNotificationTracker _hardwareBellTracker; private readonly TenantNotificationService _tenantNotificationService; @@ -77,7 +77,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest IDistressReportRepository repository, TelemetryClient telemetryClient, HardwareRejectionThrottleService hardwareThrottle, - NotificationChannelDispatcher channelDispatcher, + INotificationDispatchProducer notificationProducer, IHardwareRejectionNotificationTracker hardwareBellTracker, TenantNotificationService tenantNotificationService) { @@ -87,7 +87,7 @@ namespace AutopilotMonitor.Functions.Functions.Ingest _repository = repository; _telemetryClient = telemetryClient; _hardwareThrottle = hardwareThrottle; - _channelDispatcher = channelDispatcher; + _notificationProducer = notificationProducer; _hardwareBellTracker = hardwareBellTracker; _tenantNotificationService = tenantNotificationService; } @@ -217,7 +217,9 @@ namespace AutopilotMonitor.Functions.Functions.Ingest await _repository.SaveDistressReportAsync(tenantId, entry); - // Notify tenant webhook for hardware whitelist rejections (fire-and-forget, opt-in) + // Notify tenant channels for hardware whitelist rejections (opt-in). Durable via + // the notification-dispatch queue — a fire-and-forget send after the response was + // not counted as open work by the host and could be dropped on scale-in. if (report.ErrorType == DistressErrorType.HardwareNotAllowed && _hardwareThrottle.ShouldNotify(tenantId, manufacturer, model)) { @@ -232,7 +234,13 @@ namespace AutopilotMonitor.Functions.Functions.Ingest if (hwChannels.Count > 0) { var alert = NotificationAlertBuilder.BuildHardwareRejectedAlert(manufacturer, model, serialNumber); - _ = _channelDispatcher.SendToChannelsAsync(hwChannels, alert); + await _notificationProducer.EnqueueAsync(new NotificationDispatchEnvelope + { + TenantId = tenantId, + ChannelIds = hwChannels.Select(c => c.Id).ToList(), + Alert = alert, + EnqueuedAt = DateTime.UtcNow, + }); } } } diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Queue/AnalyzeOnEnrollmentEndQueueFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/AnalyzeOnEnrollmentEndQueueFunction.cs new file mode 100644 index 00000000..59d70204 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/AnalyzeOnEnrollmentEndQueueFunction.cs @@ -0,0 +1,47 @@ +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services.Analyze; +using AutopilotMonitor.Functions.Services.Queueing; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.Models; +using Azure.Storage.Queues.Models; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Functions.Queue +{ + /// + /// Host-managed consumer of the analyze-on-enrollment-end queue. The Functions host + /// owns the lease (renewed while the handler runs), the retry ladder (throw ⇒ redelivery after + /// extensions.queues.visibilityTimeout) and the poison move after + /// maxDequeueCount attempts to analyze-on-enrollment-end-poison. On Flex + /// Consumption the queue length is the scale signal, so a message enqueued while no instance + /// is running wakes one — the property the earlier self-polling BackgroundService lacked. + /// The rule-engine work lives unchanged in . + /// + public sealed class AnalyzeOnEnrollmentEndQueueFunction + { + private readonly AnalyzeOnEnrollmentEndHandler _handler; + private readonly ILogger _logger; + + public AnalyzeOnEnrollmentEndQueueFunction( + AnalyzeOnEnrollmentEndHandler handler, + ILogger logger) + { + _handler = handler; + _logger = logger; + } + + [Function("AnalyzeOnEnrollmentEndQueue")] + public Task Run( + [QueueTrigger(Constants.QueueNames.AnalyzeOnEnrollmentEnd, Connection = Constants.QueueNames.TriggerConnection)] + QueueMessage message, + CancellationToken cancellationToken) + { + if (!QueueEnvelopeReader.TryRead(message, _logger, out var envelope)) + return Task.CompletedTask; + + return _handler.HandleAsync(envelope, cancellationToken); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Queue/ImeMsiArchiveQueueFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/ImeMsiArchiveQueueFunction.cs new file mode 100644 index 00000000..175117af --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/ImeMsiArchiveQueueFunction.cs @@ -0,0 +1,115 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services; +using AutopilotMonitor.Functions.Services.Ime; +using AutopilotMonitor.Functions.Services.Queueing; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.DataAccess; +using AutopilotMonitor.Shared.Models; +using Azure.Storage.Queues.Models; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Functions.Queue +{ + /// + /// Host-managed consumer of the ime-msi-archive queue: downloads a newly sighted IME + /// installer into the permanent ime-archive container and merges the outcome onto the + /// ImeVersionHistory row. Volume is ~one message per Microsoft IME release. + /// + /// Retry semantics: the archiver never throws — it classifies. Permanent rejections (bad + /// version, size cap) complete the message with their Failed:* status persisted; + /// transient failures (download/timeout) throw so the host's retry → poison ladder applies, + /// with the last status visible on the row. The host renews the message lease for the + /// up-to-10-minute download window. The status row-merge happens on EVERY attempt so + /// operators see a stuck download as Failed:Download instead of nothing. + /// + /// + /// Pause ( off): a trigger + /// cannot leave a message untouched, so the envelope is re-enqueued with + /// and the current message completes. The message parks in + /// the queue until the flag returns; its dequeue count starts over, which is right for a + /// pause — it is not a failure. The re-enqueue is fail-soft like every other archive step; + /// operator-side blob backfill remains the fallback. + /// + /// + public sealed class ImeMsiArchiveQueueFunction + { + /// How long a parked message stays invisible while archiving is switched off. + internal static readonly TimeSpan PauseRequeueDelay = TimeSpan.FromMinutes(10); + + private readonly ImeMsiArchiver _archiver; + private readonly ISessionRepository _sessionRepo; + private readonly AdminConfigurationService _adminConfigService; + private readonly IImeMsiArchiveProducer _producer; + private readonly ILogger _logger; + + public ImeMsiArchiveQueueFunction( + ImeMsiArchiver archiver, + ISessionRepository sessionRepo, + AdminConfigurationService adminConfigService, + IImeMsiArchiveProducer producer, + ILogger logger) + { + _archiver = archiver ?? throw new ArgumentNullException(nameof(archiver)); + _sessionRepo = sessionRepo ?? throw new ArgumentNullException(nameof(sessionRepo)); + _adminConfigService = adminConfigService ?? throw new ArgumentNullException(nameof(adminConfigService)); + _producer = producer ?? throw new ArgumentNullException(nameof(producer)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + } + + [Function("ImeMsiArchiveQueue")] + public Task Run( + [QueueTrigger(Constants.QueueNames.ImeMsiArchive, Connection = Constants.QueueNames.TriggerConnection)] + QueueMessage message, + CancellationToken cancellationToken) + => ProcessAsync(message, cancellationToken); + + /// Testable core — the trigger entry above only forwards. + internal async Task ProcessAsync(QueueMessage message, CancellationToken cancellationToken) + { + if (!QueueEnvelopeReader.TryRead( + message, _logger, out var envelope, + validate: e => !string.IsNullOrEmpty(e.Version))) + return; + + if (!await IsArchivingEnabledAsync().ConfigureAwait(false)) + { + _logger.LogWarning( + "ImeMsiArchive paused (ImeMsiArchivingEnabled=false) — re-enqueuing version {Version} with {Delay} delay", + envelope.Version, PauseRequeueDelay); + await _producer.EnqueueAsync(envelope, PauseRequeueDelay, cancellationToken).ConfigureAwait(false); + return; + } + + var result = await _archiver.ArchiveAsync(envelope, cancellationToken).ConfigureAwait(false); + + // Merge the outcome first — also for failures, so the row always tells the truth. + await _sessionRepo.UpdateImeVersionArchiveInfoAsync( + envelope.Version, result.Status, result.BlobPath, result.Sha256, result.SizeBytes, result.SourceUrl) + .ConfigureAwait(false); + + if (!result.Success && result.Retryable) + { + // Host retry → poison after maxDequeueCount. The thrown message lands in the + // function's failure log line. + throw new InvalidOperationException( + $"IME MSI archive attempt failed with {result.Status} for version {envelope.Version} (dequeue {message.DequeueCount})"); + } + } + + private async Task IsArchivingEnabledAsync() + { + try + { + var config = await _adminConfigService.GetConfigurationAsync().ConfigureAwait(false); + return config.ImeMsiArchivingEnabled; + } + catch + { + return true; // config unreadable → keep working (archive is fail-soft anyway) + } + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Queue/IndexReconcileQueueFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/IndexReconcileQueueFunction.cs new file mode 100644 index 00000000..6fcc0300 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/IndexReconcileQueueFunction.cs @@ -0,0 +1,44 @@ +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services.Indexing; +using AutopilotMonitor.Functions.Services.Queueing; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.Models; +using Azure.Storage.Queues.Models; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Functions.Queue +{ + /// + /// Host-managed consumer of the telemetry-index-reconcile queue (Plan §M5.d.3; lease, + /// retry ladder and poison move owned by the Functions host — see + /// ). Index writes live unchanged in + /// ; the 2h IndexReconcileTimer remains the safety net. + /// + public sealed class IndexReconcileQueueFunction + { + private readonly IndexReconcileHandler _handler; + private readonly ILogger _logger; + + public IndexReconcileQueueFunction( + IndexReconcileHandler handler, + ILogger logger) + { + _handler = handler; + _logger = logger; + } + + [Function("IndexReconcileQueue")] + public Task Run( + [QueueTrigger(Constants.QueueNames.TelemetryIndexReconcile, Connection = Constants.QueueNames.TriggerConnection)] + QueueMessage message, + CancellationToken cancellationToken) + { + if (!QueueEnvelopeReader.TryRead(message, _logger, out var envelope)) + return Task.CompletedTask; + + return _handler.HandleAsync(envelope, cancellationToken); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Queue/NotificationDispatchQueueFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/NotificationDispatchQueueFunction.cs new file mode 100644 index 00000000..f095e219 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/NotificationDispatchQueueFunction.cs @@ -0,0 +1,45 @@ +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services.Notifications; +using AutopilotMonitor.Functions.Services.Queueing; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.Models.Notifications; +using Azure.Storage.Queues.Models; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Functions.Queue +{ + /// + /// Host-managed consumer of the notification-dispatch queue (lease, retry ladder and + /// poison move owned by the Functions host — see ). + /// Channel resolution and the send live in . + /// + public sealed class NotificationDispatchQueueFunction + { + private readonly NotificationDispatchHandler _handler; + private readonly ILogger _logger; + + public NotificationDispatchQueueFunction( + NotificationDispatchHandler handler, + ILogger logger) + { + _handler = handler; + _logger = logger; + } + + [Function("NotificationDispatchQueue")] + public Task Run( + [QueueTrigger(Constants.QueueNames.NotificationDispatch, Connection = Constants.QueueNames.TriggerConnection)] + QueueMessage message, + CancellationToken cancellationToken) + { + if (!QueueEnvelopeReader.TryRead( + message, _logger, out var envelope, + validate: e => !string.IsNullOrEmpty(e.TenantId) && e.Alert is not null)) + return Task.CompletedTask; + + return _handler.HandleAsync(envelope, cancellationToken); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Queue/TenantAutoApproveQueueFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/TenantAutoApproveQueueFunction.cs new file mode 100644 index 00000000..b8858814 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/TenantAutoApproveQueueFunction.cs @@ -0,0 +1,45 @@ +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services.Activation; +using AutopilotMonitor.Functions.Services.Queueing; +using AutopilotMonitor.Shared; +using Azure.Storage.Queues.Models; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Functions.Queue +{ + /// + /// Host-managed consumer of the tenant-auto-approve queue (lease, retry ladder and + /// poison move owned by the Functions host — see ). + /// The producer's ~1-minute visibility delay still applies; every decision lives in + /// . + /// + public sealed class TenantAutoApproveQueueFunction + { + private readonly TenantAutoApproveHandler _handler; + private readonly ILogger _logger; + + public TenantAutoApproveQueueFunction( + TenantAutoApproveHandler handler, + ILogger logger) + { + _handler = handler; + _logger = logger; + } + + [Function("TenantAutoApproveQueue")] + public Task Run( + [QueueTrigger(Constants.QueueNames.TenantAutoApprove, Connection = Constants.QueueNames.TriggerConnection)] + QueueMessage message, + CancellationToken cancellationToken) + { + if (!QueueEnvelopeReader.TryRead( + message, _logger, out var envelope, + validate: e => !string.IsNullOrWhiteSpace(e.TenantId))) + return Task.CompletedTask; + + return _handler.HandleAsync(envelope, cancellationToken); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Functions/Queue/VulnerabilityCorrelateQueueFunction.cs b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/VulnerabilityCorrelateQueueFunction.cs new file mode 100644 index 00000000..ebf85ea4 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Functions/Queue/VulnerabilityCorrelateQueueFunction.cs @@ -0,0 +1,43 @@ +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Functions.Services.Queueing; +using AutopilotMonitor.Functions.Services.Vulnerability; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.Models; +using Azure.Storage.Queues.Models; +using Microsoft.Azure.Functions.Worker; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Functions.Queue +{ + /// + /// Host-managed consumer of the vulnerability-correlate queue (lease, retry ladder and + /// poison move owned by the Functions host — see ). + /// Correlation itself lives unchanged in . + /// + public sealed class VulnerabilityCorrelateQueueFunction + { + private readonly VulnerabilityCorrelateHandler _handler; + private readonly ILogger _logger; + + public VulnerabilityCorrelateQueueFunction( + VulnerabilityCorrelateHandler handler, + ILogger logger) + { + _handler = handler; + _logger = logger; + } + + [Function("VulnerabilityCorrelateQueue")] + public Task Run( + [QueueTrigger(Constants.QueueNames.VulnerabilityCorrelate, Connection = Constants.QueueNames.TriggerConnection)] + QueueMessage message, + CancellationToken cancellationToken) + { + if (!QueueEnvelopeReader.TryRead(message, _logger, out var envelope)) + return Task.CompletedTask; + + return _handler.HandleAsync(envelope, cancellationToken); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Program.cs b/src/Backend/AutopilotMonitor.Functions/Program.cs index 8acebce6..1e7436cf 100644 --- a/src/Backend/AutopilotMonitor.Functions/Program.cs +++ b/src/Backend/AutopilotMonitor.Functions/Program.cs @@ -322,32 +322,27 @@ builder.Services.AddSingleton< builder.Services.AddSingleton< AutopilotMonitor.Functions.Services.Indexing.IndexReconcileHandler>(); -// Background poll-loop for the telemetry-index-reconcile queue (Plan §M5.d.3). Replaces the -// earlier QueueTrigger function, which required a Functions-host-specific -// `__queueServiceUri` app-setting that diverged from the rest of the project's -// AzureStorageAccountName + DefaultAzureCredential pattern. This worker uses the same -// resolution as AzureQueueIndexReconcileProducer — Managed Identity by account name, with -// connection-string fallback — and provides full QueueTrigger parity (visibility-timeout -// retries, poison-queue move after 5 failed attempts). -builder.Services.AddHostedService< - AutopilotMonitor.Functions.Services.Indexing.IndexReconcileQueueWorker>(); +// Consumer: Functions/Queue/IndexReconcileQueueFunction — a [QueueTrigger] on the +// Constants.QueueNames.TriggerConnection connection (host-resolved app setting +// `DataStorage__queueServiceUri`; see host.json extensions.queues). Stateless queue consumers +// are host-managed so the Flex scale controller wakes an instance for a message and renews the +// lease; only the cascade workers with lease/CAS state machines stay BackgroundServices. // Auto-analyze fan-out at session end. Replaces the previous in-function fire-and-forget // Task.Run that ran the rule engine after enrollment_complete / enrollment_failed / async // vulnerability correlation — Functions scale-in could kill the Task.Run mid-flight, leaving // rule results un-persisted (manual "Analyze Now" was the only recovery). Same producer + -// worker pattern as IndexReconcile above. +// handler pattern as IndexReconcile above; consumer = Functions/Queue/AnalyzeOnEnrollmentEndQueueFunction. builder.Services.AddSingleton< AutopilotMonitor.Functions.Services.Analyze.IAnalyzeOnEnrollmentEndProducer, AutopilotMonitor.Functions.Services.Analyze.AzureQueueAnalyzeOnEnrollmentEndProducer>(); builder.Services.AddSingleton< AutopilotMonitor.Functions.Services.Analyze.AnalyzeOnEnrollmentEndHandler>(); -builder.Services.AddHostedService< - AutopilotMonitor.Functions.Services.Analyze.AnalyzeOnEnrollmentEndQueueWorker>(); // Vulnerability-correlate fan-out triggered by the shutdown software_inventory_analysis // event. Replaces the in-function fire-and-forget Task.Run inside EventIngestProcessor. -// Same producer + worker pattern as the analyze queue. Inventory loader is DI-shared with +// Same producer + handler pattern as the analyze queue; consumer = +// Functions/Queue/VulnerabilityCorrelateQueueFunction. Inventory loader is DI-shared with // the manual rescan endpoint (GetVulnerabilityReportFunction ?rescan=true). builder.Services.AddSingleton< AutopilotMonitor.Functions.Services.Vulnerability.IVulnerabilityCorrelateProducer, @@ -357,13 +352,12 @@ builder.Services.AddSingleton< AutopilotMonitor.Functions.Services.Vulnerability.VulnerabilityInventoryLoader>(); builder.Services.AddSingleton< AutopilotMonitor.Functions.Services.Vulnerability.VulnerabilityCorrelateHandler>(); -builder.Services.AddHostedService< - AutopilotMonitor.Functions.Services.Vulnerability.VulnerabilityCorrelateQueueWorker>(); // IME-installer archiving on first fleet-wide sighting of a new IME version (~monthly). -// Producer sits in EventIngestProcessor's new-version continuation; the worker downloads -// the MSI from the CSP-reported (allowlisted) URL into the permanent ime-archive container -// and merges the outcome onto the ImeVersionHistory row. Typed HttpClient: no redirects +// Producer sits in EventIngestProcessor's new-version continuation; the consumer +// (Functions/Queue/ImeMsiArchiveQueueFunction) downloads the MSI from the CSP-reported +// (allowlisted) URL into the permanent ime-archive container and merges the outcome onto the +// ImeVersionHistory row. Typed HttpClient: no redirects // (SSRF posture — the allowlisted host must serve directly), long timeout owned by the // archiver's own CTS, transient retry via the shared external-data policy. builder.Services.AddHttpClient() @@ -377,20 +371,17 @@ builder.Services.AddHttpClient(); -builder.Services.AddHostedService< - AutopilotMonitor.Functions.Services.Ime.ImeMsiArchiveQueueWorker>(); // Delayed tenant auto-approve (public availability). Producer fires at first-login signup -// with a ~1-minute visibility delay; the worker activates the tenant via the shared -// TenantApprovalService only if AdminConfiguration.AutoApproveNewTenants is enabled at -// processing time. Same producer + worker pattern as the analyze queue above. +// with a ~1-minute visibility delay; the consumer (Functions/Queue/TenantAutoApproveQueueFunction) +// activates the tenant via the shared TenantApprovalService only if +// AdminConfiguration.AutoApproveNewTenants is enabled at processing time. Same producer + +// handler pattern as the analyze queue above. builder.Services.AddSingleton< AutopilotMonitor.Functions.Services.Activation.ITenantAutoApproveEnqueuer, AutopilotMonitor.Functions.Services.Activation.AzureQueueTenantAutoApproveEnqueuer>(); builder.Services.AddSingleton< AutopilotMonitor.Functions.Services.Activation.TenantAutoApproveHandler>(); -builder.Services.AddHostedService< - AutopilotMonitor.Functions.Services.Activation.TenantAutoApproveQueueWorker>(); // ── Critical-Table Backup (plan §PR1) ───────────────────────────────────────── // Lease-aware service registered separately from the timer/worker so the lease @@ -406,8 +397,9 @@ builder.Services.AddSingleton(); -// Worker = BackgroundService, not QueueTrigger (matches AnalyzeOnEnrollmentEndQueueWorker -// pattern per plan §Wave4 #1 — avoids the QueueTrigger-specific connection app-setting). +// Worker = self-managed BackgroundService, not QueueTrigger: the lease/CAS state machine +// (Running→Completed under lease, Failed only on poison-move) is a per-message lifecycle the +// host cannot express. See QueuePollingWorkerBase for the two-class rule. builder.Services.AddHostedService< AutopilotMonitor.Functions.Services.Backup.Queue.CriticalTableBackupQueueWorker>(); // PR2: single-row restore. Both validators are pure / I/O-only; the restore @@ -469,6 +461,14 @@ builder.Services.AddHttpClient() // Channel-level send API — routes each NotificationChannel to its transport (webhook renderer // vs. the platform Telegram bot). Transient: both transports are typed HttpClients. builder.Services.AddTransient(); +// Durable channel notifications from the agent hot paths (ingest terminal alerts, hardware +// rejection): the producer enqueues channel ids + alert, Functions/Queue/NotificationDispatchQueueFunction +// re-resolves the channels from the tenant config and sends. Handler is transient because the +// dispatcher it wraps is. +builder.Services.AddSingleton< + AutopilotMonitor.Functions.Services.Notifications.INotificationDispatchProducer, + AutopilotMonitor.Functions.Services.Notifications.AzureQueueNotificationDispatchProducer>(); +builder.Services.AddTransient(); // Transactional email (welcome / farewell). Typed client like the other outbound notifiers; // the provider is an implementation detail of EmailService + the Email:* settings. builder.Services.AddSingleton(); diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Activation/TenantAutoApproveQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Activation/TenantAutoApproveQueueWorker.cs deleted file mode 100644 index 2298684b..00000000 --- a/src/Backend/AutopilotMonitor.Functions/Services/Activation/TenantAutoApproveQueueWorker.cs +++ /dev/null @@ -1,51 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using Azure.Storage.Queues; -using AutopilotMonitor.Functions.Services.Queueing; -using AutopilotMonitor.Shared; -using Microsoft.Extensions.Logging; - -namespace AutopilotMonitor.Functions.Services.Activation -{ - /// - /// Background poll-loop for the tenant-auto-approve queue. Thin queue shell — - /// all decisions live in . Defaults of - /// (batch 32, 5-min visibility, 10-s poll, - /// poison after 5 attempts) fit this tiny workload. - /// - public sealed class TenantAutoApproveQueueWorker : QueuePollingWorker - { - private readonly TenantAutoApproveHandler _handler; - - public TenantAutoApproveQueueWorker( - QueueClientFactory queueFactory, - TenantAutoApproveHandler handler, - ILogger logger) - : base(queueFactory, Constants.QueueNames.TenantAutoApprove, logger) - { - _handler = handler; - } - - /// Test seam: inject pre-built (possibly in-memory) queues. - internal TenantAutoApproveQueueWorker( - QueueClient mainQueue, - QueueClient poisonQueue, - TenantAutoApproveHandler handler, - ILogger logger, - TimeSpan? pollIntervalOverride = null) - : base(mainQueue, poisonQueue, logger, pollIntervalOverride) - { - _handler = handler; - } - - protected override bool TryValidate(TenantAutoApproveEnvelope envelope) - => !string.IsNullOrWhiteSpace(envelope.TenantId); - - protected override string DescribeForLog(TenantAutoApproveEnvelope envelope) - => $"tenant={envelope.TenantId}"; - - protected override Task HandleAsync(TenantAutoApproveEnvelope envelope, CancellationToken ct) - => _handler.HandleAsync(envelope, ct); - } -} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AnalyzeOnEnrollmentEndQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AnalyzeOnEnrollmentEndQueueWorker.cs deleted file mode 100644 index ce1cca53..00000000 --- a/src/Backend/AutopilotMonitor.Functions/Services/Analyze/AnalyzeOnEnrollmentEndQueueWorker.cs +++ /dev/null @@ -1,34 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using AutopilotMonitor.Functions.Services.Queueing; -using AutopilotMonitor.Shared; -using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Logging; - -namespace AutopilotMonitor.Functions.Services.Analyze -{ - /// - /// Background worker for the analyze-on-enrollment-end queue. Replaces the previous - /// in-function fire-and-forget Task.Run that ran the rule engine after session-terminal - /// events — that pattern could be killed mid-flight by Azure Functions scale-in, leaving rule - /// results un-persisted. Uses the shared poll loop - /// with vanilla defaults (batch 32, 5-min visibility, 5 attempts → poison). - /// - public sealed class AnalyzeOnEnrollmentEndQueueWorker : QueuePollingWorker - { - private readonly AnalyzeOnEnrollmentEndHandler _handler; - - public AnalyzeOnEnrollmentEndQueueWorker( - QueueClientFactory queueFactory, - AnalyzeOnEnrollmentEndHandler handler, - ILogger logger) - : base(queueFactory, Constants.QueueNames.AnalyzeOnEnrollmentEnd, logger) - { - _handler = handler ?? throw new ArgumentNullException(nameof(handler)); - } - - protected override Task HandleAsync(AnalyzeOnEnrollmentEndEnvelope envelope, CancellationToken ct) - => _handler.HandleAsync(envelope, ct); - } -} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs index 775a01f9..0c79f776 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Backup/Queue/CriticalTableBackupQueueWorker.cs @@ -31,7 +31,7 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue /// CAS Running→Completed (under lease) → Release Lease → SafeDelete. /// /// - /// Queue-config differs from short-lived workers (plan Wave6): =1, + /// Queue-config differs from short-lived workers (plan Wave6): /// =60min. The per-run service budget is 50min so the 60min /// visibility gives a 10min cushion. /// @@ -61,7 +61,6 @@ namespace AutopilotMonitor.Functions.Services.Backup.Queue _clock = clock ?? TimeProvider.System; } - protected override int BatchSize => 1; protected override TimeSpan VisibilityTimeout => TimeSpan.FromMinutes(60); protected override async Task ProcessOneAsync(QueueMessage msg, CancellationToken ct) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionMaintenanceQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionMaintenanceQueueWorker.cs index a390f6a4..59c4eaf2 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionMaintenanceQueueWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionMaintenanceQueueWorker.cs @@ -20,7 +20,7 @@ namespace AutopilotMonitor.Functions.Services.Deletion /// retried — the operator asked for "a run", and one is already active. /// /// - /// BatchSize=1 + VisibilityTimeout=60min (run budget is 50min plus cushion — same rationale + /// VisibilityTimeout=60min (run budget is 50min plus cushion — same rationale /// as CriticalTableBackupQueueWorker). Real exceptions rethrow → visibility-timeout /// retry → poison after . /// @@ -38,7 +38,6 @@ namespace AutopilotMonitor.Functions.Services.Deletion _maintenance = maintenance ?? throw new ArgumentNullException(nameof(maintenance)); } - protected override int BatchSize => 1; protected override TimeSpan VisibilityTimeout => TimeSpan.FromMinutes(60); protected override bool TryValidate(SessionDeletionMaintenanceTriggerEnvelope envelope) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs index 756fdc47..a3bff177 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Deletion/SessionDeletionWorker.cs @@ -15,10 +15,9 @@ namespace AutopilotMonitor.Functions.Services.Deletion { /// /// Background poll-loop for the session-deletion queue (plan §5 PR4). Built on the - /// shared with four explicit deviations: + /// shared with three explicit deviations (one + /// cascade per receive — tens of MB and minutes of wall-time — is the base default): /// - /// = 1 (not 32) — a single cascade can be tens of MB - /// and minutes of wall-time; one-at-a-time bounds memory and poison blast radius. /// Kill-switch on entry via : every receive checks /// AdminConfiguration.SessionDeletionKillSwitch; when active the worker idles /// without dequeuing. @@ -83,13 +82,6 @@ namespace AutopilotMonitor.Functions.Services.Deletion _opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents)); } - /// - /// Explicit deviation from sibling workers — one cascade per receive bounds the worker's - /// memory footprint (a 35MB snapshot blob × 32 messages = 1.1GB) and limits poison-queue - /// blast radius on a corruption signal. Plan §5 PR4. - /// - protected override int BatchSize => 1; - protected override bool UseHeartbeat => true; protected override async ValueTask ShouldPauseAsync(CancellationToken ct) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Notifications.cs b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Notifications.cs index 2e4bd6f7..e579dd2c 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Notifications.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.Notifications.cs @@ -72,9 +72,7 @@ namespace AutopilotMonitor.Functions.Services sessionUrl: sessionUrl); NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults); - _ = _channelDispatcher.SendToChannelsAsync(targets, alert) - .ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException, - "Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted); + await EnqueueNotificationAsync(request, targets, alert).ConfigureAwait(false); } } @@ -94,9 +92,7 @@ namespace AutopilotMonitor.Functions.Services sessionUrl: sessionUrl); NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults); - _ = _channelDispatcher.SendToChannelsAsync(successChannels, alert) - .ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException, - "Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted); + await EnqueueNotificationAsync(request, successChannels, alert).ConfigureAwait(false); } if (c.EspFailureEvent != null && updatedSession?.IsPreProvisioned == true && failureChannels.Count > 0) @@ -115,12 +111,28 @@ namespace AutopilotMonitor.Functions.Services sessionUrl: sessionUrl); NotificationAlertBuilder.AddRuleResultSections(alert, ruleResults); - _ = _channelDispatcher.SendToChannelsAsync(failureChannels, alert) - .ContinueWith(t => _logger.LogWarning(t.Exception?.InnerException, - "Fire-and-forget webhook notification failed"), TaskContinuationOptions.OnlyOnFaulted); + await EnqueueNotificationAsync(request, failureChannels, alert).ConfigureAwait(false); } } + /// + /// Hands an alert to the notification-dispatch queue. The send used to be fire-and-forget + /// after the HTTP response; the host does not count such work at shutdown, so a Flex + /// scale-in could drop a customer's enrollment webhook. The envelope carries channel ids + /// only — the consumer re-resolves URLs and secrets from the tenant config at send time. + /// The producer is fail-soft, so the agent's response is never blocked. + /// + private Task EnqueueNotificationAsync( + IngestEventsRequest request, List channels, NotificationAlert alert) + => _notificationProducer.EnqueueAsync(new NotificationDispatchEnvelope + { + TenantId = request.TenantId, + SessionId = request.SessionId, + ChannelIds = channels.Select(c => c.Id).ToList(), + Alert = alert, + EnqueuedAt = DateTime.UtcNow, + }); + private SignalRMessageAction[] BuildSignalRMessages( IngestEventsRequest request, SessionSummary? updatedSession, int processedCount, List newRuleResults) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs index 80ce2be9..938c60f6 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/EventIngestProcessor.cs @@ -38,7 +38,7 @@ namespace AutopilotMonitor.Functions.Services private readonly TenantConfigurationService _configService; private readonly AnalyzeRuleService _analyzeRuleService; private readonly GatherRuleService _gatherRuleService; - private readonly NotificationChannelDispatcher _channelDispatcher; + private readonly INotificationDispatchProducer _notificationProducer; private readonly AdminConfigurationService _adminConfigService; private readonly OpsEventService _opsEventService; private readonly SlaBreachEvaluationService _slaBreachService; @@ -61,7 +61,7 @@ namespace AutopilotMonitor.Functions.Services TenantConfigurationService configService, AnalyzeRuleService analyzeRuleService, GatherRuleService gatherRuleService, - NotificationChannelDispatcher channelDispatcher, + INotificationDispatchProducer notificationProducer, AdminConfigurationService adminConfigService, OpsEventService opsEventService, SlaBreachEvaluationService slaBreachService, @@ -81,7 +81,7 @@ namespace AutopilotMonitor.Functions.Services _configService = configService; _analyzeRuleService = analyzeRuleService; _gatherRuleService = gatherRuleService; - _channelDispatcher = channelDispatcher; + _notificationProducer = notificationProducer; _adminConfigService = adminConfigService; _opsEventService = opsEventService; _slaBreachService = slaBreachService; diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Ime/AzureQueueImeMsiArchiveProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Ime/AzureQueueImeMsiArchiveProducer.cs index 35cf61be..3a1ad219 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Ime/AzureQueueImeMsiArchiveProducer.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Ime/AzureQueueImeMsiArchiveProducer.cs @@ -33,7 +33,7 @@ namespace AutopilotMonitor.Functions.Services.Ime ILogger logger) { _logger = logger; - // Base64 encoding matches the consumer (BackgroundService) so messages round-trip. + // Base64 encoding matches the host's queue-trigger default so messages round-trip. _queueClient = queueFactory.Create(Constants.QueueNames.ImeMsiArchive); } @@ -49,7 +49,10 @@ namespace AutopilotMonitor.Functions.Services.Ime _logger = logger; } - public async Task EnqueueAsync(ImeMsiArchiveEnvelope envelope, CancellationToken cancellationToken = default) + public async Task EnqueueAsync( + ImeMsiArchiveEnvelope envelope, + TimeSpan? visibilityDelay = null, + CancellationToken cancellationToken = default) { if (envelope is null) return; if (string.IsNullOrEmpty(envelope.Version)) @@ -63,10 +66,13 @@ namespace AutopilotMonitor.Functions.Services.Ime try { var body = JsonConvert.SerializeObject(envelope); - await _queueClient.SendMessageAsync(body, cancellationToken: cancellationToken).ConfigureAwait(false); + await _queueClient.SendMessageAsync( + body, + visibilityTimeout: visibilityDelay, + cancellationToken: cancellationToken).ConfigureAwait(false); _logger.LogInformation( - "ImeMsiArchive enqueued (version={Version} urlPresent={UrlPresent} tenant={Tenant} session={Session})", - envelope.Version, !string.IsNullOrEmpty(envelope.MsiDownloadUrl), envelope.TenantId, envelope.SessionId); + "ImeMsiArchive enqueued (version={Version} urlPresent={UrlPresent} tenant={Tenant} session={Session} delay={Delay})", + envelope.Version, !string.IsNullOrEmpty(envelope.MsiDownloadUrl), envelope.TenantId, envelope.SessionId, visibilityDelay); } catch (Exception ex) { diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Ime/IImeMsiArchiveProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Ime/IImeMsiArchiveProducer.cs index d349fe66..0163117a 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Ime/IImeMsiArchiveProducer.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Ime/IImeMsiArchiveProducer.cs @@ -1,3 +1,4 @@ +using System; using System.Threading; using System.Threading.Tasks; using AutopilotMonitor.Shared.Models; @@ -8,9 +9,14 @@ namespace AutopilotMonitor.Functions.Services.Ime /// Enqueues an IME-installer archive job when a new IME version is first sighted. /// Implementations must be fail-soft: the caller sits on the ingest hot path's /// fire-and-forget continuation and must never observe an exception. + /// parks the message: the queue function uses it to + /// re-enqueue while archiving is switched off. /// public interface IImeMsiArchiveProducer { - Task EnqueueAsync(ImeMsiArchiveEnvelope envelope, CancellationToken cancellationToken = default); + Task EnqueueAsync( + ImeMsiArchiveEnvelope envelope, + TimeSpan? visibilityDelay = null, + CancellationToken cancellationToken = default); } } diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Ime/ImeMsiArchiveQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Ime/ImeMsiArchiveQueueWorker.cs deleted file mode 100644 index ff673f00..00000000 --- a/src/Backend/AutopilotMonitor.Functions/Services/Ime/ImeMsiArchiveQueueWorker.cs +++ /dev/null @@ -1,107 +0,0 @@ -using AutopilotMonitor.Functions.Services.Queueing; -using AutopilotMonitor.Shared; -using AutopilotMonitor.Shared.DataAccess; -using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Logging; - -namespace AutopilotMonitor.Functions.Services.Ime -{ - /// - /// Background worker for the ime-msi-archive queue: downloads a newly sighted IME - /// installer into the permanent ime-archive container and merges the outcome onto - /// the ImeVersionHistory row. Volume is ~one message per Microsoft IME release. - /// - /// Retry semantics: the archiver never throws — it classifies. Permanent rejections - /// (bad version, size cap) complete the message with their Failed:* status - /// persisted; transient failures (download/timeout) rethrow so the visibility-timeout - /// retry → poison ladder applies, with the last status visible on the row. - /// keeps the message invisible across the up-to-10-minute - /// download window. The status row-merge happens on EVERY attempt so operators can see - /// a stuck download as Failed:Download instead of nothing. - /// - /// - public sealed class ImeMsiArchiveQueueWorker : QueuePollingWorker - { - private readonly ImeMsiArchiver _archiver; - private readonly ISessionRepository _sessionRepo; - private readonly AdminConfigurationService _adminConfigService; - - public ImeMsiArchiveQueueWorker( - QueueClientFactory queueFactory, - ImeMsiArchiver archiver, - ISessionRepository sessionRepo, - AdminConfigurationService adminConfigService, - ILogger logger) - : base(queueFactory, Constants.QueueNames.ImeMsiArchive, logger) - { - _archiver = archiver ?? throw new ArgumentNullException(nameof(archiver)); - _sessionRepo = sessionRepo ?? throw new ArgumentNullException(nameof(sessionRepo)); - _adminConfigService = adminConfigService ?? throw new ArgumentNullException(nameof(adminConfigService)); - } - - /// - /// Test seam: construct directly with mock - /// instances so the poll loop can be exercised without Azurite. Mirrors - /// 's internal test ctor. - /// - internal ImeMsiArchiveQueueWorker( - Azure.Storage.Queues.QueueClient mainQueue, - Azure.Storage.Queues.QueueClient poisonQueue, - ImeMsiArchiver archiver, - ISessionRepository sessionRepo, - AdminConfigurationService adminConfigService, - ILogger logger, - TimeSpan? pollInterval = null) - : base(mainQueue, poisonQueue, logger, pollInterval) - { - _archiver = archiver ?? throw new ArgumentNullException(nameof(archiver)); - _sessionRepo = sessionRepo ?? throw new ArgumentNullException(nameof(sessionRepo)); - _adminConfigService = adminConfigService ?? throw new ArgumentNullException(nameof(adminConfigService)); - } - - /// Download can legitimately take minutes — extend visibility while working. - protected override bool UseHeartbeat => true; - - /// - /// Paused while is off — - /// messages stay parked (not dropped) and are processed when the flag returns. - /// Uses the cached config read: this is a pause, not a kill-switch, and the worker - /// only wakes for rare messages anyway. - /// - protected override async ValueTask ShouldPauseAsync(CancellationToken ct) - { - try - { - var config = await _adminConfigService.GetConfigurationAsync(); - return !config.ImeMsiArchivingEnabled; - } - catch - { - return false; // config unreadable → keep working (archive is fail-soft anyway) - } - } - - protected override bool TryValidate(ImeMsiArchiveEnvelope envelope) - => !string.IsNullOrEmpty(envelope.Version); - - protected override string DescribeForLog(ImeMsiArchiveEnvelope envelope) - => $"version {envelope.Version}"; - - protected override async Task HandleAsync(ImeMsiArchiveEnvelope envelope, CancellationToken ct) - { - var result = await _archiver.ArchiveAsync(envelope, ct); - - // Merge the outcome first — also for failures, so the row always tells the truth. - await _sessionRepo.UpdateImeVersionArchiveInfoAsync( - envelope.Version, result.Status, result.BlobPath, result.Sha256, result.SizeBytes, result.SourceUrl); - - if (!result.Success && result.Retryable) - { - // Visibility-timeout retry → poison after MaxDequeueCount. The thrown message - // lands in the worker's handler-failure log line. - throw new InvalidOperationException( - $"IME MSI archive attempt failed with {result.Status} for version {envelope.Version}"); - } - } - } -} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Indexing/IndexReconcileQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Indexing/IndexReconcileQueueWorker.cs deleted file mode 100644 index 8e6215ca..00000000 --- a/src/Backend/AutopilotMonitor.Functions/Services/Indexing/IndexReconcileQueueWorker.cs +++ /dev/null @@ -1,34 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using AutopilotMonitor.Functions.Services.Queueing; -using AutopilotMonitor.Shared; -using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Logging; - -namespace AutopilotMonitor.Functions.Services.Indexing -{ - /// - /// Background worker for the telemetry-index-reconcile queue (Plan §2.8, §M5.d.3). - /// Replaces the earlier IndexReconcileFunction QueueTrigger (whose binding required a - /// Functions-host-specific connection app-setting the project's storage access does not need) - /// with the shared poll loop. Vanilla defaults: - /// batch 32, 5-min visibility, 5 attempts → poison. - /// - public sealed class IndexReconcileQueueWorker : QueuePollingWorker - { - private readonly IndexReconcileHandler _handler; - - public IndexReconcileQueueWorker( - QueueClientFactory queueFactory, - IndexReconcileHandler handler, - ILogger logger) - : base(queueFactory, Constants.QueueNames.TelemetryIndexReconcile, logger) - { - _handler = handler ?? throw new ArgumentNullException(nameof(handler)); - } - - protected override Task HandleAsync(IndexReconcileEnvelope envelope, CancellationToken ct) - => _handler.HandleAsync(envelope, ct); - } -} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Notifications/AzureQueueNotificationDispatchProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Notifications/AzureQueueNotificationDispatchProducer.cs new file mode 100644 index 00000000..df141cce --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Services/Notifications/AzureQueueNotificationDispatchProducer.cs @@ -0,0 +1,89 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using Azure.Storage.Queues; +using AutopilotMonitor.Functions.Services.Queueing; +using AutopilotMonitor.Shared; +using AutopilotMonitor.Shared.Models.Notifications; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; + +namespace AutopilotMonitor.Functions.Services.Notifications +{ + /// + /// Azure Storage Queue implementation of . Mirrors + /// : Managed Identity preferred, + /// connection-string fallback, Base64 encoding (matches the host's queue trigger default). + /// + /// Fault tolerance: queue-side exceptions are logged but never rethrown — the agent's + /// HTTP 200 must not be blocked. A missed enqueue means the alert is not delivered; the + /// warning names tenant and event type so an operator can see it. + /// + /// + public sealed class AzureQueueNotificationDispatchProducer : INotificationDispatchProducer + { + private readonly QueueClient _queueClient; + private readonly ILogger _logger; + + private int _queueEnsured; // 0 = not yet ensured, 1 = CreateIfNotExistsAsync has run + + public AzureQueueNotificationDispatchProducer( + QueueClientFactory queueFactory, + ILogger logger) + { + _logger = logger; + _queueClient = queueFactory.Create(Constants.QueueNames.NotificationDispatch); + } + + /// Test seam: construct directly with a mock . + internal AzureQueueNotificationDispatchProducer( + QueueClient queueClient, + ILogger logger) + { + _queueClient = queueClient; + _logger = logger; + } + + public async Task EnqueueAsync(NotificationDispatchEnvelope envelope, CancellationToken cancellationToken = default) + { + if (envelope is null) return; + if (string.IsNullOrEmpty(envelope.TenantId) || envelope.Alert is null || envelope.ChannelIds.Count == 0) + { + _logger.LogWarning( + "NotificationDispatch enqueue skipped — missing TenantId/Alert/ChannelIds (tenant={Tenant} event={EventType})", + envelope.TenantId, envelope.Alert?.EventType); + return; + } + + await EnsureQueueExistsAsync(cancellationToken).ConfigureAwait(false); + + try + { + var body = JsonConvert.SerializeObject(envelope); + await _queueClient.SendMessageAsync(body, cancellationToken: cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "NotificationDispatch enqueue failed (tenant={Tenant} session={Session} event={EventType} channels={Channels}) — alert not delivered", + envelope.TenantId, envelope.SessionId, envelope.Alert.EventType, envelope.ChannelIds.Count); + } + } + + private async Task EnsureQueueExistsAsync(CancellationToken cancellationToken) + { + if (_queueEnsured == 1) return; + + try + { + await _queueClient.CreateIfNotExistsAsync(cancellationToken: cancellationToken).ConfigureAwait(false); + Interlocked.Exchange(ref _queueEnsured, 1); + } + catch (Exception ex) + { + // Leave _queueEnsured=0 so a later call retries. + _logger.LogWarning(ex, "NotificationDispatch queue CreateIfNotExists failed — will retry next enqueue"); + } + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Notifications/INotificationDispatchProducer.cs b/src/Backend/AutopilotMonitor.Functions/Services/Notifications/INotificationDispatchProducer.cs new file mode 100644 index 00000000..b39aa40f --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Services/Notifications/INotificationDispatchProducer.cs @@ -0,0 +1,16 @@ +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Shared.Models.Notifications; + +namespace AutopilotMonitor.Functions.Services.Notifications +{ + /// + /// Enqueues a channel notification for durable delivery by + /// NotificationDispatchQueueFunction. Implementations must be fail-soft: callers sit + /// on the agent's hot path (ingest, distress) and must never observe an exception. + /// + public interface INotificationDispatchProducer + { + Task EnqueueAsync(NotificationDispatchEnvelope envelope, CancellationToken cancellationToken = default); + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Notifications/NotificationDispatchHandler.cs b/src/Backend/AutopilotMonitor.Functions/Services/Notifications/NotificationDispatchHandler.cs new file mode 100644 index 00000000..e4ee5f5d --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Services/Notifications/NotificationDispatchHandler.cs @@ -0,0 +1,64 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using AutopilotMonitor.Shared.Models.Notifications; +using Microsoft.Extensions.Logging; + +namespace AutopilotMonitor.Functions.Services.Notifications +{ + /// + /// Consumer side of the notification-dispatch queue: resolves the envelope's channel + /// ids against the tenant's CURRENT configuration (so secrets never travel in the queue and + /// a channel disabled meanwhile is skipped) and hands the alert to + /// . The dispatcher never throws — a failing + /// destination only logs — so the queue guarantees the attempt, not the delivery; per-request + /// retries stay with the transport's resilience policy. + /// + public class NotificationDispatchHandler + { + private readonly TenantConfigurationService _configService; + private readonly NotificationChannelDispatcher _dispatcher; + private readonly ILogger _logger; + + public NotificationDispatchHandler( + TenantConfigurationService configService, + NotificationChannelDispatcher dispatcher, + ILogger logger) + { + _configService = configService ?? throw new ArgumentNullException(nameof(configService)); + _dispatcher = dispatcher ?? throw new ArgumentNullException(nameof(dispatcher)); + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + } + + public virtual async Task HandleAsync(NotificationDispatchEnvelope envelope, CancellationToken cancellationToken) + { + if (envelope is null) throw new ArgumentNullException(nameof(envelope)); + + var (config, exists) = await _configService.TryGetConfigurationAsync(envelope.TenantId).ConfigureAwait(false); + if (!exists) + { + _logger.LogWarning( + "NotificationDispatch: tenant {Tenant} has no configuration — alert {EventType} dropped", + envelope.TenantId, envelope.Alert?.EventType); + return; + } + + var wanted = new HashSet(envelope.ChannelIds ?? new List(), StringComparer.Ordinal); + var targets = config.GetNotificationChannels() + .Where(c => c.Enabled && wanted.Contains(c.Id)) + .ToList(); + + if (targets.Count == 0) + { + _logger.LogWarning( + "NotificationDispatch: none of {Requested} channel(s) is still enabled for tenant {Tenant} — alert {EventType} dropped", + wanted.Count, envelope.TenantId, envelope.Alert?.EventType); + return; + } + + await _dispatcher.SendToChannelsAsync(targets, envelope.Alert).ConfigureAwait(false); + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs index 6fb667dd..e64c23fa 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Offboarding/TenantOffboardingWorker.cs @@ -15,9 +15,9 @@ namespace AutopilotMonitor.Functions.Services.Offboarding /// Background poll-loop for the tenant-offboarding queue (Plan §7.3). Built on the /// shared : /// - /// = 1 — one tenant cascade per receive bounds memory - /// (per-tenant enumerate + per-session enqueue is long-running) and limits the blast - /// radius of a poison. + /// One tenant cascade per receive (the base default) bounds memory — per-tenant + /// enumerate + per-session enqueue is long-running — and limits the blast radius of a + /// poison. /// Heartbeat () extends visibility while the handler runs so /// concurrent re-delivery can't spawn a parallel offboarding for the same tenant. /// Max-dequeue → poison queue, but persists the @@ -57,8 +57,6 @@ namespace AutopilotMonitor.Functions.Services.Offboarding _opsEvents = opsEvents ?? throw new ArgumentNullException(nameof(opsEvents)); } - protected override int BatchSize => 1; - protected override bool UseHeartbeat => true; protected override bool TryValidate(TenantOffboardingEnvelope envelope) diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueEnvelopeReader.cs b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueEnvelopeReader.cs new file mode 100644 index 00000000..fce91054 --- /dev/null +++ b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueueEnvelopeReader.cs @@ -0,0 +1,67 @@ +using System; +using Azure.Storage.Queues.Models; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; + +namespace AutopilotMonitor.Functions.Services.Queueing +{ + /// + /// Envelope decoding shared by the [QueueTrigger] functions under Functions/Queue/. + /// The functions bind the raw rather than a POCO so the producers' + /// Newtonsoft contract stays the single wire format and + /// reaches the log line. + /// + /// A false result means "permanently unusable": malformed JSON, a null literal, + /// or a failed validation. The caller returns normally, the host deletes the message — the + /// same drop the self-managed poll loop performs, without burning a dequeue attempt. + /// + /// + public static class QueueEnvelopeReader + { + public static bool TryRead( + QueueMessage message, + ILogger logger, + out TEnvelope envelope, + Func? validate = null) + where TEnvelope : class + { + if (message is null) throw new ArgumentNullException(nameof(message)); + if (logger is null) throw new ArgumentNullException(nameof(logger)); + + envelope = null!; + var name = typeof(TEnvelope).Name; + + TEnvelope? parsed; + try + { + parsed = JsonConvert.DeserializeObject(message.Body.ToString()); + } + catch (JsonException ex) + { + logger.LogWarning(ex, + "{Envelope}: malformed envelope JSON — dropping (msg {Id}, dequeue {N})", + name, message.MessageId, message.DequeueCount); + return false; + } + + if (parsed is null) + { + logger.LogWarning( + "{Envelope}: null envelope after deserialization — dropping (msg {Id}, dequeue {N})", + name, message.MessageId, message.DequeueCount); + return false; + } + + if (validate is not null && !validate(parsed)) + { + logger.LogWarning( + "{Envelope}: envelope failed validation (missing required fields) — dropping (msg {Id}, dequeue {N})", + name, message.MessageId, message.DequeueCount); + return false; + } + + envelope = parsed; + return true; + } + } +} diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorkerBase.cs b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorkerBase.cs index 85779c42..10d2191a 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorkerBase.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/Queueing/QueuePollingWorkerBase.cs @@ -9,13 +9,20 @@ using Microsoft.Extensions.Logging; namespace AutopilotMonitor.Functions.Services.Queueing { /// - /// Shared poll-loop skeleton for the project's Storage-Queue - /// workers (code-quality audit 2026-05-29, finding D2). Replaces the QueueTrigger binding — - /// which needs a Functions-host-specific <Connection>__queueServiceUri app-setting - /// the rest of the project's storage access does not use — with a pure DI poll loop that - /// reproduces the platform's retry semantics: + /// Shared poll-loop skeleton for the project's self-managed Storage-Queue + /// workers (code-quality audit 2026-05-29, finding D2). + /// + /// Two classes of queue consumer exist on purpose. Stateless "envelope in, handler, done" + /// queues are [QueueTrigger] functions under Functions/Queue/: the Functions host + /// owns lease renewal, poison-move and, on Flex Consumption, the scale signal that wakes an + /// instance for a message. This base serves the cascade workers whose per-message lifecycle + /// the host cannot express — lease/CAS state machines, a poison-move veto, a kill-switch pause + /// and an in-flight heartbeat. Their accepted limit: after a scale-in the cascade resumes on the + /// next instance, at the latest when the next timer wakes one. The loop reproduces the + /// platform's retry semantics: + /// /// - /// Receive up to messages with . + /// Receive message(s) with . /// Handler success → DeleteMessageAsync. /// Handler failure → no delete; the message reappears after the visibility timeout and /// DequeueCount increments (identical to platform retry). @@ -75,8 +82,14 @@ namespace AutopilotMonitor.Functions.Services.Queueing // ── Tunables (override per worker) ─────────────────────────────────────── - /// Max messages received per poll. Storage Queue caps batch-receive at 32. - protected virtual int BatchSize => 32; + /// + /// Messages received per poll. Default 1: the loop processes a batch sequentially and + /// the heartbeat only covers the in-flight message, so every message received beyond the + /// first would sit invisible with its clock running — with 32 × a 10-second handler the last + /// one reappears for another worker before this loop reaches it (audit 2026-09-05, F02). + /// A non-empty receive is followed by an immediate re-receive, so 1 costs no throughput. + /// + protected virtual int BatchSize => 1; /// Visibility timeout per received message — must cover the slowest handler path. protected virtual TimeSpan VisibilityTimeout => TimeSpan.FromMinutes(5); diff --git a/src/Backend/AutopilotMonitor.Functions/Services/RuleEngine.cs b/src/Backend/AutopilotMonitor.Functions/Services/RuleEngine.cs index c51b211a..742935f0 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/RuleEngine.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/RuleEngine.cs @@ -60,7 +60,7 @@ namespace AutopilotMonitor.Functions.Services /// Failure semantics: storage-layer exceptions from rule loading, event reading /// or existing-results lookup propagate to the caller. The queue worker relies on this /// to leave its message un-deleted so a transient Table Storage failure can be retried - /// via visibility-timeout (). + /// via visibility-timeout (AnalyzeOnEnrollmentEndQueueFunction). /// The on-demand HTTP path () /// wraps the call in its own try/catch and logs failures as warnings — the user sees /// the previously-stored results until they re-trigger. diff --git a/src/Backend/AutopilotMonitor.Functions/Services/TenantApprovalService.cs b/src/Backend/AutopilotMonitor.Functions/Services/TenantApprovalService.cs index 08daadbf..a688ef79 100644 --- a/src/Backend/AutopilotMonitor.Functions/Services/TenantApprovalService.cs +++ b/src/Backend/AutopilotMonitor.Functions/Services/TenantApprovalService.cs @@ -7,7 +7,7 @@ namespace AutopilotMonitor.Functions.Services; /// Shared tenant-activation path: whitelist approval plus the follow-up side effects /// (auto-promote of the onboarding user to TenantAdmin, welcome email). Used by both /// the Global Admin approve endpoint (PreviewWhitelistFunction) and the -/// auto-approve queue worker (TenantAutoApproveQueueWorker) so the two +/// auto-approve queue worker (TenantAutoApproveQueueFunction) so the two /// activation routes cannot drift. /// public class TenantApprovalService diff --git a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelateQueueWorker.cs b/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelateQueueWorker.cs deleted file mode 100644 index ce923fa3..00000000 --- a/src/Backend/AutopilotMonitor.Functions/Services/Vulnerability/VulnerabilityCorrelateQueueWorker.cs +++ /dev/null @@ -1,35 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; -using AutopilotMonitor.Functions.Services.Queueing; -using AutopilotMonitor.Shared; -using AutopilotMonitor.Shared.Models; -using Microsoft.Extensions.Logging; - -namespace AutopilotMonitor.Functions.Services.Vulnerability -{ - /// - /// Background worker for the vulnerability-correlate queue. Replaces the previous - /// in-function fire-and-forget Task.Run in EventIngestProcessor that ran - /// vulnerability correlation after a shutdown software_inventory_analysis event — that - /// pattern could be killed mid-flight by Azure Functions scale-in, leaving sessions without a - /// stored report. Uses the shared poll loop with - /// vanilla defaults (batch 32, 5-min visibility, 5 attempts → poison). - /// - public sealed class VulnerabilityCorrelateQueueWorker : QueuePollingWorker - { - private readonly VulnerabilityCorrelateHandler _handler; - - public VulnerabilityCorrelateQueueWorker( - QueueClientFactory queueFactory, - VulnerabilityCorrelateHandler handler, - ILogger logger) - : base(queueFactory, Constants.QueueNames.VulnerabilityCorrelate, logger) - { - _handler = handler ?? throw new ArgumentNullException(nameof(handler)); - } - - protected override Task HandleAsync(VulnerabilityCorrelateEnvelope envelope, CancellationToken ct) - => _handler.HandleAsync(envelope, ct); - } -} diff --git a/src/Backend/AutopilotMonitor.Functions/host.json b/src/Backend/AutopilotMonitor.Functions/host.json index b73e502e..eb89b5dd 100644 --- a/src/Backend/AutopilotMonitor.Functions/host.json +++ b/src/Backend/AutopilotMonitor.Functions/host.json @@ -27,6 +27,14 @@ "extensions": { "http": { "routePrefix": "api" + }, + "queues": { + "messageEncoding": "base64", + "batchSize": 4, + "newBatchThreshold": 2, + "maxDequeueCount": 5, + "visibilityTimeout": "00:01:00", + "maxPollingInterval": "00:00:10" } } } \ No newline at end of file diff --git a/src/Backend/AutopilotMonitor.Functions/local.settings.json.template b/src/Backend/AutopilotMonitor.Functions/local.settings.json.template index e3505da7..eb34fc64 100644 --- a/src/Backend/AutopilotMonitor.Functions/local.settings.json.template +++ b/src/Backend/AutopilotMonitor.Functions/local.settings.json.template @@ -6,6 +6,9 @@ "AzureTableStorageConnectionString": "UseDevelopmentStorage=true", "AzureBlobStorageConnectionString": "UseDevelopmentStorage=true", + "// DataStorage": "Connection the [QueueTrigger] functions under Functions/Queue bind to (Constants.QueueNames.TriggerConnection). Locally a connection string; in Azure the identity form 'DataStorage__queueServiceUri=https://.queue.core.windows.net'. The Functions host resolves it, so it cannot be derived from AzureStorageAccountName in code.", + "DataStorage": "UseDevelopmentStorage=true", + "// SignalR quota watcher": "Optional. Set to the full Azure resource id of the SignalR Service to enable hourly OpsEvents when the plan limits are approached (defaults: Standard S1 x1 unit = 1000 connections / 1M messages per day; override the limits when adding units or changing tier). Function App's managed identity needs 'Monitoring Reader' on this resource. Leave unset to skip the watcher.", "SignalRResourceId": "", "SignalRConnectionLimit": "1000", diff --git a/src/Shared/AutopilotMonitor.Shared/Constants.cs b/src/Shared/AutopilotMonitor.Shared/Constants.cs index 17cefd80..b296bcfe 100644 --- a/src/Shared/AutopilotMonitor.Shared/Constants.cs +++ b/src/Shared/AutopilotMonitor.Shared/Constants.cs @@ -1566,6 +1566,27 @@ namespace AutopilotMonitor.Shared /// public static class QueueNames { + /// + /// Name of the host connection every [QueueTrigger] function binds to + /// (Connection = TriggerConnection). Resolved by the Functions HOST, not by the + /// worker's DI, so it is its own app setting: in Azure the identity form + /// DataStorage__queueServiceUri pointing at the data storage account (the + /// Managed Identity holds Storage Queue Data Contributor there); locally a plain + /// connection string under DataStorage. Producers keep using + /// QueueClientFactory — same account, same queues. + /// + public const string TriggerConnection = "DataStorage"; + + /// + /// Durable channel notifications from the agent hot paths (terminal enrollment alerts + /// in ingest, hardware rejection in distress). Producer = INotificationDispatchProducer; + /// consumer = NotificationDispatchQueueFunction, which re-resolves the envelope's + /// channel ids against the tenant config and sends. Replaced fire-and-forget sends the + /// host did not count as open work at shutdown. Poison suffix -poison, + /// max-dequeue 5. + /// + public const string NotificationDispatch = "notification-dispatch"; + /// /// V2 Decision Engine index-table fan-out (Plan §2.8, §M5.d). One message per /// committed primary row (Signal or DecisionTransition); consumer writes the @@ -1621,7 +1642,7 @@ namespace AutopilotMonitor.Shared /// /// Delayed tenant auto-approve queue. Producer (AuthFunction first-login signup /// handler) enqueues one envelope per new tenant with a ~1-minute visibility delay; - /// consumer (TenantAutoApproveQueueWorker) activates the tenant via the shared + /// consumer (TenantAutoApproveQueueFunction) activates the tenant via the shared /// approval path if AdminConfiguration.AutoApproveNewTenants is enabled at /// processing time — otherwise the message is dropped and the tenant waits for /// manual approval. Poison suffix -poison, max-dequeue 5. @@ -1631,8 +1652,8 @@ namespace AutopilotMonitor.Shared /// /// Critical-table backup job queue. Producer = HTTP trigger /// /api/global/backups/trigger (fail-hard); consumer = - /// CriticalTableBackupQueueWorker (BackgroundService, BatchSize=1, - /// VisibilityTimeout=60min with 25min PopReceipt-renewal). Carries a small + /// CriticalTableBackupQueueWorker (self-managed BackgroundService, one message + /// per receive, VisibilityTimeout=60min with 25min PopReceipt-renewal). Carries a small /// envelope { jobId } — full state lives in . /// Poison-suffix -poison, max-dequeue 5; Failed-state is persisted on /// Poison-Move (NOT on first throw — that would defeat retry). @@ -1645,8 +1666,8 @@ namespace AutopilotMonitor.Shared /// /// Manual-trigger queue for the session-deletion maintenance run. Producer = HTTP /// trigger /api/global/session-deletions/maintenance/trigger (fail-hard); - /// consumer = SessionDeletionMaintenanceQueueWorker (BackgroundService, - /// BatchSize=1, VisibilityTimeout=60min — the run budget is 50min plus cushion). + /// consumer = SessionDeletionMaintenanceQueueWorker (self-managed + /// BackgroundService, VisibilityTimeout=60min — the run budget is 50min plus cushion). /// Carries a tiny envelope { triggeredBy }; concurrency against the 12h timer /// is serialized by the session-deletion maintenance blob lease, not the queue. /// Poison-suffix -poison, max-dequeue 5. @@ -1659,7 +1680,7 @@ namespace AutopilotMonitor.Shared /// /// IME-MSI archive fan-out. Producer = EventIngestProcessor when /// RecordImeVersionAsync reports a genuinely NEW IME version (first sighting - /// across the fleet — roughly monthly); consumer = ImeMsiArchiveQueueWorker, + /// across the fleet — roughly monthly); consumer = ImeMsiArchiveQueueFunction, /// which downloads the installer from the CSP-reported /// msiDownloadUrl (host-allowlisted, size-capped) into /// and merges the outcome into the diff --git a/src/Shared/AutopilotMonitor.Shared/Models/Notifications/NotificationDispatchEnvelope.cs b/src/Shared/AutopilotMonitor.Shared/Models/Notifications/NotificationDispatchEnvelope.cs new file mode 100644 index 00000000..f0ad9e0f --- /dev/null +++ b/src/Shared/AutopilotMonitor.Shared/Models/Notifications/NotificationDispatchEnvelope.cs @@ -0,0 +1,38 @@ +using System; +using System.Collections.Generic; + +namespace AutopilotMonitor.Shared.Models.Notifications +{ + /// + /// Queue-message envelope for the notification-dispatch queue: an enrollment or + /// hardware-rejection alert that must reach the tenant's channels even though the HTTP + /// request that produced it has already been answered. Replaces the fire-and-forget sends + /// in the ingest and distress paths, which the Functions host does not count as open work + /// at shutdown and which a Flex scale-in could therefore drop. + /// + /// The envelope names the target channels by only. The consumer + /// re-resolves them from the tenant configuration at send time, so no webhook URL, custom + /// header or signing secret is ever copied into a queue message, and a channel disabled + /// between enqueue and send is skipped. + /// + /// + public sealed class NotificationDispatchEnvelope + { + /// Schema version — bump on breaking envelope changes so consumers can reject or migrate. + public string EnvelopeVersion { get; set; } = "1"; + + public string TenantId { get; set; } = string.Empty; + + /// Session the alert is about; null for alerts without one (hardware rejection). + public string? SessionId { get; set; } + + /// Ids of the channels the producer selected (already filtered by their NotifyOn* toggle). + public List ChannelIds { get; set; } = new(); + + /// The fully built, channel-agnostic alert; renderers run in the consumer. + public NotificationAlert Alert { get; set; } = default!; + + /// UTC time the producer enqueued the message; useful for measuring queue lag. + public DateTime EnqueuedAt { get; set; } + } +}