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