// @dart=2.9 /* * Famedly Matrix SDK * Copyright (C) 2019, 2020, 2021 Famedly GmbH * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero General Public License as * published by the Free Software Foundation, either version 3 of the * License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ import 'dart:async'; import '../matrix.dart'; import 'event.dart'; import 'room.dart'; import 'utils/event_update.dart'; /// Represents the timeline of a room. The callback [onUpdate] will be triggered /// automatically. The initial /// event list will be retreived when created by the `room.getTimeline()` method. class Timeline { final Room room; final List events; /// Map of event ID to map of type to set of aggregated events final Map>> aggregatedEvents = {}; final void Function() onUpdate; final void Function(int insertID) onInsert; StreamSubscription sub; StreamSubscription roomSub; StreamSubscription sessionIdReceivedSub; bool isRequestingHistory = false; final Map _eventCache = {}; /// Searches for the event in this timeline. If not /// found, requests from the server. Requested events /// are cached. Future getEventById(String id) async { for (final event in events) { if (event.eventId == id) return event; } if (_eventCache.containsKey(id)) return _eventCache[id]; final requestedEvent = await room.getEventById(id); if (requestedEvent == null) return null; _eventCache[id] = requestedEvent; return _eventCache[id]; } // When fetching history, we will collect them into the `_historyUpdates` set // first, and then only process all events at once, once we have the full history. // This ensures that the entire history fetching only triggers `onUpdate` only *once*, // even if /sync's complete while history is being proccessed. bool _collectHistoryUpdates = false; bool get canRequestHistory { if (events.isEmpty) return true; return events.last.type != EventTypes.RoomCreate; } Future requestHistory( {int historyCount = Room.defaultHistoryCount}) async { if (isRequestingHistory) { return; } isRequestingHistory = true; onUpdate?.call(); try { // Look up for events in hive first final eventsFromStore = await room.client.database?.getEventList( room.client.id, room, start: events.length, limit: Room.defaultHistoryCount, ); if (eventsFromStore.isNotEmpty) { events.addAll(eventsFromStore); } else { Logs().v('No more events found in the store. Request from server...'); await room.requestHistory( historyCount: historyCount, onHistoryReceived: () { _collectHistoryUpdates = true; }, ); } } finally { _collectHistoryUpdates = false; isRequestingHistory = false; onUpdate?.call(); } } Timeline({this.room, List events, this.onUpdate, this.onInsert}) : events = events ?? [] { sub ??= room.client.onEvent.stream.listen(_handleEventUpdate); // If the timeline is limited we want to clear our events cache roomSub ??= room.client.onSync.stream .where((sync) => sync.rooms?.join != null && sync.rooms.join.containsKey(room.id) && sync.rooms.join[room.id]?.timeline?.limited == true) .listen((_) { events.clear(); aggregatedEvents.clear(); }); sessionIdReceivedSub ??= room.onSessionKeyReceived.stream.listen(_sessionKeyReceived); // we want to populate our aggregated events for (final e in events) { addAggregatedEvent(e); } } /// Don't forget to call this before you dismiss this object! void cancelSubscriptions() { sub?.cancel(); roomSub?.cancel(); sessionIdReceivedSub?.cancel(); } void _sessionKeyReceived(String sessionId) async { var decryptAtLeastOneEvent = false; final decryptFn = () async { if (!room.client.encryptionEnabled) { return; } for (var i = 0; i < events.length; i++) { if (events[i].type == EventTypes.Encrypted && events[i].messageType == MessageTypes.BadEncrypted && events[i].content['session_id'] == sessionId) { events[i] = await room.client.encryption .decryptRoomEvent(room.id, events[i], store: true); if (events[i].type != EventTypes.Encrypted) { decryptAtLeastOneEvent = true; } } } }; if (room.client.database != null) { await room.client.database.transaction(decryptFn); } else { await decryptFn(); } if (decryptAtLeastOneEvent) onUpdate(); } /// Request the keys for undecryptable events of this timeline void requestKeys() { for (final event in events) { if (event.type == EventTypes.Encrypted && event.messageType == MessageTypes.BadEncrypted && event.content['can_request_session'] == true) { try { room.client.encryption.keyManager.maybeAutoRequest(room.id, event.content['session_id'], event.content['sender_key']); } catch (_) { // dispose } } } } int _findEvent({String event_id, String unsigned_txid}) { // we want to find any existing event where either the passed event_id or the passed unsigned_txid // matches either the event_id or transaction_id of the existing event. // For that we create two sets, searchNeedle, what we search, and searchHaystack, where we check if there is a match. // Now, after having these two sets, if the intersect between them is non-empty, we know that we have at least one match in one pair, // thus meaning we found our element. final searchNeedle = {}; if (event_id != null) { searchNeedle.add(event_id); } if (unsigned_txid != null) { searchNeedle.add(unsigned_txid); } int i; for (i = 0; i < events.length; i++) { final searchHaystack = {}; if (events[i].eventId != null) { searchHaystack.add(events[i].eventId); } if (events[i].unsigned != null && events[i].unsigned['transaction_id'] != null) { searchHaystack.add(events[i].unsigned['transaction_id']); } if (searchNeedle.intersection(searchHaystack).isNotEmpty) { break; } } return i; } void _removeEventFromSet(Set eventSet, Event event) { eventSet.removeWhere((e) => e.matchesEventOrTransactionId(event.eventId) || (event.unsigned != null && e.matchesEventOrTransactionId(event.unsigned['transaction_id']))); } void addAggregatedEvent(Event event) { // we want to add an event to the aggregation tree if (event.relationshipType == null || event.relationshipEventId == null) { return; // nothing to do } if (!aggregatedEvents.containsKey(event.relationshipEventId)) { aggregatedEvents[event.relationshipEventId] = >{}; } if (!aggregatedEvents[event.relationshipEventId] .containsKey(event.relationshipType)) { aggregatedEvents[event.relationshipEventId] [event.relationshipType] = {}; } // remove a potential old event _removeEventFromSet( aggregatedEvents[event.relationshipEventId][event.relationshipType], event); // add the new one aggregatedEvents[event.relationshipEventId][event.relationshipType] .add(event); } void removeAggregatedEvent(Event event) { aggregatedEvents.remove(event.eventId); if (event.unsigned != null) { aggregatedEvents.remove(event.unsigned['transaction_id']); } for (final types in aggregatedEvents.values) { for (final events in types.values) { _removeEventFromSet(events, event); } } } void _handleEventUpdate(EventUpdate eventUpdate, {bool update = true}) { try { if (eventUpdate.roomID != room.id) return; if (eventUpdate.type != EventUpdateType.timeline && eventUpdate.type != EventUpdateType.history) { return; } final status = eventUpdate.content['status'] ?? (eventUpdate.content['unsigned'] is Map ? eventUpdate.content['unsigned'][messageSendingStatusKey] : null) ?? 2; // Redaction events are handled as modification for existing events. if (eventUpdate.content['type'] == EventTypes.Redaction) { final eventId = _findEvent(event_id: eventUpdate.content['redacts']); if (eventId < events.length) { removeAggregatedEvent(events[eventId]); events[eventId].setRedactionEvent(Event.fromJson( eventUpdate.content, room, )); } } else if (status == -2) { final i = _findEvent(event_id: eventUpdate.content['event_id']); if (i < events.length) { removeAggregatedEvent(events[i]); events.removeAt(i); } } else { final i = _findEvent( event_id: eventUpdate.content['event_id'], unsigned_txid: eventUpdate.content['unsigned'] is Map ? eventUpdate.content['unsigned']['transaction_id'] : null); if (i < events.length) { // if the old status is larger than the new one, we also want to preserve the old status final oldStatus = events[i].status; events[i] = Event.fromJson( eventUpdate.content, room, ); // do we preserve the status? we should allow 0 -> -1 updates and status increases if (status < oldStatus && !(status == -1 && oldStatus == 0)) { events[i].status = oldStatus; } addAggregatedEvent(events[i]); } else { final newEvent = Event.fromJson( eventUpdate.content, room, ); if (eventUpdate.type == EventUpdateType.history && events.indexWhere( (e) => e.eventId == eventUpdate.content['event_id']) != -1) return; if (eventUpdate.type == EventUpdateType.history) { events.add(newEvent); } else if (status == -1) { events.insert(events.firstIndexWhereNotError, newEvent); } else { events.insert(events.firstIndexWhereNotError, newEvent); } addAggregatedEvent(newEvent); if (onInsert != null) onInsert(0); } } if (update && !_collectHistoryUpdates) { if (onUpdate != null) onUpdate(); } } catch (e, s) { Logs().w('Handle event update failed', e, s); } } } extension on List { int get firstIndexWhereNotError { if (isEmpty) return 0; final index = indexWhere((e) => e.status != -1); if (index == -1) return length; return index; } }