Merge remote-tracking branch 'origin/test'

README.md

@@ -1,2 +1,8 @@
 # Lykke.Tools.AssetMigrator
 Generic tool for assets migrations
+
+**NB!** You should either cleanup migrations table before each migration session, or specify unique migration id.
+
+## Usage example
+
+dotnet migrate-asset.dll Transfer --balances-conn-string "azure-connection-string" --from 94853f34-3524-486d-923e-3275907aecb4 --to 2149a271-f94b-464a-8d23-673905f5feed --me-endpoint host:port --operations-url http://host/ --accuracy 5 --multiplier 3 --migration-id aa48914b-499a-480a-9334-900932b3dc07

src/Lykke.Tools.AssetMigrator/ICommandLineArguments.cs

@@ -0,0 +1,15 @@
+using Microsoft.Extensions.CommandLineUtils;
+
+namespace Lykke.Tools.AssetMigrator
+{
+    public interface ICommandLineArguments
+    {
+        MigrationMode Mode { get; }
+
+
+        void Configure(
+            CommandLineApplication app);
+
+        bool Validate();
+    }
+}

src/Lykke.Tools.AssetMigrator/Implementations/CommandLineArguments.cs

@@ -0,0 +1,30 @@
+using System;
+using Microsoft.Extensions.CommandLineUtils;
+
+namespace Lykke.Tools.AssetMigrator.Implementations
+{
+    public class CommandLineArguments : ICommandLineArguments
+    {
+        private CommandArgument _mode;
+
+
+        public MigrationMode Mode 
+            => Enum.Parse<MigrationMode>(_mode.Value);
+
+
+        public void Configure(
+            CommandLineApplication app)
+        {
+            _mode = app.Argument
+            (
+                "mode",
+                "Migration mode, either 'Transfer' or 'Copy'"
+            );
+        }
+
+        public bool Validate()
+        {
+            return Enum.TryParse<MigrationMode>(_mode.Value, out _);
+        }
+    }
+}

src/Lykke.Tools.AssetMigrator/Implementations/Migrator.cs

@@ -20,13 +20,16 @@ public sealed class Migrator : IMigrator
     {
         private readonly ILog _log;
         private readonly ILogFactory _logFactory;
+        private readonly ICommandLineArguments _arguments;
         private readonly ICommandLineOptions _options;
 
 
         public Migrator(
+            ICommandLineArguments arguments,
             ILogFactory logFactory,
             ICommandLineOptions options)
         {
+            _arguments = arguments;
             _log = logFactory.CreateLog(this);
             _logFactory = logFactory;
             _options = options;
@@ -38,10 +41,20 @@ public async Task RunAsync()
             _log.Info("Migration started");
 
             var balances = await GetBalancesAsync();
-
-            _log.Info($"Migrating {balances.Length} balances...");
 
-            await MigrateAsync(balances);
+            _log.Info($"Migrating {balances.Length} balances...");
+
+            switch (_arguments.Mode)
+            {
+                case MigrationMode.Copy:
+                    await CopyAsync(balances);
+                    break;
+                case MigrationMode.Transfer:
+                    await TransferAsync(balances);
+                    break;
+                default:
+                    throw new ArgumentOutOfRangeException($"Migration mode [{_arguments.Mode.ToString()}] is not supported.");
+            }
 
             _log.Info("Migration completed");
         }
@@ -78,7 +91,80 @@ private async Task<BalanceEntity[]> GetBalancesAsync()
             return balances.Where(x => x.Balance > 0).ToArray();
         }
 
-        private async Task MigrateAsync(
+        private async Task CopyAsync(
+            IReadOnlyList<BalanceEntity> balances)
+        {            
+            var meClient = new TcpMatchingEngineClient(_options.MEEndPoint, EmptyLogFactory.Instance);
+
+            var migrationsRepository = new MigrationsRepository
+            (
+                _options.BalancesConnectionString,
+                _logFactory,
+                _options.MigrationId
+            );
+
+            var operationsRepository = new CashOperationsRepositoryClient
+            (
+                _options.OperationsUrl,
+                _logFactory.CreateLog(this),
+                600
+            );
+
+            meClient.Start();
+
+            for (var i = 0; i < balances.Count; i++)
+            {
+                var balance = balances[i];
+
+                try
+                {
+                    var cashInId = await migrationsRepository.GetOrCreateCashInOutIdAsync(balance.ClientId, _options.TargetAssetId);
+                    var cashInAmount = ((double) balance.Balance * _options.Multiplier).TruncateDecimalPlaces((int) _options.TargetAssetAccuracy);
+
+                    var cashInResult = await meClient.CashInOutAsync
+                    (
+                        id: cashInId.ToString(),
+                        clientId: balance.ClientId,
+                        assetId: _options.TargetAssetId,
+                        amount: cashInAmount
+                    );
+
+                    if (cashInResult.Status == MeStatusCodes.Ok || cashInResult.Status == MeStatusCodes.Duplicate)
+                    {
+                        var operationId = cashInId.ToString();
+
+                        if (await operationsRepository.GetAsync(balance.ClientId, operationId) == null)
+                        {
+                            await operationsRepository.RegisterAsync(new CashInOutOperation
+                            {
+                                Amount = cashInAmount,
+                                AssetId = _options.TargetAssetId,
+                                ClientId = balance.ClientId,
+                                DateTime = DateTime.UtcNow,
+                                Id = operationId,
+                                TransactionId = operationId,
+                                BlockChainHash = _options.MigrationMessage,
+
+                                Type = CashOperationType.None,
+                                State = TransactionStates.SettledOffchain
+                            });
+                        }
+                    }
+                    else
+                    {
+                        _log.Warning($"CashIn for client [{balance.ClientId}] completed with [{cashInResult.Status.ToString()}] status");
+                    }
+
+                    _log.Info($"Copied {i + 1} of {balances.Count} balances.");
+                }
+                catch (Exception e)
+                {
+                    _log.Error(e, $"Failed to copy balance for client [{balance.ClientId}]");
+                }
+            }
+        }
+
+        private async Task TransferAsync(
             IReadOnlyList<BalanceEntity> balances)
         {
             var meClient = new TcpMatchingEngineClient(_options.MEEndPoint, EmptyLogFactory.Instance);
@@ -178,11 +264,11 @@ private async Task<BalanceEntity[]> GetBalancesAsync()
                         _log.Warning($"CashOut completed for client {balance.ClientId} with [{cashOutResult.Status.ToString()}] status");
                     }
 
-                    _log.Info($"Completed {i + 1} of {balances.Count} migrations");
+                    _log.Info($"Completed {i + 1} of {balances.Count} transfers");
                 }
                 catch (Exception e)
                 {
-                    _log.Error(e, $"Failed to migrate balance for client [{balance.ClientId}]");
+                    _log.Error(e, $"Failed to transfer balance for client [{balance.ClientId}]");
                 }
             }
         }

src/Lykke.Tools.AssetMigrator/Implementations/RootCommand.cs

@@ -12,17 +12,20 @@ namespace Lykke.Tools.AssetMigrator.Implementations
     public sealed class RootCommand : IRootCommand
     {
         private readonly CommandLineApplication _app;
+        private readonly ICommandLineArguments _arguments;
         private readonly ILog _log;
         private readonly IMigrator _migrator;
         private readonly ICommandLineOptions _options;
 
 
         public RootCommand(
+            ICommandLineArguments arguments,
             ILogFactory logFactory,
             IMigrator migrator,
             ICommandLineOptions options)
         {
             _app = new CommandLineApplication(throwOnUnexpectedArg: false);
+            _arguments = arguments;
             _log = logFactory.CreateLog(this);
             _migrator = migrator;
             _options = options;
@@ -31,6 +34,7 @@ public sealed class RootCommand : IRootCommand
 
         public CommandLineApplication Configure()
         {
+            _arguments.Configure(_app);
             _options.Configure(_app);
 
             _app.OnExecute(() => ExecuteAsync());
@@ -44,7 +48,7 @@ private async Task<int> ExecuteAsync()
         {
             try
             {
-                if (_options.ShowHelp || !_options.Validate())
+                if (_options.ShowHelp || !_arguments.Validate() || !_options.Validate())
                 {
                     _app.ShowHelp();
                 }

src/Lykke.Tools.AssetMigrator/MigrationMode.cs

@@ -0,0 +1,8 @@
+namespace Lykke.Tools.AssetMigrator
+{
+    public enum MigrationMode
+    {
+        Copy,
+        Transfer
+    }
+}

src/Lykke.Tools.AssetMigrator/Program.cs

@@ -48,6 +48,9 @@ private static IServiceCollection ConfigureServices()
                         .AddConsole()
                 );
 
+            serviceCollection
+                .AddSingleton<ICommandLineArguments, CommandLineArguments>();
+
             serviceCollection
                 .AddSingleton<ICommandLineOptions, CommandLineOptions>();