use crate::events::{Event, EventCallback, PresenceStatus};
use crate::telemetry::{dispatch_record, scrub_event, TelemetryContext, TelemetryRecord};
use crate::Error;
use chrono::{DateTime, Utc};
use offline_protocol_core::{
ContentType, ForwardInfo, MediaMetadata, Message, MessageId, MessagePriority, ReplyContext,
};
use offline_protocol_transport::TransportType;
use serde::{Deserialize, Serialize};
use std::collections::{HashSet, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::Instant;
use tracing::warn;
pub(crate) const CONFIRMATION_RETRY_INTERVAL_SECS: i64 = 5;
pub(crate) const CONFIRMATION_PROBE_INTERVAL_SECS: i64 = 5;
pub(crate) const REKEY_INTERVAL_SECS: i64 = 30;
pub(crate) const WELCOME_RETRY_BATCH_SIZE: usize = 20;
pub(crate) const WELCOME_LIFECYCLE_TTL_SECS: i64 = 300;
pub(crate) const WELCOME_RETRY_JITTER_RATIO: f64 = 0.2;
pub(crate) const WELCOME_INTERNET_CONFIRM_TIMEOUT_SECS: i64 = 10;
pub(crate) const WELCOME_MESH_CONFIRM_TIMEOUT_SECS: i64 = 15;
pub(crate) const WELCOME_NO_CARRIER_RETRY_SECS: i64 = 15;
pub(crate) const WELCOME_UNREACHABLE_RETRY_CAP_SECS: i64 = 600;
pub(crate) const WELCOME_WATCHLIST_MAX_AGE_SECS: i64 = 14 * 24 * 60 * 60;
pub(crate) const WELCOME_PRESENCE_RESCUE_BASE_SECS: i64 = 40;
pub(crate) const WELCOME_PRESENCE_RESCUE_MAX_SECS: i64 = 600;
pub(crate) const SEND_FAIL_REASON_RECIPIENT_UNREACHABLE: &str = "recipient_unreachable";
pub(crate) const SEND_FAIL_REASON_TRANSPORT: &str = "transport_send_failed";
pub(crate) const SEND_FAIL_REASON_CONFIRM_TIMEOUT: &str = "send_confirmation_timed_out";
pub(crate) const SEND_FAIL_REASON_TOKENS: &[&str] = &[
SEND_FAIL_REASON_RECIPIENT_UNREACHABLE,
SEND_FAIL_REASON_TRANSPORT,
SEND_FAIL_REASON_CONFIRM_TIMEOUT,
"transport_not_connected",
"peer_not_reachable",
"message_too_large",
"serialization_failed",
"crypto_failed",
"session_not_ready",
"recipient_blocked",
"media_transfer_limit",
"group_not_found",
"invalid_request",
"protocol_state_invalid",
];
pub(crate) fn classify_transport_send_error(raw: &str) -> &'static str {
if raw.starts_with(SEND_FAIL_REASON_RECIPIENT_UNREACHABLE) {
return SEND_FAIL_REASON_RECIPIENT_UNREACHABLE;
}
if let Some(token) = SEND_FAIL_REASON_TOKENS.iter().find(|t| **t == raw) {
return token;
}
match raw {
"Internet transport send failed"
| "Reticulum transport send failed"
| "Nostr transport send failed" => SEND_FAIL_REASON_TRANSPORT,
"Welcome send confirmation timed out" => SEND_FAIL_REASON_CONFIRM_TIMEOUT,
_ => SEND_FAIL_REASON_TRANSPORT,
}
}
pub(crate) fn send_failure_token(err: &crate::Error) -> &'static str {
use crate::Error as E;
match err {
E::Transport(inner) => match inner.code() {
"TRANSPORT_NOT_AVAILABLE" => "transport_not_connected",
"PEER_NOT_REACHABLE" => "peer_not_reachable",
"MESSAGE_TOO_LARGE" => "message_too_large",
"SERIALIZATION_ERROR" => "serialization_failed",
"CRYPTO_ERROR" => "crypto_failed",
_ => SEND_FAIL_REASON_TRANSPORT,
},
E::NotStarted | E::AlreadyStarted | E::InvalidState(_) => "protocol_state_invalid",
E::InvalidConfiguration(_) | E::InvalidArgument(_) | E::PermissionDenied(_) => {
"invalid_request"
}
E::Core(_) | E::Serialization(_) => "serialization_failed",
E::Router(_) | E::Reliability(_) | E::Service(_) => SEND_FAIL_REASON_TRANSPORT,
E::Mls(_) | E::EncryptFailed(_) => "crypto_failed",
E::MlsNotInitialized | E::SessionNotReady(_) | E::NoKeyPackage(_) => "session_not_ready",
E::UserBlocked(_) => "recipient_blocked",
E::MediaTransferLimit(_) => "media_transfer_limit",
E::GroupNotFound(_) => "group_not_found",
E::Other(_) => SEND_FAIL_REASON_TRANSPORT,
}
}
pub(crate) const RECONCILIATION_THROTTLE_MS: u64 = 2_000;
pub(crate) const LAMPORT_PERSIST_INTERVAL: u64 = 64;
pub(crate) const NOSTR_WATERMARK_PERSIST_INTERVAL_SECS: i64 = 300;
pub(crate) const MEDIA_TRANSFER_STALE_TIMEOUT_SECS: u64 = 300;
pub(crate) const MAX_KNOWN_PEERS: usize = 1000;
pub(crate) const KNOWN_PEER_TTL_SECS: u64 = 1800;
pub(crate) use offline_protocol_router::CRITICAL_RELAY_BATTERY_LEVEL;
pub(crate) const CTRL_SIG_META_KEY: &str = "__ctrl_sig";
pub(crate) const CTRL_PK_META_KEY: &str = "__ctrl_pk";
pub(crate) const CTRL_SIGN_DOMAIN: &[u8] = b"offline-ctrl-v1";
pub(crate) const MAX_ENCRYPTION_CAPABLE_PEERS: usize = 8192;
pub(crate) const MAX_BLOCKED_USERS: usize = 10_000;
pub(crate) const MAX_PLAINTEXT_RECEIVE_WARNED_PEERS: usize = 1000;
pub(crate) const MAX_CONTROL_GATE_WARNED_PEERS: usize = 1000;
pub(crate) const MAX_PENDING_KEY_PACKAGES: usize = 1000;
pub(crate) const MAX_PENDING_MESSAGES_PER_PEER: usize = 64;
pub(crate) const MAX_PENDING_MESSAGES_GLOBAL: usize = 4096;
pub(crate) const MAX_PENDING_EXPIRIES_PER_PASS: usize = 64;
pub(crate) const MAX_MIGRATED_PENDING_WRITES_PER_LAUNCH: usize = 512;
pub(crate) const MAX_DEFERRED_RESTORE_SETTLEMENTS: usize = 2 * MAX_PENDING_MESSAGES_GLOBAL;
pub(crate) const MAX_MESSAGE_CONTENT_BYTES: usize = 256 * 1024;
pub(crate) const MAX_PENDING_MESSAGE_BYTES_PER_PEER: usize = 2 * 1024 * 1024;
pub(crate) const MAX_PENDING_MESSAGE_BYTES_GLOBAL: usize = 16 * 1024 * 1024;
pub(crate) const MAX_PROTOCOL_STATE_RECORD_BYTES: usize = 4 * 1024 * 1024;
pub(crate) const MAX_KEY_PACKAGE_SENT_TO: usize = 1000;
pub(crate) const MAX_REKEY_TRACKED_PEERS: usize = 1000;
pub(crate) const MAX_KEY_PACKAGE_LIFETIME_MS: u64 = 30 * 24 * 60 * 60 * 1000;
#[derive(Clone, Debug, Serialize, Deserialize)]
pub(crate) struct EncryptionCapableEntry {
pub(crate) last_seen_ms: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct KeyPackagePayload {
pub(crate) user_id: String,
pub(crate) key_package_data: Vec<u8>,
#[serde(default)]
pub(crate) remaining_lifetime_ms: u64,
#[serde(default)]
pub(crate) timestamp_ms: u64,
#[serde(default)]
pub(crate) session_reset: bool,
#[serde(default)]
pub(crate) wire_versions: Vec<u8>,
#[serde(default)]
pub(crate) env_versions: Vec<u8>,
#[serde(default)]
pub(crate) rich_versions: Vec<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) nostr_pubkey: Option<String>,
}
pub(crate) const MLS_ENVELOPE_COMPACT_V1: u8 = 1;
pub(crate) const RICH_PAYLOAD_V1: u8 = 1;
pub(crate) const MAX_RICH_EXTRAS_BYTES: usize = 32 * 1024;
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub(crate) struct RichSendExtras {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) reply_context: Option<ReplyContext>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) media_metadata: Option<MediaMetadata>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) forward_info: Option<ForwardInfo>,
}
impl RichSendExtras {
pub(crate) fn is_any(&self) -> bool {
self.reply_context.is_some() || self.media_metadata.is_some() || self.forward_info.is_some()
}
pub(crate) fn check_size(&self) -> crate::Result<()> {
if !self.is_any() {
return Ok(());
}
let extras_len = serde_json::to_vec(self)
.map_err(|e| Error::Serialization(e.to_string()))?
.len();
if extras_len > MAX_RICH_EXTRAS_BYTES {
return Err(Error::InvalidArgument(format!(
"Rich extras too large: {} bytes serialized (max {})",
extras_len, MAX_RICH_EXTRAS_BYTES
)));
}
Ok(())
}
}
#[derive(Debug, Clone, Default)]
pub struct SendMessageOptions {
pub priority: Option<MessagePriority>,
pub reply_to_msg: Option<String>,
pub content_type: Option<ContentType>,
pub reply_context: Option<ReplyContext>,
pub media_metadata: Option<MediaMetadata>,
pub forward_info: Option<ForwardInfo>,
pub via_transport: Option<TransportType>,
}
#[derive(Debug, Clone, Default)]
pub struct MediaSendOptions {
pub media_metadata: Option<MediaMetadata>,
pub caption: Option<String>,
pub reply_to_msg: Option<String>,
pub reply_context: Option<ReplyContext>,
pub forward_info: Option<ForwardInfo>,
pub file_id: Option<String>,
}
#[derive(Clone, Serialize, Deserialize)]
pub(crate) struct RichPayloadV1 {
pub(crate) text: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) reply_context: Option<ReplyContext>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) media_metadata: Option<MediaMetadata>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) forward_info: Option<ForwardInfo>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) content_type: Option<ContentType>,
}
impl RichPayloadV1 {
pub(crate) fn parse_sealed(plaintext: &str, sender: &str) -> Option<Self> {
let sealed = plaintext.strip_prefix(super::internal_prefixes::RICH_V1)?;
match serde_json::from_str::<Self>(sealed) {
Ok(mut rich) => {
if rich.content_type == Some(ContentType::FileChunk) {
warn!(
sender = %sender,
"Sealed rich payload claims the internal FileChunk content type, ignoring the hint"
);
rich.content_type = None;
}
Some(rich)
}
Err(e) => {
warn!(
sender = %sender,
error = %e,
"Failed to parse sealed rich payload, surfacing raw text"
);
None
}
}
}
}
#[derive(Debug, Clone)]
pub(crate) struct PendingConnectionRequest {
pub(crate) recipient: String,
pub(crate) sent_at: std::time::Instant,
}
pub(crate) const PENDING_CONNECTION_REQUEST_TTL: std::time::Duration =
std::time::Duration::from_secs(1800);
pub(crate) const MAX_PENDING_CONNECTION_REQUESTS: usize = 64;
pub(crate) const MAX_INITIAL_MESSAGE_BYTES: usize = 4096;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct ConnectionRequestPayload {
pub(crate) sender_name: String,
pub(crate) timestamp_ms: i64,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) key_package: Option<Vec<u8>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) initial_message: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct ConnectionAcceptedPayload {
pub(crate) accepted_by_name: String,
#[serde(default)]
pub(crate) timestamp_ms: i64,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) key_package: Option<Vec<u8>>,
}
pub(crate) const MAX_READ_RECEIPT_IDS: usize = 256;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct PresencePayload {
pub(crate) status: PresenceStatus,
pub(crate) timestamp_ms: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct TypingIndicatorPayload {
pub(crate) conversation_id: String,
pub(crate) is_typing: bool,
pub(crate) timestamp_ms: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct ReadReceiptPayload {
pub(crate) message_ids: Vec<String>,
pub(crate) timestamp_ms: i64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct GroupCreatedPayload {
pub(crate) group_id: String,
pub(crate) name: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct GroupMessageReceivedPayload {
pub(crate) group_id: String,
pub(crate) sender: String,
pub(crate) content: String,
pub(crate) timestamp: String,
pub(crate) message_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) reply_to_msg: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) forward_info: Option<offline_protocol_core::ForwardInfo>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct GroupMemberAddedPayload {
pub(crate) group_id: String,
pub(crate) user_id: String,
pub(crate) added_by: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub(crate) group_name: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct GroupMemberRemovedPayload {
pub(crate) group_id: String,
pub(crate) user_id: String,
pub(crate) removed_by: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct GroupInfoMemberPayload {
pub(crate) user_id: String,
pub(crate) role: String,
pub(crate) joined_at: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct GroupInfoPayload {
pub(crate) group_id: String,
pub(crate) name: String,
pub(crate) created_by: String,
pub(crate) created_at: String,
pub(crate) members: Vec<GroupInfoMemberPayload>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct UserGroupSummaryPayload {
pub(crate) group_id: String,
pub(crate) name: String,
pub(crate) created_at: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct UserGroupsPayload {
pub(crate) groups: Vec<UserGroupSummaryPayload>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct GroupErrorPayload {
pub(crate) reason: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) group_id: Option<String>,
}
impl GroupErrorPayload {
pub(crate) fn classify_reason(&self) -> &'static str {
match self.reason.as_str() {
"Group not found" => "not_found",
"Only admins can sync this group" => "sync_denied",
_ => "error",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct ReceivedKeyPackage {
pub(crate) key_package_data: Vec<u8>,
pub(crate) local_expires_at_ms: u64,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub(crate) struct PeerCapabilities {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub(crate) env_versions: Vec<u8>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub(crate) rich_versions: Vec<u8>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub(crate) attested_rich_versions: Vec<u8>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) nostr_pubkey: Option<String>,
}
pub(crate) const NOSTR_PUBKEY_HEX_LEN: usize = 64;
pub(crate) fn normalize_nostr_pubkey(candidate: &str) -> Option<String> {
if candidate.len() != NOSTR_PUBKEY_HEX_LEN || !candidate.bytes().all(|b| b.is_ascii_hexdigit())
{
return None;
}
Some(candidate.to_ascii_lowercase())
}
pub(crate) const MAX_PERSISTED_CAPABILITY_VERSIONS: usize = 8;
impl PeerCapabilities {
pub(crate) fn from_advertised(
env_versions: &[u8],
rich_versions: &[u8],
nostr_pubkey: Option<&str>,
) -> Self {
Self {
env_versions: env_versions
.iter()
.copied()
.take(MAX_PERSISTED_CAPABILITY_VERSIONS)
.collect(),
rich_versions: rich_versions
.iter()
.copied()
.take(MAX_PERSISTED_CAPABILITY_VERSIONS)
.collect(),
attested_rich_versions: Vec::new(),
nostr_pubkey: nostr_pubkey.and_then(normalize_nostr_pubkey),
}
}
pub(crate) fn is_any(&self) -> bool {
!self.env_versions.is_empty()
|| !self.rich_versions.is_empty()
|| !self.attested_rich_versions.is_empty()
|| self.nostr_pubkey.is_some()
}
}
#[derive(Debug)]
pub(crate) enum ControlGateOutcome {
Proceed { signed: bool },
Rejected(InternalMessageResult),
}
#[derive(Debug)]
pub(crate) enum InternalMessageResult {
Consumed,
SecurityRejected,
Deferred,
Decrypted(String),
}
#[must_use]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ChunkOutcome {
Handled,
Deferred,
Rejected,
}
#[derive(Clone, Serialize, Deserialize)]
pub(crate) struct PendingMessage {
pub(crate) content: String,
pub(crate) priority: MessagePriority,
pub(crate) message_id: MessageId,
pub(crate) reply_to_msg: Option<MessageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) forwarded_from: Option<ForwardInfo>,
#[serde(default)]
pub(crate) content_type: ContentType,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) media_metadata: Option<MediaMetadata>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub(crate) rich: Option<RichSendExtras>,
pub(crate) queued_at: DateTime<Utc>,
#[serde(skip)]
pub(crate) serialized_bytes: usize,
}
#[derive(Serialize, Deserialize)]
pub(crate) struct PendingMessageRecord {
pub(crate) recipient: String,
pub(crate) message: PendingMessage,
}
impl PendingMessage {
pub(crate) fn measure(&mut self) {
self.serialized_bytes = serde_json::to_vec(self)
.map(|encoded| encoded.len())
.unwrap_or_else(|_| self.content.len());
}
}
#[derive(Debug, Clone)]
pub(crate) struct PendingProvenance {
pub(crate) message_id: MessageId,
pub(crate) first_queued_at: Option<DateTime<Utc>>,
}
impl PendingProvenance {
pub(crate) fn requeued(message: &PendingMessage) -> Self {
Self {
message_id: message.message_id.clone(),
first_queued_at: Some(message.queued_at),
}
}
pub(crate) fn for_id(message_id: MessageId) -> Self {
Self {
message_id,
first_queued_at: None,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) enum SessionState {
Pending,
Confirmed,
}
impl SessionState {
pub(crate) fn as_str(&self) -> &'static str {
match self {
Self::Pending => "Pending",
Self::Confirmed => "Confirmed",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub(crate) enum WelcomeDeliveryState {
Created,
SendAttempted,
Sent,
Failed,
Expired,
}
impl WelcomeDeliveryState {
pub(crate) fn as_str(&self) -> &'static str {
match self {
Self::Created => "Created",
Self::SendAttempted => "SendAttempted",
Self::Sent => "Sent",
Self::Failed => "Failed",
Self::Expired => "Expired",
}
}
}
#[derive(Debug, Clone)]
pub(crate) struct PresenceRescueThrottle {
pub(crate) next_allowed_at: DateTime<Utc>,
pub(crate) rescues: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct WelcomeLifecycleRecord {
pub(crate) peer_id: String,
pub(crate) group_id: String,
pub(crate) state: WelcomeDeliveryState,
pub(crate) attempt: u32,
#[serde(default)]
pub(crate) unreachable_parks: u32,
pub(crate) welcome_message: Message,
pub(crate) next_retry_at: Option<DateTime<Utc>>,
pub(crate) last_reason_code: Option<crate::events::WelcomeReasonCode>,
pub(crate) last_transport_error: Option<String>,
pub(crate) created_at: DateTime<Utc>,
pub(crate) expires_at: DateTime<Utc>,
}
pub(crate) mod storage_keys {
pub const PENDING_MESSAGES: &str = "pending_messages";
pub const PENDING_MESSAGE_ENTRIES: &str = "pending_message_entries";
pub const SESSION_STATES: &str = "session_states";
pub const PEER_KEY_PACKAGES: &str = "peer_key_packages";
pub const PEER_CAPABILITIES: &str = "peer_capabilities";
pub const WELCOME_LIFECYCLES: &str = "welcome_lifecycles";
pub const OUTBOX: &str = "outbox";
pub const MEDIA_DESCRIPTORS: &str = "media_descriptors";
pub const LAMPORT_CLOCK: &str = "lamport_clock";
pub const LAMPORT_CLOCK_ID: &str = "current";
pub const ENCRYPTION_CAPABLE_PEERS: &str = "encryption_capable_peers";
pub const BLOCKED_USERS: &str = "blocked_users";
pub const SCRUB_SECRET: &str = "scrub_secret";
pub const SCRUB_SECRET_ID: &str = "current";
pub const NOSTR_SIGNING_SECRET: &str = "nostr_signing_secret";
pub const NOSTR_SIGNING_SECRET_ID: &str = "current";
pub const NOSTR_WATERMARK: &str = "nostr_watermark";
pub const NOSTR_WATERMARK_ID: &str = "current";
pub const NOSTR_KEY_PACKAGE_SLOTS: &str = "nostr_key_package_slots";
pub const NOSTR_KEY_PACKAGE_SLOTS_ID: &str = "current";
pub const NOSTR_DISCOVERY_CLAIM: &str = "nostr_discovery_claim";
pub const NOSTR_DISCOVERY_CLAIM_ID: &str = "current";
pub const STATE_RECORD_KEY: &str = "protocol_state_record_key";
pub const STATE_RECORD_KEY_ID: &str = "current";
pub const BOTH_CREATE_AWAITING_DECRYPT: &str = "both_create_awaiting_decrypt";
pub const STATE_ADOPTION: &str = "protocol_state_adoption";
pub const STATE_ADOPTION_ID: &str = "v1";
pub const ADOPTABLE_STATE_KEY_TYPES: &[&str] = &[
BLOCKED_USERS,
OUTBOX,
PENDING_MESSAGES,
SESSION_STATES,
WELCOME_LIFECYCLES,
PEER_KEY_PACKAGES,
PEER_CAPABILITIES,
MEDIA_DESCRIPTORS,
BOTH_CREATE_AWAITING_DECRYPT,
LAMPORT_CLOCK,
];
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProtocolState {
Stopped,
Running,
Paused,
}
pub(crate) struct SharedState {
pub(crate) state: ProtocolState,
pub(crate) event_handlers: Vec<EventCallback>,
pub(crate) received_messages: VecDeque<Message>,
pub(crate) telemetry: Option<Arc<TelemetryContext>>,
}
impl SharedState {
pub(crate) fn new() -> Self {
Self {
state: ProtocolState::Stopped,
event_handlers: Vec::new(),
received_messages: VecDeque::new(),
telemetry: None,
}
}
pub(crate) fn emit_event(&self, event: Event) {
for handler in &self.event_handlers {
let event_for_handler = event.clone();
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
handler(event_for_handler);
}));
if result.is_err() {
tracing::error!(
event = ?event,
"EventCallback panicked; continuing. Handlers must not panic — see OfflineProtocol::on_event.",
);
}
}
if let Some(ctx) = &self.telemetry {
let scrubbed = scrub_event::scrub_event(&event, &ctx.scrubber);
let record = TelemetryRecord::Protocol(Box::new(scrubbed.into_owned()));
dispatch_record(&ctx.sink, &record);
}
}
}
pub(crate) fn lock_shared_state(
state: &Arc<Mutex<SharedState>>,
) -> std::result::Result<std::sync::MutexGuard<'_, SharedState>, Error> {
state
.lock()
.map_err(|_| Error::Other("Shared state mutex poisoned".to_string()))
}
#[derive(Debug, Clone)]
pub(crate) struct OutboxReseal {
pub(crate) content: String,
pub(crate) priority: MessagePriority,
pub(crate) reply_to_msg: Option<MessageId>,
pub(crate) forwarded_from: Option<ForwardInfo>,
pub(crate) content_type: ContentType,
pub(crate) media_metadata: Option<MediaMetadata>,
pub(crate) rich: Option<RichSendExtras>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct OutboxEntry {
pub(crate) message: Message,
pub(crate) attempt_count: u32,
pub(crate) first_sent_at: DateTime<Utc>,
pub(crate) last_sent_at: DateTime<Utc>,
pub(crate) last_transport: Option<TransportType>,
#[serde(skip)]
pub(crate) reseal: Option<OutboxReseal>,
}
#[derive(Clone)]
pub(crate) struct PendingMediaMetadataEntry {
pub(crate) content_type: ContentType,
pub(crate) media_metadata: Option<MediaMetadata>,
pub(crate) last_updated_at: Instant,
pub(crate) sender: String,
pub(crate) rich_extras: Option<crate::media_envelope::MediaRichExtras>,
pub(crate) timestamp_ms: i64,
}
#[derive(Clone)]
pub(crate) struct OutboundMediaTransfer {
pub(crate) content_type: ContentType,
pub(crate) recipient: String,
pub(crate) pinned_transport: TransportType,
pub(crate) total_chunks: u32,
pub(crate) delivered_chunks: HashSet<u32>,
pub(crate) last_updated_at: Instant,
pub(crate) media_metadata: Option<MediaMetadata>,
pub(crate) rich_extras: Option<crate::media_envelope::MediaRichExtras>,
}
#[derive(Clone, Serialize, Deserialize)]
pub(crate) struct MediaTransferDescriptor {
pub(crate) file_id: String,
pub(crate) recipient: String,
pub(crate) file_name: String,
pub(crate) file_size: u64,
pub(crate) file_checksum: String,
pub(crate) content_type: ContentType,
pub(crate) queued_at: DateTime<Utc>,
}
pub(crate) enum OutboundSendPreparation {
Ready(String),
Queued(MessageId),
}
#[cfg(test)]
mod signing_domain_tests {
use offline_protocol_mls::discovery::DISCOVERY_SIGN_DOMAIN;
use offline_protocol_mls::invite::INVITE_SIGN_DOMAIN;
use super::CTRL_SIGN_DOMAIN;
const RELAY_ADDR_SIGN_DOMAIN: &[u8] = b"offline-relay-addr-v1";
#[test]
fn signing_domains_are_mutually_non_prefixing() {
let domains: [(&str, &[u8]); 4] = [
("offline-ctrl-v1", CTRL_SIGN_DOMAIN),
("offline-disc-v1", DISCOVERY_SIGN_DOMAIN),
("offline-invite-v1", INVITE_SIGN_DOMAIN),
("offline-relay-addr-v1", RELAY_ADDR_SIGN_DOMAIN),
];
for (a_name, a) in domains {
for (b_name, b) in domains {
if a_name == b_name {
continue;
}
assert!(
!a.starts_with(b),
"signing domain {} ({:?}) is a prefix of {} ({:?}), which \
lets a signature made under one verify under the other",
b_name,
String::from_utf8_lossy(b),
a_name,
String::from_utf8_lossy(a)
);
}
}
}
#[test]
fn signing_domains_have_their_published_spellings() {
assert_eq!(CTRL_SIGN_DOMAIN, b"offline-ctrl-v1");
assert_eq!(DISCOVERY_SIGN_DOMAIN, b"offline-disc-v1");
assert_eq!(INVITE_SIGN_DOMAIN, b"offline-invite-v1");
assert_eq!(RELAY_ADDR_SIGN_DOMAIN, b"offline-relay-addr-v1");
}
#[test]
fn signing_domains_are_distinct() {
let domains: [&[u8]; 4] = [
CTRL_SIGN_DOMAIN,
DISCOVERY_SIGN_DOMAIN,
INVITE_SIGN_DOMAIN,
RELAY_ADDR_SIGN_DOMAIN,
];
for (i, a) in domains.iter().enumerate() {
for b in domains.iter().skip(i + 1) {
assert_ne!(a, b, "two signing domains are the same string");
}
}
}
}
#[cfg(test)]
mod send_failure_classification_tests {
use super::*;
#[test]
fn every_send_failure_token_classifies_to_itself() {
for token in SEND_FAIL_REASON_TOKENS {
assert_eq!(
classify_transport_send_error(token),
*token,
"{token} must survive a re-classification"
);
}
}
#[test]
fn transport_error_tokens_are_in_the_vocabulary() {
use offline_protocol_transport::Error as T;
let cases = [
T::TransportNotAvailable("ble".into()),
T::PeerNotReachable("bob".into()),
T::SendFailed("x".into()),
T::ReceiveFailed("x".into()),
T::ConfigurationError("x".into()),
T::SerializationError("x".into()),
T::MessageTooLarge(2, 1),
T::CryptoError("x".into()),
T::Other("x".into()),
];
for case in cases {
let token = send_failure_token(&crate::Error::Transport(case.clone()));
assert!(
SEND_FAIL_REASON_TOKENS.contains(&token),
"{case:?} produced {token}, which is not in SEND_FAIL_REASON_TOKENS"
);
}
}
#[test]
fn relay_prose_is_dropped_and_the_token_kept() {
assert_eq!(
classify_transport_send_error(
"recipient_unreachable: Recipient is offline and push notification could not be sent"
),
SEND_FAIL_REASON_RECIPIENT_UNREACHABLE
);
for prose in [
"Recipient connection lost and push notification could not be sent",
"Push notification already sent for this message; recipient unreachable",
"Recipient unreachable and the message is too large to deliver by push",
"Group delivery did not settle for this recipient in time",
] {
assert_eq!(
classify_transport_send_error(&format!("recipient_unreachable: {prose}")),
SEND_FAIL_REASON_RECIPIENT_UNREACHABLE
);
}
}
#[test]
fn unknown_text_falls_back_and_is_never_echoed() {
for raw in [
"Relay rejected event: blocked",
"off1qyvh9kjj4vy8f943a22qvxsct5s9ydew35v2dl2c is not connected",
"",
] {
let classified = classify_transport_send_error(raw);
assert_eq!(classified, SEND_FAIL_REASON_TRANSPORT);
assert!(
SEND_FAIL_REASON_TOKENS.contains(&classified),
"the fallback must itself be a token"
);
}
}
#[test]
fn uniffi_substituted_literals_are_classified() {
for literal in [
"Internet transport send failed",
"Reticulum transport send failed",
"Nostr transport send failed",
] {
assert_eq!(
classify_transport_send_error(literal),
SEND_FAIL_REASON_TRANSPORT
);
}
assert_eq!(
classify_transport_send_error("Welcome send confirmation timed out"),
SEND_FAIL_REASON_CONFIRM_TIMEOUT
);
}
}