1package eu.siacs.conversations.services;
2
3import static eu.siacs.conversations.utils.Random.SECURE_RANDOM;
4
5import android.util.Log;
6
7import androidx.annotation.NonNull;
8
9import java.math.BigInteger;
10import java.util.ArrayList;
11import java.util.HashSet;
12import java.util.Iterator;
13import java.util.List;
14
15import eu.siacs.conversations.Config;
16import eu.siacs.conversations.R;
17import eu.siacs.conversations.entities.Account;
18import eu.siacs.conversations.entities.Conversation;
19import eu.siacs.conversations.entities.Conversational;
20import eu.siacs.conversations.entities.ReceiptRequest;
21import eu.siacs.conversations.generator.AbstractGenerator;
22import eu.siacs.conversations.xml.Element;
23import eu.siacs.conversations.xmpp.Jid;
24import eu.siacs.conversations.xmpp.OnAdvancedStreamFeaturesLoaded;
25import eu.siacs.conversations.xmpp.mam.MamReference;
26import eu.siacs.conversations.xmpp.stanzas.IqPacket;
27import eu.siacs.conversations.xmpp.stanzas.MessagePacket;
28
29public class MessageArchiveService implements OnAdvancedStreamFeaturesLoaded {
30
31 private final XmppConnectionService mXmppConnectionService;
32
33 private final HashSet<Query> queries = new HashSet<>();
34 private final ArrayList<Query> pendingQueries = new ArrayList<>();
35
36 public enum Version {
37 MAM_0("urn:xmpp:mam:0", true),
38 MAM_1("urn:xmpp:mam:1", false),
39 MAM_2("urn:xmpp:mam:2", false);
40
41 public final boolean legacy;
42 public final String namespace;
43
44 Version(String namespace, boolean legacy) {
45 this.namespace = namespace;
46 this.legacy = legacy;
47 }
48
49 public static Version get(Account account) {
50 return get(account, null);
51 }
52
53 public static Version get(Account account, Conversation conversation) {
54 if (conversation == null || conversation.getMode() == Conversation.MODE_SINGLE) {
55 return get(account.getXmppConnection().getFeatures().getAccountFeatures());
56 } else {
57 return get(conversation.getMucOptions().getFeatures());
58 }
59 }
60
61 private static Version get(List<String> features) {
62 final Version[] values = values();
63 for (int i = values.length - 1; i >= 0; --i) {
64 for (String feature : features) {
65 if (values[i].namespace.equals(feature)) {
66 return values[i];
67 }
68 }
69 }
70 return MAM_0;
71 }
72
73 public static boolean has(List<String> features) {
74 for (String feature : features) {
75 for (Version version : values()) {
76 if (version.namespace.equals(feature)) {
77 return true;
78 }
79 }
80 }
81 return false;
82 }
83
84 public static Element findResult(MessagePacket packet) {
85 for (Version version : values()) {
86 Element result = packet.findChild("result", version.namespace);
87 if (result != null) {
88 return result;
89 }
90 }
91 return null;
92 }
93
94 }
95
96 MessageArchiveService(final XmppConnectionService service) {
97 this.mXmppConnectionService = service;
98 }
99
100 private void catchup(final Account account) {
101 synchronized (this.queries) {
102 for (Iterator<Query> iterator = this.queries.iterator(); iterator.hasNext(); ) {
103 Query query = iterator.next();
104 if (query.getAccount() == account) {
105 iterator.remove();
106 }
107 }
108 }
109 MamReference mamReference = MamReference.max(
110 mXmppConnectionService.databaseBackend.getLastMessageReceived(account),
111 mXmppConnectionService.databaseBackend.getLastClearDate(account)
112 );
113 long endCatchup = account.getXmppConnection().getLastSessionEstablished();
114 final Query query;
115 if (mamReference.getTimestamp() == 0) {
116 return;
117 } else if (endCatchup - mamReference.getTimestamp() >= Config.MAM_MAX_CATCHUP) {
118 long startCatchup = endCatchup - Config.MAM_MAX_CATCHUP;
119 List<Conversation> conversations = mXmppConnectionService.getConversations();
120 for (Conversation conversation : conversations) {
121 if (conversation.getMode() == Conversation.MODE_SINGLE && conversation.getAccount() == account && startCatchup > conversation.getLastMessageTransmitted().getTimestamp()) {
122 this.query(conversation, startCatchup, true);
123 }
124 }
125 query = new Query(account, new MamReference(startCatchup), 0);
126 } else {
127 query = new Query(account, mamReference, 0);
128 }
129 synchronized (this.queries) {
130 this.queries.add(query);
131 }
132 this.execute(query);
133 }
134
135 void catchupMUC(final Conversation conversation) {
136 if (conversation.getLastMessageTransmitted().getTimestamp() < 0 && conversation.countMessages() == 0) {
137 query(conversation,
138 new MamReference(0),
139 0,
140 true);
141 } else {
142 query(conversation,
143 conversation.getLastMessageTransmitted(),
144 0,
145 true);
146 }
147 }
148
149 public Query query(final Conversation conversation) {
150 if (conversation.getLastMessageTransmitted().getTimestamp() < 0 && conversation.countMessages() == 0) {
151 return query(conversation,
152 new MamReference(0),
153 System.currentTimeMillis(),
154 false);
155 } else {
156 return query(conversation,
157 conversation.getLastMessageTransmitted(),
158 conversation.getAccount().getXmppConnection().getLastSessionEstablished(),
159 false);
160 }
161 }
162
163 public boolean isCatchingUp(Conversation conversation) {
164 final Account account = conversation.getAccount();
165 if (account.getXmppConnection().isWaitingForSmCatchup()) {
166 return true;
167 } else {
168 synchronized (this.queries) {
169 for (Query query : this.queries) {
170 if (query.getAccount() == account && query.isCatchup() && ((conversation.getMode() == Conversation.MODE_SINGLE && query.getWith() == null) || query.getConversation() == conversation)) {
171 return true;
172 }
173 }
174 }
175 return false;
176 }
177 }
178
179 public Query query(final Conversation conversation, long end, boolean allowCatchup) {
180 return this.query(conversation, conversation.getLastMessageTransmitted(), end, allowCatchup);
181 }
182
183 public Query query(Conversation conversation, MamReference start, long end, boolean allowCatchup) {
184 synchronized (this.queries) {
185 final Query query;
186 final MamReference startActual = start;
187 if (start.getTimestamp() == 0) {
188 query = new Query(conversation, startActual, end, false);
189 query.reference = conversation.getFirstMamReference();
190 } else {
191 if (allowCatchup) {
192 MamReference maxCatchup = MamReference.max(startActual, System.currentTimeMillis() - Config.MAM_MAX_CATCHUP);
193 if (maxCatchup.greaterThan(startActual)) {
194 Query reverseCatchup = new Query(conversation, startActual, maxCatchup.getTimestamp(), false);
195 this.queries.add(reverseCatchup);
196 this.execute(reverseCatchup);
197 }
198 query = new Query(conversation, maxCatchup, end, true);
199 } else {
200 query = new Query(conversation, startActual, end, false);
201 }
202 }
203 if (end != 0 && start.greaterThan(end)) {
204 return null;
205 }
206 this.queries.add(query);
207 this.execute(query);
208 return query;
209 }
210 }
211
212 void executePendingQueries(final Account account) {
213 final List<Query> pending = new ArrayList<>();
214 synchronized (this.pendingQueries) {
215 for (Iterator<Query> iterator = this.pendingQueries.iterator(); iterator.hasNext(); ) {
216 Query query = iterator.next();
217 if (query.getAccount() == account) {
218 pending.add(query);
219 iterator.remove();
220 }
221 }
222 }
223 for (Query query : pending) {
224 this.execute(query);
225 }
226 }
227
228 private void execute(final Query query) {
229 final Account account = query.getAccount();
230 if (account.getStatus() == Account.State.ONLINE) {
231 final Conversation conversation = query.getConversation();
232 if (conversation != null && conversation.getStatus() == Conversation.STATUS_ARCHIVED) {
233 throw new IllegalStateException("Attempted to run MAM query for archived conversation");
234 }
235 Log.d(Config.LOGTAG, account.getJid().asBareJid().toString() + ": running mam query " + query.toString());
236 final IqPacket packet = this.mXmppConnectionService.getIqGenerator().queryMessageArchiveManagement(query);
237 this.mXmppConnectionService.sendIqPacket(account, packet, (a, p) -> {
238 final Element fin = p.findChild("fin", query.version.namespace);
239 if (p.getType() == IqPacket.TYPE.TIMEOUT) {
240 synchronized (this.queries) {
241 this.queries.remove(query);
242 if (query.hasCallback()) {
243 query.callback(false);
244 }
245 }
246 } else if (p.getType() == IqPacket.TYPE.RESULT && fin != null) {
247 final boolean running;
248 synchronized (this.queries) {
249 running = this.queries.contains(query);
250 }
251 if (running) {
252 processFin(query, fin);
253 } else {
254 Log.d(Config.LOGTAG, account.getJid().asBareJid() + ": ignoring MAM iq result because query had been killed");
255 }
256 } else if (p.getType() == IqPacket.TYPE.RESULT && query.isLegacy()) {
257 //do nothing
258 } else {
259 Log.d(Config.LOGTAG, a.getJid().asBareJid().toString() + ": error executing mam: " + p.toString());
260 try {
261 finalizeQuery(query, true);
262 } catch (final IllegalStateException e) {
263 //ignored
264 }
265 }
266 });
267 } else {
268 synchronized (this.pendingQueries) {
269 this.pendingQueries.add(query);
270 }
271 }
272 }
273
274 private void finalizeQuery(final Query query, boolean done) {
275 synchronized (this.queries) {
276 if (!this.queries.remove(query)) {
277 throw new IllegalStateException("Unable to remove query from queries");
278 }
279 }
280 final Conversation conversation = query.getConversation();
281 if (conversation != null) {
282 conversation.sort();
283 conversation.setHasMessagesLeftOnServer(!done);
284 final var displayState = conversation.getDisplayState();
285 if (displayState != null) {
286 mXmppConnectionService.markReadUpToStanzaId(conversation, displayState);
287 }
288 } else {
289 for (final Conversation tmp : this.mXmppConnectionService.getConversations()) {
290 if (tmp.getAccount() == query.getAccount()) {
291 tmp.sort();
292 final var displayState = tmp.getDisplayState();
293 if (displayState != null) {
294 mXmppConnectionService.markReadUpToStanzaId(tmp, displayState);
295 }
296 }
297 }
298 }
299 if (query.hasCallback()) {
300 query.callback(done);
301 } else {
302 this.mXmppConnectionService.updateConversationUi();
303 }
304 }
305
306 boolean inCatchup(Account account) {
307 synchronized (this.queries) {
308 for (Query query : queries) {
309 if (query.account == account && query.isCatchup() && query.getWith() == null) {
310 return true;
311 }
312 }
313 }
314 return false;
315 }
316
317 public boolean isCatchupInProgress(Conversation conversation) {
318 synchronized (this.queries) {
319 for (Query query : queries) {
320 if (query.account == conversation.getAccount() && query.isCatchup()) {
321 final Jid with = query.getWith() == null ? null : query.getWith().asBareJid();
322 if ((conversation.getMode() == Conversational.MODE_SINGLE && with == null) || (conversation.getJid().asBareJid().equals(with))) {
323 return true;
324 }
325 }
326 }
327 }
328 return false;
329 }
330
331 boolean queryInProgress(Conversation conversation, XmppConnectionService.OnMoreMessagesLoaded callback) {
332 synchronized (this.queries) {
333 for (Query query : queries) {
334 if (query.conversation == conversation) {
335 if (!query.hasCallback() && callback != null) {
336 query.setCallback(callback);
337 }
338 return true;
339 }
340 }
341 return false;
342 }
343 }
344
345 public boolean queryInProgress(Conversation conversation) {
346 return queryInProgress(conversation, null);
347 }
348
349 public void processFinLegacy(Element fin, Jid from) {
350 Query query = findQuery(fin.getAttribute("queryid"));
351 if (query != null && query.validFrom(from)) {
352 processFin(query, fin);
353 }
354 }
355
356 private void processFin(Query query, Element fin) {
357 boolean complete = fin.getAttributeAsBoolean("complete");
358 Element set = fin.findChild("set", "http://jabber.org/protocol/rsm");
359 Element last = set == null ? null : set.findChild("last");
360 String count = set == null ? null : set.findChildContent("count");
361 Element first = set == null ? null : set.findChild("first");
362 Element relevant = query.getPagingOrder() == PagingOrder.NORMAL ? last : first;
363 boolean abort = (!query.isCatchup() && query.getTotalCount() >= Config.PAGE_SIZE) || query.getTotalCount() >= Config.MAM_MAX_MESSAGES;
364 if (query.getConversation() != null) {
365 query.getConversation().setFirstMamReference(first == null ? null : first.getContent());
366 }
367 if (complete || relevant == null || abort) {
368 //TODO: FIX done logic to look at complete. using count is probably unreliable because it can be ommited and doesn’t work with paging.
369 boolean done;
370 if (query.isCatchup()) {
371 done = false;
372 } else {
373 if (count != null) {
374 try {
375 done = Integer.parseInt(count) <= query.getTotalCount();
376 } catch (NumberFormatException e) {
377 done = false;
378 }
379 } else {
380 done = query.getTotalCount() == 0;
381 }
382 }
383 done = done || (query.getActualMessageCount() == 0 && !query.isCatchup());
384 this.finalizeQuery(query, done);
385
386 Log.d(Config.LOGTAG, query.getAccount().getJid().asBareJid() + ": finished mam after " + query.getTotalCount() + "(" + query.getActualMessageCount() + ") messages. messages left=" + !done + " count=" + count);
387 if (query.isCatchup() && query.getActualMessageCount() > 0) {
388 mXmppConnectionService.getNotificationService().finishBacklog(true, query.getAccount());
389 }
390 processPostponed(query);
391 } else {
392 final Query nextQuery;
393 if (query.getPagingOrder() == PagingOrder.NORMAL) {
394 nextQuery = query.next(last == null ? null : last.getContent());
395 } else {
396 nextQuery = query.prev(first == null ? null : first.getContent());
397 }
398 this.execute(nextQuery);
399 this.finalizeQuery(query, false);
400 synchronized (this.queries) {
401 this.queries.add(nextQuery);
402 }
403 }
404 }
405
406 void kill(final Conversation conversation) {
407 final ArrayList<Query> toBeKilled = new ArrayList<>();
408 synchronized (this.pendingQueries) {
409 for (final Iterator<Query> iterator = this.pendingQueries.iterator(); iterator.hasNext(); ) {
410 final Query query = iterator.next();
411 if (query.getConversation() == conversation) {
412 iterator.remove();
413 Log.d(Config.LOGTAG, conversation.getAccount().getJid().asBareJid() + ": killed pending MAM query for archived conversation");
414 }
415 }
416 }
417 synchronized (this.queries) {
418 for (final Query q : queries) {
419 if (q.conversation == conversation) {
420 toBeKilled.add(q);
421 }
422 }
423 }
424 for (final Query q : toBeKilled) {
425 kill(q);
426 }
427 }
428
429 private void kill(Query query) {
430 Log.d(Config.LOGTAG, query.getAccount().getJid().asBareJid() + ": killing mam query prematurely");
431 query.callback = null;
432 this.finalizeQuery(query, false);
433 if (query.isCatchup() && query.getActualMessageCount() > 0) {
434 mXmppConnectionService.getNotificationService().finishBacklog(true, query.getAccount());
435 }
436 this.processPostponed(query);
437 }
438
439 private void processPostponed(Query query) {
440 query.account.getAxolotlService().processPostponed();
441 query.pendingReceiptRequests.removeAll(query.receiptRequests);
442 Log.d(Config.LOGTAG, query.getAccount().getJid().asBareJid() + ": found " + query.pendingReceiptRequests.size() + " pending receipt requests");
443 Iterator<ReceiptRequest> iterator = query.pendingReceiptRequests.iterator();
444 while (iterator.hasNext()) {
445 ReceiptRequest rr = iterator.next();
446 mXmppConnectionService.sendMessagePacket(query.account, mXmppConnectionService.getMessageGenerator().received(query.account, rr.getJid(), rr.getId()));
447 iterator.remove();
448 }
449 }
450
451 public Query findQuery(String id) {
452 if (id == null) {
453 return null;
454 }
455 synchronized (this.queries) {
456 for (Query query : this.queries) {
457 if (query.getQueryId().equals(id)) {
458 return query;
459 }
460 }
461 return null;
462 }
463 }
464
465 @Override
466 public void onAdvancedStreamFeaturesAvailable(Account account) {
467 if (account.getXmppConnection() != null && account.getXmppConnection().getFeatures().mam()) {
468 this.catchup(account);
469 }
470 }
471
472 public enum PagingOrder {
473 NORMAL,
474 REVERSE
475 }
476
477 public class Query {
478 private HashSet<ReceiptRequest> pendingReceiptRequests = new HashSet<>();
479 private HashSet<ReceiptRequest> receiptRequests = new HashSet<>();
480 private int totalCount = 0;
481 private int actualCount = 0;
482 private int actualInThisQuery = 0;
483 private long start;
484 private final long end;
485 private final String queryId;
486 private String reference = null;
487 private final Account account;
488 private Conversation conversation;
489 private PagingOrder pagingOrder = PagingOrder.NORMAL;
490 private XmppConnectionService.OnMoreMessagesLoaded callback = null;
491 private boolean catchup = true;
492 public final Version version;
493
494
495 Query(Conversation conversation, MamReference start, long end, boolean catchup) {
496 this(conversation.getAccount(), Version.get(conversation.getAccount(), conversation), catchup ? start : start.timeOnly(), end);
497 this.conversation = conversation;
498 this.pagingOrder = catchup ? PagingOrder.NORMAL : PagingOrder.REVERSE;
499 this.catchup = catchup;
500 }
501
502 Query(Account account, MamReference start, long end) {
503 this(account, Version.get(account), start, end);
504 }
505
506 Query(Account account, Version version, MamReference start, long end) {
507 this.account = account;
508 if (start.getReference() != null) {
509 this.reference = start.getReference();
510 } else {
511 this.start = start.getTimestamp();
512 }
513 this.end = end;
514 this.queryId = new BigInteger(50, SECURE_RANDOM).toString(32);
515 this.version = version;
516 }
517
518 private Query page(String reference) {
519 Query query = new Query(this.account, this.version, new MamReference(this.start, reference), this.end);
520 query.conversation = conversation;
521 query.totalCount = totalCount;
522 query.actualCount = actualCount;
523 query.pendingReceiptRequests = pendingReceiptRequests;
524 query.receiptRequests = receiptRequests;
525 query.callback = callback;
526 query.catchup = catchup;
527 return query;
528 }
529
530 public void removePendingReceiptRequest(ReceiptRequest receiptRequest) {
531 if (!this.pendingReceiptRequests.remove(receiptRequest)) {
532 this.receiptRequests.add(receiptRequest);
533 }
534 }
535
536 public void addPendingReceiptRequest(ReceiptRequest receiptRequest) {
537 this.pendingReceiptRequests.add(receiptRequest);
538 }
539
540 public boolean isLegacy() {
541 return version.legacy;
542 }
543
544 public boolean safeToExtractTrueCounterpart() {
545 return muc() && !isLegacy();
546 }
547
548 public Query next(String reference) {
549 Query query = page(reference);
550 query.pagingOrder = PagingOrder.NORMAL;
551 return query;
552 }
553
554 Query prev(String reference) {
555 Query query = page(reference);
556 query.pagingOrder = PagingOrder.REVERSE;
557 return query;
558 }
559
560 public String getReference() {
561 return reference;
562 }
563
564 public PagingOrder getPagingOrder() {
565 return this.pagingOrder;
566 }
567
568 public String getQueryId() {
569 return queryId;
570 }
571
572 public Jid getWith() {
573 return conversation == null ? null : conversation.getJid().asBareJid();
574 }
575
576 public boolean muc() {
577 return conversation != null && conversation.getMode() == Conversation.MODE_MULTI;
578 }
579
580 public long getStart() {
581 return start;
582 }
583
584 public boolean isCatchup() {
585 return catchup;
586 }
587
588 public void setCallback(XmppConnectionService.OnMoreMessagesLoaded callback) {
589 this.callback = callback;
590 }
591
592 public void callback(boolean done) {
593 if (this.callback != null) {
594 this.callback.onMoreMessagesLoaded(actualCount, conversation);
595 if (done) {
596 this.callback.informUser(R.string.no_more_history_on_server);
597 }
598 }
599 }
600
601 public long getEnd() {
602 return end;
603 }
604
605 public Conversation getConversation() {
606 return conversation;
607 }
608
609 public Account getAccount() {
610 return this.account;
611 }
612
613 public void incrementMessageCount() {
614 this.totalCount++;
615 }
616
617 public void incrementActualMessageCount() {
618 this.actualInThisQuery++;
619 this.actualCount++;
620 }
621
622 int getTotalCount() {
623 return this.totalCount;
624 }
625
626 int getActualMessageCount() {
627 return this.actualCount;
628 }
629
630 public int getActualInThisQuery() {
631 return this.actualInThisQuery;
632 }
633
634 public boolean validFrom(Jid from) {
635 if (muc()) {
636 return getWith().equals(from);
637 } else {
638 return (from == null) || account.getJid().asBareJid().equals(from.asBareJid());
639 }
640 }
641
642 @NonNull
643 @Override
644 public String toString() {
645 StringBuilder builder = new StringBuilder();
646 if (this.muc()) {
647 builder.append("to=");
648 builder.append(this.getWith().toString());
649 } else {
650 builder.append("with=");
651 if (this.getWith() == null) {
652 builder.append("*");
653 } else {
654 builder.append(getWith().toString());
655 }
656 }
657 if (this.start != 0) {
658 builder.append(", start=");
659 builder.append(AbstractGenerator.getTimestamp(this.start));
660 }
661 if (this.end != 0) {
662 builder.append(", end=");
663 builder.append(AbstractGenerator.getTimestamp(this.end));
664 }
665 builder.append(", order=").append(pagingOrder.toString());
666 if (this.reference != null) {
667 if (this.pagingOrder == PagingOrder.NORMAL) {
668 builder.append(", after=");
669 } else {
670 builder.append(", before=");
671 }
672 builder.append(this.reference);
673 }
674 builder.append(", catchup=").append(catchup);
675 builder.append(", ns=").append(version.namespace);
676 return builder.toString();
677 }
678
679 boolean hasCallback() {
680 return this.callback != null;
681 }
682 }
683}