working hard

This commit is contained in:
OfficialDakari 2025-10-25 18:43:31 +05:00
parent 74c39f91fe
commit 98031bbb3d
6 changed files with 111 additions and 57 deletions

View File

@ -1392,7 +1392,6 @@ class MatrixSdkDatabase extends DatabaseApi with DatabaseFileStorage {
final key = TupleKey(roomId, threadRootEventId).toString(); final key = TupleKey(roomId, threadRootEventId).toString();
final thread = await _threadsBox.get(key); final thread = await _threadsBox.get(key);
if (thread == null) return null; if (thread == null) return null;
Logs().w(thread.toString());
return Thread.fromJson(thread.cast<String, dynamic>(), client); return Thread.fromJson(thread.cast<String, dynamic>(), client);
} }

View File

@ -136,6 +136,8 @@ class Room {
partial = false; partial = false;
} }
Map<String, Thread> threads = <String, Thread>{};
Future<void> _loadThreadsFromServer() async { Future<void> _loadThreadsFromServer() async {
try { try {
final response = await client.getThreadRoots(id); final response = await client.getThreadRoots(id);
@ -151,6 +153,7 @@ class Room {
1, // count 1, // count
client, client,
); );
threads[event.eventId] = (await client.database.getThread(id, event.eventId, client))!;
} }
} catch (e) { } catch (e) {
Logs().w('Failed to load threads from server', e); Logs().w('Failed to load threads from server', e);
@ -161,7 +164,8 @@ class Room {
// This should be called from the client's sync handling // This should be called from the client's sync handling
// when a thread-related event is received // when a thread-related event is received
if (event.relationshipType == RelationshipTypes.thread && event.relationshipEventId != null) { if (event.relationshipType == RelationshipTypes.thread &&
event.relationshipEventId != null) {
// Update thread metadata in database // Update thread metadata in database
final root = await getEventById(event.relationshipEventId!); final root = await getEventById(event.relationshipEventId!);
if (root == null) return; if (root == null) return;
@ -226,6 +230,18 @@ class Room {
return dict; return dict;
} }
Future<Thread> getThread(Event rootEvent) async {
final threads = await getThreads();
if (threads.containsKey(rootEvent.eventId)) return threads[rootEvent.eventId]!;
return Thread(
room: this,
rootEvent: rootEvent,
client: client,
currentUserParticipated: false,
count: 0,
);
}
/// ID of the fully read marker event. /// ID of the fully read marker event.
String get fullyRead => String get fullyRead =>
roomAccountData['m.fully_read']?.content.tryGet<String>('event_id') ?? ''; roomAccountData['m.fully_read']?.content.tryGet<String>('event_id') ?? '';

View File

@ -37,9 +37,13 @@ class RoomTimeline extends Timeline {
StreamSubscription<String>? sessionIdReceivedSub; StreamSubscription<String>? sessionIdReceivedSub;
StreamSubscription<String>? cancelSendEventSub; StreamSubscription<String>? cancelSendEventSub;
@override
bool isRequestingHistory = false; bool isRequestingHistory = false;
@override
bool isRequestingFuture = false; bool isRequestingFuture = false;
@override
bool allowNewEvent = true; bool allowNewEvent = true;
@override
bool isFragmentedTimeline = false; bool isFragmentedTimeline = false;
final Map<String, Event> _eventCache = {}; final Map<String, Event> _eventCache = {};

View File

@ -45,6 +45,16 @@ class Thread {
room, room,
); );
} }
if (json['unsigned']?['m.thread']?['latest_event'] != null) {
lastEvent = Event.fromMatrixEvent(
MatrixEvent.fromJson(
json['unsigned']?['m.thread']?['latest_event'],
),
room,
);
}
// Although I was making this part according to specification, it's a bit off
// I have no clue why
final thread = Thread( final thread = Thread(
room: room, room: room,
client: client, client: client,
@ -334,7 +344,7 @@ class Thread {
if (prev_batch == null) { if (prev_batch == null) {
throw 'Tried to request history without a prev_batch token'; throw 'Tried to request history without a prev_batch token';
} }
final resp = await client.getRelatingEventsWithRelType( final resp = await client.getRelatingEventsWithRelType(
room.id, room.id,
rootEvent.eventId, rootEvent.eventId,
@ -350,7 +360,8 @@ class Thread {
await client.database.transaction(() async { await client.database.transaction(() async {
if (storeInDatabase && direction == Direction.b) { if (storeInDatabase && direction == Direction.b) {
this.prev_batch = resp.prevBatch; this.prev_batch = resp.prevBatch;
await client.database.setThreadPrevBatch(resp.prevBatch, room.id, rootEvent.eventId, client); await client.database.setThreadPrevBatch(
resp.prevBatch, room.id, rootEvent.eventId, client);
} }
}); });

View File

@ -1,5 +1,4 @@
import 'dart:async'; import 'dart:async';
import 'dart:convert';
import 'package:matrix/matrix.dart'; import 'package:matrix/matrix.dart';
import 'package:matrix/src/models/timeline_chunk.dart'; import 'package:matrix/src/models/timeline_chunk.dart';
@ -23,17 +22,18 @@ class ThreadTimeline extends Timeline {
StreamSubscription<String>? sessionIdReceivedSub; StreamSubscription<String>? sessionIdReceivedSub;
StreamSubscription<String>? cancelSendEventSub; StreamSubscription<String>? cancelSendEventSub;
@override
bool isRequestingHistory = false; bool isRequestingHistory = false;
@override
bool isFragmentedTimeline = false; bool isFragmentedTimeline = false;
final Map<String, Event> _eventCache = {}; final Map<String, Event> _eventCache = {};
bool _fetchedAllDatabaseEvents = false; @override
bool allowNewEvent = true; bool allowNewEvent = true;
bool _collectHistoryUpdates = false; @override
bool isRequestingFuture = false; bool isRequestingFuture = false;
ThreadTimeline({ ThreadTimeline({
@ -52,6 +52,17 @@ class ThreadTimeline extends Timeline {
historySub = room.client.onHistoryEvent.stream.listen( historySub = room.client.onHistoryEvent.stream.listen(
(event) => _handleEventUpdate(event, EventUpdateType.history), (event) => _handleEventUpdate(event, EventUpdateType.history),
); );
// we want to populate our aggregated events
for (final e in events) {
addAggregatedEvent(e);
}
// we are using a fragmented timeline
if (chunk.nextBatch != '') {
allowNewEvent = false;
isFragmentedTimeline = true;
}
} }
void _handleEventUpdate( void _handleEventUpdate(
@ -63,7 +74,9 @@ class ThreadTimeline extends Timeline {
if (event.roomId != thread.room.id) return; if (event.roomId != thread.room.id) return;
// Ignore events outside of this thread // Ignore events outside of this thread
if (event.relationshipType != RelationshipTypes.thread || if (event.relationshipType != RelationshipTypes.thread ||
event.relationshipEventId != thread.rootEvent.eventId) return; event.relationshipEventId != thread.rootEvent.eventId) {
return;
}
if (type != EventUpdateType.timeline && type != EventUpdateType.history) { if (type != EventUpdateType.timeline && type != EventUpdateType.history) {
return; return;
@ -152,14 +165,21 @@ class ThreadTimeline extends Timeline {
dir: direction, dir: direction,
from: direction == Direction.b ? chunk.prevBatch : chunk.nextBatch, from: direction == Direction.b ? chunk.prevBatch : chunk.nextBatch,
limit: historyCount, limit: historyCount,
recurse: true,
);
Logs().w(
'Loading thread events from server ${resp.chunk.length} ${resp.prevBatch}',
); );
if (resp.nextBatch == null) { if (resp.nextBatch == null) {
Logs().w('We reached the end of the timeline'); Logs().w('We reached the end of the timeline');
} }
final newNextBatch = direction == Direction.b ? resp.prevBatch : resp.nextBatch; final newNextBatch =
final newPrevBatch = direction == Direction.b ? resp.nextBatch : resp.prevBatch; direction == Direction.b ? resp.prevBatch : resp.nextBatch;
final newPrevBatch =
direction == Direction.b ? resp.nextBatch : resp.prevBatch;
final type = direction == Direction.b final type = direction == Direction.b
? EventUpdateType.history ? EventUpdateType.history
@ -193,7 +213,8 @@ class ThreadTimeline extends Timeline {
if (allowNewEvent) { if (allowNewEvent) {
Logs().d('We now allow sync update into the timeline.'); Logs().d('We now allow sync update into the timeline.');
newEvents.addAll( newEvents.addAll(
await thread.client.database.getThreadEventList(thread, onlySending: true), await thread.client.database
.getThreadEventList(thread, onlySending: true),
); );
} }
} }
@ -258,7 +279,8 @@ class ThreadTimeline extends Timeline {
} }
// Fetch all users from database we have got here. // Fetch all users from database we have got here.
for (final event in events) { for (final event in events) {
if (thread.room.getState(EventTypes.RoomMember, event.senderId) != null) { if (thread.room.getState(EventTypes.RoomMember, event.senderId) !=
null) {
continue; continue;
} }
final dbUser = final dbUser =
@ -282,7 +304,6 @@ class ThreadTimeline extends Timeline {
} }
} }
} else { } else {
_fetchedAllDatabaseEvents = true;
Logs().i('No more events found in the store. Request from server...'); Logs().i('No more events found in the store. Request from server...');
if (isFragmentedTimeline) { if (isFragmentedTimeline) {
@ -298,16 +319,13 @@ class ThreadTimeline extends Timeline {
await thread.requestHistory( await thread.requestHistory(
historyCount: historyCount, historyCount: historyCount,
direction: direction, direction: direction,
onHistoryReceived: () { onHistoryReceived: () {},
_collectHistoryUpdates = true;
},
filter: filter, filter: filter,
); );
} }
} }
} }
} finally { } finally {
_collectHistoryUpdates = false;
isRequestingHistory = false; isRequestingHistory = false;
onUpdate?.call(); onUpdate?.call();
} }
@ -387,8 +405,10 @@ class ThreadTimeline extends Timeline {
} }
@override @override
Future<void> requestHistory( Future<void> requestHistory({
{int historyCount = Room.defaultHistoryCount, StateFilter? filter}) async { int historyCount = Room.defaultHistoryCount,
StateFilter? filter,
}) async {
if (isRequestingHistory) return; if (isRequestingHistory) return;
isRequestingHistory = true; isRequestingHistory = true;
await _requestEvents( await _requestEvents(
@ -421,45 +441,45 @@ class ThreadTimeline extends Timeline {
} }
@override @override
bool get canRequestFuture => chunk.nextBatch != null && chunk.nextBatch!.isNotEmpty; bool get canRequestFuture => chunk.nextBatch.isNotEmpty;
@override @override
bool get canRequestHistory => chunk.prevBatch != null && chunk.prevBatch!.isNotEmpty; bool get canRequestHistory => chunk.prevBatch.isNotEmpty;
@override @override
Future<void> requestFuture({ Future<void> requestFuture({
int historyCount = Room.defaultHistoryCount, int historyCount = Room.defaultHistoryCount,
StateFilter? filter, StateFilter? filter,
}) async { }) async {
if (isRequestingFuture || !canRequestFuture) return; if (isRequestingFuture || !canRequestFuture) return;
isRequestingFuture = true; isRequestingFuture = true;
try { try {
await getThreadEvents( await getThreadEvents(
historyCount: historyCount, historyCount: historyCount,
direction: Direction.f, direction: Direction.f,
filter: filter, filter: filter,
); );
} finally { } finally {
isRequestingFuture = false; isRequestingFuture = false;
}
} }
}
@override @override
void requestKeys({ void requestKeys({
bool tryOnlineBackup = true, bool tryOnlineBackup = true,
bool onlineKeyBackupOnly = true, bool onlineKeyBackupOnly = true,
}) { }) {
for (final event in events) { for (final event in events) {
if (event.type == EventTypes.Encrypted && if (event.type == EventTypes.Encrypted &&
event.messageType == MessageTypes.BadEncrypted && event.messageType == MessageTypes.BadEncrypted &&
event.content['can_request_session'] == true) { event.content['can_request_session'] == true) {
final sessionId = event.content.tryGet<String>('session_id'); final sessionId = event.content.tryGet<String>('session_id');
final senderKey = event.content.tryGet<String>('sender_key'); final senderKey = event.content.tryGet<String>('sender_key');
if (sessionId != null && senderKey != null) { if (sessionId != null && senderKey != null) {
thread.room.requestSessionKey(sessionId, senderKey); thread.room.requestSessionKey(sessionId, senderKey);
}
} }
} }
} }
} }
}

View File

@ -45,6 +45,10 @@ abstract class Timeline {
bool get canRequestHistory; bool get canRequestHistory;
bool get canRequestFuture; bool get canRequestFuture;
bool get allowNewEvent;
bool get isRequestingFuture;
bool get isRequestingHistory;
bool get isFragmentedTimeline;
Timeline({ Timeline({
this.onUpdate, this.onUpdate,