imsg_session/
sync.rs

1//! Store-integrated sync: message ingestion and per-folder cursor backfill.
2
3use crate::util::now_ms;
4pub use crate::util::{datetime_to_ms, ms_to_display};
5
6use map_core::client::MapClient;
7use map_core::folders::Folder;
8use map_core::messages::ListMessagesFilter;
9use store::{Direction, FolderSyncStatus, NewMessage, PhoneField, Store};
10use tokio::io::{AsyncRead, AsyncWrite};
11
12use crate::fetch::{fetch_folder, FetchedMessage};
13
14// direction is derived from sent; status is 1 for read, 0 for unread. The persist boundary —
15// store shape (NewMessage) stays out of the shared read path
16fn to_new_message(msg: FetchedMessage, synced_at: i64) -> NewMessage {
17    NewMessage {
18        map_handle: msg.handle,
19        timestamp_ms: msg.timestamp_ms,
20        folder: msg.folder,
21        direction: if msg.sent { Direction::Sent } else { Direction::Received },
22        address: PhoneField::new(&msg.address, None),
23        status: i32::from(msg.read),
24        synced_at,
25        text: msg.text,
26        outgoing_status: None,
27    }
28}
29
30// fetches and upserts all messages in folder since the per-folder cursor anchor, paging at
31// 1024 messages per request. Reads the cursor before fetching to derive since_ms (None on
32// first run = full fetch). Tracks the highest timestamp_ms seen; on success writes the cursor
33// with sync_status = Complete and highest_ts set to that value (or the previous highest_ts
34// when no new messages were found). The cursor isn't updated on error — the next run retries
35// from the same anchor
36async fn backfill_folder<T: AsyncRead + AsyncWrite + Unpin>(
37    client: &mut MapClient<T>,
38    store: &Store,
39    folder: Folder,
40    now: i64,
41) -> anyhow::Result<()> {
42    let cursor = store.get_cursor(folder.as_str()).await?;
43    let since_ms = cursor.as_ref().map(|c| c.highest_ts);
44    // Preserve the previous highest_ts when no new messages arrive this run.
45    let mut highest_ts_seen = cursor.as_ref().map_or(0, |c| c.highest_ts);
46    let folder_str = folder.as_str();
47    let is_sent = folder == Folder::Sent;
48
49    for msg in fetch_folder(client, folder, since_ms, now, &ListMessagesFilter::default()).await? {
50        if msg.timestamp_ms > highest_ts_seen {
51            highest_ts_seen = msg.timestamp_ms;
52        }
53        let handle = msg.handle.clone();
54        store.upsert(to_new_message(msg, now)).await?;
55        // Reconcile: if this Sent message matches a speculatively-created local row
56        // that was awaiting confirmation, advance its outgoing_status to sent_confirmed.
57        if is_sent {
58            store.reconcile_outgoing(&handle).await?;
59        }
60    }
61
62    // Cursor is written only on full success; errors above exit before reaching this line.
63    store.set_cursor(folder_str, now, highest_ts_seen, FolderSyncStatus::Complete).await?;
64    Ok(())
65}
66
67/// Fetches MAP messages and upserts them into the store using per-folder cursor anchors.
68///
69/// When `folder_scope` is `None`, all four folders are processed in order: `Inbox`, `Sent`,
70/// `Deleted`, `Outbox`. When `Some`, only the specified folder is processed.
71///
72/// Each folder runs independently: a failure on one folder is logged and skipped so other
73/// folders can still advance their cursors. Returns an error if any folder failed.
74///
75/// # Errors
76///
77/// Returns the first folder error encountered. Callers should treat a partial failure as
78/// "sync incomplete" and re-run to retry the failed folder(s).
79pub async fn backfill<T: AsyncRead + AsyncWrite + Unpin>(
80    client: &mut MapClient<T>,
81    store: &Store,
82    folder_scope: Option<Folder>,
83) -> anyhow::Result<()> {
84    let all_folders = [Folder::Inbox, Folder::Sent, Folder::Deleted, Folder::Outbox];
85    let single;
86    let folders: &[Folder] = if let Some(f) = folder_scope {
87        single = [f];
88        &single
89    } else {
90        &all_folders
91    };
92
93    let now = now_ms();
94    let mut first_err: Option<anyhow::Error> = None;
95
96    for &folder in folders {
97        if let Err(e) = backfill_folder(client, store, folder, now).await {
98            tracing::warn!("backfill: {} failed — {e:#}", folder.as_str());
99            if first_err.is_none() {
100                first_err = Some(e);
101            }
102        }
103    }
104
105    first_err.map_or(Ok(()), Err)
106}
107
108/// Runs an incremental backfill across all folders using per-folder cursor anchors.
109///
110/// Each folder's cursor determines the pull boundary; a folder with no cursor triggers a full
111/// fetch. This is the sync coordinator's entry point — it is not intended for read commands.
112/// Call [`backfill`] directly when a folder scope is needed.
113///
114/// # Errors
115///
116/// Returns an error if any folder's backfill fails; see [`backfill`] for continuation semantics.
117pub async fn backfill_catch_up<T: AsyncRead + AsyncWrite + Unpin>(
118    client: &mut MapClient<T>,
119    store: &Store,
120) -> anyhow::Result<()> {
121    backfill(client, store, None).await
122}
imsg_session/sync.rs

imsg_session/
sync.rs
