1use crate::control::{self, ControlServer};
2use crate::discovery::DiscoveryProvider;
3use crate::discovery::lan::LanDiscovery;
4use crate::routing::{format_explain, predict_delivery, strategy_from_kind};
5use crate::sync::{SyncOpts, SyncStats, sync_session};
6use deaddrop_core::chunk::{ErasureSpec, default_fixed};
7use deaddrop_core::config::Config;
8use deaddrop_core::crypto::PrivateIdentity;
9use deaddrop_core::event::{Bus, Event, Metrics};
10use deaddrop_core::protocol::{CreateDrop, build_drop, open_drop_at, verify_envelope};
11use deaddrop_core::receipt::Receipt;
12use deaddrop_core::store::{StorageQuotas, Store, unix_now};
13use deaddrop_core::{
14 Destination, NodeMode, Ownership, PROTOCOL_LABEL, PeerId, Priority, PublicIdentity, Result,
15 RoutingPolicy, ephemeral_discovery_id,
16};
17use std::net::SocketAddr;
18use std::path::Path;
19use std::sync::Arc;
20use tokio::net::TcpListener;
21
22pub struct Node {
23 pub store: Arc<Store>,
24 pub identity: PrivateIdentity,
25 pub cfg: Config,
26 pub metrics: Metrics,
27 pub bus: Bus,
28}
29
30impl Node {
31 pub fn open(dir: &Path, cfg: Config) -> Result<Self> {
32 let quotas = StorageQuotas {
33 maximum: Config::parse_bytes(&cfg.storage.maximum),
34 reserved_local: Config::parse_bytes(&cfg.storage.reserved_local),
35 relay_budget: Config::parse_bytes(&cfg.storage.relay_budget),
36 temporary: Config::parse_bytes(&cfg.storage.temporary),
37 };
38 let store = Store::open(dir, quotas)?;
39 let identity = store.load_identity()?;
40 Ok(Self {
41 store: Arc::new(store),
42 identity,
43 cfg,
44 metrics: Metrics::default(),
45 bus: Bus::default(),
46 })
47 }
48
49 pub fn send_file(
50 &self,
51 path: &Path,
52 to: &str,
53 public: bool,
54 ttl: Option<u64>,
55 priority: Priority,
56 ) -> Result<deaddrop_core::ObjectId> {
57 let body = std::fs::read(path)?;
58 let book = self.store.load_contacts()?;
59 let dest = if public {
60 Destination::Public
61 } else {
62 let (peer, _) = book.resolve(to)?;
63 Destination::One { peer }
64 };
65 let mut recipients = Vec::new();
66 if !public {
67 let (peer, ident) = book.resolve(to)?;
68 recipients.push((peer, ident));
69 }
70 let built = build_drop(CreateDrop {
71 author: &self.identity,
72 recipients,
73 destination: dest,
74 plaintext: body,
75 now: unix_now(),
76 ttl_secs: ttl,
77 priority,
78 routing: RoutingPolicy {
79 kind: self.cfg.strategy(),
80 replication_budget: self.cfg.routing.replication_budget,
81 trusted_only: false,
82 },
83 application: "dd.file".into(),
84 topic: None,
85 chunking: default_fixed(),
86 compress: false,
87 hop_limit: 16,
88 public,
89 seal_until: None,
90 seal_quorum: None,
91 erasure: None,
92 })?;
93 let chunks: Vec<_> = built
94 .chunks
95 .into_iter()
96 .enumerate()
97 .map(|(i, d)| (i as u32, d))
98 .collect();
99 let oid = self.store.put_object(
100 &built.envelope,
101 &built.manifest,
102 &chunks,
103 Ownership::Local,
104 unix_now(),
105 )?;
106 self.store.trace(oid, unix_now(), "created locally")?;
107 self.metrics
108 .drops_created
109 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
110 self.bus.emit(Event::drop_created(oid));
111 Ok(oid)
112 }
113
114 pub fn send_opts(&self, opts: SendOpts) -> Result<deaddrop_core::ObjectId> {
115 let book = self.store.load_contacts()?;
116 let dest = if opts.public {
117 Destination::Public
118 } else {
119 let (peer, _) = book.resolve(&opts.to)?;
120 Destination::One { peer }
121 };
122 let mut recipients = Vec::new();
123 if !opts.public {
124 let (peer, ident) = book.resolve(&opts.to)?;
125 recipients.push((peer, ident));
126 }
127 let chunking = deaddrop_core::chunk::ChunkingAlg::Fixed {
128 size: deaddrop_core::chunk::adaptive_chunk_size(opts.plaintext.len() as u64, 30, "tcp"),
129 };
130 let built = build_drop(CreateDrop {
131 author: &self.identity,
132 recipients,
133 destination: dest,
134 plaintext: opts.plaintext,
135 now: unix_now(),
136 ttl_secs: opts.ttl,
137 priority: opts.priority,
138 routing: RoutingPolicy {
139 kind: self.cfg.strategy(),
140 replication_budget: self.cfg.routing.replication_budget,
141 trusted_only: opts.trusted_only,
142 },
143 application: opts.application,
144 topic: opts.topic,
145 chunking,
146 compress: false,
147 hop_limit: 16,
148 public: opts.public,
149 seal_until: opts.seal_until,
150 seal_quorum: opts.seal_quorum,
151 erasure: opts.erasure,
152 })?;
153 let chunks: Vec<_> = built
154 .chunks
155 .into_iter()
156 .enumerate()
157 .map(|(i, d)| (i as u32, d))
158 .collect();
159 let oid = self.store.put_object(
160 &built.envelope,
161 &built.manifest,
162 &chunks,
163 Ownership::Local,
164 unix_now(),
165 )?;
166 self.store.trace(oid, unix_now(), "created locally")?;
167 self.bus.emit(Event::drop_created(oid));
168 Ok(oid)
169 }
170
171 pub fn send_payload(
172 &self,
173 to: &str,
174 plaintext: Vec<u8>,
175 public: bool,
176 ttl: Option<u64>,
177 priority: Priority,
178 application: &str,
179 topic: Option<String>,
180 trusted_only: bool,
181 ) -> Result<deaddrop_core::ObjectId> {
182 self.send_opts(SendOpts {
183 to: to.to_string(),
184 plaintext,
185 public,
186 ttl,
187 priority,
188 application: application.into(),
189 topic,
190 trusted_only,
191 seal_until: None,
192 seal_quorum: None,
193 erasure: None,
194 })
195 }
196
197 pub fn inbox(&self) -> Result<Vec<InboxItem>> {
198 let now = unix_now();
199 let mut items = Vec::new();
200 for id in self.store.inventory(now)? {
201 let Some(env) = self.store.get_envelope(&id)? else {
202 continue;
203 };
204 if !env.destination.includes(&self.identity.peer_id) && !env.destination.is_public() {
205 continue;
206 }
207 if !self.store.complete(&id)? {
208 continue;
209 }
210 let Some(man) = self.store.get_manifest(&id)? else {
211 continue;
212 };
213 let chunks = self.store.load_chunks(&id)?;
214 let receipts = self.store.receipt_issuer_count(&id).unwrap_or(0);
215 match open_drop_at(&self.identity, &env, &man, &chunks, now, receipts) {
216 Ok(pt) => items.push(InboxItem {
217 object_id: id,
218 from: env.source,
219 bytes: pt.len() as u64,
220 application: env.application,
221 body: pt,
222 }),
223 Err(_) => continue,
224 }
225 }
226 Ok(items)
227 }
228
229 pub fn explain_route(&self, object: &str, peer: &str) -> Result<String> {
230 let book = self.store.load_contacts()?;
231 let (oid, _) = resolve_object(&self.store, object)?;
232 let env = self
233 .store
234 .get_envelope(&oid)?
235 .ok_or_else(|| deaddrop_core::DdError::invalid_frame("unknown object"))?;
236 let (pid, _) = book.resolve(peer).or_else(|_| {
237 peer.parse::<PeerId>().map(|p| {
238 (
239 p,
240 PublicIdentity {
241 version: 2,
242 ed25519_pk: [0; 32],
243 x25519_pk: [0; 32],
244 signature: [0; 64],
245 },
246 )
247 })
248 })?;
249 let copies = self.store.replication(&oid)?;
250 let d = strategy_from_kind(self.cfg.strategy()).decide(
251 &env,
252 pid,
253 self.identity.peer_id,
254 unix_now(),
255 &self.store,
256 self.cfg.node.mode,
257 deaddrop_core::RelayCapacity::Full,
258 copies,
259 )?;
260 Ok(format_explain(pid, &d))
261 }
262
263 pub fn ack_delivery(&self, object: &str) -> Result<Receipt> {
264 let (oid, _) = resolve_object(&self.store, object)?;
265 let r = Receipt::issue(
266 deaddrop_core::ReceiptKind::Delivered,
267 oid,
268 &self.identity,
269 unix_now(),
270 );
271 self.store
272 .set_state(&oid, deaddrop_core::DropState::Delivered)?;
273 Ok(r)
274 }
275
276 pub async fn sync_once(&self, addr: SocketAddr) -> Result<SyncStats> {
278 let s = tokio::net::TcpStream::connect(addr).await?;
279 let opts = SyncOpts {
280 store: self.store.as_ref(),
281 identity: &self.identity,
282 mode: self.cfg.node.mode,
283 default_strategy: self.cfg.strategy(),
284 metrics: Some(&self.metrics),
285 bus: Some(&self.bus),
286 initiator: true,
287 };
288 let stats = sync_session(opts, s).await?;
289 let _ =
290 crate::discovery::remember_locator(self.store.as_ref(), stats.peer, &addr.to_string());
291 Ok(stats)
292 }
293
294 pub async fn serve(&self, listen: SocketAddr, static_peers: Vec<SocketAddr>) -> Result<()> {
295 let listener = TcpListener::bind(listen).await?;
296 tracing::info!(
297 "DDP {} listen {listen} id {}",
298 PROTOCOL_LABEL,
299 self.identity.peer_id
300 );
301 let control = ControlServer::bind(self.store.root(), listen).await?;
302 let stop = control.stop.clone();
303 if self.cfg.discovery.lan && self.cfg.discovery.mode.advertise_lan() {
304 let lan = LanDiscovery {
305 port: self.cfg.discovery.lan_port,
306 };
307 let beacon_id = if self.cfg.discovery.ephemeral_ids {
308 ephemeral_discovery_id(self.identity.peer_id, unix_now())
309 } else {
310 self.identity.peer_id
311 };
312 let stream_port = listen.port();
313 tokio::spawn(async move {
314 let _ = lan.advertise(beacon_id, stream_port).await;
315 });
316 }
317 let store = self.store.clone();
318 let identity = PrivateIdentity::from_secrets(
319 self.identity.ed25519_bytes(),
320 self.identity.x25519_bytes(),
321 );
322 let identity = Arc::new(identity);
323 let mode = self.cfg.node.mode;
324 let strat = self.cfg.strategy();
325 let store_a = store.clone();
326 let id_a = identity.clone();
327 tokio::spawn(async move {
328 loop {
329 if let Ok((s, addr)) = listener.accept().await {
330 tracing::info!("inbound {addr}");
331 let store = store_a.clone();
332 let identity = id_a.clone();
333 tokio::spawn(async move {
334 let opts = SyncOpts {
335 store: store.as_ref(),
336 identity: identity.as_ref(),
337 mode,
338 default_strategy: strat,
339 metrics: None,
340 bus: None,
341 initiator: false,
342 };
343 let _ = sync_session(opts, s).await;
344 });
345 }
346 }
347 });
348 let mut tick = tokio::time::interval(std::time::Duration::from_secs(4));
349 loop {
350 if stop.load(std::sync::atomic::Ordering::Relaxed) {
351 control::clear(self.store.root());
352 break;
353 }
354 tick.tick().await;
355 let _ = self.store.gc(unix_now());
356 for peer in &static_peers {
357 let store = store.clone();
358 let identity = identity.clone();
359 let addr = *peer;
360 tokio::spawn(async move {
361 if let Ok(s) = tokio::net::TcpStream::connect(addr).await {
362 let opts = SyncOpts {
363 store: store.as_ref(),
364 identity: identity.as_ref(),
365 mode,
366 default_strategy: strat,
367 metrics: None,
368 bus: None,
369 initiator: true,
370 };
371 let _ = sync_session(opts, s).await;
372 }
373 });
374 }
375 }
376 Ok(())
377 }
378
379 pub fn predict_route(&self, dest: &str) -> Result<crate::routing::DeliveryForecast> {
380 let book = self.store.load_contacts()?;
381 let (pid, _) = book.resolve(dest).or_else(|_| {
382 dest.parse::<PeerId>().map(|p| {
383 (
384 p,
385 PublicIdentity {
386 version: 2,
387 ed25519_pk: [0; 32],
388 x25519_pk: [0; 32],
389 signature: [0; 64],
390 },
391 )
392 })
393 })?;
394 predict_delivery(&self.store, pid, unix_now())
395 }
396
397 pub fn events(&self) -> Vec<Event> {
398 self.bus.take()
399 }
400
401 pub fn set_mode(&mut self, mode: NodeMode) {
402 self.cfg.node.mode = mode;
403 }
404}
405
406#[derive(Debug, Clone)]
407pub struct SendOpts {
408 pub to: String,
409 pub plaintext: Vec<u8>,
410 pub public: bool,
411 pub ttl: Option<u64>,
412 pub priority: Priority,
413 pub application: String,
414 pub topic: Option<String>,
415 pub trusted_only: bool,
416 pub seal_until: Option<u64>,
417 pub seal_quorum: Option<u32>,
418 pub erasure: Option<ErasureSpec>,
419}
420
421#[derive(Debug, Clone)]
422pub struct InboxItem {
423 pub object_id: deaddrop_core::ObjectId,
424 pub from: PeerId,
425 pub bytes: u64,
426 pub application: String,
427 pub body: Vec<u8>,
428}
429
430fn resolve_object(
431 store: &Store,
432 spec: &str,
433) -> Result<(deaddrop_core::ObjectId, deaddrop_core::DropEnvelope)> {
434 if let Ok(id) = spec.parse() {
435 if let Some(env) = store.get_envelope(&id)? {
436 return Ok((id, env));
437 }
438 }
439 let prefix = spec.rsplit(':').next().unwrap_or(spec).to_ascii_lowercase();
440 for id in store.inventory(unix_now())? {
441 if deaddrop_core::hex_encode(id.as_bytes()).starts_with(&prefix) {
442 if let Some(env) = store.get_envelope(&id)? {
443 return Ok((id, env));
444 }
445 }
446 }
447 Err(deaddrop_core::DdError::invalid_frame("object not found"))
448}
449
450pub fn verify_open(store: &Store, env: &deaddrop_core::DropEnvelope, now: u64) -> Result<()> {
451 verify_envelope(env, now)?;
452 let _ = store;
453 Ok(())
454}