fix: keep reconciled and replayed lists in sync with the server

crates/rustmail-tui/src/app.rs

@@ -788,9 +788,7 @@ impl App {
         self.messages = resp.messages;
         self.total = resp.total;
         self.view_stale = pending.overflowed;
-        for delta in pending.events {
-          self.replay_delta(delta);
-        }
+        self.replay_deltas(pending.events);
         self.error = None;
         self.error_ticks = 0;
         let moved = self.reanchor_selection(anchor);
@@ -1124,6 +1122,25 @@ impl App {
     self.messages.iter_mut().find(|m| m.id == id)
   }
 
+  /// Re-applies the deltas recorded during an in-flight fetch, in order, onto
+  /// the snapshot that fetch returned.
+  ///
+  /// Arrivals are announced in the order they commit, so every arrival up to
+  /// the last one the page lists is already in the snapshot, counted in its
+  /// total even when it sits past the page; only the arrivals after it are new.
+  fn replay_deltas(&mut self, deltas: Vec<WsEvent>) {
+    let listed_through = deltas.iter().rposition(
+      |delta| matches!(delta, WsEvent::MessageNew(summary) if self.is_listed(&summary.id)),
+    );
+    for (index, delta) in deltas.into_iter().enumerate() {
+      let in_snapshot = listed_through.is_some_and(|last| index <= last);
+      if in_snapshot && matches!(delta, WsEvent::MessageNew(_)) {
+        continue;
+      }
+      self.replay_delta(delta);
+    }
+  }
+
   /// Re-applies a delta recorded during an in-flight fetch onto the snapshot
   /// that fetch returned. Selection is left alone: the caller re-resolves it
   /// by message id once every delta has been replayed. A delta whose effect
@@ -1864,6 +1881,21 @@ mod tests {
     assert_eq!(app.total, 4);
   }
 
+  #[tokio::test]
+  async fn arrival_the_snapshot_holds_past_its_page_is_not_counted_again() {
+    let mut app = in_flight_fetch(0);
+    app.page_size = 2;
+
+    for id in ["live-0", "live-1", "live-2"] {
+      app.handle_ws_message(&new_message_event(id)).await;
+    }
+    land_snapshot(&mut app, &["live-2", "live-1"], 3).await;
+
+    let ids: Vec<&str> = app.messages.iter().map(|m| m.id.as_str()).collect();
+    assert_eq!(ids, vec!["live-2", "live-1"]);
+    assert_eq!(app.total, 3);
+  }
+
   #[tokio::test]
   async fn update_during_in_flight_fetch_is_replayed_onto_the_snapshot() {
     let mut app = in_flight_fetch(1);

ui/src/stores/messages.test.ts

@@ -1411,6 +1411,21 @@ describe("arrivals while the reader is scrolled away", () => {
     expect(filteredMessages().map((m) => m.is_starred)).toEqual([false, true]);
   });
 
+  it("lets in a match a held resync finds between loaded rows", async () => {
+    const starred = (n: number) => message(n, { is_starred: true });
+    serve([starred(0), starred(2), starred(4)]);
+    toggleFilter("starred");
+    await vi.waitFor(() => expect(loading()).toBe(false));
+    setLiveHeld(true);
+    serve([starred(0), starred(1), starred(2), starred(4)]);
+
+    await resync();
+
+    expect(ids()).toEqual(["id-0", "id-1", "id-2", "id-4"]);
+    expect(heldArrivals()).toBe(0);
+    expect(total()).toBe(4);
+  });
+
   it("reconciles loaded rows past the first page on a held resync", async () => {
     const rows = range(PAGE_SIZE + 50);
     serve(rows);
@@ -1491,6 +1506,25 @@ describe("arrivals while the reader is scrolled away", () => {
     expect(ids()).toEqual(["id-0", "id-1"]);
   });
 
+  it("counts arrivals a held resync finds past what it holds once", async () => {
+    setLiveHeld(true);
+    const arrivals = Array.from({ length: MAX_LIVE_ROWS + 1 }, (_, i) =>
+      message(1000 + i),
+    );
+    const rows = [...arrivals.toReversed(), ...range(2)];
+    serve(rows);
+
+    openSocket();
+    for (const m of arrivals) {
+      receive(JSON.stringify({ type: "message:new", data: m }));
+    }
+    nextFrame();
+    await vi.waitFor(() => expect(loading()).toBe(false));
+
+    expect(heldArrivals()).toBe(arrivals.length);
+    expect(total()).toBe(rows.length);
+  });
+
   it("puts the held arrivals on top once the reader is back there", () => {
     setLiveHeld(true);
     deliver(arrival(8));
@@ -1702,6 +1736,29 @@ describe("resync when the socket opens", () => {
     expect(total()).toBe(3);
   });
 
+  it("counts an arrival the read holds past its first page once", async () => {
+    const read = deferred<Page>();
+    listMessages.mockReturnValueOnce(read.promise);
+    connectWebSocket();
+    openSocket();
+    const arrivals = Array.from({ length: PAGE_SIZE + 1 }, (_, i) =>
+      message(1000 + i),
+    );
+
+    for (const m of arrivals) {
+      receive(JSON.stringify({ type: "message:new", data: m }));
+    }
+    nextFrame();
+    const newest = arrivals.slice(1).toReversed();
+    read.resolve({ messages: newest, total: arrivals.length });
+
+    await vi.waitFor(() => expect(loading()).toBe(false));
+    expect(filteredMessages().map((m) => m.id)).toEqual(
+      newest.map((m) => m.id),
+    );
+    expect(total()).toBe(arrivals.length);
+  });
+
   it("keeps a flag change made while the read is in flight", async () => {
     const read = deferred<Page>();
     listMessages.mockReturnValueOnce(read.promise);

ui/src/stores/messages.ts

@@ -828,8 +828,15 @@ async function fetchMessages(): Promise<void> {
   await readFirstPage();
 }
 
-/** A list read that has landed: the count it read, and how to show it. */
-type LandedRead = { total: number; show: () => void };
+/**
+ * A list read that has landed: the count it read, the ids of the rows it
+ * returned, and how to show it.
+ */
+type LandedRead = {
+  total: number;
+  ids: ReadonlySet<string>;
+  show: () => void;
+};
 
 async function readTopPage(
   view: View,
@@ -841,6 +848,7 @@ async function readTopPage(
   );
   return {
     total: res.total,
+    ids: new Set(res.messages.map((m) => m.id)),
     show: () => {
       replaceRows(res.messages);
       dropHeldArrivals();
@@ -854,9 +862,12 @@ async function readTopPage(
  *
  * This is the read while live mail is held: the reader is scrolled into these
  * rows, so replacing them with the first page would move the list under them.
- * Rows above the first loaded one are arrivals and wait behind the pill. A
- * loaded row the read meets takes the server's copy, and one it passes without
- * meeting is gone from the view. The read stops at the loaded tail, at the end
+ * Rows above the first loaded one are arrivals and wait behind the pill. From
+ * the first loaded row the read meets to the last, the server's rows replace
+ * the loaded ones: a loaded row the read meets takes the server's copy, one it
+ * passes without meeting is gone from the view, and a row between them that
+ * was never loaded, such as one starred elsewhere under the starred filter,
+ * joins the list where it falls. The read stops at the loaded tail, at the end
  * of the list, after {@link PAGE_SIZE} rows past the last loaded one it met,
  * which is where a deleted tail leaves it, or once more has arrived than the
  * pill keeps, which only a fresh read on the way back to the top can show.
@@ -901,6 +912,7 @@ async function readLoadedWindow(
     if (passedAll || outranLoaded) {
       return {
         total: res.total,
+        ids: new Set(read.map((m) => m.id)),
         show: () => reconcileWindow(read, passedAll, ended, view),
       };
     }
@@ -918,21 +930,14 @@ function reconcileWindow(
   ended: boolean,
   view: View,
 ): void {
-  const server = new Map(read.map((m) => [m.id, m]));
   const loaded = messages();
   const firstLoaded = read.findIndex((m) => positions.has(m.id));
+  const lastLoaded = read.findLastIndex((m) => positions.has(m.id));
   const arrivals = firstLoaded < 0 ? read : read.slice(0, firstLoaded);
-  let passedUpTo = passedAll ? loaded.length - 1 : -1;
-  if (!passedAll) {
-    loaded.forEach((m, i) => {
-      if (server.has(m.id)) passedUpTo = i;
-    });
-  }
-  const kept = loaded.flatMap((m, i) => {
-    const fresh = server.get(m.id);
-    if (fresh !== undefined) return [fresh];
-    return i <= passedUpTo ? [] : [m];
-  });
+  const met = firstLoaded < 0 ? [] : read.slice(firstLoaded, lastLoaded + 1);
+  const lastMet = lastLoaded < 0 ? -1 : indexOfRow(read[lastLoaded].id);
+  const unmet = passedAll ? [] : loaded.slice(lastMet + 1);
+  const kept = [...met, ...unmet];
   const last = kept.at(-1) ?? arrivals.at(-1);
   const lastRead =
     last === undefined ? -1 : read.findIndex((m) => m.id === last.id);
@@ -972,7 +977,7 @@ async function readFirstPage(): Promise<boolean> {
         landed.show();
         setStoredTotal(landed.total);
         searchStale = false;
-        applyEvents(eventsDuringRead ?? []);
+        replayEvents(eventsDuringRead ?? [], landed.ids);
       });
       if (raced) void refreshTotal();
       return true;
@@ -1241,6 +1246,27 @@ function applyEvents(events: readonly WsEvent[]): void {
   admitArrivals(arrivals);
 }
 
+/**
+ * Applies the events that arrived while a list read was in flight over the
+ * list it returned.
+ *
+ * Arrivals are announced in the order they commit, so every arrival up to the
+ * last one among the rows the read returned, `read`, was already in the read,
+ * and counted in its total even when it sits past the rows it returned; only
+ * the arrivals after it are new.
+ */
+function replayEvents(
+  events: readonly ReplayableEvent[],
+  read: ReadonlySet<string>,
+): void {
+  const lastInRead = events.findLastIndex(
+    (event) => event.type === "message:new" && read.has(event.data.id),
+  );
+  applyEvents(
+    events.filter((event, i) => i > lastInRead || event.type !== "message:new"),
+  );
+}
+
 function applyEvent(event: Exclude<WsEvent, { type: "message:new" }>): void {
   switch (event.type) {
     case "message:read":