Rebalancing Job + Required services

src/Jobs/RebalanceJob.cs

@@ -0,0 +1,56 @@
+/*
+ * NodeGuard
+ * Copyright (C) 2023  Elenpay
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see http://www.gnu.org/licenses/.
+ *
+ */
+
+using NodeGuard.Services;
+using Quartz;
+
+namespace NodeGuard.Jobs;
+
+/// <summary>
+/// One-shot Quartz job that fires a queued run/retry of an existing Rebalance row.
+/// The row carries all the parameters; this job only loads it and re-runs the probe + payment.
+/// </summary>
+[DisallowConcurrentExecution]
+public class RebalanceJob : IJob
+{
+    private readonly ILogger<RebalanceJob> _logger;
+    private readonly IRebalanceService _rebalanceService;
+
+    public RebalanceJob(ILogger<RebalanceJob> logger, IRebalanceService rebalanceService)
+    {
+        _logger = logger;
+        _rebalanceService = rebalanceService;
+    }
+
+    public async Task Execute(IJobExecutionContext context)
+    {
+        var rebalanceId = context.MergedJobDataMap.GetIntValue("rebalanceId");
+        _logger.LogInformation("Firing scheduled run/retry for rebalance {Id}", rebalanceId);
+
+        try
+        {
+            await _rebalanceService.ExecuteAsync(rebalanceId, context.CancellationToken);
+        }
+        catch (Exception e)
+        {
+            _logger.LogError(e, "Error executing scheduled retry for rebalance {Id}", rebalanceId);
+            throw new JobExecutionException(e, refireImmediately: false);
+        }
+    }
+}

src/Rpc/NodeGuardService.cs

@@ -54,6 +54,12 @@ Task<GetNewWalletAddressResponse> GetNewWalletAddress(GetNewWalletAddressRequest
     Task<GetChannelResponse> GetChannel(GetChannelRequest request, ServerCallContext context);
 
     Task<AddTagsResponse> AddTags(AddTagsRequest request, ServerCallContext context);
+
+    Task<RequestRebalanceResponse> RequestRebalance(RequestRebalanceRequest request, ServerCallContext context);
+
+    Task<GetRebalanceResponse> GetRebalance(GetRebalanceRequest request, ServerCallContext context);
+
+    Task<GetRebalancesResponse> GetRebalances(GetRebalancesRequest request, ServerCallContext context);
 }
 
 /// <summary>
@@ -78,6 +84,8 @@ public class NodeGuardService : Nodeguard.NodeGuardService.NodeGuardServiceBase,
     private readonly IFMUTXORepository _fmutxoRepository;
     private readonly IUTXOTagRepository _utxoTagRepository;
     private readonly IHtlcMonitoringScheduler _htlcMonitoringScheduler;
+    private readonly IRebalanceService _rebalanceService;
+    private readonly IRebalanceRepository _rebalanceRepository;
 
     public NodeGuardService(ILogger<NodeGuardService> logger,
         ILiquidityRuleRepository liquidityRuleRepository,
@@ -94,7 +102,9 @@ public NodeGuardService(ILogger<NodeGuardService> logger,
         ILightningService lightningService,
         IFMUTXORepository fmutxoRepository,
         IUTXOTagRepository utxoTagRepository,
-        IHtlcMonitoringScheduler htlcMonitoringScheduler
+        IHtlcMonitoringScheduler htlcMonitoringScheduler,
+        IRebalanceService rebalanceService,
+        IRebalanceRepository rebalanceRepository
     )
     {
         _logger = logger;
@@ -113,6 +123,8 @@ IHtlcMonitoringScheduler htlcMonitoringScheduler
         _fmutxoRepository = fmutxoRepository;
         _utxoTagRepository = utxoTagRepository;
         _htlcMonitoringScheduler = htlcMonitoringScheduler;
+        _rebalanceService = rebalanceService;
+        _rebalanceRepository = rebalanceRepository;
         _scheduler = Task.Run(() => _schedulerFactory.GetScheduler()).Result;
     }
 
@@ -1190,6 +1202,159 @@ public override async Task<AddTagsResponse> AddTags(AddTagsRequest request, Serv
         return new AddTagsResponse();
     }
 
+    public override async Task<RequestRebalanceResponse> RequestRebalance(RequestRebalanceRequest request, ServerCallContext context)
+    {
+        if (string.IsNullOrWhiteSpace(request.NodePubkey))
+            throw new RpcException(new Status(StatusCode.InvalidArgument, "node_pubkey is required"));
+
+        if (request.AmountSats <= 0)
+            throw new RpcException(new Status(StatusCode.InvalidArgument, "amount_sats must be > 0"));
+
+        var node = await _nodeRepository.GetByPubkey(request.NodePubkey);
+        if (node == null)
+            throw new RpcException(new Status(StatusCode.NotFound, $"Node with pubkey {request.NodePubkey} not found"));
+
+        var domainRequest = new RebalanceRequest(
+            NodeId: node.Id,
+            SourceChannelId: request.HasSourceChannelId ? request.SourceChannelId : null,
+            TargetPubkey: request.HasTargetPubkey ? request.TargetPubkey : null,
+            AmountSats: request.AmountSats,
+            MaxFeePct: request.HasMaxFeePct ? request.MaxFeePct : null,
+            TimeoutSeconds: request.TimeoutSeconds,
+            IsManual: true,
+            UserRequestorId: null,
+            ProbeBackoffRatio: request.HasProbeBackoffRatio ? request.ProbeBackoffRatio : null,
+            MaxAttempts: request.HasMaxAttempts ? request.MaxAttempts : null,
+            RetryMaxFeePct: request.HasRetryMaxFeePct ? request.RetryMaxFeePct : null);
+
+        Rebalance rebalance;
+        try
+        {
+            rebalance = await _rebalanceService.RebalanceAsync(domainRequest, context.CancellationToken);
+        }
+        catch (ArgumentException ex)
+        {
+            throw new RpcException(new Status(StatusCode.InvalidArgument, ex.Message));
+        }
+        catch (Exception ex)
+        {
+            _logger.LogError(ex, "RequestRebalance failed for node {NodePubkey}", request.NodePubkey);
+            throw new RpcException(new Status(StatusCode.Internal, $"Rebalance failed: {ex.Message}"));
+        }
+
+        var response = new RequestRebalanceResponse
+        {
+            RebalanceId = rebalance.Id,
+            Status = MapRebalanceStatus(rebalance.Status),
+            AttemptNumber = rebalance.AttemptNumber,
+        };
+        if (rebalance.FeePaidSats.HasValue) response.FeePaidSats = rebalance.FeePaidSats.Value;
+        if (rebalance.EffectivePpm.HasValue) response.EffectivePpm = rebalance.EffectivePpm.Value;
+        if (rebalance.SatsAmount != rebalance.RequestedAmountSats) response.ActualAmountSats = rebalance.SatsAmount;
+        return response;
+    }
+
+    public override async Task<GetRebalanceResponse> GetRebalance(GetRebalanceRequest request, ServerCallContext context)
+    {
+        if (request.RebalanceId <= 0)
+            throw new RpcException(new Status(StatusCode.InvalidArgument, "rebalance_id is required"));
+
+        var rebalance = await _rebalanceRepository.GetById(request.RebalanceId);
+        if (rebalance == null)
+            throw new RpcException(new Status(StatusCode.NotFound, $"Rebalance {request.RebalanceId} not found"));
+
+        return MapRebalance(rebalance);
+    }
+
+    public override async Task<GetRebalancesResponse> GetRebalances(GetRebalancesRequest request, ServerCallContext context)
+    {
+        if (request.PageNumber < 1)
+            throw new RpcException(new Status(StatusCode.InvalidArgument, "page_number must be >= 1"));
+        if (request.PageSize <= 0)
+            throw new RpcException(new Status(StatusCode.InvalidArgument, "page_size must be > 0"));
+
+        int? nodeIdFilter = null;
+        if (request.HasNodePubkey && !string.IsNullOrWhiteSpace(request.NodePubkey))
+        {
+            var node = await _nodeRepository.GetByPubkey(request.NodePubkey);
+            if (node == null)
+                return new GetRebalancesResponse { TotalCount = 0 };
+            nodeIdFilter = node.Id;
+        }
+
+        var statusFilter = request.HasStatus ? MapRebalanceStatus(request.Status) : (RebalanceStatus?)null;
+        var fromDate = request.HasFromUnix ? DateTimeOffset.FromUnixTimeSeconds(request.FromUnix) : (DateTimeOffset?)null;
+        var toDate = request.HasToUnix ? DateTimeOffset.FromUnixTimeSeconds(request.ToUnix) : (DateTimeOffset?)null;
+
+        var (rebalances, totalCount) = await _rebalanceRepository.GetPaginatedAsync(
+            pageNumber: request.PageNumber,
+            pageSize: request.PageSize,
+            status: statusFilter,
+            nodeId: nodeIdFilter,
+            sourceChannelId: request.HasSourceChannelId ? request.SourceChannelId : null,
+            userId: request.HasUserId ? request.UserId : null,
+            isManual: request.HasIsManual ? request.IsManual : null,
+            fromDate: fromDate,
+            toDate: toDate);
+
+        var response = new GetRebalancesResponse { TotalCount = totalCount };
+        response.Rebalances.AddRange(rebalances.Select(MapRebalance));
+        return response;
+    }
+
+    private static GetRebalanceResponse MapRebalance(Rebalance rebalance)
+    {
+        var response = new GetRebalanceResponse
+        {
+            RebalanceId = rebalance.Id,
+            NodePubkey = rebalance.SourceNodePubKey ?? rebalance.Node?.PubKey ?? string.Empty,
+            Status = MapRebalanceStatus(rebalance.Status),
+            IsManual = rebalance.IsManual,
+            AttemptNumber = rebalance.AttemptNumber,
+            RequestedAmountSats = rebalance.RequestedAmountSats,
+            AmountSats = rebalance.SatsAmount,
+            MaxFeePct = rebalance.MaxFeePct,
+            CreationDatetimeUnix = rebalance.CreationDatetime.ToUnixTimeSeconds(),
+            UpdateDatetimeUnix = rebalance.UpdateDatetime.ToUnixTimeSeconds(),
+        };
+        if (rebalance.RetryMaxFeePct.HasValue) response.RetryMaxFeePct = rebalance.RetryMaxFeePct.Value;
+        if (rebalance.FeePaidSats.HasValue) response.FeePaidSats = rebalance.FeePaidSats.Value;
+        if (rebalance.EffectivePpm.HasValue) response.EffectivePpm = rebalance.EffectivePpm.Value;
+        if (rebalance.SourceChanIdLnd.HasValue) response.SourceChanId = rebalance.SourceChanIdLnd.Value;
+        if (rebalance.TargetPubkey != null) response.TargetPubkey = rebalance.TargetPubkey;
+        if (rebalance.ProbeBackoffRatio.HasValue) response.ProbeBackoffRatio = rebalance.ProbeBackoffRatio.Value;
+        if (rebalance.MaxAttempts.HasValue) response.MaxAttempts = rebalance.MaxAttempts.Value;
+        return response;
+    }
+
+    private static REBALANCE_STATUS MapRebalanceStatus(RebalanceStatus status) => status switch
+    {
+        RebalanceStatus.Pending => REBALANCE_STATUS.RebalancePending,
+        RebalanceStatus.Probing => REBALANCE_STATUS.RebalanceProbing,
+        RebalanceStatus.InFlight => REBALANCE_STATUS.RebalanceInFlight,
+        RebalanceStatus.Succeeded => REBALANCE_STATUS.RebalanceSucceeded,
+        RebalanceStatus.Failed => REBALANCE_STATUS.RebalanceFailed,
+        RebalanceStatus.NoRoute => REBALANCE_STATUS.RebalanceNoRoute,
+        RebalanceStatus.Timeout => REBALANCE_STATUS.RebalanceTimeout,
+        RebalanceStatus.InsufficientBalance => REBALANCE_STATUS.RebalanceInsufficientBalance,
+        RebalanceStatus.ExceededFeeLimit => REBALANCE_STATUS.RebalanceExceededFeeLimit,
+        _ => REBALANCE_STATUS.RebalanceFailed,
+    };
+
+    private static RebalanceStatus MapRebalanceStatus(REBALANCE_STATUS status) => status switch
+    {
+        REBALANCE_STATUS.RebalancePending => RebalanceStatus.Pending,
+        REBALANCE_STATUS.RebalanceProbing => RebalanceStatus.Probing,
+        REBALANCE_STATUS.RebalanceInFlight => RebalanceStatus.InFlight,
+        REBALANCE_STATUS.RebalanceSucceeded => RebalanceStatus.Succeeded,
+        REBALANCE_STATUS.RebalanceFailed => RebalanceStatus.Failed,
+        REBALANCE_STATUS.RebalanceNoRoute => RebalanceStatus.NoRoute,
+        REBALANCE_STATUS.RebalanceTimeout => RebalanceStatus.Timeout,
+        REBALANCE_STATUS.RebalanceInsufficientBalance => RebalanceStatus.InsufficientBalance,
+        REBALANCE_STATUS.RebalanceExceededFeeLimit => RebalanceStatus.ExceededFeeLimit,
+        _ => RebalanceStatus.Failed,
+    };
+
     private bool ValidateBitcoinAddress(string address)
     {
         try

src/Services/LightningClientService.cs

@@ -36,6 +36,8 @@ public interface ILightningClientService
     public Task<ListChannelsResponse?> ListChannels(Node node, Lightning.LightningClient? client = null);
     public Task<ChannelBalanceResponse?> ChannelBalanceAsync(Node node, Lightning.LightningClient? client = null);
     public Task<ChannelEdge?> GetChanInfo(Node node, ulong chanId, Lightning.LightningClient? client = null);
+    public Task<AddInvoiceResponse?> AddInvoice(Node node, Invoice invoice, Lightning.LightningClient? client = null);
+    public Task<QueryRoutesResponse?> QueryRoutes(Node node, QueryRoutesRequest request, Lightning.LightningClient? client = null);
     public AsyncServerStreamingCall<CloseStatusUpdate>? CloseChannel(Node node, Channel channel, bool forceClose = false, Lightning.LightningClient? client = null);
     public AsyncServerStreamingCall<ChannelEventUpdate> SubscribeChannelEvents(Node node, Lightning.LightningClient? client = null);
     public Task<LightningNode?> GetNodeInfo(Node node, string pubKey, Lightning.LightningClient? client = null);
@@ -68,7 +70,7 @@ private GrpcChannel CreateLightningClient(string? endpoint)
 
         var grpcChannel = GrpcChannel.ForAddress($"https://{endpoint}",
             new GrpcChannelOptions
-                {HttpHandler = httpHandler, LoggerFactory = NullLoggerFactory.Instance});
+            { HttpHandler = httpHandler, LoggerFactory = NullLoggerFactory.Instance });
 
 
         _logger.LogInformation("New grpc channel created for endpoint {endpoint}", endpoint);
@@ -166,6 +168,45 @@ public Lightning.LightningClient GetLightningClient(string? endpoint)
         }
     }
 
+    public async Task<AddInvoiceResponse?> AddInvoice(Node node, Invoice invoice, Lightning.LightningClient? client = null)
+    {
+        try
+        {
+            client ??= GetLightningClient(node.Endpoint);
+            return await client.AddInvoiceAsync(invoice, new Metadata
+            {
+                { "macaroon", node.ChannelAdminMacaroon }
+            });
+        }
+        catch (Exception e)
+        {
+            _logger.LogError(e, "Error while adding invoice for node {NodeId}", node.Id);
+            return null;
+        }
+    }
+
+    public async Task<QueryRoutesResponse?> QueryRoutes(Node node, QueryRoutesRequest request, Lightning.LightningClient? client = null)
+    {
+        try
+        {
+            client ??= GetLightningClient(node.Endpoint);
+            return await client.QueryRoutesAsync(request, new Metadata
+            {
+                { "macaroon", node.ChannelAdminMacaroon }
+            });
+        }
+        catch (RpcException e)
+        {
+            _logger.LogWarning("QueryRoutes for node {NodeId} returned no routes: {Status}", node.Id, e.Status.Detail);
+            return null;
+        }
+        catch (Exception e)
+        {
+            _logger.LogError(e, "Error while querying routes for node {NodeId}", node.Id);
+            return null;
+        }
+    }
+
     public AsyncServerStreamingCall<CloseStatusUpdate>? CloseChannel(Node node, Channel channel, bool forceClose = false, Lightning.LightningClient? client = null)
     {
         //This method is here to avoid a circular dependency between the LightningService and the ChannelRepository

src/Services/LightningRouterService.cs

@@ -1,6 +1,7 @@
 using System.Collections.Concurrent;
 using Grpc.Core;
 using Grpc.Net.Client;
+using Lnrpc;
 using Microsoft.Extensions.Logging.Abstractions;
 using NodeGuard.Data.Models;
 using NodeGuard.Helpers;
@@ -13,6 +14,8 @@ public interface ILightningRouterService
     public Router.RouterClient GetRouterClient(string? endpoint);
     public Task<RouteFeeResponse?> EstimateRouteFee(Node node, RouteFeeRequest routeFeeRequest, Router.RouterClient? client = null);
     public AsyncServerStreamingCall<HtlcEvent> SubscribeHtlcEvents(Node node, Router.RouterClient? client = null);
+    public AsyncServerStreamingCall<Payment> SendPaymentV2(Node node, SendPaymentRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null);
+    public Task<HTLCAttempt> SendToRouteV2Async(Node node, Routerrpc.SendToRouteRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null);
 }
 
 public class LightningRouterService : ILightningRouterService
@@ -35,7 +38,7 @@ private GrpcChannel CreateRouterClient(string? endpoint)
 
         var grpcChannel = GrpcChannel.ForAddress($"https://{endpoint}",
             new GrpcChannelOptions
-                {HttpHandler = httpHandler, LoggerFactory = NullLoggerFactory.Instance});
+            { HttpHandler = httpHandler, LoggerFactory = NullLoggerFactory.Instance });
 
         _logger.LogInformation("New grpc channel created for router endpoint {endpoint}", endpoint);
 
@@ -94,5 +97,27 @@ public AsyncServerStreamingCall<HtlcEvent> SubscribeHtlcEvents(Node node, Router
                 }
             });
     }
+
+    public AsyncServerStreamingCall<Payment> SendPaymentV2(Node node, SendPaymentRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null)
+    {
+        CustomArgumentNullException.ThrowIfNull(node.ChannelAdminMacaroon, nameof(node.ChannelAdminMacaroon), "LND Macaroon for {NodeName} is not well configured", node.Name);
+
+        client ??= GetRouterClient(node.Endpoint);
+
+        return client.SendPaymentV2(request,
+            new Metadata { { "macaroon", node.ChannelAdminMacaroon } },
+            cancellationToken: cancellationToken);
+    }
+
+    public async Task<HTLCAttempt> SendToRouteV2Async(Node node, Routerrpc.SendToRouteRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null)
+    {
+        CustomArgumentNullException.ThrowIfNull(node.ChannelAdminMacaroon, nameof(node.ChannelAdminMacaroon), "LND Macaroon for {NodeName} is not well configured", node.Name);
+
+        client ??= GetRouterClient(node.Endpoint);
+
+        return await client.SendToRouteV2Async(request,
+            new Metadata { { "macaroon", node.ChannelAdminMacaroon } },
+            cancellationToken: cancellationToken);
+    }
 }