@tduncan / Cortana
Code Commits Issues Pulls Cloud Agent CI/CD
main
main
dart 128 lines 3.6 KB
Raw
1 import 'dart:async';
2
3 import '../../features/chat/data/models/message.dart';
4 import '../../objectbox.g.dart';
5 import '../logger/app_logger.dart';
6 import '../storage/entities.dart';
7 import '../storage/objectbox_store.dart';
8
9 /// Batches [Message] saves and flushes them off the UI thread using
10 /// ObjectBox's [Store.runInTransactionAsync].
11 ///
12 /// Call [enqueue] from the streaming listener (fire-and-forget).
13 /// Call [flush] when the stream ends or is cancelled to ensure the
14 /// final message state is persisted before updating UI state.
15 class MessageSaveService {
16 final ObjectBoxStore _db;
17
18 final List<Message> _queue = [];
19 Timer? _flushTimer;
20 bool _disposed = false;
21 bool _isFlushing = false;
22 bool _flushAgain = false;
23 Completer<void>? _activeFlushCompleter;
24
25 MessageSaveService(this._db);
26
27 /// Enqueues [message] for a background save. Never blocks the caller.
28 ///
29 /// If the same message ID is already in the queue it is replaced
30 /// (streaming updates supersede the previous checkpoint).
31 void enqueue(Message message) {
32 if (_disposed) return;
33 final idx = _queue.indexWhere((m) => m.id == message.id);
34 if (idx != -1) {
35 _queue[idx] = message;
36 } else {
37 _queue.add(message);
38 }
39 // Start a periodic flush timer on the first enqueue.
40 _flushTimer ??= Timer.periodic(
41 const Duration(seconds: 1),
42 (_) => _doFlush(),
43 );
44 }
45
46 /// Drains the queue immediately. Awaiting this guarantees persistence
47 /// before the caller updates final UI state.
48 Future<void> flush() async {
49 _flushTimer?.cancel();
50 _flushTimer = null;
51 await _doFlush();
52 }
53
54 Future<void> _doFlush() async {
55 if (_isFlushing) {
56 _flushAgain = true;
57 await _activeFlushCompleter?.future;
58 return;
59 }
60
61 _isFlushing = true;
62 _activeFlushCompleter = Completer<void>();
63 try {
64 do {
65 _flushAgain = false;
66 if (_queue.isEmpty) break;
67
68 final batch = List<Message>.from(_queue);
69 _queue.clear();
70
71 try {
72 await _db.store.runInTransactionAsync(
73 TxMode.write,
74 _writeBatch,
75 batch,
76 );
77 } catch (e) {
78 Log.error('MessageSaveService flush error: $e');
79 }
80 } while (_flushAgain || _queue.isNotEmpty);
81 } finally {
82 _isFlushing = false;
83 _activeFlushCompleter?.complete();
84 _activeFlushCompleter = null;
85 }
86 }
87
88 /// Runs inside a background isolate provided by ObjectBox.
89 static void _writeBatch(Store store, List<Message> messages) {
90 final box = store.box<MessageEntity>();
91 final convBox = store.box<ConversationEntity>();
92
93 for (final message in messages) {
94 // Look up existing entity so we preserve the ObjectBox internal ID.
95 final query = box.query(MessageEntity_.id.equals(message.id)).build();
96 final existing = query.findFirst();
97 query.close();
98
99 final entity = MessageEntity.fromDomain(message);
100 if (existing != null) {
101 entity.internalId = existing.internalId;
102 }
103
104 // Re-establish the ToOne relation to the parent conversation.
105 final convQuery = convBox
106 .query(ConversationEntity_.id.equals(message.conversationId))
107 .build();
108 final convEntity = convQuery.findFirst();
109 convQuery.close();
110
111 if (convEntity != null) {
112 entity.conversation.target = convEntity;
113 entity.conversationUid = convEntity.id;
114 } else {
115 entity.conversationUid = message.conversationId;
116 }
117
118 box.put(entity);
119 }
120 }
121
122 void dispose() {
123 _disposed = true;
124 _flushTimer?.cancel();
125 _flushTimer = null;
126 _queue.clear();
127 }
128 }