fedimint_gw_client/

lib.rs

mod complete;
pub mod events;
pub mod pay;
#[cfg(test)]
mod tests;

use std::collections::BTreeMap;
use std::fmt;
use std::fmt::Debug;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

use async_stream::stream;
use async_trait::async_trait;
use bitcoin::hashes::{Hash, sha256};
use bitcoin::key::Secp256k1;
use bitcoin::secp256k1::All;
use complete::{GatewayCompleteCommon, GatewayCompleteStates, WaitForPreimageState};
use events::{IncomingPaymentStarted, OutgoingPaymentStarted};
use fedimint_api_client::api::DynModuleApi;
use fedimint_client::ClientHandleArc;
use fedimint_client_module::module::init::{ClientModuleInit, ClientModuleInitArgs};
use fedimint_client_module::module::recovery::NoModuleBackup;
use fedimint_client_module::module::{ClientContext, ClientModule, IClientModule, OutPointRange};
use fedimint_client_module::oplog::UpdateStreamOrOutcome;
use fedimint_client_module::sm::{Context, DynState, ModuleNotifier, State, StateTransition};
use fedimint_client_module::transaction::{
    ClientOutput, ClientOutputBundle, ClientOutputSM, TransactionBuilder,
};
use fedimint_client_module::{
    AddStateMachinesError, DynGlobalClientContext, sm_enum_variant_translation,
};
use fedimint_connectors::ConnectorRegistry;
use fedimint_core::config::FederationId;
use fedimint_core::core::{Decoder, IntoDynInstance, ModuleInstanceId, ModuleKind, OperationId};
use fedimint_core::db::{AutocommitError, DatabaseTransaction};
use fedimint_core::encoding::{Decodable, Encodable};
use fedimint_core::module::{Amounts, ApiVersion, ModuleInit, MultiApiVersion};
use fedimint_core::time::duration_since_epoch;
use fedimint_core::util::{FmtCompact, SafeUrl, Spanned};
use fedimint_core::{Amount, OutPoint, apply, async_trait_maybe_send, secp256k1};
use fedimint_derive_secret::ChildId;
use fedimint_lightning::{
    InterceptPaymentRequest, InterceptPaymentResponse, LightningContext, LightningRpcError,
    PayInvoiceResponse,
};
use fedimint_ln_client::api::LnFederationApi;
use fedimint_ln_client::incoming::{
    FundingOfferState, IncomingSmCommon, IncomingSmError, IncomingSmStates, IncomingStateMachine,
};
use fedimint_ln_client::pay::{PayInvoicePayload, PaymentData};
use fedimint_ln_client::{
    LightningClientContext, LightningClientInit, RealGatewayConnection,
    create_incoming_contract_output,
};
use fedimint_ln_common::config::LightningClientConfig;
use fedimint_ln_common::contracts::outgoing::OutgoingContractAccount;
use fedimint_ln_common::contracts::{ContractId, Preimage};
use fedimint_ln_common::route_hints::RouteHint;
use fedimint_ln_common::{
    GatewayRegistrationAuth, KIND, LNV1_INCOMING_HTLC_ADVERTISED_EXPIRY_DELTA,
    LNV1_INCOMING_HTLC_EXPIRY_SAFETY_MARGIN, LightningCommonInit, LightningGateway,
    LightningGatewayAnnouncement, LightningModuleTypes, LightningOutput, LightningOutputV0,
    RemoveGatewayRequest, create_gateway_registration_message, create_gateway_remove_message,
};
use fedimint_lnv2_common::GatewayApi;
use futures::StreamExt;
use lightning_invoice::RoutingFees;
use secp256k1::Keypair;
use serde::{Deserialize, Serialize};
use thiserror::Error;
use tracing::{debug, error, info, warn};

use self::complete::GatewayCompleteStateMachine;
use self::pay::{
    GatewayPayCommon, GatewayPayInvoice, GatewayPayStateMachine, GatewayPayStates,
    OutgoingContractError, OutgoingPaymentError,
};

/// Exclusive remaining-CLTV safety margin for an intercepted LNv1 HTLC.
///
/// This reserves the worst-case time to claim the HTLC on-chain after
/// federation funding and threshold decryption. Exactly the margin is
/// rejected; fresh funding requires at least one additional block. The value
/// deliberately matches the route-hint delta advertised by pre-upgrade
/// clients so their invoices remain payable; see
/// [`LNV1_INCOMING_HTLC_EXPIRY_SAFETY_MARGIN`] for the plan to raise it.
pub const LNV1_HTLC_EXPIRY_SAFETY_MARGIN: u32 = LNV1_INCOMING_HTLC_EXPIRY_SAFETY_MARGIN as u32;

/// The high-level state of a reissue operation started with
/// [`GatewayClientModule::gateway_pay_bolt11_invoice`].
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
pub enum GatewayExtPayStates {
    Created,
    Preimage {
        preimage: Preimage,
    },
    Success {
        preimage: Preimage,
        out_points: Vec<OutPoint>,
    },
    Canceled {
        error: OutgoingPaymentError,
    },
    Fail {
        error: OutgoingPaymentError,
        error_message: String,
    },
    OfferDoesNotExist {
        contract_id: ContractId,
    },
}

/// The high-level state of an intercepted HTLC operation started with
/// [`GatewayClientModule::gateway_handle_intercepted_htlc`].
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
pub enum GatewayExtReceiveStates {
    Funding,
    Preimage(Preimage),
    RefundSuccess {
        out_points: Vec<OutPoint>,
        error: IncomingSmError,
    },
    RefundError {
        error_message: String,
        error: IncomingSmError,
    },
    FundingFailed {
        error: IncomingSmError,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum GatewayMeta {
    /// Carries the `preimage_auth` of the request that started the payment.
    ///
    /// The operation id of a payment is its contract id, which is public in the
    /// funding transaction, and joining the operation yields the preimage. So
    /// the operation has to record who is allowed to join it. Entries written
    /// before this field existed decode as an error and are treated as
    /// unauthorized, which only affects payments in flight across an upgrade.
    Pay {
        preimage_auth: sha256::Hash,
    },
    Receive,
}

#[derive(Debug, Clone)]
pub struct GatewayClientInit {
    pub federation_index: u64,
    pub lightning_manager: Arc<dyn IGatewayClientV1>,
}

impl ModuleInit for GatewayClientInit {
    type Common = LightningCommonInit;

    async fn dump_database(
        &self,
        _dbtx: &mut DatabaseTransaction<'_>,
        _prefix_names: Vec<String>,
    ) -> Box<dyn Iterator<Item = (String, Box<dyn erased_serde::Serialize + Send>)> + '_> {
        Box::new(vec![].into_iter())
    }
}

#[apply(async_trait_maybe_send!)]
impl ClientModuleInit for GatewayClientInit {
    type Module = GatewayClientModule;

    fn supported_api_versions(&self) -> MultiApiVersion {
        MultiApiVersion::try_from_iter([ApiVersion { major: 0, minor: 0 }])
            .expect("no version conflicts")
    }

    async fn init(&self, args: &ClientModuleInitArgs<Self>) -> anyhow::Result<Self::Module> {
        Ok(GatewayClientModule {
            cfg: args.cfg().clone(),
            notifier: args.notifier().clone(),
            redeem_key: args
                .module_root_secret()
                .child_key(ChildId(0))
                .to_secp_key(&fedimint_core::secp256k1::Secp256k1::new()),
            module_api: args.module_api().clone(),
            federation_index: self.federation_index,
            client_ctx: args.context(),
            lightning_manager: self.lightning_manager.clone(),
            connector_registry: args.connector_registry.clone(),
            intercepted_htlc_lock_pool: lockable::LockPool::new(),
        })
    }
}

#[derive(Debug, Clone)]
pub struct GatewayClientContext {
    redeem_key: Keypair,
    secp: Secp256k1<All>,
    pub ln_decoder: Decoder,
    notifier: ModuleNotifier<GatewayClientStateMachines>,
    pub client_ctx: ClientContext<GatewayClientModule>,
    pub lightning_manager: Arc<dyn IGatewayClientV1>,
    pub connector_registry: ConnectorRegistry,
}

impl Context for GatewayClientContext {
    const KIND: Option<ModuleKind> = Some(fedimint_ln_common::KIND);
}

impl From<&GatewayClientContext> for LightningClientContext {
    fn from(ctx: &GatewayClientContext) -> Self {
        let gateway_conn = RealGatewayConnection {
            api: GatewayApi::new(None, ctx.connector_registry.clone()),
        };
        LightningClientContext {
            ln_decoder: ctx.ln_decoder.clone(),
            redeem_key: ctx.redeem_key,
            gateway_conn: Arc::new(gateway_conn),
            client_ctx: None,
        }
    }
}

/// Client side Lightning module **for the gateway**.
///
/// For the client side Lightning module for normal clients,
/// see [`fedimint_ln_client::LightningClientModule`]
pub struct GatewayClientModule {
    cfg: LightningClientConfig,
    pub notifier: ModuleNotifier<GatewayClientStateMachines>,
    pub redeem_key: Keypair,
    federation_index: u64,
    module_api: DynModuleApi,
    client_ctx: ClientContext<Self>,
    pub lightning_manager: Arc<dyn IGatewayClientV1>,
    connector_registry: ConnectorRegistry,
    intercepted_htlc_lock_pool: lockable::LockPool<HtlcCircuitKey>,
}

impl fmt::Debug for GatewayClientModule {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("GatewayClientModule")
            .finish_non_exhaustive()
    }
}

impl ClientModule for GatewayClientModule {
    type Init = LightningClientInit;
    type Common = LightningModuleTypes;
    type Backup = NoModuleBackup;
    type ModuleStateMachineContext = GatewayClientContext;
    type States = GatewayClientStateMachines;

    fn context(&self) -> Self::ModuleStateMachineContext {
        Self::ModuleStateMachineContext {
            redeem_key: self.redeem_key,
            secp: Secp256k1::new(),
            ln_decoder: self.decoder(),
            notifier: self.notifier.clone(),
            client_ctx: self.client_ctx.clone(),
            lightning_manager: self.lightning_manager.clone(),
            connector_registry: self.connector_registry.clone(),
        }
    }

    fn input_fee(
        &self,
        _amount: &Amounts,
        _input: &<Self::Common as fedimint_core::module::ModuleCommon>::Input,
    ) -> Option<Amounts> {
        Some(Amounts::new_bitcoin(self.cfg.fee_consensus.contract_input))
    }

    fn output_fee(
        &self,
        _amount: &Amounts,
        output: &<Self::Common as fedimint_core::module::ModuleCommon>::Output,
    ) -> Option<Amounts> {
        match output.maybe_v0_ref()? {
            LightningOutputV0::Contract(_) => {
                Some(Amounts::new_bitcoin(self.cfg.fee_consensus.contract_output))
            }
            LightningOutputV0::Offer(_) | LightningOutputV0::CancelOutgoing { .. } => {
                Some(Amounts::ZERO)
            }
        }
    }
}

impl GatewayClientModule {
    fn to_gateway_registration_info(
        &self,
        route_hints: Vec<RouteHint>,
        ttl: Duration,
        fees: RoutingFees,
        lightning_context: LightningContext,
        api: SafeUrl,
        gateway_keypair: Keypair,
    ) -> LightningGatewayAnnouncement {
        let info = LightningGateway {
            federation_index: self.federation_index,
            gateway_redeem_key: self.redeem_key.public_key(),
            node_pub_key: lightning_context.lightning_public_key,
            lightning_alias: lightning_context.lightning_alias,
            api,
            route_hints,
            fees,
            gateway_id: gateway_keypair.public_key(),
            supports_private_payments: lightning_context.lnrpc.supports_private_payments(),
        };

        // Proves to the guardians that we hold the key behind `gateway_id`, so
        // nobody else can replace our registration. Wall-clock milliseconds give
        // a nonce that keeps increasing across restarts without persisting
        // state; guardians only compare it to the nonce they already hold for
        // us, so our clock need not agree with theirs.
        let nonce = u64::try_from(duration_since_epoch().as_millis())
            .expect("milliseconds since the epoch do not exceed u64 for another 500m years");
        let signature = gateway_keypair.sign_schnorr(create_gateway_registration_message(
            self.cfg.threshold_pub_key,
            nonce,
            &info,
        ));

        LightningGatewayAnnouncement {
            info,
            ttl,
            vetted: false,
            auth: Some(GatewayRegistrationAuth { nonce, signature }),
        }
    }

    async fn create_funding_incoming_contract_output_from_htlc(
        &self,
        htlc: Htlc,
    ) -> Result<
        (
            OperationId,
            Amount,
            ClientOutput<LightningOutputV0>,
            ClientOutputSM<GatewayClientStateMachines>,
            ContractId,
        ),
        IncomingSmError,
    > {
        let operation_id = OperationId(htlc.payment_hash.to_byte_array());
        let (incoming_output, amount, contract_id) = create_incoming_contract_output(
            &self.module_api,
            htlc.payment_hash,
            htlc.outgoing_amount_msat,
            &self.redeem_key,
        )
        .await?;

        let client_output = ClientOutput::<LightningOutputV0> {
            output: incoming_output,
            amounts: Amounts::new_bitcoin(amount),
        };
        let client_output_sm = ClientOutputSM::<GatewayClientStateMachines> {
            state_machines: Arc::new(move |out_point_range: OutPointRange| {
                assert_eq!(out_point_range.count(), 1);
                vec![
                    GatewayClientStateMachines::Receive(IncomingStateMachine {
                        common: IncomingSmCommon {
                            operation_id,
                            contract_id,
                            payment_hash: htlc.payment_hash,
                        },
                        state: IncomingSmStates::FundingOffer(FundingOfferState {
                            txid: out_point_range.txid(),
                        }),
                    }),
                    GatewayClientStateMachines::Complete(GatewayCompleteStateMachine {
                        common: GatewayCompleteCommon {
                            operation_id,
                            payment_hash: htlc.payment_hash,
                            incoming_chan_id: htlc.incoming_chan_id,
                            htlc_id: htlc.htlc_id,
                        },
                        state: GatewayCompleteStates::WaitForPreimage(WaitForPreimageState),
                    }),
                ]
            }),
        };
        Ok((
            operation_id,
            amount,
            client_output,
            client_output_sm,
            contract_id,
        ))
    }

    async fn create_funding_incoming_contract_output_from_swap(
        &self,
        swap: SwapParameters,
    ) -> Result<
        (
            OperationId,
            ClientOutput<LightningOutputV0>,
            ClientOutputSM<GatewayClientStateMachines>,
        ),
        IncomingSmError,
    > {
        let payment_hash = swap.payment_hash;
        let operation_id = OperationId(payment_hash.to_byte_array());
        let (incoming_output, amount, contract_id) = create_incoming_contract_output(
            &self.module_api,
            payment_hash,
            swap.amount_msat,
            &self.redeem_key,
        )
        .await?;

        let client_output = ClientOutput::<LightningOutputV0> {
            output: incoming_output,
            amounts: Amounts::new_bitcoin(amount),
        };
        let client_output_sm = ClientOutputSM::<GatewayClientStateMachines> {
            state_machines: Arc::new(move |out_point_range| {
                assert_eq!(out_point_range.count(), 1);
                vec![GatewayClientStateMachines::Receive(IncomingStateMachine {
                    common: IncomingSmCommon {
                        operation_id,
                        contract_id,
                        payment_hash,
                    },
                    state: IncomingSmStates::FundingOffer(FundingOfferState {
                        txid: out_point_range.txid(),
                    }),
                })]
            }),
        };
        Ok((operation_id, client_output, client_output_sm))
    }

    /// Registers the gateway with a federation and reports whether it
    /// succeeded.
    ///
    /// Detailed errors remain in gateway logs; callers can retain the boolean
    /// result without exposing federation internals.
    pub async fn try_register_with_federation(
        &self,
        route_hints: Vec<RouteHint>,
        time_to_live: Duration,
        fees: RoutingFees,
        lightning_context: LightningContext,
        api: SafeUrl,
        gateway_keypair: Keypair,
    ) -> bool {
        let registration_info = self.to_gateway_registration_info(
            route_hints,
            time_to_live,
            fees,
            lightning_context,
            api,
            gateway_keypair,
        );
        let gateway_id = registration_info.info.gateway_id;

        let federation_id = self
            .client_ctx
            .get_config()
            .await
            .global
            .calculate_federation_id();
        match self.module_api.register_gateway(&registration_info).await {
            Err(e) => {
                warn!(
                    e = %e.fmt_compact(),
                    "Failed to register gateway {gateway_id} with federation {federation_id}"
                );
                false
            }
            _ => {
                info!(
                    "Successfully registered gateway {gateway_id} with federation {federation_id}"
                );
                true
            }
        }
    }

    /// Attempts to remove a gateway's registration from the federation. Since
    /// removing gateway registrations is best effort, this does not return
    /// an error and simply emits a warning when the registration cannot be
    /// removed.
    pub async fn remove_from_federation(&self, gateway_keypair: Keypair) {
        // Removing gateway registrations is best effort, so just emit a warning if it
        // fails
        if let Err(e) = self.remove_from_federation_inner(gateway_keypair).await {
            let gateway_id = gateway_keypair.public_key();
            let federation_id = self
                .client_ctx
                .get_config()
                .await
                .global
                .calculate_federation_id();
            warn!("Failed to remove gateway {gateway_id} from federation {federation_id}: {e:?}");
        }
    }

    /// Retrieves the signing challenge from each federation peer. Since each
    /// peer maintains their own list of registered gateways, the gateway
    /// needs to provide a signature that is signed by the private key of the
    /// gateway id to remove the registration.
    async fn remove_from_federation_inner(&self, gateway_keypair: Keypair) -> anyhow::Result<()> {
        let gateway_id = gateway_keypair.public_key();
        let challenges = self
            .module_api
            .get_remove_gateway_challenge(gateway_id)
            .await;

        let fed_public_key = self.cfg.threshold_pub_key;
        let signatures = challenges
            .into_iter()
            .filter_map(|(peer_id, challenge)| {
                let msg = create_gateway_remove_message(fed_public_key, peer_id, challenge?);
                let signature = gateway_keypair.sign_schnorr(msg);
                Some((peer_id, signature))
            })
            .collect::<BTreeMap<_, _>>();

        let remove_gateway_request = RemoveGatewayRequest {
            gateway_id,
            signatures,
        };

        self.module_api.remove_gateway(remove_gateway_request).await;

        Ok(())
    }

    /// Attempt to fulfill an HTLC by buying its preimage from the federation.
    ///
    /// LND can replay a still-pending HTLC after interceptor reconnect or
    /// gatewayd restart. Since the operation id is deterministic from the
    /// payment hash, the replay must not re-fetch the consumed federation offer
    /// or submit the same funding transaction again.
    ///
    /// We only short-circuit if a `GatewayCompleteStateMachine` is or was
    /// handling the exact same LND circuit. Direct swaps with the same payment
    /// hash and different circuits fall through to the normal failure/cancel
    /// path. Exact-circuit replays bypass fresh-funding expiry validation so an
    /// already-funded operation can still settle after reconnect or restart.
    ///
    /// `current_block_height` resolves to the Lightning backend's absolute best
    /// Bitcoin block height. It is awaited only after replay detection, before
    /// any fresh funding.
    pub async fn gateway_handle_intercepted_htlc(
        &self,
        htlc: Htlc,
        current_block_height: impl Future<Output = anyhow::Result<u32>>,
    ) -> anyhow::Result<OperationId> {
        debug!("Handling intercepted HTLC {htlc:?}");

        let operation_id = OperationId(htlc.payment_hash.to_byte_array());
        let circuit_key = HtlcCircuitKey {
            operation_id,
            incoming_chan_id: htlc.incoming_chan_id,
            htlc_id: htlc.htlc_id,
        };

        // Serialize same-circuit handling so the active-state replay scan and
        // operation creation are atomic for LND stream-reconnect replays.
        let _circuit_lock_guard = self
            .intercepted_htlc_lock_pool
            .async_lock(circuit_key)
            .await;

        // Check before the funding helper: the first handling consumes the
        // federation offer. Match the full circuit key, not just `operation_id`.
        let replay_of_active_circuit = self
            .client_ctx
            .get_own_operation_active_states(operation_id)
            .await
            .into_iter()
            .any(|(state, _)| circuit_key.matches_state(&state));
        if replay_of_active_circuit {
            debug!(
                ?operation_id,
                incoming_chan_id = htlc.incoming_chan_id,
                htlc_id = htlc.htlc_id,
                "HTLC circuit already being handled by an active completion state machine, treating as in-flight (likely an LND stream-reconnect replay)"
            );
            return Ok(operation_id);
        }
        let replay_of_inactive_circuit = self
            .client_ctx
            .get_own_operation_inactive_states(operation_id)
            .await
            .into_iter()
            .any(|(state, _)| circuit_key.matches_state(&state));
        if replay_of_inactive_circuit {
            debug!(
                ?operation_id,
                incoming_chan_id = htlc.incoming_chan_id,
                htlc_id = htlc.htlc_id,
                "HTLC circuit was already handled by a completion state machine, treating as idempotent replay"
            );
            return Ok(operation_id);
        }

        let current_block_height = current_block_height.await?;
        htlc.ensure_safe_expiry(current_block_height)?;
        let remaining_blocks = htlc.incoming_expiry.saturating_sub(current_block_height);
        if remaining_blocks <= u32::from(LNV1_INCOMING_HTLC_ADVERTISED_EXPIRY_DELTA) {
            // Tracks how much traffic still arrives via pre-upgrade invoices;
            // enforcement can be raised to the advertised delta once this stops
            // firing in the wild.
            warn!(
                payment_hash = %htlc.payment_hash,
                remaining_blocks,
                advertised_delta = LNV1_INCOMING_HTLC_ADVERTISED_EXPIRY_DELTA,
                "Accepting LNv1 HTLC with less remaining expiry than newly created invoices advertise, likely paid to a pre-upgrade invoice"
            );
        }

        let (op_id_from_funding, amount, client_output, client_output_sm, contract_id) = self
            .create_funding_incoming_contract_output_from_htlc(htlc.clone())
            .await?;
        // Keep the direct derivation above in sync with the funding helper. Return
        // an error instead of panicking so the caller can fail back the HTLC cleanly.
        anyhow::ensure!(
            op_id_from_funding == operation_id,
            "operation id derivation must match: {op_id_from_funding:?} != {operation_id:?}"
        );

        let output = ClientOutput {
            output: LightningOutput::V0(client_output.output),
            amounts: Amounts::new_bitcoin(amount),
        };

        let tx = TransactionBuilder::new().with_outputs(self.client_ctx.make_client_outputs(
            ClientOutputBundle::new(vec![output], vec![client_output_sm]),
        ));
        let operation_meta_gen = |_: OutPointRange| GatewayMeta::Receive;
        self.client_ctx
            .finalize_and_submit_transaction(operation_id, KIND.as_str(), operation_meta_gen, tx)
            .await?;
        debug!(?operation_id, "Submitted transaction for HTLC {htlc:?}");
        let mut dbtx = self.client_ctx.module_db().begin_transaction().await;
        self.client_ctx
            .log_event(
                &mut dbtx,
                IncomingPaymentStarted {
                    contract_id,
                    payment_hash: htlc.payment_hash,
                    invoice_amount: htlc.outgoing_amount_msat,
                    contract_amount: amount,
                    operation_id,
                },
            )
            .await;
        dbtx.commit_tx().await;
        Ok(operation_id)
    }

    /// Attempt buying preimage from this federation in order to fulfill a pay
    /// request in another federation served by this gateway. In direct swap
    /// scenario, the gateway DOES NOT send payment over the lightning network
    ///
    /// The operation is keyed on the payment hash, so an existing one means the
    /// incoming contract for this swap is already funded and only ever pays out
    /// one preimage. A second call therefore has nothing to fund; it wants the
    /// preimage the first call is buying, and gets it by being handed the
    /// operation already in progress for
    /// [`pay::GatewayPayWaitForSwapPreimage`] to follow. Reporting a failure
    /// instead would cancel the outgoing contract that this swap is the other
    /// half of, refunding the sender while the recipient still gets paid out of
    /// the gateway's own funds.
    pub async fn gateway_handle_direct_swap(
        &self,
        swap_params: SwapParameters,
    ) -> anyhow::Result<OperationId> {
        debug!("Handling direct swap {swap_params:?}");

        let payment_hash = swap_params.payment_hash;
        let operation_id = OperationId(payment_hash.to_byte_array());

        // Check before the funding helper: funding the incoming contract consumes
        // the federation's offer, and the await-offer endpoint waits for one to
        // appear rather than reporting that it is gone, so a re-entrant call gets
        // no further than that helper's timeout.
        if self.client_ctx.operation_exists(operation_id).await {
            debug!(
                operation_id = %operation_id.fmt_short(),
                %payment_hash,
                "Direct swap already in progress, returning the operation already funding it"
            );

            return Ok(operation_id);
        }

        let (op_id_from_funding, client_output, client_output_sm) = self
            .create_funding_incoming_contract_output_from_swap(swap_params.clone())
            .await?;
        // Keep the direct derivation above in sync with the funding helper.
        anyhow::ensure!(
            op_id_from_funding == operation_id,
            "operation id derivation must match: {op_id_from_funding:?} != {operation_id:?}"
        );

        self.client_ctx
            .module_db()
            .autocommit(
                |dbtx, _| {
                    let client_output = client_output.clone();
                    let client_output_sm = client_output_sm.clone();
                    Box::pin(async move {
                        // The check above raced anyone who got here first; this one
                        // shares a transaction with the write that would settle the
                        // race, so exactly one of us funds the contract and the
                        // other joins that operation.
                        if self
                            .client_ctx
                            .get_operation_dbtx(dbtx, operation_id)
                            .await
                            .is_some()
                        {
                            debug!(
                                operation_id = %operation_id.fmt_short(),
                                %payment_hash,
                                "Concurrent direct swap won the race, returning the operation already funding it"
                            );

                            return Ok(operation_id);
                        }

                        let output = ClientOutput {
                            output: LightningOutput::V0(client_output.output),
                            amounts: client_output.amounts,
                        };
                        let tx = TransactionBuilder::new().with_outputs(
                            self.client_ctx.make_client_outputs(ClientOutputBundle::new(
                                vec![output],
                                vec![client_output_sm],
                            )),
                        );

                        self.client_ctx
                            .finalize_and_submit_transaction_dbtx(
                                dbtx,
                                operation_id,
                                KIND.as_str(),
                                |_: OutPointRange| GatewayMeta::Receive,
                                tx,
                            )
                            .await?;

                        debug!(
                            ?operation_id,
                            %payment_hash,
                            "Submitted funding transaction for direct swap"
                        );

                        Ok(operation_id)
                    })
                },
                Some(100),
            )
            .await
            .map_err(|e| match e {
                AutocommitError::ClosureError { error, .. } => error,
                AutocommitError::CommitFailed { last_error, .. } => {
                    anyhow::anyhow!("Commit to DB failed: {last_error}")
                }
            })
    }

    /// Subscribe to updates when the gateway is handling an intercepted HTLC,
    /// or direct swap between federations
    pub async fn gateway_subscribe_ln_receive(
        &self,
        operation_id: OperationId,
    ) -> anyhow::Result<UpdateStreamOrOutcome<GatewayExtReceiveStates>> {
        let operation = self.client_ctx.get_operation(operation_id).await?;
        let mut stream = self.notifier.subscribe(operation_id).await;
        let client_ctx = self.client_ctx.clone();

        Ok(self.client_ctx.outcome_or_updates(&operation, operation_id, |state| match state {
                GatewayExtReceiveStates::Funding => false,
                GatewayExtReceiveStates::Preimage(_)
                | GatewayExtReceiveStates::RefundSuccess { .. }
                | GatewayExtReceiveStates::RefundError { .. }
                | GatewayExtReceiveStates::FundingFailed { .. } => true,
            }, move || {
            stream! {

                yield GatewayExtReceiveStates::Funding;

                let state = loop {
                    debug!("Getting next ln receive state for {}", operation_id.fmt_short());
                    if let Some(GatewayClientStateMachines::Receive(state)) = stream.next().await {
                        match state.state {
                            IncomingSmStates::Preimage(preimage) =>{
                                debug!(?operation_id, "Received preimage");
                                break GatewayExtReceiveStates::Preimage(preimage)
                            },
                            IncomingSmStates::RefundSubmitted { out_points, error } => {
                                debug!(?operation_id, "Refund submitted for {out_points:?} {error}");
                                match client_ctx.await_primary_module_outputs(operation_id, out_points.clone()).await {
                                    Ok(()) => {
                                        debug!(?operation_id, "Refund success");
                                        break GatewayExtReceiveStates::RefundSuccess { out_points, error }
                                    },
                                    Err(e) => {
                                        warn!(?operation_id, "Got failure {e:?} while awaiting for refund outputs {out_points:?}");
                                        break GatewayExtReceiveStates::RefundError{ error_message: e.to_string(), error }
                                    },
                                }
                            },
                            IncomingSmStates::FundingFailed { error } => {
                                warn!(?operation_id, "Funding failed: {error:?}");
                                break GatewayExtReceiveStates::FundingFailed{ error }
                            },
                            other => {
                                debug!("Got state {other:?} while awaiting for output of {}", operation_id.fmt_short());
                            }
                        }
                    }
                };
                yield state;
            }
        }))
    }

    /// For the given `OperationId`, this function will wait until the Complete
    /// state machine has finished or failed.
    pub async fn await_completion(&self, operation_id: OperationId) {
        let mut stream = self.notifier.subscribe(operation_id).await;
        loop {
            match stream.next().await {
                Some(GatewayClientStateMachines::Complete(state)) => match state.state {
                    GatewayCompleteStates::HtlcFinished => {
                        info!(%state, "LNv1 completion state machine finished");
                        return;
                    }
                    GatewayCompleteStates::Failure => {
                        error!(%state, "LNv1 completion state machine failed");
                        return;
                    }
                    _ => {
                        info!(%state, "Waiting for LNv1 completion state machine");
                        continue;
                    }
                },
                Some(GatewayClientStateMachines::Receive(state)) => {
                    info!(%state, "Waiting for LNv1 completion state machine");
                    continue;
                }
                Some(state) => {
                    warn!(%state, "Operation is not an LNv1 completion state machine");
                    return;
                }
                None => return,
            }
        }
    }

    /// Pay lightning invoice on behalf of federation user
    pub async fn gateway_pay_bolt11_invoice(
        &self,
        pay_invoice_payload: PayInvoicePayload,
    ) -> anyhow::Result<OperationId> {
        let payload = pay_invoice_payload.clone();

        // `payment_data` is caller-supplied on the unauthenticated `/pay_invoice`
        // route, so an amountless BOLT11 reaches us here. The state machine
        // rejects it in `validate_outgoing_account`, but that runs only after
        // this function has already recorded the invoice amount, so the amount
        // has to be resolved before any of that work starts.
        let invoice_amount = pay_invoice_payload
            .payment_data
            .amount()
            .ok_or(OutgoingContractError::InvoiceMissingAmount)?;

        self.lightning_manager
            .verify_pruned_invoice(pay_invoice_payload.payment_data)
            .await?;

        self.client_ctx.module_db()
            .autocommit(
                |dbtx, _| {
                    Box::pin(async {
                        let operation_id = OperationId(payload.contract_id.to_byte_array());

                        // The operation id is the contract id, so an existing entry means we
                        // already accepted a request to pay this very contract. The state
                        // machine's own dedupe key covers the whole payload, so a caller who
                        // varies any field of it -- `preimage_auth`, say -- slips past that
                        // and would have us buy the preimage a second time, spending the
                        // gateway's funds twice against a contract that only pays out once.
                        // Hand back the operation already under way instead.
                        if let Some(entry) =
                            self.client_ctx.get_operation_dbtx(dbtx, operation_id).await
                        {
                            // This operation id yields the preimage, so only the
                            // caller that started the payment may join it. The
                            // state machine's own check comes too late for a
                            // request that never reaches one, and it is keyed on
                            // `payment_data`, which the caller supplies
                            // independently of `contract_id` -- so it would answer
                            // for the wrong payment here.
                            if !matches!(
                                entry.try_meta::<GatewayMeta>(),
                                Ok(GatewayMeta::Pay { preimage_auth })
                                    if preimage_auth == payload.preimage_auth
                            ) {
                                anyhow::bail!(
                                    "Not authorized to receive the preimage for contract {}",
                                    payload.contract_id
                                );
                            }

                            debug!(
                                operation_id = %operation_id.fmt_short(),
                                contract_id = %payload.contract_id,
                                "Duplicate request to pay an outgoing contract, returning the operation already in progress"
                            );

                            return Ok(operation_id);
                        }

                        self.client_ctx.log_event(dbtx, OutgoingPaymentStarted {
                            contract_id: payload.contract_id,
                            invoice_amount,
                            operation_id,
                        }).await;

                        let state_machines =
                            vec![GatewayClientStateMachines::Pay(GatewayPayStateMachine {
                                common: GatewayPayCommon { operation_id },
                                state: GatewayPayStates::PayInvoice(GatewayPayInvoice {
                                    pay_invoice_payload: payload.clone(),
                                }),
                            })];

                        let dyn_states = state_machines
                            .into_iter()
                            .map(|s| self.client_ctx.make_dyn(s))
                            .collect();

                            match self.client_ctx.add_state_machines_dbtx(dbtx, dyn_states).await {
                                Ok(()) => {
                                    self.client_ctx
                                        .add_operation_log_entry_dbtx(
                                            dbtx,
                                            operation_id,
                                            KIND.as_str(),
                                            GatewayMeta::Pay {
                                                preimage_auth: payload.preimage_auth,
                                            },
                                        )
                                        .await;
                                }
                                Err(AddStateMachinesError::StateAlreadyExists) => {
                                    info!("State machine for operation {} already exists, will not add a new one", operation_id.fmt_short());
                                }
                                Err(other) => {
                                    anyhow::bail!("Failed to add state machines: {other:?}")
                                }
                            }
                            Ok(operation_id)
                    })
                },
                Some(100),
            )
            .await
            .map_err(|e| match e {
                AutocommitError::ClosureError { error, .. } => error,
                AutocommitError::CommitFailed { last_error, .. } => {
                    anyhow::anyhow!("Commit to DB failed: {last_error}")
                }
            })
    }

    pub async fn gateway_subscribe_ln_pay(
        &self,
        operation_id: OperationId,
    ) -> anyhow::Result<UpdateStreamOrOutcome<GatewayExtPayStates>> {
        let mut stream = self.notifier.subscribe(operation_id).await;
        let operation = self.client_ctx.get_operation(operation_id).await?;
        let client_ctx = self.client_ctx.clone();

        Ok(self.client_ctx.outcome_or_updates(&operation, operation_id, |state| match state {
                GatewayExtPayStates::Created | GatewayExtPayStates::Preimage { .. } => false,
                GatewayExtPayStates::Success { .. }
                | GatewayExtPayStates::Canceled { .. }
                | GatewayExtPayStates::Fail { .. }
                | GatewayExtPayStates::OfferDoesNotExist { .. } => true,
            }, move || {
            stream! {
                yield GatewayExtPayStates::Created;

                loop {
                    debug!("Getting next ln pay state for {}", operation_id.fmt_short());
                    match stream.next().await { Some(GatewayClientStateMachines::Pay(state)) => {
                        match state.state {
                            GatewayPayStates::Preimage(out_points, preimage) => {
                                yield GatewayExtPayStates::Preimage{ preimage: preimage.clone() };

                                match client_ctx.await_primary_module_outputs(operation_id, out_points.clone()).await {
                                    Ok(()) => {
                                        debug!(?operation_id, "Success");
                                        yield GatewayExtPayStates::Success{ preimage: preimage.clone(), out_points };
                                        return;

                                    }
                                    Err(e) => {
                                        warn!(?operation_id, "Got failure {e:?} while awaiting for outputs {out_points:?}");
                                        // TODO: yield something here?
                                    }
                                }
                            }
                            GatewayPayStates::Canceled { txid, contract_id, error } => {
                                debug!(?operation_id, "Trying to cancel contract {contract_id:?} due to {error:?}");
                                match client_ctx.transaction_updates(operation_id).await.await_tx_accepted(txid).await {
                                    Ok(()) => {
                                        debug!(?operation_id, "Canceled contract {contract_id:?} due to {error:?}");
                                        yield GatewayExtPayStates::Canceled{ error };
                                        return;
                                    }
                                    Err(e) => {
                                        warn!(?operation_id, "Got failure {e:?} while awaiting for transaction {txid} to be accepted for");
                                        yield GatewayExtPayStates::Fail { error, error_message: format!("Refund transaction {txid} was not accepted by the federation. OperationId: {} Error: {e:?}", operation_id.fmt_short()) };
                                    }
                                }
                            }
                            GatewayPayStates::OfferDoesNotExist(contract_id) => {
                                warn!("Yielding OfferDoesNotExist state for {} and contract {contract_id}", operation_id.fmt_short());
                                yield GatewayExtPayStates::OfferDoesNotExist { contract_id };
                            }
                            GatewayPayStates::Failed{ error, error_message } => {
                                warn!("Yielding Fail state for {} due to {error:?} {error_message:?}", operation_id.fmt_short());
                                yield GatewayExtPayStates::Fail{ error, error_message };
                            },
                            GatewayPayStates::PayInvoice(_) => {
                                debug!("Got initial state PayInvoice while awaiting for output of {}", operation_id.fmt_short());
                            }
                            other => {
                                info!("Got state {other:?} while awaiting for output of {}", operation_id.fmt_short());
                            }
                        }
                    } _ => {
                        warn!("Got None while getting next ln pay state for {}", operation_id.fmt_short());
                    }}
                }
            }
        }))
    }
}

#[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
struct HtlcCircuitKey {
    operation_id: OperationId,
    incoming_chan_id: u64,
    htlc_id: u64,
}

impl HtlcCircuitKey {
    fn matches_state(self, state: &GatewayClientStateMachines) -> bool {
        matches!(
            state,
            GatewayClientStateMachines::Complete(sm)
                if sm.common.operation_id == self.operation_id
                    && sm.common.incoming_chan_id == self.incoming_chan_id
                    && sm.common.htlc_id == self.htlc_id
        )
    }
}

#[derive(Debug, Clone, Eq, PartialEq, Hash, Decodable, Encodable)]
pub enum GatewayClientStateMachines {
    Pay(GatewayPayStateMachine),
    Receive(IncomingStateMachine),
    Complete(GatewayCompleteStateMachine),
}

impl fmt::Display for GatewayClientStateMachines {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            GatewayClientStateMachines::Pay(pay) => {
                write!(f, "{pay}")
            }
            GatewayClientStateMachines::Receive(receive) => {
                write!(f, "{receive}")
            }
            GatewayClientStateMachines::Complete(complete) => {
                write!(f, "{complete}")
            }
        }
    }
}

impl IntoDynInstance for GatewayClientStateMachines {
    type DynType = DynState;

    fn into_dyn(self, instance_id: ModuleInstanceId) -> Self::DynType {
        DynState::from_typed(instance_id, self)
    }
}

impl State for GatewayClientStateMachines {
    type ModuleContext = GatewayClientContext;

    fn transitions(
        &self,
        context: &Self::ModuleContext,
        global_context: &DynGlobalClientContext,
    ) -> Vec<StateTransition<Self>> {
        match self {
            GatewayClientStateMachines::Pay(pay_state) => {
                sm_enum_variant_translation!(
                    pay_state.transitions(context, global_context),
                    GatewayClientStateMachines::Pay
                )
            }
            GatewayClientStateMachines::Receive(receive_state) => {
                sm_enum_variant_translation!(
                    receive_state.transitions(&context.into(), global_context),
                    GatewayClientStateMachines::Receive
                )
            }
            GatewayClientStateMachines::Complete(complete_state) => {
                sm_enum_variant_translation!(
                    complete_state.transitions(context, global_context),
                    GatewayClientStateMachines::Complete
                )
            }
        }
    }

    fn operation_id(&self) -> fedimint_core::core::OperationId {
        match self {
            GatewayClientStateMachines::Pay(pay_state) => pay_state.operation_id(),
            GatewayClientStateMachines::Receive(receive_state) => receive_state.operation_id(),
            GatewayClientStateMachines::Complete(complete_state) => complete_state.operation_id(),
        }
    }
}

#[derive(Debug, Clone, Eq, PartialEq)]
pub struct Htlc {
    /// The HTLC payment hash.
    pub payment_hash: sha256::Hash,
    /// The incoming HTLC amount in millisatoshi.
    pub incoming_amount_msat: Amount,
    /// The outgoing HTLC amount in millisatoshi
    pub outgoing_amount_msat: Amount,
    /// Absolute Bitcoin block height at which the incoming HTLC expires.
    ///
    /// This uses the same coordinate system as the Lightning backend's current
    /// best block height.
    pub incoming_expiry: u32,
    /// The short channel id of the HTLC.
    pub short_channel_id: Option<u64>,
    /// The id of the incoming channel
    pub incoming_chan_id: u64,
    /// The index of the incoming htlc in the incoming channel
    pub htlc_id: u64,
}

/// An intercepted LNv1 HTLC does not leave enough time for safe settlement.
#[derive(Debug, Error, Clone, Eq, PartialEq)]
#[error(
    "incoming HTLC expiry is unsafe: expiry {incoming_expiry}, current height {current_block_height}, required remaining blocks greater than {expiry_safety_margin}"
)]
pub struct UnsafeHtlcExpiry {
    /// Absolute Bitcoin block height at which the HTLC expires.
    pub incoming_expiry: u32,
    /// Lightning backend's current best Bitcoin block height.
    pub current_block_height: u32,
    /// Reserved settlement and on-chain safety margin in blocks.
    pub expiry_safety_margin: u32,
}

impl Htlc {
    /// Rejects an HTLC unless its remaining lifetime exceeds the LNv1 safety
    /// margin.
    pub fn ensure_safe_expiry(&self, current_block_height: u32) -> Result<(), UnsafeHtlcExpiry> {
        let remaining_blocks = self.incoming_expiry.saturating_sub(current_block_height);
        if LNV1_HTLC_EXPIRY_SAFETY_MARGIN < remaining_blocks {
            return Ok(());
        }

        Err(UnsafeHtlcExpiry {
            incoming_expiry: self.incoming_expiry,
            current_block_height,
            expiry_safety_margin: LNV1_HTLC_EXPIRY_SAFETY_MARGIN,
        })
    }
}

impl TryFrom<InterceptPaymentRequest> for Htlc {
    type Error = anyhow::Error;

    fn try_from(s: InterceptPaymentRequest) -> Result<Self, Self::Error> {
        Ok(Self {
            payment_hash: s.payment_hash,
            incoming_amount_msat: Amount::from_msats(s.amount_msat),
            outgoing_amount_msat: Amount::from_msats(s.amount_msat),
            incoming_expiry: s.expiry,
            short_channel_id: s.short_channel_id,
            incoming_chan_id: s.incoming_chan_id,
            htlc_id: s.htlc_id,
        })
    }
}

#[derive(Debug, Clone)]
pub struct SwapParameters {
    pub payment_hash: sha256::Hash,
    pub amount_msat: Amount,
}

impl TryFrom<PaymentData> for SwapParameters {
    type Error = anyhow::Error;

    fn try_from(s: PaymentData) -> Result<Self, Self::Error> {
        let payment_hash = s.payment_hash();
        let amount_msat = s
            .amount()
            .ok_or_else(|| anyhow::anyhow!("Amountless invoice cannot be used in direct swap"))?;
        Ok(Self {
            payment_hash,
            amount_msat,
        })
    }
}

/// An interface between module implementation and the general `Gateway`
///
/// To abstract away and decouple the core gateway from the modules, the
/// interface between them is expressed as a trait. The gateway handles
/// operations that require Lightning node access or database access.
#[async_trait]
pub trait IGatewayClientV1: Debug + Send + Sync {
    /// Verifies that the supplied `preimage_auth` is the same as the
    /// `preimage_auth` that initiated the payment.
    ///
    /// If it is not, then this will return an error because this client is not
    /// authorized to receive the preimage.
    async fn verify_preimage_authentication(
        &self,
        payment_hash: sha256::Hash,
        preimage_auth: sha256::Hash,
        contract: OutgoingContractAccount,
    ) -> Result<(), OutgoingPaymentError>;

    /// Verify that the lightning node supports private payments if a pruned
    /// invoice is supplied.
    async fn verify_pruned_invoice(&self, payment_data: PaymentData) -> anyhow::Result<()>;

    /// Retrieves the federation's routing fees from the federation's config.
    async fn get_routing_fees(&self, federation_id: FederationId) -> Option<RoutingFees>;

    /// Retrieve a client given a federation ID, used for swapping ecash between
    /// federations.
    async fn get_client(&self, federation_id: &FederationId) -> Option<Spanned<ClientHandleArc>>;

    // Retrieve a client given an invoice.
    //
    // Checks if the invoice route hint last hop has source node id matching this
    // gateways node pubkey and if the short channel id matches one assigned by
    // this gateway to a connected federation. In this case, the gateway can
    // avoid paying the invoice over the lightning network and instead perform a
    // direct swap between the two federations.
    async fn get_client_for_invoice(
        &self,
        payment_data: PaymentData,
    ) -> Option<Spanned<ClientHandleArc>>;

    /// Pay a Lightning invoice using the gateway's lightning node.
    async fn pay(
        &self,
        payment_data: PaymentData,
        max_delay: u64,
        max_fee: Amount,
    ) -> Result<PayInvoiceResponse, LightningRpcError>;

    /// Use the gateway's lightning node to send a complete HTLC response.
    async fn complete_htlc(
        &self,
        htlc_response: InterceptPaymentResponse,
    ) -> Result<(), LightningRpcError>;

    /// Check if the gateway satisfy the LNv1 payment by funding an LNv2
    /// `IncomingContract`
    async fn is_lnv2_direct_swap(
        &self,
        payment_hash: sha256::Hash,
        amount: Amount,
    ) -> anyhow::Result<
        Option<(
            fedimint_lnv2_common::contracts::IncomingContract,
            ClientHandleArc,
        )>,
    >;
}