diff --git a/packages/realm-server/node-realm.ts b/packages/realm-server/node-realm.ts index 29efabfc740..66eb2affe1b 100644 --- a/packages/realm-server/node-realm.ts +++ b/packages/realm-server/node-realm.ts @@ -1,5 +1,6 @@ import type { RealmAdapter, + RealmEventDelivery, Kind, FileRef, DBAdapter, @@ -445,6 +446,7 @@ export class NodeAdapter implements RealmAdapter { realmUrl: string, matrixClient: MatrixClient, dbAdapter: DBAdapter, + opts?: RealmEventDelivery, ): Promise { realmEventsLog.debug('Broadcasting realm event', event); const eventWithRealmURL: RealmEventContent = { @@ -462,7 +464,7 @@ export class NodeAdapter implements RealmAdapter { event: eventWithRealmURL, newTxnId: () => matrixClient.nextTxnId(), prepare: (content) => - this.prepareRoomSends(content, realmUrl, matrixClient, dbAdapter), + this.prepareRoomSends(content, realmUrl, matrixClient, dbAdapter, opts), }); } @@ -471,6 +473,7 @@ export class NodeAdapter implements RealmAdapter { realmUrl: string, matrixClient: MatrixClient, dbAdapter: DBAdapter, + delivery?: RealmEventDelivery, ): Promise { if (dbAdapter.isClosed) { return []; @@ -502,6 +505,12 @@ export class NodeAdapter implements RealmAdapter { this.lastLookupFoundNoRooms ? 0 : undefined, ); this.lastLookupFoundNoRooms = Object.keys(dmRooms).length === 0; + if (delivery?.recipients) { + let recipients = new Set(delivery.recipients); + dmRooms = Object.fromEntries( + Object.entries(dmRooms).filter(([userId]) => recipients.has(userId)), + ); + } realmEventsLog.debug('Sending to dm rooms', Object.values(dmRooms)); diff --git a/packages/realm-server/tests/atomic-endpoints-test.ts b/packages/realm-server/tests/atomic-endpoints-test.ts index 81de36022fa..82e46402c08 100644 --- a/packages/realm-server/tests/atomic-endpoints-test.ts +++ b/packages/realm-server/tests/atomic-endpoints-test.ts @@ -1140,6 +1140,202 @@ module(basename(import.meta.filename), function () { } }); + test('a write that changes nothing announces its index event only to its own writer', async function (assert) { + let source = ` + import { field, CardDef, contains } from "@cardstack/base/card-api"; + import StringField from "@cardstack/base/string"; + export class Place extends CardDef { + static displayName = 'Place'; + @field name = contains(StringField); + } + `.trim(); + let doc = (op: 'add' | 'update') => ({ + 'atomic:operations': [ + { + op, + href: 'place-modules/place-silent.gts', + data: { + type: 'source', + attributes: { content: source }, + meta: {}, + }, + }, + ], + }); + let post = (op: 'add' | 'update', clientRequestId?: string) => { + let req = request + .post('/_atomic') + .set('Accept', SupportedMimeType.JSONAPI) + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'user', ['read', 'write'])}`, + ); + if (clientRequestId) { + req = req.set('X-Boxel-Client-Request-Id', clientRequestId); + } + return req.send(JSON.stringify(doc(op))); + }; + + let addResponse = await post('add'); + assert.strictEqual(addResponse.status, 201, 'initial write succeeds'); + // The initial write's own index event lands before the broadcasts + // below are recorded. + await testRealm.incrementalIndexing(); + + let broadcasts: { + event: RealmEventContent; + recipients: string[] | undefined; + }[] = []; + let originalBroadcast = + testRealmAdapter.broadcastRealmEvent.bind(testRealmAdapter); + testRealmAdapter.broadcastRealmEvent = (async ( + event, + realmUrl, + matrixClient, + dbAdapter, + opts, + ) => { + broadcasts.push({ event, recipients: opts?.recipients }); + return originalBroadcast( + event, + realmUrl, + matrixClient, + dbAdapter, + opts, + ); + }) as RealmAdapter['broadcastRealmEvent']; + let indexEvents = () => + broadcasts.filter( + ({ event }) => + event.eventName === 'index' && + 'indexType' in event && + event.indexType === 'incremental', + ); + + try { + let anonymous = await post('update'); + assert.strictEqual(anonymous.status, 201, 'no-op write succeeds'); + assert.deepEqual( + broadcasts.map(({ event }) => event.eventName), + [], + 'a no-op write that names no request id broadcasts nothing', + ); + + let identified = await post('update', 'test-noop-request'); + assert.strictEqual(identified.status, 201, 'no-op write succeeds'); + let events = indexEvents(); + assert.strictEqual( + events.length, + 1, + 'a no-op write that names a request id announces its index event once', + ); + assert.strictEqual( + (events[0]?.event as { clientRequestId?: string | null }) + .clientRequestId, + 'test-noop-request', + 'the event carries the request id', + ); + assert.deepEqual( + events[0]?.recipients, + ['user'], + "the event goes to the writer's own session room only", + ); + assert.strictEqual( + broadcasts.length, + 1, + 'no initiation or file update event goes out for a no-op write', + ); + + // A module that changes ahead of a byte-identical instance runs + // an index pass of its own, and that pass is every subscriber's + // news even though the instance changed nothing. + let instance = JSON.stringify({ + data: { + type: 'card', + attributes: { name: 'Silent' }, + meta: { + adoptsFrom: { + module: './place-modules/place-silent', + name: 'Place', + }, + }, + }, + }); + let mixed = (moduleSource: string) => ({ + 'atomic:operations': [ + { + op: 'update', + href: 'place-modules/place-silent.gts', + data: { + type: 'source', + attributes: { content: moduleSource }, + meta: {}, + }, + }, + { + op: 'add', + href: 'place-silent-1.json', + data: { + type: 'source', + attributes: { content: instance }, + meta: {}, + }, + }, + ], + }); + let send = (body: object) => + request + .post('/_atomic') + .set('Accept', SupportedMimeType.JSONAPI) + .set( + 'Authorization', + `Bearer ${createJWT(testRealm, 'user', ['read', 'write'])}`, + ) + .send(JSON.stringify(body)); + assert.strictEqual( + (await send(mixed(source))).status, + 201, + 'the instance is written beside the unchanged module', + ); + await testRealm.incrementalIndexing(); + + broadcasts = []; + let changedSource = source.replace( + "static displayName = 'Place';", + "static displayName = 'Silent Place';", + ); + let mixedBody = mixed(changedSource); + mixedBody['atomic:operations'][1].op = 'update'; + assert.strictEqual( + (await send(mixedBody)).status, + 201, + 'the module changes and the instance does not', + ); + await testRealm.incrementalIndexing(); + let flushEvents = indexEvents(); + assert.strictEqual( + flushEvents.length, + 1, + "the module flush's pass is announced", + ); + assert.strictEqual( + flushEvents[0]?.recipients, + undefined, + 'to every subscriber', + ); + assert.true( + ( + flushEvents[0]?.event as { invalidations?: string[] } + ).invalidations?.some((url) => + url.includes('place-modules/place-silent'), + ), + 'naming the module the flush invalidated', + ); + } finally { + testRealmAdapter.broadcastRealmEvent = originalBroadcast; + } + }); + test('waitForIndex write preserves compound field values backed by an unexported FieldDef', async function (assert) { let logSource = ` import { diff --git a/packages/realm-server/tests/node-realm-test.ts b/packages/realm-server/tests/node-realm-test.ts index d0c43dc266b..2b59025c6fe 100644 --- a/packages/realm-server/tests/node-realm-test.ts +++ b/packages/realm-server/tests/node-realm-test.ts @@ -245,6 +245,63 @@ module(basename(import.meta.filename), function (hooks) { assert.strictEqual(logins, 2, 'the login is tried again after a pause'); assert.deepEqual(sent, [staleRoomId], 'the event still reaches the room'); }); + + test('delivers only to the named recipients when the broadcast names any', async function (assert) { + await insertSessionForAlice(); + await insertUser( + dbAdapter, + '@carol:localhost', + 'cus_carol', + 'c@example.com', + ); + await upsertSessionRoom( + dbAdapter, + '@carol:localhost', + '!room-carol:localhost', + ); + await insertPermissions(dbAdapter, realmURL, { + '@carol:localhost': ['read'], + }); + + let sentTo: string[] = []; + let matrixClient = { + login: async () => undefined, + getUserId: () => '@realm_server:localhost', + isLoggedIn: () => true, + nextTxnId: () => `txn-${sentTo.length}`, + sendEvent: async (roomId: string) => { + sentTo.push(roomId); + return '$event'; + }, + } as unknown as MatrixClient; + + let adapter = new NodeAdapter('/tmp', false, new RealmEventDispatcher()); + await adapter.broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + ); + assert.deepEqual( + sentTo.sort(), + ['!room-alice:localhost', '!room-carol:localhost'], + 'with no recipients, every subscriber receives the event', + ); + + sentTo = []; + await adapter.broadcastRealmEvent( + makeEvent(), + realmURL.href, + matrixClient, + dbAdapter, + { recipients: ['@carol:localhost', '@nobody:localhost'] }, + ); + assert.deepEqual( + sentTo, + ['!room-carol:localhost'], + "only the named subscriber's room receives the event", + ); + }); }); module(`${basename(import.meta.filename)} | file stat probing`, function () { diff --git a/packages/runtime-common/index.ts b/packages/runtime-common/index.ts index 7569dc89415..894dbfe8dcb 100644 --- a/packages/runtime-common/index.ts +++ b/packages/runtime-common/index.ts @@ -2020,6 +2020,7 @@ export { RealmAuthDataSource } from './realm-auth-data-source.ts'; export type { Kind, RealmAdapter, + RealmEventDelivery, FileRef, RealmIndexCounts, RealmInfo, diff --git a/packages/runtime-common/realm.ts b/packages/runtime-common/realm.ts index 5e9dfe325ff..3f92aded0b3 100644 --- a/packages/runtime-common/realm.ts +++ b/packages/runtime-common/realm.ts @@ -2347,6 +2347,14 @@ export interface WriteOptions { onDurable?: () => void; } +// Who a realm event is delivered to. By default it goes to every session room +// subscribed to the realm (see `fetchRealmSessionRooms`). `recipients` narrows +// that to the named matrix users' session rooms, for an event that concerns +// only them; a recipient with no subscribed session room receives nothing. +export interface RealmEventDelivery { + recipients?: string[]; +} + export interface RealmAdapter { readdir( path: LocalPath, @@ -2444,6 +2452,7 @@ export interface RealmAdapter { realmUrl: string, matrixClient: MatrixClient, dbAdapter: DBAdapter, + opts?: RealmEventDelivery, ): Promise; // optional, set this to override _lint endpoint behavior in tests @@ -4164,6 +4173,9 @@ export class Realm { // False for a writer that asked not to announce its write (see // `WriteOptions.broadcastRealmEvents`). broadcastRealmEvents?: boolean; + // Narrows delivery to these matrix users' session rooms (see + // `RealmEventDelivery`). Absent, every subscriber receives the event. + recipients?: string[]; }, ): void { // A pass indexed alongside other publishes is announced once, by the @@ -4254,7 +4266,10 @@ export class Realm { versions = versionsOnlyThisWriterChanged(passes, versions); content = this.#withCoalescedWrites(content, coalescedWrites); } - this.broadcastRealmEvent(withVersionsIfItFits(content, versions)); + this.broadcastRealmEvent( + withVersionsIfItFits(content, versions), + opts?.recipients ? { recipients: opts.recipients } : undefined, + ); } // `coalescedWrites` on a shared pass's event, as much of it as fits. @@ -5793,9 +5808,26 @@ export class Realm { }); } } else { - // Nothing changed on disk (e.g., every file's content was already what - // the caller staged). Preserve the pre-existing always-broadcast - // behavior. + // Nothing changed on disk: every file's content was already what the + // caller staged. + // + // A module flush that ran ahead of byte-identical instances is the only + // pass that can stand behind this event, and what it invalidated is + // owed to every subscriber like any other pass — unless it was shared + // with a writer that announces it, whose event already carries these + // invalidations, so `passes` goes along and this broadcast stands down. + // + // Without such a pass the request changed nothing anyone else could be + // holding, so it is not announced to every subscriber of the realm: a + // public realm has a session room for every user, and a bulk writer + // re-sending unchanged content would cost each of them an event per + // request. The one party that can still be waiting on it is the writer, + // whose client may key a wait on the event carrying its request id. So + // a request that named an id hears it back in its writer's own session + // room, and one that named none hears nothing. A request admitted + // without a user — an anonymous write to a world-writable realm — names + // no room to narrow to, so its event goes to every subscriber as the + // only way to reach a writer that may be waiting on it. // // The versions are still reported, and they are still the file's own: a // write that found the bytes it staged already there left the file @@ -5806,21 +5838,28 @@ export class Realm { // whatever a mid-loop flush put there, and usually nothing at all. That // is why the member is documented as describing what the request wrote // rather than as a companion to the list beside it. - // - // A module flush that ran ahead of these byte-identical instances is the - // only pass behind this broadcast, and when it was shared with a writer - // that announces it, that writer's event already carries this one's - // invalidations — so `passes` goes along, and this broadcast stands down. - this.broadcastIncrementalInvalidationEvent([...invalidations], { - clientRequestId, - ...(clientAuthored ? { clientAuthored } : {}), - versions, - generation: indexGeneration, - invalidatedTypes: invalidatedTypes.value, - passes, - ownChanges, - broadcastRealmEvents, - }); + let passOwed = + invalidations.size > 0 || + passes.some((pass) => pass !== undefined && !pass.announcedByPeer); + if (passOwed || clientRequestId) { + this.broadcastIncrementalInvalidationEvent([...invalidations], { + clientRequestId, + ...(clientAuthored ? { clientAuthored } : {}), + versions, + generation: indexGeneration, + invalidatedTypes: invalidatedTypes.value, + passes, + ownChanges, + broadcastRealmEvents, + ...(!passOwed && initiatingUser + ? { recipients: [initiatingUser] } + : {}), + }); + } else { + this.#log.debug( + `not broadcasting index event for ${this.url}: the write changed nothing and named no request id`, + ); + } } // A mixed batch whose instances all turned out to be byte-identical // leaves the flush's deferred set with no later pass to fold it into. @@ -17651,12 +17690,16 @@ export class Realm { }); } - private async broadcastRealmEvent(event: RealmEventContent): Promise { + private async broadcastRealmEvent( + event: RealmEventContent, + delivery?: RealmEventDelivery, + ): Promise { this.#adapter.broadcastRealmEvent( event, this.url, this.#matrixClient, this.#dbAdapter, + delivery, ); }