use std::sync::Arc;
use anyhow::{anyhow, bail, Result};
use iroh_base::{NodeId, SecretKey};
use iroh_relay::node_info::NodeInfo;
use n0_future::{
boxed::BoxStream,
task::{self, AbortOnDropHandle},
time::{self, Duration, Instant},
};
use pkarr::SignedPacket;
use tracing::{debug, error_span, warn, Instrument};
use url::Url;
use crate::{
discovery::{Discovery, DiscoveryItem, NodeData},
endpoint::force_staging_infra,
watchable::{Disconnected, Watchable, Watcher},
Endpoint,
};
#[cfg(feature = "discovery-pkarr-dht")]
pub mod dht;
pub const N0_DNS_PKARR_RELAY_PROD: &str = "https://dns.iroh.link/pkarr";
pub const N0_DNS_PKARR_RELAY_STAGING: &str = "https://staging-dns.iroh.link/pkarr";
pub const DEFAULT_PKARR_TTL: u32 = 30;
pub const DEFAULT_REPUBLISH_INTERVAL: Duration = Duration::from_secs(60 * 5);
#[derive(derive_more::Debug, Clone)]
pub struct PkarrPublisher {
node_id: NodeId,
watchable: Watchable<Option<NodeInfo>>,
_drop_guard: Arc<AbortOnDropHandle<()>>,
}
impl PkarrPublisher {
pub fn new(secret_key: SecretKey, pkarr_relay: Url) -> Self {
Self::with_options(
secret_key,
pkarr_relay,
DEFAULT_PKARR_TTL,
DEFAULT_REPUBLISH_INTERVAL,
)
}
pub fn with_options(
secret_key: SecretKey,
pkarr_relay: Url,
ttl: u32,
republish_interval: Duration,
) -> Self {
debug!("creating pkarr publisher that publishes to {pkarr_relay}");
let node_id = secret_key.public();
let pkarr_client = PkarrRelayClient::new(pkarr_relay);
let watchable = Watchable::default();
let service = PublisherService {
ttl,
watcher: watchable.watch(),
secret_key,
pkarr_client,
republish_interval,
};
let join_handle = task::spawn(
service
.run()
.instrument(error_span!("pkarr_publish", me=%node_id.fmt_short())),
);
Self {
watchable,
node_id,
_drop_guard: Arc::new(AbortOnDropHandle::new(join_handle)),
}
}
pub fn n0_dns(secret_key: SecretKey) -> Self {
let pkarr_relay = match force_staging_infra() {
true => N0_DNS_PKARR_RELAY_STAGING,
false => N0_DNS_PKARR_RELAY_PROD,
};
let pkarr_relay: Url = pkarr_relay.parse().expect("url is valid");
Self::new(secret_key, pkarr_relay)
}
pub fn update_node_data(&self, data: &NodeData) {
let mut data = data.clone();
if data.relay_url().is_some() {
data.clear_direct_addresses();
}
let info = NodeInfo::from_parts(self.node_id, data);
self.watchable.set(Some(info)).ok();
}
}
impl Discovery for PkarrPublisher {
fn publish(&self, data: &NodeData) {
self.update_node_data(data);
}
}
#[derive(derive_more::Debug, Clone)]
struct PublisherService {
#[debug("SecretKey")]
secret_key: SecretKey,
#[debug("PkarrClient")]
pkarr_client: PkarrRelayClient,
watcher: Watcher<Option<NodeInfo>>,
ttl: u32,
republish_interval: Duration,
}
impl PublisherService {
async fn run(mut self) {
let mut failed_attempts = 0;
let republish = time::sleep(Duration::MAX);
tokio::pin!(republish);
loop {
let Ok(info) = self.watcher.get() else {
break; };
if let Some(info) = info {
if let Err(err) = self.publish_current(info).await {
failed_attempts += 1;
let retry_after = Duration::from_secs(failed_attempts);
republish.as_mut().reset(Instant::now() + retry_after);
warn!(
err = %format!("{err:#}"),
url = %self.pkarr_client.pkarr_relay_url ,
?retry_after,
%failed_attempts,
"Failed to publish to pkarr",
);
} else {
failed_attempts = 0;
republish
.as_mut()
.reset(Instant::now() + self.republish_interval);
}
}
tokio::select! {
res = self.watcher.updated() => match res {
Ok(_) => debug!("Publish node info to pkarr (info changed)"),
Err(Disconnected) => break,
},
_ = &mut republish => debug!("Publish node info to pkarr (interval elapsed)"),
}
}
}
async fn publish_current(&self, info: NodeInfo) -> Result<()> {
debug!(
data = ?info.data,
pkarr_relay = %self.pkarr_client.pkarr_relay_url,
"Publish node info to pkarr"
);
let signed_packet = info.to_pkarr_signed_packet(&self.secret_key, self.ttl)?;
self.pkarr_client.publish(&signed_packet).await?;
Ok(())
}
}
#[derive(derive_more::Debug, Clone)]
pub struct PkarrResolver {
pkarr_client: PkarrRelayClient,
}
impl PkarrResolver {
pub fn new(pkarr_relay: Url) -> Self {
Self {
pkarr_client: PkarrRelayClient::new(pkarr_relay),
}
}
pub fn n0_dns() -> Self {
let pkarr_relay = match force_staging_infra() {
true => N0_DNS_PKARR_RELAY_STAGING,
false => N0_DNS_PKARR_RELAY_PROD,
};
let pkarr_relay: Url = pkarr_relay.parse().expect("url is valid");
Self::new(pkarr_relay)
}
}
impl Discovery for PkarrResolver {
fn resolve(&self, _ep: Endpoint, node_id: NodeId) -> Option<BoxStream<Result<DiscoveryItem>>> {
let pkarr_client = self.pkarr_client.clone();
let fut = async move {
let signed_packet = pkarr_client.resolve(node_id).await?;
let info = NodeInfo::from_pkarr_signed_packet(&signed_packet)?;
let item = DiscoveryItem::new(info, "pkarr", None);
Ok(item)
};
let stream = n0_future::stream::once_future(fut);
Some(Box::pin(stream))
}
}
#[derive(Debug, Clone)]
pub struct PkarrRelayClient {
http_client: reqwest::Client,
pkarr_relay_url: Url,
}
impl PkarrRelayClient {
pub fn new(pkarr_relay_url: Url) -> Self {
Self {
http_client: reqwest::Client::new(),
pkarr_relay_url,
}
}
pub async fn resolve(&self, node_id: NodeId) -> anyhow::Result<SignedPacket> {
let public_key = pkarr::PublicKey::try_from(node_id.as_bytes())
.map_err(|e| anyhow::anyhow!(e.to_string()))?;
let mut url = self.pkarr_relay_url.clone();
url.path_segments_mut()
.map_err(|_| anyhow!("Failed to resolve: Invalid relay URL"))?
.push(&public_key.to_z32());
let response = self.http_client.get(url).send().await?;
if !response.status().is_success() {
bail!(format!(
"Resolve request failed with status {}",
response.status()
))
}
let payload = response.bytes().await?;
SignedPacket::from_relay_payload(&public_key, &payload)
.map_err(|e| anyhow::anyhow!(e.to_string()))
}
pub async fn publish(&self, signed_packet: &SignedPacket) -> anyhow::Result<()> {
let mut url = self.pkarr_relay_url.clone();
url.path_segments_mut()
.map_err(|_| anyhow!("Failed to publish: Invalid relay URL"))?
.push(&signed_packet.public_key().to_z32());
let response = self
.http_client
.put(url)
.body(signed_packet.to_relay_payload())
.send()
.await?;
let status = response.status();
if !status.is_success() {
let text = if let Ok(text) = response.text().await {
text
} else {
"(no details provided)".to_string()
};
bail!(format!(
"Publish request failed with status {status}: {text}",
))
}
Ok(())
}
}