diff --git a/packages/host/app/services/matrix-service.ts b/packages/host/app/services/matrix-service.ts index ba5dba60c42..ad73a08457a 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 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, diff --git a/packages/host/app/services/message-service.ts b/packages/host/app/services/message-service.ts index 110b955168e..5aa3aafeb97 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 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, diff --git a/packages/host/app/services/realm-server.ts b/packages/host/app/services/realm-server.ts index fd8387048fc..d811bf6f5dc 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,11 @@ 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(); private eventSubscribers: Map = 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(); + 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 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 }; + // 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; }) { diff --git a/packages/host/config/schema/1792465777331_schema.sql b/packages/host/config/schema/1792702982968_schema.sql similarity index 100% rename from packages/host/config/schema/1792465777331_schema.sql rename to packages/host/config/schema/1792702982968_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..160eecc1aff 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 private 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..92302d2fa60 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,84 @@ module('Integration | message-service | realm subscriptions', function (hooks) { } }); }); + +// A private 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 private realm joins the room its events go to', async function (assert) { + let privateRealmURL = 'http://localhost:4205/private/'; + // The realm's own user creates the room and invites the user; the host + // is not in it yet. + let roomId = mockMatrixUtils.createAndJoinRoom({ + sender: '@private_realm:localhost', + name: `Realm events: ${privateRealmURL}`, + id: '!private-realm-events:localhost', + roomType: APP_BOXEL_REALM_EVENT_ROOM_TYPE, + }); + setRealmEventRoom(privateRealmURL, 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(privateRealmURL, (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, + '@private_realm:localhost', + { + eventName: 'index', + indexType: 'incremental', + invalidations: [`${privateRealmURL}card`], + realmURL: privateRealmURL, + }, + { 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/1792702982968_create-realm-event-rooms.js b/packages/postgres/migrations/1792702982968_create-realm-event-rooms.js new file mode 100644 index 00000000000..b00522cdfa3 --- /dev/null +++ b/packages/postgres/migrations/1792702982968_create-realm-event-rooms.js @@ -0,0 +1,41 @@ +// The Matrix room a private realm sends its realm events into. Its members are +// the users the realm's grants name, so each event is stored once rather than +// once per member's session room. A realm has at most one; its row is written +// once the room's first invites have gone out. +// +// A room that stops being the realm's (a grant was revoked, the realm was +// deleted, or another process stored its room first) is retired: it moves to +// retired_realm_event_rooms until every member has been removed and the realm +// server has left it, so a removal that fails is tried again rather than +// forgotten. + +exports.up = (pgm) => { + pgm.createTable('realm_event_rooms', { + realm_url: { + type: 'varchar', + notNull: true, + primaryKey: true, + }, + room_id: { + type: 'varchar', + notNull: true, + }, + }); + pgm.createTable('retired_realm_event_rooms', { + room_id: { + type: 'varchar', + notNull: true, + primaryKey: true, + }, + realm_url: { + type: 'varchar', + notNull: true, + }, + }); + pgm.createIndex('retired_realm_event_rooms', 'realm_url'); +}; + +exports.down = (pgm) => { + pgm.dropTable('retired_realm_event_rooms', { ifExists: true }); + 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..51d2fb5538d 100755 --- a/packages/postgres/scripts/schema-dump.sh +++ b/packages/postgres/scripts/schema-dump.sh @@ -41,6 +41,8 @@ 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=retired_realm_event_rooms \ --exclude-table-and-children=host_shell_generation \ --no-tablespaces \ --no-table-access-method \ diff --git a/packages/realm-server/handlers/handle-delete-realm.ts b/packages/realm-server/handlers/handle-delete-realm.ts index e8850fe93dc..6efe5f4d2c7 100644 --- a/packages/realm-server/handlers/handle-delete-realm.ts +++ b/packages/realm-server/handlers/handle-delete-realm.ts @@ -35,6 +35,8 @@ import { deleteRegistryRowByUrl, } from '../lib/realm-registry-writes.ts'; import { normalizeRealmURL } from '../utils/realm-url.ts'; +import { realmEventDispatcher } from '../realm-event-dispatcher.ts'; +import { retireRealmEventRoom } from '../realm-event-rooms.ts'; interface DeleteRealmJSON { data: { @@ -45,6 +47,7 @@ interface DeleteRealmJSON { export default function handleDeleteRealm({ dbAdapter, + matrixClient, realmsRootPath, sendEvent, }: CreateRoutesArgs): (ctxt: Koa.Context, next: Koa.Next) => Promise { @@ -251,6 +254,26 @@ export default function handleDeleteRealm({ Sentry.captureException(error); } + // The realm's event room has no grants left to follow, so it is + // retired: its members are removed and the realm server leaves it, in + // the background, and a removal that fails is tried again when a + // realm server next starts. Through the realm server's own matrix + // client, the one every realm it serves sends its events with, and so + // the room's creator. The realm has no grantees left to tell. + try { + await retireRealmEventRoom( + { + realmURL, + matrixClient, + dbAdapter, + dispatcher: realmEventDispatcher(), + }, + { notifyGrantees: false }, + ); + } catch (error) { + Sentry.captureException(error); + } + // CS-11156. Broadcast a bulk cache-invalidation for the source realm // and each removed published realm so any peer replicas that still // have these realms mounted drop their #sourceCache / #transpiledModuleCache 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..6cd147b77e1 --- /dev/null +++ b/packages/realm-server/handlers/handle-realm-event-rooms.ts @@ -0,0 +1,133 @@ +import type Koa from 'koa'; +import { + fetchRealmEventRoom, + fetchRealmNamedGrantees, + logger, + SupportedMimeType, +} from '@cardstack/runtime-common'; +import type { CreateRoutesArgs } from '../routes.ts'; +import type { RealmServerTokenClaim } from '../utils/jwt.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, create its room and invite +// its grantees, 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 the private realms it names send +// their realm events into, so it can join them. Only a realm whose grants +// name the signed-in user is answered for, and that user is invited to the +// room before the answer goes out, so the host's join succeeds. A realm's +// room is created the first time a host asks for it. A world-readable realm, +// a realm no grant names the user on, and a realm this server doesn't serve +// are left out: their 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 without a room. +// +// 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 { user } = ctxt.state.token as RealmServerTokenClaim; + 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 { + // Checked before anything is mounted, so a user the realm doesn't + // name can't make this server mount it. + let { isPrivate } = await fetchRealmEventRoom(dbAdapter, realmURL); + if (!isPrivate) { + return; + } + let grantees = await fetchRealmNamedGrantees(dbAdapter, realmURL); + if (!grantees.includes(user)) { + return; + } + // The realm's room is created and its members invited through the + // matrix client the realm sends its realm events with. + let mounted = await reconciler.lookupOrMount(realmURL); + let roomId = await mounted?.ensureRealmEventRoom(user); + 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/handlers/handle-upsert-realm-user-permission.ts b/packages/realm-server/handlers/handle-upsert-realm-user-permission.ts index c5d37178e22..454f4b86135 100644 --- a/packages/realm-server/handlers/handle-upsert-realm-user-permission.ts +++ b/packages/realm-server/handlers/handle-upsert-realm-user-permission.ts @@ -14,6 +14,8 @@ import { } from '../middleware/index.ts'; import type { CreateRoutesArgs } from '../routes.ts'; import { normalizeRealmURL } from '../utils/realm-url.ts'; +import { realmEventDispatcher } from '../realm-event-dispatcher.ts'; +import { syncRealmEventRoom } from '../realm-event-rooms.ts'; import { appendRealmServerToUserAccountData, appendRealmToUserAccountData, @@ -137,6 +139,17 @@ export default function handleUpsertRealmUserPermission({ [user]: actions, }); + // The realm's event room holds the users its grants name, so the grant + // invites this user to it. Through the realm server's own matrix client, + // the one every realm it serves sends its events with. Not awaited: it is + // paced behind the realm's events and never rejects. + void syncRealmEventRoom({ + realmURL: normalizedRealmHref, + matrixClient, + dbAdapter, + dispatcher: realmEventDispatcher(), + }); + // Push a live "your realms changed" signal so a running session re-derives // its list from `_realm-auth` without a reload. This also covers the case // where the account-data append below is a no-op (server already present), diff --git a/packages/realm-server/main.ts b/packages/realm-server/main.ts index dd4627c5693..dd166d9a36e 100644 --- a/packages/realm-server/main.ts +++ b/packages/realm-server/main.ts @@ -54,6 +54,7 @@ import { JobsFinishedListener } from './lib/jobs-finished-listener.ts'; import { JobScopedSearchCache } from './job-scoped-search-cache.ts'; import { startHealthSampler } from './health-sampler.ts'; import { realmEventDispatcher } from './realm-event-dispatcher.ts'; +import { emptyAllRetiredRealmEventRooms } from './realm-event-rooms.ts'; import { healthCheckUserAgents } from './middleware/index.ts'; import { isDeployedServerURL, @@ -1080,6 +1081,16 @@ const reportHostShellToManager = async (dbAdapter: PgAdapter) => { // wait for first-request mount via reconciler.lookupOrMount(). await server.start(); + // A realm event room retired while a process was stopping, or whose + // removals failed, is emptied now; a deleted realm never mounts again to + // retry it. Not awaited: it is paced with the realms' events and never + // rejects. + void emptyAllRetiredRealmEventRooms({ + matrixClient, + dbAdapter, + dispatcher: realmEventDispatcher(), + }); + // Now that the HTTP listener is accepting traffic and serving the new host // shell, tell the prerender manager which shell we're serving so the fleet // recycles after a host redeploy. Reporting earlier (before the listener is diff --git a/packages/realm-server/node-realm.ts b/packages/realm-server/node-realm.ts index 66eb2affe1b..a789d26860b 100644 --- a/packages/realm-server/node-realm.ts +++ b/packages/realm-server/node-realm.ts @@ -13,7 +13,9 @@ import { unixTime, type ResponseWithNodeStream, type TokenClaims, + clearRealmEventRoom, clearSessionRoom, + fetchRealmEventRoom, fetchRealmSessionRooms, isRealmServerNotInRoomError, } from '@cardstack/runtime-common'; @@ -62,6 +64,11 @@ import { type RealmEventDispatcher, type RealmEventSend, } from './realm-event-dispatcher.ts'; +import { + ensureRealmEventRoom, + retireRealmEventRoom, + syncRealmEventRoom, +} from './realm-event-rooms.ts'; const realmEventsLog = logger('realm:events'); @@ -438,9 +445,10 @@ export class NodeAdapter implements RealmAdapter { }; } - // Resolves once the event has reached every session room subscribed to the - // realm, or has been merged or dropped by the process's realm event line - // (see realm-event-dispatcher.ts). Never rejects: a failed send is logged. + // Resolves once the event has reached every room it goes to (see + // `prepareRoomSends`), or has been merged or dropped by the process's realm + // event line (see realm-event-dispatcher.ts). Never rejects: a failed send + // is logged. async broadcastRealmEvent( event: RealmEventContent, realmUrl: string, @@ -468,6 +476,49 @@ export class NodeAdapter implements RealmAdapter { }); } + async ensureRealmEventRoom( + realmUrl: string, + matrixClient: MatrixClient, + dbAdapter: DBAdapter, + member: string, + ): Promise { + return await ensureRealmEventRoom( + { + realmURL: realmUrl, + matrixClient, + dbAdapter, + dispatcher: this.dispatcher, + }, + member, + ); + } + + async syncRealmEventRoom( + realmUrl: string, + matrixClient: MatrixClient, + dbAdapter: DBAdapter, + ): Promise { + await syncRealmEventRoom({ + realmURL: realmUrl, + matrixClient, + dbAdapter, + dispatcher: this.dispatcher, + }); + } + + async retireRealmEventRoom( + realmUrl: string, + matrixClient: MatrixClient, + dbAdapter: DBAdapter, + ): Promise { + await retireRealmEventRoom({ + realmURL: realmUrl, + matrixClient, + dbAdapter, + dispatcher: this.dispatcher, + }); + } + private async prepareRoomSends( event: RealmEventContent, realmUrl: string, @@ -499,6 +550,32 @@ export class NodeAdapter implements RealmAdapter { return []; } + // A private realm's event goes once into the realm's own room, whose + // members are the users its grants name (see realm-event-rooms.ts). An + // event meant only for some of them, a world-readable realm's event, and + // a private realm's event before any host has asked for its room go to + // session rooms instead. + if (!delivery?.recipients) { + let realmEventRoom; + try { + realmEventRoom = await fetchRealmEventRoom(dbAdapter, realmUrl); + } catch (e) { + realmEventsLog.error('Error looking up the realm event room', e); + return []; + } + if (realmEventRoom.isPrivate && realmEventRoom.roomId) { + return [ + this.realmEventRoomSend( + event, + realmUrl, + realmEventRoom.roomId, + matrixClient, + dbAdapter, + ), + ]; + } + } + let dmRooms = await this.waitForSessionRooms( dbAdapter, realmUrl, @@ -562,6 +639,58 @@ export class NodeAdapter implements RealmAdapter { })); } + private realmEventRoomSend( + event: RealmEventContent, + realmUrl: string, + roomId: string, + matrixClient: MatrixClient, + dbAdapter: DBAdapter, + ): RealmEventSend { + 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 realm's events go to session + // rooms until a host asks for the realm's room and 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) + }`, + ); + }, + }; + } + private async waitForSessionRooms( dbAdapter: DBAdapter, realmUrl: string, diff --git a/packages/realm-server/realm-event-dispatcher.ts b/packages/realm-server/realm-event-dispatcher.ts index 1d262533949..271402f6c71 100644 --- a/packages/realm-server/realm-event-dispatcher.ts +++ b/packages/realm-server/realm-event-dispatcher.ts @@ -48,6 +48,11 @@ const log = logger('realm:events'); // is the one that loses events. Events whose loss a subscriber recovers // from go first (see `isCompletion`). // +// - A change to who receives a realm's events (see `schedule`) takes its turn +// on the realm's line like an event: an invite queued before an event +// completes before the event is sent. It never merges, and it is never the +// entry the bound drops. +// // The queue is in memory. A process stopping on purpose waits a bounded time // for it to empty (see `drain`); a process that exits any other way, or whose // queue outlasts that wait, does not deliver what is still queued. @@ -73,6 +78,18 @@ export interface RealmEventBroadcast { newTxnId(): string; } +// Work other than an event that goes out on a realm's line, such as the +// invites and kicks that keep a private realm's event room in step with its +// grants. +export interface RealmLineJob { + realmURL: string; + // Resolves the sends the job makes. Runs when the job reaches the head of + // its realm's line, and behaves like `RealmEventBroadcast.prepare` + // otherwise. + prepare(): Promise; + newTxnId(): string; +} + export interface RealmEventDispatcherOptions { maxInFlight?: number; sendsPerSecond?: number; @@ -102,10 +119,11 @@ export interface RealmEventDispatcherStats { interface Entry { realmURL: string; - event: RealmEventContent; + // Undefined for a job queued through `schedule`. + event: RealmEventContent | undefined; bytes: number; enqueuedAt: number; - prepare: RealmEventBroadcast['prepare']; + prepare: (event: RealmEventContent | undefined) => Promise; newTxnId: RealmEventBroadcast['newTxnId']; waiters: (() => void)[]; prepareAttempts: number; @@ -209,19 +227,17 @@ export class RealmEventDispatcher { // best-effort, and a failed send is reported through its `onFailure`. broadcast(broadcast: RealmEventBroadcast): Promise { return new Promise((resolve) => { - let line = this.#lines.get(broadcast.realmURL); - if (!line) { - line = { pending: [], active: undefined }; - this.#lines.set(broadcast.realmURL, line); - } + let line = this.#lineFor(broadcast.realmURL); let tail = line.pending[line.pending.length - 1]; - let merged = tail ? coalesce(tail.event, broadcast.event) : undefined; + let merged = tail?.event + ? coalesce(tail.event, broadcast.event) + : undefined; if (tail && merged) { let bytes = byteLength(merged); this.#queuedBytes += bytes - tail.bytes; tail.event = merged; tail.bytes = bytes; - tail.prepare = broadcast.prepare; + tail.prepare = (event) => broadcast.prepare(event!); tail.newTxnId = broadcast.newTxnId; tail.waiters.push(resolve); this.#coalesced++; @@ -231,20 +247,53 @@ export class RealmEventDispatcher { event: broadcast.event, bytes: byteLength(broadcast.event), enqueuedAt: this.#now(), - prepare: broadcast.prepare, + prepare: (event) => broadcast.prepare(event!), newTxnId: broadcast.newTxnId, waiters: [resolve], prepareAttempts: 0, }; - line.pending.push(entry); - this.#queuedEvents++; - this.#queuedBytes += entry.bytes; + this.#enqueue(line, entry); } this.#enforceBound(); this.#pump(); }); } + // Queues a job on its realm's line, behind the realm's events queued + // before it and ahead of those queued after. Resolves once its sends are + // done; like `broadcast`, it never rejects. + schedule(job: RealmLineJob): Promise { + return new Promise((resolve) => { + this.#enqueue(this.#lineFor(job.realmURL), { + realmURL: job.realmURL, + event: undefined, + bytes: JOB_BYTES, + enqueuedAt: this.#now(), + prepare: () => job.prepare(), + newTxnId: job.newTxnId, + waiters: [resolve], + prepareAttempts: 0, + }); + this.#enforceBound(); + this.#pump(); + }); + } + + #lineFor(realmURL: string): Line { + let line = this.#lines.get(realmURL); + if (!line) { + line = { pending: [], active: undefined }; + this.#lines.set(realmURL, line); + } + return line; + } + + #enqueue(line: Line, entry: Entry) { + line.pending.push(entry); + this.#queuedEvents++; + this.#queuedBytes += entry.bytes; + } + // Resolves once nothing is queued or in flight. async idle(): Promise { if (this.#isIdle()) { @@ -306,10 +355,12 @@ export class RealmEventDispatcher { #enforceBound() { while (this.#queuedBytes > this.#maxQueuedBytes) { + // Only events are dropped, so a line holding nothing but jobs is never + // the one that loses an entry. let longest: Line | undefined; for (let line of this.#lines.values()) { if ( - line.pending.length > 0 && + line.pending.some((entry) => entry.event) && (!longest || line.pending.length > longest.pending.length) ) { longest = line; @@ -319,9 +370,12 @@ export class RealmEventDispatcher { return; } let index = longest.pending.findIndex( - (entry) => !isCompletion(entry.event), + (entry) => entry.event && !isCompletion(entry.event), ); - let [dropped] = longest.pending.splice(Math.max(0, index), 1); + if (index < 0) { + index = longest.pending.findIndex((entry) => entry.event); + } + let [dropped] = longest.pending.splice(index, 1); this.#queuedEvents--; this.#queuedBytes -= dropped.bytes; this.#dropped++; @@ -446,7 +500,7 @@ export class RealmEventDispatcher { } } log.error( - `Unable to resolve the rooms for a realm event in ${entry.realmURL}; skipping it`, + `Unable to resolve the sends for a realm event or job in ${entry.realmURL}; skipping it`, e, ); return []; @@ -642,6 +696,10 @@ function union( return [...new Set([...(a ?? []), ...(b ?? [])])]; } +// What a job counts against the bound: it holds no event, but a backlog of +// them is still a backlog. +const JOB_BYTES = 256; + function byteLength(event: RealmEventContent): number { return Buffer.byteLength(JSON.stringify(event)); } diff --git a/packages/realm-server/realm-event-rooms.ts b/packages/realm-server/realm-event-rooms.ts new file mode 100644 index 00000000000..bbcbe55471c --- /dev/null +++ b/packages/realm-server/realm-event-rooms.ts @@ -0,0 +1,481 @@ +import { + clearRealmEventRoom, + clearRetiredRealmEventRoom, + createSendEvent, + fetchRealmEventRoom, + fetchRealmNamedGrantees, + fetchRealmsWithRetiredEventRooms, + fetchRetiredRealmEventRooms, + insertRealmEventRoom, + insertRetiredRealmEventRoom, + isRealmServerNotInRoomError, + logger, + retireRealmEventRoomRow, + type DBAdapter, +} from '@cardstack/runtime-common'; +import type { MatrixClient } from '@cardstack/runtime-common/matrix-client'; +import { REALMS_LIST_UPDATED_EVENT_TYPE } from '@cardstack/runtime-common/matrix-constants'; +import { + isBackPressure, + type RealmEventDispatcher, + type RealmEventSend, +} from './realm-event-dispatcher.ts'; + +const log = logger('realm:events'); + +// A private realm sends each realm event once, into a Matrix room whose +// members are exactly the users the realm's grants name (see +// `fetchRealmNamedGrantees`), rather than once into each of their session +// rooms. Users who each hold a grant naming them on the same realm may see +// each other's Matrix ids (ADR 011 §1.2), and no one else may, through room +// membership or room state (§1.1, §1.5). So the room must never show anyone a +// user they don't share a grant with. The room is created through the matrix +// client the realm sends its events with, so that client's user is the only +// one who may send, invite or kick in it (see +// `MatrixClient.createRealmEventRoom`). +// +// Membership follows the grants: +// +// - A grantee who isn't a member is invited. No one joins any other way, +// since the room is invite-only. +// - A member no grant names is never kicked out of the realm's room, because +// a kick leaves their membership in the room's state, where a member who +// joins later could read it. The room is retired instead, and the realm gets +// a new one the next time a host asks. Retiring takes effect at once for +// every process (see `retireRealmEventRoomRow`): the realm's events go to +// session rooms, which follow the grants per event, until the new room +// exists. The remaining grantees are told their realms changed, so their +// hosts ask again and join the new room. +// - A retired room is emptied: every member is removed, and then the realm +// server leaves it. Its record is kept until that has all succeeded, and the +// next sync for the realm tries again, so a removal that fails is never +// forgotten with the room. +// - A new room is stored as the realm's only once its first invites have gone +// out, so no event is sent into it before its members can see it. +// +// The Matrix work runs as jobs on the realm's line in the realm event +// dispatcher, so invites and removals are paced with the realm's events and +// back off with them. + +export interface RealmEventRoomDeps { + realmURL: string; + matrixClient: MatrixClient; + dbAdapter: DBAdapter; + dispatcher: RealmEventDispatcher; +} + +// One creation at a time per realm in this process. A room is stored only +// after its invites, so without this, a second ask arriving meanwhile would +// see no room and make another. +const creations = new Map>(); + +function oneAtATime(realmURL: string, run: () => Promise): Promise { + let previous = creations.get(realmURL) ?? Promise.resolve(); + let next = previous.then(run, run); + creations.set(realmURL, next); + // The caller handles `next` failing; this cleanup must not reject too. + let forget = () => { + if (creations.get(realmURL) === next) { + creations.delete(realmURL); + } + }; + next.then(forget, forget); + return next; +} + +// The room the realm's events go to, for a host signed in as `member`: +// created the first time it is asked for, with the realm's grantees invited. +// Every ask also invites any grantee who isn't a member, so the asking member +// is invited by the time this resolves and their host's join succeeds. +// Undefined when the realm is world-readable or no grant names `member`. +// Rejects when the room can't be provided, including when Synapse stays +// overloaded, so the caller can tell the host to ask again. +export async function ensureRealmEventRoom( + deps: RealmEventRoomDeps, + member: string, +): Promise { + let { isPrivate } = await fetchRealmEventRoom(deps.dbAdapter, deps.realmURL); + if (!isPrivate) { + return undefined; + } + let grantees = await fetchRealmNamedGrantees(deps.dbAdapter, deps.realmURL); + if (!grantees.includes(member)) { + return undefined; + } + return await oneAtATime(deps.realmURL, async () => { + let stored = await bringInStep(deps, { required: true }); + if (stored) { + return stored; + } + return await createRoom(deps); + }); +} + +// Brings the realm's room, when it has one, in step with its grants (see +// above), and retries the removals of any of its retired rooms. For after a +// grant is added, and for when a realm mounts, so a change no process saw +// (one made while the realm wasn't served, or lost with a process that +// stopped) is repaired. Resolves once the work is done; never rejects. +export async function syncRealmEventRoom( + deps: RealmEventRoomDeps, +): Promise { + try { + await bringInStep(deps, { required: false }); + await emptyRetiredRooms(deps); + } catch (e) { + log.warn( + `Unable to bring the event room of ${deps.realmURL} in step with its grants: ${message(e)}`, + ); + } +} + +// For a grant that was revoked: the realm's room stops being the realm's at +// once, in every process, and is emptied and left in the background, and the +// remaining grantees are told to ask again. Resolves once the realm's events +// no longer go to the room; the Matrix work is awaited only with +// `waitForRemovals`. +export async function retireRealmEventRoom( + deps: RealmEventRoomDeps, + opts: { notifyGrantees?: boolean; waitForRemovals?: boolean } = {}, +): Promise { + let retired = await retireRealmEventRoomRow(deps.dbAdapter, deps.realmURL); + if (!retired) { + return; + } + if (opts.notifyGrantees ?? true) { + void notifyGrantees(deps); + } + let removals = emptyRetiredRooms(deps).catch((e) => { + log.warn( + `Unable to empty the retired event rooms of ${deps.realmURL}: ${message(e)}`, + ); + }); + if (opts.waitForRemovals) { + await removals; + } +} + +// Retries the removals of every retired room, for when the realm server +// starts: a deleted realm never mounts again, so its sync can't. +export async function emptyAllRetiredRealmEventRooms({ + matrixClient, + dbAdapter, + dispatcher, +}: Omit): Promise { + try { + for (let realmURL of await fetchRealmsWithRetiredEventRooms(dbAdapter)) { + await emptyRetiredRooms({ + realmURL, + matrixClient, + dbAdapter, + dispatcher, + }); + } + } catch (e) { + log.warn(`Unable to empty the retired realm event rooms: ${message(e)}`); + } +} + +async function loginIfNeeded(matrixClient: MatrixClient) { + if (!matrixClient.isLoggedIn()) { + await matrixClient.login(); + } +} + +// One job on the realm's line: invites any grantee who isn't a member of the +// realm's room, or retires the room when it has a member no grant names. +// Resolves the room the realm's events go to afterwards, if it has one. With +// `required`, rejects when the job couldn't run (for example when Synapse +// stayed overloaded past the dispatcher's attempts) rather than answer as if +// the realm had no room. +async function bringInStep( + deps: RealmEventRoomDeps, + { required }: { required: boolean }, +): Promise { + let outcome: { roomId: string | undefined } | undefined; + let failure: unknown; + let retire = false; + await deps.dispatcher.schedule({ + realmURL: deps.realmURL, + newTxnId: () => deps.matrixClient.nextTxnId(), + prepare: async () => { + try { + let { roomId } = await fetchRealmEventRoom( + deps.dbAdapter, + deps.realmURL, + ); + if (!roomId) { + outcome = { roomId: undefined }; + return []; + } + await loginIfNeeded(deps.matrixClient); + let grantees = await fetchRealmNamedGrantees( + deps.dbAdapter, + deps.realmURL, + ); + let memberships: Map; + try { + memberships = await deps.matrixClient.getRoomMemberships(roomId); + } catch (e) { + if (isRealmServerNotInRoomError(e, roomId)) { + // The server can't send in or administer the room, as after a + // homeserver reset, so the realm needs a new one. + await forgetRoom(deps, roomId); + outcome = { roomId: undefined }; + return []; + } + throw e; + } + let wanted = wantedMembers(deps, grantees); + let unwanted = [...memberships.keys()].some( + (user) => isMember(memberships, user) && !wanted.has(user), + ); + if (unwanted) { + retire = true; + outcome = { roomId: undefined }; + return []; + } + outcome = { roomId }; + return invites(deps, roomId, wanted, memberships); + } catch (e) { + failure = e; + if (isBackPressure(e)) { + throw e; + } + return []; + } + }, + }); + if (retire) { + await retireRealmEventRoom(deps, { waitForRemovals: true }); + return undefined; + } + if (!outcome) { + // The job never ran to the end: its last attempt failed, and `failure` + // holds why. + if (required) { + throw ( + failure ?? + new Error(`Unable to look up the event room of ${deps.realmURL}`) + ); + } + return undefined; + } + return outcome.roomId; +} + +// Creates a room for the realm, invites its grantees, and only then stores +// it as the realm's. If another process stored a room first, that one is the +// realm's, and the new one is retired. +async function createRoom(deps: RealmEventRoomDeps): Promise { + let created: string | undefined; + let failure: unknown; + await deps.dispatcher.schedule({ + realmURL: deps.realmURL, + newTxnId: () => deps.matrixClient.nextTxnId(), + prepare: async () => { + try { + await loginIfNeeded(deps.matrixClient); + let grantees = await fetchRealmNamedGrantees( + deps.dbAdapter, + deps.realmURL, + ); + // An attempt prepared again after back-pressure reuses the room the + // first attempt made. + created ??= await deps.matrixClient.createRealmEventRoom(deps.realmURL); + return invites(deps, created, wantedMembers(deps, grantees), new Map()); + } catch (e) { + failure = e; + if (isBackPressure(e)) { + throw e; + } + return []; + } + }, + }); + if (!created) { + throw ( + failure ?? + new Error(`Unable to create the event room of ${deps.realmURL}`) + ); + } + let stored = await insertRealmEventRoom( + deps.dbAdapter, + deps.realmURL, + created, + ); + if (stored === created) { + return stored; + } + await insertRetiredRealmEventRoom(deps.dbAdapter, deps.realmURL, created); + await emptyRetiredRooms(deps); + return (await bringInStep(deps, { required: true })) ?? stored; +} + +// Removes every member of each of the realm's retired rooms, then leaves the +// room and forgets it. A room whose members couldn't all be removed is kept, +// with the realm server still in it, for the next sync to try again. +async function emptyRetiredRooms(deps: RealmEventRoomDeps): Promise { + let rooms = await fetchRetiredRealmEventRooms(deps.dbAdapter, deps.realmURL); + for (let roomId of rooms) { + await emptyRetiredRoom(deps, roomId); + } +} + +async function emptyRetiredRoom( + deps: RealmEventRoomDeps, + roomId: string, +): Promise { + let emptied = false; + let gone = false; + let failed = false; + await deps.dispatcher.schedule({ + realmURL: deps.realmURL, + newTxnId: () => deps.matrixClient.nextTxnId(), + prepare: async () => { + let memberships: Map; + try { + await loginIfNeeded(deps.matrixClient); + memberships = await deps.matrixClient.getRoomMemberships(roomId); + } catch (e) { + if (isRealmServerNotInRoomError(e, roomId)) { + // Nothing the server can do in a room it isn't in. + gone = true; + return []; + } + if (isBackPressure(e)) { + throw e; + } + failed = true; + log.warn( + `Unable to read the members of the retired event room ${roomId} of ${deps.realmURL}: ${message(e)}`, + ); + return []; + } + emptied = true; + return [...memberships.keys()] + .filter((user) => isMember(memberships, user)) + .map((user) => ({ + roomId, + send: async () => { + await deps.matrixClient.kickUser( + roomId, + user, + 'This room no longer carries the realm’s events', + ); + }, + onFailure: async (e: unknown) => { + failed = true; + log.warn( + `Unable to remove ${user} from the retired event room ${roomId} of ${deps.realmURL}: ${message(e)}`, + ); + }, + })); + }, + }); + if (gone) { + await clearRetiredRealmEventRoom(deps.dbAdapter, roomId); + return; + } + if (!emptied || failed) { + return; + } + let left = false; + await deps.dispatcher.schedule({ + realmURL: deps.realmURL, + newTxnId: () => deps.matrixClient.nextTxnId(), + prepare: async () => [ + { + roomId, + send: async () => { + await deps.matrixClient.leaveRoom(roomId); + left = true; + }, + onFailure: async (e) => { + log.warn( + `Unable to leave the retired event room ${roomId} of ${deps.realmURL}: ${message(e)}`, + ); + }, + }, + ], + }); + if (left) { + await clearRetiredRealmEventRoom(deps.dbAdapter, roomId); + } +} + +// Tells each of the realm's grantees that their realms changed, so their +// hosts ask again which room to listen in. Best-effort, one user at a time. +async function notifyGrantees(deps: RealmEventRoomDeps): Promise { + let sendEvent = createSendEvent({ + matrixClient: deps.matrixClient, + dbAdapter: deps.dbAdapter, + }); + let grantees: string[]; + try { + grantees = await fetchRealmNamedGrantees(deps.dbAdapter, deps.realmURL); + } catch (e) { + log.warn(`Unable to list the grantees of ${deps.realmURL}: ${message(e)}`); + return; + } + for (let user of grantees) { + try { + await sendEvent(user, REALMS_LIST_UPDATED_EVENT_TYPE); + } catch (e) { + log.warn( + `Unable to tell ${user} the event room of ${deps.realmURL} changed: ${message(e)}`, + ); + } + } +} + +function wantedMembers(deps: RealmEventRoomDeps, grantees: string[]) { + let self = deps.matrixClient.getUserId(); + return new Set(grantees.filter((user) => user !== self)); +} + +function isMember(memberships: Map, user: string) { + return ['join', 'invite'].includes(memberships.get(user) ?? ''); +} + +function invites( + deps: RealmEventRoomDeps, + roomId: string, + wanted: Set, + memberships: Map, +): RealmEventSend[] { + return [...wanted] + .filter((user) => !isMember(memberships, user)) + .map((user) => ({ + roomId, + send: async () => { + await deps.matrixClient.inviteUser(roomId, user); + }, + onFailure: async (e: unknown) => { + // The next sync for the realm tries again. + log.warn( + `Unable to invite ${user} to the event room ${roomId} of ${deps.realmURL}: ${message(e)}`, + ); + }, + })); +} + +async function forgetRoom(deps: RealmEventRoomDeps, roomId: string) { + try { + let cleared = await clearRealmEventRoom( + deps.dbAdapter, + deps.realmURL, + roomId, + ); + log.warn( + `Forgetting event room ${roomId} of ${deps.realmURL}; the realm server is no longer in the room`, + { cleared }, + ); + } catch (e) { + log.error(`Failed to forget event room ${roomId} of ${deps.realmURL}`, e); + } +} + +function message(e: unknown): string { + return e instanceof Error ? e.message : String(e); +} 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..2df72f5e572 100644 --- a/packages/realm-server/scripts/load-harness/lib/realm-events.ts +++ b/packages/realm-server/scripts/load-harness/lib/realm-events.ts @@ -1,9 +1,10 @@ // 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 -// listen on that channel rather than infer invalidation from its own writes — +// The realm server broadcasts `app.boxel.realm-event` over Matrix: a private +// realm's events into one room for the realm, whose members are the users its +// grants name, and a world-readable realm's into the DM session room of each +// user. 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. // // Long-polls `/sync` directly rather than using `matrix-js-sdk`: the harness @@ -73,6 +74,60 @@ export async function joinInvitedRooms({ }; } +// Joins the rooms the given private realms send their realm events into, the +// way the host does when it first subscribes to a realm. The realm server +// answers only for private realms whose grants name the session's user, so +// any other realm 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..28af0b401f6 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 private 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..b86fbb10290 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,53 @@ export function setupMatrixRoom( } testAuthRoomId = payload.sessionRoom; + + // A private test realm whose grants name this user 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 has one. 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 2b59025c6fe..ba766f44a82 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'; @@ -302,6 +304,149 @@ module(basename(import.meta.filename), function (hooks) { "only the named subscriber's room receives the event", ); }); + + 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 }; + } + + async function insertSessionFor(user: string, roomId: string) { + let name = user.slice(1, user.indexOf(':')); + await insertUser(dbAdapter, user, `cus_${name}`, `${name}@example.com`); + await upsertSessionRoom(dbAdapter, user, roomId); + } + + test("sends a private realm's event once, into the realm's event room", async function (assert) { + await insertSessionFor('@alice:localhost', '!alice:localhost'); + await insertSessionFor('@bob:localhost', '!bob:localhost'); + await insertPermissions(dbAdapter, realmURL, { + '@alice:localhost': ['read', 'write'], + '@bob:localhost': ['read'], + }); + 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 a private realm's event to session rooms until it has an event room", async function (assert) { + await insertSessionFor('@alice:localhost', '!alice:localhost'); + await insertSessionFor('@bob:localhost', '!bob:localhost'); + await insertPermissions(dbAdapter, realmURL, { + '@alice:localhost': ['read', 'write'], + '@bob:localhost': ['read'], + }); + + let { matrixClient, sentTo } = recordingMatrixClient(); + await new NodeAdapter('/tmp').broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + ); + + assert.deepEqual( + sentTo.sort(), + ['!alice:localhost', '!bob:localhost'], + "sent to each grantee's session room", + ); + }); + + test("sends a world-readable realm's event to session rooms even when it has an event room", async function (assert) { + await insertSessionFor('@alice:localhost', '!alice: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, + ['!alice:localhost'], + "the room holds only named users, so a world-readable realm's readers would not hear it", + ); + }); + + test('sends an event meant for named recipients to their session rooms, not to the event room', async function (assert) { + await insertSessionFor('@alice:localhost', '!alice:localhost'); + await insertSessionFor('@bob:localhost', '!bob:localhost'); + await insertPermissions(dbAdapter, realmURL, { + '@alice:localhost': ['read', 'write'], + '@bob:localhost': ['read'], + }); + await insertRealmEventRoom(dbAdapter, realmURL.href, '!events:localhost'); + + let { matrixClient, sentTo } = recordingMatrixClient(); + await new NodeAdapter('/tmp').broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + { recipients: ['@alice:localhost'] }, + ); + + assert.deepEqual(sentTo, ['!alice:localhost']); + }); + + test("forgets a private realm's event room the realm server is no longer in", async function (assert) { + await insertPermissions(dbAdapter, realmURL, { + '@alice:localhost': ['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), + { isPrivate: true, roomId: null }, + 'the next host to ask gets a new 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..c8cc094da86 100644 --- a/packages/realm-server/tests/realm-endpoints/permissions-test.ts +++ b/packages/realm-server/tests/realm-endpoints/permissions-test.ts @@ -170,6 +170,39 @@ 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 out of its event + // room and into session rooms, 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-dispatcher-test.ts b/packages/realm-server/tests/realm-event-dispatcher-test.ts index 7c455172799..9a5dfa3f3c0 100644 --- a/packages/realm-server/tests/realm-event-dispatcher-test.ts +++ b/packages/realm-server/tests/realm-event-dispatcher-test.ts @@ -594,6 +594,85 @@ module(basename(import.meta.filename), function () { ); }); + test("runs a scheduled job in its turn on the realm's line, between the events queued around it", async function (assert) { + let h = harness({ maxInFlight: 8 }); + let order: string[] = []; + void h.broadcast(REALM_A, initiation(REALM_A, `${REALM_A}card`), ['!r1']); + let job = h.dispatcher.schedule({ + realmURL: REALM_A, + newTxnId: () => 'job-txn', + prepare: async () => { + order.push('job prepared'); + return [ + { + roomId: '!r1', + send: async () => { + order.push('job sent'); + }, + onFailure: async () => {}, + }, + ]; + }, + }); + // An initiation queued after the job would merge into one queued before + // it if the job weren't between them. + let after = h.broadcast(REALM_A, initiation(REALM_A, `${REALM_A}other`), [ + '!r1', + ]); + await h.settle(); + assert.deepEqual(order, [], 'the job waits for the event ahead of it'); + assert.strictEqual(h.started.length, 1); + + h.started[0].done.resolve(); + await job; + assert.deepEqual(order, ['job prepared', 'job sent']); + await h.settle(); + assert.strictEqual( + h.started.length, + 2, + 'the event queued after the job goes out on its own, unmerged', + ); + assert.strictEqual( + (h.started[1].event as { updatedFile: string }).updatedFile, + `${REALM_A}other`, + ); + h.releaseAll(); + await after; + assert.strictEqual(h.dispatcher.takeStats().coalesced, 0); + }); + + test('past the byte bound, drops an event but never a scheduled job', async function (assert) { + let eventBytes = Buffer.byteLength( + JSON.stringify(incremental(REALM_A, 'card-0')), + ); + let h = harness({ maxInFlight: 1, maxQueuedBytes: eventBytes + 300 }); + void h.broadcast(REALM_A, incremental(REALM_A, 'sending')); + await h.settle(); + + let jobRan = false; + let job = h.dispatcher.schedule({ + realmURL: REALM_A, + newTxnId: () => 'job-txn', + prepare: async () => { + jobRan = true; + return []; + }, + }); + let dropped = false; + void h + .broadcast(REALM_A, incremental(REALM_A, 'card-0')) + .then(() => (dropped = true)); + void h.broadcast(REALM_A, incremental(REALM_A, 'card-1')); + await h.settle(); + assert.true(dropped, 'an event behind the job is dropped'); + assert.strictEqual(h.dispatcher.takeStats().dropped, 1); + + h.releaseAll(); + await job; + assert.true(jobRan, 'the job still runs'); + await h.dispatcher.idle(); + }); + module('isBackPressure', function () { test('treats overload and unreachable answers as back-pressure, and nothing else', function (assert) { for (let status of [429, 502, 503, 504]) { 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..b0863695750 --- /dev/null +++ b/packages/realm-server/tests/realm-event-rooms-test.ts @@ -0,0 +1,802 @@ +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 type { Realm } from '@cardstack/runtime-common'; +import { + fetchRealmEventRoom, + fetchRealmNamedGrantees, + fetchRetiredRealmEventRooms, + insertPermissions, + insertRealmEventRoom, +} from '@cardstack/runtime-common'; +import type { MatrixClient } from '@cardstack/runtime-common/matrix-client'; +import { + MatrixRequestError, + 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 { + createJWT, + matrixURL, + realmSecretSeed, + setupDB, + setupPermissionedRealmCached, + testRealmHref, + waitUntil, +} from './helpers/index.ts'; +import { createJWT as createRealmServerJWT } from '../utils/jwt.ts'; +import { RealmEventDispatcher } from '../realm-event-dispatcher.ts'; +import { + ensureRealmEventRoom, + retireRealmEventRoom, + syncRealmEventRoom, +} from '../realm-event-rooms.ts'; + +const SELF = '@realm_server:localhost'; + +// A matrix client that keeps room memberships in memory and records what it +// was asked to do. +function fakeMatrixClient() { + let rooms = new Map>(); + let calls: string[] = []; + let created = 0; + let matrixClient = { + login: async () => undefined, + isLoggedIn: () => true, + getUserId: () => SELF, + nextTxnId: () => 'txn', + createRealmEventRoom: async () => { + let roomId = `!created-${++created}:localhost`; + rooms.set(roomId, new Map()); + calls.push(`create ${roomId}`); + return roomId; + }, + getRoomMemberships: async (roomId: string) => { + let room = rooms.get(roomId); + if (!room) { + throw new Error( + `Unable to read the members of room ${roomId}: status 403 - {"errcode":"M_FORBIDDEN","error":"User ${SELF} not in room ${roomId}"}`, + ); + } + return new Map(room); + }, + inviteUser: async (roomId: string, userId: string) => { + rooms.get(roomId)!.set(userId, 'invite'); + calls.push(`invite ${userId}`); + }, + kickUser: async (roomId: string, userId: string) => { + rooms.get(roomId)!.set(userId, 'leave'); + calls.push(`kick ${userId}`); + }, + leaveRoom: async (roomId: string) => { + calls.push(`leave ${roomId}`); + }, + } as unknown as MatrixClient; + return { matrixClient, rooms, calls }; +} + +module(basename(import.meta.filename), function () { + module('membership', function (hooks) { + let dbAdapter: PgAdapter; + const realmURL = 'http://127.0.0.1:4444/private/'; + + setupDB(hooks, { + beforeEach: async (_dbAdapter) => { + dbAdapter = _dbAdapter; + }, + }); + + function deps(matrixClient: MatrixClient) { + return { + realmURL, + matrixClient, + dbAdapter, + dispatcher: new RealmEventDispatcher({ sendsPerSecond: 1000 }), + }; + } + + test('names every user with their own grant, and neither the public grant nor a bare username', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '*': ['read'], + mary: ['read', 'write', 'realm-owner'], + '@alice:localhost': ['read', 'write', 'realm-owner'], + '@bob:localhost': ['read'], + }); + assert.deepEqual(await fetchRealmNamedGrantees(dbAdapter, realmURL), [ + '@alice:localhost', + '@bob:localhost', + ]); + }); + + test("creates a private realm's room for a grantee, inviting every grantee but the realm server before storing it", async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + '@bob:localhost': ['read'], + [SELF]: ['read'], + }); + let { matrixClient, calls } = fakeMatrixClient(); + let storedAtInvite: (string | null)[] = []; + let realInvite = matrixClient.inviteUser.bind(matrixClient); + (matrixClient as any).inviteUser = async ( + roomId: string, + userId: string, + ) => { + storedAtInvite.push( + (await fetchRealmEventRoom(dbAdapter, realmURL)).roomId, + ); + await realInvite(roomId, userId); + }; + + let roomId = await ensureRealmEventRoom( + deps(matrixClient), + '@bob:localhost', + ); + + assert.strictEqual(roomId, '!created-1:localhost'); + assert.deepEqual( + calls, + [ + 'create !created-1:localhost', + 'invite @alice:localhost', + 'invite @bob:localhost', + ], + 'the grantees are invited before the room is answered', + ); + assert.deepEqual( + storedAtInvite, + [null, null], + 'no event can go to the room before its grantees are invited', + ); + assert.deepEqual(await fetchRealmEventRoom(dbAdapter, realmURL), { + isPrivate: true, + roomId: roomId!, + }); + + let again = await ensureRealmEventRoom( + deps(matrixClient), + '@alice:localhost', + ); + assert.strictEqual(again, roomId, 'asking again answers the same room'); + assert.strictEqual(calls.length, 3, 'and changes nothing'); + }); + + test('creates one room when two hosts ask at once', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + '@bob:localhost': ['read'], + }); + let { matrixClient, calls } = fakeMatrixClient(); + let [a, b] = await Promise.all([ + ensureRealmEventRoom(deps(matrixClient), '@alice:localhost'), + ensureRealmEventRoom(deps(matrixClient), '@bob:localhost'), + ]); + assert.strictEqual(a, b, 'both hosts get the same room'); + assert.strictEqual( + calls.filter((call) => call.startsWith('create')).length, + 1, + ); + }); + + test('answers no room to a user no grant names, and creates none', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + }); + let { matrixClient, calls } = fakeMatrixClient(); + + assert.strictEqual( + await ensureRealmEventRoom(deps(matrixClient), '@mallory:localhost'), + undefined, + ); + assert.deepEqual(calls, []); + assert.strictEqual( + (await fetchRealmEventRoom(dbAdapter, realmURL)).roomId, + null, + ); + }); + + test('answers no room for a world-readable realm, even to a grantee', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '*': ['read'], + '@alice:localhost': ['read', 'write', 'realm-owner'], + }); + let { matrixClient, calls } = fakeMatrixClient(); + + assert.strictEqual( + await ensureRealmEventRoom(deps(matrixClient), '@alice:localhost'), + undefined, + ); + assert.deepEqual(calls, []); + }); + + test('rejects, rather than answer no room, when synapse stays overloaded', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + }); + await insertRealmEventRoom(dbAdapter, realmURL, '!busy:localhost'); + let { matrixClient } = fakeMatrixClient(); + (matrixClient as any).getRoomMemberships = async () => { + throw new MatrixRequestError('overloaded', 503); + }; + + await assert.rejects( + ensureRealmEventRoom( + { + ...deps(matrixClient), + dispatcher: new RealmEventDispatcher({ + maxAttempts: 1, + initialBackoffMs: 1, + }), + }, + '@alice:localhost', + ), + /overloaded/, + ); + }); + + test('makes a new room when the stored one is a room the realm server is no longer in', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + }); + await insertRealmEventRoom(dbAdapter, realmURL, '!gone:localhost'); + let { matrixClient } = fakeMatrixClient(); + + let roomId = await ensureRealmEventRoom( + deps(matrixClient), + '@alice:localhost', + ); + + assert.strictEqual(roomId, '!created-1:localhost'); + }); + + test('uses the room another process stored first, inviting the grantees to it, and retires its own', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + }); + let { matrixClient, rooms, calls } = fakeMatrixClient(); + let realCreate = matrixClient.createRealmEventRoom.bind(matrixClient); + (matrixClient as any).createRealmEventRoom = async (url: string) => { + let created = await realCreate(url); + // Another process stores its room while this one is creating. + rooms.set('!winner:localhost', new Map()); + await insertRealmEventRoom(dbAdapter, realmURL, '!winner:localhost'); + return created; + }; + + let roomId = await ensureRealmEventRoom( + deps(matrixClient), + '@alice:localhost', + ); + + assert.strictEqual(roomId, '!winner:localhost'); + assert.deepEqual(calls, [ + 'create !created-1:localhost', + 'invite @alice:localhost', + 'kick @alice:localhost', + 'leave !created-1:localhost', + 'invite @alice:localhost', + ]); + assert.strictEqual( + rooms.get('!winner:localhost')!.get('@alice:localhost'), + 'invite', + ); + assert.deepEqual( + await fetchRetiredRealmEventRooms(dbAdapter, realmURL), + [], + 'the unused room is emptied and forgotten', + ); + }); + + test('syncing invites a new grantee without touching the members', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + }); + let { matrixClient, rooms, calls } = fakeMatrixClient(); + let roomId = (await ensureRealmEventRoom( + deps(matrixClient), + '@alice:localhost', + ))!; + rooms.get(roomId)!.set('@alice:localhost', 'join'); + calls.length = 0; + + await insertPermissions(dbAdapter, new URL(realmURL), { + '@carol:localhost': ['read'], + }); + await syncRealmEventRoom(deps(matrixClient)); + + assert.deepEqual(calls, ['invite @carol:localhost']); + calls.length = 0; + await syncRealmEventRoom(deps(matrixClient)); + assert.deepEqual(calls, [], 'a second sync changes nothing'); + }); + + test('a member no grant names retires the room, so a later grantee never sees them', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + '@bob:localhost': ['read'], + }); + let { matrixClient, rooms, calls } = fakeMatrixClient(); + let oldRoom = (await ensureRealmEventRoom( + deps(matrixClient), + '@alice:localhost', + ))!; + rooms.get(oldRoom)!.set('@alice:localhost', 'join'); + calls.length = 0; + + // Bob's grant is revoked where nothing saw it, and Carol is granted. + await insertPermissions(dbAdapter, new URL(realmURL), { + '@bob:localhost': [], + '@carol:localhost': ['read'], + }); + await syncRealmEventRoom(deps(matrixClient)); + + assert.strictEqual( + (await fetchRealmEventRoom(dbAdapter, realmURL)).roomId, + null, + 'the old room no longer carries the realm events', + ); + assert.deepEqual(calls, [ + 'kick @alice:localhost', + 'kick @bob:localhost', + `leave ${oldRoom}`, + ]); + assert.deepEqual( + await fetchRetiredRealmEventRooms(dbAdapter, realmURL), + [], + ); + + let newRoom = (await ensureRealmEventRoom( + deps(matrixClient), + '@carol:localhost', + ))!; + assert.notStrictEqual(newRoom, oldRoom); + assert.deepEqual( + [...rooms.get(newRoom)!.keys()].sort(), + ['@alice:localhost', '@carol:localhost'], + "the new room's state holds only the current grantees", + ); + }); + + test('syncing a realm without a room does nothing', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + }); + let { matrixClient, calls } = fakeMatrixClient(); + await syncRealmEventRoom(deps(matrixClient)); + assert.deepEqual(calls, []); + }); + + test('forgets a stored room the realm server is no longer in', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + }); + await insertRealmEventRoom(dbAdapter, realmURL, '!gone:localhost'); + let { matrixClient } = fakeMatrixClient(); + + await syncRealmEventRoom(deps(matrixClient)); + + assert.strictEqual( + (await fetchRealmEventRoom(dbAdapter, realmURL)).roomId, + null, + 'the next host to ask gets a new room', + ); + }); + + test("retiring a realm's room stops its events going there at once, then empties and leaves it", async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + '@bob:localhost': ['read'], + }); + let { matrixClient, calls } = fakeMatrixClient(); + let roomId = await ensureRealmEventRoom( + deps(matrixClient), + '@alice:localhost', + ); + calls.length = 0; + + await retireRealmEventRoom(deps(matrixClient), { waitForRemovals: true }); + + assert.deepEqual(calls, [ + 'kick @alice:localhost', + 'kick @bob:localhost', + `leave ${roomId}`, + ]); + assert.strictEqual( + (await fetchRealmEventRoom(dbAdapter, realmURL)).roomId, + null, + ); + assert.deepEqual( + await fetchRetiredRealmEventRooms(dbAdapter, realmURL), + [], + ); + }); + + test('keeps a retired room it could not empty, still in it, and empties it on the next sync', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + '@bob:localhost': ['read'], + }); + let { matrixClient, calls } = fakeMatrixClient(); + let roomId = (await ensureRealmEventRoom( + deps(matrixClient), + '@alice:localhost', + ))!; + calls.length = 0; + let realKick = matrixClient.kickUser.bind(matrixClient); + let failBob = true; + (matrixClient as any).kickUser = async ( + room: string, + user: string, + reason: string, + ) => { + if (user === '@bob:localhost' && failBob) { + throw new Error('kick failed'); + } + await realKick(room, user, reason); + }; + + await retireRealmEventRoom(deps(matrixClient), { waitForRemovals: true }); + + assert.false( + calls.includes(`leave ${roomId}`), + 'the realm server stays in a room it could not empty', + ); + assert.deepEqual( + await fetchRetiredRealmEventRooms(dbAdapter, realmURL), + [roomId], + 'and keeps its record', + ); + + failBob = false; + calls.length = 0; + await syncRealmEventRoom(deps(matrixClient)); + assert.deepEqual(calls, ['kick @bob:localhost', `leave ${roomId}`]); + assert.deepEqual( + await fetchRetiredRealmEventRooms(dbAdapter, realmURL), + [], + ); + }); + + test('keeps a retired room whose members could not be read', async function (assert) { + await insertPermissions(dbAdapter, new URL(realmURL), { + '@alice:localhost': ['read', 'write', 'realm-owner'], + }); + let { matrixClient, calls } = fakeMatrixClient(); + let roomId = (await ensureRealmEventRoom( + deps(matrixClient), + '@alice:localhost', + ))!; + calls.length = 0; + let realRead = matrixClient.getRoomMemberships.bind(matrixClient); + (matrixClient as any).getRoomMemberships = async () => { + throw new Error('read failed'); + }; + + await retireRealmEventRoom(deps(matrixClient), { waitForRemovals: true }); + + assert.deepEqual( + calls, + [], + 'nothing is kicked, and the room is not left', + ); + assert.deepEqual(await fetchRetiredRealmEventRooms(dbAdapter, realmURL), [ + roomId, + ]); + + (matrixClient as any).getRoomMemberships = realRead; + await syncRealmEventRoom(deps(matrixClient)); + assert.deepEqual(calls, ['kick @alice:localhost', `leave ${roomId}`]); + }); + }); + + module('realm event rooms endpoint', function (hooks) { + let dbAdapter: PgAdapter; + let request: SuperTest; + let testRealm: Realm; + // The realm's own matrix user creates its room; these are other users. + const member = '@node-test_realm:localhost'; + const newcomer = '@experiments_realm:localhost'; + const outsider = '@catalog_realm:localhost'; + + setupPermissionedRealmCached(hooks, { + fixture: 'blank', + permissions: { + mary: ['read', 'write', 'realm-owner'], + [member]: ['read', 'write'], + }, + onRealmSetup: ({ + dbAdapter: adapter, + request: req, + testRealm: realm, + }) => { + dbAdapter = adapter; + request = req; + testRealm = realm; + }, + }); + + function realmEventRooms(realms: string[], user = member) { + return request + .post('/_realm-event-rooms') + .set('Accept', 'application/json') + .set('Content-Type', 'application/json') + .set( + 'Authorization', + `Bearer ${createRealmServerJWT( + { user, sessionRoom: 'server-session-room' }, + realmSecretSeed, + )}`, + ) + .send(JSON.stringify({ realms })); + } + + function patchPermissions(permissions: Record) { + return 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 }, + }, + }); + } + + // A client for one of the test matrix users, the way a host is one. + async function loginAs(userId: string) { + let username = userId.slice(1, userId.indexOf(':')); + let response = await fetch( + new URL('_matrix/client/v3/login', matrixURL), + { + method: 'POST', + body: JSON.stringify({ + identifier: { type: 'm.id.user', user: username }, + password: await passwordFromSeed(username, 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}` }, + }); + } + + // The grantee joins the room, which only an invite allows, and then reads + // other users' memberships from its state as a member. + async function joinAsMember(roomId: string) { + let asMember = await loginAs(member); + let join = await asMember(`join/${encodeURIComponent(roomId)}`, { + method: 'POST', + body: '{}', + }); + if (!join.ok) { + throw new Error( + `${member} could not join ${roomId}: ${join.status} ${await join.text()}`, + ); + } + return asMember; + } + + async function membershipOf( + asMember: Awaited>, + roomId: string, + userId: string, + ) { + let response = await asMember( + `rooms/${encodeURIComponent(roomId)}/state/m.room.member/${encodeURIComponent(userId)}`, + ); + if (response.status === 404) { + return undefined; + } + if (!response.ok) { + throw new Error( + `Unable to read the membership of ${userId} in ${roomId}: ${response.status} ${await response.text()}`, + ); + } + return ((await response.json()) as { membership?: string }).membership; + } + + test("answers a grantee with the realm's room, the same each time, having invited them", 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 asMember = await joinAsMember(roomId); + assert.strictEqual( + await membershipOf(asMember, roomId, member), + 'join', + 'the grantee was invited by the time the answer arrived, so it could join', + ); + + let second = await realmEventRooms([testRealmHref]); + assert.strictEqual( + second.body.rooms[testRealmHref], + roomId, + 'asking again answers with the same room', + ); + }); + + test('leaves out a realm whose grants do not name the user, a world-readable realm, and an unknown realm', async function (assert) { + let response = await realmEventRooms([testRealmHref], outsider); + assert.strictEqual(response.status, 200, 'HTTP 200 status'); + assert.deepEqual(response.body, { rooms: {} }, 'no room for an outsider'); + assert.strictEqual( + (await fetchRealmEventRoom(dbAdapter, testRealmHref)).roomId, + null, + "an outsider's ask creates no room", + ); + + let publicRealmURL = new URL('http://127.0.0.1:4444/public/'); + await insertPermissions(dbAdapter, publicRealmURL, { + '*': ['read'], + [member]: ['read', 'write'], + }); + response = await realmEventRooms([ + publicRealmURL.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 only invited users join the room, and only the realm send in it', async function (assert) { + let roomId: string = (await realmEventRooms([testRealmHref])).body.rooms[ + testRealmHref + ]; + + let asOutsider = await loginAs(outsider); + let outsiderJoin = await asOutsider( + `join/${encodeURIComponent(roomId)}`, + { method: 'POST', body: '{}' }, + ); + assert.strictEqual( + outsiderJoin.status, + 403, + 'a user who was not invited may not join', + ); + let outsiderMembers = await asOutsider( + `rooms/${encodeURIComponent(roomId)}/members`, + ); + assert.strictEqual( + outsiderMembers.status, + 403, + 'a user outside the room may not list its members', + ); + + let asMember = await loginAs(member); + let join = await asMember(`join/${encodeURIComponent(roomId)}`, { + method: 'POST', + body: '{}', + }); + assert.strictEqual(join.status, 200, 'the invited grantee joins'); + + let create = await asMember( + `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 history = await asMember( + `rooms/${encodeURIComponent(roomId)}/state/m.room.history_visibility/`, + ); + assert.strictEqual( + ((await history.json()) as { history_visibility?: string }) + .history_visibility, + 'invited', + 'a member sees the history from their invite on', + ); + + let send = await asMember( + `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', + ); + }); + + test('a grant invites its user to the room, and revoking it replaces the room with one that never held them', async function (assert) { + let oldRoom: string = (await realmEventRooms([testRealmHref])).body.rooms[ + testRealmHref + ]; + let asMember = await joinAsMember(oldRoom); + assert.strictEqual( + await membershipOf(asMember, oldRoom, newcomer), + undefined, + ); + + let granted = await patchPermissions({ [newcomer]: ['read'] }); + assert.strictEqual(granted.status, 200, 'the grant succeeds'); + await waitUntil( + async () => + (await membershipOf(asMember, oldRoom, newcomer)) === 'invite', + { timeout: 10_000, timeoutMessage: 'the grantee was never invited' }, + ); + + let revoked = await patchPermissions({ [newcomer]: [] }); + assert.strictEqual(revoked.status, 200, 'the revocation succeeds'); + assert.strictEqual( + (await fetchRealmEventRoom(dbAdapter, testRealmHref)).roomId, + null, + "by the time the revocation answers, no process sends the realm's events to the old room", + ); + + let newRoom: string = (await realmEventRooms([testRealmHref])).body.rooms[ + testRealmHref + ]; + assert.ok(newRoom, 'asking again gets a room'); + assert.notStrictEqual(newRoom, oldRoom, 'a new one'); + let asMemberOfNewRoom = await joinAsMember(newRoom); + let members = (await ( + await asMemberOfNewRoom(`rooms/${encodeURIComponent(newRoom)}/members`) + ).json()) as { chunk: { state_key: string }[] }; + assert.false( + members.chunk.some((event) => event.state_key === newcomer), + "the revoked user's id is nowhere in the new room's state", + ); + }); + + test('syncing repairs a grant change made while nothing applied it', async function (assert) { + let roomId: string = (await realmEventRooms([testRealmHref])).body.rooms[ + testRealmHref + ]; + let asMember = await joinAsMember(roomId); + // Written straight to the table, the way a grant reaches a realm that + // no process is serving. + await insertPermissions(dbAdapter, new URL(testRealmHref), { + [newcomer]: ['read'], + }); + assert.strictEqual( + await membershipOf(asMember, roomId, newcomer), + undefined, + ); + + await testRealm.syncRealmEventRoom(); + + assert.strictEqual( + await membershipOf(asMember, roomId, newcomer), + 'invite', + ); + }); + }); +}); diff --git a/packages/realm-server/tests/server-endpoints/delete-realm-test.ts b/packages/realm-server/tests/server-endpoints/delete-realm-test.ts index 8ac9a928f89..ccf031ecba2 100644 --- a/packages/realm-server/tests/server-endpoints/delete-realm-test.ts +++ b/packages/realm-server/tests/server-endpoints/delete-realm-test.ts @@ -8,8 +8,10 @@ import { v4 as uuidv4 } from 'uuid'; import { asExpressions, deriveRealmName, + fetchRealmEventRoom, insert, insertPermissions, + insertRealmEventRoom, PUBLISHED_DIRECTORY_NAME, query, } from '@cardstack/runtime-common'; @@ -26,8 +28,10 @@ import { setupServerEndpointsTest } from './helpers.ts'; module(`server-endpoints/${basename(import.meta.filename)}`, function (hooks) { let context = setupServerEndpointsTest(hooks); - async function createRealmFor(ownerUserId: string) { - let endpoint = `delete-me-${uuidv4()}`; + async function createRealmFor( + ownerUserId: string, + endpoint = `delete-me-${uuidv4()}`, + ) { let response = await context.request .post('/_create-realm') .set('Accept', 'application/vnd.api+json') @@ -890,4 +894,56 @@ module(`server-endpoints/${basename(import.meta.filename)}`, function (hooks) { 'returns an invalid URL error', ); }); + + // A realm's event room is checked against its grants when the realm mounts, + // and retired when the realm is deleted. A stored room the realm server + // isn't in shows both: the mount's check forgets it, and the delete moves it + // out of the realm's hands before answering. + test('mounting a realm checks its stored event room, and deleting it retires the room', async function (assert) { + let owner = `kiwi-${uuidv4()}`; + let ownerUserId = `@${owner}:localhost`; + let first = await createRealmFor(ownerUserId); + let endpoint = `event-room-${uuidv4()}`; + let realmURL = new URL(`/${owner}/${endpoint}/`, first).href; + await insertRealmEventRoom( + context.dbAdapter, + realmURL, + '!not-a-room:localhost', + ); + + assert.strictEqual(await createRealmFor(ownerUserId, endpoint), realmURL); + await waitUntil( + async () => + (await fetchRealmEventRoom(context.dbAdapter, realmURL)).roomId === + null, + { + timeout: 10_000, + timeoutMessage: 'mounting the realm never checked its event room', + }, + ); + + await insertRealmEventRoom( + context.dbAdapter, + realmURL, + '!another-room:localhost', + ); + let response = await context.request + .delete('/_delete-realm') + .set('Accept', 'application/vnd.api+json') + .set('Content-Type', 'application/json') + .set( + 'Authorization', + `Bearer ${createRealmServerJWT( + { user: ownerUserId, sessionRoom: 'session-room-test' }, + realmSecretSeed, + )}`, + ) + .send(JSON.stringify({ data: { type: 'realm', id: realmURL } })); + assert.strictEqual(response.status, 204, 'realm deleted'); + assert.strictEqual( + (await fetchRealmEventRoom(context.dbAdapter, realmURL)).roomId, + null, + 'the deleted realm no longer has an event room', + ); + }); }); diff --git a/packages/realm-server/tests/server-endpoints/maintenance-endpoints-test.ts b/packages/realm-server/tests/server-endpoints/maintenance-endpoints-test.ts index 72796e00b95..df0569965f3 100644 --- a/packages/realm-server/tests/server-endpoints/maintenance-endpoints-test.ts +++ b/packages/realm-server/tests/server-endpoints/maintenance-endpoints-test.ts @@ -9,8 +9,10 @@ import { sumUpCreditsLedger } from '@cardstack/billing/billing-queries'; import { boxelUIChecker } from '../../lib/boxel-ui-change-checker.ts'; import { ensureTrailingSlash, + fetchRealmEventRoom, fetchRealmPermissions, insertPermissions, + insertRealmEventRoom, } from '@cardstack/runtime-common'; import { ensureMatrixAdminUser, @@ -19,6 +21,7 @@ import { matrixRegistrationSecret, matrixURL, realmSecretSeed, + waitUntil, } from '../helpers/index.ts'; import { createJWT as createRealmServerJWT } from '../../utils/jwt.ts'; import { @@ -1406,6 +1409,36 @@ module(`server-endpoints/${basename(import.meta.filename)}`, function () { ); }); + // A stored event room the realm server isn't in shows that a grant + // checks the realm's room against its grants: the check forgets it. + test("upserts: a grant checks the realm's event room against its grants", async function (assert) { + let user = '@op-test-event-room:localhost'; + await insertRealmEventRoom( + context.dbAdapter, + testRealmURL.href, + '!not-a-room:localhost', + ); + let granted = await context.request + .post( + `/_grafana-upsert-realm-user-permission` + + `?realm=${encodeURIComponent(testRealmURL.href)}` + + `&user=${encodeURIComponent(user)}` + + `&read=true&write=false`, + ) + .set('Authorization', `Bearer ${grafanaSecret}`) + .set('Content-Type', 'application/json'); + assert.strictEqual(granted.status, 200, 'the grant succeeds'); + await waitUntil( + async () => + (await fetchRealmEventRoom(context.dbAdapter, testRealmURL.href)) + .roomId === null, + { + timeout: 10_000, + timeoutMessage: "the grant never checked the realm's event room", + }, + ); + }); + test('normalises realm URL — trailing slash and stripped querystring/hash', async function (assert) { let user = '@op-test-normalize:localhost'; // Pass a URL without trailing slash + with extraneous querystring + diff --git a/packages/runtime-common/db-queries/realm-event-room-queries.ts b/packages/runtime-common/db-queries/realm-event-room-queries.ts new file mode 100644 index 00000000000..1fad7541907 --- /dev/null +++ b/packages/runtime-common/db-queries/realm-event-room-queries.ts @@ -0,0 +1,187 @@ +import type { DBAdapter } from '../db.ts'; +import { query, param } from '../expression.ts'; + +// A private realm sends its realm events once, into a Matrix room whose +// members are exactly the users the realm's grants name, instead of once into +// each of those users' session rooms. A realm is private when it has no `'*'` +// read grant: a world-readable realm has readers no grant names, who may not +// see each other or its named users, so its events keep going to session +// rooms. + +/** + * Reports whether the given realm is private, and the room its realm events + * go to when it has one. A private 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<{ isPrivate: boolean; roomId: string | null }> { + let [row] = await query(dbAdapter, [ + 'SELECT NOT EXISTS (', + ' SELECT 1 FROM realm_user_permissions', + ' WHERE realm_url =', + param(realmURL), + " AND username = '*'", + ' AND read = true', + ') AS is_private,', + '(SELECT room_id FROM realm_event_rooms WHERE realm_url =', + param(realmURL), + ') AS room_id', + ]); + return { + isPrivate: Boolean(row?.is_private), + 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 or administer, 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; +} + +/** + * The matrix users the given realm's grants name, the users its event room + * should hold: every user with their own read or write grant. The `'*'` grant + * names no one, and neither does a grant to a bare username, which has no + * matrix account to invite. + */ +export async function fetchRealmNamedGrantees( + dbAdapter: DBAdapter, + realmURL: string, +): Promise { + let rows = await query(dbAdapter, [ + 'SELECT username FROM realm_user_permissions', + 'WHERE realm_url =', + param(realmURL), + "AND username LIKE '@%:%'", + 'AND (read = true OR write = true)', + 'ORDER BY username', + ]); + return rows.map((row) => row.username as string); +} + +/** + * Stops the realm's event room being the realm's, in one statement, and keeps + * a record of it until its members have been removed (see + * `retired_realm_event_rooms`). From the moment this returns, no process sends + * the realm's events there: they go to session rooms until a host asks for + * the realm's room again and a new one is made. Returns the retired room, or + * null when the realm had none. + */ +export async function retireRealmEventRoomRow( + dbAdapter: DBAdapter, + realmURL: string, +): Promise { + let rows = await query(dbAdapter, [ + 'WITH retired AS (', + ' DELETE FROM realm_event_rooms WHERE realm_url =', + param(realmURL), + ' RETURNING realm_url, room_id', + ')', + 'INSERT INTO retired_realm_event_rooms (room_id, realm_url)', + 'SELECT room_id, realm_url FROM retired', + 'ON CONFLICT (room_id) DO NOTHING', + 'RETURNING room_id', + ]); + return (rows[0]?.room_id as string | undefined) ?? null; +} + +/** + * Records a room that never became the realm's, such as one another process + * beat to being stored, so its invites are withdrawn and it is left. + */ +export async function insertRetiredRealmEventRoom( + dbAdapter: DBAdapter, + realmURL: string, + roomId: string, +): Promise { + await query(dbAdapter, [ + 'INSERT INTO retired_realm_event_rooms (room_id, realm_url) VALUES (', + param(roomId), + ',', + param(realmURL), + ') ON CONFLICT (room_id) DO NOTHING', + ]); +} + +/** + * The realm's retired event rooms whose members haven't all been removed. + */ +export async function fetchRetiredRealmEventRooms( + dbAdapter: DBAdapter, + realmURL: string, +): Promise { + let rows = await query(dbAdapter, [ + 'SELECT room_id FROM retired_realm_event_rooms WHERE realm_url =', + param(realmURL), + 'ORDER BY room_id', + ]); + return rows.map((row) => row.room_id as string); +} + +/** + * Forgets a retired room once every member has been removed and the realm + * server has left it. + */ +export async function clearRetiredRealmEventRoom( + dbAdapter: DBAdapter, + roomId: string, +): Promise { + await query(dbAdapter, [ + 'DELETE FROM retired_realm_event_rooms WHERE room_id =', + param(roomId), + ]); +} + +/** + * Every realm with a retired event room still to be emptied. + */ +export async function fetchRealmsWithRetiredEventRooms( + dbAdapter: DBAdapter, +): Promise { + let rows = await query(dbAdapter, [ + 'SELECT DISTINCT realm_url FROM retired_realm_event_rooms ORDER BY realm_url', + ]); + return rows.map((row) => row.realm_url as string); +} diff --git a/packages/runtime-common/index.ts b/packages/runtime-common/index.ts index 894dbfe8dcb..ea6bcd0ff59 100644 --- a/packages/runtime-common/index.ts +++ b/packages/runtime-common/index.ts @@ -1997,6 +1997,7 @@ export * from './db-queries/db-types.ts'; export * from './db-queries/realm-metadata-queries.ts'; export * from './db-queries/realm-permission-queries.ts'; export * from './db-queries/session-room-queries.ts'; +export * from './db-queries/realm-event-room-queries.ts'; export * from './db-queries/user-queries.ts'; export * from './send-event.ts'; diff --git a/packages/runtime-common/matrix-client.ts b/packages/runtime-common/matrix-client.ts index 941131f75d6..5a3049ac5bd 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,138 @@ export class MatrixClient { return json.room_id; } + // The room a private realm sends its realm events into. Only a user this + // client invites may join it, and only this client's user, the one the realm + // sends its realm events as, may send in it, change its state, invite or + // kick. So its members are whoever the realm server put there, and none of + // them can forge a realm event. A member sees the room's history from the + // moment they were invited, which covers an event sent while their host was + // still joining, but nothing from before the invite. The retention policy + // bounds how far back the history goes. + async createRealmEventRoom(realmURL: string): Promise { + let response = await this.request('_matrix/client/v3/createRoom', 'POST', { + body: JSON.stringify({ + preset: 'private_chat', + visibility: 'private', + name: `Realm events: ${realmURL}`, + creation_content: { type: APP_BOXEL_REALM_EVENT_ROOM_TYPE }, + initial_state: [ + { + type: 'm.room.join_rules', + state_key: '', + content: { join_rule: 'invite' }, + }, + { + type: 'm.room.history_visibility', + state_key: '', + content: { history_visibility: 'invited' }, + }, + { + type: 'm.room.guest_access', + state_key: '', + content: { guest_access: 'forbidden' }, + }, + { + 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 }, + }, + }), + }); + if (!response.ok) { + throw await matrixRequestError( + response, + `Unable to create realm event room for ${realmURL}`, + ); + } + let json = (await response.json()) as { room_id: string }; + return json.room_id; + } + + // The membership of every user the room has a membership event for, other + // than this client's user: 'join', 'invite', 'leave', 'ban' or 'knock'. + async getRoomMemberships(roomId: string): Promise> { + let response = await this.request( + `_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/members`, + 'GET', + { signal: AbortSignal.timeout(MATRIX_WRITE_TIMEOUT_MS) }, + ); + if (!response.ok) { + throw await matrixRequestError( + response, + `Unable to read the members of room ${roomId}`, + ); + } + let json = (await response.json()) as { + chunk: { state_key: string; content: { membership?: string } }[]; + }; + let memberships = new Map(); + for (let event of json.chunk) { + if (event.state_key !== this.access?.userId && event.content.membership) { + memberships.set(event.state_key, event.content.membership); + } + } + return memberships; + } + + async inviteUser(roomId: string, userId: string) { + await this.changeMembership(roomId, 'invite', userId); + } + + async kickUser(roomId: string, userId: string, reason: string) { + await this.changeMembership(roomId, 'kick', userId, reason); + } + + async leaveRoom(roomId: string) { + let response = await this.request( + `_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/leave`, + 'POST', + { body: '{}', signal: AbortSignal.timeout(MATRIX_WRITE_TIMEOUT_MS) }, + ); + if (!response.ok) { + throw await matrixRequestError( + response, + `Unable to leave room ${roomId}`, + ); + } + } + + private async changeMembership( + roomId: string, + action: 'invite' | 'kick', + userId: string, + reason?: string, + ) { + let response = await this.request( + `_matrix/client/v3/rooms/${encodeURIComponent(roomId)}/${action}`, + 'POST', + { + body: JSON.stringify({ + user_id: userId, + ...(reason ? { reason } : {}), + }), + signal: AbortSignal.timeout(MATRIX_WRITE_TIMEOUT_MS), + }, + ); + if (!response.ok) { + throw await matrixRequestError( + response, + `Unable to ${action} ${userId} in room ${roomId}`, + ); + } + } + 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..f83423e7220 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 private realm sends its realm events into. Its +// members are the users the realm's grants name, so one copy of each event +// reaches all of them, in place of a copy in each one'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 private 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 is in the event room of every private realm +// it has opened that still names it, which can be 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 3ef885b34a2..2a2f1ed9590 100644 --- a/packages/runtime-common/realm.ts +++ b/packages/runtime-common/realm.ts @@ -2455,6 +2455,37 @@ export interface RealmAdapter { opts?: RealmEventDelivery, ): Promise; + // A private realm sends its realm events into a room of its own, whose + // members are the users its grants name. An adapter without these sends + // every realm event to session rooms. + // + // The realm's event room for a host signed in as `member`, created if need + // be, with `member` invited by the time it resolves. Undefined when the + // realm is world-readable or no grant names `member`. + ensureRealmEventRoom?( + realmUrl: string, + matrixClient: MatrixClient, + dbAdapter: DBAdapter, + member: string, + ): Promise; + + // Brings the members of the realm's event room, if it has one, in step with + // its grants. Never rejects. + syncRealmEventRoom?( + realmUrl: string, + matrixClient: MatrixClient, + dbAdapter: DBAdapter, + ): Promise; + + // For a revoked grant: the realm's event room stops receiving its events + // before this resolves, in every process, and is replaced the next time a + // host asks for it. + retireRealmEventRoom?( + realmUrl: string, + matrixClient: MatrixClient, + dbAdapter: DBAdapter, + ): Promise; + // optional, set this to override _lint endpoint behavior in tests lintStub?( request: Request, @@ -3539,6 +3570,28 @@ export class Realm { return sessionRoom; } + // The room this realm sends its realm events into, for a host signed in as + // `member`. Undefined when the realm is world-readable, no grant names + // `member`, or the adapter sends every event to session rooms. + async ensureRealmEventRoom(member: string): Promise { + return await this.#adapter.ensureRealmEventRoom?.( + this.url, + this.#matrixClient, + this.#dbAdapter, + member, + ); + } + + // Brings the members of this realm's event room in step with its grants. + // Never rejects. + async syncRealmEventRoom(): Promise { + await this.#adapter.syncRealmEventRoom?.( + this.url, + this.#matrixClient, + this.#dbAdapter, + ); + } + private async readinessCheck( request: Request, requestContext: RequestContext, @@ -4409,6 +4462,10 @@ export class Realm { // backlog of system-priority indexing work. async start(opts?: { fromScratchIndexPriority?: number }) { this.#startedUp.fulfill((() => this.#startup(opts))()); + // Repairs a grant change no process applied to the realm's event room, + // such as one made while nothing served the realm. Not awaited: it is + // paced with the realm's events and never rejects. + void this.syncRealmEventRoom(); if (this.#adapter.fileWatcherEnabled) { await this.startFileWatcher(); @@ -16575,11 +16632,44 @@ export class Realm { // permission PATCHes are admin-rare so the over-invalidation is // negligible). await this.clearRealmIndexCachesAndBroadcast(); + // The realm's event room holds the users its grants name. A revoked grant + // retires the room before this answers, so nothing more reaches the + // revoked user through it; a grant that names someone new invites them, + // paced behind the realm's events and not awaited. + let revoked = Object.entries(patch).some( + ([user, permissions]) => + user !== '*' && + (currentPermissions[user]?.includes('read') || + currentPermissions[user]?.includes('write')) && + !permissions?.includes('read') && + !permissions?.includes('write'), + ); + if (revoked) { + await this.#adapter.retireRealmEventRoom?.( + this.url, + this.#matrixClient, + this.#dbAdapter, + ); + } else { + void this.syncRealmEventRoom(); + } // Tell each affected user their accessible-realm set changed so a running // 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 also moves where the realm's events go: a + // private realm sends them into its event room, a world-readable one into + // session rooms. So every user the realm names 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); }