iris-chat 0.1.39

Iris Chat command line client and shared encrypted chat core
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

use super::*;
use std::collections::BTreeMap;

type MessageKey = (u64, String, String);

pub(super) fn collect_device_sync_messages(
    core: &AppCore,
    roster_at: u64,
    after: Option<&DeviceSyncCursor>,
    page_size: usize,
) -> (Vec<DeviceSyncMessage>, Option<DeviceSyncCursor>) {
    let now = unix_now().get();
    let limit = page_size.saturating_add(1);
    let mut messages = BTreeMap::new();
    for message in core.threads.values().flat_map(|thread| &thread.messages) {
        if eligible(message, roster_at, now)
            && after.is_none_or(|cursor| after_cursor(message, cursor))
        {
            if let Some(value) = from_snapshot(message) {
                insert_bounded(&mut messages, value, limit);
            }
        }
    }

    let mut db_after = after
        .map(|cursor| (cursor.created_at, cursor.chat_id.clone(), cursor.id.clone()))
        .unwrap_or((roster_at, String::new(), String::new()));
    loop {
        let rows = core
            .app_store
            .load_device_sync_messages_page(
                roster_at,
                now,
                db_after.0,
                &db_after.1,
                &db_after.2,
                limit,
            )
            .unwrap_or_default();
        let exhausted = rows.len() < limit;
        let Some(last) = rows.last() else { break };
        let frontier = (last.created_at_secs, last.chat_id.clone(), last.id.clone());
        for message in rows {
            if in_memory_message(core, &message.chat_id, &message.id).is_some() {
                continue;
            }
            insert_bounded(&mut messages, from_persisted(message), limit);
        }
        let page_is_known = messages
            .last_key_value()
            .is_some_and(|(last, _)| messages.len() >= limit && *last <= frontier);
        db_after = frontier;
        if exhausted || page_is_known {
            break;
        }
    }

    let has_more = messages.len() > page_size;
    while messages.len() > page_size {
        messages.pop_last();
    }
    let values = messages.into_values().collect::<Vec<_>>();
    let next = has_more
        .then(|| values.last().map(DeviceSyncCursor::from))
        .flatten();
    (values, next)
}

fn insert_bounded(
    messages: &mut BTreeMap<MessageKey, DeviceSyncMessage>,
    value: DeviceSyncMessage,
    limit: usize,
) {
    messages.insert(message_key(&value), value);
    while messages.len() > limit {
        messages.pop_last();
    }
}

fn in_memory_message<'a>(
    core: &'a AppCore,
    chat_id: &str,
    id: &str,
) -> Option<&'a ChatMessageSnapshot> {
    core.threads
        .get(chat_id)?
        .messages
        .iter()
        .find(|message| message.id == id)
}

fn eligible(message: &ChatMessageSnapshot, roster_at: u64, now: u64) -> bool {
    message.created_at_secs >= roster_at
        && message
            .expires_at_secs
            .is_none_or(|expires_at| expires_at > now)
        && matches!(message.kind, ChatMessageKind::User)
        && !matches!(
            message.delivery,
            DeliveryState::Queued | DeliveryState::Pending | DeliveryState::Failed
        )
}

fn after_cursor(message: &ChatMessageSnapshot, cursor: &DeviceSyncCursor) -> bool {
    (message.created_at_secs, &message.chat_id, &message.id)
        > (cursor.created_at, &cursor.chat_id, &cursor.id)
}

fn from_snapshot(message: &ChatMessageSnapshot) -> Option<DeviceSyncMessage> {
    Some(DeviceSyncMessage {
        chat_id: message.chat_id.clone(),
        id: message.id.clone(),
        body: message.body.clone(),
        author: message.author_owner_pubkey_hex.clone()?,
        created_at: message.created_at_secs,
        expires_at: message.expires_at_secs,
    })
}

fn from_persisted(message: PersistedMessage) -> DeviceSyncMessage {
    DeviceSyncMessage {
        chat_id: message.chat_id,
        id: message.id,
        body: message.body,
        author: message.author_owner_pubkey_hex.unwrap_or(message.author),
        created_at: message.created_at_secs,
        expires_at: message.expires_at_secs,
    }
}

fn message_key(message: &DeviceSyncMessage) -> MessageKey {
    (
        message.created_at,
        message.chat_id.clone(),
        message.id.clone(),
    )
}