1use std::time::Duration;
2
3use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
4use base64::Engine as _;
5use lxmf_core::{decide_delivery, Message, MessageMethod, TransportMethod, WireMessage};
6use lxmf_sdk::{MessageId, SdkError, SendRequest};
7use rand_core::OsRng;
8use rns_transport::destination::{DestinationDesc, DestinationName};
9use rns_transport::hash::AddressHash;
10use rns_transport::identity::PrivateIdentity;
11use rns_transport::packet::{
12 ContextFlag, DestinationType, Header, HeaderType, IfacFlag, Packet, PacketContext,
13 PacketDataBuffer, PacketType, PropagationType, LXMF_MAX_PAYLOAD,
14};
15use rns_transport::transport::{SendPacketOutcome, Transport};
16use serde_json::Value as JsonValue;
17
18use crate::link_delivery::send_link_payload;
19use crate::{
20 EXT_ACCEPTED_RESULT_ACK, EXT_DIRECT_PACKET_MAX_WIRE_BYTES, EXT_FIELDS_BASE64,
21 EXT_LINK_CONNECT_TIMEOUT_MS, EXT_PROPAGATION_RELAY_HEX, EXT_RAW_BYTES_BASE64, EXT_SEND_MODE,
22 EXT_USE_PROPAGATION_NODE,
23};
24
25#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize)]
26#[serde(rename_all = "snake_case")]
27pub enum DeliveryMethod {
28 Opportunistic,
29 Direct,
30 Propagated,
31}
32
33#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize)]
34#[serde(rename_all = "snake_case")]
35pub enum DeliveryRepresentation {
36 Packet,
37 Resource,
38}
39
40#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize)]
41#[serde(rename_all = "snake_case")]
42pub enum DeliveryOutcome {
43 SentDirect,
44 SentBroadcast,
45}
46
47#[derive(Clone, Debug, PartialEq, Eq)]
48pub struct InProcessSendReport {
49 pub message_id: MessageId,
50 pub resolved_destination: String,
51 pub method: DeliveryMethod,
52 pub representation: DeliveryRepresentation,
53 pub outcome: DeliveryOutcome,
54 pub relay_destination: Option<String>,
55 pub receipt_hash: Option<String>,
56}
57
58pub(crate) struct SendContext<'a> {
59 pub transport: &'a Transport,
60 pub identity: &'a PrivateIdentity,
61 pub source_destination: AddressHash,
62 pub propagation_relay: Option<AddressHash>,
63 pub link_connect_timeout: Duration,
64 pub link_connect_attempts: usize,
65 pub resource_transfer_timeout: Duration,
66}
67
68pub(crate) async fn send(
69 context: SendContext<'_>,
70 request: &SendRequest,
71) -> Result<InProcessSendReport, SdkError> {
72 let requested_destination = parse_hash(&request.destination)?;
73 let remote = resolve_destination(context.transport, requested_destination, "delivery").await?;
74 let wire = encode_wire(context.identity, context.source_destination, &remote, request)?;
75 let message_id = MessageId(hex::encode(
76 WireMessage::unpack(&wire)
77 .map_err(|err| internal(format!("failed to decode encoded LXMF message: {err}")))?
78 .message_id(),
79 ));
80 let mut desired_method = requested_method(request)?;
81 if matches!(desired_method, TransportMethod::Opportunistic)
82 && request_uses_auto_mode(request)
83 && context.transport.delivery_link_available(&remote.address_hash).await
84 {
85 desired_method = TransportMethod::Direct;
86 }
87 let decision = decide_delivery(desired_method, false, wire.len())
88 .map_err(|err| validation(format!("failed to select delivery representation: {err}")))?;
89 let representation =
90 forced_representation(request, decision.method, decision.representation, wire.len());
91
92 match decision.method {
93 TransportMethod::Opportunistic => {
94 send_opportunistic(context.transport, remote.address_hash, &wire, message_id).await
95 }
96 TransportMethod::Direct => {
97 send_link_payload(
98 context.transport,
99 remote,
100 &wire,
101 message_id,
102 DeliveryMethod::Direct,
103 representation,
104 context.link_connect_timeout,
105 context.link_connect_attempts,
106 context.resource_transfer_timeout,
107 None,
108 )
109 .await
110 }
111 TransportMethod::Propagated => {
112 send_propagated(context, request, &remote, &wire, message_id).await
113 }
114 TransportMethod::Paper => Err(validation("paper delivery is not supported in-process")),
115 }
116}
117
118async fn send_opportunistic(
119 transport: &Transport,
120 destination: AddressHash,
121 wire: &[u8],
122 message_id: MessageId,
123) -> Result<InProcessSendReport, SdkError> {
124 let packet = data_packet(destination, PropagationType::Transport, wire)?;
125 let receipt_hash = hex::encode(packet.hash().to_bytes());
126 let outcome =
127 ensure_sent(transport.send_packet_with_outcome(packet).await, "opportunistic send")?;
128 Ok(InProcessSendReport {
129 message_id,
130 resolved_destination: destination.to_hex_string(),
131 method: DeliveryMethod::Opportunistic,
132 representation: DeliveryRepresentation::Packet,
133 outcome,
134 relay_destination: None,
135 receipt_hash: Some(receipt_hash),
136 })
137}
138
139async fn send_propagated(
140 context: SendContext<'_>,
141 request: &SendRequest,
142 remote: &DestinationDesc,
143 wire: &[u8],
144 message_id: MessageId,
145) -> Result<InProcessSendReport, SdkError> {
146 let relay_hash = request
147 .extensions
148 .get(EXT_PROPAGATION_RELAY_HEX)
149 .and_then(JsonValue::as_str)
150 .map(parse_hash)
151 .transpose()?
152 .or(context.propagation_relay)
153 .ok_or_else(|| transport_error("no propagation relay selected"))?;
154 let relay = resolve_destination(context.transport, relay_hash, "propagation").await?;
155 let recipient = lxmf_core::identity::Identity::new_from_slices(
156 remote.identity.public_key_bytes(),
157 remote.identity.verifying_key_bytes(),
158 );
159 let propagated = WireMessage::unpack(wire)
160 .and_then(|message| {
161 message
162 .pack_propagation_with_options_and_rng(
163 &recipient,
164 crate::state::now_ms() as f64 / 1000.0,
165 Some(&[0_u8; 32]),
166 OsRng,
167 )
168 .map(|(payload, _)| payload)
169 })
170 .map_err(|err| internal(format!("failed to encode propagated LXMF payload: {err}")))?;
171 let representation = if propagated.len() > LXMF_MAX_PAYLOAD {
172 MessageMethod::Resource
173 } else {
174 MessageMethod::Packet
175 };
176 send_link_payload(
177 context.transport,
178 relay,
179 &propagated,
180 message_id,
181 DeliveryMethod::Propagated,
182 representation,
183 context.link_connect_timeout,
184 context.link_connect_attempts,
185 context.resource_transfer_timeout,
186 Some(remote.address_hash.to_hex_string()),
187 )
188 .await
189}
190
191fn encode_wire(
192 identity: &PrivateIdentity,
193 source: AddressHash,
194 remote: &DestinationDesc,
195 request: &SendRequest,
196) -> Result<Vec<u8>, SdkError> {
197 let content = decode_required_base64(request, EXT_RAW_BYTES_BASE64, "content_base64")?;
198 let fields = request
199 .extensions
200 .get(EXT_FIELDS_BASE64)
201 .and_then(JsonValue::as_str)
202 .map(|value| BASE64_STANDARD.decode(value).map_err(|_| validation("invalid fields base64")))
203 .transpose()?
204 .map(|bytes| {
205 rmp_serde::from_slice(&bytes).map_err(|_| validation("invalid msgpack fields"))
206 })
207 .transpose()?;
208 let title = request.payload.get("title").and_then(JsonValue::as_str).unwrap_or_default();
209 let mut message = Message::new();
210 message.source_hash = Some(copy_hash(source));
211 message.destination_hash = Some(copy_hash(remote.address_hash));
212 message.set_content_from_bytes(&content);
213 message.set_title_from_string(title);
214 message.fields = fields;
215 let signer = lxmf_core::identity::PrivateIdentity::from_private_key_bytes(
216 &identity.to_private_key_bytes(),
217 )
218 .map_err(|err| internal(format!("invalid local identity: {err:?}")))?;
219 message
220 .to_wire(Some(&signer))
221 .map_err(|err| internal(format!("failed to encode LXMF message: {err}")))
222}
223
224async fn resolve_destination(
225 transport: &Transport,
226 hash: AddressHash,
227 aspect: &str,
228) -> Result<DestinationDesc, SdkError> {
229 let identity = transport
230 .destination_identity(&hash)
231 .await
232 .ok_or_else(|| transport_error(format!("destination identity unavailable for {hash}")))?;
233 Ok(DestinationDesc { identity, name: DestinationName::new("lxmf", aspect), address_hash: hash })
234}
235
236pub(crate) fn requested_method(request: &SendRequest) -> Result<TransportMethod, SdkError> {
237 if request
238 .extensions
239 .get(EXT_USE_PROPAGATION_NODE)
240 .and_then(JsonValue::as_bool)
241 .unwrap_or(false)
242 {
243 return Ok(TransportMethod::Propagated);
244 }
245 let value = request
246 .extensions
247 .get(EXT_SEND_MODE)
248 .and_then(JsonValue::as_str)
249 .or(request.delivery_method.as_deref())
250 .unwrap_or("auto")
251 .to_ascii_lowercase();
252 match value.as_str() {
253 "auto" | "opportunistic" => Ok(TransportMethod::Opportunistic),
254 "direct" | "directonly" | "direct_only" => Ok(TransportMethod::Direct),
255 "propagated" | "propagationonly" | "propagation_only" => Ok(TransportMethod::Propagated),
256 _ => Err(validation(format!("unsupported delivery method: {value}"))),
257 }
258}
259
260fn request_uses_auto_mode(request: &SendRequest) -> bool {
261 request
262 .extensions
263 .get(EXT_SEND_MODE)
264 .and_then(JsonValue::as_str)
265 .or(request.delivery_method.as_deref())
266 .is_none_or(|value| value.eq_ignore_ascii_case("auto"))
267}
268
269pub(crate) fn forced_representation(
270 request: &SendRequest,
271 method: TransportMethod,
272 representation: MessageMethod,
273 wire_len: usize,
274) -> MessageMethod {
275 if matches!(method, TransportMethod::Direct)
276 && request
277 .extensions
278 .get(EXT_DIRECT_PACKET_MAX_WIRE_BYTES)
279 .and_then(JsonValue::as_u64)
280 .is_some_and(|limit| wire_len as u64 > limit)
281 {
282 MessageMethod::Resource
283 } else {
284 representation
285 }
286}
287
288pub(crate) fn request_link_timeout(request: &SendRequest, fallback: Duration) -> Duration {
289 let accepted = request
290 .extensions
291 .get(EXT_ACCEPTED_RESULT_ACK)
292 .and_then(JsonValue::as_bool)
293 .unwrap_or(false);
294 request
295 .extensions
296 .get(EXT_LINK_CONNECT_TIMEOUT_MS)
297 .and_then(JsonValue::as_u64)
298 .map(Duration::from_millis)
299 .unwrap_or_else(|| if accepted { Duration::from_secs(5) } else { fallback })
300 .clamp(Duration::from_millis(1), Duration::from_secs(120))
301}
302
303pub(crate) fn request_link_attempts(request: &SendRequest, fallback: usize) -> usize {
304 if request.extensions.get(EXT_ACCEPTED_RESULT_ACK).and_then(JsonValue::as_bool).unwrap_or(false)
305 {
306 1
307 } else {
308 fallback.max(1)
309 }
310}
311
312pub(crate) fn request_resource_timeout(request: &SendRequest, fallback: Duration) -> Duration {
313 if request.extensions.get(EXT_ACCEPTED_RESULT_ACK).and_then(JsonValue::as_bool).unwrap_or(false)
314 {
315 Duration::from_secs(8)
316 } else {
317 fallback
318 }
319}
320
321fn decode_required_base64(
322 request: &SendRequest,
323 extension: &str,
324 payload_key: &str,
325) -> Result<Vec<u8>, SdkError> {
326 let value = request
327 .extensions
328 .get(extension)
329 .and_then(JsonValue::as_str)
330 .or_else(|| request.payload.get(payload_key).and_then(JsonValue::as_str))
331 .ok_or_else(|| validation("missing raw payload"))?;
332 BASE64_STANDARD.decode(value).map_err(|_| validation("invalid payload base64"))
333}
334
335fn data_packet(
336 destination: AddressHash,
337 propagation_type: PropagationType,
338 payload: &[u8],
339) -> Result<Packet, SdkError> {
340 Ok(Packet {
341 header: Header {
342 ifac_flag: IfacFlag::Open,
343 header_type: HeaderType::Type1,
344 context_flag: ContextFlag::Unset,
345 propagation_type,
346 destination_type: DestinationType::Single,
347 packet_type: PacketType::Data,
348 hops: 0,
349 },
350 ifac: None,
351 destination,
352 transport: None,
353 context: PacketContext::None,
354 data: PacketDataBuffer::new_from_slice(payload),
355 })
356}
357
358fn parse_hash(value: &str) -> Result<AddressHash, SdkError> {
359 let value = value.trim();
360 if value.len() != 32 || !value.chars().all(|character| character.is_ascii_hexdigit()) {
361 return Err(validation("invalid destination hash"));
362 }
363 AddressHash::new_from_hex_string(value).map_err(|_| validation("invalid destination hash"))
364}
365
366fn copy_hash(hash: AddressHash) -> [u8; 16] {
367 let mut bytes = [0_u8; 16];
368 bytes.copy_from_slice(hash.as_slice());
369 bytes
370}
371
372fn ensure_sent(outcome: SendPacketOutcome, action: &str) -> Result<DeliveryOutcome, SdkError> {
373 match outcome {
374 SendPacketOutcome::SentDirect => Ok(DeliveryOutcome::SentDirect),
375 SendPacketOutcome::SentBroadcast => Ok(DeliveryOutcome::SentBroadcast),
376 _ => Err(transport_error(format!("{action} failed: {outcome:?}"))),
377 }
378}
379
380fn validation(message: impl Into<String>) -> SdkError {
381 SdkError::new(
382 lxmf_sdk::error_code::VALIDATION_INVALID_ARGUMENT,
383 lxmf_sdk::ErrorCategory::Validation,
384 message,
385 )
386 .with_user_actionable(true)
387}
388
389pub(crate) fn transport_error(message: impl Into<String>) -> SdkError {
390 SdkError::new(lxmf_sdk::error_code::INTERNAL, lxmf_sdk::ErrorCategory::Transport, message)
391}
392
393fn internal(message: impl Into<String>) -> SdkError {
394 crate::state::internal_error(message)
395}