matrix: Add room retention policy (#2040)

packages/matrix/docker/synapse/dev/homeserver.yaml

@@ -22,6 +22,9 @@ log_config: "/data/log.config"
 presence:
   enabled: false
 
+retention:
+  enabled: true
+
 rc_messages_per_second: 10000
 rc_message_burst_count: 10000
 rc_registration:

packages/matrix/docker/synapse/index.ts

@@ -429,6 +429,45 @@ export async function getJoinedRooms(accessToken: string) {
   return joined_rooms;
 }
 
+export async function getRoomStateEventType(
+  accessToken: string,
+  roomId: string,
+  eventType: string,
+) {
+  let response = await fetch(
+    `http://localhost:${SYNAPSE_PORT}/_matrix/client/v3/rooms/${roomId}/state/${eventType}`,
+    {
+      headers: {
+        Authorization: `Bearer ${accessToken}`,
+      },
+    },
+  );
+  return await response.json();
+}
+
+export async function getRoomName(accessToken: string, roomId: string) {
+  return await getRoomStateEventType(accessToken, roomId, 'm.room.name');
+}
+
+export async function getRoomRetentionPolicy(
+  accessToken: string,
+  roomId: string,
+) {
+  return await getRoomStateEventType(accessToken, roomId, 'm.room.retention');
+}
+
+export async function getRoomMembers(roomId: string, accessToken: string) {
+  let response = await fetch(
+    `http://localhost:${SYNAPSE_PORT}/_matrix/client/v3/rooms/${roomId}/joined_members`,
+    {
+      headers: {
+        Authorization: `Bearer ${accessToken}`,
+      },
+    },
+  );
+  return await response.json();
+}
+
 export async function sync(accessToken: string) {
   let response = await fetch(
     `http://localhost:${SYNAPSE_PORT}/_matrix/client/v3/sync`,

packages/matrix/docker/synapse/test-without-registration-token/homeserver.yaml

@@ -19,6 +19,9 @@ database:
 
 log_config: "/data/log.config"
 
+retention:
+  enabled: true
+
 rc_messages_per_second: 10000
 rc_message_burst_count: 10000
 rc_registration:

packages/matrix/docker/synapse/test/homeserver.yaml

@@ -22,6 +22,9 @@ log_config: "/data/log.config"
 presence:
   enabled: false
 
+retention:
+  enabled: true
+
 rc_messages_per_second: 10000
 rc_message_burst_count: 10000
 rc_registration:

packages/matrix/tests/auth-rooms.spec.ts

@@ -0,0 +1,78 @@
+import { expect, test } from '@playwright/test';
+import {
+  synapseStart,
+  synapseStop,
+  type SynapseInstance,
+  registerUser,
+  getJoinedRooms,
+  getRoomMembers,
+  getRoomRetentionPolicy,
+} from '../docker/synapse';
+import { smtpStart, smtpStop } from '../docker/smtp4dev';
+import { login, registerRealmUsers, setupUserSubscribed } from '../helpers';
+
+import {
+  appURL,
+  startServer as startRealmServer,
+  type IsolatedRealmServer,
+} from '../helpers/isolated-realm-server';
+
+test.describe('Auth rooms', () => {
+  let synapse: SynapseInstance;
+  let realmServer: IsolatedRealmServer;
+  let user: { accessToken: string };
+
+  test.beforeEach(async () => {
+    // synapse defaults to 30s for beforeEach to finish, we need a bit more time
+    // to safely start the realm
+    test.setTimeout(120_000);
+    synapse = await synapseStart({
+      template: 'test',
+    });
+    await smtpStart();
+
+    await registerRealmUsers(synapse);
+    realmServer = await startRealmServer();
+
+    user = await registerUser(synapse, 'user1', 'pass');
+    await setupUserSubscribed('@user1:localhost', realmServer);
+  });
+
+  test.afterEach(async () => {
+    await synapseStop(synapse.synapseId);
+    await smtpStop();
+    await realmServer.stop();
+  });
+
+  test('auth rooms have a retention policy', async ({ page }) => {
+    await login(page, 'user1', 'pass', { url: appURL });
+
+    let roomIds = await getJoinedRooms(user.accessToken);
+
+    let roomIdToMembers = new Map<string, any>();
+
+    for (let room of roomIds) {
+      let members = await getRoomMembers(room, user.accessToken);
+      roomIdToMembers.set(room, members);
+    }
+
+    let realmUsers = ['@base_realm:localhost', '@test_realm:localhost'];
+
+    let realmRoomIds = roomIds.filter((room) =>
+      realmUsers.some((user) => roomIdToMembers.get(room)?.joined[user]),
+    );
+
+    expect(realmRoomIds.length).toBe(realmUsers.length);
+
+    for (let room of realmRoomIds) {
+      let retentionPolicy = await getRoomRetentionPolicy(
+        user.accessToken,
+        room,
+      );
+
+      expect(retentionPolicy).toMatchObject({
+        max_lifetime: 60 * 60 * 1000,
+      });
+    }
+  });
+});

packages/runtime-common/matrix-client.ts

@@ -1,5 +1,7 @@
 import { Sha256 } from '@aws-crypto/sha256-js';
 import { uint8ArrayToHex } from './index';
+import { REALM_ROOM_RETENTION_POLICY_MAX_LIFETIME } from './realm';
+import { Deferred } from './deferred';
 
 export interface MatrixAccess {
   accessToken: string;
@@ -13,6 +15,7 @@ export class MatrixClient {
   private access: MatrixAccess | undefined;
   private password?: string;
   private seed?: string;
+  private loggedIn = new Deferred<void>();
 
   constructor({
     matrixURL,
@@ -44,6 +47,10 @@ export class MatrixClient {
     return this.access !== undefined;
   }
 
+  async waitForLogin() {
+    return this.loggedIn.promise;
+  }
+
   private async request(
     path: string,
     method: 'POST' | 'PUT' | 'PATCH' | 'DELETE' | 'GET' = 'GET',
@@ -96,18 +103,21 @@ export class MatrixClient {
     let json = await response.json();
 
     if (!response.ok) {
-      throw new Error(
+      let error = new Error(
         `Unable to login to matrix ${this.matrixURL.href} as user ${
           this.username
         }: status ${response.status} - ${JSON.stringify(json)}`,
       );
+      this.loggedIn.reject(error);
+      throw error;
     }
     let {
       access_token: accessToken,
       device_id: deviceId,
       user_id: userId,
     } = json;
     this.access = { accessToken, deviceId, userId };
+    this.loggedIn.fulfill();
   }
 
   async getJoinedRooms() {
@@ -146,9 +156,41 @@ export class MatrixClient {
         } - ${JSON.stringify(json)}`,
       );
     }
+
+    await this.setRoomRetentionPolicy(
+      json.room_id,
+      REALM_ROOM_RETENTION_POLICY_MAX_LIFETIME,
+    );
+
     return json.room_id;
   }
 
+  async setRoomRetentionPolicy(roomId: string, maxLifetimeMs: number) {
+    try {
+      let roomState = await this.request(
+        `_matrix/client/v3/rooms/${roomId}/state`,
+      );
+
+      let roomStateJson = await roomState.json();
+
+      let retentionState = roomStateJson.find(
+        (event: any) => event.type === 'm.room.retention',
+      );
+
+      let retentionStateKey = retentionState?.content.key ?? '';
+
+      await this.request(
+        `_matrix/client/v3/rooms/${roomId}/state/m.room.retention/${retentionStateKey}`,
+        'PUT',
+        {
+          body: JSON.stringify({ max_lifetime: maxLifetimeMs }),
+        },
+      );
+    } catch (e) {
+      console.error('error setting retention policy', e);
+    }
+  }
+
   async setAccountData<T>(type: string, data: T) {
     let response = await this.request(
       `_matrix/client/v3/user/${encodeURIComponent(

packages/runtime-common/realm.ts

@@ -81,6 +81,8 @@ import {
   Utils,
 } from './matrix-backend-authentication';
 
+export const REALM_ROOM_RETENTION_POLICY_MAX_LIFETIME = 60 * 60 * 1000;
+
 export interface RealmSession {
   canRead: boolean;
   canWrite: boolean;
@@ -330,6 +332,9 @@ export class Realm {
       ),
     ]);
 
+    // TODO: remove after running in all environments; CS-7875
+    this.backfillRetentionPolicies();
+
     let loader = new Loader(fetch, virtualNetwork.resolveImport);
     adapter.setLoader?.(loader);
 
@@ -449,6 +454,23 @@ export class Realm {
     });
   }
 
+  // TODO: remove after running in all environments; CS-7875
+  private async backfillRetentionPolicies() {
+    try {
+      await this.#matrixClient.waitForLogin();
+
+      let roomIds = (await this.#matrixClient.getJoinedRooms()).joined_rooms;
+      for (let roomId of roomIds) {
+        await this.#matrixClient.setRoomRetentionPolicy(
+          roomId,
+          REALM_ROOM_RETENTION_POLICY_MAX_LIFETIME,
+        );
+      }
+    } catch (e) {
+      console.error('backfillRetentionPolicies: error', e);
+    }
+  }
+
   async indexing() {
     return this.#realmIndexUpdater.indexing();
   }