1use alloc::{collections::VecDeque, vec::Vec};
2
3use rand_core::CryptoRngCore;
4
5use rns_core::{
6 destination::{self, DestinationAnnounce, SingleInputDestination},
7 hash::AddressHash,
8 identity::{Identity, PrivateIdentity},
9 packet::{
10 ContextFlag, DestinationType, Header, HeaderType, IfacFlag, Packet, PacketContext,
11 PacketDataBuffer, PacketType, PropagationType,
12 },
13};
14
15use lxmf_core::message::{Message as LxmfMessage, WireMessage};
16
17use crate::{
18 adapters::FrameLink,
19 config::MiniNodeConfig,
20 error::MiniNodeError,
21 event::NodeEvent,
22 store::{MiniNodeStore, NeighborSnapshot, NodeSnapshot},
23 telemetry::{
24 encode_recent_fields, PositionFix, TelemetryPoint, TelemetryQuery, TelemetrySample,
25 },
26};
27
28#[derive(Clone)]
29pub struct NeighborRecord {
30 pub destination_hash: [u8; 16],
31 pub identity: Identity,
32 pub last_seen_ms: u64,
33 pub app_data: Vec<u8>,
34}
35
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct MessageEnvelope {
38 pub destination_hash: [u8; 16],
39 pub message_id: [u8; 32],
40}
41
42pub struct MiniNode<S: MiniNodeStore> {
43 config: MiniNodeConfig,
44 store: S,
45 identity: PrivateIdentity,
46 delivery_destination: SingleInputDestination,
47 neighbors: VecDeque<NeighborRecord>,
48 recent_message_ids: VecDeque<[u8; 32]>,
49 outbound_frames: VecDeque<Vec<u8>>,
50 telemetry: VecDeque<TelemetryPoint>,
51 latest_position: Option<PositionFix>,
52 events: VecDeque<NodeEvent>,
53 last_announce_ms: Option<u64>,
54}
55
56impl<S: MiniNodeStore> MiniNode<S> {
57 pub fn load_or_create<R: CryptoRngCore + Copy>(
58 rng: R,
59 config: MiniNodeConfig,
60 mut store: S,
61 ) -> Result<Self, MiniNodeError> {
62 let identity = match store.load_identity()? {
63 Some(bytes) => PrivateIdentity::from_private_key_bytes(&bytes)?,
64 None => {
65 let identity = PrivateIdentity::new_from_rand(rng);
66 store.save_identity(&identity.to_private_key_bytes())?;
67 identity
68 }
69 };
70
71 let snapshot = store.load_snapshot()?.unwrap_or_default();
72 let mut node = Self::new_with_identity(identity, config, store, snapshot);
73 node.push_event(NodeEvent::IdentityReady { destination_hash: node.destination_hash() });
74 Ok(node)
75 }
76
77 pub fn new_with_identity(
78 identity: PrivateIdentity,
79 config: MiniNodeConfig,
80 store: S,
81 snapshot: NodeSnapshot,
82 ) -> Self {
83 let delivery_destination =
84 destination::new_in(identity.clone(), config.app_name, config.aspect);
85 let mut node = Self {
86 config,
87 store,
88 identity,
89 delivery_destination,
90 neighbors: VecDeque::new(),
91 recent_message_ids: VecDeque::new(),
92 outbound_frames: VecDeque::new(),
93 telemetry: VecDeque::new(),
94 latest_position: None,
95 events: VecDeque::new(),
96 last_announce_ms: snapshot.last_announce_ms,
97 };
98 node.restore_snapshot(snapshot);
99 node
100 }
101
102 pub fn tick<R: CryptoRngCore + Copy, L: FrameLink>(
103 &mut self,
104 now_ms: u64,
105 rng: R,
106 link: &mut L,
107 ) -> Result<(), MiniNodeError> {
108 self.maybe_auto_announce(now_ms, rng)?;
109 self.flush_outbound(link)?;
110 self.poll_inbound(now_ms, link)?;
111 Ok(())
112 }
113
114 pub fn destination_hash(&self) -> [u8; 16] {
115 let mut out = [0u8; 16];
116 out.copy_from_slice(self.delivery_destination.desc.address_hash.as_slice());
117 out
118 }
119
120 pub fn identity(&self) -> &PrivateIdentity {
121 &self.identity
122 }
123
124 pub fn neighbors(&self) -> impl Iterator<Item = &NeighborRecord> {
125 self.neighbors.iter()
126 }
127
128 pub fn poll_event(&mut self) -> Option<NodeEvent> {
129 self.events.pop_front()
130 }
131
132 pub fn queue_announce<R: CryptoRngCore + Copy>(
133 &mut self,
134 now_ms: u64,
135 rng: R,
136 app_data: Option<&[u8]>,
137 ) -> Result<(), MiniNodeError> {
138 if self.outbound_frames.len() >= self.config.max_outbound_frames {
139 self.push_event(NodeEvent::AnnounceSkipped {
140 destination_hash: self.destination_hash(),
141 });
142 return Ok(());
143 }
144
145 let packet = self.delivery_destination.announce(rng, app_data)?;
146 let bytes = packet.to_bytes()?;
147 self.outbound_frames.push_back(bytes);
148 self.last_announce_ms = Some(now_ms);
149 self.persist_snapshot()?;
150 self.push_event(NodeEvent::AnnounceQueued { destination_hash: self.destination_hash() });
151 Ok(())
152 }
153
154 pub fn send_message(
155 &mut self,
156 destination_hash: [u8; 16],
157 content: &[u8],
158 ) -> Result<MessageEnvelope, MiniNodeError> {
159 self.send_message_with_fields(destination_hash, content, None)
160 }
161
162 pub fn send_message_with_latest_telemetry(
163 &mut self,
164 destination_hash: [u8; 16],
165 content: &[u8],
166 limit: usize,
167 ) -> Result<MessageEnvelope, MiniNodeError> {
168 let telemetry = self.telemetry.iter().rev().take(limit).cloned().collect::<Vec<_>>();
169 let fields = if telemetry.is_empty() && self.latest_position.is_none() {
170 None
171 } else {
172 Some(encode_recent_fields(&telemetry, self.latest_position.as_ref())?)
173 };
174 self.send_message_with_fields(destination_hash, content, fields)
175 }
176
177 pub fn record_telemetry(&mut self, sample: TelemetrySample) -> Result<(), MiniNodeError> {
178 let (points, latest_position) = sample.into_points(self.destination_hash());
179 let mut inserted = 0usize;
180 let mut dropped = 0usize;
181
182 if let Some(position) = latest_position {
183 self.latest_position = Some(position);
184 }
185
186 for point in points {
187 if self.telemetry.len() >= self.config.max_telemetry_points {
188 self.telemetry.pop_front();
189 dropped += 1;
190 }
191 self.telemetry.push_back(point);
192 inserted += 1;
193 }
194
195 if inserted > 0 {
196 self.push_event(NodeEvent::TelemetryRecorded { inserted });
197 }
198 if dropped > 0 {
199 self.push_event(NodeEvent::TelemetryDropped { dropped });
200 }
201
202 self.persist_snapshot()?;
203 Ok(())
204 }
205
206 pub fn query_telemetry(&self, query: &TelemetryQuery) -> Vec<TelemetryPoint> {
207 let mut matches =
208 self.telemetry.iter().filter(|point| query.matches(point)).cloned().collect::<Vec<_>>();
209
210 if let Some(limit) = query.limit {
211 if matches.len() > limit {
212 matches = matches[matches.len() - limit..].to_vec();
213 }
214 }
215
216 matches
217 }
218
219 fn send_message_with_fields(
220 &mut self,
221 destination_hash: [u8; 16],
222 content: &[u8],
223 fields: Option<rmpv::Value>,
224 ) -> Result<MessageEnvelope, MiniNodeError> {
225 if self.outbound_frames.len() >= self.config.max_outbound_frames {
226 return Err(MiniNodeError::QueueFull);
227 }
228
229 let mut message = LxmfMessage::new();
230 message.destination_hash = Some(destination_hash);
231 message.source_hash = Some(self.destination_hash());
232 message.set_content_from_bytes(content);
233 message.fields = fields;
234
235 let wire = message.to_wire(Some(&self.identity))?;
236 let unpacked = WireMessage::unpack(&wire)?;
237 let message_id = unpacked.message_id();
238
239 let packet = Packet {
240 header: Header {
241 ifac_flag: IfacFlag::Open,
242 header_type: HeaderType::Type1,
243 context_flag: ContextFlag::Unset,
244 propagation_type: PropagationType::Broadcast,
245 destination_type: DestinationType::Single,
246 packet_type: PacketType::Data,
247 hops: 0,
248 },
249 ifac: None,
250 destination: AddressHash::new(destination_hash),
251 transport: None,
252 context: PacketContext::None,
253 data: PacketDataBuffer::new_from_slice(&wire),
254 };
255
256 let bytes = packet.to_bytes()?;
257 self.outbound_frames.push_back(bytes);
258 self.remember_message_id(message_id);
259 self.persist_snapshot()?;
260 self.push_event(NodeEvent::MessageQueued { destination_hash, message_id });
261 Ok(MessageEnvelope { destination_hash, message_id })
262 }
263
264 fn maybe_auto_announce<R: CryptoRngCore + Copy>(
265 &mut self,
266 now_ms: u64,
267 rng: R,
268 ) -> Result<(), MiniNodeError> {
269 let should_announce = match self.last_announce_ms {
270 Some(last) => now_ms.saturating_sub(last) >= self.config.announce_interval_ms,
271 None => true,
272 };
273
274 if should_announce {
275 self.queue_announce(now_ms, rng, None)?;
276 }
277 Ok(())
278 }
279
280 fn flush_outbound<L: FrameLink>(&mut self, link: &mut L) -> Result<(), MiniNodeError> {
281 while let Some(frame) = self.outbound_frames.front() {
282 if frame.len() > link.mtu() {
283 return Err(MiniNodeError::MtuExceeded { frame_len: frame.len(), mtu: link.mtu() });
284 }
285 link.send_frame(frame)?;
286 self.outbound_frames.pop_front();
287 }
288 Ok(())
289 }
290
291 fn poll_inbound<L: FrameLink>(
292 &mut self,
293 now_ms: u64,
294 link: &mut L,
295 ) -> Result<(), MiniNodeError> {
296 let mut processed = 0;
297 while let Some(frame) = link.poll_frame()? {
298 self.handle_frame(now_ms, &frame)?;
299 processed += 1;
300 if processed >= 8 {
301 break;
302 }
303 }
304 Ok(())
305 }
306
307 fn handle_frame(&mut self, now_ms: u64, frame: &[u8]) -> Result<(), MiniNodeError> {
308 let packet = Packet::from_bytes(frame)?;
309 match packet.header.packet_type {
310 PacketType::Announce => self.handle_announce(now_ms, &packet),
311 PacketType::Data
312 if packet.destination == self.delivery_destination.desc.address_hash =>
313 {
314 self.handle_message(&packet)
315 }
316 _ => Ok(()),
317 }
318 }
319
320 fn handle_announce(&mut self, now_ms: u64, packet: &Packet) -> Result<(), MiniNodeError> {
321 let info = DestinationAnnounce::validate(packet)?;
322 let mut destination_hash = [0u8; 16];
323 destination_hash.copy_from_slice(info.destination.desc.address_hash.as_slice());
324 let app_data = info.app_data.to_vec();
325
326 if let Some(existing) =
327 self.neighbors.iter_mut().find(|neighbor| neighbor.destination_hash == destination_hash)
328 {
329 existing.identity = info.destination.desc.identity;
330 existing.last_seen_ms = now_ms;
331 existing.app_data = app_data.clone();
332 } else {
333 if self.neighbors.len() >= self.config.max_neighbors {
334 self.neighbors.pop_front();
335 }
336 self.neighbors.push_back(NeighborRecord {
337 destination_hash,
338 identity: info.destination.desc.identity,
339 last_seen_ms: now_ms,
340 app_data: app_data.clone(),
341 });
342 }
343
344 self.persist_snapshot()?;
345 self.push_event(NodeEvent::AnnounceReceived { destination_hash, app_data });
346 Ok(())
347 }
348
349 fn handle_message(&mut self, packet: &Packet) -> Result<(), MiniNodeError> {
350 let wire = WireMessage::unpack(packet.data.as_slice())?;
351 let message_id = wire.message_id();
352 if self.recent_message_ids.iter().any(|entry| entry == &message_id) {
353 return Ok(());
354 }
355
356 let verified = self
357 .neighbors
358 .iter()
359 .find(|neighbor| neighbor.destination_hash == wire.source)
360 .map(|neighbor| wire.verify(&neighbor.identity).unwrap_or(false))
361 .unwrap_or(false);
362
363 let content = wire.payload.content.as_ref().map(|value| value.to_vec()).unwrap_or_default();
364 self.remember_message_id(message_id);
365 self.persist_snapshot()?;
366 self.push_event(NodeEvent::MessageReceived {
367 source_hash: wire.source,
368 message_id,
369 verified,
370 content,
371 });
372 Ok(())
373 }
374
375 fn remember_message_id(&mut self, message_id: [u8; 32]) {
376 if self.recent_message_ids.len() >= self.config.max_recent_messages {
377 self.recent_message_ids.pop_front();
378 }
379 self.recent_message_ids.push_back(message_id);
380 }
381
382 fn restore_snapshot(&mut self, snapshot: NodeSnapshot) {
383 self.last_announce_ms = snapshot.last_announce_ms;
384 self.latest_position = snapshot.latest_position;
385
386 for neighbor in snapshot.neighbors.into_iter().take(self.config.max_neighbors) {
387 self.neighbors.push_back(NeighborRecord {
388 destination_hash: neighbor.destination_hash,
389 identity: neighbor.identity,
390 last_seen_ms: neighbor.last_seen_ms,
391 app_data: neighbor.app_data,
392 });
393 }
394
395 let keep_recent_from =
396 snapshot.recent_message_ids.len().saturating_sub(self.config.max_recent_messages);
397 for message_id in snapshot.recent_message_ids.into_iter().skip(keep_recent_from) {
398 self.recent_message_ids.push_back(message_id);
399 }
400
401 for point in snapshot.telemetry.into_iter().take(self.config.max_telemetry_points) {
402 self.telemetry.push_back(point);
403 }
404 }
405
406 fn persist_snapshot(&mut self) -> Result<(), MiniNodeError> {
407 let snapshot = NodeSnapshot {
408 last_announce_ms: self.last_announce_ms,
409 neighbors: self
410 .neighbors
411 .iter()
412 .cloned()
413 .map(|neighbor| NeighborSnapshot {
414 destination_hash: neighbor.destination_hash,
415 identity: neighbor.identity,
416 last_seen_ms: neighbor.last_seen_ms,
417 app_data: neighbor.app_data,
418 })
419 .collect(),
420 recent_message_ids: self.recent_message_ids.iter().copied().collect(),
421 telemetry: self.telemetry.iter().cloned().collect(),
422 latest_position: self.latest_position,
423 };
424 self.store.save_snapshot(&snapshot)
425 }
426
427 fn push_event(&mut self, event: NodeEvent) {
428 if self.events.len() >= self.config.max_events {
429 self.events.pop_front();
430 }
431 self.events.push_back(event);
432 }
433}