write events to table storage

src/Core/Repositories/TableStorage/EventRepository.cs

@@ -12,8 +12,12 @@ namespace Bit.Core.Repositories.TableStorage
     public class EventRepository : IEventRepository
     {
         public EventRepository(GlobalSettings globalSettings)
+            : this(globalSettings.Storage.ConnectionString)
+        { }
+
+        public EventRepository(string storageConnectionString)
         {
-            var storageAccount = CloudStorageAccount.Parse(globalSettings.Storage.ConnectionString);
+            var storageAccount = CloudStorageAccount.Parse(storageConnectionString);
             var tableClient = storageAccount.CreateCloudTableClient();
             Table = tableClient.GetTableReference("event");
         }

src/Core/Services/Implementations/RepositoryEventWriteService.cs

@@ -8,14 +8,11 @@ namespace Bit.Core.Services
     public class RepositoryEventWriteService : IEventWriteService
     {
         private readonly IEventRepository _eventRepository;
-        private readonly GlobalSettings _globalSettings;
 
         public RepositoryEventWriteService(
-            IEventRepository eventRepository,
+            IEventRepository eventRepository)
-            GlobalSettings globalSettings)
         {
             _eventRepository = eventRepository;
-            _globalSettings = globalSettings;
         }
 
         public async Task CreateAsync(EventTableEntity entity)

src/Core/Utilities/ServiceCollectionExtensions.cs

@@ -20,8 +20,8 @@ using Microsoft.WindowsAzure.Storage;
 using System;
 using System.IO;
 using SqlServerRepos = Bit.Core.Repositories.SqlServer;
-using TableStorageRepos = Bit.Core.Repositories.TableStorage;
 using System.Threading.Tasks;
+using TableStorageRepos = Bit.Core.Repositories.TableStorage;
 
 namespace Bit.Core.Utilities
 {
@@ -44,7 +44,7 @@ namespace Bit.Core.Utilities
 
             if(globalSettings.SelfHosted)
             {
-                // TODO: Sql server repo
+                // TODO: Sql server event repo
             }
             else
             {
@@ -59,7 +59,7 @@ namespace Bit.Core.Utilities
             services.AddScoped<IOrganizationService, OrganizationService>();
             services.AddScoped<ICollectionService, CollectionService>();
             services.AddScoped<IGroupService, GroupService>();
-            services.AddScoped<Services.IEventService, NoopEventService>();
+            services.AddScoped<Services.IEventService, EventService>();
             services.AddSingleton<IDeviceService, DeviceService>();
         }
 
@@ -105,15 +105,24 @@ namespace Bit.Core.Utilities
             if(!globalSettings.SelfHosted && CoreHelpers.SettingHasValue(globalSettings.Storage.ConnectionString))
             {
                 services.AddSingleton<IBlockIpService, AzureQueueBlockIpService>();
-                //services.AddSingleton<IEventWriteService, AzureQueueEventWriteService>();
             }
             else
             {
                 services.AddSingleton<IBlockIpService, NoopBlockIpService>();
-                //services.AddSingleton<IEventWriteService, RepositoryEventWriteService>();
             }
 
+            if(!globalSettings.SelfHosted && CoreHelpers.SettingHasValue(globalSettings.Storage.ConnectionString))
+            {
+                services.AddSingleton<IEventWriteService, AzureQueueEventWriteService>();
+            }
+            else if(globalSettings.SelfHosted)
+            {
+                services.AddSingleton<IEventWriteService, RepositoryEventWriteService>();
+            }
+            else
+            {
                 services.AddSingleton<IEventWriteService, NoopEventWriteService>();
+            }
 
             if(CoreHelpers.SettingHasValue(globalSettings.Attachment.ConnectionString))
             {

src/EventsProcessor/EventsProcessor.csproj

@@ -13,6 +13,7 @@
     <FileAlignment>512</FileAlignment>
     <AutoGenerateBindingRedirects>true</AutoGenerateBindingRedirects>
     <TargetFrameworkProfile />
+    <IsWebBootstrapper>false</IsWebBootstrapper>
     <PublishUrl>publish\</PublishUrl>
     <Install>true</Install>
     <InstallFrom>Disk</InstallFrom>
@@ -25,7 +26,6 @@
     <MapFileExtensions>true</MapFileExtensions>
     <ApplicationRevision>0</ApplicationRevision>
     <ApplicationVersion>1.0.0.%2a</ApplicationVersion>
-    <IsWebBootstrapper>false</IsWebBootstrapper>
     <UseApplicationTrust>false</UseApplicationTrust>
     <BootstrapperEnabled>true</BootstrapperEnabled>
   </PropertyGroup>
@@ -78,6 +78,7 @@
       <HintPath>..\..\packages\Newtonsoft.Json.10.0.3\lib\net45\Newtonsoft.Json.dll</HintPath>
     </Reference>
     <Reference Include="System" />
+    <Reference Include="System.Configuration" />
     <Reference Include="System.Core" />
     <Reference Include="System.Spatial, Version=5.8.3.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
       <HintPath>..\..\packages\System.Spatial.5.8.3\lib\net40\System.Spatial.dll</HintPath>
@@ -111,6 +112,12 @@
       <Install>false</Install>
     </BootstrapperPackage>
   </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\Core\Core.csproj">
+      <Project>{3973d21b-a692-4b60-9b70-3631c057423a}</Project>
+      <Name>Core</Name>
+    </ProjectReference>
+  </ItemGroup>
   <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
   <Import Project="..\..\packages\Microsoft.Web.WebJobs.Publish.1.1.0\tools\webjobs.targets" Condition="Exists('..\..\packages\Microsoft.Web.WebJobs.Publish.1.1.0\tools\webjobs.targets')" />
   <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 

src/EventsProcessor/Functions.cs

@@ -1,17 +1,63 @@
 using System;
+using System.Collections.Generic;
+using System.Configuration;
 using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
+using Bit.Core.Models.Data;
+using Bit.Core.Services;
 using Microsoft.Azure.WebJobs;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Linq;
 
 namespace Bit.EventsProcessor
 {
     public class Functions
     {
-        public async static Task ProcessQueueMessageAsync(
+        private static IEventWriteService _eventWriteService;
-            [QueueTrigger("event")] string message, TextWriter logger, CancellationToken token)
+
+        static Functions()
         {
-            await logger.WriteLineAsync(message);
+            var storageConnectionString = ConfigurationManager.ConnectionStrings["AzureWebJobsStorage"];
+            if(storageConnectionString == null || string.IsNullOrWhiteSpace(storageConnectionString.ConnectionString))
+            {
+                return;
+            }
+
+            var repo = new Core.Repositories.TableStorage.EventRepository(storageConnectionString.ConnectionString);
+            _eventWriteService = new RepositoryEventWriteService(repo);
+        }
+
+        public async static Task ProcessQueueMessageAsync([QueueTrigger("event")] string message,
+            TextWriter logger, CancellationToken token)
+        {
+            if(_eventWriteService == null || message == null || message.Length == 0)
+            {
+                return;
+            }
+
+            try
+            {
+                var jToken = JToken.Parse(message);
+                if(jToken is JArray)
+                {
+                    var entities = jToken.ToObject<IList<EventTableEntity>>();
+                    await _eventWriteService.CreateManyAsync(entities);
+                }
+                else if(jToken is JObject)
+                {
+                    var entity = jToken.ToObject<EventTableEntity>();
+                    await _eventWriteService.CreateAsync(entity);
+                }
+            }
+            catch(JsonReaderException)
+            {
+                await logger.WriteLineAsync("JsonReaderException: Unable to parse message.");
+            }
+            catch(JsonSerializationException)
+            {
+                await logger.WriteLineAsync("JsonSerializationException: Unable to serialize token.");
+            }
         }
     }
 }