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

umsh_mobile_core/
mobile_chat.rs
