azure queue notification service

src/Core/Models/PushNotification.cs

@@ -1,43 +1,20 @@
 using Bit.Core.Enums;
-using Newtonsoft.Json;
 using System;
 
 namespace Bit.Core.Models
 {
-    public class PayloadPushNotification
+    public class PushNotificationData<T>
     {
-        [JsonProperty(PropertyName = "data")]
+        public PushNotificationData(PushType type, T payload, string contextId)
-        public DataObj Data { get; set; }
-
-        public class DataObj
-        {
-            public DataObj(PushType type, string payload)
         {
             Type = type;
             Payload = payload;
+            ContextId = contextId;
         }
         
-            [JsonProperty(PropertyName = "type")]
         public PushType Type { get; set; }
-            [JsonProperty(PropertyName = "payload")]
+        public T Payload { get; set; }
-            public string Payload { get; set; }
+        public string ContextId { get; set; }
-        }
-    }
-
-    public class ApplePayloadPushNotification : PayloadPushNotification
-    {
-        [JsonProperty(PropertyName = "aps")]
-        public AppleData Aps { get; set; } = new AppleData { ContentAvailable = 1 };
-
-        public class AppleData
-        {
-            [JsonProperty(PropertyName = "badge")]
-            public dynamic Badge { get; set; } = null;
-            [JsonProperty(PropertyName = "alert")]
-            public string Alert { get; set; }
-            [JsonProperty(PropertyName = "content-available")]
-            public int ContentAvailable { get; set; }
-        }
     }
 
     public class SyncCipherPushNotification

src/Core/Services/Implementations/AzureQueuePushNotificationService.cs

@@ -0,0 +1,165 @@
+using System;
+using System.Threading.Tasks;
+using Bit.Core.Models.Table;
+using Bit.Core.Enums;
+using Newtonsoft.Json;
+using Bit.Core.Models;
+using Microsoft.WindowsAzure.Storage.Queue;
+using Microsoft.WindowsAzure.Storage;
+using Microsoft.AspNetCore.Http;
+
+namespace Bit.Core.Services
+{
+    public class AzureQueuePushNotificationService : IPushNotificationService
+    {
+        private readonly CloudQueue _queue;
+        private readonly GlobalSettings _globalSettings;
+        private readonly IHttpContextAccessor _httpContextAccessor;
+
+        private JsonSerializerSettings _jsonSettings = new JsonSerializerSettings
+        {
+            NullValueHandling = NullValueHandling.Ignore
+        };
+
+        public AzureQueuePushNotificationService(
+            GlobalSettings globalSettings,
+            IHttpContextAccessor httpContextAccessor)
+        {
+            var storageAccount = CloudStorageAccount.Parse(globalSettings.Events.ConnectionString);
+            var queueClient = storageAccount.CreateCloudQueueClient();
+            _queue = queueClient.GetQueueReference("sync");
+            _globalSettings = globalSettings;
+            _httpContextAccessor = httpContextAccessor;
+        }
+
+        public async Task PushSyncCipherCreateAsync(Cipher cipher)
+        {
+            await PushCipherAsync(cipher, PushType.SyncCipherCreate);
+        }
+
+        public async Task PushSyncCipherUpdateAsync(Cipher cipher)
+        {
+            await PushCipherAsync(cipher, PushType.SyncCipherUpdate);
+        }
+
+        public async Task PushSyncCipherDeleteAsync(Cipher cipher)
+        {
+            await PushCipherAsync(cipher, PushType.SyncLoginDelete);
+        }
+
+        private async Task PushCipherAsync(Cipher cipher, PushType type)
+        {
+            if(cipher.OrganizationId.HasValue)
+            {
+                var message = new SyncCipherPushNotification
+                {
+                    Id = cipher.Id,
+                    OrganizationId = cipher.OrganizationId,
+                    RevisionDate = cipher.RevisionDate,
+                };
+
+                await SendMessageAsync(type, message, true);
+            }
+            else if(cipher.UserId.HasValue)
+            {
+                var message = new SyncCipherPushNotification
+                {
+                    Id = cipher.Id,
+                    UserId = cipher.UserId,
+                    RevisionDate = cipher.RevisionDate,
+                };
+
+                await SendMessageAsync(type, message, true);
+            }
+        }
+
+        public async Task PushSyncFolderCreateAsync(Folder folder)
+        {
+            await PushFolderAsync(folder, PushType.SyncFolderCreate);
+        }
+
+        public async Task PushSyncFolderUpdateAsync(Folder folder)
+        {
+            await PushFolderAsync(folder, PushType.SyncFolderUpdate);
+        }
+
+        public async Task PushSyncFolderDeleteAsync(Folder folder)
+        {
+            await PushFolderAsync(folder, PushType.SyncFolderDelete);
+        }
+
+        private async Task PushFolderAsync(Folder folder, PushType type)
+        {
+            var message = new SyncFolderPushNotification
+            {
+                Id = folder.Id,
+                UserId = folder.UserId,
+                RevisionDate = folder.RevisionDate
+            };
+
+            await SendMessageAsync(type, message, true);
+        }
+
+        public async Task PushSyncCiphersAsync(Guid userId)
+        {
+            await PushSyncUserAsync(userId, PushType.SyncCiphers);
+        }
+
+        public async Task PushSyncVaultAsync(Guid userId)
+        {
+            await PushSyncUserAsync(userId, PushType.SyncVault);
+        }
+
+        public async Task PushSyncOrgKeysAsync(Guid userId)
+        {
+            await PushSyncUserAsync(userId, PushType.SyncOrgKeys);
+        }
+
+        public async Task PushSyncSettingsAsync(Guid userId)
+        {
+            await PushSyncUserAsync(userId, PushType.SyncSettings);
+        }
+
+        private async Task PushSyncUserAsync(Guid userId, PushType type)
+        {
+            var message = new SyncUserPushNotification
+            {
+                UserId = userId,
+                Date = DateTime.UtcNow
+            };
+
+            await SendMessageAsync(type, message, false);
+        }
+
+        private async Task SendMessageAsync<T>(PushType type, T payload, bool excludeCurrentContext)
+        {
+            var contextId = GetContextIdentifier(excludeCurrentContext);
+            var message = JsonConvert.SerializeObject(new PushNotificationData<T>(type, payload, contextId), 
+                _jsonSettings);
+            var queueMessage = new CloudQueueMessage(message);
+            await _queue.AddMessageAsync(queueMessage);
+        }
+
+        private string GetContextIdentifier(bool excludeCurrentContext)
+        {
+            if(!excludeCurrentContext)
+            {
+                return null;
+            }
+
+            var currentContext = _httpContextAccessor?.HttpContext?.
+                RequestServices.GetService(typeof(CurrentContext)) as CurrentContext;
+            return currentContext?.DeviceIdentifier;
+        }
+
+        public Task SendPayloadToUserAsync(string userId, PushType type, object payload, string identifier)
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task SendPayloadToOrganizationAsync(string orgId, PushType type, object payload, string identifier)
+        {
+            throw new NotImplementedException();
+        }
+    }
+}

src/Core/Services/Implementations/MultiServicePushNotificationService.cs

@@ -0,0 +1,121 @@
+using System;
+using System.Threading.Tasks;
+using Bit.Core.Models.Table;
+using Bit.Core.Enums;
+using System.Collections.Generic;
+using Microsoft.AspNetCore.Http;
+using Bit.Core.Utilities;
+using Microsoft.Extensions.Logging;
+
+namespace Bit.Core.Services
+{
+    public class MultiServicePushNotificationService : IPushNotificationService
+    {
+        private readonly List<IPushNotificationService> _services = new List<IPushNotificationService>();
+
+        public MultiServicePushNotificationService(
+            GlobalSettings globalSettings,
+            IHttpContextAccessor httpContextAccessor,
+            ILogger<RelayPushNotificationService> relayLogger)
+        {
+            if(globalSettings.SelfHosted)
+            {
+                if(CoreHelpers.SettingHasValue(globalSettings.PushRelayBaseUri) &&
+                    globalSettings.Installation?.Id != null &&
+                    CoreHelpers.SettingHasValue(globalSettings.Installation?.Key))
+                {
+                    _services.Add(new RelayPushNotificationService(globalSettings, httpContextAccessor, relayLogger));
+                }
+                // TODO: ApiPushNotificationService for SignalR
+            }
+            else
+            {
+#if NET471
+                _services.Add(new NotificationHubPushNotificationService(globalSettings, httpContextAccessor));
+#endif
+                _services.Add(new AzureQueuePushNotificationService(globalSettings, httpContextAccessor));
+            }
+        }
+
+        public Task PushSyncCipherCreateAsync(Cipher cipher)
+        {
+            PushToServices((s) => s.PushSyncCipherCreateAsync(cipher));
+            return Task.FromResult(0);
+        }
+
+        public Task PushSyncCipherUpdateAsync(Cipher cipher)
+        {
+            PushToServices((s) => s.PushSyncCipherUpdateAsync(cipher));
+            return Task.FromResult(0);
+        }
+
+        public Task PushSyncCipherDeleteAsync(Cipher cipher)
+        {
+            PushToServices((s) => s.PushSyncCipherDeleteAsync(cipher));
+            return Task.FromResult(0);
+        }
+
+        public Task PushSyncFolderCreateAsync(Folder folder)
+        {
+            PushToServices((s) => s.PushSyncFolderCreateAsync(folder));
+            return Task.FromResult(0);
+        }
+
+        public Task PushSyncFolderUpdateAsync(Folder folder)
+        {
+            PushToServices((s) => s.PushSyncFolderUpdateAsync(folder));
+            return Task.FromResult(0);
+        }
+
+        public Task PushSyncFolderDeleteAsync(Folder folder)
+        {
+            PushToServices((s) => s.PushSyncFolderDeleteAsync(folder));
+            return Task.FromResult(0);
+        }
+
+        public Task PushSyncCiphersAsync(Guid userId)
+        {
+            PushToServices((s) => s.PushSyncCiphersAsync(userId));
+            return Task.FromResult(0);
+        }
+
+        public Task PushSyncVaultAsync(Guid userId)
+        {
+            PushToServices((s) => s.PushSyncVaultAsync(userId));
+            return Task.FromResult(0);
+        }
+
+        public Task PushSyncOrgKeysAsync(Guid userId)
+        {
+            PushToServices((s) => s.PushSyncOrgKeysAsync(userId));
+            return Task.FromResult(0);
+        }
+
+        public Task PushSyncSettingsAsync(Guid userId)
+        {
+            PushToServices((s) => s.PushSyncSettingsAsync(userId));
+            return Task.FromResult(0);
+        }
+
+        public Task SendPayloadToUserAsync(string userId, PushType type, object payload, string identifier)
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task SendPayloadToOrganizationAsync(string orgId, PushType type, object payload, string identifier)
+        {
+            throw new NotImplementedException();
+        }
+
+        private void PushToServices(Func<IPushNotificationService, Task> pushFunc)
+        {
+            if(_services != null)
+            {
+                foreach(var service in _services)
+                {
+                    pushFunc(service);
+                }
+            }
+        }
+    }
+}

src/Core/Utilities/ServiceCollectionExtensions.cs

@@ -22,6 +22,7 @@ using SqlServerRepos = Bit.Core.Repositories.SqlServer;
 using System.Threading.Tasks;
 using TableStorageRepos = Bit.Core.Repositories.TableStorage;
 using Microsoft.Extensions.DependencyInjection.Extensions;
+using System.Collections.Generic;
 
 namespace Bit.Core.Utilities
 {
@@ -82,24 +83,22 @@ namespace Bit.Core.Utilities
                 services.AddSingleton<IMailDeliveryService, NoopMailDeliveryService>();
             }
 
+            services.AddSingleton<IPushNotificationService, MultiServicePushNotificationService>();
             if(globalSettings.SelfHosted &&
                 CoreHelpers.SettingHasValue(globalSettings.PushRelayBaseUri) &&
                 globalSettings.Installation?.Id != null &&
                 CoreHelpers.SettingHasValue(globalSettings.Installation?.Key))
             {
-                services.AddSingleton<IPushNotificationService, RelayPushNotificationService>();
                 services.AddSingleton<IPushRegistrationService, RelayPushRegistrationService>();
             }
 #if NET471
             else if(!globalSettings.SelfHosted)
             {
-                services.AddSingleton<IPushNotificationService, NotificationHubPushNotificationService>();
                 services.AddSingleton<IPushRegistrationService, NotificationHubPushRegistrationService>();
             }
 #endif
             else
             {
-                services.AddSingleton<IPushNotificationService, NoopPushNotificationService>();
                 services.AddSingleton<IPushRegistrationService, NoopPushRegistrationService>();
             }
 

src/Hub/Hub.csproj

@@ -10,6 +10,7 @@
   <ItemGroup>
     <PackageReference Include="IdentityServer4.AccessTokenValidation" Version="2.6.0" />
     <PackageReference Include="Microsoft.AspNetCore.App" />
+    <PackageReference Include="WindowsAzure.Storage" Version="9.3.0" />
   </ItemGroup>
 
   <ItemGroup>

src/Hub/TimedHostedService.cs

@@ -1,9 +1,15 @@
 using System;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Bit.Core;
+using Bit.Core.Models;
 using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
+using Microsoft.WindowsAzure.Storage;
+using Microsoft.WindowsAzure.Storage.Queue;
+using Newtonsoft.Json;
 
 namespace Bit.Hub
 {
@@ -11,37 +17,110 @@ namespace Bit.Hub
     {
         private readonly ILogger _logger;
         private readonly IHubContext<SyncHub> _hubContext;
-        private Timer _timer;
+        private readonly GlobalSettings _globalSettings;
 
-        public TimedHostedService(ILogger<TimedHostedService> logger, IHubContext<SyncHub> hubContext)
+        private Task _executingTask;
+        private CancellationTokenSource _cts;
+        private CloudQueue _queue;
+
+        public TimedHostedService(ILogger<TimedHostedService> logger, IHubContext<SyncHub> hubContext,
+            GlobalSettings globalSettings)
         {
             _logger = logger;
             _hubContext = hubContext;
+            _globalSettings = globalSettings;
         }
 
         public Task StartAsync(CancellationToken cancellationToken)
         {
-            _logger.LogInformation("Timed Background Service is starting.");
+            _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-            _timer = new Timer(DoWork, null, TimeSpan.Zero, TimeSpan.FromSeconds(5));
+            _executingTask = ExecuteAsync(_cts.Token);
-            return Task.CompletedTask;
+            return _executingTask.IsCompleted ? _executingTask : Task.CompletedTask;
         }
 
-        private void DoWork(object state)
+        public async Task StopAsync(CancellationToken cancellationToken)
         {
-            _logger.LogInformation("Timed Background Service is working.");
+            if(_executingTask == null)
-            _hubContext.Clients.All.SendAsync("ReceiveMessage", "From BG!!");
+            {
+                return;
             }
-
+            _cts.Cancel();
-        public Task StopAsync(CancellationToken cancellationToken)
+            await Task.WhenAny(_executingTask, Task.Delay(-1, cancellationToken));
-        {
+            cancellationToken.ThrowIfCancellationRequested();
-            _logger.LogInformation("Timed Background Service is stopping.");
-            _timer?.Change(Timeout.Infinite, 0);
-            return Task.CompletedTask;
         }
 
         public void Dispose()
         {
-            _timer?.Dispose();
+            // TODO
+        }
+
+        private async Task ExecuteAsync(CancellationToken cancellationToken)
+        {
+            var storageAccount = CloudStorageAccount.Parse(_globalSettings.Events.ConnectionString);
+            var queueClient = storageAccount.CreateCloudQueueClient();
+            _queue = queueClient.GetQueueReference("sync");
+
+            while(!cancellationToken.IsCancellationRequested)
+            {
+                var messages = await _queue.GetMessagesAsync(32, TimeSpan.FromMinutes(1),
+                    null, null, cancellationToken);
+                if(messages.Any())
+                {
+                    foreach(var message in messages)
+                    {
+                        var notification = JsonConvert.DeserializeObject<PushNotificationData<object>>(
+                            message.AsString);
+                        switch(notification.Type)
+                        {
+                            case Core.Enums.PushType.SyncCipherUpdate:
+                            case Core.Enums.PushType.SyncCipherCreate:
+                            case Core.Enums.PushType.SyncCipherDelete:
+                            case Core.Enums.PushType.SyncLoginDelete:
+                                var cipherNotification =
+                                    JsonConvert.DeserializeObject<PushNotificationData<SyncCipherPushNotification>>(
+                                        message.AsString);
+                                if(cipherNotification.Payload.UserId.HasValue)
+                                {
+                                    await _hubContext.Clients.User(cipherNotification.Payload.UserId.ToString())
+                                        .SendAsync("ReceiveMessage", notification, cancellationToken);
+                                }
+                                else if(cipherNotification.Payload.OrganizationId.HasValue)
+                                {
+                                    await _hubContext.Clients.Group(
+                                        $"Organization_{cipherNotification.Payload.OrganizationId}")
+                                        .SendAsync("ReceiveMessage", notification, cancellationToken);
+                                }
+                                break;
+                            case Core.Enums.PushType.SyncFolderUpdate:
+                            case Core.Enums.PushType.SyncFolderCreate:
+                            case Core.Enums.PushType.SyncFolderDelete:
+                                var folderNotification =
+                                    JsonConvert.DeserializeObject<PushNotificationData<SyncFolderPushNotification>>(
+                                        message.AsString);
+                                await _hubContext.Clients.User(folderNotification.Payload.UserId.ToString())
+                                        .SendAsync("ReceiveMessage", notification, cancellationToken);
+                                break;
+                            case Core.Enums.PushType.SyncCiphers:
+                            case Core.Enums.PushType.SyncVault:
+                            case Core.Enums.PushType.SyncOrgKeys:
+                            case Core.Enums.PushType.SyncSettings:
+                                var userNotification =
+                                    JsonConvert.DeserializeObject<PushNotificationData<SyncUserPushNotification>>(
+                                        message.AsString);
+                                await _hubContext.Clients.User(userNotification.Payload.UserId.ToString())
+                                        .SendAsync("ReceiveMessage", notification, cancellationToken);
+                                break;
+                            default:
+                                break;
+                        }
+                        await _queue.DeleteMessageAsync(message);
+                    }
+                }
+                else
+                {
+                    await Task.Delay(TimeSpan.FromSeconds(5), cancellationToken);
+                }
+            }
         }
     }
 }