refactor(sdk): extract shared MAM catch-up utilities into mamCatchUpUtils

mremond · mremond · commit 71f57317daee · 2026-03-30T19:34:51.000+02:00
Consolidate duplicated findNewestMessage, connection error checks, and
  hardcoded constants from chatSideEffects, roomSideEffects, MAM, and
  backgroundSync into a single shared module.
diff --git a/apps/fluux/src-tauri/tauri.conf.json b/apps/fluux/src-tauri/tauri.conf.json
@@ -68,7 +68,7 @@
     ],
     "macOS": {
       "minimumSystemVersion": "10.13",
-      "bundleVersion": "12987ef",
+      "bundleVersion": "707f8c0",
       "entitlements": "Entitlements.plist",
       "signingIdentity": null
     },
diff --git a/packages/fluux-sdk/src/core/backgroundSync.ts b/packages/fluux-sdk/src/core/backgroundSync.ts
@@ -20,6 +20,7 @@ import { connectionStore, chatStore, roomStore } from '../stores'
 import { NS_MAM } from './namespaces'
 import { logInfo } from './logger'
 import { buildScopedStorageKey } from '../utils/storageScope'
+import { MAM_ROOM_CATCHUP_DELAY_MS } from '../utils/mamCatchUpUtils'
 
 /**
  * Sets up background sync side effects that run after a fresh session.
@@ -173,7 +174,7 @@ export function setupBackgroundSyncSideEffects(
           // Silently ignore member discovery errors
         }
       })()
-    }, 10_000)
+    }, MAM_ROOM_CATCHUP_DELAY_MS)
   }
 
   // Fresh session: 'online' fires only on fresh sessions (not SM resumption).
diff --git a/packages/fluux-sdk/src/core/chatSideEffects.ts b/packages/fluux-sdk/src/core/chatSideEffects.ts
@@ -16,24 +16,12 @@ import { chatStore, connectionStore } from '../stores'
 import { NS_MAM } from './namespaces'
 import { logInfo } from './logger'
 import { getDomain } from './jid'
-
-/**
- * Find the newest message in the array (regardless of delay status).
- *
- * Used as the catch-up cursor for MAM forward queries. Previously this
- * skipped delayed messages, but that caused a subtle bug: when ALL cached
- * messages are delayed (common after previous MAM catch-ups), the cursor
- * returned undefined, triggering a backward query. The backward query's
- * merge (`prependOlderMessages`) incorrectly placed newer messages (sent
- * from another client while offline) at the top of the list, making them
- * invisible to the user and preventing the sidebar lastMessage update.
- */
-function findNewestMessage(messages: Array<{ timestamp?: Date }>): { timestamp: Date } | undefined {
-  for (let i = messages.length - 1; i >= 0; i--) {
-    if (messages[i].timestamp) return messages[i] as { timestamp: Date }
-  }
-  return undefined
-}
+import {
+  findNewestMessage,
+  buildCatchUpStartTime,
+  isConnectionError,
+  MAM_CACHE_LOAD_LIMIT,
+} from '../utils/mamCatchUpUtils'
 
 /**
  * Options for configuring side effects behavior.
@@ -105,28 +93,24 @@ export function setupChatSideEffects(
       // handler races with the conversation subscriber's cache load, and
       // messages may be empty — causing a backward query instead of a
       // forward catch-up from the newest cached message.
-      await chatStore.getState().loadMessagesFromCache(conversationId, { limit: 100 })
+      await chatStore.getState().loadMessagesFromCache(conversationId, { limit: MAM_CACHE_LOAD_LIMIT })
 
       // Re-read messages after cache load (store was mutated)
       const cachedMessages = chatStore.getState().messages.get(conversationId) || []
       const newestMessage = findNewestMessage(cachedMessages)
 
       const queryOptions: { with: string; start?: string } = { with: conversation.id }
       if (newestMessage?.timestamp) {
-        const startTime = new Date(newestMessage.timestamp.getTime() + 1)
-        queryOptions.start = startTime.toISOString()
+        queryOptions.start = buildCatchUpStartTime(newestMessage.timestamp)
       }
 
       await client.chat.queryMAM(queryOptions)
       logInfo('Chat: MAM sync complete')
     } catch (error) {
-      // Only log if it's not a disconnection error (those are expected during reconnect)
-      const isConnectionError = error instanceof Error &&
-        (error.message.includes('disconnected') ||
-         error.message.includes('Not connected') ||
-         error.message.includes('Socket not available'))
+      // Allow retry on next conversation switch or reconnect
+      fetchInitiated.delete(conversationId)
 
-      if (isConnectionError) {
+      if (isConnectionError(error)) {
         if (debug) console.log('[SideEffects] Chat: MAM skipped - client disconnected')
       } else {
         console.error('[SideEffects] Chat: MAM fetch failed:', error)
@@ -163,7 +147,7 @@ export function setupChatSideEffects(
         // Step 1: Always load from IndexedDB cache (deduplication is handled by loadMessagesFromCache).
         // This is a fallback for cases where the hook's cache load didn't run (e.g., reconnection).
         if (debug) console.log('[SideEffects] Chat: Loading from cache')
-        await chatStore.getState().loadMessagesFromCache(activeConversationId, { limit: 100 })
+        await chatStore.getState().loadMessagesFromCache(activeConversationId, { limit: MAM_CACHE_LOAD_LIMIT })
 
         // Step 2: Background MAM fetch (skip if already initiated this session)
         if (fetchInitiated.has(activeConversationId)) {
diff --git a/packages/fluux-sdk/src/core/defaultStoreBindings.ts b/packages/fluux-sdk/src/core/defaultStoreBindings.ts
@@ -125,6 +125,7 @@ export function createDefaultStoreBindings(options: DefaultStoreBindingsOptions
       markAllNeedsCatchUp: chatStore.getState().markAllNeedsCatchUp,
       clearNeedsCatchUp: chatStore.getState().clearNeedsCatchUp,
       updateLastMessagePreview: chatStore.getState().updateLastMessagePreview,
+      loadMessagesFromCache: chatStore.getState().loadMessagesFromCache,
       getAllConversations: () => {
         const state = chatStore.getState()
         // Use activeConversations() which efficiently returns only non-archived
diff --git a/packages/fluux-sdk/src/core/modules/MAM.ts b/packages/fluux-sdk/src/core/modules/MAM.ts
@@ -33,6 +33,14 @@ import { getBareJid, getResource } from '../jid'
 import { generateUUID, generateStableMessageId } from '../../utils/uuid'
 import { executeWithConcurrency } from '../../utils/concurrencyUtils'
 import { parseRSMResponse } from '../../utils/rsm'
+import {
+  findNewestMessage,
+  buildCatchUpStartTime,
+  isConnectionError,
+  MAM_CATCHUP_FORWARD_MAX,
+  MAM_CATCHUP_BACKWARD_MAX,
+  MAM_CACHE_LOAD_LIMIT,
+} from '../../utils/mamCatchUpUtils'
 import {
   NS_MAM,
   NS_RSM,
@@ -86,21 +94,6 @@ interface UnresolvedModifications {
   reactions: { targetId: string; from: string; emojis: string[] }[]
 }
 
-/**
- * Find the newest message in an array (regardless of delay status).
- *
- * Used as the catch-up cursor for MAM forward queries. Including delayed
- * messages ensures the catch-up always uses a forward query, which merges
- * correctly via full sort. Previously skipping delayed messages caused
- * backward queries whose prepend-based merge put newer messages (sent
- * from another client while offline) at the wrong position.
- */
-function findNewestMessage(messages: Array<{ timestamp?: Date }>): { timestamp: Date } | undefined {
-  for (let i = messages.length - 1; i >= 0; i--) {
-    if (messages[i].timestamp) return messages[i] as { timestamp: Date }
-  }
-  return undefined
-}
 
 /**
  * Message Archive Management (XEP-0313) module.
@@ -272,8 +265,7 @@ export class MAM extends BaseModule {
       return { messages: allMessages, complete: isComplete, rsm: lastRsm }
     } catch (error) {
       const msg = error instanceof Error ? error.message : 'Unknown error'
-      const isConnectionError = msg.includes('Not connected') || msg.includes('Socket not available')
-      if (isConnectionError) {
+      if (isConnectionError(error)) {
         logInfo(`MAM skipped: ...@${getDomain(conversationId) || '*'} — ${msg}`)
       } else {
         logErr(`MAM error: ...@${getDomain(conversationId) || '*'} — ${msg}`)
@@ -407,8 +399,7 @@ export class MAM extends BaseModule {
       return { messages: allMessages, complete: isComplete, rsm: lastRsm }
     } catch (error) {
       const msg = error instanceof Error ? error.message : 'Unknown error'
-      const isConnectionError = msg.includes('Not connected') || msg.includes('Socket not available')
-      if (isConnectionError) {
+      if (isConnectionError(error)) {
         logInfo(`Room MAM skipped: ${roomJid} — ${msg}`)
       } else {
         logErr(`Room MAM error: ${roomJid} — ${msg}`)
@@ -822,23 +813,30 @@ export class MAM extends BaseModule {
           // Skip if disconnected (avoid queuing doomed queries)
           if (this.deps.stores?.connection.getStatus() !== 'online') return
 
-          const messages = conv.messages || []
+          // Load IndexedDB cache first so we know the newest cached message
+          // and can do a proper forward catch-up instead of fetching only latest.
+          // Without this, conv.messages is empty after app restart (runtime-only),
+          // causing a backward "before:''" query that creates gaps with old cache.
+          await this.deps.stores?.chat.loadMessagesFromCache?.(conv.id, { limit: MAM_CACHE_LOAD_LIMIT })
+
+          // Re-read messages after cache load (store was mutated)
+          const updatedConv = this.deps.stores?.chat.getAllConversations()?.find(c => c.id === conv.id)
+          const messages = updatedConv?.messages || conv.messages || []
           const newestMessage = findNewestMessage(messages)
 
           if (newestMessage?.timestamp) {
             // Forward query from the last known message
-            const startTime = new Date(newestMessage.timestamp.getTime() + 1)
             await this.queryArchive({
               with: conv.id,
-              start: startTime.toISOString(),
-              max: 100,
+              start: buildCatchUpStartTime(newestMessage.timestamp),
+              max: MAM_CATCHUP_FORWARD_MAX,
             })
           } else {
             // No messages (empty) — fetch latest from MAM
             await this.queryArchive({
               with: conv.id,
               before: '',
-              max: 50,
+              max: MAM_CATCHUP_BACKWARD_MAX,
             })
           }
         } catch (_error) {
@@ -940,10 +938,10 @@ export class MAM extends BaseModule {
           if (this.deps.stores?.connection.getStatus() !== 'online') return
 
           // Load IndexedDB cache first so we know the newest cached message
-          // and can do a proper forward catch-up instead of fetching only latest 50.
+          // and can do a proper forward catch-up instead of fetching only latest.
           // Without this, room.messages is empty after app restart (runtime-only),
           // causing a backward "before:''" query that creates gaps with old cache.
-          await this.deps.stores?.room.loadMessagesFromCache(room.jid, { limit: 100 })
+          await this.deps.stores?.room.loadMessagesFromCache(room.jid, { limit: MAM_CACHE_LOAD_LIMIT })
 
           // Re-read room after cache load (store was mutated)
           const updatedRoom = this.deps.stores?.room.getRoom(room.jid)
@@ -952,18 +950,17 @@ export class MAM extends BaseModule {
 
           if (newestMessage?.timestamp) {
             // Forward query from the last known message
-            const startTime = new Date(newestMessage.timestamp.getTime() + 1)
             await this.queryRoomArchive({
               roomJid: room.jid,
-              start: startTime.toISOString(),
-              max: 100,
+              start: buildCatchUpStartTime(newestMessage.timestamp),
+              max: MAM_CATCHUP_FORWARD_MAX,
             })
           } else {
             // No messages (empty) — fetch latest from MAM
             await this.queryRoomArchive({
               roomJid: room.jid,
               before: '',
-              max: 50,
+              max: MAM_CATCHUP_BACKWARD_MAX,
             })
           }
         } catch (_error) {
diff --git a/packages/fluux-sdk/src/core/roomSideEffects.ts b/packages/fluux-sdk/src/core/roomSideEffects.ts
@@ -18,20 +18,14 @@ import type { XMPPClient } from './XMPPClient'
 import type { SideEffectsOptions } from './chatSideEffects'
 import { roomStore, connectionStore } from '../stores'
 import { logInfo } from './logger'
-
-/**
- * Find the newest message in the array (regardless of delay status).
- *
- * Used as the catch-up cursor for MAM forward queries. Including delayed
- * messages ensures the catch-up always uses a forward query, which merges
- * correctly via full sort.
- */
-function findNewestMessage(messages: Array<{ timestamp?: Date }>): { timestamp: Date } | undefined {
-  for (let i = messages.length - 1; i >= 0; i--) {
-    if (messages[i].timestamp) return messages[i] as { timestamp: Date }
-  }
-  return undefined
-}
+import {
+  findNewestMessage,
+  buildCatchUpStartTime,
+  isConnectionError,
+  MAM_CATCHUP_FORWARD_MAX,
+  MAM_CATCHUP_BACKWARD_MAX,
+  MAM_CACHE_LOAD_LIMIT,
+} from '../utils/mamCatchUpUtils'
 
 /**
  * Sets up room-related side effects.
@@ -111,7 +105,7 @@ export function setupRoomSideEffects(
       // handler races with the conversation subscriber's cache load, and
       // room.messages may be empty — causing a backward "before:''" query
       // instead of a forward catch-up from the newest cached message.
-      await roomStore.getState().loadMessagesFromCache(roomJid, { limit: 100 })
+      await roomStore.getState().loadMessagesFromCache(roomJid, { limit: MAM_CACHE_LOAD_LIMIT })
 
       // Re-read room after cache load (store was mutated)
       const roomAfterCache = roomStore.getState().rooms.get(roomJid)
@@ -120,32 +114,25 @@ export function setupRoomSideEffects(
 
       if (newestMessage?.timestamp) {
         // Query for messages AFTER the newest known message (catchup)
-        const startTime = new Date(newestMessage.timestamp.getTime() + 1)
         await client.chat.queryRoomMAM({
           roomJid,
-          start: startTime.toISOString(),
-          max: 100,
+          start: buildCatchUpStartTime(newestMessage.timestamp),
+          max: MAM_CATCHUP_FORWARD_MAX,
         })
       } else {
         // No cached messages - fetch latest
         await client.chat.queryRoomMAM({
           roomJid,
           before: '', // Empty = get latest
-          max: 50,
+          max: MAM_CATCHUP_BACKWARD_MAX,
         })
       }
       logInfo('Room: MAM catch-up complete')
     } catch (error) {
       // Allow backup handlers (room:joined, supportsMAM watcher) to retry
       fetchInitiated.delete(roomJid)
 
-      // Only log if it's not a disconnection error (those are expected during reconnect)
-      const isConnectionError = error instanceof Error &&
-        (error.message.includes('disconnected') ||
-         error.message.includes('Not connected') ||
-         error.message.includes('Socket not available'))
-
-      if (isConnectionError) {
+      if (isConnectionError(error)) {
         if (debug) console.log('[SideEffects] Room: MAM skipped - client disconnected')
       } else {
         console.error('[SideEffects] Room: MAM catchup failed:', error)
@@ -187,7 +174,7 @@ export function setupRoomSideEffects(
         // Step 1: Always load from IndexedDB cache (deduplication is handled by loadMessagesFromCache).
         // This is a fallback for cases where the hook's cache load didn't run (e.g., reconnection).
         if (debug) console.log('[SideEffects] Room: Loading from cache')
-        await roomStore.getState().loadMessagesFromCache(activeRoomJid, { limit: 100 })
+        await roomStore.getState().loadMessagesFromCache(activeRoomJid, { limit: MAM_CACHE_LOAD_LIMIT })
 
         // Step 2: Background MAM fetch for catchup (skip if already initiated this session)
         if (fetchInitiated.has(activeRoomJid)) {
diff --git a/packages/fluux-sdk/src/core/test-utils.ts b/packages/fluux-sdk/src/core/test-utils.ts
@@ -642,6 +642,7 @@ export const createMockStores = (): MockStoreBindings => ({
     markAllNeedsCatchUp: vi.fn(),
     clearNeedsCatchUp: vi.fn(),
     updateLastMessagePreview: vi.fn(),
+    loadMessagesFromCache: vi.fn().mockResolvedValue([]),
     getAllConversations: vi.fn().mockReturnValue([]),
     getArchivedConversations: vi.fn().mockReturnValue([]),
     archiveConversation: vi.fn(),
diff --git a/packages/fluux-sdk/src/core/types/client.ts b/packages/fluux-sdk/src/core/types/client.ts
@@ -96,6 +96,8 @@ export interface StoreBindings {
     clearNeedsCatchUp: (conversationId: string) => void
     // Update sidebar preview without affecting message history
     updateLastMessagePreview: (conversationId: string, lastMessage: Message) => void
+    // Load messages from IndexedDB cache into the conversation's in-memory message array
+    loadMessagesFromCache?: (conversationId: string, options?: { limit?: number }) => Promise<unknown>
     // Get all conversations for MAM catch-up
     getAllConversations: () => Array<{ id: string; messages: Message[] }>
     // Smart MAM: archived conversation preview refresh
diff --git a/packages/fluux-sdk/src/utils/mamCatchUpUtils.test.ts b/packages/fluux-sdk/src/utils/mamCatchUpUtils.test.ts
diff --git a/packages/fluux-sdk/src/utils/mamCatchUpUtils.ts b/packages/fluux-sdk/src/utils/mamCatchUpUtils.ts

Original file line number	Diff line number	Diff line change
`@@ -20,6 +20,7 @@ import { connectionStore, chatStore, roomStore } from '../stores'`
`20`	`20`	`import { NS_MAM } from './namespaces'`
`21`	`21`	`import { logInfo } from './logger'`
`22`	`22`	`import { buildScopedStorageKey } from '../utils/storageScope'`
	`23`	`+import { MAM_ROOM_CATCHUP_DELAY_MS } from '../utils/mamCatchUpUtils'`
`23`	`24`
`24`	`25`	`/**`
`25`	`26`	`* Sets up background sync side effects that run after a fresh session.`
`@@ -173,7 +174,7 @@ export function setupBackgroundSyncSideEffects(`
`173`	`174`	`// Silently ignore member discovery errors`
`174`	`175`	`}`
`175`	`176`	`})()`
`176`		`- }, 10_000)`
	`177`	`+ }, MAM_ROOM_CATCHUP_DELAY_MS)`
`177`	`178`	`}`
`178`	`179`
`179`	`180`	`// Fresh session: 'online' fires only on fresh sessions (not SM resumption).`