Outbox blobs reach the worker over Event Grid

infra/core/dbote-gateway.ts

@@ -7,6 +7,9 @@ import { RoleAssignment } from "@cdktf/provider-azurerm/lib/role-assignment";
 import { StorageContainer } from "@cdktf/provider-azurerm/lib/storage-container";
 import { StorageTable } from "@cdktf/provider-azurerm/lib/storage-table";
 import { StorageManagementPolicy } from "@cdktf/provider-azurerm/lib/storage-management-policy";
+import { StorageQueue } from "@cdktf/provider-azurerm/lib/storage-queue";
+import { EventgridSystemTopic } from "@cdktf/provider-azurerm/lib/eventgrid-system-topic";
+import { EventgridSystemTopicEventSubscription } from "@cdktf/provider-azurerm/lib/eventgrid-system-topic-event-subscription";
 import { Environnment } from "./environment";
 import { DboteContainerRegistry, DboteWorker } from "./dbote-worker";
 
@@ -172,7 +175,7 @@ export class DboteGateway extends Construct {
             accountReplicationType: "LRS",
         });
         this.storageAccount = storageAccount;
-        // The worker's CloudOutboxBlobDetected trigger watches bote-outbox in this account.
+        // The cloud stages its attachments in bote-outbox of this account.
         this.dataBusOutboxStorageAccount = storageAccount;
 
         // DataBus staging containers. The worker's outbox blob trigger needs its container up
@@ -184,6 +187,29 @@ export class DboteGateway extends Construct {
             });
         }
 
+        // Event Grid reports every blob written to the outbox to a queue, which starts its copy to
+        // the clients (the worker's CloudOutboxBlobCreated). The blob trigger's polling scan misses
+        // blobs: extension 5.3.8 finds only those written before the host's first poll.
+        const outboxCreated = new StorageQueue(this, environment.formatName('stq', 'bote-outbox-created'), {
+            name: 'bote-outbox-created',
+            storageAccountId: storageAccount.id,
+        });
+        const storageEvents = new EventgridSystemTopic(this, environment.formatName('evgt', 'st'), {
+            name: environment.formatName('evgt', 'st'),
+            location: environment.location,
+            resourceGroupName: environment.resourceGroup,
+            sourceResourceId: storageAccount.id,
+            topicType: 'Microsoft.Storage.StorageAccounts',
+        });
+        new EventgridSystemTopicEventSubscription(this, environment.formatName('evgs', 'bote-outbox-created'), {
+            name: 'bote-outbox-created',
+            systemTopic: storageEvents.name,
+            resourceGroupName: environment.resourceGroup,
+            includedEventTypes: ['Microsoft.Storage.BlobCreated'],
+            subjectFilter: { subjectBeginsWith: '/blobServices/default/containers/bote-outbox/blobs/' },
+            storageQueueEndpoint: { storageAccountId: storageAccount.id, queueName: outboxCreated.name },
+        });
+
         // Topic subscriptions of the clients
         new StorageTable(this, environment.formatName('stt', 'subscriptions'), {
             storageAccountName: storageAccount.name,
@@ -260,6 +286,8 @@ export class DboteGateway extends Construct {
         // A connection string with the account key: the worker signs the SAS URIs for the clients
         // with it, and Storage Queues have no user delegation SAS.
         worker.addAppSetting('dbote__Worker__Storage__Connection', storageAccount.primaryConnectionString);
+        // Event Grid carries the outbox here; the blob trigger is for storages without it (Azurite).
+        worker.addAppSetting('AzureWebJobs.CloudOutboxBlobDetected.Disabled', 'true');
         worker.addAppSetting('dbote__Worker__ServiceBus__Connection__fullyQualifiedNamespace', this.fullyQualifiedNamespace);
         worker.addAppSetting('dbote__Worker__ServiceBus__Connection__credential', 'managedidentity');
         worker.addAppSetting('dbote__Worker__ServiceBus__Connection__clientId', worker.identityClientId);

src/Dbosoft.Bote.Primitives/BoteStorageConstants.cs

@@ -8,6 +8,11 @@ public static class BoteStorageConstants
     public const string Inbox = "bote-inbox";
     public const string Outbox = "bote-outbox";
 
+    /// <summary>
+    /// The queue of the gateway storage Event Grid reports the blobs written to the outbox to.
+    /// </summary>
+    public const string OutboxCreatedQueue = "bote-outbox-created";
+
     /// <summary>
     /// Blob metadata keys used for attachment tracking and validation.
     /// </summary>

src/workers/src/Dbosoft.Bote.BoteWorker/DataBusCopyFile.cs

@@ -1,7 +1,8 @@
-using Azure.Storage.Blobs;
+using Azure.Storage.Blobs;
 using Azure.Storage.Sas;
 using Dbosoft.Bote.BoteWorker.Extensions;
 using Dbosoft.Bote.BoteWorker.Services;
+using Dbosoft.Bote.Primitives;
 using JetBrains.Annotations;
 using Microsoft.Azure.Functions.Worker;
 using Microsoft.Extensions.Logging;
@@ -31,20 +32,46 @@ public class DataBusCopyRequest
 internal class DataBusCopyFile(
     ILogger<DataBusCopyFile> logger,
     DataBusCopyProcessor copyProcessor,
-    ITenantStorageResolver tenantStorageResolver)
+    ITenantStorageResolver tenantStorageResolver,
+    BlobServiceClient outbox)
 {
 
     /// <summary>
-    /// Blob trigger for cloud outbox - starts copy from cloud outbox to client
-    /// Initiates the Cloud → client copy flow.
+    /// A blob the cloud put into its outbox, as Event Grid reports it: starts its copy to where the
+    /// clients of its tenant read it. Event Grid, not the blob trigger below, is what Azure runs on:
+    /// the polling scan of the blob extension misses blobs (extension 5.3.8 finds only those written
+    /// before the host's first poll), and a message to a client waits for its attachment.
     /// </summary>
+    [PublicAPI]
+    [Function("CloudOutboxBlobCreated")]
+    public async Task CloudOutboxBlobCreated(
+        [QueueTrigger(BoteStorageConstants.OutboxCreatedQueue, Connection = "dbote:Worker:Storage:Connection")] string message)
+    {
+        if (!OutboxEvents.TryRead(message, out var tenantId, out var attachmentId))
+        {
+            logger.LogWarning("An outbox event that names no attachment of the outbox: {Message}", message);
+            return;
+        }
+
+        await CopyToClients(outbox.GetBlobContainerClient(BoteStorageConstants.Outbox)
+            .GetBlobClient($"{tenantId}/{attachmentId}"), tenantId, attachmentId);
+    }
+
+    /// <summary>
+    /// The same for a storage without Event Grid, as Azurite is in development. Turned off in Azure
+    /// (AzureWebJobs.CloudOutboxBlobDetected.Disabled; the form with __ does not turn it off), where both would copy every attachment.
+    /// </summary>
+    [PublicAPI]
     [Function("CloudOutboxBlobDetected")]
-    public async Task CloudOutboxBlobDetected(
+    public Task CloudOutboxBlobDetected(
         // The outbox lives in the gateway storage next to the inbox, not in the Functions host
         // storage (AzureWebJobsStorage), so the cloud only needs access to one account.
         [BlobTrigger("bote-outbox/{tenantId}/{attachmentId}", Connection = "dbote:Worker:Storage:Connection")] BlobClient sourceBlob,
         string tenantId,
-        string attachmentId)
+        string attachmentId) =>
+        CopyToClients(sourceBlob, tenantId, attachmentId);
+
+    private async Task CopyToClients(BlobClient sourceBlob, string tenantId, string attachmentId)
     {
         logger.LogInformation("Cloud outbox blob detected: {TenantId}/{AttachmentId}", tenantId, attachmentId);
 
@@ -78,8 +105,6 @@ public async Task CloudOutboxBlobDetected(
         };
 
         await copyProcessor.TryProcessCopyAsync(copyRequest);
-
-
     }
 
     /// <summary>

src/workers/src/Dbosoft.Bote.BoteWorker/OutboxEvents.cs

@@ -0,0 +1,45 @@
+using System.Text.Json;
+using Dbosoft.Bote.Primitives;
+
+namespace Dbosoft.Bote.BoteWorker;
+
+/// <summary>
+/// The Event Grid events (Event Grid schema) the gateway storage sends to
+/// <see cref="BoteStorageConstants.OutboxCreatedQueue"/> for the blobs written to the outbox.
+/// </summary>
+internal static class OutboxEvents
+{
+    private const string BlobCreated = "Microsoft.Storage.BlobCreated";
+    private const string OutboxBlobs = "/blobServices/default/containers/" + BoteStorageConstants.Outbox + "/blobs/";
+
+    /// <summary>
+    /// The tenant and attachment of a blob created in the outbox; false for any other event, and
+    /// for a blob whose name is not <c>{tenant}/{attachment}</c>.
+    /// </summary>
+    public static bool TryRead(string message, out string tenantId, out string attachmentId)
+    {
+        tenantId = attachmentId = "";
+        string? type, subject;
+        try
+        {
+            using var document = JsonDocument.Parse(message);
+            var root = document.RootElement;
+            if (root.ValueKind != JsonValueKind.Object)
+                return false;
+            type = root.TryGetProperty("eventType", out var t) && t.ValueKind == JsonValueKind.String ? t.GetString() : null;
+            subject = root.TryGetProperty("subject", out var s) && s.ValueKind == JsonValueKind.String ? s.GetString() : null;
+        }
+        catch (JsonException)
+        {
+            return false;
+        }
+
+        if (type != BlobCreated || subject is null || !subject.StartsWith(OutboxBlobs, StringComparison.Ordinal))
+            return false;
+        var parts = subject[OutboxBlobs.Length..].Split('/');
+        if (parts.Length != 2 || !BoteIds.IsPathSegment(parts[0]) || !BoteAttachmentIds.IsValid(parts[1]))
+            return false;
+        (tenantId, attachmentId) = (parts[0], parts[1]);
+        return true;
+    }
+}

test/Dbosoft.Bote.BoteWorker.Tests/OutboxEventsTests.cs

@@ -0,0 +1,46 @@
+using AwesomeAssertions;
+
+namespace Dbosoft.Bote.BoteWorker.Tests;
+
+public class OutboxEventsTests
+{
+    // As Event Grid wrote it to the queue on staging, for an attachment the cloud put into the outbox.
+    private const string Created = """
+        {"topic":"/subscriptions/s/resourceGroups/g/providers/Microsoft.Storage/storageAccounts/a",
+         "subject":"/blobServices/default/containers/bote-outbox/blobs/6f086007-9250-4547-8c83-dcbefaa92f2a/00ce1a07-c818-4799-87cb-bcf0af38f38f",
+         "eventType":"Microsoft.Storage.BlobCreated","id":"09e21a04-001e-00a9-1167-64ff8206c8da",
+         "data":{"api":"PutBlob","blobType":"BlockBlob","url":"https://a.blob.core.windows.net/bote-outbox/x/y"},
+         "dataVersion":"","metadataVersion":"1","eventTime":"2026-10-25T09:59:43.9381117Z"}
+        """;
+
+    [Fact]
+    public void Reads_the_tenant_and_attachment_of_a_blob_created_in_the_outbox()
+    {
+        OutboxEvents.TryRead(Created, out var tenantId, out var attachmentId).Should().BeTrue();
+
+        tenantId.Should().Be("6f086007-9250-4547-8c83-dcbefaa92f2a");
+        attachmentId.Should().Be("00ce1a07-c818-4799-87cb-bcf0af38f38f");
+    }
+
+    [Theory]
+    [InlineData("Microsoft.Storage.BlobDeleted", "/blobServices/default/containers/bote-outbox/blobs/t/a")]
+    [InlineData("Microsoft.Storage.BlobCreated", "/blobServices/default/containers/bote-inbox/blobs/t/c/a")]
+    [InlineData("Microsoft.Storage.BlobCreated", "/blobServices/default/containers/bote-outbox/blobs/t")]
+    [InlineData("Microsoft.Storage.BlobCreated", "/blobServices/default/containers/bote-outbox/blobs/t/a/b")]
+    [InlineData("Microsoft.Storage.BlobCreated", "/blobServices/default/containers/bote-outbox/blobs/t/..")]
+    [InlineData("Microsoft.Storage.BlobCreated", "/blobServices/default/containers/bote-outbox/blobs//a")]
+    public void Takes_no_other_event(string type, string subject)
+    {
+        var message = $$"""{"eventType":"{{type}}","subject":"{{subject}}"}""";
+
+        OutboxEvents.TryRead(message, out _, out _).Should().BeFalse();
+    }
+
+    [Theory]
+    [InlineData("")]
+    [InlineData("not json")]
+    [InlineData("[]")]
+    [InlineData("""{"eventType":1,"subject":2}""")]
+    public void Takes_no_message_that_is_no_event(string message) =>
+        OutboxEvents.TryRead(message, out _, out _).Should().BeFalse();
+}