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; }