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

imsg_session/
sync.rs
