cdk 0.17.0-rc.1

use std::future::Future;
use std::pin::Pin;
use std::str::FromStr;
use std::sync::Arc;
use std::time::{Duration, Instant};

use cdk_common::database::DynMintDatabase;
use cdk_common::melt::MeltQuoteRequest;
use cdk_common::mint::MeltPaymentRequest;
use cdk_common::nut00::KnownMethod;
use cdk_common::nut05::MeltMethodOptions;
use cdk_common::nuts::nut17::{Kind, NotificationPayload};
use cdk_common::payment::{
    Bolt11OutgoingPaymentOptions, Bolt12OutgoingPaymentOptions, CustomOutgoingPaymentOptions,
    OutgoingPaymentOptions, PaymentIdentifier,
};
use cdk_common::quote_id::QuoteId;
use cdk_common::subscription::Params;
use cdk_common::{
    MeltOptions, MeltQuoteBolt12Request, MeltQuoteCreateResponse, MeltQuoteCustomRequest,
    MeltQuoteCustomResponse, MeltQuoteOnchainRequest, MeltQuoteOnchainResponse, MeltQuoteResponse,
};
use lightning::offers::offer::Offer;
use tracing::instrument;

use super::{
    CurrencyUnit, MeltQuote, MeltQuoteBolt11Request, MeltQuoteBolt11Response,
    MeltQuoteBolt12Response, MeltRequest, Mint, PaymentMethod,
};
use crate::mint::verification::MAX_REQUEST_FIELD_LEN;
use crate::nuts::MeltQuoteState;
use crate::types::PaymentProcessorKey;
use crate::util::unix_time;
use crate::{ensure_cdk, Amount, Error};

pub(crate) mod melt_saga;
pub(crate) mod shared;

#[cfg(test)]
mod tests;

use melt_saga::{MeltSaga, PaymentOutcome};

fn pending_melt_wait_timeout() -> Duration {
    if cfg!(test) {
        // Bumped from 100ms to 250ms to reduce flake on loaded CI while
        // keeping the timeout test fast.
        Duration::from_millis(250)
    } else {
        Duration::from_secs(60)
    }
}

fn pending_melt_notification_wait_interval() -> Duration {
    if cfg!(test) {
        Duration::from_millis(10)
    } else {
        Duration::from_secs(1)
    }
}

fn pending_melt_status_recheck_interval() -> Duration {
    if cfg!(test) {
        Duration::from_millis(20)
    } else {
        Duration::from_secs(5)
    }
}

/// A pending mint melt that can optionally be awaited.
#[derive(Debug)]
pub struct PendingMelt {
    response: MeltQuoteResponse<QuoteId>,
    completion: tokio::task::JoinHandle<Result<MeltQuoteResponse<QuoteId>, Error>>,
}

impl PendingMelt {
    /// Return the immediate pending response (NUT-05 style) without consuming self.
    pub fn pending_response(&self) -> &MeltQuoteResponse<QuoteId> {
        &self.response
    }

    /// Return the immediate pending response (NUT-05 style).
    pub fn into_pending_response(self) -> MeltQuoteResponse<QuoteId> {
        self.response
    }

    async fn wait(self) -> Result<MeltQuoteResponse<QuoteId>, Error> {
        match self.completion.await {
            Ok(result) => result,
            Err(err) => {
                tracing::error!("Background melt task failed to join: {}", err);
                Err(Error::Internal)
            }
        }
    }
}

impl std::future::IntoFuture for PendingMelt {
    type Output = Result<MeltQuoteResponse<QuoteId>, Error>;
    type IntoFuture = Pin<Box<dyn Future<Output = Self::Output> + Send>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(self.wait())
    }
}

impl Mint {
    /// Loads a settled (non-pending, non-unknown) melt response from the database.
    ///
    /// Returns `Ok(None)` while the quote is still `Pending` or in an `Unknown`
    /// state (backend could not determine status); the caller should keep
    /// polling. Returns `Ok(Some(response))` once the quote has reached a
    /// final state (`Paid`, `Failed`, or post-attempt `Unpaid`), with any
    /// change blind signatures attached.
    ///
    /// `Unknown` is treated as "keep polling" by design: a backend may recover
    /// and report a definitive status on a later check. A permanently `Unknown`
    /// quote in the pending-wait path will eventually hit
    /// `pending_melt_wait_timeout` and return `Error::PendingMeltTimeout`;
    /// startup recovery / explicit status checks handle the long-tail case.
    async fn load_settled_melt_response(
        localstore: &DynMintDatabase,
        quote_id: &QuoteId,
    ) -> Result<Option<MeltQuoteResponse<QuoteId>>, Error> {
        let quote = localstore
            .get_melt_quote(quote_id)
            .await?
            .ok_or(Error::UnknownQuote)?;

        if matches!(
            quote.state,
            MeltQuoteState::Pending | MeltQuoteState::Unknown
        ) {
            return Ok(None);
        }

        let change = localstore.get_blind_signatures_for_quote(quote_id).await?;
        let change_opt = if change.is_empty() {
            None
        } else {
            Some(change)
        };

        Ok(Some(quote.into_response(change_opt)))
    }

    /// Attempts to resolve a pending melt quote by checking the backend and
    /// running saga outcome processing.
    ///
    /// Returns `Ok(None)` when reconciliation proceeded normally (whether or
    /// not the quote moved out of `Pending`). Returns `Ok(Some(err_string))`
    /// when the backend status check itself failed — the caller may surface
    /// this message on a subsequent `PendingMeltTimeout` for better diagnostics.
    /// Returns `Err` on database errors or on unrecoverable saga processing
    /// failures.
    async fn reconcile_pending_melt_quote(
        mint: &Mint,
        localstore: &DynMintDatabase,
        pubsub_manager: &Arc<crate::mint::subscription::PubSubManager>,
        quote_id: &QuoteId,
    ) -> Result<Option<String>, Error> {
        let Some(mut quote) = localstore.get_melt_quote(quote_id).await? else {
            return Err(Error::UnknownQuote);
        };

        if quote.state != MeltQuoteState::Pending {
            return Ok(None);
        }

        let Some(saga) = localstore.get_melt_saga_by_quote_id(quote_id).await? else {
            tracing::warn!(
                "Pending melt quote {} has no saga metadata during reconciliation",
                quote_id
            );
            return Ok(None);
        };

        let payment_response = match mint.check_melt_payment_status(&quote).await {
            Ok(payment_response) => payment_response,
            Err(err) => {
                let err_string = err.to_string();
                tracing::warn!(
                    "Failed to reconcile pending melt quote {} with backend status: {}",
                    quote_id,
                    err_string
                );
                return Ok(Some(err_string));
            }
        };

        crate::mint::saga_recovery::process_melt_saga_outcome(
            &saga,
            &mut quote,
            &payment_response,
            localstore,
            pubsub_manager,
            mint,
        )
        .await?;

        Ok(None)
    }

    #[instrument(skip_all)]
    async fn check_melt_request_acceptable(
        &self,
        amount: Amount<CurrencyUnit>,
        method: PaymentMethod,
        request: String,
        options: Option<MeltOptions>,
    ) -> Result<(), Error> {
        let unit = amount.unit().clone();
        let mint_info = self.mint_info().await?;
        let nut05 = mint_info.nuts.nut05;

        ensure_cdk!(!nut05.disabled, Error::MeltingDisabled);

        let settings = nut05
            .get_settings(&unit, &method)
            .ok_or(Error::UnsupportedUnit)?;

        match options {
            Some(MeltOptions::Mpp { mpp: _ }) => {
                let nut15 = mint_info.nuts.nut15;
                // Verify there is no corresponding mint quote.
                // Otherwise a wallet is trying to pay someone internally, but
                // with a multi-part quote. And that's just not possible.
                if (self.localstore.get_mint_quote_by_request(&request).await?).is_some() {
                    return Err(Error::InternalMultiPartMeltQuote);
                }
                // Verify MPP is enabled for unit and method
                if !nut15
                    .methods
                    .into_iter()
                    .any(|m| m.method == method && m.unit == unit)
                {
                    return Err(Error::MppUnitMethodNotSupported(unit, method));
                }
            }
            Some(MeltOptions::Amountless { amountless: _ })
                if method.is_bolt11()
                    && !matches!(
                        settings.options,
                        Some(MeltMethodOptions::Bolt11 { amountless: true })
                    ) =>
            {
                return Err(Error::AmountlessInvoiceNotSupported(unit, method));
            }
            _ => {}
        };

        // Compare using raw values since settings use Amount without unit
        let amount_value = amount.value();
        let is_above_max = matches!(settings.max_amount, Some(max) if amount_value > max.into());
        let is_below_min = matches!(settings.min_amount, Some(min) if amount_value < min.into());
        match is_above_max || is_below_min {
            true => {
                tracing::error!(
                    "Melt amount out of range: {} is not within {} and {}",
                    amount,
                    settings.min_amount.unwrap_or_default(),
                    settings.max_amount.unwrap_or_default(),
                );
                Err(Error::AmountOutofLimitRange(
                    settings.min_amount.unwrap_or_default(),
                    settings.max_amount.unwrap_or_default(),
                    amount.into(),
                ))
            }
            false => Ok(()),
        }
    }

    /// Get melt quote for BOLT11, BOLT12, or Custom payment methods
    ///
    /// This function accepts a `MeltQuoteRequest` enum and delegates to the
    /// appropriate handler based on the request type.
    #[instrument(skip_all)]
    pub async fn get_melt_quote(
        &self,
        melt_quote_request: MeltQuoteRequest,
    ) -> Result<MeltQuoteCreateResponse<QuoteId>, Error> {
        match melt_quote_request {
            MeltQuoteRequest::Bolt11(bolt11_request) => Ok(MeltQuoteCreateResponse::Bolt11(
                self.get_melt_bolt11_quote_impl(&bolt11_request).await?,
            )),
            MeltQuoteRequest::Bolt12(bolt12_request) => Ok(MeltQuoteCreateResponse::Bolt12(
                self.get_melt_bolt12_quote_impl(&bolt12_request).await?,
            )),
            MeltQuoteRequest::Onchain(onchain_request) => Ok(MeltQuoteCreateResponse::Onchain(
                self.get_melt_onchain_quote_impl(&onchain_request).await?,
            )),
            MeltQuoteRequest::Custom(request) => {
                let quote = self.get_melt_custom_quote_impl(&request).await?;
                let method = PaymentMethod::from(request.method.as_str());

                Ok(MeltQuoteCreateResponse::Custom((method, quote)))
            }
        }
    }

    /// Implementation of get_melt_bolt11_quote
    #[instrument(skip_all)]
    async fn get_melt_bolt11_quote_impl(
        &self,
        melt_request: &MeltQuoteBolt11Request,
    ) -> Result<MeltQuoteBolt11Response<QuoteId>, Error> {
        #[cfg(feature = "prometheus")]
        let metrics = super::MintMetricGuard::new("get_melt_bolt11_quote");

        let result = async {
            let MeltQuoteBolt11Request {
                request,
                unit,
                options,
                ..
            } = melt_request;

            let ln = self
                .payment_processors
                .get(&PaymentProcessorKey::new(
                    unit.clone(),
                    PaymentMethod::Known(KnownMethod::Bolt11),
                ))
                .ok_or_else(|| {
                    tracing::info!("Could not get ln backend for {}, bolt11 ", unit);

                    Error::UnsupportedUnit
                })?;

            // Pre-generate the quote id so we can pass it to the backend in both
            // `get_payment_quote` and the eventual `make_payment`, and use the same
            // id when we persist the quote below.
            let quote_id = cdk_common::QuoteId::new_uuid();

            let bolt11 = Bolt11OutgoingPaymentOptions {
                bolt11: melt_request.request.clone(),
                max_fee_amount: None,
                timeout_secs: None,
                melt_options: melt_request.options,
                quote_id: quote_id.clone(),
            };

            let payment_quote = ln
                .get_payment_quote(
                    &melt_request.unit,
                    OutgoingPaymentOptions::Bolt11(Box::new(bolt11)),
                )
                .await
                .map_err(|err| {
                    tracing::error!(
                        "Could not get payment quote for mint quote, {} bolt11, {}",
                        unit,
                        err
                    );

                    err
                })?;

            if payment_quote.unit() != unit {
                return Err(Error::UnitMismatch);
            }

            // Validate using processor quote amount for currency conversion
            self.check_melt_request_acceptable(
                payment_quote.amount.clone(),
                PaymentMethod::Known(KnownMethod::Bolt11),
                request.to_string(),
                *options,
            )
            .await?;

            // Extract values for quote creation
            let quote_amount = payment_quote.amount;
            let quote_fee = payment_quote.fee;

            let melt_ttl = self.quote_ttl().await?.melt_ttl;

            let quote = MeltQuote::new(
                Some(quote_id),
                MeltPaymentRequest::Bolt11 {
                    bolt11: request.clone(),
                },
                unit.clone(),
                quote_amount.clone(),
                quote_fee,
                unix_time() + melt_ttl,
                payment_quote.request_lookup_id.clone(),
                *options,
                PaymentMethod::Known(KnownMethod::Bolt11),
                payment_quote.extra_json,
                payment_quote.estimated_blocks,
            );

            tracing::debug!(
                "New {} melt quote {} for {} {} with request id {:?}",
                quote.payment_method,
                quote.id,
                quote_amount,
                unit,
                payment_quote.request_lookup_id
            );

            let mut tx = self.localstore.begin_transaction().await?;
            tx.add_melt_quote(quote.clone()).await?;
            tx.commit().await?;

            Ok(quote.into())
        }
        .await;

        #[cfg(feature = "prometheus")]
        {
            metrics.record(result.is_ok());
        }

        result
    }

    /// Implementation of get_melt_bolt12_quote
    #[instrument(skip_all)]
    async fn get_melt_bolt12_quote_impl(
        &self,
        melt_request: &MeltQuoteBolt12Request,
    ) -> Result<MeltQuoteBolt11Response<QuoteId>, Error> {
        #[cfg(feature = "prometheus")]
        let metrics = super::MintMetricGuard::new("get_melt_bolt12_quote");

        let result = async {
            let MeltQuoteBolt12Request {
                request,
                unit,
                options,
            } = melt_request;

            let ln = self
                .payment_processors
                .get(&PaymentProcessorKey::new(
                    unit.clone(),
                    PaymentMethod::Known(KnownMethod::Bolt12),
                ))
                .ok_or_else(|| {
                    tracing::info!("Could not get ln backend for {}, bolt12 ", unit);

                    Error::UnsupportedUnit
                })?;

            let offer = Offer::from_str(&melt_request.request).map_err(|_| Error::Bolt12parse)?;

            let quote_id = cdk_common::QuoteId::new_uuid();

            let outgoing_payment_options = Bolt12OutgoingPaymentOptions {
                offer: offer.clone(),
                max_fee_amount: None,
                timeout_secs: None,
                melt_options: *options,
                quote_id: quote_id.clone(),
            };

            let payment_quote = ln
                .get_payment_quote(
                    &melt_request.unit,
                    OutgoingPaymentOptions::Bolt12(Box::new(outgoing_payment_options)),
                )
                .await
                .map_err(|err| {
                    tracing::error!(
                        "Could not get payment quote for mint quote, {} bolt12, {}",
                        unit,
                        err
                    );

                    err
                })?;

            if payment_quote.unit() != unit {
                return Err(Error::UnitMismatch);
            }

            // Validate using processor quote amount for currency conversion
            self.check_melt_request_acceptable(
                payment_quote.amount.clone(),
                PaymentMethod::Known(KnownMethod::Bolt12),
                request.clone(),
                *options,
            )
            .await?;

            // Extract values for quote creation
            let quote_amount = payment_quote.amount;
            let quote_fee = payment_quote.fee;

            let payment_request = MeltPaymentRequest::Bolt12 {
                offer: Box::new(offer),
            };

            let quote = MeltQuote::new(
                Some(quote_id),
                payment_request,
                unit.clone(),
                quote_amount.clone(),
                quote_fee,
                unix_time() + self.quote_ttl().await?.melt_ttl,
                payment_quote.request_lookup_id.clone(),
                *options,
                PaymentMethod::Known(KnownMethod::Bolt12),
                payment_quote.extra_json,
                payment_quote.estimated_blocks,
            );

            tracing::debug!(
                "New {} melt quote {} for {} {} with request id {:?}",
                quote.payment_method,
                quote.id,
                quote_amount,
                unit,
                payment_quote.request_lookup_id
            );

            let mut tx = self.localstore.begin_transaction().await?;
            tx.add_melt_quote(quote.clone()).await?;
            tx.commit().await?;

            Ok(quote.into())
        }
        .await;

        #[cfg(feature = "prometheus")]
        {
            metrics.record(result.is_ok());
        }

        result
    }

    /// Implementation of get_melt_onchain_quote
    #[instrument(skip_all)]
    async fn get_melt_onchain_quote_impl(
        &self,
        melt_request: &MeltQuoteOnchainRequest,
    ) -> Result<MeltQuoteOnchainResponse<QuoteId>, Error> {
        #[cfg(feature = "prometheus")]
        let metrics = super::MintMetricGuard::new("get_melt_onchain_quote");

        let result = async {
            let unit = &melt_request.unit;

            let ln = self
                .payment_processors
                .get(&PaymentProcessorKey::new(
                    unit.clone(),
                    PaymentMethod::Known(KnownMethod::Onchain),
                ))
                .ok_or_else(|| {
                    tracing::info!("Could not get ln backend for {}, onchain ", unit);
                    Error::UnsupportedUnit
                })?;

            // Generate the authoritative `QuoteId` on the mint side *before*
            // calling the backend. The onchain contract
            // ([`OnchainOutgoingPaymentOptions.quote_id`]) requires backends to
            // echo this value verbatim in
            // [`PaymentQuoteResponse.request_lookup_id`] as
            // `PaymentIdentifier::QuoteId(..)`; we validate that echo below and
            // use our locally-generated id as the `MeltQuote.id` so the flow is
            // no longer self-referential via the backend response.
            let quote_id = QuoteId::new_uuid();

            let outgoing_payment_options = cdk_common::payment::OnchainOutgoingPaymentOptions {
                address: melt_request.request.clone(),
                amount: melt_request.amount.with_unit(unit.clone()),
                max_fee_amount: None,
                quote_id: quote_id.clone(),
                // No selection at quote creation: the backend enumerates all
                // available `fee_options` and the wallet picks one (echoed back
                // as `fee_index`) when executing the melt.
                fee_index: None,
                metadata: None,
            };

            let payment_quote = ln
                .get_payment_quote(
                    unit,
                    OutgoingPaymentOptions::Onchain(Box::new(outgoing_payment_options)),
                )
                .await
                .map_err(|err| {
                    tracing::error!(
                        "Could not get payment quote for mint quote, {} onchain, {}",
                        unit,
                        err
                    );
                    err
                })?;

            if payment_quote.unit() != unit {
                return Err(Error::UnitMismatch);
            }

            // Enforce the onchain quote-id echo contract: the backend MUST return
            // `Some(PaymentIdentifier::QuoteId(quote_id))` matching the value we
            // supplied. Anything else is a contract violation; reject loudly
            // instead of silently accepting whatever the backend decided to
            // synthesize.
            match &payment_quote.request_lookup_id {
                Some(PaymentIdentifier::QuoteId(id)) if id == &quote_id => {}
                got => {
                    return Err(Error::OnchainQuoteLookupIdMismatch {
                        expected: quote_id,
                        got: got.clone(),
                    });
                }
            }

            // Validate using processor quote amount for currency conversion
            self.check_melt_request_acceptable(
                payment_quote.amount.clone(),
                PaymentMethod::Known(KnownMethod::Onchain),
                melt_request.request.clone(),
                None,
            )
            .await?;

            let melt_ttl = self.quote_ttl().await?.melt_ttl;

            // Store `request_lookup_id` deterministically from the mint-generated
            // `quote_id` rather than cloning the backend response, so the
            // persisted lookup id no longer depends on backend behavior beyond
            // the validated echo.
            let request_lookup_id = Some(PaymentIdentifier::QuoteId(quote_id.clone()));

            // Onchain backends must return explicit `fee_options`; the mint
            // validates them before persisting the quote.
            let Some(fee_options) = payment_quote.fee_options.clone() else {
                return Err(Error::OnchainFeeOptionsEmpty);
            };

            // `MeltQuote::new_onchain` applies the NUT validation. Failures are
            // returned before the quote is persisted, so a backend that violates
            // the contract never leaves state behind in the mint.
            let quote = MeltQuote::new_onchain(
                Some(quote_id),
                MeltPaymentRequest::Onchain {
                    address: melt_request.request.clone(),
                },
                unit.clone(),
                payment_quote.amount,
                unix_time() + melt_ttl,
                request_lookup_id,
                payment_quote.extra_json,
                fee_options,
            )?;

            let mut tx = self.localstore.begin_transaction().await?;
            tx.add_melt_quote(quote.clone()).await?;
            tx.commit().await?;

            Ok(quote.into())
        }
        .await;

        #[cfg(feature = "prometheus")]
        {
            metrics.record(result.is_ok());
        }

        result
    }

    /// Implementation of get_melt_custom_quote
    #[instrument(skip_all)]
    async fn get_melt_custom_quote_impl(
        &self,
        melt_request: &MeltQuoteCustomRequest,
    ) -> Result<MeltQuoteCustomResponse<QuoteId>, Error> {
        #[cfg(feature = "prometheus")]
        let metrics = super::MintMetricGuard::new("get_melt_custom_quote");

        let result = async {
            let MeltQuoteCustomRequest {
                request,
                unit,
                method,
                extra,
            } = melt_request;

            if !extra.is_null() {
                let extra_str = extra.to_string();
                if extra_str.len() > MAX_REQUEST_FIELD_LEN {
                    return Err(Error::RequestFieldTooLarge {
                        field: "extra".to_string(),
                        actual: extra_str.len(),
                        max: MAX_REQUEST_FIELD_LEN,
                    });
                }
            }

            let ln = self
                .payment_processors
                .get(&PaymentProcessorKey::new(
                    unit.clone(),
                    PaymentMethod::from(method.as_str()),
                ))
                .ok_or_else(|| {
                    tracing::info!("Could not get payment processor for {}, {} ", unit, method);
                    Error::UnsupportedUnit
                })?;

            // Convert extra serde_json::Value to JSON string if not null
            let extra_json = if extra.is_null() {
                None
            } else {
                Some(extra.to_string())
            };

            let quote_id = cdk_common::QuoteId::new_uuid();

            let custom_options =
                OutgoingPaymentOptions::Custom(Box::new(CustomOutgoingPaymentOptions {
                    method: method.to_string(),
                    request: request.clone(),
                    max_fee_amount: None,
                    timeout_secs: None,
                    melt_options: None,
                    extra_json,
                    quote_id: quote_id.clone(),
                }));

            let payment_quote = ln
                .get_payment_quote(&melt_request.unit, custom_options)
                .await
                .map_err(|err| {
                    tracing::error!(
                        "Could not get payment quote for melt quote, {} {}, {}",
                        unit,
                        method,
                        err
                    );
                    Error::UnsupportedUnit
                })?;

            if payment_quote.unit() != unit {
                return Err(Error::UnitMismatch);
            }

            // For custom methods, we don't validate amount limits upfront since
            // the payment processor handles method-specific validation
            self.check_melt_request_acceptable(
                payment_quote.amount.clone(),
                PaymentMethod::from(method.as_str()),
                request.clone(),
                None, // Custom methods don't use options
            )
            .await?;

            let melt_ttl = self.quote_ttl().await?.melt_ttl;

            // Extract values for quote creation
            let quote_amount = payment_quote.amount;
            let quote_fee = payment_quote.fee;

            let quote = MeltQuote::new(
                Some(quote_id),
                MeltPaymentRequest::Custom {
                    method: method.to_string(),
                    request: request.clone(),
                },
                unit.clone(),
                quote_amount.clone(),
                quote_fee,
                unix_time() + melt_ttl,
                payment_quote.request_lookup_id.clone(),
                None, // Custom methods don't use options
                PaymentMethod::from(method.as_str()),
                payment_quote.extra_json,
                payment_quote.estimated_blocks,
            );

            tracing::debug!(
                "New {} melt quote {} for {} {} with request id {:?}",
                method,
                quote.id,
                quote_amount,
                unit,
                payment_quote.request_lookup_id
            );

            let mut tx = self.localstore.begin_transaction().await?;
            tx.add_melt_quote(quote.clone()).await?;
            tx.commit().await?;

            Ok(quote.into())
        }
        .await;

        #[cfg(feature = "prometheus")]
        {
            metrics.record(result.is_ok());
        }

        result
    }

    /// Check melt quote status
    #[instrument(skip(self))]
    pub async fn check_melt_quote(
        &self,
        quote_id: &QuoteId,
    ) -> Result<MeltQuoteResponse<QuoteId>, Error> {
        #[cfg(feature = "prometheus")]
        let metrics = super::MintMetricGuard::new("check_melt_quote");

        let result = async {
            let mut quote = self
                .localstore
                .get_melt_quote(quote_id)
                .await?
                .ok_or(Error::UnknownQuote)?;

            self.handle_pending_melt_quote(&mut quote).await?;

            let blind_signatures = self
                .localstore
                .get_blind_signatures_for_quote(quote_id)
                .await?;

            let change = (!blind_signatures.is_empty()).then_some(blind_signatures);

            let response = match quote.payment_method {
                PaymentMethod::Known(KnownMethod::Bolt11) => {
                    MeltQuoteResponse::Bolt11(MeltQuoteBolt11Response {
                        quote: quote.id.clone(),
                        state: quote.state,
                        expiry: quote.expiry,
                        amount: quote.amount().into(),
                        fee_reserve: quote.fee_reserve().into(),
                        payment_preimage: quote.payment_proof.clone(),
                        change: change.clone(),
                        request: Some(quote.request.to_string()),
                        unit: Some(quote.unit.clone()),
                    })
                }
                PaymentMethod::Known(KnownMethod::Bolt12) => {
                    MeltQuoteResponse::Bolt12(MeltQuoteBolt12Response {
                        quote: quote.id.clone(),
                        state: quote.state,
                        expiry: quote.expiry,
                        amount: quote.amount().into(),
                        fee_reserve: quote.fee_reserve().into(),
                        payment_preimage: quote.payment_proof.clone(),
                        change: change.clone(),
                        request: Some(quote.request.to_string()),
                        unit: Some(quote.unit.clone()),
                    })
                }
                PaymentMethod::Known(KnownMethod::Onchain) => {
                    let mut response: MeltQuoteOnchainResponse<QuoteId> = quote.clone().into();
                    response.change = change.clone();
                    MeltQuoteResponse::Onchain(response)
                }
                _ => {
                    let custom_response: MeltQuoteCustomResponse<QuoteId> = quote.clone().into();
                    MeltQuoteResponse::Custom((quote.payment_method.clone(), custom_response))
                }
            };

            Ok(response)
        }
        .await;

        #[cfg(feature = "prometheus")]
        {
            metrics.record(result.is_ok());
        }

        result
    }

    /// Get the configured payment method for an existing melt quote.
    #[instrument(skip(self))]
    pub async fn get_melt_quote_method(&self, quote_id: &QuoteId) -> Result<PaymentMethod, Error> {
        let quote = self
            .localstore
            .get_melt_quote(quote_id)
            .await?
            .ok_or(Error::UnknownQuote)?;

        Ok(quote.payment_method)
    }

    /// Get melt quotes
    #[instrument(skip_all)]
    pub async fn melt_quotes(&self) -> Result<Vec<MeltQuote>, Error> {
        let quotes = self.localstore.get_melt_quotes().await?;
        Ok(quotes)
    }

    /// Melt
    ///
    /// Uses MeltSaga typestate pattern for atomic transaction handling with automatic rollback on failure.
    #[instrument(skip_all)]
    pub async fn melt(&self, melt_request: &MeltRequest<QuoteId>) -> Result<PendingMelt, Error> {
        // Check max outputs limit (if change outputs are provided)
        if let Some(outputs) = melt_request.outputs() {
            let outputs_count = outputs.len();
            if outputs_count > self.max_outputs {
                tracing::warn!(
                    "Melt request exceeds max outputs limit: {} > {}",
                    outputs_count,
                    self.max_outputs
                );
                return Err(Error::MaxOutputsExceeded {
                    actual: outputs_count,
                    max: self.max_outputs,
                });
            }
        }

        let verification = self.verify_inputs(melt_request.inputs()).await?;

        // Fetch the quote to get payment_method for operation tracking
        let quote_id = melt_request.quote().clone();
        let quote = self
            .localstore
            .get_melt_quote(&quote_id)
            .await?
            .ok_or(Error::UnknownQuote)?;

        let init_saga = MeltSaga::new(
            std::sync::Arc::new(self.clone()),
            self.localstore.clone(),
            std::sync::Arc::clone(&self.pubsub_manager),
        );

        // Step 1: Setup (TX1 - reserves inputs and outputs)
        let setup_saga = init_saga
            .setup_melt(melt_request, verification, quote.payment_method.clone())
            .await?;

        let melt_request_owned = melt_request.clone();
        let quote_id_for_log = quote_id.clone();
        let localstore = self.localstore();
        let pubsub = self.pubsub_manager();

        let quote_for_spawn = quote.clone();
        let mint_for_spawn = Arc::new(self.clone());
        let completion = tokio::spawn(async move {
            tracing::debug!(
                "Starting background melt completion for quote: {}",
                quote_id_for_log
            );

            // Step 2: Attempt internal settlement (returns saga + SettlementDecision)
            // Note: Compensation is handled internally if this fails
            let result = match setup_saga
                .attempt_internal_settlement(&melt_request_owned)
                .await
            {
                Ok((setup_saga, settlement)) => {
                    // Step 3: Make payment (internal or external)
                    match setup_saga.make_payment(settlement).await {
                        Ok(PaymentOutcome::Confirmed(payment_saga)) => {
                            // Step 4: Finalize (TX2 - marks spent, issues change)
                            payment_saga.finalize().await
                        }
                        Ok(PaymentOutcome::Pending {
                            #[cfg(feature = "prometheus")]
                            metrics,
                        }) => {
                            let result = async {
                                // Wait for the background task to complete the payment
                                let kind = match quote_for_spawn.payment_method {
                                    PaymentMethod::Known(KnownMethod::Bolt11) => {
                                        Kind::Bolt11MeltQuote
                                    }
                                    PaymentMethod::Known(KnownMethod::Bolt12) => {
                                        Kind::Bolt12MeltQuote
                                    }
                                    PaymentMethod::Known(KnownMethod::Onchain) => {
                                        Kind::OnchainMeltQuote
                                    }
                                    PaymentMethod::Custom(ref method) => {
                                        Kind::Custom(format!("{}_melt_quote", method))
                                    }
                                };

                                let params = Params {
                                    id: Arc::new(crate::subscription::SubId::from(format!(
                                        "melt_sync_wait_{}",
                                        quote_id_for_log
                                    ))),
                                    kind,
                                    filters: vec![quote_id_for_log.to_string()],
                                };

                                let mut sub = match pubsub.subscribe(params) {
                                    Ok(sub) => sub,
                                    Err(_err) => return Err(Error::Internal),
                                };

                                let start = Instant::now();
                                let mut last_status_recheck = start
                                    .checked_sub(pending_melt_status_recheck_interval())
                                    .unwrap_or(start);
                                let mut last_backend_error: Option<String> = None;

                                loop {
                                    if let Some(response) = Self::load_settled_melt_response(
                                        &localstore,
                                        &quote_id_for_log,
                                    )
                                    .await?
                                    {
                                        return Ok(response);
                                    }

                                    if start.elapsed() >= pending_melt_wait_timeout() {
                                        return Err(Error::PendingMeltTimeout {
                                            last_backend_error,
                                        });
                                    }

                                    if last_status_recheck.elapsed()
                                        >= pending_melt_status_recheck_interval()
                                    {
                                        match Self::reconcile_pending_melt_quote(
                                            &mint_for_spawn,
                                            &localstore,
                                            &pubsub,
                                            &quote_id_for_log,
                                        )
                                        .await?
                                        {
                                            Some(err_string) => {
                                                last_backend_error = Some(err_string);
                                            }
                                            None => {
                                                last_backend_error = None;
                                            }
                                        }
                                        last_status_recheck = Instant::now();

                                        if let Some(response) = Self::load_settled_melt_response(
                                            &localstore,
                                            &quote_id_for_log,
                                        )
                                        .await?
                                        {
                                            return Ok(response);
                                        }
                                    }

                                    let Ok(maybe_event) = tokio::time::timeout(
                                        pending_melt_notification_wait_interval(),
                                        sub.recv(),
                                    )
                                    .await
                                    else {
                                        continue;
                                    };

                                    if let Some(event) = maybe_event {
                                        let (event_quote_id, state) = match event.into_inner() {
                                            NotificationPayload::MeltQuoteBolt11Response(r) => {
                                                (r.quote, r.state)
                                            }
                                            NotificationPayload::MeltQuoteBolt12Response(r) => {
                                                (r.quote, r.state)
                                            }
                                            NotificationPayload::CustomMeltQuoteResponse(_, r) => {
                                                (r.quote, r.state)
                                            }
                                            // Onchain is the one method where the subscription
                                            // fast-path is most valuable, because confirmation is
                                            // inherently asynchronous. We subscribe to
                                            // `OnchainMeltQuote` events above (see `Kind` mapping)
                                            // and must handle them here; otherwise onchain waiters
                                            // would rely solely on polling via
                                            // `reconcile_pending_melt_quote`.
                                            NotificationPayload::MeltQuoteOnchainResponse(r) => {
                                                (r.quote, r.state)
                                            }
                                            _ => continue,
                                        };

                                        if event_quote_id == quote_id_for_log
                                            && state != MeltQuoteState::Pending
                                        {
                                            if let Some(response) =
                                                Self::load_settled_melt_response(
                                                    &localstore,
                                                    &quote_id_for_log,
                                                )
                                                .await?
                                            {
                                                return Ok(response);
                                            }
                                        }
                                    }
                                }
                            }
                            .await;

                            #[cfg(feature = "prometheus")]
                            if let Some(metrics) = metrics {
                                metrics.record(result.is_ok());
                            }

                            result
                        }
                        Err(err) => Err(err),
                    }
                }
                Err(err) => Err(err),
            };

            match &result {
                Ok(_) => {
                    tracing::info!(
                        "Background melt completed successfully for quote: {}",
                        quote_id_for_log
                    );
                }
                Err(e) => {
                    tracing::error!(
                        "Background melt completion failed for quote {}: {}",
                        quote_id_for_log,
                        e
                    );
                }
            }

            result
        });

        // Return immediately with the quote in PENDING state and an awaitable completion future.
        let mut quote_clone = quote.clone();
        quote_clone.state = MeltQuoteState::Pending;
        let response = quote_clone.into_response(None);

        Ok(PendingMelt {
            response,
            completion,
        })
    }
}