Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "conversation_module.h"
19 :
20 : #include "account_const.h"
21 : #include "call.h"
22 : #include "client/jami_signal.h"
23 : #include "fileutils.h"
24 : #include "jamidht/account_manager.h"
25 : #include "jamidht/commit_message.h"
26 : #include "jamidht/jamiaccount.h"
27 : #include "jamidht/collaborative_editing.h"
28 : #include "jamidht/presence_manager.h"
29 : #include "manager.h"
30 : #ifdef ENABLE_PLUGIN
31 : #include "plugin/jamipluginmanager.h"
32 : #endif
33 : #include "sip/sipcall.h"
34 : #include "vcard.h"
35 : #include "json_utils.h"
36 :
37 : #include <opendht/thread_pool.h>
38 : #include <dhtnet/certstore.h>
39 :
40 : #include <algorithm>
41 : #include <fstream>
42 :
43 : namespace jami {
44 :
45 : using ConvInfoMap = std::map<std::string, ConvInfo>;
46 :
47 : struct PendingConversationFetch
48 : {
49 : bool ready {false};
50 : bool cloning {false};
51 : std::string deviceId {};
52 : std::map<std::string, std::string> preferences {};
53 : std::map<std::string, std::map<std::string, std::string>> status {};
54 : std::set<std::string> connectingTo {};
55 : std::shared_ptr<dhtnet::ChannelSocket> socket {};
56 : };
57 :
58 : constexpr std::chrono::seconds MAX_INVITE_CLOCK_SKEW {5 * 60};
59 :
60 : static TimePoint
61 498 : validateInviteTimestamp(TimePoint embedded, TimePoint localReceived)
62 : {
63 498 : if (embedded > localReceived + MAX_INVITE_CLOCK_SKEW)
64 0 : return localReceived;
65 498 : return embedded;
66 : }
67 :
68 : constexpr std::chrono::seconds MAX_FALLBACK {12 * 3600s};
69 : // Maximum attempts at cloning a conversation whose repository fails validation
70 : // before giving up. Validation failures are permanent (the remote history is
71 : // immutable), so retrying is wasted work; the counter is kept in-memory only,
72 : // so a restart allows a new bounded round of attempts.
73 : constexpr unsigned MAX_VALIDATION_FAILURES {3};
74 :
75 : struct SyncedConversation
76 : {
77 : struct DeferredFetch
78 : {
79 : std::string peer;
80 : std::string deviceId;
81 : std::string commitId;
82 : };
83 :
84 : std::mutex mtx;
85 : std::unique_ptr<asio::steady_timer> fallbackClone;
86 : std::chrono::seconds fallbackTimer {5s};
87 : // Earliest time a new clone round may start. Armed together with fallbackClone so that
88 : // triggers calling cloneConversation() directly honour the same backoff as the timer.
89 : std::chrono::steady_clock::time_point nextCloneAttempt {};
90 : // The devices whose clone failed in the current backoff window. The devices of a round
91 : // are enumerated asynchronously, and the first refusal lands before the others were
92 : // even asked: a device that does not hold the repository -- every device of a member
93 : // but the ones that opened a document -- declines the git channel in a few ms. Holding
94 : // the whole conversation back on it would throttle the very device able to serve the
95 : // clone, so the backoff binds only the devices that already failed in this window; the
96 : // others still get their one try.
97 : std::set<std::string> cloneFailedDevices;
98 : // A failure not tied to a device (the repository failed validation, the clone threw)
99 : // backs every device off, as before.
100 : bool cloneBackoffAllDevices {false};
101 : unsigned validationFailures {0};
102 : ConvInfo info;
103 : std::unique_ptr<PendingConversationFetch> pending;
104 : std::map<std::string, DeferredFetch> deferredFetches;
105 : std::shared_ptr<Conversation> conversation;
106 :
107 : // Requires mtx. Shared by clone registration and clone installation.
108 75 : bool canReplicateDocuments(const std::string& username) const
109 : {
110 148 : return !info.isRemoved() && conversation && conversation->mode() != ConversationMode::DOCUMENT
111 148 : && !conversation->isRemoving() && conversation->isMember(username);
112 : }
113 :
114 24 : bool announcesDocument(const std::string& documentId) const
115 : {
116 24 : if (!conversation)
117 0 : return false;
118 24 : const auto documents = conversation->collaborativeDocuments();
119 24 : return std::any_of(documents.begin(), documents.end(), [&](const auto& doc) {
120 69 : return doc.at("id") == documentId;
121 24 : });
122 24 : }
123 :
124 938 : bool isUnrecoverable() const { return validationFailures >= MAX_VALIDATION_FAILURES; }
125 :
126 933 : bool cloneThrottled(const std::string& deviceId) const
127 : {
128 : // conversation mtx must be locked
129 933 : if (std::chrono::steady_clock::now() >= nextCloneAttempt)
130 899 : return false;
131 34 : return cloneBackoffAllDevices || cloneFailedDevices.find(deviceId) != cloneFailedDevices.end();
132 : }
133 :
134 : // Arms the retry deadline and returns the delay until it. Callers must still async_wait()
135 : // on fallbackClone. A failure inside an armed window joins it rather than extending it,
136 : // so a round of N refusals still retries after one delay, not N doublings of it. An empty
137 : // deviceId backs every device off.
138 97 : std::chrono::seconds scheduleCloneRetry(const std::string& deviceId = {})
139 : {
140 : // conversation mtx must be locked
141 97 : const auto now = std::chrono::steady_clock::now();
142 97 : if (now >= nextCloneAttempt) {
143 : // A new window: what failed in the previous one is not held against this round.
144 90 : cloneFailedDevices.clear();
145 90 : cloneBackoffAllDevices = false;
146 90 : nextCloneAttempt = now + fallbackTimer;
147 90 : fallbackTimer = std::min(fallbackTimer * 2, MAX_FALLBACK);
148 : }
149 97 : if (deviceId.empty())
150 61 : cloneBackoffAllDevices = true;
151 : else
152 36 : cloneFailedDevices.emplace(deviceId);
153 97 : fallbackClone->expires_at(nextCloneAttempt);
154 97 : return std::chrono::ceil<std::chrono::seconds>(nextCloneAttempt - now);
155 : }
156 :
157 852 : void resetCloneRetry()
158 : {
159 : // conversation mtx must be locked
160 852 : nextCloneAttempt = {};
161 852 : fallbackTimer = 5s;
162 852 : cloneFailedDevices.clear();
163 852 : cloneBackoffAllDevices = false;
164 852 : }
165 :
166 908 : SyncedConversation(const std::string& convId)
167 908 : : info {convId}
168 : {
169 908 : fallbackClone = std::make_unique<asio::steady_timer>(*Manager::instance().ioContext());
170 908 : }
171 48 : SyncedConversation(const ConvInfo& info)
172 48 : : info {info}
173 : {
174 48 : fallbackClone = std::make_unique<asio::steady_timer>(*Manager::instance().ioContext());
175 48 : }
176 :
177 3076 : bool startFetch(const std::string& deviceId, bool checkIfConv = false)
178 : {
179 : // conversation mtx must be locked
180 3076 : if (checkIfConv && conversation)
181 3 : return false; // Already a conversation
182 3073 : if (pending) {
183 743 : if (pending->ready)
184 59 : return false; // Already doing stuff
185 684 : if (pending->connectingTo.find(deviceId) != pending->connectingTo.end())
186 500 : return false; // Already connecting to this device
187 : } else {
188 2330 : pending = std::make_unique<PendingConversationFetch>();
189 : }
190 2514 : pending->connectingTo.insert(deviceId);
191 2514 : return true;
192 : }
193 :
194 1956 : void stopFetch(const std::string& deviceId)
195 : {
196 : // conversation mtx must be locked
197 1956 : if (!pending)
198 6 : return;
199 1950 : pending->connectingTo.erase(deviceId);
200 1950 : if (pending->connectingTo.empty())
201 1866 : pending.reset();
202 : }
203 :
204 284 : void deferFetch(const std::string& peer, const std::string& deviceId, const std::string& commitId)
205 : {
206 : // conversation mtx must be locked
207 284 : deferredFetches[deviceId] = {peer, deviceId, commitId};
208 568 : }
209 :
210 1890 : std::optional<DeferredFetch> finishFetch(const std::string& deviceId)
211 : {
212 : // conversation mtx must be locked
213 1890 : stopFetch(deviceId);
214 1890 : auto deferred = deferredFetches.find(deviceId);
215 1890 : if (deferred == deferredFetches.end())
216 1613 : return std::nullopt;
217 277 : auto request = std::move(deferred->second);
218 277 : deferredFetches.erase(deferred);
219 277 : return request;
220 277 : }
221 :
222 36 : void clearFetches()
223 : {
224 : // conversation mtx must be locked
225 36 : pending.reset();
226 36 : deferredFetches.clear();
227 36 : }
228 :
229 361 : std::vector<std::map<std::string, std::string>> getMembers(bool includeLeft, bool includeBanned) const
230 : {
231 : // conversation mtx must be locked
232 361 : if (conversation)
233 285 : return conversation->getMembers(true, includeLeft, includeBanned);
234 : // If we're cloning, we can return the initial members
235 76 : std::vector<std::map<std::string, std::string>> result;
236 76 : result.reserve(info.members.size());
237 228 : for (const auto& uri : info.members) {
238 456 : result.emplace_back(std::map<std::string, std::string> {{"uri", uri}});
239 : }
240 76 : return result;
241 228 : }
242 : };
243 :
244 : class ConversationModule::Impl : public std::enable_shared_from_this<Impl>
245 : {
246 : public:
247 : Impl(std::shared_ptr<JamiAccount>&& account,
248 : std::shared_ptr<AccountManager>&& accountManager,
249 : NeedsSyncingCb&& needsSyncingCb,
250 : SengMsgCb&& sendMsgCb,
251 : NeedSocketCb&& onNeedSocket,
252 : NeedSocketCb&& onNeedSwarmSocket,
253 : OneToOneRecvCb&& oneToOneRecvCb);
254 :
255 : template<typename S, typename T>
256 353 : inline auto withConv(const S& convId, T&& cb) const
257 : {
258 706 : if (auto conv = getConversation(convId)) {
259 353 : std::lock_guard lk(conv->mtx);
260 353 : return cb(*conv);
261 353 : } else {
262 0 : JAMI_WARNING("Conversation {} not found", convId);
263 : }
264 0 : return decltype(cb(std::declval<SyncedConversation&>()))();
265 : }
266 : template<typename S, typename T>
267 3476 : inline auto withConversation(const S& convId, T&& cb)
268 : {
269 6952 : if (auto conv = getConversation(convId)) {
270 3468 : std::lock_guard lk(conv->mtx);
271 3468 : if (conv->conversation)
272 3366 : return cb(*conv->conversation);
273 3468 : } else {
274 8 : JAMI_WARNING("Conversation {} not found", convId);
275 : }
276 110 : return decltype(cb(std::declval<Conversation&>()))();
277 : }
278 :
279 : // Retrieving recent commits
280 : /**
281 : * Clone a conversation (initial) from device
282 : * @param deviceId
283 : * @param convId
284 : */
285 : void cloneConversation(const std::string& deviceId, const std::string& peer, const std::string& convId);
286 : void cloneConversation(const std::string& deviceId,
287 : const std::string& peer,
288 : const std::shared_ptr<SyncedConversation>& conv);
289 :
290 : /**
291 : * Pull remote device
292 : * @param peer Contact URI
293 : * @param deviceId Contact's device
294 : * @param conversationId
295 : * @param commitId (optional)
296 : */
297 : void fetchNewCommits(const std::string& peer,
298 : const std::string& deviceId,
299 : const std::string& conversationId,
300 : const std::string& commitId = "");
301 : /**
302 : * Handle events to receive new commits
303 : */
304 : void handlePendingConversation(const std::string& conversationId, const std::string& deviceId);
305 :
306 : // Requests
307 : std::optional<ConversationRequest> getRequest(const std::string& id) const;
308 :
309 : // Conversations
310 : /**
311 : * Get members
312 : * @param conversationId
313 : * @param includeBanned
314 : * @return a map of members with their role and details
315 : */
316 : std::vector<std::map<std::string, std::string>> getConversationMembers(const std::string& conversationId,
317 : bool includeBanned = false) const;
318 : void setConversationMembers(const std::string& convId, const std::set<std::string>& members);
319 :
320 : /**
321 : * Remove a repository and all files
322 : * @param convId
323 : * @param sync If we send an update to other account's devices
324 : * @param force True if ignore the removing flag
325 : */
326 : void removeRepository(const std::string& convId, bool sync, bool force = false);
327 : void removeRepositoryImpl(SyncedConversation& conv, bool sync, bool force = false);
328 : /**
329 : * Remove a conversation
330 : * @param conversationId
331 : */
332 : bool removeConversation(const std::string& conversationId, bool forceRemove = false);
333 : bool removeConversationImpl(SyncedConversation& conv, bool forceRemove = false);
334 : void removeDocumentReplica(const std::string& documentId);
335 : std::vector<std::string> childDocuments(const std::vector<std::string>& parents) const;
336 :
337 : /**
338 : * Send a message notification to all members
339 : * @param conversation
340 : * @param commit
341 : * @param sync If we send an update to other account's devices
342 : * @param deviceId If we need to filter a specific device
343 : */
344 : void sendMessageNotification(const std::string& conversationId,
345 : bool sync,
346 : const std::string& commitId = "",
347 : const std::string& deviceId = "",
348 : bool bootstrapReplay = false);
349 : void sendMessageNotification(Conversation& conversation,
350 : bool sync,
351 : const std::string& commitId = "",
352 : const std::string& deviceId = "",
353 : bool bootstrapReplay = false);
354 :
355 : /**
356 : * @return if a convId is a valid conversation (repository cloned & usable)
357 : */
358 191 : bool isConversation(const std::string& convId) const
359 : {
360 191 : std::lock_guard lk(conversationsMtx_);
361 191 : auto c = conversations_.find(convId);
362 382 : return c != conversations_.end() && c->second;
363 191 : }
364 :
365 : /**
366 : * @return true if convId should ignore new invites/requests, except for removed
367 : * group conversations.
368 : */
369 834 : bool isActiveOrNonReinvitableConversation(const std::string& convId) const
370 : {
371 834 : auto conv = getConversation(convId);
372 834 : if (!conv)
373 710 : return false;
374 124 : std::lock_guard lk(conv->mtx);
375 124 : return conv->info.mode == ConversationMode::ONE_TO_ONE || !conv->info.isRemoved();
376 834 : }
377 :
378 4514 : void addConvInfo(const ConvInfo& info)
379 : {
380 4514 : std::lock_guard lk(convInfosMtx_);
381 4514 : convInfos_[info.id] = info;
382 4514 : saveConvInfos();
383 4514 : }
384 :
385 : std::string getOneToOneConversation(const std::string& uri) const noexcept;
386 :
387 : bool updateConvForContact(const std::string& uri, const std::string& oldConv, const std::string& newConv);
388 :
389 : void linkReplacementConversation(Conversation& conversation);
390 :
391 1187 : std::shared_ptr<SyncedConversation> getConversation(std::string_view convId) const
392 : {
393 1187 : std::lock_guard lk(conversationsMtx_);
394 1187 : auto c = conversations_.find(convId);
395 2374 : return c != conversations_.end() ? c->second : nullptr;
396 1187 : }
397 42565 : std::shared_ptr<SyncedConversation> getConversation(std::string_view convId)
398 : {
399 42565 : std::lock_guard lk(conversationsMtx_);
400 42565 : auto c = conversations_.find(convId);
401 85130 : return c != conversations_.end() ? c->second : nullptr;
402 42565 : }
403 1311 : std::shared_ptr<SyncedConversation> startConversation(const std::string& convId)
404 : {
405 1311 : std::lock_guard lk(conversationsMtx_);
406 1311 : auto& c = conversations_[convId];
407 1311 : if (!c)
408 883 : c = std::make_shared<SyncedConversation>(convId);
409 2622 : return c;
410 1311 : }
411 175 : std::shared_ptr<SyncedConversation> startConversation(const ConvInfo& info)
412 : {
413 175 : std::lock_guard lk(conversationsMtx_);
414 175 : auto& c = conversations_[info.id];
415 175 : if (!c)
416 16 : c = std::make_shared<SyncedConversation>(info);
417 350 : return c;
418 175 : }
419 6373 : std::vector<std::shared_ptr<SyncedConversation>> getSyncedConversations() const
420 : {
421 6373 : std::lock_guard lk(conversationsMtx_);
422 6373 : std::vector<std::shared_ptr<SyncedConversation>> result;
423 6373 : result.reserve(conversations_.size());
424 13350 : for (const auto& [_, c] : conversations_)
425 6977 : result.emplace_back(c);
426 12746 : return result;
427 6373 : }
428 3146 : std::vector<std::shared_ptr<Conversation>> getConversations() const
429 : {
430 3146 : auto conversations = getSyncedConversations();
431 3146 : std::vector<std::shared_ptr<Conversation>> result;
432 3146 : result.reserve(conversations.size());
433 6734 : for (const auto& sc : conversations) {
434 3588 : std::lock_guard lk(sc->mtx);
435 3588 : if (sc->conversation)
436 3085 : result.emplace_back(sc->conversation);
437 3588 : }
438 6292 : return result;
439 3146 : }
440 :
441 : void createCommit(const std::string& conversationId,
442 : CommitMessage&& message,
443 : bool announce = true,
444 : OnCommitCb&& onCommit = {},
445 : OnDoneCb&& cb = {});
446 :
447 : void sendMessage(const std::string& conversationId,
448 : std::string message,
449 : const std::string& replyTo = "",
450 : bool announce = true,
451 : OnCommitCb&& onCommit = {},
452 : OnDoneCb&& cb = {});
453 :
454 : void editMessage(const std::string& conversationId, const std::string& newBody, const std::string& editedId);
455 :
456 : void bootstrapCb(std::string convId);
457 :
458 : // The following methods modify what is stored on the disk
459 : /**
460 : * @note convInfosMtx_ should be locked
461 : */
462 5575 : void saveConvInfos() const { ConversationModule::saveConvInfos(accountId_, convInfos_); }
463 : /**
464 : * @note conversationsRequestsMtx_ should be locked
465 : */
466 1026 : void saveConvRequests() const { ConversationModule::saveConvRequests(accountId_, conversationsRequests_); }
467 : void declineOtherConversationWith(const std::string& uri);
468 507 : bool addConversationRequest(const std::string& id, const ConversationRequest& req)
469 : {
470 : // conversationsRequestsMtx_ MUST BE LOCKED
471 507 : if (isActiveOrNonReinvitableConversation(id))
472 0 : return false;
473 507 : auto it = conversationsRequests_.find(id);
474 507 : if (it != conversationsRequests_.end()) {
475 69 : const bool existingDeclined = it->second.declined != TimePoint {};
476 69 : const bool incomingDeclined = req.declined != TimePoint {};
477 69 : if (incomingDeclined && existingDeclined) {
478 : // Keep only the latest decline and avoid sync ping-pong.
479 4 : if (req.declined <= it->second.declined)
480 4 : return false;
481 0 : conversationsRequests_[id] = req;
482 0 : saveConvRequests();
483 0 : return false;
484 65 : } else if (incomingDeclined) {
485 : // Accept a decline only if it is newer than the active request.
486 2 : if (req.declined < it->second.received)
487 1 : return false;
488 63 : } else if (existingDeclined) {
489 : // Accept a re-invitation only if it arrived after the previous decline.
490 9 : if (req.received <= it->second.declined)
491 8 : return false;
492 : } else {
493 : // Ignore duplicate active requests.
494 54 : return false;
495 : }
496 438 : } else if (req.isOneToOne()) {
497 : // Check that we're not adding a second one to one trust request
498 : // NOTE: If a new one to one request is received, we can decline the previous one.
499 269 : declineOtherConversationWith(req.from);
500 : }
501 440 : JAMI_DEBUG("[Account {}] [Conversation {}] Adding conversation request from {}", accountId_, id, req.from);
502 440 : conversationsRequests_[id] = req;
503 440 : saveConvRequests();
504 440 : return true;
505 : }
506 576 : void rmConversationRequest(const std::string& id)
507 : {
508 : // conversationsRequestsMtx_ MUST BE LOCKED
509 576 : auto it = conversationsRequests_.find(id);
510 576 : if (it != conversationsRequests_.end()) {
511 404 : auto& md = syncingMetadatas_[id];
512 404 : md = it->second.metadatas;
513 404 : md["syncing"] = "true";
514 1212 : md["created"] = std::to_string(toSecondsSinceEpoch(it->second.received));
515 : }
516 576 : saveMetadata();
517 576 : conversationsRequests_.erase(id);
518 576 : saveConvRequests();
519 576 : }
520 :
521 : std::weak_ptr<JamiAccount> account_;
522 : std::shared_ptr<AccountManager> accountManager_;
523 : const std::string accountId_ {};
524 : NeedsSyncingCb needsSyncingCb_;
525 : SengMsgCb sendMsgCb_;
526 : NeedSocketCb onNeedSocket_;
527 : NeedSocketCb onNeedSwarmSocket_;
528 : OneToOneRecvCb oneToOneRecvCb_;
529 :
530 : std::string deviceId_ {};
531 : std::string username_ {};
532 :
533 : // Requests
534 : mutable std::mutex conversationsRequestsMtx_;
535 : std::map<std::string, ConversationRequest> conversationsRequests_;
536 :
537 : // Conversations
538 : mutable std::mutex conversationsMtx_ {};
539 : std::map<std::string, std::shared_ptr<SyncedConversation>, std::less<>> conversations_;
540 :
541 : // The following information are stored on the disk
542 : mutable std::mutex convInfosMtx_; // Note, should be locked after conversationsMtx_ if needed
543 : std::map<std::string, ConvInfo> convInfos_;
544 :
545 : // When sending a new message, we need to send the notification to some peers of the
546 : // conversation However, the conversation may be not bootstrapped, so the list will be empty.
547 : // notSyncedNotification_ will store the notifiaction to announce until we have peers to sync
548 : // with.
549 : std::mutex notSyncedNotificationMtx_;
550 : std::map<std::string, std::string> notSyncedNotification_;
551 :
552 4941 : std::weak_ptr<Impl> weak() { return std::static_pointer_cast<Impl>(shared_from_this()); }
553 :
554 : std::mutex refreshMtx_;
555 : std::map<std::string, uint64_t> refreshMessage;
556 : std::atomic_int syncCnt {0};
557 :
558 : #ifdef LIBJAMI_TEST
559 : std::function<void(std::string, Conversation::BootstrapStatus)> bootstrapCbTest_;
560 : std::function<void(const std::string&, const std::string&, bool)> fetchCompletedCbTest_;
561 : std::function<void(const std::string&)> cloneReadyCbTest_;
562 : #endif
563 :
564 : uint64_t presenceListenerToken_ {0};
565 : void onBuddyOnline(const std::string& uri);
566 :
567 873 : ~Impl()
568 873 : {
569 873 : if (auto acc = account_.lock()) {
570 4 : if (auto pm = acc->presenceManager()) {
571 4 : if (presenceListenerToken_)
572 0 : pm->removeListener(presenceListenerToken_);
573 : }
574 873 : }
575 873 : }
576 :
577 : void fixStructures(std::shared_ptr<JamiAccount> account,
578 : const std::vector<std::tuple<std::string, std::string, std::string>>& updateContactConv,
579 : const std::set<std::string>& toRm);
580 :
581 : void cloneConversationFrom(const std::shared_ptr<SyncedConversation> conv, const std::string& deviceId);
582 : void bootstrap(const std::string& convId);
583 : void fallbackClone(const asio::error_code& ec, const std::string& conversationId);
584 :
585 : void cloneConversationFrom(const ConversationRequest& request);
586 :
587 : void cloneConversationFrom(const std::string& conversationId, const std::string& uri);
588 :
589 : // While syncing, we do not want to lose metadata (avatar/title and mode)
590 : std::map<std::string, std::map<std::string, std::string>> syncingMetadatas_;
591 1016 : void saveMetadata()
592 : {
593 1016 : auto path = fileutils::get_data_dir() / accountId_;
594 1016 : std::lock_guard lock(dhtnet::fileutils::getFileLock(path / "syncingMetadatas"));
595 1016 : std::ofstream file(path / "syncingMetadatas", std::ios::trunc | std::ios::binary);
596 1016 : msgpack::pack(file, syncingMetadatas_);
597 1016 : }
598 :
599 873 : void loadMetadata()
600 : {
601 : try {
602 : // read file
603 873 : auto path = fileutils::get_data_dir() / accountId_;
604 873 : std::lock_guard lock(dhtnet::fileutils::getFileLock(path / "syncingMetadatas"));
605 1744 : auto file = fileutils::loadFile("syncingMetadatas", path);
606 : // load values
607 2 : msgpack::unpacked result;
608 2 : msgpack::unpack(result, (const char*) file.data(), file.size(), 0);
609 2 : result.get().convert(syncingMetadatas_);
610 2615 : } catch (const std::exception& e) {
611 871 : JAMI_WARNING("[Account {}] [ConversationModule] unable to load syncing metadata: {}", accountId_, e.what());
612 871 : }
613 873 : }
614 :
615 18 : void initPresence()
616 : {
617 18 : if (!presenceListenerToken_)
618 16 : if (auto acc = account_.lock())
619 32 : presenceListenerToken_ = acc->presenceManager()->addListener(
620 32 : [w = weak_from_this()](const std::string& uri, bool online) {
621 5 : if (!online)
622 0 : return;
623 5 : if (auto sthis = w.lock())
624 5 : sthis->onBuddyOnline(uri);
625 16 : });
626 18 : }
627 : };
628 :
629 873 : ConversationModule::Impl::Impl(std::shared_ptr<JamiAccount>&& account,
630 : std::shared_ptr<AccountManager>&& accountManager,
631 : NeedsSyncingCb&& needsSyncingCb,
632 : SengMsgCb&& sendMsgCb,
633 : NeedSocketCb&& onNeedSocket,
634 : NeedSocketCb&& onNeedSwarmSocket,
635 873 : OneToOneRecvCb&& oneToOneRecvCb)
636 873 : : account_(account)
637 873 : , accountManager_(accountManager)
638 873 : , accountId_(account->getAccountID())
639 873 : , needsSyncingCb_(needsSyncingCb)
640 873 : , sendMsgCb_(sendMsgCb)
641 873 : , onNeedSocket_(onNeedSocket)
642 873 : , onNeedSwarmSocket_(onNeedSwarmSocket)
643 1746 : , oneToOneRecvCb_(oneToOneRecvCb)
644 : {
645 873 : if (auto accm = account->accountManager())
646 873 : if (const auto* info = accm->getInfo()) {
647 873 : deviceId_ = info->deviceId;
648 873 : username_ = info->accountId;
649 873 : }
650 873 : conversationsRequests_ = convRequests(accountId_);
651 873 : loadMetadata();
652 873 : }
653 :
654 : void
655 256 : ConversationModule::Impl::cloneConversation(const std::string& deviceId,
656 : const std::string& peerUri,
657 : const std::string& convId)
658 : {
659 256 : JAMI_DEBUG("[Account {}] [Conversation {}] [device {}] Cloning conversation", accountId_, convId, deviceId);
660 :
661 256 : auto conv = startConversation(convId);
662 256 : std::unique_lock lk(conv->mtx);
663 256 : cloneConversation(deviceId, peerUri, conv);
664 256 : }
665 :
666 : void
667 342 : ConversationModule::Impl::cloneConversation(const std::string& deviceId,
668 : const std::string& peerUri,
669 : const std::shared_ptr<SyncedConversation>& conv)
670 : {
671 : // conv->mtx must be locked
672 342 : if (conv->info.mode == ConversationMode::DOCUMENT && conv->info.isRemoved())
673 0 : return;
674 342 : if (conv->isUnrecoverable()) {
675 0 : JAMI_WARNING("[Account {}] [Conversation {}] [device {}] Conversation is marked unrecoverable, "
676 : "ignoring clone request",
677 : accountId_,
678 : conv->info.id,
679 : deviceId);
680 0 : return;
681 : }
682 : // Checked whether or not a fetch is pending: a device that already failed in this
683 : // window stays out even while another one is being tried, and one that did not is
684 : // let in either way -- startFetch() arbitrates the round.
685 342 : if (!conv->conversation && conv->cloneThrottled(deviceId)) {
686 33 : JAMI_DEBUG("[Account {}] [Conversation {}] [device {}] Clone retry is not due yet, ignoring "
687 : "clone request",
688 : accountId_,
689 : conv->info.id,
690 : deviceId);
691 33 : return;
692 : }
693 309 : if (!conv->conversation) {
694 : // Note: here we don't return and connect to all members
695 : // the first that will successfully connect will be used for
696 : // cloning.
697 : // This avoid the case when we try to clone from convInfos + sync message
698 : // at the same time.
699 309 : if (!conv->startFetch(deviceId, true)) {
700 256 : JAMI_WARNING("[Account {}] [Conversation {}] [device {}] Already fetching conversation",
701 : accountId_,
702 : conv->info.id,
703 : deviceId);
704 256 : addConvInfo(conv->info);
705 256 : return;
706 : }
707 159 : onNeedSocket_(
708 53 : conv->info.id,
709 : deviceId,
710 106 : [w = weak(), conv, deviceId](const auto& channel) {
711 53 : std::lock_guard lk(conv->mtx);
712 53 : if (conv->pending && !conv->pending->ready) {
713 53 : if (channel) {
714 53 : conv->pending->ready = true;
715 53 : conv->pending->deviceId = channel->deviceId().toString();
716 53 : conv->pending->socket = channel;
717 53 : if (!conv->pending->cloning) {
718 53 : conv->pending->cloning = true;
719 106 : dht::ThreadPool::io().run([w, convId = conv->info.id, deviceId = conv->pending->deviceId]() {
720 106 : if (auto sthis = w.lock())
721 53 : sthis->handlePendingConversation(convId, deviceId);
722 : });
723 : }
724 53 : return true;
725 : } else {
726 0 : conv->stopFetch(deviceId);
727 : }
728 : }
729 0 : return false;
730 53 : },
731 : MIME_TYPE_GIT,
732 : false);
733 :
734 53 : JAMI_LOG("[Account {}] [Conversation {}] [device {}] Requesting device", accountId_, conv->info.id, deviceId);
735 53 : conv->info.members.emplace(username_);
736 53 : conv->info.members.emplace(peerUri);
737 53 : addConvInfo(conv->info);
738 : } else {
739 0 : JAMI_DEBUG("[Account {}] [Conversation {}] Conversation already cloned", accountId_, conv->info.id);
740 : }
741 : }
742 :
743 : void
744 13041 : ConversationModule::Impl::fetchNewCommits(const std::string& peer,
745 : const std::string& deviceId,
746 : const std::string& conversationId,
747 : const std::string& commitId)
748 : {
749 13041 : auto conv = getConversation(conversationId);
750 : {
751 13041 : bool needReclone = false;
752 13041 : std::unique_lock lkInfos(convInfosMtx_);
753 :
754 13041 : auto itConvInfo = convInfos_.find(conversationId);
755 13041 : if (itConvInfo != convInfos_.end() && itConvInfo->second.isRemoved()) {
756 2 : if (!conv)
757 0 : return;
758 :
759 2 : const bool isOneToOne = (itConvInfo->second.mode == ConversationMode::ONE_TO_ONE);
760 2 : auto contactInfo = accountManager_->getContactInfo(peer);
761 0 : const bool shouldReadd = isOneToOne && contactInfo && contactInfo->confirmed && contactInfo->isActive()
762 0 : && !contactInfo->isBanned() && contactInfo->added > itConvInfo->second.removed
763 2 : && contactInfo->conversationId != conversationId;
764 :
765 2 : if (shouldReadd) {
766 0 : if (conv) {
767 0 : std::unique_lock lkSynced(conv->mtx);
768 :
769 0 : if (!conv->conversation) {
770 0 : conv->info.created = nowMs();
771 0 : conv->info.erased = TimePoint {};
772 0 : convInfos_[conversationId] = conv->info;
773 0 : saveConvInfos();
774 0 : needReclone = true;
775 : }
776 0 : }
777 :
778 0 : lkInfos.unlock();
779 :
780 0 : if (needReclone && conv) {
781 : {
782 0 : std::unique_lock lkSynced(conv->mtx);
783 0 : cloneConversation(deviceId, peer, conv);
784 0 : }
785 : }
786 :
787 0 : return;
788 : }
789 :
790 2 : JAMI_WARNING("[Account {:s}] [Conversation {}] Received a commit, but conversation is removed",
791 : accountId_,
792 : conversationId);
793 2 : return;
794 2 : }
795 13041 : }
796 13039 : std::optional<ConversationRequest> oldReq;
797 : {
798 13039 : std::lock_guard lk(conversationsRequestsMtx_);
799 13039 : oldReq = getRequest(conversationId);
800 13039 : if (oldReq != std::nullopt && oldReq->declined != TimePoint {}) {
801 0 : JAMI_DEBUG("[Account {}] [Conversation {}] Received a request for a conversation already declined.",
802 : accountId_,
803 : conversationId);
804 0 : return;
805 : }
806 13039 : }
807 13039 : if (!commitId.empty())
808 12351 : JAMI_DEBUG("[Account {:s}] [Conversation {}] [device {}] fetching '{:s}'",
809 : accountId_,
810 : conversationId,
811 : deviceId,
812 : commitId);
813 :
814 13039 : const bool shouldRequestInvite = username_ != peer;
815 13039 : if (!conv) {
816 186 : if (oldReq == std::nullopt && shouldRequestInvite) {
817 : // A commit for a repository nothing is known about. If it names a
818 : // document announced in some conversation, its announcement drives
819 : // replication; it must not become a separate conversation invite.
820 159 : if (auto acc = account_.lock()) {
821 159 : if (acc->collaborativeEditing()->knowsDocument(conversationId))
822 0 : return;
823 159 : }
824 : // We didn't find a conversation or a request with the given ID.
825 : // This suggests that someone tried to send us an invitation but
826 : // that we didn't receive it, so we ask for a new one.
827 159 : JAMI_WARNING("[Account {}] [Conversation {}] Unable to find conversation, asking for an invite",
828 : accountId_,
829 : conversationId);
830 477 : sendMsgCb_(peer, {}, std::map<std::string, std::string> {{MIME_TYPE_INVITE, conversationId}}, 0);
831 : }
832 186 : return;
833 : }
834 12853 : std::unique_lock lk(conv->mtx);
835 :
836 12853 : if (conv->conversation) {
837 : // Check if we already have the commit
838 12813 : if (not commitId.empty() && conv->conversation->hasCommit(commitId)) {
839 10923 : return;
840 : }
841 2179 : if (conv->conversation->isRemoving()) {
842 1 : JAMI_WARNING("[Account {}] [Conversation {}] conversaton is being removed", accountId_, conversationId);
843 1 : return;
844 : }
845 2178 : if (!conv->conversation->isPeerAuthorized(peer, deviceId, true)) {
846 3 : JAMI_WARNING("[Account {}] [Conversation {}] device {} is not authorized for {}",
847 : accountId_,
848 : conversationId,
849 : deviceId,
850 : peer);
851 3 : return;
852 : }
853 :
854 : // Retrieve current last message
855 2175 : auto lastMessageId = conv->conversation->lastCommitId();
856 2175 : if (lastMessageId.empty()) {
857 1 : JAMI_ERROR("[Account {}] [Conversation {}] No message detected. This is a bug", accountId_, conversationId);
858 1 : return;
859 : }
860 :
861 2174 : if (!conv->startFetch(deviceId)) {
862 284 : conv->deferFetch(peer, deviceId, commitId);
863 284 : JAMI_WARNING("[Account {}] [Conversation {}] Already fetching from {}, deferring commit {}",
864 : accountId_,
865 : conversationId,
866 : deviceId,
867 : commitId);
868 284 : return;
869 : }
870 :
871 1890 : syncCnt.fetch_add(1);
872 5670 : onNeedSocket_(
873 : conversationId,
874 : deviceId,
875 3780 : [w = weak(), conv, conversationId, peer = std::move(peer), deviceId, commitId = std::move(commitId)](
876 : const auto& channel) {
877 1890 : auto sthis = w.lock();
878 1890 : auto acc = sthis ? sthis->account_.lock() : nullptr;
879 1890 : std::unique_lock lk(conv->mtx);
880 1890 : auto conversation = conv->conversation;
881 1890 : if (!channel || !acc || !conversation) {
882 30 : auto deferred = conv->finishFetch(deviceId);
883 30 : lk.unlock();
884 30 : if (sthis && deferred)
885 0 : sthis->fetchNewCommits(deferred->peer, deferred->deviceId, conversationId, deferred->commitId);
886 60 : if (sthis && sthis->syncCnt.fetch_sub(1) == 1)
887 22 : emitSignal<libjami::ConversationSignal::ConversationSyncFinished>(sthis->accountId_);
888 30 : return false;
889 30 : }
890 1860 : conversation->addGitSocket(channel->deviceId(), channel);
891 1860 : lk.unlock();
892 3720 : conversation->sync(
893 1860 : peer,
894 1860 : deviceId,
895 3720 : [w, conv, conversationId = std::move(conversationId), peer, deviceId, commitId](bool ok) {
896 1860 : auto shared = w.lock();
897 1860 : if (!shared)
898 0 : return;
899 : #ifdef LIBJAMI_TEST
900 1860 : if (shared->fetchCompletedCbTest_)
901 2 : shared->fetchCompletedCbTest_(conversationId, commitId, ok);
902 : #endif
903 1860 : if (!ok) {
904 67 : JAMI_WARNING("[Account {}] [Conversation {}] Unable to fetch new commit from "
905 : "{}, other peer may be disconnected",
906 : shared->accountId_,
907 : conversationId,
908 : deviceId);
909 67 : JAMI_LOG("[Account {}] [Conversation {}] Relaunch sync with {}",
910 : shared->accountId_,
911 : conversationId,
912 : deviceId);
913 : }
914 :
915 1860 : std::optional<SyncedConversation::DeferredFetch> deferred;
916 : {
917 1860 : std::lock_guard lk(conv->mtx);
918 1860 : deferred = conv->finishFetch(deviceId);
919 : // Notify peers that a new commit is there (DRT)
920 1860 : if (not commitId.empty() && ok && conv->conversation && !conv->info.isRemoved()) {
921 1156 : shared->sendMessageNotification(*conv->conversation, false, commitId, deviceId);
922 : }
923 1860 : }
924 1860 : if (deferred)
925 554 : shared->fetchNewCommits(deferred->peer,
926 277 : deferred->deviceId,
927 277 : conversationId,
928 277 : deferred->commitId);
929 3720 : if (shared->syncCnt.fetch_sub(1) == 1) {
930 1775 : emitSignal<libjami::ConversationSignal::ConversationSyncFinished>(shared->accountId_);
931 : }
932 1860 : },
933 1860 : commitId);
934 1860 : return true;
935 1890 : },
936 : "",
937 : false);
938 2175 : } else {
939 40 : if (oldReq != std::nullopt)
940 0 : return;
941 40 : if (conv->pending)
942 30 : return;
943 10 : bool clone = !conv->info.isRemoved();
944 10 : if (clone) {
945 10 : cloneConversation(deviceId, peer, conv);
946 10 : return;
947 : }
948 0 : if (!shouldRequestInvite)
949 0 : return;
950 0 : lk.unlock();
951 0 : JAMI_WARNING("[Account {}] [Conversation {}] Unable to find conversation, asking for an invite",
952 : accountId_,
953 : conversationId);
954 0 : sendMsgCb_(peer, {}, std::map<std::string, std::string> {{MIME_TYPE_INVITE, conversationId}}, 0);
955 : }
956 35312 : }
957 :
958 : void
959 5 : ConversationModule::Impl::onBuddyOnline(const std::string& uri)
960 : {
961 5 : auto acc = account_.lock();
962 5 : if (!acc)
963 0 : return;
964 :
965 5 : auto toUpdate = std::make_shared<std::vector<std::shared_ptr<Conversation>>>();
966 11 : for (auto& conv : getConversations()) {
967 6 : if (conv->isMember(uri)) {
968 4 : toUpdate->emplace_back(std::move(conv));
969 : }
970 5 : }
971 :
972 5 : if (auto pm = acc->presenceManager()) {
973 5 : auto devices = pm->getDevices(uri);
974 9 : for (const auto& conv : *toUpdate)
975 4 : conv->addKnownDevices(devices, uri);
976 5 : }
977 5 : }
978 :
979 : // Clone and store conversation
980 : void
981 458 : ConversationModule::Impl::handlePendingConversation(const std::string& conversationId, const std::string& deviceId)
982 : {
983 458 : auto acc = account_.lock();
984 458 : if (!acc)
985 0 : return;
986 458 : auto conv = getConversation(conversationId);
987 458 : if (!conv)
988 0 : return;
989 458 : std::unique_lock lk(conv->mtx, std::defer_lock);
990 860 : auto erasePending = [&] {
991 860 : conv->pending.reset();
992 860 : lk.unlock();
993 1318 : };
994 : try {
995 458 : auto conversation = std::make_shared<Conversation>(acc, deviceId, conversationId);
996 : #ifdef LIBJAMI_TEST
997 427 : if (cloneReadyCbTest_)
998 4 : cloneReadyCbTest_(conversationId);
999 : #endif
1000 427 : conversation->onMembersChanged([w = weak_from_this(), conversationId](const auto& members) {
1001 : // Delay in another thread to avoid deadlocks
1002 3318 : dht::ThreadPool::io().run([w, conversationId, members = std::move(members)] {
1003 3318 : if (auto sthis = w.lock())
1004 1659 : sthis->setConversationMembers(conversationId, members);
1005 : });
1006 1659 : });
1007 427 : conversation->onMessageStatusChanged([this, conversationId](const auto& status) {
1008 733 : auto msg = std::make_shared<SyncMsg>();
1009 1466 : msg->ms = {{conversationId, status}};
1010 733 : needsSyncingCb_(std::move(msg));
1011 1466 : });
1012 427 : conversation->onNeedSocket(onNeedSwarmSocket_);
1013 427 : if (!conversation->isMember(username_, true)) {
1014 0 : JAMI_ERROR("[Account {}] [Conversation {}] Conversation cloned but we do not seem to be a valid member",
1015 : accountId_,
1016 : conversationId);
1017 0 : conversation->erase();
1018 0 : lk.lock();
1019 0 : erasePending();
1020 0 : return;
1021 : }
1022 :
1023 : // Make sure that the list of members stored in convInfos_ matches the
1024 : // one from the conversation's repository.
1025 : // (https://git.jami.net/savoirfairelinux/jami-daemon/-/issues/1026)
1026 427 : setConversationMembers(conversationId, conversation->memberUris("", {}));
1027 :
1028 : // Map lookups must precede locking the parent. A leave locks that parent
1029 : // too, so checking and installing the child are atomic with respect to it.
1030 427 : const bool isDocument = conversation->mode() == ConversationMode::DOCUMENT;
1031 427 : auto parent = isDocument ? getConversation(conversation->parentConversationId()) : nullptr;
1032 427 : std::unique_lock<std::mutex> parentLock;
1033 427 : if (parent && parent != conv)
1034 25 : parentLock = std::unique_lock(parent->mtx);
1035 427 : lk.lock();
1036 : // Pending downloads saved by older versions have no parent field.
1037 25 : if (isDocument && conv->info.parent.empty() && parent && parent != conv
1038 452 : && parent->canReplicateDocuments(username_) && parent->announcesDocument(conversationId)) {
1039 0 : conv->info.parent = conversation->parentConversationId();
1040 0 : addConvInfo(conv->info);
1041 : }
1042 427 : if (isDocument
1043 451 : && (!parent || parent == conv || !parent->canReplicateDocuments(username_)
1044 451 : || conv->info.parent != conversation->parentConversationId()
1045 24 : || !parent->announcesDocument(conversationId))) {
1046 2 : JAMI_WARNING("[Account {}] Discarding document {}: parent is inactive or no longer announces it",
1047 : accountId_,
1048 : conversationId);
1049 2 : conversation->erase();
1050 2 : conv->info.removed = std::max(nowMs(), conv->info.created);
1051 2 : conv->info.erased = conv->info.removed;
1052 2 : addConvInfo(conv->info);
1053 2 : erasePending();
1054 2 : return;
1055 : }
1056 425 : if (conv->info.mode != conversation->mode()) {
1057 0 : JAMI_ERROR(
1058 : "[Account {}] [Conversation {}] Cloned conversation mode is {}, but {} was expected from invite.",
1059 : accountId_,
1060 : conversationId,
1061 : static_cast<int>(conversation->mode()),
1062 : static_cast<int>(conv->info.mode));
1063 : // TODO: erase conversation after transition period
1064 : // conversation->erase();
1065 : // erasePending();
1066 : // return;
1067 0 : conv->info.mode = conversation->mode();
1068 0 : addConvInfo(conv->info);
1069 : }
1070 :
1071 425 : if (conv->pending && conv->pending->socket)
1072 423 : conversation->addGitSocket(DeviceId(deviceId), std::move(conv->pending->socket));
1073 425 : auto removeRepo = false;
1074 : // Note: a removeContact while cloning. In this case, the conversation
1075 : // must not be announced and removed.
1076 425 : if (conv->info.isRemoved())
1077 1 : removeRepo = true;
1078 425 : std::map<std::string, std::string> preferences;
1079 425 : std::map<std::string, std::map<std::string, std::string>> status;
1080 425 : if (conv->pending) {
1081 424 : preferences = std::move(conv->pending->preferences);
1082 424 : status = std::move(conv->pending->status);
1083 : }
1084 425 : conv->conversation = conversation;
1085 425 : if (parentLock.owns_lock())
1086 23 : parentLock.unlock();
1087 425 : conv->resetCloneRetry();
1088 425 : if (removeRepo) {
1089 1 : removeRepositoryImpl(*conv, false, true);
1090 1 : erasePending();
1091 1 : return;
1092 : }
1093 :
1094 424 : auto commitId = conversation->join();
1095 424 : if (!commitId.empty())
1096 758 : sendMessageNotification(*conversation, false, commitId);
1097 424 : erasePending(); // Will unlock
1098 :
1099 : #ifdef LIBJAMI_TEST
1100 424 : conversation->onBootstrapStatus(bootstrapCbTest_);
1101 : #endif
1102 424 : auto id = conversation->id();
1103 424 : conversation->bootstrap([w = weak(), id = std::move(id)]() {
1104 411 : if (auto sthis = w.lock())
1105 411 : sthis->bootstrapCb(id);
1106 411 : });
1107 :
1108 424 : if (auto pm = acc->presenceManager()) {
1109 2417 : for (const auto& member : conversation->memberUris()) {
1110 1145 : conversation->addKnownDevices(pm->getDevices(member), member);
1111 424 : }
1112 : }
1113 :
1114 424 : if (!preferences.empty())
1115 1 : conversation->updatePreferences(preferences);
1116 424 : if (!status.empty())
1117 20 : conversation->updateMessageStatus(status);
1118 424 : syncingMetadatas_.erase(conversationId);
1119 424 : saveMetadata();
1120 :
1121 424 : if (conversation->mode() == ConversationMode::DOCUMENT) {
1122 : // A document is not a conversation to the clients and is never
1123 : // synced to this account's other devices; what has a stake in the
1124 : // clone's completion is the CRDT session that asked for it, and a
1125 : // client waiting for the content to be on this device.
1126 44 : acc->collaborativeEditing()->onRepositoryUpdated(conversation->parentConversationId(),
1127 : conversationId,
1128 44 : !conversation->documentHistory(1).empty());
1129 44 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentReplicated>(accountId_,
1130 44 : conversation->parentConversationId(),
1131 : conversationId,
1132 : true);
1133 22 : return;
1134 : }
1135 :
1136 402 : linkReplacementConversation(*conversation);
1137 402 : acc->collaborativeEditing()->syncDocuments(conversationId);
1138 : // Inform user that the conversation is ready
1139 402 : emitSignal<libjami::ConversationSignal::ConversationReady>(accountId_, conversationId);
1140 402 : needsSyncingCb_({});
1141 : // Download members profile on first sync
1142 402 : auto isOneOne = conversation->mode() == ConversationMode::ONE_TO_ONE;
1143 402 : auto askForProfile = isOneOne;
1144 402 : if (!isOneOne) {
1145 : // If not 1:1 only download profiles from self (to avoid non checked files)
1146 166 : auto cert = acc->certStore().getCertificate(deviceId);
1147 166 : askForProfile = cert && cert->issuer && cert->issuer->getId().toString() == username_;
1148 166 : }
1149 402 : if (askForProfile) {
1150 993 : for (const auto& member : conversation->memberUris(username_)) {
1151 237 : acc->askForProfile(conversationId, deviceId, member);
1152 252 : }
1153 : }
1154 598 : } catch (const InvalidRepositoryError&) {
1155 : // Permanent failure: the remote repository contains malformed commits and
1156 : // its history is immutable, so every re-clone would fail the same way.
1157 : // Stop after a few attempts instead of retrying (and waking the peer's
1158 : // devices through the DHT proxy) indefinitely.
1159 2 : ++conv->validationFailures;
1160 2 : if (conv->isUnrecoverable()) {
1161 0 : JAMI_ERROR("[Account {}] [Conversation {}] Failed validation {} times, marking conversation "
1162 : "as unrecoverable: remove it and ask the peer for a new invitation",
1163 : accountId_,
1164 : conversationId,
1165 : conv->validationFailures);
1166 0 : emitSignal<libjami::ConversationSignal::OnConversationError>(
1167 0 : accountId_, conversationId, EUNRECOVERABLE, "Conversation repository failed validation repeatedly");
1168 : } else {
1169 2 : auto retryIn = conv->scheduleCloneRetry();
1170 2 : JAMI_WARNING(
1171 : "[Account {}] [Conversation {}] Remote conversation failed validation ({}/{}). Re-clone in {}s",
1172 : accountId_,
1173 : conversationId,
1174 : conv->validationFailures,
1175 : MAX_VALIDATION_FAILURES,
1176 : retryIn.count());
1177 4 : conv->fallbackClone->async_wait(std::bind(&ConversationModule::Impl::fallbackClone,
1178 4 : shared_from_this(),
1179 : std::placeholders::_1,
1180 : conversationId));
1181 : }
1182 31 : } catch (const std::exception& e) {
1183 29 : auto retryIn = conv->scheduleCloneRetry();
1184 29 : JAMI_WARNING(
1185 : "[Account {}] [Conversation {}] Something went wrong when cloning conversation: {}. Re-clone in {}s",
1186 : accountId_,
1187 : conversationId,
1188 : e.what(),
1189 : retryIn.count());
1190 58 : conv->fallbackClone->async_wait(std::bind(&ConversationModule::Impl::fallbackClone,
1191 58 : shared_from_this(),
1192 : std::placeholders::_1,
1193 : conversationId));
1194 29 : }
1195 433 : lk.lock();
1196 433 : erasePending();
1197 508 : }
1198 :
1199 : std::optional<ConversationRequest>
1200 13604 : ConversationModule::Impl::getRequest(const std::string& id) const
1201 : {
1202 : // ConversationsRequestsMtx MUST BE LOCKED
1203 13604 : auto it = conversationsRequests_.find(id);
1204 13604 : if (it != conversationsRequests_.end())
1205 424 : return it->second;
1206 13180 : return std::nullopt;
1207 : }
1208 :
1209 : std::string
1210 8410 : ConversationModule::Impl::getOneToOneConversation(const std::string& uri) const noexcept
1211 : {
1212 8410 : if (auto details = accountManager_->getContactInfo(uri)) {
1213 : // If contact is removed there is no conversation
1214 : // If banned, conversation is still on disk
1215 7379 : if (details->removed != TimePoint {} && details->banned == 0) {
1216 : // Check if contact is removed
1217 21 : if (details->removed > details->added)
1218 21 : return {};
1219 : }
1220 7358 : return details->conversationId;
1221 8410 : }
1222 1031 : return {};
1223 : }
1224 :
1225 : bool
1226 13 : ConversationModule::Impl::updateConvForContact(const std::string& uri,
1227 : const std::string& oldConv,
1228 : const std::string& newConv)
1229 : {
1230 13 : if (newConv != oldConv) {
1231 12 : auto conversation = getOneToOneConversation(uri);
1232 12 : if (conversation != oldConv) {
1233 0 : JAMI_DEBUG("[Account {}] [Conversation {}] Old conversation is not found in details {} - found: {}",
1234 : accountId_,
1235 : newConv,
1236 : oldConv,
1237 : conversation);
1238 0 : return false;
1239 : }
1240 12 : accountManager_->updateContactConversation(uri, newConv);
1241 12 : return true;
1242 12 : }
1243 1 : return false;
1244 : }
1245 :
1246 : void
1247 463 : ConversationModule::Impl::linkReplacementConversation(Conversation& conversation)
1248 : {
1249 463 : if (conversation.mode() != ConversationMode::ONE_TO_ONE || conversation.isRemoving())
1250 166 : return;
1251 594 : const auto members = conversation.memberUris({}, {MemberRole::LEFT, MemberRole::BANNED});
1252 297 : if (!members.contains(username_))
1253 0 : return;
1254 891 : for (const auto& member : members) {
1255 594 : if (member == username_ && members.size() != 1)
1256 591 : continue;
1257 297 : auto contact = accountManager_->getContactInfo(member);
1258 297 : if (!contact || !contact->isActive() || contact->conversationId == conversation.id())
1259 286 : continue;
1260 11 : if (!contact->conversationId.empty()) {
1261 10 : auto current = getConversation(contact->conversationId);
1262 10 : if (!current)
1263 0 : continue;
1264 10 : std::lock_guard lk(current->mtx);
1265 10 : if (!current->info.isRemoved())
1266 8 : continue;
1267 18 : }
1268 3 : updateConvForContact(member, contact->conversationId, conversation.id());
1269 297 : }
1270 297 : }
1271 :
1272 : void
1273 269 : ConversationModule::Impl::declineOtherConversationWith(const std::string& uri)
1274 : {
1275 : // conversationsRequestsMtx_ MUST BE LOCKED
1276 272 : for (auto& [id, request] : conversationsRequests_) {
1277 3 : if (request.declined != TimePoint {})
1278 1 : continue; // Ignore already declined requests
1279 2 : if (request.isOneToOne() && request.from == uri) {
1280 2 : JAMI_WARNING("[Account {}] [Conversation {}] Decline conversation request from {}", accountId_, id, uri);
1281 2 : request.declined = nowMs();
1282 2 : syncingMetadatas_.erase(id);
1283 2 : saveMetadata();
1284 2 : emitSignal<libjami::ConversationSignal::ConversationRequestDeclined>(accountId_, id);
1285 : }
1286 : }
1287 269 : }
1288 :
1289 : std::vector<std::map<std::string, std::string>>
1290 333 : ConversationModule::Impl::getConversationMembers(const std::string& conversationId, bool includeBanned) const
1291 : {
1292 999 : return withConv(conversationId, [&](const auto& conv) { return conv.getMembers(true, includeBanned); });
1293 : }
1294 :
1295 : void
1296 14 : ConversationModule::Impl::removeRepository(const std::string& conversationId, bool sync, bool force)
1297 : {
1298 14 : auto conv = getConversation(conversationId);
1299 14 : if (!conv)
1300 0 : return;
1301 14 : std::unique_lock lk(conv->mtx);
1302 14 : removeRepositoryImpl(*conv, sync, force);
1303 14 : }
1304 :
1305 : void
1306 32 : ConversationModule::Impl::removeRepositoryImpl(SyncedConversation& conv, bool sync, bool force)
1307 : {
1308 32 : if (conv.conversation && (force || conv.conversation->isRemoving())) {
1309 : // Stop fetch!
1310 32 : conv.clearFetches();
1311 : // And abort the ones already running: the repository is about to be deleted, and erase()
1312 : // below waits for them to release the repository write lock.
1313 32 : conv.conversation->shutdownConnections();
1314 :
1315 32 : JAMI_LOG("[Account {}] [Conversation {}] Remove conversation", accountId_, conv.info.id);
1316 : try {
1317 32 : if (conv.conversation->mode() == ConversationMode::ONE_TO_ONE) {
1318 51 : for (const auto& member : conv.conversation->getInitialMembers()) {
1319 34 : if (member != username_) {
1320 : // Note: this can happen while re-adding a contact.
1321 : // In this case, check that we are removing the linked conversation.
1322 17 : if (conv.info.id == getOneToOneConversation(member)) {
1323 1 : accountManager_->removeContactConversation(member);
1324 : }
1325 : }
1326 17 : }
1327 : }
1328 0 : } catch (const std::exception& e) {
1329 0 : JAMI_ERROR("{}", e.what());
1330 0 : }
1331 32 : conv.conversation->erase();
1332 32 : conv.conversation.reset();
1333 :
1334 32 : if (!sync)
1335 3 : return;
1336 :
1337 29 : conv.info.erased = nowMs();
1338 29 : needsSyncingCb_({});
1339 29 : addConvInfo(conv.info);
1340 : }
1341 : }
1342 :
1343 : bool
1344 15 : ConversationModule::Impl::removeConversation(const std::string& conversationId, bool forceRemove)
1345 : {
1346 30 : auto removed = withConv(conversationId,
1347 30 : [this, forceRemove](auto& conv) { return removeConversationImpl(conv, forceRemove); });
1348 15 : if (removed)
1349 : // Collect after marking the parent removed: no new child can register
1350 : // past that point, including a clone without a repository yet.
1351 50 : for (const auto& documentId : childDocuments({conversationId}))
1352 : // Each held document is left the way the conversation just was: a
1353 : // leave commit the other holders fetch, then the repository goes.
1354 : // A silent erasure instead would leave this device forever listed
1355 : // as a member of a document it can no longer be reached for.
1356 25 : withConv(documentId, [this](auto& conv) { return removeConversationImpl(conv, false); });
1357 15 : return removed;
1358 15 : }
1359 :
1360 : std::vector<std::string>
1361 248 : ConversationModule::Impl::childDocuments(const std::vector<std::string>& parents) const
1362 : {
1363 248 : std::vector<std::string> ids;
1364 248 : if (parents.empty())
1365 216 : return ids;
1366 88 : for (const auto& conv : getSyncedConversations()) {
1367 56 : std::lock_guard lk(conv->mtx);
1368 56 : if (conv->info.mode == ConversationMode::DOCUMENT
1369 56 : && std::find(parents.begin(), parents.end(), conv->info.parent) != parents.end())
1370 6 : ids.push_back(conv->info.id);
1371 88 : }
1372 32 : return ids;
1373 0 : }
1374 :
1375 : void
1376 15 : ConversationModule::Impl::removeDocumentReplica(const std::string& documentId)
1377 : {
1378 15 : if (!ConversationRepository::isValidConversationId(documentId)) {
1379 5 : JAMI_WARNING("[Account {}] Refusing to remove replica with invalid document id '{}'", accountId_, documentId);
1380 5 : return;
1381 : }
1382 10 : auto conv = startConversation(documentId);
1383 10 : std::lock_guard lk(conv->mtx);
1384 10 : if (conv->conversation && conv->conversation->mode() != ConversationMode::DOCUMENT)
1385 0 : return;
1386 :
1387 10 : conv->info.mode = ConversationMode::DOCUMENT;
1388 10 : conv->info.members.emplace(username_);
1389 10 : conv->info.removed = std::max(nowMs(), conv->info.created);
1390 10 : conv->info.erased = nowMs();
1391 10 : if (conv->fallbackClone)
1392 10 : conv->fallbackClone->cancel();
1393 10 : conv->pending.reset();
1394 10 : addConvInfo(conv->info);
1395 10 : if (conv->conversation) {
1396 7 : conv->conversation->shutdownConnections();
1397 7 : conv->conversation->erase();
1398 7 : conv->conversation.reset();
1399 : }
1400 10 : }
1401 :
1402 : bool
1403 20 : ConversationModule::Impl::removeConversationImpl(SyncedConversation& conv, bool forceRemove)
1404 : {
1405 20 : auto members = conv.getMembers(false, false);
1406 20 : auto isSyncing = !conv.conversation;
1407 20 : auto hasMembers = !isSyncing // If syncing there is no member to inform
1408 17 : && std::find_if(members.begin(),
1409 : members.end(),
1410 75 : [&](const auto& member) { return member.at("uri") == username_; })
1411 34 : != members.end() // We must be still a member
1412 37 : && members.size() != 1; // If there is only ourself
1413 20 : conv.info.removed = nowMs();
1414 20 : if (isSyncing)
1415 3 : conv.info.erased = nowMs();
1416 20 : if (conv.fallbackClone)
1417 20 : conv.fallbackClone->cancel();
1418 20 : if (isSyncing)
1419 3 : conv.clearFetches();
1420 20 : conv.resetCloneRetry();
1421 : // Sync now, because it can take some time to really removes the datas
1422 20 : needsSyncingCb_({});
1423 20 : addConvInfo(conv.info);
1424 : // A document is not one of the account's conversations to a client: it is
1425 : // listed and removed through the collaborative editing API.
1426 20 : if (!conv.conversation || conv.conversation->mode() != ConversationMode::DOCUMENT)
1427 17 : emitSignal<libjami::ConversationSignal::ConversationRemoved>(accountId_, conv.info.id);
1428 20 : if (isSyncing)
1429 3 : return true;
1430 :
1431 17 : if (forceRemove && conv.conversation->mode() == ConversationMode::ONE_TO_ONE) {
1432 : // skip waiting for sync
1433 1 : removeRepositoryImpl(conv, true);
1434 1 : return true;
1435 : }
1436 :
1437 16 : auto commitId = conv.conversation->leave();
1438 16 : if (hasMembers) {
1439 11 : JAMI_LOG("Wait that someone sync that user left conversation {}", conv.info.id);
1440 : // Commit that we left
1441 11 : if (!commitId.empty()) {
1442 : // Do not sync as it's synched by convInfos
1443 22 : sendMessageNotification(*conv.conversation, false, commitId);
1444 : } else {
1445 0 : JAMI_ERROR("Failed to send message to conversation {}", conv.info.id);
1446 : }
1447 : // In this case, we wait that another peer sync the conversation
1448 : // to definitely remove it from the device. This is to inform the
1449 : // peer that we left the conversation and never want to receive
1450 : // any messages
1451 11 : return true;
1452 : }
1453 :
1454 : // Else we are the last member, so we can remove
1455 5 : removeRepositoryImpl(conv, true);
1456 5 : return true;
1457 20 : }
1458 :
1459 : void
1460 743 : ConversationModule::Impl::sendMessageNotification(const std::string& conversationId,
1461 : bool sync,
1462 : const std::string& commitId,
1463 : const std::string& deviceId,
1464 : bool bootstrapReplay)
1465 : {
1466 743 : if (auto conv = getConversation(conversationId)) {
1467 743 : std::lock_guard lk(conv->mtx);
1468 743 : if (conv->conversation)
1469 743 : sendMessageNotification(*conv->conversation, sync, commitId, deviceId, bootstrapReplay);
1470 1486 : }
1471 743 : }
1472 :
1473 : void
1474 2467 : ConversationModule::Impl::sendMessageNotification(Conversation& conversation,
1475 : bool sync,
1476 : const std::string& commitId,
1477 : const std::string& deviceId,
1478 : bool bootstrapReplay)
1479 : {
1480 2467 : auto acc = account_.lock();
1481 2467 : if (!acc)
1482 0 : return;
1483 2467 : auto commit = commitId == "" ? conversation.lastCommitId() : commitId;
1484 2467 : if (bootstrapReplay && conversation.lastCommitId() != commit)
1485 7 : return;
1486 2460 : Json::Value message;
1487 2460 : message["id"] = conversation.id();
1488 2460 : message["commit"] = commit;
1489 2460 : message["deviceId"] = deviceId_;
1490 2460 : const auto text = json::toString(message);
1491 :
1492 : // Send message notification will announce the new commit in 3 steps.
1493 7380 : const auto messageMap = std::map<std::string, std::string> {{MIME_TYPE_GIT, text}};
1494 :
1495 : // First, because our account can have several devices, announce to other devices
1496 2460 : if (sync) {
1497 : // Announce to our devices
1498 914 : std::lock_guard lk(refreshMtx_);
1499 914 : auto& refresh = refreshMessage[username_];
1500 914 : refresh = sendMsgCb_(username_, {}, messageMap, refresh);
1501 914 : }
1502 :
1503 : // Then, we announce to 2 random members in the conversation that aren't in the DRT
1504 : // This allow new devices without the ability to sync to their other devices to sync with us.
1505 : // Or they can also use an old backup.
1506 2460 : std::vector<std::string> nonConnectedMembers;
1507 2460 : std::vector<NodeId> devices;
1508 : {
1509 2460 : std::lock_guard lk(notSyncedNotificationMtx_);
1510 2460 : devices = conversation.peersToSyncWith();
1511 4920 : auto members = conversation.memberUris(username_, {MemberRole::BANNED});
1512 2460 : std::vector<std::string> connectedMembers;
1513 : // print all members
1514 15046 : for (const auto& device : devices) {
1515 12586 : auto cert = acc->certStore().getCertificate(device.toString());
1516 12586 : if (cert && cert->issuer)
1517 12586 : connectedMembers.emplace_back(cert->issuer->getId().toString());
1518 12586 : }
1519 2460 : std::sort(std::begin(connectedMembers), std::end(connectedMembers));
1520 2460 : std::set_difference(members.begin(),
1521 : members.end(),
1522 : connectedMembers.begin(),
1523 : connectedMembers.end(),
1524 : std::inserter(nonConnectedMembers, nonConnectedMembers.begin()));
1525 2460 : std::shuffle(nonConnectedMembers.begin(), nonConnectedMembers.end(), acc->rand);
1526 2460 : if (nonConnectedMembers.size() > 2)
1527 54 : nonConnectedMembers.resize(2);
1528 2460 : if (!conversation.isBootstrapped()) {
1529 733 : JAMI_DEBUG("[Conversation {}] Not yet bootstrapped, save notification", conversation.id());
1530 : // Because we can get some git channels but not bootstrapped, we should keep this
1531 : // to refresh when bootstrapped.
1532 733 : if (bootstrapReplay)
1533 9 : notSyncedNotification_.try_emplace(conversation.id(), commit);
1534 : else
1535 724 : notSyncedNotification_[conversation.id()] = commit;
1536 : }
1537 2460 : }
1538 :
1539 2460 : std::lock_guard lk(refreshMtx_);
1540 3596 : for (const auto& member : nonConnectedMembers) {
1541 1136 : auto& refresh = refreshMessage[member];
1542 1136 : refresh = sendMsgCb_(member, {}, messageMap, refresh);
1543 : }
1544 :
1545 : // Finally we send to devices that the DRT choose.
1546 15046 : for (const auto& device : devices) {
1547 12586 : auto deviceIdStr = device.toString();
1548 12586 : auto memberUri = conversation.uriFromDevice(deviceIdStr);
1549 12586 : if (memberUri.empty() || deviceIdStr == deviceId)
1550 1145 : continue;
1551 11441 : auto& refresh = refreshMessage[deviceIdStr];
1552 11441 : refresh = sendMsgCb_(memberUri, device, messageMap, refresh);
1553 13731 : }
1554 :
1555 : // And we wake up the mobile devices we are responsible for (i.e. closer
1556 : // to them than any connected node). They are not actively connected to
1557 : // the DRT, so the DHT message (converted to a push notification by the
1558 : // DHT proxy) is the only way for them to learn about the new commit.
1559 : // Note: the repository only contains certificates of devices that wrote
1560 : // commits, so for read-only devices we resolve the member URI through
1561 : // the certificate store (filled when the device was last connected).
1562 2460 : for (const auto& target : conversation.mobileNodesToNotify()) {
1563 0 : const auto& device = target.device;
1564 0 : auto deviceIdStr = device.toString();
1565 0 : if (deviceIdStr == deviceId)
1566 0 : continue;
1567 0 : auto& refresh = refreshMessage[deviceIdStr];
1568 0 : refresh = sendMsgCb_(target.uri, device, messageMap, refresh);
1569 2460 : }
1570 4934 : }
1571 :
1572 : void
1573 94 : ConversationModule::Impl::sendMessage(const std::string& conversationId,
1574 : std::string message,
1575 : const std::string& replyTo,
1576 : bool announce,
1577 : OnCommitCb&& onCommit,
1578 : OnDoneCb&& cb)
1579 : {
1580 94 : auto msg = CommitMessage::text(std::move(message), replyTo);
1581 94 : createCommit(conversationId, std::move(msg), announce, std::move(onCommit), std::move(cb));
1582 94 : }
1583 :
1584 : void
1585 161 : ConversationModule::Impl::createCommit(
1586 : const std::string& conversationId, CommitMessage&& message, bool announce, OnCommitCb&& onCommit, OnDoneCb&& cb)
1587 : {
1588 161 : if (auto conv = getConversation(conversationId)) {
1589 161 : std::lock_guard lk(conv->mtx);
1590 161 : if (conv->conversation)
1591 320 : conv->conversation->createCommit(std::move(message),
1592 160 : std::move(onCommit),
1593 480 : [this,
1594 : conversationId,
1595 : announce,
1596 160 : cb = std::move(cb)](bool ok, const std::string& commitId) {
1597 159 : if (cb)
1598 44 : cb(ok, commitId);
1599 159 : if (!announce)
1600 2 : return;
1601 157 : if (ok)
1602 465 : sendMessageNotification(conversationId, true, commitId);
1603 : else
1604 2 : JAMI_ERROR("Failed to send message to conversation {}",
1605 : conversationId);
1606 : });
1607 322 : }
1608 161 : }
1609 :
1610 : void
1611 8 : ConversationModule::Impl::editMessage(const std::string& conversationId,
1612 : const std::string& newBody,
1613 : const std::string& editedId)
1614 : {
1615 : // Check that editedId is a valid commit, from ourself and plain/text
1616 8 : auto validCommit = false;
1617 8 : std::string type, fileId;
1618 8 : if (auto conv = getConversation(conversationId)) {
1619 8 : std::lock_guard lk(conv->mtx);
1620 8 : if (conv->conversation) {
1621 8 : auto commit = conv->conversation->getCommit(editedId);
1622 8 : if (commit != std::nullopt) {
1623 7 : type = commit->commitMsg.type;
1624 7 : if (type == CommitType::DATA_TRANSFER) {
1625 3 : fileId = getFileId(editedId, commit->commitMsg.tid, commit->commitMsg.displayName);
1626 : }
1627 14 : validCommit = commit->authorId == username_
1628 7 : && (type == CommitType::TEXT || type == CommitType::DATA_TRANSFER
1629 1 : || type == CommitType::COLLAB_DOC)
1630 14 : && (type != CommitType::DATA_TRANSFER || !fileId.empty());
1631 : }
1632 8 : }
1633 16 : }
1634 8 : if (!validCommit) {
1635 2 : JAMI_ERROR("Unable to edit commit {:s}", editedId);
1636 2 : return;
1637 : }
1638 : // Commit message edition
1639 6 : CommitMessage message;
1640 6 : if (type == CommitType::DATA_TRANSFER) {
1641 : // Remove file!
1642 2 : auto path = fileutils::get_data_dir() / accountId_ / "conversation_data" / conversationId / fileId;
1643 2 : dhtnet::fileutils::remove(path, true);
1644 2 : message = CommitMessage::fileDeleted(editedId);
1645 6 : } else if (type == CommitType::COLLAB_DOC) {
1646 : // A document is retired by editing its announcement, like a file. What it
1647 : // holds lives in a repository of its own, erased on each device when the
1648 : // commit lands there, so nothing is removed from here.
1649 1 : message = CommitMessage::collabDocRemoved(editedId);
1650 : } else {
1651 3 : message = CommitMessage::edit(newBody, editedId);
1652 : }
1653 6 : createCommit(conversationId, std::move(message));
1654 10 : }
1655 :
1656 : void
1657 762 : ConversationModule::Impl::bootstrapCb(std::string convId)
1658 : {
1659 762 : std::string commitId;
1660 : {
1661 762 : std::lock_guard lk(notSyncedNotificationMtx_);
1662 762 : auto it = notSyncedNotification_.find(convId);
1663 762 : if (it != notSyncedNotification_.end()) {
1664 542 : commitId = std::move(it->second);
1665 542 : notSyncedNotification_.erase(it);
1666 : }
1667 762 : }
1668 762 : if (commitId.empty())
1669 220 : return;
1670 542 : JAMI_DEBUG("[Account {}] [Conversation {}] Resend pending message notification", accountId_, convId);
1671 542 : dht::ThreadPool::io().run([w = weak(), convId, commitId = std::move(commitId)] {
1672 542 : if (auto sthis = w.lock())
1673 542 : sthis->sendMessageNotification(convId, true, commitId, {}, true);
1674 542 : });
1675 762 : }
1676 :
1677 : void
1678 888 : ConversationModule::Impl::fixStructures(
1679 : std::shared_ptr<JamiAccount> acc,
1680 : const std::vector<std::tuple<std::string, std::string, std::string>>& updateContactConv,
1681 : const std::set<std::string>& toRm)
1682 : {
1683 889 : for (const auto& [uri, oldConv, newConv] : updateContactConv) {
1684 1 : updateConvForContact(uri, oldConv, newConv);
1685 : }
1686 : ////////////////////////////////////////////////////////////////
1687 : // Note: This is only to homogenize trust and convRequests
1688 888 : std::vector<std::string> invalidPendingRequests;
1689 : {
1690 888 : auto requests = acc->getTrustRequests();
1691 888 : std::lock_guard lk(conversationsRequestsMtx_);
1692 889 : for (const auto& request : requests) {
1693 2 : auto itConvId = request.find(libjami::Account::TrustRequest::CONVERSATIONID);
1694 1 : auto itConvFrom = request.find(libjami::Account::TrustRequest::FROM);
1695 1 : if (itConvId != request.end() && itConvFrom != request.end()) {
1696 : // Check if requests exists or is declined.
1697 1 : auto itReq = conversationsRequests_.find(itConvId->second);
1698 1 : auto declined = itReq == conversationsRequests_.end() || itReq->second.declined != TimePoint {};
1699 1 : if (declined) {
1700 1 : JAMI_WARNING("Invalid trust request found: {:s}", itConvId->second);
1701 1 : invalidPendingRequests.emplace_back(itConvFrom->second);
1702 : }
1703 : }
1704 : }
1705 888 : auto requestRemoved = false;
1706 890 : for (auto it = conversationsRequests_.begin(); it != conversationsRequests_.end();) {
1707 2 : if (it->second.from == username_) {
1708 0 : JAMI_WARNING("Detected request from ourself, this makes no sense. Remove {}", it->first);
1709 0 : it = conversationsRequests_.erase(it);
1710 : } else {
1711 2 : ++it;
1712 : }
1713 : }
1714 888 : if (requestRemoved) {
1715 0 : saveConvRequests();
1716 : }
1717 888 : }
1718 889 : for (const auto& invalidPendingRequest : invalidPendingRequests)
1719 1 : acc->discardTrustRequest(invalidPendingRequest);
1720 :
1721 : ////////////////////////////////////////////////////////////////
1722 889 : for (const auto& conv : toRm) {
1723 1 : JAMI_ERROR("[Account {}] Remove conversation ({})", accountId_, conv);
1724 1 : removeConversation(conv, true);
1725 : }
1726 912 : for (const auto& conversation : getConversations())
1727 912 : acc->collaborativeEditing()->syncDocuments(conversation->id());
1728 888 : JAMI_DEBUG("[Account {}] Conversations loaded!", accountId_);
1729 888 : }
1730 :
1731 : void
1732 594 : ConversationModule::Impl::cloneConversationFrom(const std::shared_ptr<SyncedConversation> conv,
1733 : const std::string& deviceId)
1734 : {
1735 594 : std::lock_guard lk(conv->mtx);
1736 594 : const auto& conversationId = conv->info.id;
1737 594 : if (conv->info.mode == ConversationMode::DOCUMENT && conv->info.isRemoved())
1738 0 : return;
1739 594 : if (conv->isUnrecoverable()) {
1740 0 : JAMI_WARNING("[Account {}] [Conversation {}] [device {}] Conversation is marked unrecoverable, "
1741 : "ignoring clone request",
1742 : accountId_,
1743 : conversationId,
1744 : deviceId);
1745 0 : return;
1746 : }
1747 : // Checked whether or not a fetch is pending: a device that already failed in this
1748 : // window stays out even while another one is being tried, and one that did not is
1749 : // let in either way -- startFetch() arbitrates the round.
1750 594 : if (!conv->conversation && conv->cloneThrottled(deviceId)) {
1751 1 : JAMI_DEBUG("[Account {}] [Conversation {}] [device {}] Clone retry is not due yet, ignoring "
1752 : "clone request",
1753 : accountId_,
1754 : conversationId,
1755 : deviceId);
1756 1 : return;
1757 : }
1758 593 : if (!conv->startFetch(deviceId, true)) {
1759 22 : JAMI_WARNING("[Account {}] [Conversation {}] Already fetching", accountId_, conversationId);
1760 22 : return;
1761 : }
1762 :
1763 1713 : onNeedSocket_(
1764 : conversationId,
1765 : deviceId,
1766 1142 : [wthis = weak_from_this(), conv, conversationId, deviceId](const auto& channel) {
1767 571 : std::lock_guard lk(conv->mtx);
1768 571 : if (conv->pending && !conv->pending->ready) {
1769 472 : if (channel) {
1770 405 : conv->pending->ready = true;
1771 405 : conv->pending->deviceId = channel->deviceId().toString();
1772 405 : conv->pending->socket = channel;
1773 405 : if (!conv->pending->cloning) {
1774 405 : conv->pending->cloning = true;
1775 810 : dht::ThreadPool::io().run([wthis, conversationId, deviceId = conv->pending->deviceId]() {
1776 810 : if (auto sthis = wthis.lock())
1777 405 : sthis->handlePendingConversation(conversationId, deviceId);
1778 : });
1779 : }
1780 405 : return true;
1781 133 : } else if (auto sthis = wthis.lock()) {
1782 66 : conv->stopFetch(deviceId);
1783 66 : auto retryIn = conv->scheduleCloneRetry(deviceId);
1784 66 : JAMI_WARNING("[Account {}] [Conversation {}] [device {}] Clone failed. Re-clone in {}s",
1785 : sthis->accountId_,
1786 : conversationId,
1787 : deviceId,
1788 : retryIn.count());
1789 132 : conv->fallbackClone->async_wait(std::bind(&ConversationModule::Impl::fallbackClone,
1790 : sthis,
1791 : std::placeholders::_1,
1792 66 : conversationId));
1793 : }
1794 : }
1795 166 : return false;
1796 571 : },
1797 : MIME_TYPE_GIT,
1798 : false);
1799 594 : }
1800 :
1801 : void
1802 93 : ConversationModule::Impl::fallbackClone(const asio::error_code& ec, const std::string& conversationId)
1803 : {
1804 93 : if (ec == asio::error::operation_aborted)
1805 29 : return;
1806 83 : auto conv = getConversation(conversationId);
1807 83 : if (!conv || conv->conversation)
1808 19 : return;
1809 64 : auto members = getConversationMembers(conversationId);
1810 192 : for (const auto& member : members)
1811 384 : if (member.at("uri") != username_)
1812 128 : cloneConversationFrom(conversationId, member.at("uri"));
1813 83 : }
1814 :
1815 : void
1816 1650 : ConversationModule::Impl::bootstrap(const std::string& convId)
1817 : {
1818 1650 : std::vector<std::shared_ptr<SyncedConversation>> toClone;
1819 1650 : std::vector<std::shared_ptr<Conversation>> conversations;
1820 1650 : if (convId.empty()) {
1821 893 : std::vector<std::shared_ptr<SyncedConversation>> convs;
1822 : {
1823 893 : std::lock_guard lk(convInfosMtx_);
1824 973 : for (const auto& [conversationId, convInfo] : convInfos_) {
1825 80 : if (auto conv = getConversation(conversationId))
1826 80 : convs.emplace_back(std::move(conv));
1827 : }
1828 893 : }
1829 973 : for (auto& conv : convs) {
1830 80 : std::lock_guard lk(conv->mtx);
1831 80 : if (!conv->conversation && !conv->info.isRemoved()) {
1832 : // we need to ask to clone requests when bootstrapping all conversations
1833 : // otherwise it can stay syncing
1834 8 : toClone.emplace_back(std::move(conv));
1835 72 : } else if (conv->conversation) {
1836 70 : conversations.emplace_back(conv->conversation);
1837 : }
1838 80 : }
1839 1650 : } else if (auto conv = getConversation(convId)) {
1840 509 : std::lock_guard lk(conv->mtx);
1841 509 : if (conv->conversation)
1842 506 : conversations.emplace_back(conv->conversation);
1843 1266 : }
1844 :
1845 2226 : for (const auto& conversation : conversations) {
1846 576 : if (auto acc = account_.lock())
1847 576 : acc->collaborativeEditing()->syncDocuments(conversation->id());
1848 : #ifdef LIBJAMI_TEST
1849 576 : conversation->onBootstrapStatus(bootstrapCbTest_);
1850 : #endif
1851 576 : conversation->bootstrap([w = weak(), id = conversation->id()] {
1852 206 : if (auto sthis = w.lock())
1853 206 : sthis->bootstrapCb(id);
1854 206 : });
1855 :
1856 576 : if (auto acc = account_.lock()) {
1857 576 : if (auto pm = acc->presenceManager()) {
1858 2372 : for (const auto& member : conversation->memberUris()) {
1859 644 : conversation->addKnownDevices(pm->getDevices(member), member);
1860 576 : }
1861 : }
1862 576 : }
1863 : }
1864 1658 : for (const auto& conv : toClone) {
1865 24 : for (const auto& member : conv->getMembers(false, false)) {
1866 48 : if (member.at("uri") != username_)
1867 16 : cloneConversationFrom(conv->info.id, member.at("uri"));
1868 8 : }
1869 : }
1870 1650 : }
1871 :
1872 : void
1873 407 : ConversationModule::Impl::cloneConversationFrom(const ConversationRequest& request)
1874 : {
1875 407 : auto memberHash = dht::InfoHash(request.from);
1876 407 : if (!memberHash) {
1877 0 : JAMI_WARNING("Invalid member detected: {}", request.from);
1878 0 : return;
1879 : }
1880 407 : auto conv = startConversation(request.conversationId);
1881 407 : std::lock_guard lk(conv->mtx);
1882 : // Explicit (or auto-accepted) request: the peer just told us to clone, don't sit on a
1883 : // backoff armed by earlier failures.
1884 407 : conv->resetCloneRetry();
1885 407 : if (conv->info.created == TimePoint {}) {
1886 401 : conv->info = {request.conversationId};
1887 401 : conv->info.created = request.received;
1888 401 : conv->info.members.emplace(username_);
1889 401 : conv->info.members.emplace(request.from);
1890 401 : conv->info.mode = request.mode();
1891 401 : addConvInfo(conv->info);
1892 6 : } else if (conv->info.mode != ConversationMode::ONE_TO_ONE && conv->info.isRemoved()) {
1893 : // Re-invited to a group conversation we previously left.
1894 1 : conv->info.created = nowMs();
1895 1 : conv->info.erased = TimePoint {};
1896 1 : addConvInfo(conv->info);
1897 : }
1898 407 : accountManager_->forEachDevice(memberHash, [w = weak(), conv](const auto& pk) {
1899 405 : auto sthis = w.lock();
1900 405 : auto deviceId = pk->getLongId().toString();
1901 405 : if (!sthis or deviceId == sthis->deviceId_)
1902 0 : return;
1903 405 : sthis->cloneConversationFrom(conv, deviceId);
1904 405 : });
1905 407 : }
1906 :
1907 : void
1908 133 : ConversationModule::Impl::cloneConversationFrom(const std::string& conversationId, const std::string& uri)
1909 : {
1910 133 : if (!ConversationRepository::isValidConversationId(conversationId)) {
1911 0 : JAMI_WARNING("[Account {}] Invalid conversation id detected: '{}'", accountId_, conversationId);
1912 0 : return;
1913 : }
1914 133 : auto memberHash = dht::InfoHash(uri);
1915 133 : if (!memberHash) {
1916 0 : JAMI_WARNING("Invalid member detected: {}", uri);
1917 0 : return;
1918 : }
1919 133 : auto conv = startConversation(conversationId);
1920 266 : accountManager_->forEachDevice(memberHash,
1921 266 : [w = weak(), conv, conversationId](
1922 : const std::shared_ptr<dht::crypto::PublicKey>& pk) {
1923 96 : auto sthis = w.lock();
1924 96 : auto deviceId = pk->getLongId().toString();
1925 96 : if (!sthis or deviceId == sthis->deviceId_)
1926 30 : return;
1927 66 : sthis->cloneConversationFrom(conv, deviceId);
1928 126 : });
1929 133 : }
1930 :
1931 : ////////////////////////////////////////////////////////////////
1932 :
1933 : void
1934 1026 : ConversationModule::saveConvRequests(const std::string& accountId,
1935 : const std::map<std::string, ConversationRequest>& conversationsRequests)
1936 : {
1937 1026 : auto path = fileutils::get_data_dir() / accountId;
1938 1026 : saveConvRequestsToPath(path, conversationsRequests);
1939 1026 : }
1940 :
1941 : void
1942 2039 : ConversationModule::saveConvRequestsToPath(const std::filesystem::path& path,
1943 : const std::map<std::string, ConversationRequest>& conversationsRequests)
1944 : {
1945 2039 : auto p = path / "convRequests";
1946 2039 : std::lock_guard lock(dhtnet::fileutils::getFileLock(p));
1947 2039 : std::ofstream file(p, std::ios::trunc | std::ios::binary);
1948 2039 : msgpack::pack(file, conversationsRequests);
1949 2039 : }
1950 :
1951 : void
1952 5575 : ConversationModule::saveConvInfos(const std::string& accountId, const ConvInfoMap& conversations)
1953 : {
1954 5575 : auto path = fileutils::get_data_dir() / accountId;
1955 5575 : saveConvInfosToPath(path, conversations);
1956 5575 : }
1957 :
1958 : void
1959 6588 : ConversationModule::saveConvInfosToPath(const std::filesystem::path& path, const ConvInfoMap& conversations)
1960 : {
1961 6588 : std::lock_guard lock(dhtnet::fileutils::getFileLock(path / "convInfo"));
1962 6588 : std::ofstream file(path / "convInfo", std::ios::trunc | std::ios::binary);
1963 6588 : msgpack::pack(file, conversations);
1964 6588 : }
1965 :
1966 : ////////////////////////////////////////////////////////////////
1967 :
1968 873 : ConversationModule::ConversationModule(std::shared_ptr<JamiAccount> account,
1969 : std::shared_ptr<AccountManager> accountManager,
1970 : NeedsSyncingCb&& needsSyncingCb,
1971 : SengMsgCb&& sendMsgCb,
1972 : NeedSocketCb&& onNeedSocket,
1973 : NeedSocketCb&& onNeedSwarmSocket,
1974 : OneToOneRecvCb&& oneToOneRecvCb,
1975 873 : bool autoLoadConversations)
1976 873 : : pimpl_ {std::make_unique<Impl>(std::move(account),
1977 873 : std::move(accountManager),
1978 873 : std::move(needsSyncingCb),
1979 873 : std::move(sendMsgCb),
1980 873 : std::move(onNeedSocket),
1981 873 : std::move(onNeedSwarmSocket),
1982 873 : std::move(oneToOneRecvCb))}
1983 : {
1984 873 : if (autoLoadConversations) {
1985 873 : loadConversations();
1986 : }
1987 873 : }
1988 :
1989 : void
1990 18 : ConversationModule::setAccountManager(std::shared_ptr<AccountManager> accountManager)
1991 : {
1992 18 : std::unique_lock lk(pimpl_->conversationsMtx_);
1993 18 : pimpl_->accountManager_ = accountManager;
1994 18 : }
1995 :
1996 : #ifdef LIBJAMI_TEST
1997 : void
1998 26 : ConversationModule::onBootstrapStatus(const std::function<void(std::string, Conversation::BootstrapStatus)>& cb)
1999 : {
2000 26 : pimpl_->bootstrapCbTest_ = cb;
2001 33 : for (auto& c : pimpl_->getConversations())
2002 33 : c->onBootstrapStatus(pimpl_->bootstrapCbTest_);
2003 26 : }
2004 : void
2005 3 : ConversationModule::onFetchCompleted(const std::function<void(const std::string&, const std::string&, bool)>& cb)
2006 : {
2007 3 : pimpl_->fetchCompletedCbTest_ = cb;
2008 3 : }
2009 :
2010 : void
2011 4 : ConversationModule::onCloneReady(const std::function<void(const std::string&)>& cb)
2012 : {
2013 4 : pimpl_->cloneReadyCbTest_ = cb;
2014 4 : }
2015 : #endif
2016 :
2017 : void
2018 888 : ConversationModule::loadConversations()
2019 : {
2020 888 : auto acc = pimpl_->account_.lock();
2021 888 : if (!acc)
2022 0 : return;
2023 888 : JAMI_LOG("[Account {}] Start loading conversations…", pimpl_->accountId_);
2024 888 : auto conversationPath = fileutils::get_data_dir() / pimpl_->accountId_ / "conversations";
2025 :
2026 888 : std::unique_lock lk(pimpl_->conversationsMtx_);
2027 888 : auto contacts = pimpl_->accountManager_->getContacts(
2028 888 : true); // Avoid to lock configurationMtx while conv Mtx is locked
2029 888 : std::unique_lock ilk(pimpl_->convInfosMtx_);
2030 888 : pimpl_->convInfos_ = convInfos(pimpl_->accountId_);
2031 888 : pimpl_->conversations_.clear();
2032 :
2033 : struct Ctx
2034 : {
2035 : std::mutex cvMtx;
2036 : std::condition_variable cv;
2037 : std::mutex toRmMtx;
2038 : std::set<std::string> toRm;
2039 : std::mutex convMtx;
2040 : size_t convNb;
2041 : std::map<dht::InfoHash, Contact> contacts;
2042 : std::vector<std::tuple<std::string, std::string, std::string>> updateContactConv;
2043 : };
2044 : struct PendingConvCounter
2045 : {
2046 : std::shared_ptr<Ctx> ctx;
2047 25 : PendingConvCounter(std::shared_ptr<Ctx> c)
2048 25 : : ctx(std::move(c))
2049 : {
2050 25 : std::lock_guard lk {ctx->cvMtx};
2051 25 : ++ctx->convNb;
2052 25 : }
2053 25 : ~PendingConvCounter()
2054 : {
2055 25 : std::lock_guard lk {ctx->cvMtx};
2056 25 : --ctx->convNb;
2057 25 : ctx->cv.notify_all();
2058 25 : }
2059 : };
2060 888 : auto ctx = std::make_shared<Ctx>();
2061 888 : ctx->convNb = 0;
2062 888 : ctx->contacts = std::move(contacts);
2063 :
2064 888 : std::error_code ec;
2065 938 : for (const auto& convIt : std::filesystem::directory_iterator(conversationPath, ec)) {
2066 25 : auto name = convIt.path().filename().string();
2067 25 : if (!convIt.is_directory() || !isConversationId(name))
2068 0 : continue;
2069 50 : dht::ThreadPool::io().run(
2070 50 : [this, ctx, repository = std::move(name), acc, _ = std::make_shared<PendingConvCounter>(ctx)] {
2071 : try {
2072 25 : auto sconv = std::make_shared<SyncedConversation>(repository);
2073 25 : auto conv = std::make_shared<Conversation>(acc, repository);
2074 25 : conv->onMessageStatusChanged([this, repository](const auto& status) {
2075 2 : auto msg = std::make_shared<SyncMsg>();
2076 4 : msg->ms = {{repository, status}};
2077 2 : pimpl_->needsSyncingCb_(std::move(msg));
2078 4 : });
2079 25 : conv->onMembersChanged([w = pimpl_->weak_from_this(), repository](const auto& members) {
2080 : // Delay in another thread to avoid deadlocks
2081 2 : dht::ThreadPool::io().run([w, repository, members = std::move(members)] {
2082 2 : if (auto sthis = w.lock())
2083 1 : sthis->setConversationMembers(repository, members);
2084 : });
2085 1 : });
2086 25 : conv->onNeedSocket(pimpl_->onNeedSwarmSocket_);
2087 25 : auto members = conv->memberUris(acc->getUsername(), {});
2088 : // NOTE: The following if is here to protect against any incorrect state
2089 : // that can be introduced
2090 25 : if (conv->mode() == ConversationMode::ONE_TO_ONE && members.size() == 1) {
2091 : // If we got a 1:1 conversation, but not in the contact details, it's rather a
2092 : // duplicate or a weird state
2093 10 : auto otherUri = *members.begin();
2094 10 : auto itContact = ctx->contacts.find(dht::InfoHash(otherUri));
2095 10 : if (itContact == ctx->contacts.end()) {
2096 0 : JAMI_WARNING("Contact {} not found", otherUri);
2097 0 : return;
2098 : }
2099 10 : const std::string& convFromDetails = itContact->second.conversationId;
2100 10 : auto isRemoved = !itContact->second.isActive();
2101 10 : if (convFromDetails != repository) {
2102 3 : if (convFromDetails.empty()) {
2103 2 : if (isRemoved) {
2104 : // If details is empty, contact is removed and not banned.
2105 1 : JAMI_ERROR("Conversation {} detected for {} and should be removed",
2106 : repository,
2107 : otherUri);
2108 1 : std::lock_guard lkMtx {ctx->toRmMtx};
2109 1 : ctx->toRm.insert(repository);
2110 1 : } else {
2111 1 : JAMI_ERROR("No conversation detected for {} but one exists ({}). Update details",
2112 : otherUri,
2113 : repository);
2114 1 : std::lock_guard lkMtx {ctx->toRmMtx};
2115 2 : ctx->updateContactConv.emplace_back(
2116 2 : std::make_tuple(otherUri, convFromDetails, repository));
2117 1 : }
2118 : }
2119 : }
2120 10 : }
2121 : {
2122 25 : std::lock_guard lkMtx {ctx->convMtx};
2123 25 : auto convInfo = pimpl_->convInfos_.find(repository);
2124 25 : if (convInfo == pimpl_->convInfos_.end()) {
2125 2 : JAMI_ERROR("Missing conv info for {}. This is a bug!", repository);
2126 2 : sconv->info.created = nowMs();
2127 6 : sconv->info.lastDisplayed = conv->infos()[ConversationMapKeys::LAST_DISPLAYED];
2128 : } else {
2129 23 : sconv->info = convInfo->second;
2130 23 : if (convInfo->second.isRemoved()) {
2131 : // A conversation was removed, but repository still exists
2132 1 : conv->setRemovingFlag();
2133 1 : std::lock_guard lkMtx {ctx->toRmMtx};
2134 1 : ctx->toRm.insert(repository);
2135 1 : }
2136 : }
2137 : // Even if we found the conversation in convInfos_, unable to assume that the
2138 : // list of members stored in `convInfo` is correct
2139 : // (https://git.jami.net/savoirfairelinux/jami-daemon/-/issues/1025). For this
2140 : // reason, we always use the list we got from the conversation repository to set
2141 : // the value of `sconv->info.members`.
2142 25 : members.emplace(acc->getUsername());
2143 25 : sconv->info.members = std::move(members);
2144 25 : sconv->info.mode = conv->mode();
2145 25 : if (conv->mode() == ConversationMode::DOCUMENT)
2146 0 : sconv->info.parent = conv->parentConversationId();
2147 : // convInfosMtx_ is already locked
2148 25 : pimpl_->convInfos_[repository] = sconv->info;
2149 25 : }
2150 25 : auto commits = conv->commitsEndedCalls();
2151 :
2152 25 : if (!commits.empty()) {
2153 : // Note: here, this means that some calls were actives while the
2154 : // daemon finished (can be a crash).
2155 : // Notify other in the conversation that the call is finished
2156 0 : pimpl_->sendMessageNotification(*conv, true, *commits.rbegin());
2157 : }
2158 25 : sconv->conversation = conv;
2159 25 : std::lock_guard lkMtx {ctx->convMtx};
2160 25 : pimpl_->conversations_.emplace(repository, std::move(sconv));
2161 25 : } catch (const std::logic_error& e) {
2162 0 : JAMI_WARNING("[Account {}] Conversations not loaded: {}", pimpl_->accountId_, e.what());
2163 0 : }
2164 : });
2165 913 : }
2166 888 : if (ec) {
2167 868 : JAMI_ERROR("Failed to read conversations directory {}: {}", conversationPath, ec.message());
2168 : }
2169 :
2170 888 : std::unique_lock lkCv(ctx->cvMtx);
2171 1801 : ctx->cv.wait(lkCv, [&] { return ctx->convNb == 0; });
2172 :
2173 : // Prune any invalid conversations without members and
2174 : // set the removed flag if needed
2175 888 : std::set<std::string> removed;
2176 945 : for (auto itInfo = pimpl_->convInfos_.begin(); itInfo != pimpl_->convInfos_.end();) {
2177 57 : const auto& info = itInfo->second;
2178 57 : if (info.members.empty()) {
2179 0 : itInfo = pimpl_->convInfos_.erase(itInfo);
2180 0 : continue;
2181 : }
2182 57 : if (info.isRemoved())
2183 3 : removed.insert(info.id);
2184 57 : auto itConv = pimpl_->conversations_.find(info.id);
2185 57 : if (itConv == pimpl_->conversations_.end()) {
2186 : // convInfos_ can contain a conversation that is not yet cloned
2187 : // so we need to add it there.
2188 32 : itConv = pimpl_->conversations_.emplace(info.id, std::make_shared<SyncedConversation>(info)).first;
2189 : }
2190 57 : if (itConv != pimpl_->conversations_.end() && itConv->second && itConv->second->conversation && info.isRemoved())
2191 1 : itConv->second->conversation->setRemovingFlag();
2192 57 : if (!info.isRemoved() && itConv == pimpl_->conversations_.end()) {
2193 : // In this case, the conversation is not synced and we only know ourself
2194 0 : if (info.members.size() == 1 && *info.members.begin() == acc->getUsername()) {
2195 0 : JAMI_WARNING("[Account {:s}] Conversation {:s} seems not present/synced.", pimpl_->accountId_, info.id);
2196 0 : emitSignal<libjami::ConversationSignal::ConversationRemoved>(pimpl_->accountId_, info.id);
2197 0 : itInfo = pimpl_->convInfos_.erase(itInfo);
2198 0 : continue;
2199 : }
2200 : }
2201 57 : ++itInfo;
2202 : }
2203 : // On oldest version, removeConversation didn't update "appdata/contacts"
2204 : // causing a potential incorrect state between "appdata/contacts" and "appdata/convInfos"
2205 888 : if (!removed.empty())
2206 3 : acc->unlinkConversations(removed);
2207 :
2208 914 : for (const auto& [contactId, contact] : ctx->contacts) {
2209 26 : if (contact.conversationId.empty())
2210 26 : continue;
2211 :
2212 24 : if (pimpl_->convInfos_.find(contact.conversationId) != pimpl_->convInfos_.end())
2213 24 : continue;
2214 :
2215 0 : ConvInfo newInfo;
2216 0 : newInfo.id = contact.conversationId;
2217 0 : newInfo.created = nowMs();
2218 0 : newInfo.members.emplace(pimpl_->username_);
2219 0 : newInfo.members.emplace(contactId.toString());
2220 0 : pimpl_->conversations_.emplace(contact.conversationId, std::make_shared<SyncedConversation>(newInfo));
2221 0 : pimpl_->convInfos_.emplace(contact.conversationId, std::move(newInfo));
2222 0 : }
2223 :
2224 888 : pimpl_->saveConvInfos();
2225 :
2226 888 : ilk.unlock();
2227 888 : lk.unlock();
2228 :
2229 1776 : dht::ThreadPool::io().run(
2230 1776 : [w = pimpl_->weak(), acc, updateContactConv = std::move(ctx->updateContactConv), toRm = std::move(ctx->toRm)]() {
2231 : // Will lock account manager
2232 888 : if (auto shared = w.lock())
2233 888 : shared->fixStructures(acc, updateContactConv, toRm);
2234 888 : });
2235 888 : }
2236 :
2237 : void
2238 0 : ConversationModule::loadSingleConversation(const std::string& convId)
2239 : {
2240 0 : auto acc = pimpl_->account_.lock();
2241 0 : if (!acc)
2242 0 : return;
2243 0 : JAMI_LOG("[Account {}] Start loading conversation {}", pimpl_->accountId_, convId);
2244 :
2245 0 : std::unique_lock lk(pimpl_->conversationsMtx_);
2246 0 : std::unique_lock ilk(pimpl_->convInfosMtx_);
2247 : // Load convInfos to retrieve requests that have been accepted but not yet synchronized.
2248 0 : pimpl_->convInfos_ = convInfos(pimpl_->accountId_);
2249 0 : pimpl_->conversations_.clear();
2250 :
2251 : try {
2252 0 : auto infoIt = pimpl_->convInfos_.find(convId);
2253 0 : auto sconv = infoIt != pimpl_->convInfos_.end() ? std::make_shared<SyncedConversation>(infoIt->second)
2254 0 : : std::make_shared<SyncedConversation>(convId);
2255 :
2256 0 : auto conv = std::make_shared<Conversation>(acc, convId);
2257 :
2258 0 : conv->onNeedSocket(pimpl_->onNeedSwarmSocket_);
2259 :
2260 0 : sconv->conversation = conv;
2261 0 : pimpl_->conversations_.emplace(convId, std::move(sconv));
2262 0 : } catch (const std::logic_error& e) {
2263 0 : JAMI_WARNING("[Account {}] Conversations not loaded: {}", pimpl_->accountId_, e.what());
2264 0 : }
2265 :
2266 : // Add all other conversations as dummy conversations to indicate their existence so
2267 : // isConversation could detect conversations correctly.
2268 0 : auto conversationsRepositoryIds = dhtnet::fileutils::readDirectory(fileutils::get_data_dir() / pimpl_->accountId_
2269 0 : / "conversations");
2270 0 : for (auto repositoryId : conversationsRepositoryIds) {
2271 : // Only canonical conversation ids: the directory may also hold transient
2272 : // artifacts, and a dummy conversation created from one would advertise a
2273 : // bogus id to the rest of the module.
2274 0 : if (repositoryId != convId && isConversationId(repositoryId)) {
2275 0 : auto conv = std::make_shared<SyncedConversation>(repositoryId);
2276 0 : pimpl_->conversations_.emplace(repositoryId, conv);
2277 0 : }
2278 0 : }
2279 :
2280 : // Add conversations from convInfos_ so isConversation could detect conversations correctly.
2281 : // This includes conversations that have been accepted but are not yet synchronized.
2282 0 : for (auto itInfo = pimpl_->convInfos_.begin(); itInfo != pimpl_->convInfos_.end();) {
2283 0 : const auto& info = itInfo->second;
2284 0 : if (info.members.empty()) {
2285 0 : itInfo = pimpl_->convInfos_.erase(itInfo);
2286 0 : continue;
2287 : }
2288 0 : auto itConv = pimpl_->conversations_.find(info.id);
2289 0 : if (itConv == pimpl_->conversations_.end()) {
2290 : // convInfos_ can contain a conversation that is not yet cloned
2291 : // so we need to add it there.
2292 0 : pimpl_->conversations_.emplace(info.id, std::make_shared<SyncedConversation>(info));
2293 : }
2294 0 : ++itInfo;
2295 : }
2296 :
2297 0 : ilk.unlock();
2298 0 : lk.unlock();
2299 0 : }
2300 :
2301 : void
2302 1650 : ConversationModule::bootstrap(const std::string& convId)
2303 : {
2304 1650 : pimpl_->bootstrap(convId);
2305 1650 : }
2306 :
2307 : void
2308 0 : ConversationModule::monitor()
2309 : {
2310 0 : for (auto& conv : pimpl_->getConversations())
2311 0 : conv->monitor();
2312 0 : }
2313 :
2314 : void
2315 916 : ConversationModule::clearPendingFetch()
2316 : {
2317 : // Note: This is a workaround. convModule() is kept if account is disabled/re-enabled.
2318 : // iOS uses setAccountActive() a lot, and if for some reason the previous pending fetch
2319 : // is not erased (callback not called), it will block the new messages as it will not
2320 : // sync. The best way to debug this is to get logs from the last ICE connection for
2321 : // syncing the conversation. It may have been killed in some un-expected way avoiding to
2322 : // call the callbacks. This should never happen, but if it's the case, this will allow
2323 : // new messages to be synced correctly.
2324 994 : for (auto& conv : pimpl_->getSyncedConversations()) {
2325 78 : std::lock_guard lk(conv->mtx);
2326 78 : if (conv && (conv->pending || !conv->deferredFetches.empty())) {
2327 1 : JAMI_ERROR("This is a bug, seems to still fetch to some device on initializing");
2328 1 : conv->clearFetches();
2329 : }
2330 994 : }
2331 916 : }
2332 :
2333 : void
2334 0 : ConversationModule::reloadRequests()
2335 : {
2336 0 : pimpl_->conversationsRequests_ = convRequests(pimpl_->accountId_);
2337 0 : }
2338 :
2339 : std::vector<std::string>
2340 47 : ConversationModule::getConversations() const
2341 : {
2342 47 : std::vector<std::string> result;
2343 47 : std::lock_guard lk(pimpl_->convInfosMtx_);
2344 47 : result.reserve(pimpl_->convInfos_.size());
2345 152 : for (const auto& [key, conv] : pimpl_->convInfos_) {
2346 105 : if (conv.isRemoved())
2347 3 : continue;
2348 : // Collaborative documents are swarms, not conversations: the clients
2349 : // list them per conversation through the collaborative-editing API.
2350 102 : if (conv.mode == ConversationMode::DOCUMENT)
2351 0 : continue;
2352 102 : result.emplace_back(key);
2353 : }
2354 94 : return result;
2355 47 : }
2356 :
2357 : std::string
2358 8381 : ConversationModule::getOneToOneConversation(const std::string& uri) const noexcept
2359 : {
2360 8381 : return pimpl_->getOneToOneConversation(uri);
2361 : }
2362 :
2363 : bool
2364 9 : ConversationModule::updateConvForContact(const std::string& uri, const std::string& oldConv, const std::string& newConv)
2365 : {
2366 9 : return pimpl_->updateConvForContact(uri, oldConv, newConv);
2367 : }
2368 :
2369 : std::vector<std::map<std::string, std::string>>
2370 95 : ConversationModule::getConversationRequests() const
2371 : {
2372 95 : std::vector<std::map<std::string, std::string>> requests;
2373 95 : std::lock_guard lk(pimpl_->conversationsRequestsMtx_);
2374 95 : requests.reserve(pimpl_->conversationsRequests_.size());
2375 112 : for (const auto& [id, request] : pimpl_->conversationsRequests_) {
2376 17 : if (request.declined != TimePoint {})
2377 6 : continue; // Do not add declined requests
2378 11 : requests.emplace_back(request.toMap());
2379 : }
2380 190 : return requests;
2381 95 : }
2382 :
2383 : void
2384 302 : ConversationModule::onTrustRequest(const std::string& uri,
2385 : const std::string& conversationId,
2386 : const std::vector<uint8_t>& payload,
2387 : TimePoint received,
2388 : TimePoint invited)
2389 : {
2390 302 : if (!ConversationRepository::isValidConversationId(conversationId)) {
2391 6 : JAMI_WARNING("[Account {}] Ignoring trust request from {} with invalid conversation id '{}'",
2392 : pimpl_->accountId_,
2393 : uri,
2394 : conversationId);
2395 13 : return;
2396 : }
2397 296 : std::unique_lock lk(pimpl_->conversationsRequestsMtx_);
2398 296 : ConversationRequest req;
2399 296 : req.from = uri;
2400 296 : req.conversationId = conversationId;
2401 592 : req.metadatas = ConversationRepository::infosFromVCard(
2402 888 : vCard::utils::toMap(std::string_view(reinterpret_cast<const char*>(payload.data()), payload.size())));
2403 296 : if (invited != TimePoint {})
2404 294 : req.received = validateInviteTimestamp(invited, received);
2405 : else
2406 : // Older peer that doesn't embed an invite timestamp yet
2407 2 : req.received = received;
2408 296 : auto reqMap = req.toMap();
2409 :
2410 296 : auto contactInfo = pimpl_->accountManager_->getContactInfo(uri);
2411 296 : if (contactInfo && contactInfo->confirmed && !contactInfo->isBanned() && contactInfo->isActive()) {
2412 7 : JAMI_LOG("[Account {}] Contact {} is confirmed, cloning {}", pimpl_->accountId_, uri, conversationId);
2413 7 : lk.unlock();
2414 7 : updateConvForContact(uri, contactInfo->conversationId, conversationId);
2415 7 : pimpl_->cloneConversationFrom(req);
2416 7 : return;
2417 : }
2418 :
2419 289 : if (pimpl_->addConversationRequest(conversationId, std::move(req))) {
2420 267 : lk.unlock();
2421 267 : emitSignal<libjami::ConfigurationSignal::IncomingTrustRequest>(pimpl_->accountId_,
2422 : conversationId,
2423 : uri,
2424 : payload,
2425 : toSecondsSinceEpoch(received));
2426 267 : emitSignal<libjami::ConversationSignal::ConversationRequestReceived>(pimpl_->accountId_, conversationId, reqMap);
2427 267 : pimpl_->needsSyncingCb_({});
2428 : } else {
2429 22 : JAMI_DEBUG("[Account {}] Received a request for a conversation already existing. Ignore", pimpl_->accountId_);
2430 : }
2431 317 : }
2432 :
2433 : void
2434 333 : ConversationModule::onConversationRequest(const std::string& from, const Json::Value& value)
2435 : {
2436 333 : ConversationRequest req(value);
2437 333 : if (!ConversationRepository::isValidConversationId(req.conversationId)) {
2438 6 : JAMI_WARNING("[Account {}] Ignoring conversation request from {} with invalid conversation id '{}'",
2439 : pimpl_->accountId_,
2440 : from,
2441 : req.conversationId);
2442 6 : return;
2443 : }
2444 327 : auto isOneToOne = req.isOneToOne();
2445 327 : std::unique_lock lk(pimpl_->conversationsRequestsMtx_);
2446 327 : JAMI_DEBUG("[Account {}] Receive a new conversation request for conversation {} from {}",
2447 : pimpl_->accountId_,
2448 : req.conversationId,
2449 : from);
2450 327 : auto convId = req.conversationId;
2451 :
2452 : // Already accepted request, do nothing. Exception: a group conversation
2453 : // we previously left can be legitimately re-invited to (same convId
2454 : // reused).
2455 327 : if (pimpl_->isActiveOrNonReinvitableConversation(convId))
2456 122 : return;
2457 : // Prefer the sender-embedded invite timestamp (set by generateInvitation()) so that a
2458 : // passively redelivered/re-asked invite (same timestamp) can be told apart from a
2459 : // genuine new re-invitation (fresh timestamp). Fall back to now() only for older peers
2460 : // that don't embed one yet. The embedded value is untrusted peer input, so clamp it to a
2461 : // reasonable clock-skew window to prevent a far-future timestamp from suppressing later
2462 : // legitimate re-invites in the request de-duplication logic.
2463 205 : if (req.received == TimePoint {})
2464 1 : req.received = nowMs();
2465 : else
2466 204 : req.received = validateInviteTimestamp(req.received, nowMs());
2467 205 : req.from = from;
2468 :
2469 205 : if (isOneToOne) {
2470 2 : auto contactInfo = pimpl_->accountManager_->getContactInfo(from);
2471 2 : if (contactInfo && contactInfo->confirmed && !contactInfo->isBanned() && contactInfo->isActive()) {
2472 1 : JAMI_LOG("[Account {}] Contact {} is confirmed, cloning {}", pimpl_->accountId_, from, convId);
2473 1 : lk.unlock();
2474 1 : updateConvForContact(from, contactInfo->conversationId, convId);
2475 1 : pimpl_->cloneConversationFrom(req);
2476 1 : return;
2477 : }
2478 2 : }
2479 :
2480 204 : auto reqMap = req.toMap();
2481 204 : if (pimpl_->addConversationRequest(convId, std::move(req))) {
2482 168 : lk.unlock();
2483 : // Note: no need to sync here because other connected devices should receive
2484 : // the same conversation request. Will sync when the conversation will be added
2485 168 : if (isOneToOne)
2486 1 : pimpl_->oneToOneRecvCb_(convId, from);
2487 168 : emitSignal<libjami::ConversationSignal::ConversationRequestReceived>(pimpl_->accountId_, convId, reqMap);
2488 : }
2489 579 : }
2490 :
2491 : std::string
2492 3 : ConversationModule::peerFromConversationRequest(const std::string& convId) const
2493 : {
2494 3 : std::lock_guard lk(pimpl_->conversationsRequestsMtx_);
2495 3 : auto it = pimpl_->conversationsRequests_.find(convId);
2496 3 : if (it != pimpl_->conversationsRequests_.end()) {
2497 3 : return it->second.from;
2498 : }
2499 0 : return {};
2500 3 : }
2501 :
2502 : void
2503 159 : ConversationModule::onNeedConversationRequest(const std::string& from, const std::string& conversationId)
2504 : {
2505 159 : if (!ConversationRepository::isValidConversationId(conversationId))
2506 0 : return;
2507 159 : auto conv = pimpl_->getConversation(conversationId);
2508 159 : if (!conv)
2509 0 : return;
2510 159 : std::unique_lock lk(conv->mtx);
2511 159 : if (!conv->conversation)
2512 0 : return;
2513 159 : if (conv->conversation->mode() == ConversationMode::DOCUMENT) {
2514 : // A document is joined by cloning it, never by invitation.
2515 0 : JAMI_WARNING("{} is asking an invite for document {}", from, conversationId);
2516 0 : return;
2517 : }
2518 159 : if (!conv->conversation->isMember(from, true)) {
2519 0 : JAMI_WARNING("{} is asking a new invite for {}, but not a member", from, conversationId);
2520 0 : return;
2521 : }
2522 159 : JAMI_LOG("{} is asking a new invite for {}", from, conversationId);
2523 : // Reuse the original invite timestamp so passive resends aren't treated as new invitations.
2524 159 : auto it = conv->info.invited.find(from);
2525 159 : bool needsPersist = it == conv->info.invited.end();
2526 159 : auto sent = needsPersist ? nowMs() : it->second;
2527 159 : if (needsPersist)
2528 2 : conv->info.invited[from] = sent;
2529 159 : auto info = needsPersist ? conv->info : ConvInfo {};
2530 159 : auto invite = conv->conversation->generateInvitation(sent);
2531 159 : lk.unlock();
2532 159 : if (needsPersist) {
2533 : // convInfosMtx_ must not be taken while conv->mtx is held.
2534 2 : std::lock_guard lkInfos(pimpl_->convInfosMtx_);
2535 2 : pimpl_->convInfos_[conversationId] = std::move(info);
2536 2 : pimpl_->saveConvInfos();
2537 2 : }
2538 159 : pimpl_->sendMsgCb_(from, {}, std::move(invite), 0);
2539 159 : }
2540 :
2541 : void
2542 565 : ConversationModule::acceptConversationRequest(const std::string& conversationId, const std::string& deviceId)
2543 : {
2544 : // For all conversation members, try to open a git channel with this conversation ID
2545 565 : std::unique_lock lkCr(pimpl_->conversationsRequestsMtx_);
2546 565 : auto request = pimpl_->getRequest(conversationId);
2547 565 : if (request == std::nullopt) {
2548 166 : lkCr.unlock();
2549 166 : if (auto conv = pimpl_->getConversation(conversationId)) {
2550 129 : std::unique_lock lk(conv->mtx);
2551 129 : if (!conv->conversation) {
2552 123 : lk.unlock();
2553 123 : pimpl_->cloneConversationFrom(conv, deviceId);
2554 : }
2555 295 : }
2556 166 : JAMI_WARNING("[Account {}] [Conversation {}] [device {}] Request not found.",
2557 : pimpl_->accountId_,
2558 : conversationId,
2559 : deviceId);
2560 166 : return;
2561 : }
2562 399 : pimpl_->rmConversationRequest(conversationId);
2563 399 : lkCr.unlock();
2564 399 : pimpl_->accountManager_->acceptTrustRequest(request->from, true);
2565 399 : pimpl_->cloneConversationFrom(*request);
2566 731 : }
2567 :
2568 : void
2569 6 : ConversationModule::declineConversationRequest(const std::string& conversationId)
2570 : {
2571 6 : std::lock_guard lk(pimpl_->conversationsRequestsMtx_);
2572 6 : auto it = pimpl_->conversationsRequests_.find(conversationId);
2573 6 : if (it != pimpl_->conversationsRequests_.end()) {
2574 6 : it->second.declined = nowMs();
2575 6 : pimpl_->saveConvRequests();
2576 : }
2577 6 : pimpl_->syncingMetadatas_.erase(conversationId);
2578 6 : pimpl_->saveMetadata();
2579 6 : emitSignal<libjami::ConversationSignal::ConversationRequestDeclined>(pimpl_->accountId_, conversationId);
2580 6 : pimpl_->needsSyncingCb_({});
2581 6 : }
2582 :
2583 : std::string
2584 430 : ConversationModule::startConversation(ConversationMode mode, const dht::InfoHash& otherMember)
2585 : {
2586 430 : auto acc = pimpl_->account_.lock();
2587 430 : if (!acc)
2588 0 : return {};
2589 : // Create the conversation object
2590 430 : std::shared_ptr<Conversation> conversation;
2591 : try {
2592 430 : conversation = std::make_shared<Conversation>(acc, mode, otherMember.toString());
2593 430 : auto conversationId = conversation->id();
2594 430 : conversation->onMessageStatusChanged([this, conversationId](const auto& status) {
2595 1119 : auto msg = std::make_shared<SyncMsg>();
2596 2238 : msg->ms = {{conversationId, status}};
2597 1119 : pimpl_->needsSyncingCb_(std::move(msg));
2598 2238 : });
2599 430 : conversation->onMembersChanged([w = pimpl_->weak_from_this(), conversationId](const auto& members) {
2600 : // Delay in another thread to avoid deadlocks
2601 2144 : dht::ThreadPool::io().run([w, conversationId, members = std::move(members)] {
2602 2144 : if (auto sthis = w.lock())
2603 1072 : sthis->setConversationMembers(conversationId, members);
2604 : });
2605 1072 : });
2606 430 : conversation->onNeedSocket(pimpl_->onNeedSwarmSocket_);
2607 : #ifdef LIBJAMI_TEST
2608 430 : conversation->onBootstrapStatus(pimpl_->bootstrapCbTest_);
2609 : #endif
2610 430 : conversation->bootstrap([w = pimpl_->weak_from_this(), conversationId]() {
2611 126 : if (auto sthis = w.lock())
2612 126 : sthis->bootstrapCb(conversationId);
2613 126 : });
2614 430 : if (auto pm = acc->presenceManager()) {
2615 1720 : for (const auto& member : conversation->memberUris()) {
2616 430 : conversation->addKnownDevices(pm->getDevices(member), member);
2617 430 : }
2618 : }
2619 430 : } catch (const std::exception& e) {
2620 0 : JAMI_ERROR("[Account {}] Error while generating a conversation {}", pimpl_->accountId_, e.what());
2621 0 : return {};
2622 0 : }
2623 430 : auto convId = conversation->id();
2624 430 : auto conv = pimpl_->startConversation(convId);
2625 430 : std::unique_lock lk(conv->mtx);
2626 430 : conv->info.created = nowMs();
2627 430 : conv->info.mode = mode;
2628 430 : conv->info.members.emplace(pimpl_->username_);
2629 430 : if (otherMember)
2630 273 : conv->info.members.emplace(otherMember.toString());
2631 430 : conv->conversation = conversation;
2632 430 : addConvInfo(conv->info);
2633 430 : lk.unlock();
2634 :
2635 430 : pimpl_->needsSyncingCb_({});
2636 430 : emitSignal<libjami::ConversationSignal::ConversationReady>(pimpl_->accountId_, convId);
2637 430 : return convId;
2638 430 : }
2639 :
2640 : std::string
2641 25 : ConversationModule::startDocument(const std::string& parentConversationId, const std::string& mimeType)
2642 : {
2643 25 : auto acc = pimpl_->account_.lock();
2644 25 : if (!acc)
2645 0 : return {};
2646 : // The repository is created first, then loaded through the same constructor
2647 : // a held swarm uses, so the wiring below is identical whether the document
2648 : // was created here or cloned from a peer.
2649 25 : std::string docId;
2650 25 : std::shared_ptr<Conversation> conversation;
2651 : try {
2652 : {
2653 25 : auto repo = ConversationRepository::createDocument(acc, parentConversationId, mimeType);
2654 25 : if (!repo)
2655 0 : return {};
2656 25 : docId = repo->id();
2657 25 : }
2658 25 : conversation = std::make_shared<Conversation>(acc, docId);
2659 25 : conversation->onMembersChanged([w = pimpl_->weak_from_this(), docId](const auto& members) {
2660 : // Delay in another thread to avoid deadlocks
2661 154 : dht::ThreadPool::io().run([w, docId, members = std::move(members)] {
2662 154 : if (auto sthis = w.lock())
2663 77 : sthis->setConversationMembers(docId, members);
2664 : });
2665 77 : });
2666 25 : conversation->onNeedSocket(pimpl_->onNeedSwarmSocket_);
2667 : #ifdef LIBJAMI_TEST
2668 25 : conversation->onBootstrapStatus(pimpl_->bootstrapCbTest_);
2669 : #endif
2670 25 : conversation->bootstrap([w = pimpl_->weak_from_this(), docId]() {
2671 19 : if (auto sthis = w.lock())
2672 19 : sthis->bootstrapCb(docId);
2673 19 : });
2674 0 : } catch (const std::exception& e) {
2675 0 : JAMI_ERROR("[Account {}] Error while generating a document {}", pimpl_->accountId_, e.what());
2676 0 : return {};
2677 0 : }
2678 25 : auto conv = pimpl_->startConversation(docId);
2679 25 : std::lock_guard lk(conv->mtx);
2680 25 : conv->info.created = nowMs();
2681 25 : conv->info.mode = ConversationMode::DOCUMENT;
2682 25 : conv->info.parent = parentConversationId;
2683 25 : conv->info.members.emplace(pimpl_->username_);
2684 25 : conv->conversation = conversation;
2685 : // Saved so the document is reloaded as held on restart, but never handed to
2686 : // the clients as a conversation nor synced to this account's other devices:
2687 : // each device discovers documents through the parent conversation.
2688 25 : pimpl_->addConvInfo(conv->info);
2689 25 : return docId;
2690 25 : }
2691 :
2692 : bool
2693 55 : ConversationModule::cloneDocumentFrom(const std::string& parentConversationId,
2694 : const std::string& documentId,
2695 : const std::vector<std::string>& candidates,
2696 : bool reopen)
2697 : {
2698 55 : if (!ConversationRepository::isValidConversationId(documentId)) {
2699 5 : JAMI_WARNING("[Account {}] Refusing to clone invalid document id '{}'", pimpl_->accountId_, documentId);
2700 5 : return false;
2701 : }
2702 50 : if (candidates.empty())
2703 0 : return false;
2704 50 : auto parent = pimpl_->getConversation(parentConversationId);
2705 50 : if (!parent || parentConversationId == documentId)
2706 0 : return false;
2707 50 : auto conv = pimpl_->startConversation(documentId);
2708 : {
2709 50 : std::lock_guard parentLock(parent->mtx);
2710 50 : if (!parent->canReplicateDocuments(pimpl_->username_)) {
2711 1 : JAMI_DEBUG("[Account {}] Deferring document {}: parent is inactive", pimpl_->accountId_, documentId);
2712 1 : return false;
2713 : }
2714 49 : std::lock_guard lk(conv->mtx);
2715 49 : if (!conv->info.parent.empty() && conv->info.parent != parentConversationId) {
2716 0 : JAMI_WARNING("[Account {}] Refusing document {} with conflicting parent", pimpl_->accountId_, documentId);
2717 0 : return false;
2718 : }
2719 49 : if (!reopen && conv->info.removed != TimePoint {} && conv->info.isRemoved()) {
2720 5 : JAMI_DEBUG("[Account {}] [Document {}] Keeping explicitly removed replica offline",
2721 : pimpl_->accountId_,
2722 : documentId);
2723 5 : return false;
2724 : }
2725 44 : if (conv->info.parent.empty()) {
2726 26 : conv->info.parent = parentConversationId;
2727 26 : if (conv->pending)
2728 0 : pimpl_->addConvInfo(conv->info);
2729 : }
2730 44 : if (conv->conversation || conv->pending)
2731 13 : return true;
2732 : // Setting the mode up front spares the mode-mismatch complaint when the
2733 : // clone lands; a document reopened after a local removal is un-removed
2734 : // the same way a re-added conversation is.
2735 31 : conv->info.mode = ConversationMode::DOCUMENT;
2736 31 : conv->info.parent = parentConversationId;
2737 31 : conv->info.created = nowMs();
2738 31 : conv->info.removed = TimePoint {};
2739 31 : conv->info.erased = TimePoint {};
2740 31 : conv->info.members.emplace(pimpl_->username_);
2741 : // Each candidate plays the part an inviter plays for a conversation:
2742 : // it is who the pending clone is authorized to come from, which is
2743 : // what isPeerAuthorized() falls back to before the repo exists.
2744 92 : for (const auto& uri : candidates)
2745 61 : conv->info.members.emplace(uri);
2746 31 : pimpl_->addConvInfo(conv->info);
2747 68 : }
2748 : // The fetch starts from the preferred candidates only: should they fail,
2749 : // the fallback rounds walk everybody recorded above, with backoff, and
2750 : // stop once the clone lands. Asking every candidate up front instead
2751 : // would burst connection attempts at each member of a large conversation.
2752 : static constexpr size_t initialSources = 2;
2753 31 : auto initiated = std::min(candidates.size(), initialSources);
2754 92 : for (size_t i = 0; i < initiated; ++i)
2755 61 : pimpl_->cloneConversationFrom(documentId, candidates[i]);
2756 31 : return true;
2757 50 : }
2758 :
2759 : bool
2760 81 : ConversationModule::isCloningDocument(const std::string& documentId)
2761 : {
2762 81 : auto conv = pimpl_->getConversation(documentId);
2763 81 : if (!conv)
2764 1 : return false;
2765 80 : std::lock_guard lk(conv->mtx);
2766 80 : return !conv->conversation && conv->pending;
2767 81 : }
2768 :
2769 : void
2770 14 : ConversationModule::removeDocumentReplica(const std::string& documentId)
2771 : {
2772 14 : pimpl_->removeDocumentReplica(documentId);
2773 14 : }
2774 :
2775 : std::string
2776 1 : ConversationModule::addDocumentAttachment(const std::string& documentId, const std::vector<uint8_t>& data)
2777 : {
2778 1 : auto conv = pimpl_->getConversation(documentId);
2779 1 : if (!conv)
2780 0 : return {};
2781 1 : std::shared_ptr<Conversation> conversation;
2782 : {
2783 1 : std::lock_guard lk(conv->mtx);
2784 1 : conversation = conv->conversation;
2785 1 : }
2786 1 : if (!conversation || conversation->mode() != ConversationMode::DOCUMENT)
2787 0 : return {};
2788 1 : auto [attachmentId, commitId] = conversation->addDocumentAttachment(data);
2789 1 : if (!commitId.empty())
2790 3 : pimpl_->sendMessageNotification(documentId, true, commitId);
2791 1 : return attachmentId;
2792 1 : }
2793 :
2794 : void
2795 0 : ConversationModule::cloneConversationFrom(const std::string& conversationId, const std::string& uri)
2796 : {
2797 0 : pimpl_->cloneConversationFrom(conversationId, uri);
2798 0 : }
2799 :
2800 : // Message send/load
2801 : void
2802 94 : ConversationModule::sendMessage(const std::string& conversationId,
2803 : std::string message,
2804 : const std::string& replyTo,
2805 : bool announce,
2806 : OnCommitCb&& onCommit,
2807 : OnDoneCb&& cb)
2808 : {
2809 94 : pimpl_->sendMessage(conversationId, std::move(message), replyTo, announce, std::move(onCommit), std::move(cb));
2810 94 : }
2811 :
2812 : void
2813 55 : ConversationModule::createCommit(
2814 : const std::string& conversationId, CommitMessage&& message, bool announce, OnCommitCb&& onCommit, OnDoneCb&& cb)
2815 : {
2816 55 : pimpl_->createCommit(conversationId, std::move(message), announce, std::move(onCommit), std::move(cb));
2817 55 : }
2818 :
2819 : void
2820 8 : ConversationModule::editMessage(const std::string& conversationId,
2821 : const std::string& newBody,
2822 : const std::string& editedId)
2823 : {
2824 8 : pimpl_->editMessage(conversationId, newBody, editedId);
2825 8 : }
2826 :
2827 : void
2828 3 : ConversationModule::reactToMessage(const std::string& conversationId,
2829 : const std::string& newBody,
2830 : const std::string& reactToId)
2831 : {
2832 3 : auto message = CommitMessage::reaction(newBody, reactToId);
2833 3 : pimpl_->createCommit(conversationId, std::move(message));
2834 3 : }
2835 :
2836 : void
2837 106 : ConversationModule::addCallHistoryMessage(const std::string& uri, uint64_t duration_ms, const std::string& reason)
2838 : {
2839 106 : auto finalUri = uri.substr(0, uri.find("@ring.dht"));
2840 106 : finalUri = finalUri.substr(0, uri.find("@jami.dht"));
2841 106 : auto convId = getOneToOneConversation(finalUri);
2842 106 : if (!convId.empty()) {
2843 3 : auto message = CommitMessage::outgoingCallEnd(finalUri, duration_ms, reason);
2844 3 : pimpl_->createCommit(convId, std::move(message));
2845 3 : }
2846 106 : }
2847 :
2848 : bool
2849 16 : ConversationModule::onMessageDisplayed(const std::string& peer,
2850 : const std::string& conversationId,
2851 : const std::string& interactionId)
2852 : {
2853 16 : if (auto conv = getConversation(conversationId))
2854 16 : return conv->setMessageDisplayed(peer, interactionId);
2855 0 : return false;
2856 : }
2857 :
2858 : std::map<std::string, std::map<std::string, std::map<std::string, std::string>>>
2859 203 : ConversationModule::convMessageStatus() const
2860 : {
2861 203 : std::map<std::string, std::map<std::string, std::map<std::string, std::string>>> messageStatus;
2862 312 : for (const auto& conv : pimpl_->getConversations()) {
2863 109 : auto d = conv->messageStatus();
2864 109 : if (!d.empty())
2865 74 : messageStatus[conv->id()] = std::move(d);
2866 312 : }
2867 203 : return messageStatus;
2868 0 : }
2869 :
2870 : void
2871 0 : ConversationModule::clearCache(const std::string& conversationId)
2872 : {
2873 0 : if (auto conv = pimpl_->getConversation(conversationId)) {
2874 0 : std::lock_guard lk(conv->mtx);
2875 0 : if (conv->conversation) {
2876 0 : conv->conversation->clearCache();
2877 : }
2878 0 : }
2879 0 : }
2880 :
2881 : uint32_t
2882 3 : ConversationModule::loadConversation(const std::string& conversationId, const std::string& fromMessage, size_t n)
2883 : {
2884 3 : auto acc = pimpl_->account_.lock();
2885 3 : if (auto conv = pimpl_->getConversation(conversationId)) {
2886 3 : std::lock_guard lk(conv->mtx);
2887 3 : if (conv->conversation) {
2888 3 : const uint32_t id = std::uniform_int_distribution<uint32_t> {1}(acc->rand);
2889 3 : LogOptions options;
2890 3 : options.from = fromMessage;
2891 3 : options.nbOfCommits = n;
2892 3 : auto convWeak = std::weak_ptr<Conversation>(conv->conversation);
2893 6 : conv->conversation->loadMessages(
2894 9 : [accountId = pimpl_->accountId_,
2895 : conversationId,
2896 : id,
2897 : convWeak,
2898 3 : permissions = conv->conversation->permissions(pimpl_->username_)](auto&& messages) {
2899 6 : emitSignal<libjami::ConversationSignal::SwarmLoaded>(id,
2900 3 : accountId,
2901 3 : conversationId,
2902 : messages,
2903 3 : permissions);
2904 : #ifdef ENABLE_PLUGIN
2905 6 : if (const auto convPtr = convWeak.lock()) {
2906 3 : convPtr->loadMissingBodyOverwrites();
2907 : }
2908 : #endif
2909 3 : },
2910 : options);
2911 3 : return id;
2912 3 : }
2913 6 : }
2914 0 : return 0;
2915 3 : }
2916 :
2917 : uint32_t
2918 0 : ConversationModule::loadSwarmUntil(const std::string& conversationId,
2919 : const std::string& fromMessage,
2920 : const std::string& toMessage)
2921 : {
2922 0 : auto acc = pimpl_->account_.lock();
2923 0 : if (auto conv = pimpl_->getConversation(conversationId)) {
2924 0 : std::lock_guard lk(conv->mtx);
2925 0 : if (conv->conversation) {
2926 0 : const uint32_t id = std::uniform_int_distribution<uint32_t> {}(acc->rand);
2927 0 : LogOptions options;
2928 0 : options.from = fromMessage;
2929 0 : options.to = toMessage;
2930 0 : options.includeTo = true;
2931 0 : auto convWeak = std::weak_ptr<Conversation>(conv->conversation);
2932 0 : conv->conversation->loadMessages(
2933 0 : [accountId = pimpl_->accountId_,
2934 : conversationId,
2935 : id,
2936 : convWeak,
2937 0 : permissions = conv->conversation->permissions(pimpl_->username_)](auto&& messages) {
2938 0 : emitSignal<libjami::ConversationSignal::SwarmLoaded>(id,
2939 0 : accountId,
2940 0 : conversationId,
2941 : messages,
2942 0 : permissions);
2943 : #ifdef ENABLE_PLUGIN
2944 0 : if (const auto convPtr = convWeak.lock()) {
2945 0 : convPtr->loadMissingBodyOverwrites();
2946 : }
2947 : #endif
2948 0 : },
2949 : options);
2950 0 : return id;
2951 0 : }
2952 0 : }
2953 0 : return 0;
2954 0 : }
2955 :
2956 : std::shared_ptr<TransferManager>
2957 287 : ConversationModule::dataTransfer(const std::string& conversationId) const
2958 : {
2959 859 : return pimpl_->withConversation(conversationId, [](auto& conversation) { return conversation.dataTransfer(); });
2960 : }
2961 :
2962 : std::optional<bool>
2963 3 : ConversationModule::fileTransferInfo(
2964 : const std::string& conversationId, const std::string& fileId, std::string& path, int64_t& total, int64_t& progress)
2965 : {
2966 3 : auto conv = pimpl_->getConversation(conversationId);
2967 3 : if (!conv)
2968 1 : return false;
2969 2 : std::shared_ptr<TransferManager> transferManager;
2970 : {
2971 2 : std::lock_guard lk(conv->mtx);
2972 2 : if (conv->conversation)
2973 2 : transferManager = conv->conversation->dataTransfer();
2974 2 : }
2975 2 : if (!transferManager)
2976 0 : return std::nullopt;
2977 2 : return transferManager->info(fileId, path, total, progress);
2978 3 : }
2979 :
2980 : bool
2981 21 : ConversationModule::onFileChannelRequest(const std::string& conversationId,
2982 : const std::string& member,
2983 : const std::string& fileId,
2984 : bool verifyShaSum) const
2985 : {
2986 21 : if (auto conv = pimpl_->getConversation(conversationId)) {
2987 21 : std::filesystem::path path;
2988 21 : std::string sha3sum;
2989 21 : std::unique_lock lk(conv->mtx);
2990 21 : if (!conv->conversation)
2991 0 : return false;
2992 21 : if (!conv->conversation->onFileChannelRequest(member, fileId, path, sha3sum))
2993 4 : return false;
2994 :
2995 : // Release the lock here to prevent the sha3 calculation from blocking other threads.
2996 17 : lk.unlock();
2997 17 : if (!std::filesystem::is_regular_file(path)) {
2998 2 : JAMI_WARNING("[Account {:s}] [Conversation {}] {:s} asked for non existing file {}",
2999 : pimpl_->accountId_,
3000 : conversationId,
3001 : member,
3002 : fileId);
3003 2 : return false;
3004 : }
3005 : // Check that our file is correct before sending
3006 15 : if (verifyShaSum && sha3sum != fileutils::sha3File(path)) {
3007 1 : JAMI_WARNING("[Account {:s}] [Conversation {}] {:s} asked for file {:s}, but our version is not "
3008 : "complete or corrupted",
3009 : pimpl_->accountId_,
3010 : conversationId,
3011 : member,
3012 : fileId);
3013 1 : return false;
3014 : }
3015 14 : return true;
3016 42 : }
3017 0 : return false;
3018 : }
3019 :
3020 : bool
3021 26 : ConversationModule::downloadFile(const std::string& conversationId,
3022 : const std::string& interactionId,
3023 : const std::string& fileId,
3024 : const std::string& path)
3025 : {
3026 26 : if (auto conv = pimpl_->getConversation(conversationId)) {
3027 26 : std::lock_guard lk(conv->mtx);
3028 26 : if (conv->conversation)
3029 130 : return conv->conversation->downloadFile(interactionId, fileId, path, "", "");
3030 52 : }
3031 0 : return false;
3032 : }
3033 :
3034 : void
3035 1362 : ConversationModule::syncConversations(const std::string& peer, const std::string& deviceId)
3036 : {
3037 : // Sync conversations where peer is member
3038 1362 : std::set<std::string> toFetch;
3039 1362 : std::set<std::string> toClone;
3040 3620 : for (const auto& conv : pimpl_->getSyncedConversations()) {
3041 2258 : std::lock_guard lk(conv->mtx);
3042 2258 : if (conv->conversation) {
3043 1960 : if (!conv->conversation->isRemoving() && conv->conversation->isMember(peer, false)) {
3044 686 : toFetch.emplace(conv->info.id);
3045 : }
3046 298 : } else if (!conv->info.isRemoved()
3047 886 : && std::find(conv->info.members.begin(), conv->info.members.end(), peer)
3048 886 : != conv->info.members.end()) {
3049 : // In this case the conversation was never cloned (can be after an import)
3050 256 : toClone.emplace(conv->info.id);
3051 : }
3052 3620 : }
3053 2048 : for (const auto& cid : toFetch)
3054 2058 : pimpl_->fetchNewCommits(peer, deviceId, cid);
3055 1618 : for (const auto& cid : toClone)
3056 256 : pimpl_->cloneConversation(deviceId, peer, cid);
3057 2724 : if (pimpl_->syncCnt.load() == 0)
3058 700 : emitSignal<libjami::ConversationSignal::ConversationSyncFinished>(pimpl_->accountId_);
3059 1362 : }
3060 :
3061 : void
3062 218 : ConversationModule::onSyncData(const SyncMsg& msg, const std::string& peerId, const std::string& deviceId)
3063 : {
3064 218 : std::vector<std::string> toClone;
3065 218 : std::vector<std::shared_ptr<Conversation>> toLink;
3066 : // Conversations whose removal was learned from this message. Their
3067 : // documents must be forfeited too, and documents are excluded from the
3068 : // device sync (holding one is a per-device choice), so the forfeit is
3069 : // inferred here from the parent's removal.
3070 218 : std::vector<std::string> removedConversations;
3071 : // Set when the conversation list or requests actually change, so the change
3072 : // can be re-propagated to our other devices (relay). Metadata-only updates
3073 : // (preferences, message status) deliberately do not set it.
3074 218 : bool listChanged = false;
3075 393 : for (const auto& [key, convInfo] : msg.c) {
3076 175 : const auto& convId = convInfo.id;
3077 175 : if (!isConversationId(convId))
3078 0 : continue;
3079 : {
3080 175 : std::lock_guard lk(pimpl_->conversationsRequestsMtx_);
3081 175 : pimpl_->rmConversationRequest(convId);
3082 175 : }
3083 :
3084 : // Whether this conversation is new to our list. Used to decide if the
3085 : // change must be relayed: a conversation we already know but have not
3086 : // finished cloning yet must not be treated as a change, otherwise it
3087 : // would re-trigger syncing on every received sync.
3088 175 : bool isNewConv = not pimpl_->isConversation(convId);
3089 175 : auto conv = pimpl_->startConversation(convInfo);
3090 175 : std::unique_lock lk(conv->mtx);
3091 : // Skip outdated info
3092 175 : if (std::max(convInfo.created, convInfo.removed) < std::max(conv->info.created, conv->info.removed))
3093 7 : continue;
3094 168 : if (not convInfo.isRemoved()) {
3095 : // If multi devices, it can detect a conversation that was already
3096 : // removed, so just check if the convinfo contains a removed conv
3097 161 : if (conv->info.removed != TimePoint {}) {
3098 0 : if (conv->info.removed >= convInfo.created) {
3099 : // Only reclone if re-added, else the peer is not synced yet (could be
3100 : // offline before)
3101 0 : continue;
3102 : }
3103 0 : JAMI_DEBUG("Re-add previously removed conversation {:s}", convId);
3104 : }
3105 161 : conv->info = convInfo;
3106 161 : if (!conv->conversation) {
3107 76 : if (isNewConv)
3108 16 : listChanged = true;
3109 76 : if (deviceId != "") {
3110 76 : pimpl_->cloneConversation(deviceId, peerId, conv);
3111 : } else {
3112 : // In this case, information is from JAMS
3113 : // JAMS does not store the conversation itself, so we
3114 : // must use information to clone the conversation
3115 0 : addConvInfo(convInfo);
3116 0 : toClone.emplace_back(convId);
3117 : }
3118 85 : } else if (conv->info.mode == ConversationMode::ONE_TO_ONE) {
3119 60 : toLink.emplace_back(conv->conversation);
3120 : }
3121 : } else {
3122 7 : if (conv->conversation && !conv->conversation->isRemoving()) {
3123 5 : emitSignal<libjami::ConversationSignal::ConversationRemoved>(pimpl_->accountId_, convId);
3124 5 : conv->conversation->setRemovingFlag();
3125 : }
3126 7 : auto update = false;
3127 7 : if (conv->info.removed == TimePoint {}) {
3128 5 : update = true;
3129 5 : listChanged = true;
3130 5 : conv->info.removed = nowMs();
3131 5 : emitSignal<libjami::ConversationSignal::ConversationRemoved>(pimpl_->accountId_, convId);
3132 : }
3133 7 : if (convInfo.erased != TimePoint {} && conv->info.erased == TimePoint {}) {
3134 2 : update = true;
3135 2 : listChanged = true;
3136 2 : conv->info.erased = nowMs();
3137 2 : pimpl_->addConvInfo(conv->info);
3138 2 : pimpl_->removeRepositoryImpl(*conv, false);
3139 5 : } else if (update) {
3140 3 : pimpl_->addConvInfo(conv->info);
3141 : }
3142 7 : if (update)
3143 5 : removedConversations.emplace_back(convId);
3144 : }
3145 182 : }
3146 :
3147 : // The account left the removed conversations, forfeiting their documents
3148 : // on this device as well. The removal already carried the account's leave
3149 : // to the other holders, so each local replica is dropped silently.
3150 : // Collected outside the loop above: the conversation map lock and a
3151 : // conversation's own lock nest the other way around.
3152 219 : for (const auto& documentId : pimpl_->childDocuments(removedConversations))
3153 219 : pimpl_->removeDocumentReplica(documentId);
3154 :
3155 : // Reuse the normal clear path's lookup only for contacts whose linked
3156 : // conversation was removed by this sync.
3157 223 : for (const auto& removedId : removedConversations) {
3158 5 : auto removed = pimpl_->getConversation(removedId);
3159 5 : if (!removed)
3160 0 : continue;
3161 5 : std::set<std::string> members;
3162 : {
3163 5 : std::lock_guard lk(removed->mtx);
3164 5 : if (removed->info.mode != ConversationMode::ONE_TO_ONE)
3165 2 : continue;
3166 3 : members = removed->info.members;
3167 5 : }
3168 3 : members.emplace(pimpl_->username_);
3169 3 : const auto replacement = findMatchingOneToOneConversation(removedId, members);
3170 3 : if (auto conversation = getConversation(replacement))
3171 3 : toLink.emplace_back(std::move(conversation));
3172 7 : }
3173 279 : for (const auto& conversation : toLink)
3174 61 : pimpl_->linkReplacementConversation(*conversation);
3175 :
3176 218 : for (const auto& cid : toClone) {
3177 0 : auto members = getConversationMembers(cid);
3178 0 : for (const auto& member : members) {
3179 0 : if (member.at("uri") != pimpl_->username_)
3180 0 : cloneConversationFrom(cid, member.at("uri"));
3181 : }
3182 0 : }
3183 :
3184 234 : for (const auto& [convId, req] : msg.cr) {
3185 16 : if (!ConversationRepository::isValidConversationId(convId)) {
3186 0 : JAMI_WARNING("[Account {}] [device {}] Ignoring synced request with invalid conversation id '{}'",
3187 : pimpl_->accountId_,
3188 : deviceId,
3189 : convId);
3190 0 : continue;
3191 : }
3192 16 : if (req.from == pimpl_->username_) {
3193 0 : JAMI_WARNING("Detected request from ourself, ignore {}.", convId);
3194 0 : continue;
3195 : }
3196 16 : std::unique_lock lk(pimpl_->conversationsRequestsMtx_);
3197 16 : if (pimpl_->isConversation(convId)) {
3198 : // Already handled request
3199 2 : pimpl_->rmConversationRequest(convId);
3200 2 : continue;
3201 : }
3202 :
3203 : // New request
3204 14 : const auto requestChanged = pimpl_->addConversationRequest(convId, req);
3205 14 : auto currentRequest = pimpl_->conversationsRequests_.find(convId);
3206 14 : if (currentRequest != pimpl_->conversationsRequests_.end() && currentRequest->second.declined != TimePoint {})
3207 18 : pimpl_->accountManager_->discardTrustRequest(currentRequest->second.from,
3208 : convId,
3209 9 : currentRequest->second.declined);
3210 14 : if (!requestChanged)
3211 9 : continue;
3212 : // A request was added or transitioned to declined: the request list
3213 : // changed and must be re-propagated to our other devices.
3214 5 : listChanged = true;
3215 :
3216 5 : if (req.declined != TimePoint {}) {
3217 : // Request declined
3218 4 : JAMI_LOG("[Account {:s}] Declined request detected for conversation {:s} (device {:s})",
3219 : pimpl_->accountId_,
3220 : convId,
3221 : deviceId);
3222 4 : pimpl_->syncingMetadatas_.erase(convId);
3223 4 : pimpl_->saveMetadata();
3224 4 : lk.unlock();
3225 4 : emitSignal<libjami::ConversationSignal::ConversationRequestDeclined>(pimpl_->accountId_, convId);
3226 4 : continue;
3227 : }
3228 1 : lk.unlock();
3229 :
3230 1 : JAMI_LOG("[Account {:s}] New request detected for conversation {:s} (device {:s})",
3231 : pimpl_->accountId_,
3232 : convId,
3233 : deviceId);
3234 :
3235 1 : emitSignal<libjami::ConversationSignal::ConversationRequestReceived>(pimpl_->accountId_, convId, req.toMap());
3236 16 : }
3237 :
3238 : // Updates preferences for conversations
3239 222 : for (const auto& [convId, p] : msg.p) {
3240 4 : if (auto conv = pimpl_->getConversation(convId)) {
3241 4 : std::unique_lock lk(conv->mtx);
3242 4 : if (conv->conversation) {
3243 2 : auto conversation = conv->conversation;
3244 2 : lk.unlock();
3245 2 : conversation->updatePreferences(p);
3246 4 : } else if (conv->pending) {
3247 2 : conv->pending->preferences = p;
3248 : }
3249 8 : }
3250 : }
3251 :
3252 : // Updates displayed for conversations
3253 315 : for (const auto& [convId, ms] : msg.ms) {
3254 97 : if (auto conv = pimpl_->getConversation(convId)) {
3255 92 : std::unique_lock lk(conv->mtx);
3256 92 : if (conv->conversation) {
3257 44 : auto conversation = conv->conversation;
3258 44 : lk.unlock();
3259 44 : conversation->updateMessageStatus(ms);
3260 92 : } else if (conv->pending) {
3261 41 : conv->pending->status = ms;
3262 : }
3263 189 : }
3264 : }
3265 :
3266 : // If the conversation list or requests changed as a result of this sync,
3267 : // re-propagate to our other devices (relay) so a change learned from one
3268 : // device reaches the others. This only applies to peer-to-peer syncs:
3269 : // when the data comes from the JAMS server (empty deviceId), the server is
3270 : // the synchronization hub and already holds the change, so re-propagating
3271 : // would push it straight back and cause an infinite sync loop.
3272 218 : if (listChanged && !deviceId.empty())
3273 23 : pimpl_->needsSyncingCb_({});
3274 218 : }
3275 :
3276 : bool
3277 2 : ConversationModule::needsSyncingWith(const std::string& memberUri) const
3278 : {
3279 : // Check if a conversation needs to fetch remote or to be cloned
3280 2 : std::lock_guard lk(pimpl_->conversationsMtx_);
3281 2 : for (const auto& [key, ci] : pimpl_->conversations_) {
3282 1 : std::lock_guard lk(ci->mtx);
3283 1 : if (ci->conversation) {
3284 0 : if (ci->conversation->isRemoving() && ci->conversation->isMember(memberUri, false))
3285 0 : return true;
3286 1 : } else if (ci->info.removed == TimePoint {}
3287 1 : && std::find(ci->info.members.begin(), ci->info.members.end(), memberUri) != ci->info.members.end()) {
3288 : // In this case the conversation was never cloned (can be after an import)
3289 1 : return true;
3290 : }
3291 1 : }
3292 1 : return false;
3293 2 : }
3294 :
3295 : void
3296 2242 : ConversationModule::setFetched(const std::string& conversationId,
3297 : const std::string& deviceId,
3298 : const std::string& commitId)
3299 : {
3300 2242 : if (auto conv = pimpl_->getConversation(conversationId)) {
3301 2242 : std::lock_guard lk(conv->mtx);
3302 2242 : if (conv->conversation) {
3303 2242 : bool remove = conv->conversation->isRemoving();
3304 2242 : conv->conversation->hasFetched(deviceId, commitId);
3305 2242 : if (remove)
3306 9 : pimpl_->removeRepositoryImpl(*conv, true);
3307 : }
3308 4484 : }
3309 2242 : }
3310 :
3311 : void
3312 12084 : ConversationModule::fetchNewCommits(const std::string& peer,
3313 : const std::string& deviceId,
3314 : const std::string& conversationId,
3315 : const std::string& commitId)
3316 : {
3317 12084 : if (!ConversationRepository::isValidConversationId(conversationId)) {
3318 6 : JAMI_WARNING("[Account {}] [device {}] Ignoring commit notification for invalid conversation id '{}'",
3319 : pimpl_->accountId_,
3320 : deviceId,
3321 : conversationId);
3322 6 : return;
3323 : }
3324 12078 : pimpl_->fetchNewCommits(peer, deviceId, conversationId, commitId);
3325 : }
3326 :
3327 : void
3328 178 : ConversationModule::addConversationMember(const std::string& conversationId,
3329 : const dht::InfoHash& contactUri,
3330 : bool sendRequest)
3331 : {
3332 178 : auto conv = pimpl_->getConversation(conversationId);
3333 178 : if (not conv || not conv->conversation) {
3334 0 : JAMI_ERROR("Conversation {:s} does not exist", conversationId);
3335 0 : return;
3336 : }
3337 178 : std::unique_lock lk(conv->mtx);
3338 :
3339 178 : auto contactUriStr = contactUri.toString();
3340 178 : if (conv->conversation->isMember(contactUriStr, true)) {
3341 0 : JAMI_DEBUG("{:s} is already a member of {:s}, resend invite", contactUriStr, conversationId);
3342 : // Note: This should not be necessary, but if for whatever reason the other side didn't
3343 : // join we should not forbid new invites
3344 0 : auto sent = nowMs();
3345 0 : conv->info.invited[contactUriStr] = sent;
3346 0 : auto info = conv->info;
3347 0 : auto invite = conv->conversation->generateInvitation(sent);
3348 0 : lk.unlock();
3349 : // convInfosMtx_ must not be taken while conv->mtx is held
3350 : // so update convInfos_/save only after unlocking conv->mtx.
3351 : {
3352 0 : std::lock_guard lkInfos(pimpl_->convInfosMtx_);
3353 0 : pimpl_->convInfos_[conversationId] = std::move(info);
3354 0 : pimpl_->saveConvInfos();
3355 0 : }
3356 0 : pimpl_->sendMsgCb_(contactUriStr, {}, std::move(invite), 0);
3357 0 : return;
3358 0 : }
3359 :
3360 356 : conv->conversation->addMember(contactUriStr,
3361 356 : [this, conv, conversationId, sendRequest, contactUriStr](bool ok,
3362 : const std::string& commitId) {
3363 178 : if (ok) {
3364 175 : std::unique_lock lk(conv->mtx);
3365 350 : pimpl_->sendMessageNotification(*conv->conversation,
3366 : true,
3367 : commitId); // For the other members
3368 175 : if (sendRequest) {
3369 171 : auto sent = nowMs();
3370 171 : conv->info.invited[contactUriStr] = sent;
3371 171 : auto info = conv->info;
3372 171 : auto invite = conv->conversation->generateInvitation(sent);
3373 171 : lk.unlock();
3374 : // convInfosMtx_ must be taken after releasing conv->mtx
3375 : {
3376 171 : std::lock_guard lkInfos(pimpl_->convInfosMtx_);
3377 171 : pimpl_->convInfos_[conversationId] = std::move(info);
3378 171 : pimpl_->saveConvInfos();
3379 171 : }
3380 171 : pimpl_->sendMsgCb_(contactUriStr, {}, std::move(invite), 0);
3381 171 : }
3382 175 : }
3383 178 : });
3384 178 : }
3385 :
3386 : void
3387 14 : ConversationModule::removeConversationMember(const std::string& conversationId,
3388 : const dht::InfoHash& contactUri,
3389 : bool isDevice)
3390 : {
3391 14 : auto contactUriStr = contactUri.toString();
3392 14 : if (auto conv = pimpl_->getConversation(conversationId)) {
3393 14 : std::lock_guard lk(conv->mtx);
3394 14 : if (conv->conversation)
3395 14 : return conv->conversation
3396 14 : ->removeMember(contactUriStr, isDevice, [this, conversationId](bool ok, const std::string& commitId) {
3397 14 : if (ok) {
3398 36 : pimpl_->sendMessageNotification(conversationId, true, commitId);
3399 : }
3400 28 : });
3401 28 : }
3402 14 : }
3403 :
3404 : std::vector<std::map<std::string, std::string>>
3405 269 : ConversationModule::getConversationMembers(const std::string& conversationId, bool includeBanned) const
3406 : {
3407 269 : return pimpl_->getConversationMembers(conversationId, includeBanned);
3408 : }
3409 :
3410 : std::vector<std::string>
3411 2 : ConversationModule::getConversationPermissions(const std::string& conversationId) const
3412 : {
3413 2 : return pimpl_->withConversation(conversationId,
3414 5 : [&](const auto& conv) { return conv.permissions(pimpl_->username_); });
3415 : }
3416 :
3417 : uint32_t
3418 4 : ConversationModule::countInteractions(const std::string& convId,
3419 : const std::string& toId,
3420 : const std::string& fromId,
3421 : const std::string& authorUri) const
3422 : {
3423 4 : if (auto conv = pimpl_->getConversation(convId)) {
3424 4 : std::lock_guard lk(conv->mtx);
3425 4 : if (conv->conversation)
3426 4 : return conv->conversation->countInteractions(toId, fromId, authorUri);
3427 8 : }
3428 0 : return 0;
3429 : }
3430 :
3431 : void
3432 4 : ConversationModule::search(uint32_t req, const std::string& convId, const Filter& filter) const
3433 : {
3434 4 : if (convId.empty()) {
3435 0 : auto convs = pimpl_->getConversations();
3436 0 : if (convs.empty()) {
3437 0 : emitSignal<libjami::ConversationSignal::MessagesFound>(req,
3438 0 : pimpl_->accountId_,
3439 0 : std::string {},
3440 0 : std::vector<std::map<std::string, std::string>> {});
3441 0 : return;
3442 : }
3443 0 : auto finishedFlag = std::make_shared<std::atomic_int>(convs.size());
3444 0 : for (const auto& conv : convs) {
3445 0 : conv->search(req, filter, finishedFlag);
3446 : }
3447 4 : } else if (auto conv = pimpl_->getConversation(convId)) {
3448 4 : std::lock_guard lk(conv->mtx);
3449 4 : if (conv->conversation)
3450 4 : conv->conversation->search(req, filter, std::make_shared<std::atomic_int>(1));
3451 8 : }
3452 : }
3453 :
3454 : void
3455 36 : ConversationModule::updateConversationInfos(const std::string& conversationId,
3456 : const std::map<std::string, std::string>& infos,
3457 : bool sync)
3458 : {
3459 36 : auto conv = pimpl_->getConversation(conversationId);
3460 36 : if (not conv or not conv->conversation) {
3461 0 : JAMI_ERROR("Conversation {:s} does not exist", conversationId);
3462 0 : return;
3463 : }
3464 36 : std::lock_guard lk(conv->mtx);
3465 36 : conv->conversation->updateInfos(infos, [this, conversationId, sync](bool ok, const std::string& commitId) {
3466 36 : if (ok && sync) {
3467 27 : pimpl_->sendMessageNotification(conversationId, true, commitId);
3468 27 : } else if (sync)
3469 2 : JAMI_WARNING("Unable to update info on {:s}", conversationId);
3470 36 : });
3471 36 : }
3472 :
3473 : std::map<std::string, std::string>
3474 5 : ConversationModule::conversationInfos(const std::string& conversationId) const
3475 : {
3476 : {
3477 5 : std::lock_guard lk(pimpl_->conversationsRequestsMtx_);
3478 5 : auto itReq = pimpl_->conversationsRequests_.find(conversationId);
3479 5 : if (itReq != pimpl_->conversationsRequests_.end())
3480 1 : return itReq->second.metadatas;
3481 5 : }
3482 4 : if (auto conv = pimpl_->getConversation(conversationId)) {
3483 4 : std::lock_guard lk(conv->mtx);
3484 4 : std::map<std::string, std::string> md;
3485 : {
3486 4 : std::lock_guard lk(pimpl_->conversationsRequestsMtx_);
3487 4 : auto syncingMetadatasIt = pimpl_->syncingMetadatas_.find(conversationId);
3488 4 : if (syncingMetadatasIt != pimpl_->syncingMetadatas_.end()) {
3489 2 : if (conv->conversation) {
3490 0 : pimpl_->syncingMetadatas_.erase(syncingMetadatasIt);
3491 0 : pimpl_->saveMetadata();
3492 : } else {
3493 2 : md = syncingMetadatasIt->second;
3494 : }
3495 : }
3496 4 : }
3497 4 : if (conv->conversation) {
3498 2 : auto infos = conv->conversation->infos();
3499 : // Expose the locally-tracked conversation timestamps so that clients can
3500 : // display when the conversation was created/accepted (joined) and, when
3501 : // applicable, removed. Legacy keys carry seconds; *Ms keys carry
3502 : // milliseconds. Unknown keys are ignored by older devices.
3503 2 : if (conv->info.created != TimePoint {}) {
3504 6 : infos[ConversationMapKeys::CREATED] = std::to_string(toSecondsSinceEpoch(conv->info.created));
3505 6 : infos[ConversationMapKeys::CREATED_MS] = std::to_string(toMillisecondsSinceEpoch(conv->info.created));
3506 : }
3507 2 : if (conv->info.removed != TimePoint {}) {
3508 0 : infos[ConversationMapKeys::REMOVED] = std::to_string(toSecondsSinceEpoch(conv->info.removed));
3509 0 : infos[ConversationMapKeys::REMOVED_MS] = std::to_string(toMillisecondsSinceEpoch(conv->info.removed));
3510 : }
3511 2 : return infos;
3512 2 : } else
3513 2 : return md;
3514 8 : }
3515 0 : JAMI_ERROR("Conversation {:s} does not exist", conversationId);
3516 0 : return {};
3517 : }
3518 :
3519 : void
3520 5 : ConversationModule::setConversationPreferences(const std::string& conversationId,
3521 : const std::map<std::string, std::string>& prefs)
3522 : {
3523 5 : if (auto conv = pimpl_->getConversation(conversationId)) {
3524 5 : std::unique_lock lk(conv->mtx);
3525 5 : if (not conv->conversation) {
3526 0 : JAMI_ERROR("Conversation {:s} does not exist", conversationId);
3527 0 : return;
3528 : }
3529 5 : auto conversation = conv->conversation;
3530 5 : lk.unlock();
3531 5 : conversation->updatePreferences(prefs);
3532 5 : auto msg = std::make_shared<SyncMsg>();
3533 15 : msg->p = {{conversationId, conversation->preferences(true)}};
3534 5 : pimpl_->needsSyncingCb_(std::move(msg));
3535 10 : }
3536 10 : }
3537 :
3538 : std::map<std::string, std::string>
3539 14 : ConversationModule::getConversationPreferences(const std::string& conversationId, bool includeCreated) const
3540 : {
3541 14 : if (auto conv = pimpl_->getConversation(conversationId)) {
3542 14 : std::lock_guard lk(conv->mtx);
3543 14 : if (conv->conversation)
3544 13 : return conv->conversation->preferences(includeCreated);
3545 28 : }
3546 1 : return {};
3547 : }
3548 :
3549 : std::map<std::string, std::map<std::string, std::string>>
3550 203 : ConversationModule::convPreferences() const
3551 : {
3552 203 : std::map<std::string, std::map<std::string, std::string>> p;
3553 312 : for (const auto& conv : pimpl_->getConversations()) {
3554 109 : auto prefs = conv->preferences(true);
3555 109 : if (!prefs.empty())
3556 3 : p[conv->id()] = std::move(prefs);
3557 312 : }
3558 203 : return p;
3559 0 : }
3560 :
3561 : std::vector<uint8_t>
3562 0 : ConversationModule::conversationVCard(const std::string& conversationId) const
3563 : {
3564 0 : if (auto conv = pimpl_->getConversation(conversationId)) {
3565 0 : std::lock_guard lk(conv->mtx);
3566 0 : if (conv->conversation)
3567 0 : return conv->conversation->vCard();
3568 0 : }
3569 0 : JAMI_ERROR("Conversation {:s} does not exist", conversationId);
3570 0 : return {};
3571 : }
3572 :
3573 : bool
3574 0 : ConversationModule::isMemberBanned(const std::string& convId, const std::string& uri) const
3575 : {
3576 : dhtnet::tls::TrustStore::PermissionStatus status;
3577 : {
3578 0 : std::lock_guard lk(pimpl_->conversationsMtx_);
3579 0 : status = pimpl_->accountManager_->getCertificateStatus(uri);
3580 0 : }
3581 0 : if (auto conv = pimpl_->getConversation(convId)) {
3582 0 : std::lock_guard lk(conv->mtx);
3583 0 : if (!conv->conversation)
3584 0 : return true;
3585 0 : if (conv->conversation->mode() != ConversationMode::ONE_TO_ONE)
3586 0 : return conv->conversation->isMemberBanned(uri);
3587 : // If 1:1 we check the certificate status
3588 0 : return status == dhtnet::tls::TrustStore::PermissionStatus::BANNED;
3589 0 : }
3590 0 : return true;
3591 : }
3592 :
3593 : bool
3594 1788 : ConversationModule::isDeviceBanned(const std::string& convId, const std::string& deviceId) const
3595 : {
3596 : dhtnet::tls::TrustStore::PermissionStatus status;
3597 : {
3598 1788 : std::lock_guard lk(pimpl_->conversationsMtx_);
3599 1788 : status = pimpl_->accountManager_->getCertificateStatus(deviceId);
3600 1788 : }
3601 1788 : if (auto conv = pimpl_->getConversation(convId)) {
3602 1788 : std::lock_guard lk(conv->mtx);
3603 1788 : if (!conv->conversation)
3604 0 : return true;
3605 1788 : if (conv->conversation->mode() != ConversationMode::ONE_TO_ONE)
3606 1214 : return conv->conversation->isDeviceBanned(deviceId);
3607 574 : return status == dhtnet::tls::TrustStore::PermissionStatus::BANNED;
3608 3576 : }
3609 0 : return true;
3610 : }
3611 :
3612 : bool
3613 6335 : ConversationModule::isPeerAuthorized(const std::string& convId,
3614 : const std::string& uri,
3615 : const std::string& deviceId,
3616 : bool includeInvited) const
3617 : {
3618 : dhtnet::tls::TrustStore::PermissionStatus memberStatus;
3619 : dhtnet::tls::TrustStore::PermissionStatus deviceStatus;
3620 : {
3621 6335 : std::lock_guard lk(pimpl_->conversationsMtx_);
3622 6335 : memberStatus = pimpl_->accountManager_->getCertificateStatus(uri);
3623 6335 : deviceStatus = pimpl_->accountManager_->getCertificateStatus(deviceId);
3624 6335 : }
3625 6335 : if (memberStatus == dhtnet::tls::TrustStore::PermissionStatus::BANNED
3626 6335 : || deviceStatus == dhtnet::tls::TrustStore::PermissionStatus::BANNED) {
3627 0 : return false;
3628 : }
3629 6335 : if (auto conv = pimpl_->getConversation(convId)) {
3630 6314 : std::lock_guard lk(conv->mtx);
3631 6314 : if (conv->conversation)
3632 5688 : return conv->conversation->isPeerAuthorized(uri, deviceId, includeInvited);
3633 626 : if (!includeInvited)
3634 0 : return false;
3635 626 : return !conv->info.isRemoved() && conv->info.members.find(uri) != conv->info.members.end();
3636 12649 : }
3637 :
3638 21 : return false;
3639 : }
3640 :
3641 : bool
3642 455 : ConversationModule::isPeerAuthorizedByConversation(const dht::crypto::Certificate& certificate) const
3643 : {
3644 455 : if (!certificate.issuer)
3645 0 : return false;
3646 :
3647 549 : for (const auto& conversation : pimpl_->getConversations()) {
3648 513 : if (!conversation->isRemoving() && conversation->isPeerCertificateAuthorized(certificate))
3649 419 : return true;
3650 455 : }
3651 36 : return false;
3652 : }
3653 :
3654 : bool
3655 35 : ConversationModule::mayServeDocument(const std::string& documentId,
3656 : const std::string& uri,
3657 : const std::string& deviceId) const
3658 : {
3659 35 : auto conv = pimpl_->getConversation(documentId);
3660 35 : if (!conv)
3661 5 : return false; // We don't hold this document — nothing to vouch with
3662 30 : std::shared_ptr<Conversation> conversation;
3663 : {
3664 30 : std::lock_guard lk(conv->mtx);
3665 30 : conversation = conv->conversation;
3666 30 : }
3667 30 : if (!conversation || conversation->mode() != ConversationMode::DOCUMENT)
3668 4 : return false;
3669 26 : if (conversation->isMemberBanned(uri))
3670 2 : return false; // Banned from the document: only an admin's re-add can undo that
3671 24 : return isPeerAuthorized(conversation->parentConversationId(), uri, deviceId, false);
3672 35 : }
3673 :
3674 : void
3675 22 : ConversationModule::authorizeDocumentPeer(const std::string& documentId,
3676 : const std::string& uri,
3677 : const std::string& deviceId,
3678 : std::function<void(bool)>&& cb)
3679 : {
3680 22 : auto conv = pimpl_->getConversation(documentId);
3681 22 : if (!conv)
3682 0 : return cb(false); // We don't hold this document — nothing to vouch with
3683 22 : std::shared_ptr<Conversation> conversation;
3684 : {
3685 22 : std::lock_guard lk(conv->mtx);
3686 22 : conversation = conv->conversation;
3687 22 : }
3688 22 : if (!conversation || conversation->mode() != ConversationMode::DOCUMENT)
3689 0 : return cb(false);
3690 22 : if (conversation->isMemberBanned(uri))
3691 0 : return cb(false); // Banned from the document: only an admin's re-add can undo that
3692 22 : if (!isPeerAuthorized(conversation->parentConversationId(), uri, deviceId, false))
3693 0 : return cb(false);
3694 : // The add commit is what makes the clone we are about to serve
3695 : // self-justifying: the peer's membership is part of what it receives.
3696 44 : conversation->addMember(uri, [cb = std::move(cb)](bool ok, const std::string&) { cb(ok); });
3697 22 : }
3698 :
3699 : void
3700 22 : ConversationModule::removeContact(const std::string& uri, bool banned)
3701 : {
3702 : // Remove linked conversation's requests
3703 : {
3704 22 : std::lock_guard lk(pimpl_->conversationsRequestsMtx_);
3705 22 : auto update = false;
3706 26 : for (auto it = pimpl_->conversationsRequests_.begin(); it != pimpl_->conversationsRequests_.end(); ++it) {
3707 4 : if (it->second.from == uri && it->second.declined == TimePoint {}) {
3708 4 : JAMI_DEBUG("Declining conversation request {:s} from {:s}", it->first, uri);
3709 4 : pimpl_->syncingMetadatas_.erase(it->first);
3710 4 : pimpl_->saveMetadata();
3711 4 : emitSignal<libjami::ConversationSignal::ConversationRequestDeclined>(pimpl_->accountId_, it->first);
3712 4 : update = true;
3713 4 : it->second.declined = nowMs();
3714 : }
3715 : }
3716 22 : if (update) {
3717 4 : pimpl_->saveConvRequests();
3718 4 : pimpl_->needsSyncingCb_({});
3719 : }
3720 22 : }
3721 22 : if (banned) {
3722 7 : auto conversationId = getOneToOneConversation(uri);
3723 7 : pimpl_->withConversation(conversationId, [&](auto& conv) { conv.shutdownConnections(); });
3724 7 : return; // Keep the conversation in banned model but stop connections
3725 7 : }
3726 :
3727 : // Removed contacts should not be linked to any conversation
3728 30 : pimpl_->accountManager_->updateContactConversation(uri, "");
3729 :
3730 : // Remove all one-to-one conversations with the removed contact
3731 15 : auto isSelf = uri == pimpl_->username_;
3732 15 : std::vector<std::string> toRm;
3733 18 : auto removeConvInfo = [&](const auto& conv, const auto& members) {
3734 1 : if ((isSelf && members.size() == 1)
3735 19 : || (!isSelf && std::find(members.begin(), members.end(), uri) != members.end())) {
3736 : // Mark the conversation as removed if it wasn't already
3737 17 : if (!conv->info.isRemoved()) {
3738 15 : conv->info.removed = nowMs();
3739 15 : emitSignal<libjami::ConversationSignal::ConversationRemoved>(pimpl_->accountId_, conv->info.id);
3740 15 : pimpl_->addConvInfo(conv->info);
3741 15 : return true;
3742 : }
3743 : }
3744 3 : return false;
3745 15 : };
3746 : {
3747 15 : std::lock_guard lk(pimpl_->conversationsMtx_);
3748 33 : for (auto& [convId, conv] : pimpl_->conversations_) {
3749 18 : std::lock_guard lk(conv->mtx);
3750 18 : if (conv->conversation) {
3751 : try {
3752 : // Note it's important to check getUsername(), else
3753 : // removing self can remove all conversations
3754 17 : if (conv->conversation->mode() == ConversationMode::ONE_TO_ONE) {
3755 17 : auto initMembers = conv->conversation->getInitialMembers();
3756 17 : if (removeConvInfo(conv, initMembers))
3757 14 : toRm.emplace_back(convId);
3758 17 : }
3759 0 : } catch (const std::exception& e) {
3760 0 : JAMI_WARNING("{}", e.what());
3761 0 : }
3762 : } else {
3763 1 : removeConvInfo(conv, conv->info.members);
3764 : }
3765 18 : }
3766 15 : }
3767 29 : for (const auto& id : toRm)
3768 14 : pimpl_->removeRepository(id, true, true);
3769 15 : for (const auto& documentId : pimpl_->childDocuments(toRm))
3770 15 : pimpl_->removeDocumentReplica(documentId);
3771 15 : }
3772 :
3773 : bool
3774 15 : ConversationModule::removeConversation(const std::string& conversationId)
3775 : {
3776 15 : auto conversation = pimpl_->getConversation(conversationId);
3777 15 : std::string existingConvId;
3778 :
3779 15 : if (!conversation) {
3780 1 : JAMI_LOG("Conversation {} not found", conversationId);
3781 1 : return false;
3782 : }
3783 :
3784 14 : std::shared_ptr<Conversation> conv;
3785 : {
3786 14 : std::lock_guard lk(conversation->mtx);
3787 14 : conv = conversation->conversation;
3788 14 : }
3789 :
3790 3 : auto sendNotification = [&](const std::string& convId) {
3791 3 : if (auto convObj = pimpl_->getConversation(convId)) {
3792 3 : std::lock_guard lk(convObj->mtx);
3793 3 : if (convObj->conversation) {
3794 3 : auto commitId = convObj->conversation->lastCommitId();
3795 3 : if (!commitId.empty()) {
3796 9 : pimpl_->sendMessageNotification(*convObj->conversation, true, commitId);
3797 : }
3798 3 : }
3799 6 : }
3800 3 : };
3801 :
3802 3 : auto handleNewConversation = [&](const std::string& uri) {
3803 3 : std::string newConvId = startConversation(ConversationMode::ONE_TO_ONE, dht::InfoHash(uri));
3804 3 : pimpl_->accountManager_->updateContactConversation(uri, newConvId);
3805 3 : sendNotification(newConvId);
3806 3 : return newConvId;
3807 0 : };
3808 :
3809 14 : if (!conv) {
3810 1 : auto contacts = pimpl_->accountManager_->getContacts(false);
3811 2 : for (const auto& contact : contacts) {
3812 1 : if (contact.second.conversationId == conversationId) {
3813 1 : const std::string& uri = contact.first.toString();
3814 1 : handleNewConversation(uri);
3815 1 : }
3816 : }
3817 :
3818 1 : return pimpl_->removeConversation(conversationId);
3819 1 : }
3820 :
3821 13 : if (conv->mode() == ConversationMode::ONE_TO_ONE) {
3822 2 : auto members = conv->getMembers(true, false, false);
3823 2 : if (members.empty()) {
3824 0 : return false;
3825 : }
3826 :
3827 6 : for (const auto& m : members) {
3828 4 : const auto& uri = m.at("uri");
3829 :
3830 4 : if (members.size() == 1 && uri == pimpl_->username_) {
3831 0 : if (conv->getInitialMembers().size() == 1 && conv->getInitialMembers()[0] == pimpl_->username_) {
3832 : // Self conversation, create new conversation and remove the old one
3833 0 : handleNewConversation(uri);
3834 : } else {
3835 0 : existingConvId = findMatchingOneToOneConversation(conversationId, conv->memberUris("", {}));
3836 0 : if (existingConvId.empty()) {
3837 : // If left with only ended conversation of peer
3838 0 : for (const auto& otherMember : conv->getInitialMembers()) {
3839 0 : if (otherMember != pimpl_->username_) {
3840 0 : handleNewConversation(otherMember);
3841 : }
3842 0 : }
3843 : }
3844 : }
3845 0 : break;
3846 : }
3847 :
3848 4 : if (uri == pimpl_->username_)
3849 2 : continue;
3850 :
3851 2 : existingConvId = findMatchingOneToOneConversation(conversationId, conv->memberUris("", {}));
3852 2 : if (!existingConvId.empty()) {
3853 : // Found an existing conversation, just update the contact
3854 0 : pimpl_->accountManager_->updateContactConversation(uri, existingConvId);
3855 0 : sendNotification(existingConvId);
3856 : } else {
3857 : // No existing conversation found, create a new one
3858 2 : handleNewConversation(uri);
3859 : }
3860 : }
3861 2 : }
3862 :
3863 13 : return pimpl_->removeConversation(conversationId);
3864 15 : }
3865 :
3866 : std::string
3867 5 : ConversationModule::findMatchingOneToOneConversation(const std::string& excludedConversationId,
3868 : const std::set<std::string>& targetUris) const
3869 : {
3870 5 : std::lock_guard lk(pimpl_->conversationsMtx_);
3871 10 : for (const auto& [otherConvId, otherConvPtr] : pimpl_->conversations_) {
3872 6 : if (otherConvId == excludedConversationId)
3873 5 : continue;
3874 :
3875 2 : std::lock_guard lk(otherConvPtr->mtx);
3876 2 : if (!otherConvPtr->conversation || otherConvPtr->conversation->mode() != ConversationMode::ONE_TO_ONE)
3877 1 : continue;
3878 :
3879 1 : const auto& info = otherConvPtr->info;
3880 1 : if (info.removed != TimePoint {} && info.isRemoved())
3881 0 : continue;
3882 :
3883 2 : auto otherUris = otherConvPtr->conversation->memberUris({}, {MemberRole::LEFT, MemberRole::BANNED});
3884 :
3885 1 : if (otherUris == targetUris)
3886 1 : return otherConvId;
3887 3 : }
3888 :
3889 4 : return {};
3890 5 : }
3891 :
3892 : bool
3893 69 : ConversationModule::isHosting(const std::string& conversationId, const std::string& confId) const
3894 : {
3895 69 : if (conversationId.empty()) {
3896 59 : std::lock_guard lk(pimpl_->conversationsMtx_);
3897 59 : return std::find_if(pimpl_->conversations_.cbegin(),
3898 59 : pimpl_->conversations_.cend(),
3899 18 : [&](const auto& conv) {
3900 18 : return conv.second->conversation && conv.second->conversation->isHosting(confId);
3901 : })
3902 118 : != pimpl_->conversations_.cend();
3903 69 : } else if (auto conv = pimpl_->getConversation(conversationId)) {
3904 10 : if (conv->conversation) {
3905 10 : return conv->conversation->isHosting(confId);
3906 : }
3907 10 : }
3908 0 : return false;
3909 : }
3910 :
3911 : std::vector<std::map<std::string, std::string>>
3912 21 : ConversationModule::getActiveCalls(const std::string& conversationId) const
3913 : {
3914 21 : return pimpl_->withConversation(conversationId,
3915 63 : [](const auto& conversation) { return conversation.currentCalls(); });
3916 : }
3917 :
3918 : std::shared_ptr<SIPCall>
3919 22 : ConversationModule::call(const std::string& url,
3920 : const std::vector<libjami::MediaMap>& mediaList,
3921 : std::function<void(const std::string&, const DeviceId&, const std::shared_ptr<SIPCall>&)>&& cb)
3922 : {
3923 154 : std::string conversationId = "", confId = "", uri = "", deviceId = "";
3924 22 : if (url.find('/') == std::string::npos) {
3925 13 : conversationId = url;
3926 : } else {
3927 9 : auto parameters = jami::split_string(url, '/');
3928 9 : if (parameters.size() != 4) {
3929 0 : JAMI_ERROR("Incorrect url {:s}", url);
3930 0 : return {};
3931 : }
3932 9 : conversationId = parameters[0];
3933 9 : uri = parameters[1];
3934 9 : deviceId = parameters[2];
3935 9 : confId = parameters[3];
3936 9 : }
3937 :
3938 22 : auto conv = pimpl_->getConversation(conversationId);
3939 22 : if (!conv)
3940 0 : return {};
3941 22 : std::unique_lock lk(conv->mtx);
3942 22 : if (!conv->conversation) {
3943 0 : JAMI_ERROR("Conversation {:s} not found", conversationId);
3944 0 : return {};
3945 : }
3946 :
3947 : // Check if we want to join a specific conference
3948 : // So, if confId is specified or if there is some activeCalls
3949 : // or if we are the default host.
3950 22 : auto activeCalls = conv->conversation->currentCalls();
3951 22 : auto infos = conv->conversation->infos();
3952 44 : auto itRdvAccount = infos.find("rdvAccount");
3953 22 : auto itRdvDevice = infos.find("rdvDevice");
3954 22 : auto sendCallRequest = false;
3955 22 : if (!confId.empty()) {
3956 9 : sendCallRequest = true;
3957 9 : JAMI_DEBUG("Calling self, join conference");
3958 13 : } else if (!activeCalls.empty()) {
3959 : // Else, we try to join active calls
3960 0 : sendCallRequest = true;
3961 0 : auto& ac = *activeCalls.rbegin();
3962 0 : confId = ac.at("id");
3963 0 : uri = ac.at("uri");
3964 0 : deviceId = ac.at("device");
3965 13 : } else if (itRdvAccount != infos.end() && itRdvDevice != infos.end() && !itRdvAccount->second.empty()) {
3966 : // Else, creates "to" (accountId/deviceId/conversationId/confId) and ask remote host
3967 3 : sendCallRequest = true;
3968 3 : uri = itRdvAccount->second;
3969 3 : deviceId = itRdvDevice->second;
3970 3 : confId = "0";
3971 3 : JAMI_DEBUG("Remote host detected. Calling {:s} on device {:s}", uri, deviceId);
3972 : }
3973 22 : lk.unlock();
3974 :
3975 22 : auto account = pimpl_->account_.lock();
3976 22 : std::vector<libjami::MediaMap> mediaMap = mediaList.empty() ? MediaAttribute::mediaAttributesToMediaMaps(
3977 60 : pimpl_->account_.lock()->createDefaultMediaList(
3978 41 : pimpl_->account_.lock()->isVideoEnabled()))
3979 41 : : mediaList;
3980 :
3981 22 : if (!sendCallRequest || (uri == pimpl_->username_ && deviceId == pimpl_->deviceId_)) {
3982 11 : confId = confId == "0" ? Manager::instance().callFactory.getNewCallID() : confId;
3983 : // TODO attach host with media list
3984 11 : hostConference(conversationId, confId, "", mediaMap);
3985 11 : return {};
3986 : }
3987 :
3988 : // Else we need to create a call
3989 11 : auto& manager = Manager::instance();
3990 11 : std::shared_ptr<SIPCall> call = manager.callFactory.newSipCall(account, Call::CallType::OUTGOING, mediaMap);
3991 :
3992 11 : if (not call)
3993 0 : return {};
3994 :
3995 11 : auto callUri = fmt::format("{}/{}/{}/{}", conversationId, uri, deviceId, confId);
3996 44 : account->getIceOptions([call,
3997 11 : accountId = account->getAccountID(),
3998 : callUri,
3999 11 : uri = std::move(uri),
4000 : conversationId,
4001 : deviceId,
4002 11 : cb = std::move(cb)](auto&& opts) {
4003 11 : if (call->isIceEnabled()) {
4004 11 : if (not call->createIceMediaTransport(false)
4005 22 : or not call->initIceMediaTransport(true, std::forward<dhtnet::IceTransportOptions>(opts))) {
4006 0 : return;
4007 : }
4008 : }
4009 11 : JAMI_DEBUG("New outgoing call with {}", uri);
4010 11 : call->setPeerNumber(uri);
4011 11 : call->setPeerUri("swarm:" + uri);
4012 :
4013 11 : JAMI_DEBUG("Calling: {:s}", callUri);
4014 11 : call->setState(Call::ConnectionState::TRYING);
4015 11 : call->setPeerNumber(callUri);
4016 11 : call->setPeerUri("rdv:" + callUri);
4017 22 : call->addStateListener(
4018 22 : [accountId, conversationId](Call::CallState call_state, Call::ConnectionState cnx_state, int) {
4019 62 : if (cnx_state == Call::ConnectionState::DISCONNECTED && call_state == Call::CallState::MERROR) {
4020 1 : emitSignal<libjami::ConfigurationSignal::NeedsHost>(accountId, conversationId);
4021 1 : return true;
4022 : }
4023 61 : return true;
4024 : });
4025 11 : cb(callUri, DeviceId(deviceId), call);
4026 : });
4027 :
4028 11 : return call;
4029 22 : }
4030 :
4031 : void
4032 14 : ConversationModule::hostConference(const std::string& conversationId,
4033 : const std::string& confId,
4034 : const std::string& callId,
4035 : const std::vector<libjami::MediaMap>& mediaList)
4036 : {
4037 14 : auto acc = pimpl_->account_.lock();
4038 14 : if (!acc)
4039 0 : return;
4040 14 : auto conf = acc->getConference(confId);
4041 14 : auto createConf = !conf;
4042 14 : std::shared_ptr<SIPCall> call;
4043 14 : if (!callId.empty()) {
4044 3 : call = std::dynamic_pointer_cast<SIPCall>(acc->getCall(callId));
4045 3 : if (!call) {
4046 0 : JAMI_WARNING("No call with id {} found", callId);
4047 0 : return;
4048 : }
4049 : }
4050 14 : if (createConf) {
4051 14 : conf = std::make_shared<Conference>(acc, confId);
4052 14 : acc->attach(conf);
4053 : }
4054 :
4055 14 : if (!callId.empty())
4056 3 : conf->addSubCall(callId);
4057 :
4058 14 : if (callId.empty())
4059 11 : conf->attachHost(mediaList);
4060 :
4061 14 : if (createConf) {
4062 14 : emitSignal<libjami::CallSignal::ConferenceCreated>(acc->getAccountID(), conversationId, conf->getConfId());
4063 : } else {
4064 0 : conf->reportMediaNegotiationStatus();
4065 0 : emitSignal<libjami::CallSignal::ConferenceChanged>(acc->getAccountID(), conf->getConfId(), conf->getStateStr());
4066 0 : return;
4067 : }
4068 :
4069 14 : auto conv = pimpl_->getConversation(conversationId);
4070 14 : if (!conv)
4071 0 : return;
4072 14 : std::unique_lock lk(conv->mtx);
4073 14 : if (!conv->conversation) {
4074 0 : JAMI_ERROR("Conversation {} not found", conversationId);
4075 0 : return;
4076 : }
4077 : // Add commit to conversation
4078 14 : auto message = CommitMessage::conferenceHostingStart(conf->getConfId(), pimpl_->deviceId_, pimpl_->username_);
4079 28 : conv->conversation->hostConference(std::move(message),
4080 28 : [w = pimpl_->weak(), conversationId](bool ok, const std::string& commitId) {
4081 14 : if (ok) {
4082 14 : if (auto shared = w.lock())
4083 42 : shared->sendMessageNotification(conversationId, true, commitId);
4084 : } else {
4085 0 : JAMI_ERROR("Failed to send message to conversation {}", conversationId);
4086 : }
4087 14 : });
4088 :
4089 : // When conf finished = remove host & commit
4090 : // Master call, so when it's stopped, the conference will be stopped (as we use the hold
4091 : // state for detaching the call)
4092 28 : conf->onShutdown([w = pimpl_->weak(),
4093 14 : accountUri = pimpl_->username_,
4094 14 : confId = conf->getConfId(),
4095 : conversationId,
4096 : conv](int duration) {
4097 14 : auto shared = w.lock();
4098 14 : if (shared) {
4099 10 : auto message = CommitMessage::conferenceHostingEnd(confId, shared->deviceId_, accountUri, duration);
4100 :
4101 10 : std::lock_guard lk(conv->mtx);
4102 10 : if (!conv->conversation) {
4103 0 : JAMI_ERROR("Conversation {} not found", conversationId);
4104 0 : return;
4105 : }
4106 10 : conv->conversation
4107 10 : ->removeActiveConference(std::move(message), [w, conversationId](bool ok, const std::string& commitId) {
4108 10 : if (ok) {
4109 10 : if (auto shared = w.lock()) {
4110 30 : shared->sendMessageNotification(conversationId, true, commitId);
4111 10 : }
4112 : } else {
4113 0 : JAMI_ERROR("Failed to send message to conversation {}", conversationId);
4114 : }
4115 10 : });
4116 10 : }
4117 14 : });
4118 14 : }
4119 :
4120 : std::map<std::string, ConvInfo>
4121 1111 : ConversationModule::convInfos(const std::string& accountId)
4122 : {
4123 1111 : return convInfosFromPath(fileutils::get_data_dir() / accountId);
4124 : }
4125 :
4126 : std::map<std::string, ConvInfo>
4127 1157 : ConversationModule::convInfosFromPath(const std::filesystem::path& path)
4128 : {
4129 1157 : std::map<std::string, ConvInfo> convInfos;
4130 : try {
4131 : // read file
4132 1157 : std::lock_guard lock(dhtnet::fileutils::getFileLock(path / "convInfo"));
4133 1157 : auto file = fileutils::loadFile("convInfo", path);
4134 : // load values
4135 1157 : msgpack::unpacked result;
4136 1157 : msgpack::unpack(result, (const char*) file.data(), file.size());
4137 1155 : result.get().convert(convInfos);
4138 1460 : std::erase_if(convInfos, [](const auto& item) { return !isConversationId(item.first); });
4139 1163 : } catch (const std::exception& e) {
4140 2 : JAMI_WARNING("[convInfo] error loading convInfo: {}", e.what());
4141 2 : }
4142 : // Ids are used as directory names: drop anything persisted before ids were validated
4143 1462 : for (auto it = convInfos.begin(); it != convInfos.end();) {
4144 305 : if (!ConversationRepository::isValidConversationId(it->first)
4145 305 : || !ConversationRepository::isValidConversationId(it->second.id)) {
4146 0 : JAMI_WARNING("[convInfo] dropping conversation with invalid id '{}'", it->first);
4147 0 : it = convInfos.erase(it);
4148 : } else {
4149 305 : ++it;
4150 : }
4151 : }
4152 1157 : return convInfos;
4153 0 : }
4154 :
4155 : std::map<std::string, ConversationRequest>
4156 1076 : ConversationModule::convRequests(const std::string& accountId)
4157 : {
4158 1076 : return convRequestsFromPath(fileutils::get_data_dir() / accountId);
4159 : }
4160 :
4161 : std::map<std::string, ConversationRequest>
4162 1122 : ConversationModule::convRequestsFromPath(const std::filesystem::path& path)
4163 : {
4164 1122 : std::map<std::string, ConversationRequest> convRequests;
4165 : try {
4166 : // read file
4167 1122 : std::lock_guard lock(dhtnet::fileutils::getFileLock(path / "convRequests"));
4168 1122 : auto file = fileutils::loadFile("convRequests", path);
4169 : // load values
4170 1122 : msgpack::unpacked result;
4171 1122 : msgpack::unpack(result, (const char*) file.data(), file.size(), 0);
4172 1122 : result.get().convert(convRequests);
4173 1122 : } catch (const std::exception& e) {
4174 0 : JAMI_WARNING("[convInfo] error loading convInfo: {}", e.what());
4175 0 : }
4176 1134 : for (auto it = convRequests.begin(); it != convRequests.end();) {
4177 12 : if (!ConversationRepository::isValidConversationId(it->first)
4178 12 : || !ConversationRepository::isValidConversationId(it->second.conversationId)) {
4179 0 : JAMI_WARNING("[convRequests] dropping request with invalid conversation id '{}'", it->first);
4180 0 : it = convRequests.erase(it);
4181 : } else {
4182 12 : ++it;
4183 : }
4184 : }
4185 1122 : return convRequests;
4186 0 : }
4187 :
4188 : void
4189 430 : ConversationModule::addConvInfo(const ConvInfo& info)
4190 : {
4191 430 : pimpl_->addConvInfo(info);
4192 430 : }
4193 :
4194 : void
4195 3236 : ConversationModule::Impl::setConversationMembers(const std::string& convId, const std::set<std::string>& members)
4196 : {
4197 3236 : if (auto conv = getConversation(convId)) {
4198 3236 : std::lock_guard lk(conv->mtx);
4199 3236 : conv->info.members = members;
4200 3236 : addConvInfo(conv->info);
4201 6472 : }
4202 3236 : }
4203 :
4204 : std::shared_ptr<Conversation>
4205 2746 : ConversationModule::getConversation(const std::string& convId)
4206 : {
4207 2746 : if (auto conv = pimpl_->getConversation(convId)) {
4208 2689 : std::lock_guard lk(conv->mtx);
4209 2689 : return conv->conversation;
4210 5435 : }
4211 57 : return nullptr;
4212 : }
4213 :
4214 : std::shared_ptr<dhtnet::ChannelSocket>
4215 6357 : ConversationModule::gitSocket(std::string_view deviceId, std::string_view convId) const
4216 : {
4217 6357 : if (auto conv = pimpl_->getConversation(convId)) {
4218 6357 : std::lock_guard lk(conv->mtx);
4219 6357 : if (conv->conversation)
4220 5230 : return conv->conversation->gitSocket(DeviceId(deviceId));
4221 1127 : else if (conv->pending)
4222 1123 : return conv->pending->socket;
4223 12714 : }
4224 4 : return nullptr;
4225 : }
4226 :
4227 : void
4228 1509 : ConversationModule::removeGitSocket(std::string_view deviceId,
4229 : std::string_view convId,
4230 : const std::shared_ptr<dhtnet::ChannelSocket>& expected)
4231 : {
4232 2954 : pimpl_->withConversation(convId, [&](auto& conv) { conv.removeGitSocket(DeviceId(deviceId), expected); });
4233 1509 : }
4234 :
4235 : void
4236 917 : ConversationModule::shutdownConnections()
4237 : {
4238 1914 : for (const auto& c : pimpl_->getSyncedConversations()) {
4239 997 : std::lock_guard lkc(c->mtx);
4240 997 : if (c->conversation)
4241 905 : c->conversation->shutdownConnections();
4242 997 : if (c->pending)
4243 67 : c->pending->socket = {};
4244 1914 : }
4245 917 : }
4246 : void
4247 1650 : ConversationModule::addSwarmChannel(const std::string& conversationId, std::shared_ptr<dhtnet::ChannelSocket> channel)
4248 : {
4249 3264 : pimpl_->withConversation(conversationId, [&](auto& conv) { conv.addSwarmChannel(std::move(channel)); });
4250 1650 : }
4251 :
4252 : void
4253 1366 : ConversationModule::addKnownDevice(const std::string& peerUri, const DeviceId& deviceId)
4254 : {
4255 3326 : for (const auto& conv : pimpl_->getConversations())
4256 1960 : if (conv->isMember(peerUri))
4257 2054 : conv->connectNode(deviceId);
4258 1366 : }
4259 :
4260 : void
4261 0 : ConversationModule::connectivityChanged()
4262 : {
4263 0 : for (const auto& conv : pimpl_->getConversations())
4264 0 : conv->connectivityChanged();
4265 0 : }
4266 :
4267 : std::shared_ptr<Typers>
4268 9 : ConversationModule::getTypers(const std::string& convId)
4269 : {
4270 9 : if (auto c = pimpl_->getConversation(convId)) {
4271 9 : std::lock_guard lk(c->mtx);
4272 9 : if (c->conversation)
4273 9 : return c->conversation->typers();
4274 18 : }
4275 0 : return nullptr;
4276 : }
4277 :
4278 : void
4279 18 : ConversationModule::initPresence()
4280 : {
4281 18 : pimpl_->initPresence();
4282 18 : }
4283 :
4284 : } // namespace jami
|
