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.h"
19 :
20 : #include "account_const.h"
21 : #include "jamiaccount.h"
22 : #include "client/jami_signal.h"
23 : #include "swarm/swarm_manager.h"
24 : #include "conversationrepository.h"
25 : #ifdef ENABLE_PLUGIN
26 : #include "manager.h"
27 : #include "plugin/jamipluginmanager.h"
28 : #include "plugin/streamdata.h"
29 : #endif
30 : #include "jami/conversation_interface.h"
31 : #include "jami/configurationmanager_interface.h"
32 :
33 : #include "fileutils.h"
34 : #include "json_utils.h"
35 : #include "logger.h"
36 : #include "presence_manager.h"
37 :
38 : #include <opendht/thread_pool.h>
39 : #include <opendht/infohash.h>
40 : #include <fmt/compile.h>
41 : #include <asio/error_code.hpp>
42 :
43 : #include <algorithm>
44 : #include <memory>
45 : #include <charconv>
46 : #include <set>
47 : #include <string_view>
48 : #include <tuple>
49 : #include <optional>
50 : #include <utility>
51 : #include <deque>
52 : #include <chrono>
53 : #include <ctime>
54 : #include <functional>
55 : #include <atomic>
56 : #include <ranges>
57 : #include <regex>
58 :
59 : namespace jami {
60 :
61 : static const char* const LAST_MODIFIED = "lastModified";
62 :
63 : namespace {
64 :
65 : // Read a timestamp from JSON, preferring the millisecond key and falling
66 : // back to the legacy seconds key (written by older devices).
67 : TimePoint
68 571 : timePointFromJson(const Json::Value& json, const char* msKey, const char* secondsKey)
69 : {
70 571 : if (json.isMember(msKey))
71 13 : return timePointFromMilliseconds(json[msKey].asLargestInt());
72 558 : return timePointFromSeconds(json[secondsKey].asLargestInt());
73 : }
74 :
75 : // Resolve a timestamp from msgpack values, preferring milliseconds.
76 : TimePoint
77 813 : resolveTimePoint(const std::optional<int64_t>& ms, int64_t seconds)
78 : {
79 813 : return ms ? timePointFromMilliseconds(*ms) : timePointFromSeconds(seconds);
80 : }
81 :
82 : } // namespace
83 :
84 13 : ConvInfo::ConvInfo(const Json::Value& json)
85 : {
86 13 : id = json[ConversationMapKeys::ID].asString();
87 13 : created = timePointFromJson(json, ConversationMapKeys::CREATED_MS, ConversationMapKeys::CREATED);
88 13 : removed = timePointFromJson(json, ConversationMapKeys::REMOVED_MS, ConversationMapKeys::REMOVED);
89 13 : erased = timePointFromJson(json, ConversationMapKeys::ERASED_MS, ConversationMapKeys::ERASED);
90 33 : for (const auto& v : json[ConversationMapKeys::MEMBERS]) {
91 20 : members.emplace(v["uri"].asString());
92 : }
93 13 : lastDisplayed = json[ConversationMapKeys::LAST_DISPLAYED].asString();
94 13 : if (json.isMember(ConversationMapKeys::MODE))
95 12 : mode = static_cast<ConversationMode>(json[ConversationMapKeys::MODE].asInt());
96 13 : }
97 :
98 : Json::Value
99 26 : ConvInfo::toJson() const
100 : {
101 26 : Json::Value json;
102 26 : json[ConversationMapKeys::ID] = id;
103 26 : json[ConversationMapKeys::CREATED] = Json::Int64(toSecondsSinceEpoch(created));
104 26 : json[ConversationMapKeys::CREATED_MS] = Json::Int64(toMillisecondsSinceEpoch(created));
105 26 : if (removed != TimePoint {}) {
106 1 : json[ConversationMapKeys::REMOVED] = Json::Int64(toSecondsSinceEpoch(removed));
107 1 : json[ConversationMapKeys::REMOVED_MS] = Json::Int64(toMillisecondsSinceEpoch(removed));
108 : }
109 26 : if (erased != TimePoint {}) {
110 1 : json[ConversationMapKeys::ERASED] = Json::Int64(toSecondsSinceEpoch(erased));
111 1 : json[ConversationMapKeys::ERASED_MS] = Json::Int64(toMillisecondsSinceEpoch(erased));
112 : }
113 67 : for (const auto& m : members) {
114 41 : Json::Value member;
115 41 : member["uri"] = m;
116 41 : json[ConversationMapKeys::MEMBERS].append(member);
117 41 : }
118 26 : json[ConversationMapKeys::LAST_DISPLAYED] = lastDisplayed;
119 26 : json[ConversationMapKeys::MODE] = static_cast<int>(mode);
120 26 : return json;
121 0 : }
122 :
123 : void
124 251 : ConvInfo::msgpack_unpack(const msgpack::object& o)
125 : {
126 251 : if (o.type != msgpack::type::MAP)
127 0 : throw msgpack::type_error();
128 251 : int64_t createdSec = 0, removedSec = 0, erasedSec = 0;
129 251 : std::optional<int64_t> createdMs, removedMs, erasedMs;
130 2760 : for (uint32_t i = 0; i < o.via.map.size; ++i) {
131 2509 : const auto& kv = o.via.map.ptr[i];
132 2509 : if (kv.key.type != msgpack::type::STR)
133 0 : continue;
134 2509 : std::string_view key(kv.key.via.str.ptr, kv.key.via.str.size);
135 2509 : if (key == ConversationMapKeys::ID)
136 251 : kv.val.convert(id);
137 2258 : else if (key == ConversationMapKeys::CREATED)
138 251 : kv.val.convert(createdSec);
139 2007 : else if (key == ConversationMapKeys::REMOVED)
140 251 : kv.val.convert(removedSec);
141 1756 : else if (key == ConversationMapKeys::ERASED)
142 251 : kv.val.convert(erasedSec);
143 1505 : else if (key == ConversationMapKeys::CREATED_MS)
144 251 : createdMs = kv.val.as<int64_t>();
145 1255 : else if (key == ConversationMapKeys::REMOVED_MS)
146 251 : removedMs = kv.val.as<int64_t>();
147 1004 : else if (key == ConversationMapKeys::ERASED_MS)
148 251 : erasedMs = kv.val.as<int64_t>();
149 753 : else if (key == ConversationMapKeys::MEMBERS)
150 251 : kv.val.convert(members);
151 502 : else if (key == ConversationMapKeys::LAST_DISPLAYED)
152 251 : kv.val.convert(lastDisplayed);
153 251 : else if (key == ConversationMapKeys::MODE)
154 251 : kv.val.convert(mode);
155 : }
156 251 : created = resolveTimePoint(createdMs, createdSec);
157 251 : removed = resolveTimePoint(removedMs, removedSec);
158 251 : erased = resolveTimePoint(erasedMs, erasedSec);
159 251 : }
160 :
161 : void
162 0 : ConvInfo::msgpack_object(msgpack::object* o, msgpack::zone& z) const
163 : {
164 0 : int64_t createdSec = toSecondsSinceEpoch(created);
165 0 : int64_t removedSec = toSecondsSinceEpoch(removed);
166 0 : int64_t erasedSec = toSecondsSinceEpoch(erased);
167 0 : int64_t createdMs = toMillisecondsSinceEpoch(created);
168 0 : int64_t removedMs = toMillisecondsSinceEpoch(removed);
169 0 : int64_t erasedMs = toMillisecondsSinceEpoch(erased);
170 0 : msgpack::type::make_define_map(ConversationMapKeys::ID,
171 0 : id,
172 : ConversationMapKeys::CREATED,
173 : createdSec,
174 : ConversationMapKeys::REMOVED,
175 : removedSec,
176 : ConversationMapKeys::ERASED,
177 : erasedSec,
178 : ConversationMapKeys::MEMBERS,
179 0 : members,
180 : ConversationMapKeys::LAST_DISPLAYED,
181 0 : lastDisplayed,
182 : ConversationMapKeys::MODE,
183 0 : mode,
184 : ConversationMapKeys::CREATED_MS,
185 : createdMs,
186 : ConversationMapKeys::REMOVED_MS,
187 : removedMs,
188 : ConversationMapKeys::ERASED_MS,
189 : erasedMs)
190 0 : .msgpack_object(o, z);
191 0 : }
192 :
193 : // ConversationRequest
194 266 : ConversationRequest::ConversationRequest(const Json::Value& json)
195 : {
196 266 : received = timePointFromJson(json, ConversationMapKeys::RECEIVED_MS, ConversationMapKeys::RECEIVED);
197 266 : declined = timePointFromJson(json, ConversationMapKeys::DECLINED_MS, ConversationMapKeys::DECLINED);
198 266 : from = json[ConversationMapKeys::FROM].asString();
199 266 : conversationId = json[ConversationMapKeys::CONVERSATIONID].asString();
200 266 : auto& md = json[ConversationMapKeys::METADATAS];
201 535 : for (const auto& member : md.getMemberNames()) {
202 269 : metadatas.emplace(member, md[member].asString());
203 266 : }
204 266 : }
205 :
206 : Json::Value
207 2 : ConversationRequest::toJson() const
208 : {
209 2 : Json::Value json;
210 2 : json[ConversationMapKeys::CONVERSATIONID] = conversationId;
211 2 : json[ConversationMapKeys::FROM] = from;
212 2 : json[ConversationMapKeys::RECEIVED] = Json::Int64(toSecondsSinceEpoch(received));
213 2 : json[ConversationMapKeys::RECEIVED_MS] = Json::Int64(toMillisecondsSinceEpoch(received));
214 2 : if (declined != TimePoint {}) {
215 0 : json[ConversationMapKeys::DECLINED] = Json::Int64(toSecondsSinceEpoch(declined));
216 0 : json[ConversationMapKeys::DECLINED_MS] = Json::Int64(toMillisecondsSinceEpoch(declined));
217 : }
218 4 : for (const auto& [key, value] : metadatas) {
219 2 : json[ConversationMapKeys::METADATAS][key] = value;
220 : }
221 2 : return json;
222 0 : }
223 :
224 : std::map<std::string, std::string>
225 285 : ConversationRequest::toMap() const
226 : {
227 285 : auto result = metadatas;
228 570 : result[ConversationMapKeys::ID] = conversationId;
229 570 : result[ConversationMapKeys::FROM] = from;
230 285 : if (declined != TimePoint {})
231 0 : result[ConversationMapKeys::DECLINED] = std::to_string(toSecondsSinceEpoch(declined));
232 853 : result[ConversationMapKeys::RECEIVED] = std::to_string(toSecondsSinceEpoch(received));
233 285 : return result;
234 0 : }
235 :
236 : void
237 30 : ConversationRequest::msgpack_unpack(const msgpack::object& o)
238 : {
239 30 : if (o.type != msgpack::type::MAP)
240 0 : throw msgpack::type_error();
241 30 : int64_t receivedSec = 0, declinedSec = 0;
242 30 : std::optional<int64_t> receivedMs, declinedMs;
243 240 : for (uint32_t i = 0; i < o.via.map.size; ++i) {
244 210 : const auto& kv = o.via.map.ptr[i];
245 210 : if (kv.key.type != msgpack::type::STR)
246 0 : continue;
247 210 : std::string_view key(kv.key.via.str.ptr, kv.key.via.str.size);
248 210 : if (key == ConversationMapKeys::FROM)
249 30 : kv.val.convert(from);
250 180 : else if (key == ConversationMapKeys::CONVERSATIONID)
251 30 : kv.val.convert(conversationId);
252 150 : else if (key == ConversationMapKeys::METADATAS)
253 30 : kv.val.convert(metadatas);
254 120 : else if (key == ConversationMapKeys::RECEIVED)
255 30 : kv.val.convert(receivedSec);
256 90 : else if (key == ConversationMapKeys::DECLINED)
257 30 : kv.val.convert(declinedSec);
258 60 : else if (key == ConversationMapKeys::RECEIVED_MS)
259 30 : receivedMs = kv.val.as<int64_t>();
260 30 : else if (key == ConversationMapKeys::DECLINED_MS)
261 30 : declinedMs = kv.val.as<int64_t>();
262 : }
263 30 : received = resolveTimePoint(receivedMs, receivedSec);
264 30 : declined = resolveTimePoint(declinedMs, declinedSec);
265 30 : }
266 :
267 : void
268 0 : ConversationRequest::msgpack_object(msgpack::object* o, msgpack::zone& z) const
269 : {
270 0 : int64_t receivedSec = toSecondsSinceEpoch(received);
271 0 : int64_t declinedSec = toSecondsSinceEpoch(declined);
272 0 : int64_t receivedMs = toMillisecondsSinceEpoch(received);
273 0 : int64_t declinedMs = toMillisecondsSinceEpoch(declined);
274 0 : msgpack::type::make_define_map(ConversationMapKeys::FROM,
275 0 : from,
276 : ConversationMapKeys::CONVERSATIONID,
277 0 : conversationId,
278 : ConversationMapKeys::METADATAS,
279 0 : metadatas,
280 : ConversationMapKeys::RECEIVED,
281 : receivedSec,
282 : ConversationMapKeys::DECLINED,
283 : declinedSec,
284 : ConversationMapKeys::RECEIVED_MS,
285 : receivedMs,
286 : ConversationMapKeys::DECLINED_MS,
287 : declinedMs)
288 0 : .msgpack_object(o, z);
289 0 : }
290 :
291 : using MessageList = std::list<std::shared_ptr<libjami::SwarmMessage>>;
292 :
293 : struct History
294 : {
295 : // While loading the history, we need to avoid:
296 : // - reloading history (can just be ignored)
297 : // - adding new commits (should wait for history to be loaded)
298 : std::mutex mutex {};
299 : std::condition_variable cv {};
300 : bool loading {false};
301 : MessageList messageList {};
302 : std::map<std::string, std::shared_ptr<libjami::SwarmMessage>> quickAccess {};
303 : std::map<std::string, std::list<std::shared_ptr<libjami::SwarmMessage>>> pendingEditions {};
304 : std::map<std::string, std::list<std::map<std::string, std::string>>> pendingReactions {};
305 : };
306 :
307 : class Conversation::Impl
308 : {
309 : private:
310 401 : Impl(std::unique_ptr<ConversationRepository>&& repository,
311 : const std::shared_ptr<JamiAccount>& account,
312 : std::vector<ConversationCommit>&& commits = {})
313 401 : : repository_(repository ? std::move(repository) : throw std::logic_error("Invalid repository"))
314 391 : , account_(account)
315 391 : , accountId_(account->getAccountID())
316 391 : , userId_(account->getUsername())
317 782 : , deviceId_(account->currentDeviceId())
318 391 : , swarmManager_(std::make_shared<SwarmManager>(NodeId(deviceId_),
319 391 : account->isMobile(),
320 391 : Manager::instance().getSeededRandomEngine(),
321 782 : [account = account_](const DeviceId& deviceId) {
322 287 : if (auto acc = account.lock()) {
323 287 : return acc->isConnectedWith(deviceId);
324 287 : }
325 0 : return false;
326 : }))
327 391 : , transferManager_(std::make_shared<TransferManager>(accountId_,
328 : "",
329 391 : repository_->id(),
330 391 : Manager::instance().getSeededRandomEngine()))
331 391 : , repoPath_(fileutils::get_data_dir() / accountId_ / "conversations" / repository_->id())
332 391 : , conversationDataPath_(fileutils::get_data_dir() / accountId_ / "conversation_data" / repository_->id())
333 391 : , fetchedPath_(conversationDataPath_ / ConversationDirectories::FETCHED)
334 391 : , sendingPath_(conversationDataPath_ / ConversationDirectories::SENDING)
335 391 : , preferencesPath_(conversationDataPath_ / ConversationDirectories::PREFERENCES)
336 391 : , statusPath_(conversationDataPath_ / ConversationDirectories::STATUS)
337 391 : , hostedCallsPath_(conversationDataPath_ / ConversationDirectories::HOSTED_CALLS)
338 391 : , activeCallsPath_(conversationDataPath_ / ConversationDirectories::ACTIVE_CALLS)
339 391 : , ioContext_(Manager::instance().ioContext())
340 2356 : , typers_(std::make_shared<Typers>(account, repository_->id()))
341 : {
342 391 : if (!commits.empty())
343 186 : initActiveCalls(repository_->convCommitsToMap(commits));
344 391 : loadStatus();
345 391 : setupMemberCallback();
346 421 : }
347 :
348 196 : Impl(std::pair<std::unique_ptr<ConversationRepository>, std::vector<ConversationCommit>>&& repoAndCommits,
349 : const std::shared_ptr<JamiAccount>& account)
350 196 : : Impl(std::move(repoAndCommits.first), account, std::move(repoAndCommits.second))
351 186 : {}
352 :
353 : public:
354 187 : Impl(const std::shared_ptr<JamiAccount>& account, ConversationMode mode, const std::string& otherMember = "")
355 187 : : Impl(ConversationRepository::createConversation(account, mode, otherMember), account)
356 187 : {}
357 :
358 18 : Impl(const std::shared_ptr<JamiAccount>& account, const std::string& conversationId)
359 18 : : Impl(std::make_unique<ConversationRepository>(account, conversationId), account)
360 18 : {}
361 :
362 200 : Impl(const std::shared_ptr<JamiAccount>& account, const std::string& remoteDevice, const std::string& conversationId)
363 200 : : Impl(ConversationRepository::cloneConversation(account, remoteDevice, conversationId), account)
364 186 : {}
365 :
366 2658 : std::string toString() const
367 : {
368 10631 : return fmt::format(FMT_COMPILE("[Account {}] [Conversation {}]"), accountId_, repository_->id());
369 : }
370 :
371 0 : std::vector<std::map<std::string, std::string>> getConnectivity() const
372 : {
373 0 : return swarmManager_->getRoutingTableInfo();
374 : }
375 :
376 : mutable std::string fmtStr_;
377 :
378 391 : ~Impl() { stopTracking(); }
379 :
380 : /**
381 : * If, for whatever reason, the daemon is stopped while hosting a conference,
382 : * we need to announce the end of this call when restarting.
383 : * To avoid to keep active calls forever.
384 : */
385 : std::vector<std::string> commitsEndedCalls();
386 : bool isAdmin() const;
387 :
388 286 : void announce(const std::string& commitId, bool commitFromSelf = false)
389 : {
390 286 : std::vector<std::string> vec;
391 286 : if (!commitId.empty())
392 284 : vec.emplace_back(commitId);
393 286 : announce(vec, commitFromSelf);
394 286 : }
395 :
396 298 : void announce(const std::vector<std::string>& commits, bool commitFromSelf = false)
397 : {
398 298 : std::vector<ConversationCommit> convcommits;
399 298 : convcommits.reserve(commits.size());
400 606 : for (const auto& cid : commits) {
401 308 : if (auto commit = repository_->getCommit(cid)) {
402 308 : convcommits.emplace_back(*commit);
403 308 : }
404 : }
405 299 : announce(repository_->convCommitsToMap(convcommits), commitFromSelf);
406 298 : }
407 :
408 : /**
409 : * Initialize activeCalls_ from the list of commits in the repository
410 : * @param commits Commits in reverse chronological order (i.e. from newest to oldest)
411 : */
412 186 : void initActiveCalls(const std::vector<std::map<std::string, std::string>>& commits) const
413 : {
414 186 : std::unordered_set<std::string> invalidHostUris;
415 186 : std::unordered_set<std::string> invalidCallIds;
416 :
417 186 : std::lock_guard lk(activeCallsMtx_);
418 1166 : for (const auto& commit : commits) {
419 1960 : if (commit.at(CommitKey::TYPE) == CommitType::MEMBER) {
420 : // Each commit of type MEMBER has an "action" field whose value can be one
421 : // of the following: "add", "join", "remove", "ban", "unban"
422 : // In the case of "remove" and "ban", we need to add the member's URI to
423 : // invalidHostUris to ensure that any call they may have started in the past
424 : // is no longer considered active.
425 : // For the other actions, there's no harm in adding the member's URI anyway,
426 : // since it's not possible to start hosting a call before joining the swarm (or
427 : // before getting unbanned in the case of previously banned members).
428 1546 : invalidHostUris.emplace(commit.at(CommitKey::URI));
429 624 : } else if (commit.find(CommitKey::CONF_ID) != commit.end() && commit.find(CommitKey::URI) != commit.end()
430 417 : && commit.find(CommitKey::DEVICE) != commit.end()) {
431 : // The commit indicates either the end or the beginning of a call
432 : // (depending on whether there is a "duration" field or not).
433 1 : auto convId = repository_->id();
434 2 : auto confId = commit.at(CommitKey::CONF_ID);
435 2 : auto uri = commit.at(CommitKey::URI);
436 1 : auto device = commit.at(CommitKey::DEVICE);
437 :
438 3 : if (commit.find(CommitKey::DURATION) == commit.end()
439 1 : && invalidCallIds.find(confId) == invalidCallIds.end()
440 3 : && invalidHostUris.find(uri) == invalidHostUris.end()) {
441 1 : std::map<std::string, std::string> activeCall;
442 2 : activeCall["id"] = confId;
443 2 : activeCall["uri"] = uri;
444 1 : activeCall["device"] = device;
445 1 : activeCalls_.emplace_back(activeCall);
446 1 : fmt::print("swarm:{} new active call detected: {} (on device {}, account {})\n",
447 : convId,
448 : confId,
449 : device,
450 : uri);
451 1 : }
452 : // Even if the call was active, we still add its ID to invalidCallIds to make sure it
453 : // doesn't get added a second time. (This shouldn't happen normally, but in practice
454 : // there are sometimes multiple commits indicating the beginning of the same call.)
455 1 : invalidCallIds.emplace(confId);
456 1 : }
457 : }
458 186 : saveActiveCalls();
459 186 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_, repository_->id(), activeCalls_);
460 186 : }
461 :
462 : /**
463 : * Update activeCalls_ via announced commits (in load or via new commits)
464 : * @param commit Commit to check
465 : * @param eraseOnly If we want to ignore added commits
466 : * @param emitSig If we want to emit to client
467 : * @note eraseOnly is used by loadMessages. This is a fail-safe, this SHOULD NOT happen
468 : */
469 74 : void updateActiveCalls(const std::map<std::string, std::string>& commit,
470 : bool eraseOnly = false,
471 : bool emitSig = true) const
472 : {
473 74 : if (!repository_)
474 0 : return;
475 148 : if (commit.at(CommitKey::TYPE) == CommitType::MEMBER) {
476 : // In this case, we need to check if we are not removing a hosting member or device
477 17 : std::lock_guard lk(activeCallsMtx_);
478 17 : auto it = activeCalls_.begin();
479 17 : auto updateActives = false;
480 17 : while (it != activeCalls_.end()) {
481 0 : if (it->at("uri") == commit.at(CommitKey::URI) || it->at("device") == commit.at(CommitKey::URI)) {
482 0 : JAMI_DEBUG("Removing {:s} from the active calls, because {:s} left",
483 : it->at("id"),
484 : commit.at(CommitKey::URI));
485 0 : it = activeCalls_.erase(it);
486 0 : updateActives = true;
487 : } else {
488 0 : ++it;
489 : }
490 : }
491 17 : if (updateActives) {
492 0 : saveActiveCalls();
493 0 : if (emitSig)
494 0 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_,
495 0 : repository_->id(),
496 0 : activeCalls_);
497 : }
498 17 : return;
499 17 : }
500 : // Else, it's a call information
501 330 : if (commit.find(CommitKey::CONF_ID) != commit.end() && commit.find(CommitKey::URI) != commit.end()
502 275 : && commit.find(CommitKey::DEVICE) != commit.end()) {
503 53 : auto convId = repository_->id();
504 107 : auto confId = commit.at(CommitKey::CONF_ID);
505 108 : auto uri = commit.at(CommitKey::URI);
506 54 : auto device = commit.at(CommitKey::DEVICE);
507 54 : std::lock_guard lk(activeCallsMtx_);
508 54 : auto itActive = std::find_if(activeCalls_.begin(), activeCalls_.end(), [&](const auto& value) {
509 161 : return value.at("id") == confId && value.at("uri") == uri && value.at("device") == device;
510 : });
511 161 : if (commit.find(CommitKey::DURATION) == commit.end()) {
512 30 : if (itActive == activeCalls_.end() && !eraseOnly) {
513 31 : JAMI_DEBUG("swarm:{:s} new current call detected: {:s} on device {:s}, account {:s}",
514 : convId,
515 : confId,
516 : device,
517 : uri);
518 155 : activeCalls_.emplace_back(std::map<std::string, std::string> {
519 : {"id", confId},
520 : {"uri", uri },
521 : {"device", device},
522 124 : });
523 31 : saveActiveCalls();
524 31 : if (emitSig)
525 30 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_,
526 31 : repository_->id(),
527 31 : activeCalls_);
528 : }
529 : } else {
530 23 : if (itActive != activeCalls_.end()) {
531 23 : itActive = activeCalls_.erase(itActive);
532 : // Unlikely, but we must ensure that no duplicate exists
533 46 : while (itActive != activeCalls_.end()) {
534 0 : itActive = std::find_if(itActive, activeCalls_.end(), [&](const auto& value) {
535 0 : return value.at("id") == confId && value.at("uri") == uri && value.at("device") == device;
536 : });
537 0 : if (itActive != activeCalls_.end()) {
538 0 : JAMI_ERROR("Duplicate call found. (This is a bug)");
539 0 : itActive = activeCalls_.erase(itActive);
540 : }
541 : }
542 :
543 23 : if (eraseOnly) {
544 0 : JAMI_WARNING("previous swarm:{:s} call finished detected: {:s} on device "
545 : "{:s}, account {:s}",
546 : convId,
547 : confId,
548 : device,
549 : uri);
550 : } else {
551 23 : JAMI_DEBUG("swarm:{:s} call finished: {:s} on device {:s}, account {:s}",
552 : convId,
553 : confId,
554 : device,
555 : uri);
556 : }
557 : }
558 23 : saveActiveCalls();
559 23 : if (emitSig)
560 23 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_,
561 23 : repository_->id(),
562 23 : activeCalls_);
563 : }
564 53 : }
565 31 : }
566 :
567 1202 : void announce(const std::vector<std::map<std::string, std::string>>& commits, bool commitFromSelf = false)
568 : {
569 1202 : if (!repository_)
570 0 : return;
571 1203 : auto convId = repository_->id();
572 1203 : auto ok = !commits.empty();
573 2408 : auto lastId = ok ? commits.rbegin()->at(ConversationMapKeys::ID) : "";
574 1203 : addToHistory(loadedHistory_, commits, true, commitFromSelf);
575 1203 : if (ok) {
576 1201 : bool announceMember = false;
577 2435 : for (const auto& c : commits) {
578 : // Announce member events
579 2471 : if (c.at(CommitKey::TYPE) == CommitType::MEMBER) {
580 4603 : if (c.find(CommitKey::URI) != c.end() && c.find(CommitKey::ACTION) != c.end()) {
581 1842 : const auto& uri = c.at(CommitKey::URI);
582 921 : const auto& actionStr = c.at(CommitKey::ACTION);
583 921 : auto action = -1;
584 921 : if (actionStr == CommitAction::ADD)
585 443 : action = 0;
586 478 : else if (actionStr == CommitAction::JOIN)
587 460 : action = 1;
588 18 : else if (actionStr == CommitAction::REMOVE)
589 1 : action = 2;
590 17 : else if (actionStr == CommitAction::BAN)
591 16 : action = 3;
592 1 : else if (actionStr == CommitAction::UNBAN)
593 1 : action = 4;
594 921 : if (actionStr == CommitAction::BAN || actionStr == CommitAction::REMOVE) {
595 : // In this case, a potential host was removed during a call.
596 17 : updateActiveCalls(c);
597 17 : typers_->removeTyper(uri);
598 : }
599 921 : if (action != -1) {
600 921 : announceMember = true;
601 921 : emitSignal<libjami::ConversationSignal::ConversationMemberEvent>(accountId_,
602 : convId,
603 : uri,
604 : action);
605 : }
606 : }
607 630 : } else if (c.at(CommitKey::TYPE) == CommitType::CALL_HISTORY) {
608 57 : updateActiveCalls(c);
609 : }
610 : #ifdef ENABLE_PLUGIN
611 1236 : if (auto& pluginChatManager = Manager::instance().getJamiPluginManager().getChatServicesManager();
612 1236 : pluginChatManager.hasHandlers()) {
613 6 : auto cm = std::make_shared<JamiMessage>(accountId_, convId, c.at("author") != userId_, c, false);
614 3 : cm->isSwarm = true;
615 3 : pluginChatManager.publishMessage(cm);
616 :
617 3 : const auto editIt = c.find(CommitKey::EDIT);
618 3 : const bool isEdit = editIt != c.end() && !editIt->second.empty();
619 6 : const bool isText = c.at(CommitKey::TYPE) == CommitType::TEXT;
620 :
621 6 : if (const auto idIt = c.find("id"); idIt != c.end() && (isEdit || isText)) {
622 1 : const std::string msgId = idIt->second;
623 :
624 : // Snapshot under lock.
625 1 : libjami::SwarmMessage snap;
626 : {
627 1 : std::lock_guard const lk(loadedHistory_.mutex);
628 1 : if (auto it = loadedHistory_.quickAccess.find(msgId);
629 1 : it != loadedHistory_.quickAccess.end()) {
630 1 : snap = *it->second;
631 : }
632 1 : }
633 :
634 1 : if (!snap.id.empty()) {
635 : // Transform outside lock.
636 3 : std::vector<libjami::SwarmMessage> single {snap};
637 1 : pluginChatManager.transformSwarmMessages(single, accountId_, convId);
638 0 : auto& transformed = single[0];
639 0 : const auto boIt = transformed.pluginData.find("bodyOverwrite");
640 0 : const bool hasOverwrite = boIt != transformed.pluginData.end() && !boIt->second.empty();
641 :
642 : // Write back + build signal under one lock, emit outside.
643 0 : libjami::SwarmMessage signalMsg;
644 : {
645 0 : std::lock_guard const lk(loadedHistory_.mutex);
646 0 : if (isEdit || hasOverwrite) {
647 0 : signalMsg = injectEditionOverwrites(msgId, hasOverwrite ? boIt->second : "");
648 : } else {
649 0 : if (auto it = loadedHistory_.quickAccess.find(msgId);
650 0 : it != loadedHistory_.quickAccess.end()) {
651 0 : it->second->pluginData["bodyOverwrite"] = "";
652 : }
653 : }
654 0 : }
655 0 : if (!signalMsg.id.empty()) {
656 0 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_,
657 : convId,
658 : signalMsg);
659 : }
660 0 : }
661 0 : }
662 3 : } else
663 : #endif
664 : {
665 : // No active handlers: still notify clients when an edit arrives.
666 2465 : if (const auto editIt = c.find(CommitKey::EDIT); editIt != c.end() && !editIt->second.empty()) {
667 7 : libjami::SwarmMessage signalMsg;
668 : {
669 7 : std::lock_guard const lk(loadedHistory_.mutex);
670 7 : if (auto originalIt = loadedHistory_.quickAccess.find(editIt->second);
671 7 : originalIt != loadedHistory_.quickAccess.end()) {
672 7 : signalMsg = *originalIt->second;
673 : }
674 7 : }
675 7 : if (!signalMsg.id.empty()) {
676 7 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, convId, signalMsg);
677 : }
678 7 : }
679 : }
680 : }
681 :
682 1200 : if (announceMember && onMembersChanged_) {
683 921 : onMembersChanged_(repository_->memberUris("", {}));
684 : }
685 : }
686 1205 : }
687 :
688 391 : void loadStatus()
689 : {
690 : try {
691 : // read file
692 780 : auto file = fileutils::loadFile(statusPath_);
693 : // load values
694 2 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
695 2 : std::lock_guard lk {messageStatusMtx_};
696 2 : oh.get().convert(messagesStatus_);
697 391 : } catch (const std::exception& e) {
698 389 : }
699 391 : }
700 1112 : void saveStatus()
701 : {
702 1112 : std::ofstream file(statusPath_, std::ios::trunc | std::ios::binary);
703 1112 : msgpack::pack(file, messagesStatus_);
704 1112 : }
705 :
706 18 : void loadActiveCalls() const
707 : {
708 : try {
709 : // read file
710 29 : auto file = fileutils::loadFile(activeCallsPath_);
711 : // load values
712 7 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
713 7 : std::lock_guard lk {activeCallsMtx_};
714 7 : oh.get().convert(activeCalls_);
715 18 : } catch (const std::exception& e) {
716 11 : return;
717 11 : }
718 : }
719 :
720 258 : void saveActiveCalls() const
721 : {
722 258 : std::ofstream file(activeCallsPath_, std::ios::trunc | std::ios::binary);
723 258 : msgpack::pack(file, activeCalls_);
724 258 : }
725 :
726 18 : void loadHostedCalls() const
727 : {
728 : try {
729 : // read file
730 30 : auto file = fileutils::loadFile(hostedCallsPath_);
731 : // load values
732 6 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
733 6 : std::lock_guard lk {activeCallsMtx_};
734 6 : oh.get().convert(hostedCalls_);
735 18 : } catch (const std::exception& e) {
736 12 : return;
737 12 : }
738 : }
739 :
740 42 : void saveHostedCalls() const
741 : {
742 42 : std::ofstream file(hostedCallsPath_, std::ios::trunc | std::ios::binary);
743 42 : msgpack::pack(file, hostedCalls_);
744 42 : }
745 :
746 : void voteUnban(const std::string& contactUri, const std::string_view type, const OnDoneCb& cb);
747 :
748 : std::vector<std::map<std::string, std::string>> getMembers(bool includeInvited,
749 : bool includeLeft,
750 : bool includeBanned) const;
751 :
752 : std::vector<std::map<std::string, std::string>> getTrackedMembers() const;
753 :
754 5957 : std::string_view memberBanType(const std::string& uri) const
755 : {
756 5957 : auto crt = fmt::format("{}.crt", uri);
757 5958 : auto bannedMember = repoPath_ / MemberPath::BANNED / MemberPath::MEMBERS / crt;
758 5956 : if (std::filesystem::is_regular_file(bannedMember))
759 17 : return "members"sv;
760 5940 : auto bannedAdmin = repoPath_ / MemberPath::BANNED / MemberPath::ADMINS / crt;
761 5940 : if (std::filesystem::is_regular_file(bannedAdmin))
762 0 : return "admins"sv;
763 5941 : auto bannedInvited = repoPath_ / MemberPath::BANNED / MemberPath::INVITED / uri;
764 5940 : if (std::filesystem::is_regular_file(bannedInvited))
765 0 : return "invited"sv;
766 5939 : return {};
767 5956 : }
768 :
769 6671 : bool isDeviceBanned(const std::string& deviceId) const
770 : {
771 6671 : auto crt = fmt::format("{}.crt", deviceId);
772 6673 : auto bannedDevice = repoPath_ / MemberPath::BANNED / MemberPath::DEVICES / crt;
773 13339 : return std::filesystem::is_regular_file(bannedDevice);
774 6673 : }
775 :
776 4248 : std::shared_ptr<dhtnet::ChannelSocket> gitSocket(const DeviceId& deviceId) const
777 : {
778 4248 : std::lock_guard lk(gitSocketMtx_);
779 4248 : auto deviceSockets = gitSocketList_.find(deviceId);
780 8495 : return (deviceSockets != gitSocketList_.end()) ? deviceSockets->second : nullptr;
781 4248 : }
782 :
783 1821 : void addGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& socket)
784 : {
785 1821 : std::lock_guard lk(gitSocketMtx_);
786 1821 : gitSocketList_[deviceId] = socket;
787 1821 : }
788 935 : void removeGitSocket(const DeviceId& deviceId)
789 : {
790 935 : std::lock_guard lk(gitSocketMtx_);
791 934 : auto deviceSockets = gitSocketList_.find(deviceId);
792 935 : if (deviceSockets != gitSocketList_.end())
793 492 : gitSocketList_.erase(deviceSockets);
794 935 : }
795 :
796 : void disconnectFromDevice(const DeviceId& deviceId);
797 :
798 : /**
799 : * Remove all git sockets and all DRT nodes associated with the given peer.
800 : * This is used when a swarm member is banned to ensure that we stop syncing
801 : * with them or sending them message notifications.
802 : */
803 : void disconnectFromPeer(const std::string& peerUri);
804 :
805 : std::vector<std::map<std::string, std::string>> getMembers(bool includeInvited, bool includeLeft) const;
806 :
807 : std::function<void()> bootstrapCb_;
808 : std::mutex bootstrapMtx_ {};
809 : #ifdef LIBJAMI_TEST
810 : std::function<void(std::string, BootstrapStatus)> bootstrapCbTest_;
811 : #endif
812 :
813 : std::mutex writeMtx_ {};
814 : const std::unique_ptr<ConversationRepository> repository_;
815 : const std::weak_ptr<JamiAccount> account_;
816 : const std::string accountId_;
817 : const std::string userId_;
818 : const std::string deviceId_;
819 : const std::shared_ptr<SwarmManager> swarmManager_;
820 : std::atomic_bool isRemoving_ {false};
821 : std::vector<libjami::SwarmMessage> loadMessages(const LogOptions& options, History* optHistory = nullptr);
822 : void loadMissingBodyOverwrites();
823 : void reloadBodyOverwriteMessages();
824 :
825 : // Out-of-order calls are safe: only the latest edition's overwrite propagates to the original.
826 : void updateMessageBodyOverwrite(const std::string& messageId, std::string_view bodyOverwrite);
827 : // Must be called with loadedHistory_.mutex held.
828 : // Stores bodyOverwrite on messageId, propagates to original if edition, returns signal message.
829 : // Only propagates when messageId == originalMsg->latestEditionId, so out-of-order calls are safe.
830 : libjami::SwarmMessage injectEditionOverwrites(const std::string& messageId, std::string_view bodyOverwrite);
831 :
832 : void clearBodyOverwrites();
833 : void pull(const std::string& deviceId);
834 :
835 : // Avoid multiple fetch/merges at the same time.
836 : std::mutex pullcbsMtx_ {};
837 : // store current remote in fetch
838 : std::map<std::string, std::deque<std::pair<std::string, OnPullCb>>> fetchingRemotes_ {};
839 : const std::shared_ptr<TransferManager> transferManager_ {};
840 : const std::filesystem::path repoPath_ {};
841 : const std::filesystem::path conversationDataPath_ {};
842 : const std::filesystem::path fetchedPath_ {};
843 :
844 : // Manage last message displayed and status
845 : const std::filesystem::path sendingPath_ {};
846 : const std::filesystem::path preferencesPath_ {};
847 : const std::filesystem::path statusPath_ {};
848 :
849 : OnMembersChanged onMembersChanged_ {};
850 : struct TrackedMember
851 : {
852 : std::set<DeviceId> devices;
853 : std::set<DeviceId> failedDevices;
854 : };
855 : std::map<std::string, TrackedMember> trackedMembers_;
856 : mutable std::mutex trackedMembersMtx_;
857 : bool isTracking_ {false};
858 : void setupMemberCallback();
859 : void startTracking(std::weak_ptr<Conversation> w);
860 : void stopTracking();
861 : void rotateTrackedMembers(const std::string& memberUri = "", const DeviceId& deviceId = {});
862 : void monitorConnection(std::weak_ptr<Conversation> w);
863 : void onConnectionFailed(const DeviceId& deviceId, const std::string& memberUri = "");
864 :
865 : uint64_t presenceDeviceListenerToken_ {0};
866 :
867 : // Manage hosted calls on this device
868 : std::filesystem::path hostedCallsPath_ {};
869 : mutable std::map<std::string, uint64_t /* start time */> hostedCalls_ {};
870 : // Manage active calls for this conversation (can be hosted by other devices)
871 : std::filesystem::path activeCallsPath_ {};
872 : mutable std::mutex activeCallsMtx_ {};
873 : mutable std::vector<std::map<std::string, std::string>> activeCalls_ {};
874 :
875 : mutable std::mutex gitSocketMtx_ {};
876 : GitSocketList gitSocketList_ {};
877 :
878 : // Bootstrap
879 : const std::shared_ptr<asio::io_context> ioContext_;
880 :
881 : /**
882 : * Loaded history represents the linearized history to show for clients
883 : */
884 : History loadedHistory_ {};
885 : std::vector<std::shared_ptr<libjami::SwarmMessage>> addToHistory(
886 : History& history,
887 : const std::vector<std::map<std::string, std::string>>& commits,
888 : bool messageReceived = false,
889 : bool commitFromSelf = false);
890 :
891 : void handleReaction(History& history, const std::shared_ptr<libjami::SwarmMessage>& sharedCommit) const;
892 : void handleEdition(History& history,
893 : const std::shared_ptr<libjami::SwarmMessage>& sharedCommit,
894 : bool messageReceived) const;
895 : bool handleMessage(History& history,
896 : const std::shared_ptr<libjami::SwarmMessage>& sharedCommit,
897 : bool messageReceived) const;
898 : void rectifyStatus(const std::shared_ptr<libjami::SwarmMessage>& message, History& history) const;
899 : /**
900 : * {uri, {
901 : * {"fetch", "commitId"},
902 : * {"fetched_ts", "timestamp"},
903 : * {"read", "commitId"},
904 : * {"read_ts", "timestamp"}
905 : * }
906 : * }
907 : */
908 : mutable std::mutex messageStatusMtx_;
909 : std::function<void(const std::map<std::string, std::map<std::string, std::string>>&)> messageStatusCb_ {};
910 : std::map<std::string, std::map<std::string, std::string>> messagesStatus_ {};
911 : /**
912 : * Status: 0 = commited, 1 = fetched, 2 = read
913 : * This cache the curent status to add in the messages
914 : */
915 : // Note: only store int32_t cause it's easy to pass to dbus this way
916 : // memberToStatus serves as a cache for loading messages
917 : std::map<std::string, int32_t> memberToStatus;
918 :
919 : // futureStatus is used to store the status for receiving messages
920 : // (because we're not sure to fetch the commit before receiving a status change for this)
921 : std::map<std::string, std::map<std::string, int32_t>> futureStatus;
922 : // Update internal structures regarding status
923 : void updateStatus(const std::string& uri,
924 : libjami::Account::MessageStates status,
925 : const std::string& commitId,
926 : const std::string& ts,
927 : bool emit = false);
928 :
929 : std::shared_ptr<Typers> typers_;
930 : };
931 :
932 : /**
933 : * Stores bodyOverwrite on messageId, propagates to original if messageId is an edition commit,
934 : * and returns the message to signal (original if edition, messageId's message otherwise),
935 : * with each edition entry's "bodyOverwrite" injected from the corresponding commit's pluginData.
936 : * Must be called with loadedHistory_.mutex held.
937 : * @param messageId ID of the edition commit or original message to update.
938 : * @param bodyOverwrite New overwrite value; empty string marks the message as processed with no overwrite.
939 : */
940 : libjami::SwarmMessage
941 0 : Conversation::Impl::injectEditionOverwrites(const std::string& messageId, const std::string_view bodyOverwrite)
942 : {
943 0 : const auto it = loadedHistory_.quickAccess.find(messageId);
944 0 : if (it == loadedHistory_.quickAccess.end()) {
945 0 : return {};
946 : }
947 :
948 0 : libjami::SwarmMessage* signalTarget = it->second.get();
949 :
950 0 : if (const auto editIt = it->second->body.find(CommitKey::EDIT); editIt != it->second->body.end()) {
951 0 : it->second->pluginData["bodyOverwrite"] = bodyOverwrite;
952 0 : if (editIt->second.empty()) {
953 0 : return {};
954 : }
955 0 : const auto originalIt = loadedHistory_.quickAccess.find(editIt->second);
956 0 : if (originalIt == loadedHistory_.quickAccess.end()) {
957 0 : return {};
958 : }
959 0 : signalTarget = originalIt->second.get();
960 0 : } else if (it->second->editions.empty()) {
961 0 : it->second->pluginData["bodyOverwrite"] = bodyOverwrite;
962 : }
963 :
964 0 : if (const auto msgEditionIt = std::ranges::find_if(signalTarget->editions,
965 0 : [&](const auto& e) {
966 0 : const auto idIt = e.find("id");
967 0 : return idIt != e.end() && idIt->second == messageId;
968 : });
969 0 : msgEditionIt != signalTarget->editions.end()) {
970 0 : (*msgEditionIt)["bodyOverwrite"] = bodyOverwrite;
971 : }
972 :
973 0 : if (const auto latestEdIt = loadedHistory_.quickAccess.find(signalTarget->latestEditionId);
974 0 : latestEdIt != loadedHistory_.quickAccess.end()) {
975 0 : if (const auto latestEdBoIt = latestEdIt->second->pluginData.find("bodyOverwrite");
976 0 : latestEdBoIt != latestEdIt->second->pluginData.end() && !latestEdBoIt->second.empty()) {
977 0 : signalTarget->pluginData["bodyOverwrite"] = latestEdBoIt->second;
978 : }
979 : }
980 0 : return *signalTarget;
981 0 : }
982 :
983 : void
984 391 : Conversation::Impl::setupMemberCallback()
985 : {
986 391 : repository_->onMembersChanged([this](const std::set<std::string>& memberUris) {
987 : {
988 1083 : std::lock_guard lk(trackedMembersMtx_);
989 1083 : if (isTracking_) {
990 230 : if (auto acc = account_.lock()) {
991 502 : for (auto it = trackedMembers_.begin(); it != trackedMembers_.end();) {
992 272 : if (memberUris.find(it->first) == memberUris.end()) {
993 7 : acc->presenceManager()->untrackBuddy(it->first);
994 7 : it = trackedMembers_.erase(it);
995 : } else {
996 265 : ++it;
997 : }
998 : }
999 230 : }
1000 : }
1001 1083 : }
1002 :
1003 1083 : if (onMembersChanged_)
1004 1083 : onMembersChanged_(memberUris);
1005 1083 : });
1006 391 : }
1007 :
1008 : void
1009 551 : Conversation::Impl::startTracking(std::weak_ptr<Conversation> w)
1010 : {
1011 551 : auto acc = account_.lock();
1012 551 : if (!acc)
1013 0 : return;
1014 :
1015 : {
1016 551 : std::lock_guard lk(trackedMembersMtx_);
1017 551 : if (isTracking_)
1018 150 : return;
1019 401 : isTracking_ = true;
1020 802 : presenceDeviceListenerToken_ = acc->presenceManager()->addDeviceListener(
1021 802 : [w](const std::string& uri, const DeviceId& deviceId, bool online) {
1022 688 : if (auto sthis = w.lock()) {
1023 688 : if (online && sthis->isMember(uri)) {
1024 1650 : sthis->addKnownDevices({deviceId}, uri);
1025 : }
1026 688 : }
1027 688 : });
1028 551 : }
1029 :
1030 802 : rotateTrackedMembers();
1031 551 : }
1032 :
1033 : void
1034 771 : Conversation::Impl::stopTracking()
1035 : {
1036 : // Collect data under trackedMembersMtx_, then release it before calling
1037 : // into PresenceManager to avoid lock-ordering inversion with the
1038 : // PresenceManager mutex (which is held when notifyListeners calls
1039 : // addKnownDevices, which in turn acquires trackedMembersMtx_).
1040 771 : uint64_t token = 0;
1041 771 : std::vector<std::string> urisToUntrack;
1042 : {
1043 771 : std::lock_guard lk(trackedMembersMtx_);
1044 772 : if (!isTracking_)
1045 371 : return;
1046 401 : isTracking_ = false;
1047 401 : token = presenceDeviceListenerToken_;
1048 401 : presenceDeviceListenerToken_ = 0;
1049 1157 : for (const auto& [uri, _] : trackedMembers_) {
1050 755 : urisToUntrack.push_back(uri);
1051 : }
1052 401 : trackedMembers_.clear();
1053 772 : }
1054 :
1055 401 : auto acc = account_.lock();
1056 401 : if (!acc)
1057 182 : return;
1058 :
1059 219 : if (token) {
1060 219 : acc->presenceManager()->removeDeviceListener(token);
1061 : }
1062 :
1063 701 : for (const auto& uri : urisToUntrack) {
1064 482 : acc->presenceManager()->untrackBuddy(uri);
1065 : }
1066 954 : }
1067 :
1068 : void
1069 681 : Conversation::Impl::rotateTrackedMembers(const std::string& memberUri, const DeviceId& deviceId)
1070 : {
1071 681 : auto acc = account_.lock();
1072 681 : if (!acc)
1073 0 : return;
1074 :
1075 681 : std::lock_guard lk(trackedMembersMtx_);
1076 681 : if (!isTracking_)
1077 221 : return;
1078 :
1079 460 : if (!memberUri.empty()) {
1080 59 : if (auto it = trackedMembers_.find(memberUri); it != trackedMembers_.end()) {
1081 54 : JAMI_WARNING("{} [device {}] Rotating tracked members after connection failure", toString(), deviceId);
1082 54 : auto& info = it->second;
1083 54 : info.failedDevices.insert(deviceId);
1084 54 : if (std::includes(info.failedDevices.begin(),
1085 : info.failedDevices.end(),
1086 : info.devices.begin(),
1087 : info.devices.end())) {
1088 31 : acc->presenceManager()->untrackBuddy(it->first);
1089 31 : trackedMembers_.erase(it);
1090 : }
1091 : } else {
1092 5 : return;
1093 : }
1094 : }
1095 :
1096 455 : auto members = repository_->members();
1097 455 : size_t N = members.size();
1098 455 : size_t K = std::min(N, 3 + (size_t) std::log2(N));
1099 :
1100 : // If we are already trying enough devices, don't rotate
1101 455 : auto activeDevices = swarmManager_->getActiveNodesCount();
1102 455 : if (activeDevices >= 2 * K)
1103 0 : return;
1104 :
1105 455 : JAMI_WARNING("{} Refreshing tracked members: {}/{} active ({}/{} devices)",
1106 : toString(),
1107 : trackedMembers_.size(),
1108 : K,
1109 : activeDevices,
1110 : 2 * K);
1111 :
1112 : // Add new members if we have space
1113 455 : if (trackedMembers_.size() < K) {
1114 432 : std::vector<std::string> candidates;
1115 432 : candidates.reserve(N - trackedMembers_.size());
1116 1495 : for (const auto& m : members) {
1117 1063 : if (m.uri != memberUri && trackedMembers_.find(m.uri) == trackedMembers_.end()) {
1118 995 : candidates.push_back(m.uri);
1119 : }
1120 : }
1121 432 : if (!candidates.empty()) {
1122 403 : std::vector<std::string> chosen;
1123 403 : std::sample(candidates.begin(),
1124 : candidates.end(),
1125 : std::back_inserter(chosen),
1126 403 : K - trackedMembers_.size(),
1127 403 : Manager::instance().getSeededRandomEngine());
1128 1197 : for (const auto& uri : chosen) {
1129 794 : acc->presenceManager()->trackBuddy(uri);
1130 794 : trackedMembers_.emplace(uri, TrackedMember {});
1131 : }
1132 403 : }
1133 432 : }
1134 907 : }
1135 :
1136 : void
1137 280 : Conversation::Impl::onConnectionFailed(const DeviceId& deviceId, const std::string& memberUri)
1138 : {
1139 280 : rotateTrackedMembers(memberUri, deviceId);
1140 280 : }
1141 :
1142 : void
1143 513 : Conversation::Impl::monitorConnection(std::weak_ptr<Conversation> w)
1144 : {
1145 513 : if (!swarmManager_->isConnected()) {
1146 505 : startTracking(w);
1147 : }
1148 :
1149 513 : swarmManager_->onConnectionChanged([w](bool ok) {
1150 427 : dht::ThreadPool::io().run([w, ok] {
1151 427 : if (auto sthis = w.lock()) {
1152 : // Check the current connection state rather than relying on the
1153 : // captured `ok` value, because the thread pool does not guarantee
1154 : // execution order. Two rapid state changes (connected then
1155 : // disconnected) could otherwise cause stopTracking to run after
1156 : // startTracking, leaving the conversation untracked while the DRT
1157 : // has no connected nodes.
1158 427 : if (sthis->pimpl_->swarmManager_->isConnected()) {
1159 380 : sthis->pimpl_->stopTracking();
1160 : } else {
1161 46 : sthis->pimpl_->startTracking(w);
1162 : }
1163 427 : if (ok) {
1164 376 : if (sthis->pimpl_->bootstrapCb_)
1165 376 : sthis->pimpl_->bootstrapCb_();
1166 : }
1167 : #ifdef LIBJAMI_TEST
1168 427 : if (sthis->pimpl_->bootstrapCbTest_)
1169 30 : sthis->pimpl_->bootstrapCbTest_(sthis->id(),
1170 15 : ok ? BootstrapStatus::SUCCESS : BootstrapStatus::FAILED);
1171 : #endif
1172 427 : }
1173 427 : });
1174 427 : });
1175 513 : }
1176 :
1177 : bool
1178 16 : Conversation::Impl::isAdmin() const
1179 : {
1180 16 : auto adminsPath = repoPath_ / MemberPath::ADMINS;
1181 32 : return std::filesystem::is_regular_file(fileutils::getFullPath(adminsPath, userId_ + ".crt"));
1182 16 : }
1183 :
1184 : void
1185 11 : Conversation::Impl::disconnectFromDevice(const DeviceId& deviceId)
1186 : {
1187 22 : swarmManager_->deleteNode({deviceId});
1188 :
1189 11 : std::shared_ptr<dhtnet::ChannelSocket> socket;
1190 : {
1191 11 : std::lock_guard lk(gitSocketMtx_);
1192 11 : if (auto it = gitSocketList_.find(deviceId); it != gitSocketList_.end()) {
1193 11 : socket = std::move(it->second);
1194 11 : gitSocketList_.erase(it);
1195 : }
1196 11 : }
1197 11 : if (socket)
1198 11 : socket->shutdown();
1199 11 : }
1200 :
1201 : void
1202 16 : Conversation::Impl::disconnectFromPeer(const std::string& peerUri)
1203 : {
1204 16 : std::set<DeviceId> devicesToRemove;
1205 :
1206 16 : const auto nodes = swarmManager_->getAllNodes();
1207 26 : for (const auto node : nodes)
1208 10 : if (peerUri == repository_->uriFromDevice(node.toString()))
1209 5 : devicesToRemove.emplace(node);
1210 :
1211 : {
1212 16 : std::lock_guard lk(gitSocketMtx_);
1213 37 : for (const auto& [deviceId, socket] : gitSocketList_) {
1214 21 : auto cert = socket ? socket->peerCertificate() : nullptr;
1215 42 : if ((cert && cert->issuer && cert->issuer->getId().toString() == peerUri)
1216 42 : || peerUri == repository_->uriFromDevice(deviceId.toString())) {
1217 11 : devicesToRemove.emplace(deviceId);
1218 : }
1219 21 : }
1220 16 : }
1221 :
1222 27 : for (const auto& deviceId : devicesToRemove)
1223 11 : disconnectFromDevice(deviceId);
1224 16 : }
1225 :
1226 : std::vector<std::map<std::string, std::string>>
1227 101 : Conversation::Impl::getMembers(bool includeInvited, bool includeLeft, bool includeBanned) const
1228 : {
1229 101 : std::vector<std::map<std::string, std::string>> result;
1230 101 : auto members = repository_->members();
1231 101 : std::lock_guard lk(messageStatusMtx_);
1232 304 : for (const auto& member : members) {
1233 203 : if (member.role == MemberRole::BANNED && !includeBanned) {
1234 0 : continue;
1235 : }
1236 203 : if (member.role == MemberRole::INVITED && !includeInvited)
1237 0 : continue;
1238 203 : if (member.role == MemberRole::LEFT && !includeLeft)
1239 0 : continue;
1240 203 : auto mm = member.map();
1241 203 : auto it = messagesStatus_.find(member.uri);
1242 203 : if (it != messagesStatus_.end()) {
1243 274 : auto readIt = it->second.find("read");
1244 137 : if (readIt != it->second.end())
1245 255 : mm[ConversationMapKeys::LAST_DISPLAYED] = readIt->second;
1246 : }
1247 203 : result.emplace_back(std::move(mm));
1248 203 : }
1249 202 : return result;
1250 101 : }
1251 :
1252 : std::vector<std::map<std::string, std::string>>
1253 0 : Conversation::Impl::getTrackedMembers() const
1254 : {
1255 0 : std::vector<std::map<std::string, std::string>> result;
1256 0 : std::lock_guard lk(trackedMembersMtx_);
1257 0 : for (const auto& [uri, member] : trackedMembers_) {
1258 0 : std::map<std::string, std::string> map;
1259 0 : map["uri"] = uri;
1260 0 : std::string devicesStr;
1261 0 : for (const auto& dev : member.devices) {
1262 0 : if (!devicesStr.empty())
1263 0 : devicesStr += ";";
1264 0 : devicesStr += dev.toString();
1265 : }
1266 0 : map["devices"] = devicesStr;
1267 0 : result.emplace_back(std::move(map));
1268 0 : }
1269 0 : return result;
1270 0 : }
1271 :
1272 : std::vector<std::string>
1273 18 : Conversation::Impl::commitsEndedCalls()
1274 : {
1275 : // Handle current calls
1276 18 : std::vector<std::string> commits {};
1277 18 : std::unique_lock lk(writeMtx_);
1278 18 : std::unique_lock lkA(activeCallsMtx_);
1279 18 : for (const auto& hostedCall : hostedCalls_) {
1280 : // In this case, this means that we left
1281 : // the conference while still hosting it, so activeCalls
1282 : // will not be correctly updated
1283 : // We don't need to send notifications there, as peers will sync with presence
1284 0 : auto confId = hostedCall.first;
1285 0 : auto now = std::chrono::system_clock::now();
1286 0 : uint64_t nowConverted = std::chrono::duration_cast<std::chrono::seconds>(now.time_since_epoch()).count();
1287 0 : auto duration = (nowConverted > hostedCall.second) ? (nowConverted - hostedCall.second) * 1000 : 0;
1288 0 : auto commitMessage = CommitMessage::conferenceHostingEnd(confId, deviceId_, userId_, duration);
1289 :
1290 0 : auto itActive = std::find_if(activeCalls_.begin(),
1291 : activeCalls_.end(),
1292 0 : [this, confId = hostedCall.first](const auto& value) {
1293 0 : return value.at("id") == confId && value.at("uri") == userId_
1294 0 : && value.at("device") == deviceId_;
1295 : });
1296 0 : if (itActive != activeCalls_.end())
1297 0 : activeCalls_.erase(itActive);
1298 0 : commits.emplace_back(repository_->commitMessage(commitMessage.toString()));
1299 0 : JAMI_DEBUG("Removing hosted conference... {:s}", hostedCall.first);
1300 0 : }
1301 18 : hostedCalls_.clear();
1302 18 : saveActiveCalls();
1303 18 : saveHostedCalls();
1304 36 : return commits;
1305 18 : }
1306 :
1307 : std::vector<libjami::SwarmMessage>
1308 1704 : Conversation::Impl::loadMessages(const LogOptions& options, History* optHistory)
1309 : {
1310 1704 : auto history = optHistory ? optHistory : &loadedHistory_;
1311 :
1312 : // history->mutex is locked by the caller
1313 1704 : if (!repository_ || history->loading) {
1314 0 : return {};
1315 : }
1316 1704 : history->loading = true;
1317 :
1318 : // By convention, if options.nbOfCommits is zero, then we
1319 : // don't impose a limit on the number of commits returned.
1320 1704 : bool limitNbOfCommits = options.nbOfCommits > 0;
1321 :
1322 1704 : auto startLogging = options.from == "";
1323 1704 : auto breakLogging = false;
1324 1704 : auto currentHistorySize = loadedHistory_.messageList.size();
1325 1704 : std::vector<std::string> replies;
1326 1704 : std::vector<std::shared_ptr<libjami::SwarmMessage>> msgList;
1327 3408 : repository_->log(
1328 : /* preCondition */
1329 3408 : [&](const auto& id, const auto& author, const auto& commit) {
1330 8007 : if (options.skipMerge && git_commit_parentcount(commit.get()) > 1) {
1331 2 : return CallbackResult::Skip;
1332 : }
1333 8006 : if (id == options.to) {
1334 702 : if (options.includeTo)
1335 0 : breakLogging = true; // For the next commit
1336 : }
1337 7997 : if (replies.empty()) { // This avoid load until
1338 : // NOTE: in the future, we may want to add "Reply-Body" in commit to avoid to load
1339 : // until this commit
1340 7999 : if ((limitNbOfCommits
1341 7999 : && (loadedHistory_.messageList.size() - currentHistorySize) == options.nbOfCommits))
1342 1 : return CallbackResult::Break; // Stop logging
1343 7998 : if (breakLogging)
1344 0 : return CallbackResult::Break; // Stop logging
1345 7998 : if (id == options.to && !options.includeTo) {
1346 701 : return CallbackResult::Break; // Stop logging
1347 : }
1348 : }
1349 :
1350 7294 : if (!startLogging && options.from != "" && options.from == id)
1351 1104 : startLogging = true;
1352 7296 : if (!startLogging)
1353 24 : return CallbackResult::Skip; // Start logging after this one
1354 :
1355 7272 : if (options.fastLog) {
1356 6196 : if (options.authorUri != "") {
1357 1 : if (options.authorUri == repository_->uriFromDevice(author.email)) {
1358 1 : return CallbackResult::Break; // Found author, stop
1359 : }
1360 : }
1361 : }
1362 :
1363 7269 : return CallbackResult::Ok; // Continue
1364 : },
1365 : /* emplaceCb */
1366 3408 : [&](auto&& cc) {
1367 7277 : if (limitNbOfCommits && (msgList.size() == options.nbOfCommits))
1368 483 : return;
1369 6795 : auto optMessage = repository_->convCommitToMap(cc);
1370 6795 : if (!optMessage.has_value())
1371 1 : return;
1372 6790 : auto message = optMessage.value();
1373 20379 : if (message.find(CommitKey::REPLY_TO) != message.end()) {
1374 1 : auto it = std::find(replies.begin(), replies.end(), message.at(CommitKey::REPLY_TO));
1375 1 : if (it == replies.end()) {
1376 3 : replies.emplace_back(message.at(CommitKey::REPLY_TO));
1377 : }
1378 : }
1379 6791 : auto it = std::find(replies.begin(), replies.end(), message.at("id"));
1380 6789 : if (it != replies.end()) {
1381 1 : replies.erase(it);
1382 : }
1383 6795 : std::shared_ptr<libjami::SwarmMessage> firstMsg;
1384 6795 : if ((history == &loadedHistory_) && msgList.empty() && !loadedHistory_.messageList.empty()) {
1385 0 : firstMsg = *loadedHistory_.messageList.rbegin();
1386 : }
1387 20383 : auto added = addToHistory(*history, {message}, false, false);
1388 6794 : if (!added.empty() && firstMsg) {
1389 0 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, repository_->id(), *firstMsg);
1390 : }
1391 6796 : msgList.insert(msgList.end(), added.begin(), added.end());
1392 13595 : },
1393 : /* postCondition */
1394 0 : [&](auto, auto, auto) {
1395 : // Stop logging if there was a limit set on the number of commits
1396 : // to return and we reached it. This isn't strictly necessary since
1397 : // the check at the beginning of `emplaceCb` ensures that we won't
1398 : // return too many messages, but it prevents us from needlessly
1399 : // iterating over a (potentially) large number of commits.
1400 7273 : return limitNbOfCommits && (msgList.size() == options.nbOfCommits);
1401 : },
1402 1704 : options.from,
1403 1704 : options.logIfNotFound);
1404 :
1405 1704 : history->loading = false;
1406 1704 : history->cv.notify_all();
1407 :
1408 : // Convert for client (remove ptr)
1409 1704 : std::vector<libjami::SwarmMessage> ret;
1410 1704 : ret.reserve(msgList.size());
1411 8475 : for (const auto& msg : msgList) {
1412 6777 : ret.emplace_back(*msg);
1413 : }
1414 1702 : return ret;
1415 1704 : }
1416 :
1417 : void
1418 3 : Conversation::Impl::handleReaction(History& history, const std::shared_ptr<libjami::SwarmMessage>& sharedCommit) const
1419 : {
1420 6 : auto it = history.quickAccess.find(sharedCommit->body.at(CommitKey::REACT_TO));
1421 3 : auto peditIt = history.pendingEditions.find(sharedCommit->id);
1422 3 : if (peditIt != history.pendingEditions.end()) {
1423 0 : auto oldBody = sharedCommit->body;
1424 0 : sharedCommit->body[CommitKey::BODY] = peditIt->second.front()->body[CommitKey::BODY];
1425 0 : if (sharedCommit->body.at(CommitKey::BODY).empty())
1426 0 : return;
1427 0 : history.pendingEditions.erase(peditIt);
1428 0 : }
1429 3 : if (it != history.quickAccess.end()) {
1430 3 : it->second->reactions.emplace_back(sharedCommit->body);
1431 3 : emitSignal<libjami::ConversationSignal::ReactionAdded>(accountId_,
1432 3 : repository_->id(),
1433 3 : it->second->id,
1434 3 : sharedCommit->body);
1435 : } else {
1436 0 : history.pendingReactions[sharedCommit->body.at(CommitKey::REACT_TO)].emplace_back(sharedCommit->body);
1437 : }
1438 : }
1439 :
1440 : void
1441 9 : Conversation::Impl::handleEdition(History& history,
1442 : const std::shared_ptr<libjami::SwarmMessage>& sharedCommit,
1443 : bool messageReceived) const
1444 : {
1445 18 : auto editId = sharedCommit->body.at(CommitKey::EDIT);
1446 9 : auto it = history.quickAccess.find(editId);
1447 9 : if (it != history.quickAccess.end()) {
1448 7 : auto baseCommit = it->second;
1449 7 : if (baseCommit) {
1450 21 : auto itReact = baseCommit->body.find(CommitKey::REACT_TO);
1451 7 : std::string toReplace = (baseCommit->type == CommitType::DATA_TRANSFER) ? CommitKey::TID : CommitKey::BODY;
1452 7 : auto body = sharedCommit->body.at(toReplace);
1453 : // Edit reaction
1454 7 : if (itReact != baseCommit->body.end()) {
1455 1 : baseCommit->body[toReplace] = body; // Replace body if pending
1456 1 : it = history.quickAccess.find(itReact->second);
1457 1 : auto itPending = history.pendingReactions.find(itReact->second);
1458 1 : if (it != history.quickAccess.end()) {
1459 1 : baseCommit = it->second; // Base commit
1460 1 : auto itPreviousReact = std::find_if(baseCommit->reactions.begin(),
1461 1 : baseCommit->reactions.end(),
1462 1 : [&](const auto& reaction) {
1463 3 : return reaction.at("id") == editId;
1464 : });
1465 1 : if (itPreviousReact != baseCommit->reactions.end()) {
1466 1 : (*itPreviousReact)[toReplace] = body;
1467 1 : if (body.empty()) {
1468 1 : baseCommit->reactions.erase(itPreviousReact);
1469 2 : emitSignal<libjami::ConversationSignal::ReactionRemoved>(accountId_,
1470 1 : repository_->id(),
1471 1 : baseCommit->id,
1472 : editId);
1473 : }
1474 : }
1475 0 : } else if (itPending != history.pendingReactions.end()) {
1476 : // Else edit if pending
1477 0 : auto itReaction = std::find_if(itPending->second.begin(),
1478 0 : itPending->second.end(),
1479 0 : [&](const auto& reaction) { return reaction.at("id") == editId; });
1480 0 : if (itReaction != itPending->second.end()) {
1481 0 : (*itReaction)[toReplace] = body;
1482 0 : if (body.empty())
1483 0 : itPending->second.erase(itReaction);
1484 : }
1485 : } else {
1486 : // Add to pending edtions
1487 0 : messageReceived ? history.pendingEditions[editId].emplace_front(sharedCommit)
1488 0 : : history.pendingEditions[editId].emplace_back(sharedCommit);
1489 : }
1490 : } else {
1491 : // Normal message
1492 6 : auto editionBody = it->second->body;
1493 : // Tag the superseded body with the commit id that introduced it, so the
1494 : // client can look up per-edition plugin overwrites via quickAccess.
1495 12 : editionBody["id"] = it->second->latestEditionId.empty() ? it->second->id : it->second->latestEditionId;
1496 6 : it->second->editions.emplace(it->second->editions.begin(), std::move(editionBody));
1497 : #ifdef ENABLE_PLUGIN
1498 12 : if (const auto boIt = it->second->pluginData.find("bodyOverwrite");
1499 6 : boIt != it->second->pluginData.end()) {
1500 0 : it->second->editions.front()["bodyOverwrite"] = boIt->second;
1501 : }
1502 : #endif
1503 6 : it->second->body[toReplace] = sharedCommit->body[toReplace];
1504 6 : it->second->latestEditionId = sharedCommit->id;
1505 6 : if (toReplace == CommitKey::TID) {
1506 : // Avoid to replace fileId in client
1507 6 : it->second->body["fileId"] = "";
1508 : }
1509 : // Remove reactions
1510 6 : if (sharedCommit->body.at(toReplace).empty())
1511 3 : it->second->reactions.clear();
1512 6 : }
1513 7 : }
1514 7 : } else {
1515 2 : messageReceived ? history.pendingEditions[editId].emplace_front(sharedCommit)
1516 2 : : history.pendingEditions[editId].emplace_back(sharedCommit);
1517 : }
1518 9 : }
1519 :
1520 : bool
1521 8015 : Conversation::Impl::handleMessage(History& history,
1522 : const std::shared_ptr<libjami::SwarmMessage>& sharedCommit,
1523 : bool messageReceived) const
1524 : {
1525 8015 : if (messageReceived) {
1526 : // For a received message, we place it at the beginning of the list
1527 1207 : if (!history.messageList.empty())
1528 962 : sharedCommit->linearizedParent = (*history.messageList.begin())->id;
1529 1207 : history.messageList.emplace_front(sharedCommit);
1530 : } else {
1531 : // For a loaded message, we load from newest to oldest
1532 : // So we change the parent of the last message.
1533 6808 : if (!history.messageList.empty())
1534 5114 : (*history.messageList.rbegin())->linearizedParent = sharedCommit->id;
1535 6797 : history.messageList.emplace_back(sharedCommit);
1536 : }
1537 : // Handle pending reactions/editions
1538 8015 : auto reactIt = history.pendingReactions.find(sharedCommit->id);
1539 8014 : if (reactIt != history.pendingReactions.end()) {
1540 0 : for (const auto& commitBody : reactIt->second)
1541 0 : sharedCommit->reactions.emplace_back(commitBody);
1542 0 : history.pendingReactions.erase(reactIt);
1543 : }
1544 8016 : auto peditIt = history.pendingEditions.find(sharedCommit->id);
1545 8012 : if (peditIt != history.pendingEditions.end()) {
1546 0 : auto oldBody = sharedCommit->body;
1547 : // Tag original body with its own message id for client identification.
1548 0 : oldBody["id"] = sharedCommit->id;
1549 0 : sharedCommit->latestEditionId = peditIt->second.front()->id;
1550 0 : if (sharedCommit->type == CommitType::DATA_TRANSFER) {
1551 0 : sharedCommit->body[CommitKey::TID] = peditIt->second.front()->body[CommitKey::TID];
1552 0 : sharedCommit->body["fileId"] = "";
1553 : } else {
1554 0 : sharedCommit->body[CommitKey::BODY] = peditIt->second.front()->body[CommitKey::BODY];
1555 : }
1556 0 : peditIt->second.pop_front();
1557 0 : for (const auto& commit : peditIt->second) {
1558 0 : auto edBody = commit->body;
1559 : // Tag each superseded body with its edition commit id, mirroring
1560 : // handleEdition so injectEditionOverwrites can find plugin overwrites.
1561 0 : edBody["id"] = commit->id;
1562 0 : sharedCommit->editions.emplace_back(std::move(edBody));
1563 0 : }
1564 0 : sharedCommit->editions.emplace_back(std::move(oldBody));
1565 0 : history.pendingEditions.erase(peditIt);
1566 0 : }
1567 : // Announce to client
1568 8012 : if (messageReceived)
1569 1207 : emitSignal<libjami::ConversationSignal::SwarmMessageReceived>(accountId_, repository_->id(), *sharedCommit);
1570 8011 : return !messageReceived;
1571 : }
1572 :
1573 : void
1574 8028 : Conversation::Impl::rectifyStatus(const std::shared_ptr<libjami::SwarmMessage>& message, History& history) const
1575 : {
1576 8028 : auto parentIt = history.quickAccess.find(message->linearizedParent);
1577 8020 : auto currentMessage = message;
1578 :
1579 19935 : while (parentIt != history.quickAccess.end()) {
1580 11941 : const auto& parent = parentIt->second;
1581 13377 : for (const auto& [peer, value] : message->status) {
1582 12291 : auto parentStatusIt = parent->status.find(peer);
1583 12198 : if (parentStatusIt == parent->status.end() || parentStatusIt->second < value) {
1584 1428 : parent->status[peer] = value;
1585 2855 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(accountId_,
1586 1422 : repository_->id(),
1587 : peer,
1588 1424 : parent->id,
1589 : value);
1590 10857 : } else if (parentStatusIt->second >= value) {
1591 10853 : break;
1592 : }
1593 : }
1594 11952 : currentMessage = parent;
1595 11968 : parentIt = history.quickAccess.find(parent->linearizedParent);
1596 : }
1597 8021 : }
1598 :
1599 : std::vector<std::shared_ptr<libjami::SwarmMessage>>
1600 8018 : Conversation::Impl::addToHistory(History& history,
1601 : const std::vector<std::map<std::string, std::string>>& commits,
1602 : bool messageReceived,
1603 : bool commitFromSelf)
1604 : {
1605 : //
1606 : // NOTE: This function makes the following assumptions on its arguments:
1607 : // - The messages in "history" are in reverse chronological order (newest message
1608 : // first, oldest message last).
1609 : // - If messageReceived is true, then the commits in "commits" are assumed to be in
1610 : // chronological order (oldest to newest) and to be newer than the ones in "history".
1611 : // They are therefore inserted at the beginning of the message list.
1612 : // - If messageReceived is false, then the commits in "commits" are assumed to be in
1613 : // reverse chronological order (newest to oldest) and to be older than the ones in
1614 : // "history". They are therefore appended at the end of the message list.
1615 : //
1616 8018 : auto acc = account_.lock();
1617 8016 : if (!acc)
1618 0 : return {};
1619 8012 : auto username = acc->getUsername();
1620 8008 : if (messageReceived && (&history == &loadedHistory_ && history.loading)) {
1621 0 : std::unique_lock lk(history.mutex);
1622 0 : history.cv.wait(lk, [&] { return !history.loading; });
1623 0 : }
1624 :
1625 : // Only set messages' status on history for client
1626 8008 : bool needToSetMessageStatus = !commitFromSelf && &history == &loadedHistory_;
1627 :
1628 8008 : std::vector<std::shared_ptr<libjami::SwarmMessage>> sharedCommits;
1629 16050 : for (const auto& commit : commits) {
1630 8049 : auto commitId = commit.at("id");
1631 8046 : if (history.quickAccess.find(commitId) != history.quickAccess.end())
1632 1 : continue; // Already present
1633 8046 : auto typeIt = commit.find(CommitKey::TYPE);
1634 : // Nothing to show for the client, skip
1635 8045 : if (typeIt != commit.end() && typeIt->second == CommitType::MERGE)
1636 23 : continue;
1637 :
1638 8026 : auto sharedCommit = std::make_shared<libjami::SwarmMessage>();
1639 8024 : sharedCommit->fromMapStringString(commit);
1640 :
1641 8027 : if (needToSetMessageStatus) {
1642 915 : std::lock_guard lk(messageStatusMtx_);
1643 : // Check if we already have status information for the commit.
1644 917 : auto itFuture = futureStatus.find(sharedCommit->id);
1645 916 : if (itFuture != futureStatus.end()) {
1646 9 : sharedCommit->status = std::move(itFuture->second);
1647 9 : futureStatus.erase(itFuture);
1648 : }
1649 916 : }
1650 8028 : sharedCommits.emplace_back(sharedCommit);
1651 8048 : }
1652 :
1653 8015 : if (needToSetMessageStatus) {
1654 912 : constexpr int32_t SENDING = static_cast<int32_t>(libjami::Account::MessageStates::SENDING);
1655 912 : constexpr int32_t SENT = static_cast<int32_t>(libjami::Account::MessageStates::SENT);
1656 912 : constexpr int32_t DISPLAYED = static_cast<int32_t>(libjami::Account::MessageStates::DISPLAYED);
1657 :
1658 912 : std::lock_guard lk(messageStatusMtx_);
1659 13481 : for (const auto& member : repository_->members()) {
1660 : // For each member, we iterate over the commits to add in reverse chronological
1661 : // order (i.e. from newest to oldest) and set their status from the point of view
1662 : // of that member (as best we can given the information we have).
1663 : //
1664 : // The key assumption made in order to compute the status is that it can never decrease
1665 : // (with respect to the ordering SENDING < SENT < DISPLAYED) as we go back in time. We
1666 : // therefore start by setting the "status" variable below to the lowest possible value,
1667 : // and increase it when we encounter a commit for which it is justified to do so.
1668 : //
1669 : // If messageReceived is true, then the commits we are adding are the most recent in the
1670 : // conversation history, so the lowest possible value is SENDING.
1671 : //
1672 : // If messageReceived is false, then the commits we are adding are older than the ones
1673 : // that are already in the history, so the lowest possible value is the status of the
1674 : // oldest message in the history so far, which is stored in memberToStatus.
1675 12569 : auto status = SENDING;
1676 12569 : if (!messageReceived) {
1677 12 : auto cache = memberToStatus[member.uri];
1678 12 : if (cache > status)
1679 9 : status = cache;
1680 : }
1681 12569 : auto& messagesStatus = messagesStatus_[member.uri];
1682 :
1683 25121 : for (auto it = sharedCommits.rbegin(); it != sharedCommits.rend(); it++) {
1684 12581 : auto sharedCommit = *it;
1685 12577 : auto previousStatus = status;
1686 12577 : auto& commitStatus = sharedCommit->status[member.uri];
1687 :
1688 : // Compute status for the current commit.
1689 37639 : if (status < SENT && messagesStatus["fetched"] == sharedCommit->id) {
1690 9 : status = SENT;
1691 : }
1692 37593 : if (messagesStatus["read"] == sharedCommit->id) {
1693 1 : status = DISPLAYED;
1694 : }
1695 37604 : if (member.uri == sharedCommit->body.at("author")) {
1696 915 : status = DISPLAYED;
1697 : }
1698 12519 : if (status < commitStatus) {
1699 0 : status = commitStatus;
1700 : }
1701 :
1702 : // Store computed value.
1703 12519 : commitStatus = status;
1704 :
1705 : // Update messagesStatus_ if needed.
1706 12519 : if (previousStatus == SENDING && status >= SENT) {
1707 2734 : messagesStatus["fetched"] = sharedCommit->id;
1708 : }
1709 12520 : if (previousStatus <= SENT && status == DISPLAYED) {
1710 2712 : messagesStatus["read"] = sharedCommit->id;
1711 : }
1712 12520 : }
1713 :
1714 12568 : if (!messageReceived) {
1715 : // Update memberToStatus with the status of the last (i.e. oldest) added commit.
1716 12 : memberToStatus[member.uri] = status;
1717 : }
1718 911 : }
1719 913 : }
1720 :
1721 8016 : std::vector<std::shared_ptr<libjami::SwarmMessage>> messages;
1722 16041 : for (const auto& sharedCommit : sharedCommits) {
1723 8025 : history.quickAccess[sharedCommit->id] = sharedCommit;
1724 :
1725 16054 : auto reactToIt = sharedCommit->body.find(CommitKey::REACT_TO);
1726 16043 : auto editIt = sharedCommit->body.find(CommitKey::EDIT);
1727 8022 : if (reactToIt != sharedCommit->body.end() && !reactToIt->second.empty()) {
1728 3 : handleReaction(history, sharedCommit);
1729 8025 : } else if (editIt != sharedCommit->body.end() && !editIt->second.empty()) {
1730 9 : handleEdition(history, sharedCommit, messageReceived);
1731 8014 : } else if (handleMessage(history, sharedCommit, messageReceived)) {
1732 6805 : messages.emplace_back(sharedCommit);
1733 : }
1734 8025 : rectifyStatus(sharedCommit, history);
1735 : }
1736 :
1737 8010 : return messages;
1738 8016 : }
1739 :
1740 187 : Conversation::Conversation(const std::shared_ptr<JamiAccount>& account,
1741 : ConversationMode mode,
1742 187 : const std::string& otherMember)
1743 187 : : pimpl_ {new Impl {account, mode, otherMember}}
1744 187 : {}
1745 :
1746 18 : Conversation::Conversation(const std::shared_ptr<JamiAccount>& account, const std::string& conversationId)
1747 18 : : pimpl_ {new Impl {account, conversationId}}
1748 18 : {}
1749 :
1750 200 : Conversation::Conversation(const std::shared_ptr<JamiAccount>& account,
1751 : const std::string& remoteDevice,
1752 200 : const std::string& conversationId)
1753 200 : : pimpl_ {new Impl {account, remoteDevice, conversationId}}
1754 200 : {}
1755 :
1756 391 : Conversation::~Conversation() {}
1757 :
1758 : std::string
1759 5978 : Conversation::id() const
1760 : {
1761 5978 : return pimpl_->repository_ ? pimpl_->repository_->id() : "";
1762 : }
1763 :
1764 : void
1765 139 : Conversation::addMember(const std::string& contactUri, const OnDoneCb& cb)
1766 : {
1767 : try {
1768 139 : if (mode() == ConversationMode::ONE_TO_ONE) {
1769 : // Only authorize to add left members
1770 1 : auto initialMembers = getInitialMembers();
1771 1 : auto it = std::find(initialMembers.begin(), initialMembers.end(), contactUri);
1772 1 : if (it == initialMembers.end()) {
1773 1 : JAMI_WARNING("Unable to add new member in one to one conversation");
1774 1 : cb(false, "");
1775 1 : return;
1776 : }
1777 1 : }
1778 0 : } catch (const std::exception& e) {
1779 0 : JAMI_WARNING("Unable to get mode: {}", e.what());
1780 0 : cb(false, "");
1781 0 : return;
1782 0 : }
1783 138 : if (isMember(contactUri, true)) {
1784 0 : JAMI_WARNING("Unable to add member {} because it's already a member", contactUri);
1785 0 : cb(false, "");
1786 0 : return;
1787 : }
1788 138 : if (isMemberBanned(contactUri)) {
1789 2 : if (pimpl_->isAdmin()) {
1790 1 : dht::ThreadPool::io().run([w = weak(), contactUri = std::move(contactUri), cb = std::move(cb)] {
1791 1 : if (auto sthis = w.lock()) {
1792 1 : auto members = sthis->pimpl_->repository_->members();
1793 1 : auto type = sthis->pimpl_->memberBanType(contactUri);
1794 1 : if (type.empty()) {
1795 0 : cb(false, {});
1796 0 : return;
1797 : }
1798 1 : sthis->pimpl_->voteUnban(contactUri, type, cb);
1799 2 : }
1800 : });
1801 : } else {
1802 1 : JAMI_WARNING("Unable to add member {} because this member is blocked", contactUri);
1803 2 : cb(false, "");
1804 : }
1805 2 : return;
1806 : }
1807 :
1808 136 : dht::ThreadPool::io().run([w = weak(), contactUri = std::move(contactUri), cb = std::move(cb)] {
1809 136 : if (auto sthis = w.lock()) {
1810 : // Add member files and commit
1811 136 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
1812 136 : auto commit = sthis->pimpl_->repository_->addMember(contactUri);
1813 136 : if (not commit.empty())
1814 136 : sthis->pimpl_->announce(commit, true);
1815 136 : lk.unlock();
1816 136 : if (cb)
1817 136 : cb(!commit.empty(), commit);
1818 272 : }
1819 136 : });
1820 : }
1821 :
1822 : std::shared_ptr<dhtnet::ChannelSocket>
1823 4248 : Conversation::gitSocket(const DeviceId& deviceId) const
1824 : {
1825 4248 : return pimpl_->gitSocket(deviceId);
1826 : }
1827 :
1828 : void
1829 1820 : Conversation::addGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& socket)
1830 : {
1831 1820 : pimpl_->addGitSocket(deviceId, socket);
1832 1821 : }
1833 :
1834 : void
1835 935 : Conversation::removeGitSocket(const DeviceId& deviceId)
1836 : {
1837 935 : pimpl_->removeGitSocket(deviceId);
1838 935 : }
1839 :
1840 : void
1841 382 : Conversation::shutdownConnections()
1842 : {
1843 : {
1844 382 : std::lock_guard lk(pimpl_->gitSocketMtx_);
1845 382 : pimpl_->gitSocketList_.clear();
1846 382 : }
1847 382 : if (pimpl_->swarmManager_)
1848 382 : pimpl_->swarmManager_->shutdown();
1849 382 : }
1850 :
1851 : void
1852 0 : Conversation::connectivityChanged()
1853 : {
1854 0 : if (pimpl_->swarmManager_)
1855 0 : pimpl_->swarmManager_->maintainBuckets();
1856 0 : }
1857 :
1858 : std::vector<jami::DeviceId>
1859 0 : Conversation::getDeviceIdList() const
1860 : {
1861 0 : return pimpl_->swarmManager_->getAllNodes();
1862 : }
1863 :
1864 : std::shared_ptr<Typers>
1865 9 : Conversation::typers() const
1866 : {
1867 9 : return pimpl_->typers_;
1868 : }
1869 :
1870 : std::vector<std::map<std::string, std::string>>
1871 0 : Conversation::getConnectivity() const
1872 : {
1873 0 : return pimpl_->getConnectivity();
1874 : }
1875 :
1876 : void
1877 1 : Conversation::Impl::voteUnban(const std::string& contactUri, const std::string_view type, const OnDoneCb& cb)
1878 : {
1879 : // Check if admin
1880 1 : if (!isAdmin()) {
1881 0 : JAMI_WARNING("You're not an admin of this repo. Unable to unblock {}", contactUri);
1882 0 : cb(false, {});
1883 0 : return;
1884 : }
1885 :
1886 : // Vote for removal
1887 1 : std::unique_lock lk(writeMtx_);
1888 1 : auto voteCommit = repository_->voteUnban(contactUri, type);
1889 1 : if (voteCommit.empty()) {
1890 0 : JAMI_WARNING("Unbanning {} failed", contactUri);
1891 0 : cb(false, "");
1892 0 : return;
1893 : }
1894 :
1895 1 : auto lastId = voteCommit;
1896 1 : std::vector<std::string> commits;
1897 1 : commits.emplace_back(voteCommit);
1898 :
1899 : // If admin, check vote
1900 2 : auto resolveCommit = repository_->resolveVote(contactUri, type, CommitAction::UNBAN);
1901 1 : if (!resolveCommit.empty()) {
1902 1 : commits.emplace_back(resolveCommit);
1903 1 : lastId = resolveCommit;
1904 1 : JAMI_WARNING("Vote solved for unbanning {}.", contactUri);
1905 : }
1906 1 : announce(commits, true);
1907 1 : lk.unlock();
1908 1 : if (cb)
1909 1 : cb(!lastId.empty(), lastId);
1910 1 : }
1911 :
1912 : void
1913 13 : Conversation::removeMember(const std::string& contactUri, bool isDevice, const OnDoneCb& cb)
1914 : {
1915 26 : dht::ThreadPool::io().run(
1916 26 : [w = weak(), contactUri = std::move(contactUri), isDevice = std::move(isDevice), cb = std::move(cb)] {
1917 13 : if (auto sthis = w.lock()) {
1918 : // Check if admin
1919 13 : if (!sthis->pimpl_->isAdmin()) {
1920 1 : JAMI_WARNING("You're not an admin of this repo. Unable to block {}", contactUri);
1921 1 : cb(false, {});
1922 2 : return;
1923 : }
1924 :
1925 : // Get current user type
1926 12 : std::string type;
1927 12 : if (isDevice) {
1928 0 : type = "devices";
1929 : } else {
1930 12 : auto members = sthis->pimpl_->repository_->members();
1931 26 : for (const auto& member : members) {
1932 26 : if (member.uri == contactUri) {
1933 12 : if (member.role == MemberRole::INVITED) {
1934 2 : type = "invited";
1935 10 : } else if (member.role == MemberRole::ADMIN) {
1936 1 : type = "admins";
1937 9 : } else if (member.role == MemberRole::MEMBER) {
1938 9 : type = "members";
1939 : }
1940 12 : break;
1941 : }
1942 : }
1943 12 : if (type.empty()) {
1944 0 : cb(false, {});
1945 0 : return;
1946 : }
1947 12 : }
1948 :
1949 : // Vote for removal
1950 12 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
1951 12 : auto voteCommit = sthis->pimpl_->repository_->voteKick(contactUri, type);
1952 12 : if (voteCommit.empty()) {
1953 1 : JAMI_WARNING("Kicking {} failed", contactUri);
1954 2 : cb(false, "");
1955 1 : return;
1956 : }
1957 :
1958 11 : auto lastId = voteCommit;
1959 11 : std::vector<std::string> commits;
1960 11 : commits.emplace_back(voteCommit);
1961 :
1962 : // If admin, check vote
1963 22 : auto resolveCommit = sthis->pimpl_->repository_->resolveVote(contactUri, type, CommitAction::BAN);
1964 11 : if (!resolveCommit.empty()) {
1965 11 : commits.emplace_back(resolveCommit);
1966 11 : lastId = resolveCommit;
1967 11 : JAMI_WARNING("Vote solved for {}. {} banned", contactUri, isDevice ? "Device" : "Member");
1968 11 : if (isDevice)
1969 0 : sthis->pimpl_->disconnectFromDevice(DeviceId(contactUri));
1970 : else
1971 11 : sthis->pimpl_->disconnectFromPeer(contactUri);
1972 : }
1973 :
1974 11 : sthis->pimpl_->announce(commits, true);
1975 11 : lk.unlock();
1976 11 : cb(!lastId.empty(), lastId);
1977 27 : }
1978 : });
1979 13 : }
1980 :
1981 : std::vector<std::map<std::string, std::string>>
1982 101 : Conversation::getMembers(bool includeInvited, bool includeLeft, bool includeBanned) const
1983 : {
1984 101 : return pimpl_->getMembers(includeInvited, includeLeft, includeBanned);
1985 : }
1986 :
1987 : std::vector<std::map<std::string, std::string>>
1988 0 : Conversation::getTrackedMembers() const
1989 : {
1990 0 : return pimpl_->getTrackedMembers();
1991 : }
1992 :
1993 : std::set<std::string>
1994 2538 : Conversation::memberUris(std::string_view filter, const std::set<MemberRole>& filteredRoles) const
1995 : {
1996 2538 : return pimpl_->repository_->memberUris(filter, filteredRoles);
1997 : }
1998 :
1999 : std::vector<NodeId>
2000 1747 : Conversation::peersToSyncWith() const
2001 : {
2002 1747 : auto s = pimpl_->swarmManager_->getConnectedNodes();
2003 1747 : std::lock_guard lk(pimpl_->gitSocketMtx_);
2004 15902 : for (const auto& [deviceId, _] : pimpl_->gitSocketList_)
2005 14152 : if (std::find(s.cbegin(), s.cend(), deviceId) == s.cend())
2006 1331 : s.emplace_back(deviceId);
2007 3495 : return s;
2008 1747 : }
2009 :
2010 : bool
2011 1748 : Conversation::isBootstrapped() const
2012 : {
2013 1748 : return pimpl_->swarmManager_->isConnected();
2014 : }
2015 :
2016 : std::string
2017 15748 : Conversation::uriFromDevice(const std::string& deviceId) const
2018 : {
2019 15748 : return pimpl_->repository_->uriFromDevice(deviceId);
2020 : }
2021 :
2022 : void
2023 0 : Conversation::monitor()
2024 : {
2025 0 : pimpl_->swarmManager_->getRoutingTable().printRoutingTable();
2026 0 : }
2027 :
2028 : std::string
2029 186 : Conversation::join()
2030 : {
2031 186 : return pimpl_->repository_->join();
2032 : }
2033 :
2034 : bool
2035 8719 : Conversation::isMember(const std::string& uri, bool includeInvited) const
2036 : {
2037 8719 : auto uriCrt = uri + ".crt"sv;
2038 17436 : if (std::filesystem::is_regular_file(pimpl_->repoPath_ / MemberPath::ADMINS / uriCrt)
2039 17438 : || std::filesystem::is_regular_file(pimpl_->repoPath_ / MemberPath::MEMBERS / uriCrt)) {
2040 6661 : return true;
2041 : }
2042 2053 : if (includeInvited) {
2043 1478 : if (std::filesystem::is_regular_file(pimpl_->repoPath_ / MemberPath::INVITED / uri)) {
2044 1198 : return true;
2045 : }
2046 280 : if (mode() == ConversationMode::ONE_TO_ONE) {
2047 3 : for (const auto& member : getInitialMembers()) {
2048 2 : if (member == uri)
2049 0 : return true;
2050 1 : }
2051 : }
2052 : }
2053 855 : return false;
2054 8714 : }
2055 :
2056 : bool
2057 5957 : Conversation::isMemberBanned(const std::string& uri) const
2058 : {
2059 5957 : return !pimpl_->memberBanType(uri).empty();
2060 : }
2061 :
2062 : bool
2063 6672 : Conversation::isDeviceBanned(const std::string& deviceId) const
2064 : {
2065 6672 : return pimpl_->isDeviceBanned(deviceId);
2066 : }
2067 :
2068 : bool
2069 5819 : Conversation::isPeerAuthorized(const std::string& uri, const std::string& deviceId, bool includeInvited) const
2070 : {
2071 5819 : return !isMemberBanned(uri) && !isDeviceBanned(deviceId) && isMember(uri, includeInvited);
2072 : }
2073 :
2074 : void
2075 142 : Conversation::createCommit(CommitMessage&& message, OnCommitCb&& onCommit, OnDoneCb&& cb)
2076 : {
2077 142 : if (!message.replyTo.empty()) {
2078 2 : if (!pimpl_->repository_->hasCommit(message.replyTo)) {
2079 1 : JAMI_ERROR("Replying to invalid commit {}", message.replyTo);
2080 1 : return;
2081 : }
2082 : }
2083 282 : dht::ThreadPool::io().run(
2084 282 : [w = weak(), message = std::move(message), onCommit = std::move(onCommit), cb = std::move(cb)] {
2085 141 : if (auto sthis = w.lock()) {
2086 141 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
2087 141 : auto commit = sthis->pimpl_->repository_->commitMessage(message.toString());
2088 141 : lk.unlock();
2089 141 : if (onCommit)
2090 14 : onCommit(commit);
2091 141 : sthis->pimpl_->announce(commit, true);
2092 140 : if (cb)
2093 140 : cb(!commit.empty(), commit);
2094 283 : }
2095 140 : });
2096 : }
2097 :
2098 : bool
2099 14266 : Conversation::hasCommit(const std::string& commitId) const
2100 : {
2101 14266 : return pimpl_->repository_->hasCommit(commitId);
2102 : }
2103 :
2104 : std::optional<ConversationCommit>
2105 35 : Conversation::getCommit(const std::string& commitId) const
2106 : {
2107 35 : return pimpl_->repository_->getCommit(commitId);
2108 : }
2109 :
2110 : void
2111 2 : Conversation::loadMessages(const OnLoadMessages& cb, const LogOptions& options)
2112 : {
2113 2 : if (!cb)
2114 0 : return;
2115 2 : dht::ThreadPool::io().run([w = weak(), cb = std::move(cb), options] {
2116 2 : if (auto sthis = w.lock()) {
2117 2 : std::unique_lock lk(sthis->pimpl_->loadedHistory_.mutex);
2118 2 : auto result = sthis->pimpl_->loadMessages(options);
2119 2 : lk.unlock();
2120 2 : cb(std::move(result));
2121 4 : }
2122 2 : });
2123 : }
2124 :
2125 : void
2126 3 : Conversation::Impl::loadMissingBodyOverwrites()
2127 : {
2128 : #ifdef ENABLE_PLUGIN
2129 3 : auto& pluginChatManager = Manager::instance().getJamiPluginManager().getChatServicesManager();
2130 3 : if (!pluginChatManager.hasHandlers()) {
2131 2 : return;
2132 : }
2133 :
2134 1 : const auto& convId = repository_->id();
2135 :
2136 : // Collect all TEXT commits (originals and editions) missing a bodyOverwrite.
2137 1 : std::vector<libjami::SwarmMessage> batch;
2138 : {
2139 1 : std::lock_guard const lk(loadedHistory_.mutex);
2140 4 : for (const auto& msg : loadedHistory_.quickAccess | std::views::values) {
2141 3 : if (msg->type == CommitType::TEXT
2142 11 : && ((msg->editions.empty() && !msg->pluginData.contains("bodyOverwrite"))
2143 3 : || (!msg->editions.empty() && !msg->editions.back().contains("bodyOverwrite")))) {
2144 1 : batch.push_back(*msg);
2145 : }
2146 : }
2147 1 : }
2148 :
2149 1 : if (batch.empty()) {
2150 0 : return;
2151 : }
2152 :
2153 : // Single plugin call for all messages at once.
2154 1 : pluginChatManager.transformSwarmMessages(batch, accountId_, convId);
2155 :
2156 : // Write results back, deduplicate by signal-message id (multiple editions → same original).
2157 1 : std::map<std::string, libjami::SwarmMessage> signalMap;
2158 : {
2159 1 : std::lock_guard const lk(loadedHistory_.mutex);
2160 2 : for (const auto& transformed : batch) {
2161 2 : const auto boIt = transformed.pluginData.find("bodyOverwrite");
2162 1 : if (boIt == transformed.pluginData.end() || boIt->second.empty()) {
2163 1 : if (const auto it = loadedHistory_.quickAccess.find(transformed.id);
2164 1 : it != loadedHistory_.quickAccess.end()) {
2165 3 : it->second->pluginData["bodyOverwrite"] = "";
2166 : }
2167 1 : continue;
2168 1 : }
2169 0 : if (auto signalMsg = injectEditionOverwrites(transformed.id, boIt->second); !signalMsg.id.empty()) {
2170 0 : signalMap[signalMsg.id] = std::move(signalMsg);
2171 0 : }
2172 : }
2173 1 : }
2174 1 : for (const auto& msg : signalMap | std::views::values) {
2175 0 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, convId, msg);
2176 : }
2177 : #endif
2178 1 : }
2179 :
2180 : void
2181 3 : Conversation::loadMissingBodyOverwrites() const
2182 : {
2183 3 : pimpl_->loadMissingBodyOverwrites();
2184 3 : }
2185 :
2186 : void
2187 0 : Conversation::Impl::reloadBodyOverwriteMessages()
2188 : {
2189 : #ifdef ENABLE_PLUGIN
2190 : // Clear all bodyOverwrite* entries so loadMissingBodyOverwrites() reload all body overwrites.
2191 : {
2192 0 : std::lock_guard const lk(loadedHistory_.mutex);
2193 0 : for (const auto& msg : loadedHistory_.quickAccess | std::views::values) {
2194 0 : if (msg->editions.empty()) {
2195 0 : msg->pluginData.erase("bodyOverwrite");
2196 : } else {
2197 0 : msg->pluginData.erase("bodyOverwrite");
2198 0 : for (auto& edition : msg->editions) {
2199 0 : edition.erase("bodyOverwrite");
2200 : }
2201 : }
2202 : }
2203 0 : }
2204 0 : loadMissingBodyOverwrites();
2205 : #endif
2206 0 : }
2207 :
2208 : void
2209 0 : Conversation::reloadBodyOverwriteMessages() const
2210 : {
2211 0 : pimpl_->reloadBodyOverwriteMessages();
2212 0 : }
2213 :
2214 : void
2215 0 : Conversation::Impl::updateMessageBodyOverwrite(const std::string& messageId, const std::string_view bodyOverwrite)
2216 : {
2217 : #ifdef ENABLE_PLUGIN
2218 0 : const auto& convId = repository_->id();
2219 0 : libjami::SwarmMessage signalMsg;
2220 : {
2221 0 : std::lock_guard const lk(loadedHistory_.mutex);
2222 0 : signalMsg = injectEditionOverwrites(messageId, bodyOverwrite);
2223 0 : }
2224 0 : if (!signalMsg.id.empty()) {
2225 0 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, convId, signalMsg);
2226 : }
2227 : #endif
2228 0 : }
2229 :
2230 : void
2231 0 : Conversation::updateMessageBodyOverwrite(const std::string& messageId, const std::string& bodyOverwrite) const
2232 : {
2233 0 : pimpl_->updateMessageBodyOverwrite(messageId, bodyOverwrite);
2234 0 : }
2235 :
2236 : void
2237 1 : Conversation::Impl::clearBodyOverwrites()
2238 : {
2239 : #ifdef ENABLE_PLUGIN
2240 1 : const auto convId = repository_->id();
2241 1 : std::map<std::string, libjami::SwarmMessage> signalMap;
2242 : {
2243 1 : std::lock_guard const lk(loadedHistory_.mutex);
2244 : // Track which message ids need a signal: their own overwrite was cleared,
2245 : // or one of their edition commits' overwrite was cleared.
2246 4 : for (const auto& [id, msg] : loadedHistory_.quickAccess) {
2247 3 : if (msg->editions.empty()) {
2248 6 : msg->pluginData.erase("bodyOverwrite");
2249 9 : if (!msg->body.contains(CommitKey::EDIT)) {
2250 3 : signalMap[id] = *msg;
2251 : }
2252 : } else {
2253 0 : msg->pluginData.erase("bodyOverwrite");
2254 0 : for (auto& edition : msg->editions) {
2255 0 : edition.erase("bodyOverwrite");
2256 : }
2257 0 : signalMap[id] = *msg;
2258 : }
2259 : }
2260 1 : }
2261 4 : for (const auto& msg : signalMap | std::views::values) {
2262 3 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, convId, msg);
2263 : }
2264 : #endif
2265 1 : }
2266 :
2267 : void
2268 1 : Conversation::clearBodyOverwrites() const
2269 : {
2270 1 : pimpl_->clearBodyOverwrites();
2271 1 : }
2272 :
2273 : void
2274 0 : Conversation::clearCache()
2275 : {
2276 0 : std::lock_guard lk(pimpl_->loadedHistory_.mutex);
2277 0 : pimpl_->loadedHistory_.messageList.clear();
2278 0 : pimpl_->loadedHistory_.quickAccess.clear();
2279 0 : pimpl_->loadedHistory_.pendingEditions.clear();
2280 0 : pimpl_->loadedHistory_.pendingReactions.clear();
2281 : {
2282 0 : std::lock_guard lk(pimpl_->messageStatusMtx_);
2283 0 : pimpl_->memberToStatus.clear();
2284 0 : }
2285 0 : }
2286 :
2287 : std::string
2288 1978 : Conversation::lastCommitId() const
2289 : {
2290 : {
2291 1978 : std::lock_guard lk(pimpl_->loadedHistory_.mutex);
2292 1977 : if (!pimpl_->loadedHistory_.messageList.empty())
2293 1391 : return (*pimpl_->loadedHistory_.messageList.begin())->id;
2294 1978 : }
2295 586 : LogOptions options;
2296 586 : options.nbOfCommits = 1;
2297 586 : options.skipMerge = true;
2298 586 : History optHistory;
2299 586 : std::scoped_lock lock(pimpl_->writeMtx_, optHistory.mutex);
2300 586 : auto res = pimpl_->loadMessages(options, &optHistory);
2301 586 : if (res.empty())
2302 1 : return {};
2303 585 : return (*optHistory.messageList.begin())->id;
2304 586 : }
2305 :
2306 : bool
2307 1635 : Conversation::pull(const std::string& deviceId, OnPullCb&& cb, std::string commitId)
2308 : {
2309 1635 : std::lock_guard lk(pimpl_->pullcbsMtx_);
2310 4905 : auto [it, notInProgress] = pimpl_->fetchingRemotes_.emplace(deviceId,
2311 3270 : std::deque<std::pair<std::string, OnPullCb>>());
2312 1635 : auto& pullcbs = it->second;
2313 1635 : auto itPull = std::find_if(pullcbs.begin(), pullcbs.end(), [&](const auto& elem) {
2314 6 : return std::get<0>(elem) == commitId;
2315 3270 : });
2316 1634 : if (itPull != pullcbs.end()) {
2317 6 : JAMI_DEBUG("{} Ignoring request to pull from {:s} with commit {:s}: pull already in progress",
2318 : pimpl_->toString(),
2319 : deviceId,
2320 : commitId);
2321 6 : cb(false);
2322 6 : return false;
2323 : }
2324 1629 : JAMI_DEBUG("{} [device {}] Pulling '{:s}'", pimpl_->toString(), deviceId, commitId);
2325 1629 : pullcbs.emplace_back(std::move(commitId), std::move(cb));
2326 1629 : if (notInProgress)
2327 1621 : dht::ThreadPool::io().run([w = weak(), deviceId] {
2328 1621 : if (auto sthis_ = w.lock())
2329 1620 : sthis_->pimpl_->pull(deviceId);
2330 1621 : });
2331 1629 : return true;
2332 1635 : }
2333 :
2334 : void
2335 1620 : Conversation::Impl::pull(const std::string& deviceId)
2336 : {
2337 1620 : auto& repo = repository_;
2338 :
2339 1620 : std::string commitId;
2340 1621 : OnPullCb cb;
2341 : while (true) {
2342 : {
2343 3250 : std::lock_guard lk(pullcbsMtx_);
2344 3250 : auto it = fetchingRemotes_.find(deviceId);
2345 3250 : if (it == fetchingRemotes_.end()) {
2346 0 : JAMI_ERROR("Could not find device {:s} in fetchingRemotes", deviceId);
2347 0 : break;
2348 : }
2349 3250 : auto& pullcbs = it->second;
2350 3250 : if (pullcbs.empty()) {
2351 1621 : fetchingRemotes_.erase(it);
2352 1621 : break;
2353 : }
2354 1629 : auto& elem = pullcbs.front();
2355 1629 : commitId = std::move(std::get<0>(elem));
2356 1629 : cb = std::move(std::get<1>(elem));
2357 1629 : pullcbs.pop_front();
2358 3250 : }
2359 : // If recently fetched, the commit can already be there, so no need to do complex operations
2360 1629 : if (commitId != "" && repo->hasCommit(commitId)) {
2361 19 : cb(true);
2362 39 : continue;
2363 : }
2364 : // Pull from remote
2365 1610 : auto fetched = repo->fetch(deviceId);
2366 1610 : if (!fetched) {
2367 20 : cb(false);
2368 20 : continue;
2369 : }
2370 :
2371 1590 : auto oldHead = repo->getHead();
2372 :
2373 1590 : std::unique_lock lk(writeMtx_);
2374 : auto commits = repo->mergeHistory(deviceId,
2375 1594 : [this](const std::string& peerUri) { this->disconnectFromPeer(peerUri); });
2376 1590 : if (!commits.empty()) {
2377 904 : announce(commits);
2378 : }
2379 1590 : lk.unlock();
2380 :
2381 1590 : bool commitFound = false;
2382 1590 : if (commitId != "") {
2383 : // If `commitId` is non-empty, then we were attempting to pull a specific commit.
2384 : // We need to check if we actually got it; the fact that the fetch above was
2385 : // successful doesn't guarantee that we did.
2386 1059 : for (const auto& commit : commits) {
2387 1813 : if (commit.at("id") == commitId) {
2388 899 : commitFound = true;
2389 899 : break;
2390 : }
2391 : }
2392 : } else {
2393 538 : commitFound = true;
2394 : }
2395 1590 : if (!commitFound)
2396 153 : JAMI_WARNING("Successfully fetched from device {} but didn't receive expected commit {}",
2397 : deviceId,
2398 : commitId);
2399 : // WARNING: If its argument is `true`, this callback will attempt to send a message notification
2400 : // for commit `commitId` to other members of the swarm. It's important that we only
2401 : // send these notifications if we actually have the commit. Otherwise, we can end up
2402 : // in a situation where the members of the swarm keep sending notifications to each
2403 : // other for a commit that none of them have (note that we are unable to rule this out, as
2404 : // nothing prevents a malicious user from intentionally sending a notification with
2405 : // a fake commit ID).
2406 1590 : if (cb)
2407 1590 : cb(commitFound);
2408 :
2409 : // Announce if profile changed
2410 1590 : if (!commits.empty()) {
2411 1810 : auto diffStats = repo->diffStats("HEAD", oldHead);
2412 905 : auto changedFiles = repo->changedFiles(diffStats);
2413 905 : if (find(changedFiles.begin(), changedFiles.end(), "profile.vcf") != changedFiles.end()) {
2414 5 : emitSignal<libjami::ConversationSignal::ConversationProfileUpdated>(accountId_,
2415 5 : repo->id(),
2416 10 : repo->infos());
2417 : }
2418 905 : }
2419 3219 : }
2420 1621 : }
2421 :
2422 : void
2423 1635 : Conversation::sync(const std::string& member, const std::string& deviceId, OnPullCb&& cb, std::string commitId)
2424 : {
2425 1635 : pull(deviceId, std::move(cb), commitId);
2426 1635 : dht::ThreadPool::io().run([member, deviceId, w = weak_from_this()] {
2427 1635 : auto sthis = w.lock();
2428 : // For waiting request, downloadFile
2429 1636 : for (const auto& wr : sthis->dataTransfer()->waitingRequests()) {
2430 1 : sthis->downloadFile(wr.interactionId, wr.fileId, wr.path, member, deviceId);
2431 1633 : }
2432 1633 : });
2433 1635 : }
2434 :
2435 : std::map<std::string, std::string>
2436 270 : Conversation::generateInvitation() const
2437 : {
2438 : // Invite the new member to the conversation
2439 270 : Json::Value root;
2440 270 : auto& metadata = root[ConversationMapKeys::METADATAS];
2441 546 : for (const auto& [k, v] : infos()) {
2442 276 : if (v.size() >= 64000) {
2443 0 : JAMI_WARNING("Cutting invite because the SIP message will be too long");
2444 0 : continue;
2445 : }
2446 276 : metadata[k] = v;
2447 270 : }
2448 270 : root[ConversationMapKeys::CONVERSATIONID] = id();
2449 1080 : return {{"application/invite+json", json::toString(root)}};
2450 540 : }
2451 :
2452 : std::string
2453 7 : Conversation::leave()
2454 : {
2455 7 : setRemovingFlag();
2456 7 : std::lock_guard lk(pimpl_->writeMtx_);
2457 14 : return pimpl_->repository_->leave();
2458 7 : }
2459 :
2460 : void
2461 10 : Conversation::setRemovingFlag()
2462 : {
2463 10 : pimpl_->isRemoving_ = true;
2464 10 : }
2465 :
2466 : bool
2467 3722 : Conversation::isRemoving()
2468 : {
2469 3722 : return pimpl_->isRemoving_;
2470 : }
2471 :
2472 : void
2473 21 : Conversation::erase()
2474 : {
2475 21 : if (pimpl_->conversationDataPath_ != "")
2476 21 : dhtnet::fileutils::removeAll(pimpl_->conversationDataPath_, true);
2477 21 : if (!pimpl_->repository_)
2478 0 : return;
2479 21 : std::lock_guard lk(pimpl_->writeMtx_);
2480 21 : pimpl_->repository_->erase();
2481 21 : }
2482 :
2483 : ConversationMode
2484 1869 : Conversation::mode() const
2485 : {
2486 1869 : return pimpl_->repository_->mode();
2487 : }
2488 :
2489 : std::vector<std::string>
2490 30 : Conversation::getInitialMembers() const
2491 : {
2492 30 : return pimpl_->repository_->getInitialMembers();
2493 : }
2494 :
2495 : bool
2496 0 : Conversation::isInitialMember(const std::string& uri) const
2497 : {
2498 0 : auto members = getInitialMembers();
2499 0 : return std::find(members.begin(), members.end(), uri) != members.end();
2500 0 : }
2501 :
2502 : void
2503 9 : Conversation::updateInfos(const std::map<std::string, std::string>& map, const OnDoneCb& cb)
2504 : {
2505 9 : dht::ThreadPool::io().run([w = weak(), map = std::move(map), cb = std::move(cb)] {
2506 9 : if (auto sthis = w.lock()) {
2507 9 : auto& repo = sthis->pimpl_->repository_;
2508 9 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
2509 9 : auto commit = repo->updateInfos(map);
2510 9 : sthis->pimpl_->announce(commit, true);
2511 9 : lk.unlock();
2512 9 : if (cb)
2513 9 : cb(!commit.empty(), commit);
2514 9 : emitSignal<libjami::ConversationSignal::ConversationProfileUpdated>(sthis->pimpl_->accountId_,
2515 9 : repo->id(),
2516 18 : repo->infos());
2517 18 : }
2518 9 : });
2519 9 : }
2520 :
2521 : std::map<std::string, std::string>
2522 316 : Conversation::infos() const
2523 : {
2524 316 : return pimpl_->repository_->infos();
2525 : }
2526 :
2527 : void
2528 8 : Conversation::updatePreferences(const std::map<std::string, std::string>& map)
2529 : {
2530 8 : const auto& filePath = pimpl_->preferencesPath_;
2531 8 : auto prefs = map;
2532 8 : auto itLast = prefs.find(LAST_MODIFIED);
2533 8 : if (itLast != prefs.end()) {
2534 3 : std::error_code ec;
2535 3 : if (std::filesystem::is_regular_file(filePath, ec)) {
2536 1 : auto lastModified = fileutils::lastWriteTimeInSeconds(filePath);
2537 : try {
2538 1 : if (lastModified >= to_int<uint64_t>(itLast->second))
2539 0 : return;
2540 0 : } catch (...) {
2541 0 : return;
2542 0 : }
2543 : }
2544 3 : prefs.erase(itLast);
2545 : }
2546 :
2547 8 : std::ofstream file(filePath, std::ios::trunc | std::ios::binary);
2548 8 : msgpack::pack(file, prefs);
2549 8 : emitSignal<libjami::ConversationSignal::ConversationPreferencesUpdated>(pimpl_->accountId_, id(), std::move(prefs));
2550 8 : }
2551 :
2552 : std::map<std::string, std::string>
2553 81 : Conversation::preferences(bool includeLastModified) const
2554 : {
2555 : try {
2556 81 : std::map<std::string, std::string> preferences;
2557 81 : const auto& filePath = pimpl_->preferencesPath_;
2558 151 : auto file = fileutils::loadFile(filePath);
2559 11 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
2560 11 : oh.get().convert(preferences);
2561 11 : if (includeLastModified)
2562 24 : preferences[LAST_MODIFIED] = std::to_string(fileutils::lastWriteTimeInSeconds(filePath));
2563 11 : return preferences;
2564 151 : } catch (const std::exception& e) {
2565 70 : }
2566 70 : return {};
2567 : }
2568 :
2569 : std::vector<uint8_t>
2570 0 : Conversation::vCard() const
2571 : {
2572 : try {
2573 0 : return fileutils::loadFile(pimpl_->repoPath_ / "profile.vcf");
2574 0 : } catch (...) {
2575 0 : }
2576 0 : return {};
2577 : }
2578 :
2579 : std::shared_ptr<TransferManager>
2580 1751 : Conversation::dataTransfer() const
2581 : {
2582 1751 : return pimpl_->transferManager_;
2583 : }
2584 :
2585 : bool
2586 13 : Conversation::onFileChannelRequest(const std::string& member,
2587 : const std::string& fileId,
2588 : std::filesystem::path& path,
2589 : std::string& sha3sum) const
2590 : {
2591 13 : if (!isMember(member))
2592 0 : return false;
2593 :
2594 13 : auto sep = fileId.find('_');
2595 13 : if (sep == std::string::npos)
2596 0 : return false;
2597 :
2598 13 : auto interactionId = fileId.substr(0, sep);
2599 13 : auto commit = getCommit(interactionId);
2600 26 : if (commit == std::nullopt || commit->commitMsg.tid.empty() || commit->commitMsg.sha3sum.empty()
2601 26 : || commit->commitMsg.type != CommitType::DATA_TRANSFER) {
2602 0 : JAMI_WARNING("[Account {:s}] {} requested invalid file transfer commit {}",
2603 : pimpl_->accountId_,
2604 : member,
2605 : interactionId);
2606 0 : return false;
2607 : }
2608 :
2609 13 : path = dataTransfer()->path(fileId);
2610 13 : sha3sum = commit->commitMsg.sha3sum;
2611 :
2612 13 : return true;
2613 13 : }
2614 :
2615 : bool
2616 14 : Conversation::downloadFile(const std::string& interactionId,
2617 : const std::string& fileId,
2618 : const std::string& path,
2619 : const std::string&,
2620 : const std::string& deviceId)
2621 : {
2622 14 : auto commit = getCommit(interactionId);
2623 14 : if (commit == std::nullopt || commit->commitMsg.type != CommitType::DATA_TRANSFER) {
2624 0 : JAMI_ERROR("Commit doesn't exists or is not a file transfer {} (Conversation: {}) ", interactionId, id());
2625 0 : return false;
2626 : }
2627 14 : auto tid = commit->commitMsg.tid;
2628 14 : auto sha3sum = commit->commitMsg.sha3sum;
2629 14 : auto totalSize = commit->commitMsg.totalSize;
2630 :
2631 14 : if (tid.empty() || sha3sum.empty() || totalSize < 0) {
2632 0 : JAMI_ERROR("Invalid file transfer commit (missing tid, size or sha3)");
2633 0 : return false;
2634 : }
2635 :
2636 : // Be sure to not lock conversation
2637 14 : dht::ThreadPool().io().run([w = weak(), deviceId, fileId, interactionId, sha3sum, path, totalSize] {
2638 14 : if (auto shared = w.lock()) {
2639 14 : std::filesystem::path filePath(path);
2640 14 : if (filePath.empty()) {
2641 0 : filePath = shared->dataTransfer()->path(fileId);
2642 : }
2643 :
2644 14 : std::error_code ec;
2645 14 : if (std::filesystem::file_size(filePath, ec) == static_cast<size_t>(totalSize)) {
2646 1 : if (fileutils::sha3File(filePath) == sha3sum) {
2647 1 : JAMI_WARNING("Ignoring request to download existing file: {}", filePath);
2648 1 : return;
2649 : }
2650 : }
2651 :
2652 13 : std::filesystem::path tempFilePath(filePath);
2653 13 : tempFilePath += ".tmp";
2654 13 : auto start = std::filesystem::file_size(tempFilePath, ec);
2655 13 : if (ec || start == static_cast<decltype(start)>(-1)) {
2656 11 : start = 0;
2657 : }
2658 13 : size_t end = 0;
2659 :
2660 13 : auto acc = shared->pimpl_->account_.lock();
2661 13 : if (!acc)
2662 0 : return;
2663 13 : shared->dataTransfer()->waitForTransfer(fileId, interactionId, sha3sum, path, totalSize);
2664 13 : acc->askForFileChannel(shared->id(), deviceId, interactionId, fileId, start, end);
2665 28 : }
2666 : });
2667 14 : return true;
2668 14 : }
2669 :
2670 : void
2671 1112 : Conversation::hasFetched(const std::string& deviceId, const std::string& commitId)
2672 : {
2673 1112 : dht::ThreadPool::io().run([w = weak(), deviceId, commitId]() {
2674 1112 : auto sthis = w.lock();
2675 1112 : if (!sthis)
2676 0 : return;
2677 : // Update fetched for Uri
2678 1112 : auto uri = sthis->uriFromDevice(deviceId);
2679 1112 : if (uri.empty() || uri == sthis->pimpl_->userId_)
2680 43 : return;
2681 : // When a user fetches a commit, the message is sent for this person
2682 1069 : sthis->pimpl_->updateStatus(uri,
2683 : libjami::Account::MessageStates::SENT,
2684 1069 : commitId,
2685 2138 : std::to_string(std::time(nullptr)),
2686 : true);
2687 1155 : });
2688 1112 : }
2689 :
2690 : void
2691 1114 : Conversation::Impl::updateStatus(const std::string& uri,
2692 : libjami::Account::MessageStates st,
2693 : const std::string& commitId,
2694 : const std::string& ts,
2695 : bool emit)
2696 : {
2697 : // This method can be called if peer send us a status or if another device sync. Emit will be true if a peer
2698 : // send us a status and will emit to other connected devices.
2699 1114 : LogOptions options;
2700 1114 : std::map<std::string, std::map<std::string, std::string>> newStatus;
2701 : {
2702 : // Update internal structures.
2703 1114 : std::lock_guard lk(messageStatusMtx_);
2704 1114 : auto& status = messagesStatus_[uri];
2705 1114 : auto& oldStatus = status[st == libjami::Account::MessageStates::SENT ? "fetched" : "read"];
2706 1114 : if (oldStatus == commitId)
2707 2 : return; // Nothing to do
2708 1112 : options.to = oldStatus;
2709 1112 : options.from = commitId;
2710 1112 : oldStatus = commitId;
2711 1112 : status[st == libjami::Account::MessageStates::SENT ? "fetched_ts" : "read_ts"] = ts;
2712 1112 : saveStatus();
2713 1112 : if (emit)
2714 1083 : newStatus[uri].insert(status.begin(), status.end());
2715 1114 : }
2716 1112 : if (emit && messageStatusCb_) {
2717 1083 : messageStatusCb_(newStatus);
2718 : }
2719 : // Update messages status for all commit between the old and new one
2720 1112 : options.logIfNotFound = false;
2721 1112 : options.fastLog = true;
2722 1112 : History optHistory;
2723 1112 : std::unique_lock lk(optHistory.mutex); // Avoid to announce messages while updating status.
2724 1112 : auto res = loadMessages(options, &optHistory);
2725 1112 : std::unique_lock mlk(messageStatusMtx_);
2726 1112 : std::vector<std::pair<std::string, int32_t>> statusToUpdate;
2727 1112 : if (res.size() == 0) {
2728 : // In this case, commit is not received yet, so we cache it
2729 11 : futureStatus[commitId][uri] = static_cast<int32_t>(st);
2730 : }
2731 7307 : for (const auto& [cid, _] : optHistory.quickAccess) {
2732 6195 : auto message = loadedHistory_.quickAccess.find(cid);
2733 6195 : if (message != loadedHistory_.quickAccess.end()) {
2734 : // Update message and emit to client,
2735 2832 : if (static_cast<int32_t>(st) > message->second->status[uri]) {
2736 2828 : message->second->status[uri] = static_cast<int32_t>(st);
2737 2828 : statusToUpdate.emplace_back(cid, static_cast<int32_t>(st));
2738 : }
2739 : } else {
2740 : // In this case, commit is not loaded by client, so we cache it
2741 : // No need to emit to client, they will get a correct status on load.
2742 3363 : futureStatus[cid][uri] = static_cast<int32_t>(st);
2743 : }
2744 : }
2745 1112 : mlk.unlock();
2746 1112 : lk.unlock();
2747 3940 : for (const auto& [cid, status] : statusToUpdate)
2748 5655 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(accountId_,
2749 2828 : repository_->id(),
2750 : uri,
2751 : cid,
2752 : static_cast<int>(status));
2753 1116 : }
2754 :
2755 : bool
2756 18 : Conversation::setMessageDisplayed(const std::string& uri, const std::string& interactionId)
2757 : {
2758 18 : std::lock_guard lk(pimpl_->messageStatusMtx_);
2759 54 : if (pimpl_->messagesStatus_[uri]["read"] == interactionId)
2760 2 : return false; // Nothing to do
2761 16 : dht::ThreadPool::io().run([w = weak(), uri, interactionId]() {
2762 16 : auto sthis = w.lock();
2763 16 : if (!sthis)
2764 0 : return;
2765 16 : sthis->pimpl_->updateStatus(uri,
2766 : libjami::Account::MessageStates::DISPLAYED,
2767 16 : interactionId,
2768 32 : std::to_string(std::time(nullptr)),
2769 : true);
2770 16 : });
2771 16 : return true;
2772 18 : }
2773 :
2774 : std::map<std::string, std::map<std::string, std::string>>
2775 63 : Conversation::messageStatus() const
2776 : {
2777 63 : std::lock_guard lk(pimpl_->messageStatusMtx_);
2778 126 : return pimpl_->messagesStatus_;
2779 63 : }
2780 :
2781 : void
2782 49 : Conversation::updateMessageStatus(const std::map<std::string, std::map<std::string, std::string>>& messageStatus)
2783 : {
2784 49 : std::unique_lock lk(pimpl_->messageStatusMtx_);
2785 49 : std::vector<std::tuple<libjami::Account::MessageStates, std::string, std::string, std::string>> stVec;
2786 106 : for (const auto& [uri, status] : messageStatus) {
2787 57 : auto& oldMs = pimpl_->messagesStatus_[uri];
2788 363 : if (status.find("fetched_ts") != status.end() && status.at("fetched") != oldMs["fetched"]) {
2789 107 : if (oldMs["fetched_ts"].empty() || std::stol(oldMs["fetched_ts"]) <= std::stol(status.at("fetched_ts"))) {
2790 0 : stVec.emplace_back(libjami::Account::MessageStates::SENT,
2791 : uri,
2792 25 : status.at("fetched"),
2793 75 : status.at("fetched_ts"));
2794 : }
2795 : }
2796 199 : if (status.find("read_ts") != status.end() && status.at("read") != oldMs["read"]) {
2797 12 : if (oldMs["read_ts"].empty() || std::stol(oldMs["read_ts"]) <= std::stol(status.at("read_ts"))) {
2798 0 : stVec.emplace_back(libjami::Account::MessageStates::DISPLAYED,
2799 : uri,
2800 4 : status.at("read"),
2801 12 : status.at("read_ts"));
2802 : }
2803 : }
2804 : }
2805 49 : lk.unlock();
2806 :
2807 78 : for (const auto& [status, uri, commitId, ts] : stVec) {
2808 29 : pimpl_->updateStatus(uri, status, commitId, ts);
2809 : }
2810 49 : }
2811 :
2812 : void
2813 391 : Conversation::onMessageStatusChanged(
2814 : const std::function<void(const std::map<std::string, std::map<std::string, std::string>>&)>& cb)
2815 : {
2816 391 : std::unique_lock lk(pimpl_->messageStatusMtx_);
2817 391 : pimpl_->messageStatusCb_ = cb;
2818 391 : }
2819 :
2820 : #ifdef LIBJAMI_TEST
2821 : void
2822 514 : Conversation::onBootstrapStatus(const std::function<void(std::string, BootstrapStatus)>& cb)
2823 : {
2824 514 : std::lock_guard lock(pimpl_->bootstrapMtx_);
2825 514 : pimpl_->bootstrapCbTest_ = cb;
2826 514 : }
2827 :
2828 : std::vector<libjami::SwarmMessage>
2829 0 : Conversation::loadMessagesSync(const LogOptions& options)
2830 : {
2831 0 : std::lock_guard lk(pimpl_->loadedHistory_.mutex);
2832 0 : auto result = pimpl_->loadMessages(options);
2833 0 : return result;
2834 0 : }
2835 :
2836 : void
2837 0 : Conversation::announce(const std::vector<std::map<std::string, std::string>>& commits, bool commitFromSelf)
2838 : {
2839 0 : pimpl_->announce(commits, commitFromSelf);
2840 0 : }
2841 :
2842 : void
2843 0 : Conversation::announce(const std::string& commitId, bool commitFromSelf)
2844 : {
2845 0 : pimpl_->announce(commitId, commitFromSelf);
2846 0 : }
2847 : #endif
2848 :
2849 : void
2850 513 : Conversation::bootstrap(std::function<void()> onBootstrapped, const std::vector<DeviceId>& knownDevices)
2851 : {
2852 513 : std::lock_guard lock(pimpl_->bootstrapMtx_);
2853 513 : if (!pimpl_ || !pimpl_->repository_ || !pimpl_->swarmManager_)
2854 0 : return;
2855 : // Bootstrap the DRT from currently known devices. Since the per-device
2856 : // presence monitoring (monitorConnection/startTracking below), callers do
2857 : // not pass any device list here: candidates are injected as members get
2858 : // online, through addKnownDevices() with devices reported by the
2859 : // PresenceManager (i.e. announced on the DHT), and rotated on connection
2860 : // failure. The knownDevices parameter remains for callers/tests that
2861 : // already hold a list of live devices.
2862 : // If a connection succeeds, onConnectionChanged will be called with ok=true
2863 513 : pimpl_->bootstrapCb_ = std::move(onBootstrapped);
2864 513 : std::vector<DeviceId> devices = knownDevices;
2865 513 : JAMI_DEBUG("{} Bootstrap with {} device(s)", pimpl_->toString(), devices.size());
2866 :
2867 513 : if (!devices.empty()) {
2868 0 : pimpl_->swarmManager_->setKnownNodes(devices);
2869 : }
2870 :
2871 513 : pimpl_->monitorConnection(weak_from_this());
2872 :
2873 : // If is shutdown, the conversation was re-added, causing no new nodes to be connected, but just a classic
2874 : // connectivity change
2875 513 : if (pimpl_->swarmManager_->isShutdown()) {
2876 16 : pimpl_->swarmManager_->restart();
2877 16 : pimpl_->swarmManager_->maintainBuckets();
2878 : }
2879 513 : }
2880 :
2881 : void
2882 1568 : Conversation::addKnownDevices(const std::vector<DeviceId>& devices, const std::string& memberUri)
2883 : {
2884 1568 : if (devices.empty())
2885 880 : return;
2886 688 : if (!memberUri.empty()) {
2887 : // JAMI_WARNING("{} Adding {} known devices for member {}", pimpl_->toString(), devices.size(), memberUri);
2888 688 : std::lock_guard lk(pimpl_->trackedMembersMtx_);
2889 688 : auto it = pimpl_->trackedMembers_.find(memberUri);
2890 688 : if (it != pimpl_->trackedMembers_.end()) {
2891 681 : it->second.devices.insert(devices.begin(), devices.end());
2892 : }
2893 688 : } else {
2894 0 : JAMI_ERROR("{} Adding {} known devices without member URI", pimpl_->toString(), devices.size());
2895 : }
2896 688 : pimpl_->swarmManager_->setKnownNodes(devices);
2897 : }
2898 :
2899 : void
2900 568 : Conversation::connectNode(const DeviceId& deviceId)
2901 : {
2902 568 : pimpl_->swarmManager_->connectNode(deviceId);
2903 568 : }
2904 :
2905 : std::vector<std::string>
2906 18 : Conversation::commitsEndedCalls()
2907 : {
2908 18 : pimpl_->loadActiveCalls();
2909 18 : pimpl_->loadHostedCalls();
2910 18 : auto commits = pimpl_->commitsEndedCalls();
2911 18 : if (!commits.empty()) {
2912 : // Announce to client
2913 0 : dht::ThreadPool::io().run([w = weak(), commits] {
2914 0 : if (auto sthis = w.lock())
2915 0 : sthis->pimpl_->announce(commits, true);
2916 0 : });
2917 : }
2918 18 : return commits;
2919 0 : }
2920 :
2921 : void
2922 391 : Conversation::onMembersChanged(OnMembersChanged&& cb)
2923 : {
2924 391 : pimpl_->onMembersChanged_ = std::move(cb);
2925 391 : }
2926 :
2927 : void
2928 391 : Conversation::onNeedSocket(NeedSocketCb needSocket)
2929 : {
2930 782 : pimpl_->swarmManager_->needSocketCb_ = [needSocket = std::move(needSocket),
2931 : w = weak()](const std::string& deviceId, ChannelCb&& cb, bool noNewSocket) {
2932 1041 : if (auto sthis = w.lock()) {
2933 1041 : auto wrappedCb = [cb = std::move(cb), w, deviceId](const std::shared_ptr<dhtnet::ChannelSocket>& socket) {
2934 1039 : if (auto sthis = w.lock()) {
2935 1033 : if (!socket) {
2936 277 : if (auto acc = sthis->pimpl_->account_.lock()) {
2937 278 : auto cert = acc->certStore().getCertificate(deviceId);
2938 281 : if (cert && cert->issuer) {
2939 280 : sthis->pimpl_->onConnectionFailed(DeviceId(deviceId), cert->issuer->getId().toString());
2940 : } else {
2941 1 : JAMI_WARNING("{} Unable to get member URI from device ID {}",
2942 : sthis->pimpl_->toString(),
2943 : deviceId);
2944 : }
2945 281 : } else {
2946 0 : return false;
2947 281 : }
2948 : }
2949 1041 : }
2950 1041 : return cb(socket);
2951 1041 : };
2952 2082 : needSocket(sthis->id(), deviceId, std::move(wrappedCb), "application/im-gitmessage-id", noNewSocket);
2953 2082 : }
2954 1432 : };
2955 391 : }
2956 :
2957 : void
2958 1018 : Conversation::addSwarmChannel(std::shared_ptr<dhtnet::ChannelSocket> channel)
2959 : {
2960 1018 : auto deviceId = channel->deviceId();
2961 : // Transmit avatar if necessary
2962 : // We do this here, because at this point we know both sides are connected and in
2963 : // the same conversation
2964 : // addSwarmChannel is a bit more complex, but it should be the best moment to do this.
2965 1018 : auto cert = channel->peerCertificate();
2966 1018 : if (!cert || !cert->issuer)
2967 0 : return;
2968 1018 : auto member = cert->issuer->getId().toString();
2969 1018 : pimpl_->swarmManager_->addChannel(std::move(channel));
2970 1018 : dht::ThreadPool::io().run([member, deviceId, a = pimpl_->account_, w = weak_from_this()] {
2971 1018 : auto sthis = w.lock();
2972 1018 : if (auto account = a.lock()) {
2973 1018 : account->sendProfile(sthis->id(), member, deviceId.toString());
2974 1018 : }
2975 1018 : });
2976 1018 : }
2977 :
2978 : uint32_t
2979 4 : Conversation::countInteractions(const std::string& toId, const std::string& fromId, const std::string& authorUri) const
2980 : {
2981 4 : LogOptions options;
2982 4 : options.to = toId;
2983 4 : options.from = fromId;
2984 4 : options.authorUri = authorUri;
2985 4 : options.logIfNotFound = false;
2986 4 : options.fastLog = true;
2987 4 : History history;
2988 4 : std::lock_guard lk(history.mutex);
2989 4 : auto res = pimpl_->loadMessages(options, &history);
2990 8 : return res.size();
2991 4 : }
2992 :
2993 : void
2994 4 : Conversation::search(uint32_t req, const Filter& filter, const std::shared_ptr<std::atomic_int>& flag) const
2995 : {
2996 : // Because logging a conversation can take quite some time,
2997 : // do it asynchronously
2998 4 : dht::ThreadPool::io().run([w = weak(), req, filter, flag] {
2999 4 : if (auto sthis = w.lock()) {
3000 4 : History history;
3001 4 : std::vector<std::map<std::string, std::string>> commits {};
3002 : // std::regex_constants::ECMAScript is the default flag.
3003 4 : auto re = std::regex(filter.regexSearch,
3004 4 : filter.caseSensitive ? std::regex_constants::ECMAScript : std::regex_constants::icase);
3005 12 : sthis->pimpl_->repository_->log(
3006 8 : [&](const std::string& /*id*/, const GitAuthor& author, const GitCommit& commit) {
3007 20 : if (!filter.author.empty() && filter.author != sthis->uriFromDevice(author.email)) {
3008 : // Filter author
3009 0 : return CallbackResult::Skip;
3010 : }
3011 20 : auto commitTime = git_commit_time(commit.get());
3012 20 : if (filter.before && filter.before < commitTime) {
3013 : // Only get commits before this date
3014 0 : return CallbackResult::Skip;
3015 : }
3016 20 : if (filter.after && filter.after > commitTime) {
3017 : // Only get commits before this date
3018 0 : if (git_commit_parentcount(commit.get()) <= 1)
3019 0 : return CallbackResult::Break;
3020 : else
3021 0 : return CallbackResult::Skip; // Because we are sorting it with
3022 : // GIT_SORT_TOPOLOGICAL | GIT_SORT_TIME
3023 : }
3024 :
3025 20 : return CallbackResult::Ok; // Continue
3026 : },
3027 8 : [&](ConversationCommit&& cc) {
3028 20 : if (auto optMessage = sthis->pimpl_->repository_->convCommitToMap(cc))
3029 80 : sthis->pimpl_->addToHistory(history, {optMessage.value()}, false, false);
3030 40 : },
3031 8 : [&](const std::string& id, const GitAuthor&, ConversationCommit&) {
3032 20 : if (id == filter.lastId)
3033 0 : return true;
3034 20 : return false;
3035 : },
3036 : "",
3037 : false);
3038 : // Search on generated history
3039 24 : for (auto& message : history.messageList) {
3040 20 : auto contentType = message->type;
3041 20 : auto isSearchable = contentType == CommitType::TEXT || contentType == CommitType::DATA_TRANSFER;
3042 20 : if (filter.type.empty() && !isSearchable) {
3043 : // Not searchable, at least for now
3044 8 : continue;
3045 12 : } else if (contentType == filter.type || filter.type.empty()) {
3046 12 : if (isSearchable) {
3047 : // If it's a text match the body, else the display name
3048 60 : auto body = contentType == CommitType::TEXT ? message->body.at(CommitKey::BODY)
3049 36 : : message->body.at(CommitKey::DISPLAY_NAME);
3050 12 : std::smatch body_match;
3051 12 : if (std::regex_search(body, body_match, re)) {
3052 5 : auto commit = message->body;
3053 10 : commit["id"] = message->id;
3054 10 : commit[CommitKey::TYPE] = message->type;
3055 5 : commits.emplace_back(commit);
3056 5 : }
3057 12 : } else {
3058 : // Matching type, just add it to the results
3059 0 : commits.emplace_back(message->body);
3060 : }
3061 :
3062 12 : if (filter.maxResult != 0 && commits.size() == filter.maxResult)
3063 0 : break;
3064 : }
3065 20 : }
3066 :
3067 4 : if (commits.size() > 0)
3068 9 : emitSignal<libjami::ConversationSignal::MessagesFound>(req,
3069 3 : sthis->pimpl_->accountId_,
3070 6 : sthis->id(),
3071 3 : std::move(commits));
3072 : // If we're the latest thread, inform client that the search is finished
3073 4 : if ((*flag)-- == 1 /* decrement return the old value */) {
3074 8 : emitSignal<libjami::ConversationSignal::MessagesFound>(
3075 8 : req, sthis->pimpl_->accountId_, std::string {}, std::vector<std::map<std::string, std::string>> {});
3076 : }
3077 8 : }
3078 4 : });
3079 4 : }
3080 :
3081 : void
3082 14 : Conversation::hostConference(CommitMessage&& message, OnDoneCb&& cb)
3083 : {
3084 14 : if (message.confId.empty()) {
3085 0 : JAMI_ERROR("{}Malformed commit: no confId", pimpl_->toString());
3086 0 : return;
3087 : }
3088 :
3089 14 : auto now = std::chrono::system_clock::now();
3090 14 : auto nowSecs = std::chrono::duration_cast<std::chrono::seconds>(now.time_since_epoch()).count();
3091 : {
3092 14 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3093 14 : pimpl_->hostedCalls_[message.confId] = nowSecs;
3094 14 : pimpl_->saveHostedCalls();
3095 14 : }
3096 :
3097 14 : createCommit(std::move(message), {}, std::move(cb));
3098 : }
3099 :
3100 : bool
3101 20 : Conversation::isHosting(const std::string& confId) const
3102 : {
3103 20 : auto info = infos();
3104 64 : if (info["rdvDevice"] == pimpl_->deviceId_ && info["rdvHost"] == pimpl_->userId_)
3105 0 : return true; // We are the current device Host
3106 20 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3107 20 : return pimpl_->hostedCalls_.find(confId) != pimpl_->hostedCalls_.end();
3108 20 : }
3109 :
3110 : void
3111 10 : Conversation::removeActiveConference(CommitMessage&& message, OnDoneCb&& cb)
3112 : {
3113 10 : if (message.confId.empty()) {
3114 0 : JAMI_ERROR("{}Malformed commit: no confId", pimpl_->toString());
3115 0 : return;
3116 : }
3117 :
3118 10 : auto erased = false;
3119 : {
3120 10 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3121 10 : erased = pimpl_->hostedCalls_.erase(message.confId);
3122 10 : }
3123 10 : if (erased) {
3124 10 : pimpl_->saveHostedCalls();
3125 10 : createCommit(std::move(message), {}, std::move(cb));
3126 : } else
3127 0 : cb(false, "");
3128 : }
3129 :
3130 : std::vector<std::map<std::string, std::string>>
3131 38 : Conversation::currentCalls() const
3132 : {
3133 38 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3134 76 : return pimpl_->activeCalls_;
3135 38 : }
3136 : } // namespace jami
|
