LM-3196 implemented API trading blocking

settings.yaml

@@ -84,6 +84,9 @@ RabbitMq:
       settings-key: RabbitSpotPricesConnString
     ExchangeName:
       settings-key: PublicTradesExchangeName
+  ClientAccountFeedConnectionString:
+    settings-key: HFT-ClientAccountFeed
+    types: [RabbitMq]
 SagasRabbitMq:
   RabbitConnectionString:
     settings-key: RabbitSagas

src/HftApi.Common/Configuration/RabbitMqConfig.cs

@@ -1,4 +1,4 @@
-namespace HftApi.Common.Configuration
+namespace HftApi.Common.Configuration
 {
     public class RabbitMqConfig
     {
@@ -7,11 +7,6 @@ public class RabbitMqConfig
         public RabbitMqConnection Balances { get; set; }
         public RabbitMqConnection Orders { get; set; }
         public RabbitMqConnection PublicTrades { get; set; }
-    }
-
-    public class RabbitMqConnection
-    {
-        public string ConnectionString { get; set; }
-        public string ExchangeName { get; set; }
+        public string ClientAccountFeedConnectionString { get; set; }
     }
 }

src/HftApi.Common/Configuration/RabbitMqConnection.cs

@@ -0,0 +1,8 @@
+namespace HftApi.Common.Configuration
+{
+    public class RabbitMqConnection
+    {
+        public string ConnectionString { get; set; }
+        public string ExchangeName { get; set; }
+    }
+}

src/HftApi.Common/Configuration/ServicesConfig.cs

@@ -1,4 +1,4 @@
-using Lykke.Service.Kyc.Client;
+using Lykke.Service.Kyc.Client;
 
 namespace HftApi.Common.Configuration
 {

src/HftApi/Extensions/UserExxtensions.cs → src/HftApi/Extensions/UserExtensions.cs

@@ -1,9 +1,9 @@
-using System.Linq;
+using System.Linq;
 using System.Security.Claims;
 
 namespace HftApi.Extensions
 {
-    public static class UserExxtensions
+    public static class UserExtensions
     {
         public static string GetClientId(this ClaimsPrincipal user)
         {

src/HftApi/HftApi.csproj

@@ -21,7 +21,7 @@
     <PackageReference Include="Lykke.Messaging.RabbitMq" Version="2.2.2" />
     <PackageReference Include="Lykke.Service.Balances.Client" Version="2.3.2" />
     <PackageReference Include="Lykke.Service.ClientDialogs.Client" Version="1.0.1" />
-    <PackageReference Include="Lykke.Service.HftInternalService.Client" Version="2.1.1" />
+    <PackageReference Include="Lykke.Service.HftInternalService.Client" Version="2.4.0" />
     <PackageReference Include="Lykke.Service.Kyc.Client" Version="1.5.0" />
     <PackageReference Include="Lykke.Service.Operations.Client" Version="3.1.0" />
     <PackageReference Include="Lykke.Service.Operations.Contracts" Version="3.1.1" />

src/HftApi/Modules/AutofacModule.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 using Autofac;
 using AzureStorage;
 using AzureStorage.Tables;
@@ -52,6 +52,9 @@ protected override void Load(ContainerBuilder builder)
                 .WithParameter(TypedParameter.From(_config.Redis.OrderBooksCacheKeyPattern))
                 .SingleInstance();
 
+            builder.RegisterType<BlockedClientsService>()
+                .As<IBlockedClientsService>();
+
             var cache = new RedisCache(new RedisCacheOptions
             {
                 Configuration = _config.Redis.RedisConfiguration,
@@ -80,6 +83,12 @@ protected override void Load(ContainerBuilder builder)
                 .WithParameter("exchangeName", _config.RabbitMq.HftInternal.ExchangeName)
                 .SingleInstance();
 
+            builder.RegisterType<ClientSettingsUpdatesSubscriber>()
+                .As<IStartable>()
+                .AutoActivate()
+                .WithParameter("connectionString", _config.RabbitMq.ClientAccountFeedConnectionString)
+                .SingleInstance();
+
             builder.RegisterHftInternalClient(_config.Services.HftInternalServiceUrl);
 
             builder.RegisterType<TokenService>()
@@ -187,7 +196,7 @@ protected override void Load(ContainerBuilder builder)
             builder.RegisterOperationsClient(_config.Services.OperationsServiceUrl);
 
             builder.RegisterClientDialogsClient(_config.Services.ClientDialogsServiceUrl);
-            
+
             builder.RegisterInstance(
                 new Swisschain.Sirius.Api.ApiClient.ApiClient(_config.Services.SiriusApiServiceClient.GrpcServiceUrl, _config.Services.SiriusApiServiceClient.ApiKey)
             ).As<Swisschain.Sirius.Api.ApiClient.IApiClient>();

src/HftApi/RabbitSubscribers/ClientSettingsUpdatesSubscriber.cs

@@ -0,0 +1,102 @@
+using System;
+using System.Threading.Tasks;
+using Autofac;
+using Common.Log;
+using HftApi.RabbitSubscribers.Messages;
+using Lykke.Common.Log;
+using Lykke.HftApi.Domain.Services;
+using Lykke.RabbitMqBroker;
+using Lykke.RabbitMqBroker.Subscriber;
+using Lykke.Service.HftInternalService.Client;
+
+namespace HftApi.RabbitSubscribers
+{
+    public class ClientSettingsUpdatesSubscriber : IStartable, IDisposable
+    {
+        private readonly RabbitMqSubscriber<ClientSettingsCashoutBlockUpdated> _subscriber;
+        private readonly IHftInternalClient _hftInternalClient;
+        private readonly ITokenService _tokenService;
+        private readonly ILog _log;
+
+        public ClientSettingsUpdatesSubscriber(ILogFactory logFactory,
+            string connectionString,
+            IHftInternalClient hftInternalClient,
+            ITokenService tokenService)
+        {
+            _hftInternalClient = hftInternalClient;
+            _tokenService = tokenService;
+            _log = logFactory.CreateLog(this);
+
+            var subscriptionSettings = RabbitMqSubscriptionSettings
+                .ForSubscriber(
+                    connectionString,
+                    "client-account.client-settings-updated",
+                    $"client-account.client-settings-updated.hft-api-v2-{Environment.MachineName}")
+                .UseRoutingKey(nameof(ClientSettingsCashoutBlockUpdated));
+
+            subscriptionSettings.DeadLetterExchangeName = null;
+
+            var strategy = new DefaultErrorHandlingStrategy(logFactory, subscriptionSettings);
+
+            _subscriber = new RabbitMqSubscriber<ClientSettingsCashoutBlockUpdated>(logFactory, subscriptionSettings, strategy)
+                .SetMessageDeserializer(new JsonMessageDeserializer<ClientSettingsCashoutBlockUpdated>())
+                .SetMessageReadStrategy(new MessageReadWithTemporaryQueueStrategy())
+                .Subscribe(HandleMessage);
+        }
+
+        public async Task HandleMessage(ClientSettingsCashoutBlockUpdated evt)
+        {
+            // Race condition with KeyUpdatedEvent event handling is possible, but it is decided
+            // that it's acceptable since these events are not very frequent
+
+            _log.Info($"Got client trades blocking update. Trades are blocked: {evt.TradesBlocked}", context: new
+            {
+                ClientId = evt.ClientId
+            });           
+
+            var enabledClientApiKeys = await _hftInternalClient.Keys.GetKeys(evt.ClientId);
+
+            if (evt.TradesBlocked)
+            {
+                foreach (var key in enabledClientApiKeys)
+                {
+                    _tokenService.Remove(key.ApiKey);
+
+                    var apiKeyStart = key.ApiKey.Substring(0, 4);
+
+                    _log.Info($"API key has been cached", context: new
+                    {
+                        ClientId = evt.ClientId,
+                        ApiKeyStart = apiKeyStart
+                    });
+                }
+            }
+            else
+            {
+                foreach (var key in enabledClientApiKeys)
+                {
+                    _tokenService.Add(key.ApiKey);
+
+                    var apiKeyStart = key.ApiKey.Substring(0, 4);
+
+                    _log.Info($"API key has been evicted from the cache", context: new
+                    {
+                        ClientId = evt.ClientId,
+                        ApiKeyStart = apiKeyStart
+                    });
+                }                
+            }
+        }
+
+        public void Start()
+        {
+            _subscriber.Start();
+        }
+
+        public void Dispose()
+        {
+            _subscriber?.Stop();
+            _subscriber?.Dispose();
+        }
+    }
+}

src/HftApi/RabbitSubscribers/KeyUpdateSubscriber.cs

@@ -1,12 +1,12 @@
-using System;
+using System;
 using System.Threading.Tasks;
 using Autofac;
+using Common.Log;
 using Lykke.Common.Log;
 using Lykke.HftApi.Domain.Services;
 using Lykke.RabbitMqBroker;
 using Lykke.RabbitMqBroker.Subscriber;
 using Lykke.Service.HftInternalService.Client.Messages;
-using Microsoft.Extensions.Logging;
 
 namespace HftApi.RabbitSubscribers
 {
@@ -16,51 +16,85 @@ public class KeyUpdateSubscriber : IStartable, IDisposable
         private readonly string _exchangeName;
         private readonly ITokenService _tokenService;
         private readonly ILogFactory _logFactory;
+        private readonly ILog _log;
+        private readonly IBlockedClientsService _blockedClients;
         private RabbitMqSubscriber<KeyUpdatedEvent> _subscriber;
 
         public KeyUpdateSubscriber(
             string connectionString,
             string exchangeName,
             ITokenService tokenService,
-            ILoggerFactory loggerFactory,
-            ILogFactory logFactory)
+            ILogFactory logFactory,
+            IBlockedClientsService blockedClients)
         {
+            _log = logFactory.CreateLog(this);
             _connectionString = connectionString;
             _exchangeName = exchangeName;
             _tokenService = tokenService;
             _logFactory = logFactory;
+            _blockedClients = blockedClients;
         }
 
         public void Start()
         {
-            _tokenService.InitAsync().GetAwaiter().GetResult();
-
             var settings = RabbitMqSubscriptionSettings
                 .ForSubscriber(_connectionString, _exchangeName, $"{nameof(KeyUpdateSubscriber)}-{Environment.MachineName}");
 
             settings.DeadLetterExchangeName = null;
 
             _subscriber = new RabbitMqSubscriber<KeyUpdatedEvent>(_logFactory,
                     settings,
-                    new ResilientErrorHandlingStrategy(_logFactory, settings, TimeSpan.FromSeconds(10)))
+                    new ResilientErrorHandlingStrategy(_log, settings, TimeSpan.FromSeconds(10)))
                 .SetMessageDeserializer(new JsonMessageDeserializer<KeyUpdatedEvent>())
                 .Subscribe(ProcessMessageAsync)
                 .CreateDefaultBinding()
                 .Start();
+
+            _tokenService.InitAsync().GetAwaiter().GetResult();
         }
 
-        private Task ProcessMessageAsync(KeyUpdatedEvent message)
+        private async Task ProcessMessageAsync(KeyUpdatedEvent message)
         {
-            if (message.IsDeleted)
-                _tokenService.Remove(message.Id);
-            else
+            // Race condition with ClientSettingsCashoutBlockUpdated event handling is possible, but it is decided
+            // that it's acceptable since these events are not very frequent
+
+            var isClientBlocked = await _blockedClients.IsClientBlocked(message.ClientId);
+            var apiKeyStart = message.Id.Substring(0, 4);
+
+            _log.Info($"API key deleted: {message.IsDeleted}. Client blocked: {isClientBlocked}", context: new
+            {
+                ClientId = message.ClientId,
+                WalletId = message.WalletId,
+                ApiKeyStart = apiKeyStart
+            });
+
+            if (!message.IsDeleted && !isClientBlocked)
+            {
                 _tokenService.Add(message.Id);
 
-            return Task.CompletedTask;
+                _log.Info($"API key has been cached", context: new
+                {
+                    ClientId = message.ClientId,
+                    WalletId = message.WalletId,
+                    ApiKeyStart = apiKeyStart
+                });
+            }
+            else
+            {
+                _tokenService.Remove(message.Id);
+
+                _log.Info($"API key has been evicted from the cache", context: new
+                {
+                    ClientId = message.ClientId,
+                    WalletId = message.WalletId,
+                    ApiKeyStart = apiKeyStart
+                });
+            }
         }
 
         public void Dispose()
         {
+            _subscriber?.Stop();
             _subscriber?.Dispose();
         }
     }

src/HftApi/RabbitSubscribers/Messages/ClientSettingsCashoutBlockUpdated.cs

@@ -0,0 +1,10 @@
+namespace HftApi.RabbitSubscribers.Messages
+{
+    // Owned by Lykke.Service.ClientAccount service
+    public class ClientSettingsCashoutBlockUpdated
+    {
+        public string ClientId { get; set; }
+        public bool CashOutBlocked { get; set; }
+        public bool TradesBlocked { get; set; }
+    }
+}

src/Lykke.HftApi.Domain/Services/IBlockedClientsService.cs

@@ -0,0 +1,9 @@
+using System.Threading.Tasks;
+
+namespace Lykke.HftApi.Domain.Services
+{
+    public interface IBlockedClientsService
+    {
+        Task<bool> IsClientBlocked(string clientId);
+    }
+}

src/Lykke.HftApi.Domain/Services/ITokenService.cs

@@ -1,4 +1,4 @@
-using System.Threading.Tasks;
+using System.Threading.Tasks;
 
 namespace Lykke.HftApi.Domain.Services
 {

src/Lykke.HftApi.Services/AssetsService.cs

@@ -1,12 +1,10 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using Autofac;
-using Lykke.HftApi.Domain;
 using Lykke.HftApi.Domain.Entities;
 using Lykke.HftApi.Domain.Entities.Assets;
-using Lykke.HftApi.Domain.Exceptions;
 using Lykke.HftApi.Domain.Services;
 using Lykke.HftApi.Services.AssetsClient;
 

src/Lykke.HftApi.Services/BlockedClientsService.cs

@@ -0,0 +1,38 @@
+using System.Threading.Tasks;
+using Common.Log;
+using Lykke.Common.ApiLibrary.Exceptions;
+using Lykke.Common.Log;
+using Lykke.HftApi.Domain.Services;
+using Lykke.Service.ClientAccount.Client;
+
+namespace Lykke.HftApi.Services
+{
+    public class BlockedClientsService : IBlockedClientsService
+    {
+        private readonly ILog _log;
+        private readonly IClientAccountClient _clientAccountClient;
+
+        public BlockedClientsService(ILogFactory logFactory, IClientAccountClient clientAccountClient)
+        {
+            _log = logFactory.CreateLog(this);
+            _clientAccountClient = clientAccountClient;
+        }
+
+        public async Task<bool> IsClientBlocked(string clientId)
+        {
+            try
+            {
+                return (await _clientAccountClient.ClientSettings.GetCashOutBlockSettingsAsync(clientId))?.TradesBlocked == true;
+            }
+            catch (ClientApiException ex) when (ex.Message == "Client not found")
+            {
+                _log.Warning("Client not found. It will be treated as blocked one", context: new
+                {
+                    ClientId = clientId
+                });
+
+                return false;
+            }
+        }
+    }
+}