diff --git a/src/Jobs/RebalanceJob.cs b/src/Jobs/RebalanceJob.cs new file mode 100644 index 00000000..1fd1a30b --- /dev/null +++ b/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; + +/// +/// 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. +/// +[DisallowConcurrentExecution] +public class RebalanceJob : IJob +{ + private readonly ILogger _logger; + private readonly IRebalanceService _rebalanceService; + + public RebalanceJob(ILogger 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); + } + } +} diff --git a/src/Rpc/NodeGuardService.cs b/src/Rpc/NodeGuardService.cs index c8b0a755..38867072 100644 --- a/src/Rpc/NodeGuardService.cs +++ b/src/Rpc/NodeGuardService.cs @@ -54,6 +54,12 @@ Task GetNewWalletAddress(GetNewWalletAddressRequest Task GetChannel(GetChannelRequest request, ServerCallContext context); Task AddTags(AddTagsRequest request, ServerCallContext context); + + Task RequestRebalance(RequestRebalanceRequest request, ServerCallContext context); + + Task GetRebalance(GetRebalanceRequest request, ServerCallContext context); + + Task GetRebalances(GetRebalancesRequest request, ServerCallContext context); } /// @@ -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 logger, ILiquidityRuleRepository liquidityRuleRepository, @@ -94,7 +102,9 @@ public NodeGuardService(ILogger 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 AddTags(AddTagsRequest request, Serv return new AddTagsResponse(); } + public override async Task 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 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 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 diff --git a/src/Services/LightningClientService.cs b/src/Services/LightningClientService.cs index 742302c7..96ae7c10 100644 --- a/src/Services/LightningClientService.cs +++ b/src/Services/LightningClientService.cs @@ -36,6 +36,8 @@ public interface ILightningClientService public Task ListChannels(Node node, Lightning.LightningClient? client = null); public Task ChannelBalanceAsync(Node node, Lightning.LightningClient? client = null); public Task GetChanInfo(Node node, ulong chanId, Lightning.LightningClient? client = null); + public Task AddInvoice(Node node, Invoice invoice, Lightning.LightningClient? client = null); + public Task QueryRoutes(Node node, QueryRoutesRequest request, Lightning.LightningClient? client = null); public AsyncServerStreamingCall? CloseChannel(Node node, Channel channel, bool forceClose = false, Lightning.LightningClient? client = null); public AsyncServerStreamingCall SubscribeChannelEvents(Node node, Lightning.LightningClient? client = null); public Task 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 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 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? 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 diff --git a/src/Services/LightningRouterService.cs b/src/Services/LightningRouterService.cs index 472f1653..67d89c58 100644 --- a/src/Services/LightningRouterService.cs +++ b/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 EstimateRouteFee(Node node, RouteFeeRequest routeFeeRequest, Router.RouterClient? client = null); public AsyncServerStreamingCall SubscribeHtlcEvents(Node node, Router.RouterClient? client = null); + public AsyncServerStreamingCall SendPaymentV2(Node node, SendPaymentRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null); + public Task 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 SubscribeHtlcEvents(Node node, Router } }); } + + public AsyncServerStreamingCall 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 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); + } } diff --git a/src/Services/LightningService.cs b/src/Services/LightningService.cs index 2aa882d3..4828058b 100644 --- a/src/Services/LightningService.cs +++ b/src/Services/LightningService.cs @@ -142,6 +142,49 @@ public interface ILightningService /// Optional timeout in seconds /// public Task EstimateRouteFee(string destPubkey, long amountSat, string? paymentRequest = null, uint timeout = 30); + + /// + /// Adds a self-payable invoice on the given node. Used for circular rebalancing where + /// the source node pays itself + /// + Task AddInvoiceAsync(Node node, long amountSats, string memo, long expirySeconds); + + /// + /// Probes a candidate route at the requested amount, reducing the amount until either a route + /// works (signaled by IncorrectOrUnknownPaymentDetails at the last hop) or the + /// minimum probe amount is reached. Returns the probed amount and the validated + /// route, or NoRoute if no route was found. + /// + Task ProbeRouteAsync(Node node, long amountSats, long feeLimitMsat, + ulong? outgoingChanId, string? lastHopPubkeyHex, double probeBackoffRatio, CancellationToken ct); + + /// + /// Sends a self-payment via Routerrpc.SendPaymentV2 with the given constraints and + /// drains the streaming response to the terminal Payment update. LND drives MPP and + /// internal route iteration / retries; we provide the cost cap (feeLimitMsat) and + /// the optional last-hop peer pin (lastHopPubkeyHex; null lets LND pick any inbound + /// peer). The dest is implicit in paymentRequest (self-issued invoice). + /// + Task SendPaymentV2Async(Node node, string paymentRequest, long feeLimitMsat, + ulong[]? outgoingChanIds, string? lastHopPubkeyHex, int timeoutSeconds, CancellationToken ct); + + /// + /// Returns the local-outbound fee rate (ppm) of *some* channel with the given peer. + /// Used to estimate "the fee we'd charge to forward through this peer" for ppm-cap + /// suggestions when the LND request only constrains the peer (LastHopPubkey), not + /// a specific channel. If the peer has multiple channels with us, the first one + /// returned by ListChannels is used as a representative. + /// + Task GetLocalOutboundFeeRatePpmByPeerAsync(Node node, string peerPubkey); + } + + /// + /// Result of . + /// + public abstract record ProbeResult + { + public sealed record Success(long AmountSats, Lnrpc.Route Route) : ProbeResult; + public sealed record NoRoute(string? Reason = null) : ProbeResult; } public class LightningService : ILightningService @@ -402,8 +445,8 @@ await ValidatePSBTInputsAreSpendable(channelOperationRequest, fundedPSBT, } } - var finalSignedPSBT = await SignWithInternalWallet(channelOperationRequest, fundedPSBT, derivationStrategyBase, channelfundingTx, network); - + var finalSignedPSBT = await SignWithInternalWallet(channelOperationRequest, fundedPSBT, derivationStrategyBase, channelfundingTx, network); + //Null check if (finalSignedPSBT is null) { @@ -787,9 +830,9 @@ public async Task CreateOpenChannelRequest(ChannelOperationR var upfrontShutdownScriptReq = remoteNodeInfo.Features.ContainsKey((uint)FeatureBit.UpfrontShutdownScriptReq); if (upfrontShutdownScriptOpt && remoteNodeInfo.Features[(uint)FeatureBit.UpfrontShutdownScriptOpt] is - { IsKnown: true } || + { IsKnown: true } || upfrontShutdownScriptReq && remoteNodeInfo.Features[(uint)FeatureBit.UpfrontShutdownScriptReq] is - { IsKnown: true }) + { IsKnown: true }) { var address = await GetCloseAddress(channelOperationRequest, derivationStrategyBase, _nbXplorerService, _logger); @@ -853,7 +896,7 @@ public static void CheckArgumentsAreValid(ChannelOperationRequest channelOperati OperationRequestType requestype, ILogger? _logger = null) { if (channelOperationRequest == null) throw new ArgumentNullException(nameof(channelOperationRequest)); - + //If the wallet is watch only, we cannot open a channel as there is no way securely sign the PSBT with SIGHASH_NONE (the internal wallet is not there) if (channelOperationRequest.Wallet != null && channelOperationRequest.Wallet.IsWatchOnly) { @@ -871,8 +914,8 @@ public static void CheckArgumentsAreValid(ChannelOperationRequest channelOperati $"Invalid request. Requested ${channelOperationRequest.RequestType.ToString()} on ${requestype.ToString()} method"; _logger?.LogError(requestInvalid); - - + + throw new ArgumentOutOfRangeException(requestInvalid); } @@ -1528,5 +1571,186 @@ public async Task> GetChannelsState() return null; } } + + public async Task AddInvoiceAsync(Node node, long amountSats, string memo, long expirySeconds) + { + return await _lightningClientService.AddInvoice(node, new Invoice + { + Memo = memo, + Value = amountSats, + Expiry = expirySeconds, + Private = false, + }); + } + + public async Task ProbeRouteAsync(Node node, long amountSats, long feeLimitMsat, + ulong? outgoingChanId, string? lastHopPubkeyHex, double probeBackoffRatio, CancellationToken ct) + { + // Ratio must be in the closed interval [0, 1]. 0 zeroes the next try; 1 never shrinks. + var ratio = probeBackoffRatio >= 0.0 && probeBackoffRatio <= 1.0 + ? probeBackoffRatio + : Constants.REBALANCE_PROBE_BACKOFF_RATIO; + var amount = amountSats; + while (amount >= Constants.REBALANCE_MIN_PROBE_AMOUNT_SATS) + { + ct.ThrowIfCancellationRequested(); + + // Scale fee_limit proportionally to the current probe amount so the cap + // tracks the ratio the user accepted at the requested amount. + var scaledFeeLimitMsat = amountSats > 0 + ? (long)((double)feeLimitMsat * amount / amountSats) + : feeLimitMsat; + + var routesRequest = new QueryRoutesRequest + { + Amt = amount, + FeeLimit = new FeeLimit { FixedMsat = scaledFeeLimitMsat }, + }; + + if (outgoingChanId.HasValue) + routesRequest.OutgoingChanIds.Add(outgoingChanId.Value); + + if (!string.IsNullOrEmpty(lastHopPubkeyHex)) + { + routesRequest.LastHopPubkey = ByteString.CopyFrom(Convert.FromHexString(lastHopPubkeyHex)); + routesRequest.PubKey = node.PubKey; // circular: destination is self + } + else + { + // No explicit last hop: dest must still be set; default to self for circular. + routesRequest.PubKey = node.PubKey; + } + + var routesResponse = await _lightningClientService.QueryRoutes(node, routesRequest); + if (routesResponse == null || routesResponse.Routes.Count == 0) + { + amount = (long)(amount * ratio); + continue; + } + + var routesToTry = routesResponse.Routes + .Take(Constants.REBALANCE_MAX_PROBE_ROUTES_PER_AMOUNT) + .ToList(); + + foreach (var route in routesToTry) + { + ct.ThrowIfCancellationRequested(); + + var randomHash = new byte[32]; + RandomNumberGenerator.Fill(randomHash); + + var sendToRouteRequest = new Routerrpc.SendToRouteRequest + { + PaymentHash = ByteString.CopyFrom(randomHash), + Route = route, + }; + + HTLCAttempt? attempt; + try + { + attempt = await _lightningRouterService.SendToRouteV2Async(node, sendToRouteRequest, ct); + } + catch (Exception e) + { + _logger.LogWarning(e, "Probe SendToRouteV2 failed for node {NodeId} amount {Amount}", node.Id, amount); + continue; + } + + if (attempt?.Failure?.Code == Failure.Types.FailureCode.IncorrectOrUnknownPaymentDetails) + { + // The route worked all the way to the destination; the random hash + // failed at the final hop, which is the probe success signal. + return new ProbeResult.Success(amount, route); + } + + // Mid-route failure - skip this route and try the next. + } + + amount = (long)(amount * ratio); + } + + return new ProbeResult.NoRoute("Probe exhausted before reaching minimum amount"); + } + public async Task SendPaymentV2Async(Node node, string paymentRequest, long feeLimitMsat, + ulong[]? outgoingChanIds, string? lastHopPubkeyHex, int timeoutSeconds, CancellationToken ct) + { + if (string.IsNullOrEmpty(paymentRequest)) + throw new ArgumentException("Payment request must not be empty", nameof(paymentRequest)); + + var request = new SendPaymentRequest + { + PaymentRequest = paymentRequest, + FeeLimitMsat = feeLimitMsat, + TimeoutSeconds = timeoutSeconds > 0 ? timeoutSeconds : 60, + AllowSelfPayment = true, // mandatory for circular self-pay + NoInflightUpdates = true, // we only care about the terminal Payment update + }; + + if (outgoingChanIds is { Length: > 0 }) + request.OutgoingChanIds.AddRange(outgoingChanIds); + + if (!string.IsNullOrEmpty(lastHopPubkeyHex)) + request.LastHopPubkey = ByteString.CopyFrom(Convert.FromHexString(lastHopPubkeyHex)); + + try + { + using var stream = _lightningRouterService.SendPaymentV2(node, request, ct); + Payment? terminal = null; + await foreach (var update in stream.ResponseStream.ReadAllAsync(ct)) + { + terminal = update; + if (update.Status != Payment.Types.PaymentStatus.InFlight + && update.Status != Payment.Types.PaymentStatus.Initiated) + { + break; + } + } + + return terminal ?? new Payment + { + Status = Payment.Types.PaymentStatus.Failed, + FailureReason = PaymentFailureReason.FailureReasonError, + }; + } + catch (Exception e) + { + _logger.LogWarning(e, "SendPaymentV2 failed for node {NodeId}", node.Id); + return new Payment + { + Status = Payment.Types.PaymentStatus.Failed, + FailureReason = PaymentFailureReason.FailureReasonError, + }; + } + } + + public async Task GetLocalOutboundFeeRatePpmAsync(Node node, ulong chanId) + { + var edge = await _lightningClientService.GetChanInfo(node, chanId); + if (edge == null) return null; + + // The "local" side is the policy whose pubkey matches the node we're rebalancing on. + RoutingPolicy? policy = null; + if (edge.Node1Pub == node.PubKey) policy = edge.Node1Policy; + else if (edge.Node2Pub == node.PubKey) policy = edge.Node2Policy; + + // FeeRateMilliMsat is "milli-msat per msat" which numerically equals ppm. + return policy?.FeeRateMilliMsat; + } + + public async Task GetLocalOutboundFeeRatePpmByPeerAsync(Node node, string peerPubkey) + { + if (string.IsNullOrEmpty(peerPubkey)) return null; + + var listed = await _lightningClientService.ListChannels(node); + if (listed == null) return null; + + // Find the first active channel with that peer. LND's LastHopPubkey constrains + // only the peer, not a specific channel; this is purely a representative rate. + var match = listed.Channels.FirstOrDefault(c => c.Active && c.RemotePubkey == peerPubkey) + ?? listed.Channels.FirstOrDefault(c => c.RemotePubkey == peerPubkey); + if (match == null) return null; + + return await GetLocalOutboundFeeRatePpmAsync(node, match.ChanId); + } } } \ No newline at end of file diff --git a/src/Services/RebalanceService.cs b/src/Services/RebalanceService.cs new file mode 100644 index 00000000..eace6f2e --- /dev/null +++ b/src/Services/RebalanceService.cs @@ -0,0 +1,404 @@ +/* + * 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 Lnrpc; +using NodeGuard.Data.Models; +using NodeGuard.Data.Repositories.Interfaces; +using NodeGuard.Helpers; +using NodeGuard.Jobs; +using Quartz; + +namespace NodeGuard.Services; + +public record RebalanceRequest( + int NodeId, + int? SourceChannelId, + string? TargetPubkey, + long AmountSats, + double? MaxFeePct, + int TimeoutSeconds = 60, + bool IsManual = true, + string? UserRequestorId = null, + double? ProbeBackoffRatio = null, + int? MaxAttempts = null, + double? RetryMaxFeePct = null); + +public interface IRebalanceService +{ + /// + /// Persists a new Rebalance row, runs the first attempt synchronously, and returns + /// the row in its post-attempt state. Eligible failures are queued for retry via Quartz. + /// + Task RebalanceAsync(RebalanceRequest request, CancellationToken ct = default); + + /// + /// Runs (or re-runs) the probe + payment for an existing Rebalance row. Used by + /// . + /// + Task ExecuteAsync(int rebalanceId, CancellationToken ct = default); +} + +public class RebalanceService : IRebalanceService +{ + private readonly ILogger _logger; + private readonly INodeRepository _nodeRepository; + private readonly IChannelRepository _channelRepository; + private readonly IRebalanceRepository _rebalanceRepository; + private readonly ILightningService _lightningService; + private readonly IAuditService _auditService; + private readonly ISchedulerFactory _schedulerFactory; + + public RebalanceService( + ILogger logger, + INodeRepository nodeRepository, + IChannelRepository channelRepository, + IRebalanceRepository rebalanceRepository, + ILightningService lightningService, + IAuditService auditService, + ISchedulerFactory schedulerFactory) + { + _logger = logger; + _nodeRepository = nodeRepository; + _channelRepository = channelRepository; + _rebalanceRepository = rebalanceRepository; + _lightningService = lightningService; + _auditService = auditService; + _schedulerFactory = schedulerFactory; + } + + public async Task RebalanceAsync(RebalanceRequest request, CancellationToken ct = default) + { + if (request.AmountSats <= 0) + throw new ArgumentException( + "Rebalance amount is zero — channel is already at or above the requested inbound ratio, no rebalance needed.", + nameof(request.AmountSats)); + + if (request.ProbeBackoffRatio is { } ratio && (ratio <= 0.0 || ratio >= 1.0)) + throw new ArgumentException( + "Probe backoff ratio must be in the open interval (0, 1). 0 zeroes the next try; 1 never shrinks.", + nameof(request.ProbeBackoffRatio)); + + if (request.MaxAttempts is { } attempts && attempts <= 0) + throw new ArgumentException( + "Max attempts must be at least 1.", + nameof(request.MaxAttempts)); + + if (request.RetryMaxFeePct is { } retryPct && retryPct <= 0.0) + throw new ArgumentException( + "Retry max fee % must be greater than 0.", + nameof(request.RetryMaxFeePct)); + + var node = await _nodeRepository.GetById(request.NodeId); + if (node == null) + throw new ArgumentException($"Node {request.NodeId} not found", nameof(request.NodeId)); + + ulong? sourceChanIdLnd = null; + if (request.SourceChannelId.HasValue) + { + var sourceChannel = await _channelRepository.GetById(request.SourceChannelId.Value); + if (sourceChannel == null) + throw new ArgumentException($"Source channel {request.SourceChannelId} not found"); + sourceChanIdLnd = sourceChannel.ChanId; + } + + // LastHopPubkey constrains the receiving peer, not a specific channel — LND picks + // which of that peer's channels to use. We don't accept or persist a target chan_id. + var targetPubkey = request.TargetPubkey; + + var maxFeePct = ResolveMaxFeePct(request.MaxFeePct, Constants.REBALANCE_DEFAULT_MAX_FEE_PCT); + + var rebalance = new Rebalance + { + NodeId = node.Id, + SourceNodePubKey = node.PubKey, + Status = RebalanceStatus.Pending, + IsManual = request.IsManual, + AttemptNumber = 1, + RequestedAmountSats = request.AmountSats, + SatsAmount = request.AmountSats, + MaxFeePct = maxFeePct, + SourceChannelId = request.SourceChannelId, + SourceChanIdLnd = sourceChanIdLnd, + TargetPubkey = targetPubkey, + UserRequestorId = request.UserRequestorId, + TimeoutSeconds = request.TimeoutSeconds == 0 ? 60 : request.TimeoutSeconds, + ProbeBackoffRatio = request.ProbeBackoffRatio, + MaxAttempts = request.MaxAttempts, + RetryMaxFeePct = request.RetryMaxFeePct, + }; + + var (added, addError) = await _rebalanceRepository.AddAsync(rebalance); + if (!added) + { + _logger.LogError("Failed to persist rebalance: {Error}", addError); + throw new InvalidOperationException($"Failed to persist rebalance: {addError}"); + } + + await _auditService.LogAsync( + AuditActionType.RebalanceInitiated, + AuditEventType.Attempt, + AuditObjectType.Rebalance, + rebalance.Id.ToString(), + new + { + rebalance.NodeId, + rebalance.RequestedAmountSats, + rebalance.MaxFeePct, + rebalance.SourceChanIdLnd, + rebalance.TargetPubkey, + rebalance.IsManual, + rebalance.AttemptNumber, + }); + + return await ExecuteAsync(rebalance.Id, ct); + } + + public async Task ExecuteAsync(int rebalanceId, CancellationToken ct = default) + { + var rebalance = await _rebalanceRepository.GetById(rebalanceId); + if (rebalance == null) + throw new InvalidOperationException($"Rebalance {rebalanceId} not found"); + + var node = rebalance.Node ?? await _nodeRepository.GetById(rebalance.NodeId); + if (node == null) + throw new InvalidOperationException($"Node {rebalance.NodeId} not found for rebalance {rebalanceId}"); + + try + { + var memo = $"NG rebalance #{rebalance.Id} attempt {rebalance.AttemptNumber}"; + var invoice = await _lightningService.AddInvoiceAsync(node, rebalance.SatsAmount, memo, + rebalance.TimeoutSeconds + 60); + if (invoice == null || string.IsNullOrEmpty(invoice.PaymentRequest)) + { + rebalance.Status = RebalanceStatus.Failed; + _rebalanceRepository.Update(rebalance); + await _auditService.LogAsync(AuditActionType.RebalanceCompleted, AuditEventType.Failure, + AuditObjectType.Rebalance, rebalance.Id.ToString(), + new { Reason = "Failed to create self-invoice" }); + await ScheduleRetryIfEligibleAsync(rebalance); + return rebalance; + } + + var feeLimitMsat = ComputeFeeLimitMsat(rebalance.SatsAmount, rebalance.MaxFeePct); + + rebalance.Status = RebalanceStatus.Probing; + _rebalanceRepository.Update(rebalance); + await _auditService.LogAsync(AuditActionType.RebalanceProbing, AuditEventType.Attempt, + AuditObjectType.Rebalance, rebalance.Id.ToString(), + new { rebalance.AttemptNumber, rebalance.SatsAmount, FeeLimitMsat = feeLimitMsat }); + + var probeBackoffRatio = rebalance.ProbeBackoffRatio ?? Constants.REBALANCE_PROBE_BACKOFF_RATIO; + var probe = await _lightningService.ProbeRouteAsync(node, rebalance.SatsAmount, feeLimitMsat, + rebalance.SourceChanIdLnd, rebalance.TargetPubkey, probeBackoffRatio, ct); + + if (probe is ProbeResult.NoRoute noRoute) + { + rebalance.Status = RebalanceStatus.NoRoute; + _rebalanceRepository.Update(rebalance); + await _auditService.LogAsync(AuditActionType.RebalanceProbing, AuditEventType.Failure, + AuditObjectType.Rebalance, rebalance.Id.ToString(), + new { Reason = noRoute.Reason }); + await _auditService.LogAsync(AuditActionType.RebalanceCompleted, AuditEventType.Failure, + AuditObjectType.Rebalance, rebalance.Id.ToString(), + new { rebalance.Status }); + await ScheduleRetryIfEligibleAsync(rebalance); + return rebalance; + } + + var success = (ProbeResult.Success)probe; + await _auditService.LogAsync(AuditActionType.RebalanceProbing, AuditEventType.Success, + AuditObjectType.Rebalance, rebalance.Id.ToString(), + new { ProbedAmountSats = success.AmountSats, RouteHops = success.Route.Hops.Count }); + + if (success.AmountSats < rebalance.SatsAmount) + { + rebalance.SatsAmount = success.AmountSats; + feeLimitMsat = ComputeFeeLimitMsat(rebalance.SatsAmount, rebalance.MaxFeePct); + } + + rebalance.Status = RebalanceStatus.InFlight; + _rebalanceRepository.Update(rebalance); + + var outgoingChanIds = rebalance.SourceChanIdLnd.HasValue + ? [rebalance.SourceChanIdLnd.Value] + : Array.Empty(); + + // Probe gave us feasibility + (possibly shrunk) amount; the actual settle goes + // through LND's full pathfinder via SendPaymentV2 — that gives us MPP and + // built-in per-route retries inside one payment, which the SendToRouteV2 path + // didn't. The probed Route itself is informational only. + var payment = await _lightningService.SendPaymentV2Async( + node, + invoice.PaymentRequest, + feeLimitMsat, + outgoingChanIds, + rebalance.TargetPubkey, + rebalance.TimeoutSeconds, + ct); + + ApplyTerminalPayment(rebalance, payment); + + // Defensive post-hoc ppm guard. Should never fire because feeLimitMsat clamps, + // but if it does we surface a distinct status for operators / Grafana. + var maxFeePpmCap = (long)Math.Round(rebalance.MaxFeePct * 10_000d, MidpointRounding.AwayFromZero); + if (rebalance.Status == RebalanceStatus.Succeeded + && rebalance.EffectivePpm.HasValue + && rebalance.EffectivePpm.Value > maxFeePpmCap) + { + rebalance.Status = RebalanceStatus.ExceededFeeLimit; + } + + _rebalanceRepository.Update(rebalance); + + if (rebalance.Status == RebalanceStatus.Succeeded) + { + await _auditService.LogAsync(AuditActionType.RebalanceCompleted, AuditEventType.Success, + AuditObjectType.Rebalance, rebalance.Id.ToString(), + new + { + rebalance.FeePaidSats, + rebalance.EffectivePpm, + ActualAmountSats = rebalance.SatsAmount, + rebalance.AttemptNumber, + }); + } + else + { + await _auditService.LogAsync(AuditActionType.RebalanceCompleted, AuditEventType.Failure, + AuditObjectType.Rebalance, rebalance.Id.ToString(), + new + { + rebalance.Status, + FailureReason = payment.FailureReason.ToString(), + }); + await ScheduleRetryIfEligibleAsync(rebalance); + } + + return rebalance; + } + catch (Exception ex) + { + _logger.LogError(ex, "Rebalance {Id} attempt {Attempt} failed unexpectedly", + rebalance.Id, rebalance.AttemptNumber); + rebalance.Status = RebalanceStatus.Failed; + _rebalanceRepository.Update(rebalance); + await _auditService.LogAsync(AuditActionType.RebalanceCompleted, AuditEventType.Failure, + AuditObjectType.Rebalance, rebalance.Id.ToString(), + new { ExceptionType = ex.GetType().Name, Message = ex.Message }); + await ScheduleRetryIfEligibleAsync(rebalance); + return rebalance; + } + } + + /// + /// Resolves the max fee percentage to use. Uses the caller-supplied value when positive, + /// otherwise the supplied default. Convention: 0.05 means 0.05%. + /// + private static double ResolveMaxFeePct(double? userSupplied, decimal defaultPct) + => userSupplied is { } v && v > 0 ? v : (double)defaultPct; + + /// + /// fee_msat = sats × (pct / 100) × 1000 = sats × pct × 10. Decimal math, banker-safe rounding. + /// + private static long ComputeFeeLimitMsat(long satsAmount, double maxFeePct) + => (long)Math.Round(satsAmount * (decimal)maxFeePct * 10m, MidpointRounding.AwayFromZero); + + + private static void ApplyTerminalPayment(Rebalance rebalance, Payment payment) + { + switch (payment.Status) + { + case Payment.Types.PaymentStatus.Succeeded: + rebalance.Status = RebalanceStatus.Succeeded; + rebalance.FeePaidMsat = payment.FeeMsat; + rebalance.FeePaidSats = payment.FeeMsat / 1_000L; + rebalance.PreimageHex = payment.PaymentPreimage; + break; + case Payment.Types.PaymentStatus.Failed: + rebalance.Status = payment.FailureReason switch + { + PaymentFailureReason.FailureReasonNoRoute => RebalanceStatus.NoRoute, + PaymentFailureReason.FailureReasonTimeout => RebalanceStatus.Timeout, + PaymentFailureReason.FailureReasonInsufficientBalance => RebalanceStatus.InsufficientBalance, + _ => RebalanceStatus.Failed, + }; + break; + default: + rebalance.Status = RebalanceStatus.Failed; + break; + } + } + + private async Task ScheduleRetryIfEligibleAsync(Rebalance rebalance) + { + if (rebalance.Status is RebalanceStatus.Succeeded + or RebalanceStatus.InsufficientBalance + or RebalanceStatus.ExceededFeeLimit) + { + return; + } + + var maxAttempts = rebalance.MaxAttempts ?? Constants.REBALANCE_MAX_ATTEMPTS; + if (rebalance.AttemptNumber >= maxAttempts) + { + return; + } + + var nextAttempt = rebalance.AttemptNumber + 1; + var retryDefaultPct = rebalance.RetryMaxFeePct.HasValue + ? (decimal)rebalance.RetryMaxFeePct.Value + : Constants.REBALANCE_DEFAULT_RETRY_MAX_FEE_PCT; + var retryPct = ResolveMaxFeePct(null, retryDefaultPct); + + rebalance.AttemptNumber = nextAttempt; + rebalance.MaxFeePct = Math.Max(rebalance.MaxFeePct, retryPct); + rebalance.Status = RebalanceStatus.Pending; + _rebalanceRepository.Update(rebalance); + + var delaySeconds = Constants.REBALANCE_INITIAL_RETRY_DELAY_SECONDS + * Math.Pow(Constants.REBALANCE_RETRY_BACKOFF_MULTIPLIER, nextAttempt - 2); + var fireAt = DateTimeOffset.UtcNow.AddSeconds(delaySeconds); + + var scheduler = await _schedulerFactory.GetScheduler(); + var jobData = new JobDataMap(); + jobData.Put("rebalanceId", rebalance.Id); + + var job = JobBuilder.Create() + .WithIdentity($"{nameof(RebalanceJob)}-{rebalance.Id}-a{nextAttempt}") + .UsingJobData(jobData) + .RequestRecovery() + .Build(); + + var trigger = TriggerBuilder.Create() + .WithIdentity($"{nameof(RebalanceJob)}Trigger-{rebalance.Id}-a{nextAttempt}") + .StartAt(fireAt) + .Build(); + + await scheduler.ScheduleJob(job, trigger); + + await _auditService.LogAsync(AuditActionType.RebalanceRetryScheduled, AuditEventType.Attempt, + AuditObjectType.Rebalance, rebalance.Id.ToString(), + new + { + NewAttemptNumber = nextAttempt, + NewMaxFeePct = rebalance.MaxFeePct, + FireAt = fireAt, + }); + } +} diff --git a/test/NodeGuard.Tests/Rpc/NodeGuardServiceTests.cs b/test/NodeGuard.Tests/Rpc/NodeGuardServiceTests.cs index 8037b704..d972dad9 100644 --- a/test/NodeGuard.Tests/Rpc/NodeGuardServiceTests.cs +++ b/test/NodeGuard.Tests/Rpc/NodeGuardServiceTests.cs @@ -75,7 +75,9 @@ private NodeGuardService CreateNodeGuardService( ILightningService? lightningService = null, IFMUTXORepository? fmutxoRepository = null, IUTXOTagRepository? utxoTagRepository = null, - IHtlcMonitoringScheduler? htlcMonitoringScheduler = null) + IHtlcMonitoringScheduler? htlcMonitoringScheduler = null, + IRebalanceService? rebalanceService = null, + IRebalanceRepository? rebalanceRepository = null) { return new NodeGuardService( logger ?? _logger.Object, @@ -93,7 +95,9 @@ private NodeGuardService CreateNodeGuardService( lightningService ?? new Mock().Object, fmutxoRepository ?? new Mock().Object, utxoTagRepository ?? new Mock().Object, - htlcMonitoringScheduler ?? CreateHtlcMonitoringSchedulerMock().Object); + htlcMonitoringScheduler ?? CreateHtlcMonitoringSchedulerMock().Object, + rebalanceService ?? new Mock().Object, + rebalanceRepository ?? new Mock().Object); } [Fact] @@ -1283,5 +1287,301 @@ await act .WithMessage( "Status(StatusCode=\"Internal\", Detail=\"You can't select wallets by type and id at the same time\")"); } + + [Fact] + public async Task RequestRebalance_NodePubkeyMissing_ThrowsInvalidArgument() + { + var service = CreateNodeGuardService(); + var request = new RequestRebalanceRequest { NodePubkey = "", AmountSats = 1000 }; + + var act = () => service.RequestRebalance(request, TestServerCallContext.Create()); + + (await act.Should().ThrowAsync()) + .Which.StatusCode.Should().Be(StatusCode.InvalidArgument); + } + + [Fact] + public async Task RequestRebalance_AmountSatsNotPositive_ThrowsInvalidArgument() + { + var service = CreateNodeGuardService(); + var request = new RequestRebalanceRequest { NodePubkey = "pubkey1", AmountSats = 0 }; + + var act = () => service.RequestRebalance(request, TestServerCallContext.Create()); + + (await act.Should().ThrowAsync()) + .Which.StatusCode.Should().Be(StatusCode.InvalidArgument); + } + + [Fact] + public async Task RequestRebalance_NodeNotFound_ThrowsNotFound() + { + var nodeRepoMock = new Mock(); + nodeRepoMock.Setup(r => r.GetByPubkey("pubkey1")).ReturnsAsync((Node?)null); + + var service = CreateNodeGuardService(nodeRepository: nodeRepoMock.Object); + var request = new RequestRebalanceRequest { NodePubkey = "pubkey1", AmountSats = 1000 }; + + var act = () => service.RequestRebalance(request, TestServerCallContext.Create()); + + (await act.Should().ThrowAsync()) + .Which.StatusCode.Should().Be(StatusCode.NotFound); + } + + [Fact] + public async Task RequestRebalance_Success_PassesOptionalsAndReturnsResponse() + { + var node = new Node { Id = 42, PubKey = "pubkey1" }; + var nodeRepoMock = new Mock(); + nodeRepoMock.Setup(r => r.GetByPubkey("pubkey1")).ReturnsAsync(node); + + RebalanceRequest? capturedRequest = null; + var rebalanceServiceMock = new Mock(); + rebalanceServiceMock + .Setup(s => s.RebalanceAsync(It.IsAny(), It.IsAny())) + .Callback((r, _) => capturedRequest = r) + .ReturnsAsync(new Rebalance + { + Id = 7, + NodeId = node.Id, + SourceNodePubKey = node.PubKey, + Status = RebalanceStatus.Succeeded, + AttemptNumber = 1, + RequestedAmountSats = 1000, + SatsAmount = 950, + FeePaidSats = 3, + FeePaidMsat = 3000, + }); + + var service = CreateNodeGuardService( + nodeRepository: nodeRepoMock.Object, + rebalanceService: rebalanceServiceMock.Object); + + var request = new RequestRebalanceRequest + { + NodePubkey = "pubkey1", + AmountSats = 1000, + MaxFeePct = 0.05, + RetryMaxFeePct = 0.1, + SourceChannelId = 11, + TargetPubkey = "peer-pubkey", + TimeoutSeconds = 90, + ProbeBackoffRatio = 0.5, + MaxAttempts = 4, + }; + + var response = await service.RequestRebalance(request, TestServerCallContext.Create()); + + response.RebalanceId.Should().Be(7); + response.Status.Should().Be(REBALANCE_STATUS.RebalanceSucceeded); + response.AttemptNumber.Should().Be(1); + response.FeePaidSats.Should().Be(3); + response.HasActualAmountSats.Should().BeTrue(); + response.ActualAmountSats.Should().Be(950); + + capturedRequest.Should().NotBeNull(); + capturedRequest!.NodeId.Should().Be(node.Id); + capturedRequest.AmountSats.Should().Be(1000); + capturedRequest.MaxFeePct.Should().Be(0.05); + capturedRequest.RetryMaxFeePct.Should().Be(0.1); + capturedRequest.SourceChannelId.Should().Be(11); + capturedRequest.TargetPubkey.Should().Be("peer-pubkey"); + capturedRequest.TimeoutSeconds.Should().Be(90); + capturedRequest.ProbeBackoffRatio.Should().Be(0.5); + capturedRequest.MaxAttempts.Should().Be(4); + capturedRequest.IsManual.Should().BeTrue(); + } + + [Fact] + public async Task RequestRebalance_DomainArgumentException_ThrowsInvalidArgument() + { + var node = new Node { Id = 1, PubKey = "pubkey1" }; + var nodeRepoMock = new Mock(); + nodeRepoMock.Setup(r => r.GetByPubkey("pubkey1")).ReturnsAsync(node); + + var rebalanceServiceMock = new Mock(); + rebalanceServiceMock + .Setup(s => s.RebalanceAsync(It.IsAny(), It.IsAny())) + .ThrowsAsync(new ArgumentException("bad ratio")); + + var service = CreateNodeGuardService( + nodeRepository: nodeRepoMock.Object, + rebalanceService: rebalanceServiceMock.Object); + + var request = new RequestRebalanceRequest { NodePubkey = "pubkey1", AmountSats = 1000 }; + + var act = () => service.RequestRebalance(request, TestServerCallContext.Create()); + + (await act.Should().ThrowAsync()) + .Which.StatusCode.Should().Be(StatusCode.InvalidArgument); + } + + [Fact] + public async Task GetRebalance_InvalidId_ThrowsInvalidArgument() + { + var service = CreateNodeGuardService(); + var request = new GetRebalanceRequest { RebalanceId = 0 }; + + var act = () => service.GetRebalance(request, TestServerCallContext.Create()); + + (await act.Should().ThrowAsync()) + .Which.StatusCode.Should().Be(StatusCode.InvalidArgument); + } + + [Fact] + public async Task GetRebalance_NotFound_ThrowsNotFound() + { + var rebalanceRepoMock = new Mock(); + rebalanceRepoMock.Setup(r => r.GetById(99)).ReturnsAsync((Rebalance?)null); + + var service = CreateNodeGuardService(rebalanceRepository: rebalanceRepoMock.Object); + + var act = () => service.GetRebalance(new GetRebalanceRequest { RebalanceId = 99 }, + TestServerCallContext.Create()); + + (await act.Should().ThrowAsync()) + .Which.StatusCode.Should().Be(StatusCode.NotFound); + } + + [Fact] + public async Task GetRebalance_Found_ReturnsMappedResponse() + { + var rebalance = new Rebalance + { + Id = 5, + NodeId = 1, + SourceNodePubKey = "pubkey1", + Status = RebalanceStatus.Succeeded, + IsManual = true, + AttemptNumber = 2, + RequestedAmountSats = 1000, + SatsAmount = 950, + MaxFeePct = 0.05, + RetryMaxFeePct = 0.1, + FeePaidSats = 3, + FeePaidMsat = 3000, + SourceChanIdLnd = 123456UL, + TargetPubkey = "peer-pubkey", + ProbeBackoffRatio = 0.5, + MaxAttempts = 4, + CreationDatetime = DateTimeOffset.FromUnixTimeSeconds(1_700_000_000), + UpdateDatetime = DateTimeOffset.FromUnixTimeSeconds(1_700_000_100), + }; + + var rebalanceRepoMock = new Mock(); + rebalanceRepoMock.Setup(r => r.GetById(5)).ReturnsAsync(rebalance); + + var service = CreateNodeGuardService(rebalanceRepository: rebalanceRepoMock.Object); + + var response = await service.GetRebalance(new GetRebalanceRequest { RebalanceId = 5 }, + TestServerCallContext.Create()); + + response.RebalanceId.Should().Be(5); + response.NodePubkey.Should().Be("pubkey1"); + response.Status.Should().Be(REBALANCE_STATUS.RebalanceSucceeded); + response.IsManual.Should().BeTrue(); + response.AttemptNumber.Should().Be(2); + response.RequestedAmountSats.Should().Be(1000); + response.AmountSats.Should().Be(950); + response.MaxFeePct.Should().Be(0.05); + response.RetryMaxFeePct.Should().Be(0.1); + response.FeePaidSats.Should().Be(3); + response.EffectivePpm.Should().Be(3000L * 1000L / 950L); + response.SourceChanId.Should().Be(123456UL); + response.TargetPubkey.Should().Be("peer-pubkey"); + response.ProbeBackoffRatio.Should().Be(0.5); + response.MaxAttempts.Should().Be(4); + response.CreationDatetimeUnix.Should().Be(1_700_000_000); + response.UpdateDatetimeUnix.Should().Be(1_700_000_100); + } + + [Fact] + public async Task GetRebalances_InvalidPageNumber_ThrowsInvalidArgument() + { + var service = CreateNodeGuardService(); + var request = new GetRebalancesRequest { PageNumber = 0, PageSize = 10 }; + + var act = () => service.GetRebalances(request, TestServerCallContext.Create()); + + (await act.Should().ThrowAsync()) + .Which.StatusCode.Should().Be(StatusCode.InvalidArgument); + } + + [Fact] + public async Task GetRebalances_UnknownNodePubkey_ReturnsEmptyPageWithoutQueryingRepo() + { + var nodeRepoMock = new Mock(); + nodeRepoMock.Setup(r => r.GetByPubkey("missing")).ReturnsAsync((Node?)null); + + var rebalanceRepoMock = new Mock(); + + var service = CreateNodeGuardService( + nodeRepository: nodeRepoMock.Object, + rebalanceRepository: rebalanceRepoMock.Object); + + var response = await service.GetRebalances( + new GetRebalancesRequest { PageNumber = 1, PageSize = 10, NodePubkey = "missing" }, + TestServerCallContext.Create()); + + response.TotalCount.Should().Be(0); + response.Rebalances.Should().BeEmpty(); + rebalanceRepoMock.Verify(r => r.GetPaginatedAsync( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task GetRebalances_ResolvesNodePubkeyAndForwardsFilters() + { + var node = new Node { Id = 42, PubKey = "pubkey1" }; + var nodeRepoMock = new Mock(); + nodeRepoMock.Setup(r => r.GetByPubkey("pubkey1")).ReturnsAsync(node); + + var rebalances = new List + { + new Rebalance + { + Id = 1, NodeId = node.Id, SourceNodePubKey = node.PubKey, + Status = RebalanceStatus.Succeeded, AttemptNumber = 1, + RequestedAmountSats = 100, SatsAmount = 100, MaxFeePct = 0.05, + CreationDatetime = DateTimeOffset.FromUnixTimeSeconds(1_700_000_000), + UpdateDatetime = DateTimeOffset.FromUnixTimeSeconds(1_700_000_050), + }, + }; + var rebalanceRepoMock = new Mock(); + rebalanceRepoMock + .Setup(r => r.GetPaginatedAsync( + 1, 10, + RebalanceStatus.Succeeded, + node.Id, + 77, + "user-1", + true, + It.IsAny(), + It.IsAny())) + .ReturnsAsync((rebalances, 1)); + + var service = CreateNodeGuardService( + nodeRepository: nodeRepoMock.Object, + rebalanceRepository: rebalanceRepoMock.Object); + + var response = await service.GetRebalances(new GetRebalancesRequest + { + PageNumber = 1, + PageSize = 10, + Status = REBALANCE_STATUS.RebalanceSucceeded, + NodePubkey = "pubkey1", + SourceChannelId = 77, + UserId = "user-1", + IsManual = true, + FromUnix = 1_699_000_000, + ToUnix = 1_701_000_000, + }, TestServerCallContext.Create()); + + response.TotalCount.Should().Be(1); + response.Rebalances.Should().HaveCount(1); + response.Rebalances[0].RebalanceId.Should().Be(1); + response.Rebalances[0].NodePubkey.Should().Be("pubkey1"); + } } }