umsh_mobile_core/
mobile_chat.rs

1//! Owned mobile facade records for the sans-I/O text engine.
2//!
3//! Conversations are addressed by an opaque string that discriminates the two
4//! kinds the platform can hold: a peer's base58 address for a direct
5//! conversation, and `ch:<hex tag>` for a channel's group conversation. Room
6//! profiles are still outside this facade, and are dropped rather than
7//! exported in a form Swift would have to interpret.
8
9use std::cell::RefCell;
10use std::collections::{BTreeMap, BTreeSet};
11use std::rc::Rc;
12
13use rand::Rng;
14use umsh_core::{ChannelKey, ChannelTag, NodeHint, PublicKey};
15use umsh_text::engine::sequence::MessageHandle;
16use umsh_text::engine::{
17    ArchiveKey, CompletionStatus, ComposeIntent, ComposeRef, DeliveryState, Direction, Engine,
18    EngineConfig, Event, MessageMutation, MutationKind, Output, Presence, RegardingRef,
19    ResolvedRef, StreamCheckpoint, Transmission,
20};
21use umsh_text::model::{ConversationKey, SenderScope, WireRef};
22use umsh_text::validate::DirectChannelProfile;
23
24pub(crate) type ChatEngine = Engine<DirectChannelProfile>;
25
26/// Prefix marking a channel group conversation address.
27const CHANNEL_ADDRESS_PREFIX: &str = "ch:";
28/// Prefix marking a blind-unicast conversation over a channel key.
29const CHANNEL_DIRECT_ADDRESS_PREFIX: &str = "chd:";
30
31#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
32pub struct MobileChatCheckpointRecord {
33    pub conversation_address: String,
34    pub next_id: u8,
35    pub epoch: u16,
36}
37
38#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
39pub struct MobileChatArchiveRecord {
40    pub conversation_address: String,
41    pub message_id: u8,
42    pub fragment_index: Option<u8>,
43    pub payload: Vec<u8>,
44}
45
46/// Retire every archived fragment stored under one message ID. Emitted when
47/// an edit or delete supersedes that ID's content; the platform must apply
48/// these *before* the batch's archive upserts so an edit's replacement
49/// payloads land on a clean slate and the superseded content can never be
50/// served to a resend request again.
51#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
52pub struct MobileChatArchiveDeleteRecord {
53    pub conversation_address: String,
54    pub message_id: u8,
55}
56
57#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Enum)]
58pub enum MobileChatMutationKind {
59    Insert,
60    UpdateBody,
61    Edit,
62    Delete,
63}
64
65#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Enum)]
66pub enum MobileChatDirection {
67    Inbound,
68    Outbound,
69}
70
71/// Ordered-slot presence of a transcript row (mirrors the engine's
72/// [`Presence`]). A `GapPending` row is a reserved spinner placeholder; an
73/// `Unavailable` row is a permanent loss marker.
74#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Enum)]
75pub enum MobileChatPresence {
76    Present,
77    GapPending,
78    Unavailable,
79}
80
81/// Physical-layer metadata for a record produced by a live received frame.
82///
83/// The engine is transport-agnostic and carries none of this, so the facade
84/// attaches it alongside the mutation the frame produced. Absent on records
85/// that came from a timer, a compose, or a repair drain.
86#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
87pub struct MobileChatRxMetadataRecord {
88    pub rssi_dbm: Option<i16>,
89    pub snr_centibels: Option<i16>,
90    pub lqi: Option<u8>,
91    /// Radio links the frame crossed to get here, counting the final one into
92    /// this device: a frame heard directly from its sender is one hop. The
93    /// same count a ping reply reports, and absent for the same reason — a
94    /// frame source-routed without a trace route crossed hops nobody
95    /// recorded.
96    pub hop_count: Option<u8>,
97    /// Intermediate-router hints in trace-route order: each forwarding
98    /// repeater prepends its own hint, so the list starts nearest us and ends
99    /// nearest the sender. That is return-path order — usable directly as a
100    /// source route back — and the reverse of the path the frame travelled.
101    pub route_hints: Vec<Vec<u8>>,
102    pub source_authenticated: bool,
103}
104
105/// A channel member previously known only by their claimed hint has been
106/// resolved to a full public key.
107///
108/// Multicast senders are identified on the wire by a 3-byte hint. The platform
109/// renders those as anonymous members until a frame carries the full key;
110/// this record lets it upgrade the rows it already stored, keyed by
111/// conversation and hint.
112#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
113pub struct MobileChatSenderResolutionRecord {
114    pub conversation_address: String,
115    pub sender_hint: Vec<u8>,
116    pub sender_address: String,
117}
118
119#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
120pub struct MobileChatMutationRecord {
121    /// A facade-session namespace prevents the engine's process-local u32
122    /// handles from colliding after restart.
123    pub session_id: u64,
124    pub handle: u32,
125    pub revision: u32,
126    pub kind: MobileChatMutationKind,
127    pub conversation_address: Option<String>,
128    pub sender_address: Option<String>,
129    /// The sender's 3-byte claimed hint, for a channel group message. The
130    /// only sender identity a multicast frame is required to carry; present
131    /// whether or not `sender_address` could be resolved.
132    pub sender_hint: Option<Vec<u8>>,
133    pub direction: Option<MobileChatDirection>,
134    pub message_type: Option<u8>,
135    pub wire_id: Option<u8>,
136    pub epoch: Option<u16>,
137    pub client_token: Option<u32>,
138    pub sender_handle: Option<String>,
139    pub regarding_handle: Option<u32>,
140    /// When a reply or emote references a message the engine holds no live
141    /// handle for, these export the wire reference the same way the
142    /// `original_*` fields do for edits: the target's wire ID within
143    /// `regarding_direction`'s stream, plus the hint identifying which
144    /// member's stream in a channel group.
145    pub regarding_wire_id: Option<u8>,
146    pub regarding_direction: Option<MobileChatDirection>,
147    pub regarding_sender_hint: Option<Vec<u8>>,
148    pub background_color: Option<Vec<u8>>,
149    pub text_color: Option<Vec<u8>>,
150    pub original_handle: Option<u32>,
151    /// When an edit/delete references a message the engine no longer holds a
152    /// live handle for (composed before a restart), these export the wire
153    /// reference so the platform can resolve it against persisted rows:
154    /// the original's wire ID within `original_direction`'s stream of the
155    /// record's `conversation_address` conversation.
156    pub original_wire_id: Option<u8>,
157    pub original_direction: Option<MobileChatDirection>,
158    /// The claimed hint of the original's sender, for a channel group
159    /// conversation. Inbound group streams are per-member, so matching a wire
160    /// reference there needs the hint as well as the ID and direction.
161    pub original_sender_hint: Option<Vec<u8>>,
162    pub body: Option<String>,
163    pub complete: Option<bool>,
164    pub present_fragments: Option<u16>,
165    pub fragment_count: Option<u8>,
166    pub finalized: Option<bool>,
167    /// Ordered-slot presence for `Insert` records (spinner placeholder, real
168    /// message, or loss marker). `UpdateBody`/`Edit`/`Delete` leave it at
169    /// `Present`; the host does not reinterpret presence on those.
170    pub presence: MobileChatPresence,
171    /// The record fills a slot reserved earlier by a gap, so it arrived out of
172    /// order and should render a "received late" caption.
173    pub received_late: bool,
174    /// The host should raise a user notification for this record (single-frame
175    /// arrival, fragment completion, or notify deadline; never placeholders).
176    ///
177    /// Whether a notification is actually shown remains the host's decision —
178    /// a muted conversation still produces records with this set, and still
179    /// counts as unread.
180    pub notify: bool,
181    /// Radio metadata for the frame that produced this record, when one did.
182    pub rx: Option<MobileChatRxMetadataRecord>,
183}
184
185/// Platform-persisted identity of a previously composed outbound message,
186/// used to target an edit or delete. `session_id`/`handle` identify it when
187/// composed by the current facade session; `wire_id`/`epoch` are the durable
188/// fallback for messages composed before a restart.
189#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
190pub struct MobileChatOriginalRef {
191    pub session_id: u64,
192    pub handle: u32,
193    pub wire_id: Option<u8>,
194    pub epoch: Option<u16>,
195}
196
197/// Platform-persisted identity of the message a reply or emote is about.
198///
199/// Unlike [`MobileChatOriginalRef`], which only ever names a message we
200/// composed, this can name either party's: `direction` says whose stream the
201/// wire ID belongs to, and `sender_hint` says which member's within a channel
202/// group. Reacting to a message read before the app last launched is the
203/// ordinary case, so the wire fields carry the weight here.
204#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
205pub struct MobileChatRegardingRef {
206    pub session_id: u64,
207    pub handle: u32,
208    pub wire_id: Option<u8>,
209    pub direction: Option<MobileChatDirection>,
210    pub sender_hint: Option<Vec<u8>>,
211    pub epoch: Option<u16>,
212}
213
214#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Enum)]
215pub enum MobileChatDeliveryState {
216    Sent,
217    Acknowledged,
218    Failed,
219}
220
221#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
222pub struct MobileChatDeliveryRecord {
223    pub session_id: u64,
224    pub handle: u32,
225    pub fragment_index: Option<u8>,
226    pub state: MobileChatDeliveryState,
227}
228
229#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
230pub struct MobileChatArchiveLookupRecord {
231    pub request_id: u32,
232    pub conversation_address: String,
233    pub message_id: u8,
234    pub fragment_index: Option<u8>,
235}
236
237#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Enum)]
238pub enum MobileChatArchiveResultKind {
239    Found,
240    Deleted,
241    Evicted,
242    Unknown,
243}
244
245#[derive(Clone, Debug, PartialEq, Eq, uniffi::Record)]
246pub struct MobileChatComposeBatchRecord {
247    pub batch_id: u64,
248    pub checkpoint: MobileChatCheckpointRecord,
249    /// Applied before `archives`: archive retirements for superseded
250    /// (edited or deleted) message IDs.
251    pub archive_deletes: Vec<MobileChatArchiveDeleteRecord>,
252    /// These exact payloads must be committed with the checkpoint before the
253    /// batch is released to the radio.
254    pub archives: Vec<MobileChatArchiveRecord>,
255    pub mutations: Vec<MobileChatMutationRecord>,
256}
257
258pub(crate) struct PendingChatBatch {
259    pub transmissions: Vec<Transmission>,
260}
261
262pub(crate) struct ComposedChatBatch {
263    pub record: MobileChatComposeBatchRecord,
264    pub deliveries: Vec<MobileChatDeliveryRecord>,
265    pub diagnostics: Vec<String>,
266}
267
268pub(crate) struct ChatDrain {
269    pub checkpoint: Option<MobileChatCheckpointRecord>,
270    pub transmissions: Vec<Transmission>,
271    pub archive_deletes: Vec<MobileChatArchiveDeleteRecord>,
272    pub archives: Vec<MobileChatArchiveRecord>,
273    pub mutations: Vec<MobileChatMutationRecord>,
274    pub deliveries: Vec<MobileChatDeliveryRecord>,
275    pub lookups: Vec<MobileChatArchiveLookupRecord>,
276    pub resolutions: Vec<MobileChatSenderResolutionRecord>,
277    pub diagnostics: Vec<String>,
278}
279
280impl ChatDrain {
281    fn new() -> Self {
282        Self {
283            checkpoint: None,
284            transmissions: Vec::new(),
285            archive_deletes: Vec::new(),
286            archives: Vec::new(),
287            mutations: Vec::new(),
288            deliveries: Vec::new(),
289            lookups: Vec::new(),
290            resolutions: Vec::new(),
291            diagnostics: Vec::new(),
292        }
293    }
294}
295
296/// The channels this session can hold a conversation in, and the keys to
297/// reach them.
298///
299/// Channel membership already lives in the MAC, which resolves an inbound
300/// frame to a key by authenticating it. This is the chat layer's own view:
301/// it maps between the tag a conversation is keyed by and the key a send
302/// needs, in both directions.
303#[derive(Default)]
304pub(crate) struct ChannelRegistry {
305    entries: BTreeMap<ChannelTag, ChannelKey>,
306}
307
308impl ChannelRegistry {
309    pub fn register(&mut self, tag: ChannelTag, key: ChannelKey) {
310        self.entries.insert(tag, key);
311    }
312
313    pub fn remove(&mut self, tag: &ChannelTag) {
314        self.entries.remove(tag);
315    }
316
317    pub fn key(&self, tag: &ChannelTag) -> Option<ChannelKey> {
318        self.entries.get(tag).copied()
319    }
320
321    pub fn contains(&self, tag: &ChannelTag) -> bool {
322        self.entries.contains_key(tag)
323    }
324}
325
326pub(crate) struct MobileChatState {
327    /// Keep the reducer off the worker future's stack. The mobile MAC/host is
328    /// already a large bounded value, and combining both inline can exhaust a
329    /// debug-build thread stack.
330    pub engine: Box<ChatEngine>,
331    pub session_id: u64,
332    next_batch_id: u64,
333    pub pending_batches: BTreeMap<u64, PendingChatBatch>,
334    /// Shared with the worker so a batch rejection — which rebuilds the
335    /// reducer — cannot lose the channels the platform registered.
336    pub channels: Rc<RefCell<ChannelRegistry>>,
337    /// Full keys learned for claimed member hints, per channel. A hint is only
338    /// 3 bytes and two members could in principle claim the same one, so this
339    /// records the first key seen for a hint and leaves it there.
340    resolved_members: BTreeMap<(ChannelTag, [u8; 3]), PublicKey>,
341    /// Hints already reported to the platform, so one resolution is announced
342    /// once rather than on every frame.
343    announced_members: BTreeSet<(ChannelTag, [u8; 3])>,
344}
345
346impl MobileChatState {
347    pub fn new(local_key: PublicKey, channels: Rc<RefCell<ChannelRegistry>>) -> Self {
348        Self {
349            engine: Box::new(ChatEngine::new(
350                DirectChannelProfile,
351                local_key,
352                EngineConfig::default(),
353                rand::rng().next_u64(),
354            )),
355            session_id: rand::rng().next_u64().max(1),
356            next_batch_id: 1,
357            pending_batches: BTreeMap::new(),
358            channels,
359            resolved_members: BTreeMap::new(),
360            announced_members: BTreeSet::new(),
361        }
362    }
363
364    pub fn restore(
365        &mut self,
366        checkpoints: &[MobileChatCheckpointRecord],
367        now_ms: u64,
368    ) -> Vec<String> {
369        let mut diagnostics = Vec::new();
370        let mut restored = Vec::new();
371        for record in checkpoints {
372            match self.checkpoint_from_record(record) {
373                Some(checkpoint) => restored.push(checkpoint),
374                None => diagnostics.push(format!(
375                    "chat checkpoint for unknown conversation {}",
376                    record.conversation_address
377                )),
378            }
379        }
380        self.engine.restore(&restored, now_ms);
381        let _ = self.drain();
382        diagnostics
383    }
384
385    /// Note a channel member's full key, learned from a frame that carried
386    /// both it and the claimed hint. Returns a record the first time a hint
387    /// resolves, so the platform can upgrade rows it stored anonymously.
388    pub fn resolve_member(
389        &mut self,
390        channel: ChannelTag,
391        hint: NodeHint,
392        key: PublicKey,
393    ) -> Option<MobileChatSenderResolutionRecord> {
394        self.resolved_members
395            .entry((channel, hint.0))
396            .or_insert(key);
397        if !self.announced_members.insert((channel, hint.0)) {
398            return None;
399        }
400        Some(MobileChatSenderResolutionRecord {
401            conversation_address: channel_address(channel),
402            sender_hint: hint.0.to_vec(),
403            sender_address: address(key),
404        })
405    }
406
407    pub fn compose_text(
408        &mut self,
409        conversation: ConversationKey,
410        client_token: u32,
411        body: &str,
412        now_ms: u64,
413    ) -> Result<ComposedChatBatch, ()> {
414        self.compose_batch(
415            conversation,
416            client_token,
417            ComposeIntent::Text {
418                body,
419                status: false,
420            },
421            now_ms,
422        )
423    }
424
425    pub fn compose_edit(
426        &mut self,
427        conversation: ConversationKey,
428        client_token: u32,
429        original: &MobileChatOriginalRef,
430        body: &str,
431        now_ms: u64,
432    ) -> Result<ComposedChatBatch, ()> {
433        let original = self.compose_ref(original).ok_or(())?;
434        self.compose_batch(
435            conversation,
436            client_token,
437            ComposeIntent::Edit { original, body },
438            now_ms,
439        )
440    }
441
442    pub fn compose_delete(
443        &mut self,
444        conversation: ConversationKey,
445        client_token: u32,
446        original: &MobileChatOriginalRef,
447        now_ms: u64,
448    ) -> Result<ComposedChatBatch, ()> {
449        let original = self.compose_ref(original).ok_or(())?;
450        self.compose_batch(
451            conversation,
452            client_token,
453            ComposeIntent::Delete { original },
454            now_ms,
455        )
456    }
457
458    /// Emote about a message: a reaction, or — with an empty body — the
459    /// withdrawal of one. Replacing a reaction is another emote, not an edit
460    /// of the previous one; the newest emote from a sender is the one that
461    /// counts.
462    pub fn compose_reaction(
463        &mut self,
464        conversation: ConversationKey,
465        client_token: u32,
466        target: &MobileChatRegardingRef,
467        body: &str,
468        now_ms: u64,
469    ) -> Result<ComposedChatBatch, ()> {
470        let regarding = self.regarding_ref(target).ok_or(())?;
471        self.compose_batch(
472            conversation,
473            client_token,
474            ComposeIntent::Reply {
475                body,
476                regarding,
477                status: true,
478            },
479            now_ms,
480        )
481    }
482
483    /// Resolve the platform's persisted identity of a regarding target to an
484    /// engine reference, on the same terms as [`Self::compose_ref`].
485    fn regarding_ref(&self, target: &MobileChatRegardingRef) -> Option<RegardingRef> {
486        if target.session_id == self.session_id {
487            return Some(RegardingRef::Handle(MessageHandle(target.handle)));
488        }
489        let sender_hint = match &target.sender_hint {
490            None => None,
491            Some(bytes) => Some(NodeHint(<[u8; 3]>::try_from(bytes.as_slice()).ok()?)),
492        };
493        Some(RegardingRef::Wire {
494            message_id: target.wire_id?,
495            direction: match target.direction? {
496                MobileChatDirection::Inbound => Direction::Inbound,
497                MobileChatDirection::Outbound => Direction::Outbound,
498            },
499            sender_hint,
500            epoch: target.epoch,
501        })
502    }
503
504    /// Resolve the platform's persisted identity of an original message to
505    /// an engine compose reference. Same facade session: the engine handle
506    /// is still live. Earlier session: fall back to the persisted wire
507    /// identity, which the engine validates against stream continuity.
508    fn compose_ref(&self, original: &MobileChatOriginalRef) -> Option<ComposeRef> {
509        if original.session_id == self.session_id {
510            return Some(ComposeRef::Handle(MessageHandle(original.handle)));
511        }
512        match (original.wire_id, original.epoch) {
513            (Some(message_id), Some(epoch)) => Some(ComposeRef::Wire { message_id, epoch }),
514            _ => None,
515        }
516    }
517
518    fn compose_batch(
519        &mut self,
520        conversation: ConversationKey,
521        client_token: u32,
522        intent: ComposeIntent<'_>,
523        now_ms: u64,
524    ) -> Result<ComposedChatBatch, ()> {
525        self.engine
526            .compose(conversation, client_token, intent, now_ms)
527            .map_err(|_| ())?;
528        let mut drain = self.drain();
529        let checkpoint = drain.checkpoint.ok_or(())?;
530        let fragment_count = u8::try_from(drain.archives.len()).map_err(|_| ())?;
531        // What is actually going on the air. For an original message it
532        // matches the archive count, but an edit's frames carry the editing
533        // option while its replacement archive is re-encoded without it, so
534        // the two can fragment differently.
535        let transmission_count = u8::try_from(drain.transmissions.len()).map_err(|_| ())?;
536        for mutation in &mut drain.mutations {
537            match mutation.kind {
538                MobileChatMutationKind::Insert
539                    if mutation.direction == Some(MobileChatDirection::Outbound) =>
540                {
541                    mutation.fragment_count = Some(fragment_count.max(1));
542                }
543                // Stamped so the platform can re-key the edited message's
544                // delivery tracking to the edit's transmissions and still
545                // know how many acknowledgments make it delivered.
546                MobileChatMutationKind::Edit => {
547                    mutation.fragment_count = Some(transmission_count.max(1));
548                }
549                _ => {}
550            }
551        }
552        let batch_id = self.next_batch_id;
553        self.next_batch_id = self.next_batch_id.wrapping_add(1).max(1);
554        self.pending_batches.insert(
555            batch_id,
556            PendingChatBatch {
557                transmissions: drain.transmissions,
558            },
559        );
560        Ok(ComposedChatBatch {
561            record: MobileChatComposeBatchRecord {
562                batch_id,
563                checkpoint,
564                archive_deletes: drain.archive_deletes,
565                archives: drain.archives,
566                mutations: drain.mutations,
567            },
568            deliveries: drain.deliveries,
569            diagnostics: drain.diagnostics,
570        })
571    }
572
573    pub fn drain(&mut self) -> ChatDrain {
574        let mut drained = ChatDrain::new();
575        while let Some(output) = self.engine.poll_output() {
576            match output {
577                Output::Transmit(transmission) => {
578                    if let Some(archive) = transmission.archive {
579                        if let Some(record) =
580                            self.archive_record(archive, transmission.payload.as_slice())
581                        {
582                            drained.archives.push(record);
583                        }
584                    }
585                    drained.transmissions.push(transmission);
586                }
587                Output::StoreCheckpoint {
588                    conversation,
589                    next_id,
590                    epoch,
591                } => {
592                    drained.checkpoint =
593                        self.conversation_address(conversation)
594                            .map(|conversation_address| MobileChatCheckpointRecord {
595                                conversation_address,
596                                next_id,
597                                epoch,
598                            });
599                }
600                Output::LookupOutbound {
601                    request_id,
602                    conversation,
603                    sequence,
604                } => {
605                    if let Some(conversation_address) = self.conversation_address(conversation) {
606                        drained.lookups.push(MobileChatArchiveLookupRecord {
607                            request_id,
608                            conversation_address,
609                            message_id: sequence.message_id,
610                            fragment_index: sequence.fragment.map(|fragment| fragment.index),
611                        });
612                    }
613                }
614                Output::StoreArchive { key, payload } => {
615                    if let Some(record) = self.archive_record(key, payload.as_slice()) {
616                        drained.archives.push(record);
617                    }
618                }
619                Output::DeleteArchive {
620                    conversation,
621                    message_id,
622                } => {
623                    if let Some(conversation_address) = self.conversation_address(conversation) {
624                        drained.archive_deletes.push(MobileChatArchiveDeleteRecord {
625                            conversation_address,
626                            message_id,
627                        });
628                    }
629                }
630                Output::StoreMessage(mutation) => {
631                    if let Some(record) = self.mutation_record(mutation) {
632                        drained.mutations.push(record);
633                    }
634                }
635                Output::Event(Event::DeliveryStateChanged {
636                    handle,
637                    fragment,
638                    state,
639                }) => drained.deliveries.push(MobileChatDeliveryRecord {
640                    session_id: self.session_id,
641                    handle: handle.0,
642                    fragment_index: fragment,
643                    state: match state {
644                        DeliveryState::Sent => MobileChatDeliveryState::Sent,
645                        DeliveryState::Acked => MobileChatDeliveryState::Acknowledged,
646                        DeliveryState::Failed => MobileChatDeliveryState::Failed,
647                    },
648                }),
649                Output::Event(event) => drained.diagnostics.push(format!("{event:?}")),
650                Output::Diagnostic(diagnostic) => {
651                    drained.diagnostics.push(format!("{diagnostic:?}"));
652                }
653            }
654        }
655        drained
656    }
657
658    fn mutation_record(&self, mutation: MessageMutation) -> Option<MobileChatMutationRecord> {
659        let mut record = MobileChatMutationRecord {
660            session_id: self.session_id,
661            handle: mutation.handle.0,
662            revision: mutation.revision,
663            kind: MobileChatMutationKind::Insert,
664            conversation_address: None,
665            sender_address: None,
666            sender_hint: None,
667            direction: None,
668            message_type: None,
669            wire_id: None,
670            epoch: None,
671            client_token: None,
672            sender_handle: None,
673            regarding_handle: None,
674            regarding_wire_id: None,
675            regarding_direction: None,
676            regarding_sender_hint: None,
677            background_color: None,
678            text_color: None,
679            original_handle: None,
680            original_wire_id: None,
681            original_direction: None,
682            original_sender_hint: None,
683            body: None,
684            complete: None,
685            present_fragments: None,
686            fragment_count: None,
687            finalized: None,
688            presence: MobileChatPresence::Present,
689            received_late: false,
690            notify: false,
691            rx: None,
692        };
693        match mutation.kind {
694            MutationKind::Insert {
695                conversation,
696                sender,
697                direction,
698                message_type,
699                wire_id,
700                epoch,
701                client_token,
702                sender_handle,
703                regarding,
704                bg_color,
705                text_color,
706                body,
707                status,
708                presence,
709                late,
710                notify,
711            } => {
712                record.conversation_address = self.conversation_address(conversation);
713                record.sender_address = self.sender_address(conversation, sender);
714                record.sender_hint = claimed_hint(sender);
715                record.direction = Some(match direction {
716                    Direction::Inbound => MobileChatDirection::Inbound,
717                    Direction::Outbound => MobileChatDirection::Outbound,
718                });
719                record.message_type = Some(message_type.to_byte());
720                record.wire_id = wire_id;
721                record.epoch = Some(epoch);
722                record.client_token = client_token;
723                record.sender_handle =
724                    sender_handle.map(|value| self.engine.body(&value).to_owned());
725                if let Some(regarding) = regarding {
726                    apply_regarding(&mut record, regarding);
727                }
728                record.background_color = bg_color.map(|color| color.to_vec());
729                record.text_color = text_color.map(|color| color.to_vec());
730                record.body = Some(self.engine.body(&body).to_owned());
731                record.presence = mobile_presence(presence);
732                record.received_late = late;
733                record.notify = notify;
734                apply_completion(&mut record, status);
735            }
736            MutationKind::UpdateBody {
737                body,
738                status,
739                late,
740                notify,
741            } => {
742                record.kind = MobileChatMutationKind::UpdateBody;
743                record.body = Some(self.engine.body(&body).to_owned());
744                record.received_late = late;
745                record.notify = notify;
746                apply_completion(&mut record, status);
747            }
748            MutationKind::Edit {
749                conversation,
750                original,
751                body,
752            } => {
753                record.kind = MobileChatMutationKind::Edit;
754                record.conversation_address = self.conversation_address(conversation);
755                apply_original(&mut record, original);
756                record.body = Some(self.engine.body(&body).to_owned());
757            }
758            MutationKind::Delete {
759                conversation,
760                original,
761            } => {
762                record.kind = MobileChatMutationKind::Delete;
763                record.conversation_address = self.conversation_address(conversation);
764                apply_original(&mut record, original);
765            }
766        }
767        Some(record)
768    }
769
770    /// The platform-facing address of a conversation: a peer's base58 address
771    /// for a direct one, a prefixed tag for a channel. Rooms have no address
772    /// in this facade.
773    fn conversation_address(&self, conversation: ConversationKey) -> Option<String> {
774        match conversation {
775            ConversationKey::Direct { peer } => Some(address(peer)),
776            ConversationKey::ChannelGroup { channel } => Some(channel_address(channel)),
777            ConversationKey::ChannelDirect { channel, peer } => Some(format!(
778                "{CHANNEL_DIRECT_ADDRESS_PREFIX}{}:{}",
779                hex(&channel.0),
780                address(peer)
781            )),
782            ConversationKey::Room { .. } => None,
783        }
784    }
785
786    /// The sender's full address when one is known. A multicast member claims
787    /// only a hint, so this is whatever key that hint has resolved to.
788    fn sender_address(&self, conversation: ConversationKey, sender: SenderScope) -> Option<String> {
789        match sender {
790            SenderScope::Peer(peer) => Some(address(peer)),
791            SenderScope::Local => None,
792            SenderScope::ClaimedMember(hint) => {
793                let ConversationKey::ChannelGroup { channel } = conversation else {
794                    return None;
795                };
796                self.resolved_members
797                    .get(&(channel, hint.0))
798                    .map(|peer| address(*peer))
799            }
800        }
801    }
802
803    fn archive_record(&self, key: ArchiveKey, payload: &[u8]) -> Option<MobileChatArchiveRecord> {
804        Some(MobileChatArchiveRecord {
805            conversation_address: self.conversation_address(key.conversation)?,
806            message_id: key.message_id,
807            fragment_index: key.fragment,
808            payload: payload.to_vec(),
809        })
810    }
811
812    fn checkpoint_from_record(
813        &self,
814        record: &MobileChatCheckpointRecord,
815    ) -> Option<StreamCheckpoint> {
816        let conversation = self.parse_conversation_address(&record.conversation_address)?;
817        Some(StreamCheckpoint {
818            conversation,
819            next_id: record.next_id,
820            epoch: record.epoch,
821        })
822    }
823
824    /// Resolve a stored address back to a conversation. A channel address only
825    /// resolves while its key is registered — the platform registers channels
826    /// before restoring chat, so an unresolvable one means the channel was
827    /// left.
828    pub fn parse_conversation_address(&self, value: &str) -> Option<ConversationKey> {
829        if let Some(rest) = value.strip_prefix(CHANNEL_DIRECT_ADDRESS_PREFIX) {
830            let (tag, peer) = rest.split_once(':')?;
831            let channel = parse_channel_tag(tag)?;
832            if !self.channels.borrow().contains(&channel) {
833                return None;
834            }
835            return Some(ConversationKey::ChannelDirect {
836                channel,
837                peer: decode_address(peer)?,
838            });
839        }
840        if let Some(tag) = value.strip_prefix(CHANNEL_ADDRESS_PREFIX) {
841            let channel = parse_channel_tag(tag)?;
842            if !self.channels.borrow().contains(&channel) {
843                return None;
844            }
845            return Some(ConversationKey::ChannelGroup { channel });
846        }
847        Some(ConversationKey::Direct {
848            peer: decode_address(value)?,
849        })
850    }
851}
852
853fn mobile_presence(presence: Presence) -> MobileChatPresence {
854    match presence {
855        Presence::Present => MobileChatPresence::Present,
856        Presence::GapPending => MobileChatPresence::GapPending,
857        Presence::Unavailable => MobileChatPresence::Unavailable,
858    }
859}
860
861fn apply_completion(record: &mut MobileChatMutationRecord, status: CompletionStatus) {
862    match status {
863        CompletionStatus::Complete => record.complete = Some(true),
864        CompletionStatus::Partial {
865            present,
866            count,
867            finalized,
868        } => {
869            record.complete = Some(false);
870            record.present_fragments = Some(present);
871            record.fragment_count = Some(count);
872            record.finalized = Some(finalized);
873        }
874    }
875}
876
877/// Export an edit/delete target: a live handle when resolved, otherwise the
878/// wire reference for the platform to match against its persisted rows.
879/// Room-scoped reference forms are outside this facade.
880fn apply_original(record: &mut MobileChatMutationRecord, reference: ResolvedRef) {
881    match reference {
882        ResolvedRef::Handle(MessageHandle(handle)) => {
883            record.original_handle = Some(handle);
884        }
885        ResolvedRef::Unresolved(WireRef::SenderScoped { sender, message_id }) => {
886            record.original_wire_id = Some(message_id);
887            record.original_direction = Some(wire_ref_direction(sender));
888            record.original_sender_hint = claimed_hint(sender);
889        }
890        ResolvedRef::Unresolved(WireRef::RoomCanonical { .. }) => {}
891    }
892}
893
894/// Export a reply/emote target, on the same terms as [`apply_original`].
895/// Reactions overwhelmingly target messages received long before this
896/// process started, so the unresolved form is the common case here rather
897/// than the exception.
898fn apply_regarding(record: &mut MobileChatMutationRecord, reference: ResolvedRef) {
899    match reference {
900        ResolvedRef::Handle(MessageHandle(handle)) => {
901            record.regarding_handle = Some(handle);
902        }
903        ResolvedRef::Unresolved(WireRef::SenderScoped { sender, message_id }) => {
904            record.regarding_wire_id = Some(message_id);
905            record.regarding_direction = Some(wire_ref_direction(sender));
906            record.regarding_sender_hint = claimed_hint(sender);
907        }
908        ResolvedRef::Unresolved(WireRef::RoomCanonical { .. }) => {}
909    }
910}
911
912fn wire_ref_direction(sender: SenderScope) -> MobileChatDirection {
913    match sender {
914        SenderScope::Local => MobileChatDirection::Outbound,
915        // A channel member's message is inbound like any other peer's; the
916        // hint is what tells the platform whose stream it was.
917        SenderScope::Peer(_) | SenderScope::ClaimedMember(_) => MobileChatDirection::Inbound,
918    }
919}
920
921fn claimed_hint(sender: SenderScope) -> Option<Vec<u8>> {
922    match sender {
923        SenderScope::ClaimedMember(hint) => Some(hint.0.to_vec()),
924        SenderScope::Local | SenderScope::Peer(_) => None,
925    }
926}
927
928/// The conversation address of a channel's group conversation.
929pub(crate) fn channel_address(channel: ChannelTag) -> String {
930    format!("{CHANNEL_ADDRESS_PREFIX}{}", hex(&channel.0))
931}
932
933fn parse_channel_tag(value: &str) -> Option<ChannelTag> {
934    if value.len() != 32 {
935        return None;
936    }
937    let mut bytes = [0u8; 16];
938    for (index, byte) in bytes.iter_mut().enumerate() {
939        *byte = u8::from_str_radix(value.get(index * 2..index * 2 + 2)?, 16).ok()?;
940    }
941    Some(ChannelTag(bytes))
942}
943
944fn hex(bytes: &[u8]) -> String {
945    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
946}
947
948fn address(key: PublicKey) -> String {
949    umsh_core::base58::encode(&key.0)
950        .into_iter()
951        .map(char::from)
952        .collect()
953}
954
955fn decode_address(value: &str) -> Option<PublicKey> {
956    umsh_core::base58::decode(value.as_bytes())
957        .ok()
958        .map(PublicKey)
959}
960
961#[cfg(test)]
962mod tests {
963    use super::*;
964
965    const LOCAL: PublicKey = PublicKey([0xAA; 32]);
966    const PEER: PublicKey = PublicKey([0x11; 32]);
967    const CHANNEL_KEY: ChannelKey = ChannelKey([0x42; 32]);
968
969    fn direct() -> ConversationKey {
970        ConversationKey::Direct { peer: PEER }
971    }
972
973    /// A state whose registry already holds the test channel.
974    fn state_with_channel() -> (MobileChatState, ChannelTag) {
975        let tag = crate::channel_tag(&CHANNEL_KEY);
976        let registry = Rc::new(RefCell::new(ChannelRegistry::default()));
977        registry.borrow_mut().register(tag, CHANNEL_KEY);
978        (MobileChatState::new(LOCAL, registry), tag)
979    }
980
981    fn empty_state() -> MobileChatState {
982        MobileChatState::new(LOCAL, Rc::new(RefCell::new(ChannelRegistry::default())))
983    }
984
985    /// A channel conversation survives a facade restart: its checkpoint is
986    /// addressed by tag, and restoring it resumes the same outbound stream
987    /// rather than starting a fresh one that would replay wire IDs.
988    #[test]
989    fn a_channel_checkpoint_round_trips_through_restore() {
990        let (mut first, tag) = state_with_channel();
991        let conversation = ConversationKey::ChannelGroup { channel: tag };
992        let composed = first
993            .compose_text(conversation, 5, "on my way", 0)
994            .expect("composing into a held channel succeeds");
995        let checkpoint = composed.record.checkpoint;
996        assert_eq!(checkpoint.conversation_address, channel_address(tag));
997
998        let (mut restarted, _) = state_with_channel();
999        assert!(
1000            restarted
1001                .restore(std::slice::from_ref(&checkpoint), 0)
1002                .is_empty(),
1003            "a checkpoint for a held channel restores without complaint"
1004        );
1005        // Continuity proves the restore landed on the same stream: a cold
1006        // engine would hand out the ID the first message already used.
1007        let next = restarted
1008            .compose_text(conversation, 6, "still moving", 1)
1009            .expect("composing after restore succeeds");
1010        assert_ne!(
1011            next.record.mutations[0].wire_id, composed.record.mutations[0].wire_id,
1012            "the restored stream must not reissue a spent wire ID"
1013        );
1014    }
1015
1016    /// A checkpoint whose channel this session no longer holds is reported
1017    /// rather than dropped in silence — the channel was left, and the stream
1018    /// it belonged to cannot be resumed without the key.
1019    #[test]
1020    fn a_checkpoint_for_an_unheld_channel_is_diagnosed() {
1021        let (mut held, tag) = state_with_channel();
1022        let checkpoint = held
1023            .compose_text(ConversationKey::ChannelGroup { channel: tag }, 1, "hi", 0)
1024            .expect("composing into a held channel succeeds")
1025            .record
1026            .checkpoint;
1027
1028        let diagnostics = empty_state().restore(std::slice::from_ref(&checkpoint), 0);
1029        assert_eq!(diagnostics.len(), 1, "{diagnostics:?}");
1030        assert!(
1031            diagnostics[0].contains(&checkpoint.conversation_address),
1032            "the diagnostic must name the conversation: {}",
1033            diagnostics[0]
1034        );
1035    }
1036
1037    /// The full restart round trip at the facade level: the persisted
1038    /// (wire_id, epoch) of a message composed by one facade session lets a
1039    /// fresh session — restored from the persisted checkpoint — compose an
1040    /// edit whose mutation record exports a platform-resolvable reference.
1041    #[test]
1042    fn edit_by_persisted_reference_after_facade_restart() {
1043        let mut first = empty_state();
1044        let composed = first
1045            .compose_text(direct(), 7, "v1", 0)
1046            .expect("compose succeeds");
1047        let insert = composed
1048            .record
1049            .mutations
1050            .iter()
1051            .find(|mutation| mutation.kind == MobileChatMutationKind::Insert)
1052            .expect("insert mutation");
1053        let original = MobileChatOriginalRef {
1054            session_id: insert.session_id,
1055            handle: insert.handle,
1056            wire_id: insert.wire_id,
1057            epoch: insert.epoch,
1058        };
1059        let checkpoint = composed.record.checkpoint;
1060
1061        let mut restarted = empty_state();
1062        assert_ne!(
1063            restarted.session_id, first.session_id,
1064            "sessions must not collide"
1065        );
1066        let _ = restarted.restore(std::slice::from_ref(&checkpoint), 0);
1067        let edited = restarted
1068            .compose_edit(direct(), 8, &original, "v2", 1)
1069            .expect("wire-referenced edit composes after restart");
1070        let edit = edited
1071            .record
1072            .mutations
1073            .iter()
1074            .find(|mutation| mutation.kind == MobileChatMutationKind::Edit)
1075            .expect("edit mutation");
1076        assert_eq!(edit.original_handle, None);
1077        assert_eq!(edit.original_wire_id, insert.wire_id);
1078        assert_eq!(edit.original_direction, Some(MobileChatDirection::Outbound));
1079        assert_eq!(edit.conversation_address, insert.conversation_address);
1080        assert_eq!(edit.body.as_deref(), Some("v2"));
1081
1082        // Superseded content is retired and re-issued under the original wire
1083        // ID: a resend request served from the archive can only carry "v2".
1084        assert!(
1085            edited
1086                .record
1087                .archive_deletes
1088                .iter()
1089                .any(|delete| Some(delete.message_id) == insert.wire_id)
1090        );
1091        assert!(
1092            edited
1093                .record
1094                .archives
1095                .iter()
1096                .any(|archive| Some(archive.message_id) == insert.wire_id)
1097        );
1098
1099        // Deleting retires the archive without replacing it.
1100        let deleted = restarted
1101            .compose_delete(direct(), 9, &original, 2)
1102            .expect("delete composes");
1103        assert!(
1104            deleted
1105                .record
1106                .archive_deletes
1107                .iter()
1108                .any(|delete| Some(delete.message_id) == insert.wire_id)
1109        );
1110        assert!(
1111            !deleted
1112                .record
1113                .archives
1114                .iter()
1115                .any(|archive| Some(archive.message_id) == insert.wire_id)
1116        );
1117
1118        // Without continuity (no restored checkpoint) the same reference is
1119        // rejected instead of silently starting a dangling edit.
1120        let mut cold = empty_state();
1121        assert!(cold.compose_delete(direct(), 9, &original, 0).is_err());
1122    }
1123
1124    /// The ordinary reaction: the target is a message the peer sent, and no
1125    /// live handle backs it, so the record must carry the wire coordinates
1126    /// for the platform to match against its own rows.
1127    #[test]
1128    fn compose_reaction_exports_wire_regarding() {
1129        let (mut state, tag) = state_with_channel();
1130        let conversation = ConversationKey::ChannelGroup { channel: tag };
1131        let hint = vec![0x33, 0x44, 0x55];
1132        let target = MobileChatRegardingRef {
1133            // A session that is not ours: the handle cannot be trusted.
1134            session_id: state.session_id.wrapping_add(1),
1135            handle: 0,
1136            wire_id: Some(9),
1137            direction: Some(MobileChatDirection::Inbound),
1138            sender_hint: Some(hint.clone()),
1139            epoch: Some(0),
1140        };
1141        let composed = state
1142            .compose_reaction(conversation, 1, &target, "+1", 0)
1143            .expect("reaction composes against a persisted target");
1144        let insert = composed
1145            .record
1146            .mutations
1147            .iter()
1148            .find(|mutation| mutation.kind == MobileChatMutationKind::Insert)
1149            .expect("insert mutation");
1150        assert_eq!(insert.message_type, Some(1), "an emote is status text");
1151        assert_eq!(insert.body.as_deref(), Some("+1"));
1152        assert_eq!(insert.regarding_handle, None);
1153        assert_eq!(insert.regarding_wire_id, Some(9));
1154        assert_eq!(
1155            insert.regarding_direction,
1156            Some(MobileChatDirection::Inbound)
1157        );
1158        assert_eq!(insert.regarding_sender_hint, Some(hint));
1159
1160        // Withdrawal is the same message with nothing in it — never an edit
1161        // or a delete, which would target the emote row instead.
1162        let withdrawn = state
1163            .compose_reaction(conversation, 2, &target, "", 1)
1164            .expect("withdrawal composes");
1165        let insert = withdrawn
1166            .record
1167            .mutations
1168            .iter()
1169            .find(|mutation| mutation.kind == MobileChatMutationKind::Insert)
1170            .expect("withdrawal is an insert");
1171        assert_eq!(insert.body.as_deref(), Some(""));
1172        assert_eq!(insert.regarding_wire_id, Some(9));
1173    }
1174
1175    /// Reacting to one of our own messages from an earlier launch: the handle
1176    /// belongs to a dead session, so the persisted wire identity carries it.
1177    #[test]
1178    fn reaction_target_from_prior_session_uses_wire_ref() {
1179        let mut first = empty_state();
1180        let composed = first
1181            .compose_text(direct(), 7, "mine", 0)
1182            .expect("compose succeeds");
1183        let insert = composed
1184            .record
1185            .mutations
1186            .iter()
1187            .find(|mutation| mutation.kind == MobileChatMutationKind::Insert)
1188            .expect("insert mutation");
1189        let target = MobileChatRegardingRef {
1190            session_id: insert.session_id,
1191            handle: insert.handle,
1192            wire_id: insert.wire_id,
1193            direction: Some(MobileChatDirection::Outbound),
1194            sender_hint: None,
1195            epoch: insert.epoch,
1196        };
1197        let checkpoint = composed.record.checkpoint;
1198
1199        let mut restarted = empty_state();
1200        let _ = restarted.restore(std::slice::from_ref(&checkpoint), 0);
1201        let reacted = restarted
1202            .compose_reaction(direct(), 8, &target, "<3", 1)
1203            .expect("wire-referenced reaction composes after restart");
1204        let emote = reacted
1205            .record
1206            .mutations
1207            .iter()
1208            .find(|mutation| mutation.kind == MobileChatMutationKind::Insert)
1209            .expect("insert mutation");
1210        assert_eq!(emote.regarding_handle, None);
1211        assert_eq!(emote.regarding_wire_id, insert.wire_id);
1212        assert_eq!(
1213            emote.regarding_direction,
1214            Some(MobileChatDirection::Outbound)
1215        );
1216
1217        // A cold session has no continuity, so the reference is refused
1218        // rather than aimed at whatever now holds that wire ID.
1219        let mut cold = empty_state();
1220        assert!(
1221            cold.compose_reaction(direct(), 9, &target, "<3", 0)
1222                .is_err()
1223        );
1224    }
1225}
umsh_mobile_core/mobile_chat.rs

umsh_mobile_core/
mobile_chat.rs
