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