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 689 : timePointFromJson(const Json::Value& json, const char* msKey, const char* secondsKey)
73 : {
74 689 : if (json.isMember(msKey))
75 334 : return timePointFromMilliseconds(json[msKey].asLargestInt());
76 355 : return timePointFromSeconds(json[secondsKey].asLargestInt());
77 : }
78 :
79 : // Resolve a timestamp from msgpack values, preferring milliseconds.
80 : TimePoint
81 947 : resolveTimePoint(const std::optional<int64_t>& ms, int64_t seconds)
82 : {
83 947 : 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 297 : ConvInfo::msgpack_unpack(const msgpack::object& o)
140 : {
141 297 : if (o.type != msgpack::type::MAP)
142 0 : throw msgpack::type_error();
143 297 : int64_t createdSec = 0, removedSec = 0, erasedSec = 0;
144 297 : std::optional<int64_t> createdMs, removedMs, erasedMs;
145 297 : std::map<std::string, int64_t> invitedMs;
146 3560 : for (uint32_t i = 0; i < o.via.map.size; ++i) {
147 3263 : const auto& kv = o.via.map.ptr[i];
148 3263 : if (kv.key.type != msgpack::type::STR)
149 0 : continue;
150 3263 : std::string_view key(kv.key.via.str.ptr, kv.key.via.str.size);
151 3262 : if (key == ConversationMapKeys::ID)
152 297 : kv.val.convert(id);
153 2965 : else if (key == ConversationMapKeys::CREATED)
154 297 : kv.val.convert(createdSec);
155 2669 : else if (key == ConversationMapKeys::REMOVED)
156 297 : kv.val.convert(removedSec);
157 2372 : else if (key == ConversationMapKeys::ERASED)
158 297 : kv.val.convert(erasedSec);
159 2075 : else if (key == ConversationMapKeys::CREATED_MS)
160 296 : createdMs = kv.val.as<int64_t>();
161 1778 : else if (key == ConversationMapKeys::REMOVED_MS)
162 296 : removedMs = kv.val.as<int64_t>();
163 1482 : else if (key == ConversationMapKeys::ERASED_MS)
164 296 : erasedMs = kv.val.as<int64_t>();
165 1187 : else if (key == ConversationMapKeys::MEMBERS)
166 297 : kv.val.convert(members);
167 890 : else if (key == ConversationMapKeys::LAST_DISPLAYED)
168 297 : kv.val.convert(lastDisplayed);
169 593 : else if (key == ConversationMapKeys::MODE)
170 297 : kv.val.convert(mode);
171 296 : else if (key == ConversationMapKeys::INVITED)
172 296 : kv.val.convert(invitedMs);
173 : }
174 297 : created = resolveTimePoint(createdMs, createdSec);
175 297 : removed = resolveTimePoint(removedMs, removedSec);
176 297 : erased = resolveTimePoint(erasedMs, erasedSec);
177 297 : invited.clear();
178 335 : for (const auto& [uri, ms] : invitedMs)
179 38 : invited[uri] = timePointFromMilliseconds(ms);
180 297 : }
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 319 : ConversationRequest::ConversationRequest(const Json::Value& json)
221 : {
222 319 : received = timePointFromJson(json, ConversationMapKeys::RECEIVED_MS, ConversationMapKeys::RECEIVED);
223 319 : declined = timePointFromJson(json, ConversationMapKeys::DECLINED_MS, ConversationMapKeys::DECLINED);
224 319 : from = json[ConversationMapKeys::FROM].asString();
225 319 : conversationId = json[ConversationMapKeys::CONVERSATIONID].asString();
226 319 : auto& md = json[ConversationMapKeys::METADATAS];
227 641 : for (const auto& member : md.getMemberNames()) {
228 322 : metadatas.emplace(member, md[member].asString());
229 319 : }
230 319 : }
231 :
232 : Json::Value
233 8 : ConversationRequest::toJson() const
234 : {
235 8 : Json::Value json;
236 8 : json[ConversationMapKeys::CONVERSATIONID] = conversationId;
237 8 : json[ConversationMapKeys::FROM] = from;
238 8 : json[ConversationMapKeys::RECEIVED] = Json::Int64(toSecondsSinceEpoch(received));
239 8 : json[ConversationMapKeys::RECEIVED_MS] = Json::Int64(toMillisecondsSinceEpoch(received));
240 8 : if (declined != TimePoint {}) {
241 0 : json[ConversationMapKeys::DECLINED] = Json::Int64(toSecondsSinceEpoch(declined));
242 0 : json[ConversationMapKeys::DECLINED_MS] = Json::Int64(toMillisecondsSinceEpoch(declined));
243 : }
244 16 : for (const auto& [key, value] : metadatas) {
245 8 : json[ConversationMapKeys::METADATAS][key] = value;
246 : }
247 8 : return json;
248 0 : }
249 :
250 : std::map<std::string, std::string>
251 310 : ConversationRequest::toMap() const
252 : {
253 310 : auto result = metadatas;
254 620 : result[ConversationMapKeys::ID] = conversationId;
255 620 : result[ConversationMapKeys::FROM] = from;
256 310 : if (declined != TimePoint {})
257 3 : result[ConversationMapKeys::DECLINED] = std::to_string(toSecondsSinceEpoch(declined));
258 930 : result[ConversationMapKeys::RECEIVED] = std::to_string(toSecondsSinceEpoch(received));
259 310 : return result;
260 0 : }
261 :
262 : void
263 28 : ConversationRequest::msgpack_unpack(const msgpack::object& o)
264 : {
265 28 : if (o.type != msgpack::type::MAP)
266 0 : throw msgpack::type_error();
267 28 : int64_t receivedSec = 0, declinedSec = 0;
268 28 : std::optional<int64_t> receivedMs, declinedMs;
269 222 : for (uint32_t i = 0; i < o.via.map.size; ++i) {
270 194 : const auto& kv = o.via.map.ptr[i];
271 194 : if (kv.key.type != msgpack::type::STR)
272 0 : continue;
273 194 : std::string_view key(kv.key.via.str.ptr, kv.key.via.str.size);
274 194 : if (key == ConversationMapKeys::FROM)
275 28 : kv.val.convert(from);
276 166 : else if (key == ConversationMapKeys::CONVERSATIONID)
277 28 : kv.val.convert(conversationId);
278 138 : else if (key == ConversationMapKeys::METADATAS)
279 28 : kv.val.convert(metadatas);
280 110 : else if (key == ConversationMapKeys::RECEIVED)
281 28 : kv.val.convert(receivedSec);
282 82 : else if (key == ConversationMapKeys::DECLINED)
283 28 : kv.val.convert(declinedSec);
284 54 : else if (key == ConversationMapKeys::RECEIVED_MS)
285 27 : receivedMs = kv.val.as<int64_t>();
286 27 : else if (key == ConversationMapKeys::DECLINED_MS)
287 27 : declinedMs = kv.val.as<int64_t>();
288 : }
289 28 : received = resolveTimePoint(receivedMs, receivedSec);
290 28 : declined = resolveTimePoint(declinedMs, declinedSec);
291 28 : }
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 505 : Impl(std::unique_ptr<ConversationRepository>&& repository,
337 : const std::shared_ptr<JamiAccount>& account,
338 : std::vector<ConversationCommit>&& commits = {})
339 505 : : repository_(repository ? std::move(repository) : throw std::logic_error("Invalid repository"))
340 492 : , account_(account)
341 492 : , accountId_(account->getAccountID())
342 492 : , userId_(account->getUsername())
343 984 : , deviceId_(account->currentDeviceId())
344 492 : , swarmManager_(std::make_shared<SwarmManager>(
345 492 : NodeId(deviceId_),
346 492 : account->isMobile(),
347 492 : Manager::instance().getSeededRandomEngine(),
348 492 : [account = account_](const DeviceId& deviceId) {
349 338 : if (auto acc = account.lock()) {
350 338 : return acc->isConnectedWith(deviceId);
351 337 : }
352 0 : return false;
353 : },
354 492 : repository_->id(),
355 984 : [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 492 : , transferManager_(std::make_shared<TransferManager>(accountId_,
399 : "",
400 492 : repository_->id(),
401 492 : Manager::instance().getSeededRandomEngine()))
402 492 : , repoPath_(fileutils::get_data_dir() / accountId_ / "conversations" / repository_->id())
403 492 : , conversationDataPath_(fileutils::get_data_dir() / accountId_ / "conversation_data" / repository_->id())
404 492 : , fetchedPath_(conversationDataPath_ / ConversationDirectories::FETCHED)
405 492 : , sendingPath_(conversationDataPath_ / ConversationDirectories::SENDING)
406 492 : , preferencesPath_(conversationDataPath_ / ConversationDirectories::PREFERENCES)
407 492 : , statusPath_(conversationDataPath_ / ConversationDirectories::STATUS)
408 492 : , mobileNodesPath_(conversationDataPath_ / ConversationDirectories::MOBILE_NODES)
409 492 : , hostedCallsPath_(conversationDataPath_ / ConversationDirectories::HOSTED_CALLS)
410 492 : , activeCallsPath_(conversationDataPath_ / ConversationDirectories::ACTIVE_CALLS)
411 492 : , ioContext_(Manager::instance().ioContext())
412 2965 : , typers_(std::make_shared<Typers>(account, repository_->id()))
413 : {
414 492 : if (!commits.empty())
415 237 : initActiveCalls(repository_->convCommitsToMap(commits));
416 492 : 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 492 : 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 984 : swarmManager_->onMobileNodeInfosChanged(
425 984 : [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 492 : setupMemberCallback();
431 531 : }
432 :
433 250 : Impl(std::pair<std::unique_ptr<ConversationRepository>, std::vector<ConversationCommit>>&& repoAndCommits,
434 : const std::shared_ptr<JamiAccount>& account)
435 250 : : Impl(std::move(repoAndCommits.first), account, std::move(repoAndCommits.second))
436 237 : {}
437 :
438 : public:
439 221 : Impl(const std::shared_ptr<JamiAccount>& account, ConversationMode mode, const std::string& otherMember = "")
440 221 : : Impl(ConversationRepository::createConversation(account, mode, otherMember), account)
441 221 : {}
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 254 : Impl(const std::shared_ptr<JamiAccount>& account, const std::string& remoteDevice, const std::string& conversationId)
448 254 : : Impl(ConversationRepository::cloneConversation(account, remoteDevice, conversationId), account)
449 237 : {}
450 :
451 3100 : std::string toString() const
452 : {
453 12399 : 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 492 : ~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 378 : void announce(const std::string& commitId, bool commitFromSelf = false)
474 : {
475 378 : std::vector<std::string> vec;
476 378 : if (!commitId.empty())
477 375 : vec.emplace_back(commitId);
478 378 : announce(vec, commitFromSelf);
479 378 : }
480 :
481 395 : void announce(const std::vector<std::string>& commits, bool commitFromSelf = false)
482 : {
483 395 : std::vector<ConversationCommit> convcommits;
484 395 : convcommits.reserve(commits.size());
485 804 : for (const auto& cid : commits) {
486 409 : if (auto commit = repository_->getCommit(cid)) {
487 409 : convcommits.emplace_back(*commit);
488 409 : }
489 : }
490 396 : announce(repository_->convCommitsToMap(convcommits), commitFromSelf);
491 395 : }
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 237 : void initActiveCalls(const std::vector<std::map<std::string, std::string>>& commits) const
498 : {
499 237 : std::unordered_set<std::string> invalidHostUris;
500 237 : std::unordered_set<std::string> invalidCallIds;
501 :
502 237 : std::lock_guard lk(activeCallsMtx_);
503 1356 : for (const auto& commit : commits) {
504 2238 : 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 1662 : invalidHostUris.emplace(commit.at(CommitKey::URI));
514 867 : } else if (commit.find(CommitKey::CONF_ID) != commit.end() && commit.find(CommitKey::URI) != commit.end()
515 579 : && 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 237 : saveActiveCalls();
544 237 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_, repository_->id(), activeCalls_);
545 237 : }
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 82 : void updateActiveCalls(const std::map<std::string, std::string>& commit,
555 : bool eraseOnly = false,
556 : bool emitSig = true) const
557 : {
558 82 : if (!repository_)
559 0 : return;
560 164 : 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 22 : while (it != activeCalls_.end()) {
566 0 : if (it->at("uri") == commit.at(CommitKey::URI) || it->at("device") == commit.at(CommitKey::URI)) {
567 0 : JAMI_DEBUG("Removing {:s} from the active calls, because {:s} left",
568 : it->at("id"),
569 : commit.at(CommitKey::URI));
570 0 : it = activeCalls_.erase(it);
571 0 : updateActives = true;
572 : } else {
573 0 : ++it;
574 : }
575 : }
576 22 : if (updateActives) {
577 0 : saveActiveCalls();
578 0 : if (emitSig)
579 0 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_,
580 0 : repository_->id(),
581 0 : activeCalls_);
582 : }
583 22 : return;
584 22 : }
585 : // Else, it's a call information
586 339 : if (commit.find(CommitKey::CONF_ID) != commit.end() && commit.find(CommitKey::URI) != commit.end()
587 279 : && commit.find(CommitKey::DEVICE) != commit.end()) {
588 53 : auto convId = repository_->id();
589 106 : auto confId = commit.at(CommitKey::CONF_ID);
590 106 : auto uri = commit.at(CommitKey::URI);
591 53 : auto device = commit.at(CommitKey::DEVICE);
592 53 : std::lock_guard lk(activeCallsMtx_);
593 53 : auto itActive = std::find_if(activeCalls_.begin(), activeCalls_.end(), [&](const auto& value) {
594 154 : return value.at("id") == confId && value.at("uri") == uri && value.at("device") == device;
595 : });
596 159 : 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 22 : if (itActive != activeCalls_.end()) {
616 22 : itActive = activeCalls_.erase(itActive);
617 : // Unlikely, but we must ensure that no duplicate exists
618 44 : 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 22 : 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 22 : JAMI_DEBUG("swarm:{:s} call finished: {:s} on device {:s}, account {:s}",
637 : convId,
638 : confId,
639 : device,
640 : uri);
641 : }
642 : }
643 22 : saveActiveCalls();
644 22 : if (emitSig)
645 22 : emitSignal<libjami::ConfigurationSignal::ActiveCallsChanged>(accountId_,
646 22 : repository_->id(),
647 22 : activeCalls_);
648 : }
649 53 : }
650 31 : }
651 :
652 1378 : void announce(const std::vector<std::map<std::string, std::string>>& commits, bool commitFromSelf = false)
653 : {
654 1378 : if (!repository_)
655 0 : return;
656 1378 : auto convId = repository_->id();
657 1378 : 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 1318 : auto ok = !commits.empty();
681 2637 : auto lastId = ok ? commits.rbegin()->at(ConversationMapKeys::ID) : "";
682 1317 : addToHistory(loadedHistory_, commits, true, commitFromSelf);
683 1317 : if (ok) {
684 1315 : bool announceMember = false;
685 2686 : for (const auto& c : commits) {
686 : // Announce member events
687 2746 : if (c.at(CommitKey::TYPE) == CommitType::MEMBER) {
688 4933 : if (c.find(CommitKey::URI) != c.end() && c.find(CommitKey::ACTION) != c.end()) {
689 1973 : const auto& uri = c.at(CommitKey::URI);
690 987 : const auto& actionStr = c.at(CommitKey::ACTION);
691 987 : auto action = -1;
692 987 : if (actionStr == CommitAction::ADD)
693 470 : action = 0;
694 517 : else if (actionStr == CommitAction::JOIN)
695 493 : action = 1;
696 24 : else if (actionStr == CommitAction::REMOVE)
697 3 : action = 2;
698 21 : else if (actionStr == CommitAction::BAN)
699 19 : action = 3;
700 2 : else if (actionStr == CommitAction::UNBAN)
701 2 : action = 4;
702 987 : 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 987 : if (action != -1) {
708 987 : announceMember = true;
709 987 : emitSignal<libjami::ConversationSignal::ConversationMemberEvent>(accountId_,
710 : convId,
711 : uri,
712 : action);
713 : }
714 : }
715 772 : } else if (c.at(CommitKey::TYPE) == CommitType::CALL_HISTORY) {
716 60 : updateActiveCalls(c);
717 : }
718 : #ifdef ENABLE_PLUGIN
719 1373 : if (auto& pluginChatManager = Manager::instance().getJamiPluginManager().getChatServicesManager();
720 1372 : 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 2739 : 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 1314 : if (announceMember && onMembersChanged_) {
791 986 : onMembersChanged_(repository_->memberUris("", {}));
792 : }
793 : }
794 1379 : }
795 :
796 492 : void loadStatus()
797 : {
798 : try {
799 : // read file
800 982 : 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 492 : } catch (const std::exception& e) {
806 490 : }
807 492 : }
808 1787 : void saveStatus()
809 : {
810 1787 : std::ofstream file(statusPath_, std::ios::trunc | std::ios::binary);
811 1787 : msgpack::pack(file, messagesStatus_);
812 1787 : }
813 :
814 492 : void loadMobileNodes()
815 : {
816 : try {
817 984 : 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 492 : } catch (const std::exception& e) {
829 492 : return;
830 492 : }
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 308 : void saveActiveCalls() const
848 : {
849 308 : std::ofstream file(activeCallsPath_, std::ios::trunc | std::ios::binary);
850 308 : msgpack::pack(file, activeCalls_);
851 308 : }
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 42 : void saveHostedCalls() const
868 : {
869 42 : std::ofstream file(hostedCallsPath_, std::ios::trunc | std::ios::binary);
870 42 : msgpack::pack(file, hostedCalls_);
871 42 : }
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 6575 : std::string_view memberBanType(const std::string& uri) const
882 : {
883 6575 : auto crt = fmt::format("{}.crt", uri);
884 6575 : auto bannedMember = repoPath_ / MemberPath::BANNED / MemberPath::MEMBERS / crt;
885 6574 : if (std::filesystem::is_regular_file(bannedMember))
886 23 : return "members"sv;
887 6553 : auto bannedAdmin = repoPath_ / MemberPath::BANNED / MemberPath::ADMINS / crt;
888 6551 : if (std::filesystem::is_regular_file(bannedAdmin))
889 0 : return "admins"sv;
890 6553 : auto bannedInvited = repoPath_ / MemberPath::BANNED / MemberPath::INVITED / uri;
891 6552 : if (std::filesystem::is_regular_file(bannedInvited))
892 2 : return "invited"sv;
893 6550 : return {};
894 6575 : }
895 :
896 7241 : bool isDeviceBanned(const std::string& deviceId) const
897 : {
898 7241 : auto crt = fmt::format("{}.crt", deviceId);
899 7242 : auto bannedDevice = repoPath_ / MemberPath::BANNED / MemberPath::DEVICES / crt;
900 14480 : return std::filesystem::is_regular_file(bannedDevice);
901 7242 : }
902 :
903 4678 : std::shared_ptr<dhtnet::ChannelSocket> gitSocket(const DeviceId& deviceId) const
904 : {
905 4678 : std::lock_guard lk(gitSocketMtx_);
906 4678 : auto deviceSockets = gitSocketList_.find(deviceId);
907 9358 : return (deviceSockets != gitSocketList_.end()) ? deviceSockets->second.get() : nullptr;
908 4679 : }
909 :
910 2055 : void addGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& socket)
911 : {
912 2055 : GitSocket replaced;
913 : {
914 2055 : std::lock_guard lk(gitSocketMtx_);
915 2055 : auto& slot = gitSocketList_[deviceId];
916 : // Re-registering the channel we already own must not close it.
917 2055 : if (slot.get() == socket)
918 1028 : return;
919 1027 : replaced = std::move(slot);
920 1027 : slot = socket;
921 2055 : }
922 : // Closing the replaced channel, if any, happens here, outside the lock.
923 2055 : }
924 989 : void removeGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& expected = {})
925 : {
926 989 : GitSocket socket;
927 : {
928 989 : std::lock_guard lk(gitSocketMtx_);
929 989 : auto deviceSockets = gitSocketList_.find(deviceId);
930 988 : if (deviceSockets == gitSocketList_.end())
931 460 : return;
932 : // A dead channel must not evict the one that replaced it.
933 529 : if (expected && deviceSockets->second.get() != expected)
934 0 : return;
935 527 : socket = std::move(deviceSockets->second);
936 528 : gitSocketList_.erase(deviceSockets);
937 988 : }
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 988 : }
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 492 : Conversation::Impl::setupMemberCallback()
1136 : {
1137 492 : repository_->onMembersChanged([this](const std::set<std::string>& memberUris) {
1138 : {
1139 1235 : std::lock_guard lk(trackedMembersMtx_);
1140 1235 : if (isTracking_) {
1141 298 : if (auto acc = account_.lock()) {
1142 633 : for (auto it = trackedMembers_.begin(); it != trackedMembers_.end();) {
1143 335 : if (memberUris.find(it->first) == memberUris.end()) {
1144 7 : acc->presenceManager()->untrackBuddy(it->first);
1145 7 : it = trackedMembers_.erase(it);
1146 : } else {
1147 328 : ++it;
1148 : }
1149 : }
1150 298 : }
1151 : }
1152 1235 : }
1153 :
1154 1236 : if (onMembersChanged_)
1155 1235 : onMembersChanged_(memberUris);
1156 1236 : });
1157 492 : }
1158 :
1159 : void
1160 682 : Conversation::Impl::startTracking(std::weak_ptr<Conversation> w)
1161 : {
1162 682 : auto acc = account_.lock();
1163 682 : if (!acc)
1164 0 : return;
1165 :
1166 : {
1167 682 : std::lock_guard lk(trackedMembersMtx_);
1168 682 : if (isTracking_)
1169 172 : return;
1170 510 : isTracking_ = true;
1171 1020 : presenceDeviceListenerToken_ = acc->presenceManager()->addDeviceListener(
1172 1020 : [w](const std::string& uri, const DeviceId& deviceId, bool online) {
1173 849 : if (auto sthis = w.lock()) {
1174 849 : if (online && sthis->isMember(uri)) {
1175 2022 : sthis->addKnownDevices({deviceId}, uri);
1176 : }
1177 849 : }
1178 849 : });
1179 682 : }
1180 :
1181 1020 : rotateTrackedMembers();
1182 682 : }
1183 :
1184 : void
1185 778 : 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 778 : uint64_t token = 0;
1192 778 : std::vector<std::string> urisToUntrack;
1193 : {
1194 778 : std::lock_guard lk(trackedMembersMtx_);
1195 778 : if (!isTracking_)
1196 268 : return;
1197 510 : isTracking_ = false;
1198 510 : token = presenceDeviceListenerToken_;
1199 510 : presenceDeviceListenerToken_ = 0;
1200 1439 : for (const auto& [uri, _] : trackedMembers_) {
1201 929 : urisToUntrack.push_back(uri);
1202 : }
1203 510 : trackedMembers_.clear();
1204 778 : }
1205 :
1206 510 : auto acc = account_.lock();
1207 510 : if (!acc)
1208 217 : return;
1209 :
1210 293 : if (token) {
1211 293 : acc->presenceManager()->removeDeviceListener(token);
1212 : }
1213 :
1214 902 : for (const auto& uri : urisToUntrack) {
1215 610 : acc->presenceManager()->untrackBuddy(uri);
1216 : }
1217 995 : }
1218 :
1219 : void
1220 879 : Conversation::Impl::rotateTrackedMembers(const std::string& memberUri, const DeviceId& deviceId)
1221 : {
1222 879 : auto acc = account_.lock();
1223 879 : if (!acc)
1224 0 : return;
1225 :
1226 879 : std::lock_guard lk(trackedMembersMtx_);
1227 879 : if (!isTracking_)
1228 298 : return;
1229 :
1230 581 : if (!memberUri.empty()) {
1231 71 : if (auto it = trackedMembers_.find(memberUri); it != trackedMembers_.end()) {
1232 69 : JAMI_WARNING("{} [device {}] Rotating tracked members after connection failure", toString(), deviceId);
1233 69 : auto& info = it->second;
1234 69 : info.failedDevices.insert(deviceId);
1235 69 : if (std::includes(info.failedDevices.begin(),
1236 : info.failedDevices.end(),
1237 : info.devices.begin(),
1238 : info.devices.end())) {
1239 37 : acc->presenceManager()->untrackBuddy(it->first);
1240 37 : trackedMembers_.erase(it);
1241 : }
1242 : } else {
1243 2 : return;
1244 : }
1245 : }
1246 :
1247 579 : auto members = repository_->members();
1248 579 : size_t N = members.size();
1249 579 : 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 579 : auto activeDevices = swarmManager_->getActiveNodesCount();
1253 579 : if (activeDevices >= 2 * K)
1254 0 : return;
1255 :
1256 579 : 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 579 : if (trackedMembers_.size() < K) {
1265 547 : std::vector<std::string> candidates;
1266 547 : candidates.reserve(N - trackedMembers_.size());
1267 1799 : for (const auto& m : members) {
1268 1252 : if (m.uri != memberUri && trackedMembers_.find(m.uri) == trackedMembers_.end()) {
1269 1174 : candidates.push_back(m.uri);
1270 : }
1271 : }
1272 547 : if (!candidates.empty()) {
1273 512 : std::vector<std::string> chosen;
1274 512 : std::sample(candidates.begin(),
1275 : candidates.end(),
1276 : std::back_inserter(chosen),
1277 512 : K - trackedMembers_.size(),
1278 512 : Manager::instance().getSeededRandomEngine());
1279 1485 : for (const auto& uri : chosen) {
1280 973 : acc->presenceManager()->trackBuddy(uri);
1281 973 : trackedMembers_.emplace(uri, TrackedMember {});
1282 : }
1283 512 : }
1284 547 : }
1285 1179 : }
1286 :
1287 : void
1288 369 : Conversation::Impl::onConnectionFailed(const DeviceId& deviceId, const std::string& memberUri)
1289 : {
1290 369 : rotateTrackedMembers(memberUri, deviceId);
1291 369 : }
1292 :
1293 : void
1294 634 : Conversation::Impl::monitorConnection(std::weak_ptr<Conversation> w)
1295 : {
1296 634 : if (!swarmManager_->isConnected()) {
1297 623 : startTracking(w);
1298 : }
1299 :
1300 634 : swarmManager_->onConnectionChanged([w](bool ok) {
1301 345 : dht::ThreadPool::io().run([w, ok] {
1302 345 : 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 345 : if (sthis->pimpl_->swarmManager_->isConnected()) {
1310 286 : sthis->pimpl_->stopTracking();
1311 : } else {
1312 59 : sthis->pimpl_->startTracking(w);
1313 : }
1314 345 : if (ok) {
1315 287 : if (sthis->pimpl_->bootstrapCb_)
1316 287 : sthis->pimpl_->bootstrapCb_();
1317 : }
1318 : #ifdef LIBJAMI_TEST
1319 345 : if (sthis->pimpl_->bootstrapCbTest_)
1320 40 : sthis->pimpl_->bootstrapCbTest_(sthis->id(),
1321 20 : ok ? BootstrapStatus::SUCCESS : BootstrapStatus::FAILED);
1322 : #endif
1323 345 : }
1324 345 : });
1325 345 : });
1326 634 : }
1327 :
1328 : bool
1329 22 : Conversation::Impl::isAdmin() const
1330 : {
1331 22 : auto adminsPath = repoPath_ / MemberPath::ADMINS;
1332 44 : return std::filesystem::is_regular_file(fileutils::getFullPath(adminsPath, userId_ + ".crt"));
1333 22 : }
1334 :
1335 : void
1336 13 : Conversation::Impl::disconnectFromDevice(const DeviceId& deviceId)
1337 : {
1338 26 : swarmManager_->deleteNode({deviceId});
1339 :
1340 13 : GitSocket socket;
1341 : {
1342 13 : std::lock_guard lk(gitSocketMtx_);
1343 13 : if (auto it = gitSocketList_.find(deviceId); it != gitSocketList_.end()) {
1344 13 : socket = std::move(it->second);
1345 13 : gitSocketList_.erase(it);
1346 : }
1347 13 : }
1348 : // The channel is closed here, outside the lock.
1349 13 : }
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 29 : for (const auto node : nodes)
1358 10 : if (peerUri == repository_->uriFromDevice(node.toString()))
1359 5 : devicesToRemove.emplace(node);
1360 :
1361 : {
1362 19 : std::lock_guard lk(gitSocketMtx_);
1363 42 : for (const auto& [deviceId, socket] : gitSocketList_) {
1364 23 : auto cert = socket ? socket->peerCertificate() : nullptr;
1365 46 : if ((cert && cert->issuer && cert->issuer->getId().toString() == peerUri)
1366 46 : || peerUri == repository_->uriFromDevice(deviceId.toString())) {
1367 12 : devicesToRemove.emplace(deviceId);
1368 : }
1369 23 : }
1370 19 : }
1371 :
1372 31 : for (const auto& deviceId : devicesToRemove)
1373 12 : disconnectFromDevice(deviceId);
1374 19 : }
1375 :
1376 : std::vector<std::map<std::string, std::string>>
1377 133 : Conversation::Impl::getMembers(bool includeInvited, bool includeLeft, bool includeBanned) const
1378 : {
1379 133 : std::vector<std::map<std::string, std::string>> result;
1380 133 : auto members = repository_->members();
1381 133 : std::lock_guard lk(messageStatusMtx_);
1382 401 : for (const auto& member : members) {
1383 268 : if (member.role == MemberRole::BANNED && !includeBanned) {
1384 0 : continue;
1385 : }
1386 268 : if (member.role == MemberRole::INVITED && !includeInvited)
1387 0 : continue;
1388 268 : if (member.role == MemberRole::LEFT && !includeLeft)
1389 0 : continue;
1390 268 : auto mm = member.map();
1391 268 : auto it = messagesStatus_.find(member.uri);
1392 267 : if (it != messagesStatus_.end()) {
1393 386 : auto readIt = it->second.find("read");
1394 193 : if (readIt != it->second.end())
1395 393 : mm[ConversationMapKeys::LAST_DISPLAYED] = readIt->second;
1396 : }
1397 267 : result.emplace_back(std::move(mm));
1398 268 : }
1399 266 : return result;
1400 133 : }
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 2458 : Conversation::Impl::loadMessages(const LogOptions& options, History* optHistory)
1459 : {
1460 2458 : auto history = optHistory ? optHistory : &loadedHistory_;
1461 :
1462 : // history->mutex is locked by the caller
1463 2458 : if (!repository_ || history->loading) {
1464 0 : return {};
1465 : }
1466 2458 : 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 2458 : bool limitNbOfCommits = options.nbOfCommits > 0;
1471 :
1472 2458 : auto startLogging = options.from == "";
1473 2458 : auto breakLogging = false;
1474 2458 : auto currentHistorySize = loadedHistory_.messageList.size();
1475 2458 : std::vector<std::string> replies;
1476 2458 : std::vector<std::shared_ptr<libjami::SwarmMessage>> msgList;
1477 4916 : repository_->log(
1478 : /* preCondition */
1479 4916 : [&](const auto& id, const auto& author, const auto& commit) {
1480 18652 : if (options.skipMerge && git_commit_parentcount(commit.get()) > 1) {
1481 4 : return CallbackResult::Skip;
1482 : }
1483 18647 : if (id == options.to) {
1484 1032 : if (options.includeTo)
1485 0 : breakLogging = true; // For the next commit
1486 : }
1487 18631 : 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 18642 : if ((limitNbOfCommits
1491 18642 : && (loadedHistory_.messageList.size() - currentHistorySize) == options.nbOfCommits))
1492 2 : return CallbackResult::Break; // Stop logging
1493 18640 : if (breakLogging)
1494 0 : return CallbackResult::Break; // Stop logging
1495 18640 : if (id == options.to && !options.includeTo) {
1496 1031 : return CallbackResult::Break; // Stop logging
1497 : }
1498 : }
1499 :
1500 17601 : if (!startLogging && options.from != "" && options.from == id)
1501 1771 : startLogging = true;
1502 17606 : if (!startLogging)
1503 27 : return CallbackResult::Skip; // Start logging after this one
1504 :
1505 17579 : if (options.fastLog) {
1506 16328 : 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 17576 : return CallbackResult::Ok; // Continue
1514 : },
1515 : /* emplaceCb */
1516 4916 : [&](auto&& cc) {
1517 17586 : if (limitNbOfCommits && (msgList.size() == options.nbOfCommits))
1518 581 : return;
1519 17008 : auto optMessage = repository_->convCommitToMap(cc);
1520 17002 : if (!optMessage.has_value())
1521 3 : return;
1522 16991 : auto message = optMessage.value();
1523 50984 : 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 16987 : auto it = std::find(replies.begin(), replies.end(), message.at("id"));
1530 16996 : if (it != replies.end()) {
1531 1 : replies.erase(it);
1532 : }
1533 16998 : std::shared_ptr<libjami::SwarmMessage> firstMsg;
1534 16998 : if ((history == &loadedHistory_) && msgList.empty() && !loadedHistory_.messageList.empty()) {
1535 0 : firstMsg = *loadedHistory_.messageList.rbegin();
1536 : }
1537 51009 : auto added = addToHistory(*history, {message}, false, false);
1538 17004 : if (!added.empty() && firstMsg) {
1539 0 : emitSignal<libjami::ConversationSignal::SwarmMessageUpdated>(accountId_, repository_->id(), *firstMsg);
1540 : }
1541 17004 : msgList.insert(msgList.end(), added.begin(), added.end());
1542 34008 : },
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 17570 : return limitNbOfCommits && (msgList.size() == options.nbOfCommits);
1551 : },
1552 2458 : options.from,
1553 2458 : options.logIfNotFound);
1554 :
1555 2458 : history->loading = false;
1556 2458 : history->cv.notify_all();
1557 :
1558 : // Convert for client (remove ptr)
1559 2458 : std::vector<libjami::SwarmMessage> ret;
1560 2458 : ret.reserve(msgList.size());
1561 19447 : for (const auto& msg : msgList) {
1562 16995 : ret.emplace_back(*msg);
1563 : }
1564 2455 : return ret;
1565 2456 : }
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 18331 : Conversation::Impl::handleMessage(History& history,
1678 : const std::shared_ptr<libjami::SwarmMessage>& sharedCommit,
1679 : bool messageReceived) const
1680 : {
1681 18331 : if (messageReceived) {
1682 : // For a received message, we place it at the beginning of the list
1683 1326 : if (!history.messageList.empty())
1684 1027 : sharedCommit->linearizedParent = (*history.messageList.begin())->id;
1685 1326 : 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 17005 : if (!history.messageList.empty())
1690 14572 : (*history.messageList.rbegin())->linearizedParent = sharedCommit->id;
1691 17001 : history.messageList.emplace_back(sharedCommit);
1692 : }
1693 : // Handle pending reactions/editions
1694 18331 : auto reactIt = history.pendingReactions.find(sharedCommit->id);
1695 18332 : 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 18337 : auto peditIt = history.pendingEditions.find(sharedCommit->id);
1701 18334 : 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 18329 : if (messageReceived)
1725 1325 : emitSignal<libjami::ConversationSignal::SwarmMessageReceived>(accountId_, repository_->id(), *sharedCommit);
1726 18336 : return !messageReceived;
1727 : }
1728 :
1729 : void
1730 18352 : Conversation::Impl::rectifyStatus(const std::shared_ptr<libjami::SwarmMessage>& message, History& history) const
1731 : {
1732 18352 : auto parentIt = history.quickAccess.find(message->linearizedParent);
1733 18353 : auto currentMessage = message;
1734 :
1735 30384 : while (parentIt != history.quickAccess.end()) {
1736 12094 : const auto& parent = parentIt->second;
1737 12371 : for (const auto& [peer, value] : message->status) {
1738 11204 : auto parentStatusIt = parent->status.find(peer);
1739 11131 : if (parentStatusIt == parent->status.end() || parentStatusIt->second < value) {
1740 290 : parent->status[peer] = value;
1741 580 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(accountId_,
1742 290 : repository_->id(),
1743 : peer,
1744 290 : parent->id,
1745 : value);
1746 10926 : } else if (parentStatusIt->second >= value) {
1747 10930 : break;
1748 : }
1749 : }
1750 12116 : currentMessage = parent;
1751 12127 : parentIt = history.quickAccess.find(parent->linearizedParent);
1752 : }
1753 18349 : }
1754 :
1755 : std::vector<std::shared_ptr<libjami::SwarmMessage>>
1756 18346 : 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 18346 : auto acc = account_.lock();
1773 18345 : if (!acc)
1774 0 : return {};
1775 18346 : auto username = acc->getUsername();
1776 18332 : 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 18332 : bool needToSetMessageStatus = !commitFromSelf && &history == &loadedHistory_;
1783 :
1784 18332 : 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 36719 : for (const auto& commit : commits) {
1792 18398 : auto typeIt = commit.find(CommitKey::TYPE);
1793 18386 : if (typeIt == commit.end() || typeIt->second != CommitType::COLLAB_DOC)
1794 18384 : continue;
1795 56 : auto editIt = commit.find(CommitKey::EDIT);
1796 56 : if (editIt == commit.end() || editIt->second.empty())
1797 53 : 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 36714 : for (const auto& commit : commits) {
1808 18390 : auto commitId = commit.at("id");
1809 18388 : if (history.quickAccess.find(commitId) != history.quickAccess.end())
1810 1 : continue; // Already present
1811 18398 : auto typeIt = commit.find(CommitKey::TYPE);
1812 : // Nothing to show for the client, skip
1813 18386 : if (typeIt != commit.end() && typeIt->second == CommitType::MERGE)
1814 44 : 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 18347 : if (typeIt != commit.end() && typeIt->second == CommitType::COLLAB_DOC) {
1820 : // Removals were applied above, before any announcement of this batch.
1821 56 : auto editIt = commit.find(CommitKey::EDIT);
1822 56 : bool isRemoval = editIt != commit.end() && !editIt->second.empty();
1823 56 : if (!isRemoval) {
1824 106 : if (auto uriIt = commit.find(CommitKey::URI); uriIt != commit.end() && !uriIt->second.empty())
1825 53 : acc->collaborativeEditing()->onDocumentAnnounced(repository_->id(), uriIt->second);
1826 : }
1827 : }
1828 :
1829 18352 : auto sharedCommit = std::make_shared<libjami::SwarmMessage>();
1830 18351 : sharedCommit->fromMapStringString(commit);
1831 :
1832 18347 : if (needToSetMessageStatus) {
1833 977 : std::lock_guard lk(messageStatusMtx_);
1834 : // Check if we already have status information for the commit.
1835 978 : auto itFuture = futureStatus.find(sharedCommit->id);
1836 977 : if (itFuture != futureStatus.end()) {
1837 13 : sharedCommit->status = std::move(itFuture->second);
1838 13 : futureStatus.erase(itFuture);
1839 : }
1840 976 : }
1841 18348 : sharedCommits.emplace_back(sharedCommit);
1842 18403 : }
1843 :
1844 18331 : if (needToSetMessageStatus) {
1845 971 : constexpr int32_t SENDING = static_cast<int32_t>(libjami::Account::MessageStates::SENDING);
1846 971 : constexpr int32_t SENT = static_cast<int32_t>(libjami::Account::MessageStates::SENT);
1847 971 : constexpr int32_t DISPLAYED = static_cast<int32_t>(libjami::Account::MessageStates::DISPLAYED);
1848 :
1849 971 : std::lock_guard lk(messageStatusMtx_);
1850 13683 : 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 12708 : auto status = SENDING;
1867 12708 : if (!messageReceived) {
1868 12 : auto cache = memberToStatus[member.uri];
1869 12 : if (cache > status)
1870 9 : status = cache;
1871 : }
1872 12708 : auto& messagesStatus = messagesStatus_[member.uri];
1873 :
1874 25391 : for (auto it = sharedCommits.rbegin(); it != sharedCommits.rend(); it++) {
1875 12727 : auto sharedCommit = *it;
1876 12712 : auto previousStatus = status;
1877 12712 : auto& commitStatus = sharedCommit->status[member.uri];
1878 :
1879 : // Compute status for the current commit.
1880 38048 : if (status < SENT && messagesStatus["fetched"] == sharedCommit->id) {
1881 12 : status = SENT;
1882 : }
1883 37964 : if (messagesStatus["read"] == sharedCommit->id) {
1884 1 : status = DISPLAYED;
1885 : }
1886 37977 : if (member.uri == sharedCommit->body.at("author")) {
1887 975 : status = DISPLAYED;
1888 : }
1889 12674 : if (status < commitStatus) {
1890 2 : status = commitStatus;
1891 : }
1892 :
1893 : // Store computed value.
1894 12674 : commitStatus = status;
1895 :
1896 : // Update messagesStatus_ if needed.
1897 12674 : if (previousStatus == SENDING && status >= SENT) {
1898 2918 : messagesStatus["fetched"] = sharedCommit->id;
1899 : }
1900 12673 : if (previousStatus <= SENT && status == DISPLAYED) {
1901 2883 : messagesStatus["read"] = sharedCommit->id;
1902 : }
1903 12672 : }
1904 :
1905 12711 : if (!messageReceived) {
1906 : // Update memberToStatus with the status of the last (i.e. oldest) added commit.
1907 12 : memberToStatus[member.uri] = status;
1908 : }
1909 970 : }
1910 971 : }
1911 :
1912 18332 : std::vector<std::shared_ptr<libjami::SwarmMessage>> messages;
1913 36688 : for (const auto& sharedCommit : sharedCommits) {
1914 18355 : history.quickAccess[sharedCommit->id] = sharedCommit;
1915 :
1916 36696 : auto reactToIt = sharedCommit->body.find(CommitKey::REACT_TO);
1917 36676 : auto editIt = sharedCommit->body.find(CommitKey::EDIT);
1918 18328 : if (reactToIt != sharedCommit->body.end() && !reactToIt->second.empty()) {
1919 3 : handleReaction(history, sharedCommit);
1920 18344 : } else if (editIt != sharedCommit->body.end() && !editIt->second.empty()) {
1921 12 : handleEdition(history, sharedCommit, messageReceived);
1922 18329 : } else if (handleMessage(history, sharedCommit, messageReceived)) {
1923 17005 : messages.emplace_back(sharedCommit);
1924 : }
1925 18348 : rectifyStatus(sharedCommit, history);
1926 : }
1927 :
1928 18343 : return messages;
1929 18344 : }
1930 :
1931 221 : Conversation::Conversation(const std::shared_ptr<JamiAccount>& account,
1932 : ConversationMode mode,
1933 221 : const std::string& otherMember)
1934 221 : : pimpl_ {new Impl {account, mode, otherMember}}
1935 221 : {}
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 254 : Conversation::Conversation(const std::shared_ptr<JamiAccount>& account,
1942 : const std::string& remoteDevice,
1943 254 : const std::string& conversationId)
1944 254 : : pimpl_ {new Impl {account, remoteDevice, conversationId}}
1945 254 : {}
1946 :
1947 492 : Conversation::~Conversation() {}
1948 :
1949 : std::string
1950 6519 : Conversation::id() const
1951 : {
1952 6519 : return pimpl_->repository_ ? pimpl_->repository_->id() : "";
1953 : }
1954 :
1955 : void
1956 181 : Conversation::addMember(const std::string& contactUri, const OnDoneCb& cb)
1957 : {
1958 : try {
1959 181 : 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 180 : 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 180 : if (isMemberBanned(contactUri)) {
1980 3 : if (pimpl_->isAdmin()) {
1981 2 : dht::ThreadPool::io().run([w = weak(), contactUri = std::move(contactUri), cb = std::move(cb)] {
1982 2 : if (auto sthis = w.lock()) {
1983 2 : auto members = sthis->pimpl_->repository_->members();
1984 2 : auto type = sthis->pimpl_->memberBanType(contactUri);
1985 2 : if (type.empty()) {
1986 0 : cb(false, {});
1987 0 : return;
1988 : }
1989 2 : sthis->pimpl_->voteUnban(contactUri, type, cb);
1990 4 : }
1991 : });
1992 : } else {
1993 1 : JAMI_WARNING("Unable to add member {} because this member is blocked", contactUri);
1994 2 : cb(false, "");
1995 : }
1996 3 : return;
1997 : }
1998 :
1999 177 : dht::ThreadPool::io().run([w = weak(), contactUri = std::move(contactUri), cb = std::move(cb)] {
2000 177 : if (auto sthis = w.lock()) {
2001 : // Add member files and commit
2002 177 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
2003 177 : auto commit = sthis->pimpl_->repository_->addMember(contactUri);
2004 177 : if (not commit.empty())
2005 177 : sthis->pimpl_->announce(commit, true);
2006 177 : lk.unlock();
2007 177 : if (cb)
2008 177 : cb(!commit.empty(), commit);
2009 354 : }
2010 177 : });
2011 : }
2012 :
2013 : std::shared_ptr<dhtnet::ChannelSocket>
2014 4678 : Conversation::gitSocket(const DeviceId& deviceId) const
2015 : {
2016 4678 : return pimpl_->gitSocket(deviceId);
2017 : }
2018 :
2019 : void
2020 2055 : Conversation::addGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& socket)
2021 : {
2022 2055 : pimpl_->addGitSocket(deviceId, socket);
2023 2054 : }
2024 :
2025 : void
2026 988 : Conversation::removeGitSocket(const DeviceId& deviceId, const std::shared_ptr<dhtnet::ChannelSocket>& expected)
2027 : {
2028 988 : pimpl_->removeGitSocket(deviceId, expected);
2029 988 : }
2030 :
2031 : void
2032 505 : Conversation::shutdownConnections()
2033 : {
2034 505 : GitSocketList gitSockets;
2035 : {
2036 505 : std::lock_guard lk(pimpl_->gitSocketMtx_);
2037 505 : gitSockets = std::move(pimpl_->gitSocketList_);
2038 505 : pimpl_->gitSocketList_.clear();
2039 505 : }
2040 : // Closing the channels wakes up any fetch currently blocked reading from them.
2041 505 : gitSockets.clear();
2042 505 : if (pimpl_->swarmManager_)
2043 505 : pimpl_->swarmManager_->shutdown();
2044 505 : }
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 2 : Conversation::Impl::voteUnban(const std::string& contactUri, const std::string_view type, const OnDoneCb& cb)
2073 : {
2074 : // Check if admin
2075 2 : 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 2 : std::unique_lock lk(writeMtx_);
2083 2 : auto voteCommit = repository_->voteUnban(contactUri, type);
2084 2 : if (voteCommit.empty()) {
2085 0 : JAMI_WARNING("Unbanning {} failed", contactUri);
2086 0 : cb(false, "");
2087 0 : return;
2088 : }
2089 :
2090 2 : auto lastId = voteCommit;
2091 2 : std::vector<std::string> commits;
2092 2 : commits.emplace_back(voteCommit);
2093 :
2094 : // If admin, check vote
2095 4 : auto resolveCommit = repository_->resolveVote(contactUri, type, CommitAction::UNBAN);
2096 2 : if (!resolveCommit.empty()) {
2097 2 : commits.emplace_back(resolveCommit);
2098 2 : lastId = resolveCommit;
2099 2 : JAMI_WARNING("Vote solved for unbanning {}.", contactUri);
2100 : }
2101 2 : announce(commits, true);
2102 2 : lk.unlock();
2103 2 : if (cb)
2104 2 : cb(!lastId.empty(), lastId);
2105 2 : }
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 30 : for (const auto& member : members) {
2127 30 : if (member.uri == contactUri) {
2128 15 : if (member.role == MemberRole::INVITED) {
2129 3 : type = "invited";
2130 12 : } else if (member.role == MemberRole::ADMIN) {
2131 1 : type = "admins";
2132 11 : } else if (member.role == MemberRole::MEMBER) {
2133 11 : 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 133 : Conversation::getMembers(bool includeInvited, bool includeLeft, bool includeBanned) const
2178 : {
2179 133 : 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 2729 : Conversation::memberUris(std::string_view filter, const std::set<MemberRole>& filteredRoles) const
2190 : {
2191 2729 : return pimpl_->repository_->memberUris(filter, filteredRoles);
2192 : }
2193 :
2194 : std::vector<NodeId>
2195 1797 : Conversation::peersToSyncWith() const
2196 : {
2197 1797 : auto s = pimpl_->swarmManager_->getConnectedNodes();
2198 1797 : std::lock_guard lk(pimpl_->gitSocketMtx_);
2199 12953 : for (const auto& [deviceId, _] : pimpl_->gitSocketList_)
2200 11158 : if (std::find(s.cbegin(), s.cend(), deviceId) == s.cend())
2201 1471 : s.emplace_back(deviceId);
2202 3594 : return s;
2203 1797 : }
2204 :
2205 : std::vector<MobileNodeTarget>
2206 1775 : Conversation::mobileNodesToNotify() const
2207 : {
2208 1775 : std::vector<MobileNodeTarget> targets;
2209 1775 : auto account = pimpl_->account_.lock();
2210 1774 : if (!account)
2211 0 : return targets;
2212 1774 : 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 1775 : return targets;
2228 1775 : }
2229 :
2230 : bool
2231 1775 : Conversation::isBootstrapped() const
2232 : {
2233 1775 : return pimpl_->swarmManager_->isConnected();
2234 : }
2235 :
2236 : std::string
2237 13763 : Conversation::uriFromDevice(const std::string& deviceId) const
2238 : {
2239 13763 : 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 237 : Conversation::join()
2256 : {
2257 237 : return pimpl_->repository_->join();
2258 : }
2259 :
2260 : bool
2261 9575 : Conversation::isMember(const std::string& uri, bool includeInvited) const
2262 : {
2263 9575 : auto uriCrt = uri + ".crt"sv;
2264 19146 : if (std::filesystem::is_regular_file(pimpl_->repoPath_ / MemberPath::ADMINS / uriCrt)
2265 19146 : || std::filesystem::is_regular_file(pimpl_->repoPath_ / MemberPath::MEMBERS / uriCrt)) {
2266 6847 : return true;
2267 : }
2268 2726 : if (includeInvited) {
2269 1961 : if (std::filesystem::is_regular_file(pimpl_->repoPath_ / MemberPath::INVITED / uri)) {
2270 1577 : return true;
2271 : }
2272 383 : 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 1148 : return false;
2280 9572 : }
2281 :
2282 : bool
2283 6572 : Conversation::isMemberBanned(const std::string& uri) const
2284 : {
2285 6572 : return !pimpl_->memberBanType(uri).empty();
2286 : }
2287 :
2288 : bool
2289 7239 : Conversation::isDeviceBanned(const std::string& deviceId) const
2290 : {
2291 7239 : return pimpl_->isDeviceBanned(deviceId);
2292 : }
2293 :
2294 : bool
2295 6354 : Conversation::isPeerAuthorized(const std::string& uri, const std::string& deviceId, bool includeInvited) const
2296 : {
2297 6354 : return !isMemberBanned(uri) && !isDeviceBanned(deviceId) && isMember(uri, includeInvited);
2298 : }
2299 :
2300 : void
2301 174 : Conversation::createCommit(CommitMessage&& message, OnCommitCb&& onCommit, OnDoneCb&& cb)
2302 : {
2303 174 : 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 346 : dht::ThreadPool::io().run(
2310 346 : [w = weak(), message = std::move(message), onCommit = std::move(onCommit), cb = std::move(cb)] {
2311 173 : if (auto sthis = w.lock()) {
2312 173 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
2313 173 : auto commit = sthis->pimpl_->repository_->commitMessage(message.toString());
2314 173 : lk.unlock();
2315 173 : if (onCommit)
2316 15 : onCommit(commit);
2317 173 : sthis->pimpl_->announce(commit, true);
2318 172 : if (cb)
2319 172 : cb(!commit.empty(), commit);
2320 347 : }
2321 172 : });
2322 : }
2323 :
2324 : bool
2325 11543 : Conversation::hasCommit(const std::string& commitId) const
2326 : {
2327 11543 : return pimpl_->repository_->hasCommit(commitId);
2328 : }
2329 :
2330 : std::optional<ConversationCommit>
2331 51 : Conversation::getCommit(const std::string& commitId) const
2332 : {
2333 51 : 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 2178 : Conversation::lastCommitId() const
2574 : {
2575 : {
2576 2178 : std::lock_guard lk(pimpl_->loadedHistory_.mutex);
2577 2179 : if (!pimpl_->loadedHistory_.messageList.empty())
2578 1514 : return (*pimpl_->loadedHistory_.messageList.begin())->id;
2579 2179 : }
2580 665 : LogOptions options;
2581 665 : options.nbOfCommits = 1;
2582 665 : options.skipMerge = true;
2583 665 : History optHistory;
2584 665 : std::scoped_lock lock(pimpl_->writeMtx_, optHistory.mutex);
2585 665 : auto res = pimpl_->loadMessages(options, &optHistory);
2586 665 : if (res.empty())
2587 2 : return {};
2588 663 : return (*optHistory.messageList.begin())->id;
2589 665 : }
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 5454 : auto [it, notInProgress] = pimpl_->fetchingRemotes_.emplace(deviceId,
2596 3633 : std::deque<std::pair<std::string, OnPullCb>>());
2597 1818 : 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 3636 : });
2601 1818 : 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 1818 : JAMI_DEBUG("{} [device {}] Pulling '{:s}'", pimpl_->toString(), deviceId, commitId);
2610 1818 : pullcbs.emplace_back(std::move(commitId), std::move(cb));
2611 1818 : if (notInProgress)
2612 1811 : dht::ThreadPool::io().run([w = weak(), deviceId] {
2613 1811 : if (auto sthis_ = w.lock())
2614 1811 : sthis_->pimpl_->pull(deviceId);
2615 1810 : });
2616 1818 : return true;
2617 1818 : }
2618 :
2619 : void
2620 1811 : Conversation::Impl::pull(const std::string& deviceId)
2621 : {
2622 1811 : auto& repo = repository_;
2623 :
2624 1811 : std::string commitId;
2625 1811 : OnPullCb cb;
2626 : while (true) {
2627 : {
2628 3627 : std::lock_guard lk(pullcbsMtx_);
2629 3628 : auto it = fetchingRemotes_.find(deviceId);
2630 3628 : if (it == fetchingRemotes_.end()) {
2631 0 : JAMI_ERROR("Could not find device {:s} in fetchingRemotes", deviceId);
2632 0 : break;
2633 : }
2634 3628 : auto& pullcbs = it->second;
2635 3629 : if (pullcbs.empty()) {
2636 1811 : fetchingRemotes_.erase(it);
2637 1810 : break;
2638 : }
2639 1818 : auto& elem = pullcbs.front();
2640 1818 : commitId = std::move(std::get<0>(elem));
2641 1818 : cb = std::move(std::get<1>(elem));
2642 1818 : pullcbs.pop_front();
2643 3628 : }
2644 : // If recently fetched, the commit can already be there, so no need to do complex operations
2645 1818 : if (commitId != "" && repo->hasCommit(commitId)) {
2646 15 : cb(true);
2647 56 : continue;
2648 : }
2649 : // Pull from remote
2650 1803 : auto fetched = repo->fetch(deviceId);
2651 1803 : if (!fetched) {
2652 41 : cb(false);
2653 41 : continue;
2654 : }
2655 :
2656 1762 : auto oldHead = repo->getHead();
2657 :
2658 1762 : std::unique_lock lk(writeMtx_);
2659 : auto commits = repo->mergeHistory(deviceId,
2660 1767 : [this](const std::string& peerUri) { this->disconnectFromPeer(peerUri); });
2661 1762 : if (!commits.empty()) {
2662 983 : announce(commits);
2663 : }
2664 1762 : lk.unlock();
2665 :
2666 1762 : bool commitFound = false;
2667 1762 : 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 1296 : for (const auto& commit : commits) {
2672 1970 : if (commit.at("id") == commitId) {
2673 974 : commitFound = true;
2674 974 : break;
2675 : }
2676 : }
2677 : } else {
2678 477 : commitFound = true;
2679 : }
2680 1761 : if (!commitFound)
2681 311 : 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 1761 : if (cb)
2692 1762 : cb(commitFound);
2693 :
2694 : // Announce if profile changed
2695 1761 : if (!commits.empty()) {
2696 1966 : auto diffStats = repo->diffStats("HEAD", oldHead);
2697 982 : auto changedFiles = repo->changedFiles(diffStats);
2698 983 : 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 983 : }
2704 3576 : }
2705 1809 : }
2706 :
2707 : void
2708 1817 : Conversation::sync(const std::string& member, const std::string& deviceId, OnPullCb&& cb, std::string commitId)
2709 : {
2710 1817 : 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 1817 : if (!sthis)
2714 0 : return;
2715 : // For waiting request, downloadFile
2716 1818 : for (const auto& wr : sthis->dataTransfer()->waitingRequests()) {
2717 1 : sthis->downloadFile(wr.interactionId, wr.fileId, wr.path, member, deviceId);
2718 1816 : }
2719 1817 : });
2720 1817 : }
2721 :
2722 : std::map<std::string, std::string>
2723 316 : Conversation::generateInvitation(TimePoint sent) const
2724 : {
2725 : // Invite the new member to the conversation
2726 316 : Json::Value root;
2727 316 : auto& metadata = root[ConversationMapKeys::METADATAS];
2728 638 : for (const auto& [k, v] : infos()) {
2729 322 : if (v.size() >= 64000) {
2730 0 : JAMI_WARNING("Cutting invite because the SIP message will be too long");
2731 0 : continue;
2732 : }
2733 322 : metadata[k] = v;
2734 316 : }
2735 316 : root[ConversationMapKeys::CONVERSATIONID] = id();
2736 316 : root[ConversationMapKeys::RECEIVED] = Json::Int64(toSecondsSinceEpoch(sent));
2737 316 : root[ConversationMapKeys::RECEIVED_MS] = Json::Int64(toMillisecondsSinceEpoch(sent));
2738 1264 : return {{"application/invite+json", json::toString(root)}};
2739 632 : }
2740 :
2741 : std::string
2742 13 : Conversation::leave()
2743 : {
2744 13 : setRemovingFlag();
2745 13 : std::lock_guard lk(pimpl_->writeMtx_);
2746 26 : return pimpl_->repository_->leave();
2747 13 : }
2748 :
2749 : void
2750 17 : Conversation::setRemovingFlag()
2751 : {
2752 17 : pimpl_->isRemoving_ = true;
2753 17 : }
2754 :
2755 : bool
2756 4815 : Conversation::isRemoving()
2757 : {
2758 4815 : return pimpl_->isRemoving_;
2759 : }
2760 :
2761 : void
2762 35 : Conversation::erase()
2763 : {
2764 35 : if (pimpl_->conversationDataPath_ != "")
2765 35 : dhtnet::fileutils::removeAll(pimpl_->conversationDataPath_, true);
2766 35 : if (!pimpl_->repository_)
2767 0 : return;
2768 35 : std::lock_guard lk(pimpl_->writeMtx_);
2769 35 : pimpl_->repository_->erase();
2770 35 : }
2771 :
2772 : ConversationMode
2773 2891 : Conversation::mode() const
2774 : {
2775 2891 : return pimpl_->repository_->mode();
2776 : }
2777 :
2778 : std::string
2779 54 : Conversation::parentConversationId() const
2780 : {
2781 54 : return pimpl_->repository_->parentConversationId();
2782 : }
2783 :
2784 : std::string
2785 0 : Conversation::documentMimeType() const
2786 : {
2787 0 : return pimpl_->repository_->documentMimeType();
2788 : }
2789 :
2790 : namespace {
2791 : // The base64 update lines of every checkpoint commit in a document log,
2792 : // oldest first, i.e. in the order the updates must be replayed.
2793 : std::vector<std::string>
2794 36 : collectUpdates(const std::vector<ConversationCommit>& commits)
2795 : {
2796 36 : std::vector<std::string> updates;
2797 202 : for (auto it = commits.rbegin(); it != commits.rend(); ++it) {
2798 166 : if (it->commitMsg.type != CommitType::CHECKPOINT)
2799 154 : continue;
2800 23 : for (const auto& line : split_string(it->commitMsg.body, '\n'))
2801 11 : if (!line.empty())
2802 23 : updates.emplace_back(line);
2803 : }
2804 36 : return updates;
2805 0 : }
2806 : } // namespace
2807 :
2808 : std::vector<std::string>
2809 36 : Conversation::documentUpdates() const
2810 : {
2811 36 : LogOptions options;
2812 36 : options.skipMerge = true;
2813 72 : return collectUpdates(pimpl_->repository_->log(options));
2814 36 : }
2815 :
2816 : std::optional<std::vector<std::string>>
2817 0 : Conversation::documentUpdatesAt(const std::string& commitId) const
2818 : {
2819 0 : if (!getCommit(commitId))
2820 0 : return std::nullopt;
2821 0 : LogOptions options;
2822 0 : options.from = commitId;
2823 0 : options.skipMerge = true;
2824 0 : return collectUpdates(pimpl_->repository_->log(options));
2825 0 : }
2826 :
2827 : std::vector<std::map<std::string, std::string>>
2828 61 : Conversation::documentHistory(size_t max) const
2829 : {
2830 61 : LogOptions options;
2831 61 : options.skipMerge = true;
2832 61 : auto commits = pimpl_->repository_->log(options);
2833 61 : std::vector<std::map<std::string, std::string>> result;
2834 236 : for (const auto& commit : commits) {
2835 175 : if (commit.commitMsg.type != CommitType::CHECKPOINT)
2836 157 : continue;
2837 18 : size_t deltas = 0;
2838 36 : for (const auto& line : split_string(commit.commitMsg.body, '\n'))
2839 18 : if (!line.empty())
2840 36 : ++deltas;
2841 126 : result.emplace_back(std::map<std::string, std::string> {
2842 18 : {"id", commit.id},
2843 18 : {"author", commit.authorId},
2844 18 : {"device", commit.author.email},
2845 0 : {"timestamp", std::to_string(commit.timestamp)},
2846 18 : {"deltas", std::to_string(deltas)},
2847 108 : });
2848 18 : if (max != 0 && result.size() >= max)
2849 0 : break;
2850 : }
2851 122 : return result;
2852 79 : }
2853 :
2854 : std::pair<std::string, std::string>
2855 1 : Conversation::addDocumentAttachment(const std::vector<uint8_t>& data)
2856 : {
2857 1 : std::unique_lock lk(pimpl_->writeMtx_);
2858 1 : auto headBefore = pimpl_->repository_->getHead();
2859 1 : auto attachmentId = pimpl_->repository_->addAttachment(data);
2860 1 : if (attachmentId.empty())
2861 0 : return {};
2862 1 : auto head = pimpl_->repository_->getHead();
2863 1 : if (head == headBefore)
2864 0 : return {attachmentId, {}}; // Same content already attached, nothing new to announce
2865 1 : pimpl_->announce(head, true);
2866 1 : return {attachmentId, head};
2867 1 : }
2868 :
2869 : std::vector<uint8_t>
2870 3 : Conversation::documentAttachment(const std::string& attachmentId) const
2871 : {
2872 3 : return pimpl_->repository_->attachment(attachmentId);
2873 : }
2874 :
2875 : std::vector<std::string>
2876 36 : Conversation::documentAttachmentIds() const
2877 : {
2878 36 : return pimpl_->repository_->attachmentIds();
2879 : }
2880 :
2881 : std::vector<std::string>
2882 32 : Conversation::getInitialMembers() const
2883 : {
2884 32 : return pimpl_->repository_->getInitialMembers();
2885 : }
2886 :
2887 : bool
2888 0 : Conversation::isInitialMember(const std::string& uri) const
2889 : {
2890 0 : auto members = getInitialMembers();
2891 0 : return std::find(members.begin(), members.end(), uri) != members.end();
2892 0 : }
2893 :
2894 : void
2895 27 : Conversation::updateInfos(const std::map<std::string, std::string>& map, const OnDoneCb& cb)
2896 : {
2897 27 : dht::ThreadPool::io().run([w = weak(), map = std::move(map), cb = std::move(cb)] {
2898 27 : if (auto sthis = w.lock()) {
2899 27 : auto& repo = sthis->pimpl_->repository_;
2900 27 : std::unique_lock lk(sthis->pimpl_->writeMtx_);
2901 27 : auto commit = repo->updateInfos(map);
2902 27 : sthis->pimpl_->announce(commit, true);
2903 27 : lk.unlock();
2904 27 : if (cb)
2905 27 : cb(!commit.empty(), commit);
2906 27 : if (repo->mode() == ConversationMode::DOCUMENT)
2907 18 : return; // A document is not a conversation for the client; a rename
2908 : // is reported through CollaborativeDocumentRenamed instead
2909 9 : emitSignal<libjami::ConversationSignal::ConversationProfileUpdated>(sthis->pimpl_->accountId_,
2910 9 : repo->id(),
2911 18 : repo->infos());
2912 72 : }
2913 : });
2914 27 : }
2915 :
2916 : std::map<std::string, std::string>
2917 401 : Conversation::infos() const
2918 : {
2919 401 : return pimpl_->repository_->infos();
2920 : }
2921 :
2922 : void
2923 7 : Conversation::updatePreferences(const std::map<std::string, std::string>& map)
2924 : {
2925 7 : const auto& filePath = pimpl_->preferencesPath_;
2926 7 : auto prefs = map;
2927 7 : auto itLast = prefs.find(LAST_MODIFIED);
2928 7 : if (itLast != prefs.end()) {
2929 2 : std::error_code ec;
2930 2 : if (std::filesystem::is_regular_file(filePath, ec)) {
2931 0 : auto lastModified = fileutils::lastWriteTimeInSeconds(filePath);
2932 : try {
2933 0 : if (lastModified >= to_int<uint64_t>(itLast->second))
2934 0 : return;
2935 0 : } catch (...) {
2936 0 : return;
2937 0 : }
2938 : }
2939 2 : prefs.erase(itLast);
2940 : }
2941 :
2942 7 : std::ofstream file(filePath, std::ios::trunc | std::ios::binary);
2943 7 : msgpack::pack(file, prefs);
2944 7 : emitSignal<libjami::ConversationSignal::ConversationPreferencesUpdated>(pimpl_->accountId_, id(), std::move(prefs));
2945 7 : }
2946 :
2947 : std::map<std::string, std::string>
2948 94 : Conversation::preferences(bool includeLastModified) const
2949 : {
2950 : try {
2951 94 : std::map<std::string, std::string> preferences;
2952 94 : const auto& filePath = pimpl_->preferencesPath_;
2953 178 : auto file = fileutils::loadFile(filePath);
2954 10 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
2955 10 : oh.get().convert(preferences);
2956 10 : if (includeLastModified)
2957 21 : preferences[LAST_MODIFIED] = std::to_string(fileutils::lastWriteTimeInSeconds(filePath));
2958 10 : return preferences;
2959 178 : } catch (const std::exception& e) {
2960 84 : }
2961 84 : return {};
2962 : }
2963 :
2964 : std::vector<uint8_t>
2965 0 : Conversation::vCard() const
2966 : {
2967 : try {
2968 0 : return fileutils::loadFile(pimpl_->repoPath_ / "profile.vcf");
2969 0 : } catch (...) {
2970 0 : }
2971 0 : return {};
2972 : }
2973 :
2974 : std::shared_ptr<TransferManager>
2975 1966 : Conversation::dataTransfer() const
2976 : {
2977 1966 : return pimpl_->transferManager_;
2978 : }
2979 :
2980 : bool
2981 19 : Conversation::onFileChannelRequest(const std::string& member,
2982 : const std::string& fileId,
2983 : std::filesystem::path& path,
2984 : std::string& sha3sum) const
2985 : {
2986 19 : if (!isMember(member))
2987 0 : return false;
2988 :
2989 19 : if (!isValidFileId(fileId)) {
2990 2 : JAMI_WARNING("[Account {:s}] {} requested file with invalid id {}", pimpl_->accountId_, member, fileId);
2991 2 : return false;
2992 : }
2993 17 : auto sep = fileId.find('_');
2994 17 : auto interactionId = fileId.substr(0, sep);
2995 17 : auto commit = getCommit(interactionId);
2996 34 : if (commit == std::nullopt || commit->commitMsg.tid.empty() || commit->commitMsg.sha3sum.empty()
2997 34 : || commit->commitMsg.type != CommitType::DATA_TRANSFER) {
2998 0 : JAMI_WARNING("[Account {:s}] {} requested invalid file transfer commit {}",
2999 : pimpl_->accountId_,
3000 : member,
3001 : interactionId);
3002 0 : return false;
3003 : }
3004 : // The commit is the only source of truth for the file name
3005 17 : if (fileId != getFileId(interactionId, commit->commitMsg.tid, commit->commitMsg.displayName)) {
3006 2 : JAMI_WARNING("[Account {:s}] {} requested file {} not matching commit {}",
3007 : pimpl_->accountId_,
3008 : member,
3009 : fileId,
3010 : interactionId);
3011 2 : return false;
3012 : }
3013 :
3014 15 : path = dataTransfer()->path(fileId);
3015 15 : sha3sum = commit->commitMsg.sha3sum;
3016 :
3017 15 : return true;
3018 17 : }
3019 :
3020 : bool
3021 24 : Conversation::downloadFile(const std::string& interactionId,
3022 : const std::string& fileId,
3023 : const std::string& path,
3024 : const std::string&,
3025 : const std::string& deviceId)
3026 : {
3027 24 : auto commit = getCommit(interactionId);
3028 24 : if (commit == std::nullopt || commit->commitMsg.type != CommitType::DATA_TRANSFER) {
3029 0 : JAMI_ERROR("Commit doesn't exists or is not a file transfer {} (Conversation: {}) ", interactionId, id());
3030 0 : return false;
3031 : }
3032 24 : auto tid = commit->commitMsg.tid;
3033 24 : auto sha3sum = commit->commitMsg.sha3sum;
3034 24 : auto totalSize = commit->commitMsg.totalSize;
3035 :
3036 24 : if (tid.empty() || sha3sum.empty() || totalSize < 0) {
3037 0 : JAMI_ERROR("Invalid file transfer commit (missing tid, size or sha3)");
3038 0 : return false;
3039 : }
3040 24 : if (!isValidFileId(fileId) || fileId != getFileId(interactionId, tid, commit->commitMsg.displayName)) {
3041 7 : JAMI_ERROR("File id {} does not match file transfer commit {}", fileId, interactionId);
3042 7 : return false;
3043 : }
3044 17 : if (!path.empty() && std::filesystem::path(path).is_relative()) {
3045 1 : JAMI_ERROR("Refusing relative file transfer destination {}", path);
3046 1 : return false;
3047 : }
3048 :
3049 : // Be sure to not lock conversation
3050 16 : dht::ThreadPool::io().run([w = weak(), deviceId, fileId, interactionId, sha3sum, path, totalSize] {
3051 16 : auto shared = w.lock();
3052 16 : if (!shared)
3053 0 : return;
3054 16 : auto transferManager = shared->dataTransfer();
3055 16 : const auto emitTransferEvent = [accountId = shared->pimpl_->accountId_,
3056 : conversationId = shared->id(),
3057 16 : interactionId,
3058 16 : fileId](libjami::DataTransferEventCode code) {
3059 2 : emitSignal<libjami::DataTransferSignal::DataTransferEvent>(accountId,
3060 2 : conversationId,
3061 2 : interactionId,
3062 2 : fileId,
3063 : uint32_t(code));
3064 18 : };
3065 16 : switch (transferManager->waitForTransfer(fileId, interactionId, sha3sum, path, totalSize)) {
3066 2 : case TransferManager::WaitResult::complete:
3067 2 : JAMI_LOG("Ignoring request to download available file: {}", fileId);
3068 2 : emitTransferEvent(libjami::DataTransferEventCode::finished);
3069 2 : return;
3070 0 : case TransferManager::WaitResult::conflict:
3071 0 : emitTransferEvent(libjami::DataTransferEventCode::invalid_pathname);
3072 0 : return;
3073 14 : case TransferManager::WaitResult::waiting:
3074 14 : break;
3075 : }
3076 14 : auto acc = shared->pimpl_->account_.lock();
3077 14 : if (!acc)
3078 0 : return;
3079 : // Resume from the partial file left by a previous attempt, if any
3080 14 : auto destination = path.empty() ? transferManager->path(fileId) : std::filesystem::path(path);
3081 14 : std::error_code ec;
3082 14 : auto start = std::filesystem::file_size(transferManager->temporaryPath(fileId, destination), ec);
3083 14 : if (ec || start == static_cast<decltype(start)>(-1))
3084 12 : start = 0;
3085 14 : size_t end = 0;
3086 14 : acc->askForFileChannel(shared->id(), deviceId, interactionId, fileId, start, end);
3087 20 : });
3088 16 : return true;
3089 24 : }
3090 :
3091 : void
3092 1985 : Conversation::hasFetched(const std::string& deviceId, const std::string& commitId)
3093 : {
3094 1985 : dht::ThreadPool::io().run([w = weak(), deviceId, commitId]() {
3095 1986 : auto sthis = w.lock();
3096 1986 : if (!sthis)
3097 0 : return;
3098 : // Update fetched for Uri
3099 1986 : auto uri = sthis->uriFromDevice(deviceId);
3100 1986 : if (uri.empty() || uri == sthis->pimpl_->userId_)
3101 45 : return;
3102 : // When a user fetches a commit, the message is sent for this person
3103 1941 : sthis->pimpl_->updateStatus(uri,
3104 : libjami::Account::MessageStates::SENT,
3105 1941 : commitId,
3106 3882 : std::to_string(std::time(nullptr)),
3107 : true);
3108 2031 : });
3109 1986 : }
3110 :
3111 : void
3112 1994 : Conversation::Impl::updateStatus(const std::string& uri,
3113 : libjami::Account::MessageStates st,
3114 : const std::string& commitId,
3115 : const std::string& ts,
3116 : bool emit)
3117 : {
3118 : // This method can be called if peer send us a status or if another device sync. Emit will be true if a peer
3119 : // send us a status and will emit to other connected devices.
3120 1994 : LogOptions options;
3121 1993 : std::map<std::string, std::map<std::string, std::string>> newStatus;
3122 : {
3123 : // Update internal structures.
3124 1994 : std::lock_guard lk(messageStatusMtx_);
3125 1994 : auto& status = messagesStatus_[uri];
3126 1994 : auto& oldStatus = status[st == libjami::Account::MessageStates::SENT ? "fetched" : "read"];
3127 1994 : if (oldStatus == commitId)
3128 207 : return; // Nothing to do
3129 1787 : options.to = oldStatus;
3130 1787 : options.from = commitId;
3131 1787 : oldStatus = commitId;
3132 1787 : status[st == libjami::Account::MessageStates::SENT ? "fetched_ts" : "read_ts"] = ts;
3133 1787 : saveStatus();
3134 1787 : if (emit)
3135 1750 : newStatus[uri].insert(status.begin(), status.end());
3136 1994 : }
3137 1787 : if (emit && messageStatusCb_) {
3138 1730 : messageStatusCb_(newStatus);
3139 : }
3140 : // Update messages status for all commit between the old and new one
3141 1787 : options.logIfNotFound = false;
3142 1787 : options.fastLog = true;
3143 1787 : History optHistory;
3144 1787 : std::unique_lock lk(optHistory.mutex); // Avoid to announce messages while updating status.
3145 1787 : auto res = loadMessages(options, &optHistory);
3146 1785 : std::unique_lock mlk(messageStatusMtx_);
3147 1787 : std::vector<std::pair<std::string, int32_t>> statusToUpdate;
3148 1787 : if (res.size() == 0) {
3149 : // In this case, commit is not received yet, so we cache it
3150 21 : futureStatus[commitId][uri] = static_cast<int32_t>(st);
3151 : }
3152 18112 : for (const auto& [cid, _] : optHistory.quickAccess) {
3153 16325 : auto message = loadedHistory_.quickAccess.find(cid);
3154 16325 : if (message != loadedHistory_.quickAccess.end()) {
3155 : // Update message and emit to client,
3156 4879 : if (static_cast<int32_t>(st) > message->second->status[uri]) {
3157 4873 : message->second->status[uri] = static_cast<int32_t>(st);
3158 4872 : statusToUpdate.emplace_back(cid, static_cast<int32_t>(st));
3159 : }
3160 : } else {
3161 : // In this case, commit is not loaded by client, so we cache it
3162 : // No need to emit to client, they will get a correct status on load.
3163 11447 : futureStatus[cid][uri] = static_cast<int32_t>(st);
3164 : }
3165 : }
3166 1787 : mlk.unlock();
3167 1787 : lk.unlock();
3168 6658 : for (const auto& [cid, status] : statusToUpdate)
3169 9741 : emitSignal<libjami::ConfigurationSignal::AccountMessageStatusChanged>(accountId_,
3170 4872 : repository_->id(),
3171 : uri,
3172 : cid,
3173 : static_cast<int>(status));
3174 2199 : }
3175 :
3176 : bool
3177 18 : Conversation::setMessageDisplayed(const std::string& uri, const std::string& interactionId)
3178 : {
3179 18 : std::lock_guard lk(pimpl_->messageStatusMtx_);
3180 54 : if (pimpl_->messagesStatus_[uri]["read"] == interactionId)
3181 2 : return false; // Nothing to do
3182 16 : dht::ThreadPool::io().run([w = weak(), uri, interactionId]() {
3183 16 : auto sthis = w.lock();
3184 16 : if (!sthis)
3185 0 : return;
3186 16 : sthis->pimpl_->updateStatus(uri,
3187 : libjami::Account::MessageStates::DISPLAYED,
3188 16 : interactionId,
3189 32 : std::to_string(std::time(nullptr)),
3190 : true);
3191 16 : });
3192 16 : return true;
3193 18 : }
3194 :
3195 : std::map<std::string, std::map<std::string, std::string>>
3196 76 : Conversation::messageStatus() const
3197 : {
3198 76 : std::lock_guard lk(pimpl_->messageStatusMtx_);
3199 152 : return pimpl_->messagesStatus_;
3200 76 : }
3201 :
3202 : void
3203 51 : Conversation::updateMessageStatus(const std::map<std::string, std::map<std::string, std::string>>& messageStatus)
3204 : {
3205 51 : std::unique_lock lk(pimpl_->messageStatusMtx_);
3206 51 : std::vector<std::tuple<libjami::Account::MessageStates, std::string, std::string, std::string>> stVec;
3207 118 : for (const auto& [uri, status] : messageStatus) {
3208 67 : auto& oldMs = pimpl_->messagesStatus_[uri];
3209 401 : if (status.find("fetched_ts") != status.end() && status.at("fetched") != oldMs["fetched"]) {
3210 155 : if (oldMs["fetched_ts"].empty() || std::stol(oldMs["fetched_ts"]) <= std::stol(status.at("fetched_ts"))) {
3211 0 : stVec.emplace_back(libjami::Account::MessageStates::SENT,
3212 : uri,
3213 33 : status.at("fetched"),
3214 99 : status.at("fetched_ts"));
3215 : }
3216 : }
3217 229 : if (status.find("read_ts") != status.end() && status.at("read") != oldMs["read"]) {
3218 12 : if (oldMs["read_ts"].empty() || std::stol(oldMs["read_ts"]) <= std::stol(status.at("read_ts"))) {
3219 0 : stVec.emplace_back(libjami::Account::MessageStates::DISPLAYED,
3220 : uri,
3221 4 : status.at("read"),
3222 12 : status.at("read_ts"));
3223 : }
3224 : }
3225 : }
3226 51 : lk.unlock();
3227 :
3228 88 : for (const auto& [status, uri, commitId, ts] : stVec) {
3229 37 : pimpl_->updateStatus(uri, status, commitId, ts);
3230 : }
3231 51 : }
3232 :
3233 : void
3234 476 : Conversation::onMessageStatusChanged(
3235 : const std::function<void(const std::map<std::string, std::map<std::string, std::string>>&)>& cb)
3236 : {
3237 476 : std::unique_lock lk(pimpl_->messageStatusMtx_);
3238 476 : pimpl_->messageStatusCb_ = cb;
3239 476 : }
3240 :
3241 : #ifdef LIBJAMI_TEST
3242 : void
3243 635 : Conversation::onBootstrapStatus(const std::function<void(std::string, BootstrapStatus)>& cb)
3244 : {
3245 635 : std::lock_guard lock(pimpl_->bootstrapMtx_);
3246 635 : pimpl_->bootstrapCbTest_ = cb;
3247 635 : }
3248 :
3249 : std::vector<libjami::SwarmMessage>
3250 0 : Conversation::loadMessagesSync(const LogOptions& options)
3251 : {
3252 0 : std::lock_guard lk(pimpl_->loadedHistory_.mutex);
3253 0 : auto result = pimpl_->loadMessages(options);
3254 0 : return result;
3255 0 : }
3256 :
3257 : void
3258 0 : Conversation::announce(const std::vector<std::map<std::string, std::string>>& commits, bool commitFromSelf)
3259 : {
3260 0 : pimpl_->announce(commits, commitFromSelf);
3261 0 : }
3262 :
3263 : void
3264 0 : Conversation::announce(const std::string& commitId, bool commitFromSelf)
3265 : {
3266 0 : pimpl_->announce(commitId, commitFromSelf);
3267 0 : }
3268 : #endif
3269 :
3270 : void
3271 634 : Conversation::bootstrap(std::function<void()> onBootstrapped, const std::vector<DeviceId>& knownDevices)
3272 : {
3273 634 : std::lock_guard lock(pimpl_->bootstrapMtx_);
3274 634 : if (!pimpl_ || !pimpl_->repository_ || !pimpl_->swarmManager_)
3275 0 : return;
3276 : // Bootstrap the DRT from currently known devices. Since the per-device
3277 : // presence monitoring (monitorConnection/startTracking below), callers do
3278 : // not pass any device list here: candidates are injected as members get
3279 : // online, through addKnownDevices() with devices reported by the
3280 : // PresenceManager (i.e. announced on the DHT), and rotated on connection
3281 : // failure. The knownDevices parameter remains for callers/tests that
3282 : // already hold a list of live devices.
3283 : // If a connection succeeds, onConnectionChanged will be called with ok=true
3284 634 : pimpl_->bootstrapCb_ = std::move(onBootstrapped);
3285 634 : std::vector<DeviceId> devices = knownDevices;
3286 634 : JAMI_DEBUG("{} Bootstrap with {} device(s)", pimpl_->toString(), devices.size());
3287 :
3288 634 : if (!devices.empty()) {
3289 0 : pimpl_->swarmManager_->setKnownNodes(devices);
3290 : }
3291 :
3292 634 : pimpl_->monitorConnection(weak_from_this());
3293 :
3294 : // If is shutdown, the conversation was re-added, causing no new nodes to be connected, but just a classic
3295 : // connectivity change
3296 634 : if (pimpl_->swarmManager_->isShutdown()) {
3297 20 : pimpl_->swarmManager_->restart();
3298 20 : pimpl_->swarmManager_->maintainBuckets();
3299 614 : } else if (!pimpl_->swarmManager_->isConnected()) {
3300 : // A swarm manager that is up but holds no connected node will never get
3301 : // one on its own: setKnownNodes() only acts on ids it has never seen
3302 : // before, so members it already knows are simply skipped.
3303 : //
3304 : // A mobile client reaches that state every time it leaves and returns
3305 : // to the foreground: setAccountActive() leaves established connections
3306 : // alone by default, so the swarm manager is not shut down, yet the
3307 : // links die with the network. Bootstrapping then finds nothing to do,
3308 : // and the conversation stays silent until the process is restarted.
3309 603 : pimpl_->swarmManager_->maintainBuckets();
3310 : }
3311 634 : }
3312 :
3313 : void
3314 1855 : Conversation::addKnownDevices(const std::vector<DeviceId>& devices, const std::string& memberUri)
3315 : {
3316 1855 : if (devices.empty())
3317 1017 : return;
3318 838 : if (!memberUri.empty()) {
3319 : // JAMI_WARNING("{} Adding {} known devices for member {}", pimpl_->toString(), devices.size(), memberUri);
3320 838 : std::lock_guard lk(pimpl_->trackedMembersMtx_);
3321 838 : auto it = pimpl_->trackedMembers_.find(memberUri);
3322 838 : if (it != pimpl_->trackedMembers_.end()) {
3323 833 : it->second.devices.insert(devices.begin(), devices.end());
3324 : }
3325 838 : } else {
3326 0 : JAMI_ERROR("{} Adding {} known devices without member URI", pimpl_->toString(), devices.size());
3327 : }
3328 838 : pimpl_->swarmManager_->setKnownNodes(devices);
3329 : }
3330 :
3331 : void
3332 518 : Conversation::connectNode(const DeviceId& deviceId)
3333 : {
3334 518 : pimpl_->swarmManager_->connectNode(deviceId);
3335 518 : }
3336 :
3337 : std::vector<std::string>
3338 18 : Conversation::commitsEndedCalls()
3339 : {
3340 18 : pimpl_->loadActiveCalls();
3341 18 : pimpl_->loadHostedCalls();
3342 18 : auto commits = pimpl_->commitsEndedCalls();
3343 18 : if (!commits.empty()) {
3344 : // Announce to client
3345 0 : dht::ThreadPool::io().run([w = weak(), commits] {
3346 0 : if (auto sthis = w.lock())
3347 0 : sthis->pimpl_->announce(commits, true);
3348 0 : });
3349 : }
3350 18 : return commits;
3351 0 : }
3352 :
3353 : void
3354 492 : Conversation::onMembersChanged(OnMembersChanged&& cb)
3355 : {
3356 492 : pimpl_->onMembersChanged_ = std::move(cb);
3357 492 : }
3358 :
3359 : void
3360 492 : Conversation::onNeedSocket(NeedSocketCb needSocket)
3361 : {
3362 984 : pimpl_->swarmManager_->needSocketCb_ = [needSocket = std::move(needSocket),
3363 : w = weak()](const std::string& deviceId, ChannelCb&& cb, bool noNewSocket) {
3364 1128 : if (auto sthis = w.lock()) {
3365 1128 : auto wrappedCb = [cb = std::move(cb), w, deviceId](const std::shared_ptr<dhtnet::ChannelSocket>& socket) {
3366 1125 : if (auto sthis = w.lock()) {
3367 1121 : if (!socket) {
3368 368 : if (auto acc = sthis->pimpl_->account_.lock()) {
3369 369 : auto cert = acc->certStore().getCertificate(deviceId);
3370 369 : if (cert && cert->issuer) {
3371 369 : sthis->pimpl_->onConnectionFailed(DeviceId(deviceId), cert->issuer->getId().toString());
3372 : } else {
3373 0 : JAMI_WARNING("{} Unable to get member URI from device ID {}",
3374 : sthis->pimpl_->toString(),
3375 : deviceId);
3376 : }
3377 369 : } else {
3378 0 : return false;
3379 369 : }
3380 : }
3381 1128 : }
3382 1128 : return cb(socket);
3383 1128 : };
3384 2254 : needSocket(sthis->id(), deviceId, std::move(wrappedCb), "application/im-gitmessage-id", noNewSocket);
3385 2256 : }
3386 1620 : };
3387 492 : }
3388 :
3389 : void
3390 986 : Conversation::addSwarmChannel(std::shared_ptr<dhtnet::ChannelSocket> channel)
3391 : {
3392 986 : auto deviceId = channel->deviceId();
3393 : // Transmit avatar if necessary
3394 : // We do this here, because at this point we know both sides are connected and in
3395 : // the same conversation
3396 : // addSwarmChannel is a bit more complex, but it should be the best moment to do this.
3397 987 : auto cert = channel->peerCertificate();
3398 987 : if (!cert || !cert->issuer)
3399 0 : return;
3400 986 : auto member = cert->issuer->getId().toString();
3401 : // The TLS handshake authenticated this certificate: pin it so that any mobile
3402 : // lease this device gossips can be verified without a lookup.
3403 987 : if (auto account = pimpl_->account_.lock())
3404 986 : account->certStore().pinCertificate(cert);
3405 987 : pimpl_->swarmManager_->addChannel(std::move(channel));
3406 987 : dht::ThreadPool::io().run([member, deviceId, a = pimpl_->account_, w = weak_from_this()] {
3407 987 : auto sthis = w.lock();
3408 987 : if (auto account = a.lock()) {
3409 987 : account->sendProfile(sthis->id(), member, deviceId.toString());
3410 986 : }
3411 987 : });
3412 987 : }
3413 :
3414 : uint32_t
3415 4 : Conversation::countInteractions(const std::string& toId, const std::string& fromId, const std::string& authorUri) const
3416 : {
3417 4 : LogOptions options;
3418 4 : options.to = toId;
3419 4 : options.from = fromId;
3420 4 : options.authorUri = authorUri;
3421 4 : options.logIfNotFound = false;
3422 4 : options.fastLog = true;
3423 4 : History history;
3424 4 : std::lock_guard lk(history.mutex);
3425 4 : auto res = pimpl_->loadMessages(options, &history);
3426 8 : return res.size();
3427 4 : }
3428 :
3429 : void
3430 4 : Conversation::search(uint32_t req, const Filter& filter, const std::shared_ptr<std::atomic_int>& flag) const
3431 : {
3432 : // Because logging a conversation can take quite some time,
3433 : // do it asynchronously
3434 4 : dht::ThreadPool::io().run([w = weak(), req, filter, flag] {
3435 4 : if (auto sthis = w.lock()) {
3436 4 : History history;
3437 4 : std::vector<std::map<std::string, std::string>> commits {};
3438 : // std::regex_constants::ECMAScript is the default flag.
3439 4 : auto re = std::regex(filter.regexSearch,
3440 4 : filter.caseSensitive ? std::regex_constants::ECMAScript : std::regex_constants::icase);
3441 12 : sthis->pimpl_->repository_->log(
3442 8 : [&](const std::string& /*id*/, const GitAuthor& author, const GitCommit& commit) {
3443 20 : if (!filter.author.empty() && filter.author != sthis->uriFromDevice(author.email)) {
3444 : // Filter author
3445 0 : return CallbackResult::Skip;
3446 : }
3447 20 : auto commitTime = git_commit_time(commit.get());
3448 20 : if (filter.before && filter.before < commitTime) {
3449 : // Only get commits before this date
3450 0 : return CallbackResult::Skip;
3451 : }
3452 20 : if (filter.after && filter.after > commitTime) {
3453 : // Only get commits before this date
3454 0 : if (git_commit_parentcount(commit.get()) <= 1)
3455 0 : return CallbackResult::Break;
3456 : else
3457 0 : return CallbackResult::Skip; // Because we are sorting it with
3458 : // GIT_SORT_TOPOLOGICAL | GIT_SORT_TIME
3459 : }
3460 :
3461 20 : return CallbackResult::Ok; // Continue
3462 : },
3463 8 : [&](ConversationCommit&& cc) {
3464 20 : if (auto optMessage = sthis->pimpl_->repository_->convCommitToMap(cc))
3465 80 : sthis->pimpl_->addToHistory(history, {optMessage.value()}, false, false);
3466 40 : },
3467 8 : [&](const std::string& id, const GitAuthor&, ConversationCommit&) {
3468 20 : if (id == filter.lastId)
3469 0 : return true;
3470 20 : return false;
3471 : },
3472 : "",
3473 : false);
3474 : // Search on generated history
3475 24 : for (auto& message : history.messageList) {
3476 20 : auto contentType = message->type;
3477 20 : auto isSearchable = contentType == CommitType::TEXT || contentType == CommitType::DATA_TRANSFER;
3478 20 : if (filter.type.empty() && !isSearchable) {
3479 : // Not searchable, at least for now
3480 8 : continue;
3481 12 : } else if (contentType == filter.type || filter.type.empty()) {
3482 12 : if (isSearchable) {
3483 : // If it's a text match the body, else the display name
3484 60 : auto body = contentType == CommitType::TEXT ? message->body.at(CommitKey::BODY)
3485 36 : : message->body.at(CommitKey::DISPLAY_NAME);
3486 12 : std::smatch body_match;
3487 12 : if (std::regex_search(body, body_match, re)) {
3488 5 : auto commit = message->body;
3489 10 : commit["id"] = message->id;
3490 10 : commit[CommitKey::TYPE] = message->type;
3491 5 : commits.emplace_back(commit);
3492 5 : }
3493 12 : } else {
3494 : // Matching type, just add it to the results
3495 0 : commits.emplace_back(message->body);
3496 : }
3497 :
3498 12 : if (filter.maxResult != 0 && commits.size() == filter.maxResult)
3499 0 : break;
3500 : }
3501 20 : }
3502 :
3503 4 : if (commits.size() > 0)
3504 9 : emitSignal<libjami::ConversationSignal::MessagesFound>(req,
3505 3 : sthis->pimpl_->accountId_,
3506 6 : sthis->id(),
3507 3 : std::move(commits));
3508 : // If we're the latest thread, inform client that the search is finished
3509 4 : if ((*flag)-- == 1 /* decrement return the old value */) {
3510 8 : emitSignal<libjami::ConversationSignal::MessagesFound>(
3511 8 : req, sthis->pimpl_->accountId_, std::string {}, std::vector<std::map<std::string, std::string>> {});
3512 : }
3513 8 : }
3514 4 : });
3515 4 : }
3516 :
3517 : void
3518 14 : Conversation::hostConference(CommitMessage&& message, OnDoneCb&& cb)
3519 : {
3520 14 : if (message.confId.empty()) {
3521 0 : JAMI_ERROR("{}Malformed commit: no confId", pimpl_->toString());
3522 0 : return;
3523 : }
3524 :
3525 14 : auto now = std::chrono::system_clock::now();
3526 14 : auto nowSecs = std::chrono::duration_cast<std::chrono::seconds>(now.time_since_epoch()).count();
3527 : {
3528 14 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3529 14 : pimpl_->hostedCalls_[message.confId] = nowSecs;
3530 14 : pimpl_->saveHostedCalls();
3531 14 : }
3532 :
3533 14 : createCommit(std::move(message), {}, std::move(cb));
3534 : }
3535 :
3536 : bool
3537 20 : Conversation::isHosting(const std::string& confId) const
3538 : {
3539 20 : auto info = infos();
3540 64 : if (info["rdvDevice"] == pimpl_->deviceId_ && info["rdvHost"] == pimpl_->userId_)
3541 0 : return true; // We are the current device Host
3542 20 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3543 20 : return pimpl_->hostedCalls_.find(confId) != pimpl_->hostedCalls_.end();
3544 20 : }
3545 :
3546 : void
3547 10 : Conversation::removeActiveConference(CommitMessage&& message, OnDoneCb&& cb)
3548 : {
3549 10 : if (message.confId.empty()) {
3550 0 : JAMI_ERROR("{}Malformed commit: no confId", pimpl_->toString());
3551 0 : return;
3552 : }
3553 :
3554 10 : auto erased = false;
3555 : {
3556 10 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3557 10 : erased = pimpl_->hostedCalls_.erase(message.confId);
3558 10 : }
3559 10 : if (erased) {
3560 10 : pimpl_->saveHostedCalls();
3561 10 : createCommit(std::move(message), {}, std::move(cb));
3562 : } else
3563 0 : cb(false, "");
3564 : }
3565 :
3566 : std::vector<std::map<std::string, std::string>>
3567 38 : Conversation::currentCalls() const
3568 : {
3569 38 : std::lock_guard lk(pimpl_->activeCallsMtx_);
3570 76 : return pimpl_->activeCalls_;
3571 38 : }
3572 : } // namespace jami
|
