Merge branch 'henri/fragmented-timeline' into 'main'

feat: non cached fragmented timeline

Closes #278

See merge request famedly/company/frontend/famedlysdk!1009

lib/src/client.dart

@@ -1384,7 +1384,7 @@ class Client extends MatrixApi {
         await roomsLoading;
         await _accountDataLoading;
         _currentTransaction = database.transaction(() async {
-          await _handleSync(syncResp);
+          await _handleSync(syncResp, direction: Direction.f);
           if (prevBatch != syncResp.nextBatch) {
             await database.storePrevBatch(syncResp.nextBatch);
           }
@@ -1396,7 +1396,7 @@ class Client extends MatrixApi {
         );
         onSyncStatus.add(SyncStatusUpdate(SyncStatus.cleaningUp));
       } else {
-        await _handleSync(syncResp);
+        await _handleSync(syncResp, direction: Direction.f);
       }
       if (_disposed || _aborted) return;
       if (prevBatch == null) {
@@ -1440,13 +1440,13 @@ class Client extends MatrixApi {
   }
 
   /// Use this method only for testing utilities!
-  Future<void> handleSync(SyncUpdate sync, {bool sortAtTheEnd = false}) async {
+  Future<void> handleSync(SyncUpdate sync, {Direction? direction}) async {
     // ensure we don't upload keys because someone forgot to set a key count
     sync.deviceOneTimeKeysCount ??= {'signed_curve25519': 100};
-    await _handleSync(sync, sortAtTheEnd: sortAtTheEnd);
+    await _handleSync(sync, direction: direction);
   }
 
-  Future<void> _handleSync(SyncUpdate sync, {bool sortAtTheEnd = false}) async {
+  Future<void> _handleSync(SyncUpdate sync, {Direction? direction}) async {
     final syncToDevice = sync.toDevice;
     if (syncToDevice != null) {
       await _handleToDeviceEvents(syncToDevice);
@@ -1455,15 +1455,15 @@ class Client extends MatrixApi {
     if (sync.rooms != null) {
       final join = sync.rooms?.join;
       if (join != null) {
-        await _handleRooms(join, sortAtTheEnd: sortAtTheEnd);
+        await _handleRooms(join, direction: direction);
       }
       final invite = sync.rooms?.invite;
       if (invite != null) {
-        await _handleRooms(invite, sortAtTheEnd: sortAtTheEnd);
+        await _handleRooms(invite, direction: direction);
       }
       final leave = sync.rooms?.leave;
       if (leave != null) {
-        await _handleRooms(leave, sortAtTheEnd: sortAtTheEnd);
+        await _handleRooms(leave, direction: direction);
       }
     }
     for (final newPresence in sync.presence ?? []) {
@@ -1524,7 +1524,7 @@ class Client extends MatrixApi {
   }
 
   Future<void> _handleRooms(Map<String, SyncRoomUpdate> rooms,
-      {bool sortAtTheEnd = false}) async {
+      {Direction? direction}) async {
     var handledRooms = 0;
     for (final entry in rooms.entries) {
       onSyncStatus.add(SyncStatusUpdate(
@@ -1540,11 +1540,14 @@ class Client extends MatrixApi {
       /// Handle now all room events and save them in the database
       if (room is JoinedRoomUpdate) {
         final state = room.state;
+
         if (state != null && state.isNotEmpty) {
           // TODO: This method seems to be comperatively slow for some updates
           await _handleRoomEvents(
-              id, state.map((i) => i.toJson()).toList(), EventUpdateType.state,
+            id,
-              sortAtTheEnd: sortAtTheEnd);
+            state.map((i) => i.toJson()).toList(),
+            EventUpdateType.state,
+          );
         }
 
         final timelineEvents = room.timeline?.events;
@@ -1552,8 +1555,11 @@ class Client extends MatrixApi {
           await _handleRoomEvents(
               id,
               timelineEvents.map((i) => i.toJson()).toList(),
-              sortAtTheEnd ? EventUpdateType.history : EventUpdateType.timeline,
+              direction != null
-              sortAtTheEnd: sortAtTheEnd);
+                  ? (direction == Direction.b
+                      ? EventUpdateType.history
+                      : EventUpdateType.timeline)
+                  : EventUpdateType.timeline);
         }
 
         final ephemeral = room.ephemeral;
@@ -1576,10 +1582,10 @@ class Client extends MatrixApi {
         final timelineEvents = room.timeline?.events;
         if (timelineEvents != null && timelineEvents.isNotEmpty) {
           await _handleRoomEvents(
-              id,
+            id,
-              timelineEvents.map((i) => i.toJson()).toList(),
+            timelineEvents.map((i) => i.toJson()).toList(),
-              EventUpdateType.timeline,
+            EventUpdateType.timeline,
-              sortAtTheEnd: sortAtTheEnd);
+          );
         }
         final accountData = room.accountData;
         if (accountData != null && accountData.isNotEmpty) {
@@ -1650,16 +1656,14 @@ class Client extends MatrixApi {
   }
 
   Future<void> _handleRoomEvents(
-      String chat_id, List<dynamic> events, EventUpdateType type,
+      String chat_id, List<dynamic> events, EventUpdateType type) async {
-      {bool sortAtTheEnd = false}) async {
     for (final event in events) {
-      await _handleEvent(event, chat_id, type, sortAtTheEnd: sortAtTheEnd);
+      await _handleEvent(event, chat_id, type);
     }
   }
 
   Future<void> _handleEvent(
-      Map<String, dynamic> event, String roomID, EventUpdateType type,
+      Map<String, dynamic> event, String roomID, EventUpdateType type) async {
-      {bool sortAtTheEnd = false}) async {
     if (event['type'] is String && event['content'] is Map<String, dynamic>) {
       // The client must ignore any new m.room.encryption event to prevent
       // man-in-the-middle attacks!
@@ -1672,11 +1676,7 @@ class Client extends MatrixApi {
         return;
       }
 
-      var update = EventUpdate(
+      var update = EventUpdate(roomID: roomID, type: type, content: event);
-        roomID: roomID,
-        type: type,
-        content: event,
-      );
       if (event['type'] == EventTypes.Encrypted && encryptionEnabled) {
         update = await update.decrypt(room);
       }

lib/src/database/database_api.dart

@@ -83,6 +83,7 @@ abstract class DatabaseApi {
   Future<List<Event>> getEventList(
     Room room, {
     int start = 0,
+    bool onlySending,
     int limit,
   });
 

lib/src/database/fluffybox_database.dart

@@ -345,6 +345,7 @@ class FluffyBoxDatabase extends DatabaseApi {
   Future<List<Event>> getEventList(
     Room room, {
     int start = 0,
+    bool onlySending = false,
     int? limit,
   }) =>
       runBenchmarked<List<Event>>('Get event list', () async {
@@ -366,10 +367,11 @@ class FluffyBoxDatabase extends DatabaseApi {
         // Combine those two lists while respecting the start and limit parameters.
         final end = min(timelineEventIds.length,
             start + (limit ?? timelineEventIds.length));
-        final eventIds = sendingEventIds +
+
-            (start < timelineEventIds.length
+        final eventIds = sendingEventIds;
-                ? timelineEventIds.getRange(start, end).toList()
+        if (start < timelineEventIds.length && !onlySending) {
-                : []);
+          eventIds.addAll(timelineEventIds.getRange(start, end).toList());
+        }
 
         return await _getEventsByIds(eventIds.cast<String>(), room);
       });

lib/src/database/hive_database.dart

@@ -387,6 +387,7 @@ class FamedlySdkHiveDatabase extends DatabaseApi {
   Future<List<Event>> getEventList(
     Room room, {
     int start = 0,
+    bool onlySending = false,
     int? limit,
   }) =>
       runBenchmarked<List<Event>>('Get event list', () async {
@@ -410,7 +411,7 @@ class FamedlySdkHiveDatabase extends DatabaseApi {
         final end = min(timelineEventIds.length,
             start + (limit ?? timelineEventIds.length));
         final eventIds = sendingEventIds +
-            (start < timelineEventIds.length
+            (start < timelineEventIds.length && !onlySending
                 ? timelineEventIds.getRange(start, end).toList()
                 : []);
 

lib/src/models/timeline_chunk.dart

@@ -0,0 +1,10 @@
+import '../../matrix.dart';
+
+class TimelineChunk {
+  String prevBatch; // pos of the first event of the database timeline chunk
+  String nextBatch;
+
+  List<Event> events;
+  TimelineChunk(
+      {required this.events, this.prevBatch = '', this.nextBatch = ''});
+}

lib/src/room.dart

@@ -22,6 +22,7 @@ import 'dart:typed_data';
 
 import 'package:collection/collection.dart';
 import 'package:html_unescape/html_unescape.dart';
+import 'package:matrix/src/models/timeline_chunk.dart';
 import 'package:matrix/src/utils/crypto/crypto.dart';
 import 'package:matrix/src/utils/file_send_request_credentials.dart';
 import 'package:matrix/src/utils/space_child.dart';
@@ -1080,7 +1081,8 @@ class Room {
   /// Returns the actual count of received timeline events.
   Future<int> requestHistory(
       {int historyCount = defaultHistoryCount,
-      void Function()? onHistoryReceived}) async {
+      void Function()? onHistoryReceived,
+      direction = Direction.b}) async {
     final prev_batch = this.prev_batch;
     if (prev_batch == null) {
       throw 'Tried to request history without a prev_batch token';
@@ -1088,7 +1090,7 @@ class Room {
     final resp = await client.getRoomEvents(
       id,
       prev_batch,
-      Direction.b,
+      direction,
       limit: historyCount,
       filter: jsonEncode(StateFilter(lazyLoadMembers: true).toJson()),
     );
@@ -1109,8 +1111,12 @@ class Room {
                           state: resp.state,
                           timeline: TimelineUpdate(
                             limited: false,
-                            events: resp.chunk,
+                            events: direction == Direction.b
-                            prevBatch: resp.end,
+                                ? resp.chunk
+                                : resp.chunk?.reversed.toList(),
+                            prevBatch: direction == Direction.b
+                                ? resp.end
+                                : resp.start,
                           ),
                         )
                       }
@@ -1122,13 +1128,15 @@ class Room {
                           timeline: TimelineUpdate(
                             limited: false,
                             events: resp.chunk,
-                            prevBatch: resp.end,
+                            prevBatch: direction == Direction.b
+                                ? resp.end
+                                : resp.start,
                           ),
                         ),
                       }
                     : null),
           ),
-          sortAtTheEnd: true);
+          direction: Direction.b);
     };
 
     if (client.database != null) {
@@ -1207,6 +1215,34 @@ class Room {
     return;
   }
 
+  Future<TimelineChunk?> getEventContext(String eventId) async {
+    final resp = await client.getEventContext(id, eventId,
+        limit: Room.defaultHistoryCount
+        // filter: jsonEncode(StateFilter(lazyLoadMembers: true).toJson()),
+        );
+
+    final events = [
+      if (resp.eventsAfter != null) ...resp.eventsAfter!.reversed.toList(),
+      if (resp.event != null) resp.event!,
+      if (resp.eventsBefore != null) ...resp.eventsBefore!
+    ].map((e) => Event.fromMatrixEvent(e, this)).toList();
+
+    // Try again to decrypt encrypted events but don't update the database.
+    if (encrypted && client.database != null && client.encryptionEnabled) {
+      for (var i = 0; i < events.length; i++) {
+        if (events[i].type == EventTypes.Encrypted &&
+            events[i].content['can_request_session'] == true) {
+          events[i] = await client.encryption!.decryptRoomEvent(id, events[i]);
+        }
+      }
+    }
+
+    final chunk = TimelineChunk(
+        nextBatch: resp.end ?? '', prevBatch: resp.start ?? '', events: events);
+
+    return chunk;
+  }
+
   /// This API updates the marker for the given receipt type to the event ID
   /// specified.
   Future<void> postReceipt(String eventId) async {
@@ -1225,47 +1261,80 @@ class Room {
   /// just want to update the whole timeline on every change, use the [onUpdate]
   /// callback. For updating only the parts that have changed, use the
   /// [onChange], [onRemove], [onInsert] and the [onHistoryReceived] callbacks.
-  Future<Timeline> getTimeline({
+  /// This method can also retrieve the timeline at a specific point by setting
-    void Function(int index)? onChange,
+  /// the [eventContextId]
-    void Function(int index)? onRemove,
+  Future<Timeline> getTimeline(
-    void Function(int insertID)? onInsert,
+      {void Function(int index)? onChange,
-    void Function()? onUpdate,
+      void Function(int index)? onRemove,
-  }) async {
+      void Function(int insertID)? onInsert,
+      void Function()? onNewEvent,
+      void Function()? onUpdate,
+      String? eventContextId}) async {
     await postLoad();
-    final events = await client.database?.getEventList(
-          this,
-          limit: defaultHistoryCount,
-        ) ??
-        <Event>[];
 
-    // Try again to decrypt encrypted events and update the database.
+    final _events = await client.database?.getEventList(
-    if (encrypted && client.database != null && client.encryptionEnabled) {
+      this,
-      await client.database?.transaction(() async {
+      limit: defaultHistoryCount,
-        for (var i = 0; i < events.length; i++) {
+    );
-          if (events[i].type == EventTypes.Encrypted &&
+
-              events[i].content['can_request_session'] == true) {
+    var chunk = TimelineChunk(events: _events ?? []);
-            events[i] = await client.encryption!
+
-                .decryptRoomEvent(id, events[i], store: true);
+    if (_events != null) {
-          }
+      if (eventContextId != null) {
+        if (_events
+                .firstWhereOrNull((event) => event.eventId == eventContextId) !=
+            null) {
+          chunk = TimelineChunk(events: _events);
+        } else {
+          chunk = await getEventContext(eventContextId) ??
+              TimelineChunk(events: []);
         }
-      });
+      }
+
+      // Fetch all users from database we have got here.
+      if (eventContextId != null) {
+        for (final event in _events) {
+          if (getState(EventTypes.RoomMember, event.senderId) != null) continue;
+          final dbUser = await client.database?.getUser(event.senderId, this);
+          if (dbUser != null) setState(dbUser);
+        }
+      }
     }
 
-    // Fetch all users from database we have got here.
+    if (encrypted && client.encryptionEnabled) {
-    for (final event in events) {
+      // decrypt messages
-      if (getState(EventTypes.RoomMember, event.senderId) != null) continue;
+      for (var i = 0; i < chunk.events.length; i++) {
-      final dbUser = await client.database?.getUser(event.senderId, this);
+        if (chunk.events[i].type == EventTypes.Encrypted) {
-      if (dbUser != null) setState(dbUser);
+          if (eventContextId != null) {
+            // for the fragmented timeline, we don't cache the decrypted
+            //message in the database
+            chunk.events[i] = await client.encryption!.decryptRoomEvent(
+              id,
+              chunk.events[i],
+            );
+          } else if (client.database != null) {
+            // else, we need the database
+            await client.database?.transaction(() async {
+              for (var i = 0; i < chunk.events.length; i++) {
+                if (chunk.events[i].content['can_request_session'] == true) {
+                  chunk.events[i] = await client.encryption!
+                      .decryptRoomEvent(id, chunk.events[i], store: true);
+                }
+              }
+            });
+          }
+        }
+      }
     }
 
     final timeline = Timeline(
-      room: this,
+        room: this,
-      events: events,
+        chunk: chunk,
-      onChange: onChange,
+        onChange: onChange,
-      onRemove: onRemove,
+        onRemove: onRemove,
-      onInsert: onInsert,
+        onInsert: onInsert,
-      onUpdate: onUpdate,
+        onNewEvent: onNewEvent,
-    );
+        onUpdate: onUpdate);
     return timeline;
   }
 
@@ -1793,13 +1862,13 @@ class Room {
   }
 
   Future<void> _handleFakeSync(SyncUpdate syncUpdate,
-      {bool sortAtTheEnd = false}) async {
+      {Direction? direction}) async {
     if (client.database != null) {
       await client.database?.transaction(() async {
-        await client.handleSync(syncUpdate, sortAtTheEnd: sortAtTheEnd);
+        await client.handleSync(syncUpdate, direction: direction);
       });
     } else {
-      await client.handleSync(syncUpdate, sortAtTheEnd: sortAtTheEnd);
+      await client.handleSync(syncUpdate, direction: direction);
     }
   }
 

lib/src/timeline.dart

@@ -17,17 +17,20 @@
  */
 
 import 'dart:async';
+import 'dart:convert';
 
 import 'package:collection/src/iterable_extensions.dart';
 
 import '../matrix.dart';
+import 'models/timeline_chunk.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<Event> events;
+  List<Event> get events => chunk.events;
 
   /// Map of event ID to map of type to set of aggregated events
   final Map<String, Map<String, Set<Event>>> aggregatedEvents = {};
@@ -36,14 +39,21 @@ class Timeline {
   final void Function(int index)? onChange;
   final void Function(int index)? onInsert;
   final void Function(int index)? onRemove;
+  final void Function()? onNewEvent;
 
   StreamSubscription<EventUpdate>? sub;
   StreamSubscription<SyncUpdate>? roomSub;
   StreamSubscription<String>? sessionIdReceivedSub;
   bool isRequestingHistory = false;
+  bool isRequestingFuture = false;
+
+  bool allowNewEvent = true;
+  bool isFragmentedTimeline = false;
 
   final Map<String, Event> _eventCache = {};
 
+  TimelineChunk chunk;
+
   /// Searches for the event in this timeline. If not
   /// found, requests from the server. Requested events
   /// are cached.
@@ -74,16 +84,41 @@ class Timeline {
     if (isRequestingHistory) {
       return;
     }
+
     isRequestingHistory = true;
+    await _requestEvents(direction: Direction.b, historyCount: historyCount);
+    isRequestingHistory = false;
+  }
+
+  bool get canRequestFuture => !allowNewEvent;
+
+  Future<void> requestFuture(
+      {int historyCount = Room.defaultHistoryCount}) async {
+    if (allowNewEvent) {
+      return; // we shouldn't force to add new events if they will autatically be added
+    }
+
+    if (isRequestingFuture) return;
+    isRequestingFuture = true;
+    await _requestEvents(direction: Direction.f, historyCount: historyCount);
+    isRequestingFuture = false;
+  }
+
+  Future<void> _requestEvents(
+      {int historyCount = Room.defaultHistoryCount,
+      required Direction direction}) async {
     onUpdate?.call();
 
     try {
-      // Look up for events in hive first
+      // Look up for events in the database first. With fragmented view, we should delete the database cache
-      final eventsFromStore = await room.client.database?.getEventList(
+      final eventsFromStore = isFragmentedTimeline
-        room,
+          ? null
-        start: events.length,
+          : await room.client.database?.getEventList(
-        limit: Room.defaultHistoryCount,
+              room,
-      );
+              start: events.length,
+              limit: Room.defaultHistoryCount,
+            );
+
       if (eventsFromStore != null && eventsFromStore.isNotEmpty) {
         // Fetch all users from database we have got here.
         for (final event in events) {
@@ -95,20 +130,37 @@ class Timeline {
           if (dbUser != null) room.setState(dbUser);
         }
 
-        events.addAll(eventsFromStore);
+        if (direction == Direction.b) {
-        final startIndex = events.length - eventsFromStore.length;
+          events.addAll(eventsFromStore);
-        final endIndex = events.length;
+          final startIndex = events.length - eventsFromStore.length;
-        for (var i = startIndex; i < endIndex; i++) {
+          final endIndex = events.length;
-          onInsert?.call(i);
+          for (var i = startIndex; i < endIndex; i++) {
+            onInsert?.call(i);
+          }
+        } else {
+          events.insertAll(0, eventsFromStore);
+          final startIndex = eventsFromStore.length;
+          final endIndex = 0;
+          for (var i = startIndex; i > endIndex; i--) {
+            onInsert?.call(i);
+          }
         }
       } else {
-        Logs().v('No more events found in the store. Request from server...');
+        Logs().i('No more events found in the store. Request from server...');
-        await room.requestHistory(
+        if (isFragmentedTimeline) {
-          historyCount: historyCount,
+          await getRoomEvents(
-          onHistoryReceived: () {
+            historyCount: historyCount,
-            _collectHistoryUpdates = true;
+            direction: direction,
-          },
+          );
-        );
+        } else {
+          await room.requestHistory(
+            historyCount: historyCount,
+            direction: direction,
+            onHistoryReceived: () {
+              _collectHistoryUpdates = true;
+            },
+          );
+        }
       }
     } finally {
       _collectHistoryUpdates = false;
@@ -117,14 +169,103 @@ class Timeline {
     }
   }
 
-  Timeline({
+  /// Request more previous events from the server. [historyCount] defines how much events should
-    required this.room,
+  /// be received maximum. When the request is answered, [onHistoryReceived] will be triggered **before**
-    List<Event>? events,
+  /// the historical events will be published in the onEvent stream.
-    this.onUpdate,
+  /// Returns the actual count of received timeline events.
-    this.onChange,
+  Future<int> getRoomEvents(
-    this.onInsert,
+      {int historyCount = Room.defaultHistoryCount,
-    this.onRemove,
+      direction = Direction.b}) async {
-  }) : events = events ?? [] {
+    final resp = await room.client.getRoomEvents(
+      room.id,
+      direction == Direction.b ? chunk.prevBatch : chunk.nextBatch,
+      direction,
+      limit: historyCount,
+      filter: jsonEncode(StateFilter(lazyLoadMembers: true).toJson()),
+    );
+
+    if (resp.end == null || resp.start == null) {
+      Logs().w('end or start parameters where not set in the response');
+    }
+
+    final newNextBatch = direction == Direction.b ? resp.start : resp.end;
+    final newPrevBatch = direction == Direction.b ? resp.end : resp.start;
+
+    final type = direction == Direction.b
+        ? EventUpdateType.history
+        : EventUpdateType.timeline;
+
+    if ((resp.state?.length ?? 0) == 0 && resp.start != resp.end) {
+      if (type == EventUpdateType.history) {
+        Logs().w(
+            '[nav] we can still request history prevBatch: $type $newPrevBatch');
+      } else {
+        Logs().w(
+            '[nav] we can still request history nextBatch: $type $newNextBatch');
+      }
+    }
+
+    final newEvents =
+        resp.chunk?.map((e) => Event.fromMatrixEvent(e, room)).toList() ?? [];
+
+    if (!allowNewEvent) {
+      if (resp.start == resp.end) allowNewEvent = true;
+
+      if (allowNewEvent) {
+        Logs().d('We now allow sync update into the timeline.');
+        allowNewEvent = true;
+        newEvents.addAll(
+            await room.client.database?.getEventList(room, onlySending: true) ??
+                []);
+      }
+    }
+
+    // Try to decrypt encrypted events but don't update the database.
+    if (room.encrypted &&
+        room.client.database != null &&
+        room.client.encryptionEnabled) {
+      for (var i = 0; i < newEvents.length; i++) {
+        if (newEvents[i].type == EventTypes.Encrypted) {
+          newEvents[i] = await room.client.encryption!
+              .decryptRoomEvent(room.id, newEvents[i]);
+        }
+      }
+    }
+
+    // update chunk anchors
+    if (type == EventUpdateType.history) {
+      chunk.prevBatch = newPrevBatch ?? '';
+
+      final offset = chunk.events.length;
+
+      chunk.events.addAll(newEvents);
+
+      for (var i = 0; i < newEvents.length; i++) {
+        onInsert?.call(i + offset);
+      }
+    } else {
+      chunk.nextBatch = newNextBatch ?? '';
+      chunk.events.insertAll(0, newEvents.reversed);
+
+      for (var i = 0; i < newEvents.length; i++) {
+        onInsert?.call(i);
+      }
+    }
+
+    if (onUpdate != null) {
+      onUpdate!();
+    }
+    return resp.chunk?.length ?? 0;
+  }
+
+  Timeline(
+      {required this.room,
+      this.onUpdate,
+      this.onChange,
+      this.onInsert,
+      this.onRemove,
+      this.onNewEvent,
+      required this.chunk}) {
     sub = room.client.onEvent.stream.listen(_handleEventUpdate);
 
     // If the timeline is limited we want to clear our events cache
@@ -136,9 +277,15 @@ class Timeline {
         room.onSessionKeyReceived.stream.listen(_sessionKeyReceived);
 
     // we want to populate our aggregated events
-    for (final e in this.events) {
+    for (final e in events) {
       addAggregatedEvent(e);
     }
+
+    // we are using a fragmented timeline
+    if (chunk.nextBatch != '') {
+      allowNewEvent = false;
+      isFragmentedTimeline = true;
+    }
   }
 
   /// Removes all entries from [events] which are not in this SyncUpdate.
@@ -281,6 +428,13 @@ class Timeline {
           eventUpdate.type != EventUpdateType.history) {
         return;
       }
+
+      if (eventUpdate.type == EventUpdateType.timeline) {
+        onNewEvent?.call();
+      }
+
+      if (!allowNewEvent) return;
+
       final status = eventStatusFromInt(eventUpdate.content['status'] ??
           (eventUpdate.content['unsigned'] is Map<String, dynamic>
               ? eventUpdate.content['unsigned'][messageSendingStatusKey]

lib/src/utils/event_update.dart

@@ -39,11 +39,8 @@ class EventUpdate {
   // The json payload of the content of this event.
   final Map<String, dynamic> content;
 
-  EventUpdate({
+  EventUpdate(
-    required this.roomID,
+      {required this.roomID, required this.type, required this.content});
-    required this.type,
-    required this.content,
-  });
 
   Future<EventUpdate> decrypt(Room room, {bool store = false}) async {
     final encryption = room.client.encryption;
@@ -57,10 +54,7 @@ class EventUpdate {
           room.id, Event.fromJson(content, room),
           store: store, updateType: type);
       return EventUpdate(
-        roomID: roomID,
+          roomID: roomID, type: type, content: decrpytedEvent.toJson());
-        type: type,
-        content: decrpytedEvent.toJson(),
-      );
     } catch (e, s) {
       Logs().e('[LibOlm] Could not decrypt megolm event', e, s);
       return this;

test/database_api_test.dart

@@ -232,7 +232,8 @@ void testDatabase(
   });
   test('getEventList', () async {
     final events = await database.getEventList(
-        Room(id: '!testroom:example.com', client: Client('testclient')));
+      Room(id: '!testroom:example.com', client: Client('testclient')),
+    );
     expect(events.single.type, EventTypes.Message);
   });
   test('getUser', () async {

test/event_test.dart

@@ -21,6 +21,7 @@ import 'dart:typed_data';
 
 import 'package:matrix/encryption.dart';
 import 'package:matrix/matrix.dart';
+import 'package:matrix/src/models/timeline_chunk.dart';
 import 'package:olm/olm.dart' as olm;
 import 'package:test/test.dart';
 
@@ -1107,8 +1108,9 @@ void main() {
         'sender': '@example:example.org',
         'event_id': '\$edit2',
       }, room);
-      final timeline =
+      final timeline = Timeline(
-          Timeline(events: <Event>[event, edit1, edit2], room: room);
+          chunk: TimelineChunk(events: <Event>[event, edit1, edit2]),
+          room: room);
       expect(event.hasAggregatedEvents(timeline, RelationshipTypes.edit), true);
       expect(event.aggregatedEvents(timeline, RelationshipTypes.edit),
           {edit1, edit2});
@@ -1204,24 +1206,26 @@ void main() {
         'sender': '@bob:example.org',
       }, room);
       // no edits
-      var displayEvent =
+      var displayEvent = event.getDisplayEvent(
-          event.getDisplayEvent(Timeline(events: <Event>[event], room: room));
+          Timeline(chunk: TimelineChunk(events: <Event>[event]), room: room));
       expect(displayEvent.body, 'blah');
       // one edit
-      displayEvent = event
+      displayEvent = event.getDisplayEvent(Timeline(
-          .getDisplayEvent(Timeline(events: <Event>[event, edit1], room: room));
+          chunk: TimelineChunk(events: <Event>[event, edit1]), room: room));
       expect(displayEvent.body, 'edit 1');
       // two edits
-      displayEvent = event.getDisplayEvent(
+      displayEvent = event.getDisplayEvent(Timeline(
-          Timeline(events: <Event>[event, edit1, edit2], room: room));
+          chunk: TimelineChunk(events: <Event>[event, edit1, edit2]),
+          room: room));
       expect(displayEvent.body, 'edit 2');
       // foreign edit
-      displayEvent = event
+      displayEvent = event.getDisplayEvent(Timeline(
-          .getDisplayEvent(Timeline(events: <Event>[event, edit3], room: room));
+          chunk: TimelineChunk(events: <Event>[event, edit3]), room: room));
       expect(displayEvent.body, 'blah');
       // mixed foreign and non-foreign
-      displayEvent = event.getDisplayEvent(
+      displayEvent = event.getDisplayEvent(Timeline(
-          Timeline(events: <Event>[event, edit1, edit2, edit3], room: room));
+          chunk: TimelineChunk(events: <Event>[event, edit1, edit2, edit3]),
+          room: room));
       expect(displayEvent.body, 'edit 2');
       event = Event.fromJson({
         'type': EventTypes.Message,
@@ -1239,8 +1243,9 @@ void main() {
           },
         },
       }, room);
-      displayEvent = event.getDisplayEvent(
+      displayEvent = event.getDisplayEvent(Timeline(
-          Timeline(events: <Event>[event, edit1, edit2, edit3], room: room));
+          chunk: TimelineChunk(events: <Event>[event, edit1, edit2, edit3]),
+          room: room));
       expect(displayEvent.body, 'Redacted');
     });
     test('attachments', () async {

test/fake_matrix_api.dart

@@ -148,7 +148,7 @@ class FakeMatrixApi extends MockClient {
           return Response.bytes(utf8.encode(json.encode(res)), statusCode);
         });
 
-  static Map<String, dynamic> messagesResponse = {
+  static Map<String, dynamic> messagesResponsePast = {
     'start': 't47429-4392820_219380_26003_2265',
     'end': 't47409-4357353_219380_26003_2265',
     'chunk': [
@@ -206,6 +206,70 @@ class FakeMatrixApi extends MockClient {
     ],
     'state': [],
   };
+  static Map<String, dynamic> messagesResponseFuture = {
+    'start': 't456',
+    'end': 't789',
+    'chunk': [
+      {
+        'content': {
+          'body': 'Gangnam Style',
+          'url': 'mxc://example.org/a526eYUSFFxlgbQYZmo442',
+          'info': {
+            'thumbnail_url': 'mxc://example.org/FHyPlCeYUSFFxlgbQYZmoEoe',
+            'thumbnail_info': {
+              'mimetype': 'image/jpeg',
+              'size': 46144,
+              'w': 300,
+              'h': 300
+            },
+            'w': 480,
+            'h': 320,
+            'duration': 2140786,
+            'size': 1563685,
+            'mimetype': 'video/mp4'
+          },
+          'msgtype': 'm.video'
+        },
+        'type': 'm.room.message',
+        'event_id': '1143273582443PhrSn:example.org',
+        'room_id': '!1234:example.com',
+        'sender': '@example:example.org',
+        'origin_server_ts': 1432735824653,
+        'unsigned': {'age': 1234}
+      },
+      {
+        'content': {'name': 'The room name'},
+        'type': 'm.room.name',
+        'event_id': '2143273582443PhrSn:example.org',
+        'room_id': '!1234:example.com',
+        'sender': '@example:example.org',
+        'origin_server_ts': 1432735824653,
+        'unsigned': {'age': 1234},
+        'state_key': ''
+      },
+      {
+        'content': {
+          'body': 'This is an example text message',
+          'msgtype': 'm.text',
+          'format': 'org.matrix.custom.html',
+          'formatted_body': '<b>This is an example text message</b>'
+        },
+        'type': 'm.room.message',
+        'event_id': '3143273582443PhrSn:example.org',
+        'room_id': '!1234:example.com',
+        'sender': '@example:example.org',
+        'origin_server_ts': 1432735824653,
+        'unsigned': {'age': 1234}
+      }
+    ],
+    'state': [],
+  };
+  static Map<String, dynamic> messagesResponseFutureEnd = {
+    'start': 't789',
+    'end': 't789',
+    'chunk': [],
+    'state': [],
+  };
 
   static Map<String, dynamic> syncResponse = {
     'next_batch': Random().nextDouble().toString(),
@@ -1337,11 +1401,19 @@ class FakeMatrixApi extends MockClient {
             'unsigned': {'age': 1234}
           },
       '/client/r0/rooms/!localpart%3Aserver.abc/messages?from=1234&dir=b&to=1234&limit=10&filter=%7B%22lazy_load_members%22%3Atrue%7D':
-          (var req) => messagesResponse,
+          (var req) => messagesResponsePast,
       '/client/r0/rooms/!localpart%3Aserver.abc/messages?from=&dir=b&limit=10&filter=%7B%22lazy_load_members%22%3Atrue%7D':
-          (var req) => messagesResponse,
+          (var req) => messagesResponsePast,
       '/client/r0/rooms/!1234%3Aexample.com/messages?from=1234&dir=b&limit=30&filter=%7B%22lazy_load_members%22%3Atrue%7D':
-          (var req) => messagesResponse,
+          (var req) => messagesResponsePast,
+      '/client/r0/rooms/!localpart%3Aserver.abc/messages?from=t456&dir=f&to=1234&limit=10&filter=%7B%22lazy_load_members%22%3Atrue%7D':
+          (var req) => messagesResponseFuture,
+      '/client/r0/rooms/!1234%3Aexample.com/messages?from=t456&dir=f&limit=30&filter=%7B%22lazy_load_members%22%3Atrue%7D':
+          (var req) => messagesResponseFuture,
+      '/client/r0/rooms/!localpart%3Aserver.abc/messages?from=t789&dir=f&to=1234&limit=10&filter=%7B%22lazy_load_members%22%3Atrue%7D':
+          (var req) => messagesResponseFutureEnd,
+      '/client/r0/rooms/!1234%3Aexample.com/messages?from=t789&dir=f&limit=30&filter=%7B%22lazy_load_members%22%3Atrue%7D':
+          (var req) => messagesResponseFutureEnd,
       '/client/versions': (var req) => {
             'versions': [
               'r0.0.1',

test/timeline_context_test.dart

@@ -0,0 +1,589 @@
+/*
+ *   Famedly Matrix SDK
+ *   Copyright (C) 2019, 2020 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 <https://www.gnu.org/licenses/>.
+ */
+
+import 'package:matrix/matrix.dart';
+import 'package:matrix/src/models/timeline_chunk.dart';
+
+import 'package:test/test.dart';
+import 'package:olm/olm.dart' as olm;
+import 'fake_client.dart';
+
+void main() {
+  group('Timeline context', () {
+    Logs().level = Level.error;
+    final roomID = '!1234:example.com';
+    final testTimeStamp = DateTime.now().millisecondsSinceEpoch;
+    var updateCount = 0;
+    final insertList = <int>[];
+    final changeList = <int>[];
+    final removeList = <int>[];
+    var olmEnabled = true;
+
+    late Client client;
+    late Room room;
+    late Timeline timeline;
+    test('create stuff', () async {
+      try {
+        await olm.init();
+        olm.get_library_version();
+      } catch (e) {
+        olmEnabled = false;
+        Logs().w('[LibOlm] Failed to load LibOlm', e);
+      }
+      Logs().i('[LibOlm] Enabled: $olmEnabled');
+      client = await getClient();
+      client.sendMessageTimeoutSeconds = 5;
+
+      room = Room(
+          id: roomID, client: client, prev_batch: 't123', roomAccountData: {});
+      timeline = Timeline(
+        room: room,
+        chunk: TimelineChunk(events: [], nextBatch: 't456', prevBatch: 't123'),
+        onUpdate: () {
+          updateCount++;
+        },
+        onInsert: insertList.add,
+        onChange: changeList.add,
+        onRemove: removeList.add,
+      );
+
+      expect(timeline.isFragmentedTimeline, true);
+      expect(timeline.allowNewEvent, false);
+    });
+
+    test('Request future', () async {
+      timeline.events.clear();
+      await timeline.requestFuture();
+
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 3);
+      expect(insertList, [0, 1, 2]);
+      expect(timeline.events.length, 3);
+      expect(timeline.events[0].eventId, '3143273582443PhrSn:example.org');
+      expect(timeline.events[1].eventId, '2143273582443PhrSn:example.org');
+      expect(timeline.events[2].eventId, '1143273582443PhrSn:example.org');
+      expect(timeline.chunk.nextBatch, 't789');
+
+      expect(timeline.isFragmentedTimeline, true);
+      expect(timeline.allowNewEvent, false);
+    });
+
+    /// We send a message in a fragmented timeline, it didn't reached the end so we shouldn't be displayed.
+    test('Send message not displayed', () async {
+      updateCount = 0;
+
+      await room.sendTextEvent('test', txid: '1234');
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 0);
+      expect(insertList, [0, 1, 2]);
+      expect(insertList.length,
+          timeline.events.length); // expect no new events to have been added
+
+      final eventId = '1844295642248BcDkn:example.org';
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'test'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.synced.intValue,
+          'event_id': eventId,
+          'unsigned': {'transaction_id': '1234'},
+          'origin_server_ts': DateTime.now().millisecondsSinceEpoch
+        },
+      )); // just assume that it was on the server for this call but not for the following.
+
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 0);
+      expect(insertList, [0, 1, 2]);
+      expect(timeline.events.length,
+          3); // we still expect the timeline to contain the same numbre of elements
+    });
+
+    test('Request future end of timeline', () async {
+      await timeline.requestFuture();
+
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 3);
+      expect(insertList, [0, 1, 2]);
+      expect(insertList.length, timeline.events.length);
+      expect(timeline.events[0].eventId, '3143273582443PhrSn:example.org');
+      expect(timeline.events[1].eventId, '2143273582443PhrSn:example.org');
+      expect(timeline.events[2].eventId, '1143273582443PhrSn:example.org');
+      expect(timeline.chunk.nextBatch, 't789');
+
+      expect(timeline.isFragmentedTimeline, true);
+      expect(timeline.allowNewEvent, true);
+    });
+
+    test('Send message', () async {
+      await room.sendTextEvent('test', txid: '1234');
+
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(updateCount, 5);
+      expect(insertList, [0, 1, 2, 0]);
+      expect(insertList.length, timeline.events.length);
+      final eventId = timeline.events[0].eventId;
+      expect(eventId.startsWith('\$event'), true);
+      expect(timeline.events[0].status, EventStatus.sent);
+
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'test'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.synced.intValue,
+          'event_id': eventId,
+          'unsigned': {'transaction_id': '1234'},
+          'origin_server_ts': DateTime.now().millisecondsSinceEpoch
+        },
+      ));
+
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 6);
+      expect(insertList, [0, 1, 2, 0]);
+      expect(insertList.length, timeline.events.length);
+      expect(timeline.events[0].eventId, eventId);
+      expect(timeline.events[0].status, EventStatus.synced);
+    });
+
+    test('Send message with error', () async {
+      updateCount = 0;
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.sending.intValue,
+          'event_id': 'abc',
+          'origin_server_ts': testTimeStamp
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 1);
+      await room.sendTextEvent('test', txid: 'errortxid');
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 3);
+      await room.sendTextEvent('test', txid: 'errortxid2');
+      await Future.delayed(Duration(milliseconds: 50));
+      await room.sendTextEvent('test', txid: 'errortxid3');
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 7);
+      expect(insertList, [0, 1, 2, 0, 0, 0, 1, 2]);
+      expect(insertList.length, timeline.events.length);
+      expect(changeList, [0, 0, 0, 1, 2]);
+      expect(removeList, []);
+      expect(timeline.events[0].status, EventStatus.error);
+      expect(timeline.events[1].status, EventStatus.error);
+      expect(timeline.events[2].status, EventStatus.error);
+    });
+
+    test('Remove message', () async {
+      updateCount = 0;
+      await timeline.events[0].remove();
+
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 1);
+
+      expect(insertList, [0, 1, 2, 0, 0, 0, 1, 2]);
+      expect(changeList, [0, 0, 0, 1, 2]);
+      expect(removeList, [0]);
+      expect(timeline.events.length, 7);
+      expect(timeline.events[0].status, EventStatus.error);
+    });
+
+    test('getEventById', () async {
+      var event = await timeline.getEventById('abc');
+      expect(event?.content, {'msgtype': 'm.text', 'body': 'Testcase'});
+
+      event = await timeline.getEventById('not_found');
+      expect(event, null);
+
+      event = await timeline.getEventById('unencrypted_event');
+      expect(event?.body, 'This is an example text message');
+
+      if (olmEnabled) {
+        event = await timeline.getEventById('encrypted_event');
+        // the event is invalid but should have traces of attempting to decrypt
+        expect(event?.messageType, MessageTypes.BadEncrypted);
+      }
+    });
+
+    test('Resend message', () async {
+      timeline.events.clear();
+      updateCount = 0;
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.error.intValue,
+          'event_id': 'new-test-event',
+          'origin_server_ts': testTimeStamp,
+          'unsigned': {'transaction_id': 'newresend'},
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.error);
+      await timeline.events[0].sendAgain();
+
+      await Future.delayed(Duration(milliseconds: 50));
+
+      expect(updateCount, 3);
+
+      expect(insertList, [0, 1, 2, 0, 0, 0, 1, 2, 0]);
+      expect(changeList, [0, 0, 0, 1, 2, 0, 0]);
+      expect(removeList, [0]);
+      expect(timeline.events.length, 1);
+      expect(timeline.events[0].status, EventStatus.sent);
+    });
+
+    test('Clear cache on limited timeline', () async {
+      client.onSync.add(
+        SyncUpdate(
+          nextBatch: '1234',
+          rooms: RoomsUpdate(
+            join: {
+              roomID: JoinedRoomUpdate(
+                timeline: TimelineUpdate(
+                  limited: true,
+                  prevBatch: 'blah',
+                ),
+                unreadNotifications: UnreadNotificationCounts(
+                  highlightCount: 0,
+                  notificationCount: 0,
+                ),
+              ),
+            },
+          ),
+        ),
+      );
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events.isEmpty, true);
+    });
+
+    test('sort errors on top', () async {
+      timeline.events.clear();
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.error.intValue,
+          'event_id': 'abc',
+          'origin_server_ts': testTimeStamp
+        },
+      ));
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.synced.intValue,
+          'event_id': 'def',
+          'origin_server_ts': testTimeStamp + 5
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.error);
+      expect(timeline.events[1].status, EventStatus.synced);
+    });
+
+    test('sending event to failed update', () async {
+      timeline.events.clear();
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.sending.intValue,
+          'event_id': 'will-fail',
+          'origin_server_ts': DateTime.now().millisecondsSinceEpoch,
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.sending);
+      expect(timeline.events.length, 1);
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.error.intValue,
+          'event_id': 'will-fail',
+          'origin_server_ts': testTimeStamp
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.error);
+      expect(timeline.events.length, 1);
+    });
+    test('setReadMarker', () async {
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.synced.intValue,
+          'event_id': 'will-work',
+          'origin_server_ts': DateTime.now().millisecondsSinceEpoch,
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      room.notificationCount = 1;
+      await timeline.setReadMarker();
+      expect(room.notificationCount, 0);
+    });
+    test('sending an event and the http request finishes first, 0 -> 1 -> 2',
+        () async {
+      timeline.events.clear();
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.sending.intValue,
+          'event_id': 'transaction',
+          'origin_server_ts': DateTime.now().millisecondsSinceEpoch,
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.sending);
+      expect(timeline.events.length, 1);
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.sent.intValue,
+          'event_id': '\$event',
+          'origin_server_ts': testTimeStamp,
+          'unsigned': {'transaction_id': 'transaction'}
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.sent);
+      expect(timeline.events.length, 1);
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.synced.intValue,
+          'event_id': '\$event',
+          'origin_server_ts': testTimeStamp,
+          'unsigned': {'transaction_id': 'transaction'}
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.synced);
+      expect(timeline.events.length, 1);
+    });
+    test('sending an event where the sync reply arrives first, 0 -> 2 -> 1',
+        () async {
+      timeline.events.clear();
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'event_id': 'transaction',
+          'origin_server_ts': DateTime.now().millisecondsSinceEpoch,
+          'unsigned': {
+            messageSendingStatusKey: EventStatus.sending.intValue,
+            'transaction_id': 'transaction',
+          },
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.sending);
+      expect(timeline.events.length, 1);
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'event_id': '\$event',
+          'origin_server_ts': testTimeStamp,
+          'unsigned': {
+            'transaction_id': 'transaction',
+            messageSendingStatusKey: EventStatus.synced.intValue,
+          },
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.synced);
+      expect(timeline.events.length, 1);
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'event_id': '\$event',
+          'origin_server_ts': testTimeStamp,
+          'unsigned': {
+            'transaction_id': 'transaction',
+            messageSendingStatusKey: EventStatus.sent.intValue,
+          },
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.synced);
+      expect(timeline.events.length, 1);
+    });
+    test('sending an event 0 -> -1 -> 2', () async {
+      timeline.events.clear();
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.sending.intValue,
+          'event_id': 'transaction',
+          'origin_server_ts': DateTime.now().millisecondsSinceEpoch,
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.sending);
+      expect(timeline.events.length, 1);
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.error.intValue,
+          'origin_server_ts': testTimeStamp,
+          'unsigned': {'transaction_id': 'transaction'},
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.error);
+      expect(timeline.events.length, 1);
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.synced.intValue,
+          'event_id': '\$event',
+          'origin_server_ts': testTimeStamp,
+          'unsigned': {'transaction_id': 'transaction'},
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.synced);
+      expect(timeline.events.length, 1);
+    });
+    test('sending an event 0 -> 2 -> -1', () async {
+      timeline.events.clear();
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.sending.intValue,
+          'event_id': 'transaction',
+          'origin_server_ts': DateTime.now().millisecondsSinceEpoch,
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.sending);
+      expect(timeline.events.length, 1);
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.synced.intValue,
+          'event_id': '\$event',
+          'origin_server_ts': testTimeStamp,
+          'unsigned': {'transaction_id': 'transaction'},
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.synced);
+      expect(timeline.events.length, 1);
+      client.onEvent.add(EventUpdate(
+        type: EventUpdateType.timeline,
+        roomID: roomID,
+        content: {
+          'type': 'm.room.message',
+          'content': {'msgtype': 'm.text', 'body': 'Testcase'},
+          'sender': '@alice:example.com',
+          'status': EventStatus.error.intValue,
+          'origin_server_ts': testTimeStamp,
+          'unsigned': {'transaction_id': 'transaction'},
+        },
+      ));
+      await Future.delayed(Duration(milliseconds: 50));
+      expect(timeline.events[0].status, EventStatus.synced);
+      expect(timeline.events.length, 1);
+    });
+    test('logout', () async {
+      await client.logout();
+    });
+  });
+}

test/timeline_test.dart

@@ -17,6 +17,7 @@
  */
 
 import 'package:matrix/matrix.dart';
+import 'package:matrix/src/models/timeline_chunk.dart';
 
 import 'package:test/test.dart';
 import 'package:olm/olm.dart' as olm;
@@ -52,7 +53,7 @@ void main() {
           id: roomID, client: client, prev_batch: '1234', roomAccountData: {});
       timeline = Timeline(
         room: room,
-        events: [],
+        chunk: TimelineChunk(events: []),
         onUpdate: () {
           updateCount++;
         },