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 "jamidht/collaborative_editing.h"
23 : #include "client/jami_signal.h"
24 : #include "swarm/swarm_manager.h"
25 : #include "conversationrepository.h"
26 : #include "timestamp.h"
27 :
28 : #ifdef ENABLE_PLUGIN
29 : #include "manager.h"
30 : #include "plugin/jamipluginmanager.h"
31 : #include "plugin/streamdata.h"
32 : #endif
33 : #include "jami/conversation_interface.h"
34 : #include "jami/configurationmanager_interface.h"
35 :
36 : #include "fileutils.h"
37 : #include "json_utils.h"
38 : #include "logger.h"
39 : #include "presence_manager.h"
40 : #include "string_utils.h"
41 :
42 : #include <opendht/thread_pool.h>
43 : #include <opendht/infohash.h>
44 : #include <fmt/compile.h>
45 : #include <asio/error_code.hpp>
46 :
47 : #include <algorithm>
48 : #include <memory>
49 : #include <charconv>
50 : #include <set>
51 : #include <string_view>
52 : #include <tuple>
53 : #include <optional>
54 : #include <utility>
55 : #include <deque>
56 : #include <chrono>
57 : #include <ctime>
58 : #include <functional>
59 : #include <atomic>
60 : #include <ranges>
61 : #include <regex>
62 :
63 : namespace jami {
64 :
65 : static const char* const LAST_MODIFIED = "lastModified";
66 :
67 : namespace {
68 :
69 : // Read a timestamp from JSON, preferring the millisecond key and falling
70 : // back to the legacy seconds key (written by older devices).
71 : TimePoint
72 758 : timePointFromJson(const Json::Value& json, const char* msKey, const char* secondsKey)
73 : {
74 758 : if (json.isMember(msKey))
75 362 : return timePointFromMilliseconds(json[msKey].asLargestInt());
76 396 : return timePointFromSeconds(json[secondsKey].asLargestInt());
77 : }
78 :
79 : // Resolve a timestamp from msgpack values, preferring milliseconds.
80 : TimePoint
81 1461 : resolveTimePoint(const std::optional<int64_t>& ms, int64_t seconds)
82 : {
83 1461 : return ms ? timePointFromMilliseconds(*ms) : timePointFromSeconds(seconds);
84 : }
85 :
86 : } // namespace
87 :
88 30 : ConvInfo::ConvInfo(const Json::Value& json)
89 : {
90 30 : id = json[ConversationMapKeys::ID].asString();
91 30 : created = timePointFromJson(json, ConversationMapKeys::CREATED_MS, ConversationMapKeys::CREATED);
92 30 : removed = timePointFromJson(json, ConversationMapKeys::REMOVED_MS, ConversationMapKeys::REMOVED);
93 30 : erased = timePointFromJson(json, ConversationMapKeys::ERASED_MS, ConversationMapKeys::ERASED);
94 81 : for (const auto& v : json[ConversationMapKeys::MEMBERS]) {
95 51 : members.emplace(v["uri"].asString());
96 : }
97 30 : lastDisplayed = json[ConversationMapKeys::LAST_DISPLAYED].asString();
98 30 : parent = json[ConversationMapKeys::PARENT].asString();
99 30 : if (json.isMember(ConversationMapKeys::MODE))
100 28 : mode = static_cast<ConversationMode>(json[ConversationMapKeys::MODE].asInt());
101 30 : if (json.isMember(ConversationMapKeys::INVITED)) {
102 3 : const auto& invitedJson = json[ConversationMapKeys::INVITED];
103 7 : for (const auto& uri : invitedJson.getMemberNames())
104 7 : invited[uri] = timePointFromMilliseconds(invitedJson[uri].asLargestInt());
105 : }
106 30 : }
107 :
108 : Json::Value
109 57 : ConvInfo::toJson() const
110 : {
111 57 : Json::Value json;
112 57 : json[ConversationMapKeys::ID] = id;
113 57 : json[ConversationMapKeys::CREATED] = Json::Int64(toSecondsSinceEpoch(created));
114 57 : json[ConversationMapKeys::CREATED_MS] = Json::Int64(toMillisecondsSinceEpoch(created));
115 57 : if (removed != TimePoint {}) {
116 2 : json[ConversationMapKeys::REMOVED] = Json::Int64(toSecondsSinceEpoch(removed));
117 2 : json[ConversationMapKeys::REMOVED_MS] = Json::Int64(toMillisecondsSinceEpoch(removed));
118 : }
119 57 : if (erased != TimePoint {}) {
120 1 : json[ConversationMapKeys::ERASED] = Json::Int64(toSecondsSinceEpoch(erased));
121 1 : json[ConversationMapKeys::ERASED_MS] = Json::Int64(toMillisecondsSinceEpoch(erased));
122 : }
123 159 : for (const auto& m : members) {
124 102 : Json::Value member;
125 102 : member["uri"] = m;
126 102 : json[ConversationMapKeys::MEMBERS].append(member);
127 102 : }
128 57 : json[ConversationMapKeys::LAST_DISPLAYED] = lastDisplayed;
129 57 : json[ConversationMapKeys::MODE] = static_cast<int>(mode);
130 57 : if (!parent.empty())
131 1 : json[ConversationMapKeys::PARENT] = parent;
132 57 : if (!invited.empty()) {
133 7 : Json::Value invitedJson;
134 16 : for (const auto& [uri, t] : invited)
135 9 : invitedJson[uri] = Json::Int64(toMillisecondsSinceEpoch(t));
136 7 : json[ConversationMapKeys::INVITED] = std::move(invitedJson);
137 7 : }
138 57 : return json;
139 0 : }
140 :
141 : void
142 471 : ConvInfo::msgpack_unpack(const msgpack::object& o)
143 : {
144 471 : if (o.type != msgpack::type::MAP)
145 0 : throw msgpack::type_error();
146 471 : int64_t createdSec = 0, removedSec = 0, erasedSec = 0;
147 471 : std::optional<int64_t> createdMs, removedMs, erasedMs;
148 471 : std::map<std::string, int64_t> invitedMs;
149 6118 : for (uint32_t i = 0; i < o.via.map.size; ++i) {
150 5647 : const auto& kv = o.via.map.ptr[i];
151 5647 : if (kv.key.type != msgpack::type::STR)
152 0 : continue;
153 5647 : std::string_view key(kv.key.via.str.ptr, kv.key.via.str.size);
154 5647 : if (key == ConversationMapKeys::ID)
155 471 : kv.val.convert(id);
156 5176 : else if (key == ConversationMapKeys::CREATED)
157 471 : kv.val.convert(createdSec);
158 4705 : else if (key == ConversationMapKeys::REMOVED)
159 471 : kv.val.convert(removedSec);
160 4234 : else if (key == ConversationMapKeys::ERASED)
161 471 : kv.val.convert(erasedSec);
162 3763 : else if (key == ConversationMapKeys::CREATED_MS)
163 470 : createdMs = kv.val.as<int64_t>();
164 3293 : else if (key == ConversationMapKeys::REMOVED_MS)
165 470 : removedMs = kv.val.as<int64_t>();
166 2823 : else if (key == ConversationMapKeys::ERASED_MS)
167 470 : erasedMs = kv.val.as<int64_t>();
168 2353 : else if (key == ConversationMapKeys::MEMBERS)
169 471 : kv.val.convert(members);
170 1882 : else if (key == ConversationMapKeys::LAST_DISPLAYED)
171 471 : kv.val.convert(lastDisplayed);
172 1411 : else if (key == ConversationMapKeys::MODE)
173 471 : kv.val.convert(mode);
174 940 : else if (key == ConversationMapKeys::PARENT)
175 470 : kv.val.convert(parent);
176 470 : else if (key == ConversationMapKeys::INVITED)
177 470 : kv.val.convert(invitedMs);
178 : }
179 471 : created = resolveTimePoint(createdMs, createdSec);
180 471 : removed = resolveTimePoint(removedMs, removedSec);
181 471 : erased = resolveTimePoint(erasedMs, erasedSec);
182 471 : invited.clear();
183 509 : for (const auto& [uri, ms] : invitedMs)
184 38 : invited[uri] = timePointFromMilliseconds(ms);
185 471 : }
186 :
187 : void
188 1 : ConvInfo::msgpack_object(msgpack::object* o, msgpack::zone& z) const
189 : {
190 1 : int64_t createdSec = toSecondsSinceEpoch(created);
191 1 : int64_t removedSec = toSecondsSinceEpoch(removed);
192 1 : int64_t erasedSec = toSecondsSinceEpoch(erased);
193 1 : int64_t createdMs = toMillisecondsSinceEpoch(created);
194 1 : int64_t removedMs = toMillisecondsSinceEpoch(removed);
195 1 : int64_t erasedMs = toMillisecondsSinceEpoch(erased);
196 1 : std::map<std::string, int64_t> invitedMs;
197 1 : for (const auto& [uri, t] : invited)
198 0 : invitedMs[uri] = toMillisecondsSinceEpoch(t);
199 0 : msgpack::type::make_define_map(ConversationMapKeys::ID,
200 1 : id,
201 : ConversationMapKeys::CREATED,
202 : createdSec,
203 : ConversationMapKeys::REMOVED,
204 : removedSec,
205 : ConversationMapKeys::ERASED,
206 : erasedSec,
207 : ConversationMapKeys::MEMBERS,
208 1 : members,
209 : ConversationMapKeys::LAST_DISPLAYED,
210 1 : lastDisplayed,
211 : ConversationMapKeys::MODE,
212 1 : mode,
213 : ConversationMapKeys::PARENT,
214 1 : parent,
215 : ConversationMapKeys::CREATED_MS,
216 : createdMs,
217 : ConversationMapKeys::REMOVED_MS,
218 : removedMs,
219 : ConversationMapKeys::ERASED_MS,
220 : erasedMs,
221 : ConversationMapKeys::INVITED,
222 : invitedMs)
223 1 : .msgpack_object(o, z);
224 1 : }
225 :
226 : // ConversationRequest
227 334 : ConversationRequest::ConversationRequest(const Json::Value& json)
228 : {
229 334 : received = timePointFromJson(json, ConversationMapKeys::RECEIVED_MS, ConversationMapKeys::RECEIVED);
230 334 : declined = timePointFromJson(json, ConversationMapKeys::DECLINED_MS, ConversationMapKeys::DECLINED);
231 334 : from = json[ConversationMapKeys::FROM].asString();
232 334 : conversationId = json[ConversationMapKeys::CONVERSATIONID].asString();
233 334 : auto& md = json[ConversationMapKeys::METADATAS];
234 671 : for (const auto& member : md.getMemberNames()) {
235 337 : metadatas.emplace(member, md[member].asString());
236 334 : }
237 334 : }
238 :
239 : Json::Value
240 8 : ConversationRequest::toJson() const
241 : {
242 8 : Json::Value json;
243 8 : json[ConversationMapKeys::CONVERSATIONID] = conversationId;
244 8 : json[ConversationMapKeys::FROM] = from;
245 8 : json[ConversationMapKeys::RECEIVED] = Json::Int64(toSecondsSinceEpoch(received));
246 8 : json[ConversationMapKeys::RECEIVED_MS] = Json::Int64(toMillisecondsSinceEpoch(received));
247 8 : if (declined != TimePoint {}) {
248 0 : json[ConversationMapKeys::DECLINED] = Json::Int64(toSecondsSinceEpoch(declined));
249 0 : json[ConversationMapKeys::DECLINED_MS] = Json::Int64(toMillisecondsSinceEpoch(declined));
250 : }
251 16 : for (const auto& [key, value] : metadatas) {
252 8 : json[ConversationMapKeys::METADATAS][key] = value;
253 : }
254 8 : return json;
255 0 : }
256 :
257 : std::map<std::string, std::string>
258 513 : ConversationRequest::toMap() const
259 : {
260 513 : auto result = metadatas;
261 1026 : result[ConversationMapKeys::ID] = conversationId;
262 1026 : result[ConversationMapKeys::FROM] = from;
263 513 : if (declined != TimePoint {})
264 3 : result[ConversationMapKeys::DECLINED] = std::to_string(toSecondsSinceEpoch(declined));
265 1539 : result[ConversationMapKeys::RECEIVED] = std::to_string(toSecondsSinceEpoch(received));
266 513 : return result;
267 0 : }
268 :
269 : void
270 24 : ConversationRequest::msgpack_unpack(const msgpack::object& o)
271 : {
272 24 : if (o.type != msgpack::type::MAP)
273 0 : throw msgpack::type_error();
274 24 : int64_t receivedSec = 0, declinedSec = 0;
275 24 : std::optional<int64_t> receivedMs, declinedMs;
276 190 : for (uint32_t i = 0; i < o.via.map.size; ++i) {
277 166 : const auto& kv = o.via.map.ptr[i];
278 166 : if (kv.key.type != msgpack::type::STR)
279 0 : continue;
280 166 : std::string_view key(kv.key.via.str.ptr, kv.key.via.str.size);
281 166 : if (key == ConversationMapKeys::FROM)
282 24 : kv.val.convert(from);
283 142 : else if (key == ConversationMapKeys::CONVERSATIONID)
284 24 : kv.val.convert(conversationId);
285 118 : else if (key == ConversationMapKeys::METADATAS)
286 24 : kv.val.convert(metadatas);
287 94 : else if (key == ConversationMapKeys::RECEIVED)
288 24 : kv.val.convert(receivedSec);
289 70 : else if (key == ConversationMapKeys::DECLINED)
290 24 : kv.val.convert(declinedSec);
291 46 : else if (key == ConversationMapKeys::RECEIVED_MS)
292 23 : receivedMs = kv.val.as<int64_t>();
293 23 : else if (key == ConversationMapKeys::DECLINED_MS)
294 23 : declinedMs = kv.val.as<int64_t>();
295 : }
296 24 : received = resolveTimePoint(receivedMs, receivedSec);
297 24 : declined = resolveTimePoint(declinedMs, declinedSec);
298 24 : }
299 :
300 : void
301 0 : ConversationRequest::msgpack_object(msgpack::object* o, msgpack::zone& z) const
302 : {
303 0 : int64_t receivedSec = toSecondsSinceEpoch(received);
304 0 : int64_t declinedSec = toSecondsSinceEpoch(declined);
305 0 : int64_t receivedMs = toMillisecondsSinceEpoch(received);
306 0 : int64_t declinedMs = toMillisecondsSinceEpoch(declined);
307 0 : msgpack::type::make_define_map(ConversationMapKeys::FROM,
308 0 : from,
309 : ConversationMapKeys::CONVERSATIONID,
310 0 : conversationId,
311 : ConversationMapKeys::METADATAS,
312 0 : metadatas,
313 : ConversationMapKeys::RECEIVED,
314 : receivedSec,
315 : ConversationMapKeys::DECLINED,
316 : declinedSec,
317 : ConversationMapKeys::RECEIVED_MS,
318 : receivedMs,
319 : ConversationMapKeys::DECLINED_MS,
320 : declinedMs)
321 0 : .msgpack_object(o, z);
322 0 : }
323 :
324 : using MessageList = std::list<std::shared_ptr<libjami::SwarmMessage>>;
325 :
326 : struct History
327 : {
328 : // While loading the history, we need to avoid:
329 : // - reloading history (can just be ignored)
330 : // - adding new commits (should wait for history to be loaded)
331 : std::mutex mutex {};
332 : std::condition_variable cv {};
333 : bool loading {false};
334 : MessageList messageList {};
335 : std::map<std::string, std::shared_ptr<libjami::SwarmMessage>> quickAccess {};
336 : std::map<std::string, std::list<std::shared_ptr<libjami::SwarmMessage>>> pendingEditions {};
337 : std::map<std::string, std::list<std::map<std::string, std::string>>> pendingReactions {};
338 : };
339 :
340 : class Conversation::Impl
341 : {
342 : private:
343 : #ifdef LIBJAMI_TEST
344 : // Allow the test-only Conversation constructor to forward to the private Impl constructor.
345 : friend class Conversation;
346 : #endif
347 935 : Impl(std::unique_ptr<ConversationRepository>&& repository,
348 : const std::shared_ptr<JamiAccount>& account,
349 : std::vector<ConversationCommit>&& commits = {})
350 935 : : repository_(repository ? std::move(repository) : throw std::logic_error("Invalid repository"))
351 907 : , account_(account)
352 907 : , accountId_(account->getAccountID())
353 907 : , userId_(account->getUsername())
354 1814 : , deviceId_(account->currentDeviceId())
355 907 : , swarmManager_(std::make_shared<SwarmManager>(
356 907 : NodeId(deviceId_),
357 907 : account->isMobile(),
358 907 : Manager::instance().getSeededRandomEngine(),
359 907 : [account = account_](const DeviceId& deviceId) {
360 613 : if (auto acc = account.lock()) {
361 613 : return acc->isConnectedWith(deviceId);
362 613 : }
363 0 : return false;
364 : },
365 907 : repository_->id(),
366 1814 : [account = account_, conversationId = repository_->id(), deviceId = NodeId(deviceId_)]()
367 : -> std::optional<MobileNodeInfo> {
368 0 : auto acc = account.lock();
369 0 : if (!acc || !acc->isMobile())
370 0 : return std::nullopt;
371 0 : const auto& identity = acc->identity();
372 0 : if (!identity.first || !identity.second || !identity.second->issuer)
373 0 : return std::nullopt;
374 :
375 0 : const auto now = std::chrono::system_clock::now();
376 0 : const auto certificateExpiry = identity.second->getExpiration();
377 0 : const auto maximumExpiry = now + MAX_MOBILE_LEASE_DURATION;
378 0 : const auto expiry = std::min(certificateExpiry, maximumExpiry);
379 0 : if (expiry <= now)
380 0 : return std::nullopt;
381 :
382 0 : MobileLease lease {1,
383 0 : conversationId,
384 0 : identity.second->issuer->getId(),
385 : deviceId,
386 0 : toSecondsSinceEpoch(now),
387 0 : toSecondsSinceEpoch(expiry),
388 0 : {}};
389 0 : lease.signature = identity.first->sign(mobileLeasePayload(lease));
390 0 : return MobileNodeInfo {deviceId, std::move(lease)};
391 0 : },
392 0 : [this](const dht::InfoHash& issuerId) {
393 : const auto members
394 0 : = repository_->memberUris("", {MemberRole::INVITED, MemberRole::BANNED, MemberRole::LEFT});
395 0 : return members.contains(issuerId.toString());
396 0 : },
397 0 : [account = account_](const NodeId& deviceId) -> std::shared_ptr<dht::crypto::Certificate> {
398 0 : if (auto acc = account.lock())
399 0 : return acc->certStore().getCertificate(deviceId.toString());
400 0 : return {};
401 : },
402 0 : [account = account_](const NodeId& deviceId,
403 : std::function<void(const std::shared_ptr<dht::crypto::Certificate>&)>&& cb) {
404 0 : if (auto acc = account.lock())
405 0 : acc->findCertificate(deviceId, std::move(cb));
406 : else
407 0 : cb(nullptr);
408 0 : }))
409 907 : , transferManager_(std::make_shared<TransferManager>(accountId_,
410 : "",
411 907 : repository_->id(),
412 907 : Manager::instance().getSeededRandomEngine()))
413 907 : , repoPath_(fileutils::get_data_dir() / accountId_ / "conversations" / repository_->id())
414 907 : , conversationDataPath_(fileutils::get_data_dir() / accountId_ / "conversation_data" / repository_->id())
415 907 : , fetchedPath_(conversationDataPath_ / ConversationDirectories::FETCHED)
416 907 : , sendingPath_(conversationDataPath_ / ConversationDirectories::SENDING)
417 907 : , preferencesPath_(conversationDataPath_ / ConversationDirectories::PREFERENCES)
418 907 : , statusPath_(conversationDataPath_ / ConversationDirectories::STATUS)
419 907 : , mobileNodesPath_(conversationDataPath_ / ConversationDirectories::MOBILE_NODES)
420 907 : , hostedCallsPath_(conversationDataPath_ / ConversationDirectories::HOSTED_CALLS)
421 907 : , activeCallsPath_(conversationDataPath_ / ConversationDirectories::ACTIVE_CALLS)
422 907 : , ioContext_(Manager::instance().ioContext())
423 5470 : , typers_(std::make_shared<Typers>(account, repository_->id()))
424 : {
425 907 : if (!commits.empty())
426 427 : initActiveCalls(repository_->convCommitsToMap(commits));
427 907 : loadStatus();
428 : // Restore mobility knowledge: in a swarm made of mostly mobile
429 : // devices, gossip alone is unavailable after a restart (nobody is
430 : // connected), so the persisted list is the only way to know which
431 : // devices need a wake-up notification.
432 907 : loadMobileNodes();
433 : // Self-contained callback: the swarm manager can outlive this Impl
434 : // (its timers hold shared_from_this), so do not capture `this`.
435 1814 : swarmManager_->onMobileNodeInfosChanged(
436 1814 : [path = mobileNodesPath_](const std::vector<MobileNodeInfo>& mobileNodes) {
437 0 : std::lock_guard lk {dhtnet::fileutils::getFileLock(path)};
438 0 : std::ofstream file(path, std::ios::trunc | std::ios::binary);
439 0 : msgpack::pack(file, mobileNodes);
440 0 : });
441 907 : setupMemberCallback();
442 991 : }
443 :
444 455 : Impl(std::pair<std::unique_ptr<ConversationRepository>, std::vector<ConversationCommit>>&& repoAndCommits,
445 : const std::shared_ptr<JamiAccount>& account)
446 455 : : Impl(std::move(repoAndCommits.first), account, std::move(repoAndCommits.second))
447 427 : {}
448 :
449 : public:
450 430 : Impl(const std::shared_ptr<JamiAccount>& account, ConversationMode mode, const std::string& otherMember = "")
451 430 : : Impl(ConversationRepository::createConversation(account, mode, otherMember), account)
452 430 : {}
453 :
454 50 : Impl(const std::shared_ptr<JamiAccount>& account, const std::string& conversationId)
455 50 : : Impl(std::make_unique<ConversationRepository>(account, conversationId), account)
456 50 : {}
457 :
458 458 : Impl(const std::shared_ptr<JamiAccount>& account, const std::string& remoteDevice, const std::string& conversationId)
459 458 : : Impl(ConversationRepository::cloneConversation(account, remoteDevice, conversationId), account)
460 427 : {}
461 :
462 3235 : std::string toString() const
463 : {
464 12940 : return fmt::format(FMT_COMPILE("[Account {}] [Conversation {}]"), accountId_, repository_->id());
465 : }
466 :
467 0 : std::vector<std::map<std::string, std::string>> getConnectivity() const
468 : {
469 0 : return swarmManager_->getRoutingTableInfo();
470 : }
471 :
472 : mutable std::string fmtStr_;
473 :
474 907 : ~Impl() { stopTracking(); }
475 :
476 : /**
477 : * If, for whatever reason, the daemon is stopped while hosting a conference,
478 : * we need to announce the end of this call when restarting.
479 : * To avoid to keep active calls forever.
480 : */
481 : std::vector<std::string> commitsEndedCalls();
482 : bool isAdmin() const;
483 :
484 416 : void announce(const std::string& commitId, bool commitFromSelf = false)
485 : {
486 416 : std::vector<std::string> vec;
487 416 : if (!commitId.empty())
488 412 : vec.emplace_back(commitId);
489 416 : announce(vec, commitFromSelf);
490 416 : }
491 :
492 432 : void announce(const std::vector<std::string>& commits, bool commitFromSelf = false)
493 : {
494 432 : std::vector<ConversationCommit> convcommits;
495 432 : convcommits.reserve(commits.size());
496 876 : for (const auto& cid : commits) {
497 444 : if (auto commit = repository_->getCommit(cid)) {
498 444 : convcommits.emplace_back(*commit);
499 444 : }
500 : }
501 432 : announce(repository_->convCommitsToMap(convcommits), commitFromSelf);
502 432 : }
503 :
504 : /**
505 : * Initialize activeCalls_ from the list of commits in the repository
506 : * @param commits Commits in reverse chronological order (i.e. from newest to oldest)
507 : */
508 427 : void initActiveCalls(const std::vector<std::map<std::string, std::string>>& commits) const
509 : {
510 427 : std::unordered_set<std::string> invalidHostUris;
511 427 : std::unordered_set<std::string> invalidCallIds;
512 :
513 427 : std::lock_guard lk(activeCallsMtx_);
514 1773 : for (const auto& commit : commits) {
515 2692 : if (commit.at(CommitKey::TYPE) == CommitType::MEMBER) {
516 : // Each commit of type MEMBER has an "action" field whose value can be one
517 : // of the following: "add", "join", "remove", "ban", "unban"
518 : // In the case of "remove" and "ban", we need to add the member's URI to
519 : // invalidHostUris to ensure that any call they may have started in the past
520 : // is no longer considered active.
521 : // For the other actions, there's no harm in adding the member's URI anyway,
522 : // since it's not possible to start hosting a call before joining the swarm (or
523 : // before getting unbanned in the case of previously banned members).
524 1716 : invalidHostUris.emplace(commit.at(CommitKey::URI));
525 1467 : } else if (commit.find(CommitKey::CONF_ID) != commit.end() && commit.find(CommitKey::URI) != commit.end()
526 979 : && commit.find(CommitKey::DEVICE) != commit.end()) {
527 : // The commit indicates either the end or the beginning of a call
528 : // (depending on whether there is a "duration" field or not).
529 1 : auto convId = repository_->id();
530 2 : auto confId = commit.at(CommitKey::CONF_ID);
531 2 : auto uri = commit.at(CommitKey::URI);
532 1 : auto device = commit.at(CommitKey::DEVICE);
533 :
534 3 : if (commit.find(CommitKey::DURATION) == commit.end()
535 1 : && invalidCallIds.find(confId) == invalidCallIds.end()
536 3 : && invalidHostUris.find(uri) == invalidHostUris.end()) {
537 1 : std::map<std::string, std::string> activeCall;
538 2 : activeCall["id"] = confId;
539 2 : activeCall["uri"] = uri;
540 1 : activeCall["device"] = device;
541 1 : activeCalls_.emplace_back(activeCall);
542 1 : JAMI_LOG("swarm:{} new active call detected: {} (on device {}, account {})",
543 : convId,
544 : confId,
545 : device,
546 : uri);
547 1 : }
548 : // Even if the call was active, we still add its ID to invalidCallIds to make sure it
549 : // doesn't get added a second time. (This shouldn't happen normally, but in practice
550 : // there are sometimes multiple commits indicating the beginning of the same call.)
551 1 : invalidCallIds.emplace(confId);
552 1 : }
553 : }
554 427 : saveActiveCalls();
555 427 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_, repository_->id(), activeCalls_);
556 427 : }
557 :
558 : /**
559 : * Update activeCalls_ via announced commits (in load or via new commits)
560 : * @param commit Commit to check
561 : * @param eraseOnly If we want to ignore added commits
562 : * @param emitSig If we want to emit to client
563 : * @note eraseOnly is used by loadMessages. This is a fail-safe, this SHOULD NOT happen
564 : */
565 81 : void updateActiveCalls(const std::map<std::string, std::string>& commit,
566 : bool eraseOnly = false,
567 : bool emitSig = true) const
568 : {
569 81 : if (!repository_)
570 0 : return;
571 162 : if (commit.at(CommitKey::TYPE) == CommitType::MEMBER) {
572 : // In this case, we need to check if we are not removing a hosting member or device
573 21 : std::lock_guard lk(activeCallsMtx_);
574 21 : auto it = activeCalls_.begin();
575 21 : auto updateActives = false;
576 22 : while (it != activeCalls_.end()) {
577 5 : if (it->at("uri") == commit.at(CommitKey::URI) || it->at("device") == commit.at(CommitKey::URI)) {
578 5 : JAMI_DEBUG("Removing {:s} from the active calls, because {:s} left",
579 : it->at("id"),
580 : commit.at(CommitKey::URI));
581 1 : it = activeCalls_.erase(it);
582 1 : updateActives = true;
583 : } else {
584 0 : ++it;
585 : }
586 : }
587 21 : if (updateActives) {
588 1 : saveActiveCalls();
589 1 : if (emitSig)
590 1 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_,
591 1 : repository_->id(),
592 1 : activeCalls_);
593 : }
594 21 : return;
595 21 : }
596 : // Else, it's a call information
597 339 : if (commit.find(CommitKey::CONF_ID) != commit.end() && commit.find(CommitKey::URI) != commit.end()
598 279 : && commit.find(CommitKey::DEVICE) != commit.end()) {
599 53 : auto convId = repository_->id();
600 106 : auto confId = commit.at(CommitKey::CONF_ID);
601 106 : auto uri = commit.at(CommitKey::URI);
602 53 : auto device = commit.at(CommitKey::DEVICE);
603 53 : std::lock_guard lk(activeCallsMtx_);
604 53 : auto itActive = std::find_if(activeCalls_.begin(), activeCalls_.end(), [&](const auto& value) {
605 154 : return value.at("id") == confId && value.at("uri") == uri && value.at("device") == device;
606 : });
607 159 : if (commit.find(CommitKey::DURATION) == commit.end()) {
608 31 : if (itActive == activeCalls_.end() && !eraseOnly) {
609 31 : JAMI_DEBUG("swarm:{:s} new current call detected: {:s} on device {:s}, account {:s}",
610 : convId,
611 : confId,
612 : device,
613 : uri);
614 155 : activeCalls_.emplace_back(std::map<std::string, std::string> {
615 : {"id", confId},
616 : {"uri", uri},
617 : {"device", device},
618 124 : });
619 31 : saveActiveCalls();
620 31 : if (emitSig)
621 31 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_,
622 31 : repository_->id(),
623 31 : activeCalls_);
624 : }
625 : } else {
626 22 : if (itActive != activeCalls_.end()) {
627 22 : itActive = activeCalls_.erase(itActive);
628 : // Unlikely, but we must ensure that no duplicate exists
629 44 : while (itActive != activeCalls_.end()) {
630 0 : itActive = std::find_if(itActive, activeCalls_.end(), [&](const auto& value) {
631 0 : return value.at("id") == confId && value.at("uri") == uri && value.at("device") == device;
632 : });
633 0 : if (itActive != activeCalls_.end()) {
634 0 : JAMI_ERROR("Duplicate call found. (This is a bug)");
635 0 : itActive = activeCalls_.erase(itActive);
636 : }
637 : }
638 :
639 22 : if (eraseOnly) {
640 0 : JAMI_WARNING("previous swarm:{:s} call finished detected: {:s} on device "
641 : "{:s}, account {:s}",
642 : convId,
643 : confId,
644 : device,
645 : uri);
646 : } else {
647 22 : JAMI_DEBUG("swarm:{:s} call finished: {:s} on device {:s}, account {:s}",
648 : convId,
649 : confId,
650 : device,
651 : uri);
652 : }
653 : }
654 22 : saveActiveCalls();
655 22 : if (emitSig)
656 22 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_,
657 22 : repository_->id(),
658 22 : activeCalls_);
659 : }
660 53 : }
661 31 : }
662 :
663 1596 : void announce(const std::vector<std::map<std::string, std::string>>& commits, bool commitFromSelf = false)
664 : {
665 1596 : if (!repository_)
666 0 : return;
667 1596 : auto convId = repository_->id();
668 1596 : if (repository_->mode() == ConversationMode::DOCUMENT) {
669 : // A document's commits are not messages: nothing here goes to the
670 : // clients' conversation views or to the plugins. What a merge brought
671 : // in -- checkpoints, attachments, a rename -- is replayed into the
672 : // live CRDT session instead. Member events still feed the internal
673 : // callback so the swarm's view of who to sync with stays fresh.
674 89 : bool memberEvent = false;
675 89 : bool contentChanged = false;
676 186 : for (const auto& c : commits) {
677 194 : memberEvent |= c.at(CommitKey::TYPE) == CommitType::MEMBER;
678 194 : contentChanged |= c.at(CommitKey::TYPE) == CommitType::CHECKPOINT;
679 : }
680 89 : if (memberEvent && onMembersChanged_)
681 40 : onMembersChanged_(repository_->memberUris("", {}));
682 : // Nothing to replay for our own commits: what this device wrote came
683 : // out of the live session in the first place. From another thread:
684 : // announce() can run under writeMtx_ (a pull holds it) and the replay
685 : // asks the module for the conversation, which takes locks of its own.
686 89 : if (!commits.empty() && !commitFromSelf)
687 50 : dht::ThreadPool::io().run(
688 50 : [w = account_, parentId = repository_->parentConversationId(), convId, contentChanged] {
689 25 : if (auto acc = w.lock())
690 25 : if (auto collab = acc->collaborativeEditing())
691 50 : collab->onRepositoryUpdated(parentId, convId, contentChanged);
692 25 : });
693 89 : return;
694 : }
695 1507 : auto ok = !commits.empty();
696 3017 : auto lastId = ok ? commits.rbegin()->at(ConversationMapKeys::ID) : "";
697 1507 : addToHistory(loadedHistory_, commits, true, commitFromSelf);
698 : // A merge may have interleaved these commits with ones already shown.
699 3048 : if (std::any_of(commits.begin(), commits.end(), [](const auto& c) {
700 1541 : auto typeIt = c.find(CommitKey::TYPE);
701 1541 : return typeIt != c.end() && typeIt->second == CommitType::MERGE;
702 : }))
703 23 : relinearizeAfterMerge(commits);
704 1507 : if (ok) {
705 1504 : bool announceMember = false;
706 3055 : for (const auto& c : commits) {
707 : // Announce member events
708 3102 : if (c.at(CommitKey::TYPE) == CommitType::MEMBER) {
709 5835 : if (c.find(CommitKey::URI) != c.end() && c.find(CommitKey::ACTION) != c.end()) {
710 2334 : const auto& uri = c.at(CommitKey::URI);
711 1167 : const auto& actionStr = c.at(CommitKey::ACTION);
712 1167 : auto event = libjami::MemberEvent::INVALID;
713 1167 : if (actionStr == CommitAction::ADD)
714 482 : event = libjami::MemberEvent::ADD;
715 685 : else if (actionStr == CommitAction::JOIN)
716 662 : event = libjami::MemberEvent::JOIN;
717 23 : else if (actionStr == CommitAction::REMOVE)
718 5 : event = libjami::MemberEvent::REMOVE;
719 18 : else if (actionStr == CommitAction::BAN)
720 16 : event = libjami::MemberEvent::BAN;
721 2 : else if (actionStr == CommitAction::UNBAN)
722 2 : event = libjami::MemberEvent::UNBAN;
723 1167 : if (event == libjami::MemberEvent::BAN || event == libjami::MemberEvent::REMOVE) {
724 : // In this case, a potential host was removed during a call.
725 21 : updateActiveCalls(c);
726 21 : typers_->removeTyper(uri);
727 : }
728 1167 : if (event != libjami::MemberEvent::INVALID) {
729 1167 : announceMember = true;
730 1167 : emitSignal<libjami::ConversationSignal::ConversationMemberEvent>(accountId_,
731 : convId,
732 : uri,
733 : static_cast<int>(event));
734 : }
735 : }
736 768 : } else if (c.at(CommitKey::TYPE) == CommitType::CALL_HISTORY) {
737 60 : updateActiveCalls(c);
738 : }
739 : #ifdef ENABLE_PLUGIN
740 1551 : if (auto& pluginChatManager = Manager::instance().getJamiPluginManager().getChatServicesManager();
741 1551 : pluginChatManager.hasHandlers()) {
742 6 : auto cm = std::make_shared<JamiMessage>(accountId_, convId, c.at("author") != userId_, c, false);
743 3 : cm->isSwarm = true;
744 3 : pluginChatManager.publishMessage(cm);
745 :
746 3 : const auto editIt = c.find(CommitKey::EDIT);
747 3 : const bool isEdit = editIt != c.end() && !editIt->second.empty();
748 6 : const bool isText = c.at(CommitKey::TYPE) == CommitType::TEXT;
749 :
750 6 : if (const auto idIt = c.find("id"); idIt != c.end() && (isEdit || isText)) {
751 0 : const std::string msgId = idIt->second;
752 :
753 : // Snapshot under lock.
754 0 : libjami::SwarmMessage snap;
755 : {
756 0 : std::lock_guard const lk(loadedHistory_.mutex);
757 0 : if (auto it = loadedHistory_.quickAccess.find(msgId);
758 0 : it != loadedHistory_.quickAccess.end()) {
759 0 : snap = *it->second;
760 : }
761 0 : }
762 :
763 0 : if (!snap.id.empty()) {
764 : // Transform outside lock.
765 0 : std::vector<libjami::SwarmMessage> single {snap};
766 0 : pluginChatManager.transformSwarmMessages(single, accountId_, convId);
767 0 : auto& transformed = single[0];
768 0 : const auto boIt = transformed.pluginData.find("bodyOverwrite");
769 0 : const bool hasOverwrite = boIt != transformed.pluginData.end() && !boIt->second.empty();
770 :
771 : // Write back + build signal under one lock, emit outside.
772 0 : libjami::SwarmMessage signalMsg;
773 : {
774 0 : std::lock_guard const lk(loadedHistory_.mutex);
775 0 : if (isEdit || hasOverwrite) {
776 0 : signalMsg = injectEditionOverwrites(msgId, hasOverwrite ? boIt->second : "");
777 : } else {
778 0 : if (auto it = loadedHistory_.quickAccess.find(msgId);
779 0 : it != loadedHistory_.quickAccess.end()) {
780 0 : it->second->pluginData["bodyOverwrite"] = "";
781 : }
782 : }
783 0 : }
784 0 : if (!signalMsg.id.empty() && !signalMsg.body.contains(CommitKey::REACT_TO)) {
785 0 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_,
786 : convId,
787 : signalMsg);
788 : }
789 0 : }
790 0 : }
791 3 : } else
792 : #endif
793 : {
794 : // No active handlers: still notify clients when an edit arrives.
795 3096 : if (const auto editIt = c.find(CommitKey::EDIT); editIt != c.end() && !editIt->second.empty()) {
796 7 : libjami::SwarmMessage signalMsg;
797 : {
798 7 : std::lock_guard const lk(loadedHistory_.mutex);
799 7 : if (auto originalIt = loadedHistory_.quickAccess.find(editIt->second);
800 7 : originalIt != loadedHistory_.quickAccess.end()) {
801 7 : signalMsg = *originalIt->second;
802 : }
803 7 : }
804 21 : if (!signalMsg.id.empty() && !signalMsg.body.contains(CommitKey::REACT_TO)) {
805 6 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, convId, signalMsg);
806 : }
807 7 : }
808 : }
809 : }
810 :
811 1504 : if (announceMember && onMembersChanged_) {
812 1167 : onMembersChanged_(repository_->memberUris("", {}));
813 : }
814 : }
815 1596 : }
816 :
817 907 : void loadStatus()
818 : {
819 : try {
820 : // read file
821 1805 : auto file = fileutils::loadFile(statusPath_);
822 : // load values
823 9 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
824 9 : std::lock_guard lk {messageStatusMtx_};
825 9 : oh.get().convert(messagesStatus_);
826 907 : } catch (const std::exception& e) {
827 898 : }
828 907 : }
829 1921 : void saveStatus()
830 : {
831 1921 : std::ofstream file(statusPath_, std::ios::trunc | std::ios::binary);
832 1921 : msgpack::pack(file, messagesStatus_);
833 1921 : }
834 :
835 907 : void loadMobileNodes()
836 : {
837 : try {
838 1814 : auto file = fileutils::loadFile(mobileNodesPath_);
839 0 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
840 : try {
841 0 : std::vector<MobileNodeInfo> nodes;
842 0 : oh.get().convert(nodes);
843 0 : swarmManager_->setMobileNodes(nodes);
844 0 : } catch (const std::exception&) {
845 0 : std::vector<NodeId> nodes;
846 0 : oh.get().convert(nodes);
847 0 : swarmManager_->setMobileNodes(nodes);
848 0 : }
849 907 : } catch (const std::exception& e) {
850 907 : return;
851 907 : }
852 : }
853 :
854 25 : void loadActiveCalls() const
855 : {
856 : try {
857 : // read file
858 40 : auto file = fileutils::loadFile(activeCallsPath_);
859 : // load values
860 10 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
861 10 : std::lock_guard lk {activeCallsMtx_};
862 10 : oh.get().convert(activeCalls_);
863 25 : } catch (const std::exception& e) {
864 15 : return;
865 15 : }
866 : }
867 :
868 506 : void saveActiveCalls() const
869 : {
870 506 : std::ofstream file(activeCallsPath_, std::ios::trunc | std::ios::binary);
871 506 : msgpack::pack(file, activeCalls_);
872 506 : }
873 :
874 25 : void loadHostedCalls() const
875 : {
876 : try {
877 : // read file
878 45 : auto file = fileutils::loadFile(hostedCallsPath_);
879 : // load values
880 5 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
881 5 : std::lock_guard lk {activeCallsMtx_};
882 5 : oh.get().convert(hostedCalls_);
883 25 : } catch (const std::exception& e) {
884 20 : return;
885 20 : }
886 : }
887 :
888 49 : void saveHostedCalls() const
889 : {
890 49 : std::ofstream file(hostedCallsPath_, std::ios::trunc | std::ios::binary);
891 49 : msgpack::pack(file, hostedCalls_);
892 49 : }
893 :
894 : void voteUnban(const std::string& contactUri, const std::string_view type, const OnDoneCb& cb);
895 :
896 : std::vector<std::map<std::string, std::string>> getMembers(bool includeInvited,
897 : bool includeLeft,
898 : bool includeBanned) const;
899 :
900 : std::vector<std::map<std::string, std::string>> getTrackedMembers() const;
901 :
902 8633 : std::string_view memberBanType(const std::string& uri) const
903 : {
904 8633 : auto crt = fmt::format("{}.crt", uri);
905 8633 : auto bannedMember = repoPath_ / MemberPath::BANNED / MemberPath::MEMBERS / crt;
906 8633 : if (std::filesystem::is_regular_file(bannedMember))
907 32 : return "members"sv;
908 8601 : auto bannedAdmin = repoPath_ / MemberPath::BANNED / MemberPath::ADMINS / crt;
909 8601 : if (std::filesystem::is_regular_file(bannedAdmin))
910 0 : return "admins"sv;
911 8601 : auto bannedInvited = repoPath_ / MemberPath::BANNED / MemberPath::INVITED / uri;
912 8601 : if (std::filesystem::is_regular_file(bannedInvited))
913 2 : return "invited"sv;
914 8599 : return {};
915 8633 : }
916 :
917 9571 : bool isDeviceBanned(const std::string& deviceId) const
918 : {
919 9571 : auto crt = fmt::format("{}.crt", deviceId);
920 9571 : auto bannedDevice = repoPath_ / MemberPath::BANNED / MemberPath::DEVICES / crt;
921 19142 : return std::filesystem::is_regular_file(bannedDevice);
922 9571 : }
923 :
924 5230 : std::shared_ptr<dhtnet::ChannelSocket> gitSocket(const DeviceId& deviceId) const
925 : {
926 5230 : std::lock_guard lk(gitSocketMtx_);
927 5230 : auto deviceSockets = gitSocketList_.find(deviceId);
928 10460 : return (deviceSockets != gitSocketList_.end()) ? deviceSockets->second.get() : nullptr;
929 5230 : }
930 :
931 2284 : void addGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& socket)
932 : {
933 2284 : GitSocket replaced;
934 : {
935 2284 : std::lock_guard lk(gitSocketMtx_);
936 2284 : auto& slot = gitSocketList_[deviceId];
937 : // Re-registering the channel we already own must not close it.
938 2284 : if (slot.get() == socket)
939 806 : return;
940 1478 : replaced = std::move(slot);
941 1478 : slot = socket;
942 2284 : }
943 : // Closing the replaced channel, if any, happens here, outside the lock.
944 2284 : }
945 1445 : void removeGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& expected = {})
946 : {
947 1445 : GitSocket socket;
948 : {
949 1445 : std::lock_guard lk(gitSocketMtx_);
950 1445 : auto deviceSockets = gitSocketList_.find(deviceId);
951 1445 : if (deviceSockets == gitSocketList_.end())
952 695 : return;
953 : // A dead channel must not evict the one that replaced it.
954 750 : if (expected && deviceSockets->second.get() != expected)
955 1 : return;
956 749 : socket = std::move(deviceSockets->second);
957 749 : gitSocketList_.erase(deviceSockets);
958 1445 : }
959 : // Closing the channel outside the lock tells the peer to stop serving it, and wakes up
960 : // any fetch blocked reading from it, which would otherwise hold
961 : // ConversationRepository::opMtx_ until its read times out.
962 1445 : }
963 :
964 : void disconnectFromDevice(const DeviceId& deviceId);
965 :
966 : /**
967 : * Remove all git sockets and all DRT nodes associated with the given peer.
968 : * This is used when a swarm member is banned to ensure that we stop syncing
969 : * with them or sending them message notifications.
970 : */
971 : void disconnectFromPeer(const std::string& peerUri);
972 :
973 : std::vector<std::map<std::string, std::string>> getMembers(bool includeInvited, bool includeLeft) const;
974 :
975 : std::function<void()> bootstrapCb_;
976 : std::mutex bootstrapMtx_ {};
977 : #ifdef LIBJAMI_TEST
978 : std::function<void(std::string, BootstrapStatus)> bootstrapCbTest_;
979 : #endif
980 :
981 : std::mutex writeMtx_ {};
982 : const std::unique_ptr<ConversationRepository> repository_;
983 : const std::weak_ptr<JamiAccount> account_;
984 : const std::string accountId_;
985 : const std::string userId_;
986 : const std::string deviceId_;
987 : const std::shared_ptr<SwarmManager> swarmManager_;
988 : std::atomic_bool isRemoving_ {false};
989 : std::vector<libjami::SwarmMessage> loadMessages(const LogOptions& options, History* optHistory = nullptr);
990 : void loadMissingBodyOverwrites();
991 : void reloadBodyOverwriteMessages();
992 :
993 : // Out-of-order calls are safe: only the latest edition's overwrite propagates to the original.
994 : void updateMessageBodyOverwrite(const std::string& messageId, std::string_view bodyOverwrite);
995 : // Must be called with loadedHistory_.mutex held.
996 : // Stores bodyOverwrite on messageId, propagates to original if edition, returns signal message.
997 : // Only propagates when messageId == originalMsg->latestEditionId, so out-of-order calls are safe.
998 : libjami::SwarmMessage injectEditionOverwrites(const std::string& messageId, std::string_view bodyOverwrite);
999 :
1000 : void clearBodyOverwrites();
1001 : void pull(const std::string& deviceId);
1002 :
1003 : // Avoid multiple fetch/merges at the same time.
1004 : std::mutex pullcbsMtx_ {};
1005 : // store current remote in fetch
1006 : std::map<std::string, std::deque<std::pair<std::string, OnPullCb>>> fetchingRemotes_ {};
1007 : const std::shared_ptr<TransferManager> transferManager_ {};
1008 : const std::filesystem::path repoPath_ {};
1009 : const std::filesystem::path conversationDataPath_ {};
1010 : const std::filesystem::path fetchedPath_ {};
1011 :
1012 : // Manage last message displayed and status
1013 : const std::filesystem::path sendingPath_ {};
1014 : const std::filesystem::path preferencesPath_ {};
1015 : const std::filesystem::path statusPath_ {};
1016 :
1017 : // Devices known to be mobile, persisted so that wake-up notifications
1018 : // can be sent right after a restart, before any DRT gossip is received
1019 : const std::filesystem::path mobileNodesPath_ {};
1020 :
1021 : OnMembersChanged onMembersChanged_ {};
1022 : struct TrackedMember
1023 : {
1024 : std::set<DeviceId> devices;
1025 : std::set<DeviceId> failedDevices;
1026 : std::chrono::steady_clock::time_point trackedSince {std::chrono::steady_clock::now()};
1027 : bool presenceReported {false};
1028 : };
1029 : std::map<std::string, TrackedMember> trackedMembers_;
1030 : mutable std::mutex trackedMembersMtx_;
1031 : bool isTracking_ {false};
1032 : void setupMemberCallback();
1033 : void startTracking(std::weak_ptr<Conversation> w);
1034 : void stopTracking();
1035 : void rotateTrackedMembers(const std::string& memberUri = "", const DeviceId& deviceId = {});
1036 : void monitorConnection(std::weak_ptr<Conversation> w);
1037 : void onConnectionFailed(const DeviceId& deviceId, const std::string& memberUri = "");
1038 : void onDeviceOffline(const DeviceId& deviceId, const std::string& memberUri);
1039 :
1040 : uint64_t presenceDeviceListenerToken_ {0};
1041 :
1042 : // Manage hosted calls on this device
1043 : std::filesystem::path hostedCallsPath_ {};
1044 : mutable std::map<std::string, uint64_t /* start time */> hostedCalls_ {};
1045 : // Manage active calls for this conversation (can be hosted by other devices)
1046 : std::filesystem::path activeCallsPath_ {};
1047 : mutable std::mutex activeCallsMtx_ {};
1048 : mutable std::vector<std::map<std::string, std::string>> activeCalls_ {};
1049 :
1050 : mutable std::mutex gitSocketMtx_ {};
1051 : GitSocketList gitSocketList_ {};
1052 :
1053 : // Bootstrap
1054 : const std::shared_ptr<asio::io_context> ioContext_;
1055 :
1056 : /**
1057 : * Loaded history represents the linearized history to show for clients
1058 : */
1059 : History loadedHistory_ {};
1060 : std::vector<std::shared_ptr<libjami::SwarmMessage>> addToHistory(
1061 : History& history,
1062 : const std::vector<std::map<std::string, std::string>>& commits,
1063 : bool messageReceived = false,
1064 : bool commitFromSelf = false);
1065 :
1066 : void handleReaction(History& history, const std::shared_ptr<libjami::SwarmMessage>& sharedCommit) const;
1067 : void handleEdition(History& history,
1068 : const std::shared_ptr<libjami::SwarmMessage>& sharedCommit,
1069 : bool messageReceived) const;
1070 : bool handleMessage(History& history,
1071 : const std::shared_ptr<libjami::SwarmMessage>& sharedCommit,
1072 : bool messageReceived) const;
1073 : void rectifyStatus(const std::shared_ptr<libjami::SwarmMessage>& message, History& history) const;
1074 : /**
1075 : * After a non fast-forward merge, reorder the newest part of loadedHistory_ so
1076 : * that it matches the repository log order, and notify clients of moved messages.
1077 : * @param commits Commits announced by the merge, in chronological order
1078 : */
1079 : void relinearizeAfterMerge(const std::vector<std::map<std::string, std::string>>& commits);
1080 : /**
1081 : * {uri, {
1082 : * {"fetch", "commitId"},
1083 : * {"fetched_ts", "timestamp"},
1084 : * {"read", "commitId"},
1085 : * {"read_ts", "timestamp"}
1086 : * }
1087 : * }
1088 : */
1089 : mutable std::mutex messageStatusMtx_;
1090 : std::function<void(const std::map<std::string, std::map<std::string, std::string>>&)> messageStatusCb_ {};
1091 : std::map<std::string, std::map<std::string, std::string>> messagesStatus_ {};
1092 : /**
1093 : * Status: 0 = commited, 1 = fetched, 2 = read
1094 : * This cache the curent status to add in the messages
1095 : */
1096 : // Note: only store int32_t cause it's easy to pass to dbus this way
1097 : // memberToStatus serves as a cache for loading messages
1098 : std::map<std::string, int32_t> memberToStatus;
1099 :
1100 : // futureStatus is used to store the status for receiving messages
1101 : // (because we're not sure to fetch the commit before receiving a status change for this)
1102 : std::map<std::string, std::map<std::string, int32_t>> futureStatus;
1103 : // Update internal structures regarding status
1104 : void updateStatus(const std::string& uri,
1105 : libjami::Account::MessageStates status,
1106 : const std::string& commitId,
1107 : const std::string& ts,
1108 : bool emit = false);
1109 :
1110 : std::shared_ptr<Typers> typers_;
1111 : };
1112 :
1113 : /**
1114 : * Stores bodyOverwrite on messageId, propagates to original if messageId is an edition commit,
1115 : * and returns the message to signal (original if edition, messageId's message otherwise),
1116 : * with each edition entry's "bodyOverwrite" injected from the corresponding commit's pluginData.
1117 : * Must be called with loadedHistory_.mutex held.
1118 : * @param messageId ID of the edition commit or original message to update.
1119 : * @param bodyOverwrite New overwrite value; empty string marks the message as processed with no overwrite.
1120 : */
1121 : libjami::SwarmMessage
1122 0 : Conversation::Impl::injectEditionOverwrites(const std::string& messageId, const std::string_view bodyOverwrite)
1123 : {
1124 0 : const auto it = loadedHistory_.quickAccess.find(messageId);
1125 0 : if (it == loadedHistory_.quickAccess.end()) {
1126 0 : return {};
1127 : }
1128 :
1129 0 : libjami::SwarmMessage* signalTarget = it->second.get();
1130 :
1131 0 : if (const auto editIt = it->second->body.find(CommitKey::EDIT); editIt != it->second->body.end()) {
1132 0 : it->second->pluginData["bodyOverwrite"] = bodyOverwrite;
1133 0 : if (editIt->second.empty()) {
1134 0 : return {};
1135 : }
1136 0 : const auto originalIt = loadedHistory_.quickAccess.find(editIt->second);
1137 0 : if (originalIt == loadedHistory_.quickAccess.end()) {
1138 0 : return {};
1139 : }
1140 0 : signalTarget = originalIt->second.get();
1141 0 : } else if (it->second->editions.empty()) {
1142 0 : it->second->pluginData["bodyOverwrite"] = bodyOverwrite;
1143 : }
1144 :
1145 0 : if (const auto msgEditionIt = std::ranges::find_if(signalTarget->editions,
1146 0 : [&](const auto& e) {
1147 0 : const auto idIt = e.find("id");
1148 0 : return idIt != e.end() && idIt->second == messageId;
1149 : });
1150 0 : msgEditionIt != signalTarget->editions.end()) {
1151 0 : (*msgEditionIt)["bodyOverwrite"] = bodyOverwrite;
1152 : }
1153 :
1154 0 : if (const auto latestEdIt = loadedHistory_.quickAccess.find(signalTarget->latestEditionId);
1155 0 : latestEdIt != loadedHistory_.quickAccess.end()) {
1156 0 : if (const auto latestEdBoIt = latestEdIt->second->pluginData.find("bodyOverwrite");
1157 0 : latestEdBoIt != latestEdIt->second->pluginData.end() && !latestEdBoIt->second.empty()) {
1158 0 : signalTarget->pluginData["bodyOverwrite"] = latestEdBoIt->second;
1159 : }
1160 : }
1161 0 : return *signalTarget;
1162 0 : }
1163 :
1164 : void
1165 907 : Conversation::Impl::setupMemberCallback()
1166 : {
1167 907 : repository_->onMembersChanged([this](const std::set<std::string>& memberUris) {
1168 : {
1169 1602 : std::lock_guard lk(trackedMembersMtx_);
1170 1602 : if (isTracking_) {
1171 249 : if (auto acc = account_.lock()) {
1172 571 : for (auto it = trackedMembers_.begin(); it != trackedMembers_.end();) {
1173 322 : if (memberUris.find(it->first) == memberUris.end()) {
1174 4 : acc->presenceManager()->untrackBuddy(it->first);
1175 4 : it = trackedMembers_.erase(it);
1176 : } else {
1177 318 : ++it;
1178 : }
1179 : }
1180 249 : }
1181 : }
1182 1602 : }
1183 :
1184 1602 : if (onMembersChanged_)
1185 1602 : onMembersChanged_(memberUris);
1186 1602 : });
1187 907 : }
1188 :
1189 : void
1190 1509 : Conversation::Impl::startTracking(std::weak_ptr<Conversation> w)
1191 : {
1192 1509 : auto acc = account_.lock();
1193 1509 : if (!acc)
1194 0 : return;
1195 :
1196 : {
1197 1509 : std::lock_guard lk(trackedMembersMtx_);
1198 1509 : if (isTracking_)
1199 557 : return;
1200 952 : isTracking_ = true;
1201 1904 : presenceDeviceListenerToken_ = acc->presenceManager()->addDeviceListener(
1202 1904 : [w](const std::string& uri, const DeviceId& deviceId, bool online) {
1203 1673 : if (auto sthis = w.lock()) {
1204 1673 : if (online) {
1205 1543 : if (sthis->isMember(uri))
1206 3645 : sthis->addKnownDevices({deviceId}, uri);
1207 : } else {
1208 130 : sthis->pimpl_->onDeviceOffline(deviceId, uri);
1209 : }
1210 1673 : }
1211 1673 : });
1212 1509 : }
1213 :
1214 1904 : rotateTrackedMembers();
1215 1509 : }
1216 :
1217 : void
1218 1668 : Conversation::Impl::stopTracking()
1219 : {
1220 : // Collect data under trackedMembersMtx_, then release it before calling
1221 : // into PresenceManager to avoid lock-ordering inversion with the
1222 : // PresenceManager mutex (which is held when notifyListeners calls
1223 : // addKnownDevices, which in turn acquires trackedMembersMtx_).
1224 1668 : uint64_t token = 0;
1225 1668 : std::vector<std::string> urisToUntrack;
1226 : {
1227 1668 : std::lock_guard lk(trackedMembersMtx_);
1228 1668 : if (!isTracking_)
1229 716 : return;
1230 952 : isTracking_ = false;
1231 952 : token = presenceDeviceListenerToken_;
1232 952 : presenceDeviceListenerToken_ = 0;
1233 2774 : for (const auto& [uri, _] : trackedMembers_) {
1234 1822 : urisToUntrack.push_back(uri);
1235 : }
1236 952 : trackedMembers_.clear();
1237 1668 : }
1238 :
1239 952 : auto acc = account_.lock();
1240 952 : if (!acc)
1241 199 : return;
1242 :
1243 753 : if (token) {
1244 753 : acc->presenceManager()->removeDeviceListener(token);
1245 : }
1246 :
1247 2230 : for (const auto& uri : urisToUntrack) {
1248 1477 : acc->presenceManager()->untrackBuddy(uri);
1249 : }
1250 1867 : }
1251 :
1252 : void
1253 1756 : Conversation::Impl::rotateTrackedMembers(const std::string& memberUri, const DeviceId& deviceId)
1254 : {
1255 1756 : auto acc = account_.lock();
1256 1756 : if (!acc)
1257 0 : return;
1258 :
1259 1756 : std::lock_guard lk(trackedMembersMtx_);
1260 1756 : if (!isTracking_)
1261 589 : return;
1262 :
1263 1167 : auto members = repository_->members();
1264 :
1265 1167 : if (!memberUri.empty()) {
1266 215 : auto it = trackedMembers_.find(memberUri);
1267 215 : if (it == trackedMembers_.end())
1268 4 : return;
1269 211 : auto& info = it->second;
1270 211 : if (deviceId) {
1271 103 : JAMI_WARNING("{} [device {}] Rotating tracked members after connection failure", toString(), deviceId);
1272 103 : info.failedDevices.insert(deviceId);
1273 : } else {
1274 108 : JAMI_WARNING("{} Rotating tracked members, {} has no device online", toString(), memberUri);
1275 : }
1276 : // std::includes() is vacuously true on an empty second range, so a
1277 : // member whose devices are still unknown would be dropped on its very
1278 : // first failure. That set is only filled by addKnownDevices() once
1279 : // the presence manager reports the member online, while a failure
1280 : // is reported for any device the DRT happens to try, so it is
1281 : // routinely empty right after the member was picked. Give presence
1282 : // time to answer; past that, an empty set means nobody is online
1283 : // and the member is as good as fully failed.
1284 211 : constexpr auto PRESENCE_GRACE = std::chrono::seconds(30);
1285 330 : auto presencePending = info.devices.empty() && !info.presenceReported
1286 330 : && std::chrono::steady_clock::now() - info.trackedSince < PRESENCE_GRACE;
1287 422 : if (!presencePending
1288 211 : && std::includes(info.failedDevices.begin(),
1289 : info.failedDevices.end(),
1290 : info.devices.begin(),
1291 : info.devices.end())) {
1292 : // Rotating only makes sense if another member can take the slot. In
1293 : // a one-to-one conversation there is none: dropping the peer would
1294 : // leave presence watched for nobody, so addKnownDevices() is never
1295 : // called again and the swarm stays silent until the account reloads.
1296 169 : auto replaceable = std::any_of(members.begin(), members.end(), [&](const auto& m) {
1297 350 : return m.uri != memberUri && trackedMembers_.find(m.uri) == trackedMembers_.end();
1298 : });
1299 169 : if (replaceable) {
1300 0 : acc->presenceManager()->untrackBuddy(it->first);
1301 0 : trackedMembers_.erase(it);
1302 : }
1303 : }
1304 : }
1305 :
1306 1163 : size_t N = members.size();
1307 1163 : size_t K = std::min(N, 3 + (size_t) std::log2(N));
1308 :
1309 : // If we are already trying enough devices, don't rotate
1310 1163 : auto activeDevices = swarmManager_->getActiveNodesCount();
1311 1163 : if (activeDevices >= 2 * K)
1312 0 : return;
1313 :
1314 1163 : JAMI_WARNING("{} Refreshing tracked members: {}/{} active ({}/{} devices)",
1315 : toString(),
1316 : trackedMembers_.size(),
1317 : K,
1318 : activeDevices,
1319 : 2 * K);
1320 :
1321 : // Add new members if we have space
1322 1163 : if (trackedMembers_.size() < K) {
1323 952 : std::vector<std::string> candidates;
1324 952 : candidates.reserve(N - trackedMembers_.size());
1325 2979 : for (const auto& m : members) {
1326 2027 : if (m.uri != memberUri && trackedMembers_.find(m.uri) == trackedMembers_.end()) {
1327 2027 : candidates.push_back(m.uri);
1328 : }
1329 : }
1330 952 : if (!candidates.empty()) {
1331 952 : std::vector<std::string> chosen;
1332 952 : std::sample(candidates.begin(),
1333 : candidates.end(),
1334 : std::back_inserter(chosen),
1335 952 : K - trackedMembers_.size(),
1336 952 : Manager::instance().getSeededRandomEngine());
1337 2778 : for (const auto& uri : chosen) {
1338 1826 : acc->presenceManager()->trackBuddy(uri);
1339 1826 : trackedMembers_.emplace(uri, TrackedMember {});
1340 : }
1341 952 : }
1342 952 : }
1343 2353 : }
1344 :
1345 : void
1346 696 : Conversation::Impl::onConnectionFailed(const DeviceId& deviceId, const std::string& memberUri)
1347 : {
1348 696 : rotateTrackedMembers(memberUri, deviceId);
1349 696 : }
1350 :
1351 : void
1352 130 : Conversation::Impl::onDeviceOffline(const DeviceId& deviceId, const std::string& memberUri)
1353 : {
1354 : // A device that left is neither a target nor a failure to hold against its member.
1355 130 : bool lastDevice = false;
1356 : {
1357 130 : std::lock_guard lk(trackedMembersMtx_);
1358 130 : auto it = trackedMembers_.find(memberUri);
1359 130 : if (it == trackedMembers_.end())
1360 8 : return;
1361 122 : it->second.presenceReported = true;
1362 122 : it->second.devices.erase(deviceId);
1363 122 : it->second.failedDevices.erase(deviceId);
1364 122 : lastDevice = it->second.devices.empty();
1365 130 : }
1366 : // Nobody left to reach: watch someone else rather than wait for the DRT
1367 : // to fail on ids it may never retry.
1368 122 : if (lastDevice)
1369 108 : rotateTrackedMembers(memberUri);
1370 : }
1371 :
1372 : void
1373 1455 : Conversation::Impl::monitorConnection(std::weak_ptr<Conversation> w)
1374 : {
1375 1455 : if (!swarmManager_->isConnected()) {
1376 1410 : startTracking(w);
1377 : }
1378 :
1379 1455 : swarmManager_->onConnectionChanged([w](bool ok) {
1380 860 : dht::ThreadPool::io().run([w, ok] {
1381 860 : if (auto sthis = w.lock()) {
1382 : // Check the current connection state rather than relying on the
1383 : // captured `ok` value, because the thread pool does not guarantee
1384 : // execution order. Two rapid state changes (connected then
1385 : // disconnected) could otherwise cause stopTracking to run after
1386 : // startTracking, leaving the conversation untracked while the DRT
1387 : // has no connected nodes.
1388 860 : if (sthis->pimpl_->swarmManager_->isConnected()) {
1389 761 : sthis->pimpl_->stopTracking();
1390 : } else {
1391 99 : sthis->pimpl_->startTracking(w);
1392 : }
1393 860 : if (ok) {
1394 762 : if (sthis->pimpl_->bootstrapCb_)
1395 762 : sthis->pimpl_->bootstrapCb_();
1396 : }
1397 : #ifdef LIBJAMI_TEST
1398 860 : if (sthis->pimpl_->bootstrapCbTest_)
1399 98 : sthis->pimpl_->bootstrapCbTest_(sthis->id(),
1400 49 : ok ? BootstrapStatus::SUCCESS : BootstrapStatus::FAILED);
1401 : #endif
1402 860 : }
1403 860 : });
1404 860 : });
1405 1455 : }
1406 :
1407 : bool
1408 20 : Conversation::Impl::isAdmin() const
1409 : {
1410 20 : auto adminsPath = repoPath_ / MemberPath::ADMINS;
1411 40 : return std::filesystem::is_regular_file(fileutils::getFullPath(adminsPath, userId_ + ".crt"));
1412 20 : }
1413 :
1414 : void
1415 13 : Conversation::Impl::disconnectFromDevice(const DeviceId& deviceId)
1416 : {
1417 26 : swarmManager_->deleteNode({deviceId});
1418 :
1419 13 : GitSocket socket;
1420 : {
1421 13 : std::lock_guard lk(gitSocketMtx_);
1422 13 : if (auto it = gitSocketList_.find(deviceId); it != gitSocketList_.end()) {
1423 13 : socket = std::move(it->second);
1424 13 : gitSocketList_.erase(it);
1425 : }
1426 13 : }
1427 : // The channel is closed here, outside the lock.
1428 13 : }
1429 :
1430 : void
1431 16 : Conversation::Impl::disconnectFromPeer(const std::string& peerUri)
1432 : {
1433 16 : std::set<DeviceId> devicesToRemove;
1434 :
1435 16 : const auto nodes = swarmManager_->getAllNodes();
1436 35 : for (const auto node : nodes)
1437 19 : if (peerUri == repository_->uriFromDevice(node.toString()))
1438 12 : devicesToRemove.emplace(node);
1439 :
1440 : {
1441 16 : std::lock_guard lk(gitSocketMtx_);
1442 35 : for (const auto& [deviceId, socket] : gitSocketList_) {
1443 19 : auto cert = socket ? socket->peerCertificate() : nullptr;
1444 38 : if ((cert && cert->issuer && cert->issuer->getId().toString() == peerUri)
1445 38 : || peerUri == repository_->uriFromDevice(deviceId.toString())) {
1446 12 : devicesToRemove.emplace(deviceId);
1447 : }
1448 19 : }
1449 16 : }
1450 :
1451 28 : for (const auto& deviceId : devicesToRemove)
1452 12 : disconnectFromDevice(deviceId);
1453 16 : }
1454 :
1455 : std::vector<std::map<std::string, std::string>>
1456 287 : Conversation::Impl::getMembers(bool includeInvited, bool includeLeft, bool includeBanned) const
1457 : {
1458 287 : std::vector<std::map<std::string, std::string>> result;
1459 287 : auto members = repository_->members();
1460 287 : std::lock_guard lk(messageStatusMtx_);
1461 860 : for (const auto& member : members) {
1462 573 : if (member.role == MemberRole::BANNED && !includeBanned) {
1463 0 : continue;
1464 : }
1465 573 : if (member.role == MemberRole::INVITED && !includeInvited)
1466 0 : continue;
1467 573 : if (member.role == MemberRole::LEFT && !includeLeft)
1468 0 : continue;
1469 573 : auto mm = member.map();
1470 573 : auto it = messagesStatus_.find(member.uri);
1471 573 : if (it != messagesStatus_.end()) {
1472 650 : auto readIt = it->second.find("read");
1473 325 : if (readIt != it->second.end())
1474 303 : mm[ConversationMapKeys::LAST_DISPLAYED] = readIt->second;
1475 : }
1476 573 : result.emplace_back(std::move(mm));
1477 573 : }
1478 574 : return result;
1479 287 : }
1480 :
1481 : std::vector<std::map<std::string, std::string>>
1482 1 : Conversation::Impl::getTrackedMembers() const
1483 : {
1484 1 : std::vector<std::map<std::string, std::string>> result;
1485 1 : std::lock_guard lk(trackedMembersMtx_);
1486 3 : for (const auto& [uri, member] : trackedMembers_) {
1487 2 : std::map<std::string, std::string> map;
1488 2 : map["uri"] = uri;
1489 2 : std::string devicesStr;
1490 3 : for (const auto& dev : member.devices) {
1491 1 : if (!devicesStr.empty())
1492 0 : devicesStr += ";";
1493 1 : devicesStr += dev.toString();
1494 : }
1495 2 : map["devices"] = devicesStr;
1496 2 : result.emplace_back(std::move(map));
1497 2 : }
1498 2 : return result;
1499 1 : }
1500 :
1501 : std::vector<std::string>
1502 25 : Conversation::Impl::commitsEndedCalls()
1503 : {
1504 : // Handle current calls
1505 25 : std::vector<std::string> commits {};
1506 25 : std::unique_lock lk(writeMtx_);
1507 25 : std::unique_lock lkA(activeCallsMtx_);
1508 25 : for (const auto& hostedCall : hostedCalls_) {
1509 : // In this case, this means that we left
1510 : // the conference while still hosting it, so activeCalls
1511 : // will not be correctly updated
1512 : // We don't need to send notifications there, as peers will sync with presence
1513 0 : auto confId = hostedCall.first;
1514 0 : auto now = std::chrono::system_clock::now();
1515 0 : uint64_t nowConverted = std::chrono::duration_cast<std::chrono::seconds>(now.time_since_epoch()).count();
1516 0 : auto duration = (nowConverted > hostedCall.second) ? (nowConverted - hostedCall.second) * 1000 : 0;
1517 0 : auto commitMessage = CommitMessage::conferenceHostingEnd(confId, deviceId_, userId_, duration);
1518 :
1519 0 : auto itActive = std::find_if(activeCalls_.begin(),
1520 : activeCalls_.end(),
1521 0 : [this, confId = hostedCall.first](const auto& value) {
1522 0 : return value.at("id") == confId && value.at("uri") == userId_
1523 0 : && value.at("device") == deviceId_;
1524 : });
1525 0 : if (itActive != activeCalls_.end())
1526 0 : activeCalls_.erase(itActive);
1527 0 : commits.emplace_back(repository_->commitMessage(commitMessage.toString()));
1528 0 : JAMI_DEBUG("Removing hosted conference... {:s}", hostedCall.first);
1529 0 : }
1530 25 : hostedCalls_.clear();
1531 25 : saveActiveCalls();
1532 25 : saveHostedCalls();
1533 50 : return commits;
1534 25 : }
1535 :
1536 : std::vector<libjami::SwarmMessage>
1537 3124 : Conversation::Impl::loadMessages(const LogOptions& options, History* optHistory)
1538 : {
1539 3124 : auto history = optHistory ? optHistory : &loadedHistory_;
1540 :
1541 : // history->mutex is locked by the caller
1542 3124 : if (!repository_ || history->loading) {
1543 0 : return {};
1544 : }
1545 3124 : history->loading = true;
1546 :
1547 : // By convention, if options.nbOfCommits is zero, then we
1548 : // don't impose a limit on the number of commits returned.
1549 3124 : bool limitNbOfCommits = options.nbOfCommits > 0;
1550 :
1551 3124 : auto startLogging = options.from == "";
1552 3124 : auto breakLogging = false;
1553 3124 : auto currentHistorySize = loadedHistory_.messageList.size();
1554 3124 : std::vector<std::string> replies;
1555 3124 : std::vector<std::shared_ptr<libjami::SwarmMessage>> msgList;
1556 6248 : repository_->log(
1557 : /* preCondition */
1558 6248 : [&](const auto& id, const auto& author, const auto& commit) {
1559 17271 : if (options.skipMerge && git_commit_parentcount(commit.get()) > 1) {
1560 0 : return CallbackResult::Skip;
1561 : }
1562 17271 : if (id == options.to) {
1563 821 : if (options.includeTo)
1564 0 : breakLogging = true; // For the next commit
1565 : }
1566 17271 : if (replies.empty()) { // This avoid load until
1567 : // NOTE: in the future, we may want to add "Reply-Body" in commit to avoid to load
1568 : // until this commit
1569 17270 : if ((limitNbOfCommits
1570 17270 : && (loadedHistory_.messageList.size() - currentHistorySize) == options.nbOfCommits))
1571 1 : return CallbackResult::Break; // Stop logging
1572 17269 : if (breakLogging)
1573 0 : return CallbackResult::Break; // Stop logging
1574 17269 : if (id == options.to && !options.includeTo) {
1575 820 : return CallbackResult::Break; // Stop logging
1576 : }
1577 : }
1578 :
1579 16450 : if (!startLogging && options.from != "" && options.from == id)
1580 1911 : startLogging = true;
1581 16450 : if (!startLogging)
1582 22 : return CallbackResult::Skip; // Start logging after this one
1583 :
1584 16428 : if (options.fastLog) {
1585 14370 : if (options.authorUri != "") {
1586 1 : if (options.authorUri == repository_->uriFromDevice(author.email)) {
1587 1 : return CallbackResult::Break; // Found author, stop
1588 : }
1589 : }
1590 : }
1591 :
1592 16427 : return CallbackResult::Ok; // Continue
1593 : },
1594 : /* emplaceCb */
1595 6248 : [&](auto&& cc) {
1596 16427 : if (limitNbOfCommits && (msgList.size() == options.nbOfCommits))
1597 857 : return;
1598 15573 : auto optMessage = repository_->convCommitToMap(cc);
1599 15573 : if (!optMessage.has_value())
1600 3 : return;
1601 15570 : auto message = optMessage.value();
1602 46710 : if (message.find(CommitKey::REPLY_TO) != message.end()) {
1603 1 : auto it = std::find(replies.begin(), replies.end(), message.at(CommitKey::REPLY_TO));
1604 1 : if (it == replies.end()) {
1605 3 : replies.emplace_back(message.at(CommitKey::REPLY_TO));
1606 : }
1607 : }
1608 15570 : auto it = std::find(replies.begin(), replies.end(), message.at("id"));
1609 15570 : if (it != replies.end()) {
1610 1 : replies.erase(it);
1611 : }
1612 15570 : std::shared_ptr<libjami::SwarmMessage> firstMsg;
1613 15570 : if ((history == &loadedHistory_) && msgList.empty() && !loadedHistory_.messageList.empty()) {
1614 0 : firstMsg = *loadedHistory_.messageList.rbegin();
1615 : }
1616 46710 : auto added = addToHistory(*history, {message}, false, false);
1617 15570 : if (!added.empty() && firstMsg) {
1618 0 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, repository_->id(), *firstMsg);
1619 : }
1620 15570 : msgList.insert(msgList.end(), added.begin(), added.end());
1621 31143 : },
1622 : /* postCondition */
1623 0 : [&](auto, auto, auto) {
1624 : // Stop logging if there was a limit set on the number of commits
1625 : // to return and we reached it. This isn't strictly necessary since
1626 : // the check at the beginning of `emplaceCb` ensures that we won't
1627 : // return too many messages, but it prevents us from needlessly
1628 : // iterating over a (potentially) large number of commits.
1629 16427 : return limitNbOfCommits && (msgList.size() == options.nbOfCommits);
1630 : },
1631 3124 : options.from,
1632 3124 : options.logIfNotFound);
1633 :
1634 3124 : history->loading = false;
1635 3124 : history->cv.notify_all();
1636 :
1637 : // Convert for client (remove ptr)
1638 3124 : std::vector<libjami::SwarmMessage> ret;
1639 3124 : ret.reserve(msgList.size());
1640 18683 : for (const auto& msg : msgList) {
1641 15559 : ret.emplace_back(*msg);
1642 : }
1643 3124 : return ret;
1644 3124 : }
1645 :
1646 : void
1647 3 : Conversation::Impl::handleReaction(History& history, const std::shared_ptr<libjami::SwarmMessage>& sharedCommit) const
1648 : {
1649 6 : auto it = history.quickAccess.find(sharedCommit->body.at(CommitKey::REACT_TO));
1650 3 : auto peditIt = history.pendingEditions.find(sharedCommit->id);
1651 3 : if (peditIt != history.pendingEditions.end()) {
1652 0 : auto oldBody = sharedCommit->body;
1653 0 : sharedCommit->body[CommitKey::BODY] = peditIt->second.front()->body[CommitKey::BODY];
1654 0 : if (sharedCommit->body.at(CommitKey::BODY).empty())
1655 0 : return;
1656 0 : history.pendingEditions.erase(peditIt);
1657 0 : }
1658 3 : if (it != history.quickAccess.end()) {
1659 : // A deleted message (its body, or tid for a file, has been emptied by an edition) displays
1660 : // no reactions.
1661 3 : const auto& target = it->second;
1662 3 : const std::string& bodyKey = (target->type == CommitType::DATA_TRANSFER) ? CommitKey::TID : CommitKey::BODY;
1663 3 : auto bodyIt = target->body.find(bodyKey);
1664 3 : if (bodyIt != target->body.end() && bodyIt->second.empty())
1665 0 : return;
1666 3 : target->reactions.emplace_back(sharedCommit->body);
1667 3 : emitSignal<libjami::ConversationSignal::ReactionAdded>(accountId_,
1668 3 : repository_->id(),
1669 3 : target->id,
1670 3 : sharedCommit->body);
1671 3 : } else {
1672 0 : history.pendingReactions[sharedCommit->body.at(CommitKey::REACT_TO)].emplace_back(sharedCommit->body);
1673 : }
1674 : }
1675 :
1676 : void
1677 9 : Conversation::Impl::handleEdition(History& history,
1678 : const std::shared_ptr<libjami::SwarmMessage>& sharedCommit,
1679 : bool messageReceived) const
1680 : {
1681 18 : auto editId = sharedCommit->body.at(CommitKey::EDIT);
1682 9 : auto it = history.quickAccess.find(editId);
1683 9 : if (it != history.quickAccess.end()) {
1684 7 : auto baseCommit = it->second;
1685 7 : if (baseCommit) {
1686 21 : auto itReact = baseCommit->body.find(CommitKey::REACT_TO);
1687 7 : std::string toReplace = (baseCommit->type == CommitType::DATA_TRANSFER) ? CommitKey::TID : CommitKey::BODY;
1688 : // An edition need not carry the field it replaces: retiring a
1689 : // collaborative document says nothing but which commit it retires.
1690 : // Absent means empty, which is what an edition to nothing already
1691 : // means everywhere below.
1692 7 : std::string body;
1693 7 : if (auto itBody = sharedCommit->body.find(toReplace); itBody != sharedCommit->body.end())
1694 5 : body = itBody->second;
1695 : // Edit reaction
1696 7 : if (itReact != baseCommit->body.end()) {
1697 1 : baseCommit->body[toReplace] = body; // Replace body if pending
1698 1 : it = history.quickAccess.find(itReact->second);
1699 1 : auto itPending = history.pendingReactions.find(itReact->second);
1700 1 : if (it != history.quickAccess.end()) {
1701 1 : baseCommit = it->second; // Base commit
1702 1 : auto itPreviousReact = std::find_if(baseCommit->reactions.begin(),
1703 1 : baseCommit->reactions.end(),
1704 1 : [&](const auto& reaction) {
1705 3 : return reaction.at("id") == editId;
1706 : });
1707 1 : if (itPreviousReact != baseCommit->reactions.end()) {
1708 1 : (*itPreviousReact)[toReplace] = body;
1709 1 : if (body.empty()) {
1710 1 : baseCommit->reactions.erase(itPreviousReact);
1711 2 : emitSignal<libjami::ConversationSignal::ReactionRemoved>(accountId_,
1712 1 : repository_->id(),
1713 1 : baseCommit->id,
1714 : editId);
1715 : }
1716 : }
1717 0 : } else if (itPending != history.pendingReactions.end()) {
1718 : // Else edit if pending
1719 0 : auto itReaction = std::find_if(itPending->second.begin(),
1720 0 : itPending->second.end(),
1721 0 : [&](const auto& reaction) { return reaction.at("id") == editId; });
1722 0 : if (itReaction != itPending->second.end()) {
1723 0 : (*itReaction)[toReplace] = body;
1724 0 : if (body.empty())
1725 0 : itPending->second.erase(itReaction);
1726 : }
1727 : } else {
1728 : // Add to pending edtions
1729 0 : messageReceived ? history.pendingEditions[editId].emplace_front(sharedCommit)
1730 0 : : history.pendingEditions[editId].emplace_back(sharedCommit);
1731 : }
1732 : } else {
1733 : // Normal message
1734 6 : auto editionBody = it->second->body;
1735 : // Tag the superseded body with the commit id that introduced it, so the
1736 : // client can look up per-edition plugin overwrites via quickAccess.
1737 12 : editionBody["id"] = it->second->latestEditionId.empty() ? it->second->id : it->second->latestEditionId;
1738 6 : it->second->editions.emplace(it->second->editions.begin(), std::move(editionBody));
1739 : #ifdef ENABLE_PLUGIN
1740 12 : if (const auto boIt = it->second->pluginData.find("bodyOverwrite");
1741 6 : boIt != it->second->pluginData.end()) {
1742 0 : it->second->editions.front()["bodyOverwrite"] = boIt->second;
1743 : }
1744 : #endif
1745 6 : it->second->body[toReplace] = sharedCommit->body[toReplace];
1746 6 : it->second->latestEditionId = sharedCommit->id;
1747 6 : if (toReplace == CommitKey::TID) {
1748 : // Avoid to replace fileId in client
1749 6 : it->second->body["fileId"] = "";
1750 : }
1751 : // Remove reactions
1752 6 : if (sharedCommit->body.at(toReplace).empty())
1753 5 : it->second->reactions.clear();
1754 6 : }
1755 7 : }
1756 7 : } else {
1757 2 : messageReceived ? history.pendingEditions[editId].emplace_front(sharedCommit)
1758 2 : : history.pendingEditions[editId].emplace_back(sharedCommit);
1759 : }
1760 9 : }
1761 :
1762 : bool
1763 17091 : Conversation::Impl::handleMessage(History& history,
1764 : const std::shared_ptr<libjami::SwarmMessage>& sharedCommit,
1765 : bool messageReceived) const
1766 : {
1767 17091 : if (messageReceived) {
1768 : // For a received message, we place it at the beginning of the list
1769 1512 : if (!history.messageList.empty())
1770 1034 : sharedCommit->linearizedParent = (*history.messageList.begin())->id;
1771 1512 : history.messageList.emplace_front(sharedCommit);
1772 : } else {
1773 : // For a loaded message, we load from newest to oldest
1774 : // So we change the parent of the last message.
1775 15579 : if (!history.messageList.empty())
1776 12468 : (*history.messageList.rbegin())->linearizedParent = sharedCommit->id;
1777 15579 : history.messageList.emplace_back(sharedCommit);
1778 : }
1779 : // Handle pending reactions/editions
1780 17091 : auto reactIt = history.pendingReactions.find(sharedCommit->id);
1781 17091 : if (reactIt != history.pendingReactions.end()) {
1782 0 : for (const auto& commitBody : reactIt->second)
1783 0 : sharedCommit->reactions.emplace_back(commitBody);
1784 0 : history.pendingReactions.erase(reactIt);
1785 : }
1786 17091 : auto peditIt = history.pendingEditions.find(sharedCommit->id);
1787 17091 : if (peditIt != history.pendingEditions.end()) {
1788 0 : std::string toReplace = (sharedCommit->type == CommitType::DATA_TRANSFER) ? CommitKey::TID : CommitKey::BODY;
1789 : // Pending editions are stored newest first: replay them oldest first, as handleEdition would.
1790 0 : for (auto it = peditIt->second.rbegin(); it != peditIt->second.rend(); ++it) {
1791 0 : const auto& editCommit = *it;
1792 : // Retiring a collaborative document carries no body: absent means empty.
1793 0 : std::string body;
1794 0 : if (auto itBody = editCommit->body.find(toReplace); itBody != editCommit->body.end())
1795 0 : body = itBody->second;
1796 0 : auto editionBody = sharedCommit->body;
1797 0 : editionBody["id"] = sharedCommit->latestEditionId.empty() ? sharedCommit->id
1798 0 : : sharedCommit->latestEditionId;
1799 0 : sharedCommit->editions.emplace(sharedCommit->editions.begin(), std::move(editionBody));
1800 0 : sharedCommit->body[toReplace] = body;
1801 0 : sharedCommit->latestEditionId = editCommit->id;
1802 0 : if (toReplace == CommitKey::TID)
1803 0 : sharedCommit->body["fileId"] = "";
1804 0 : if (body.empty())
1805 0 : sharedCommit->reactions.clear();
1806 0 : }
1807 0 : history.pendingEditions.erase(peditIt);
1808 0 : }
1809 : // Announce to client
1810 17091 : if (messageReceived)
1811 1512 : emitSignal<libjami::ConversationSignal::SwarmMessageReceived>(accountId_, repository_->id(), *sharedCommit);
1812 17091 : return !messageReceived;
1813 : }
1814 :
1815 : void
1816 17103 : Conversation::Impl::rectifyStatus(const std::shared_ptr<libjami::SwarmMessage>& message, History& history) const
1817 : {
1818 17103 : auto parentIt = history.quickAccess.find(message->linearizedParent);
1819 17103 : auto currentMessage = message;
1820 :
1821 29253 : while (parentIt != history.quickAccess.end()) {
1822 12150 : const auto& parent = parentIt->second;
1823 12843 : for (const auto& [peer, value] : message->status) {
1824 11646 : auto parentStatusIt = parent->status.find(peer);
1825 11646 : if (parentStatusIt == parent->status.end() || parentStatusIt->second < value) {
1826 693 : parent->status[peer] = value;
1827 1386 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(accountId_,
1828 693 : repository_->id(),
1829 : peer,
1830 693 : parent->id,
1831 : value);
1832 10953 : } else if (parentStatusIt->second >= value) {
1833 10953 : break;
1834 : }
1835 : }
1836 12150 : currentMessage = parent;
1837 12150 : parentIt = history.quickAccess.find(parent->linearizedParent);
1838 : }
1839 17103 : }
1840 :
1841 : void
1842 23 : Conversation::Impl::relinearizeAfterMerge(const std::vector<std::map<std::string, std::string>>& commits)
1843 : {
1844 23 : auto& history = loadedHistory_;
1845 :
1846 23 : std::set<std::string> newIds;
1847 76 : for (const auto& c : commits)
1848 106 : newIds.emplace(c.at(ConversationMapKeys::ID));
1849 23 : if (newIds.empty())
1850 0 : return;
1851 :
1852 : // Walk the log from HEAD, in the order loadMessages() will use, until every
1853 : // announced commit has been seen plus one more commit, the anchor. Everything
1854 : // older than the anchor is left alone, so the anchor must also be older than
1855 : // every message currently shown: commits sharing a timestamp are not guaranteed
1856 : // to keep their relative order once the graph changes.
1857 23 : auto& list = history.messageList;
1858 23 : auto oldIt = list.begin();
1859 23 : std::set<std::string> visited;
1860 23 : std::vector<std::string> order; // newest first, merges excluded
1861 23 : size_t remaining = newIds.size();
1862 23 : bool anchorFound = false;
1863 69 : repository_->log(
1864 46 : [&](const auto& id, const auto&, const auto& commit) {
1865 106 : if (git_commit_parentcount(commit.get()) <= 1)
1866 71 : return CallbackResult::Ok;
1867 : // A batch of merges alone can still reorder commits sharing a timestamp.
1868 35 : if (newIds.erase(id))
1869 29 : --remaining;
1870 35 : return CallbackResult::Skip;
1871 : },
1872 117 : [&](auto&& cc) { order.emplace_back(std::move(cc.id)); },
1873 46 : [&](const auto& id, const auto&, const auto&) {
1874 71 : visited.emplace(id);
1875 71 : if (newIds.erase(id))
1876 24 : --remaining;
1877 71 : if (remaining != 0)
1878 24 : return false;
1879 107 : while (oldIt != list.end() && visited.count((*oldIt)->id))
1880 60 : ++oldIt;
1881 47 : anchorFound = oldIt == list.end() || (*oldIt)->id == id;
1882 47 : return anchorFound;
1883 : },
1884 : "",
1885 : false);
1886 23 : if (remaining != 0)
1887 0 : return; // Should not happen: announced commits are always reachable from HEAD
1888 23 : std::string anchorId;
1889 23 : if (anchorFound) {
1890 23 : anchorId = std::move(order.back());
1891 23 : order.pop_back();
1892 : }
1893 :
1894 : // Only displayed messages are part of the linear chain; reactions and edits are not.
1895 37 : auto isChained = [](const libjami::SwarmMessage& m) {
1896 74 : auto reactIt = m.body.find(CommitKey::REACT_TO);
1897 74 : auto editIt = m.body.find(CommitKey::EDIT);
1898 37 : return (reactIt == m.body.end() || reactIt->second.empty())
1899 37 : && (editIt == m.body.end() || editIt->second.empty());
1900 : };
1901 23 : std::vector<std::shared_ptr<libjami::SwarmMessage>> chain; // oldest first
1902 23 : chain.reserve(order.size());
1903 71 : for (auto it = order.rbegin(); it != order.rend(); ++it) {
1904 48 : auto q = history.quickAccess.find(*it);
1905 48 : if (q != history.quickAccess.end() && isChained(*q->second))
1906 37 : chain.emplace_back(q->second);
1907 : }
1908 23 : if (chain.empty())
1909 11 : return;
1910 :
1911 : // These messages are the newest ones in the history: detach them from the front
1912 : // and put them back in log order.
1913 12 : std::set<const libjami::SwarmMessage*> detaching;
1914 49 : for (const auto& msg : chain)
1915 37 : detaching.emplace(msg.get());
1916 49 : for (auto it = list.begin(); it != list.end() && !detaching.empty();) {
1917 37 : if (detaching.erase(it->get()))
1918 37 : it = list.erase(it);
1919 : else
1920 0 : ++it;
1921 : }
1922 49 : for (const auto& msg : chain)
1923 37 : list.emplace_front(msg);
1924 :
1925 : // Announce the chain from the first message whose predecessor changed, oldest
1926 : // first: a client that places a message right after its parent needs the whole
1927 : // tail replayed, since a message may keep its parent while that parent moves.
1928 12 : std::vector<libjami::SwarmMessage> moved;
1929 12 : const std::string* parentId = &anchorId;
1930 49 : for (const auto& msg : chain) {
1931 37 : if (!moved.empty() || msg->linearizedParent != *parentId) {
1932 16 : msg->linearizedParent = *parentId;
1933 16 : moved.emplace_back(*msg);
1934 : }
1935 37 : parentId = &msg->id;
1936 : }
1937 28 : for (const auto& msg : moved)
1938 16 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, repository_->id(), msg);
1939 67 : }
1940 :
1941 : std::vector<std::shared_ptr<libjami::SwarmMessage>>
1942 17097 : Conversation::Impl::addToHistory(History& history,
1943 : const std::vector<std::map<std::string, std::string>>& commits,
1944 : bool messageReceived,
1945 : bool commitFromSelf)
1946 : {
1947 : //
1948 : // NOTE: This function makes the following assumptions on its arguments:
1949 : // - The messages in "history" are in reverse chronological order (newest message
1950 : // first, oldest message last).
1951 : // - If messageReceived is true, then the commits in "commits" are assumed to be in
1952 : // chronological order (oldest to newest) and to be newer than the ones in "history".
1953 : // They are therefore inserted at the beginning of the message list.
1954 : // - If messageReceived is false, then the commits in "commits" are assumed to be in
1955 : // reverse chronological order (newest to oldest) and to be older than the ones in
1956 : // "history". They are therefore appended at the end of the message list.
1957 : //
1958 17097 : auto acc = account_.lock();
1959 17097 : if (!acc)
1960 0 : return {};
1961 17097 : auto username = acc->getUsername();
1962 17097 : if (messageReceived && (&history == &loadedHistory_ && history.loading)) {
1963 0 : std::unique_lock lk(history.mutex);
1964 0 : history.cv.wait(lk, [&] { return !history.loading; });
1965 0 : }
1966 :
1967 : // Only set messages' status on history for client
1968 17097 : bool needToSetMessageStatus = !commitFromSelf && &history == &loadedHistory_;
1969 :
1970 17097 : std::vector<std::shared_ptr<libjami::SwarmMessage>> sharedCommits;
1971 : // Apply the document removals of this batch before anything else in it.
1972 : //
1973 : // A removal is always newer than the announcement it retires, but the batch is
1974 : // walked oldest first when messages come in and newest first when older ones
1975 : // are paged back in. Acting in batch order would therefore recreate the
1976 : // repository of every deleted document each time the user scrolls up.
1977 34238 : for (const auto& commit : commits) {
1978 17141 : auto typeIt = commit.find(CommitKey::TYPE);
1979 17141 : if (typeIt == commit.end() || typeIt->second != CommitType::COLLAB_DOC)
1980 17137 : continue;
1981 77 : auto editIt = commit.find(CommitKey::EDIT);
1982 77 : if (editIt == commit.end() || editIt->second.empty())
1983 73 : continue;
1984 : // A removal names no document of its own: which one it retires is read from
1985 : // the announcement it edits, the only commit the swarm ties to its author.
1986 : // Otherwise a member could retire somebody else's document.
1987 4 : if (auto announcement = repository_->getCommit(editIt->second)) {
1988 4 : const auto& docId = announcement->commitMsg.uri;
1989 4 : if (!docId.empty())
1990 4 : acc->collaborativeEditing()->onDocumentRemoved(repository_->id(), docId);
1991 4 : }
1992 : }
1993 34238 : for (const auto& commit : commits) {
1994 17141 : auto commitId = commit.at("id");
1995 17141 : if (history.quickAccess.find(commitId) != history.quickAccess.end())
1996 0 : continue; // Already present
1997 17141 : auto typeIt = commit.find(CommitKey::TYPE);
1998 : // Nothing to show for the client, skip
1999 17141 : if (typeIt != commit.end() && typeIt->second == CommitType::MERGE)
2000 38 : continue;
2001 : // A collaborative document is announced here, but its content lives in a
2002 : // separate repository: make sure that repository exists locally so it can
2003 : // be replicated. The commit itself falls through and is displayed like a
2004 : // shared file.
2005 17103 : if (typeIt != commit.end() && typeIt->second == CommitType::COLLAB_DOC) {
2006 : // Removals were applied above, before any announcement of this batch.
2007 77 : auto editIt = commit.find(CommitKey::EDIT);
2008 77 : bool isRemoval = editIt != commit.end() && !editIt->second.empty();
2009 77 : if (!isRemoval) {
2010 146 : if (auto uriIt = commit.find(CommitKey::URI); uriIt != commit.end() && !uriIt->second.empty())
2011 73 : acc->collaborativeEditing()->onDocumentAnnounced(repository_->id(), uriIt->second);
2012 : }
2013 : }
2014 :
2015 17103 : auto sharedCommit = std::make_shared<libjami::SwarmMessage>();
2016 17103 : sharedCommit->fromMapStringString(commit);
2017 :
2018 17103 : if (needToSetMessageStatus) {
2019 1151 : std::lock_guard lk(messageStatusMtx_);
2020 : // Check if we already have status information for the commit.
2021 1151 : auto itFuture = futureStatus.find(sharedCommit->id);
2022 1151 : if (itFuture != futureStatus.end()) {
2023 9 : sharedCommit->status = std::move(itFuture->second);
2024 9 : futureStatus.erase(itFuture);
2025 : }
2026 1151 : }
2027 17103 : sharedCommits.emplace_back(sharedCommit);
2028 17141 : }
2029 :
2030 17097 : if (needToSetMessageStatus) {
2031 1148 : constexpr int32_t SENDING = static_cast<int32_t>(libjami::Account::MessageStates::SENDING);
2032 1148 : constexpr int32_t SENT = static_cast<int32_t>(libjami::Account::MessageStates::SENT);
2033 1148 : constexpr int32_t DISPLAYED = static_cast<int32_t>(libjami::Account::MessageStates::DISPLAYED);
2034 :
2035 1148 : std::lock_guard lk(messageStatusMtx_);
2036 14225 : for (const auto& member : repository_->members()) {
2037 : // For each member, we iterate over the commits to add in reverse chronological
2038 : // order (i.e. from newest to oldest) and set their status from the point of view
2039 : // of that member (as best we can given the information we have).
2040 : //
2041 : // The key assumption made in order to compute the status is that it can never decrease
2042 : // (with respect to the ordering SENDING < SENT < DISPLAYED) as we go back in time. We
2043 : // therefore start by setting the "status" variable below to the lowest possible value,
2044 : // and increase it when we encounter a commit for which it is justified to do so.
2045 : //
2046 : // If messageReceived is true, then the commits we are adding are the most recent in the
2047 : // conversation history, so the lowest possible value is SENDING.
2048 : //
2049 : // If messageReceived is false, then the commits we are adding are older than the ones
2050 : // that are already in the history, so the lowest possible value is the status of the
2051 : // oldest message in the history so far, which is stored in memberToStatus.
2052 13077 : auto status = SENDING;
2053 13077 : if (!messageReceived) {
2054 13 : auto cache = memberToStatus[member.uri];
2055 13 : if (cache > status)
2056 9 : status = cache;
2057 : }
2058 13077 : auto& messagesStatus = messagesStatus_[member.uri];
2059 :
2060 26160 : for (auto it = sharedCommits.rbegin(); it != sharedCommits.rend(); it++) {
2061 13083 : auto sharedCommit = *it;
2062 13083 : auto previousStatus = status;
2063 13083 : auto& commitStatus = sharedCommit->status[member.uri];
2064 :
2065 : // Compute status for the current commit.
2066 39219 : if (status < SENT && messagesStatus["fetched"] == sharedCommit->id) {
2067 10 : status = SENT;
2068 : }
2069 39249 : if (messagesStatus["read"] == sharedCommit->id) {
2070 1 : status = DISPLAYED;
2071 : }
2072 39249 : if (member.uri == sharedCommit->body.at("author")) {
2073 1147 : status = DISPLAYED;
2074 : }
2075 13083 : if (status < commitStatus) {
2076 0 : status = commitStatus;
2077 : }
2078 :
2079 : // Store computed value.
2080 13083 : commitStatus = status;
2081 :
2082 : // Update messagesStatus_ if needed.
2083 13083 : if (previousStatus == SENDING && status >= SENT) {
2084 3435 : messagesStatus["fetched"] = sharedCommit->id;
2085 : }
2086 13083 : if (previousStatus <= SENT && status == DISPLAYED) {
2087 3408 : messagesStatus["read"] = sharedCommit->id;
2088 : }
2089 13083 : }
2090 :
2091 13077 : if (!messageReceived) {
2092 : // Update memberToStatus with the status of the last (i.e. oldest) added commit.
2093 13 : memberToStatus[member.uri] = status;
2094 : }
2095 1148 : }
2096 1148 : }
2097 :
2098 17097 : std::vector<std::shared_ptr<libjami::SwarmMessage>> messages;
2099 34200 : for (const auto& sharedCommit : sharedCommits) {
2100 17103 : history.quickAccess[sharedCommit->id] = sharedCommit;
2101 :
2102 34206 : auto reactToIt = sharedCommit->body.find(CommitKey::REACT_TO);
2103 34206 : auto editIt = sharedCommit->body.find(CommitKey::EDIT);
2104 17103 : if (reactToIt != sharedCommit->body.end() && !reactToIt->second.empty()) {
2105 3 : handleReaction(history, sharedCommit);
2106 17100 : } else if (editIt != sharedCommit->body.end() && !editIt->second.empty()) {
2107 9 : handleEdition(history, sharedCommit, messageReceived);
2108 17091 : } else if (handleMessage(history, sharedCommit, messageReceived)) {
2109 15579 : messages.emplace_back(sharedCommit);
2110 : }
2111 17103 : rectifyStatus(sharedCommit, history);
2112 : }
2113 :
2114 17097 : return messages;
2115 17097 : }
2116 :
2117 430 : Conversation::Conversation(const std::shared_ptr<JamiAccount>& account,
2118 : ConversationMode mode,
2119 430 : const std::string& otherMember)
2120 430 : : pimpl_ {new Impl {account, mode, otherMember}}
2121 430 : {}
2122 :
2123 50 : Conversation::Conversation(const std::shared_ptr<JamiAccount>& account, const std::string& conversationId)
2124 50 : : pimpl_ {new Impl {account, conversationId}}
2125 50 : {}
2126 :
2127 458 : Conversation::Conversation(const std::shared_ptr<JamiAccount>& account,
2128 : const std::string& remoteDevice,
2129 458 : const std::string& conversationId)
2130 458 : : pimpl_ {new Impl {account, remoteDevice, conversationId}}
2131 458 : {}
2132 :
2133 : #ifdef LIBJAMI_TEST
2134 0 : Conversation::Conversation(std::unique_ptr<ConversationRepository>&& repository,
2135 : const std::shared_ptr<JamiAccount>& account,
2136 0 : std::vector<ConversationCommit>&& commits)
2137 0 : : pimpl_ {new Impl {std::move(repository), account, std::move(commits)}}
2138 0 : {}
2139 : #endif
2140 :
2141 907 : Conversation::~Conversation() {}
2142 :
2143 : std::string
2144 10691 : Conversation::id() const
2145 : {
2146 10691 : return pimpl_->repository_ ? pimpl_->repository_->id() : "";
2147 : }
2148 :
2149 : void
2150 200 : Conversation::addMember(const std::string& contactUri, const OnDoneCb& cb)
2151 : {
2152 : try {
2153 200 : if (mode() == ConversationMode::ONE_TO_ONE) {
2154 : // Only authorize to add left members
2155 1 : auto initialMembers = getInitialMembers();
2156 1 : auto it = std::find(initialMembers.begin(), initialMembers.end(), contactUri);
2157 1 : if (it == initialMembers.end()) {
2158 1 : JAMI_WARNING("Unable to add new member in one to one conversation");
2159 1 : cb(false, "");
2160 1 : return;
2161 : }
2162 1 : }
2163 0 : } catch (const std::exception& e) {
2164 0 : JAMI_WARNING("Unable to get mode: {}", e.what());
2165 0 : cb(false, "");
2166 0 : return;
2167 0 : }
2168 199 : if (isMember(contactUri, true)) {
2169 0 : JAMI_WARNING("Unable to add member {} because it's already a member", contactUri);
2170 0 : cb(false, "");
2171 0 : return;
2172 : }
2173 199 : if (isMemberBanned(contactUri)) {
2174 2 : if (pimpl_->isAdmin()) {
2175 2 : dht::ThreadPool::io().run([w = weak(), contactUri = std::move(contactUri), cb = std::move(cb)] {
2176 2 : if (auto sthis = w.lock()) {
2177 2 : auto members = sthis->pimpl_->repository_->members();
2178 2 : auto type = sthis->pimpl_->memberBanType(contactUri);
2179 2 : if (type.empty()) {
2180 0 : cb(false, {});
2181 0 : return;
2182 : }
2183 2 : sthis->pimpl_->voteUnban(contactUri, type, cb);
2184 4 : }
2185 : });
2186 : } else {
2187 0 : JAMI_WARNING("Unable to add member {} because this member is blocked", contactUri);
2188 0 : cb(false, "");
2189 : }
2190 2 : return;
2191 : }
2192 :
2193 197 : dht::ThreadPool::io().run([w = weak(), contactUri = std::move(contactUri), cb = std::move(cb)] {
2194 197 : if (auto sthis = w.lock()) {
2195 : // Add member files and commit
2196 197 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
2197 197 : auto commit = sthis->pimpl_->repository_->addMember(contactUri);
2198 197 : if (not commit.empty())
2199 195 : sthis->pimpl_->announce(commit, true);
2200 197 : lk.unlock();
2201 197 : if (cb)
2202 197 : cb(!commit.empty(), commit);
2203 394 : }
2204 197 : });
2205 : }
2206 :
2207 : std::shared_ptr<dhtnet::ChannelSocket>
2208 5230 : Conversation::gitSocket(const DeviceId& deviceId) const
2209 : {
2210 5230 : return pimpl_->gitSocket(deviceId);
2211 : }
2212 :
2213 : void
2214 2284 : Conversation::addGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& socket)
2215 : {
2216 2284 : pimpl_->addGitSocket(deviceId, socket);
2217 2284 : }
2218 :
2219 : void
2220 1445 : Conversation::removeGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& expected)
2221 : {
2222 1445 : pimpl_->removeGitSocket(deviceId, expected);
2223 1445 : }
2224 :
2225 : void
2226 944 : Conversation::shutdownConnections()
2227 : {
2228 944 : GitSocketList gitSockets;
2229 : {
2230 944 : std::lock_guard lk(pimpl_->gitSocketMtx_);
2231 944 : gitSockets = std::move(pimpl_->gitSocketList_);
2232 944 : pimpl_->gitSocketList_.clear();
2233 944 : }
2234 : // Closing the channels wakes up any fetch currently blocked reading from them.
2235 944 : gitSockets.clear();
2236 944 : if (pimpl_->swarmManager_)
2237 944 : pimpl_->swarmManager_->shutdown();
2238 944 : }
2239 :
2240 : void
2241 0 : Conversation::connectivityChanged()
2242 : {
2243 0 : if (pimpl_->swarmManager_)
2244 0 : pimpl_->swarmManager_->maintainBuckets();
2245 0 : }
2246 :
2247 : std::vector<jami::DeviceId>
2248 0 : Conversation::getDeviceIdList() const
2249 : {
2250 0 : return pimpl_->swarmManager_->getAllNodes();
2251 : }
2252 :
2253 : std::shared_ptr<Typers>
2254 9 : Conversation::typers() const
2255 : {
2256 9 : return pimpl_->typers_;
2257 : }
2258 :
2259 : std::vector<std::map<std::string, std::string>>
2260 0 : Conversation::getConnectivity() const
2261 : {
2262 0 : return pimpl_->getConnectivity();
2263 : }
2264 :
2265 : void
2266 2 : Conversation::Impl::voteUnban(const std::string& contactUri, const std::string_view type, const OnDoneCb& cb)
2267 : {
2268 : // Check if admin
2269 2 : if (!isAdmin()) {
2270 0 : JAMI_WARNING("You're not an admin of this repo. Unable to unblock {}", contactUri);
2271 0 : cb(false, {});
2272 0 : return;
2273 : }
2274 :
2275 : // Vote for removal
2276 2 : std::unique_lock lk(writeMtx_);
2277 2 : auto voteCommit = repository_->voteUnban(contactUri, type);
2278 2 : if (voteCommit.empty()) {
2279 0 : JAMI_WARNING("Unbanning {} failed", contactUri);
2280 0 : cb(false, "");
2281 0 : return;
2282 : }
2283 :
2284 2 : auto lastId = voteCommit;
2285 2 : std::vector<std::string> commits;
2286 2 : commits.emplace_back(voteCommit);
2287 :
2288 : // If admin, check vote
2289 4 : auto resolveCommit = repository_->resolveVote(contactUri, type, CommitAction::UNBAN);
2290 2 : if (!resolveCommit.empty()) {
2291 2 : commits.emplace_back(resolveCommit);
2292 2 : lastId = resolveCommit;
2293 2 : JAMI_WARNING("Vote solved for unbanning {}.", contactUri);
2294 : }
2295 2 : announce(commits, true);
2296 2 : lk.unlock();
2297 2 : if (cb)
2298 2 : cb(!lastId.empty(), lastId);
2299 2 : }
2300 :
2301 : void
2302 16 : Conversation::removeMember(const std::string& contactUri, bool isDevice, const OnDoneCb& cb)
2303 : {
2304 32 : dht::ThreadPool::io().run(
2305 32 : [w = weak(), contactUri = std::move(contactUri), isDevice = std::move(isDevice), cb = std::move(cb)] {
2306 16 : if (auto sthis = w.lock()) {
2307 : // Check if admin
2308 16 : if (!sthis->pimpl_->isAdmin()) {
2309 1 : JAMI_WARNING("You're not an admin of this repo. Unable to block {}", contactUri);
2310 1 : cb(false, {});
2311 2 : return;
2312 : }
2313 :
2314 : // Get current user type
2315 15 : std::string type;
2316 15 : if (isDevice) {
2317 1 : type = "devices";
2318 : } else {
2319 14 : auto members = sthis->pimpl_->repository_->members();
2320 31 : for (const auto& member : members) {
2321 31 : if (member.uri == contactUri) {
2322 14 : if (member.role == MemberRole::INVITED) {
2323 2 : type = "invited";
2324 12 : } else if (member.role == MemberRole::ADMIN) {
2325 1 : type = "admins";
2326 11 : } else if (member.role == MemberRole::MEMBER) {
2327 11 : type = "members";
2328 : }
2329 14 : break;
2330 : }
2331 : }
2332 14 : if (type.empty()) {
2333 0 : cb(false, {});
2334 0 : return;
2335 : }
2336 14 : }
2337 :
2338 : // Vote for removal
2339 15 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
2340 15 : auto voteCommit = sthis->pimpl_->repository_->voteKick(contactUri, type);
2341 15 : if (voteCommit.empty()) {
2342 1 : JAMI_WARNING("Kicking {} failed", contactUri);
2343 2 : cb(false, "");
2344 1 : return;
2345 : }
2346 :
2347 14 : auto lastId = voteCommit;
2348 14 : std::vector<std::string> commits;
2349 14 : commits.emplace_back(voteCommit);
2350 :
2351 : // If admin, check vote
2352 28 : auto resolveCommit = sthis->pimpl_->repository_->resolveVote(contactUri, type, CommitAction::BAN);
2353 14 : if (!resolveCommit.empty()) {
2354 14 : commits.emplace_back(resolveCommit);
2355 14 : lastId = resolveCommit;
2356 14 : JAMI_WARNING("Vote solved for {}. {} banned", contactUri, isDevice ? "Device" : "Member");
2357 14 : if (isDevice)
2358 1 : sthis->pimpl_->disconnectFromDevice(DeviceId(contactUri));
2359 : else
2360 13 : sthis->pimpl_->disconnectFromPeer(contactUri);
2361 : }
2362 :
2363 14 : sthis->pimpl_->announce(commits, true);
2364 14 : lk.unlock();
2365 14 : cb(!lastId.empty(), lastId);
2366 33 : }
2367 : });
2368 16 : }
2369 :
2370 : std::vector<std::map<std::string, std::string>>
2371 287 : Conversation::getMembers(bool includeInvited, bool includeLeft, bool includeBanned) const
2372 : {
2373 287 : return pimpl_->getMembers(includeInvited, includeLeft, includeBanned);
2374 : }
2375 :
2376 : std::vector<std::map<std::string, std::string>>
2377 1 : Conversation::getTrackedMembers() const
2378 : {
2379 1 : return pimpl_->getTrackedMembers();
2380 : }
2381 :
2382 : std::set<std::string>
2383 4940 : Conversation::memberUris(std::string_view filter, const std::set<MemberRole>& filteredRoles) const
2384 : {
2385 4940 : return pimpl_->repository_->memberUris(filter, filteredRoles);
2386 : }
2387 :
2388 : std::vector<NodeId>
2389 2510 : Conversation::peersToSyncWith() const
2390 : {
2391 2510 : auto s = pimpl_->swarmManager_->getConnectedNodes();
2392 2510 : std::lock_guard lk(pimpl_->gitSocketMtx_);
2393 15007 : for (const auto& [deviceId, _] : pimpl_->gitSocketList_)
2394 12497 : if (std::find(s.cbegin(), s.cend(), deviceId) == s.cend())
2395 693 : s.emplace_back(deviceId);
2396 5020 : return s;
2397 2510 : }
2398 :
2399 : std::vector<MobileNodeTarget>
2400 2460 : Conversation::mobileNodesToNotify() const
2401 : {
2402 2460 : std::vector<MobileNodeTarget> targets;
2403 2460 : auto account = pimpl_->account_.lock();
2404 2460 : if (!account)
2405 0 : return targets;
2406 2460 : for (const auto& info : pimpl_->swarmManager_->getMobileNodeInfosToNotify()) {
2407 0 : const auto deviceId = info.id.toString();
2408 0 : auto uri = uriFromDevice(deviceId);
2409 : // A verified lease binds the device to its issuer, so the account URI is
2410 : // already known without touching the certificate.
2411 0 : if (uri.empty() && info.lease)
2412 0 : uri = info.lease->issuer_id.toString();
2413 0 : if (uri.empty()) {
2414 0 : auto cert = account->certStore().getCertificate(deviceId);
2415 0 : if (cert && cert->issuer)
2416 0 : uri = cert->issuer->getId().toString();
2417 0 : }
2418 0 : if (!uri.empty() && isPeerAuthorized(uri, deviceId, false))
2419 0 : targets.emplace_back(MobileNodeTarget {info.id, std::move(uri)});
2420 0 : }
2421 2460 : return targets;
2422 2460 : }
2423 :
2424 : bool
2425 2460 : Conversation::isBootstrapped() const
2426 : {
2427 2460 : return pimpl_->swarmManager_->isConnected();
2428 : }
2429 :
2430 : std::string
2431 14874 : Conversation::uriFromDevice(const std::string& deviceId) const
2432 : {
2433 14874 : return pimpl_->repository_->uriFromDevice(deviceId);
2434 : }
2435 :
2436 : std::map<std::string, std::vector<DeviceId>>
2437 28 : Conversation::memberDevices() const
2438 : {
2439 28 : return pimpl_->repository_->devices();
2440 : }
2441 :
2442 : void
2443 0 : Conversation::monitor()
2444 : {
2445 0 : pimpl_->swarmManager_->getRoutingTable().printRoutingTable();
2446 0 : }
2447 :
2448 : std::string
2449 424 : Conversation::join()
2450 : {
2451 424 : return pimpl_->repository_->join();
2452 : }
2453 :
2454 : bool
2455 16022 : Conversation::isMember(const std::string& uri, bool includeInvited) const
2456 : {
2457 16022 : auto uriCrt = uri + ".crt"sv;
2458 32044 : if (std::filesystem::is_regular_file(pimpl_->repoPath_ / MemberPath::ADMINS / uriCrt)
2459 32044 : || std::filesystem::is_regular_file(pimpl_->repoPath_ / MemberPath::MEMBERS / uriCrt)) {
2460 10002 : return true;
2461 : }
2462 6020 : if (includeInvited) {
2463 3047 : if (std::filesystem::is_regular_file(pimpl_->repoPath_ / MemberPath::INVITED / uri)) {
2464 2622 : return true;
2465 : }
2466 425 : if (mode() == ConversationMode::ONE_TO_ONE) {
2467 3 : for (const auto& member : getInitialMembers()) {
2468 2 : if (member == uri)
2469 0 : return true;
2470 1 : }
2471 : }
2472 : }
2473 3398 : return false;
2474 16022 : }
2475 :
2476 : bool
2477 8631 : Conversation::isMemberBanned(const std::string& uri) const
2478 : {
2479 8631 : return !pimpl_->memberBanType(uri).empty();
2480 : }
2481 :
2482 : bool
2483 9571 : Conversation::isDeviceBanned(const std::string& deviceId) const
2484 : {
2485 9571 : return pimpl_->isDeviceBanned(deviceId);
2486 : }
2487 :
2488 : std::vector<std::string>
2489 4 : Conversation::permissions(const std::string& uri) const
2490 : {
2491 4 : return pimpl_->repository_->permissions(uri);
2492 : }
2493 :
2494 : bool
2495 8380 : Conversation::isPeerAuthorized(const std::string& uri, const std::string& deviceId, bool includeInvited) const
2496 : {
2497 8380 : return !isMemberBanned(uri) && !isDeviceBanned(deviceId) && isMember(uri, includeInvited);
2498 : }
2499 :
2500 : bool
2501 511 : Conversation::isPeerCertificateAuthorized(const dht::crypto::Certificate& certificate) const
2502 : {
2503 511 : if (!certificate.issuer)
2504 0 : return false;
2505 :
2506 511 : const auto memberUri = certificate.issuer->getId().toString();
2507 511 : const auto deviceId = certificate.getLongId().toString();
2508 511 : return isPeerAuthorized(memberUri, deviceId, false)
2509 511 : && pimpl_->repository_->isDeviceCertificateFromMember(certificate);
2510 511 : }
2511 :
2512 : void
2513 185 : Conversation::createCommit(CommitMessage&& message, OnCommitCb&& onCommit, OnDoneCb&& cb)
2514 : {
2515 185 : if (!message.replyTo.empty()) {
2516 2 : if (!pimpl_->repository_->hasCommit(message.replyTo)) {
2517 1 : JAMI_ERROR("Replying to invalid commit {}", message.replyTo);
2518 1 : return;
2519 : }
2520 : }
2521 368 : dht::ThreadPool::io().run(
2522 368 : [w = weak(), message = std::move(message), onCommit = std::move(onCommit), cb = std::move(cb)] {
2523 184 : if (auto sthis = w.lock()) {
2524 184 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
2525 184 : auto commit = sthis->pimpl_->repository_->commitMessage(message.toString());
2526 184 : lk.unlock();
2527 184 : if (onCommit)
2528 15 : onCommit(commit);
2529 184 : sthis->pimpl_->announce(commit, true);
2530 184 : if (cb)
2531 184 : cb(!commit.empty(), commit);
2532 368 : }
2533 184 : });
2534 : }
2535 :
2536 : bool
2537 12127 : Conversation::hasCommit(const std::string& commitId) const
2538 : {
2539 12127 : return pimpl_->repository_->hasCommit(commitId);
2540 : }
2541 :
2542 : std::optional<ConversationCommit>
2543 57 : Conversation::getCommit(const std::string& commitId) const
2544 : {
2545 57 : return pimpl_->repository_->getCommit(commitId);
2546 : }
2547 :
2548 : namespace {
2549 : /**
2550 : * Announcement commit ids retired by a removal, from a full conversation log.
2551 : *
2552 : * Which document a removal retires is read from the announcement it edits, never from
2553 : * the removal itself: the swarm only checks that an edition carries the author of the
2554 : * commit it edits, so trusting an id carried by the removal would let a member retire
2555 : * a document somebody else created.
2556 : */
2557 : std::set<std::string>
2558 1148 : retiredAnnouncements(const std::vector<std::map<std::string, std::string>>& commits)
2559 : {
2560 1148 : std::set<std::string> retired;
2561 4077 : for (const auto& commit : commits) {
2562 2929 : auto typeIt = commit.find(CommitKey::TYPE);
2563 2929 : if (typeIt == commit.end() || typeIt->second != CommitType::COLLAB_DOC)
2564 2819 : continue;
2565 220 : if (auto editIt = commit.find(CommitKey::EDIT); editIt != commit.end() && !editIt->second.empty())
2566 4 : retired.emplace(editIt->second);
2567 : }
2568 1148 : return retired;
2569 0 : }
2570 : } // namespace
2571 :
2572 : std::vector<std::map<std::string, std::string>>
2573 1148 : Conversation::collaborativeDocuments() const
2574 : {
2575 1148 : if (!pimpl_->repository_)
2576 0 : return {};
2577 : // Read straight from git so documents are found even when their announcing commit
2578 : // is not (or no longer) in the loaded message window.
2579 1148 : LogOptions options;
2580 1148 : options.skipMerge = true;
2581 1148 : auto commits = pimpl_->repository_->convCommitsToMap(pimpl_->repository_->log(options));
2582 1148 : auto retired = retiredAnnouncements(commits);
2583 1148 : std::vector<std::map<std::string, std::string>> result;
2584 4077 : for (auto& commit : commits) {
2585 2929 : auto typeIt = commit.find(CommitKey::TYPE);
2586 2929 : if (typeIt == commit.end() || typeIt->second != CommitType::COLLAB_DOC)
2587 2827 : continue;
2588 110 : auto uriIt = commit.find(CommitKey::URI);
2589 110 : if (uriIt == commit.end() || uriIt->second.empty())
2590 4 : continue;
2591 106 : auto idIt = commit.find("id");
2592 106 : if (idIt != commit.end() && retired.count(idIt->second) != 0)
2593 4 : continue;
2594 : // A document is addressed by its own id everywhere -- opening, renaming,
2595 : // removing -- so that is what "id" carries here; the commit's own id only
2596 : // says which timeline interaction announced it.
2597 816 : result.push_back({{"id", uriIt->second},
2598 204 : {"announcement", idIt != commit.end() ? idIt->second : ""},
2599 204 : {"displayName", commit[CommitKey::DISPLAY_NAME]},
2600 204 : {"mimeType", commit[CommitKey::MIME_TYPE]},
2601 204 : {"author", commit["author"]},
2602 204 : {"timestamp", commit["timestamp"]}});
2603 : }
2604 1148 : return result;
2605 1760 : }
2606 :
2607 : void
2608 3 : Conversation::loadMessages(const OnLoadMessages& cb, const LogOptions& options)
2609 : {
2610 3 : if (!cb)
2611 0 : return;
2612 3 : dht::ThreadPool::io().run([w = weak(), cb = std::move(cb), options] {
2613 3 : if (auto sthis = w.lock()) {
2614 3 : std::unique_lock lk(sthis->pimpl_->loadedHistory_.mutex);
2615 3 : auto result = sthis->pimpl_->loadMessages(options);
2616 3 : lk.unlock();
2617 3 : cb(std::move(result));
2618 6 : }
2619 3 : });
2620 : }
2621 :
2622 : void
2623 4 : Conversation::Impl::loadMissingBodyOverwrites()
2624 : {
2625 : #ifdef ENABLE_PLUGIN
2626 4 : auto& pluginChatManager = Manager::instance().getJamiPluginManager().getChatServicesManager();
2627 4 : if (!pluginChatManager.hasHandlers()) {
2628 4 : return;
2629 : }
2630 :
2631 1 : const auto& convId = repository_->id();
2632 :
2633 : // Collect all TEXT commits (originals and editions) missing a bodyOverwrite.
2634 1 : std::vector<libjami::SwarmMessage> batch;
2635 : {
2636 1 : std::lock_guard const lk(loadedHistory_.mutex);
2637 3 : for (const auto& msg : loadedHistory_.quickAccess | std::views::values) {
2638 2 : if (msg->type == CommitType::TEXT
2639 6 : && ((msg->editions.empty() && !msg->pluginData.contains("bodyOverwrite"))
2640 2 : || (!msg->editions.empty() && !msg->editions.back().contains("bodyOverwrite")))) {
2641 0 : batch.push_back(*msg);
2642 : }
2643 : }
2644 1 : }
2645 :
2646 1 : if (batch.empty()) {
2647 1 : return;
2648 : }
2649 :
2650 : // Single plugin call for all messages at once.
2651 0 : pluginChatManager.transformSwarmMessages(batch, accountId_, convId);
2652 :
2653 : // Write results back, deduplicate by signal-message id (multiple editions → same original).
2654 0 : std::map<std::string, libjami::SwarmMessage> signalMap;
2655 : {
2656 0 : std::lock_guard const lk(loadedHistory_.mutex);
2657 0 : for (const auto& transformed : batch) {
2658 0 : const auto boIt = transformed.pluginData.find("bodyOverwrite");
2659 0 : if (boIt == transformed.pluginData.end() || boIt->second.empty()) {
2660 0 : if (const auto it = loadedHistory_.quickAccess.find(transformed.id);
2661 0 : it != loadedHistory_.quickAccess.end()) {
2662 0 : it->second->pluginData["bodyOverwrite"] = "";
2663 : }
2664 0 : continue;
2665 0 : }
2666 0 : if (auto signalMsg = injectEditionOverwrites(transformed.id, boIt->second); !signalMsg.id.empty()) {
2667 0 : signalMap[signalMsg.id] = std::move(signalMsg);
2668 0 : }
2669 : }
2670 0 : }
2671 0 : for (const auto& msg : signalMap | std::views::values) {
2672 0 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, convId, msg);
2673 : }
2674 : #endif
2675 1 : }
2676 :
2677 : void
2678 4 : Conversation::loadMissingBodyOverwrites() const
2679 : {
2680 4 : pimpl_->loadMissingBodyOverwrites();
2681 4 : }
2682 :
2683 : void
2684 0 : Conversation::Impl::reloadBodyOverwriteMessages()
2685 : {
2686 : #ifdef ENABLE_PLUGIN
2687 : // Clear all bodyOverwrite* entries so loadMissingBodyOverwrites() reload all body overwrites.
2688 : {
2689 0 : std::lock_guard const lk(loadedHistory_.mutex);
2690 0 : for (const auto& msg : loadedHistory_.quickAccess | std::views::values) {
2691 0 : if (msg->editions.empty()) {
2692 0 : msg->pluginData.erase("bodyOverwrite");
2693 : } else {
2694 0 : msg->pluginData.erase("bodyOverwrite");
2695 0 : for (auto& edition : msg->editions) {
2696 0 : edition.erase("bodyOverwrite");
2697 : }
2698 : }
2699 : }
2700 0 : }
2701 0 : loadMissingBodyOverwrites();
2702 : #endif
2703 0 : }
2704 :
2705 : void
2706 0 : Conversation::reloadBodyOverwriteMessages() const
2707 : {
2708 0 : pimpl_->reloadBodyOverwriteMessages();
2709 0 : }
2710 :
2711 : void
2712 0 : Conversation::Impl::updateMessageBodyOverwrite(const std::string& messageId, const std::string_view bodyOverwrite)
2713 : {
2714 : #ifdef ENABLE_PLUGIN
2715 0 : const auto& convId = repository_->id();
2716 0 : libjami::SwarmMessage signalMsg;
2717 : {
2718 0 : std::lock_guard const lk(loadedHistory_.mutex);
2719 0 : signalMsg = injectEditionOverwrites(messageId, bodyOverwrite);
2720 0 : }
2721 0 : if (!signalMsg.id.empty()) {
2722 0 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, convId, signalMsg);
2723 : }
2724 : #endif
2725 0 : }
2726 :
2727 : void
2728 0 : Conversation::updateMessageBodyOverwrite(const std::string& messageId, const std::string& bodyOverwrite) const
2729 : {
2730 0 : pimpl_->updateMessageBodyOverwrite(messageId, bodyOverwrite);
2731 0 : }
2732 :
2733 : void
2734 1 : Conversation::Impl::clearBodyOverwrites()
2735 : {
2736 : #ifdef ENABLE_PLUGIN
2737 1 : const auto convId = repository_->id();
2738 1 : std::map<std::string, libjami::SwarmMessage> signalMap;
2739 : {
2740 1 : std::lock_guard const lk(loadedHistory_.mutex);
2741 : // Track which message ids need a signal: their own overwrite was cleared,
2742 : // or one of their edition commits' overwrite was cleared.
2743 3 : for (const auto& [id, msg] : loadedHistory_.quickAccess) {
2744 2 : if (msg->editions.empty()) {
2745 4 : msg->pluginData.erase("bodyOverwrite");
2746 6 : if (!msg->body.contains(CommitKey::EDIT)) {
2747 2 : signalMap[id] = *msg;
2748 : }
2749 : } else {
2750 0 : msg->pluginData.erase("bodyOverwrite");
2751 0 : for (auto& edition : msg->editions) {
2752 0 : edition.erase("bodyOverwrite");
2753 : }
2754 0 : signalMap[id] = *msg;
2755 : }
2756 : }
2757 1 : }
2758 3 : for (const auto& msg : signalMap | std::views::values) {
2759 2 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, convId, msg);
2760 : }
2761 : #endif
2762 1 : }
2763 :
2764 : void
2765 1 : Conversation::clearBodyOverwrites() const
2766 : {
2767 1 : pimpl_->clearBodyOverwrites();
2768 1 : }
2769 :
2770 : void
2771 0 : Conversation::clearCache()
2772 : {
2773 0 : std::lock_guard lk(pimpl_->loadedHistory_.mutex);
2774 0 : pimpl_->loadedHistory_.messageList.clear();
2775 0 : pimpl_->loadedHistory_.quickAccess.clear();
2776 0 : pimpl_->loadedHistory_.pendingEditions.clear();
2777 0 : pimpl_->loadedHistory_.pendingReactions.clear();
2778 : {
2779 0 : std::lock_guard lk(pimpl_->messageStatusMtx_);
2780 0 : pimpl_->memberToStatus.clear();
2781 0 : }
2782 0 : }
2783 :
2784 : std::string
2785 2720 : Conversation::lastCommitId() const
2786 : {
2787 : {
2788 2720 : std::lock_guard lk(pimpl_->loadedHistory_.mutex);
2789 2720 : if (!pimpl_->loadedHistory_.messageList.empty())
2790 1524 : return (*pimpl_->loadedHistory_.messageList.begin())->id;
2791 2720 : }
2792 1196 : LogOptions options;
2793 1196 : options.nbOfCommits = 1;
2794 1196 : options.skipMerge = true;
2795 1196 : History optHistory;
2796 1196 : std::scoped_lock lock(pimpl_->writeMtx_, optHistory.mutex);
2797 1196 : auto res = pimpl_->loadMessages(options, &optHistory);
2798 1196 : if (res.empty())
2799 1 : return {};
2800 1195 : return (*optHistory.messageList.begin())->id;
2801 1196 : }
2802 :
2803 : bool
2804 1861 : Conversation::pull(const std::string& deviceId, OnPullCb&& cb, std::string commitId)
2805 : {
2806 1861 : std::lock_guard lk(pimpl_->pullcbsMtx_);
2807 5583 : auto [it, notInProgress] = pimpl_->fetchingRemotes_.emplace(deviceId,
2808 3722 : std::deque<std::pair<std::string, OnPullCb>>());
2809 1861 : auto& pullcbs = it->second;
2810 0 : auto itPull = std::find_if(pullcbs.begin(), pullcbs.end(), [&](const auto& elem) {
2811 0 : return std::get<0>(elem) == commitId;
2812 3722 : });
2813 1861 : if (itPull != pullcbs.end()) {
2814 0 : JAMI_DEBUG("{} Ignoring request to pull from {:s} with commit {:s}: pull already in progress",
2815 : pimpl_->toString(),
2816 : deviceId,
2817 : commitId);
2818 0 : cb(false);
2819 0 : return false;
2820 : }
2821 1861 : JAMI_DEBUG("{} [device {}] Pulling '{:s}'", pimpl_->toString(), deviceId, commitId);
2822 1861 : pullcbs.emplace_back(std::move(commitId), std::move(cb));
2823 1861 : if (notInProgress)
2824 1854 : dht::ThreadPool::io().run([w = weak(), deviceId] {
2825 1854 : if (auto sthis_ = w.lock())
2826 1854 : sthis_->pimpl_->pull(deviceId);
2827 1854 : });
2828 1861 : return true;
2829 1861 : }
2830 :
2831 : void
2832 1854 : Conversation::Impl::pull(const std::string& deviceId)
2833 : {
2834 1854 : auto& repo = repository_;
2835 :
2836 1854 : std::string commitId;
2837 1854 : OnPullCb cb;
2838 : while (true) {
2839 : {
2840 3715 : std::lock_guard lk(pullcbsMtx_);
2841 3715 : auto it = fetchingRemotes_.find(deviceId);
2842 3715 : if (it == fetchingRemotes_.end()) {
2843 0 : JAMI_ERROR("Could not find device {:s} in fetchingRemotes", deviceId);
2844 0 : break;
2845 : }
2846 3715 : auto& pullcbs = it->second;
2847 3715 : if (pullcbs.empty()) {
2848 1854 : fetchingRemotes_.erase(it);
2849 1854 : break;
2850 : }
2851 1861 : auto& elem = pullcbs.front();
2852 1861 : commitId = std::move(std::get<0>(elem));
2853 1861 : cb = std::move(std::get<1>(elem));
2854 1861 : pullcbs.pop_front();
2855 3715 : }
2856 : // If recently fetched, the commit can already be there, so no need to do complex operations
2857 1861 : if (commitId != "" && repo->hasCommit(commitId)) {
2858 1 : cb(true);
2859 52 : continue;
2860 : }
2861 : // Pull from remote
2862 1860 : auto fetched = repo->fetch(deviceId);
2863 1860 : if (!fetched) {
2864 51 : cb(false);
2865 51 : continue;
2866 : }
2867 :
2868 1809 : auto oldHead = repo->getHead();
2869 :
2870 1809 : std::unique_lock lk(writeMtx_);
2871 : auto commits = repo->mergeHistory(deviceId,
2872 1812 : [this](const std::string& peerUri) { this->disconnectFromPeer(peerUri); });
2873 1809 : if (!commits.empty()) {
2874 1164 : announce(commits);
2875 : }
2876 1809 : lk.unlock();
2877 :
2878 1809 : bool commitFound = false;
2879 1809 : if (commitId != "") {
2880 : // If `commitId` is non-empty, then we were attempting to pull a specific commit.
2881 : // We need to check if we actually got it; the fact that the fetch above was
2882 : // successful doesn't guarantee that we did.
2883 1181 : for (const auto& commit : commits) {
2884 2328 : if (commit.at("id") == commitId) {
2885 1156 : commitFound = true;
2886 1156 : break;
2887 : }
2888 : }
2889 : } else {
2890 636 : commitFound = true;
2891 : }
2892 1809 : if (!commitFound)
2893 17 : JAMI_WARNING("Successfully fetched from device {} but didn't receive expected commit {}",
2894 : deviceId,
2895 : commitId);
2896 : // WARNING: If its argument is `true`, this callback will attempt to send a message notification
2897 : // for commit `commitId` to other members of the swarm. It's important that we only
2898 : // send these notifications if we actually have the commit. Otherwise, we can end up
2899 : // in a situation where the members of the swarm keep sending notifications to each
2900 : // other for a commit that none of them have (note that we are unable to rule this out, as
2901 : // nothing prevents a malicious user from intentionally sending a notification with
2902 : // a fake commit ID).
2903 1809 : if (cb)
2904 1809 : cb(commitFound);
2905 :
2906 : // Announce if profile changed
2907 1809 : if (!commits.empty()) {
2908 2328 : auto diffStats = repo->diffStats("HEAD", oldHead);
2909 1164 : auto changedFiles = repo->changedFiles(diffStats);
2910 1164 : if (find(changedFiles.begin(), changedFiles.end(), "profile.vcf") != changedFiles.end()) {
2911 7 : emitSignal<libjami::ConversationSignal::ConversationProfileUpdated>(accountId_,
2912 7 : repo->id(),
2913 14 : repo->infos());
2914 : }
2915 1164 : }
2916 3670 : }
2917 : // Merge the packs the fetches stored, if they are too many. This runs once
2918 : // the pulls are done, and without any lock, so that nobody waits for it.
2919 1854 : repo->repack();
2920 1854 : }
2921 :
2922 : void
2923 1860 : Conversation::sync(const std::string& member, const std::string& deviceId, OnPullCb&& cb, std::string commitId)
2924 : {
2925 1860 : pull(deviceId, std::move(cb), commitId);
2926 1860 : dht::ThreadPool::io().run([member, deviceId, w = weak_from_this()] {
2927 1860 : auto sthis = w.lock();
2928 1860 : if (!sthis)
2929 0 : return;
2930 : // For waiting request, downloadFile
2931 1861 : for (const auto& wr : sthis->dataTransfer()->waitingRequests()) {
2932 1 : sthis->downloadFile(wr.interactionId, wr.fileId, wr.path, member, deviceId);
2933 1860 : }
2934 1860 : });
2935 1860 : }
2936 :
2937 : std::map<std::string, std::string>
2938 330 : Conversation::generateInvitation(TimePoint sent) const
2939 : {
2940 : // Invite the new member to the conversation
2941 330 : Json::Value root;
2942 330 : auto& metadata = root[ConversationMapKeys::METADATAS];
2943 666 : for (const auto& [k, v] : infos()) {
2944 336 : if (v.size() >= 64000) {
2945 0 : JAMI_WARNING("Cutting invite because the SIP message will be too long");
2946 0 : continue;
2947 : }
2948 336 : metadata[k] = v;
2949 330 : }
2950 330 : root[ConversationMapKeys::CONVERSATIONID] = id();
2951 330 : root[ConversationMapKeys::RECEIVED] = Json::Int64(toSecondsSinceEpoch(sent));
2952 330 : root[ConversationMapKeys::RECEIVED_MS] = Json::Int64(toMillisecondsSinceEpoch(sent));
2953 1320 : return {{"application/invite+json", json::toString(root)}};
2954 660 : }
2955 :
2956 : std::string
2957 16 : Conversation::leave()
2958 : {
2959 16 : setRemovingFlag();
2960 16 : std::lock_guard lk(pimpl_->writeMtx_);
2961 32 : return pimpl_->repository_->leave();
2962 16 : }
2963 :
2964 : void
2965 24 : Conversation::setRemovingFlag()
2966 : {
2967 24 : pimpl_->isRemoving_ = true;
2968 24 : }
2969 :
2970 : bool
2971 8390 : Conversation::isRemoving()
2972 : {
2973 8390 : return pimpl_->isRemoving_;
2974 : }
2975 :
2976 : void
2977 41 : Conversation::erase()
2978 : {
2979 41 : if (pimpl_->conversationDataPath_ != "")
2980 41 : dhtnet::fileutils::removeAll(pimpl_->conversationDataPath_, true);
2981 41 : if (!pimpl_->repository_)
2982 0 : return;
2983 41 : std::lock_guard lk(pimpl_->writeMtx_);
2984 41 : pimpl_->repository_->erase();
2985 41 : }
2986 :
2987 : ConversationMode
2988 6443 : Conversation::mode() const
2989 : {
2990 6443 : return pimpl_->repository_->mode();
2991 : }
2992 :
2993 : std::string
2994 139 : Conversation::parentConversationId() const
2995 : {
2996 139 : return pimpl_->repository_->parentConversationId();
2997 : }
2998 :
2999 : std::string
3000 0 : Conversation::documentMimeType() const
3001 : {
3002 0 : return pimpl_->repository_->documentMimeType();
3003 : }
3004 :
3005 : namespace {
3006 : // The base64 update lines of every checkpoint commit in a document log,
3007 : // oldest first, i.e. in the order the updates must be replayed.
3008 : std::vector<std::string>
3009 40 : collectUpdates(const std::vector<ConversationCommit>& commits)
3010 : {
3011 40 : std::vector<std::string> updates;
3012 222 : for (auto it = commits.rbegin(); it != commits.rend(); ++it) {
3013 182 : if (it->commitMsg.type != CommitType::CHECKPOINT)
3014 162 : continue;
3015 38 : for (const auto& line : split_string(it->commitMsg.body, '\n'))
3016 18 : if (!line.empty())
3017 38 : updates.emplace_back(line);
3018 : }
3019 40 : return updates;
3020 0 : }
3021 : } // namespace
3022 :
3023 : std::vector<std::string>
3024 37 : Conversation::documentUpdates() const
3025 : {
3026 37 : LogOptions options;
3027 37 : options.skipMerge = true;
3028 74 : return collectUpdates(pimpl_->repository_->log(options));
3029 37 : }
3030 :
3031 : std::optional<std::vector<std::string>>
3032 3 : Conversation::documentUpdatesAt(const std::string& commitId) const
3033 : {
3034 3 : if (!getCommit(commitId))
3035 0 : return std::nullopt;
3036 3 : LogOptions options;
3037 3 : options.from = commitId;
3038 3 : options.skipMerge = true;
3039 3 : return collectUpdates(pimpl_->repository_->log(options));
3040 3 : }
3041 :
3042 : std::vector<std::map<std::string, std::string>>
3043 106 : Conversation::documentHistory(size_t max) const
3044 : {
3045 106 : LogOptions options;
3046 106 : options.skipMerge = true;
3047 106 : auto commits = pimpl_->repository_->log(options);
3048 106 : std::vector<std::map<std::string, std::string>> result;
3049 499 : for (const auto& commit : commits) {
3050 403 : if (commit.commitMsg.type != CommitType::CHECKPOINT)
3051 365 : continue;
3052 38 : size_t deltas = 0;
3053 76 : for (const auto& line : split_string(commit.commitMsg.body, '\n'))
3054 38 : if (!line.empty())
3055 76 : ++deltas;
3056 266 : result.emplace_back(std::map<std::string, std::string> {
3057 38 : {"id", commit.id},
3058 38 : {"author", commit.authorId},
3059 38 : {"device", commit.author.email},
3060 0 : {"timestamp", std::to_string(commit.timestamp)},
3061 38 : {"deltas", std::to_string(deltas)},
3062 228 : });
3063 38 : if (max != 0 && result.size() >= max)
3064 10 : break;
3065 : }
3066 212 : return result;
3067 144 : }
3068 :
3069 : std::pair<std::string, std::string>
3070 1 : Conversation::addDocumentAttachment(const std::vector<uint8_t>& data)
3071 : {
3072 1 : std::unique_lock lk(pimpl_->writeMtx_);
3073 1 : auto headBefore = pimpl_->repository_->getHead();
3074 1 : auto attachmentId = pimpl_->repository_->addAttachment(data);
3075 1 : if (attachmentId.empty())
3076 0 : return {};
3077 1 : auto head = pimpl_->repository_->getHead();
3078 1 : if (head == headBefore)
3079 0 : return {attachmentId, {}}; // Same content already attached, nothing new to announce
3080 1 : pimpl_->announce(head, true);
3081 1 : return {attachmentId, head};
3082 1 : }
3083 :
3084 : std::vector<uint8_t>
3085 3 : Conversation::documentAttachment(const std::string& attachmentId) const
3086 : {
3087 3 : return pimpl_->repository_->attachment(attachmentId);
3088 : }
3089 :
3090 : std::vector<std::string>
3091 37 : Conversation::documentAttachmentIds() const
3092 : {
3093 37 : return pimpl_->repository_->attachmentIds();
3094 : }
3095 :
3096 : std::vector<std::string>
3097 36 : Conversation::getInitialMembers() const
3098 : {
3099 36 : return pimpl_->repository_->getInitialMembers();
3100 : }
3101 :
3102 : bool
3103 0 : Conversation::isInitialMember(const std::string& uri) const
3104 : {
3105 0 : auto members = getInitialMembers();
3106 0 : return std::find(members.begin(), members.end(), uri) != members.end();
3107 0 : }
3108 :
3109 : void
3110 36 : Conversation::updateInfos(const std::map<std::string, std::string>& map, const OnDoneCb& cb)
3111 : {
3112 36 : dht::ThreadPool::io().run([w = weak(), map = std::move(map), cb = std::move(cb)] {
3113 36 : if (auto sthis = w.lock()) {
3114 36 : auto& repo = sthis->pimpl_->repository_;
3115 36 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
3116 36 : auto commit = repo->updateInfos(map);
3117 36 : sthis->pimpl_->announce(commit, true);
3118 36 : lk.unlock();
3119 36 : if (cb)
3120 36 : cb(!commit.empty(), commit);
3121 36 : if (repo->mode() == ConversationMode::DOCUMENT)
3122 27 : return; // A document is not a conversation for the client; a rename
3123 : // is reported through CollaborativeDocumentRenamed instead
3124 9 : emitSignal<libjami::ConversationSignal::ConversationProfileUpdated>(sthis->pimpl_->accountId_,
3125 9 : repo->id(),
3126 18 : repo->infos());
3127 99 : }
3128 : });
3129 36 : }
3130 :
3131 : std::map<std::string, std::string>
3132 426 : Conversation::infos() const
3133 : {
3134 426 : return pimpl_->repository_->infos();
3135 : }
3136 :
3137 : void
3138 8 : Conversation::updatePreferences(const std::map<std::string, std::string>& map)
3139 : {
3140 8 : const auto& filePath = pimpl_->preferencesPath_;
3141 8 : auto prefs = map;
3142 8 : auto itLast = prefs.find(LAST_MODIFIED);
3143 8 : if (itLast != prefs.end()) {
3144 3 : std::error_code ec;
3145 3 : if (std::filesystem::is_regular_file(filePath, ec)) {
3146 1 : auto lastModified = fileutils::lastWriteTimeInSeconds(filePath);
3147 : try {
3148 1 : if (lastModified >= to_int<uint64_t>(itLast->second))
3149 0 : return;
3150 0 : } catch (...) {
3151 0 : return;
3152 0 : }
3153 : }
3154 3 : prefs.erase(itLast);
3155 : }
3156 :
3157 8 : std::ofstream file(filePath, std::ios::trunc | std::ios::binary);
3158 8 : msgpack::pack(file, prefs);
3159 8 : emitSignal<libjami::ConversationSignal::ConversationPreferencesUpdated>(pimpl_->accountId_, id(), std::move(prefs));
3160 8 : }
3161 :
3162 : std::map<std::string, std::string>
3163 127 : Conversation::preferences(bool includeLastModified) const
3164 : {
3165 : try {
3166 127 : std::map<std::string, std::string> preferences;
3167 127 : const auto& filePath = pimpl_->preferencesPath_;
3168 243 : auto file = fileutils::loadFile(filePath);
3169 11 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
3170 11 : oh.get().convert(preferences);
3171 11 : if (includeLastModified)
3172 24 : preferences[LAST_MODIFIED] = std::to_string(fileutils::lastWriteTimeInSeconds(filePath));
3173 11 : return preferences;
3174 243 : } catch (const std::exception& e) {
3175 116 : }
3176 116 : return {};
3177 : }
3178 :
3179 : std::vector<uint8_t>
3180 0 : Conversation::vCard() const
3181 : {
3182 : try {
3183 0 : return fileutils::loadFile(pimpl_->repoPath_ / "profile.vcf");
3184 0 : } catch (...) {
3185 0 : }
3186 0 : return {};
3187 : }
3188 :
3189 : std::shared_ptr<TransferManager>
3190 2182 : Conversation::dataTransfer() const
3191 : {
3192 2182 : return pimpl_->transferManager_;
3193 : }
3194 :
3195 : bool
3196 21 : Conversation::onFileChannelRequest(const std::string& member,
3197 : const std::string& fileId,
3198 : std::filesystem::path& path,
3199 : std::string& sha3sum) const
3200 : {
3201 21 : if (!isMember(member))
3202 0 : return false;
3203 :
3204 21 : if (!isValidFileId(fileId)) {
3205 2 : JAMI_WARNING("[Account {:s}] {} requested file with invalid id {}", pimpl_->accountId_, member, fileId);
3206 2 : return false;
3207 : }
3208 19 : auto sep = fileId.find('_');
3209 19 : auto interactionId = fileId.substr(0, sep);
3210 19 : auto commit = getCommit(interactionId);
3211 38 : if (commit == std::nullopt || commit->commitMsg.tid.empty() || commit->commitMsg.sha3sum.empty()
3212 38 : || commit->commitMsg.type != CommitType::DATA_TRANSFER) {
3213 0 : JAMI_WARNING("[Account {:s}] {} requested invalid file transfer commit {}",
3214 : pimpl_->accountId_,
3215 : member,
3216 : interactionId);
3217 0 : return false;
3218 : }
3219 : // The commit is the only source of truth for the file name
3220 19 : if (fileId != getFileId(interactionId, commit->commitMsg.tid, commit->commitMsg.displayName)) {
3221 2 : JAMI_WARNING("[Account {:s}] {} requested file {} not matching commit {}",
3222 : pimpl_->accountId_,
3223 : member,
3224 : fileId,
3225 : interactionId);
3226 2 : return false;
3227 : }
3228 :
3229 17 : path = dataTransfer()->path(fileId);
3230 17 : sha3sum = commit->commitMsg.sha3sum;
3231 :
3232 17 : return true;
3233 19 : }
3234 :
3235 : bool
3236 27 : Conversation::downloadFile(const std::string& interactionId,
3237 : const std::string& fileId,
3238 : const std::string& path,
3239 : const std::string&,
3240 : const std::string& deviceId)
3241 : {
3242 27 : auto commit = getCommit(interactionId);
3243 27 : if (commit == std::nullopt || commit->commitMsg.type != CommitType::DATA_TRANSFER) {
3244 1 : JAMI_ERROR("Commit doesn't exists or is not a file transfer {} (Conversation: {}) ", interactionId, id());
3245 1 : return false;
3246 : }
3247 26 : auto tid = commit->commitMsg.tid;
3248 26 : auto sha3sum = commit->commitMsg.sha3sum;
3249 26 : auto totalSize = commit->commitMsg.totalSize;
3250 :
3251 26 : if (tid.empty() || sha3sum.empty() || totalSize < 0) {
3252 0 : JAMI_ERROR("Invalid file transfer commit (missing tid, size or sha3)");
3253 0 : return false;
3254 : }
3255 26 : if (!isValidFileId(fileId) || fileId != getFileId(interactionId, tid, commit->commitMsg.displayName)) {
3256 7 : JAMI_ERROR("File id {} does not match file transfer commit {}", fileId, interactionId);
3257 7 : return false;
3258 : }
3259 19 : if (!path.empty() && std::filesystem::path(path).is_relative()) {
3260 1 : JAMI_ERROR("Refusing relative file transfer destination {}", path);
3261 1 : return false;
3262 : }
3263 :
3264 : // Be sure to not lock conversation
3265 18 : dht::ThreadPool::io().run([w = weak(), deviceId, fileId, interactionId, sha3sum, path, totalSize] {
3266 18 : auto shared = w.lock();
3267 18 : if (!shared)
3268 0 : return;
3269 18 : auto transferManager = shared->dataTransfer();
3270 18 : const auto emitTransferEvent = [accountId = shared->pimpl_->accountId_,
3271 : conversationId = shared->id(),
3272 18 : interactionId,
3273 18 : fileId](libjami::DataTransferEventCode code) {
3274 4 : emitSignal<libjami::DataTransferSignal::DataTransferEvent>(accountId,
3275 4 : conversationId,
3276 4 : interactionId,
3277 4 : fileId,
3278 : uint32_t(code));
3279 22 : };
3280 18 : switch (transferManager->waitForTransfer(fileId, interactionId, sha3sum, path, totalSize)) {
3281 2 : case TransferManager::WaitResult::complete:
3282 2 : JAMI_LOG("Ignoring request to download available file: {}", fileId);
3283 2 : emitTransferEvent(libjami::DataTransferEventCode::finished);
3284 2 : return;
3285 2 : case TransferManager::WaitResult::conflict:
3286 2 : emitTransferEvent(libjami::DataTransferEventCode::invalid_pathname);
3287 2 : return;
3288 14 : case TransferManager::WaitResult::waiting:
3289 14 : break;
3290 : }
3291 14 : auto acc = shared->pimpl_->account_.lock();
3292 14 : if (!acc)
3293 0 : return;
3294 : // Resume from the partial file left by a previous attempt, if any
3295 14 : auto destination = path.empty() ? transferManager->path(fileId) : std::filesystem::path(path);
3296 14 : std::error_code ec;
3297 14 : auto start = std::filesystem::file_size(transferManager->temporaryPath(fileId, destination), ec);
3298 14 : if (ec || start == static_cast<decltype(start)>(-1))
3299 11 : start = 0;
3300 14 : size_t end = 0;
3301 14 : acc->askForFileChannel(shared->id(), deviceId, interactionId, fileId, start, end);
3302 26 : });
3303 18 : return true;
3304 27 : }
3305 :
3306 : void
3307 2242 : Conversation::hasFetched(const std::string& deviceId, const std::string& commitId)
3308 : {
3309 2242 : dht::ThreadPool::io().run([w = weak(), deviceId, commitId]() {
3310 2242 : auto sthis = w.lock();
3311 2242 : if (!sthis)
3312 0 : return;
3313 : // Update fetched for Uri
3314 2242 : auto uri = sthis->uriFromDevice(deviceId);
3315 2242 : if (uri.empty() || uri == sthis->pimpl_->userId_)
3316 56 : return;
3317 : // When a user fetches a commit, the message is sent for this person
3318 2186 : sthis->pimpl_->updateStatus(uri,
3319 : libjami::Account::MessageStates::SENT,
3320 2186 : commitId,
3321 4372 : std::to_string(std::time(nullptr)),
3322 : true);
3323 2298 : });
3324 2242 : }
3325 :
3326 : void
3327 2236 : Conversation::Impl::updateStatus(const std::string& uri,
3328 : libjami::Account::MessageStates st,
3329 : const std::string& commitId,
3330 : const std::string& ts,
3331 : bool emit)
3332 : {
3333 : // This method can be called if peer send us a status or if another device sync. Emit will be true if a peer
3334 : // send us a status and will emit to other connected devices.
3335 2236 : LogOptions options;
3336 2236 : std::map<std::string, std::map<std::string, std::string>> newStatus;
3337 : {
3338 : // Update internal structures.
3339 2236 : std::lock_guard lk(messageStatusMtx_);
3340 2236 : auto& status = messagesStatus_[uri];
3341 2236 : auto& oldStatus = status[st == libjami::Account::MessageStates::SENT ? "fetched" : "read"];
3342 2236 : if (oldStatus == commitId)
3343 315 : return; // Nothing to do
3344 1921 : options.to = oldStatus;
3345 1921 : options.from = commitId;
3346 1921 : oldStatus = commitId;
3347 1921 : status[st == libjami::Account::MessageStates::SENT ? "fetched_ts" : "read_ts"] = ts;
3348 1921 : saveStatus();
3349 1921 : if (emit)
3350 1885 : newStatus[uri].insert(status.begin(), status.end());
3351 2236 : }
3352 1921 : if (emit && messageStatusCb_) {
3353 1854 : messageStatusCb_(newStatus);
3354 : }
3355 : // Update messages status for all commit between the old and new one
3356 1921 : options.logIfNotFound = false;
3357 1921 : options.fastLog = true;
3358 1921 : History optHistory;
3359 1921 : std::unique_lock lk(optHistory.mutex); // Avoid to announce messages while updating status.
3360 1921 : auto res = loadMessages(options, &optHistory);
3361 1921 : std::unique_lock mlk(messageStatusMtx_);
3362 1921 : std::vector<std::pair<std::string, int32_t>> statusToUpdate;
3363 1921 : if (res.size() == 0) {
3364 : // In this case, commit is not received yet, so we cache it
3365 14 : futureStatus[commitId][uri] = static_cast<int32_t>(st);
3366 : }
3367 16273 : for (const auto& [cid, _] : optHistory.quickAccess) {
3368 14352 : auto message = loadedHistory_.quickAccess.find(cid);
3369 14352 : if (message != loadedHistory_.quickAccess.end()) {
3370 : // Update message and emit to client,
3371 1704 : if (static_cast<int32_t>(st) > message->second->status[uri]) {
3372 1699 : message->second->status[uri] = static_cast<int32_t>(st);
3373 1699 : statusToUpdate.emplace_back(cid, static_cast<int32_t>(st));
3374 : }
3375 : } else {
3376 : // In this case, commit is not loaded by client, so we cache it
3377 : // No need to emit to client, they will get a correct status on load.
3378 12648 : futureStatus[cid][uri] = static_cast<int32_t>(st);
3379 : }
3380 : }
3381 1921 : mlk.unlock();
3382 1921 : lk.unlock();
3383 3620 : for (const auto& [cid, status] : statusToUpdate)
3384 3398 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(accountId_,
3385 1699 : repository_->id(),
3386 : uri,
3387 : cid,
3388 : static_cast<int>(status));
3389 2551 : }
3390 :
3391 : bool
3392 16 : Conversation::setMessageDisplayed(const std::string& uri, const std::string& interactionId)
3393 : {
3394 16 : std::lock_guard lk(pimpl_->messageStatusMtx_);
3395 48 : if (pimpl_->messagesStatus_[uri]["read"] == interactionId)
3396 2 : return false; // Nothing to do
3397 14 : dht::ThreadPool::io().run([w = weak(), uri, interactionId]() {
3398 14 : auto sthis = w.lock();
3399 14 : if (!sthis)
3400 0 : return;
3401 14 : sthis->pimpl_->updateStatus(uri,
3402 : libjami::Account::MessageStates::DISPLAYED,
3403 14 : interactionId,
3404 28 : std::to_string(std::time(nullptr)),
3405 : true);
3406 14 : });
3407 14 : return true;
3408 16 : }
3409 :
3410 : std::map<std::string, std::map<std::string, std::string>>
3411 109 : Conversation::messageStatus() const
3412 : {
3413 109 : std::lock_guard lk(pimpl_->messageStatusMtx_);
3414 218 : return pimpl_->messagesStatus_;
3415 109 : }
3416 :
3417 : void
3418 64 : Conversation::updateMessageStatus(const std::map<std::string, std::map<std::string, std::string>>& messageStatus)
3419 : {
3420 64 : std::unique_lock lk(pimpl_->messageStatusMtx_);
3421 64 : std::vector<std::tuple<libjami::Account::MessageStates, std::string, std::string, std::string>> stVec;
3422 149 : for (const auto& [uri, status] : messageStatus) {
3423 85 : auto& oldMs = pimpl_->messagesStatus_[uri];
3424 507 : if (status.find("fetched_ts") != status.end() && status.at("fetched") != oldMs["fetched"]) {
3425 128 : if (oldMs["fetched_ts"].empty() || std::stol(oldMs["fetched_ts"]) <= std::stol(status.at("fetched_ts"))) {
3426 0 : stVec.emplace_back(libjami::Account::MessageStates::SENT,
3427 : uri,
3428 32 : status.at("fetched"),
3429 96 : status.at("fetched_ts"));
3430 : }
3431 : }
3432 283 : if (status.find("read_ts") != status.end() && status.at("read") != oldMs["read"]) {
3433 12 : if (oldMs["read_ts"].empty() || std::stol(oldMs["read_ts"]) <= std::stol(status.at("read_ts"))) {
3434 0 : stVec.emplace_back(libjami::Account::MessageStates::DISPLAYED,
3435 : uri,
3436 4 : status.at("read"),
3437 12 : status.at("read_ts"));
3438 : }
3439 : }
3440 : }
3441 64 : lk.unlock();
3442 :
3443 100 : for (const auto& [status, uri, commitId, ts] : stVec) {
3444 36 : pimpl_->updateStatus(uri, status, commitId, ts);
3445 : }
3446 64 : }
3447 :
3448 : void
3449 882 : Conversation::onMessageStatusChanged(
3450 : const std::function<void(const std::map<std::string, std::map<std::string, std::string>>&)>& cb)
3451 : {
3452 882 : std::unique_lock lk(pimpl_->messageStatusMtx_);
3453 882 : pimpl_->messageStatusCb_ = cb;
3454 882 : }
3455 :
3456 : #ifdef LIBJAMI_TEST
3457 : void
3458 1462 : Conversation::onBootstrapStatus(const std::function<void(std::string, BootstrapStatus)>& cb)
3459 : {
3460 1462 : std::lock_guard lock(pimpl_->bootstrapMtx_);
3461 1462 : pimpl_->bootstrapCbTest_ = cb;
3462 1462 : }
3463 :
3464 : std::vector<libjami::SwarmMessage>
3465 0 : Conversation::loadMessagesSync(const LogOptions& options)
3466 : {
3467 0 : std::lock_guard lk(pimpl_->loadedHistory_.mutex);
3468 0 : auto result = pimpl_->loadMessages(options);
3469 0 : return result;
3470 0 : }
3471 :
3472 : void
3473 0 : Conversation::announce(const std::vector<std::map<std::string, std::string>>& commits, bool commitFromSelf)
3474 : {
3475 0 : pimpl_->announce(commits, commitFromSelf);
3476 0 : }
3477 :
3478 : void
3479 0 : Conversation::announce(const std::string& commitId, bool commitFromSelf)
3480 : {
3481 0 : pimpl_->announce(commitId, commitFromSelf);
3482 0 : }
3483 : #endif
3484 :
3485 : void
3486 1455 : Conversation::bootstrap(std::function<void()> onBootstrapped, const std::vector<DeviceId>& knownDevices)
3487 : {
3488 1455 : std::lock_guard lock(pimpl_->bootstrapMtx_);
3489 1455 : if (!pimpl_ || !pimpl_->repository_ || !pimpl_->swarmManager_)
3490 0 : return;
3491 : // Bootstrap the DRT from currently known devices. Since the per-device
3492 : // presence monitoring (monitorConnection/startTracking below), callers do
3493 : // not pass any device list here: candidates are injected as members get
3494 : // online, through addKnownDevices() with devices reported by the
3495 : // PresenceManager (i.e. announced on the DHT), and rotated on connection
3496 : // failure. The knownDevices parameter remains for callers/tests that
3497 : // already hold a list of live devices.
3498 : // If a connection succeeds, onConnectionChanged will be called with ok=true
3499 1455 : pimpl_->bootstrapCb_ = std::move(onBootstrapped);
3500 1455 : std::vector<DeviceId> devices = knownDevices;
3501 :
3502 1455 : if (!devices.empty()) {
3503 0 : JAMI_DEBUG("{} Bootstrap with {} device(s)", pimpl_->toString(), devices.size());
3504 0 : pimpl_->swarmManager_->setKnownNodes(devices);
3505 : }
3506 :
3507 1455 : pimpl_->monitorConnection(weak_from_this());
3508 :
3509 : // If is shutdown, the conversation was re-added, causing no new nodes to be connected, but just a classic
3510 : // connectivity change
3511 1455 : if (pimpl_->swarmManager_->isShutdown()) {
3512 33 : pimpl_->swarmManager_->restart();
3513 33 : pimpl_->swarmManager_->maintainBuckets();
3514 1422 : } else if (!pimpl_->swarmManager_->isConnected()) {
3515 : // A swarm manager that is up but holds no connected node will never get
3516 : // one on its own: setKnownNodes() only acts on ids it has never seen
3517 : // before, so members it already knows are simply skipped.
3518 : //
3519 : // A mobile client reaches that state every time it leaves and returns
3520 : // to the foreground: setAccountActive() leaves established connections
3521 : // alone by default, so the swarm manager is not shut down, yet the
3522 : // links die with the network. Bootstrapping then finds nothing to do,
3523 : // and the conversation stays silent until the process is restarted.
3524 1379 : pimpl_->swarmManager_->maintainBuckets();
3525 : }
3526 1455 : }
3527 :
3528 : void
3529 3438 : Conversation::addKnownDevices(const std::vector<DeviceId>& devices, const std::string& memberUri)
3530 : {
3531 3438 : if (devices.empty())
3532 1511 : return;
3533 1927 : if (!memberUri.empty()) {
3534 : // JAMI_WARNING("{} Adding {} known devices for member {}", pimpl_->toString(), devices.size(), memberUri);
3535 1927 : std::lock_guard lk(pimpl_->trackedMembersMtx_);
3536 1927 : auto it = pimpl_->trackedMembers_.find(memberUri);
3537 1927 : if (it != pimpl_->trackedMembers_.end()) {
3538 1924 : it->second.presenceReported = true;
3539 1924 : it->second.devices.insert(devices.begin(), devices.end());
3540 : // Presence only reports a device once it was absent, so a failure
3541 : // recorded before this announce says nothing about the device now.
3542 3856 : for (const auto& device : devices)
3543 1932 : it->second.failedDevices.erase(device);
3544 : }
3545 1927 : } else {
3546 0 : JAMI_ERROR("{} Adding {} known devices without member URI", pimpl_->toString(), devices.size());
3547 : }
3548 1927 : pimpl_->swarmManager_->setKnownNodes(devices);
3549 : }
3550 :
3551 : void
3552 688 : Conversation::connectNode(const DeviceId& deviceId)
3553 : {
3554 688 : pimpl_->swarmManager_->connectNode(deviceId);
3555 688 : }
3556 :
3557 : std::vector<std::string>
3558 25 : Conversation::commitsEndedCalls()
3559 : {
3560 25 : pimpl_->loadActiveCalls();
3561 25 : pimpl_->loadHostedCalls();
3562 25 : auto commits = pimpl_->commitsEndedCalls();
3563 25 : if (!commits.empty()) {
3564 : // Announce to client
3565 0 : dht::ThreadPool::io().run([w = weak(), commits] {
3566 0 : if (auto sthis = w.lock())
3567 0 : sthis->pimpl_->announce(commits, true);
3568 0 : });
3569 : }
3570 25 : return commits;
3571 0 : }
3572 :
3573 : void
3574 907 : Conversation::onMembersChanged(OnMembersChanged&& cb)
3575 : {
3576 907 : pimpl_->onMembersChanged_ = std::move(cb);
3577 907 : }
3578 :
3579 : void
3580 907 : Conversation::onNeedSocket(NeedSocketCb needSocket)
3581 : {
3582 1814 : pimpl_->swarmManager_->needSocketCb_ = [needSocket = std::move(needSocket),
3583 : w = weak()](const std::string& deviceId, ChannelCb&& cb, bool noNewSocket) {
3584 1894 : if (auto sthis = w.lock()) {
3585 1894 : auto wrappedCb = [cb = std::move(cb), w, deviceId](const std::shared_ptr<dhtnet::ChannelSocket>& socket) {
3586 1893 : if (auto sthis = w.lock()) {
3587 1885 : if (!socket) {
3588 696 : if (auto acc = sthis->pimpl_->account_.lock()) {
3589 696 : auto cert = acc->certStore().getCertificate(deviceId);
3590 696 : if (cert && cert->issuer) {
3591 696 : sthis->pimpl_->onConnectionFailed(DeviceId(deviceId), cert->issuer->getId().toString());
3592 : } else {
3593 0 : JAMI_WARNING("{} Unable to get member URI from device ID {}",
3594 : sthis->pimpl_->toString(),
3595 : deviceId);
3596 : }
3597 696 : } else {
3598 0 : return false;
3599 696 : }
3600 : }
3601 1893 : }
3602 1893 : return cb(socket);
3603 1894 : };
3604 3788 : needSocket(sthis->id(), deviceId, std::move(wrappedCb), "application/im-gitmessage-id", noNewSocket);
3605 3788 : }
3606 2801 : };
3607 907 : }
3608 :
3609 : void
3610 1614 : Conversation::addSwarmChannel(std::shared_ptr<dhtnet::ChannelSocket> channel)
3611 : {
3612 1614 : auto deviceId = channel->deviceId();
3613 : // Transmit avatar if necessary
3614 : // We do this here, because at this point we know both sides are connected and in
3615 : // the same conversation
3616 : // addSwarmChannel is a bit more complex, but it should be the best moment to do this.
3617 1614 : auto cert = channel->peerCertificate();
3618 1614 : if (!cert || !cert->issuer)
3619 0 : return;
3620 1614 : auto member = cert->issuer->getId().toString();
3621 : // The TLS handshake authenticated this certificate: pin it so that any mobile
3622 : // lease this device gossips can be verified without a lookup.
3623 1614 : if (auto account = pimpl_->account_.lock())
3624 1614 : account->certStore().pinCertificate(cert);
3625 1614 : pimpl_->swarmManager_->addChannel(std::move(channel));
3626 1614 : dht::ThreadPool::io().run([member, deviceId, a = pimpl_->account_, w = weak_from_this()] {
3627 1614 : auto sthis = w.lock();
3628 1614 : if (auto account = a.lock()) {
3629 1614 : account->sendProfile(sthis->id(), member, deviceId.toString());
3630 1614 : }
3631 1614 : });
3632 1614 : }
3633 :
3634 : uint32_t
3635 4 : Conversation::countInteractions(const std::string& toId, const std::string& fromId, const std::string& authorUri) const
3636 : {
3637 4 : LogOptions options;
3638 4 : options.to = toId;
3639 4 : options.from = fromId;
3640 4 : options.authorUri = authorUri;
3641 4 : options.logIfNotFound = false;
3642 4 : options.fastLog = true;
3643 4 : History history;
3644 4 : std::lock_guard lk(history.mutex);
3645 4 : auto res = pimpl_->loadMessages(options, &history);
3646 8 : return res.size();
3647 4 : }
3648 :
3649 : void
3650 4 : Conversation::search(uint32_t req, const Filter& filter, const std::shared_ptr<std::atomic_int>& flag) const
3651 : {
3652 : // Because logging a conversation can take quite some time,
3653 : // do it asynchronously
3654 4 : dht::ThreadPool::io().run([w = weak(), req, filter, flag] {
3655 4 : if (auto sthis = w.lock()) {
3656 4 : History history;
3657 4 : std::vector<std::map<std::string, std::string>> commits {};
3658 : // std::regex_constants::ECMAScript is the default flag.
3659 4 : auto re = std::regex(filter.regexSearch,
3660 4 : filter.caseSensitive ? std::regex_constants::ECMAScript : std::regex_constants::icase);
3661 12 : sthis->pimpl_->repository_->log(
3662 8 : [&](const std::string& /*id*/, const GitAuthor& author, const GitCommit& commit) {
3663 20 : if (!filter.author.empty() && filter.author != sthis->uriFromDevice(author.email)) {
3664 : // Filter author
3665 0 : return CallbackResult::Skip;
3666 : }
3667 20 : auto commitTime = git_commit_time(commit.get());
3668 20 : if (filter.before && filter.before < commitTime) {
3669 : // Only get commits before this date
3670 0 : return CallbackResult::Skip;
3671 : }
3672 20 : if (filter.after && filter.after > commitTime) {
3673 : // Only get commits before this date
3674 0 : if (git_commit_parentcount(commit.get()) <= 1)
3675 0 : return CallbackResult::Break;
3676 : else
3677 0 : return CallbackResult::Skip; // Because we are sorting it with
3678 : // GIT_SORT_TOPOLOGICAL | GIT_SORT_TIME
3679 : }
3680 :
3681 20 : return CallbackResult::Ok; // Continue
3682 : },
3683 8 : [&](ConversationCommit&& cc) {
3684 20 : if (auto optMessage = sthis->pimpl_->repository_->convCommitToMap(cc))
3685 80 : sthis->pimpl_->addToHistory(history, {optMessage.value()}, false, false);
3686 40 : },
3687 8 : [&](const std::string& id, const GitAuthor&, ConversationCommit&) {
3688 20 : if (id == filter.lastId)
3689 0 : return true;
3690 20 : return false;
3691 : },
3692 : "",
3693 : false);
3694 : // Search on generated history
3695 24 : for (auto& message : history.messageList) {
3696 20 : auto contentType = message->type;
3697 20 : auto isSearchable = contentType == CommitType::TEXT || contentType == CommitType::DATA_TRANSFER;
3698 20 : if (filter.type.empty() && !isSearchable) {
3699 : // Not searchable, at least for now
3700 8 : continue;
3701 12 : } else if (contentType == filter.type || filter.type.empty()) {
3702 12 : if (isSearchable) {
3703 : // If it's a text match the body, else the display name
3704 60 : auto body = contentType == CommitType::TEXT ? message->body.at(CommitKey::BODY)
3705 36 : : message->body.at(CommitKey::DISPLAY_NAME);
3706 12 : std::smatch body_match;
3707 12 : if (std::regex_search(body, body_match, re)) {
3708 5 : auto commit = message->body;
3709 10 : commit["id"] = message->id;
3710 10 : commit[CommitKey::TYPE] = message->type;
3711 5 : commits.emplace_back(commit);
3712 5 : }
3713 12 : } else {
3714 : // Matching type, just add it to the results
3715 0 : commits.emplace_back(message->body);
3716 : }
3717 :
3718 12 : if (filter.maxResult != 0 && commits.size() == filter.maxResult)
3719 0 : break;
3720 : }
3721 20 : }
3722 :
3723 4 : if (commits.size() > 0)
3724 9 : emitSignal<libjami::ConversationSignal::MessagesFound>(req,
3725 3 : sthis->pimpl_->accountId_,
3726 6 : sthis->id(),
3727 3 : std::move(commits));
3728 : // If we're the latest thread, inform client that the search is finished
3729 4 : if ((*flag)-- == 1 /* decrement return the old value */) {
3730 8 : emitSignal<libjami::ConversationSignal::MessagesFound>(
3731 8 : req, sthis->pimpl_->accountId_, std::string {}, std::vector<std::map<std::string, std::string>> {});
3732 : }
3733 8 : }
3734 4 : });
3735 4 : }
3736 :
3737 : void
3738 14 : Conversation::hostConference(CommitMessage&& message, OnDoneCb&& cb)
3739 : {
3740 14 : if (message.confId.empty()) {
3741 0 : JAMI_ERROR("{}Malformed commit: no confId", pimpl_->toString());
3742 0 : return;
3743 : }
3744 :
3745 14 : auto now = std::chrono::system_clock::now();
3746 14 : auto nowSecs = std::chrono::duration_cast<std::chrono::seconds>(now.time_since_epoch()).count();
3747 : {
3748 14 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3749 14 : pimpl_->hostedCalls_[message.confId] = nowSecs;
3750 14 : pimpl_->saveHostedCalls();
3751 14 : }
3752 :
3753 14 : createCommit(std::move(message), {}, std::move(cb));
3754 : }
3755 :
3756 : bool
3757 28 : Conversation::isHosting(const std::string& confId) const
3758 : {
3759 28 : auto info = infos();
3760 88 : if (info["rdvDevice"] == pimpl_->deviceId_ && info["rdvHost"] == pimpl_->userId_)
3761 0 : return true; // We are the current device Host
3762 28 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3763 28 : return pimpl_->hostedCalls_.find(confId) != pimpl_->hostedCalls_.end();
3764 28 : }
3765 :
3766 : void
3767 10 : Conversation::removeActiveConference(CommitMessage&& message, OnDoneCb&& cb)
3768 : {
3769 10 : if (message.confId.empty()) {
3770 0 : JAMI_ERROR("{}Malformed commit: no confId", pimpl_->toString());
3771 0 : return;
3772 : }
3773 :
3774 10 : auto erased = false;
3775 : {
3776 10 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3777 10 : erased = pimpl_->hostedCalls_.erase(message.confId);
3778 10 : }
3779 10 : if (erased) {
3780 10 : pimpl_->saveHostedCalls();
3781 10 : createCommit(std::move(message), {}, std::move(cb));
3782 : } else
3783 0 : cb(false, "");
3784 : }
3785 :
3786 : std::vector<std::map<std::string, std::string>>
3787 43 : Conversation::currentCalls() const
3788 : {
3789 43 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3790 86 : return pimpl_->activeCalls_;
3791 43 : }
3792 : } // namespace jami
|
