Send a private realm's events once, into a room of its named grantees

packages/host/app/services/matrix-service.ts

@@ -59,6 +59,7 @@ import {
   APP_BOXEL_REALM_EVENT_TYPE,
   APP_BOXEL_REALM_SERVER_EVENT_MSGTYPE,
   APP_BOXEL_REALMS_EVENT_TYPE,
+  APP_BOXEL_REALM_EVENT_ROOM_TYPE,
   APP_BOXEL_REALM_SERVERS_EVENT_TYPE,
   REALMS_LIST_UPDATED_EVENT_TYPE,
   APP_BOXEL_WORKSPACE_FAVORITES_EVENT_TYPE,
@@ -74,6 +75,8 @@ import {
   SLIDING_SYNC_AUTH_ROOM_LIST_NAME,
   SLIDING_SYNC_AUTH_ROOM_TIMELINE_LIMIT,
   SLIDING_SYNC_LIST_RANGE_END,
+  SLIDING_SYNC_REALM_EVENT_ROOM_LIST_NAME,
+  SLIDING_SYNC_REALM_EVENT_ROOM_LIST_RANGE,
   SLIDING_SYNC_TIMEOUT,
   type LLMMode,
   getToolRequests,
@@ -1627,6 +1630,8 @@ export default class MatrixService extends Service {
       ranges: [[0, SLIDING_SYNC_LIST_RANGE_END]],
       filters: {
         is_dm: false,
+        // A private realm's event room isn't a DM either.
+        not_room_types: [APP_BOXEL_REALM_EVENT_ROOM_TYPE],
       },
       timeline_limit: timelineLimit,
       required_state: [['*', '*']],
@@ -1647,6 +1652,18 @@ export default class MatrixService extends Service {
       timeline_limit: SLIDING_SYNC_AUTH_ROOM_TIMELINE_LIMIT,
       required_state: [['*', '*']],
     });
+    // The rooms private realms send their realm events into. Like the session
+    // rooms they aren't AI rooms, so their events are relayed as realm events.
+    lists.set(SLIDING_SYNC_REALM_EVENT_ROOM_LIST_NAME, {
+      ranges: [SLIDING_SYNC_REALM_EVENT_ROOM_LIST_RANGE],
+      filters: {
+        room_types: [APP_BOXEL_REALM_EVENT_ROOM_TYPE],
+      },
+      timeline_limit: SLIDING_SYNC_AUTH_ROOM_TIMELINE_LIMIT,
+      // Only the user's own membership, which event handling checks, rather
+      // than every member of a room many users share.
+      required_state: [['m.room.member', '$ME']],
+    });
     this.slidingSync = new this.matrixSdkLoader.SlidingSync(
       this.client.baseUrl,
       lists,

packages/host/app/services/message-service.ts

@@ -13,6 +13,9 @@ export default class MessageService extends Service {
     new Map();
   @service declare private network: NetworkService;
   @service declare private session: SessionService;
+  // App-scoped wiring, like RealmServerService.eventSubscribers: a service
+  // registers once in its constructor, so logout leaves these in place.
+  private firstSubscriptionListeners = new Set<(realm: string) => void>();
 
   constructor(...args: ConstructorParameters<typeof Service>) {
     super(...args);
@@ -50,6 +53,9 @@ export default class MessageService extends Service {
     let key = this.keyFor(realmURL);
     if (!this.listenerCallbacks.has(key)) {
       this.listenerCallbacks.set(key, []);
+      for (let listener of this.firstSubscriptionListeners) {
+        listener(key);
+      }
     }
 
     // TODO might want to consider making separate subscription methods so that
@@ -65,6 +71,24 @@ export default class MessageService extends Service {
     };
   }
 
+  // The realms that have at least one subscriber.
+  subscribedRealms(): string[] {
+    return [...this.listenerCallbacks.keys()];
+  }
+
+  // Calls `listener` with each realm as it gains its first subscriber, starting
+  // with the realms subscribed to already. A private realm's events arrive in
+  // a room of their own, which RealmServerService joins this way.
+  onFirstSubscription(listener: (realm: string) => void): () => void {
+    this.firstSubscriptionListeners.add(listener);
+    for (let realm of this.listenerCallbacks.keys()) {
+      listener(realm);
+    }
+    return () => {
+      this.firstSubscriptionListeners.delete(listener);
+    };
+  }
+
   private removeSubscriptionCallback(
     realmURL: string,
     cb: (ev: RealmEventContent) => void,

packages/host/app/services/realm-server.ts

@@ -35,6 +35,7 @@ import {
   type PublicRealmSummary,
   type RealmServerTokenSource,
 } from '@cardstack/runtime-common';
+import { REALMS_LIST_UPDATED_EVENT_TYPE } from '@cardstack/runtime-common/matrix-constants';
 import {
   joinDMRoom,
   RealmAuthClient,
@@ -50,13 +51,19 @@ import {
 import { createServerRequestTimingMiddleware } from './client-telemetry';
 
 import type { ExtendedClient } from './matrix-sdk-loader';
+import type MessageService from './message-service';
 import type NetworkService from './network';
 import type RealmService from './realm';
 import type SessionService from './session';
 import type { IEvent } from 'matrix-js-sdk';
 
 const { hostsOwnAssets, resolvedBaseRealmURL } = ENV;
 
+const REALM_EVENT_ROOMS_FIRST_RETRY_MS = 2_000;
+const REALM_EVENT_ROOMS_MAX_RETRY_MS = 60_000;
+// The most realms `/_realm-event-rooms` answers for in one request.
+const REALM_EVENT_ROOMS_PER_REQUEST = 100;
+
 export interface RealmServerTokenClaims {
   user: string;
   sessionRoom: string;
@@ -121,6 +128,7 @@ export default class RealmServerService
   extends Service
   implements RealmServerTokenSource
 {
+  @service declare private messageService: MessageService;
   @service declare private network: NetworkService;
   @service declare private session: SessionService;
   @service declare private realm: RealmService;
@@ -145,11 +153,31 @@ export default class RealmServerService
   private _ready = new Deferred<void>();
   private eventSubscribers: Map<string, RealmServerEventSubscriber[]> =
     new Map();
+  // A private realm sends its realm events into a room of its own, whose
+  // members are the users its grants name; the host joins it when something
+  // first subscribes to the realm (see
+  // `joinRealmEventRoom`). The realms still to be asked about, and every realm
+  // asked about this session.
+  private realmEventRoomsPending = new Set<string>();
+  private realmEventRoomsRequested = new Set<string>();
+  // How long to wait before asking again about realms whose rooms could not
+  // be provided. Doubles on each failure up to a minute; reset on success.
+  private realmEventRoomsRetryMs = REALM_EVENT_ROOMS_FIRST_RETRY_MS;
 
   constructor(owner: Owner) {
     super(owner);
     this.session.register(this);
     this.fetchCatalogRealms();
+    this.messageService.onFirstSubscription((realm) =>
+      this.joinRealmEventRoom(realm),
+    );
+    // A change to which realms the user can read can also change whether a
+    // realm they have open names them, or is world-readable, and so which
+    // room its events arrive in. Ask again about every realm that has a
+    // subscriber.
+    this.subscribeEvent(REALMS_LIST_UPDATED_EVENT_TYPE, async () => {
+      this.reaskRealmEventRooms();
+    });
   }
 
   get ready() {
@@ -178,6 +206,13 @@ export default class RealmServerService
       0,
       this.unreachableRealmServersList.length,
     );
+    // A run in flight belongs to the session that is ending; it must not put
+    // that session's realms back in line for the next one.
+    this.joinRealmEventRoomsTask.cancelAll();
+    this.retryRealmEventRoomsTask.cancelAll();
+    this.realmEventRoomsPending.clear();
+    this.realmEventRoomsRequested.clear();
+    this.realmEventRoomsRetryMs = REALM_EVENT_ROOMS_FIRST_RETRY_MS;
     // Do NOT clear eventSubscribers here: subscriber *wiring* is app-scoped.
     // Services (e.g. BillingService) call subscribeEvent() once in their
     // constructor, for the app's lifetime. Logout is an in-app reset, not a
@@ -696,8 +731,109 @@ export default class RealmServerService
     for (let [realmURL, token] of Object.entries(tokens)) {
       this.realm.getOrCreateRealmResource(realmURL, token);
     }
+    // Realms subscribed to before there was a session to join with.
+    void this.joinRealmEventRoomsTask.perform();
+  }
+
+  // Joins the room the given realm sends its realm events into, when it is a
+  // private realm on this realm server whose grants name the user; the realm
+  // server invites the user before it answers. Other realms' events arrive in
+  // the user's session room. Joining once is enough: the host stays in the
+  // room until a revoked grant removes it, and catches up on what it missed
+  // through sync. Realms asked for in the same turn are asked about together.
+  private joinRealmEventRoom(realmIdentifier: string) {
+    let realmURL = ensureTrailingSlash(
+      resolvedRealmURLHref(this.network.virtualNetwork, realmIdentifier),
+    );
+    if (this.realmEventRoomsRequested.has(realmURL)) {
+      return;
+    }
+    this.realmEventRoomsRequested.add(realmURL);
+    this.realmEventRoomsPending.add(realmURL);
+    void this.joinRealmEventRoomsTask.perform();
   }
 
+  private reaskRealmEventRooms() {
+    this.realmEventRoomsRequested.clear();
+    for (let realm of this.messageService.subscribedRealms()) {
+      this.joinRealmEventRoom(realm);
+    }
+  }
+
+  private joinRealmEventRoomsTask = task({ keepLatest: true }, async () => {
+    await rawTimeout(0);
+    // Without a matrix client there's no session to join with. Signing in
+    // runs this again (see `authenticateToAllAccessibleRealms`).
+    let client = this.client;
+    if (!client || this.realmEventRoomsPending.size === 0) {
+      return;
+    }
+    // Asked about a request's worth at a time; the rest wait for the next run,
+    // started below once this one succeeds.
+    let realms = [...this.realmEventRoomsPending].slice(
+      0,
+      REALM_EVENT_ROOMS_PER_REQUEST,
+    );
+    for (let realm of realms) {
+      this.realmEventRoomsPending.delete(realm);
+    }
+    try {
+      await this.login();
+      // TODO(ADR 004 §1.7–1.9): every realm is asked about at the home realm
+      // server, which answers only for its own realms. That holds while the
+      // host serves realms from its home server alone (see
+      // `assertOwnRealmServer`). Once realms come from several servers, ask
+      // each server about its own realms with that server's session, and
+      // join the rooms through that server's homeserver.
+      let response = await this.network.fetch(
+        `${this.url.href}_realm-event-rooms`,
+        {
+          method: 'POST',
+          headers: {
+            Accept: SupportedMimeType.JSON,
+            'Content-Type': 'application/json',
+            Authorization: `Bearer ${this.token}`,
+          },
+          body: JSON.stringify({ realms }),
+        },
+      );
+      // A 503 still carries the rooms the server could provide; the rest are
+      // asked about again below.
+      let { rooms } = (
+        response.ok || response.status === 503 ? await response.json() : {}
+      ) as { rooms?: Record<string, string> };
+      // Joining a room the user is already in changes nothing, and the SDK
+      // answers it locally once sync has the room.
+      for (let roomId of new Set(Object.values(rooms ?? {}))) {
+        await client.joinRoom(roomId);
+      }
+      if (!response.ok) {
+        throw new Error(`${response.status} from _realm-event-rooms`);
+      }
+      this.realmEventRoomsRetryMs = REALM_EVENT_ROOMS_FIRST_RETRY_MS;
+      if (this.realmEventRoomsPending.size > 0) {
+        void this.joinRealmEventRoomsTask.perform();
+      }
+    } catch (e) {
+      // Asked about again after a pause, and again on the next subscription.
+      for (let realm of realms) {
+        this.realmEventRoomsPending.add(realm);
+      }
+      console.warn(`Unable to join the realm event rooms of ${realms}`, e);
+      let retryMs = this.realmEventRoomsRetryMs;
+      this.realmEventRoomsRetryMs = Math.min(
+        retryMs * 2,
+        REALM_EVENT_ROOMS_MAX_RETRY_MS,
+      );
+      void this.retryRealmEventRoomsTask.perform(retryMs);
+    }
+  });
+
+  private retryRealmEventRoomsTask = restartableTask(async (ms: number) => {
+    await rawTimeout(ms);
+    void this.joinRealmEventRoomsTask.perform();
+  });
+
   private async ensureJoinedSessionRoom(tokens: {
     [realmUrl: string]: string;
   }) {

packages/host/tests/helpers/index.gts

@@ -101,6 +101,7 @@ export {
   resetCatalogRealmURL,
   setRealmArchived,
   setRealmAuthFailure,
+  setRealmEventRoom,
   setupAuthEndpoints,
   setupPublicReadableRealms,
   setCatalogRealmURL,

packages/host/tests/helpers/mock-matrix/_client.ts

@@ -1022,14 +1022,26 @@ export class MockClient implements ExtendedClient {
     for (const [listKey, list] of Object.entries(req.lists || {})) {
       for (let i = 0; i < list.ranges.length; i++) {
         let [start, end] = list.ranges[i];
-        //currently we only filter rooms using is_dm
+        // Filters on is_dm, room_types and not_room_types, the filters the
+        // host's lists use.
         let dmRooms = (await this.getAccountDataFromServer('m.direct')) ?? {};
+        // getRoomState answers with the state event's content.
+        let roomType = (roomId: string): string =>
+          this.serverState.getRoomState(roomId, 'm.room.create', '')?.type;
         let roomsInRange = this.serverState.rooms
           .filter((r) =>
             list.filters?.is_dm
               ? dmRooms[this.loggedInAs!]?.includes(r.id)
               : !dmRooms[this.loggedInAs!]?.includes(r.id),
           )
+          .filter(
+            (r) =>
+              !list.filters?.room_types ||
+              list.filters.room_types.includes(roomType(r.id)),
+          )
+          .filter(
+            (r) => !list.filters?.not_room_types?.includes(roomType(r.id)),
+          )
           .slice(start, end + 1);
 
         for (let j = 0; j < roomsInRange.length; j++) {

packages/host/tests/helpers/mock-matrix/_server-state.ts

@@ -67,6 +67,7 @@ export class ServerState {
     name?: string,
     timestamp: number = this.#now(),
     id?: string,
+    roomType?: string,
   ): string {
     if (document.querySelector('[data-test-throw-room-error]')) {
       throw new Error('Intentional error thrown');
@@ -92,6 +93,7 @@ export class ServerState {
         content: {
           creator: sender,
           room_version: '0',
+          ...(roomType ? { type: roomType } : {}),
         },
       },
       { origin_server_ts: timestamp, state_key: '' },

packages/host/tests/helpers/mock-matrix/_sliding-sync.ts

@@ -98,11 +98,37 @@ export class MockSlidingSync extends SlidingSync {
       return;
     }
 
+    // The room is reported in the list whose room-type filters admit it, as
+    // the homeserver would: a typed room goes to the list that asks for its
+    // type, and to the AI room list only when that list doesn't exclude it.
+    let roomType = (() => {
+      try {
+        // getRoomState answers with the state event's content.
+        return serverState?.getRoomState?.(roomId, 'm.room.create', '')
+          ?.type as string | undefined;
+      } catch {
+        return undefined;
+      }
+    })();
+    let listKey =
+      (roomType &&
+        Object.keys(this._lists).find((key) =>
+          this._lists[key]?.filters?.room_types?.includes(roomType),
+        )) ||
+      'ai-room';
+    if (
+      listKey === 'ai-room' &&
+      roomType &&
+      this._lists['ai-room']?.filters?.not_room_types?.includes(roomType)
+    ) {
+      return;
+    }
+
     // Create a mock sliding sync response that includes the new room
     let mockResponse = {
       pos: String(Date.now()),
       lists: {
-        ['ai-room']: {
+        [listKey]: {
           count: 1,
           ops: [
             {

packages/host/tests/helpers/mock-matrix/_utils.ts

@@ -213,17 +213,21 @@ export class MockUtils {
     name,
     id,
     timestamp,
+    roomType,
   }: {
     sender: string;
     name: string;
     id?: string;
     timestamp?: number;
+    // The room's `m.room.create` type, which sliding sync lists filter on.
+    roomType?: string;
   }) => {
     let roomId = this.testState.sdk!.serverState.createRoom(
       sender,
       name,
       timestamp,
       id,
+      roomType,
     );
     return roomId;
   };