im_core/messages/
dto.rs

1use serde::{Deserialize, Serialize};
2use std::collections::VecDeque;
3use std::sync::Arc;
4
5#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
6pub struct SendMessageRequest {
7    pub target: MessageTarget,
8    pub body: MessageBody,
9    pub security: MessageSecurityMode,
10    pub client_message_id: Option<crate::ids::MessageId>,
11    pub delivery: MessageDeliveryOptions,
12    #[serde(default, skip_serializing_if = "Option::is_none")]
13    pub delegated_signing: Option<DelegatedSigningOptions>,
14}
15
16#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
17pub struct DelegatedSigningOptions {
18    #[serde(default, skip_serializing_if = "Option::is_none")]
19    pub logical_sender_did: Option<String>,
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    pub signing_verification_method: Option<String>,
22    #[serde(default, skip_serializing_if = "Option::is_none")]
23    pub signing_key_ref: Option<String>,
24    #[serde(default, skip_serializing_if = "Option::is_none")]
25    pub actor_agent_did: Option<String>,
26}
27
28#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
29pub enum MessageTarget {
30    Direct(crate::ids::PeerRef),
31    Group(crate::ids::GroupRef),
32}
33
34#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
35pub enum MessageBody {
36    Text {
37        text: String,
38        kind: MessageKind,
39    },
40    Payload {
41        payload: serde_json::Value,
42    },
43    Attachment {
44        input: crate::attachments::AttachmentInput,
45        caption: Option<String>,
46        #[serde(default, skip_serializing_if = "Option::is_none")]
47        mention_payload: Option<serde_json::Value>,
48        mime_type: Option<String>,
49        filename: Option<String>,
50    },
51}
52
53#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
54pub enum MessageKind {
55    Text,
56    Markdown,
57}
58
59#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
60pub enum MessageSecurityMode {
61    DefaultPlain,
62    Plain,
63    E2eeRequired,
64    SecureDirect,
65    GroupE2ee,
66}
67
68pub type MessageSecurityPolicy = MessageSecurityMode;
69
70#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
71pub struct MessageDeliveryOptions {
72    pub idempotency_key: Option<String>,
73    pub wait_for_final_acceptance: bool,
74}
75
76#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
77pub struct SendMessageResult {
78    pub message: Message,
79    pub delivery: DeliveryState,
80    pub warnings: Vec<String>,
81}
82
83#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
84pub enum DeliveryState {
85    Accepted,
86    Sent,
87    StoredLocally,
88    Failed { reason: String },
89}
90
91#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
92pub struct Message {
93    pub id: crate::ids::MessageId,
94    pub thread: ThreadRef,
95    pub direction: MessageDirection,
96    pub sender: crate::ids::PeerRef,
97    pub receiver: Option<crate::ids::PeerRef>,
98    pub group: Option<crate::ids::GroupRef>,
99    pub body: MessageBodyView,
100    pub sent_at: Option<String>,
101    pub received_at: Option<String>,
102    pub metadata: MessageMetadata,
103}
104
105#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
106pub struct MessagePage {
107    pub items: Vec<Message>,
108    pub next_cursor: Option<crate::ids::Cursor>,
109    pub has_more: bool,
110    pub source: Option<String>,
111    pub resolved_dids: Vec<crate::ids::Did>,
112    pub warnings: Vec<String>,
113}
114
115impl MessagePage {
116    pub fn into_page(self) -> crate::ids::Page<Message> {
117        crate::ids::Page {
118            items: self.items,
119            next_cursor: self.next_cursor,
120            has_more: self.has_more,
121        }
122    }
123}
124
125#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
126pub enum MessageDirection {
127    Outgoing,
128    Incoming,
129    Unknown,
130}
131
132#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
133pub enum MessageBodyView {
134    Text { text: String, kind: MessageKind },
135    Payload { payload: serde_json::Value },
136    Unsupported { content_type: Option<String> },
137}
138
139#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
140pub struct MessageMetadata {
141    #[serde(default)]
142    pub operation_id: Option<String>,
143    #[serde(default)]
144    pub delivery_state: Option<String>,
145    #[serde(default)]
146    pub send_state: Option<MessageSendState>,
147    #[serde(default)]
148    pub retry_plan: Option<MessageRetryPlan>,
149    #[serde(default)]
150    pub server_sequence: Option<i64>,
151    #[serde(default)]
152    pub content_type: Option<String>,
153    #[serde(default, skip_serializing_if = "Option::is_none")]
154    pub conversation_identity: Option<ConversationIdentity>,
155    #[serde(default)]
156    pub attributes: Vec<MessageMetadataAttribute>,
157}
158
159#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
160pub struct ConversationIdentity {
161    pub conversation_id: String,
162    pub canonical_thread_kind: String,
163    pub canonical_thread_id: String,
164    pub storage_thread_ref: ConversationStorageThreadRef,
165    #[serde(default)]
166    pub aliases: Vec<ConversationAlias>,
167    pub identity_scope: ConversationIdentityScope,
168    pub migration_state: ConversationMigrationState,
169}
170
171#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
172pub struct ConversationStorageThreadRef {
173    pub kind: String,
174    pub id: String,
175}
176
177#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
178pub struct ConversationAlias {
179    pub kind: String,
180    pub id: String,
181    pub source: ConversationAliasSource,
182}
183
184#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
185#[serde(rename_all = "snake_case")]
186pub enum ConversationAliasSource {
187    LegacyDirectDid,
188    OldFlutterSortedDirect,
189    PeerScopeStorage,
190    GroupStorage,
191    ThreadStorage,
192    Unknown,
193}
194
195#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
196#[serde(rename_all = "snake_case")]
197pub enum ConversationIdentityScope {
198    Direct,
199    Group,
200    Thread,
201    Mail,
202    Unknown,
203}
204
205#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
206#[serde(rename_all = "snake_case")]
207pub enum ConversationMigrationState {
208    Canonical,
209    AliasResolved,
210    LegacyInput,
211    Unknown,
212}
213
214#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
215pub struct MessageSendState {
216    pub state: MessageSendStateKind,
217    pub operation_id: Option<String>,
218    pub message_id: Option<crate::ids::MessageId>,
219    pub reason: Option<String>,
220    pub updated_at: Option<String>,
221}
222
223#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
224#[serde(rename_all = "snake_case")]
225pub enum MessageSendStateKind {
226    Pending,
227    Accepted,
228    Sent,
229    StoredLocally,
230    Failed,
231}
232
233#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
234pub struct MessageRetryPlan {
235    pub retryable: bool,
236    pub action: MessageRetryAction,
237    pub operation_id: Option<String>,
238    pub message_id: Option<crate::ids::MessageId>,
239    pub reason: Option<String>,
240}
241
242#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
243#[serde(rename_all = "snake_case")]
244pub enum MessageRetryAction {
245    None,
246    RetryDirectText,
247    RetryGroupText,
248    RetryDirectPayload,
249    RetryGroupPayload,
250}
251
252#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
253pub struct MessageMetadataAttribute {
254    pub key: String,
255    pub value: String,
256}
257
258#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
259pub enum ThreadRef {
260    Direct(crate::ids::PeerRef),
261    Group(crate::ids::GroupRef),
262    Thread(crate::ids::ThreadId),
263}
264
265impl ConversationIdentity {
266    pub fn from_thread_ref(thread: &ThreadRef) -> Self {
267        let (kind, id) = thread_ref_parts(thread);
268        Self::from_storage_parts(kind, id)
269    }
270
271    pub fn from_thread_ref_for_owner(thread: &ThreadRef, owner_did: &str) -> Self {
272        let (kind, id) = thread_ref_parts(thread);
273        Self::from_storage_parts_for_owner(kind, id, owner_did)
274    }
275
276    pub fn from_storage_parts(kind: impl Into<String>, id: impl Into<String>) -> Self {
277        Self::from_storage_parts_with_owner(kind, id, None)
278    }
279
280    pub fn from_storage_parts_for_owner(
281        kind: impl Into<String>,
282        id: impl Into<String>,
283        owner_did: &str,
284    ) -> Self {
285        Self::from_storage_parts_with_owner(kind, id, Some(owner_did))
286    }
287
288    fn from_storage_parts_with_owner(
289        kind: impl Into<String>,
290        id: impl Into<String>,
291        owner_did: Option<&str>,
292    ) -> Self {
293        let kind = kind.into();
294        let id = id.into();
295        let conversation_id = canonical_conversation_id_for_storage_parts(&kind, &id, owner_did);
296        let identity_scope = identity_scope_for_kind_id(&kind, &id);
297        let migration_state =
298            migration_state_for_storage_parts(&kind, &id, &conversation_id, &identity_scope);
299        let aliases = aliases_for_storage_parts(&kind, &id, &conversation_id, &identity_scope);
300        Self {
301            conversation_id: conversation_id.clone(),
302            canonical_thread_kind: canonical_thread_kind(&identity_scope).to_owned(),
303            canonical_thread_id: conversation_id,
304            storage_thread_ref: ConversationStorageThreadRef { kind, id },
305            aliases,
306            identity_scope,
307            migration_state,
308        }
309    }
310}
311
312pub fn direct_peer_scope_thread_id(
313    user_id: impl AsRef<str>,
314    full_handle: impl AsRef<str>,
315) -> crate::ImResult<crate::ids::ThreadId> {
316    let scope = crate::internal::local_state::owner_scope::DirectPeerScope::new(
317        user_id.as_ref().to_owned(),
318        full_handle.as_ref().to_owned(),
319    )?;
320    crate::ids::ThreadId::parse(
321        crate::internal::local_state::owner_scope::direct_conversation_id_for_peer_scope(&scope),
322    )
323}
324
325#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
326pub struct InboxQuery {
327    pub scope: InboxScope,
328    pub limit: crate::ids::PageLimit,
329    pub cursor: Option<crate::ids::Cursor>,
330    pub unread_only: bool,
331    #[serde(default, skip_serializing_if = "Option::is_none")]
332    pub inbox_history_options: Option<InboxHistoryOptions>,
333}
334
335#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
336pub enum InboxScope {
337    All,
338    DirectOnly,
339    GroupOnly,
340}
341
342#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
343pub struct HistoryQuery {
344    pub limit: crate::ids::PageLimit,
345    pub cursor: Option<crate::ids::Cursor>,
346    #[serde(default, skip_serializing_if = "Option::is_none")]
347    pub inbox_history_options: Option<InboxHistoryOptions>,
348}
349
350#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
351pub struct LocalHistoryQuery {
352    pub limit: crate::ids::PageLimit,
353    pub cursor: Option<crate::ids::Cursor>,
354}
355
356#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
357pub struct InboxHistoryOptions {
358    #[serde(default, skip_serializing_if = "Option::is_none")]
359    pub inbox_owner_did: Option<String>,
360    #[serde(default, skip_serializing_if = "Option::is_none")]
361    pub inbox_auth_verification_method: Option<String>,
362    #[serde(default, skip_serializing_if = "Option::is_none")]
363    pub inbox_auth_key_ref: Option<String>,
364    #[serde(default, skip_serializing_if = "Option::is_none")]
365    pub inbox_auth: Option<InboxAuth>,
366}
367
368#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
369#[serde(tag = "kind", rename_all = "snake_case")]
370pub enum InboxAuth {
371    ScopedInboxToken { token: ScopedInboxToken },
372}
373
374#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
375pub struct ScopedInboxToken {
376    pub token: String,
377}
378
379#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
380pub struct MarkReadResult {
381    pub updated_count: u32,
382    pub message_ids: Vec<crate::ids::MessageId>,
383    pub warnings: Vec<String>,
384}
385
386#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
387pub struct MarkThreadReadRequest {
388    pub thread: ThreadRef,
389    #[serde(default, skip_serializing_if = "Option::is_none")]
390    pub watermark: Option<ReadWatermark>,
391    #[serde(default, skip_serializing_if = "Option::is_none")]
392    pub fallback_max_message_ids: Option<u32>,
393}
394
395#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
396pub struct MarkConversationReadRequest {
397    pub conversation: ConversationReadRef,
398    #[serde(default, skip_serializing_if = "Option::is_none")]
399    pub watermark: Option<ReadWatermark>,
400    #[serde(default, skip_serializing_if = "Option::is_none")]
401    pub fallback_max_message_ids: Option<u32>,
402}
403
404#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
405pub struct ReadWatermark {
406    #[serde(default, skip_serializing_if = "Option::is_none")]
407    pub last_read_message_id: Option<crate::ids::MessageId>,
408    #[serde(default, skip_serializing_if = "Option::is_none")]
409    pub last_read_thread_seq: Option<String>,
410    #[serde(default, skip_serializing_if = "Option::is_none")]
411    pub read_at: Option<chrono::DateTime<chrono::Utc>>,
412}
413
414#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
415pub struct MarkThreadReadResult {
416    pub updated_count: u32,
417    pub remote_acknowledged: bool,
418    pub partial: bool,
419    pub fallback_used: bool,
420    pub pending_remote_ack: bool,
421    #[serde(default, skip_serializing_if = "Option::is_none")]
422    pub effective_watermark: Option<ReadWatermark>,
423    pub legacy_message_ids: Vec<crate::ids::MessageId>,
424    pub warnings: Vec<String>,
425}
426
427#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
428pub struct ConversationReadRef {
429    pub conversation_id: String,
430}
431
432impl ConversationReadRef {
433    pub fn new(conversation_id: impl Into<String>) -> crate::ImResult<Self> {
434        let conversation_id = conversation_id.into();
435        let conversation_id = conversation_id.trim();
436        if conversation_id.is_empty() {
437            return Err(crate::ImError::invalid_input(
438                Some("conversation_id".to_owned()),
439                "conversation_id must not be empty",
440            ));
441        }
442        Ok(Self {
443            conversation_id: conversation_id.to_owned(),
444        })
445    }
446
447    pub fn as_thread_ref(&self) -> crate::ImResult<ThreadRef> {
448        crate::ids::ThreadId::parse(&self.conversation_id).map(ThreadRef::Thread)
449    }
450}
451
452#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
453pub struct SendConversationTextRequest {
454    pub conversation: ConversationReadRef,
455    pub text: String,
456    pub markdown: bool,
457    pub security: MessageSecurityMode,
458    pub client_message_id: Option<crate::ids::MessageId>,
459    pub idempotency_key: Option<String>,
460    pub wait_for_final_acceptance: bool,
461    #[serde(default, skip_serializing_if = "Option::is_none")]
462    pub delegated_signing: Option<DelegatedSigningOptions>,
463}
464
465#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
466pub struct SendConversationPayloadRequest {
467    pub conversation: ConversationReadRef,
468    pub payload: serde_json::Value,
469    pub security: MessageSecurityMode,
470    pub client_message_id: Option<crate::ids::MessageId>,
471    pub idempotency_key: Option<String>,
472    pub wait_for_final_acceptance: bool,
473    #[serde(default, skip_serializing_if = "Option::is_none")]
474    pub delegated_signing: Option<DelegatedSigningOptions>,
475}
476
477#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
478pub struct SyncThreadAfterRequest {
479    pub thread: ThreadRef,
480    #[serde(default, skip_serializing_if = "Option::is_none")]
481    pub after_server_seq: Option<String>,
482    #[serde(default, skip_serializing_if = "Option::is_none")]
483    pub limit: Option<u32>,
484}
485
486#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
487pub struct SyncConversationAfterRequest {
488    pub conversation: ConversationReadRef,
489    #[serde(default, skip_serializing_if = "Option::is_none")]
490    pub after_server_seq: Option<String>,
491    #[serde(default, skip_serializing_if = "Option::is_none")]
492    pub limit: Option<u32>,
493}
494
495#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
496pub struct SyncThreadAfterResult {
497    pub messages: Vec<Message>,
498    pub next_after_server_seq: Option<String>,
499    pub has_more: bool,
500    pub warnings: Vec<String>,
501}
502
503#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
504pub struct SyncDeltaRequest {
505    #[serde(default, skip_serializing_if = "Option::is_none")]
506    pub limit: Option<u32>,
507    #[serde(default, skip_serializing_if = "Option::is_none")]
508    pub device_id: Option<String>,
509    #[serde(default, skip_serializing_if = "Option::is_none")]
510    pub reason: Option<String>,
511}
512
513#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
514pub struct SyncDeltaResult {
515    pub events_applied: u32,
516    pub pages_fetched: u32,
517    pub last_applied_event_seq: Option<String>,
518    pub has_more: bool,
519    pub snapshot_required: bool,
520    #[serde(default, skip_serializing_if = "Option::is_none")]
521    pub retention_floor_event_seq: Option<String>,
522    pub warnings: Vec<String>,
523}
524
525#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
526pub struct ConversationListSnapshot {
527    pub format_version: u32,
528    pub im_schema_version: i64,
529    pub owner_identity_id: String,
530    pub owner_did: String,
531    pub generated_at_ms: i64,
532    #[serde(default, skip_serializing_if = "Option::is_none")]
533    pub summary_version: Option<String>,
534    pub unread_total: u32,
535    pub items: Vec<ConversationSnapshotItem>,
536}
537
538#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
539#[serde(tag = "type", rename_all = "snake_case")]
540pub enum ConversationStorePatch {
541    Reset {
542        owner_identity_id: String,
543        owner_did: String,
544        version: u64,
545        unread_total: u32,
546        items: Vec<ConversationSnapshotItem>,
547    },
548    Upsert {
549        owner_identity_id: String,
550        owner_did: String,
551        version: u64,
552        unread_total: u32,
553        item: ConversationSnapshotItem,
554        index: u32,
555    },
556    Remove {
557        owner_identity_id: String,
558        owner_did: String,
559        version: u64,
560        unread_total: u32,
561        thread_kind: String,
562        thread_id: String,
563        #[serde(default, skip_serializing_if = "Option::is_none")]
564        conversation_identity: Option<ConversationIdentity>,
565    },
566    Reorder {
567        owner_identity_id: String,
568        owner_did: String,
569        version: u64,
570        unread_total: u32,
571        thread_kind: String,
572        thread_id: String,
573        #[serde(default, skip_serializing_if = "Option::is_none")]
574        conversation_identity: Option<ConversationIdentity>,
575        index: u32,
576    },
577    RepairRequired {
578        owner_identity_id: String,
579        owner_did: String,
580        version: u64,
581        unread_total: u32,
582        reason: String,
583    },
584}
585
586#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
587#[serde(tag = "type", rename_all = "snake_case")]
588pub enum ThreadMessageStorePatch {
589    Reset {
590        owner_identity_id: String,
591        owner_did: String,
592        version: u64,
593        thread_kind: String,
594        thread_id: String,
595        #[serde(default, skip_serializing_if = "Option::is_none")]
596        conversation_identity: Option<ConversationIdentity>,
597        items: Vec<Message>,
598    },
599    Upsert {
600        owner_identity_id: String,
601        owner_did: String,
602        version: u64,
603        thread_kind: String,
604        thread_id: String,
605        #[serde(default, skip_serializing_if = "Option::is_none")]
606        conversation_identity: Option<ConversationIdentity>,
607        message: Message,
608        index: u32,
609    },
610    Remove {
611        owner_identity_id: String,
612        owner_did: String,
613        version: u64,
614        thread_kind: String,
615        thread_id: String,
616        #[serde(default, skip_serializing_if = "Option::is_none")]
617        conversation_identity: Option<ConversationIdentity>,
618        message_id: String,
619    },
620    RepairRequired {
621        owner_identity_id: String,
622        owner_did: String,
623        version: u64,
624        thread_kind: String,
625        thread_id: String,
626        #[serde(default, skip_serializing_if = "Option::is_none")]
627        conversation_identity: Option<ConversationIdentity>,
628        reason: String,
629    },
630}
631
632pub struct ConversationPatchSession {
633    pub(crate) store: Arc<crate::internal::runtime_store::conversation_store::ConversationStore>,
634    pub(crate) receiver: tokio::sync::broadcast::Receiver<ConversationStorePatch>,
635    pub(crate) initial: VecDeque<ConversationStorePatch>,
636    closed: bool,
637}
638
639impl ConversationPatchSession {
640    pub(crate) fn new(
641        store: Arc<crate::internal::runtime_store::conversation_store::ConversationStore>,
642        receiver: tokio::sync::broadcast::Receiver<ConversationStorePatch>,
643        initial: Vec<ConversationStorePatch>,
644    ) -> Self {
645        Self {
646            store,
647            receiver,
648            initial: initial.into(),
649            closed: false,
650        }
651    }
652
653    pub async fn next_patch(&mut self) -> Option<ConversationStorePatch> {
654        if self.closed {
655            return None;
656        }
657        if let Some(patch) = self.initial.pop_front() {
658            return Some(patch);
659        }
660        loop {
661            match self.receiver.recv().await {
662                Ok(patch) => return Some(patch),
663                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
664                    return Some(self.store.repair_required_patch("subscriber_lag"));
665                }
666                Err(tokio::sync::broadcast::error::RecvError::Closed) => return None,
667            }
668        }
669    }
670
671    pub async fn stop(&mut self) -> crate::ImResult<()> {
672        self.closed = true;
673        self.receiver.resubscribe();
674        Ok(())
675    }
676}
677
678pub struct ThreadMessagePatchSession {
679    pub(crate) store: Arc<crate::internal::runtime_store::message_store::MessageStore>,
680    pub(crate) receiver: tokio::sync::broadcast::Receiver<ThreadMessageStorePatch>,
681    pub(crate) initial: VecDeque<ThreadMessageStorePatch>,
682    pub(crate) thread: ThreadRef,
683    pub(crate) limit: u32,
684    closed: bool,
685}
686
687impl ThreadMessagePatchSession {
688    pub(crate) fn new(
689        store: Arc<crate::internal::runtime_store::message_store::MessageStore>,
690        receiver: tokio::sync::broadcast::Receiver<ThreadMessageStorePatch>,
691        initial: Vec<ThreadMessageStorePatch>,
692        thread: ThreadRef,
693        limit: u32,
694    ) -> Self {
695        Self {
696            store,
697            receiver,
698            initial: initial.into(),
699            thread,
700            limit,
701            closed: false,
702        }
703    }
704
705    pub async fn next_patch(&mut self) -> Option<ThreadMessageStorePatch> {
706        if self.closed {
707            return None;
708        }
709        if let Some(patch) = self.initial.pop_front() {
710            return Some(patch);
711        }
712        loop {
713            match self.receiver.recv().await {
714                Ok(patch) if thread_patch_matches(&patch, &self.thread) => return Some(patch),
715                Ok(_) => continue,
716                Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
717                    return Some(self.store.repair_required_patch(
718                        &self.thread,
719                        self.limit,
720                        "subscriber_lag",
721                    ));
722                }
723                Err(tokio::sync::broadcast::error::RecvError::Closed) => return None,
724            }
725        }
726    }
727
728    pub async fn stop(&mut self) -> crate::ImResult<()> {
729        self.closed = true;
730        self.receiver.resubscribe();
731        Ok(())
732    }
733}
734
735fn thread_patch_matches(patch: &ThreadMessageStorePatch, thread: &ThreadRef) -> bool {
736    let (expected_kind, expected_id) = thread_ref_parts(thread);
737    match patch {
738        ThreadMessageStorePatch::Reset {
739            thread_kind,
740            thread_id,
741            ..
742        }
743        | ThreadMessageStorePatch::Upsert {
744            thread_kind,
745            thread_id,
746            ..
747        }
748        | ThreadMessageStorePatch::Remove {
749            thread_kind,
750            thread_id,
751            ..
752        }
753        | ThreadMessageStorePatch::RepairRequired {
754            thread_kind,
755            thread_id,
756            ..
757        } => thread_kind == &expected_kind && thread_id == &expected_id,
758    }
759}
760
761pub(crate) fn thread_ref_parts(thread: &ThreadRef) -> (String, String) {
762    match thread {
763        ThreadRef::Direct(peer) => ("direct".to_owned(), peer.as_str().to_owned()),
764        ThreadRef::Group(group) => ("group".to_owned(), group.as_str().to_owned()),
765        ThreadRef::Thread(thread) => ("thread".to_owned(), thread.as_str().to_owned()),
766    }
767}
768
769fn canonical_conversation_id_for_storage_parts(
770    kind: &str,
771    id: &str,
772    owner_did: Option<&str>,
773) -> String {
774    match kind.trim() {
775        "direct" => crate::internal::local_state::owner_scope::direct_conversation_id(id),
776        "group" => crate::internal::local_state::owner_scope::group_conversation_id(id),
777        "thread" if id.trim().starts_with("dm:peer-scope:") => id.trim().to_owned(),
778        "thread" if id.trim().starts_with("dm:") => owner_did
779            .and_then(|owner_did| {
780                crate::internal::local_state::owner_scope::direct_conversation_id_from_thread_alias(
781                    id, owner_did,
782                )
783            })
784            .unwrap_or_else(|| id.trim().to_owned()),
785        "thread" if id.trim().starts_with("group:") => id.trim().to_owned(),
786        "thread" if id.trim().starts_with("mail:") => id.trim().to_owned(),
787        "thread" => id.trim().to_owned(),
788        _ if !id.trim().is_empty() => id.trim().to_owned(),
789        _ => "unknown".to_owned(),
790    }
791}
792
793fn identity_scope_for_kind_id(kind: &str, id: &str) -> ConversationIdentityScope {
794    match kind.trim() {
795        "direct" => ConversationIdentityScope::Direct,
796        "group" => ConversationIdentityScope::Group,
797        "thread" if id.trim().starts_with("dm:") => ConversationIdentityScope::Direct,
798        "thread" if id.trim().starts_with("group:") => ConversationIdentityScope::Group,
799        "thread" if id.trim().starts_with("mail:") => ConversationIdentityScope::Mail,
800        "thread" => ConversationIdentityScope::Thread,
801        _ => ConversationIdentityScope::Unknown,
802    }
803}
804
805fn canonical_thread_kind(scope: &ConversationIdentityScope) -> &'static str {
806    match scope {
807        ConversationIdentityScope::Direct => "direct",
808        ConversationIdentityScope::Group => "group",
809        ConversationIdentityScope::Mail => "mail",
810        ConversationIdentityScope::Thread => "thread",
811        ConversationIdentityScope::Unknown => "unknown",
812    }
813}
814
815fn migration_state_for_storage_parts(
816    kind: &str,
817    id: &str,
818    conversation_id: &str,
819    scope: &ConversationIdentityScope,
820) -> ConversationMigrationState {
821    match (kind.trim(), scope) {
822        ("direct", ConversationIdentityScope::Direct) => ConversationMigrationState::LegacyInput,
823        ("group", ConversationIdentityScope::Group) => ConversationMigrationState::LegacyInput,
824        ("thread", ConversationIdentityScope::Direct)
825            if id.trim().starts_with("dm:") && !id.trim().starts_with("dm:peer-scope:") =>
826        {
827            ConversationMigrationState::LegacyInput
828        }
829        ("thread", _) if id.trim() == conversation_id => ConversationMigrationState::Canonical,
830        ("thread", _) => ConversationMigrationState::AliasResolved,
831        _ => ConversationMigrationState::Unknown,
832    }
833}
834
835fn aliases_for_storage_parts(
836    kind: &str,
837    id: &str,
838    conversation_id: &str,
839    scope: &ConversationIdentityScope,
840) -> Vec<ConversationAlias> {
841    let source = match (kind.trim(), scope) {
842        ("direct", ConversationIdentityScope::Direct) => ConversationAliasSource::LegacyDirectDid,
843        ("group", ConversationIdentityScope::Group) => ConversationAliasSource::GroupStorage,
844        ("thread", ConversationIdentityScope::Direct)
845            if id.trim().starts_with("dm:peer-scope:") =>
846        {
847            ConversationAliasSource::PeerScopeStorage
848        }
849        ("thread", ConversationIdentityScope::Direct) if is_old_flutter_direct_alias(id) => {
850            ConversationAliasSource::OldFlutterSortedDirect
851        }
852        ("thread", ConversationIdentityScope::Direct) if id.trim().starts_with("dm:") => {
853            ConversationAliasSource::LegacyDirectDid
854        }
855        ("thread", _) => ConversationAliasSource::ThreadStorage,
856        _ => ConversationAliasSource::Unknown,
857    };
858    if id.trim() == conversation_id.trim() && source == ConversationAliasSource::PeerScopeStorage {
859        return Vec::new();
860    }
861    vec![ConversationAlias {
862        kind: kind.trim().to_owned(),
863        id: id.trim().to_owned(),
864        source,
865    }]
866}
867
868fn is_old_flutter_direct_alias(id: &str) -> bool {
869    id.trim()
870        .strip_prefix("dm:")
871        .map(|value| value.contains(":did:"))
872        .unwrap_or(false)
873}
874
875#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
876pub struct ConversationSnapshotItem {
877    pub thread_kind: String,
878    pub thread_id: String,
879    #[serde(default, skip_serializing_if = "Option::is_none")]
880    pub conversation_identity: Option<ConversationIdentity>,
881    pub participants: Vec<String>,
882    #[serde(default, skip_serializing_if = "Option::is_none")]
883    pub last_message: Option<ConversationSnapshotMessage>,
884    pub unread_count: u32,
885    #[serde(default)]
886    pub unread_mention_count: u32,
887    #[serde(default, skip_serializing_if = "Option::is_none")]
888    pub first_unread_mention_message_id: Option<String>,
889    pub message_count: u32,
890    #[serde(default, skip_serializing_if = "Option::is_none")]
891    pub last_message_at: Option<String>,
892}
893
894#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
895pub struct ConversationSnapshotMessage {
896    pub id: String,
897    pub thread_kind: String,
898    pub thread_id: String,
899    #[serde(default, skip_serializing_if = "Option::is_none")]
900    pub conversation_identity: Option<ConversationIdentity>,
901    pub direction: String,
902    pub sender: String,
903    #[serde(default, skip_serializing_if = "Option::is_none")]
904    pub receiver: Option<String>,
905    #[serde(default, skip_serializing_if = "Option::is_none")]
906    pub group: Option<String>,
907    pub body: ConversationSnapshotMessageBody,
908    #[serde(default, skip_serializing_if = "Option::is_none")]
909    pub sent_at: Option<String>,
910    #[serde(default, skip_serializing_if = "Option::is_none")]
911    pub received_at: Option<String>,
912    #[serde(default, skip_serializing_if = "Option::is_none")]
913    pub server_sequence: Option<i64>,
914    #[serde(default, skip_serializing_if = "Option::is_none")]
915    pub content_type: Option<String>,
916    #[serde(default)]
917    pub attributes: Vec<MessageMetadataAttribute>,
918}
919
920#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
921pub struct ConversationSnapshotMessageBody {
922    #[serde(default, skip_serializing_if = "Option::is_none")]
923    pub text: Option<String>,
924    #[serde(default, skip_serializing_if = "Option::is_none")]
925    pub kind: Option<String>,
926    #[serde(default, skip_serializing_if = "Option::is_none")]
927    pub payload_json: Option<String>,
928    #[serde(default, skip_serializing_if = "Option::is_none")]
929    pub unsupported_content_type: Option<String>,
930}
931
932#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
933pub struct Conversation {
934    pub thread: ThreadRef,
935    #[serde(default, skip_serializing_if = "Option::is_none")]
936    pub conversation_identity: Option<ConversationIdentity>,
937    pub title: Option<String>,
938    pub participants: Vec<crate::ids::PeerRef>,
939    pub last_message: Option<Message>,
940    pub unread_count: u32,
941    #[serde(default)]
942    pub unread_mention_count: u32,
943    #[serde(default, skip_serializing_if = "Option::is_none")]
944    pub first_unread_mention_message_id: Option<crate::ids::MessageId>,
945    pub message_count: u32,
946    pub last_message_at: Option<String>,
947}
948
949#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
950pub struct ConversationQuery {
951    pub limit: crate::ids::PageLimit,
952    #[serde(default, skip_serializing_if = "Option::is_none")]
953    pub cursor: Option<crate::ids::Cursor>,
954    pub include_groups: bool,
955    pub include_direct: bool,
956    pub unread_only: bool,
957}
958
959#[cfg(test)]
960mod conversation_identity_tests {
961    use super::*;
962
963    #[test]
964    fn conversation_identity_direct_ref_uses_ownerless_conversation_id() {
965        let peer = crate::ids::PeerRef::parse("did:example:bob", "").unwrap();
966        let identity = ConversationIdentity::from_thread_ref(&ThreadRef::Direct(peer));
967
968        assert_eq!(identity.conversation_id, "dm:did:example:bob");
969        assert_eq!(identity.canonical_thread_kind, "direct");
970        assert_eq!(identity.canonical_thread_id, "dm:did:example:bob");
971        assert_eq!(identity.storage_thread_ref.kind, "direct");
972        assert_eq!(identity.storage_thread_ref.id, "did:example:bob");
973        assert_eq!(identity.identity_scope, ConversationIdentityScope::Direct);
974        assert_eq!(
975            identity.migration_state,
976            ConversationMigrationState::LegacyInput
977        );
978        assert_eq!(identity.aliases.len(), 1);
979        assert_eq!(
980            identity.aliases[0].source,
981            ConversationAliasSource::LegacyDirectDid
982        );
983    }
984
985    #[test]
986    fn conversation_identity_peer_scope_thread_is_canonical_direct_storage() {
987        let thread_id =
988            crate::messages::direct_peer_scope_thread_id("user-1", "alice.anpclaw.com").unwrap();
989        let identity = ConversationIdentity::from_thread_ref(&ThreadRef::Thread(thread_id));
990
991        assert!(identity.conversation_id.starts_with("dm:peer-scope:v1:"));
992        assert_eq!(identity.canonical_thread_kind, "direct");
993        assert_eq!(identity.canonical_thread_id, identity.conversation_id);
994        assert_eq!(identity.storage_thread_ref.kind, "thread");
995        assert_eq!(identity.storage_thread_ref.id, identity.conversation_id);
996        assert_eq!(identity.identity_scope, ConversationIdentityScope::Direct);
997        assert_eq!(
998            identity.migration_state,
999            ConversationMigrationState::Canonical
1000        );
1001        assert!(identity.aliases.is_empty());
1002    }
1003
1004    #[test]
1005    fn conversation_identity_old_flutter_sorted_direct_is_alias_input() {
1006        let identity = ConversationIdentity::from_storage_parts(
1007            "thread",
1008            "dm:did:example:alice:did:example:bob",
1009        );
1010
1011        assert_eq!(
1012            identity.conversation_id,
1013            "dm:did:example:alice:did:example:bob"
1014        );
1015        assert_eq!(identity.canonical_thread_kind, "direct");
1016        assert_eq!(identity.identity_scope, ConversationIdentityScope::Direct);
1017        assert_eq!(
1018            identity.migration_state,
1019            ConversationMigrationState::LegacyInput
1020        );
1021        assert_eq!(identity.aliases.len(), 1);
1022        assert_eq!(
1023            identity.aliases[0].source,
1024            ConversationAliasSource::OldFlutterSortedDirect
1025        );
1026    }
1027
1028    #[test]
1029    fn conversation_identity_old_flutter_sorted_direct_resolves_with_owner() {
1030        let identity = ConversationIdentity::from_storage_parts_for_owner(
1031            "thread",
1032            "dm:did:example:alice:did:example:bob",
1033            "did:example:alice",
1034        );
1035
1036        assert_eq!(identity.conversation_id, "dm:did:example:bob");
1037        assert_eq!(identity.canonical_thread_kind, "direct");
1038        assert_eq!(identity.canonical_thread_id, "dm:did:example:bob");
1039        assert_eq!(identity.storage_thread_ref.kind, "thread");
1040        assert_eq!(
1041            identity.storage_thread_ref.id,
1042            "dm:did:example:alice:did:example:bob"
1043        );
1044        assert_eq!(identity.identity_scope, ConversationIdentityScope::Direct);
1045        assert_eq!(
1046            identity.migration_state,
1047            ConversationMigrationState::LegacyInput
1048        );
1049        assert_eq!(identity.aliases.len(), 1);
1050        assert_eq!(
1051            identity.aliases[0].source,
1052            ConversationAliasSource::OldFlutterSortedDirect
1053        );
1054    }
1055
1056    #[test]
1057    fn conversation_identity_group_ref_uses_group_conversation_id() {
1058        let group = crate::ids::GroupRef::parse("did:example:group").unwrap();
1059        let identity = ConversationIdentity::from_thread_ref(&ThreadRef::Group(group));
1060
1061        assert_eq!(identity.conversation_id, "group:did:example:group");
1062        assert_eq!(identity.canonical_thread_kind, "group");
1063        assert_eq!(identity.identity_scope, ConversationIdentityScope::Group);
1064        assert_eq!(
1065            identity.migration_state,
1066            ConversationMigrationState::LegacyInput
1067        );
1068    }
1069}
im_core/messages/dto.rs

im_core/messages/
dto.rs
