fedimint_gateway_server/

lib.rs

#![deny(clippy::pedantic)]
#![allow(clippy::cast_possible_truncation)]
#![allow(clippy::cast_possible_wrap)]
#![allow(clippy::cast_sign_loss)]
#![allow(clippy::default_trait_access)]
#![allow(clippy::doc_markdown)]
#![allow(clippy::missing_errors_doc)]
#![allow(clippy::missing_panics_doc)]
#![allow(clippy::module_name_repetitions)]
#![allow(clippy::must_use_candidate)]
#![allow(clippy::return_self_not_must_use)]
#![allow(clippy::similar_names)]
#![allow(clippy::too_many_lines)]
#![allow(clippy::large_futures)]
#![allow(clippy::struct_field_names)]

pub mod client;
pub mod config;
pub mod envs;
mod error;
mod events;
mod federation_manager;
mod federation_status;
mod iroh_server;
mod metrics;
mod rate_limit;
mod registration_health;
pub mod rpc_server;
mod types;

use std::collections::{BTreeMap, BTreeSet};
use std::env;
use std::fmt::Display;
use std::net::SocketAddr;
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, UNIX_EPOCH};

use anyhow::{Context, anyhow};
use async_trait::async_trait;
use bitcoin::hashes::sha256;
use bitcoin::{Address, Network, Txid, secp256k1};
use clap::Parser;
use client::GatewayClientBuilder;
pub use config::GatewayParameters;
use config::{DatabaseBackend, GatewayOpts};
use envs::FM_GATEWAY_SKIP_WAIT_FOR_SYNC_ENV;
use error::FederationNotConnected;
use events::ALL_GATEWAY_EVENTS;
use federation_manager::FederationManager;
use fedimint_bip39::{Bip39RootSecretStrategy, Language, Mnemonic};
use fedimint_bitcoind::bitcoincore::BitcoindClient;
use fedimint_bitcoind::{EsploraClient, IBitcoindRpc};
use fedimint_client::module_init::ClientModuleInitRegistry;
use fedimint_client::secret::RootSecretStrategy;
use fedimint_client::{Client, ClientHandleArc};
use fedimint_core::base32::{self, FEDIMINT_PREFIX};
use fedimint_core::config::FederationId;
use fedimint_core::core::OperationId;
use fedimint_core::db::{Committable, Database, DatabaseTransaction, apply_migrations};
use fedimint_core::envs::is_env_var_set;
use fedimint_core::invite_code::InviteCode;
use fedimint_core::module::CommonModuleInit;
use fedimint_core::module::registry::ModuleDecoderRegistry;
use fedimint_core::rustls::install_crypto_provider;
use fedimint_core::secp256k1::PublicKey;
use fedimint_core::secp256k1::schnorr::Signature;
use fedimint_core::task::{TaskGroup, TaskHandle, TaskShutdownToken, sleep, timeout};
use fedimint_core::time::duration_since_epoch;
use fedimint_core::util::backoff_util::fibonacci_max_one_hour;
use fedimint_core::util::{FmtCompact, SafeUrl, Spanned, retry};
use fedimint_core::{
    Amount, BitcoinAmountOrAll, PeerId, TieredCounts, crit, fedimint_build_code_version_env,
    get_network_for_address,
};
use fedimint_eventlog::{DBTransactionEventLogExt, EventLogId, StructuredPaymentEvents};
use fedimint_gateway_common::{
    BackupPayload, ChainSource, CloseChannelsWithPeerRequest, CloseChannelsWithPeerResponse,
    ConnectFedPayload, ConnectPeerRequest, ConnectorType, CreateInvoiceForOperatorPayload,
    CreateOfferPayload, CreateOfferResponse, DepositAddressPayload, DepositAddressRecheckPayload,
    FederationBalanceInfo, FederationConfig, FederationInfo, GatewayBalances, GatewayFedConfig,
    GatewayInfo, GetInvoiceRequest, GetInvoiceResponse, LeaveFedPayload, LightningInfo,
    LightningMode, ListTransactionsPayload, ListTransactionsResponse, MnemonicResponse,
    OpenChannelRequest, PayInvoiceForOperatorPayload, PayOfferPayload, PayOfferResponse,
    PaymentLogPayload, PaymentLogResponse, PaymentPolicy, PaymentPolicyGuard, PaymentStats,
    PaymentSummaryPayload, PaymentSummaryResponse, PeginFromOnchainPayload, ReceiveEcashPayload,
    ReceiveEcashResponse, RegisteredProtocol, SendOnchainRequest, SetChannelFeesRequest,
    SetFeesPayload, SetMnemonicPayload, SetPaymentPolicyPayload, SpendEcashPayload,
    SpendEcashResponse, V1_API_ENDPOINT, WithdrawPayload, WithdrawPreviewPayload,
    WithdrawPreviewResponse, WithdrawResponse, WithdrawToOnchainPayload,
};
use fedimint_gateway_server_db::{GatewayDbtxNcExt as _, get_gatewayd_database_migrations};
pub use fedimint_gateway_ui::IAdminGateway;
use fedimint_gw_client::events::compute_lnv1_stats;
use fedimint_gw_client::pay::{OutgoingPaymentError, OutgoingPaymentErrorType};
use fedimint_gw_client::{
    GatewayClientModule, GatewayClientV1Error, GatewayExtPayStates, GatewayExtReceiveStates, Htlc,
    IGatewayClientV1, SwapParameters,
};
use fedimint_gwv2_client::events::compute_lnv2_stats;
use fedimint_gwv2_client::{
    EXPIRATION_DELTA_MINIMUM_V2, FinalReceiveState, GatewayClientModuleV2, GatewayClientV2Error,
    IGatewayClientV2,
};
use fedimint_lightning::ldk::MnemonicSource;
use fedimint_lightning::lnd::GatewayLndClient;
use fedimint_lightning::{
    CreateInvoiceRequest, ILnRpcClient, InterceptPaymentRequest, InterceptPaymentResponse,
    InvoiceDescription, LightningContext, LightningRpcError, LnRpcTracked, Lnv2HoldInvoiceFilter,
    PayInvoiceResponse, PaymentAction, RouteHtlcStream, ldk,
};
use fedimint_ln_client::pay::PaymentData;
use fedimint_ln_common::config::LightningClientConfig;
use fedimint_ln_common::contracts::outgoing::OutgoingContractAccount;
use fedimint_ln_common::contracts::{IdentifiableContract, Preimage};
use fedimint_ln_common::{LightningCommonInit, PreimageAuth};
use fedimint_lnurl::VerifyResponse;
use fedimint_lnv2_common::Bolt11InvoiceDescription;
use fedimint_lnv2_common::contracts::{IncomingContract, PaymentImage};
use fedimint_lnv2_common::gateway_api::{
    CreateBolt11InvoicePayload, MAX_INVOICE_EXPIRY_SECS, PaymentFee, RoutingInfo,
    SendPaymentPayload,
};
use fedimint_logging::LOG_GATEWAY;
use fedimint_mint_client::{MintClientInit, MintClientModule, OOBNotes, ReissueExternalNotesState};
use fedimint_mintv2_client::{
    MintClientInit as MintV2ClientInit, MintClientModule as MintV2ClientModule,
};
use fedimint_wallet_client::{PegOutFees, WalletClientInit, WalletClientModule, WithdrawState};
use futures::stream::StreamExt;
use lightning_invoice::{Bolt11Invoice, RoutingFees};
use rand::rngs::OsRng;
use tokio::sync::RwLock;
use tracing::{debug, info, info_span, warn};

use crate::envs::FM_GATEWAY_MNEMONIC_ENV;
use crate::error::{AdminGatewayError, LNv1Error, LNv2Error, PublicGatewayError};
use crate::events::get_events_for_duration;
use crate::rate_limit::TokenBucketRateLimiter;
use crate::registration_health::RegistrationHealthTracker;
use crate::rpc_server::run_webserver;
use crate::types::PrettyInterceptPaymentRequest;

/// How long a gateway announcement stays valid
const GW_ANNOUNCEMENT_TTL: Duration = Duration::from_mins(10);

/// The default number of route hints that the legacy gateway provides for
/// invoice creation.
const DEFAULT_NUM_ROUTE_HINTS: u32 = 1;

/// Default maximum burst of requests to the public invoice creation endpoint.
const DEFAULT_INVOICE_RATE_LIMIT_BURST: u32 = 50;

/// Default sustained number of requests per second to the public invoice
/// creation endpoint.
const DEFAULT_INVOICE_RATE_LIMIT_PER_SECOND: u32 = 5;

/// Default Bitcoin network for testing purposes.
pub const DEFAULT_NETWORK: Network = Network::Regtest;

/// How long code that needs to talk to the lightning node backs off before
/// re-checking whether the gateway has (re)connected to it.
const LIGHTNING_CONTEXT_RETRY_INTERVAL: Duration = Duration::from_secs(5);

/// How long an LNURL-verify request that asked to wait blocks before reporting
/// the payment as not settled yet.
///
/// The payment being waited for may never arrive, so the wait needs an upper
/// bound to stop callers from parking a request handler indefinitely. Reporting
/// "not settled" on expiry is a regular LNURL-verify response, so clients
/// simply poll again.
const VERIFY_WAIT_TIMEOUT: Duration = Duration::from_secs(30);

pub type Result<T> = std::result::Result<T, PublicGatewayError>;
pub type AdminResult<T> = std::result::Result<T, AdminGatewayError>;

/// Name of the gateway's database that is used for metadata and configuration
/// storage.
const DB_FILE: &str = "gatewayd.db";

/// Name of the folder that the gateway uses to store its node database when
/// running in LDK mode.
const LDK_NODE_DB_FOLDER: &str = "ldk_node";

/// ```mermaid
/// graph LR
/// classDef virtual fill:#fff,stroke-dasharray: 5 5
///
///    NotConfigured -- create or recover wallet --> Disconnected
///    Disconnected -- establish lightning connection --> Connected
///    Connected -- load federation clients --> Running
///    Connected -- not synced to chain --> Syncing
///    Syncing -- load federation clients --> Running
///    Running -- disconnected from lightning node --> Disconnected
///    Running -- shutdown initiated --> ShuttingDown
/// ```
#[derive(Clone, Debug)]
pub enum GatewayState {
    NotConfigured {
        // Broadcast channel to alert gateway background threads that the mnemonic has been
        // created/set, and how it was obtained.
        mnemonic_sender: tokio::sync::broadcast::Sender<MnemonicSource>,
    },
    Disconnected,
    Syncing,
    Connected,
    Running {
        lightning_context: LightningContext,
    },
    ShuttingDown {
        lightning_context: LightningContext,
    },
}

impl Display for GatewayState {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        match self {
            GatewayState::NotConfigured { .. } => write!(f, "NotConfigured"),
            GatewayState::Disconnected => write!(f, "Disconnected"),
            GatewayState::Syncing => write!(f, "Syncing"),
            GatewayState::Connected => write!(f, "Connected"),
            GatewayState::Running { .. } => write!(f, "Running"),
            GatewayState::ShuttingDown { .. } => write!(f, "ShuttingDown"),
        }
    }
}

/// Helper struct for storing the registration parameters for LNv1 for each
/// network protocol.
#[derive(Debug, Clone)]
struct Registration {
    /// The url to advertise in the registration that clients can use to connect
    endpoint_url: SafeUrl,

    /// Keypair that was used to register the gateway registration
    keypair: secp256k1::Keypair,
}

impl Registration {
    pub async fn new(db: &Database, endpoint_url: SafeUrl, protocol: RegisteredProtocol) -> Self {
        let keypair = Gateway::load_or_create_gateway_keypair(db, protocol).await;
        Self {
            endpoint_url,
            keypair,
        }
    }
}

#[bon::bon]
impl Gateway {
    /// Construct a [`Gateway`] using a fluent builder API.
    ///
    /// # Example
    /// ```ignore
    /// let gateway = Gateway::builder(lightning_mode, client_builder, gateway_db)
    ///     .listen(addr)
    ///     .api_addr(url)
    ///     .bcrypt_password_hash(hash)
    ///     .network(Network::Regtest)
    ///     .gateway_state(state)
    ///     .chain_source(chain_source)
    ///     .build()
    ///     .await?;
    /// ```
    #[builder(start_fn = builder, finish_fn = build)]
    pub async fn new_with_builder(
        #[builder(start_fn)] lightning_mode: LightningMode,
        #[builder(start_fn)] client_builder: GatewayClientBuilder,
        #[builder(start_fn)] gateway_db: Database,
        bcrypt_password_hash: bcrypt::HashParts,
        bcrypt_liquidity_manager_password_hash: Option<bcrypt::HashParts>,
        gateway_state: GatewayState,
        chain_source: ChainSource,
        #[builder(default = ([127, 0, 0, 1], 80).into())] listen: SocketAddr,
        api_addr: Option<SafeUrl>,
        #[builder(default = DEFAULT_NETWORK)] network: Network,
        #[builder(default = DEFAULT_NUM_ROUTE_HINTS)] num_route_hints: u32,
        #[builder(default = PaymentFee::TRANSACTION_FEE_DEFAULT)] default_routing_fees: PaymentFee,
        #[builder(default = PaymentFee::TRANSACTION_FEE_DEFAULT)]
        default_transaction_fees: PaymentFee,
        iroh_listen: Option<SocketAddr>,
        iroh_dns: Option<SafeUrl>,
        #[builder(default)] iroh_relays: Vec<SafeUrl>,
        metrics_listen: Option<SocketAddr>,
    ) -> anyhow::Result<Gateway> {
        let versioned_api = api_addr.map(|addr| {
            addr.join(V1_API_ENDPOINT)
                .expect("Failed to version gateway API address")
        });

        let metrics_listen = metrics_listen.unwrap_or_else(|| {
            SocketAddr::new(
                std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST),
                listen.port() + 1,
            )
        });

        Gateway::new(
            lightning_mode,
            GatewayParameters {
                listen,
                versioned_api,
                bcrypt_password_hash,
                bcrypt_liquidity_manager_password_hash,
                network,
                num_route_hints,
                default_routing_fees,
                default_transaction_fees,
                iroh_listen,
                iroh_dns,
                iroh_relays,
                skip_setup: true,
                metrics_listen,
                invoice_rate_limit_burst: DEFAULT_INVOICE_RATE_LIMIT_BURST,
                invoice_rate_limit_per_second: DEFAULT_INVOICE_RATE_LIMIT_PER_SECOND,
            },
            gateway_db,
            client_builder,
            gateway_state,
            chain_source,
            MnemonicSource::Existing,
        )
        .await
    }
}

/// The action to take after handling a payment stream.
enum ReceivePaymentStreamAction {
    RetryAfterDelay,
    NoRetry,
}

#[derive(Clone)]
pub struct Gateway {
    /// The gateway's federation manager.
    federation_manager: Arc<RwLock<FederationManager>>,

    /// The mode that specifies the lightning connection parameters
    lightning_mode: LightningMode,

    /// The current state of the Gateway.
    state: Arc<RwLock<GatewayState>>,

    /// Builder struct that allows the gateway to build a Fedimint client, which
    /// handles the communication with a federation.
    client_builder: GatewayClientBuilder,

    /// Database for Gateway metadata.
    gateway_db: Database,

    /// The socket the gateway listens on.
    listen: SocketAddr,

    /// The socket the gateway's metrics server listens on.
    metrics_listen: SocketAddr,

    /// The task group for all tasks related to the gateway.
    task_group: TaskGroup,

    /// The bcrypt password hash used to authenticate the gateway.
    bcrypt_password_hash: String,

    /// The bcrypt password hash used to authenticate the gateway liquidity
    /// manager.
    bcrypt_liquidity_manager_password_hash: Option<String>,

    /// The number of route hints to include in LNv1 invoices.
    num_route_hints: u32,

    /// The Bitcoin network that the Lightning network is configured to.
    network: Network,

    /// The source of the Bitcoin blockchain data
    chain_source: ChainSource,

    /// The default routing fees for new federations
    default_routing_fees: PaymentFee,

    /// The default transaction fees for new federations
    default_transaction_fees: PaymentFee,

    /// The secret key for the Iroh `Endpoint`
    iroh_sk: iroh::SecretKey,

    /// The socket that the gateway listens on for the Iroh `Endpoint`
    iroh_listen: Option<SocketAddr>,

    /// Optional DNS server used for discovery of the Iroh `Endpoint`
    iroh_dns: Option<SafeUrl>,

    /// List of additional relays that can be used to establish a connection to
    /// the Iroh `Endpoint`
    iroh_relays: Vec<SafeUrl>,

    /// A map of the network protocols the gateway supports to the data needed
    /// for registering with a federation.
    registrations: BTreeMap<RegisteredProtocol, Registration>,

    /// Detail-free retained results of LNv1 federation registration attempts.
    registration_health: RegistrationHealthTracker,

    /// Rate limiter for the public invoice creation endpoint.
    invoice_rate_limiter: Arc<TokenBucketRateLimiter>,

    /// Locks the payment hash of each HOLD invoice whose delivery is being
    /// handled, so that a duplicate delivery is dropped; see
    /// [`Self::handle_lightning_payment`].
    lnv2_hold_invoice_lock_pool: Arc<lockable::LockPool<sha256::Hash>>,

    /// Serializes payment policy changes against the funding of the payments
    /// those policies admit.
    payment_policy_lock: Arc<RwLock<()>>,

    /// How the mnemonic was obtained at startup. When the gateway starts
    /// unconfigured, the source of the mnemonic set later arrives through the
    /// `mnemonic_sender` channel instead.
    mnemonic_source: MnemonicSource,
}

impl std::fmt::Debug for Gateway {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Gateway")
            .field("federation_manager", &self.federation_manager)
            .field("state", &self.state)
            .field("client_builder", &self.client_builder)
            .field("gateway_db", &self.gateway_db)
            .field("listen", &self.listen)
            .field("registrations", &self.registrations)
            .finish_non_exhaustive()
    }
}

/// Internal helper for on-chain withdrawal calculations
struct WithdrawDetails {
    amount: Amount,
    mint_fees: Option<Amount>,
    peg_out_fees: PegOutFees,
    flat_peg_out_fee: bool,
}

/// Waits for the ecash change of a successful withdrawal to be issued.
///
/// A withdrawal is reported successful once its transaction is accepted, but
/// the primary module issues the change from funding it asynchronously. Until
/// then the change is missing from the balance, so callers that read the
/// balance right after a withdrawal would see it understated by the change.
/// The withdrawal itself has already succeeded, so a failure here is only
/// logged.
async fn await_withdraw_change(
    client: &ClientHandleArc,
    operation_id: OperationId,
    change: Vec<fedimint_core::OutPoint>,
) {
    if let Err(err) = client
        .await_primary_bitcoin_module_outputs(operation_id, change)
        .await
    {
        warn!(
            target: LOG_GATEWAY,
            err = %err.fmt_compact(),
            ?operation_id,
            "Failed to await withdrawal change outputs"
        );
    }
}

/// Executes a withdrawal using the walletv2 module
async fn withdraw_v2(
    client: &ClientHandleArc,
    wallet_module: &fedimint_walletv2_client::WalletClientModule,
    address: &Address,
    amount: BitcoinAmountOrAll,
    quoted_fee: Option<bitcoin::Amount>,
) -> AdminResult<WithdrawResponse> {
    let current_fee =
        wallet_module
            .send_fee()
            .await
            .map_err(|e| AdminGatewayError::WithdrawError {
                failure_reason: e.fmt_compact().to_string(),
            })?;

    // The federation rejects any send whose fee is below its current send fee,
    // so a quote that has gone stale would only produce a doomed transaction.
    // Sending with the quoted fee otherwise keeps the operator paying exactly
    // what they confirmed in the preview.
    let fee = match quoted_fee {
        Some(quoted_fee) if quoted_fee < current_fee => {
            return Err(AdminGatewayError::WithdrawError {
                failure_reason: format!(
                    "The on-chain fee increased from {} to {} sats since the preview, please preview the withdrawal again",
                    quoted_fee.to_sat(),
                    current_fee.to_sat()
                ),
            });
        }
        Some(quoted_fee) => quoted_fee,
        None => current_fee,
    };

    let withdraw_amount = match amount {
        BitcoinAmountOrAll::All => {
            let balance = client.get_balance_for_btc().await.map_err(|err| {
                AdminGatewayError::Unexpected(anyhow!(
                    "Balance not available: {}",
                    err.fmt_compact()
                ))
            })?;

            // The on-chain fee is only part of the cost: funding the wallet
            // output also incurs the federation's per-note fees. `fee` is
            // passed on to `send` below so both are computed against the same
            // on-chain fee.
            wallet_module
                .max_sendable_amount(balance, fee)
                .await
                .map_err(|err| AdminGatewayError::WithdrawError {
                    failure_reason: format!(
                        "Insufficient funds. Balance: {balance} Fee: {fee}: {}",
                        err.fmt_compact()
                    ),
                })?
        }
        BitcoinAmountOrAll::Amount(a) => a,
    };

    let operation_id = wallet_module
        .send(
            address.as_unchecked().clone(),
            withdraw_amount,
            Some(fee),
            serde_json::Value::Null,
        )
        .await
        .map_err(|e| AdminGatewayError::WithdrawError {
            failure_reason: e.fmt_compact().to_string(),
        })?;

    let result = wallet_module
        .await_final_send_operation_state(operation_id)
        .await
        .map_err(|e| AdminGatewayError::WithdrawError {
            failure_reason: e.fmt_compact().to_string(),
        })?;

    let fees = PegOutFees::from_amount(fee);

    match result {
        fedimint_walletv2_client::FinalSendOperationState::Success(txid) => {
            if let Some(Ok(fedimint_walletv2_client::WalletOperationMeta::Send(meta))) = client
                .operation_log()
                .get_operation(operation_id)
                .await
                .map(|operation| operation.try_meta())
            {
                await_withdraw_change(
                    client,
                    operation_id,
                    meta.change_outpoint_range.into_iter().collect(),
                )
                .await;
            }
            info!(target: LOG_GATEWAY, amount = %withdraw_amount, address = %address, "Sent funds via walletv2");
            Ok(WithdrawResponse { txid, fees })
        }
        fedimint_walletv2_client::FinalSendOperationState::Aborted => {
            Err(AdminGatewayError::WithdrawError {
                failure_reason: "Withdrawal transaction was aborted".to_string(),
            })
        }
        fedimint_walletv2_client::FinalSendOperationState::Failure => {
            Err(AdminGatewayError::WithdrawError {
                failure_reason: "Withdrawal failed".to_string(),
            })
        }
    }
}

/// Calculates an estimated max withdrawable amount on-chain
async fn calculate_max_withdrawable(
    client: &ClientHandleArc,
    address: &Address,
) -> AdminResult<WithdrawDetails> {
    let balance = client.get_balance_for_btc().await.map_err(|err| {
        AdminGatewayError::Unexpected(anyhow!("Balance not available: {}", err.fmt_compact()))
    })?;

    if let Ok(wallet_module) =
        client.get_first_module::<fedimint_walletv2_client::WalletClientModule>()
    {
        let fee = wallet_module
            .send_fee()
            .await
            .map_err(|e| AdminGatewayError::WithdrawError {
                failure_reason: e.fmt_compact().to_string(),
            })?;

        let max_withdrawable = wallet_module
            .max_sendable_amount(balance, fee)
            .await
            .map_err(|err| AdminGatewayError::WithdrawError {
                failure_reason: err.fmt_compact().to_string(),
            })?;

        // Everything the balance does not become an on-chain payment or miner
        // fee is the federation fee: the wallet output fee, the mint input
        // fees on the funding notes, any change output fees and dust. This is
        // exact by construction, so it needs no separate estimate.
        let federation_fees = balance
            .saturating_sub(Amount::from_sats(max_withdrawable.to_sat()))
            .saturating_sub(Amount::from_sats(fee.to_sat()));

        return Ok(WithdrawDetails {
            amount: Amount::from_sats(max_withdrawable.to_sat()),
            mint_fees: Some(federation_fees),
            peg_out_fees: PegOutFees::from_amount(fee),
            flat_peg_out_fee: true,
        });
    }

    let Ok(wallet_module) = client.get_first_module::<WalletClientModule>() else {
        return Err(AdminGatewayError::Unexpected(anyhow!(
            "No wallet module found"
        )));
    };

    let (max_withdrawable, peg_out_fees) = wallet_module
        .max_withdrawable_amount(address, balance)
        .await
        .map_err(|err| AdminGatewayError::WithdrawError {
            failure_reason: err.fmt_compact().to_string(),
        })?;

    // Everything the balance does not become an on-chain payment or miner fee
    // is the federation fee: the peg-out output fee, the mint input fees on the
    // funding notes, any change output fees and dust. This is exact by
    // construction, so it needs no separate estimate.
    let federation_fees = balance
        .saturating_sub(Amount::from_sats(max_withdrawable.to_sat()))
        .saturating_sub(Amount::from_sats(peg_out_fees.amount().to_sat()));

    Ok(WithdrawDetails {
        amount: Amount::from_sats(max_withdrawable.to_sat()),
        mint_fees: Some(federation_fees),
        peg_out_fees,
        flat_peg_out_fee: false,
    })
}

impl Gateway {
    /// Returns a bitcoind client using the credentials that were passed in from
    /// the environment variables.
    fn get_bitcoind_client(
        opts: &GatewayOpts,
        network: bitcoin::Network,
        gateway_id: &PublicKey,
    ) -> anyhow::Result<(BitcoindClient, ChainSource)> {
        let bitcoind_username = opts
            .bitcoind_username
            .clone()
            .expect("FM_BITCOIND_URL is set but FM_BITCOIND_USERNAME is not");
        let url = opts.bitcoind_url.clone().expect("No bitcoind url set");
        let password = opts
            .bitcoind_password
            .clone()
            .expect("FM_BITCOIND_URL is set but FM_BITCOIND_PASSWORD is not");

        let chain_source = ChainSource::Bitcoind {
            username: bitcoind_username.clone(),
            password: password.clone(),
            server_url: url.clone(),
        };
        let wallet_name = format!("gatewayd-{gateway_id}");
        let client = BitcoindClient::new(&url, bitcoind_username, password, &wallet_name, network)?;
        Ok((client, chain_source))
    }

    /// Default function for creating a gateway with the `Mint`, `Wallet`, and
    /// `Gateway` modules.
    pub async fn new_with_default_modules(
        mnemonic_sender: tokio::sync::broadcast::Sender<MnemonicSource>,
    ) -> anyhow::Result<Gateway> {
        let opts = GatewayOpts::parse();
        let gateway_parameters = opts.to_gateway_parameters()?;
        let decoders = ModuleDecoderRegistry::default();

        let db_path = opts.data_dir.join(DB_FILE);
        let gateway_db = match opts.db_backend {
            DatabaseBackend::RocksDb => {
                debug!(target: LOG_GATEWAY, "Using RocksDB database backend");
                Database::new(
                    fedimint_rocksdb::RocksDb::build(db_path).open().await?,
                    decoders,
                )
            }
            DatabaseBackend::CursedRedb => {
                debug!(target: LOG_GATEWAY, "Using CursedRedb database backend");
                Database::new(
                    fedimint_cursed_redb::MemAndRedb::new(db_path).await?,
                    decoders,
                )
            }
        };

        // Apply database migrations before using the database to ensure old database
        // structures are readable.
        apply_migrations(
            &gateway_db,
            (),
            "gatewayd".to_string(),
            get_gatewayd_database_migrations(),
            None,
            None,
        )
        .await?;

        // For legacy reasons, we use the http id for the unique identifier of the
        // bitcoind watch-only wallet
        let http_id = Self::load_or_create_gateway_keypair(&gateway_db, RegisteredProtocol::Http)
            .await
            .public_key();
        let (dyn_bitcoin_rpc, chain_source) =
            match (opts.bitcoind_url.as_ref(), opts.esplora_url.as_ref()) {
                (Some(_), None) => {
                    let (client, chain_source) =
                        Self::get_bitcoind_client(&opts, gateway_parameters.network, &http_id)?;
                    (client.into_dyn(), chain_source)
                }
                (None, Some(url)) => {
                    let client = EsploraClient::new(url)
                        .expect("Could not create EsploraClient")
                        .into_dyn();
                    let chain_source = ChainSource::Esplora {
                        server_url: url.clone(),
                    };
                    (client, chain_source)
                }
                (Some(_), Some(_)) => {
                    // Use bitcoind by default if both are set
                    let (client, chain_source) =
                        Self::get_bitcoind_client(&opts, gateway_parameters.network, &http_id)?;
                    (client.into_dyn(), chain_source)
                }
                _ => unreachable!("ArgGroup already enforced XOR relation"),
            };

        // Gateway module will be attached when the federation clients are created
        // because the LN RPC will be injected with `GatewayClientGen`.
        let mut registry = ClientModuleInitRegistry::new();
        registry.attach(MintClientInit);
        registry.attach(MintV2ClientInit);
        registry.attach(WalletClientInit::new(dyn_bitcoin_rpc));
        registry.attach(fedimint_walletv2_client::WalletClientInit);

        let client_builder =
            GatewayClientBuilder::new(opts.data_dir.clone(), registry, opts.db_backend).await?;

        let (gateway_state, mnemonic_source) = if Self::load_mnemonic(&gateway_db).await.is_some() {
            (GatewayState::Disconnected, MnemonicSource::Existing)
        } else {
            // Generate a mnemonic or use one from an environment variable if `skip_setup`
            // is true
            if gateway_parameters.skip_setup {
                let (mnemonic, mnemonic_source) = if let Ok(words) =
                    std::env::var(FM_GATEWAY_MNEMONIC_ENV)
                {
                    info!(target: LOG_GATEWAY, "Using provided mnemonic from environment variable");
                    let mnemonic = Mnemonic::parse_in_normalized(Language::English, words.as_str())
                        .map_err(|e| {
                            AdminGatewayError::MnemonicError(anyhow!(format!(
                                "Seed phrase provided in environment was invalid {e:?}"
                            )))
                        })?;
                    (mnemonic, MnemonicSource::Existing)
                } else {
                    debug!(target: LOG_GATEWAY, "Generating mnemonic and writing entropy to client storage");
                    (
                        Bip39RootSecretStrategy::<12>::random(&mut OsRng),
                        MnemonicSource::Generated,
                    )
                };

                Client::store_encodable_client_secret(&gateway_db, mnemonic.to_entropy())
                    .await
                    .map_err(|err| AdminGatewayError::MnemonicError(err.into()))?;
                (GatewayState::Disconnected, mnemonic_source)
            } else {
                // The mnemonic's source arrives through `mnemonic_sender` once
                // it is set
                (
                    GatewayState::NotConfigured { mnemonic_sender },
                    MnemonicSource::Existing,
                )
            }
        };

        info!(
            target: LOG_GATEWAY,
            version = %fedimint_build_code_version_env!(),
            "Starting gatewayd",
        );

        Gateway::new(
            opts.mode,
            gateway_parameters,
            gateway_db,
            client_builder,
            gateway_state,
            chain_source,
            mnemonic_source,
        )
        .await
    }

    /// Helper function for creating a gateway from either
    /// `new_with_default_modules` or `Gateway::builder`.
    async fn new(
        lightning_mode: LightningMode,
        gateway_parameters: GatewayParameters,
        gateway_db: Database,
        client_builder: GatewayClientBuilder,
        gateway_state: GatewayState,
        chain_source: ChainSource,
        mnemonic_source: MnemonicSource,
    ) -> anyhow::Result<Gateway> {
        let num_route_hints = gateway_parameters.num_route_hints;
        let network = gateway_parameters.network;

        let task_group = TaskGroup::new();
        task_group.install_kill_handler();

        let mut registrations = BTreeMap::new();
        if let Some(http_url) = gateway_parameters.versioned_api {
            registrations.insert(
                RegisteredProtocol::Http,
                Registration::new(&gateway_db, http_url, RegisteredProtocol::Http).await,
            );
        }

        let iroh_sk = Self::load_or_create_iroh_key(&gateway_db).await;
        if gateway_parameters.iroh_listen.is_some() {
            let endpoint_url = SafeUrl::parse(&format!("iroh://{}", iroh_sk.public()))?;
            registrations.insert(
                RegisteredProtocol::Iroh,
                Registration::new(&gateway_db, endpoint_url, RegisteredProtocol::Iroh).await,
            );
        }

        Ok(Self {
            federation_manager: Arc::new(RwLock::new(FederationManager::new())),
            lightning_mode,
            state: Arc::new(RwLock::new(gateway_state)),
            client_builder,
            gateway_db: gateway_db.clone(),
            listen: gateway_parameters.listen,
            metrics_listen: gateway_parameters.metrics_listen,
            task_group,
            bcrypt_password_hash: gateway_parameters.bcrypt_password_hash.to_string(),
            bcrypt_liquidity_manager_password_hash: gateway_parameters
                .bcrypt_liquidity_manager_password_hash
                .map(|h| h.to_string()),
            num_route_hints,
            network,
            chain_source,
            default_routing_fees: gateway_parameters.default_routing_fees,
            default_transaction_fees: gateway_parameters.default_transaction_fees,
            iroh_sk,
            iroh_dns: gateway_parameters.iroh_dns,
            iroh_relays: gateway_parameters.iroh_relays,
            iroh_listen: gateway_parameters.iroh_listen,
            registrations,
            registration_health: RegistrationHealthTracker::default(),
            invoice_rate_limiter: Arc::new(TokenBucketRateLimiter::new(
                gateway_parameters.invoice_rate_limit_burst,
                gateway_parameters.invoice_rate_limit_per_second,
            )),
            lnv2_hold_invoice_lock_pool: Arc::new(lockable::LockPool::new()),
            payment_policy_lock: Arc::new(RwLock::new(())),
            mnemonic_source,
        })
    }

    async fn load_or_create_gateway_keypair(
        gateway_db: &Database,
        protocol: RegisteredProtocol,
    ) -> secp256k1::Keypair {
        let mut dbtx = gateway_db.begin_transaction().await;
        let keypair = dbtx.load_or_create_gateway_keypair(protocol).await;
        dbtx.commit_tx().await;
        keypair
    }

    /// Returns `iroh::SecretKey` and saves it to the database if it does not
    /// exist
    async fn load_or_create_iroh_key(gateway_db: &Database) -> iroh::SecretKey {
        let mut dbtx = gateway_db.begin_transaction().await;
        let iroh_sk = dbtx.load_or_create_iroh_key().await;
        dbtx.commit_tx().await;
        iroh_sk
    }

    pub async fn http_gateway_id(&self) -> PublicKey {
        Self::load_or_create_gateway_keypair(&self.gateway_db, RegisteredProtocol::Http)
            .await
            .public_key()
    }

    async fn get_state(&self) -> GatewayState {
        self.state.read().await.clone()
    }

    /// Reads and serializes structures from the Gateway's database for the
    /// purpose for serializing to JSON for inspection.
    pub async fn dump_database(
        dbtx: &mut DatabaseTransaction<'_>,
        prefix_names: Vec<String>,
    ) -> BTreeMap<String, Box<dyn erased_serde::Serialize + Send>> {
        dbtx.dump_database(prefix_names).await
    }

    /// Main entrypoint into the gateway that starts the client registration
    /// timer, loads the federation clients from the persisted config,
    /// begins listening for intercepted payments, and starts the webserver
    /// to service requests.
    pub async fn run(
        self,
        runtime: Arc<tokio::runtime::Runtime>,
        mnemonic_receiver: tokio::sync::broadcast::Receiver<MnemonicSource>,
    ) -> anyhow::Result<TaskShutdownToken> {
        install_crypto_provider().await;
        self.register_clients_timer();
        self.load_clients().await?;
        self.start_gateway(runtime, mnemonic_receiver.resubscribe());
        self.spawn_backup_task();
        self.spawn_prune_registered_contracts_task();
        // start metrics server
        fedimint_metrics::spawn_api_server(self.metrics_listen, self.task_group.clone()).await?;
        // start webserver last to avoid handling requests before fully initialized
        let handle = self.task_group.make_handle();
        run_webserver(Arc::new(self), mnemonic_receiver.resubscribe()).await?;
        let shutdown_receiver = handle.make_shutdown_rx();
        Ok(shutdown_receiver)
    }

    /// Spawns a background task that checks every `BACKUP_UPDATE_INTERVAL` to
    /// see if any federations need to be backed up.
    fn spawn_backup_task(&self) {
        let self_copy = self.clone();
        self.task_group
            .spawn_cancellable_silent("backup ecash", async move {
                const BACKUP_UPDATE_INTERVAL: Duration = Duration::from_hours(1);
                let mut interval = tokio::time::interval(BACKUP_UPDATE_INTERVAL);
                interval.tick().await;
                loop {
                    {
                        let mut dbtx = self_copy.gateway_db.begin_transaction().await;
                        self_copy.backup_all_federations(&mut dbtx).await;
                        dbtx.commit_tx().await;
                        interval.tick().await;
                    }
                }
            });
    }

    /// Spawns a background task that periodically deletes registered incoming
    /// contract records whose invoice has long expired, so unpaid invoice
    /// registrations cannot grow the database without bound.
    fn spawn_prune_registered_contracts_task(&self) {
        let self_copy = self.clone();
        self.task_group.spawn_cancellable_silent(
            "prune registered incoming contracts",
            async move {
                const PRUNE_INTERVAL: Duration = Duration::from_hours(1);
                // Records are kept for a day past invoice expiry so payments
                // settled shortly before expiry can still complete and be
                // verified via the preimage verification endpoint.
                const RETENTION_AFTER_EXPIRY: Duration = Duration::from_hours(24);

                let mut interval = tokio::time::interval(PRUNE_INTERVAL);
                loop {
                    interval.tick().await;

                    let cutoff_secs = duration_since_epoch()
                        .saturating_sub(RETENTION_AFTER_EXPIRY)
                        .as_secs();

                    let mut dbtx = self_copy.gateway_db.begin_transaction().await;
                    let num_pruned = dbtx.prune_registered_incoming_contracts(cutoff_secs).await;
                    match dbtx.commit_tx_result().await {
                        Ok(()) => {
                            if num_pruned > 0 {
                                info!(
                                    target: LOG_GATEWAY,
                                    num_pruned,
                                    "Pruned expired incoming contract records"
                                );
                            }
                        }
                        Err(err) => {
                            warn!(
                                target: LOG_GATEWAY,
                                err = %err.fmt_compact(),
                                "Failed to prune expired incoming contract records"
                            );
                        }
                    }
                }
            },
        );
    }

    /// Loops through all federations and checks their last save backup time. If
    /// the last saved backup time is past the threshold time, backup the
    /// federation.
    pub async fn backup_all_federations(&self, dbtx: &mut DatabaseTransaction<'_, Committable>) {
        /// How long the federation manager should wait to backup the ecash for
        /// each federation
        const BACKUP_THRESHOLD_DURATION: Duration = Duration::from_hours(24);

        let now = fedimint_core::time::now();
        let threshold = now
            .checked_sub(BACKUP_THRESHOLD_DURATION)
            .expect("Cannot be negative");
        for (id, last_backup) in dbtx.load_backup_records().await {
            match last_backup {
                Some(backup_time) if backup_time < threshold => {
                    let fed_manager = self.federation_manager.read().await;
                    fed_manager.backup_federation(&id, dbtx, now).await;
                }
                None => {
                    let fed_manager = self.federation_manager.read().await;
                    fed_manager.backup_federation(&id, dbtx, now).await;
                }
                _ => {}
            }
        }
    }

    /// Begins the task for listening for intercepted payments from the
    /// lightning node.
    fn start_gateway(
        &self,
        runtime: Arc<tokio::runtime::Runtime>,
        mut mnemonic_receiver: tokio::sync::broadcast::Receiver<MnemonicSource>,
    ) {
        const PAYMENT_STREAM_RETRY_SECONDS: u64 = 60;

        let self_copy = self.clone();
        let tg = self.task_group.clone();
        self.task_group.spawn(
            "Subscribe to intercepted lightning payments in stream",
            |handle| async move {
                let mut mnemonic_source = self_copy.mnemonic_source;
                // Repeatedly attempt to establish a connection to the lightning node and create a payment stream, re-trying if the connection is broken.
                loop {
                    if handle.is_shutting_down() {
                        info!(target: LOG_GATEWAY, "Gateway lightning payment stream handler loop is shutting down");
                        break;
                    }

                    if let GatewayState::NotConfigured{ .. } = self_copy.get_state().await {
                        info!(
                            target: LOG_GATEWAY,
                            "Waiting for the mnemonic to be set before starting lightning receive loop."
                        );
                        info!(
                            target: LOG_GATEWAY,
                            "You might need to provide it from the UI or refer to documentation w.r.t how to initialize it."
                        );

                        if let Ok(source) = mnemonic_receiver.recv().await {
                            mnemonic_source = source;
                        }
                        info!(
                            target: LOG_GATEWAY,
                            ?mnemonic_source,
                            "Received mnemonic, attempting to start lightning receive loop"
                        );
                    }

                    let payment_stream_task_group = tg.make_subgroup();
                    let lnrpc_route = self_copy.create_lightning_client(runtime.clone(), mnemonic_source).await;

                    debug!(target: LOG_GATEWAY, "Establishing lightning payment stream...");
                    let (stream, ln_client) = match lnrpc_route.route_htlcs(&payment_stream_task_group).await
                    {
                        Ok((stream, ln_client)) => (stream, ln_client),
                        Err(err) => {
                            warn!(target: LOG_GATEWAY, err = %err.fmt_compact(), "Failed to open lightning payment stream");
                            // `route_htlcs` may have already spawned tasks into the
                            // subgroup before failing (e.g. the LNv1 interceptor is
                            // spawned before LNv2 setup, which can fail). Tear the
                            // subgroup down so no stale task keeps owning the LND HTLC
                            // stream, which would prevent the retry from taking over and
                            // could cause it to cancel real HTLCs after `gateway_receiver`
                            // is dropped.
                            if let Err(err) = payment_stream_task_group.shutdown_join_all(None).await {
                                crit!(target: LOG_GATEWAY, err = %err.fmt_compact(), "Lightning payment stream task group shutdown");
                            }
                            sleep(Duration::from_secs(PAYMENT_STREAM_RETRY_SECONDS)).await;
                            continue
                        }
                    };

                    // Successful calls to `route_htlcs` establish a connection
                    self_copy.set_gateway_state(GatewayState::Connected).await;
                    info!(target: LOG_GATEWAY, "Established lightning payment stream");

                    let route_payments_response =
                        self_copy.route_lightning_payments(&handle, stream, ln_client).await;

                    self_copy.set_gateway_state(GatewayState::Disconnected).await;
                    if let Err(err) = payment_stream_task_group.shutdown_join_all(None).await {
                        crit!(target: LOG_GATEWAY, err = %err.fmt_compact(), "Lightning payment stream task group shutdown");
                    }

                    self_copy.unannounce_from_all_federations().await;

                    match route_payments_response {
                        ReceivePaymentStreamAction::RetryAfterDelay => {
                            warn!(target: LOG_GATEWAY, retry_interval = %PAYMENT_STREAM_RETRY_SECONDS, "Disconnected from lightning node");
                            sleep(Duration::from_secs(PAYMENT_STREAM_RETRY_SECONDS)).await;
                        }
                        ReceivePaymentStreamAction::NoRetry => break,
                    }
                }
            },
        );
    }

    /// Handles a stream of incoming payments from the lightning node after
    /// ensuring the gateway is properly configured. Awaits until the stream
    /// is closed, then returns with the appropriate action to take.
    async fn route_lightning_payments<'a>(
        &'a self,
        handle: &TaskHandle,
        mut stream: RouteHtlcStream<'a>,
        ln_client: Arc<dyn ILnRpcClient>,
    ) -> ReceivePaymentStreamAction {
        let LightningInfo::Connected {
            public_key: lightning_public_key,
            alias: lightning_alias,
            network: lightning_network,
            block_height: _,
            synced_to_chain,
        } = ln_client.parsed_node_info().await
        else {
            warn!(target: LOG_GATEWAY, "Failed to retrieve Lightning info");
            return ReceivePaymentStreamAction::RetryAfterDelay;
        };

        assert!(
            self.network == lightning_network,
            "Lightning node network does not match Gateway's network. LN: {lightning_network} Gateway: {}",
            self.network
        );

        if synced_to_chain || is_env_var_set(FM_GATEWAY_SKIP_WAIT_FOR_SYNC_ENV) {
            info!(target: LOG_GATEWAY, "Gateway is already synced to chain");
        } else {
            self.set_gateway_state(GatewayState::Syncing).await;
            info!(target: LOG_GATEWAY, "Waiting for chain sync");
            if let Err(err) = ln_client.wait_for_chain_sync().await {
                warn!(target: LOG_GATEWAY, err = %err.fmt_compact(), "Failed to wait for chain sync");
                return ReceivePaymentStreamAction::RetryAfterDelay;
            }
        }

        let lightning_context = LightningContext {
            lnrpc: LnRpcTracked::new(ln_client, "gateway"),
            lightning_public_key,
            lightning_alias,
            lightning_network,
        };
        if let GatewayState::ShuttingDown { .. } = self
            .set_gateway_state(GatewayState::Running { lightning_context })
            .await
        {
            info!(
                target: LOG_GATEWAY,
                "Reconnected to the lightning node while shutting down, not accepting payments"
            );
        } else {
            info!(target: LOG_GATEWAY, "Gateway is running");
        }

        if matches!(self.lightning_mode, LightningMode::Lnd { .. }) {
            // Re-register the gateway with all federations after connecting to the
            // lightning node
            let mut dbtx = self.gateway_db.begin_transaction_nc().await;
            let all_federations_configs =
                dbtx.load_federation_configs().await.into_iter().collect();
            self.register_federations(&all_federations_configs, &self.task_group)
                .await;
        }

        // Runs until the connection to the lightning node breaks or we receive the
        // shutdown signal.
        let htlc_task_group = self.task_group.make_subgroup();
        if handle
            .cancel_on_shutdown(async move {
                loop {
                    let payment_request_or = tokio::select! {
                        payment_request_or = stream.next() => {
                            payment_request_or
                        }
                        () = self.is_shutting_down_safely() => {
                            break;
                        }
                    };

                    let Some(payment_request) = payment_request_or else {
                        warn!(
                            target: LOG_GATEWAY,
                            "Unexpected response from incoming lightning payment stream. Shutting down payment processor"
                        );
                        break;
                    };

                    let state_guard = self.state.read().await;
                    if let GatewayState::Running { ref lightning_context } = *state_guard {
                        // Spawn a subtask to handle each payment in parallel
                        let gateway = self.clone();
                        let lightning_context = lightning_context.clone();
                        htlc_task_group.spawn_cancellable_silent(
                            "handle_lightning_payment",
                            async move {
                                let start = fedimint_core::time::now();
                                let outcome = gateway
                                    .handle_lightning_payment(payment_request, &lightning_context)
                                    .await;
                                metrics::HTLC_HANDLING_DURATION_SECONDS
                                    .with_label_values(&[outcome])
                                    .observe(
                                        fedimint_core::time::now()
                                            .duration_since(start)
                                            .unwrap_or_default()
                                            .as_secs_f64(),
                                    );
                            },
                        );
                    } else {
                        warn!(
                            target: LOG_GATEWAY,
                            state = %state_guard,
                            "Gateway isn't in a running state, cannot handle incoming payments."
                        );
                        break;
                    }
                }
            })
            .await
            .is_ok()
        {
            warn!(target: LOG_GATEWAY, "Lightning payment stream connection broken. Gateway is disconnected");
            ReceivePaymentStreamAction::RetryAfterDelay
        } else {
            info!(target: LOG_GATEWAY, "Received shutdown signal");
            ReceivePaymentStreamAction::NoRetry
        }
    }

    /// Polls the Gateway's state waiting for it to shutdown so the thread
    /// processing payment requests can exit.
    async fn is_shutting_down_safely(&self) {
        loop {
            if let GatewayState::ShuttingDown { .. } = self.get_state().await {
                return;
            }

            fedimint_core::task::sleep(Duration::from_secs(1)).await;
        }
    }

    /// Handles an intercepted lightning payment. If the payment is part of an
    /// incoming payment to a federation, spawns a state machine and hands the
    /// payment off to it. If the payment's last-hop short channel id maps to
    /// a known federation but no LNv1 or LNv2 offer matched, cancels (fails
    /// back) the HTLC so the sender can retry rather than treating the
    /// gateway as a dead route. Otherwise (real-channel forwards), resumes
    /// the HTLC so LND can route it as a normal forward. A duplicate delivery
    /// of a HOLD invoice that is still being handled is dropped.
    ///
    /// Returns the outcome label for metrics tracking.
    pub async fn handle_lightning_payment(
        &self,
        payment_request: InterceptPaymentRequest,
        lightning_context: &LightningContext,
    ) -> &'static str {
        info!(
            target: LOG_GATEWAY,
            lightning_payment = %PrettyInterceptPaymentRequest(&payment_request),
            "Intercepting lightning payment",
        );

        // A payment without an incoming circuit is held by a HOLD invoice, which
        // the node can deliver more than once. Only one delivery is handled at a
        // time, from here through settling or cancelling the invoice. A
        // duplicate is dropped rather than made to wait: whatever the first
        // delivery decides resolves the whole invoice, and a duplicate handled
        // afterwards could fund an invoice the first has just cancelled. LND,
        // for one, replays every accepted invoice when the gateway resubscribes
        // after a reconnect. Forwards are not deduplicated: each has
        // its own circuit, and a replay after a reconnect may be the only
        // delivery still able to resolve it.
        let _hold_invoice_guard = if payment_request.incoming_circuit().is_none() {
            let Some(guard) = self
                .lnv2_hold_invoice_lock_pool
                .try_lock(payment_request.payment_hash)
            else {
                info!(
                    target: LOG_GATEWAY,
                    payment_hash = %payment_request.payment_hash,
                    "Dropping duplicate delivery of a HOLD invoice that is already being handled",
                );
                return "duplicate";
            };
            Some(guard)
        } else {
            None
        };

        let lnv2_start = fedimint_core::time::now();
        let lnv2_result = self
            .try_handle_lightning_payment_lnv2(&payment_request, lightning_context)
            .await;
        let lnv2_outcome = if lnv2_result.is_ok() {
            "success"
        } else {
            "error"
        };
        metrics::HTLC_LNV2_ATTEMPT_DURATION_SECONDS
            .with_label_values(&[lnv2_outcome])
            .observe(
                fedimint_core::time::now()
                    .duration_since(lnv2_start)
                    .unwrap_or_default()
                    .as_secs_f64(),
            );
        if lnv2_result.is_ok() {
            return "lnv2";
        }

        // A payment held for a registered LNv2 contract is left to that
        // contract's receive, or to the Lightning node's expiry, even if its
        // amount does not match: cancelling it would also fail back HTLCs the
        // receive may already be settling.
        if payment_request.incoming_circuit().is_none()
            && let Err(PublicGatewayError::LNv2(err @ LNv2Error::IncomingAmountMismatch { .. })) =
                &lnv2_result
        {
            warn!(
                target: LOG_GATEWAY,
                payment_hash = %payment_request.payment_hash,
                err = %err.fmt_compact(),
                "Ignoring payment for a registered LNv2 contract whose amount does not match",
            );
            return "ignore";
        }

        let lnv1_start = fedimint_core::time::now();
        let lnv1_result = self
            .try_handle_lightning_payment_ln_legacy(&payment_request, lightning_context)
            .await;
        let lnv1_outcome = if lnv1_result.is_ok() {
            "success"
        } else {
            "error"
        };
        metrics::HTLC_LNV1_ATTEMPT_DURATION_SECONDS
            .with_label_values(&[lnv1_outcome])
            .observe(
                fedimint_core::time::now()
                    .duration_since(lnv1_start)
                    .unwrap_or_default()
                    .as_secs_f64(),
            );
        if lnv1_result.is_ok() {
            return "lnv1";
        }

        // Neither LNv1 nor LNv2 matched. If the last-hop scid is one of our
        // federation virtual scids, cancel so the sender gets a non-permanent
        // failure (avoiding `UNKNOWN_NEXT_PEER` blacklisting). If the scid is
        // for a real channel, resume so LND forwards normally.
        let is_federation_scid = match payment_request.short_channel_id {
            Some(scid) => self
                .federation_manager
                .read()
                .await
                .get_client_for_index(scid)
                .is_some(),
            None => false,
        };

        if is_federation_scid {
            // The HTLC targeted a federation we serve but we couldn't claim
            // it (no LNv1 offer / no LNv2 contract / receives turned off /
            // underfunded gateway / federation timeout / etc.). Surface the
            // underlying error variants so operators can diagnose the cause;
            // otherwise both `Err` values are dropped on the floor and the
            // only visible signal is the metric label `"error"`.
            warn!(
                target: LOG_GATEWAY,
                payment_hash = %payment_request.payment_hash,
                short_channel_id = ?payment_request.short_channel_id,
                amount_msat = payment_request.amount_msat,
                incoming_chan_id = payment_request.incoming_chan_id,
                htlc_id = payment_request.htlc_id,
                lnv2_err = ?lnv2_result.as_ref().err(),
                lnv1_err = ?lnv1_result.as_ref().err(),
                "Lightning payment for a served federation could not be accepted: cancelling HTLC",
            );
            Self::cancel_unmatched_lightning_payment(payment_request, lightning_context).await;
            "cancel"
        } else {
            // Normal route-through traffic: the gateway's LND interceptor
            // sees every HTLC, but only federation-scid HTLCs are ours to
            // handle. Resume so LND forwards the rest as a regular routing
            // node — no warning needed since this is the expected path.
            Self::forward_lightning_payment(payment_request, lightning_context).await;
            "forward"
        }
    }

    /// Tries to handle a lightning payment using the LNv2 protocol.
    /// Returns `Ok` if the payment was handled, `Err` otherwise.
    async fn try_handle_lightning_payment_lnv2(
        &self,
        htlc_request: &InterceptPaymentRequest,
        lightning_context: &LightningContext,
    ) -> Result<()> {
        // If `payment_hash` has been registered as a LNv2 payment, we try to complete
        // the payment by getting the preimage from the federation
        // using the LNv2 protocol. If the `payment_hash` is not registered,
        // this payment is either a legacy Lightning payment or the end destination is
        // not a Fedimint.
        // Match and fund against the amount actually locked in the incoming
        // HTLC, not the sender-controlled onion forward amount, so a forged
        // `amt_to_forward` cannot satisfy the registered contract's amount
        // check while only a token amount is really locked.
        let (contract, client) = self
            .get_registered_incoming_contract_and_client_v2(
                PaymentImage::Hash(htlc_request.payment_hash),
                htlc_request.incoming_amount_msat,
            )
            .await?;

        // An intercepted forward, or an unknown current height, leaves no time
        // to fund a fresh contract; see `lnv2_blocks_to_claim_deadline`.
        let current_block_height = lightning_context
            .lnrpc
            .info()
            .await
            .ok()
            .map(|info| info.block_height);
        let blocks_to_claim_deadline =
            htlc_request.lnv2_blocks_to_claim_deadline(current_block_height);

        if let Err(err) = client
            .get_first_module::<GatewayClientModuleV2>()
            .expect("Must have client module")
            .relay_incoming_htlc(
                htlc_request.payment_hash,
                htlc_request.incoming_chan_id,
                htlc_request.htlc_id,
                contract,
                htlc_request.incoming_amount_msat,
                blocks_to_claim_deadline,
            )
            .await
        {
            warn!(
                target: LOG_GATEWAY,
                err = %err.fmt_compact(),
                "Error relaying incoming lightning payment"
            );

            let outcome = InterceptPaymentResponse {
                action: PaymentAction::Cancel,
                payment_hash: htlc_request.payment_hash,
                incoming_chan_id: htlc_request.incoming_chan_id,
                htlc_id: htlc_request.htlc_id,
            };

            if let Err(err) = lightning_context.lnrpc.complete_htlc(outcome).await {
                warn!(target: LOG_GATEWAY, err = %err.fmt_compact(), "Error sending HTLC response to lightning node");
            }
        }

        Ok(())
    }

    /// Tries to handle a lightning payment using the legacy lightning protocol.
    /// Returns `Ok` if the payment was handled, `Err` otherwise.
    async fn try_handle_lightning_payment_ln_legacy(
        &self,
        htlc_request: &InterceptPaymentRequest,
        lightning_context: &LightningContext,
    ) -> Result<()> {
        // Check if the payment corresponds to a federation supporting legacy Lightning.
        let Some(federation_index) = htlc_request.short_channel_id else {
            return Err(PublicGatewayError::LNv1(LNv1Error::IncomingPayment(
                "Incoming payment has not last hop short channel id".to_string(),
            )));
        };

        let Some(client) = self
            .federation_manager
            .read()
            .await
            .get_client_for_index(federation_index)
        else {
            return Err(PublicGatewayError::LNv1(LNv1Error::IncomingPayment("Incoming payment has a last hop short channel id that does not map to a known federation".to_string())));
        };

        // Both LND's `incoming_expiry` and LDK's `claim_deadline` are absolute
        // Bitcoin heights. LDK does not currently produce LNv1 forwards (it has
        // no federation short-channel id), but using the backend's own best
        // height keeps the unit and chain view consistent for every backend.
        client
            .borrow()
            .with(|client| async {
                let htlc = Htlc::from(htlc_request.clone());
                let lnv1 = client
                    .get_first_module::<GatewayClientModule>()
                    .map_err(|_| {
                        PublicGatewayError::LNv1(LNv1Error::IncomingPayment(
                            "Federation does not have LNv1 module".to_string(),
                        ))
                    })?;
                match lnv1
                    .gateway_handle_intercepted_htlc(htlc, async {
                        Ok(lightning_context.lnrpc.info().await?.block_height)
                    })
                    .await
                {
                    Ok(_) => Ok(()),
                    Err(e) => Err(PublicGatewayError::LNv1(LNv1Error::IncomingPayment(
                        format!("Error intercepting lightning payment {}", e.fmt_compact()),
                    ))),
                }
            })
            .await
    }

    /// Cancels (fails back) a lightning payment whose last-hop scid maps to a
    /// known federation but matched no LNv1 or LNv2 offer.
    ///
    /// Returning `PaymentAction::Forward` here would tell LND to resume the
    /// HTLC as a normal forward, but the last-hop short channel id is a
    /// virtual scid (no real channel exists), so LND would fail it back with
    /// the permanent error `UNKNOWN_NEXT_PEER`. Senders' mission control
    /// treats that as a permanent blacklist signal against the gateway,
    /// breaking future payments across all federations.
    ///
    /// `PaymentAction::Cancel` maps to `ResolveHoldForwardAction::Fail`, which
    /// fails the HTLC back with a non-permanent reason so the sender can
    /// retry instead of blacklisting the gateway.
    async fn cancel_unmatched_lightning_payment(
        htlc_request: InterceptPaymentRequest,
        lightning_context: &LightningContext,
    ) {
        let outcome = InterceptPaymentResponse {
            action: PaymentAction::Cancel,
            payment_hash: htlc_request.payment_hash,
            incoming_chan_id: htlc_request.incoming_chan_id,
            htlc_id: htlc_request.htlc_id,
        };

        if let Err(err) = lightning_context.lnrpc.complete_htlc(outcome).await {
            warn!(target: LOG_GATEWAY, err = %err.fmt_compact(), "Error sending lightning payment response to lightning node");
        }
    }

    /// Forwards a lightning payment to the next hop like a normal lightning
    /// node. Used when the intercepted HTLC is not destined for any federation
    /// this gateway serves, so LND should route it normally over a real
    /// channel.
    async fn forward_lightning_payment(
        htlc_request: InterceptPaymentRequest,
        lightning_context: &LightningContext,
    ) {
        let outcome = InterceptPaymentResponse {
            action: PaymentAction::Forward,
            payment_hash: htlc_request.payment_hash,
            incoming_chan_id: htlc_request.incoming_chan_id,
            htlc_id: htlc_request.htlc_id,
        };

        if let Err(err) = lightning_context.lnrpc.complete_htlc(outcome).await {
            warn!(target: LOG_GATEWAY, err = %err.fmt_compact(), "Error sending lightning payment response to lightning node");
        }
    }

    /// Helper function for atomically changing the Gateway's internal state.
    ///
    /// Shutting down is one-way. The lightning connection loop keeps running
    /// while `handle_shutdown_msg` drains the payments that are still in
    /// flight, and a reconnect in that window must not move the gateway back
    /// into a state that accepts new payments. A reconnect may still hand over
    /// a fresh `LightningContext`, which the drain needs to complete the
    /// payments it is waiting for.
    /// Returns the state that is in effect afterwards, which is not the
    /// requested one if the gateway is shutting down.
    async fn set_gateway_state(&self, state: GatewayState) -> GatewayState {
        let mut lock = self.state.write().await;

        if let GatewayState::ShuttingDown { .. } = *lock {
            match state {
                GatewayState::Running { lightning_context } => {
                    *lock = GatewayState::ShuttingDown { lightning_context };
                }
                ignored => {
                    info!(
                        target: LOG_GATEWAY,
                        ignored_state = %ignored,
                        "Gateway is shutting down, ignoring state change"
                    );
                }
            }
        } else {
            *lock = state;
        }

        lock.clone()
    }

    /// Drives the gateway's state directly, bypassing the lightning connection
    /// loop in [`Self::start_gateway`] that owns every real transition.
    ///
    /// Tests need to observe behaviour in states the builder cannot start them
    /// in -- notably "connected later than the federation clients" -- and have
    /// no lightning node to get there with. Nothing in production should call
    /// this.
    #[doc(hidden)]
    pub async fn set_gateway_state_out_of_band(&self, state: GatewayState) {
        self.set_gateway_state(state).await;
    }

    /// If the Gateway is connected to the Lightning node, returns the
    /// `ClientConfig` for each federation that the Gateway is connected to.
    pub async fn handle_get_federation_config(
        &self,
        federation_id_or: Option<FederationId>,
    ) -> AdminResult<GatewayFedConfig> {
        if !matches!(self.get_state().await, GatewayState::Running { .. }) {
            return Ok(GatewayFedConfig {
                federations: BTreeMap::new(),
            });
        }

        let federations = if let Some(federation_id) = federation_id_or {
            let mut federations = BTreeMap::new();
            federations.insert(
                federation_id,
                self.federation_manager
                    .read()
                    .await
                    .get_federation_config(federation_id)
                    .await?,
            );
            federations
        } else {
            self.federation_manager
                .read()
                .await
                .get_all_federation_configs()
                .await
        };

        Ok(GatewayFedConfig { federations })
    }

    /// Returns a Bitcoin deposit on-chain address for pegging in Bitcoin for a
    /// specific connected federation.
    pub async fn handle_address_msg(&self, payload: DepositAddressPayload) -> AdminResult<Address> {
        let client = self.select_client(payload.federation_id).await?;

        if let Ok(wallet_module) = client.value().get_first_module::<WalletClientModule>() {
            let address = wallet_module
                .allocate_deposit_address_expert_only(())
                .await
                .map_err(|e| AdminGatewayError::Unexpected(e.into()))?
                .address;
            Ok(address)
        } else if let Ok(wallet_module) = client
            .value()
            .get_first_module::<fedimint_walletv2_client::WalletClientModule>()
        {
            Ok(wallet_module.receive().await)
        } else {
            Err(AdminGatewayError::Unexpected(anyhow!(
                "No wallet module found"
            )))
        }
    }

    /// Requests the gateway to pay an outgoing LN invoice on behalf of a
    /// Fedimint client. Returns the payment hash's preimage on success.
    async fn handle_pay_invoice_msg(
        &self,
        payload: fedimint_ln_client::pay::PayInvoicePayload,
    ) -> Result<Preimage> {
        let GatewayState::Running { .. } = self.get_state().await else {
            return Err(PublicGatewayError::Lightning(
                LightningRpcError::FailedToConnect,
            ));
        };

        debug!(target: LOG_GATEWAY, "Handling pay invoice message");
        let client = self.select_client(payload.federation_id).await?;
        let contract_id = payload.contract_id;
        let gateway_module = &client
            .value()
            .get_first_module::<GatewayClientModule>()
            .map_err(|err| LNv1Error::OutgoingPayment(err.into()))
            .map_err(PublicGatewayError::LNv1)?;
        let operation_id = gateway_module
            .gateway_pay_bolt11_invoice(payload)
            .await
            .map_err(|err| LNv1Error::OutgoingPayment(err.into()))
            .map_err(PublicGatewayError::LNv1)?;
        let mut updates = gateway_module
            .gateway_subscribe_ln_pay(operation_id)
            .await
            .map_err(|err| LNv1Error::OutgoingPayment(err.into()))
            .map_err(PublicGatewayError::LNv1)?
            .into_stream();
        while let Some(update) = updates.next().await {
            match update {
                GatewayExtPayStates::Success { preimage, .. } => {
                    debug!(target: LOG_GATEWAY, contract_id = %contract_id, "Successfully paid invoice");
                    return Ok(preimage);
                }
                GatewayExtPayStates::Fail {
                    error,
                    error_message,
                } => {
                    return Err(PublicGatewayError::LNv1(LNv1Error::OutgoingContract {
                        error: Box::new(error),
                        message: format!(
                            "{error_message} while paying invoice with contract id {contract_id}"
                        ),
                    }));
                }
                GatewayExtPayStates::Canceled { error } => {
                    return Err(PublicGatewayError::LNv1(LNv1Error::OutgoingContract {
                        error: Box::new(error.clone()),
                        message: format!(
                            "Cancelled with {error} while paying invoice with contract id {contract_id}"
                        ),
                    }));
                }
                GatewayExtPayStates::Created => {
                    debug!(target: LOG_GATEWAY, contract_id = %contract_id, "Start pay invoice state machine");
                }
                other => {
                    debug!(target: LOG_GATEWAY, state = ?other, contract_id = %contract_id, "Got state while paying invoice");
                }
            }
        }

        Err(PublicGatewayError::LNv1(LNv1Error::OutgoingPayment(
            anyhow!("Ran out of state updates while paying invoice"),
        )))
    }

    /// Handles a request for the gateway to backup a connected federation's
    /// ecash.
    pub async fn handle_backup_msg(
        &self,
        BackupPayload { federation_id }: BackupPayload,
    ) -> AdminResult<()> {
        let federation_manager = self.federation_manager.read().await;
        let client = federation_manager
            .client(&federation_id)
            .ok_or(AdminGatewayError::ClientCreationError(anyhow::anyhow!(
                format!("Gateway has not connected to {federation_id}")
            )))?
            .value();
        let metadata: BTreeMap<String, String> = BTreeMap::new();
        #[allow(deprecated)]
        client
            .backup_to_federation(fedimint_client::backup::Metadata::from_json_serialized(
                metadata,
            ))
            .await
            .map_err(anyhow::Error::from)?;
        Ok(())
    }

    /// Trigger rechecking for deposits on an address
    pub async fn handle_recheck_address_msg(
        &self,
        payload: DepositAddressRecheckPayload,
    ) -> AdminResult<()> {
        let client = self.select_client(payload.federation_id).await?;

        if let Ok(wallet_module) = client.value().get_first_module::<WalletClientModule>() {
            wallet_module
                .recheck_pegin_address_by_address(payload.address)
                .await
                .map_err(|e| AdminGatewayError::Unexpected(e.into()))?;
            Ok(())
        } else if client
            .value()
            .get_first_module::<fedimint_walletv2_client::WalletClientModule>()
            .is_ok()
        {
            // Walletv2 auto-claims deposits, so this is a no-op
            Ok(())
        } else {
            Err(AdminGatewayError::Unexpected(anyhow!(
                "No wallet module found"
            )))
        }
    }

    /// Handles a request to receive ecash into the gateway.
    pub async fn handle_receive_ecash_msg(
        &self,
        payload: ReceiveEcashPayload,
    ) -> Result<ReceiveEcashResponse> {
        // Extract federation_id_prefix from either format
        let federation_id_prefix = base32::decode_prefixed::<fedimint_mintv2_client::ECash>(
            FEDIMINT_PREFIX,
            &payload.notes,
        )
        .ok()
        .and_then(|e| e.mint())
        .map(|id| id.to_prefix())
        .or_else(|| {
            OOBNotes::from_str(&payload.notes)
                .ok()
                .map(|n| n.federation_id_prefix())
        })
        .ok_or_else(|| PublicGatewayError::ReceiveEcashError {
            failure_reason: "Invalid ecash format: could not parse as ECash or OOBNotes"
                .to_string(),
        })?;

        let client = self
            .federation_manager
            .read()
            .await
            .get_client_for_federation_id_prefix(federation_id_prefix)
            .ok_or(FederationNotConnected {
                federation_id_prefix,
            })?;

        // Check which module is present and parse accordingly
        if let Ok(mint) = client.value().get_first_module::<MintClientModule>() {
            let notes = OOBNotes::from_str(&payload.notes).map_err(|e| {
                PublicGatewayError::ReceiveEcashError {
                    failure_reason: format!(
                        "Expected OOBNotes for MintV1 federation: {}",
                        e.fmt_compact()
                    ),
                }
            })?;
            let amount = notes.total_amount();

            let operation_id = mint.reissue_external_notes(notes, ()).await.map_err(|e| {
                PublicGatewayError::ReceiveEcashError {
                    failure_reason: e.fmt_compact().to_string(),
                }
            })?;

            let mut updates = mint
                .subscribe_reissue_external_notes(operation_id)
                .await
                .map_err(|e| PublicGatewayError::ReceiveEcashError {
                    failure_reason: format!(
                        "Could not subscribe to reissue operation: {}",
                        e.fmt_compact()
                    ),
                })?
                .into_stream();

            // Only `Done` and `Failed` are terminal for this stream. Ending on
            // `Created` or `Issuing` means the outputs were never finalized, so
            // their blind signatures were never verified, and reporting the notes'
            // claimed amount would credit e-cash the gateway does not hold.
            let mut reissued = false;
            while let Some(update) = updates.next().await {
                match update {
                    ReissueExternalNotesState::Failed(failure_reason) => {
                        return Err(PublicGatewayError::ReceiveEcashError { failure_reason });
                    }
                    ReissueExternalNotesState::Done => reissued = true,
                    ReissueExternalNotesState::Created | ReissueExternalNotesState::Issuing => {}
                }
            }

            if !reissued {
                return Err(PublicGatewayError::ReceiveEcashError {
                    failure_reason: "Reissue operation ended before the notes were reissued"
                        .to_string(),
                });
            }

            Ok(ReceiveEcashResponse { amount })
        } else if let Ok(mint) = client
            .value()
            .get_primary_module_for_unit::<MintV2ClientModule>(
                fedimint_core::module::AmountUnit::BITCOIN,
            )
        {
            let ecash: fedimint_mintv2_client::ECash =
                base32::decode_prefixed(FEDIMINT_PREFIX, &payload.notes).map_err(|e| {
                    PublicGatewayError::ReceiveEcashError {
                        failure_reason: format!("Expected ECash for MintV2 federation: {e}"),
                    }
                })?;
            let amount = ecash.amount();

            let operation_id = mint
                .receive(ecash, serde_json::Value::Null)
                .await
                .map_err(|e| PublicGatewayError::ReceiveEcashError {
                    failure_reason: e.fmt_compact().to_string(),
                })?;

            let final_state = mint
                .await_final_receive_operation_state(operation_id)
                .await
                .map_err(|e| PublicGatewayError::ReceiveEcashError {
                    failure_reason: e.fmt_compact().to_string(),
                })?;
            match final_state {
                fedimint_mintv2_client::FinalReceiveOperationState::Success => {}
                fedimint_mintv2_client::FinalReceiveOperationState::Rejected => {
                    return Err(PublicGatewayError::ReceiveEcashError {
                        failure_reason: "ECash receive was rejected".to_string(),
                    });
                }
            }

            Ok(ReceiveEcashResponse { amount })
        } else {
            Err(PublicGatewayError::ReceiveEcashError {
                failure_reason: "No mint module found".to_string(),
            })
        }
    }

    /// Retrieves an invoice by the payment hash if it exists, otherwise returns
    /// `None`.
    pub async fn handle_get_invoice_msg(
        &self,
        payload: GetInvoiceRequest,
    ) -> AdminResult<Option<GetInvoiceResponse>> {
        let lightning_context = self.get_lightning_context().await?;
        let invoice = lightning_context.lnrpc.get_invoice(payload).await?;
        Ok(invoice)
    }

    /// Withdraws ecash from a federation and pegs-out to the Lightning node's
    /// onchain wallet
    pub async fn handle_withdraw_to_onchain_msg(
        &self,
        payload: WithdrawToOnchainPayload,
    ) -> AdminResult<WithdrawResponse> {
        let address = self.handle_get_ln_onchain_address_msg().await?;
        let withdraw = WithdrawPayload {
            address: address.into_unchecked(),
            federation_id: payload.federation_id,
            amount: payload.amount,
            quoted_fees: None,
        };
        self.handle_withdraw_msg(withdraw).await
    }

    /// Deposits the specified amount from the gateway's onchain wallet into the
    /// Federation's ecash wallet
    pub async fn handle_pegin_from_onchain_msg(
        &self,
        payload: PeginFromOnchainPayload,
    ) -> AdminResult<Txid> {
        let deposit = DepositAddressPayload {
            federation_id: payload.federation_id,
        };
        let address = self.handle_address_msg(deposit).await?;
        let send_onchain = SendOnchainRequest {
            address: address.into_unchecked(),
            amount: payload.amount,
            fee_rate_sats_per_vbyte: payload.fee_rate_sats_per_vbyte,
        };
        let txid = self.handle_send_onchain_msg(send_onchain).await?;

        Ok(txid)
    }

    /// Registers the gateway with each specified federation.
    ///
    /// Does nothing once the gateway is shutting down: the lightning connection
    /// loop keeps reconnecting while `handle_shutdown_msg` drains the payments
    /// that are still in flight, and re-announcing there would advertise a
    /// route that is about to disappear, undoing the
    /// `unannounce_from_all_federations` the shutdown just performed.
    async fn register_federations(
        &self,
        federations: &BTreeMap<FederationId, FederationConfig>,
        register_task_group: &TaskGroup,
    ) {
        if let GatewayState::ShuttingDown { .. } = self.get_state().await {
            info!(
                target: LOG_GATEWAY,
                "Gateway is shutting down, skipping federation registration"
            );
            return;
        }

        if let Ok(lightning_context) = self.get_lightning_context().await {
            let route_hints = lightning_context
                .lnrpc
                .parsed_route_hints(self.num_route_hints)
                .await;
            if route_hints.is_empty() {
                warn!(target: LOG_GATEWAY, "Gateway did not retrieve any route hints, may reduce receive success rate.");
            }

            for (federation_id, federation_config) in federations {
                // A fee that predates the fee limits may be too large to announce. Skip
                // that federation rather than failing the whole registration pass: this
                // runs on the root task group at startup, so an error here would keep
                // the gateway from booting at all.
                let routing_fees = match RoutingFees::try_from(federation_config.lightning_fee) {
                    Ok(routing_fees) => routing_fees,
                    Err(err) => {
                        warn!(
                            target: LOG_GATEWAY,
                            %federation_id,
                            err = %err.fmt_compact(),
                            "Skipping registration, the configured lightning fee cannot be announced. Set a smaller fee with `set_fees`."
                        );
                        continue;
                    }
                };

                let fed_manager = self.federation_manager.read().await;
                if let Some(client) = fed_manager.client(federation_id) {
                    let federation_id = *federation_id;
                    let client_arc = client.clone().into_value();
                    let route_hints = route_hints.clone();
                    let lightning_context = lightning_context.clone();
                    let registration_health = self.registration_health.clone();
                    let registrations = self
                        .registrations
                        .clone()
                        .into_iter()
                        .map(|(protocol, registration)| {
                            let attempt =
                                registration_health.begin_lnv1_attempt(federation_id, protocol);
                            (registration, attempt)
                        })
                        .collect::<Vec<_>>();

                    register_task_group.spawn_cancellable_silent(
                        "register federation",
                        async move {
                            let Ok(gateway_client) =
                                client_arc.get_first_module::<GatewayClientModule>()
                            else {
                                return;
                            };

                            for (registration, attempt) in registrations {
                                let succeeded = gateway_client
                                    .try_register_with_federation(
                                        route_hints.clone(),
                                        GW_ANNOUNCEMENT_TTL,
                                        routing_fees,
                                        lightning_context.clone(),
                                        registration.endpoint_url,
                                        registration.keypair,
                                    )
                                    .await;
                                registration_health
                                    .complete_attempt(
                                        attempt,
                                        succeeded,
                                        fedimint_core::time::now(),
                                        fedimint_core::runtime::Instant::now(),
                                    )
                                    .await;
                            }
                        },
                    );
                }
            }
        }
    }

    /// Retrieves a `ClientHandleArc` from the Gateway's in memory structures
    /// that keep track of available clients, given a `federation_id`.
    pub async fn select_client(
        &self,
        federation_id: FederationId,
    ) -> std::result::Result<Spanned<fedimint_client::ClientHandleArc>, FederationNotConnected>
    {
        self.federation_manager
            .read()
            .await
            .client(&federation_id)
            .cloned()
            .ok_or(FederationNotConnected {
                federation_id_prefix: federation_id.to_prefix(),
            })
    }

    async fn load_mnemonic(gateway_db: &Database) -> Option<Mnemonic> {
        let secret = Client::load_decodable_client_secret::<Vec<u8>>(gateway_db)
            .await
            .ok()?;
        Mnemonic::from_entropy(&secret).ok()
    }

    /// Reads the connected federation client configs from the Gateway's
    /// database and reconstructs the clients necessary for interacting with
    /// connection federations.
    async fn load_clients(&self) -> AdminResult<()> {
        if let GatewayState::NotConfigured { .. } = self.get_state().await {
            return Ok(());
        }

        let mut federation_manager = self.federation_manager.write().await;

        let configs = {
            let mut dbtx = self.gateway_db.begin_transaction_nc().await;
            dbtx.load_federation_configs().await
        };

        if let Some(max_federation_index) = configs.values().map(|cfg| cfg.federation_index).max() {
            federation_manager.set_next_index(max_federation_index + 1);
        }

        let mnemonic = Self::load_mnemonic(&self.gateway_db)
            .await
            .expect("mnemonic should be set");

        for (federation_id, config) in configs {
            let federation_index = config.federation_index;
            match Box::pin(Spanned::try_new(
                info_span!(target: LOG_GATEWAY, "client", federation_id  = %federation_id.clone()),
                self.client_builder
                    .build(config, Arc::new(self.clone()), &mnemonic),
            ))
            .await
            {
                Ok(client) => {
                    federation_manager.add_client(federation_index, client);
                }
                _ => {
                    warn!(target: LOG_GATEWAY, federation_id = %federation_id, "Failed to load client");
                }
            }
        }

        Ok(())
    }

    /// Legacy mechanism for registering the Gateway with connected federations.
    /// This will spawn a task that will re-register the Gateway with
    /// connected federations every 8.5 mins. Only registers the Gateway if it
    /// has successfully connected to the Lightning node, so that it can
    /// include route hints in the registration.
    fn register_clients_timer(&self) {
        // Only spawn background registration thread if gateway is LND
        if matches!(self.lightning_mode, LightningMode::Lnd { .. }) {
            info!(target: LOG_GATEWAY, "Spawning register task...");
            let gateway = self.clone();
            let register_task_group = self.task_group.make_subgroup();
            self.task_group.spawn_cancellable("register clients", async move {
                loop {
                    let gateway_state = gateway.get_state().await;
                    if let GatewayState::Running { .. } = &gateway_state {
                        let mut dbtx = gateway.gateway_db.begin_transaction_nc().await;
                        let all_federations_configs = dbtx.load_federation_configs().await.into_iter().collect();
                        gateway.register_federations(&all_federations_configs, &register_task_group).await;
                    } else {
                        // We need to retry more often if the gateway is not in the Running state
                        const NOT_RUNNING_RETRY: Duration = Duration::from_secs(10);
                        warn!(target: LOG_GATEWAY, gateway_state = %gateway_state, retry_interval = ?NOT_RUNNING_RETRY, "Will not register federation yet because gateway still not in Running state");
                        sleep(NOT_RUNNING_RETRY).await;
                        continue;
                    }

                    // Allow a 15% buffer of the TTL before the re-registering gateway
                    // with the federations.
                    sleep(GW_ANNOUNCEMENT_TTL.mul_f32(0.85)).await;
                }
            });
        }
    }

    /// Verifies that the federation has at least one lightning module (LNv1 or
    /// LNv2) and that the network matches the gateway's network.
    async fn check_federation_network(
        client: &ClientHandleArc,
        network: Network,
    ) -> AdminResult<()> {
        let federation_id = client.federation_id();
        let config = client.config().await;

        let lnv1_cfg = config
            .modules
            .values()
            .find(|m| LightningCommonInit::KIND == m.kind);

        let lnv2_cfg = config
            .modules
            .values()
            .find(|m| fedimint_lnv2_common::LightningCommonInit::KIND == m.kind);

        // Ensure the federation has at least one lightning module
        if lnv1_cfg.is_none() && lnv2_cfg.is_none() {
            return Err(AdminGatewayError::ClientCreationError(anyhow!(
                "Federation {federation_id} does not have any lightning module (LNv1 or LNv2)"
            )));
        }

        // Verify the LNv1 network if present
        if let Some(cfg) = lnv1_cfg {
            let ln_cfg: &LightningClientConfig = cfg.cast().map_err(anyhow::Error::from)?;

            if ln_cfg.network.0 != network {
                crit!(
                    target: LOG_GATEWAY,
                    federation_id = %federation_id,
                    network = %network,
                    "Incorrect LNv1 network for federation",
                );
                return Err(AdminGatewayError::ClientCreationError(anyhow!(format!(
                    "Unsupported LNv1 network {}",
                    ln_cfg.network
                ))));
            }
        }

        // Verify the LNv2 network if present
        if let Some(cfg) = lnv2_cfg {
            let ln_cfg: &fedimint_lnv2_common::config::LightningClientConfig =
                cfg.cast().map_err(anyhow::Error::from)?;

            if ln_cfg.network != network {
                crit!(
                    target: LOG_GATEWAY,
                    federation_id = %federation_id,
                    network = %network,
                    "Incorrect LNv2 network for federation",
                );
                return Err(AdminGatewayError::ClientCreationError(anyhow!(format!(
                    "Unsupported LNv2 network {}",
                    ln_cfg.network
                ))));
            }
        }

        Ok(())
    }

    /// Checks the Gateway's current state and returns the proper
    /// `LightningContext` if it is available.
    ///
    /// The error is synthesised from the gateway's own state: no RPC is
    /// attempted, so `Err` means "this process does not currently hold a
    /// session with the lightning node", never "the lightning node was asked
    /// and answered no". Callers that would turn a failure here into a
    /// decision about a payment must use `await_lightning_context`
    /// instead.
    pub async fn get_lightning_context(
        &self,
    ) -> std::result::Result<LightningContext, LightningRpcError> {
        match self.get_state().await {
            GatewayState::Running { lightning_context }
            | GatewayState::ShuttingDown { lightning_context } => Ok(lightning_context),
            _ => Err(LightningRpcError::FailedToConnect),
        }
    }

    /// Waits until the gateway holds a `LightningContext` and returns it.
    ///
    /// The lightning node is the only oracle for whether an HTLC of ours is in
    /// flight, so code deciding the fate of a payment must actually ask it.
    /// [`Self::get_lightning_context`] cannot stand in for that: its `Err` is
    /// produced locally, and the gateway spends part of every startup without
    /// a context. [`Self::run`] awaits `load_clients` before `start_gateway`,
    /// and building a client starts its executor, so payment state machines
    /// persisted across a restart re-enter while the state is still
    /// `Disconnected`. Reading that as a payment failure cancels an outgoing
    /// contract whose HTLC the previous process may already have settled,
    /// leaving the gateway out of pocket for a payment it did make.
    ///
    /// Waiting is the conservative side of that trade. It ends when the
    /// gateway connects, or when the caller is dropped: every caller runs
    /// inside a client state machine transition or a webserver request, both
    /// of which are cancelled when the gateway shuts down. It does not strand
    /// the payer either, since the outgoing contract's timelock refunds them
    /// without the gateway's cooperation, whereas a cancellation is final (see
    /// `LightningInput` processing in `fedimint-ln-server`).
    async fn await_lightning_context(&self) -> LightningContext {
        loop {
            match self.get_lightning_context().await {
                Ok(lightning_context) => return lightning_context,
                Err(err) => {
                    let state = self.get_state().await;

                    warn!(
                        target: LOG_GATEWAY,
                        err = %err.fmt_compact(),
                        %state,
                        retry_interval_secs = LIGHTNING_CONTEXT_RETRY_INTERVAL.as_secs(),
                        "Not connected to the lightning node, waiting before asking it again",
                    );

                    sleep(LIGHTNING_CONTEXT_RETRY_INTERVAL).await;
                }
            }
        }
    }

    /// Iterates through all of the federations the gateway is registered with
    /// and requests to remove the registration record.
    pub async fn unannounce_from_all_federations(&self) {
        if matches!(self.lightning_mode, LightningMode::Lnd { .. }) {
            for registration in self.registrations.values() {
                self.federation_manager
                    .read()
                    .await
                    .unannounce_from_all_federations(registration.keypair)
                    .await;
            }
        }
    }

    async fn create_lightning_client(
        &self,
        runtime: Arc<tokio::runtime::Runtime>,
        mnemonic_source: MnemonicSource,
    ) -> Box<dyn ILnRpcClient> {
        match self.lightning_mode.clone() {
            LightningMode::Lnd {
                lnd_rpc_addr,
                lnd_tls_cert,
                lnd_macaroon,
                lnd_time_pref,
                lnd_payment_timeout_secs,
            } => {
                // The LND backend uses this to ignore HOLD invoices on the
                // shared LND node that aren't federation-bound. Returns true
                // iff there is a registered LNv2 incoming contract for the
                // given payment hash.
                let gateway_db = self.gateway_db.clone();
                let lnv2_filter: Lnv2HoldInvoiceFilter = Arc::new(move |hash| {
                    let gateway_db = gateway_db.clone();
                    Box::pin(async move {
                        gateway_db
                            .begin_transaction_nc()
                            .await
                            .load_registered_incoming_contract(PaymentImage::Hash(hash))
                            .await
                            .is_some()
                    })
                });

                Box::new(GatewayLndClient::new(
                    lnd_rpc_addr,
                    lnd_tls_cert,
                    lnd_macaroon,
                    lnd_time_pref,
                    lnd_payment_timeout_secs,
                    None,
                    lnv2_filter,
                ))
            }
            LightningMode::Ldk {
                lightning_port,
                alias,
            } => {
                let mnemonic = Self::load_mnemonic(&self.gateway_db)
                    .await
                    .expect("mnemonic should be set");
                // Retrieving the fees inside of LDK can sometimes fail/time out. To prevent
                // crashing the gateway, we wait a bit and just try
                // to re-create the client. The gateway cannot proceed until this succeeds.
                retry("create LDK Node", fibonacci_max_one_hour(), || async {
                    ldk::GatewayLdkClient::new(
                        &self.client_builder.data_dir().join(LDK_NODE_DB_FOLDER),
                        self.chain_source.clone(),
                        self.network,
                        lightning_port,
                        alias.clone(),
                        mnemonic.clone(),
                        mnemonic_source,
                        runtime.clone(),
                    )
                    .map(Box::new)
                    // `retry` logs each failure with `{:#}`, which prints the
                    // cause chain only for `anyhow::Error`.
                    .map_err(anyhow::Error::from)
                })
                .await
                .expect("Could not create LDK Node")
            }
        }
    }
}

#[async_trait]
impl IAdminGateway for Gateway {
    type Error = AdminGatewayError;

    /// Returns information about the Gateway back to the client when requested
    /// via the webserver.
    async fn handle_get_info(&self) -> AdminResult<GatewayInfo> {
        let GatewayState::Running { lightning_context } = self.get_state().await else {
            return Ok(GatewayInfo {
                federations: vec![],
                federation_fake_scids: None,
                version_hash: fedimint_build_code_version_env!().to_string(),
                gateway_state: self.state.read().await.to_string(),
                lightning_info: LightningInfo::NotConnected,
                lightning_mode: self.lightning_mode.clone(),
                registrations: self
                    .registrations
                    .iter()
                    .map(|(k, v)| (k.clone(), (v.endpoint_url.clone(), v.keypair.public_key())))
                    .collect(),
            });
        };

        let dbtx = self.gateway_db.begin_transaction_nc().await;
        let federations = self
            .federation_manager
            .read()
            .await
            .federation_info_all_federations(dbtx)
            .await;

        let channels: BTreeMap<u64, FederationId> = federations
            .iter()
            .map(|federation_info| {
                (
                    federation_info.config.federation_index,
                    federation_info.federation_id,
                )
            })
            .collect();

        let lightning_info = lightning_context.lnrpc.parsed_node_info().await;

        Ok(GatewayInfo {
            federations,
            federation_fake_scids: Some(channels),
            version_hash: fedimint_build_code_version_env!().to_string(),
            gateway_state: self.state.read().await.to_string(),
            lightning_info,
            lightning_mode: self.lightning_mode.clone(),
            registrations: self
                .registrations
                .iter()
                .map(|(k, v)| (k.clone(), (v.endpoint_url.clone(), v.keypair.public_key())))
                .collect(),
        })
    }

    /// Returns a list of Lightning network channels from the Gateway's
    /// Lightning node.
    async fn handle_list_channels_msg(
        &self,
    ) -> AdminResult<Vec<fedimint_gateway_common::ChannelInfo>> {
        let context = self.get_lightning_context().await?;
        let response = context.lnrpc.list_channels().await?;
        Ok(response.channels)
    }

    /// Computes the 24 hour payment summary statistics for this gateway.
    /// Combines the LNv1 and LNv2 stats together.
    async fn handle_payment_summary_msg(
        &self,
        PaymentSummaryPayload {
            start_millis,
            end_millis,
        }: PaymentSummaryPayload,
    ) -> AdminResult<PaymentSummaryResponse> {
        let federation_manager = self.federation_manager.read().await;
        let fed_configs = federation_manager.get_all_federation_configs().await;
        let federation_ids = fed_configs.keys().collect::<Vec<_>>();
        let start = UNIX_EPOCH + Duration::from_millis(start_millis);
        let end = UNIX_EPOCH + Duration::from_millis(end_millis);

        if start > end {
            return Err(AdminGatewayError::Unexpected(anyhow!("Invalid time range")));
        }

        let mut outgoing = StructuredPaymentEvents::default();
        let mut incoming = StructuredPaymentEvents::default();
        for fed_id in federation_ids {
            let client = federation_manager
                .client(fed_id)
                .expect("No client available")
                .value();
            let all_events = &get_events_for_duration(client, start, end).await;

            let (mut lnv1_outgoing, mut lnv1_incoming) = compute_lnv1_stats(all_events);
            let (mut lnv2_outgoing, mut lnv2_incoming) = compute_lnv2_stats(all_events);
            outgoing.combine(&mut lnv1_outgoing);
            incoming.combine(&mut lnv1_incoming);
            outgoing.combine(&mut lnv2_outgoing);
            incoming.combine(&mut lnv2_incoming);
        }

        Ok(PaymentSummaryResponse {
            outgoing: PaymentStats::compute(&outgoing),
            incoming: PaymentStats::compute(&incoming),
        })
    }

    /// Handle a request to have the Gateway leave a federation. The Gateway
    /// will request the federation to remove the registration record and
    /// the gateway will remove the configuration needed to construct the
    /// federation client.
    async fn handle_leave_federation(
        &self,
        payload: LeaveFedPayload,
    ) -> AdminResult<FederationInfo> {
        // Lock the federation manager before starting the db transaction to reduce the
        // chance of db write conflicts.
        let mut federation_manager = self.federation_manager.write().await;
        let mut dbtx = self.gateway_db.begin_transaction().await;

        let federation_info = federation_manager
            .leave_federation(
                payload.federation_id,
                &mut dbtx.to_ref_nc(),
                self.registrations.values().collect(),
            )
            .await?;

        dbtx.remove_federation_config(payload.federation_id).await;
        dbtx.commit_tx().await;
        self.registration_health
            .clear_federation(payload.federation_id)
            .await;
        Ok(federation_info)
    }

    /// Handles a connection request to join a new federation. The gateway will
    /// download the federation's client configuration, construct a new
    /// client, registers, the gateway with the federation, and persists the
    /// necessary config to reconstruct the client when restarting the gateway.
    async fn handle_connect_federation(
        &self,
        payload: ConnectFedPayload,
    ) -> AdminResult<FederationInfo> {
        let GatewayState::Running { lightning_context } = self.get_state().await else {
            return Err(AdminGatewayError::Lightning(
                LightningRpcError::FailedToConnect,
            ));
        };

        let invite_code = InviteCode::from_str(&payload.invite_code).map_err(|e| {
            AdminGatewayError::ClientCreationError(anyhow!(format!(
                "Invalid federation member string {e:?}"
            )))
        })?;

        let federation_id = invite_code.federation_id();

        let mut federation_manager = self.federation_manager.write().await;

        // Check if this federation has already been registered
        if federation_manager.has_federation(federation_id) {
            return Err(AdminGatewayError::ClientCreationError(anyhow!(
                "Federation has already been registered"
            )));
        }

        // The gateway deterministically assigns a unique identifier (u64) to each
        // federation connected.
        let federation_index = federation_manager.pop_next_index()?;

        let federation_config = FederationConfig {
            invite_code,
            federation_index,
            lightning_fee: self.default_routing_fees,
            transaction_fee: self.default_transaction_fees,
            payment_policies: BTreeSet::new(),
            // Note: deprecated, unused
            _connector: ConnectorType::Tcp,
        };

        // The default fees are validated at startup, so this only fails if the gateway
        // is running with fees that predate that check. Refuse to join a federation
        // whose fee could not be announced rather than persisting its config.
        let routing_fees = RoutingFees::try_from(federation_config.lightning_fee)
            .map_err(|err| AdminGatewayError::GatewayConfigurationError(err.to_string()))?;

        let mnemonic = Self::load_mnemonic(&self.gateway_db)
            .await
            .expect("mnemonic should be set");
        let recover = payload.recover.unwrap_or(false);
        if recover {
            self.client_builder
                .recover(federation_config.clone(), Arc::new(self.clone()), &mnemonic)
                .await?;
        }

        let client = self
            .client_builder
            .build(federation_config.clone(), Arc::new(self.clone()), &mnemonic)
            .await?;

        if recover {
            client.wait_for_all_active_state_machines().await;
        }

        // Instead of using `FederationManager::federation_info`, we manually create
        // federation info here because short channel id is not yet persisted.
        let federation_info = FederationInfo {
            federation_id,
            federation_name: federation_manager.federation_name(&client).await,
            balance_msat: client.get_balance_for_btc().await.unwrap_or_else(|err| {
                warn!(
                    target: LOG_GATEWAY,
                    err = %err.fmt_compact(),
                    %federation_id,
                    "Balance not immediately available after joining/recovering."
                );
                Amount::default()
            }),
            config: federation_config.clone(),
            last_backup_time: None,
        };

        Self::check_federation_network(&client, self.network).await?;
        if matches!(self.lightning_mode, LightningMode::Lnd { .. })
            && let Ok(lnv1) = client.get_first_module::<GatewayClientModule>()
        {
            for (protocol, registration) in &self.registrations {
                let attempt = self
                    .registration_health
                    .begin_lnv1_attempt(federation_id, protocol.clone());
                let succeeded = lnv1
                    .try_register_with_federation(
                        // Route hints will be updated in the background
                        Vec::new(),
                        GW_ANNOUNCEMENT_TTL,
                        routing_fees,
                        lightning_context.clone(),
                        registration.endpoint_url.clone(),
                        registration.keypair,
                    )
                    .await;
                self.registration_health
                    .complete_attempt(
                        attempt,
                        succeeded,
                        fedimint_core::time::now(),
                        fedimint_core::runtime::Instant::now(),
                    )
                    .await;
            }
        }

        // no need to enter span earlier, because connect-fed has a span
        federation_manager.add_client(
            federation_index,
            Spanned::new(
                info_span!(target: LOG_GATEWAY, "client", federation_id=%federation_id.clone()),
                async { client },
            )
            .await,
        );

        let mut dbtx = self.gateway_db.begin_transaction().await;
        dbtx.save_federation_config(&federation_config).await;
        dbtx.save_federation_backup_record(federation_id, None)
            .await;
        dbtx.commit_tx().await;
        debug!(
            target: LOG_GATEWAY,
            federation_id = %federation_id,
            federation_index = %federation_index,
            "Federation connected"
        );

        Ok(federation_info)
    }

    /// Handles a request to change the lightning or transaction fees for all
    /// federations or a federation specified by the `FederationId`.
    async fn handle_set_fees_msg(
        &self,
        SetFeesPayload {
            federation_id,
            lightning_base,
            lightning_parts_per_million,
            transaction_base,
            transaction_parts_per_million,
        }: SetFeesPayload,
    ) -> AdminResult<()> {
        let mut dbtx = self.gateway_db.begin_transaction().await;
        let mut fed_configs = if let Some(fed_id) = federation_id {
            dbtx.load_federation_configs()
                .await
                .into_iter()
                .filter(|(id, _)| *id == fed_id)
                .collect::<BTreeMap<_, _>>()
        } else {
            dbtx.load_federation_configs().await
        };

        // Silently succeeding on an unknown federation would let a mistyped id
        // pass for fees that were never changed.
        if let Some(fed_id) = federation_id
            && fed_configs.is_empty()
        {
            return Err(FederationNotConnected {
                federation_id_prefix: fed_id.to_prefix(),
            }
            .into());
        }

        let federation_manager = self.federation_manager.read().await;

        for (federation_id, config) in &mut fed_configs {
            let mut lightning_fee = config.lightning_fee;
            if let Some(lightning_base) = lightning_base {
                lightning_fee.base = lightning_base;
            }

            if let Some(lightning_ppm) = lightning_parts_per_million {
                lightning_fee.parts_per_million = lightning_ppm;
            }

            let mut transaction_fee = config.transaction_fee;
            if let Some(transaction_base) = transaction_base {
                transaction_fee.base = transaction_base;
            }

            if let Some(transaction_ppm) = transaction_parts_per_million {
                transaction_fee.parts_per_million = transaction_ppm;
            }

            // Changing the fees of a federation the gateway is not connected to is
            // rejected, as it was before the fee limits applied to every federation.
            federation_manager
                .client(federation_id)
                .ok_or(FederationNotConnected {
                    federation_id_prefix: federation_id.to_prefix(),
                })?;

            // The limits are enforced for every federation, not just the ones running
            // LNv2. An unchecked fee is persisted verbatim and is converted into
            // `RoutingFees` on every LNv1 payment and on every registration, including
            // the one at startup, so a fee that does not fit into the `u32` components
            // of `RoutingFees` would leave the gateway unable to boot.
            let send_fees = lightning_fee.checked_add(transaction_fee).ok_or_else(|| {
                AdminGatewayError::GatewayConfigurationError(format!(
                    "Total Send fees overflowed, they may not exceed {}",
                    PaymentFee::SEND_FEE_LIMIT
                ))
            })?;

            // Check if the lightning fee + transaction fee is higher than the send limit
            if !send_fees.is_within(&PaymentFee::SEND_FEE_LIMIT) {
                return Err(AdminGatewayError::GatewayConfigurationError(format!(
                    "Total Send fees exceeded {}",
                    PaymentFee::SEND_FEE_LIMIT
                )));
            }

            // Check if the transaction fee is higher than the receive limit
            if !transaction_fee.is_within(&PaymentFee::RECEIVE_FEE_LIMIT) {
                return Err(AdminGatewayError::GatewayConfigurationError(format!(
                    "Transaction fees exceeded RECEIVE LIMIT {}",
                    PaymentFee::RECEIVE_FEE_LIMIT
                )));
            }

            config.lightning_fee = lightning_fee;
            config.transaction_fee = transaction_fee;
            dbtx.save_federation_config(config).await;
        }

        dbtx.commit_tx().await;

        if matches!(self.lightning_mode, LightningMode::Lnd { .. }) {
            let register_task_group = TaskGroup::new();

            self.register_federations(&fed_configs, &register_task_group)
                .await;
        }

        Ok(())
    }

    /// Handles a request to change which payments the gateway performs on
    /// behalf of the clients of all federations or of the federation specified
    /// by the `FederationId`. Like `set_fees`, only the settings present in
    /// the payload change.
    async fn handle_set_payment_policy_msg(
        &self,
        SetPaymentPolicyPayload {
            federation_id,
            receive_enabled,
        }: SetPaymentPolicyPayload,
    ) -> AdminResult<()> {
        // Taken before the policy is read anywhere else: every payment admitted
        // under the old policy funds before this returns, and none is admitted
        // afterwards.
        let _policy_change = self.payment_policy_lock.write().await;

        let mut dbtx = self.gateway_db.begin_transaction().await;
        let mut fed_configs = if let Some(fed_id) = federation_id {
            dbtx.load_federation_configs()
                .await
                .into_iter()
                .filter(|(id, _)| *id == fed_id)
                .collect::<BTreeMap<_, _>>()
        } else {
            dbtx.load_federation_configs().await
        };

        // Silently succeeding on an unknown federation would let a mistyped id
        // pass for a switch that was never flipped.
        if let Some(fed_id) = federation_id
            && fed_configs.is_empty()
        {
            return Err(FederationNotConnected {
                federation_id_prefix: fed_id.to_prefix(),
            }
            .into());
        }

        let federation_manager = self.federation_manager.read().await;

        for (federation_id, config) in &mut fed_configs {
            federation_manager
                .client(federation_id)
                .ok_or(FederationNotConnected {
                    federation_id_prefix: federation_id.to_prefix(),
                })?;

            if let Some(receive_enabled) = receive_enabled {
                if receive_enabled {
                    config
                        .payment_policies
                        .remove(&PaymentPolicy::ReceivesDisabled);
                } else {
                    config
                        .payment_policies
                        .insert(PaymentPolicy::ReceivesDisabled);
                }
            }

            info!(
                target: LOG_GATEWAY,
                %federation_id,
                payment_policies = ?config.payment_policies,
                "Updated federation payment policy"
            );

            dbtx.save_federation_config(config).await;
        }

        dbtx.commit_tx().await;

        Ok(())
    }

    /// Handles an authenticated request for the gateway's mnemonic. This also
    /// returns a vector of federations that are not using the mnemonic
    /// backup strategy.
    async fn handle_mnemonic_msg(&self) -> AdminResult<MnemonicResponse> {
        let mnemonic = Self::load_mnemonic(&self.gateway_db)
            .await
            .expect("mnemonic should be set");
        let words = mnemonic
            .words()
            .map(std::string::ToString::to_string)
            .collect::<Vec<_>>();
        let all_federations = self
            .federation_manager
            .read()
            .await
            .get_all_federation_configs()
            .await
            .keys()
            .copied()
            .collect::<BTreeSet<_>>();
        let legacy_federations = self.client_builder.legacy_federations(all_federations);
        let mnemonic_response = MnemonicResponse {
            mnemonic: words,
            legacy_federations,
        };
        Ok(mnemonic_response)
    }

    /// Instructs the Gateway's Lightning node to open a channel to a peer
    /// specified by `pubkey`.
    async fn handle_open_channel_msg(&self, payload: OpenChannelRequest) -> AdminResult<Txid> {
        info!(target: LOG_GATEWAY, pubkey = %payload.pubkey, host = %payload.host, amount = %payload.channel_size_sats, "Opening Lightning channel...");
        let context = self.get_lightning_context().await?;
        let res = context.lnrpc.open_channel(payload).await?;
        info!(target: LOG_GATEWAY, txid = %res.funding_txid, "Initiated channel open");
        Txid::from_str(&res.funding_txid).map_err(|e| {
            AdminGatewayError::Lightning(LightningRpcError::InvalidMetadata {
                failure_reason: format!("Received invalid channel funding txid string {e}"),
            })
        })
    }

    /// Instructs the Gateway's Lightning node to connect to a peer specified by
    /// `pubkey`.
    async fn handle_connect_peer_msg(&self, payload: ConnectPeerRequest) -> AdminResult<()> {
        info!(
            target: LOG_GATEWAY,
            pubkey = %payload.node_address.pubkey,
            host = %payload.node_address.host_with_port(),
            "Connecting to Lightning peer..."
        );
        let context = self.get_lightning_context().await?;
        context.lnrpc.connect_peer(payload).await?;
        info!(target: LOG_GATEWAY, "Connected to Lightning peer");
        Ok(())
    }

    /// Instructs the Gateway's Lightning node to close all channels with a peer
    /// specified by `pubkey`.
    async fn handle_close_channels_with_peer_msg(
        &self,
        payload: CloseChannelsWithPeerRequest,
    ) -> AdminResult<CloseChannelsWithPeerResponse> {
        info!(target: LOG_GATEWAY, close_channel_request = %payload, "Closing lightning channel...");
        let context = self.get_lightning_context().await?;
        let response = context
            .lnrpc
            .close_channels_with_peer(payload.clone())
            .await?;
        info!(target: LOG_GATEWAY, close_channel_request = %payload, "Initiated channel closure");
        Ok(response)
    }

    /// Updates the local-side routing fees (base + ppm) on a single channel
    /// identified by funding outpoint.
    async fn handle_set_channel_fees_msg(&self, payload: SetChannelFeesRequest) -> AdminResult<()> {
        info!(
            target: LOG_GATEWAY,
            funding_outpoint = %payload.funding_outpoint,
            base_fee_msat = payload.base_fee_msat,
            parts_per_million = payload.parts_per_million,
            "Updating channel fees..."
        );
        let context = self.get_lightning_context().await?;
        context.lnrpc.set_channel_fees(payload).await?;
        Ok(())
    }

    /// Returns the ecash, lightning, and onchain balances for the gateway and
    /// the gateway's lightning node.
    async fn handle_get_balances_msg(&self) -> AdminResult<GatewayBalances> {
        let dbtx = self.gateway_db.begin_transaction_nc().await;
        let federation_infos = self
            .federation_manager
            .read()
            .await
            .federation_info_all_federations(dbtx)
            .await;

        let ecash_balances: Vec<FederationBalanceInfo> = federation_infos
            .iter()
            .map(|federation_info| FederationBalanceInfo {
                federation_id: federation_info.federation_id,
                ecash_balance_msats: Amount {
                    msats: federation_info.balance_msat.msats,
                },
            })
            .collect();

        let context = self.get_lightning_context().await?;
        let lightning_node_balances = context.lnrpc.get_balances().await?;

        Ok(GatewayBalances {
            onchain_balance_sats: lightning_node_balances.onchain_balance_sats,
            lightning_balance_msats: lightning_node_balances.lightning_balance_msats,
            ecash_balances,
            inbound_lightning_liquidity_msats: lightning_node_balances
                .inbound_lightning_liquidity_msats,
        })
    }

    /// Send funds from the gateway's lightning node on-chain wallet.
    async fn handle_send_onchain_msg(&self, payload: SendOnchainRequest) -> AdminResult<Txid> {
        let context = self.get_lightning_context().await?;
        let response = context.lnrpc.send_onchain(payload.clone()).await?;
        let txid =
            Txid::from_str(&response.txid).map_err(|e| AdminGatewayError::WithdrawError {
                failure_reason: format!("Failed to parse withdrawal TXID: {e}"),
            })?;
        info!(onchain_request = %payload, txid = %txid, "Sent onchain transaction");
        Ok(txid)
    }

    /// Generates an onchain address to fund the gateway's lightning node.
    async fn handle_get_ln_onchain_address_msg(&self) -> AdminResult<Address> {
        let context = self.get_lightning_context().await?;
        let response = context.lnrpc.get_ln_onchain_address().await?;

        let address = Address::from_str(&response.address).map_err(|e| {
            AdminGatewayError::Lightning(LightningRpcError::InvalidMetadata {
                failure_reason: e.to_string(),
            })
        })?;

        address.require_network(self.network).map_err(|e| {
            AdminGatewayError::Lightning(LightningRpcError::InvalidMetadata {
                failure_reason: e.to_string(),
            })
        })
    }

    async fn handle_deposit_address_msg(
        &self,
        payload: DepositAddressPayload,
    ) -> AdminResult<Address> {
        self.handle_address_msg(payload).await
    }

    async fn handle_receive_ecash_msg(
        &self,
        payload: ReceiveEcashPayload,
    ) -> AdminResult<ReceiveEcashResponse> {
        Self::handle_receive_ecash_msg(self, payload)
            .await
            .map_err(|e| AdminGatewayError::Unexpected(anyhow::anyhow!("{e}")))
    }

    /// Creates an invoice that is directly payable to the gateway's lightning
    /// node.
    async fn handle_create_invoice_for_operator_msg(
        &self,
        payload: CreateInvoiceForOperatorPayload,
    ) -> AdminResult<Bolt11Invoice> {
        let GatewayState::Running { lightning_context } = self.get_state().await else {
            return Err(AdminGatewayError::Lightning(
                LightningRpcError::FailedToConnect,
            ));
        };

        Bolt11Invoice::from_str(
            &lightning_context
                .lnrpc
                .create_invoice(CreateInvoiceRequest {
                    payment_hash: None, /* Empty payment hash indicates an invoice payable
                                         * directly to the gateway. */
                    amount_msat: payload.amount_msats,
                    expiry_secs: payload.expiry_secs.unwrap_or(3600),
                    description: payload.description.map(InvoiceDescription::Direct),
                })
                .await?
                .invoice,
        )
        .map_err(|e| {
            AdminGatewayError::Lightning(LightningRpcError::InvalidMetadata {
                failure_reason: e.to_string(),
            })
        })
    }

    /// Requests the gateway to pay an outgoing LN invoice using its own funds.
    /// Returns the payment hash's preimage on success.
    async fn handle_pay_invoice_for_operator_msg(
        &self,
        payload: PayInvoiceForOperatorPayload,
    ) -> AdminResult<Preimage> {
        // Those are the ldk defaults
        const BASE_FEE: u64 = 50;
        const FEE_DENOMINATOR: u64 = 100;
        const MAX_DELAY: u64 = 1008;

        let GatewayState::Running { lightning_context } = self.get_state().await else {
            return Err(AdminGatewayError::Lightning(
                LightningRpcError::FailedToConnect,
            ));
        };

        let max_fee = BASE_FEE
            + payload
                .invoice
                .amount_milli_satoshis()
                .context("Invoice is missing amount")?
                .saturating_div(FEE_DENOMINATOR);

        let res = lightning_context
            .lnrpc
            .pay(payload.invoice, MAX_DELAY, Amount::from_msats(max_fee))
            .await?;
        Ok(res.preimage)
    }

    /// Lists the transactions that the lightning node has made.
    async fn handle_list_transactions_msg(
        &self,
        payload: ListTransactionsPayload,
    ) -> AdminResult<ListTransactionsResponse> {
        let lightning_context = self.get_lightning_context().await?;
        let response = lightning_context
            .lnrpc
            .list_transactions(payload.start_secs, payload.end_secs)
            .await?;
        Ok(response)
    }

    // Handles a request the spend the gateway's ecash for a given federation.
    async fn handle_spend_ecash_msg(
        &self,
        payload: SpendEcashPayload,
    ) -> AdminResult<SpendEcashResponse> {
        let client = self
            .select_client(payload.federation_id)
            .await?
            .into_value();

        if let Ok(mint_module) = client.get_first_module::<MintClientModule>() {
            let notes = mint_module
                .send_oob_notes(payload.amount, ())
                .await
                .map_err(|e| AdminGatewayError::Unexpected(e.into()))?;
            debug!(target: LOG_GATEWAY, ?notes, "Spend ecash notes");
            Ok(SpendEcashResponse {
                notes: notes.to_string(),
            })
        } else if let Ok(mint_module) = client.get_primary_module_for_unit::<MintV2ClientModule>(
            fedimint_core::module::AmountUnit::BITCOIN,
        ) {
            let (_, ecash) = mint_module
                .send(payload.amount, serde_json::Value::Null, true)
                .await
                .map_err(|e| AdminGatewayError::Unexpected(e.into()))?;

            Ok(SpendEcashResponse {
                notes: base32::encode_prefixed(FEDIMINT_PREFIX, &ecash),
            })
        } else {
            Err(AdminGatewayError::Unexpected(anyhow::anyhow!(
                "No mint module available"
            )))
        }
    }

    /// Instructs the gateway to shutdown, but only after all incoming payments
    /// have been handled.
    async fn handle_shutdown_msg(&self, task_group: TaskGroup) -> AdminResult<()> {
        // Take the write lock on the state so that no additional payments are
        // processed. `ShuttingDown` is terminal, so the state cannot move back to
        // `Running` once this returns.
        let was_running = {
            let mut state_guard = self.state.write().await;
            if let GatewayState::Running { lightning_context } = state_guard.clone() {
                *state_guard = GatewayState::ShuttingDown { lightning_context };
                true
            } else {
                false
            }
        };

        // The guard has to be released before waiting. Finishing an incoming payment
        // that already bought the preimage from the federation goes through
        // `complete_htlc`, which loops on `get_lightning_context` and would block on
        // the write guard forever. `/stop` would never return, the HTLC would expire,
        // and the gateway would be left having spent ecash for a payment its sender
        // gets refunded. `get_lightning_context` accepts `ShuttingDown`, so the
        // in-flight payments can complete while the gateway drains.
        if was_running {
            self.federation_manager
                .read()
                .await
                .wait_for_incoming_payments()
                .await?;
        }

        let tg = task_group.clone();
        tg.spawn("Kill Gateway", |_task_handle| async {
            if let Err(err) = task_group.shutdown_join_all(Duration::from_mins(3)).await {
                warn!(target: LOG_GATEWAY, err = %err.fmt_compact(), "Error shutting down gateway");
            }
        });
        Ok(())
    }

    fn get_task_group(&self) -> TaskGroup {
        self.task_group.clone()
    }

    /// Returns a Bitcoin TXID from a peg-out transaction for a specific
    /// connected federation.
    async fn handle_withdraw_msg(&self, payload: WithdrawPayload) -> AdminResult<WithdrawResponse> {
        let WithdrawPayload {
            amount,
            address,
            federation_id,
            quoted_fees,
        } = payload;

        let address_network = get_network_for_address(&address);
        let gateway_network = self.network;
        let Ok(address) = address.require_network(gateway_network) else {
            return Err(AdminGatewayError::WithdrawError {
                failure_reason: format!(
                    "Gateway is running on network {gateway_network}, but provided withdraw address is for network {address_network}"
                ),
            });
        };

        let client = self.select_client(federation_id).await?;

        if let Ok(wallet_module) = client
            .value()
            .get_first_module::<fedimint_walletv2_client::WalletClientModule>()
        {
            // The UI round-trips the walletv2 flat fee through
            // `PegOutFees::from_amount`, so `amount()` recovers it exactly.
            let quoted_fee = quoted_fees.map(|fees| fees.amount());
            return withdraw_v2(client.value(), &wallet_module, &address, amount, quoted_fee).await;
        }

        let wallet_module = client
            .value()
            .get_first_module::<WalletClientModule>()
            .map_err(anyhow::Error::from)?;

        // If fees are provided (from UI preview flow), use them directly
        // Otherwise fetch fees (CLI backwards compatibility)
        let (withdraw_amount, fees) = match quoted_fees {
            // UI flow: user confirmed these exact values, just use them
            Some(fees) => {
                let amt = match amount {
                    BitcoinAmountOrAll::Amount(a) => a,
                    BitcoinAmountOrAll::All => {
                        // UI always resolves "all" to specific amount in preview - reject if not
                        return Err(AdminGatewayError::WithdrawError {
                            failure_reason:
                                "Cannot use 'all' with quoted fees - amount must be resolved first"
                                    .to_string(),
                        });
                    }
                };
                (amt, fees)
            }
            // CLI flow: fetch fees (existing behavior for backwards compatibility)
            None => match amount {
                // The on-chain fee is only part of the cost of withdrawing
                // everything: funding the peg-out output also incurs the
                // federation's per-note fees. The returned fees are quoted at
                // the returned amount, so they must be used together.
                BitcoinAmountOrAll::All => {
                    let balance = client.value().get_balance_for_btc().await.map_err(|err| {
                        AdminGatewayError::Unexpected(anyhow!(
                            "Balance not available: {}",
                            err.fmt_compact()
                        ))
                    })?;

                    wallet_module
                        .max_withdrawable_amount(&address, balance)
                        .await
                        .map_err(|err| AdminGatewayError::WithdrawError {
                            failure_reason: format!(
                                "Insufficient funds. Balance: {balance}: {}",
                                err.fmt_compact()
                            ),
                        })?
                }
                BitcoinAmountOrAll::Amount(amount) => (
                    amount,
                    wallet_module
                        .get_withdraw_fees(&address, amount)
                        .await
                        .map_err(|e| AdminGatewayError::Unexpected(e.into()))?,
                ),
            },
        };

        let operation_id = wallet_module
            .withdraw(&address, withdraw_amount, fees, ())
            .await
            .map_err(|e| AdminGatewayError::Unexpected(e.into()))?;
        let mut updates = wallet_module
            .subscribe_withdraw_updates(operation_id)
            .await
            .map_err(|e| AdminGatewayError::Unexpected(e.into()))?
            .into_stream();

        while let Some(update) = updates.next().await {
            match update {
                WithdrawState::Succeeded(txid) => {
                    info!(target: LOG_GATEWAY, amount = %withdraw_amount, address = %address, "Sent funds");
                    return Ok(WithdrawResponse { txid, fees });
                }
                WithdrawState::Failed(e) => {
                    return Err(AdminGatewayError::WithdrawError { failure_reason: e });
                }
                WithdrawState::Created => {}
            }
        }

        Err(AdminGatewayError::WithdrawError {
            failure_reason: "Ran out of state updates while withdrawing".to_string(),
        })
    }

    /// Returns a preview of the withdrawal fees without executing the
    /// withdrawal. Used by the UI for two-step withdrawal confirmation.
    async fn handle_withdraw_preview_msg(
        &self,
        payload: WithdrawPreviewPayload,
    ) -> AdminResult<WithdrawPreviewResponse> {
        // Deserialized amounts can bypass BitcoinAmountOrAll::from_str.
        // Reject them before converting to millisatoshis or querying either wallet.
        if let BitcoinAmountOrAll::Amount(amount) = payload.amount
            && amount > bitcoin::Amount::MAX_MONEY
        {
            return Err(AdminGatewayError::WithdrawError {
                failure_reason: "Withdrawal amount exceeds maximum Bitcoin supply".to_string(),
            });
        }

        let gateway_network = self.network;
        let address_checked = payload
            .address
            .clone()
            .require_network(gateway_network)
            .map_err(|_| AdminGatewayError::WithdrawError {
                failure_reason: "Address network mismatch".to_string(),
            })?;

        let client = self.select_client(payload.federation_id).await?;

        let WithdrawDetails {
            amount,
            mint_fees,
            peg_out_fees,
            flat_peg_out_fee,
        } = match payload.amount {
            BitcoinAmountOrAll::All => {
                calculate_max_withdrawable(client.value(), &address_checked).await?
            }
            BitcoinAmountOrAll::Amount(btc_amount) => {
                if let Ok(wallet_module) = client.value().get_first_module::<WalletClientModule>() {
                    WithdrawDetails {
                        amount: btc_amount.into(),
                        mint_fees: None,
                        peg_out_fees: wallet_module
                            .get_withdraw_fees(&address_checked, btc_amount)
                            .await
                            .map_err(|e| AdminGatewayError::Unexpected(e.into()))?,
                        flat_peg_out_fee: false,
                    }
                } else if let Ok(wallet_module) = client
                    .value()
                    .get_first_module::<fedimint_walletv2_client::WalletClientModule>(
                ) {
                    let fee = wallet_module.send_fee().await.map_err(|e| {
                        AdminGatewayError::WithdrawError {
                            failure_reason: e.fmt_compact().to_string(),
                        }
                    })?;
                    // The wallet output carries the payment plus the on-chain
                    // fee; funding it costs the federation fees quoted here.
                    let output_amount = btc_amount.checked_add(fee).ok_or_else(|| {
                        AdminGatewayError::WithdrawError {
                            failure_reason: "Withdrawal amount overflow".to_string(),
                        }
                    })?;
                    let federation_fees = wallet_module
                        .send_fee_quote(output_amount)
                        .await
                        .map_err(|e| AdminGatewayError::WithdrawError {
                            failure_reason: e.fmt_compact().to_string(),
                        })?
                        .total()
                        .get_bitcoin();
                    WithdrawDetails {
                        amount: btc_amount.into(),
                        mint_fees: Some(federation_fees),
                        peg_out_fees: PegOutFees::from_amount(fee),
                        flat_peg_out_fee: true,
                    }
                } else {
                    return Err(AdminGatewayError::Unexpected(anyhow!(
                        "No wallet module found"
                    )));
                }
            }
        };

        let total_cost = amount
            .checked_add(peg_out_fees.amount().into())
            .and_then(|a| a.checked_add(mint_fees.unwrap_or(Amount::ZERO)))
            .ok_or_else(|| AdminGatewayError::Unexpected(anyhow!("Total cost overflow")))?;

        Ok(WithdrawPreviewResponse {
            withdraw_amount: amount,
            address: payload.address.assume_checked().to_string(),
            peg_out_fees,
            total_cost,
            mint_fees,
            flat_peg_out_fee,
        })
    }

    /// Queries the client log for payment events and returns to the user.
    /// Returns a paginated list of gateway payment-related events, ordered from
    /// newest to oldest.
    ///
    /// If `event_kinds` is empty, only events matching `ALL_GATEWAY_EVENTS`
    /// are returned — this is **not** equivalent to "all events". Other
    /// internal events (e.g. `tx-created`, `NoteCreated`) share the same event
    /// log and consume IDs, so returned event IDs may be non-contiguous.
    ///
    /// Pagination works backwards from `end_position` (or the log tip if
    /// `None`), returning at most `pagination_size` matching events.
    async fn handle_payment_log_msg(
        &self,
        PaymentLogPayload {
            end_position,
            pagination_size,
            federation_id,
            event_kinds,
        }: PaymentLogPayload,
    ) -> AdminResult<PaymentLogResponse> {
        const BATCH_SIZE: u64 = 10_000;
        let federation_manager = self.federation_manager.read().await;
        let client = federation_manager
            .client(&federation_id)
            .ok_or(FederationNotConnected {
                federation_id_prefix: federation_id.to_prefix(),
            })?
            .value();

        // An empty `event_kinds` defaults to gateway payment-related events, not
        // "all events". This means returned event IDs may be non-contiguous since
        // other internal events share the same ID space.
        let event_kinds = if event_kinds.is_empty() {
            ALL_GATEWAY_EVENTS.to_vec()
        } else {
            event_kinds
        };

        let end_position = if let Some(position) = end_position {
            position
        } else {
            let mut dbtx = client.db().begin_transaction_nc().await;
            dbtx.get_next_event_log_id().await
        };

        let mut start_position = end_position.saturating_sub(BATCH_SIZE);

        let mut payment_log = Vec::new();

        while payment_log.len() < pagination_size {
            let batch = client.get_event_log(Some(start_position), BATCH_SIZE).await;
            let mut filtered_batch = batch
                .into_iter()
                .filter(|e| e.id() <= end_position && event_kinds.contains(&e.as_raw().kind))
                .collect::<Vec<_>>();
            filtered_batch.reverse();
            payment_log.extend(filtered_batch);

            // Compute the start position for the next batch query
            start_position = start_position.saturating_sub(BATCH_SIZE);

            if start_position == EventLogId::LOG_START {
                break;
            }
        }

        // Truncate the payment log to the expected pagination size
        payment_log.truncate(pagination_size);

        Ok(PaymentLogResponse(payment_log))
    }

    /// Set the gateway's root mnemonic by generating a new one or using the
    /// words provided in `SetMnemonicPayload`.
    async fn handle_set_mnemonic_msg(&self, payload: SetMnemonicPayload) -> AdminResult<()> {
        // The state lock must be held from the `NotConfigured` check until the
        // transition to `Disconnected`, otherwise a concurrent call could pass
        // the check, lose the race for the seed write, and leave the gateway
        // stuck in `NotConfigured` with a seed already stored
        let mut state_guard = self.state.write().await;

        // Verify the state is NotConfigured
        let GatewayState::NotConfigured { mnemonic_sender } = state_guard.clone() else {
            return Err(AdminGatewayError::MnemonicError(anyhow!(
                "Gateway is not is NotConfigured state"
            )));
        };

        let (mnemonic, mnemonic_source) = if let Some(words) = payload.words {
            info!(target: LOG_GATEWAY, "Using user provided mnemonic");
            let mnemonic = Mnemonic::parse_in_normalized(Language::English, words.as_str())
                .map_err(|e| {
                    AdminGatewayError::MnemonicError(anyhow!(format!(
                        "Seed phrase provided in environment was invalid {e:?}"
                    )))
                })?;
            (mnemonic, MnemonicSource::Existing)
        } else {
            debug!(target: LOG_GATEWAY, "Generating mnemonic and writing entropy to client storage");
            (
                Bip39RootSecretStrategy::<12>::random(&mut OsRng),
                MnemonicSource::Generated,
            )
        };

        Client::store_encodable_client_secret(&self.gateway_db, mnemonic.to_entropy())
            .await
            .map_err(|err| AdminGatewayError::MnemonicError(err.into()))?;

        *state_guard = GatewayState::Disconnected;
        drop(state_guard);

        // Alert the gateway background threads that the mnemonic has been set
        let _ = mnemonic_sender.send(mnemonic_source);

        Ok(())
    }

    /// Creates a BOLT12 offer using the gateway's lightning node
    async fn handle_create_offer_for_operator_msg(
        &self,
        payload: CreateOfferPayload,
    ) -> AdminResult<CreateOfferResponse> {
        let lightning_context = self.get_lightning_context().await?;
        let offer = lightning_context.lnrpc.create_offer(
            payload.amount,
            payload.description,
            payload.expiry_secs,
            payload.quantity,
        )?;
        Ok(CreateOfferResponse { offer })
    }

    /// Pays a BOLT12 offer using the gateway's lightning node
    async fn handle_pay_offer_for_operator_msg(
        &self,
        payload: PayOfferPayload,
    ) -> AdminResult<PayOfferResponse> {
        let lightning_context = self.get_lightning_context().await?;
        let preimage = lightning_context
            .lnrpc
            .pay_offer(
                payload.offer,
                payload.quantity,
                payload.amount,
                payload.payer_note,
            )
            .await?;
        Ok(PayOfferResponse {
            preimage: preimage.to_string(),
        })
    }

    /// Returns a `BTreeMap` that is keyed by the `FederationId` and contains
    /// all the invite codes (with peer names) for the federation.
    async fn handle_export_invite_codes(
        &self,
    ) -> BTreeMap<FederationId, BTreeMap<PeerId, (String, InviteCode)>> {
        let fed_manager = self.federation_manager.read().await;
        fed_manager.all_invite_codes().await
    }

    /// Returns `TieredCounts` which describes the breakdown of notes in the
    /// gateway's wallet for the given `FederationId`
    async fn handle_get_note_summary_msg(
        &self,
        federation_id: &FederationId,
    ) -> AdminResult<TieredCounts> {
        let fed_manager = self.federation_manager.read().await;
        fed_manager.get_note_summary(federation_id).await
    }

    fn get_password_hash(&self) -> String {
        self.bcrypt_password_hash.clone()
    }

    fn gatewayd_version(&self) -> String {
        let gatewayd_version = env!("CARGO_PKG_VERSION");
        gatewayd_version.to_string()
    }

    async fn get_chain_source(&self) -> (ChainSource, Network) {
        (self.chain_source.clone(), self.network)
    }

    fn lightning_mode(&self) -> LightningMode {
        self.lightning_mode.clone()
    }

    async fn is_configured(&self) -> bool {
        !matches!(self.get_state().await, GatewayState::NotConfigured { .. })
    }
}

// LNv2 Gateway implementation
impl Gateway {
    /// Retrieves the `PublicKey` of the Gateway module for a given federation
    /// for LNv2. This is NOT the same as the `gateway_id`, it is different
    /// per-connected federation.
    async fn public_key_v2(&self, federation_id: &FederationId) -> Option<PublicKey> {
        self.federation_manager
            .read()
            .await
            .client(federation_id)
            .and_then(|client| {
                // A federation only has to offer one of the two lightning modules, so a
                // client we serve over LNv1 may well have no LNv2 module at all.
                client
                    .value()
                    .get_first_module::<GatewayClientModuleV2>()
                    .ok()
                    .map(|module| module.keypair.public_key())
            })
    }

    /// Whether the gateway currently accepts incoming payments on behalf of
    /// the clients of `federation_id`, as set with `set_payment_policy`. A
    /// federation without a stored config is treated as accepting them, so the
    /// check never turns a missing record into a refused payment.
    async fn receive_enabled(&self, federation_id: FederationId) -> bool {
        self.gateway_db
            .begin_transaction_nc()
            .await
            .load_federation_config(federation_id)
            .await
            .is_none_or(|config| config.receive_enabled())
    }

    /// Admits a fresh incoming payment when the receive policy allows it,
    /// returning a guard that holds off a policy change until the caller has
    /// funded the payment.
    async fn admit_fresh_receive(&self, federation_id: FederationId) -> Option<PaymentPolicyGuard> {
        // The policy is read under the guard, so it cannot change between this
        // read and the funding the caller performs while holding it.
        let guard = self.payment_policy_lock.clone().read_owned().await;

        self.receive_enabled(federation_id)
            .await
            .then(|| PaymentPolicyGuard::new(guard))
    }

    /// Returns payment information that LNv2 clients can use to instruct this
    /// Gateway to pay an invoice or receive a payment.
    pub async fn routing_info_v2(
        &self,
        federation_id: &FederationId,
    ) -> Result<Option<RoutingInfo>> {
        let context = self.get_lightning_context().await?;

        let mut dbtx = self.gateway_db.begin_transaction_nc().await;
        let fed_config = dbtx.load_federation_config(*federation_id).await.ok_or(
            PublicGatewayError::FederationNotConnected(FederationNotConnected {
                federation_id_prefix: federation_id.to_prefix(),
            }),
        )?;

        let lightning_fee = fed_config.lightning_fee;
        let transaction_fee = fed_config.transaction_fee;
        let receive_enabled = fed_config.receive_enabled();

        // This route is public and unauthenticated, so the sum of two fees stored
        // before the fee limits applied must not be able to panic here.
        let send_fee_default = lightning_fee.checked_add(transaction_fee).ok_or_else(|| {
            PublicGatewayError::Unexpected(anyhow!(
                "The configured fees of federation {federation_id} cannot be added"
            ))
        })?;

        Ok(self
            .public_key_v2(federation_id)
            .await
            .map(|module_public_key| RoutingInfo {
                lightning_public_key: context.lightning_public_key,
                lightning_alias: Some(context.lightning_alias.clone()),
                module_public_key,
                send_fee_default,
                // The base fee ensures that the gateway does not loose sats sending the payment due
                // to fees paid on the transaction claiming the outgoing contract or
                // subsequent transactions spending the newly issued ecash
                send_fee_minimum: transaction_fee,
                expiration_delta_default: 1440,
                expiration_delta_minimum: EXPIRATION_DELTA_MINIMUM_V2,
                // The base fee ensures that the gateway does not loose sats receiving the payment
                // due to fees paid on the transaction funding the incoming contract
                receive_fee: transaction_fee,
                receive_enabled,
            }))
    }

    /// Instructs this gateway to pay a Lightning network invoice via the LNv2
    /// protocol.
    pub async fn send_payment_v2(
        &self,
        payload: SendPaymentPayload,
    ) -> Result<std::result::Result<[u8; 32], Signature>> {
        let client = self.select_client(payload.federation_id).await?;
        // A federation only has to offer one of the two lightning modules, so a
        // client we serve over LNv1 may well have no LNv2 module at all.
        let module = client
            .value()
            .get_first_module::<GatewayClientModuleV2>()
            .map_err(|err| PublicGatewayError::LNv2(LNv2Error::OutgoingPayment(err.into())))?;

        module
            .send_payment(payload)
            .await
            .map_err(|err| LNv2Error::OutgoingPayment(err.into()))
            .map_err(PublicGatewayError::LNv2)
    }

    /// For the LNv2 protocol, this will create an invoice by fetching it from
    /// the connected Lightning node, then save the payment hash so that
    /// incoming lightning payments can be matched as a receive attempt to a
    /// specific federation.
    pub async fn create_bolt11_invoice_v2(
        &self,
        payload: CreateBolt11InvoicePayload,
    ) -> Result<Bolt11Invoice> {
        // Unauthenticated invoice creation consumes resources on the Lightning
        // node and burns CPU on contract verification, so the request rate is
        // limited before any other work is done.
        if !self.invoice_rate_limiter.try_acquire() {
            return Err(PublicGatewayError::RateLimited);
        }

        // Verifying the contract is two pairings; keep them off the async workers.
        let contract = payload.contract.clone();

        if !fedimint_core::runtime::spawn_blocking(move || contract.verify()).await {
            return Err(PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                "The contract is invalid".to_string(),
            )));
        }

        let payment_info = self.routing_info_v2(&payload.federation_id).await?.ok_or(
            LNv2Error::IncomingPayment(format!(
                "Federation {} does not exist",
                payload.federation_id
            )),
        )?;

        if !payment_info.receive_enabled {
            return Err(PublicGatewayError::ReceiveDisabled {
                federation_id_prefix: payload.federation_id.to_prefix(),
            });
        }

        if payload.contract.commitment.refund_pk != payment_info.module_public_key {
            return Err(PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                "The incoming contract is keyed to another gateway".to_string(),
            )));
        }

        let contract_amount = payment_info.receive_fee.subtract_from(payload.amount.msats);

        if contract_amount == Amount::ZERO {
            return Err(PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                "Zero amount incoming contracts are not supported".to_string(),
            )));
        }

        if contract_amount != payload.contract.commitment.amount {
            return Err(PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                "The contract amount does not pay the correct amount of fees".to_string(),
            )));
        }

        if payload.contract.commitment.expiration_or_fee <= duration_since_epoch().as_secs() {
            return Err(PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                "The contract has already expired".to_string(),
            )));
        }

        if payload.expiry_secs > MAX_INVOICE_EXPIRY_SECS {
            return Err(PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                "The invoice expiry exceeds the maximum of one day".to_string(),
            )));
        }

        let payment_hash = match payload.contract.commitment.payment_image {
            PaymentImage::Hash(payment_hash) => payment_hash,
            PaymentImage::Point(..) => {
                return Err(PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                    "PaymentImage is not a payment hash".to_string(),
                )));
            }
        };

        // Reserve the payment hash in the database before requesting the
        // invoice so a replayed payment hash is rejected without creating any
        // state on the Lightning node, and so the contract is guaranteed to be
        // registered by the time the invoice is payable.
        let mut dbtx = self.gateway_db.begin_transaction().await;

        let invoice_expires_at_secs = duration_since_epoch()
            .as_secs()
            .saturating_add(u64::from(payload.expiry_secs));

        if dbtx
            .save_registered_incoming_contract(
                payload.federation_id,
                payload.amount,
                invoice_expires_at_secs,
                payload.contract,
            )
            .await
            .is_some()
        {
            return Err(PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                "PaymentHash is already registered".to_string(),
            )));
        }

        dbtx.commit_tx_result().await.map_err(|_| {
            PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                "Payment hash is already registered".to_string(),
            ))
        })?;

        match self
            .create_invoice_via_lnrpc_v2(
                payment_hash,
                payload.amount,
                payload.description.clone(),
                payload.expiry_secs,
            )
            .await
        {
            Ok(invoice) => Ok(invoice),
            Err(err) => {
                // Release the reservation so the payment hash is not burned by
                // a transient Lightning node failure.
                let mut dbtx = self.gateway_db.begin_transaction().await;
                dbtx.delete_registered_incoming_contract(PaymentImage::Hash(payment_hash))
                    .await;
                if let Err(db_err) = dbtx.commit_tx_result().await {
                    warn!(
                        target: LOG_GATEWAY,
                        err = %db_err.fmt_compact(),
                        %payment_hash,
                        "Failed to release incoming contract reservation after Lightning error"
                    );
                }

                Err(err.into())
            }
        }
    }

    /// Retrieves a BOLT11 invoice from the connected Lightning node with a
    /// specific `payment_hash`.
    pub async fn create_invoice_via_lnrpc_v2(
        &self,
        payment_hash: sha256::Hash,
        amount: Amount,
        description: Bolt11InvoiceDescription,
        expiry_time: u32,
    ) -> std::result::Result<Bolt11Invoice, LightningRpcError> {
        let lnrpc = self.get_lightning_context().await?.lnrpc;

        let response = match description {
            Bolt11InvoiceDescription::Direct(description) => {
                lnrpc
                    .create_invoice(CreateInvoiceRequest {
                        payment_hash: Some(payment_hash),
                        amount_msat: amount.msats,
                        expiry_secs: expiry_time,
                        description: Some(InvoiceDescription::Direct(description)),
                    })
                    .await?
            }
            Bolt11InvoiceDescription::Hash(hash) => {
                lnrpc
                    .create_invoice(CreateInvoiceRequest {
                        payment_hash: Some(payment_hash),
                        amount_msat: amount.msats,
                        expiry_secs: expiry_time,
                        description: Some(InvoiceDescription::Hash(hash)),
                    })
                    .await?
            }
        };

        Bolt11Invoice::from_str(&response.invoice).map_err(|e| {
            LightningRpcError::FailedToGetInvoice {
                failure_reason: e.to_string(),
            }
        })
    }

    pub async fn verify_bolt11_preimage_v2(
        &self,
        payment_hash: sha256::Hash,
        wait: bool,
    ) -> std::result::Result<VerifyResponse, String> {
        let registered_contract = self
            .gateway_db
            .begin_transaction_nc()
            .await
            .load_registered_incoming_contract(PaymentImage::Hash(payment_hash))
            .await
            .ok_or("Unknown payment hash".to_string())?;

        let client = self
            .select_client(registered_contract.federation_id)
            .await
            .map_err(|_| "Not connected to federation".to_string())?
            .into_value();

        let operation_id = OperationId::from_encodable(&registered_contract.contract);

        if !(wait || client.operation_exists(operation_id).await) {
            return Ok(VerifyResponse::pending());
        }

        let module = client
            .get_first_module::<GatewayClientModuleV2>()
            .expect("Must have client module");

        let Ok(state) = timeout(VERIFY_WAIT_TIMEOUT, module.await_receive(operation_id)).await
        else {
            return Ok(VerifyResponse::pending());
        };

        let preimage = match state {
            FinalReceiveState::Success(preimage) => Ok(preimage),
            FinalReceiveState::Failure => Err("Payment has failed".to_string()),
            FinalReceiveState::Refunded => Err("Payment has been refunded".to_string()),
            FinalReceiveState::Rejected => Err("Payment has been rejected".to_string()),
        }?;

        Ok(VerifyResponse::settled(preimage))
    }

    /// Retrieves the persisted `CreateInvoicePayload` from the database
    /// specified by the `payment_hash` and the `ClientHandleArc` specified
    /// by the payload's `federation_id`.
    pub async fn get_registered_incoming_contract_and_client_v2(
        &self,
        payment_image: PaymentImage,
        amount_msats: u64,
    ) -> Result<(IncomingContract, ClientHandleArc)> {
        let registered_incoming_contract = self
            .gateway_db
            .begin_transaction_nc()
            .await
            .load_registered_incoming_contract(payment_image)
            .await
            .ok_or(PublicGatewayError::LNv2(LNv2Error::IncomingPayment(
                "No corresponding decryption contract available".to_string(),
            )))?;

        if registered_incoming_contract.incoming_amount_msats != amount_msats {
            return Err(PublicGatewayError::LNv2(
                LNv2Error::IncomingAmountMismatch {
                    registered_msats: registered_incoming_contract.incoming_amount_msats,
                    payment_msats: amount_msats,
                },
            ));
        }

        let client = self
            .select_client(registered_incoming_contract.federation_id)
            .await?
            .into_value();

        Ok((registered_incoming_contract.contract, client))
    }

    /// Completes (settles or cancels) an intercepted HTLC, retrying transient
    /// failures until Lightning reports a terminal outcome.
    ///
    /// Serves [`IGatewayClientV1::complete_htlc`] and
    /// [`IGatewayClientV2::complete_htlc`]: by now the incoming contract is
    /// funded, so giving up on a transient failure would strand it. Only
    /// [`LightningRpcError::HtlcCompletionRejected`], a permanent state, is
    /// returned to the caller.
    async fn await_complete_htlc(
        &self,
        htlc: InterceptPaymentResponse,
    ) -> std::result::Result<(), LightningRpcError> {
        loop {
            let lightning_context = self.await_lightning_context().await;

            match lightning_context.lnrpc.complete_htlc(htlc.clone()).await {
                Ok(()) => return Ok(()),
                Err(err @ LightningRpcError::HtlcCompletionRejected { .. }) => {
                    warn!(
                        target: LOG_GATEWAY,
                        err = %err.fmt_compact(),
                        "Lightning cannot reach the requested terminal HTLC outcome",
                    );
                    return Err(err);
                }
                Err(err) => {
                    warn!(
                        target: LOG_GATEWAY,
                        err = %err.fmt_compact(),
                        "Failure trying to complete HTLC, retrying",
                    );
                }
            }

            sleep(LIGHTNING_CONTEXT_RETRY_INTERVAL).await;
        }
    }

    /// Answers whether the connected Lightning node has any record of an
    /// outbound payment for `payment_hash`, retrying transient lookup
    /// failures until the node itself can answer.
    ///
    /// Serves [`IGatewayClientV1::outbound_payment_exists`] and
    /// [`IGatewayClientV2::outbound_payment_exists`]: a wrong `false` lets a
    /// resumed state machine cancel a contract whose payment is still in
    /// flight, so no answer is synthesised from a failure.
    async fn await_outbound_payment_exists(&self, payment_hash: sha256::Hash) -> bool {
        loop {
            let lightning_context = self.await_lightning_context().await;

            match lightning_context
                .lnrpc
                .outbound_payment_exists(payment_hash)
                .await
            {
                Ok(exists) => return exists,
                Err(err) => {
                    warn!(
                        target: LOG_GATEWAY,
                        err = %err.fmt_compact(),
                        %payment_hash,
                        "Failed to check for a dispatched payment, retrying",
                    );
                }
            }

            sleep(LIGHTNING_CONTEXT_RETRY_INTERVAL).await;
        }
    }
}

#[async_trait]
impl IGatewayClientV2 for Gateway {
    async fn complete_htlc(
        &self,
        htlc_response: InterceptPaymentResponse,
    ) -> std::result::Result<(), LightningRpcError> {
        self.await_complete_htlc(htlc_response).await
    }

    async fn is_direct_swap(
        &self,
        invoice: &Bolt11Invoice,
    ) -> std::result::Result<Option<(IncomingContract, ClientHandleArc)>, GatewayClientV2Error>
    {
        // Deciding this from a locally synthesised "not connected" would route a
        // direct swap onto the lightning network, or -- once the send state
        // machine turns the error into a cancellation -- forfeit a contract we
        // may already be committed to. Ask once we can actually answer.
        let lightning_context = self.await_lightning_context().await;
        if lightning_context.lightning_public_key == invoice.get_payee_pub_key() {
            let (contract, client) = self
                .get_registered_incoming_contract_and_client_v2(
                    PaymentImage::Hash(*invoice.payment_hash()),
                    invoice
                        .amount_milli_satoshis()
                        .expect("The amount invoice has been previously checked"),
                )
                .await
                .map_err(GatewayClientV2Error::new)?;
            Ok(Some((contract, client)))
        } else {
            Ok(None)
        }
    }

    async fn pay(
        &self,
        invoice: Bolt11Invoice,
        max_delay: u64,
        max_fee: Amount,
    ) -> std::result::Result<[u8; 32], LightningRpcError> {
        // The send state machine forfeits the outgoing contract on any error from
        // here, so only the lightning node gets to say this payment failed.
        let lightning_context = self.await_lightning_context().await;
        lightning_context
            .lnrpc
            .pay(invoice, max_delay, max_fee)
            .await
            .map(|response| response.preimage.0)
    }

    async fn outbound_payment_exists(&self, payment_hash: sha256::Hash) -> bool {
        self.await_outbound_payment_exists(payment_hash).await
    }

    async fn begin_fresh_receive(
        &self,
        federation_id: &FederationId,
    ) -> Option<PaymentPolicyGuard> {
        self.admit_fresh_receive(*federation_id).await
    }

    async fn min_contract_amount(
        &self,
        federation_id: &FederationId,
        amount: u64,
    ) -> std::result::Result<Amount, GatewayClientV2Error> {
        Ok(self
            .routing_info_v2(federation_id)
            .await
            .map_err(GatewayClientV2Error::new)?
            .ok_or_else(|| GatewayClientV2Error::new("Routing Info not available"))?
            .send_fee_minimum
            .add_to(amount))
    }

    async fn is_lnv1_invoice(&self, invoice: &Bolt11Invoice) -> Option<Spanned<ClientHandleArc>> {
        let rhints = invoice.route_hints();
        let hop = rhints.first().and_then(|rh| rh.0.last())?;

        // Answering `None` because we happen to be between lightning connections
        // sends a swap that never needed the lightning network out over it.
        let lightning_context = self.await_lightning_context().await;
        if hop.src_node_id != lightning_context.lightning_public_key {
            return None;
        }

        self.federation_manager
            .read()
            .await
            .get_client_for_index(hop.short_channel_id)
    }

    async fn relay_lnv1_swap(
        &self,
        client: &ClientHandleArc,
        invoice: &Bolt11Invoice,
        allow_fresh_dispatch: bool,
    ) -> std::result::Result<Option<FinalReceiveState>, GatewayClientV2Error> {
        let swap_params =
            SwapParameters {
                payment_hash: *invoice.payment_hash(),
                amount_msat: Amount::from_msats(invoice.amount_milli_satoshis().ok_or_else(
                    || GatewayClientV2Error::new("Amountless invoice not supported"),
                )?),
            };
        // Nothing has been paid out yet, so failing here only cancels the send.
        let lnv1 = client
            .get_first_module::<GatewayClientModule>()
            .map_err(|_| GatewayClientV2Error::new("Federation does not have an LNv1 module"))?;
        let Some(operation_id) = lnv1
            .gateway_handle_direct_swap(swap_params, allow_fresh_dispatch)
            .await
            .map_err(GatewayClientV2Error::new)?
        else {
            return Ok(None);
        };
        let mut stream = lnv1
            .gateway_subscribe_ln_receive(operation_id)
            .await
            .map_err(GatewayClientV2Error::new)?
            .into_stream();
        let mut final_state = FinalReceiveState::Failure;
        while let Some(update) = stream.next().await {
            match update {
                GatewayExtReceiveStates::Funding => {}
                GatewayExtReceiveStates::FundingFailed { error: _ } => {
                    final_state = FinalReceiveState::Rejected;
                }
                GatewayExtReceiveStates::Preimage(preimage) => {
                    final_state = FinalReceiveState::Success(preimage.0);
                }
                GatewayExtReceiveStates::RefundError {
                    error_message: _,
                    error: _,
                } => {
                    final_state = FinalReceiveState::Failure;
                }
                GatewayExtReceiveStates::RefundSuccess {
                    out_points: _,
                    error: _,
                } => {
                    final_state = FinalReceiveState::Refunded;
                }
            }
        }

        Ok(Some(final_state))
    }

    async fn claim_payment_image(
        &self,
        payment_image: &PaymentImage,
        operation_id: OperationId,
    ) -> bool {
        // `autocommit` retries on write-write conflicts, so concurrent claims for
        // the same payment image are serialized: exactly one records itself as the
        // claimer, the rest observe it and forfeit.
        self.gateway_db
            .autocommit(
                |dbtx, _| {
                    let payment_image = payment_image.clone();
                    Box::pin(async move {
                        let claimer = dbtx
                            .claim_outgoing_payment_image(payment_image, operation_id)
                            .await;
                        Ok::<_, std::convert::Infallible>(claimer == operation_id)
                    })
                },
                None,
            )
            .await
            .expect("Retries until the transaction commits")
    }

    async fn is_lightning_connected(&self) -> bool {
        self.get_lightning_context().await.is_ok()
    }

    async fn await_lightning_connected(&self) {
        self.await_lightning_context().await;
    }
}

#[async_trait]
impl IGatewayClientV1 for Gateway {
    async fn claim_payment_image(
        &self,
        payment_image: &PaymentImage,
        operation_id: OperationId,
    ) -> bool {
        IGatewayClientV2::claim_payment_image(self, payment_image, operation_id).await
    }

    async fn verify_preimage_authentication(
        &self,
        payment_hash: sha256::Hash,
        preimage_auth: sha256::Hash,
        contract: OutgoingContractAccount,
    ) -> std::result::Result<(), OutgoingPaymentError> {
        let mut dbtx = self.gateway_db.begin_transaction().await;
        if let Some(secret_hash) = dbtx.load_preimage_authentication(payment_hash).await {
            if !PreimageAuth::new(secret_hash).verifies(preimage_auth) {
                return Err(OutgoingPaymentError {
                    error_type: OutgoingPaymentErrorType::InvalidInvoicePreimage,
                    contract_id: contract.contract.contract_id(),
                    contract: Some(contract),
                });
            }
        } else {
            // Committing the `preimage_auth` to the database can fail if two users try to
            // pay the same invoice at the same time.
            dbtx.save_new_preimage_authentication(payment_hash, preimage_auth)
                .await;
            return dbtx
                .commit_tx_result()
                .await
                .map_err(|_| OutgoingPaymentError {
                    error_type: OutgoingPaymentErrorType::InvoiceAlreadyPaid,
                    contract_id: contract.contract.contract_id(),
                    contract: Some(contract),
                });
        }

        Ok(())
    }

    async fn verify_pruned_invoice(
        &self,
        payment_data: PaymentData,
    ) -> std::result::Result<(), GatewayClientV1Error> {
        if matches!(payment_data, PaymentData::PrunedInvoice { .. }) {
            let lightning_context = self
                .get_lightning_context()
                .await
                .map_err(GatewayClientV1Error::new)?;

            if !lightning_context.lnrpc.supports_private_payments() {
                return Err(GatewayClientV1Error::new(
                    "Private payments are not supported by the lightning node",
                ));
            }
        }

        Ok(())
    }

    async fn begin_fresh_receive(&self, federation_id: FederationId) -> Option<PaymentPolicyGuard> {
        self.admit_fresh_receive(federation_id).await
    }

    async fn get_routing_fees(&self, federation_id: FederationId) -> Option<RoutingFees> {
        let mut gateway_dbtx = self.gateway_db.begin_transaction_nc().await;
        let lightning_fee = gateway_dbtx
            .load_federation_config(federation_id)
            .await?
            .lightning_fee;

        // A fee stored before the fee limits applied may not be announceable. The
        // caller treats `None` as a federation configuration error, which is
        // preferable to panicking in the middle of a payment.
        RoutingFees::try_from(lightning_fee)
            .inspect_err(|err| {
                warn!(
                    target: LOG_GATEWAY,
                    %federation_id,
                    err = %err.fmt_compact(),
                    "Configured lightning fee cannot be used. Set a smaller fee with `set_fees`."
                );
            })
            .ok()
    }

    async fn get_client(&self, federation_id: &FederationId) -> Option<Spanned<ClientHandleArc>> {
        self.federation_manager
            .read()
            .await
            .client(federation_id)
            .cloned()
    }

    async fn get_client_for_invoice(
        &self,
        payment_data: PaymentData,
    ) -> Option<Spanned<ClientHandleArc>> {
        let rhints = payment_data.route_hints();
        let hop = rhints.first().and_then(|rh| rh.0.last())?;

        // Answering `None` because we happen to be between lightning connections
        // sends a swap that never needed the lightning network out over it.
        let lightning_context = self.await_lightning_context().await;
        if hop.src_node_id != lightning_context.lightning_public_key {
            return None;
        }

        self.federation_manager
            .read()
            .await
            .get_client_for_index(hop.short_channel_id)
    }

    async fn pay(
        &self,
        payment_data: PaymentData,
        max_delay: u64,
        max_fee: Amount,
    ) -> std::result::Result<PayInvoiceResponse, LightningRpcError> {
        // `GatewayPayInvoice` cancels the outgoing contract on any error from
        // here, so only the lightning node gets to say this payment failed.
        let lightning_context = self.await_lightning_context().await;

        match payment_data {
            PaymentData::Invoice(invoice) => {
                lightning_context
                    .lnrpc
                    .pay(invoice, max_delay, max_fee)
                    .await
            }
            PaymentData::PrunedInvoice(invoice) => {
                lightning_context
                    .lnrpc
                    .pay_private(invoice, max_delay, max_fee)
                    .await
            }
        }
    }

    async fn outbound_payment_exists(&self, payment_hash: sha256::Hash) -> bool {
        self.await_outbound_payment_exists(payment_hash).await
    }

    async fn complete_htlc(
        &self,
        htlc: InterceptPaymentResponse,
    ) -> std::result::Result<(), LightningRpcError> {
        self.await_complete_htlc(htlc).await
    }

    async fn is_lnv2_direct_swap(
        &self,
        payment_hash: sha256::Hash,
        amount: Amount,
    ) -> std::result::Result<
        Option<(
            fedimint_lnv2_common::contracts::IncomingContract,
            ClientHandleArc,
        )>,
        GatewayClientV1Error,
    > {
        let (contract, client) = self
            .get_registered_incoming_contract_and_client_v2(
                PaymentImage::Hash(payment_hash),
                amount.msats,
            )
            .await
            .map_err(GatewayClientV1Error::new)?;
        Ok(Some((contract, client)))
    }
}