From 7d13d4fcfd688b7ef4c1d447f5a9e76145d7ee65 Mon Sep 17 00:00:00 2001 From: tajniak81 <13187254+tajniak81@users.noreply.github.com> Date: Sun, 18 Oct 2026 23:22:01 +0200 Subject: [PATCH] Mobile: Simplified Sliding Sync (MSC4186) The famedly SDK only knows /sync, but its sync loop asks Client.sync for each response. WarpClient answers that with a sliding sync request and rewrites the response into the /sync shape (slidingToSyncJson), so decryption, the database, timelines and notifications work unchanged. - Two lists, rooms and direct chats (with all their members, so DM names come out right), over a window that starts at the 20 most recent rooms for a quick first screen and widens to every room (up to 1000). - The state the screens and encryption need, $LAZY members and $ME, and call memberships; timeline_limit 10, with prev_batch for history. - Extensions: to_device (its since rides in the token), e2ee, account data, receipts, typing. - Rooms the user left while the connection was open become leaves; invite_state and knock_state become invites and knocks. - The SDK's token is ss1||. A classic token, none, or M_UNKNOWN_POS starts a new connection; a server without org.matrix.simplified_msc3575 gets plain /sync. Tested against Warp: first sync, encryption setup (account data and oneShotSync through the connection), live messages, unread counts and typing. Co-Authored-By: Claude Opus 5.5 --- apps/mobile/lib/matrix/client.dart | 4 +- apps/mobile/lib/matrix/sliding_sync.dart | 430 +++++++++++++++++++++++ apps/mobile/test/sliding_sync_test.dart | 240 +++++++++++++ 3 files changed, 673 insertions(+), 1 deletion(-) create mode 100644 apps/mobile/lib/matrix/sliding_sync.dart create mode 100644 apps/mobile/test/sliding_sync_test.dart diff --git a/apps/mobile/lib/matrix/client.dart b/apps/mobile/lib/matrix/client.dart index 93f6364..9b5f902 100644 --- a/apps/mobile/lib/matrix/client.dart +++ b/apps/mobile/lib/matrix/client.dart @@ -9,6 +9,8 @@ import 'package:path/path.dart' as p; import 'package:path_provider/path_provider.dart'; import 'package:sqflite/sqflite.dart' as sqflite; +import 'sliding_sync.dart'; + /// The name the session's device gets on the homeserver. const deviceDisplayName = 'Warp Android'; @@ -42,7 +44,7 @@ Future createClient({bool startSync = true}) async { ), ); - final client = Client( + final client = WarpClient( 'Warp', database: database, verificationMethods: { diff --git a/apps/mobile/lib/matrix/sliding_sync.dart b/apps/mobile/lib/matrix/sliding_sync.dart new file mode 100644 index 0000000..7e00e91 --- /dev/null +++ b/apps/mobile/lib/matrix/sliding_sync.dart @@ -0,0 +1,430 @@ +import 'package:matrix/matrix.dart'; + +/// Simplified Sliding Sync (MSC4186) for the famedly SDK, which only knows +/// `/sync`. +/// +/// The SDK's sync loop calls [Client.sync] and processes what it returns: +/// rooms, timelines, to-device messages, device lists, key counts. This +/// client answers that call with a sliding sync request instead, rewritten +/// into the `/sync` response the loop expects. Everything downstream — +/// decryption, the database, timelines, notifications — is unchanged. +/// +/// What it buys is the first sync. A classic initial sync sends every room +/// with its full state before anything can be drawn; here the first +/// request asks for the 20 most recent rooms with a handful of state events +/// each, so the room list appears at once, and the next requests widen the +/// window to every room. +/// +/// The SDK's token (`prevBatch`) becomes `ss1||`. A +/// classic token, or none, starts a new connection; so does +/// `M_UNKNOWN_POS`, the server saying it has forgotten the connection +/// (after a restart, or 30 idle minutes). A server that doesn't advertise +/// `org.matrix.simplified_msc3575` gets plain `/sync`. +class WarpClient extends Client { + WarpClient( + super.clientName, { + required super.database, + super.verificationMethods, + super.supportedLoginTypes, + super.nativeImplementations, + super.logLevel, + super.importantStateEvents, + super.onSoftLogout, + }); + + /// Turns sliding sync off, for a server whose implementation misbehaves. + bool slidingSyncEnabled = true; + + /// Whether the homeserver has sliding sync; null until asked. + bool? _serverHasSliding; + Uri? _askedHomeserver; + + /// How many rooms each list asks for; grows to the list's count. + int _window = firstWindow; + + static const firstWindow = 20; + static const maxWindow = 1000; + static const _connId = 'warp-mobile'; + + /// Whether the last sync used sliding sync, for the settings screen. + bool get usingSlidingSync => _serverHasSliding == true && slidingSyncEnabled; + + @override + Future sync({ + String? filter, + String? since, + bool? fullState, + PresenceType? setPresence, + int? timeout, + bool? useStateAfter, + }) async { + if (slidingSyncEnabled && await _hasSliding()) { + final token = SlidingToken.parse(since); + try { + return await _slidingSync(token, timeout, setPresence); + } on MatrixException catch (e) { + if (e.errcode != 'M_UNKNOWN_POS') rethrow; + Logs().i('Sliding sync connection expired; starting a new one'); + _window = firstWindow; + return _slidingSync( + SlidingToken(null, token.toDeviceSince), + null, + setPresence, + ); + } + } + // A sliding sync token means nothing to /sync: start over. + final classicSince = since != null && SlidingToken.isSliding(since) + ? null + : since; + return super.sync( + filter: filter, + since: classicSince, + fullState: fullState, + setPresence: setPresence, + timeout: classicSince == null ? null : timeout, + useStateAfter: useStateAfter, + ); + } + + Future _hasSliding() async { + if (_askedHomeserver != homeserver) { + _serverHasSliding = null; + _askedHomeserver = homeserver; + } + if (_serverHasSliding != null) return _serverHasSliding!; + try { + final versions = await getVersions(); + _serverHasSliding = + versions.unstableFeatures?['org.matrix.simplified_msc3575'] == true; + } catch (e) { + Logs().w('Could not ask the server about sliding sync', e); + return false; + } + Logs().i( + _serverHasSliding! + ? 'Using Simplified Sliding Sync' + : 'The server has no sliding sync; using /sync', + ); + return _serverHasSliding!; + } + + Future _slidingSync( + SlidingToken token, + int? timeout, + PresenceType? setPresence, + ) async { + final response = await request( + RequestType.POST, + '/client/unstable/org.matrix.simplified_msc3575/sync', + query: { + 'pos': ?token.pos, + // A new connection answers at once; only an established one waits. + if (token.pos != null && timeout != null) 'timeout': '$timeout', + if (setPresence != null) 'set_presence': setPresence.name, + }, + data: slidingRequest( + window: _window, + toDeviceSince: token.toDeviceSince, + connId: _connId, + ), + ); + + // Widen the window to every room, once the first rooms are drawn. + final lists = response['lists']; + if (lists is Map) { + var count = 0; + for (final list in lists.values) { + if (list is Map && list['count'] is int) { + count = count > list['count'] ? count : list['count'] as int; + } + } + if (count > _window) _window = count > maxWindow ? maxWindow : count; + } + + final extensions = response['extensions']; + final toDevice = extensions is Map ? extensions['to_device'] : null; + final next = SlidingToken( + response['pos'] as String?, + (toDevice is Map ? toDevice['next_batch'] as String? : null) ?? + token.toDeviceSince, + ); + return SyncUpdate.fromJson( + slidingToSyncJson(response, userId: userID!, nextBatch: next.toString()), + ); + } +} + +/// The state each room is sent with: what the room list and the room +/// screen draw, what encryption needs, and calls. The members are the +/// timeline's senders and the user; direct chats get all of theirs (two +/// people), so their names come out right. +const slidingRequiredState = [ + ['m.room.create', ''], + ['m.room.name', ''], + ['m.room.avatar', ''], + ['m.room.topic', ''], + ['m.room.canonical_alias', ''], + ['m.room.encryption', ''], + ['m.room.power_levels', ''], + ['m.room.join_rules', ''], + ['m.room.history_visibility', ''], + ['m.room.guest_access', ''], + ['m.room.tombstone', ''], + ['m.room.pinned_events', ''], + ['m.space.child', '*'], + ['m.space.parent', '*'], + ['m.room.member', r'$LAZY'], + ['m.room.member', r'$ME'], + ['org.matrix.msc3401.call.member', '*'], +]; + +/// The request body: two lists (rooms, and direct chats with all their +/// members) over the same window, and the extensions encryption needs. +Map slidingRequest({ + required int window, + required String? toDeviceSince, + required String connId, +}) { + final ranges = [ + [0, window - 1], + ]; + return { + 'conn_id': connId, + 'lists': { + 'rooms': { + 'ranges': ranges, + 'timeline_limit': 10, + 'required_state': slidingRequiredState, + 'filters': {'is_dm': false}, + }, + 'dms': { + 'ranges': ranges, + 'timeline_limit': 10, + 'required_state': [ + ...slidingRequiredState, + ['m.room.member', '*'], + ], + 'filters': {'is_dm': true}, + }, + }, + 'extensions': { + 'to_device': {'enabled': true, 'since': ?toDeviceSince}, + 'e2ee': {'enabled': true}, + 'account_data': {'enabled': true}, + 'receipts': {'enabled': true}, + 'typing': {'enabled': true}, + }, + }; +} + +/// The SDK's sync token while sliding sync is in use. +class SlidingToken { + const SlidingToken(this.pos, this.toDeviceSince); + + final String? pos; + final String? toDeviceSince; + + static const _prefix = 'ss1|'; + + static bool isSliding(String token) => token.startsWith(_prefix); + + /// Reads a token; a classic `/sync` token, or none, is a new connection. + factory SlidingToken.parse(String? token) { + if (token == null || !isSliding(token)) { + return const SlidingToken(null, null); + } + final parts = token.substring(_prefix.length).split('|'); + String? part(int i) => + parts.length > i && parts[i].isNotEmpty ? parts[i] : null; + return SlidingToken(part(0), part(1)); + } + + @override + String toString() => '$_prefix${pos ?? ''}|${toDeviceSince ?? ''}'; +} + +/// Rewrites a sliding sync response into the `/sync` response with the +/// same news, for [SyncUpdate.fromJson]. +/// +/// A room with `invite_state` is an invite and one with `knock_state` a +/// knock. Any other room is joined, unless the user's own latest membership +/// in what was sent is a leave or a ban: sliding sync keeps rooms the user +/// left while the connection was open in its lists, which is how a client +/// hears that one is gone. +Map slidingToSyncJson( + Map response, { + required String userId, + required String nextBatch, +}) { + final join = >{}; + final invite = {}; + final knock = {}; + final leave = {}; + + Map joined(String roomId) => + join.putIfAbsent(roomId, () => {}); + + final rooms = response['rooms']; + if (rooms is Map) { + for (final entry in rooms.entries) { + final roomId = entry.key as String; + final room = entry.value; + if (room is! Map) continue; + + final inviteState = room['invite_state']; + if (inviteState is List) { + invite[roomId] = { + 'invite_state': {'events': inviteState}, + }; + continue; + } + final knockState = room['knock_state']; + if (knockState is List) { + knock[roomId] = { + 'knock_state': {'events': knockState}, + }; + continue; + } + + final state = room['required_state'] is List + ? room['required_state'] as List + : const []; + final timeline = room['timeline'] is List + ? room['timeline'] as List + : const []; + final timelineJson = { + 'events': timeline, + 'limited': room['limited'] == true, + if (room['prev_batch'] is String) 'prev_batch': room['prev_batch'], + }; + + final own = _ownMembership([...state, ...timeline], userId); + if (own == 'leave' || own == 'ban') { + leave[roomId] = { + 'state': {'events': state}, + 'timeline': timelineJson, + }; + continue; + } + + final heroes = room['heroes'] is List + ? [ + for (final h in room['heroes'] as List) + if (h is Map && h['user_id'] is String) h['user_id'], + ] + : null; + joined(roomId).addAll({ + 'state': {'events': state}, + 'timeline': timelineJson, + 'summary': { + 'm.heroes': ?heroes, + if (room['joined_count'] is int) + 'm.joined_member_count': room['joined_count'], + if (room['invited_count'] is int) + 'm.invited_member_count': room['invited_count'], + }, + 'unread_notifications': { + 'notification_count': room['notification_count'] is int + ? room['notification_count'] + : 0, + 'highlight_count': room['highlight_count'] is int + ? room['highlight_count'] + : 0, + }, + }); + } + } + + final out = {'next_batch': nextBatch}; + final extensions = response['extensions']; + if (extensions is Map) { + // Typing and receipts are per room ephemeral events. + for (final name in ['typing', 'receipts']) { + final ext = extensions[name]; + final byRoom = ext is Map ? ext['rooms'] : null; + if (byRoom is! Map) continue; + for (final entry in byRoom.entries) { + final roomId = entry.key as String; + if (!join.containsKey(roomId) && + rooms is Map && + rooms[roomId] != null) { + // Sent as an invite, knock or leave: no ephemeral events there. + continue; + } + final ephemeral = + (joined(roomId)['ephemeral'] ??= {'events': []}) + as Map; + (ephemeral['events'] as List).add(entry.value); + } + } + + final accountData = extensions['account_data']; + if (accountData is Map) { + if (accountData['global'] is List) { + out['account_data'] = {'events': accountData['global']}; + } + final byRoom = accountData['rooms']; + if (byRoom is Map) { + for (final entry in byRoom.entries) { + final roomId = entry.key as String; + if (leave.containsKey(roomId)) { + (leave[roomId] as Map)['account_data'] = { + 'events': entry.value, + }; + } else if (!invite.containsKey(roomId) && + !knock.containsKey(roomId)) { + joined(roomId)['account_data'] = {'events': entry.value}; + } + } + } + } + + final toDevice = extensions['to_device']; + if (toDevice is Map && toDevice['events'] is List) { + out['to_device'] = {'events': toDevice['events']}; + } + + final e2ee = extensions['e2ee']; + if (e2ee is Map) { + if (e2ee['device_lists'] is Map) { + out['device_lists'] = e2ee['device_lists']; + } + if (e2ee['device_one_time_keys_count'] is Map) { + out['device_one_time_keys_count'] = e2ee['device_one_time_keys_count']; + } + if (e2ee['device_unused_fallback_key_types'] is List) { + out['device_unused_fallback_key_types'] = + e2ee['device_unused_fallback_key_types']; + } + } + } + + if (join.isNotEmpty || + invite.isNotEmpty || + knock.isNotEmpty || + leave.isNotEmpty) { + out['rooms'] = { + if (join.isNotEmpty) 'join': join, + if (invite.isNotEmpty) 'invite': invite, + if (knock.isNotEmpty) 'knock': knock, + if (leave.isNotEmpty) 'leave': leave, + }; + } + return out; +} + +/// The user's membership as of the last of [events] about them. +String? _ownMembership(List events, String userId) { + String? membership; + for (final e in events) { + if (e is Map && + e['type'] == 'm.room.member' && + e['state_key'] == userId && + e['content'] is Map) { + final m = (e['content'] as Map)['membership']; + if (m is String) membership = m; + } + } + return membership; +} diff --git a/apps/mobile/test/sliding_sync_test.dart b/apps/mobile/test/sliding_sync_test.dart new file mode 100644 index 0000000..b23c45a --- /dev/null +++ b/apps/mobile/test/sliding_sync_test.dart @@ -0,0 +1,240 @@ +import 'package:flutter_test/flutter_test.dart'; +import 'package:matrix/matrix.dart'; +import 'package:warp/matrix/sliding_sync.dart'; + +const me = '@alice:example.org'; + +Map member(String userId, String membership, {int ts = 1}) => + { + 'type': 'm.room.member', + 'state_key': userId, + 'sender': userId, + 'event_id': '\$m$ts$membership', + 'origin_server_ts': ts, + 'content': {'membership': membership}, + }; + +Map message(String body, {int ts = 2}) => { + 'type': 'm.room.message', + 'sender': '@bob:example.org', + 'event_id': '\$msg$ts', + 'origin_server_ts': ts, + 'content': {'msgtype': 'm.text', 'body': body}, +}; + +void main() { + group('SlidingToken', () { + test('no token is a new connection', () { + final t = SlidingToken.parse(null); + expect(t.pos, isNull); + expect(t.toDeviceSince, isNull); + }); + + test('a /sync token is a new connection', () { + expect(SlidingToken.parse('s72594_4483_1934').pos, isNull); + }); + + test('round trip', () { + final t = SlidingToken.parse(const SlidingToken('42', 'td7').toString()); + expect(t.pos, '42'); + expect(t.toDeviceSince, 'td7'); + }); + + test('a token without a to-device position', () { + final t = SlidingToken.parse(const SlidingToken('3', null).toString()); + expect(t.pos, '3'); + expect(t.toDeviceSince, isNull); + }); + }); + + group('slidingRequest', () { + test('asks for both lists over the window, with the extensions', () { + final body = slidingRequest(window: 20, toDeviceSince: 'x', connId: 'c'); + final lists = body['lists'] as Map; + expect((lists['rooms'] as Map)['ranges'], [ + [0, 19], + ]); + expect(((lists['dms'] as Map)['filters'] as Map)['is_dm'], true); + final ext = body['extensions'] as Map; + expect((ext['to_device'] as Map)['since'], 'x'); + expect((ext['e2ee'] as Map)['enabled'], true); + }); + }); + + group('slidingToSyncJson', () { + test('a joined room becomes a join with state, timeline and counts', () { + final json = slidingToSyncJson( + { + 'pos': '1', + 'rooms': { + '!a:x': { + 'required_state': [member(me, 'join')], + 'timeline': [message('hi')], + 'prev_batch': 'p1', + 'limited': true, + 'joined_count': 2, + 'notification_count': 3, + 'highlight_count': 1, + 'heroes': [ + {'user_id': '@bob:example.org', 'displayname': 'Bob'}, + ], + }, + }, + }, + userId: me, + nextBatch: 'nb', + ); + final sync = SyncUpdate.fromJson(json); + expect(sync.nextBatch, 'nb'); + final room = sync.rooms!.join!['!a:x']!; + expect(room.state!.single.type, 'm.room.member'); + expect(room.timeline!.events!.single.content['body'], 'hi'); + expect(room.timeline!.limited, true); + expect(room.timeline!.prevBatch, 'p1'); + expect(room.unreadNotifications!.notificationCount, 3); + expect(room.unreadNotifications!.highlightCount, 1); + expect(room.summary!.mJoinedMemberCount, 2); + expect(room.summary!.mHeroes, ['@bob:example.org']); + }); + + test('invite and knock state', () { + final json = slidingToSyncJson( + { + 'rooms': { + '!i:x': { + 'invite_state': [ + { + 'type': 'm.room.member', + 'state_key': me, + 'sender': '@bob:example.org', + 'content': {'membership': 'invite'}, + }, + ], + }, + '!k:x': {'knock_state': []}, + }, + }, + userId: me, + nextBatch: 'nb', + ); + final sync = SyncUpdate.fromJson(json); + expect(sync.rooms!.invite!['!i:x']!.inviteState!.single.stateKey, me); + expect(sync.rooms!.knock!.containsKey('!k:x'), true); + expect(sync.rooms!.join, isNull); + }); + + test('a room the user left is a leave', () { + final json = slidingToSyncJson( + { + 'rooms': { + '!l:x': { + 'required_state': [member(me, 'join', ts: 1)], + 'timeline': [member(me, 'leave', ts: 5)], + }, + }, + }, + userId: me, + nextBatch: 'nb', + ); + final sync = SyncUpdate.fromJson(json); + expect(sync.rooms!.leave!.containsKey('!l:x'), true); + expect(sync.rooms!.join, isNull); + }); + + test('a rejoin after a leave is a join', () { + final json = slidingToSyncJson( + { + 'rooms': { + '!r:x': { + 'timeline': [ + member(me, 'leave', ts: 1), + member(me, 'join', ts: 2), + ], + }, + }, + }, + userId: me, + nextBatch: 'nb', + ); + expect(SyncUpdate.fromJson(json).rooms!.join!.containsKey('!r:x'), true); + }); + + test('extensions: to-device, keys, account data, typing, receipts', () { + final json = slidingToSyncJson( + { + 'rooms': { + '!a:x': {'timeline': [message('hi')]}, + }, + 'extensions': { + 'to_device': { + 'next_batch': 't2', + 'events': [ + { + 'type': 'm.room_key_request', + 'sender': '@bob:example.org', + 'content': {}, + }, + ], + }, + 'e2ee': { + 'device_lists': { + 'changed': ['@bob:example.org'], + 'left': [], + }, + 'device_one_time_keys_count': {'signed_curve25519': 50}, + 'device_unused_fallback_key_types': ['signed_curve25519'], + }, + 'account_data': { + 'global': [ + {'type': 'm.direct', 'content': {}}, + ], + 'rooms': { + '!a:x': [ + {'type': 'm.fully_read', 'content': {'event_id': '\$x'}}, + ], + }, + }, + 'typing': { + 'rooms': { + // A room with no other news still gets its typing list. + '!quiet:x': { + 'type': 'm.typing', + 'content': { + 'user_ids': ['@bob:example.org'], + }, + }, + }, + }, + 'receipts': { + 'rooms': { + '!a:x': {'type': 'm.receipt', 'content': {}}, + }, + }, + }, + }, + userId: me, + nextBatch: 'nb', + ); + final sync = SyncUpdate.fromJson(json); + expect(sync.toDevice!.single.type, 'm.room_key_request'); + expect(sync.deviceLists!.changed, ['@bob:example.org']); + expect(sync.deviceOneTimeKeysCount!['signed_curve25519'], 50); + expect(sync.deviceUnusedFallbackKeyTypes, ['signed_curve25519']); + expect(sync.accountData!.single.type, 'm.direct'); + final a = sync.rooms!.join!['!a:x']!; + expect(a.accountData!.single.type, 'm.fully_read'); + expect(a.ephemeral!.single.type, 'm.receipt'); + final quiet = sync.rooms!.join!['!quiet:x']!; + expect(quiet.ephemeral!.single.type, 'm.typing'); + }); + + test('an empty response is just a token', () { + final json = slidingToSyncJson( + {'pos': '9', 'rooms': {}}, + userId: me, + nextBatch: 'nb', + ); + expect(json, {'next_batch': 'nb'}); + }); + }); +}