Skip to content

Retry and subscription fixes #836

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tompro merged 3 commits into from
Mar 2, 2026
Merged

Retry and subscription fixes #836

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8978513
Retry and subscription fixes
tompro Mar 2, 2026
03c804f
Merge branch 'master' into retry-fixes
tompro Mar 2, 2026
a285c85
Review fixes
tompro Mar 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion crates/bcr-ebill-api/src/constants.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ pub const MAX_FILE_NAME_CHARACTERS: usize = 200;
pub const VALID_FILE_MIME_TYPES: [&str; 3] = ["image/jpeg", "image/png", "application/pdf"];

// When subscribing events we subtract this from the last received event time
pub const NOSTR_EVENT_TIME_SLACK: u64 = 3600 * 24 * 7; // 1 week
pub const DEFAULT_INITIAL_SUBSCRIPTION_DELAY_SECONDS: u32 = 1;
pub const NOSTR_MAX_RELAYS: usize = 200;

Expand Down
37 changes: 32 additions & 5 deletions crates/bcr-ebill-transport/src/block_transport.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,10 @@ use bcr_ebill_core::application::ServiceTraitBounds;
use bcr_ebill_core::application::company::Company;
use bcr_ebill_core::protocol::blockchain::BlockchainType;
use bcr_ebill_core::protocol::crypto::BcrKeys;
use bcr_ebill_core::protocol::event::{BillChainEvent, CompanyChainEvent, IdentityChainEvent};
use bcr_ebill_core::protocol::event::{
BillChainEvent, CompanyChainEvent, EventEnvelope, IdentityChainEvent,
};
use bitcoin::base58;
use log::{debug, error};

use bcr_ebill_api::service::transport_service::{Error, Result};
Expand Down Expand Up @@ -156,8 +159,20 @@ impl BlockTransportServiceApi for BlockTransportService {
if let Some((recipient, invite)) = events.generate_company_invite_message()
&& let Some(identity) = self.nostr_transport.resolve_identity(&recipient).await
{
node.send_private_event(&events.sender(), &identity, invite.try_into()?)
.await?;
let message: EventEnvelope = invite.try_into()?;
if let Err(e) = node
.send_private_event(&events.sender(), &identity, message.clone())
.await
{
error!("Failed to send company invite, queuing for retry: {e}");
self.nostr_transport
.queue_retry_message(
&events.sender(),
Some(&recipient),
base58::encode(&borsh::to_vec(&message)?),
)
.await?;
}
}

Ok(())
Expand Down Expand Up @@ -216,8 +231,20 @@ impl BlockTransportServiceApi for BlockTransportService {
if !invites.is_empty() {
for (recipient, event) in invites {
if let Some(identity) = self.nostr_transport.resolve_identity(&recipient).await {
node.send_private_event(&events.sender(), &identity, event.try_into()?)
.await?;
let message: EventEnvelope = event.try_into()?;
if let Err(e) = node
.send_private_event(&events.sender(), &identity, message.clone())
.await
{
error!("Failed to send bill invite, queuing for retry: {e}");
self.nostr_transport
.queue_retry_message(
&events.sender(),
Some(&recipient),
base58::encode(&borsh::to_vec(&message)?),
)
.await?;
}
}
}
}
Expand Down
19 changes: 3 additions & 16 deletions crates/bcr-ebill-transport/src/nostr.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ use std::{
};

use bcr_ebill_api::{
constants::{NOSTR_EVENT_TIME_SLACK, NOSTR_MAX_RELAYS},
constants::NOSTR_MAX_RELAYS,
service::{
contact_service::ContactServiceApi,
transport_service::{
Expand Down Expand Up @@ -1077,15 +1077,10 @@ pub async fn should_process(

fn valid_time(kind: Kind, created: nostr::Timestamp, since: Option<Timestamp>) -> bool {
match since {
Some(ts) => {
let time = if matches!(kind, Kind::EncryptedDirectMessage | Kind::GiftWrap) {
add_time_slack(ts)
} else {
ts
};
Some(time) if !matches!(kind, Kind::EncryptedDirectMessage | Kind::GiftWrap) => {
created.as_u64() >= time.inner()
}
None => true,
_ => true,
}
}

Expand Down Expand Up @@ -1163,14 +1158,6 @@ async fn get_offset(db: &Arc<dyn NostrEventOffsetStoreApi>, node_id: &NodeId) ->
.unwrap_or(Timestamp::zero())
}

fn add_time_slack(ts: Timestamp) -> Timestamp {
if ts.inner() <= NOSTR_EVENT_TIME_SLACK {
ts
} else {
ts - NOSTR_EVENT_TIME_SLACK
}
}

pub async fn add_offset(
db: &Arc<dyn NostrEventOffsetStoreApi>,
event_id: EventId,
Expand Down
44 changes: 34 additions & 10 deletions crates/bcr-ebill-transport/src/nostr_transport.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,13 @@ use log::{debug, error, warn};
use bcr_ebill_api::service::transport_service::{Error, Result};
use bcr_ebill_core::protocol::PostalAddress;

#[derive(borsh::BorshDeserialize)]
struct LegacyBillEventMessage {
event_type: bcr_ebill_core::protocol::event::EventType,
version: String,
data: BillChainEventPayload,
}

/// Transport implementation for Nostr
pub struct NostrTransportService {
nostr_client: Arc<dyn TransportClientApi>,
Expand Down Expand Up @@ -147,18 +154,16 @@ impl NostrTransportService {
let node = self.get_node_transport(sender);
for (node_id, event_to_process) in events.iter() {
if let Some(identity) = self.resolve_identity(node_id).await {
let message: EventEnvelope = event_to_process.clone().try_into()?;
if let Err(e) = node
.send_private_event(sender, &identity, event_to_process.clone().try_into()?)
.send_private_event(sender, &identity, message.clone())
.await
{
error!("Failed to send block notification, will add it to retry queue: {e}");
self.queue_retry_message(
sender,
Some(node_id),
base58::encode(
&borsh::to_vec(event_to_process)
.map_err(|e| Error::Message(e.to_string()))?,
),
base58::encode(&borsh::to_vec(&message)?),
)
.await?;
}
Expand Down Expand Up @@ -287,7 +292,29 @@ impl NostrTransportService {
continue;
}
};
match borsh::from_slice::<EventEnvelope>(&decoded) {
let message = match borsh::from_slice::<EventEnvelope>(&decoded) {
Ok(message) => Ok(message),
Err(envelope_err) => {
match borsh::from_slice::<LegacyBillEventMessage>(&decoded) {
Ok(legacy_event) => match borsh::to_vec(&legacy_event.data) {
Ok(data) => Ok(EventEnvelope {
event_type: legacy_event.event_type,
version: legacy_event.version,
data,
}),
Err(e) => Err(Error::Message(e.to_string())),
},
Err(legacy_err) => {
error!(
"Failed to deserialize private retry payload as envelope ({envelope_err}) or legacy event ({legacy_err})"
);
Err(Error::Message(envelope_err.to_string()))
}
}
}
};

match message {
Ok(message) => {
self.send_retry_private_message(
&queued_message.sender_id,
Expand All @@ -296,10 +323,7 @@ impl NostrTransportService {
)
.await
}
Err(e) => {
error!("Failed to deserialize private retry payload: {e}");
Err(Error::Message(e.to_string()))
}
Err(e) => Err(e),
}
}
None => {
Expand Down
66 changes: 66 additions & 0 deletions crates/bcr-ebill-transport/src/transport_service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2444,6 +2444,72 @@ mod tests {
assert!(result.is_ok());
}

#[tokio::test]
async fn test_send_retry_private_message_with_legacy_payload_format() {
init_test_cfg();

let (service, _) = expect_service(
|mock_transport, mock_contact_store, _, mock_queue, _, _, _, _| {
let recipient = node_id_test_other();
let message_id = "private_legacy_msg_id";
let sender = node_id_test();

let legacy_event = Event::new_bill(BillChainEventPayload {
event_type: BillEventType::BillMintingRequested,
bill_id: bill_id_test(),
action_type: Some(ActionType::CheckBill),
sum: Some(Sum::new_sat(1).unwrap()),
});

let payload = base58::encode(&borsh::to_vec(&legacy_event).unwrap());

let queued_message = NostrQueuedMessage {
id: message_id.to_string(),
sender_id: sender.to_owned(),
recipient: Some(recipient.to_owned()),
payload,
};

let identity = get_identity_public_data(
&recipient,
&Email::new("test@example.com").unwrap(),
vec![],
);

mock_contact_store
.expect_get()
.with(eq(recipient.to_owned()))
.returning(move |_| Ok(Some(as_contact(&identity))))
.once();

mock_transport
.expect_send_private_event()
.returning(|_, _, _| Ok(()))
.once();

mock_queue
.expect_get_retry_messages()
.with(eq(1))
.returning(move |_| Ok(vec![queued_message.clone()]))
.once();
mock_queue
.expect_get_retry_messages()
.with(eq(1))
.returning(|_| Ok(vec![]))
.once();

mock_queue
.expect_succeed_retry()
.with(eq(message_id))
.returning(|_| Ok(()))
.once();
},
);

let result = service.send_retry_messages().await;
assert!(result.is_ok());
}

#[tokio::test]
async fn test_send_retry_private_message_with_invalid_base58() {
init_test_cfg();
Expand Down
44 changes: 8 additions & 36 deletions crates/bcr-ebill-wasm/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,49 +147,21 @@ async fn ensure_transport_contact_data_published(ctx: &Context, default_mint_nod
return;
}

if let Ok(full_identity) = ctx.identity_service.get_full_identity().await {
match ctx
.transport_service
.contact_transport()
.resolve_contact(&full_identity.identity.node_id)
if let Ok(full_identity) = ctx.identity_service.get_full_identity().await
&& let Err(e) = ctx
.identity_service
.publish_contact(&full_identity.identity, &full_identity.key_pair)
.await
{
Ok(None) => {
if let Err(e) = ctx
.identity_service
.publish_contact(&full_identity.identity, &full_identity.key_pair)
.await
{
warn!("Could not publish identity details to Nostr: {e}")
}
}
Ok(Some(_)) => (),
Err(e) => {
warn!("Could not resolve personal identity details on Nostr: {e}")
}
}
{
warn!("Could not publish identity details to Nostr: {e}")
}

if let Ok(companies) = ctx.company_service.get_list_of_companies().await {
for c in companies.iter() {
if let Ok((company, keys)) = ctx.company_service.get_company_and_keys_by_id(&c.id).await
&& let Err(e) = ctx.company_service.publish_contact(&company, &keys).await
{
match ctx
.transport_service
.contact_transport()
.resolve_contact(&company.id)
.await
{
Ok(None) => {
if let Err(e) = ctx.company_service.publish_contact(&company, &keys).await {
warn!("Could not publish company details to Nostr: {e}")
}
}
Ok(Some(_)) => (),
Err(e) => {
warn!("Could not resolve company details on Nostr: {e}")
}
}
warn!("Could not publish company details to Nostr: {e}")
}
}
}
Expand Down