use std::collections::{HashMap, HashSet};
use std::time::{Duration, Instant};
use base64::Engine as _;
use offline_protocol_core::Username;
use offline_protocol_mls::discovery::{
parse_discovery_body, verify_discovery_record, DiscoveryBody, DiscoveryRecordV1,
DiscoveryTombstoneV1,
};
use offline_protocol_transport::nostr::ResolveRequest;
use offline_protocol_transport::nostr_crypto::discovery_tag_for_username;
use serde::{Deserialize, Serialize};
use tracing::{debug, warn};
use super::{lock_shared_state, storage_keys, OfflineProtocol, ProtocolState};
use crate::events::{Event, UsernameClaim};
use crate::{Error, Result};
const DISCOVERY_REFRESH_INTERVAL: Duration = Duration::from_secs(60);
const DISCOVERY_MAX_BACKOFF: Duration = Duration::from_secs(1800);
const RESOLUTION_TIMEOUT: Duration = Duration::from_secs(30);
const MAX_CONCURRENT_RESOLUTIONS: usize = 32;
const MAX_PENDING_RETRACTIONS: usize = 8;
const MAX_CLAIMS_PER_RESOLUTION: usize = 32;
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub(crate) struct DiscoveryClaimState {
#[serde(default)]
pub(crate) claim: Option<Username>,
#[serde(default)]
pub(crate) retracting: Vec<Username>,
}
pub(crate) struct PendingResolution {
username: Username,
claims: HashMap<String, DiscoveryRecordV1>,
tombstoned: HashSet<String>,
rejected: u32,
truncated: u32,
started: Instant,
}
impl OfflineProtocol {
pub(crate) fn refresh_nostr_discovery_claim(&mut self) {
if !self.nostr_discovery_refresh_due() {
return;
}
let now = Instant::now();
let confirmed = self
.transport_manager
.take_confirmed_nostr_discovery_publications();
self.retire_confirmed_retractions(&confirmed);
for tag in self
.transport_manager
.take_failed_nostr_discovery_publications()
{
if self.discovery_tag_is_claimed(&tag) {
self.nostr_discovery_published = false;
}
self.note_discovery_failure(&tag, now);
}
let enabled = self.transport_manager.nostr_discovery_active();
let desired = if enabled {
match self.config.profile.parse::<Username>() {
Ok(username) => Some(username),
Err(e) => {
debug!(
error = %e,
"Profile is not a claimable username; discovery claim not published"
);
None
}
}
} else {
None
};
if let Some(standing) = self.nostr_discovery_claim.clone() {
if desired.as_ref() != Some(&standing) {
self.owe_retraction(standing);
}
}
self.queue_owed_retractions(desired.as_ref(), now);
self.prune_discovery_backoff(desired.as_ref());
let Some(username) = desired else {
return;
};
if self.nostr_discovery_claim.as_ref() == Some(&username) && self.nostr_discovery_published
{
return;
}
if self.discovery_backoff_active(&username, now) {
return;
}
if self.mls_manager.is_none()
|| self.local_address().is_none()
|| self.transport_manager.nostr_public_key().is_none()
{
debug!("Identity not ready yet; discovery claim deferred to a later tick");
return;
}
if let Err(e) = self.publish_discovery_claim(username) {
warn!(error = %e, "Failed to publish the Nostr username discovery claim");
}
}
fn publish_discovery_claim(&mut self, username: Username) -> Result<()> {
let address = self
.local_address()
.ok_or_else(|| Error::Other("No local address yet".to_string()))?
.parse()
.map_err(|e| Error::Other(format!("Local address is not canonical: {}", e)))?;
let nostr_author = self
.transport_manager
.nostr_public_key()
.ok_or_else(|| Error::Other("Nostr transport has no signing key yet".to_string()))?;
let author_bytes = hex::decode(&nostr_author)
.map_err(|e| Error::Other(format!("Nostr public key is not hex: {}", e)))?;
let mls = self.mls_manager.as_ref().ok_or(Error::MlsNotInitialized)?;
let manager = mls
.read()
.map_err(|_| Error::Other("MLS lock poisoned".to_string()))?;
let public_key = manager
.get_identity_public_key()
.map_err(|e| Error::Other(format!("Failed to get identity public key: {}", e)))?;
let record = DiscoveryRecordV1::unsigned(
username.clone(),
address,
public_key,
author_bytes,
chrono::Utc::now().timestamp_millis(),
)
.sign_with(|payload| manager.sign_data(payload))
.map_err(|e| Error::Other(format!("Failed to sign the discovery record: {}", e)))?;
drop(manager);
let payload =
serde_json::to_vec(&record).map_err(|e| Error::Serialization(e.to_string()))?;
self.transport_manager
.publish_nostr_discovery_record(username.clone(), payload);
self.nostr_discovery_published = true;
let claim_changed = self.nostr_discovery_claim.as_ref() != Some(&username);
let debt_cleared = self.remove_owed_retraction(&username);
if claim_changed {
self.nostr_discovery_claim = Some(username);
}
if claim_changed || debt_cleared {
self.persist_nostr_discovery_claim();
}
Ok(())
}
fn owe_retraction(&mut self, username: Username) {
if !self.nostr_discovery_retracting.contains(&username) {
if self.nostr_discovery_retracting.len() >= MAX_PENDING_RETRACTIONS {
self.nostr_discovery_retracting.remove(0);
warn!(
cap = MAX_PENDING_RETRACTIONS,
"Too many un-landed username retractions; giving up on the \
oldest. Its claim may stand until its key packages expire"
);
}
self.nostr_discovery_retracting.push(username);
}
self.nostr_discovery_claim = None;
self.nostr_discovery_published = false;
self.persist_nostr_discovery_claim();
}
fn remove_owed_retraction(&mut self, username: &Username) -> bool {
let before = self.nostr_discovery_retracting.len();
self.nostr_discovery_retracting
.retain(|owed| owed != username);
self.nostr_discovery_retracting.len() != before
}
fn retire_confirmed_retractions(&mut self, tags: &[String]) {
if tags.is_empty() || self.nostr_discovery_retracting.is_empty() {
return;
}
let before = self.nostr_discovery_retracting.len();
self.nostr_discovery_retracting.retain(|username| {
match discovery_tag_for_username(username) {
Ok(tag) => !tags.contains(&tag),
Err(_) => false,
}
});
if self.nostr_discovery_retracting.len() != before {
debug!("A username retraction reached a relay; no longer owed");
self.persist_nostr_discovery_claim();
}
}
fn queue_owed_retractions(&mut self, desired: Option<&Username>, now: Instant) {
if self.nostr_discovery_retracting.is_empty() {
return;
}
if !self.transport_manager.has_nostr_transport() {
debug!("No Nostr transport; username retractions stay owed");
return;
}
let owed: Vec<Username> = self
.nostr_discovery_retracting
.iter()
.filter(|username| Some(*username) != desired)
.filter(|username| !self.discovery_backoff_active(username, now))
.cloned()
.collect();
for username in owed {
match serde_json::to_vec(&DiscoveryTombstoneV1::new()) {
Ok(payload) => {
self.transport_manager
.retract_nostr_discovery_record(username, payload);
}
Err(e) => warn!(error = %e, "Failed to build a discovery tombstone"),
}
}
}
fn discovery_tag_is_claimed(&self, tag: &str) -> bool {
self.nostr_discovery_claim
.as_ref()
.and_then(|username| discovery_tag_for_username(username).ok())
.is_some_and(|claimed| claimed == tag)
}
fn prune_discovery_backoff(&mut self, desired: Option<&Username>) {
if self.nostr_discovery_backoff.is_empty() {
return;
}
let mut live: HashSet<String> = self
.nostr_discovery_retracting
.iter()
.filter_map(|username| discovery_tag_for_username(username).ok())
.collect();
for username in desired.into_iter().chain(self.nostr_discovery_claim.iter()) {
if let Ok(tag) = discovery_tag_for_username(username) {
live.insert(tag);
}
}
self.nostr_discovery_backoff
.retain(|tag, _| live.contains(tag));
}
fn nostr_discovery_refresh_due(&mut self) -> bool {
let now = Instant::now();
if let Some(last) = self.last_nostr_discovery_refresh {
if now.duration_since(last) < DISCOVERY_REFRESH_INTERVAL {
return false;
}
}
self.last_nostr_discovery_refresh = Some(now);
true
}
fn note_discovery_failure(&mut self, tag: &str, now: Instant) {
let entry = self
.nostr_discovery_backoff
.entry(tag.to_string())
.or_insert(super::nostr_publication::PublicationBackoff {
failures: 0,
last_failure: now,
retry_at: now,
});
if now.duration_since(entry.last_failure) > DISCOVERY_MAX_BACKOFF {
entry.failures = 1;
} else {
entry.failures = entry.failures.saturating_add(1);
}
entry.last_failure = now;
let delay = if entry.failures <= 1 {
Duration::ZERO
} else {
let shift = (entry.failures - 2).min(16);
DISCOVERY_REFRESH_INTERVAL
.saturating_mul(1u32 << shift)
.min(DISCOVERY_MAX_BACKOFF)
};
entry.retry_at = now + delay;
}
fn discovery_backoff_active(&self, username: &Username, now: Instant) -> bool {
let Ok(tag) =
offline_protocol_transport::nostr_crypto::discovery_tag_for_username(username)
else {
return false;
};
self.nostr_discovery_backoff
.get(&tag)
.is_some_and(|backoff| now < backoff.retry_at)
}
pub fn resolve_username(&mut self, username: &str) -> Result<bool> {
let username: Username = username
.parse()
.map_err(|e| Error::InvalidArgument(format!("Not a resolvable username: {}", e)))?;
if !self.transport_manager.nostr_discovery_active() {
return Err(Error::InvalidConfiguration(
"Username discovery is disabled; enable nostr_username_discovery_enabled \
(which also requires nostr_cold_contact_enabled)"
.to_string(),
));
}
{
let state = lock_shared_state(&self.shared_state)?;
if state.state != ProtocolState::Running {
return Err(Error::NotStarted);
}
}
if self.nostr_resolution_requests.contains_key(&username)
|| self
.nostr_resolutions
.values()
.any(|resolution| resolution.username == username)
{
return Ok(false);
}
match self
.transport_manager
.resolve_nostr_username(username.clone())
{
ResolveRequest::Queued => {}
ResolveRequest::AlreadyQueued => return Ok(false),
ResolveRequest::Disabled => {
return Err(Error::InvalidConfiguration(
"Username discovery is disabled".to_string(),
))
}
ResolveRequest::QueueFull => {
return Err(Error::InvalidState(
"Too many username lookups in flight; retry shortly".to_string(),
))
}
}
self.nostr_resolution_requests
.entry(username)
.or_insert_with(Instant::now);
Ok(true)
}
pub fn begin_username_resolution(&mut self, query_id: String, username: Username) {
if self.nostr_resolution_requests.remove(&username).is_none() {
debug!("Discovery query minted for an already-answered request; not accumulating");
return;
}
if self.nostr_resolutions.len() >= MAX_CONCURRENT_RESOLUTIONS {
if let Some(stale) = self
.nostr_resolutions
.iter()
.min_by_key(|(_, resolution)| resolution.started)
.map(|(id, _)| id.clone())
{
self.flush_username_resolution(&stale);
}
}
if self.nostr_resolutions.contains_key(&query_id) {
warn!(
query_id = %query_id,
"Discovery query id collision; flushing the resolution it displaces"
);
self.flush_username_resolution(&query_id);
}
self.nostr_resolutions.insert(
query_id,
PendingResolution {
username,
claims: HashMap::new(),
tombstoned: HashSet::new(),
rejected: 0,
truncated: 0,
started: Instant::now(),
},
);
}
pub fn handle_resolved_discovery_record(&mut self, query_id: &str, author: &str, data: &[u8]) {
let Some(resolution) = self.nostr_resolutions.get_mut(query_id) else {
debug!(query_id = %query_id, "Discovery record for an unknown resolution");
return;
};
let body = match parse_discovery_body(data) {
Ok(body) => body,
Err(e) => {
debug!(error = %e, "Undecodable discovery record");
resolution.rejected = resolution.rejected.saturating_add(1);
return;
}
};
let record = match body {
DiscoveryBody::Record(record) => *record,
DiscoveryBody::Tombstone => {
resolution.claims.remove(author);
resolution.tombstoned.insert(author.to_string());
return;
}
};
if resolution.tombstoned.contains(author) {
debug!("Discovery record from an author that has retracted; ignoring");
return;
}
let author_bytes = match hex::decode(author) {
Ok(bytes) => bytes,
Err(_) => {
resolution.rejected = resolution.rejected.saturating_add(1);
return;
}
};
if let Err(rejection) =
verify_discovery_record(&record, &resolution.username, &author_bytes)
{
debug!(reason = %rejection, "Discovery record refused");
resolution.rejected = resolution.rejected.saturating_add(1);
return;
}
if resolution.claims.len() >= MAX_CLAIMS_PER_RESOLUTION
&& !resolution.claims.contains_key(author)
{
debug!(
cap = MAX_CLAIMS_PER_RESOLUTION,
"Username resolution at its claim ceiling; ignoring the rest"
);
resolution.truncated = resolution.truncated.saturating_add(1);
return;
}
match resolution.claims.get(author) {
Some(existing) if existing.issued_at_ms >= record.issued_at_ms => {}
_ => {
resolution.claims.insert(author.to_string(), record);
}
}
}
pub fn flush_username_resolution(&mut self, query_id: &str) {
let Some(resolution) = self.nostr_resolutions.remove(query_id) else {
return;
};
let claims: Vec<UsernameClaim> = resolution
.claims
.into_values()
.map(|record| UsernameClaim {
address: record.address.to_string(),
public_key: base64::engine::general_purpose::STANDARD.encode(&record.pubkey),
issued_at_ms: record.issued_at_ms,
})
.collect();
self.emit_event(Event::UsernameResolved {
username: resolution.username.into_string(),
claims,
rejected: resolution.rejected,
truncated: resolution.truncated,
});
}
pub(crate) fn sweep_username_resolutions(&mut self) {
let now = Instant::now();
if !self.nostr_resolutions.is_empty() {
let expired: Vec<String> = self
.nostr_resolutions
.iter()
.filter(|(_, resolution)| {
now.duration_since(resolution.started) > RESOLUTION_TIMEOUT
})
.map(|(id, _)| id.clone())
.collect();
for query_id in expired {
debug!(query_id = %query_id, "Username resolution timed out; emitting what it has");
self.flush_username_resolution(&query_id);
}
}
if self.nostr_resolution_requests.is_empty() {
return;
}
let unminted: Vec<Username> = self
.nostr_resolution_requests
.iter()
.filter(|(_, requested_at)| now.duration_since(**requested_at) > RESOLUTION_TIMEOUT)
.map(|(username, _)| username.clone())
.collect();
for username in unminted {
debug!("Username lookup never reached a relay; emitting an empty answer");
self.nostr_resolution_requests.remove(&username);
self.transport_manager
.cancel_nostr_username_resolution(&username);
self.emit_event(Event::UsernameResolved {
username: username.into_string(),
claims: Vec::new(),
rejected: 0,
truncated: 0,
});
}
}
pub(crate) fn flush_username_resolutions_for_stop(&mut self) {
let in_flight: Vec<String> = self.nostr_resolutions.keys().cloned().collect();
for query_id in in_flight {
debug!(query_id = %query_id, "Protocol stopping; emitting what this resolution has");
self.flush_username_resolution(&query_id);
}
let unminted: Vec<Username> = self.nostr_resolution_requests.keys().cloned().collect();
for username in unminted {
debug!("Protocol stopping; answering a lookup that never reached a relay");
self.nostr_resolution_requests.remove(&username);
self.transport_manager
.cancel_nostr_username_resolution(&username);
self.emit_event(Event::UsernameResolved {
username: username.into_string(),
claims: Vec::new(),
rejected: 0,
truncated: 0,
});
}
}
pub(crate) fn persist_nostr_discovery_claim(&mut self) {
let Some(storage) = self.protocol_state_storage.clone() else {
return;
};
let state = DiscoveryClaimState {
claim: self.nostr_discovery_claim.clone(),
retracting: self.nostr_discovery_retracting.clone(),
};
let bytes = match serde_json::to_vec(&state) {
Ok(bytes) => bytes,
Err(e) => {
warn!(error = %e, "Failed to serialize the Nostr discovery claim");
return;
}
};
if let Err(e) = self.write_state_record(
storage.as_ref(),
storage_keys::NOSTR_DISCOVERY_CLAIM,
storage_keys::NOSTR_DISCOVERY_CLAIM_ID,
&bytes,
) {
warn!(error = %e, "Failed to persist the Nostr discovery claim");
}
}
pub(crate) fn restore_nostr_discovery_claim(&mut self) {
let Some(storage) = self.protocol_state_storage.clone() else {
return;
};
let data = match self.read_state_record(
storage.as_ref(),
storage_keys::NOSTR_DISCOVERY_CLAIM,
storage_keys::NOSTR_DISCOVERY_CLAIM_ID,
) {
Ok(Some(data)) => data,
Ok(None) => return,
Err(e) => {
warn!(error = %e, "Failed to read the Nostr discovery claim; starting fresh");
return;
}
};
let restored = serde_json::from_slice::<DiscoveryClaimState>(&data).or_else(|_| {
serde_json::from_slice::<Option<Username>>(&data).map(|claim| DiscoveryClaimState {
claim,
retracting: Vec::new(),
})
});
match restored {
Ok(state) => {
self.nostr_discovery_claim = state.claim;
self.nostr_discovery_retracting = state.retracting;
self.nostr_discovery_published = false;
}
Err(e) => warn!(error = %e, "Corrupted Nostr discovery claim; starting fresh"),
}
}
}