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