Ticket #797 : Store the scim message in blob storage

src/Scim/SimpleIdServer.Scim.Startup/Configurations/MassTransitStorageConfiguration.cs

@@ -0,0 +1,19 @@
+// Copyright (c) SimpleIdServer. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See LICENSE in the project root for license information.
+using MassTransit;
+
+namespace SimpleIdServer.Scim.Startup.Configurations;
+
+public class MassTransitStorageConfiguration
+{
+    public bool IsEnabled { get; set; }
+    public MassTransitStorageTypes Type { get; set; }
+    public string ConnectionString { get; set; }
+}
+
+public enum MassTransitStorageTypes
+{
+    INMEMORY = 0,
+    DIRECTORY = 1,
+    AZURESTORAGE = 2
+}

src/Scim/SimpleIdServer.Scim.Startup/Consumers/BigMessageConsumer.cs

@@ -0,0 +1,33 @@
+using MassTransit;
+using Newtonsoft.Json;
+using SimpleIdServer.Scim.ExternalEvents;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading.Tasks;
+
+namespace SimpleIdServer.Scim.Startup.Consumers
+{
+    public class BigMessageConsumer : IConsumer<BigMessage>
+    {
+        private static List<Type> _types = new List<Type>
+        {
+            typeof(RepresentationAddedEvent),
+            typeof(RepresentationRemovedEvent),
+            typeof(RepresentationUpdatedEvent),
+            typeof(RepresentationRefAttributeAddedEvent),
+            typeof(RepresentationRefAttributeRemovedEvent),
+            typeof(RepresentationRefAttributeUpdatedEvent)
+        };
+
+
+        public async Task Consume(ConsumeContext<BigMessage> context)
+        {
+            var bigPayload = await context.Message.Payload.Value;
+            var type = _types.Single(t => t.Name == context.Message.Name);
+            var message = JsonConvert.DeserializeObject(Encoding.UTF8.GetString(bigPayload), type);
+            // CONTINUE...
+        }
+    }
+}

src/Scim/SimpleIdServer.Scim.Startup/Program.cs

@@ -1,7 +1,10 @@
 // Copyright (c) SimpleIdServer. All rights reserved.
 // Licensed under the Apache License, Version 2.0. See LICENSE in the project root for license information.
 using AspNetCore.Authentication.ApiKey;
+using Azure.Storage.Blobs;
 using MassTransit;
+using MassTransit.AzureStorage.MessageData;
+using MassTransit.MessageData;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Mvc;
 using Microsoft.AspNetCore.Routing;
@@ -12,6 +15,7 @@
 using Microsoft.Extensions.Options;
 using MongoDB.Driver;
 using SimpleIdServer.Scim.Domains;
+using SimpleIdServer.Scim.ExternalEvents;
 using SimpleIdServer.Scim.Infrastructure;
 using SimpleIdServer.Scim.Persistence.MongoDB.Extensions;
 using SimpleIdServer.Scim.Persistence.MongoDB.Infrastructures;
@@ -110,15 +114,25 @@ private static void ConfigureSwagger(WebApplicationBuilder builder)
 
     private static void ConfigureScim(WebApplicationBuilder builder)
     {
+        var massTransitSection = builder.Configuration.GetSection(nameof(MassTransitStorageConfiguration));
+        var massTransitConf = massTransitSection.Get<MassTransitStorageConfiguration>();
+        var repository = ConfigureMassTransitConfiguration(builder.Services, massTransitConf);
         builder.Services.AddSIDScim(_ =>
         {
             _.IgnoreUnsupportedCanonicalValues = false;
             _.EnableRealm = bool.Parse(builder.Configuration["IsRealmEnabled"]);
+            _.IsBigMessagePublished = massTransitConf.IsEnabled;
         }, massTransitOptions: _ =>
         {
-            _.AddConsumer<IntegrationEventConsumer>();
+            if(repository == null) _.AddConsumer<IntegrationEventConsumer>();
+            else _.AddConsumer<BigMessageConsumer>();
             _.UsingInMemory((context, cfg) =>
             {
+                if (repository != null)
+                {
+                    cfg.UseMessageData(repository);
+                }
+
                 cfg.ConfigureEndpoints(context);
             });
         });
@@ -170,6 +184,28 @@ void ConfigureEFStorage(StorageConfiguration conf)
             }, supportSqlite: conf.Type == StorageTypes.SQLITE);
         }
 
+        IMessageDataRepository ConfigureMassTransitConfiguration(IServiceCollection services, MassTransitStorageConfiguration conf)
+        {
+            if (!conf.IsEnabled) return null;
+            IMessageDataRepository repository = null;
+            switch (conf.Type)
+            {
+                case MassTransitStorageTypes.INMEMORY:
+                    repository = new InMemoryMessageDataRepository();
+                    break;
+                case MassTransitStorageTypes.DIRECTORY:
+                    repository = new FileSystemMessageDataRepository(new DirectoryInfo(AppDomain.CurrentDomain.BaseDirectory));
+                    break;
+                case MassTransitStorageTypes.AZURESTORAGE:
+                    var client = new BlobServiceClient(conf.ConnectionString);
+                    repository = client.CreateMessageDataRepository("message-data");
+                    break;
+            }
+
+            services.AddSingleton(repository);
+            return repository;
+        }
+
         void ConfigureMongoDbStorage(StorageConfiguration conf)
         {
             var basePath = Path.Combine(builder.Environment.ContentRootPath, "Schemas");

src/Scim/SimpleIdServer.Scim.Startup/SimpleIdServer.Scim.Startup.csproj

@@ -13,6 +13,7 @@
 		<PackageReference Include="Microsoft.AspNetCore.Mvc.NewtonsoftJson" Version="8.0.4" />
 		<PackageReference Include="Swashbuckle.AspNetCore" Version="6.3.0 " />
 		<PackageReference Include="AspNetCore.Authentication.ApiKey" Version="8.0.1" />
+		<PackageReference Include="MassTransit.Azure.Storage" Version="8.2.1" />
 	</ItemGroup>
 	<ItemGroup>
 		<ProjectReference Include="..\SimpleIdServer.Scim.SqliteMigrations\SimpleIdServer.Scim.SqliteMigrations.csproj" />

src/Scim/SimpleIdServer.Scim.Startup/appsettings.Docker.json

@@ -25,5 +25,9 @@
     "ConnectionString": "Data Source=localhost;Initial Catalog=Scim;TrustServerCertificate=True;User Id=sa;Password=a7Nyz2WWcmhqs4nq9a7Z;",
     "Type": "SQLSERVER"
   },
-  "IsRealmEnabled": false
+  "IsRealmEnabled": false,
+  "MassTransitStorageConfiguration": {
+    "IsEnabled": false,
+    "Type": "INMEMORY"
+  }
 }

src/Scim/SimpleIdServer.Scim.Startup/appsettings.json

@@ -25,5 +25,9 @@
     "ConnectionString": "Data Source=.;Initial Catalog=SCIM;Integrated Security=True;TrustServerCertificate=True",
     "Type": "SQLSERVER"
   },
-  "IsRealmEnabled": false
+  "IsRealmEnabled": false,
+  "MassTransitStorageConfiguration": {
+    "IsEnabled": true,
+    "Type": "DIRECTORY"
+  }
 }

src/Scim/SimpleIdServer.Scim/Api/BaseApiController.cs

@@ -44,7 +44,7 @@ public abstract class BaseApiController : Controller
         private readonly IAttributeReferenceEnricher _attributeReferenceEnricher;
         private readonly SCIMHostOptions _options;
         private readonly ILogger _logger;
-        private readonly IBusControl _busControl;
+        private readonly IBusHelper _busControl;
         private readonly IResourceTypeResolver _resourceTypeResolver;
         private readonly IUriProvider _uriProvider;
         private readonly IRealmRepository _realmRepository;
@@ -58,8 +58,8 @@ public BaseApiController(string resourceType,
             IGetRepresentationQueryHandler getRepresentationQueryHandler, 
             IAttributeReferenceEnricher attributeReferenceEnricher, 
             IOptionsMonitor<SCIMHostOptions> options, 
-            ILogger logger, 
-            IBusControl busControl, 
+            ILogger logger,
+            IBusHelper busControl, 
             IResourceTypeResolver resourceTypeResolver, 
             IUriProvider uriProvider,
             IRealmRepository realmRepository)
@@ -357,7 +357,15 @@ protected async Task<IActionResult> InternalAdd(string prefix, RepresentationPar
                 representation.ApplyEmptyArray();
                 var location = GetLocation(representation);
                 var content = representation.ToResponse(location, true, mergeExtensionAttributes: _options.MergeExtensionAttributes);
-                if (IsPublishEvtsEnabled) await _busControl.Publish(new RepresentationAddedEvent(representation.Id, representation.Version, GetResourceType(_resourceType), content, _options.IncludeToken ? Request.GetToken() : string.Empty));
+                if (IsPublishEvtsEnabled)
+                {
+                    var message = new BigMessage
+                    {
+
+                    };
+                    await _busControl.Publish(new RepresentationAddedEvent(representation.Id, representation.Version, GetResourceType(_resourceType), content, _options.IncludeToken ? Request.GetToken() : string.Empty));
+                }
+
                 return BuildHTTPResult(HttpStatusCode.Created, location, representation.Version, content);
             }
             catch (SCIMSchemaViolatedException ex)

src/Scim/SimpleIdServer.Scim/Api/GroupsController.cs

@@ -14,7 +14,7 @@ namespace SimpleIdServer.Scim.Api
 {
     public class GroupsController : BaseApiController
     {
-        public GroupsController(IAddRepresentationCommandHandler addRepresentationCommandHandler, IDeleteRepresentationCommandHandler deleteRepresentationCommandHandler, IReplaceRepresentationCommandHandler replaceRepresentationCommandHandler, IPatchRepresentationCommandHandler patchRepresentationCommandHandler, ISearchRepresentationsQueryHandler searchRepresentationsQueryHandler, IGetRepresentationQueryHandler getRepresentationQueryHandler, IAttributeReferenceEnricher attributeReferenceEnricher, IOptionsMonitor<SCIMHostOptions> options, ILogger<GroupsController> logger, IBusControl busControl, IResourceTypeResolver resourceTypeResolver, IUriProvider uriProvider, IRealmRepository realmRepository) : base(SCIMResourceTypes.Group, addRepresentationCommandHandler, deleteRepresentationCommandHandler, replaceRepresentationCommandHandler, patchRepresentationCommandHandler, searchRepresentationsQueryHandler, getRepresentationQueryHandler, attributeReferenceEnricher, options, logger, busControl, resourceTypeResolver, uriProvider, realmRepository)
+        public GroupsController(IAddRepresentationCommandHandler addRepresentationCommandHandler, IDeleteRepresentationCommandHandler deleteRepresentationCommandHandler, IReplaceRepresentationCommandHandler replaceRepresentationCommandHandler, IPatchRepresentationCommandHandler patchRepresentationCommandHandler, ISearchRepresentationsQueryHandler searchRepresentationsQueryHandler, IGetRepresentationQueryHandler getRepresentationQueryHandler, IAttributeReferenceEnricher attributeReferenceEnricher, IOptionsMonitor<SCIMHostOptions> options, ILogger<GroupsController> logger, IBusHelper busControl, IResourceTypeResolver resourceTypeResolver, IUriProvider uriProvider, IRealmRepository realmRepository) : base(SCIMResourceTypes.Group, addRepresentationCommandHandler, deleteRepresentationCommandHandler, replaceRepresentationCommandHandler, patchRepresentationCommandHandler, searchRepresentationsQueryHandler, getRepresentationQueryHandler, attributeReferenceEnricher, options, logger, busControl, resourceTypeResolver, uriProvider, realmRepository)
         {
         }
 

src/Scim/SimpleIdServer.Scim/Api/UsersController.cs

@@ -13,7 +13,7 @@ namespace SimpleIdServer.Scim.Api
 {
     public class UsersController : BaseApiController
     {
-        public UsersController(IAddRepresentationCommandHandler addRepresentationCommandHandler, IDeleteRepresentationCommandHandler deleteRepresentationCommandHandler, IReplaceRepresentationCommandHandler replaceRepresentationCommandHandler, IPatchRepresentationCommandHandler patchRepresentationCommandHandler, ISearchRepresentationsQueryHandler searchRepresentationsQueryHandler, IGetRepresentationQueryHandler getRepresentationQueryHandler, IAttributeReferenceEnricher attributeReferenceEnricher, IOptionsMonitor<SCIMHostOptions> options, ILogger<UsersController> logger, IBusControl busControl, IResourceTypeResolver resourceTypeResolver, IUriProvider uriProvider, IRealmRepository realmRepository) : base(SCIMResourceTypes.User, addRepresentationCommandHandler, deleteRepresentationCommandHandler, replaceRepresentationCommandHandler, patchRepresentationCommandHandler, searchRepresentationsQueryHandler, getRepresentationQueryHandler, attributeReferenceEnricher, options, logger, busControl, resourceTypeResolver, uriProvider, realmRepository)
+        public UsersController(IAddRepresentationCommandHandler addRepresentationCommandHandler, IDeleteRepresentationCommandHandler deleteRepresentationCommandHandler, IReplaceRepresentationCommandHandler replaceRepresentationCommandHandler, IPatchRepresentationCommandHandler patchRepresentationCommandHandler, ISearchRepresentationsQueryHandler searchRepresentationsQueryHandler, IGetRepresentationQueryHandler getRepresentationQueryHandler, IAttributeReferenceEnricher attributeReferenceEnricher, IOptionsMonitor<SCIMHostOptions> options, ILogger<UsersController> logger, IBusHelper busControl, IResourceTypeResolver resourceTypeResolver, IUriProvider uriProvider, IRealmRepository realmRepository) : base(SCIMResourceTypes.User, addRepresentationCommandHandler, deleteRepresentationCommandHandler, replaceRepresentationCommandHandler, patchRepresentationCommandHandler, searchRepresentationsQueryHandler, getRepresentationQueryHandler, attributeReferenceEnricher, options, logger, busControl, resourceTypeResolver, uriProvider, realmRepository)
         {
         }
 

src/Scim/SimpleIdServer.Scim/Commands/Handlers/AddRepresentationCommandHandler.cs

@@ -28,7 +28,7 @@ public AddRepresentationCommandHandler(
             ISCIMRepresentationCommandRepository scimRepresentationCommandRepository,
             IRepresentationReferenceSync representationReferenceSync,
             IRepresentationHelper representationHelper,
-            IBusControl busControl) : base(busControl)
+            IBusHelper busControl) : base(busControl)
         {
             _scimSchemaCommandRepository = scimSchemaCommandRepository;
             _scimRepresentationCommandRepository = scimRepresentationCommandRepository;

src/Scim/SimpleIdServer.Scim/Commands/Handlers/BaseCommandHandler.cs

@@ -12,9 +12,9 @@ namespace SimpleIdServer.Scim.Commands.Handlers
 {
     public class BaseCommandHandler
     {
-        private readonly IBusControl _busControl;
+        private readonly IBusHelper _busControl;
 
-        public BaseCommandHandler(IBusControl busControl)
+        public BaseCommandHandler(IBusHelper busControl)
         {
             _busControl = busControl;
         }

src/Scim/SimpleIdServer.Scim/Commands/Handlers/DeleteRepresentationCommandHandler.cs

@@ -23,7 +23,7 @@ public class DeleteRepresentationCommandHandler : BaseCommandHandler, IDeleteRep
         public DeleteRepresentationCommandHandler(ISCIMSchemaCommandRepository scimSchemaCommandRepository,
             ISCIMRepresentationCommandRepository scimRepresentationCommandRepository,
             IRepresentationReferenceSync representationReferenceSync,
-            IBusControl busControl) : base(busControl)
+            IBusHelper busControl) : base(busControl)
         {
             _scimSchemaCommandRepository = scimSchemaCommandRepository;
             _scimRepresentationCommandRepository = scimRepresentationCommandRepository;

src/Scim/SimpleIdServer.Scim/Commands/Handlers/PatchRepresentationCommandHandler.cs

@@ -34,7 +34,7 @@ public PatchRepresentationCommandHandler(
             IRepresentationReferenceSync representationReferenceSync,
             IRepresentationHelper representationHelper,
             IOptions<SCIMHostOptions> options,
-            IBusControl busControl) : base(busControl)
+            IBusHelper busControl) : base(busControl)
         {
             _scimSchemaCommandRepository = scimSchemaCommandRepository;
             _scimAttributeMappingQueryRepository = scimAttributeMappingQueryRepository;

src/Scim/SimpleIdServer.Scim/Commands/Handlers/ReplaceRepresentationCommandHandler.cs

@@ -30,7 +30,7 @@ public ReplaceRepresentationCommandHandler(
             ISCIMRepresentationCommandRepository scimRepresentationCommandRepository,
             IRepresentationReferenceSync representationReferenceSync,
             IRepresentationHelper representationHelper,
-            IBusControl busControl) : base(busControl)
+            IBusHelper busControl) : base(busControl)
         {
             _scimAttributeMappingQueryRepository = scimAttributeMappingQueryRepository;
             _scimSchemaCommandRepository = scimSchemaCommandRepository;

src/Scim/SimpleIdServer.Scim/ExternalEvents/BigMessage.cs

@@ -0,0 +1,12 @@
+// Copyright (c) SimpleIdServer. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See LICENSE in the project root for license information.
+using MassTransit;
+
+namespace SimpleIdServer.Scim.ExternalEvents
+{
+    public class BigMessage
+    {
+        public string Name { get; set; }
+        public MessageData<byte[]> Payload { get; set; }
+    }
+}

src/Scim/SimpleIdServer.Scim/Helpers/BusHelper.cs

@@ -0,0 +1,59 @@
+// Copyright (c) SimpleIdServer. All rights reserved.
+// Licensed under the Apache License, Version 2.0. See LICENSE in the project root for license information.
+
+using MassTransit;
+using MassTransit.MessageData;
+using Microsoft.Extensions.Options;
+using Newtonsoft.Json;
+using SimpleIdServer.Scim.ExternalEvents;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace SimpleIdServer.Scim.Helpers
+{
+    public interface IBusHelper
+    {
+        Task Publish<T>(T evt, CancellationToken cancellationToken = default) where T : IntegrationEvent;
+    }
+
+    public class BusHelper : IBusHelper
+    {
+        private readonly SCIMHostOptions _options;
+        private readonly IBusControl _busControl;
+        private readonly IMessageDataRepository _messageDataRepository;
+
+        public BusHelper(
+            IOptionsMonitor<SCIMHostOptions> options, 
+            IBusControl busControl, 
+            IMessageDataRepository messageDataRepository)
+        {
+            _options = options.CurrentValue;
+            _busControl = busControl;
+            _messageDataRepository = messageDataRepository;
+        }
+
+        public async Task Publish<T>(T evt, CancellationToken cancellationToken = default) where T : IntegrationEvent
+        {
+            if (!_options.IsBigMessagePublished)
+            {
+                await _busControl.Publish(evt);
+                return;
+            }
+
+            var bigPayload = await _messageDataRepository.PutBytes(Serialize(evt), cancellationToken);
+            var message = new BigMessage
+            {
+                Name = typeof(T).Name,
+                Payload = bigPayload
+            };
+            await _busControl.Publish(message, cancellationToken);
+        }
+
+        public static byte[] Serialize(object obj)
+        {
+            var json = JsonConvert.SerializeObject(obj);
+            return Encoding.UTF8.GetBytes(json);
+        }
+    }
+}

src/Scim/SimpleIdServer.Scim/SCIMHostOptions.cs

@@ -86,5 +86,6 @@ public SCIMHostOptions()
         /// </summary>
         public bool EnableRealm { get; set; } = false;
         public bool IsFullRepresentationReturned { get; set; }
+        public bool IsBigMessagePublished { get; set; }
     }
 }

src/Scim/SimpleIdServer.Scim/ServiceCollectionExtensions.cs

@@ -99,6 +99,7 @@ private static IServiceCollection AddHelpers(this IServiceCollection services)
             services.AddTransient<IRepresentationHelper, RepresentationHelper>();
             services.AddHttpContextAccessor();
             services.AddTransient<IUriProvider, UriProvider>();
+            services.AddTransient<IBusHelper, BusHelper>();
             return services;
         }