From 68433412a44ed2b6c54ef36d9c380f1cfee6f4a4 Mon Sep 17 00:00:00 2001 From: Harsh Date: Sun, 30 Aug 2026 05:50:41 +0530 Subject: [PATCH 1/3] feat(payments): add with_server_payments, the server payments registration entry point --- src/payments/mod.rs | 8 +- src/payments/server_explicit_gating.rs | 5 +- src/payments/server_transport_payments.rs | 642 ++++++++++++++++++++++ src/transport/server/mod.rs | 222 ++++++++ 4 files changed, 873 insertions(+), 4 deletions(-) create mode 100644 src/payments/server_transport_payments.rs diff --git a/src/payments/mod.rs b/src/payments/mod.rs index df1cba9..e44ae50 100644 --- a/src/payments/mod.rs +++ b/src/payments/mod.rs @@ -7,8 +7,10 @@ //! [`PaymentError`] taxonomy, the canonical invocation identity used for //! explicit-gating authorization matching, the bounded [`AuthorizationStore`] of //! pending and granted authorizations, and deterministic fakes behind the -//! `test-utils` feature. It carries no transport wiring and no network; the -//! middleware that consumes these arrives later. +//! `test-utils` feature. The one piece of transport wiring here is the +//! [`with_server_payments`](crate::payments::server_transport_payments::with_server_payments) +//! registration entry point, which composes the middlewares, tags, and senders +//! onto a server transport. //! //! Constants and tag builders stay reachable under their module paths //! ([`crate::payments::constants`] / [`crate::payments::tags`]); the wire/config @@ -22,6 +24,7 @@ pub mod errors; pub mod server_explicit_gating; pub mod server_payments; pub(crate) mod server_payments_utils; +pub mod server_transport_payments; pub mod tags; pub mod traits; pub mod types; @@ -41,6 +44,7 @@ pub use server_explicit_gating::{ pub use server_payments::{ create_server_payments_middleware, ServerPaymentsMiddlewareParams, ServerPaymentsOptions, }; +pub use server_transport_payments::with_server_payments; pub use traits::{PaymentHandler, PaymentProcessor, ResolvePrice}; pub use types::{ Meta, PaymentAcceptedParams, PaymentHandlerRequest, PaymentInteractionPolicy, PaymentOption, diff --git a/src/payments/server_explicit_gating.rs b/src/payments/server_explicit_gating.rs index 1c198a3..efb11a6 100644 --- a/src/payments/server_explicit_gating.rs +++ b/src/payments/server_explicit_gating.rs @@ -63,8 +63,9 @@ pub struct ExplicitGatingMiddlewareParams { /// ([`NostrServerTransport::targeted_response_sender`](crate::transport::server::NostrServerTransport::targeted_response_sender)). pub sender: TargetedResponseSender, /// The store of pending and granted payment authorizations. Injected, not constructed - /// here, so the configuration entry point can share one store across registrations and - /// tests can observe lifecycle state. + /// here, so tests can observe lifecycle state and a hand-wiring caller that takes on + /// the composition itself may share a store (the configuration entry point constructs + /// a fresh one and registers at most once). pub authorization_store: AuthorizationStore, /// Pre-built PMI-to-processor map, shared across middlewares. Built locally when `None`. pub processors_by_pmi: Option>>>, diff --git a/src/payments/server_transport_payments.rs b/src/payments/server_transport_payments.rs new file mode 100644 index 0000000..cb6ac8c --- /dev/null +++ b/src/payments/server_transport_payments.rs @@ -0,0 +1,642 @@ +//! CEP-8 payments registration for the server transport. +//! +//! [`with_server_payments`] is the one production entry point that composes the +//! payment stack onto a [`NostrServerTransport`]: it derives the announcement +//! `pmi` / `cap` / `payment_interaction` tag surface from the configuration, +//! records the negotiation policy, and registers the transparent payment +//! middleware plus, under the permissive default policy, the explicit-gating +//! middleware. The pieces it wires (the middleware factories, the tag builders, +//! the injected senders) are all public and can be hand-wired, but only this +//! entry point guarantees they agree with each other and with the wire. + +use std::sync::Arc; + +use crate::core::types::PaymentInteractionMode; +use crate::payments::authorization_store::AuthorizationStore; +use crate::payments::server_explicit_gating::{ + create_explicit_gating_middleware, ExplicitGatingMiddlewareParams, +}; +use crate::payments::server_payments::{ + create_server_payments_middleware, ServerPaymentsMiddlewareParams, ServerPaymentsOptions, +}; +use crate::payments::server_payments_utils::build_processors_by_pmi; +use crate::payments::tags::{cap_tags_from_priced_capabilities, payment_interaction_tag, pmi_tag}; +use crate::payments::types::PaymentInteractionPolicy; +use crate::transport::server::NostrServerTransport; +use nostr_sdk::prelude::Tag; + +const LOG_TARGET: &str = "contextvm_sdk::payments::server_transport_payments"; + +/// Attach CEP-8 payments to a [`NostrServerTransport`]. +/// +/// This is the production registration path for server-side payments. In order, it: +/// builds the PMI-to-processor map once and shares it across both lifecycles; sets the +/// announcement extra tags (one `pmi` tag per processor in registration order, plus a +/// single `payment_interaction=explicit_gating` availability tag when the policy is +/// [`PaymentInteractionPolicy::Optional`]); sets the announcement pricing tags (one +/// `cap` tag per advertisable priced capability); records the payment-interaction +/// policy for session negotiation; and registers the transparent payment middleware, +/// followed (under `Optional`) by the explicit-gating middleware with a fresh +/// [`AuthorizationStore`]. A `Transparent` policy advertises no `payment_interaction` +/// tag and registers no gating middleware; an `explicit_gating` session request is then +/// rejected with a JSON-RPC `-32602`. +/// +/// # Registration contract +/// +/// Call this exactly once, after constructing the transport and before +/// [`start`](NostrServerTransport::start). Both misuses are refused with an error +/// before any state changes: on a started transport the middleware chain and policy +/// are already frozen, so registration would silently take no effect while the live +/// tag setters still advertise payments (priced requests would execute for free); and +/// a second registration would append a second middleware pair that charges every +/// priced request twice. The transport is not restartable after +/// [`close`](NostrServerTransport::close), so a post-close registration is inert. +/// Register before [`announce`](NostrServerTransport::announce), or the first kind +/// 11316 announcement ships without payment tags. +/// +/// # Tag ownership +/// +/// This function owns the announcement extra-tag slot: it replaces any extra tags set +/// earlier through +/// [`set_announcement_extra_tags`](NostrServerTransport::set_announcement_extra_tags), +/// exactly as the reference implementation does. Calling either tag setter after +/// registration is also unsupported: announcements and the normal response path read +/// the new set live, but the payment senders registered here capture the tag sets at +/// registration time and keep emitting the originals, so payment events diverge from +/// responses (and the replacement wipes the `pmi` and availability tags from the live +/// paths). Let this function own both tag slots. +/// +/// Pricing advertisement has one deliberate asymmetry, shared with the reference +/// implementation: a priced capability whose `name` is `None` or whose `method` is not +/// one of `tools/call` / `prompts/get` / `resources/read` still prices matching +/// requests but produces no `cap` tag, so it is billable without being advertised. +/// +/// # Configuration warnings +/// +/// Configuration is not validated, matching the reference implementation: empty +/// processor or priced-capability lists register anyway and surface failures at +/// request time. Two configurations log a warning here instead of failing. A +/// `payment_ttl` above the transport's session timeout warns because the paying +/// client's session can expire before the payment resolves, which costs the client the +/// acceptance notification even though the paid result still delivers from the +/// captured route snapshot. Priced capabilities with no processors warn because every +/// priced request will fail processor selection at request time and be dropped. +/// +/// # Rate limiting +/// +/// Every successful payment offer, in either lifecycle, spawns one detached +/// verification task, unbounded across identities and unauthenticated peers. +/// `max_pending_payments` and the authorization store's entry caps bound state, not +/// tasks. CEP-8 explicitly leaves rate limiting and abuse prevention to +/// implementations and sanctions only discretionary eviction, so deployments that +/// cannot trust their peer set should bound intake upstream: `allowed_public_keys`, +/// relay-side policy, or an external limiter. +/// +/// # State lifetime +/// +/// All payment state is in-memory and single-process: the pending-payment dedup, the +/// authorization store's pending and granted entries, and the payment route snapshots +/// are all forgotten on restart, and an LRU-evicted authorization re-invoices its +/// payer on retry. +/// +/// # Errors +/// +/// Fails without mutating the transport when the transport is already started, or +/// when a payment-interaction policy is already recorded (a prior registration, or a +/// hand-set policy via +/// [`set_supported_payment_interaction`](NostrServerTransport::set_supported_payment_interaction); +/// this function registers once and owns the policy). +pub fn with_server_payments( + transport: &mut NostrServerTransport, + options: ServerPaymentsOptions, +) -> crate::Result<()> { + // Both guards run before any mutation, so a failed call leaves the transport + // exactly as it was and the caller can correct and retry. + if transport.is_started() { + return Err(crate::Error::Other( + "with_server_payments must be called before start()".to_string(), + )); + } + if transport.supported_payment_interaction().is_some() { + return Err(crate::Error::Other( + "a payment interaction policy is already recorded on this transport; \ + with_server_payments registers once and owns the policy" + .to_string(), + )); + } + + // Build the PMI-to-processor map once and share it across both middlewares, so + // the duplicate-PMI warning fires once per duplicate occurrence in total. + let shared_processors = Arc::new(build_processors_by_pmi(&options.processors)); + + // Log-only configuration warnings; neither changes behavior or the wire. + let session_timeout = transport.session_timeout(); + if options.payment_ttl > session_timeout { + tracing::warn!( + target: LOG_TARGET, + payment_ttl = ?options.payment_ttl, + session_timeout = ?session_timeout, + "payment_ttl exceeds the transport's session timeout: a paying client's \ + session can expire before its payment resolves, costing the client the \ + acceptance notification (the paid result itself still delivers from the \ + captured route snapshot)" + ); + } + if options.processors.is_empty() && !options.priced_capabilities.is_empty() { + tracing::warn!( + target: LOG_TARGET, + priced_capabilities = options.priced_capabilities.len(), + "priced capabilities are configured but no payment processors are: every \ + priced request will fail processor selection at request time and be dropped" + ); + } + + let policy = options.payment_interaction; + let supports_explicit_gating = policy == PaymentInteractionPolicy::Optional; + + transport.set_announcement_extra_tags(compose_payment_extra_tags( + &options, + supports_explicit_gating, + )); + transport.set_announcement_pricing_tags(cap_tags_from_priced_capabilities( + &options.priced_capabilities, + )); + transport.set_supported_payment_interaction(policy); + + // Both senders capture the announcement tag sets at the moment they are built, so + // they are built only after both tag setters above have run. A sender built + // earlier would ship an empty discovery replay on every payment event for the + // life of the process. + let notification_sender = transport.payment_notification_sender(options.payment_ttl); + transport.add_inbound_middleware(create_server_payments_middleware( + ServerPaymentsMiddlewareParams { + options: options.clone(), + sender: notification_sender, + processors_by_pmi: Some(Arc::clone(&shared_processors)), + }, + )); + + // The transparent middleware self-gates on the per-session effective mode, so it + // is safe to register the explicit-gating middleware alongside it: each request is + // routed to exactly one lifecycle based on the negotiated mode. + if supports_explicit_gating { + let targeted_sender = transport.targeted_response_sender(); + transport.add_inbound_middleware(create_explicit_gating_middleware( + ExplicitGatingMiddlewareParams { + options, + sender: targeted_sender, + authorization_store: AuthorizationStore::new(), + processors_by_pmi: Some(shared_processors), + }, + )); + } + + Ok(()) +} + +/// The announcement extra-tag segment: one `pmi` tag per processor in registration +/// order (duplicates preserved, mirroring the reference implementation's wire), and, +/// when explicit gating is supported (the [`PaymentInteractionPolicy::Optional`] +/// policy), a single `payment_interaction=explicit_gating` availability tag pushed +/// last. A `Transparent` policy advertises no `payment_interaction` tag at all. The +/// caller passes the same `supports_explicit_gating` bool that drives the policy +/// recording and the conditional gating registration, so the three consumers cannot +/// drift apart. +fn compose_payment_extra_tags( + options: &ServerPaymentsOptions, + supports_explicit_gating: bool, +) -> Vec { + let mut tags: Vec = options + .processors + .iter() + .map(|processor| pmi_tag(processor.pmi())) + .collect(); + if supports_explicit_gating { + tags.push(payment_interaction_tag( + PaymentInteractionMode::ExplicitGating, + )); + } + tags +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::core::constants::SERVER_ANNOUNCEMENT_KIND; + use crate::core::types::ServerInfo; + use crate::payments::errors::PaymentError; + use crate::payments::traits::PaymentProcessor; + use crate::payments::types::{ + PaymentProcessorCreateParams, PaymentProcessorVerifyParams, PaymentRequiredParams, + PricedCapability, VerifyOutcome, + }; + use crate::relay::mock::MockRelayPool; + use crate::relay::RelayPoolTrait; + use crate::transport::server::NostrServerTransportConfig; + use async_trait::async_trait; + use std::io::Write; + use std::sync::Mutex as StdMutex; + use std::time::Duration; + use tracing_subscriber::fmt::MakeWriter; + + /// A minimal local processor double, so these tests run in every feature + /// configuration (the deterministic fakes are `test-utils`-gated). + struct StubProcessor { + pmi: String, + } + + impl StubProcessor { + fn arc(pmi: &str) -> Arc { + Arc::new(Self { + pmi: pmi.to_string(), + }) + } + } + + #[async_trait] + impl PaymentProcessor for StubProcessor { + fn pmi(&self) -> &str { + &self.pmi + } + + async fn create_payment_required( + &self, + params: PaymentProcessorCreateParams, + ) -> Result { + Ok(PaymentRequiredParams { + amount: params.amount, + pay_req: format!("invoice-{}", params.request_event_id), + pmi: self.pmi.clone(), + description: params.description, + ttl: None, + meta: None, + }) + } + + async fn verify_payment( + &self, + _params: PaymentProcessorVerifyParams, + ) -> Result { + Ok(VerifyOutcome::default()) + } + } + + fn priced_tool(name: &str, amount: i64) -> PricedCapability { + PricedCapability { + method: "tools/call".to_string(), + name: Some(name.to_string()), + amount, + max_amount: None, + currency_unit: "sats".to_string(), + description: None, + } + } + + fn options( + pmis: &[&str], + priced: Vec, + policy: PaymentInteractionPolicy, + ) -> ServerPaymentsOptions { + ServerPaymentsOptions::new(pmis.iter().map(|p| StubProcessor::arc(p)).collect(), priced) + .with_payment_interaction(policy) + } + + async fn transport() -> NostrServerTransport { + transport_with(NostrServerTransportConfig::default()).await + } + + async fn transport_with(config: NostrServerTransportConfig) -> NostrServerTransport { + NostrServerTransport::with_relay_pool( + config, + Arc::new(MockRelayPool::new()) as Arc, + ) + .await + .expect("server transport") + } + + fn tag_tuples(tags: &[Tag]) -> Vec> { + tags.iter().map(|t| t.clone().to_vec()).collect() + } + + /// A thread-local tracing capture, so warn-count asserts do not race parallel tests. + #[derive(Clone, Default)] + struct Capture(Arc>>); + + impl Capture { + fn contents(&self) -> String { + String::from_utf8_lossy(&self.0.lock().unwrap()).into_owned() + } + } + + impl Write for Capture { + fn write(&mut self, buf: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(buf); + Ok(buf.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'a> MakeWriter<'a> for Capture { + type Writer = Capture; + fn make_writer(&'a self) -> Capture { + self.clone() + } + } + + fn warn_capture() -> (Capture, tracing::subscriber::DefaultGuard) { + let capture = Capture::default(); + let subscriber = tracing_subscriber::fmt() + .with_max_level(tracing::Level::WARN) + .with_writer(capture.clone()) + .finish(); + let guard = tracing::subscriber::set_default(subscriber); + (capture, guard) + } + + // ── tag composition ───────────────────────────────────────────── + + /// Compose with the same policy-to-bool binding the entry point uses. + fn compose(options: &ServerPaymentsOptions) -> Vec { + compose_payment_extra_tags( + options, + options.payment_interaction == PaymentInteractionPolicy::Optional, + ) + } + + /// The composed extra segment, asserted as an ordered sequence per policy: the + /// availability tag exists only under `Optional` and is pushed last; duplicates + /// and registration order are preserved; empty processors yield no `pmi` tags. + #[test] + fn composition_per_policy() { + let optional = options( + &["pmi:A", "pmi:B"], + vec![], + PaymentInteractionPolicy::Optional, + ); + assert_eq!( + tag_tuples(&compose(&optional)), + vec![ + vec!["pmi".to_string(), "pmi:A".to_string()], + vec!["pmi".to_string(), "pmi:B".to_string()], + vec![ + "payment_interaction".to_string(), + "explicit_gating".to_string() + ], + ], + "the availability tag must be present exactly once, ordered last" + ); + + let transparent = options( + &["pmi:A", "pmi:B"], + vec![], + PaymentInteractionPolicy::Transparent, + ); + assert_eq!( + tag_tuples(&compose(&transparent)), + vec![ + vec!["pmi".to_string(), "pmi:A".to_string()], + vec!["pmi".to_string(), "pmi:B".to_string()], + ], + "a transparent-only policy advertises no payment_interaction tag" + ); + + let duplicated = options( + &["pmi:A", "pmi:A"], + vec![], + PaymentInteractionPolicy::Transparent, + ); + assert_eq!( + tag_tuples(&compose(&duplicated)), + vec![ + vec!["pmi".to_string(), "pmi:A".to_string()], + vec!["pmi".to_string(), "pmi:A".to_string()], + ], + "duplicate processors emit duplicate pmi tags (reference parity)" + ); + + let empty = options(&[], vec![], PaymentInteractionPolicy::Optional); + assert_eq!( + tag_tuples(&compose(&empty)), + vec![vec![ + "payment_interaction".to_string(), + "explicit_gating".to_string() + ]], + "no processors yield no pmi tags" + ); + } + + // ── guards ────────────────────────────────────────────────────── + + /// A post-start registration is refused with a clean error before any mutation: + /// the policy stays unrecorded and the announcement carries no payment tags. + #[tokio::test] + async fn post_start_call_errors_without_mutating() { + let pool = Arc::new(MockRelayPool::new()); + let mut server = NostrServerTransport::with_relay_pool( + NostrServerTransportConfig::default().with_server_info(ServerInfo { + name: Some("guard-test".to_string()), + ..Default::default() + }), + Arc::clone(&pool) as Arc, + ) + .await + .expect("server transport"); + server.start().await.expect("start"); + + let result = with_server_payments( + &mut server, + options( + &["pmi:A"], + vec![priced_tool("paid-tool", 21)], + PaymentInteractionPolicy::Optional, + ), + ); + let error = result.expect_err("a post-start registration must be refused"); + assert!( + error.to_string().contains("before start()"), + "unexpected error: {error}" + ); + + assert_eq!( + server.supported_payment_interaction(), + None, + "the refused call must not record a policy" + ); + server.announce().await.expect("announce"); + let announcement = pool + .stored_events() + .await + .into_iter() + .find(|e| e.kind.as_u16() == SERVER_ANNOUNCEMENT_KIND) + .expect("announcement published"); + let announcement_tags: Vec = announcement.tags.iter().cloned().collect(); + for tag in tag_tuples(&announcement_tags) { + assert!( + !matches!( + tag.first().map(String::as_str), + Some("pmi") | Some("cap") | Some("payment_interaction") + ), + "the refused call must not set announcement tags, found {tag:?}" + ); + } + server.close().await.expect("close"); + } + + /// A second registration is refused and the first registration's policy stands. + #[tokio::test] + async fn second_call_errors_and_first_registration_stands() { + let mut server = transport().await; + with_server_payments( + &mut server, + options( + &["pmi:A"], + vec![priced_tool("paid-tool", 21)], + PaymentInteractionPolicy::Optional, + ), + ) + .expect("the first registration succeeds"); + + let error = with_server_payments( + &mut server, + options( + &["pmi:B"], + vec![priced_tool("other-tool", 5)], + PaymentInteractionPolicy::Transparent, + ), + ) + .expect_err("a second registration must be refused"); + assert!( + error + .to_string() + .contains("a payment interaction policy is already recorded"), + "unexpected error: {error}" + ); + + assert_eq!( + server.supported_payment_interaction(), + Some(PaymentInteractionPolicy::Optional), + "the refused second call must not overwrite the first policy" + ); + } + + // ── warns ─────────────────────────────────────────────────────── + + /// The duplicate-PMI warning fires once per duplicate occurrence in total, because + /// the processor map is built once and shared across both registered middlewares. + #[tokio::test] + async fn duplicate_pmi_warns_once_across_both_middlewares() { + let (capture, _guard) = warn_capture(); + let mut server = transport().await; + with_server_payments( + &mut server, + options( + &["pmi:A", "pmi:A"], + vec![priced_tool("paid-tool", 21)], + PaymentInteractionPolicy::Optional, + ), + ) + .expect("registers"); + + let logs = capture.contents(); + assert_eq!( + logs.matches("duplicate PMI processor registered").count(), + 1, + "the shared map must be built exactly once, logs:\n{logs}" + ); + assert!( + !logs.contains("no payment processors"), + "no other configuration warning applies here, logs:\n{logs}" + ); + } + + /// The TTL warning fires exactly when `payment_ttl` strictly exceeds the session + /// timeout, naming both durations; the equal-defaults configuration stays silent. + #[tokio::test] + async fn ttl_warn_fires_only_above_session_timeout() { + let above = { + let (capture, _guard) = warn_capture(); + let mut server = transport_with( + NostrServerTransportConfig::default() + .with_session_timeout(Duration::from_secs(300)), + ) + .await; + with_server_payments( + &mut server, + options( + &["pmi:A"], + vec![priced_tool("paid-tool", 21)], + PaymentInteractionPolicy::Optional, + ) + .with_payment_ttl(Duration::from_secs(301)), + ) + .expect("registers"); + capture.contents() + }; + assert_eq!( + above + .matches("exceeds the transport's session timeout") + .count(), + 1, + "one warning per registration, logs:\n{above}" + ); + assert!( + above.contains("301") && above.contains("300"), + "the warning must name both durations, logs:\n{above}" + ); + + let equal = { + let (capture, _guard) = warn_capture(); + let mut server = transport_with( + NostrServerTransportConfig::default() + .with_session_timeout(Duration::from_secs(300)), + ) + .await; + with_server_payments( + &mut server, + options( + &["pmi:A"], + vec![priced_tool("paid-tool", 21)], + PaymentInteractionPolicy::Optional, + ) + .with_payment_ttl(Duration::from_secs(300)), + ) + .expect("registers"); + capture.contents() + }; + assert!( + !equal.contains("exceeds the transport's session timeout"), + "an equal (default) configuration must stay silent, logs:\n{equal}" + ); + } + + /// Priced capabilities with no processors warn loudly at registration time but + /// still register, preserving the reference implementation's no-validation posture. + #[tokio::test] + async fn empty_processors_with_priced_caps_warns_and_still_registers() { + let (capture, _guard) = warn_capture(); + let mut server = transport().await; + let result = with_server_payments( + &mut server, + options( + &[], + vec![priced_tool("paid-tool", 21)], + PaymentInteractionPolicy::Optional, + ), + ); + assert!(result.is_ok(), "registration must not validate"); + + let logs = capture.contents(); + assert_eq!( + logs.matches("no payment processors are").count(), + 1, + "the misconfiguration must be loud at registration time, logs:\n{logs}" + ); + assert_eq!( + server.supported_payment_interaction(), + Some(PaymentInteractionPolicy::Optional), + "the policy is recorded despite the warning" + ); + } +} diff --git a/src/transport/server/mod.rs b/src/transport/server/mod.rs index 2a7ddb5..a55cb17 100644 --- a/src/transport/server/mod.rs +++ b/src/transport/server/mod.rs @@ -868,6 +868,22 @@ impl NostrServerTransport { self.supported_payment_interaction = Some(policy); } + /// Whether [`start`](Self::start) has run (the same predicate the pre-start + /// `debug_assert!`s above use). + pub(crate) fn is_started(&self) -> bool { + !self.task_handles.is_empty() + } + + /// The configured session timeout. + pub(crate) fn session_timeout(&self) -> Duration { + self.config.session_timeout + } + + /// The recorded payment-interaction policy, or `None` when payments were never configured. + pub(crate) fn supported_payment_interaction(&self) -> Option { + self.supported_payment_interaction + } + /// Start listening for incoming requests. pub async fn start(&mut self) -> Result<()> { self.base @@ -7425,6 +7441,212 @@ mod tests { transport.close().await.expect("close"); } + /// A minimal local processor double for the registration wiring tests below, so + /// they run in every feature configuration (the deterministic fakes are + /// `test-utils`-gated). + struct RegistrationStubProcessor { + pmi: String, + } + + #[async_trait::async_trait] + impl crate::payments::PaymentProcessor for RegistrationStubProcessor { + fn pmi(&self) -> &str { + &self.pmi + } + + async fn create_payment_required( + &self, + params: crate::payments::types::PaymentProcessorCreateParams, + ) -> std::result::Result< + crate::payments::types::PaymentRequiredParams, + crate::payments::PaymentError, + > { + Ok(crate::payments::types::PaymentRequiredParams { + amount: params.amount, + pay_req: format!("invoice-{}", params.request_event_id), + pmi: self.pmi.clone(), + description: params.description, + ttl: None, + meta: None, + }) + } + + async fn verify_payment( + &self, + _params: crate::payments::types::PaymentProcessorVerifyParams, + ) -> std::result::Result + { + Ok(crate::payments::types::VerifyOutcome::default()) + } + } + + fn registration_options(payment_ttl: Duration) -> crate::payments::ServerPaymentsOptions { + crate::payments::ServerPaymentsOptions::new( + vec![Arc::new(RegistrationStubProcessor { + pmi: "stub-pmi".to_string(), + })], + vec![crate::payments::types::PricedCapability { + method: "tools/call".to_string(), + name: Some("paid-tool".to_string()), + amount: 21, + max_amount: None, + currency_unit: "sats".to_string(), + description: None, + }], + ) + .with_payment_ttl(payment_ttl) + } + + /// The payments registration entry point threads the configured `payment_ttl` into + /// the notification sender's snapshot horizon: the snapshot recorded at + /// `payment_required` publication expires at capture time plus the CONFIGURED TTL, + /// not the crate default. The recorded `expires_at` is asserted directly against + /// the configured horizon, bounded by instants taken around the flow, so the test + /// has no timing dependence: a default-stamping wiring (300 s) and a + /// capped-at-default wiring both record a horizon below the lower bound and fail. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn snapshot_expiry_tracks_the_configured_payment_ttl() { + let configured_ttl = Duration::from_secs(1234); // distinctive, above the 300 s default + let (client_pool, server_pool) = MockRelayPool::create_pair(); + let server_pubkey = server_pool.mock_public_key(); + let s_pool = Arc::new(server_pool); + + let mut transport = NostrServerTransport::with_relay_pool( + NostrServerTransportConfig::default().with_encryption_mode(EncryptionMode::Disabled), + Arc::clone(&s_pool) as Arc, + ) + .await + .expect("server transport"); + crate::payments::with_server_payments(&mut transport, registration_options(configured_ttl)) + .expect("register payments"); + + let mut server_rx = transport.take_message_receiver().expect("receiver"); + let before_publish = Instant::now(); + transport.start().await.expect("start"); + tokio::time::sleep(Duration::from_millis(20)).await; + + let client_keys = Keys::generate(); + let request = JsonRpcMessage::Request(JsonRpcRequest { + jsonrpc: "2.0".to_string(), + id: serde_json::json!("snap-ttl-1"), + method: "tools/call".to_string(), + params: Some(serde_json::json!({ "name": "paid-tool" })), + }); + let request_event = crate::core::serializers::mcp_to_nostr_event( + &request, + CTXVM_MESSAGES_KIND, + BaseTransport::create_recipient_tags(&server_pubkey), + ) + .expect("serialize the priced call") + .sign_with_keys(&client_keys) + .expect("sign the priced call"); + let request_event_id = request_event.id.to_hex(); + client_pool + .publish_event(&request_event) + .await + .expect("publish the priced call"); + + // Wait for the middleware to publish payment_required (which records the snapshot). + let deadline = tokio::time::Instant::now() + Duration::from_secs(3); + loop { + let published = s_pool + .stored_events() + .await + .into_iter() + .any(|e| e.pubkey == server_pubkey && e.content.contains("payment_required")); + if published { + break; + } + assert!( + tokio::time::Instant::now() < deadline, + "payment_required must be published for the priced call" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } + let after_publish = Instant::now(); + + { + let cache = NostrServerTransport::lock_payment_route_snapshots( + &transport.payment_route_snapshots, + ); + let expires_at = cache + .peek(&request_event_id) + .expect("the publish must have recorded a route snapshot") + .expires_at; + assert!( + expires_at >= before_publish + configured_ttl, + "the snapshot horizon must be the CONFIGURED payment_ttl, not the crate \ + default (or a default-capped value)" + ); + assert!( + expires_at <= after_publish + configured_ttl, + "the snapshot horizon must not exceed capture time plus the configured TTL" + ); + } + + // Drain whatever the settled payment forwarded, then shut down. + let _ = server_rx.try_recv(); + transport.close().await.expect("close"); + } + + /// The payments registration entry point routes each composed tag segment to its + /// own announcement slot: the `cap` tags land in the pricing slot and the + /// `pmi`/availability tags in the extra-common slot. The slots are asserted + /// directly because they are what the pricing-only list kinds and the common-only + /// first-response replay consume, where a swap has real wire consequences; the + /// kind 11316 announcement concatenates both slots and only its segment ORDER + /// betrays a swap. + #[tokio::test] + async fn tags_land_in_their_slots() { + let pool: Arc = Arc::new(MockRelayPool::new()); + let mut transport = + NostrServerTransport::with_relay_pool(NostrServerTransportConfig::default(), pool) + .await + .expect("server transport"); + crate::payments::with_server_payments( + &mut transport, + registration_options(Duration::from_secs(300)), + ) + .expect("register payments"); + + let tuples = |tags: &[Tag]| -> Vec> { + tags.iter().map(|t| t.clone().to_vec()).collect() + }; + + assert_eq!( + tuples(transport.announcement_manager.get_pricing_tags()), + vec![vec![ + "cap".to_string(), + "tool:paid-tool".to_string(), + "21".to_string(), + "sats".to_string() + ]], + "the pricing slot must hold exactly the cap segment" + ); + + let payment_tuples: Vec> = + tuples(&transport.announcement_manager.get_common_tags()) + .into_iter() + .filter(|t| { + matches!( + t.first().map(String::as_str), + Some("pmi") | Some("payment_interaction") | Some("cap") + ) + }) + .collect(); + assert_eq!( + payment_tuples, + vec![ + vec!["pmi".to_string(), "stub-pmi".to_string()], + vec![ + "payment_interaction".to_string(), + "explicit_gating".to_string() + ], + ], + "the extra-common slot must hold the pmi and availability tags, and no cap tag" + ); + } + // ── Targeted response sender ───────────────────────────────────────────── /// A server transport wired to `pool`, with the CEP-8 knobs the targeted-send tests need. From 0428340e5e932be2854cfd30df4ac5f624efe6d7 Mon Sep 17 00:00:00 2001 From: Harsh Date: Sun, 30 Aug 2026 05:50:45 +0530 Subject: [PATCH 2/3] feat(gateway): register payments from GatewayConfig payment_options at both transport-construction sites --- src/gateway/mod.rs | 50 +++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 45 insertions(+), 5 deletions(-) diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 5027d5b..0a4a47b 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -5,6 +5,7 @@ use crate::core::error::{Error, Result}; use crate::core::types::JsonRpcMessage; +use crate::payments::{with_server_payments, ServerPaymentsOptions}; use crate::transport::server::{IncomingRequest, NostrServerTransport, NostrServerTransportConfig}; /// Configuration for the gateway. @@ -13,12 +14,26 @@ use crate::transport::server::{IncomingRequest, NostrServerTransport, NostrServe pub struct GatewayConfig { /// Nostr server transport configuration. pub nostr_config: NostrServerTransportConfig, + /// Optional CEP-8 payments configuration. When set, the gateway registers + /// payments on its transport via [`with_server_payments`] between + /// constructing the transport and starting it. + pub payment_options: Option, } impl GatewayConfig { - /// Create a new gateway configuration. + /// Create a new gateway configuration (payments stay opt-in via + /// [`with_payment_options`](Self::with_payment_options)). pub fn new(nostr_config: NostrServerTransportConfig) -> Self { - Self { nostr_config } + Self { + nostr_config, + payment_options: None, + } + } + + /// Set the CEP-8 payments configuration. + pub fn with_payment_options(mut self, payment_options: ServerPaymentsOptions) -> Self { + self.payment_options = Some(payment_options); + self } } @@ -37,7 +52,10 @@ impl NostrMCPGateway { where T: nostr_sdk::prelude::IntoNostrSigner, { - let transport = NostrServerTransport::new(signer, config.nostr_config).await?; + let mut transport = NostrServerTransport::new(signer, config.nostr_config).await?; + if let Some(payment_options) = config.payment_options { + with_server_payments(&mut transport, payment_options)?; + } Ok(Self { transport, @@ -74,6 +92,11 @@ impl NostrMCPGateway { } /// Stop the gateway. + /// + /// The underlying transport is not restartable after close, so a stopped + /// gateway cannot be started again (a later `start` panics at the transport's + /// message-channel expect); payments registration neither causes nor changes + /// this. pub async fn stop(&mut self) -> Result<()> { if !self.is_running { return Ok(()); @@ -107,7 +130,10 @@ impl NostrMCPGateway { use crate::NostrServerTransport; use rmcp::ServiceExt; - let transport = NostrServerTransport::new(signer, config.nostr_config).await?; + let mut transport = NostrServerTransport::new(signer, config.nostr_config).await?; + if let Some(payment_options) = config.payment_options { + with_server_payments(&mut transport, payment_options)?; + } handler .serve(transport) .await @@ -148,12 +174,16 @@ mod tests { open_stream: Default::default(), }; - let config = GatewayConfig { nostr_config }; + let config = GatewayConfig { + nostr_config, + payment_options: None, + }; assert_eq!( config.nostr_config.relay_urls, vec!["wss://relay.example.com"] ); + assert!(config.payment_options.is_none()); assert_eq!( config.nostr_config.encryption_mode, EncryptionMode::Required @@ -177,6 +207,7 @@ mod tests { fn test_gateway_config_with_defaults() { let config = GatewayConfig { nostr_config: NostrServerTransportConfig::default(), + payment_options: None, }; assert_eq!( config.nostr_config.encryption_mode, @@ -184,4 +215,13 @@ mod tests { ); assert!(!config.nostr_config.is_announced_server); } + + #[test] + fn test_gateway_config_builder_sets_payment_options() { + let config = + GatewayConfig::new(NostrServerTransportConfig::default()).with_payment_options( + crate::payments::ServerPaymentsOptions::new(Vec::new(), Vec::new()), + ); + assert!(config.payment_options.is_some()); + } } From 88b03e4df7df501c9d442f65127a56c2440a2f95 Mon Sep 17 00:00:00 2001 From: Harsh Date: Sun, 30 Aug 2026 05:50:50 +0530 Subject: [PATCH 3/3] test(payments): drive both payment lifecycles through the production registration entry point --- CHANGELOG.md | 15 + Cargo.toml | 4 + tests/conformance_cep8_wire_format.rs | 62 ++ tests/payments_server_e2e.rs | 990 ++++++++++++++++++++++++++ 4 files changed, 1071 insertions(+) create mode 100644 tests/payments_server_e2e.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index be778a8..0a01443 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -70,6 +70,21 @@ the two checks can never mint a second invoice against an unclaimed paid grant. Registration arrives with the payments configuration entry point, alongside the transparent middleware's. + - The server payments registration entry point: + `contextvm_sdk::payments::with_server_payments(&mut transport, options)` composes + the whole server-side payment stack in one call, before `start()`. It advertises + the configured payment methods and prices on announcements (`pmi` and `cap` tags, + plus a `payment_interaction=explicit_gating` availability tag when the policy is + `Optional`), records the negotiation policy, and registers the transparent payment + middleware plus, under `Optional`, the explicit-gating middleware with a fresh + authorization store, with the payment senders built after the tags so their + captured discovery replay is complete. It refuses (with an error, before touching + any state) the two calls that would silently lose money: registration after + `start()`, which would advertise payments while gating nothing, and a second + registration, which would charge every priced request twice. The gateway joins in: + `GatewayConfig` gains `payment_options` (with a `with_payment_options` builder), + consumed when the gateway builds its transport in both `NostrMCPGateway::new` and + `serve_handler`. ### Fixed diff --git a/Cargo.toml b/Cargo.toml index f2a4266..2d6cdf4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -105,6 +105,10 @@ required-features = ["test-utils"] name = "payments_explicit_gating_e2e" required-features = ["test-utils"] +[[test]] +name = "payments_server_e2e" +required-features = ["test-utils"] + [[test]] name = "rmcp_handshake_survival" required-features = ["rmcp", "test-utils"] diff --git a/tests/conformance_cep8_wire_format.rs b/tests/conformance_cep8_wire_format.rs index 97528c0..32d4509 100644 --- a/tests/conformance_cep8_wire_format.rs +++ b/tests/conformance_cep8_wire_format.rs @@ -437,3 +437,65 @@ fn cep8_payment_required_error_data_shape() { }) ); } + +// ── announcement segment composition ───────────────────────────────────────── + +/// The composed announcement segments a payments-configured server advertises, built +/// from the public builders exactly as the registration entry point composes them. +/// +/// The per-tag formats are pinned above; the delta pinned here is the COMPOSITION, +/// asserted as ordered sequences: the extra segment is one `pmi` tag per processor in +/// registration order (duplicates preserved), followed by exactly one +/// `payment_interaction=explicit_gating` availability tag ordered LAST when explicit +/// gating is available as an opt-in mode; a transparent-only server advertises no +/// `payment_interaction` tag at all. The pricing segment is the `cap` tags in +/// declaration order. +#[test] +fn cep8_announcement_segments_compose_per_policy() { + let pmis = ["pmi:A".to_string(), "pmi:B".to_string()]; + let caps = vec![ + priced("tools/call", "add", 1, None, "sats"), + priced("prompts/get", "summarize", 5, Some(20), "sats"), + ]; + + // Explicit gating available (the permissive default policy): availability tag last. + let mut extra_optional: Vec = pmis.iter().map(|p| pmi_tag(p)).collect(); + extra_optional.push(payment_interaction_tag( + PaymentInteractionMode::ExplicitGating, + )); + assert_eq!( + wire_all(&extra_optional), + vec![ + vec!["pmi", "pmi:A"], + vec!["pmi", "pmi:B"], + vec!["payment_interaction", "explicit_gating"], + ], + "the availability tag must be present exactly once and ordered last" + ); + + // Transparent-only: the pmi tags alone, no payment_interaction tag. + let extra_transparent: Vec = pmis.iter().map(|p| pmi_tag(p)).collect(); + assert_eq!( + wire_all(&extra_transparent), + vec![vec!["pmi", "pmi:A"], vec!["pmi", "pmi:B"]], + "a transparent-only server advertises no payment_interaction tag" + ); + + // Duplicate processors emit duplicate pmi tags (ts-sdk parity; the tag list maps + // the raw processor list, not the deduplicated selection map). + let duplicated = ["pmi:A".to_string(), "pmi:A".to_string()]; + let extra_duplicated: Vec = duplicated.iter().map(|p| pmi_tag(p)).collect(); + assert_eq!( + wire_all(&extra_duplicated), + vec![vec!["pmi", "pmi:A"], vec!["pmi", "pmi:A"]], + ); + + // The pricing segment, in declaration order, range prices hyphenated. + assert_eq!( + wire_all(&cap_tags_from_priced_capabilities(&caps)), + vec![ + vec!["cap", "tool:add", "1", "sats"], + vec!["cap", "prompt:summarize", "5-20", "sats"], + ], + ); +} diff --git a/tests/payments_server_e2e.rs b/tests/payments_server_e2e.rs new file mode 100644 index 0000000..13ba837 --- /dev/null +++ b/tests/payments_server_e2e.rs @@ -0,0 +1,990 @@ +//! CEP-8 server payments through the production registration entry point, end to end +//! over `MockRelayPool`. +//! +//! Every server in this suite is configured via `with_server_payments`, never by +//! hand-registering middlewares: the suite exists to prove the production wiring (tag +//! composition, policy recording, sender capture order, conditional gating +//! registration, the snapshot TTL threading) drives both payment lifecycles +//! correctly. The factory-level suites keep their hand-registration; this one must +//! not. +//! +//! Clock discipline: the authorization store, the stale-route sweep and session expiry +//! all run on `std::time::Instant`, which paused tokio time does not advance, so every +//! test here runs on the real clock with tiny configured timeouts. A test that mixes +//! the paused tokio clock with a store-TTL assertion proves nothing while green, which +//! is why none does. + +use std::sync::Arc; +use std::time::Duration; + +use contextvm_sdk::core::constants::CTXVM_MESSAGES_KIND; +use contextvm_sdk::core::serializers; +use contextvm_sdk::core::types::EncryptionMode; +use contextvm_sdk::payments::fakes::{FakePaymentProcessor, FakePaymentProcessorOptions}; +use contextvm_sdk::payments::types::PricedCapability; +use contextvm_sdk::payments::{ + with_server_payments, PaymentInteractionPolicy, ServerPaymentsOptions, +}; +use contextvm_sdk::relay::mock::MockRelayPool; +use contextvm_sdk::transport::base::BaseTransport; +use contextvm_sdk::transport::client::{NostrClientTransport, NostrClientTransportConfig}; +use contextvm_sdk::transport::server::{ + IncomingRequest, NostrServerTransport, NostrServerTransportConfig, +}; +use contextvm_sdk::{ + JsonRpcMessage, JsonRpcRequest, JsonRpcResponse, PaymentInteractionMode, RelayPoolTrait, + ServerInfo, +}; +use nostr_sdk::prelude::*; + +fn as_pool(pool: &Arc) -> Arc { + Arc::clone(pool) as Arc +} + +fn paid_call(id: &str) -> JsonRpcMessage { + JsonRpcMessage::Request(JsonRpcRequest { + jsonrpc: "2.0".to_string(), + id: serde_json::json!(id), + method: "tools/call".to_string(), + params: Some(serde_json::json!({ "name": "paid-tool" })), + }) +} + +fn result_response(id: &str) -> JsonRpcMessage { + JsonRpcMessage::Response(JsonRpcResponse { + jsonrpc: "2.0".to_string(), + id: serde_json::json!(id), + result: serde_json::json!({ "content": [] }), + }) +} + +fn priced_tool(amount: i64) -> PricedCapability { + PricedCapability { + method: "tools/call".to_string(), + name: Some("paid-tool".to_string()), + amount, + max_amount: None, + currency_unit: "sats".to_string(), + description: None, + } +} + +fn fake_processor(pmi: &str, verify_delay_ms: u64) -> Arc { + Arc::new(FakePaymentProcessor::with_options( + FakePaymentProcessorOptions { + pmi: pmi.to_string(), + verify_delay_ms, + create_delay_ms: 0, + ttl: None, + }, + )) +} + +/// The standard one-processor, one-priced-tool payments configuration. +fn payments_options(verify_delay_ms: u64) -> ServerPaymentsOptions { + ServerPaymentsOptions::new( + vec![fake_processor("fake", verify_delay_ms)], + vec![priced_tool(21)], + ) +} + +fn all_tags(event: &Event) -> Vec> { + event.tags.iter().map(|t| t.clone().to_vec()).collect() +} + +/// Every stored server-authored ContextVM event whose content contains `needle`. +async fn server_events_containing( + pool: &Arc, + server: PublicKey, + needle: &str, +) -> Vec { + pool.stored_events() + .await + .into_iter() + .filter(|e| { + e.kind == Kind::Custom(CTXVM_MESSAGES_KIND) + && e.pubkey == server + && e.content.contains(needle) + }) + .collect() +} + +/// Poll for one server event containing `needle`, within `deadline`. +async fn wait_for_server_event( + pool: &Arc, + server: PublicKey, + needle: &str, + deadline: Duration, +) -> Event { + let end = tokio::time::Instant::now() + deadline; + loop { + let found = server_events_containing(pool, server, needle).await; + if let Some(event) = found.into_iter().next() { + return event; + } + assert!( + tokio::time::Instant::now() < end, + "no server event containing {needle:?} within {deadline:?}" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +/// Poll `cond` for up to `deadline`. +async fn wait_until(what: &str, deadline: Duration, mut cond: impl AsyncFnMut() -> bool) { + let end = tokio::time::Instant::now() + deadline; + loop { + if cond().await { + return; + } + assert!( + tokio::time::Instant::now() < end, + "condition not reached within {deadline:?}: {what}" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } +} + +/// Every stored client request event whose content contains the quoted `id`. +async fn client_request_events( + pool: &Arc, + client: PublicKey, + id: &str, +) -> Vec { + let needle = format!("\"{id}\""); + pool.stored_events() + .await + .into_iter() + .filter(|e| { + e.kind == Kind::Custom(CTXVM_MESSAGES_KIND) + && e.pubkey == client + && e.content.contains(&needle) + }) + .collect() +} + +struct Fx { + server: NostrServerTransport, + server_rx: tokio::sync::mpsc::UnboundedReceiver, + client: NostrClientTransport, + pool: Arc, + client_pubkey: PublicKey, + server_pubkey: PublicKey, +} + +/// A paired client/server with payments registered through the PRODUCTION entry +/// point before `start()`. The client's mode and PMIs are per test; a `None` mode is +/// the default (transparent) client. +async fn fixture( + options: ServerPaymentsOptions, + configure_config: impl FnOnce(NostrServerTransportConfig) -> NostrServerTransportConfig, + client_mode: Option, +) -> Fx { + let (client_pool, server_pool) = MockRelayPool::create_pair(); + let server_pubkey = server_pool.mock_public_key(); + let client_pubkey = client_pool.mock_public_key(); + let pool = Arc::new(server_pool); + + let mut server = NostrServerTransport::with_relay_pool( + configure_config( + NostrServerTransportConfig::default().with_encryption_mode(EncryptionMode::Disabled), + ), + as_pool(&pool), + ) + .await + .expect("server transport"); + with_server_payments(&mut server, options).expect("register payments"); + + let mut client_config = NostrClientTransportConfig::default() + .with_relay_urls(vec!["wss://mock.relay".to_string()]) + .with_server_pubkey(server_pubkey.to_hex()) + .with_encryption_mode(EncryptionMode::Disabled) + .with_timeout(Duration::from_secs(30)); + if let Some(mode) = client_mode { + client_config = client_config + .with_payment_interaction(mode) + .with_pmis(vec!["fake".to_string()]); + } + let mut client = NostrClientTransport::with_relay_pool(client_config, Arc::new(client_pool)) + .await + .expect("client transport"); + + let server_rx = server.take_message_receiver().expect("server rx"); + let _client_rx = client.take_message_receiver().expect("client rx"); + server.start().await.expect("server start"); + client.start().await.expect("client start"); + tokio::time::sleep(Duration::from_millis(20)).await; + + Fx { + server, + server_rx, + client, + pool, + client_pubkey, + server_pubkey, + } +} + +/// A raw signed `tools/call` for `paid-tool` from `keys`, with optional extra tags +/// (used to place `payment_interaction` upserts on individual messages). +fn signed_paid_call( + keys: &Keys, + server_pubkey: PublicKey, + id: &str, + extra_tags: Vec, +) -> Event { + let mut tags = BaseTransport::create_recipient_tags(&server_pubkey); + tags.extend(extra_tags); + serializers::mcp_to_nostr_event(&paid_call(id), CTXVM_MESSAGES_KIND, tags) + .expect("serialize the priced call") + .sign_with_keys(keys) + .expect("sign the priced call") +} + +fn pi_tag(value: &str) -> Tag { + Tag::custom( + TagKind::Custom("payment_interaction".into()), + vec![value.to_string()], + ) +} + +// ── the two lifecycles through the entry point ────────────────────────────── + +/// A default-mode client's priced call runs the full transparent lifecycle through +/// the production wiring: the invoice carries the discovery replay INCLUDING the +/// payment discovery tags (proof the notification sender was built after the tag +/// setters ran), settlement is acknowledged, and the paid request reaches the +/// handler. +/// +/// Fixture rule, load-bearing: the priced call MUST be the session's first message +/// and the replay assertion MUST include the `pmi` tags. The one-shot discovery +/// latch burns on the session's first outbound event regardless of content, and the +/// normal response path reads the tag sets live, so a fixture that lets any response +/// precede the priced call would leave a build-senders-before-tags wiring invisible. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn transparent_lifecycle_engages_through_the_entry_point() { + let mut fx = fixture(payments_options(50), |c| c, None).await; + + fx.client.send(&paid_call("pay-t1")).await.expect("send"); + let request_event_id = client_request_events(&fx.pool, fx.client_pubkey, "pay-t1").await[0] + .id + .to_hex(); + + let required = wait_for_server_event( + &fx.pool, + fx.server_pubkey, + "payment_required", + Duration::from_secs(2), + ) + .await; + // The FULL tag list in order: recipient, correlation, then the one-shot discovery + // replay (payment discovery tags first, transport-internal capability last). + assert_eq!( + all_tags(&required), + vec![ + vec!["p".to_string(), fx.client_pubkey.to_hex()], + vec!["e".to_string(), request_event_id.clone()], + vec!["pmi".to_string(), "fake".to_string()], + vec![ + "payment_interaction".to_string(), + "explicit_gating".to_string() + ], + vec!["support_oversized_transfer".to_string()], + ], + "the invoice must replay the full discovery set captured at registration" + ); + + // The request reaches the handler only after the fake settles. + let incoming = tokio::time::timeout(Duration::from_secs(3), fx.server_rx.recv()) + .await + .expect("the paid request must reach the handler") + .expect("channel open"); + assert_eq!(incoming.event_id, request_event_id); + let accepted = server_events_containing(&fx.pool, fx.server_pubkey, "payment_accepted").await; + assert_eq!(accepted.len(), 1, "settlement must be acknowledged"); + + fx.server + .send_response(&request_event_id, result_response("pay-t1")) + .await + .expect("respond"); + let response = wait_for_server_event( + &fx.pool, + fx.server_pubkey, + "\"content\"", + Duration::from_secs(2), + ) + .await; + assert!( + all_tags(&response).contains(&vec!["e".to_string(), request_event_id]), + "the response must correlate to the request" + ); + + fx.server.close().await.expect("close"); +} + +/// A gating client's first priced call is answered `-32042` with the full first +/// response tag surface, the handler stays silent until the payment settles, and the +/// paid retry claims the grant and delivers the result. +/// +/// Fixture rule, load-bearing: the priced call MUST be the session's first message +/// and the offer's tag assertion MUST include the `pmi` tags (see the transparent +/// twin above for why an earlier response would disarm this assert). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn gating_lifecycle_engages_through_the_entry_point() { + let mut fx = fixture( + payments_options(50), + |c| c, + Some(PaymentInteractionMode::ExplicitGating), + ) + .await; + + fx.client.send(&paid_call("pay-g1")).await.expect("send"); + let request_event_id = client_request_events(&fx.pool, fx.client_pubkey, "pay-g1").await[0] + .id + .to_hex(); + + let offer = wait_for_server_event( + &fx.pool, + fx.server_pubkey, + "Payment Required", + Duration::from_secs(2), + ) + .await; + // The FULL tag list in order: recipient, correlation, the discovery replay + // including the payment discovery tags, with the effective-mode disclosure + // deduplicated against the replayed availability advertisement. + assert_eq!( + all_tags(&offer), + vec![ + vec!["p".to_string(), fx.client_pubkey.to_hex()], + vec!["e".to_string(), request_event_id.clone()], + vec!["pmi".to_string(), "fake".to_string()], + vec![ + "payment_interaction".to_string(), + "explicit_gating".to_string() + ], + vec!["support_oversized_transfer".to_string()], + ], + "the offer must compose the full first-response tag surface" + ); + // The payload keeps the client's own inner request id and offers the fake PMI. + assert!( + offer.content.contains("\"id\":\"pay-g1\""), + "the error id must be the original inner request id, got {}", + offer.content + ); + assert!(offer.content.contains("\"code\":-32042")); + assert!(offer.content.contains("\"pmi\":\"fake\"")); + + // The gated request never reaches the handler before payment. + assert!( + fx.server_rx.try_recv().is_err(), + "the gated request must not reach the handler unpaid" + ); + + // The fake settles 50 ms after the offer; the margin is generous (real clock). + tokio::time::sleep(Duration::from_millis(500)).await; + + // The retry is a fresh request event with the same method and params: it claims + // the grant and forwards. + fx.client.send(&paid_call("pay-g1")).await.expect("retry"); + let incoming = tokio::time::timeout(Duration::from_secs(3), fx.server_rx.recv()) + .await + .expect("the paid retry must reach the handler") + .expect("channel open"); + assert_ne!( + incoming.event_id, request_event_id, + "the claiming invocation is a new event" + ); + + fx.server + .send_response(&incoming.event_id, result_response("pay-g1")) + .await + .expect("respond"); + let response = wait_for_server_event( + &fx.pool, + fx.server_pubkey, + "\"content\"", + Duration::from_secs(2), + ) + .await; + assert!( + all_tags(&response).contains(&vec!["e".to_string(), incoming.event_id.clone()]), + "the result rides the claiming invocation's own correlation" + ); + + fx.server.close().await.expect("close"); +} + +// ── the transparent-only policy ───────────────────────────────────────────── + +/// A transparent-only server rejects the gating request with the whole `-32602` +/// object, still gates the same client's subsequent priced call through the +/// transparent lifecycle, and advertises no `payment_interaction` tag. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn transparent_policy_rejects_gating_and_still_gates() { + let mut fx = fixture( + payments_options(50).with_payment_interaction(PaymentInteractionPolicy::Transparent), + |c| c.with_server_info(ServerInfo::default().with_name("transparent-only")), + Some(PaymentInteractionMode::ExplicitGating), + ) + .await; + + // Request 1: the mode request draws the whole -32602 object. + fx.client.send(&paid_call("rej-1")).await.expect("send"); + let rejection = wait_for_server_event( + &fx.pool, + fx.server_pubkey, + "Unsupported payment_interaction", + Duration::from_secs(2), + ) + .await; + let payload: serde_json::Value = + serde_json::from_str(&rejection.content).expect("rejection parses"); + assert_eq!( + payload, + serde_json::json!({ + "jsonrpc": "2.0", + "id": "rej-1", + "error": { + "code": -32602, + "message": "Unsupported payment_interaction mode: explicit_gating", + "data": { + "requested": "explicit_gating", + "supported": ["transparent"], + }, + }, + }), + "the whole rejection object must match the negotiation wire shape" + ); + assert!( + fx.server_rx.try_recv().is_err(), + "the rejected request must not reach the handler" + ); + + // Request 2: the client's one-shot mode latch is spent, so this runs under the + // default transparent mode, and the priced call is still gated. + fx.client.send(&paid_call("rej-2")).await.expect("send"); + wait_for_server_event( + &fx.pool, + fx.server_pubkey, + "payment_required", + Duration::from_secs(2), + ) + .await; + let incoming = tokio::time::timeout(Duration::from_secs(3), fx.server_rx.recv()) + .await + .expect("the paid transparent request must reach the handler") + .expect("channel open"); + fx.server + .send_response(&incoming.event_id, result_response("rej-2")) + .await + .expect("respond"); + + // The announcement advertises the PMIs and prices but never a mode. + fx.server.announce().await.expect("announce"); + let announcement = fx + .pool + .stored_events() + .await + .into_iter() + .find(|e| e.kind == Kind::Custom(11316)) + .expect("announcement"); + assert!( + !all_tags(&announcement) + .iter() + .any(|t| t.first().map(String::as_str) == Some("payment_interaction")), + "a transparent-only server must not advertise a payment_interaction tag" + ); + + fx.server.close().await.expect("close"); +} + +// ── the oversized re-inject dispatch site ─────────────────────────────────── + +/// A CEP-22 reassembled priced call in a gating session is gated on the re-inject +/// dispatch site: the offer correlates to the end frame's event and takes the first +/// configured processor (the re-injected context presents no client PMIs; they rode +/// the start frame), and the paid oversized retry claims and round-trips. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn oversized_reassembled_priced_call_is_gated() { + let (client_pool, server_pool) = MockRelayPool::create_pair(); + let server_pubkey = server_pool.mock_public_key(); + let pool = Arc::new(server_pool); + + let mut server = NostrServerTransport::with_relay_pool( + NostrServerTransportConfig::default().with_encryption_mode(EncryptionMode::Disabled), + as_pool(&pool), + ) + .await + .expect("server transport"); + with_server_payments(&mut server, payments_options(50)).expect("register payments"); + let mut server_rx = server.take_message_receiver().expect("rx"); + server.start().await.expect("start"); + + let mut client = NostrClientTransport::with_relay_pool( + NostrClientTransportConfig::default() + .with_relay_urls(vec!["wss://mock.relay".to_string()]) + .with_server_pubkey(server_pubkey.to_hex()) + .with_encryption_mode(EncryptionMode::Disabled) + .with_payment_interaction(PaymentInteractionMode::ExplicitGating) + .with_pmis(vec!["fake".to_string()]) + .with_timeout(Duration::from_secs(30)), + Arc::new(client_pool), + ) + .await + .expect("client transport"); + client.start().await.expect("client start"); + tokio::time::sleep(Duration::from_millis(20)).await; + + // An oversized priced request as the FIRST send, with a progressToken (nothing + // fragments without one). + let blob = "x".repeat(200_000); + let oversized = JsonRpcMessage::Request(JsonRpcRequest { + jsonrpc: "2.0".to_string(), + id: serde_json::json!("big-1"), + method: "tools/call".to_string(), + params: Some(serde_json::json!({ + "name": "paid-tool", + "arguments": { "blob": blob }, + "_meta": { "progressToken": "tok-big" }, + })), + }); + client.send(&oversized).await.expect("oversized send"); + + let offer = wait_for_server_event( + &pool, + server_pubkey, + "Payment Required", + Duration::from_secs(5), + ) + .await; + assert!(offer.content.contains("\"code\":-32042")); + assert!( + offer.content.contains("\"pmi\":\"fake\""), + "with no PMIs on the re-injected context the offer takes the first processor" + ); + assert!( + server_rx.try_recv().is_err(), + "the gated reassembled request must not reach the handler unpaid" + ); + + // The fake settles 50 ms after the offer; then the oversized retry claims. + tokio::time::sleep(Duration::from_millis(500)).await; + client.send(&oversized).await.expect("oversized retry"); + let incoming = tokio::time::timeout(Duration::from_secs(5), server_rx.recv()) + .await + .expect("the paid reassembled retry must reach the handler") + .expect("channel open"); + server + .send_response(&incoming.event_id, result_response("big-1")) + .await + .expect("respond"); + let response = + wait_for_server_event(&pool, server_pubkey, "\"content\"", Duration::from_secs(2)).await; + assert!(response.content.contains("\"big-1\"")); + + server.close().await.expect("close"); +} + +// ── sweep survival with the threaded snapshot TTL ─────────────────────────── + +/// A payment that outlives the stale-route sweep still delivers its result through +/// the production wiring: the snapshot recorded at invoice time (with the TTL the +/// entry point threads from the configured `payment_ttl`) is the only delivery path +/// once the route is swept. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn long_payment_survives_the_sweep_with_the_threaded_snapshot_ttl() { + let mut fx = fixture( + payments_options(400).with_payment_ttl(Duration::from_secs(120)), + |c| { + c.with_request_timeout(Duration::from_millis(100)) + .with_cleanup_interval(Duration::from_millis(50)) + }, + None, + ) + .await; + + fx.client.send(&paid_call("sweep-1")).await.expect("send"); + let request_event_id = client_request_events(&fx.pool, fx.client_pubkey, "sweep-1").await[0] + .id + .to_hex(); + + wait_for_server_event( + &fx.pool, + fx.server_pubkey, + "payment_required", + Duration::from_secs(2), + ) + .await; + + let incoming = tokio::time::timeout(Duration::from_secs(3), fx.server_rx.recv()) + .await + .expect("the paid request must reach the handler") + .expect("channel open"); + assert_eq!(incoming.event_id, request_event_id); + + // The route must be genuinely gone before the response is sent, so the delivery + // below is attributable to the snapshot path alone. + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + while fx.server.has_event_route(&request_event_id).await { + assert!( + tokio::time::Instant::now() < deadline, + "the stale-route sweep must have reaped the route during the payment" + ); + tokio::time::sleep(Duration::from_millis(25)).await; + } + + fx.server + .send_response(&request_event_id, result_response("sweep-1")) + .await + .expect("the swept-route response must deliver from the snapshot"); + let response = wait_for_server_event( + &fx.pool, + fx.server_pubkey, + "\"content\"", + Duration::from_secs(2), + ) + .await; + let tags = all_tags(&response); + assert!( + tags.contains(&vec!["e".to_string(), request_event_id.clone()]), + "the delivered response must carry the request's correlation, got {tags:?}" + ); + assert!( + response.content.contains("\"sweep-1\""), + "the delivered response must restore the client's own request id" + ); + + fx.server.close().await.expect("close"); +} + +// ── no authorization migration across lifecycles ──────────────────────────── + +/// One client public key, one server, the session mode flipped by mid-session +/// `payment_interaction` updates: a completed transparent payment mints no gating +/// state, and a gating grant is neither consumed nor honored by the transparent +/// lifecycle, surviving intact for the gating retry. Wire-observable throughout; no +/// store access. The single-pubkey shape is what makes this falsifiable: grants key +/// on the client public key plus the canonical invocation identity, so a two-client +/// fixture could never observe a cross-lifecycle migration. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn grants_do_not_migrate_across_lifecycles() { + let (client_pool, server_pool) = MockRelayPool::create_pair(); + let server_pubkey = server_pool.mock_public_key(); + let pool = Arc::new(server_pool); + let client_pool = Arc::new(client_pool); + + let mut server = NostrServerTransport::with_relay_pool( + NostrServerTransportConfig::default().with_encryption_mode(EncryptionMode::Disabled), + as_pool(&pool), + ) + .await + .expect("server transport"); + with_server_payments(&mut server, payments_options(50)).expect("register payments"); + let mut server_rx = server.take_message_receiver().expect("rx"); + server.start().await.expect("start"); + tokio::time::sleep(Duration::from_millis(20)).await; + + // One client keypair for the whole test: raw signed events, so each message + // controls its own payment_interaction tag (the client transport's one-shot + // emission latch cannot express a mid-session update). + let keys = Keys::generate(); + + // 1) Under the default transparent mode, complete a paid call. + let m1 = signed_paid_call(&keys, server_pubkey, "mig-1", vec![]); + client_pool.publish_event(&m1).await.expect("publish"); + wait_for_server_event( + &pool, + server_pubkey, + "payment_required", + Duration::from_secs(2), + ) + .await; + let incoming = tokio::time::timeout(Duration::from_secs(3), server_rx.recv()) + .await + .expect("the paid transparent request must reach the handler") + .expect("channel open"); + assert_eq!(incoming.event_id, m1.id.to_hex()); + server + .send_response(&incoming.event_id, result_response("mig-1")) + .await + .expect("respond"); + + // 2) Update the session to explicit gating and re-send the identical + // invocation: the completed transparent payment must have minted no gating + // grant, so this is offered, not forwarded. + let m2 = signed_paid_call( + &keys, + server_pubkey, + "mig-2", + vec![pi_tag("explicit_gating")], + ); + client_pool.publish_event(&m2).await.expect("publish"); + let offer = wait_for_server_event( + &pool, + server_pubkey, + "Payment Required", + Duration::from_secs(2), + ) + .await; + assert!( + all_tags(&offer).contains(&vec!["e".to_string(), m2.id.to_hex()]), + "the offer must answer the gating invocation" + ); + assert!(offer.content.contains("\"code\":-32042")); + assert!( + server_rx.try_recv().is_err(), + "a transparent execution must never mint an explicit-gating authorization" + ); + + // The fake settles the offered payment 50 ms later: a gating grant now exists. + tokio::time::sleep(Duration::from_millis(500)).await; + + // 3) Update back to transparent WITHOUT retrying: the identical invocation must + // run the transparent lifecycle (a fresh invoice), not consume the gating grant + // as a free forward. + let m3 = signed_paid_call(&keys, server_pubkey, "mig-3", vec![pi_tag("transparent")]); + client_pool.publish_event(&m3).await.expect("publish"); + wait_until( + "the transparent lifecycle must re-invoice the granted identity", + Duration::from_secs(2), + async || { + server_events_containing(&pool, server_pubkey, "payment_required") + .await + .iter() + .any(|e| all_tags(e).contains(&vec!["e".to_string(), m3.id.to_hex()])) + }, + ) + .await; + // This transparent payment settles and forwards on its own; drain it. + let incoming = tokio::time::timeout(Duration::from_secs(3), server_rx.recv()) + .await + .expect("the second transparent payment must forward after settling") + .expect("channel open"); + assert_eq!(incoming.event_id, m3.id.to_hex()); + server + .send_response(&incoming.event_id, result_response("mig-3")) + .await + .expect("respond"); + + // 4) Update to explicit gating again: the grant minted in step 2 must still be + // intact, so the retry claims and forwards with no new offer. + let m4 = signed_paid_call( + &keys, + server_pubkey, + "mig-4", + vec![pi_tag("explicit_gating")], + ); + client_pool.publish_event(&m4).await.expect("publish"); + let incoming = tokio::time::timeout(Duration::from_secs(3), server_rx.recv()) + .await + .expect("the gating retry must claim the intact grant") + .expect("channel open"); + assert_eq!(incoming.event_id, m4.id.to_hex()); + assert!( + !server_events_containing(&pool, server_pubkey, "Payment Required") + .await + .iter() + .any(|e| all_tags(e).contains(&vec!["e".to_string(), m4.id.to_hex()])), + "the claiming retry must not draw a fresh offer" + ); + + server.close().await.expect("close"); +} + +// ── the announcement surface ──────────────────────────────────────────────── + +/// The kind 11316 announcement carries the composed payment surface in order: the +/// `pmi` tags in registration order, the availability tag last in the extra +/// segment (present only under the permissive policy), and the `cap` pricing tags. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn announcement_carries_the_payment_surface() { + for (policy, expect_availability) in [ + (PaymentInteractionPolicy::Optional, true), + (PaymentInteractionPolicy::Transparent, false), + ] { + let pool = Arc::new(MockRelayPool::new()); + let mut server = NostrServerTransport::with_relay_pool( + NostrServerTransportConfig::default() + .with_server_info(ServerInfo::default().with_name("announcer")), + as_pool(&pool), + ) + .await + .expect("server transport"); + + let options = ServerPaymentsOptions::new( + vec![fake_processor("pmi:A", 0), fake_processor("pmi:B", 0)], + vec![ + PricedCapability { + method: "tools/call".to_string(), + name: Some("add".to_string()), + amount: 1, + max_amount: None, + currency_unit: "sats".to_string(), + description: None, + }, + PricedCapability { + method: "prompts/get".to_string(), + name: Some("summarize".to_string()), + amount: 5, + max_amount: Some(20), + currency_unit: "sats".to_string(), + description: None, + }, + ], + ) + .with_payment_interaction(policy); + with_server_payments(&mut server, options).expect("register payments"); + + server.announce().await.expect("announce"); + let announcement = pool + .stored_events() + .await + .into_iter() + .find(|e| e.kind == Kind::Custom(11316)) + .expect("announcement"); + + // The payment tags, in composition order, filtered out of the surrounding + // server-info and capability tags. + let payment_tags: Vec> = all_tags(&announcement) + .into_iter() + .filter(|t| { + matches!( + t.first().map(String::as_str), + Some("pmi") | Some("payment_interaction") | Some("cap") + ) + }) + .collect(); + let mut expected = vec![ + vec!["pmi".to_string(), "pmi:A".to_string()], + vec!["pmi".to_string(), "pmi:B".to_string()], + ]; + if expect_availability { + expected.push(vec![ + "payment_interaction".to_string(), + "explicit_gating".to_string(), + ]); + } + expected.push(vec![ + "cap".to_string(), + "tool:add".to_string(), + "1".to_string(), + "sats".to_string(), + ]); + expected.push(vec![ + "cap".to_string(), + "prompt:summarize".to_string(), + "5-20".to_string(), + "sats".to_string(), + ]); + assert_eq!( + payment_tags, expected, + "policy {policy:?}: the announcement's payment surface must match" + ); + } +} + +// ── double registration refused ───────────────────────────────────────────── + +/// A second registration on the same transport is refused, and a priced call is +/// charged exactly once: the double-charge a silently appended second middleware +/// pair would produce is structurally closed. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn double_registration_is_refused() { + let (client_pool, server_pool) = MockRelayPool::create_pair(); + let server_pubkey = server_pool.mock_public_key(); + let pool = Arc::new(server_pool); + + let mut server = NostrServerTransport::with_relay_pool( + NostrServerTransportConfig::default().with_encryption_mode(EncryptionMode::Disabled), + as_pool(&pool), + ) + .await + .expect("server transport"); + with_server_payments(&mut server, payments_options(10_000)) + .expect("the first registration succeeds"); + let error = with_server_payments(&mut server, payments_options(10_000)) + .expect_err("a second registration must be refused"); + assert!( + error + .to_string() + .contains("a payment interaction policy is already recorded"), + "unexpected error: {error}" + ); + + let mut server_rx = server.take_message_receiver().expect("rx"); + server.start().await.expect("start"); + + let mut client = NostrClientTransport::with_relay_pool( + NostrClientTransportConfig::default() + .with_relay_urls(vec!["wss://mock.relay".to_string()]) + .with_server_pubkey(server_pubkey.to_hex()) + .with_encryption_mode(EncryptionMode::Disabled) + .with_timeout(Duration::from_secs(30)), + Arc::new(client_pool), + ) + .await + .expect("client transport"); + client.start().await.expect("client start"); + tokio::time::sleep(Duration::from_millis(20)).await; + + // The verify is parked, so within this window every charge stays visible as its + // own payment_required. + client.send(&paid_call("once-1")).await.expect("send"); + wait_for_server_event( + &pool, + server_pubkey, + "payment_required", + Duration::from_secs(2), + ) + .await; + tokio::time::sleep(Duration::from_millis(300)).await; + let required = server_events_containing(&pool, server_pubkey, "payment_required").await; + assert_eq!( + required.len(), + 1, + "one registered lifecycle charges one priced request exactly once" + ); + assert!(server_rx.try_recv().is_err()); + + server.close().await.expect("close"); +} + +// ── advertisement and disclosure dedup ────────────────────────────────────── + +/// The first response of a gating session carries exactly one `payment_interaction` +/// tag: the replayed availability advertisement and the effective-mode disclosure +/// deduplicate through the production wiring. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn advertisement_and_disclosure_dedup_on_first_response() { + let mut fx = fixture( + payments_options(10_000), // parked: this test is about the offer's tags + |c| c, + Some(PaymentInteractionMode::ExplicitGating), + ) + .await; + + fx.client.send(&paid_call("dedup-1")).await.expect("send"); + let offer = wait_for_server_event( + &fx.pool, + fx.server_pubkey, + "Payment Required", + Duration::from_secs(2), + ) + .await; + let mode_tags: Vec> = all_tags(&offer) + .into_iter() + .filter(|t| t.first().map(String::as_str) == Some("payment_interaction")) + .collect(); + assert_eq!( + mode_tags, + vec![vec![ + "payment_interaction".to_string(), + "explicit_gating".to_string() + ]], + "the advertisement and the disclosure must collapse to one tag" + ); + + fx.server.close().await.expect("close"); +}