diff --git a/infra/core/dbote-gateway.ts b/infra/core/dbote-gateway.ts index 23bfe48..d410b62 100644 --- a/infra/core/dbote-gateway.ts +++ b/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); diff --git a/src/Dbosoft.Bote.Primitives/BoteStorageConstants.cs b/src/Dbosoft.Bote.Primitives/BoteStorageConstants.cs index e70d844..5ad948f 100644 --- a/src/Dbosoft.Bote.Primitives/BoteStorageConstants.cs +++ b/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"; + /// + /// The queue of the gateway storage Event Grid reports the blobs written to the outbox to. + /// + public const string OutboxCreatedQueue = "bote-outbox-created"; + /// /// Blob metadata keys used for attachment tracking and validation. /// diff --git a/src/workers/src/Dbosoft.Bote.BoteWorker/DataBusCopyFile.cs b/src/workers/src/Dbosoft.Bote.BoteWorker/DataBusCopyFile.cs index 53669dc..97872b3 100644 --- a/src/workers/src/Dbosoft.Bote.BoteWorker/DataBusCopyFile.cs +++ b/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 logger, DataBusCopyProcessor copyProcessor, - ITenantStorageResolver tenantStorageResolver) + ITenantStorageResolver tenantStorageResolver, + BlobServiceClient outbox) { /// - /// 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. /// + [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); + } + + /// + /// 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. + /// + [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); - - } /// diff --git a/src/workers/src/Dbosoft.Bote.BoteWorker/OutboxEvents.cs b/src/workers/src/Dbosoft.Bote.BoteWorker/OutboxEvents.cs new file mode 100644 index 0000000..a12d7ef --- /dev/null +++ b/src/workers/src/Dbosoft.Bote.BoteWorker/OutboxEvents.cs @@ -0,0 +1,45 @@ +using System.Text.Json; +using Dbosoft.Bote.Primitives; + +namespace Dbosoft.Bote.BoteWorker; + +/// +/// The Event Grid events (Event Grid schema) the gateway storage sends to +/// for the blobs written to the outbox. +/// +internal static class OutboxEvents +{ + private const string BlobCreated = "Microsoft.Storage.BlobCreated"; + private const string OutboxBlobs = "/blobServices/default/containers/" + BoteStorageConstants.Outbox + "/blobs/"; + + /// + /// The tenant and attachment of a blob created in the outbox; false for any other event, and + /// for a blob whose name is not {tenant}/{attachment}. + /// + 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; + } +} diff --git a/test/Dbosoft.Bote.BoteWorker.Tests/OutboxEventsTests.cs b/test/Dbosoft.Bote.BoteWorker.Tests/OutboxEventsTests.cs new file mode 100644 index 0000000..b5c52a6 --- /dev/null +++ b/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(); +}