feat: Update how messages load and sync, allow plugins to import messages

toju-app/src/app/store/messages/messages-incoming.handlers.spec.ts

@@ -95,7 +95,7 @@ describe('dispatchIncomingMessage room-scoped sync', () => {
 
     expect(getMessages).toHaveBeenCalledWith('room-b', expect.any(Number), 0);
     expect(sendToPeer).toHaveBeenCalledWith('peer-2', {
-      type: 'chat-sync-full',
+      type: 'chat-sync-batch',
       roomId: 'room-b',
       messages: roomBMessages
     });

toju-app/src/app/store/messages/messages-incoming.handlers.ts

@@ -289,6 +289,12 @@ async function processSyncBatch(
   attachments: AttachmentFacade
 ): Promise<Message[]> {
   const toUpsert: Message[] = [];
+  // Yield to the event loop every YIELD_EVERY messages so Angular change
+  // detection and user input aren't starved while a large sync batch
+  // (e.g. from a bulk plugin import) drains serial DB writes.
+  const YIELD_EVERY = 50;
+
+  let processed = 0;
 
   for (const incoming of event.messages) {
     attachments.rememberMessageRoom(incoming.id, incoming.roomId);
@@ -305,6 +311,12 @@ async function processSyncBatch(
 
     if (changed)
       toUpsert.push(message);
+
+    processed += 1;
+
+    if (processed % YIELD_EVERY === 0) {
+      await new Promise<void>((resolve) => setTimeout(resolve, 0));
+    }
   }
 
   if (hasAttachmentMetaMap(event.attachments)) {
@@ -603,13 +615,20 @@ function handleSyncRequest(
   return from(
     (async () => {
       const all = await db.getMessages(targetRoomId, FULL_SYNC_LIMIT, 0);
-      const syncFullEvent: ChatEvent = {
-        type: 'chat-sync-full',
-        roomId: targetRoomId,
-        messages: all
-      };
 
-      webrtc.sendToPeer(fromPeerId, syncFullEvent);
+      // Ship as chunked chat-sync-batch events instead of a single
+      // chat-sync-full payload. A monolithic dump of up to FULL_SYNC_LIMIT
+      // messages can exceed the WebRTC SCTP per-message size ceiling and be
+      // silently dropped - especially after bulk plugin imports.
+      for (const chunk of chunkArray(all, CHUNK_SIZE)) {
+        const syncBatchEvent: ChatEvent = {
+          type: 'chat-sync-batch',
+          roomId: targetRoomId,
+          messages: chunk
+        };
+
+        webrtc.sendToPeer(fromPeerId, syncBatchEvent);
+      }
     })()
   ).pipe(mergeMap(() => EMPTY));
 }

toju-app/src/app/store/messages/messages.actions.ts

@@ -23,6 +23,24 @@ export const MessagesActions = createActionGroup({
     'Load Messages Success': props<{ messages: Message[] }>(),
     'Load Messages Failure': props<{ error: string }>(),
 
+    /**
+     * Fetches a page of messages strictly older than `beforeTimestamp` for a
+     * given conversation (room + channel). Used by the chat scroll-up handler
+     * to backfill history from the local database on demand.
+     */
+    'Load Older Messages': props<{
+      roomId: string;
+      channelId: string;
+      beforeTimestamp: number;
+      limit: number;
+    }>(),
+    'Load Older Messages Success': props<{
+      conversationKey: string;
+      messages: Message[];
+      reachedEnd: boolean;
+    }>(),
+    'Load Older Messages Failure': props<{ error: string }>(),
+
     /** Sends a new chat message to the current room and broadcasts to peers. */
     'Send Message': props<{ content: string; replyToId?: string; channelId?: string }>(),
     'Send Message Success': props<{ message: Message }>(),

toju-app/src/app/store/messages/messages.effects.ts

@@ -43,7 +43,8 @@ import { TimeSyncService } from '../../core/services/time-sync.service';
 import {
   DELETED_MESSAGE_CONTENT,
   Message,
-  Reaction
+  Reaction,
+  Room
 } from '../../shared-kernel';
 import { hydrateMessages } from './messages.helpers';
 import { canEditMessage } from '../../domains/chat/domain/rules/message.rules';
@@ -67,8 +68,9 @@ export class MessagesEffects {
   loadMessages$ = createEffect(() =>
     this.actions$.pipe(
       ofType(MessagesActions.loadMessages),
-      switchMap(({ roomId }) =>
-        from(this.db.getMessages(roomId, INITIAL_ROOM_MESSAGE_LIMIT, 0)).pipe(
+      withLatestFrom(this.store.select(selectCurrentRoom)),
+      switchMap(([{ roomId }, currentRoom]) =>
+        from(this.loadInitialMessages(roomId, currentRoom)).pipe(
           mergeMap(async (messages) => {
             const hydrated = await hydrateMessages(messages, this.db);
 
@@ -88,6 +90,58 @@ export class MessagesEffects {
     )
   );
 
+  /** Paginates older messages from the local DB for scroll-up history loading. */
+  loadOlderMessages$ = createEffect(() =>
+    this.actions$.pipe(
+      ofType(MessagesActions.loadOlderMessages),
+      mergeMap(({ roomId, channelId, beforeTimestamp, limit }) =>
+        from(
+          this.db.getMessages(roomId, limit, 0, channelId, beforeTimestamp)
+        ).pipe(
+          mergeMap(async (messages) => {
+            const hydrated = await hydrateMessages(messages, this.db);
+
+            for (const message of hydrated) {
+              this.attachments.rememberMessageRoom(message.id, message.roomId);
+            }
+
+            return MessagesActions.loadOlderMessagesSuccess({
+              conversationKey: `${roomId}:${channelId}`,
+              messages: hydrated,
+              reachedEnd: hydrated.length < limit
+            });
+          }),
+          catchError((error) =>
+            of(MessagesActions.loadOlderMessagesFailure({ error: error.message }))
+          )
+        )
+      )
+    )
+  );
+
+  private async loadInitialMessages(roomId: string, currentRoom: Room | null): Promise<Message[]> {
+    const textChannels = currentRoom?.id === roomId
+      ? (currentRoom.channels ?? []).filter((channel) => channel.type === 'text')
+      : [];
+
+    if (textChannels.length <= 1) {
+      return this.db.getMessages(roomId, INITIAL_ROOM_MESSAGE_LIMIT, 0, textChannels[0]?.id);
+    }
+
+    const channelMessageSets = await Promise.all(
+      textChannels.map((channel) => this.db.getMessages(roomId, INITIAL_ROOM_MESSAGE_LIMIT, 0, channel.id))
+    );
+    const messagesById = new Map<string, Message>();
+
+    for (const messages of channelMessageSets) {
+      for (const message of messages) {
+        messagesById.set(message.id, message);
+      }
+    }
+
+    return [...messagesById.values()].sort((first, second) => first.timestamp - second.timestamp);
+  }
+
   /** Constructs a new message, persists it locally, and broadcasts to all peers. */
   sendMessage$ = createEffect(() =>
     this.actions$.pipe(

toju-app/src/app/store/messages/messages.helpers.ts

@@ -45,10 +45,17 @@ export async function hydrateMessages(
   return messages.map((msg) => msg.isDeleted ? normaliseDeletedMessage(msg) : msg);
 }
 
-/** Builds a sync inventory item from a message and its reaction count. */
+/** Builds a sync inventory item from a message and its reaction count.
+ *
+ * Reactions are read from the already-hydrated `msg.reactions` array (the
+ * persistence layer joins them in via `getMessages`), and attachment counts
+ * only come from the in-memory override. We deliberately avoid per-message
+ * DB lookups here so a whole-room inventory stays O(1) DB calls even when
+ * the room contains tens of thousands of messages.
+ */
 export async function buildInventoryItem(
   msg: Message,
-  db: DatabaseService,
+  _db: DatabaseService,
   attachmentCountOverride?: number
 ): Promise<InventoryItem> {
   if (msg.isDeleted) {
@@ -60,50 +67,49 @@ export async function buildInventoryItem(
     };
   }
 
-  const reactions = await db.getReactionsForMessage(msg.id);
-  const attachments =
-    attachmentCountOverride === undefined
-      ? await db.getAttachmentsForMessage(msg.id)
-      : [];
-
-  return { id: msg.id,
+  const item: InventoryItem = {
+    id: msg.id,
     ts: getMessageTimestamp(msg),
-    rc: reactions.length,
-    ac: attachmentCountOverride ?? attachments.length };
+    rc: msg.reactions?.length ?? 0
+  };
+
+  if (attachmentCountOverride !== undefined) {
+    item.ac = attachmentCountOverride;
+  }
+
+  return item;
 }
 
-/** Builds a local map of `{timestamp, reactionCount, attachmentCount}` keyed by message ID. */
+/** Builds a local map of `{timestamp, reactionCount, attachmentCount}` keyed by message ID.
+ *
+ * As with {@link buildInventoryItem}, reactions come from the already-hydrated
+ * `msg.reactions` array and attachment counts only come from the in-memory
+ * override map.
+ */
 export async function buildLocalInventoryMap(
   messages: Message[],
-  db: DatabaseService,
+  _db: DatabaseService,
   attachmentCountOverrides?: ReadonlyMap<string, number>
 ): Promise<Map<string, { ts: number; rc: number; ac: number }>> {
   const map = new Map<string, { ts: number; rc: number; ac: number }>();
 
-  await Promise.all(
-    messages.map(async (msg) => {
-      if (msg.isDeleted) {
-        map.set(msg.id, {
-          ts: getMessageTimestamp(msg),
-          rc: 0,
-          ac: 0
-        });
+  for (const msg of messages) {
+    if (msg.isDeleted) {
+      map.set(msg.id, {
+        ts: getMessageTimestamp(msg),
+        rc: 0,
+        ac: 0
+      });
 
-        return;
-      }
+      continue;
+    }
 
-      const reactions = await db.getReactionsForMessage(msg.id);
-      const attachmentCountOverride = attachmentCountOverrides?.get(msg.id);
-      const attachments =
-        attachmentCountOverride === undefined
-          ? await db.getAttachmentsForMessage(msg.id)
-          : [];
-
-      map.set(msg.id, { ts: getMessageTimestamp(msg),
-        rc: reactions.length,
-        ac: attachmentCountOverride ?? attachments.length });
-    })
-  );
+    map.set(msg.id, {
+      ts: getMessageTimestamp(msg),
+      rc: msg.reactions?.length ?? 0,
+      ac: attachmentCountOverrides?.get(msg.id) ?? 0
+    });
+  }
 
   return map;
 }

toju-app/src/app/store/messages/messages.reducer.ts

@@ -13,10 +13,18 @@ export interface MessagesState extends EntityState<Message> {
   loading: boolean;
   /** Whether a peer-to-peer sync cycle is in progress. */
   syncing: boolean;
+  /** Whether a scroll-up older-page fetch is currently in flight. */
+  loadingOlder: boolean;
   /** Most recent error message from message operations. */
   error: string | null;
   /** ID of the room whose messages are currently loaded. */
   currentRoomId: string | null;
+  /**
+   * Conversation keys (`${roomId}:${channelId}`) that have been paginated
+   * all the way back to the start of the local DB history. Used by the
+   * scroll-up handler to stop issuing further DB pages.
+   */
+  exhaustedConversations: Record<string, true>;
 }
 
 export const messagesAdapter: EntityAdapter<Message> = createEntityAdapter<Message>({
@@ -27,8 +35,10 @@ export const messagesAdapter: EntityAdapter<Message> = createEntityAdapter<Messa
 export const initialState: MessagesState = messagesAdapter.getInitialState({
   loading: false,
   syncing: false,
+  loadingOlder: false,
   error: null,
-  currentRoomId: null
+  currentRoomId: null,
+  exhaustedConversations: {}
 });
 
 export const messagesReducer = createReducer(
@@ -41,7 +51,8 @@ export const messagesReducer = createReducer(
         ...state,
         loading: true,
         error: null,
-        currentRoomId: roomId
+        currentRoomId: roomId,
+        exhaustedConversations: {}
       });
     }
 
@@ -66,6 +77,30 @@ export const messagesReducer = createReducer(
     error
   })),
 
+  // Load older messages - paginate backwards from the DB on scroll-up.
+  on(MessagesActions.loadOlderMessages, (state) => ({
+    ...state,
+    loadingOlder: true,
+    error: null
+  })),
+
+  on(MessagesActions.loadOlderMessagesSuccess, (state, { conversationKey, messages, reachedEnd }) =>
+    messagesAdapter.upsertMany(messages, {
+      ...state,
+      loadingOlder: false,
+      exhaustedConversations: reachedEnd
+        ? { ...state.exhaustedConversations,
+          [conversationKey]: true }
+        : state.exhaustedConversations
+    })
+  ),
+
+  on(MessagesActions.loadOlderMessagesFailure, (state, { error }) => ({
+    ...state,
+    loadingOlder: false,
+    error
+  })),
+
   // Send message
   on(MessagesActions.sendMessage, (state) => ({
     ...state,
@@ -202,7 +237,10 @@ export const messagesReducer = createReducer(
 
     return messagesAdapter.upsertMany(merged, {
       ...state,
-      syncing: false
+      syncing: false,
+      // Peer sync may have inserted messages older than our current oldest;
+      // reopen pagination so the scroll-up handler revisits the DB.
+      exhaustedConversations: {}
     });
   }),
 
@@ -221,7 +259,8 @@ export const messagesReducer = createReducer(
   on(MessagesActions.clearMessages, (state) =>
     messagesAdapter.removeAll({
       ...state,
-      currentRoomId: null
+      currentRoomId: null,
+      exhaustedConversations: {}
     })
   )
 );

toju-app/src/app/store/messages/messages.selectors.ts

@@ -36,6 +36,21 @@ export const selectMessagesSyncing = createSelector(
   (state) => state.syncing
 );
 
+/** Whether a scroll-up older-page DB fetch is currently in flight. */
+export const selectMessagesLoadingOlder = createSelector(
+  selectMessagesState,
+  (state) => state.loadingOlder
+);
+
+/** Whether the given conversation (`${roomId}:${channelId}`) has been
+ * paginated all the way back to the start of the local DB history.
+ */
+export const selectConversationExhausted = (conversationKey: string) =>
+  createSelector(
+    selectMessagesState,
+    (state) => state.exhaustedConversations[conversationKey] === true
+  );
+
 /** Selects the ID of the room whose messages are currently loaded. */
 export const selectCurrentRoomId = createSelector(
   selectMessagesState,