diff --git a/packages/host/app/services/matrix-service.ts b/packages/host/app/services/matrix-service.ts index ba5dba60c42..d2166e3d22a 100644 --- a/packages/host/app/services/matrix-service.ts +++ b/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 world-readable 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,19 @@ export default class MatrixService extends Service { timeline_limit: SLIDING_SYNC_AUTH_ROOM_TIMELINE_LIMIT, required_state: [['*', '*']], }); + // The rooms world-readable 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, diff --git a/packages/host/app/services/message-service.ts b/packages/host/app/services/message-service.ts index 110b955168e..0a365781428 100644 --- a/packages/host/app/services/message-service.ts +++ b/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) { 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 world-readable 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, diff --git a/packages/host/app/services/realm-server.ts b/packages/host/app/services/realm-server.ts index fd8387048fc..12de3603e55 100644 --- a/packages/host/app/services/realm-server.ts +++ b/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,6 +51,7 @@ 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'; @@ -57,6 +59,9 @@ 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; + export interface RealmServerTokenClaims { user: string; sessionRoom: string; @@ -121,6 +126,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 +151,29 @@ export default class RealmServerService private _ready = new Deferred(); private eventSubscribers: Map = new Map(); + // A world-readable realm sends its realm events into a room of its own, + // which the host joins 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(); + private realmEventRoomsRequested = new Set(); + // 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 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 +202,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 +727,98 @@ 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 + // world-readable realm on this realm server; other realms' events arrive in + // the user's session room. Joining once is enough: the host stays in the + // room, 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; + } + let realms = [...this.realmEventRoomsPending]; + this.realmEventRoomsPending.clear(); + 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 }; + // 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; + } 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; }) { diff --git a/packages/host/config/schema/1792465777331_schema.sql b/packages/host/config/schema/1792619085178_schema.sql similarity index 100% rename from packages/host/config/schema/1792465777331_schema.sql rename to packages/host/config/schema/1792619085178_schema.sql diff --git a/packages/host/tests/helpers/index.gts b/packages/host/tests/helpers/index.gts index 0c4bd6267ff..82d89503730 100644 --- a/packages/host/tests/helpers/index.gts +++ b/packages/host/tests/helpers/index.gts @@ -101,6 +101,7 @@ export { resetCatalogRealmURL, setRealmArchived, setRealmAuthFailure, + setRealmEventRoom, setupAuthEndpoints, setupPublicReadableRealms, setCatalogRealmURL, diff --git a/packages/host/tests/helpers/mock-matrix/_client.ts b/packages/host/tests/helpers/mock-matrix/_client.ts index 013e6493e31..e1dd7f002ac 100644 --- a/packages/host/tests/helpers/mock-matrix/_client.ts +++ b/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++) { diff --git a/packages/host/tests/helpers/mock-matrix/_server-state.ts b/packages/host/tests/helpers/mock-matrix/_server-state.ts index ede825abdbd..557a25df8ba 100644 --- a/packages/host/tests/helpers/mock-matrix/_server-state.ts +++ b/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: '' }, diff --git a/packages/host/tests/helpers/mock-matrix/_sliding-sync.ts b/packages/host/tests/helpers/mock-matrix/_sliding-sync.ts index 65790665820..5a9465cd2bc 100644 --- a/packages/host/tests/helpers/mock-matrix/_sliding-sync.ts +++ b/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: [ { diff --git a/packages/host/tests/helpers/mock-matrix/_utils.ts b/packages/host/tests/helpers/mock-matrix/_utils.ts index 1525b449281..e8a73680f14 100644 --- a/packages/host/tests/helpers/mock-matrix/_utils.ts +++ b/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; }; diff --git a/packages/host/tests/helpers/realm-server-mock/index.ts b/packages/host/tests/helpers/realm-server-mock/index.ts index e7957f390a7..80d0a41b437 100644 --- a/packages/host/tests/helpers/realm-server-mock/index.ts +++ b/packages/host/tests/helpers/realm-server-mock/index.ts @@ -58,6 +58,7 @@ function ensureRealmServerMockState( realmPermissions, publicReadableRealms: new Map(), archivedRealms: new Map(), + realmEventRooms: new Map(), handler, ensureSessionRoom: sessionRoomEnsurer, }; @@ -125,6 +126,14 @@ export function setRealmArchived(realmURL: string, archived: boolean) { // Simulate a trusted realm server going unreachable (or recovering) at its // `_realm-auth` endpoint, so boot-assembly graceful-degradation and retry can // be exercised deterministically. +// Makes `/_realm-event-rooms` answer for the given world-readable realm with +// the given room, the one the host should join for its realm events. +export function setRealmEventRoom(realmURL: string, roomId: string) { + let network = getService('network') as NetworkService; + let state = ensureRealmServerMockState(network); + state.realmEventRooms.set(ensureTrailingSlash(realmURL), roomId); +} + export function setRealmAuthFailure(shouldFail: boolean) { let network = getService('network') as NetworkService; let state = ensureRealmServerMockState(network); diff --git a/packages/host/tests/helpers/realm-server-mock/routes.ts b/packages/host/tests/helpers/realm-server-mock/routes.ts index aa2b1bf7980..03bcd88c0dd 100644 --- a/packages/host/tests/helpers/realm-server-mock/routes.ts +++ b/packages/host/tests/helpers/realm-server-mock/routes.ts @@ -630,6 +630,24 @@ function registerAuthRoutes() { }, }); + registerRealmServerRoute({ + path: '/_realm-event-rooms', + handler: async (req, _url, state: RealmServerMockState) => { + let { realms } = (await req.json()) as { realms: string[] }; + let rooms: Record = {}; + for (let realm of realms) { + let roomId = state.realmEventRooms.get(ensureTrailingSlash(realm)); + if (roomId) { + rooms[realm] = roomId; + } + } + return new Response(JSON.stringify({ rooms }), { + status: 200, + headers: { 'content-type': SupportedMimeType.JSON }, + }); + }, + }); + registerRealmServerRoute({ path: '/_server-session', handler: async (req) => { diff --git a/packages/host/tests/helpers/realm-server-mock/types.ts b/packages/host/tests/helpers/realm-server-mock/types.ts index 285af6e94d9..12d8206a5dc 100644 --- a/packages/host/tests/helpers/realm-server-mock/types.ts +++ b/packages/host/tests/helpers/realm-server-mock/types.ts @@ -16,6 +16,10 @@ export type RealmServerMockState = { // Realms archived via POST /_archive-realm, keyed by normalized URL. The // value carries the archived-at timestamp surfaced by GET /_archived-realms. archivedRealms: Map; + // The rooms `/_realm-event-rooms` answers with, keyed by normalized realm + // URL. Empty unless a test sets one, so the host joins no room and every + // realm's events keep arriving in its session room. + realmEventRooms: Map; mountedVirtualNetwork?: unknown; ensureSessionRoom?: EnsureSessionRoom; // When true, `_realm-auth` responds 503 — used to simulate a trusted realm diff --git a/packages/host/tests/integration/matrix-service-sliding-sync-test.ts b/packages/host/tests/integration/matrix-service-sliding-sync-test.ts index 1bcfb958e65..bd6485061c9 100644 --- a/packages/host/tests/integration/matrix-service-sliding-sync-test.ts +++ b/packages/host/tests/integration/matrix-service-sliding-sync-test.ts @@ -5,6 +5,7 @@ import { module, test } from 'qunit'; import { baseRealm } from '@cardstack/runtime-common'; import { + APP_BOXEL_REALM_EVENT_ROOM_TYPE, SLIDING_SYNC_AI_ROOM_LIST_NAME, SLIDING_SYNC_AI_ROOM_TIMELINE_LIMIT, } from '@cardstack/runtime-common/matrix-constants'; @@ -63,7 +64,7 @@ module( ); assert.deepEqual( aiListCalls[0]?.list.filters, - { is_dm: false }, + { is_dm: false, not_room_types: [APP_BOXEL_REALM_EVENT_ROOM_TYPE] }, 'AI-room list filters preserved across the bump', ); }); diff --git a/packages/host/tests/integration/message-service-test.ts b/packages/host/tests/integration/message-service-test.ts index b623f8a9ee9..d23e0591db9 100644 --- a/packages/host/tests/integration/message-service-test.ts +++ b/packages/host/tests/integration/message-service-test.ts @@ -1,11 +1,21 @@ +import { waitUntil } from '@ember/test-helpers'; + import { getService } from '@universal-ember/test-support'; import { module, test } from 'qunit'; import { baseRealm } from '@cardstack/runtime-common'; +import { + APP_BOXEL_REALM_EVENT_ROOM_TYPE, + APP_BOXEL_REALM_EVENT_TYPE, +} from '@cardstack/runtime-common/matrix-constants'; import ENV from '@cardstack/host/config/environment'; -import { testRealmURL, setupIntegrationTestRealm } from '../helpers'; +import { + testRealmURL, + setRealmEventRoom, + setupIntegrationTestRealm, +} from '../helpers'; import { setupMockMatrix } from '../helpers/mock-matrix'; import { setupRenderingTest } from '../helpers/setup'; @@ -117,3 +127,83 @@ module('Integration | message-service | realm subscriptions', function (hooks) { } }); }); + +// A world-readable realm sends its realm events into a room of its own, which +// the host joins when something first subscribes to the realm. +module('Integration | message-service | realm event rooms', function (hooks) { + setupRenderingTest(hooks); + + let mockMatrixUtils = setupMockMatrix(hooks, { + loggedInAs: '@testuser:localhost', + activeRealms: [baseRealm.url, testRealmURL], + autostart: true, + }); + + hooks.beforeEach(async function () { + await setupIntegrationTestRealm({ mockMatrixUtils, contents: {} }); + }); + + test('subscribing to a world-readable realm joins the room its events go to', async function (assert) { + let publicRealmURL = 'http://localhost:4205/public/'; + // The realm's own user creates the room; the host is not in it yet. + let roomId = mockMatrixUtils.createAndJoinRoom({ + sender: '@public_realm:localhost', + name: `Realm events: ${publicRealmURL}`, + id: '!public-realm-events:localhost', + roomType: APP_BOXEL_REALM_EVENT_ROOM_TYPE, + }); + setRealmEventRoom(publicRealmURL, roomId); + let membership = () => { + try { + return mockMatrixUtils.getRoomState( + roomId, + 'm.room.member', + '@testuser:localhost', + ).membership; + } catch { + return undefined; + } + }; + assert.strictEqual(membership(), undefined, 'the host starts outside it'); + let messageService = getService('message-service'); + + let heard: RealmEventContent[] = []; + let unsubscribe = messageService.subscribe(publicRealmURL, (event) => { + heard.push(event); + }); + try { + await waitUntil(() => membership() === 'join', { + timeoutMessage: `the host did not join ${roomId}`, + }); + assert.strictEqual( + membership(), + 'join', + 'the host joined the realm event room', + ); + + // An event the realm sends into the room reaches the realm's + // subscribers, which it would not if the room were taken for an AI room. + mockMatrixUtils.simulateRemoteMessage( + roomId, + '@public_realm:localhost', + { + eventName: 'index', + indexType: 'incremental', + invalidations: [`${publicRealmURL}card`], + realmURL: publicRealmURL, + }, + { type: APP_BOXEL_REALM_EVENT_TYPE }, + ); + await waitUntil(() => heard.length > 0, { + timeoutMessage: `no realm event arrived through ${roomId}`, + }); + assert.deepEqual( + heard.map((event) => event.eventName), + ['index'], + "the realm's subscriber heard the event sent into its room", + ); + } finally { + unsubscribe(); + } + }); +}); diff --git a/packages/postgres/migrations/1792619085178_create-realm-event-rooms.js b/packages/postgres/migrations/1792619085178_create-realm-event-rooms.js new file mode 100644 index 00000000000..bd18d0fdc16 --- /dev/null +++ b/packages/postgres/migrations/1792619085178_create-realm-event-rooms.js @@ -0,0 +1,21 @@ +// The Matrix room a world-readable realm sends its realm events into, shared by +// every host that has joined it. A realm has at most one; its row is written +// the first time a host asks for the room. + +exports.up = (pgm) => { + pgm.createTable('realm_event_rooms', { + realm_url: { + type: 'varchar', + notNull: true, + primaryKey: true, + }, + room_id: { + type: 'varchar', + notNull: true, + }, + }); +}; + +exports.down = (pgm) => { + pgm.dropTable('realm_event_rooms', { ifExists: true }); +}; diff --git a/packages/postgres/scripts/schema-dump.sh b/packages/postgres/scripts/schema-dump.sh index 83e77530011..f769bb5ccea 100755 --- a/packages/postgres/scripts/schema-dump.sh +++ b/packages/postgres/scripts/schema-dump.sh @@ -41,6 +41,7 @@ docker exec boxel-pg pg_dump \ --exclude-table-and-children=billable_call_reservations \ --exclude-table-and-children=claimed_domains_for_sites \ --exclude-table-and-children=session_rooms \ + --exclude-table-and-children=realm_event_rooms \ --exclude-table-and-children=host_shell_generation \ --no-tablespaces \ --no-table-access-method \ diff --git a/packages/realm-server/handlers/handle-realm-event-rooms.ts b/packages/realm-server/handlers/handle-realm-event-rooms.ts new file mode 100644 index 00000000000..712d493fa0a --- /dev/null +++ b/packages/realm-server/handlers/handle-realm-event-rooms.ts @@ -0,0 +1,127 @@ +import type Koa from 'koa'; +import { + fetchRealmEventRoom, + logger, + SupportedMimeType, +} from '@cardstack/runtime-common'; +import type { CreateRoutesArgs } from '../routes.ts'; +import { normalizeRealmURL } from '../utils/realm-url.ts'; +import { + fetchRequestFromContext, + sendResponseForBadRequest, + setContextResponse, +} from '../middleware/index.ts'; + +const log = logger('realm:events'); + +// A host asks about the realms it has open, batched as they open; this is +// well above that, and bounds what one request can make the server do. +const MAX_REALMS_PER_REQUEST = 100; +// Looking a realm's room up can mount the realm and create its room, so a +// request works through its realms a few at a time. +const REALMS_IN_FLIGHT = 4; + +// Answers a signed-in host with the rooms that the world-readable realms it +// names send their realm events into, so it can join them. A realm's room is +// created the first time a host asks for it. A realm that isn't +// world-readable, or that this server doesn't serve, is left out of the +// answer: its events reach the host through the user's session room, if they +// reach it at all. A realm is named by its URL or by a registered prefix. +// +// When a realm's room can't be looked up or created, the answer is a 503 +// carrying the rooms that were found, so the host asks again rather than take +// the realm for one that isn't world-readable. +// +// Body: `{ realms: string[] }`. Answer: `{ rooms: { [realm]: roomId } }`, +// keyed by each realm as the request spelled it. +export default function handleRealmEventRooms({ + dbAdapter, + reconciler, + virtualNetwork, +}: CreateRoutesArgs): (ctxt: Koa.Context, next: Koa.Next) => Promise { + return async function (ctxt: Koa.Context, _next: Koa.Next) { + let request = await fetchRequestFromContext(ctxt); + let body: { realms?: unknown }; + try { + body = JSON.parse(await request.text()); + } catch { + await sendResponseForBadRequest(ctxt, 'Request body is not valid JSON'); + return; + } + let realms = body?.realms; + if ( + !Array.isArray(realms) || + !realms.every((realm) => typeof realm === 'string') + ) { + await sendResponseForBadRequest( + ctxt, + 'Request body must be { "realms": string[] }', + ); + return; + } + if (realms.length > MAX_REALMS_PER_REQUEST) { + await sendResponseForBadRequest( + ctxt, + `A request may name at most ${MAX_REALMS_PER_REQUEST} realms`, + ); + return; + } + + let rooms: Record = {}; + let failed = false; + let lookUp = async (realm: string) => { + let identifier = virtualNetwork.isRegisteredPrefix(realm) + ? virtualNetwork.toURL(realm).href + : realm; + let realmURL = normalizeRealmURL(identifier)?.href; + if (!realmURL) { + return; + } + try { + let { worldReadable, roomId } = await fetchRealmEventRoom( + dbAdapter, + realmURL, + ); + if (!worldReadable) { + return; + } + if (!roomId) { + // The realm creates its room, through the matrix client it sends + // its realm events with, so that client's user is the one allowed + // to send in it. + let mounted = await reconciler.lookupOrMount(realmURL); + roomId = (await mounted?.ensureRealmEventRoom()) ?? null; + } + if (roomId) { + rooms[realm] = roomId; + } + } catch (e) { + failed = true; + log.error(`Unable to provide the realm event room of ${realmURL}`, e); + } + }; + let queue = [...new Set(realms as string[])]; + await Promise.all( + Array.from( + { length: Math.min(REALMS_IN_FLIGHT, queue.length) }, + async () => { + for ( + let realm = queue.shift(); + realm !== undefined; + realm = queue.shift() + ) { + await lookUp(realm); + } + }, + ), + ); + + await setContextResponse( + ctxt, + new Response(JSON.stringify({ rooms }), { + status: failed ? 503 : 200, + headers: { 'content-type': SupportedMimeType.JSON }, + }), + ); + }; +} diff --git a/packages/realm-server/node-realm.ts b/packages/realm-server/node-realm.ts index 29efabfc740..c483beb952d 100644 --- a/packages/realm-server/node-realm.ts +++ b/packages/realm-server/node-realm.ts @@ -12,7 +12,9 @@ import { unixTime, type ResponseWithNodeStream, type TokenClaims, + clearRealmEventRoom, clearSessionRoom, + fetchRealmEventRoom, fetchRealmSessionRooms, isRealmServerNotInRoomError, } from '@cardstack/runtime-common'; @@ -496,6 +498,72 @@ export class NodeAdapter implements RealmAdapter { return []; } + // A world-readable realm's events go once into its own room, which every + // host using the realm joins, instead of into the session room of every + // user who can read it. Until a host asks for that room it doesn't exist, + // and no one can be listening. + let realmEventRoom; + try { + realmEventRoom = await fetchRealmEventRoom(dbAdapter, realmUrl); + } catch (e) { + realmEventsLog.error('Error looking up the realm event room', e); + return []; + } + if (realmEventRoom.worldReadable) { + let roomId = realmEventRoom.roomId; + if (!roomId) { + realmEventsLog.debug( + `${realmUrl} has no realm event room yet, skipping broadcast`, + ); + return []; + } + return [ + { + roomId, + send: async (txnId: string) => { + if (!matrixClient.isLoggedIn()) { + await matrixClient.login(); + } + await matrixClient.sendEvent( + roomId, + APP_BOXEL_REALM_EVENT_TYPE, + event, + txnId, + ); + }, + onFailure: async (e: unknown) => { + // A stored room the server user is no longer in, as after a + // homeserver reset, is forgotten: the next host that asks for the + // realm's room gets a new one. + if (isRealmServerNotInRoomError(e, roomId)) { + try { + let cleared = await clearRealmEventRoom( + dbAdapter, + realmUrl, + roomId, + ); + realmEventsLog.warn( + `Forgetting realm event room ${roomId} of ${realmUrl}; realm server is no longer in the room`, + { cleared }, + ); + return; + } catch (cleanupError) { + realmEventsLog.error( + `Failed to forget realm event room ${roomId} of ${realmUrl}`, + cleanupError, + ); + } + } + realmEventsLog.error( + `Unable to send event in realm event room ${roomId} for ${realmUrl}: ${ + e instanceof Error ? e.message : String(e) + }`, + ); + }, + }, + ]; + } + let dmRooms = await this.waitForSessionRooms( dbAdapter, realmUrl, diff --git a/packages/realm-server/routes.ts b/packages/realm-server/routes.ts index eee42268efb..74c7a36dcfc 100644 --- a/packages/realm-server/routes.ts +++ b/packages/realm-server/routes.ts @@ -61,6 +61,7 @@ import handleOpenRouterPassthrough from './handlers/handle-openrouter-passthroug import handlePostDeployment from './handlers/handle-post-deployment.ts'; import { handleCheckBoxelDomainAvailabilityRequest } from './handlers/handle-check-boxel-domain-availability.ts'; import handleRealmAuth from './handlers/handle-realm-auth.ts'; +import handleRealmEventRooms from './handlers/handle-realm-event-rooms.ts'; import handleSetRealmCatalog from './handlers/handle-set-realm-catalog.ts'; import handleDelegateSession from './handlers/handle-delegate-session.ts'; import handleWorkerRequest from './handlers/handle-worker-request.ts'; @@ -618,6 +619,11 @@ export function createRoutes(args: CreateRoutesArgs) { jwtMiddleware(args.realmSecretSeed, args.dbAdapter), handleRealmAuth(args), ); + router.post( + '/_realm-event-rooms', + jwtMiddleware(args.realmSecretSeed, args.dbAdapter), + handleRealmEventRooms(args), + ); // Shared-secret authenticated (HMAC over body + timestamp); auth is handled // inside the handler because the signature covers the request body. router.post('/_delegate-session', handleDelegateSession(args)); diff --git a/packages/realm-server/scripts/load-harness/lib/realm-events.ts b/packages/realm-server/scripts/load-harness/lib/realm-events.ts index 1025d4aa581..0fdfa751c97 100644 --- a/packages/realm-server/scripts/load-harness/lib/realm-events.ts +++ b/packages/realm-server/scripts/load-harness/lib/realm-events.ts @@ -1,8 +1,9 @@ // Subscribe to a realm's index events the way the host does. // -// The realm server broadcasts `app.boxel.realm-event` into each user's DM -// session room over Matrix, and the host relays what arrives there into its -// live-search resources. A driver that wants to behave like a browser has to +// The realm server broadcasts `app.boxel.realm-event` over Matrix: a +// world-readable realm's events into one room for the realm that every host +// using it joins, and any other realm's into the DM session room of each user +// it grants. The host relays what arrives into its live-search resources. A driver that wants to behave like a browser has to // listen on that channel rather than infer invalidation from its own writes — // which is the difference between modelling the fan-out and measuring it. // @@ -73,6 +74,60 @@ export async function joinInvitedRooms({ }; } +// Joins the rooms the given world-readable realms send their realm events +// into, the way the host does when it first subscribes to a realm. The realm +// server answers only for world-readable realms, so a realm that isn't joins +// nothing and its events arrive in the session room. Returns how many rooms +// were joined. +export async function joinRealmEventRooms({ + realmServerUrl, + serverToken, + matrixUrl, + accessToken, + realms, +}: { + realmServerUrl: string; + serverToken: string; + matrixUrl: string; + accessToken: string; + realms: string[]; +}): Promise { + let response = await fetch( + `${realmServerUrl.replace(/\/$/, '')}/_realm-event-rooms`, + { + method: 'POST', + headers: { + Accept: 'application/json', + 'Content-Type': 'application/json', + Authorization: serverToken, + }, + body: JSON.stringify({ realms }), + }, + ); + if (!response.ok) { + throw new Error(`_realm-event-rooms failed: ${response.status}`); + } + let { rooms } = (await response.json()) as { + rooms?: Record; + }; + let roomIds = [...new Set(Object.values(rooms ?? {}))]; + let base = matrixUrl.replace(/\/$/, ''); + for (let roomId of roomIds) { + let join = await fetch( + `${base}/_matrix/client/v3/join/${encodeURIComponent(roomId)}`, + { + method: 'POST', + headers: { Authorization: `Bearer ${accessToken}` }, + body: '{}', + }, + ); + if (!join.ok) { + throw new Error(`join ${roomId} failed: ${join.status}`); + } + } + return roomIds.length; +} + export async function initialSyncToken({ matrixUrl, accessToken, diff --git a/packages/realm-server/scripts/load-harness/run-load.ts b/packages/realm-server/scripts/load-harness/run-load.ts index af8a52ac6d0..366f256d622 100644 --- a/packages/realm-server/scripts/load-harness/run-load.ts +++ b/packages/realm-server/scripts/load-harness/run-load.ts @@ -51,6 +51,7 @@ import { eventCannotMatch, initialSyncToken, joinInvitedRooms, + joinRealmEventRooms, streamRealmEvents, type RealmEvent, } from './lib/realm-events.ts'; @@ -580,10 +581,9 @@ async function modelCall(session: Session): Promise { } } -// One Matrix long-poll per session, fanned out to that session's reader. The -// realm broadcasts into each user's own DM session room, so every simulated -// user needs its own stream — which is also what makes the event volume here -// comparable to the cohort's. +// One Matrix long-poll per session, fanned out to that session's reader. Every +// simulated user reads its own stream, as each browser does, whether the realm +// broadcasts into each user's DM session room or into one room they all share. let subscriptions = new Map void)[]>(); let subscribeAbort = new AbortController(); @@ -610,10 +610,19 @@ function subscribeSession( matrixUrl: args.matrixUrl, accessToken: session.accessToken, }); + // A world-readable realm's events go to a room of its own instead, + // which the host joins when it first subscribes to the realm. + joined += await joinRealmEventRooms({ + realmServerUrl: args.realmServerUrl, + serverToken: session.serverToken, + matrixUrl: args.matrixUrl, + accessToken: session.accessToken, + realms: [realmUrl], + }); if (joined === 0) { stats.sessionsWithoutRooms++; console.error( - ` ${session.username} has no realm session room, so it will receive\n` + + ` ${session.username} is in no room realm events go to, so it will receive\n` + ` no events — a low re-run count would mean nothing.`, ); } diff --git a/packages/realm-server/tests/helpers/index.ts b/packages/realm-server/tests/helpers/index.ts index a2a6528f924..19e7f795e86 100644 --- a/packages/realm-server/tests/helpers/index.ts +++ b/packages/realm-server/tests/helpers/index.ts @@ -2355,6 +2355,7 @@ export function setupMatrixRoom( }); let testAuthRoomId: string | undefined; + let testRealmEventRoomId: string | undefined; hooks.beforeEach(async function () { await matrixClient.login(); @@ -2388,15 +2389,52 @@ export function setupMatrixRoom( } testAuthRoomId = payload.sessionRoom; + + // A world-readable test realm sends its realm events into a room of its + // own instead, which a host joins the way this does. + let eventRoomsResponse = await ( + realmSetup.serverRequest ?? realmSetup.request + ) + .post('/_realm-event-rooms') + .set('Accept', 'application/json') + .set('Content-Type', 'application/json') + .set('Authorization', jwt) + .send(JSON.stringify({ realms: [realmSetup.testRealm.url] })); + if (eventRoomsResponse.status !== 200) { + throw new Error( + `Realm server did not answer /_realm-event-rooms: ${eventRoomsResponse.status} ${eventRoomsResponse.text}`, + ); + } + testRealmEventRoomId = ( + eventRoomsResponse.body as { rooms: Record } + ).rooms[realmSetup.testRealm.url]; + if (testRealmEventRoomId && !rooms.includes(testRealmEventRoomId)) { + await matrixClient.joinRoom(testRealmEventRoomId); + } }); return { matrixClient, - // Every event the room received at or after `since`, however many that - // is. The comparison is inclusive so an event sent in the same millisecond - // the caller recorded its start time still counts. + // Every event the realm's events reach this user through at or after + // `since`, however many that is, newest first: the session room, and the + // realm's event room when it is world-readable. The comparison is + // inclusive so an event sent in the same millisecond the caller recorded + // its start time still counts. getMessagesSince: async function (since: number) { - return await matrixClient.roomMessagesSince(testAuthRoomId!, since); + let messages = await matrixClient.roomMessagesSince( + testAuthRoomId!, + since, + ); + if (testRealmEventRoomId) { + messages = [ + ...messages, + ...(await matrixClient.roomMessagesSince( + testRealmEventRoomId, + since, + )), + ].sort((a, b) => b.origin_server_ts - a.origin_server_ts); + } + return messages; }, }; } diff --git a/packages/realm-server/tests/node-realm-test.ts b/packages/realm-server/tests/node-realm-test.ts index d0c43dc266b..44020f656d9 100644 --- a/packages/realm-server/tests/node-realm-test.ts +++ b/packages/realm-server/tests/node-realm-test.ts @@ -5,8 +5,10 @@ import { tmpdir } from 'os'; import fsExtra from 'fs-extra'; import type { PgAdapter } from '@cardstack/postgres'; import { + fetchRealmEventRoom, fetchSessionRoom, insertPermissions, + insertRealmEventRoom, upsertSessionRoom, } from '@cardstack/runtime-common'; import type { MatrixClient } from '@cardstack/runtime-common/matrix-client'; @@ -245,6 +247,121 @@ module(basename(import.meta.filename), function (hooks) { assert.strictEqual(logins, 2, 'the login is tried again after a pause'); assert.deepEqual(sent, [staleRoomId], 'the event still reaches the room'); }); + + function recordingMatrixClient() { + let sentTo: string[] = []; + let matrixClient = { + login: async () => undefined, + getUserId: () => '@realm_server:localhost', + isLoggedIn: () => true, + nextTxnId: () => 'txn', + sendEvent: async (roomId: string) => { + sentTo.push(roomId); + return '$event'; + }, + } as unknown as MatrixClient; + return { matrixClient, sentTo }; + } + + test("sends a world-readable realm's event once, into the realm's event room", async function (assert) { + await insertUser( + dbAdapter, + '@alice:localhost', + 'cus_alice', + 'alice@example.com', + ); + await upsertSessionRoom(dbAdapter, '@alice:localhost', '!alice:localhost'); + await insertUser(dbAdapter, '@bob:localhost', 'cus_bob', 'bob@example.com'); + await upsertSessionRoom(dbAdapter, '@bob:localhost', '!bob:localhost'); + await insertPermissions(dbAdapter, realmURL, { + '*': ['read'], + '@alice:localhost': ['read', 'write'], + }); + await insertRealmEventRoom(dbAdapter, realmURL.href, '!events:localhost'); + + let { matrixClient, sentTo } = recordingMatrixClient(); + await new NodeAdapter('/tmp').broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + ); + + assert.deepEqual( + sentTo, + ['!events:localhost'], + 'sent to the realm event room and to no session room', + ); + }); + + test('sends nothing for a world-readable realm that has no event room yet', async function (assert) { + await insertUser( + dbAdapter, + '@alice:localhost', + 'cus_alice', + 'alice@example.com', + ); + await upsertSessionRoom(dbAdapter, '@alice:localhost', '!alice:localhost'); + await insertPermissions(dbAdapter, realmURL, { '*': ['read'] }); + + let { matrixClient, sentTo } = recordingMatrixClient(); + await new NodeAdapter('/tmp').broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + ); + + assert.deepEqual(sentTo, [], 'no room was sent to'); + }); + + test("forgets a world-readable realm's event room the realm server is no longer in", async function (assert) { + await insertPermissions(dbAdapter, realmURL, { '*': ['read'] }); + await insertRealmEventRoom(dbAdapter, realmURL.href, '!events:localhost'); + + let matrixClient = { + login: async () => undefined, + getUserId: () => '@realm_server:localhost', + isLoggedIn: () => true, + nextTxnId: () => 'txn-1', + sendEvent: async () => { + throw new Error( + `Unable to send room event 'app.boxel.realm-event' to room !events:localhost: status 403 - {"errcode":"M_FORBIDDEN","error":"User @realm_server:localhost not in room !events:localhost"}`, + ); + }, + } as unknown as MatrixClient; + + await new NodeAdapter( + '/tmp', + false, + new RealmEventDispatcher(), + ).broadcastRealmEvent(makeEvent(), realmURL.href, matrixClient, dbAdapter); + + assert.deepEqual( + await fetchRealmEventRoom(dbAdapter, realmURL.href), + { worldReadable: true, roomId: null }, + 'the next host to ask gets a new room', + ); + }); + + test("sends a realm's event to session rooms once it is no longer world-readable", async function (assert) { + await insertSessionForAlice(); + await insertRealmEventRoom(dbAdapter, realmURL.href, '!events:localhost'); + + let { matrixClient, sentTo } = recordingMatrixClient(); + await new NodeAdapter('/tmp').broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + ); + + assert.deepEqual( + sentTo, + [staleRoomId], + "sent to alice's session room and not to the realm's old event room", + ); + }); }); module(`${basename(import.meta.filename)} | file stat probing`, function () { diff --git a/packages/realm-server/tests/realm-endpoints/permissions-test.ts b/packages/realm-server/tests/realm-endpoints/permissions-test.ts index 3c8d4f7d19b..ff97d1857ec 100644 --- a/packages/realm-server/tests/realm-endpoints/permissions-test.ts +++ b/packages/realm-server/tests/realm-endpoints/permissions-test.ts @@ -170,6 +170,38 @@ module(`realm-endpoints/${basename(import.meta.filename)}`, function () { ); }); + test("PATCH /_permissions of the '*' grant pushes realms-list-updated to every user the realm names", async function (assert) { + // Making a realm world-readable moves its events into a room of its + // own, so a host that has it open must ask again where to listen. + await insertPermissions(dbAdapter, testRealmURL, { + [granteeUserId]: ['read'], + }); + let response = await request + .patch('/_permissions') + .set('Accept', 'application/vnd.api+json') + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'mary', [ + 'read', + 'write', + 'realm-owner', + ])}`, + ) + .send({ + data: { + id: testRealmHref, + type: 'permissions', + attributes: { permissions: { '*': ['read'] } }, + }, + }); + assert.strictEqual(response.status, 200, 'the grant succeeds'); + + assert.ok( + await receivedRealmsListUpdated(matrixClient, sessionRoom), + 'a realms-list-updated event reached the session room of a user the realm already named', + ); + }); + test('PATCH /_permissions still succeeds when the grantee has no session room', async function (assert) { // A user who never established a session room (e.g. a fresh // Synapse-registered account) has nowhere to deliver to. The push is a diff --git a/packages/realm-server/tests/realm-event-rooms-test.ts b/packages/realm-server/tests/realm-event-rooms-test.ts new file mode 100644 index 00000000000..3d65c66d64c --- /dev/null +++ b/packages/realm-server/tests/realm-event-rooms-test.ts @@ -0,0 +1,155 @@ +import QUnit from 'qunit'; +const { module, test } = QUnit; +import type { SuperTest, Test as SupertestTest } from 'supertest'; +import { basename } from 'path'; + +import type { PgAdapter } from '@cardstack/postgres'; +import { insertPermissions } from '@cardstack/runtime-common'; +import { passwordFromSeed } from '@cardstack/runtime-common/matrix-client'; +import { + APP_BOXEL_REALM_EVENT_ROOM_TYPE, + APP_BOXEL_REALM_EVENT_TYPE, +} from '@cardstack/runtime-common/matrix-constants'; + +import { + matrixURL, + realmSecretSeed, + setupPermissionedRealmCached, + testRealmHref, +} from './helpers/index.ts'; +import { createJWT as createRealmServerJWT } from '../utils/jwt.ts'; + +module(basename(import.meta.filename), function () { + module('realm event rooms handler', function (hooks) { + let dbAdapter: PgAdapter; + let request: SuperTest; + const matrixUserId = '@test_realm:localhost'; + + setupPermissionedRealmCached(hooks, { + fixture: 'blank', + permissions: { + '*': ['read'], + '@node-test_realm:localhost': ['read', 'realm-owner'], + }, + onRealmSetup: ({ dbAdapter: adapter, request: req }) => { + dbAdapter = adapter; + request = req; + }, + }); + + function realmEventRooms(realms: string[]) { + return request + .post('/_realm-event-rooms') + .set('Accept', 'application/json') + .set('Content-Type', 'application/json') + .set( + 'Authorization', + `Bearer ${createRealmServerJWT( + { user: matrixUserId, sessionRoom: 'server-session-room' }, + realmSecretSeed, + )}`, + ) + .send(JSON.stringify({ realms })); + } + + // A client for a user other than the realm's, the way a host is one. + async function loginAsMember() { + let response = await fetch( + new URL('_matrix/client/v3/login', matrixURL), + { + method: 'POST', + body: JSON.stringify({ + identifier: { type: 'm.id.user', user: 'test_realm' }, + password: await passwordFromSeed('test_realm', realmSecretSeed), + type: 'm.login.password', + }), + }, + ); + let { access_token } = (await response.json()) as { + access_token: string; + }; + return (path: string, init: RequestInit = {}) => + fetch(new URL(`_matrix/client/v3/${path}`, matrixURL), { + ...init, + headers: { Authorization: `Bearer ${access_token}` }, + }); + } + + test('answers a world-readable realm with the same room each time', async function (assert) { + let first = await realmEventRooms([testRealmHref]); + assert.strictEqual(first.status, 200, 'HTTP 200 status'); + let roomId = first.body.rooms[testRealmHref]; + assert.ok(roomId, 'the realm has an event room'); + + let second = await realmEventRooms([testRealmHref]); + assert.strictEqual( + second.body.rooms[testRealmHref], + roomId, + 'asking again answers with the same room', + ); + }); + + test('leaves out a realm that is not world-readable', async function (assert) { + let privateRealmURL = new URL('http://127.0.0.1:4444/private/'); + await insertPermissions(dbAdapter, privateRealmURL, { + [matrixUserId]: ['read', 'write'], + }); + + let response = await realmEventRooms([ + privateRealmURL.href, + 'http://127.0.0.1:4444/unknown/', + ]); + + assert.strictEqual(response.status, 200, 'HTTP 200 status'); + assert.deepEqual(response.body, { rooms: {} }, 'no room is answered'); + }); + + test('rejects a request without a session', async function (assert) { + let response = await request + .post('/_realm-event-rooms') + .set('Accept', 'application/json') + .set('Content-Type', 'application/json') + .send(JSON.stringify({ realms: [testRealmHref] })); + + assert.strictEqual(response.status, 401, 'HTTP 401 status'); + }); + + test('rejects a body without a list of realms', async function (assert) { + let response = await realmEventRooms('not a list' as unknown as string[]); + + assert.strictEqual(response.status, 400, 'HTTP 400 status'); + }); + + test('lets any user join the room, and only the realm send in it', async function (assert) { + let roomId: string = (await realmEventRooms([testRealmHref])).body.rooms[ + testRealmHref + ]; + let member = await loginAsMember(); + + let join = await member(`join/${encodeURIComponent(roomId)}`, { + method: 'POST', + body: '{}', + }); + assert.strictEqual(join.status, 200, 'a user who was not invited joins'); + + let create = await member( + `rooms/${encodeURIComponent(roomId)}/state/m.room.create/`, + ); + assert.strictEqual( + ((await create.json()) as { type?: string }).type, + APP_BOXEL_REALM_EVENT_ROOM_TYPE, + 'the room has the realm event room type', + ); + + let send = await member( + `rooms/${encodeURIComponent(roomId)}/send/${APP_BOXEL_REALM_EVENT_TYPE}/forged-${Date.now()}`, + { method: 'PUT', body: JSON.stringify({ eventName: 'index' }) }, + ); + assert.strictEqual( + send.status, + 403, + 'a member may not send a realm event into the room', + ); + }); + }); +}); diff --git a/packages/realm-server/tests/session-room-queries-test.ts b/packages/realm-server/tests/session-room-queries-test.ts index 9bc2d561caa..f7303c2f441 100644 --- a/packages/realm-server/tests/session-room-queries-test.ts +++ b/packages/realm-server/tests/session-room-queries-test.ts @@ -5,8 +5,11 @@ import type { PgAdapter } from '@cardstack/postgres'; import { insertPermissions } from '@cardstack/runtime-common'; import { clearSessionRoom, + fetchRealmEventRoom, fetchSessionRoom, fetchRealmSessionRooms, + clearRealmEventRoom, + insertRealmEventRoom, upsertSessionRoom, } from '@cardstack/runtime-common/db-queries/session-room-queries'; import { setupDB, insertUser } from './helpers/index.ts'; @@ -199,8 +202,7 @@ module(basename(import.meta.filename), function () { ); }); - test('returns all users with session rooms when the realm is world-readable', async function (assert) { - // Create multiple users with session rooms + test('a world-readable grant adds no users', async function (assert) { await insertUser( dbAdapter, '@alice:localhost', @@ -213,70 +215,6 @@ module(basename(import.meta.filename), function () { '!room-alice:localhost', ); - await insertUser( - dbAdapter, - '@bob:localhost', - 'cus_bob', - 'bob@example.com', - ); - await upsertSessionRoom( - dbAdapter, - '@bob:localhost', - '!room-bob:localhost', - ); - - await insertUser( - dbAdapter, - '@charlie:localhost', - 'cus_charlie', - 'charlie@example.com', - ); - await upsertSessionRoom( - dbAdapter, - '@charlie:localhost', - '!room-charlie:localhost', - ); - - // World-readable realm: username='*' with read=true - // No per-user permission rows for alice, bob, or charlie - await insertPermissions(dbAdapter, realmURL, { - '*': ['read'], - }); - - let result = await fetchRealmSessionRooms(dbAdapter, realmURL.href); - - assert.deepEqual( - result, - { - '@alice:localhost': '!room-alice:localhost', - '@bob:localhost': '!room-bob:localhost', - '@charlie:localhost': '!room-charlie:localhost', - }, - 'returns ALL users with session rooms for world-readable realm', - ); - }); - - test('world-readable realm still excludes users without session rooms', async function (assert) { - await insertUser( - dbAdapter, - '@alice:localhost', - 'cus_alice', - 'alice@example.com', - ); - await upsertSessionRoom( - dbAdapter, - '@alice:localhost', - '!room-alice:localhost', - ); - - // bob has no session room - await insertUser( - dbAdapter, - '@bob:localhost', - 'cus_bob', - 'bob@example.com', - ); - await insertPermissions(dbAdapter, realmURL, { '*': ['read'], }); @@ -285,10 +223,8 @@ module(basename(import.meta.filename), function () { assert.deepEqual( result, - { - '@alice:localhost': '!room-alice:localhost', - }, - 'excludes users without session rooms even for world-readable realm', + {}, + "a world-readable realm's events go to its own room, not to session rooms", ); }); @@ -340,7 +276,7 @@ module(basename(import.meta.filename), function () { ); }); - test('combines explicit permissions and world-readable access', async function (assert) { + test('returns only users with their own grant on a world-readable realm', async function (assert) { await insertUser( dbAdapter, '@alice:localhost', @@ -365,7 +301,6 @@ module(basename(import.meta.filename), function () { '!room-bob:localhost', ); - // alice has explicit write permission AND the realm is world-readable await insertPermissions(dbAdapter, realmURL, { '*': ['read'], '@alice:localhost': ['read', 'write'], @@ -373,17 +308,110 @@ module(basename(import.meta.filename), function () { let result = await fetchRealmSessionRooms(dbAdapter, realmURL.href); - // Both users should be returned: alice via explicit + world-readable, bob via world-readable - assert.strictEqual(Object.keys(result).length, 2, 'returns 2 users'); + assert.deepEqual( + result, + { '@alice:localhost': '!room-alice:localhost' }, + 'alice is included through her own grant, and bob is not', + ); + }); + }); + + module('realm event rooms', function (hooks) { + let dbAdapter: PgAdapter; + const realmURL = new URL('http://127.0.0.1:4444/test/'); + + setupDB(hooks, { + beforeEach: async (_dbAdapter) => { + dbAdapter = _dbAdapter; + }, + }); + + test('a realm that is not world-readable has no event room', async function (assert) { + await insertPermissions(dbAdapter, realmURL, { + '@alice:localhost': ['read', 'write'], + }); + + assert.deepEqual(await fetchRealmEventRoom(dbAdapter, realmURL.href), { + worldReadable: false, + roomId: null, + }); + }); + + test('a world-readable realm has no event room until one is stored', async function (assert) { + await insertPermissions(dbAdapter, realmURL, { '*': ['read'] }); + + assert.deepEqual( + await fetchRealmEventRoom(dbAdapter, realmURL.href), + { worldReadable: true, roomId: null }, + 'no room before one is stored', + ); + + let stored = await insertRealmEventRoom( + dbAdapter, + realmURL.href, + '!events:localhost', + ); + assert.strictEqual(stored, '!events:localhost', 'the room is stored'); + assert.deepEqual( + await fetchRealmEventRoom(dbAdapter, realmURL.href), + { worldReadable: true, roomId: '!events:localhost' }, + 'the stored room is reported', + ); + }); + + test('storing a second room for a realm keeps the first', async function (assert) { + await insertRealmEventRoom(dbAdapter, realmURL.href, '!first:localhost'); + let stored = await insertRealmEventRoom( + dbAdapter, + realmURL.href, + '!second:localhost', + ); + assert.strictEqual( - result['@alice:localhost'], - '!room-alice:localhost', - 'alice is included', + stored, + '!first:localhost', + 'the room stored first is the one returned', + ); + }); + + test('rooms stored for a realm at once all come back as the one kept', async function (assert) { + let stored = await Promise.all( + ['!a:localhost', '!b:localhost', '!c:localhost', '!d:localhost'].map( + (roomId) => insertRealmEventRoom(dbAdapter, realmURL.href, roomId), + ), + ); + let { roomId } = await fetchRealmEventRoom(dbAdapter, realmURL.href); + assert.ok(roomId, 'a room is stored'); + assert.deepEqual( + stored, + stored.map(() => roomId), + 'every caller gets the stored room back', + ); + }); + + test('forgets the stored room only when it is the one named', async function (assert) { + await insertPermissions(dbAdapter, realmURL, { '*': ['read'] }); + await insertRealmEventRoom(dbAdapter, realmURL.href, '!events:localhost'); + + assert.false( + await clearRealmEventRoom(dbAdapter, realmURL.href, '!other:localhost'), + 'another room is not forgotten', ); assert.strictEqual( - result['@bob:localhost'], - '!room-bob:localhost', - 'bob is included via world-readable access', + (await fetchRealmEventRoom(dbAdapter, realmURL.href)).roomId, + '!events:localhost', + ); + assert.true( + await clearRealmEventRoom( + dbAdapter, + realmURL.href, + '!events:localhost', + ), + 'the stored room is forgotten', + ); + assert.strictEqual( + (await fetchRealmEventRoom(dbAdapter, realmURL.href)).roomId, + null, ); }); }); diff --git a/packages/runtime-common/db-queries/session-room-queries.ts b/packages/runtime-common/db-queries/session-room-queries.ts index abaef15d054..5127486188a 100644 --- a/packages/runtime-common/db-queries/session-room-queries.ts +++ b/packages/runtime-common/db-queries/session-room-queries.ts @@ -67,15 +67,12 @@ export async function clearSessionRoom( } /** - * Returns a mapping of matrix user id to session room id for all known sessions - * that should be notified about changes to the given realm. + * Returns a mapping of matrix user id to session room id for the users with + * their own read or write permission on the given realm. * - * Includes: - * - Users with explicit read/write permissions for this realm. - * - All registered users when the realm is world-readable (username = '*' - * with read = true). This is intentional: world-readable realms are visible - * to everyone, so all users need incremental-index notifications to keep - * their UI up to date. + * A world-readable realm's events don't go to session rooms; they go to the + * realm's event room (see `fetchRealmEventRoom`), so a `'*'` grant adds no one + * here. */ export async function fetchRealmSessionRooms( dbAdapter: DBAdapter, @@ -85,21 +82,12 @@ export async function fetchRealmSessionRooms( 'SELECT u.matrix_user_id, u.session_room_id', 'FROM users u', 'WHERE u.session_room_id IS NOT NULL', - 'AND (', - ' EXISTS (', - ' SELECT 1 FROM realm_user_permissions', - ' WHERE realm_url =', + 'AND EXISTS (', + ' SELECT 1 FROM realm_user_permissions', + ' WHERE realm_url =', param(realmURL), - ' AND username = u.matrix_user_id', - ' AND (read = true OR write = true)', - ' )', - ' OR EXISTS (', - ' SELECT 1 FROM realm_user_permissions', - ' WHERE realm_url =', - param(realmURL), - " AND username = '*'", - ' AND read = true', - ' )', + ' AND username = u.matrix_user_id', + ' AND (read = true OR write = true)', ')', ]); @@ -111,3 +99,74 @@ export async function fetchRealmSessionRooms( } return result; } + +/** + * Reports whether the given realm is world-readable, and the room its realm + * events go to when it has one. A world-readable realm's room is created the + * first time a host asks for it, so until then `roomId` is null. + */ +export async function fetchRealmEventRoom( + dbAdapter: DBAdapter, + realmURL: string, +): Promise<{ worldReadable: boolean; roomId: string | null }> { + let [row] = await query(dbAdapter, [ + 'SELECT EXISTS (', + ' SELECT 1 FROM realm_user_permissions', + ' WHERE realm_url =', + param(realmURL), + " AND username = '*'", + ' AND read = true', + ') AS world_readable,', + '(SELECT room_id FROM realm_event_rooms WHERE realm_url =', + param(realmURL), + ') AS room_id', + ]); + return { + worldReadable: Boolean(row?.world_readable), + roomId: (row?.room_id as string | null | undefined) ?? null, + }; +} + +/** + * Stores the given room as the realm's event room, unless the realm already + * has one, and returns the room that is stored. When two processes create a + * room for the same realm at once, the first stored wins and the other's room + * goes unused. A conflicting insert waits for the row it conflicts with and + * reads that row's room, so every caller gets the winner back. + */ +export async function insertRealmEventRoom( + dbAdapter: DBAdapter, + realmURL: string, + roomId: string, +): Promise { + let rows = await query(dbAdapter, [ + 'INSERT INTO realm_event_rooms (realm_url, room_id) VALUES (', + param(realmURL), + ',', + param(roomId), + ') ON CONFLICT (realm_url) DO UPDATE', + 'SET room_id = realm_event_rooms.room_id', + 'RETURNING room_id', + ]); + return rows[0].room_id as string; +} + +/** + * Forgets the realm's event room when it is the given room, so the next host + * that asks for the realm's room gets a new one. For a stored room the realm + * server can no longer send in, such as after its homeserver was reset. + */ +export async function clearRealmEventRoom( + dbAdapter: DBAdapter, + realmURL: string, + roomId: string, +): Promise { + let rows = await query(dbAdapter, [ + 'DELETE FROM realm_event_rooms WHERE realm_url =', + param(realmURL), + 'AND room_id =', + param(roomId), + 'RETURNING realm_url', + ]); + return rows.length > 0; +} diff --git a/packages/runtime-common/matrix-client.ts b/packages/runtime-common/matrix-client.ts index 941131f75d6..d21622aafe3 100644 --- a/packages/runtime-common/matrix-client.ts +++ b/packages/runtime-common/matrix-client.ts @@ -1,6 +1,7 @@ import { Sha256 } from '@aws-crypto/sha256-js'; import { uint8ArrayToHex } from './index.ts'; import { REALM_ROOM_RETENTION_POLICY_MAX_LIFETIME } from './realm.ts'; +import { APP_BOXEL_REALM_EVENT_ROOM_TYPE } from './matrix-constants.ts'; import type { MatrixEvent } from '@cardstack/base/matrix-event'; type JoinedRoomsResponse = { joined_rooms: string[] }; @@ -313,6 +314,56 @@ export class MatrixClient { return json.room_id; } + // The room a world-readable realm sends its realm events into. Any user on + // the homeserver may join it. A member can read the room's history, so a + // host that is still joining when an event is sent still receives it; the + // host ignores events from before it started, and the retention policy + // bounds how far back the history goes. The events are about a realm anyone + // can read. Only this client's user, the one the realm sends its realm + // events as, may send in it or change its state. A realm server's realms + // share that user, so a member can't forge a realm event either way. + async createRealmEventRoom(realmURL: string): Promise { + let response = await this.request('_matrix/client/v3/createRoom', 'POST', { + body: JSON.stringify({ + preset: 'public_chat', + visibility: 'private', + name: `Realm events: ${realmURL}`, + creation_content: { type: APP_BOXEL_REALM_EVENT_ROOM_TYPE }, + initial_state: [ + { + type: 'm.room.history_visibility', + state_key: '', + content: { history_visibility: 'shared' }, + }, + { + type: 'm.room.retention', + state_key: '', + content: { max_lifetime: REALM_ROOM_RETENTION_POLICY_MAX_LIFETIME }, + }, + ], + power_level_content_override: { + users_default: 0, + events_default: 100, + state_default: 100, + invite: 100, + kick: 100, + ban: 100, + redact: 100, + notifications: { room: 100 }, + }, + }), + }); + let json = (await response.json()) as { room_id: string }; + if (!response.ok) { + throw new Error( + `Unable to create realm event room for ${realmURL}: status ${ + response.status + } - ${JSON.stringify(json)}`, + ); + } + return json.room_id; + } + async setRoomRetentionPolicy(roomId: string, maxLifetimeMs: number) { try { let roomState = await this.request( diff --git a/packages/runtime-common/matrix-constants.ts b/packages/runtime-common/matrix-constants.ts index 5b3aabf3528..b19d63b39d2 100644 --- a/packages/runtime-common/matrix-constants.ts +++ b/packages/runtime-common/matrix-constants.ts @@ -161,6 +161,10 @@ export const APP_BOXEL_WORKSPACE_FAVORITES_EVENT_TYPE = export const APP_BOXEL_CATALOG_REALMS_EVENT_TYPE = 'app.boxel.catalog-realms'; export const APP_BOXEL_SYSTEM_CARD_EVENT_TYPE = 'app.boxel.system-card'; export const APP_BOXEL_REALM_EVENT_TYPE = 'app.boxel.realm-event'; +// The room type of the room a world-readable realm sends its realm events +// into. One room per realm is shared by every host that has joined it, in +// place of a copy of each event in every user's session room. +export const APP_BOXEL_REALM_EVENT_ROOM_TYPE = 'app.boxel.realm-events'; export const APP_BOXEL_ACTIVE_LLM = 'app.boxel.active-llm'; export const BOT_TRIGGER_EVENT_TYPE = 'app.boxel.bot-trigger'; export const APP_BOXEL_REASONING_CONTENT_KEY = 'app.boxel.reasoning'; @@ -269,6 +273,7 @@ export const DEFAULT_FALLBACK_MODEL_ID = 'openai/gpt-6-luna'; export const SLIDING_SYNC_AI_ROOM_LIST_NAME = 'ai-room'; export const SLIDING_SYNC_AUTH_ROOM_LIST_NAME = 'auth-room'; +export const SLIDING_SYNC_REALM_EVENT_ROOM_LIST_NAME = 'realm-event-room'; export const SLIDING_SYNC_LIST_RANGE_END = 9; // Sliding-sync per-room timeline_limit is sized differently per list, and the // AI-room list also uses a two-stage value: small during the very first sync @@ -299,7 +304,9 @@ export const SLIDING_SYNC_LIST_RANGE_END = 9; // initiation → update → incremental, plus an extra initiation per file in // multi-file batches). The `update` event sits in the middle of that burst and // is what DirectoryResource subscribes to for live file-tree refresh; without -// the higher limit it gets silently dropped from the live Room.timeline. +// the higher limit it gets silently dropped from the live Room.timeline. The +// rooms of world-readable realms (realm-event-room list) receive the same +// bursts, so they use the same limit. export const INITIAL_SLIDING_SYNC_LIST_TIMELINE_LIMIT = 1; // The ai-bot's room list window. Sliding sync reports changes only for rooms // inside the requested range, and matrix-js-sdk does not backfill a `limited` @@ -312,4 +319,11 @@ export const INITIAL_SLIDING_SYNC_LIST_TIMELINE_LIMIT = 1; export const SLIDING_SYNC_AI_BOT_ROOM_LIST_RANGE: [number, number] = [0, 99]; export const SLIDING_SYNC_AI_ROOM_TIMELINE_LIMIT = 20; export const SLIDING_SYNC_AUTH_ROOM_TIMELINE_LIMIT = 20; +// Sliding sync orders a list by recent activity and reports changes only for +// rooms inside its range. A host stays in every realm event room it has +// joined, so it can hold more rooms than this window, but a room moves back +// into the window as soon as it gets an event. +export const SLIDING_SYNC_REALM_EVENT_ROOM_LIST_RANGE: [number, number] = [ + 0, 99, +]; export const SLIDING_SYNC_TIMEOUT = 30000; diff --git a/packages/runtime-common/realm.ts b/packages/runtime-common/realm.ts index 9d95ca35619..3b17dae1d52 100644 --- a/packages/runtime-common/realm.ts +++ b/packages/runtime-common/realm.ts @@ -577,7 +577,9 @@ import { type PopulateCoordinator, } from './definition-lookup.ts'; import { + fetchRealmEventRoom, fetchSessionRoom, + insertRealmEventRoom, upsertSessionRoom, } from './db-queries/session-room-queries.ts'; import { REALMS_LIST_UPDATED_EVENT_TYPE } from './matrix-constants.ts'; @@ -2888,6 +2890,9 @@ export class Realm { // `/_federated-index-counts` callers share one aggregate instead of each // running the full per-url scan. Nulled once it settles (and on index swap). #indexCountsPromise: Promise | null = null; + // The in-flight creation of this realm's event room, so concurrent callers + // on this process create one room between them. + #realmEventRoomCreation: Promise | null = null; // Cached host routing map, derived from the indexed RealmConfig card. // `getHostRoutingMap()` is called on every host-mode index request // (serve-index), so re-querying the index each time is wasteful — the map @@ -3494,6 +3499,39 @@ export class Realm { return sessionRoom; } + // The room this realm sends its realm events into while it is + // world-readable, created the first time it is asked for. Undefined for a + // realm that isn't world-readable, whose events go to the session rooms of + // the users it grants. + async ensureRealmEventRoom(): Promise { + let { worldReadable, roomId } = await fetchRealmEventRoom( + this.#dbAdapter, + this.url, + ); + if (!worldReadable) { + return undefined; + } + if (roomId) { + return roomId; + } + if (!this.#realmEventRoomCreation) { + this.#realmEventRoomCreation = (async () => { + // A caller that read the table before an earlier creation stored its + // room arrives here after that creation settled. + let stored = await fetchRealmEventRoom(this.#dbAdapter, this.url); + if (stored.roomId) { + return stored.roomId; + } + await this.#matrixClient.login(); + let created = await this.#matrixClient.createRealmEventRoom(this.url); + return await insertRealmEventRoom(this.#dbAdapter, this.url, created); + })().finally(() => { + this.#realmEventRoomCreation = null; + }); + } + return await this.#realmEventRoomCreation; + } + private async readinessCheck( request: Request, requestContext: RequestContext, @@ -16373,7 +16411,19 @@ export class Realm { // session re-derives it from `_realm-auth` (which reads the permissions // written just above) without a reload. Nothing else notifies a grantee: // the index_updated broadcast above is server-to-server only. - await this.notifyRealmsListUpdated(Object.keys(patch)); + // + // A change to the `'*'` grant moves where the realm's events go: a + // world-readable realm sends them into a room of its own, any other realm + // into its grantees' session rooms. So every user the realm grants by + // name hears of it too, and their hosts ask again which room to listen in. + let affected = new Set(Object.keys(patch)); + if (affected.has('*')) { + for (let user of Object.keys(currentPermissions)) { + affected.add(user); + } + } + affected.delete('*'); + await this.notifyRealmsListUpdated([...affected]); return await this.getRealmPermissions(request, requestContext); }