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