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 : #include "collaborative_editing.h"
18 :
19 : #include "jamidht/jamiaccount.h"
20 : #include "jamidht/conversation_module.h"
21 : #include "jamidht/conversation.h"
22 : #include "jamidht/commit_message.h"
23 : #include "manager.h"
24 : #include "client/jami_signal.h"
25 : #include "base64.h"
26 : #include "string_utils.h"
27 :
28 : #include <dhtnet/multiplexed_socket.h>
29 : #include <msgpack.hpp>
30 : #include <opendht/thread_pool.h>
31 :
32 : #include <algorithm>
33 : #include <chrono>
34 : #include <future>
35 : #include <random>
36 :
37 : namespace jami {
38 :
39 : // Checkpoint policy. Writing a commit for every burst of keystrokes multiplies
40 : // the number of git objects for no benefit: on a 6000-keystroke session, the
41 : // commit cadence is by far the dominant factor of the on-disk footprint, ahead
42 : // of what is actually stored in each commit. Updates are therefore accumulated
43 : // and flushed either after a pause in typing or once enough of them have piled
44 : // up, which bounds both the footprint and the amount of work lost on a crash.
45 : static constexpr std::chrono::seconds CHECKPOINT_IDLE {10};
46 : static constexpr size_t CHECKPOINT_MAX_PENDING {200};
47 : /// Ceiling for a session that has no repository to drain into.
48 : static constexpr size_t PENDING_HARD_CAP {CHECKPOINT_MAX_PENDING * 50};
49 :
50 : // What a document holds when its creator did not say: the simplest thing an
51 : // editor can be pointed at.
52 : static constexpr const char DEFAULT_DOC_MIME_TYPE[] = "text/plain";
53 :
54 : // Ceilings on what the repository is asked to hold per item. A name is a label,
55 : // not a place to keep content; an attachment is bounded because it is stored
56 : // and replicated whole.
57 : static constexpr size_t MAX_DOCUMENT_NAME_SIZE {256};
58 : static constexpr size_t MAX_ATTACHMENT_SIZE {16 * 1024 * 1024};
59 :
60 : // Key added to a document listing, alongside those read from the announcing
61 : // commit: whether this device still holds the document. Part of the client API,
62 : // documented on getCollaborativeDocuments().
63 : static constexpr char DOCUMENT_STORED_LOCALLY[] = "storedLocally";
64 :
65 : // Ceilings on what a single message may carry. Both the client API and the
66 : // channels hand us opaque blobs, and both are held in memory before the engine
67 : // gets a say. The point is not to guess a "correct" size but to keep one message
68 : // from being able to allocate without bound: a CRDT update stays well under a
69 : // megabyte even when it carries the whole state of a large document, and an
70 : // awareness state is a cursor plus a display name.
71 : static constexpr size_t MAX_UPDATE_SIZE {8 * 1024 * 1024};
72 : static constexpr size_t MAX_AWARENESS_SIZE {8 * 1024};
73 : /// A whole awareness message may carry one entry per peer, not just ours.
74 : static constexpr size_t MAX_AWARENESS_MESSAGE_SIZE {64 * 1024};
75 :
76 : // Awareness upkeep, with the cadence the yjs protocol settled on. A state is
77 : // re-announced well before it is due to expire, so that losing one announcement
78 : // does not make an editor blink out of the document; a peer that announces
79 : // nothing for a whole timeout is one whose device is gone, not one who stopped
80 : // typing, and its cursor is withdrawn.
81 : static constexpr std::chrono::seconds AWARENESS_TIMEOUT {30};
82 : static constexpr std::chrono::seconds AWARENESS_RENEW {AWARENESS_TIMEOUT / 2};
83 : static constexpr std::chrono::seconds AWARENESS_SWEEP {AWARENESS_TIMEOUT / 10};
84 : /// How many peers may hold a state in one document at once. An authorized member
85 : /// picks its own client ids, so nothing but this stops one from filling the
86 : /// table with ids nobody is behind.
87 : static constexpr size_t MAX_AWARENESS_PEERS {256};
88 :
89 : namespace {
90 :
91 : // What a real-time frame carries: a Y-CRDT update or an awareness message. One
92 : // byte on the wire, so a frame type from a newer daemon is skipped over rather
93 : // than choked on.
94 : constexpr uint8_t FRAME_UPDATE {0};
95 : constexpr uint8_t FRAME_AWARENESS {1};
96 :
97 : // A frame is: varuint payload length, one tag byte, the payload. The length is
98 : // variable so the framing costs one byte on the frames that matter -- a
99 : // keystroke's update is a few dozen bytes -- while still naming sizes up to the
100 : // caps above.
101 : void
102 3 : appendVarUint(std::vector<uint8_t>& out, uint64_t value)
103 : {
104 3 : while (value >= 0x80) {
105 0 : out.push_back(static_cast<uint8_t>(0x80 | (value & 0x7F)));
106 0 : value >>= 7;
107 : }
108 3 : out.push_back(static_cast<uint8_t>(value));
109 3 : }
110 :
111 : enum class VarUint { OK, INCOMPLETE, MALFORMED };
112 :
113 : // Reads seven bits per byte, least significant group first, the high bit
114 : // marking that another byte follows. What is being read comes from a peer, so
115 : // an integer that never terminates has to end the parse, not the process.
116 : VarUint
117 3 : readVarUint(const uint8_t* data, size_t size, uint64_t& value, size_t& consumed)
118 : {
119 3 : value = 0;
120 3 : for (size_t i = 0; i < size; ++i) {
121 3 : if (i >= 10)
122 0 : return VarUint::MALFORMED; // longer than any uint64 ever encodes to
123 3 : value |= static_cast<uint64_t>(data[i] & 0x7F) << (7 * i);
124 3 : if ((data[i] & 0x80) == 0) {
125 3 : consumed = i + 1;
126 3 : return VarUint::OK;
127 : }
128 : }
129 0 : return size >= 10 ? VarUint::MALFORMED : VarUint::INCOMPLETE;
130 : }
131 :
132 : /// One client's slice of an awareness message, as it travels the wire.
133 : struct AwarenessWire
134 : {
135 : uint64_t clientId {0};
136 : /// Bumped by its owner on every change. An entry whose clock is not greater
137 : /// than the one already held is ignored, which is what keeps a message that
138 : /// took a longer route from resurrecting a state everyone has moved past.
139 : uint64_t clock {0};
140 : /// The client's state as a JSON document, or "null" for a client that is
141 : /// gone. Opaque here: its shape is the editors' agreement.
142 : std::string state;
143 4 : MSGPACK_DEFINE(clientId, clock, state)
144 : };
145 :
146 : std::vector<uint8_t>
147 2 : encodeAwareness(const std::vector<AwarenessWire>& entries)
148 : {
149 2 : msgpack::sbuffer buffer;
150 2 : msgpack::pack(buffer, entries);
151 2 : const auto* data = reinterpret_cast<const uint8_t*>(buffer.data());
152 6 : return {data, data + buffer.size()};
153 2 : }
154 :
155 : // Cap a document name, cutting on a code point boundary so the result is still
156 : // valid UTF-8 and can be put back into JSON.
157 : std::string
158 27 : truncatedName(std::string name)
159 : {
160 27 : if (name.size() <= MAX_DOCUMENT_NAME_SIZE)
161 27 : return name;
162 0 : size_t cut = MAX_DOCUMENT_NAME_SIZE;
163 0 : while (cut > 0 && (static_cast<unsigned char>(name[cut]) & 0xC0) == 0x80)
164 0 : --cut;
165 0 : name.resize(cut);
166 0 : return name;
167 : }
168 :
169 : } // namespace
170 :
171 : /// What one client id is currently sharing in a document.
172 : struct AwarenessPeer
173 : {
174 : uint64_t clock {0};
175 : /// JSON, or empty once the client withdrew its state.
176 : std::string state;
177 : std::chrono::steady_clock::time_point lastSeen;
178 : /// The account that announced this client id. Held so that a member cannot
179 : /// speak for a client id another one already claimed, and so that a state
180 : /// can still be attributed to a person when it reaches the clients.
181 : std::string owner;
182 : };
183 :
184 : struct CollaborativeEditing::Session
185 : {
186 : std::string conversationId;
187 : std::string documentId;
188 : std::unique_ptr<YrsDocument> doc;
189 : // The last name handed to the clients. A remote rename cannot be spotted by
190 : // reading the repository before and after a synchronization -- the merge is
191 : // already done by the time we hear about it -- so this is what tells a name
192 : // that changed from one the clients have seen. Empty until the document is
193 : // opened: announcing a rename to a client that was never told the first name
194 : // would be a phantom event. Guarded by mutex_.
195 : std::optional<std::string> announcedName;
196 : // Whether a client currently has the document open. A closed holder keeps
197 : // replicating -- that is what holding is -- but its client is not told about
198 : // updates it is not looking at; reopening hands the converged state over
199 : // instead. Guarded by mutex_.
200 : bool open {false};
201 : std::unique_ptr<asio::steady_timer> checkpointTimer;
202 : // Set once the repository's stored updates have been replayed into this session.
203 : bool persistedLoaded {false};
204 : // Attachment ids the local clients already know about, so a synchronization
205 : // only announces what it actually brought. Seeded on open with what the
206 : // repository already holds: a client reads those itself, and re-announcing
207 : // them would make every editor redraw its images on every sync. Guarded by
208 : // mutex_.
209 : std::set<std::string> knownAttachments;
210 :
211 : // Local updates produced since the last checkpoint, base64-encoded. Guarded by
212 : // its own mutex: it is filled from a YrsDocument callback, which already holds
213 : // the document's lock, so it must not reach for the manager-wide mutex.
214 : std::mutex pendingMutex;
215 : std::vector<std::string> pending;
216 :
217 : // The timer is rearmed both from the thread producing the edits and from the
218 : // io thread retrying a failed checkpoint; asio timers are not thread-safe.
219 : std::mutex timerMutex;
220 :
221 : // Set when the pending batch reached its cap, so that continued typing stops
222 : // pushing the debounce timer further away and the checkpoint actually runs.
223 : std::atomic_bool checkpointDue {false};
224 :
225 : // Real-time state. Kept under its own lock: the upkeep timer walks it from
226 : // the io thread while the clients write to it, and neither has any business
227 : // waiting on the manager-wide lock to do so.
228 : std::mutex protocolMutex;
229 : // The live channels to the peer devices that also have the document open,
230 : // per device. Two devices opening towards each other at once can end up
231 : // with a channel each, which is why this holds a list: dropping one of the
232 : // pair would have each side keep the one the other just closed. Frames are
233 : // sent on every one of them and a duplicate merges as a no-op -- that is
234 : // what a CRDT is for.
235 : std::map<std::string, std::vector<std::shared_ptr<dhtnet::ChannelSocket>>> channels;
236 : std::map<uint64_t, AwarenessPeer> awareness;
237 : /// This device's own entry in the table above. Its clock is what tells peers
238 : /// which of two states they hold is the later one, so it only ever grows.
239 : uint64_t localClock {0};
240 : std::string localState;
241 : std::chrono::steady_clock::time_point localAnnounced;
242 : std::unique_ptr<asio::steady_timer> awarenessTimer;
243 : bool upkeepRunning {false};
244 : };
245 :
246 453 : CollaborativeEditing::CollaborativeEditing(const std::shared_ptr<JamiAccount>& account)
247 453 : : account_(account)
248 453 : , accountId_(account->getAccountID())
249 906 : , ioContext_(Manager::instance().ioContext())
250 453 : {}
251 :
252 453 : CollaborativeEditing::~CollaborativeEditing() = default;
253 :
254 : uint64_t
255 46 : CollaborativeEditing::replicaId()
256 : {
257 : // Drawn at random, never derived from the device id. The client id is only
258 : // half of an item id: the other half is a per-replica counter that starts at
259 : // zero and is not persisted on its own. A replica that reuses an id after
260 : // producing items its peers already hold would restart that counter, and the
261 : // peers would silently drop the new items as ones they had already seen,
262 : // diverging for good.
263 : //
264 : // The daemon's replica never produces an item of its own -- it only applies
265 : // updates coming from the clients and from the peers -- so its id never
266 : // reaches any document. Drawing it costs nothing and keeps that property from
267 : // depending on the fact that nothing writes here today.
268 46 : if (clientId_ != 0)
269 3 : return clientId_;
270 43 : std::random_device rd;
271 : // 53 bits: the yjs ecosystem carries these ids through JSON, where integers
272 : // above 2^53 are no longer exact.
273 43 : std::uniform_int_distribution<uint64_t> dist(1, (uint64_t(1) << 53) - 1);
274 43 : std::mt19937_64 gen(rd());
275 43 : clientId_ = dist(gen);
276 43 : return clientId_;
277 43 : }
278 :
279 : std::string
280 429 : CollaborativeEditing::key(const std::string& conversationId, const std::string& documentId)
281 : {
282 429 : return conversationId + '/' + documentId;
283 : }
284 :
285 : std::shared_ptr<Conversation>
286 621 : CollaborativeEditing::documentConversation(const std::string& documentId)
287 : {
288 621 : auto account = account_.lock();
289 621 : if (!account)
290 0 : return nullptr;
291 621 : auto* cm = account->convModule(true);
292 621 : if (!cm)
293 0 : return nullptr;
294 621 : auto conversation = cm->getConversation(documentId);
295 621 : if (!conversation || conversation->mode() != ConversationMode::DOCUMENT)
296 179 : return nullptr;
297 442 : return conversation;
298 621 : }
299 :
300 : std::shared_ptr<CollaborativeEditing::Session>
301 84 : CollaborativeEditing::findSession(const std::string& conversationId, const std::string& documentId)
302 : {
303 84 : std::lock_guard<std::mutex> lk(mutex_);
304 84 : auto it = sessions_.find(key(conversationId, documentId));
305 168 : return it != sessions_.end() ? it->second : nullptr;
306 84 : }
307 :
308 : std::shared_ptr<CollaborativeEditing::Session>
309 26 : CollaborativeEditing::findSessionByDocument(const std::string& documentId)
310 : {
311 26 : std::lock_guard<std::mutex> lk(mutex_);
312 26 : for (const auto& [_, session] : sessions_)
313 25 : if (session && session->documentId == documentId)
314 25 : return session;
315 1 : return nullptr;
316 26 : }
317 :
318 : std::shared_ptr<CollaborativeEditing::Session>
319 49 : CollaborativeEditing::ensureSession(const std::string& conversationId, const std::string& documentId)
320 : {
321 49 : std::lock_guard<std::mutex> lk(mutex_);
322 49 : auto k = key(conversationId, documentId);
323 49 : if (auto it = sessions_.find(k); it != sessions_.end())
324 6 : return it->second;
325 :
326 43 : auto session = std::make_shared<Session>();
327 43 : session->conversationId = conversationId;
328 43 : session->documentId = documentId;
329 43 : session->doc = std::make_unique<YrsDocument>(replicaId());
330 43 : session->checkpointTimer = std::make_unique<asio::steady_timer>(*ioContext_);
331 43 : session->awarenessTimer = std::make_unique<asio::steady_timer>(*ioContext_);
332 43 : sessions_.emplace(k, session);
333 43 : return session;
334 49 : }
335 :
336 : std::string
337 25 : CollaborativeEditing::createDocument(const std::string& conversationId,
338 : const std::string& name,
339 : const std::string& mimeType)
340 : {
341 : // Settle the media type here rather than in each of the two places that
342 : // record it: an empty one in the initial commit and none at all in the
343 : // announcement would list the document as having no type while its
344 : // repository claimed one.
345 25 : const std::string type = mimeType.empty() ? DEFAULT_DOC_MIME_TYPE : mimeType;
346 :
347 25 : auto account = account_.lock();
348 25 : if (!account)
349 0 : return {};
350 25 : auto* cm = account->convModule();
351 25 : if (!cm)
352 0 : return {};
353 : // The document gets a swarm repository of its own, holding its content and
354 : // history; its id is the repository's. The creator is its first member.
355 25 : auto documentId = cm->startDocument(conversationId, type);
356 25 : if (documentId.empty()) {
357 0 : JAMI_ERROR("[Account {}] Unable to create a document repository in conversation {}", accountId_, conversationId);
358 0 : return {};
359 : }
360 :
361 : // The name describes the document, it is not part of its content: it lives
362 : // in the repository's profile, like a conversation's title, and reaches the
363 : // other holders through the ordinary repository synchronization.
364 25 : const auto stored = truncatedName(name);
365 25 : if (!stored.empty())
366 75 : cm->updateConversationInfos(documentId, {{"title", stored}}, false);
367 :
368 25 : auto session = ensureSession(conversationId, documentId);
369 : {
370 25 : std::lock_guard<std::mutex> lk(mutex_);
371 25 : session->persistedLoaded = true; // the repository is newborn: nothing to replay
372 25 : session->announcedName = stored;
373 25 : nameCache_[key(conversationId, documentId)] = stored;
374 25 : ++nameEpoch_;
375 25 : }
376 : {
377 : // Record it as announced right away. The commit below is asynchronous,
378 : // and a client that creates a document then opens it must not lose that
379 : // race against its own announcement.
380 25 : std::lock_guard<std::mutex> lk(announcedMtx_);
381 25 : announced_[conversationId].emplace(documentId);
382 25 : }
383 : // Announce the document in the conversation so that members discover it and
384 : // can replicate its repository. The announcement carries no content.
385 : //
386 : // Without it the document exists here and nowhere else: the other members
387 : // never learn about it, and the authorization that lets them replicate it is
388 : // derived from that very announcement. Report the failure rather than hand
389 : // back the id of a document nobody else can reach.
390 25 : auto announceResult = std::make_shared<std::promise<bool>>();
391 25 : auto announcedDone = announceResult->get_future();
392 75 : cm->createCommit(conversationId,
393 50 : CommitMessage::collabDocCreated(documentId, stored, type),
394 : true,
395 : {},
396 75 : [w = weak_from_this(),
397 : conversationId,
398 : documentId,
399 25 : accountId = accountId_,
400 : announceResult](bool ok, const std::string&) {
401 24 : announceResult->set_value(ok);
402 24 : if (ok)
403 24 : return;
404 0 : JAMI_ERROR("[Account {}] Document {} was not announced in conversation {}: "
405 : "the other members cannot reach it",
406 : accountId,
407 : documentId,
408 : conversationId);
409 0 : if (auto sthis = w.lock()) {
410 : {
411 0 : std::lock_guard<std::mutex> lk(sthis->announcedMtx_);
412 0 : if (auto it = sthis->announced_.find(conversationId); it != sthis->announced_.end())
413 0 : it->second.erase(documentId);
414 0 : }
415 0 : sthis->closeDocument(conversationId, documentId);
416 0 : sthis->dropLocalReplica(conversationId, documentId);
417 : // Undo startDocument() too: a repository nobody was
418 : // ever told about would still be reloaded as held on
419 : // every restart.
420 0 : if (auto account = sthis->account_.lock())
421 0 : if (auto* cm = account->convModule())
422 0 : cm->removeDocumentReplica(documentId);
423 0 : }
424 : });
425 : // The commit is written on another thread; wait for it, so that a document
426 : // this call hands back is really there: listed in the conversation, and
427 : // announced to the members. Without this, creating a document then listing
428 : // the conversation's documents would be a race against our own commit.
429 25 : if (announcedDone.wait_for(std::chrono::seconds(30)) != std::future_status::ready) {
430 : // Never resolving would mean the conversation vanished under us and the
431 : // commit callback was dropped; the rollback above cannot run either, so
432 : // all that is left is to say the announcement never happened.
433 1 : JAMI_ERROR("[Account {}] Document {} announcement never completed in conversation {}",
434 : accountId_,
435 : documentId,
436 : conversationId);
437 1 : return {};
438 : }
439 24 : if (!announcedDone.get())
440 0 : return {};
441 24 : return documentId;
442 50 : }
443 :
444 : bool
445 1 : CollaborativeEditing::removeDocument(const std::string& conversationId, const std::string& documentId)
446 : {
447 1 : auto account = account_.lock();
448 1 : if (!account)
449 0 : return false;
450 1 : auto* cm = account->convModule();
451 1 : if (!cm)
452 0 : return false;
453 : // Removing a document means retiring the commit that announced it, so that
454 : // commit has to be found first. Its id is also the only thing the removal
455 : // carries: the swarm ties an edition to the author of what it edits, which is
456 : // what keeps a member from retiring somebody else's document.
457 1 : std::string announcementId;
458 1 : for (const auto& doc : documents(conversationId)) {
459 1 : auto idIt = doc.find("id");
460 1 : if (idIt == doc.end() || idIt->second != documentId)
461 0 : continue;
462 2 : if (auto annIt = doc.find("announcement"); annIt != doc.end())
463 1 : announcementId = annIt->second;
464 1 : break;
465 1 : }
466 1 : if (announcementId.empty()) {
467 0 : JAMI_WARNING("[Account {}] [Document {}] Not removing: no announcement found in conversation {}",
468 : accountId_,
469 : documentId,
470 : conversationId);
471 0 : return false;
472 : }
473 : // The peers apply the removal when the commit reaches them, and this device
474 : // does the same through addToHistory() once the commit lands locally: nothing
475 : // is erased here, so a commit that never happens leaves the document intact.
476 : // editMessage() is what checks that we authored the announcement, exactly as
477 : // it does for a shared file.
478 1 : cm->editMessage(conversationId, {}, announcementId);
479 1 : return true;
480 1 : }
481 :
482 : bool
483 4 : CollaborativeEditing::removeDocumentLocally(const std::string& conversationId, const std::string& documentId)
484 : {
485 4 : if (!isAnnouncedDocument(conversationId, documentId)) {
486 0 : JAMI_ERROR("[Account {}] [Document {}] Not removing from this device: it was not announced in "
487 : "conversation {}",
488 : accountId_,
489 : documentId,
490 : conversationId);
491 0 : return false;
492 : }
493 :
494 4 : if (auto account = account_.lock())
495 4 : if (auto* cm = account->convModule())
496 4 : cm->removeDocumentReplica(documentId);
497 4 : dropLocalReplica(conversationId, documentId);
498 4 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentRemoved>(accountId_, conversationId, documentId, false);
499 4 : return true;
500 : }
501 :
502 : void
503 2 : CollaborativeEditing::setName(const std::string& conversationId, const std::string& documentId, const std::string& name)
504 : {
505 2 : auto account = account_.lock();
506 2 : if (!account)
507 0 : return;
508 2 : auto* cm = account->convModule();
509 2 : if (!cm)
510 0 : return;
511 2 : if (!documentConversation(documentId)) {
512 0 : JAMI_WARNING("[Account {}] [Document {}] Unable to rename: this device does not hold the document",
513 : accountId_,
514 : documentId);
515 0 : return;
516 : }
517 : // The name describes the document, it is not part of its content: keeping it
518 : // in the repository's profile rather than inside the CRDT is what lets the
519 : // daemon stay blind to what the document holds. It reaches the other holders
520 : // through the ordinary repository synchronization, and the swarm's own
521 : // validation is what restricts who may write it.
522 : //
523 : // The repository caps what it stores, so remember and announce what it really
524 : // holds, or the cache would answer a name no other member will ever see.
525 2 : const auto stored = truncatedName(name);
526 6 : cm->updateConversationInfos(documentId, {{"title", stored}}, true);
527 : {
528 2 : std::lock_guard<std::mutex> lk(mutex_);
529 2 : if (auto it = sessions_.find(key(conversationId, documentId)); it != sessions_.end())
530 2 : it->second->announcedName = stored;
531 2 : nameCache_[key(conversationId, documentId)] = stored;
532 2 : ++nameEpoch_;
533 2 : }
534 2 : emitRename(conversationId, documentId, stored);
535 4 : }
536 :
537 : std::string
538 17 : CollaborativeEditing::documentName(const std::string& conversationId, const std::string& documentId)
539 : {
540 : // Reading a name must stay cheap. Clients ask for it constantly: once per
541 : // document to list a conversation's documents, and once per message delegate
542 : // built while scrolling a conversation. Reading it from the repository's
543 : // profile means git lookups on the caller's thread -- the client's UI thread
544 : // -- so the answer is cached, and the cache is refreshed wherever the name
545 : // can change: in setName and on synchronization.
546 17 : const auto k = key(conversationId, documentId);
547 17 : uint64_t epoch = 0;
548 : {
549 17 : std::lock_guard<std::mutex> lk(mutex_);
550 17 : if (auto it = nameCache_.find(k); it != nameCache_.end())
551 3 : return it->second;
552 14 : epoch = nameEpoch_;
553 17 : }
554 14 : auto conversation = documentConversation(documentId);
555 14 : if (!conversation)
556 14 : return {}; // not held here: nothing worth remembering
557 0 : std::string name;
558 0 : auto infos = conversation->infos();
559 0 : if (auto it = infos.find("title"); it != infos.end())
560 0 : name = it->second;
561 0 : std::lock_guard<std::mutex> lk(mutex_);
562 : // Only remember it if nothing invalidated the cache while we were reading.
563 : // A synchronization landing during the read above would otherwise be
564 : // overwritten by the name we read just before it, and stay wrong for good.
565 0 : if (epoch == nameEpoch_)
566 0 : nameCache_[k] = name;
567 0 : return name;
568 0 : }
569 :
570 : std::vector<std::map<std::string, std::string>>
571 98 : CollaborativeEditing::documents(const std::string& conversationId)
572 : {
573 98 : auto account = account_.lock();
574 98 : if (!account)
575 0 : return {};
576 98 : auto* cm = account->convModule();
577 98 : if (!cm)
578 0 : return {};
579 98 : auto conversation = cm->getConversation(conversationId);
580 98 : if (!conversation)
581 14 : return {};
582 84 : auto docs = conversation->collaborativeDocuments();
583 : // The conversation knows which documents exist; whether this device still
584 : // holds one is whether its repository is here. A client has to be able to
585 : // tell them apart: one opens on what is already here, the other has to be
586 : // fetched back first.
587 110 : for (auto& doc : docs) {
588 26 : auto it = doc.find("id");
589 78 : doc[DOCUMENT_STORED_LOCALLY] = (it != doc.end() && documentConversation(it->second)) ? TRUE_STR : FALSE_STR;
590 : }
591 84 : return docs;
592 98 : }
593 :
594 : bool
595 109 : CollaborativeEditing::isAnnouncedDocument(const std::string& conversationId, const std::string& documentId)
596 : {
597 : // A document only exists once a COLLAB_DOC commit announced it in the
598 : // conversation. Without this check, a client naming arbitrary ids in an
599 : // open call would have us create a repository on disk for each.
600 : {
601 109 : std::lock_guard<std::mutex> lk(announcedMtx_);
602 109 : if (auto rmIt = removed_.find(conversationId); rmIt != removed_.end() && rmIt->second.count(documentId) != 0)
603 0 : return false;
604 109 : if (auto it = announced_.find(conversationId); it != announced_.end())
605 109 : return it->second.count(documentId) != 0;
606 109 : }
607 : // First question asked about this conversation: walking its whole history is
608 : // expensive, so do it once and let onDocumentAnnounced() keep the set fresh.
609 0 : std::set<std::string> ids;
610 0 : for (const auto& doc : documents(conversationId)) {
611 0 : if (auto it = doc.find("id"); it != doc.end())
612 0 : ids.emplace(it->second);
613 0 : }
614 0 : std::lock_guard<std::mutex> lk(announcedMtx_);
615 0 : auto& set = announced_[conversationId];
616 0 : set.merge(ids);
617 : // documents() already drops the retired ones, so a removal met earlier cannot
618 : // be undone by this merge; but a removal recorded meanwhile still wins.
619 0 : if (auto rmIt = removed_.find(conversationId); rmIt != removed_.end() && rmIt->second.count(documentId) != 0)
620 0 : return false;
621 0 : return set.count(documentId) != 0;
622 0 : }
623 :
624 : bool
625 159 : CollaborativeEditing::knowsDocument(const std::string& documentId)
626 : {
627 159 : std::lock_guard<std::mutex> lk(announcedMtx_);
628 159 : for (const auto& [_, ids] : announced_)
629 0 : if (ids.count(documentId) != 0)
630 0 : return true;
631 159 : return false;
632 159 : }
633 :
634 : bool
635 0 : CollaborativeEditing::isRemovedDocument(const std::string& conversationId, const std::string& documentId)
636 : {
637 0 : std::lock_guard<std::mutex> lk(announcedMtx_);
638 0 : auto it = removed_.find(conversationId);
639 0 : return it != removed_.end() && it->second.count(documentId) != 0;
640 0 : }
641 :
642 : void
643 8 : CollaborativeEditing::dropLocalReplica(const std::string& conversationId, const std::string& documentId)
644 : {
645 8 : std::shared_ptr<Session> session;
646 : {
647 8 : std::lock_guard<std::mutex> lk(mutex_);
648 8 : auto k = key(conversationId, documentId);
649 8 : if (auto it = sessions_.find(k); it != sessions_.end()) {
650 5 : session = it->second;
651 5 : sessions_.erase(it);
652 : }
653 8 : nameCache_.erase(k);
654 8 : ++nameEpoch_;
655 8 : }
656 : // Drop the live replica without checkpointing it first: the repository is
657 : // going with it, and writing to it would only race with the erase. The
658 : // timer is cancelled for the same reason -- it would fire on a repository
659 : // that no longer exists.
660 8 : if (session && session->checkpointTimer) {
661 5 : std::lock_guard<std::mutex> lk(session->timerMutex);
662 5 : session->checkpointTimer->cancel();
663 5 : }
664 8 : if (session)
665 5 : closeRealtimeChannels(session);
666 8 : }
667 :
668 : YrsDocument::Bytes
669 24 : CollaborativeEditing::openDocument(const std::string& conversationId, const std::string& documentId)
670 : {
671 24 : if (!ensureDocumentReplica(conversationId, documentId, true))
672 0 : return {};
673 24 : auto session = ensureSession(conversationId, documentId);
674 24 : auto conversation = documentConversation(documentId);
675 42 : const auto name = conversation ? conversation->infos()["title"] : documentName(conversationId, documentId);
676 : {
677 24 : std::lock_guard<std::mutex> lk(mutex_);
678 24 : session->open = true;
679 24 : if (!session->announcedName)
680 18 : session->announcedName = name;
681 24 : }
682 24 : loadPersistedState(session);
683 24 : auto state = session->doc->encodeStateAsUpdate();
684 : // Reach out to the other devices editing the document. What they produced
685 : // while nothing was open here is not asked for -- there is no handshake --
686 : // and does not need to be: it arrives with its producer's next checkpoint,
687 : // while everything from here on arrives live.
688 24 : connectRealtimeChannels(session);
689 24 : return state;
690 24 : }
691 :
692 : std::vector<std::string>
693 46 : CollaborativeEditing::cloneSources(const std::shared_ptr<Conversation>& parent, const std::string& documentId)
694 : {
695 46 : auto docs = parent->collaborativeDocuments();
696 184 : auto doc = std::find_if(docs.begin(), docs.end(), [&](const auto& item) { return item.at("id") == documentId; });
697 46 : if (doc == docs.end())
698 0 : return {};
699 92 : const auto& author = doc->at("author");
700 46 : std::vector<std::string> candidates;
701 46 : if (parent->isMember(author))
702 46 : candidates.push_back(author);
703 92 : auto members = parent->memberUris(author, {MemberRole::INVITED, MemberRole::BANNED, MemberRole::LEFT});
704 46 : candidates.insert(candidates.end(), members.begin(), members.end());
705 46 : std::set<std::string> online;
706 92 : for (const auto& device : parent->peersToSyncWith())
707 92 : online.emplace(parent->uriFromDevice(device.toString()));
708 138 : std::stable_partition(candidates.begin(), candidates.end(), [&](const auto& uri) { return online.count(uri) != 0; });
709 46 : return candidates;
710 46 : }
711 :
712 : bool
713 105 : CollaborativeEditing::ensureDocumentReplica(const std::string& conversationId,
714 : const std::string& documentId,
715 : bool reopen)
716 : {
717 105 : if (!isAnnouncedDocument(conversationId, documentId)) {
718 0 : JAMI_WARNING("[Account {}] Refusing unannounced document {}", accountId_, documentId);
719 0 : return false;
720 : }
721 105 : auto account = account_.lock();
722 105 : auto* cm = account ? account->convModule() : nullptr;
723 105 : auto parent = cm ? cm->getConversation(conversationId) : nullptr;
724 105 : if (!parent || parent->isRemoving() || !parent->isMember(account->getUsername())) {
725 1 : JAMI_DEBUG("[Account {}] Deferring document {}: parent is inactive", accountId_, documentId);
726 1 : return false;
727 : }
728 104 : if (documentConversation(documentId))
729 58 : return true;
730 46 : return cm->cloneDocumentFrom(conversationId, documentId, cloneSources(parent, documentId), reopen);
731 105 : }
732 :
733 : void
734 15 : CollaborativeEditing::closeDocument(const std::string& conversationId, const std::string& documentId)
735 : {
736 15 : auto session = findSession(conversationId, documentId);
737 15 : if (!session)
738 0 : return;
739 : {
740 15 : std::lock_guard<std::mutex> lk(mutex_);
741 15 : session->open = false;
742 15 : }
743 : // Flush pending edits, but keep the in-memory CRDT replica so that reopening
744 : // the document shows its current content. The session stays consistent via
745 : // persisted commits (replayed on load) and live updates from other members.
746 15 : if (session->checkpointTimer) {
747 15 : std::lock_guard<std::mutex> lk(session->timerMutex);
748 15 : session->checkpointTimer->cancel();
749 15 : }
750 15 : checkpointNow(session);
751 : // Withdraw this device's awareness state, which is what clears its cursor
752 : // for the other members. Done explicitly rather than left to expire so that
753 : // closing an editor is seen at once instead of a timeout later.
754 15 : publishAwareness(session, {});
755 : // The channels only exist between devices that are editing, and this one no
756 : // longer is. A reopen starts them over.
757 15 : closeRealtimeChannels(session);
758 15 : }
759 :
760 : void
761 13 : CollaborativeEditing::applyUpdate(const std::string& conversationId,
762 : const std::string& documentId,
763 : const YrsDocument::Bytes& update)
764 : {
765 13 : auto session = findSession(conversationId, documentId);
766 13 : if (!session)
767 0 : return;
768 13 : if (update.size() > MAX_UPDATE_SIZE) {
769 0 : JAMI_WARNING("[Account {}] [Document {}] Discarding a {} byte update from the client: "
770 : "over the {} byte limit",
771 : accountId_,
772 : documentId,
773 : update.size(),
774 : MAX_UPDATE_SIZE);
775 0 : return;
776 : }
777 : // Merge before forwarding: an update the engine rejects must not be sent to
778 : // the members nor written to the repository.
779 13 : if (!session->doc->applyUpdate(update))
780 0 : return;
781 13 : onLocalUpdate(session, update);
782 13 : }
783 :
784 : YrsDocument::Bytes
785 7 : CollaborativeEditing::documentState(const std::string& conversationId, const std::string& documentId)
786 : {
787 7 : auto session = findSession(conversationId, documentId);
788 14 : return session ? session->doc->encodeStateAsUpdate() : YrsDocument::Bytes {};
789 7 : }
790 :
791 : namespace {
792 :
793 : /// Frame a payload and write it to one channel.
794 : void
795 3 : writeFrame(const std::shared_ptr<dhtnet::ChannelSocket>& socket, uint8_t tag, const std::vector<uint8_t>& payload)
796 : {
797 3 : std::vector<uint8_t> frame;
798 3 : frame.reserve(payload.size() + 11);
799 3 : appendVarUint(frame, payload.size());
800 3 : frame.push_back(tag);
801 3 : frame.insert(frame.end(), payload.begin(), payload.end());
802 3 : std::error_code ec;
803 3 : socket->write(frame.data(), frame.size(), ec);
804 3 : if (ec)
805 0 : JAMI_WARNING("Unable to send a {} byte collaborative frame: {}", frame.size(), ec.message());
806 3 : }
807 :
808 : } // namespace
809 :
810 : bool
811 20 : CollaborativeEditing::acceptsRealtimeChannel(const std::string& documentId,
812 : const std::string& peer,
813 : const std::string& deviceId)
814 : {
815 : // Only for a document being edited here. A closed holder replicates through
816 : // the swarm and has no use for live frames; a device that never held the
817 : // document has nothing to accept them into; and answering for ids nobody
818 : // opened would let a peer probe what this device knows.
819 20 : auto session = findSessionByDocument(documentId);
820 20 : if (!session)
821 1 : return false;
822 : {
823 19 : std::lock_guard<std::mutex> lk(mutex_);
824 19 : if (!session->open)
825 16 : return false;
826 19 : }
827 : // Membership per the document's own replica, never the parent
828 : // conversation's. Invited counts as in: a document member is only ever
829 : // invited by a holder having just vouched for it while serving its clone,
830 : // and its join follows by itself -- there is no pending-invitation state to
831 : // wrongly admit. A joiner whose commits have not reached us at all is still
832 : // refused, and reached out to the moment they are merged.
833 3 : auto conversation = documentConversation(documentId);
834 3 : if (!conversation)
835 0 : return false; // the clone is still in flight: nothing to check against
836 3 : return conversation->isPeerAuthorized(peer, deviceId, true);
837 20 : }
838 :
839 : void
840 6 : CollaborativeEditing::onRealtimeChannel(const std::string& documentId,
841 : const std::string& peer,
842 : const std::string& deviceId,
843 : std::shared_ptr<dhtnet::ChannelSocket> socket)
844 : {
845 6 : auto session = findSessionByDocument(documentId);
846 6 : bool open = false;
847 6 : if (session) {
848 6 : std::lock_guard<std::mutex> lk(mutex_);
849 6 : open = session->open;
850 6 : }
851 6 : if (!open) {
852 : // The document was closed while the channel was being set up.
853 0 : socket->shutdown();
854 0 : return;
855 : }
856 : {
857 6 : std::lock_guard<std::mutex> lk(session->protocolMutex);
858 6 : session->channels[deviceId].push_back(socket);
859 6 : }
860 :
861 : // The parsing state of this one channel. Only its own receive callback
862 : // touches it, and those are delivered in order, so it needs no lock.
863 6 : auto buffer = std::make_shared<std::vector<uint8_t>>();
864 6 : std::weak_ptr<CollaborativeEditing> wthis = weak_from_this();
865 6 : std::weak_ptr<Session> wsession = session;
866 6 : std::weak_ptr<dhtnet::ChannelSocket> wsocket = socket;
867 6 : socket->setOnRecv([wthis, wsession, wsocket, peer, buffer](const uint8_t* data, size_t size) {
868 3 : auto sthis = wthis.lock();
869 3 : auto session = wsession.lock();
870 3 : if (!sthis || !session)
871 0 : return size;
872 3 : buffer->insert(buffer->end(), data, data + size);
873 3 : size_t pos = 0;
874 6 : while (pos < buffer->size()) {
875 3 : uint64_t length = 0;
876 3 : size_t consumed = 0;
877 3 : const auto res = readVarUint(buffer->data() + pos, buffer->size() - pos, length, consumed);
878 3 : if (res == VarUint::INCOMPLETE)
879 0 : break;
880 : // The length is judged before anything is held against it: a peer
881 : // must not have us buffer megabytes of a frame that could only be
882 : // refused once complete. Nothing recovers a framing violation --
883 : // there is no way back into sync with a stream that lies about its
884 : // lengths -- so the channel goes down with it.
885 3 : if (res == VarUint::MALFORMED || length > MAX_UPDATE_SIZE) {
886 0 : buffer->clear();
887 0 : if (auto socket = wsocket.lock())
888 0 : socket->shutdown();
889 0 : return size;
890 : }
891 3 : if (buffer->size() - pos - consumed < 1 + length)
892 0 : break; // the rest of the frame is still in flight
893 3 : const uint8_t tag = (*buffer)[pos + consumed];
894 3 : sthis->onFrame(session, peer, tag, buffer->data() + pos + consumed + 1, length);
895 3 : pos += consumed + 1 + length;
896 : }
897 3 : buffer->erase(buffer->begin(), buffer->begin() + pos);
898 3 : return size;
899 3 : });
900 6 : socket->onShutdown([wsession, wsocket, deviceId](const std::error_code& /*ec*/) {
901 6 : auto session = wsession.lock();
902 6 : if (!session)
903 0 : return;
904 6 : auto socket = wsocket.lock();
905 6 : std::lock_guard<std::mutex> lk(session->protocolMutex);
906 6 : auto it = session->channels.find(deviceId);
907 6 : if (it == session->channels.end())
908 1 : return;
909 5 : auto& list = it->second;
910 10 : list.erase(std::remove_if(list.begin(), list.end(), [&](const auto& s) { return !socket || s == socket; }),
911 5 : list.end());
912 5 : if (list.empty())
913 5 : session->channels.erase(it);
914 8 : });
915 :
916 : // Tell the newcomer at once who this device is in the document: our
917 : // awareness state only re-announces itself at the renewal cadence, and an
918 : // editor joining a session should not stare at an empty document for
919 : // fifteen seconds before the cursors appear.
920 6 : std::vector<uint8_t> hello;
921 : {
922 6 : std::lock_guard<std::mutex> lk(session->protocolMutex);
923 6 : if (!session->localState.empty())
924 0 : hello = encodeAwareness({{clientId_, session->localClock, session->localState}});
925 6 : }
926 6 : if (!hello.empty())
927 0 : writeFrame(socket, FRAME_AWARENESS, hello);
928 6 : }
929 :
930 : void
931 57 : CollaborativeEditing::connectRealtimeChannels(const std::shared_ptr<Session>& session)
932 : {
933 57 : auto account = account_.lock();
934 57 : if (!account)
935 0 : return;
936 : {
937 57 : std::lock_guard<std::mutex> lk(mutex_);
938 57 : if (!session->open)
939 14 : return;
940 57 : }
941 43 : auto conversation = documentConversation(session->documentId);
942 43 : if (!conversation)
943 15 : return; // the clone is still in flight: its completion comes back here
944 : // Every device the repository lists is asked, not just the ones the
945 : // document's swarm is connected to right now: the channel dials through the
946 : // connection manager, which reaches a device the DRT has not settled on
947 : // yet. Most are declined -- the other end only accepts while it has the
948 : // document open -- and the ones that matter are the editors.
949 28 : const auto ownDevice = std::string(account->currentDeviceId());
950 79 : for (const auto& [member, devices] : conversation->memberDevices()) {
951 102 : for (const auto& device : devices) {
952 51 : const auto deviceId = device.toString();
953 51 : if (deviceId == ownDevice)
954 27 : continue;
955 : {
956 24 : std::lock_guard<std::mutex> lk(session->protocolMutex);
957 24 : if (session->channels.count(deviceId) != 0)
958 2 : continue; // already talking to it
959 24 : }
960 : // The socket is handled where the accepting side's is, in
961 : // YdocChannelHandler::onReady; a refusal needs nothing done.
962 22 : account->connectYdocDevice(device, session->documentId);
963 51 : }
964 28 : }
965 72 : }
966 :
967 : void
968 20 : CollaborativeEditing::closeRealtimeChannels(const std::shared_ptr<Session>& session)
969 : {
970 20 : decltype(session->channels) channels;
971 : {
972 20 : std::lock_guard<std::mutex> lk(session->protocolMutex);
973 20 : channels = std::move(session->channels);
974 20 : session->channels.clear();
975 20 : }
976 : // Outside the lock: shutting a socket down runs its shutdown handler, which
977 : // takes the same lock to unregister -- and finds nothing left to.
978 21 : for (auto& [_, list] : channels)
979 2 : for (auto& socket : list)
980 1 : socket->shutdown();
981 20 : }
982 :
983 : void
984 15 : CollaborativeEditing::broadcastFrame(const std::shared_ptr<Session>& session,
985 : uint8_t tag,
986 : const std::vector<uint8_t>& payload)
987 : {
988 15 : if (payload.empty())
989 0 : return;
990 : // Written outside the lock: a send can stall on a congested peer, and the
991 : // receive path needs the lock to route what the others are saying.
992 15 : std::vector<std::shared_ptr<dhtnet::ChannelSocket>> sockets;
993 : {
994 15 : std::lock_guard<std::mutex> lk(session->protocolMutex);
995 18 : for (const auto& [_, list] : session->channels)
996 3 : sockets.insert(sockets.end(), list.begin(), list.end());
997 15 : }
998 18 : for (const auto& socket : sockets)
999 3 : writeFrame(socket, tag, payload);
1000 15 : }
1001 :
1002 : void
1003 3 : CollaborativeEditing::onFrame(
1004 : const std::shared_ptr<Session>& session, const std::string& from, uint8_t tag, const uint8_t* payload, size_t size)
1005 : {
1006 3 : switch (tag) {
1007 1 : case FRAME_UPDATE:
1008 1 : onRemoteUpdate(session, YrsDocument::Bytes(payload, payload + size));
1009 1 : break;
1010 2 : case FRAME_AWARENESS:
1011 2 : onAwarenessPayload(session, from, payload, size);
1012 2 : break;
1013 0 : default:
1014 : // A frame type from a newer daemon. Skipping it -- its length is known
1015 : // -- is what lets one be added without every peer upgrading first.
1016 0 : break;
1017 : }
1018 3 : }
1019 :
1020 : void
1021 1 : CollaborativeEditing::onRemoteUpdate(const std::shared_ptr<Session>& session, const YrsDocument::Bytes& update)
1022 : {
1023 1 : if (update.empty())
1024 0 : return;
1025 : // Cleared before the update rather than read after it alone, so that what
1026 : // this update brought is not confused with what an earlier one did.
1027 1 : session->doc->takeChanged();
1028 1 : if (!session->doc->applyUpdate(update))
1029 0 : return; // malformed: don't hand it to the clients
1030 : // Nothing when the update taught the replica nothing: a frame that raced a
1031 : // checkpoint fetch carries what the replay already merged, and forwarding
1032 : // it would light an "unread" badge on a document nobody touched.
1033 1 : if (!session->doc->takeChanged())
1034 0 : return;
1035 : // Channels only exist while the document is open here, but a frame can slip
1036 : // in between the close and the sockets going down; reopening hands the
1037 : // merged state over instead.
1038 : {
1039 1 : std::lock_guard<std::mutex> lk(mutex_);
1040 1 : if (!session->open)
1041 0 : return;
1042 1 : }
1043 : // Not persisted here: the device that produced it checkpoints it into its own
1044 : // repository and it reaches ours through synchronization. Storing it again
1045 : // would keep one copy per member of every single edit.
1046 1 : emitUpdate(session->conversationId, session->documentId, update);
1047 : }
1048 :
1049 : void
1050 2 : CollaborativeEditing::onAwarenessPayload(const std::shared_ptr<Session>& session,
1051 : const std::string& from,
1052 : const uint8_t* data,
1053 : size_t size)
1054 : {
1055 2 : if (size > MAX_AWARENESS_MESSAGE_SIZE)
1056 0 : return;
1057 2 : std::vector<AwarenessWire> entries;
1058 : try {
1059 2 : msgpack::object_handle oh = msgpack::unpack(reinterpret_cast<const char*>(data), size);
1060 2 : oh.get().convert(entries);
1061 2 : } catch (const std::exception&) {
1062 0 : return; // a peer sent something that is not an awareness message
1063 0 : }
1064 :
1065 : // What the local clients have to be told, gathered while the table is locked
1066 : // and emitted once it is not: a signal handler is entitled to call back into
1067 : // this manager.
1068 2 : std::vector<std::pair<uint64_t, std::string>> changed;
1069 2 : std::vector<uint64_t> left;
1070 2 : bool contested = false;
1071 2 : const auto now = std::chrono::steady_clock::now();
1072 : {
1073 2 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1074 4 : for (const auto& entry : entries) {
1075 2 : if (entry.state.size() > MAX_AWARENESS_SIZE)
1076 0 : continue; // a presence state, not a payload
1077 2 : if (entry.clientId == clientId_) {
1078 : // Someone is speaking for this device. Nothing legitimate does
1079 : // that, and the protocol's own answer -- outrun it so that peers
1080 : // keep our real state -- is also the right one here.
1081 0 : contested = true;
1082 0 : continue;
1083 : }
1084 2 : auto it = session->awareness.find(entry.clientId);
1085 2 : if (it != session->awareness.end()) {
1086 1 : if (it->second.owner != from)
1087 0 : continue; // a member may not speak for another's client id
1088 : // Strictly greater: two states with the same clock are the same
1089 : // state having taken two routes, and re-emitting one of them
1090 : // would make a cursor jump back to where it already was.
1091 1 : if (entry.clock <= it->second.clock)
1092 0 : continue;
1093 : } else {
1094 1 : if (session->awareness.size() >= MAX_AWARENESS_PEERS)
1095 0 : continue;
1096 1 : if (entry.state.empty() || entry.state == "null")
1097 0 : continue; // a client that is gone and was never here
1098 : }
1099 2 : const bool gone = entry.state.empty() || entry.state == "null";
1100 2 : if (gone) {
1101 1 : session->awareness.erase(entry.clientId);
1102 1 : left.push_back(entry.clientId);
1103 : } else {
1104 1 : auto& peer = session->awareness[entry.clientId];
1105 1 : peer.clock = entry.clock;
1106 1 : peer.state = entry.state;
1107 1 : peer.lastSeen = now;
1108 1 : peer.owner = from;
1109 1 : changed.emplace_back(entry.clientId, entry.state);
1110 : }
1111 : }
1112 2 : }
1113 :
1114 3 : for (const auto& [clientId, state] : changed)
1115 2 : emitSignal<libjami::ConversationSignal::CollaborativeAwarenessChanged>(accountId_,
1116 1 : session->conversationId,
1117 1 : session->documentId,
1118 : from,
1119 : clientId,
1120 : state);
1121 3 : for (auto clientId : left)
1122 2 : emitSignal<libjami::ConversationSignal::CollaborativeParticipantLeft>(accountId_,
1123 1 : session->conversationId,
1124 1 : session->documentId,
1125 : from,
1126 : clientId);
1127 2 : if (contested) {
1128 0 : JAMI_WARNING("[Account {}] [Document {}] {} announced this device's client id; re-announcing",
1129 : accountId_,
1130 : session->documentId,
1131 : from);
1132 0 : std::string state;
1133 : {
1134 0 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1135 0 : state = session->localState;
1136 0 : }
1137 0 : publishAwareness(session, state);
1138 0 : }
1139 2 : scheduleAwarenessUpkeep(session);
1140 2 : }
1141 :
1142 : void
1143 1 : CollaborativeEditing::setAwareness(const std::string& conversationId,
1144 : const std::string& documentId,
1145 : const std::string& state)
1146 : {
1147 1 : if (state.size() > MAX_AWARENESS_SIZE) {
1148 0 : JAMI_WARNING("[Account {}] [Document {}] Refusing to broadcast an oversized awareness state",
1149 : accountId_,
1150 : documentId);
1151 0 : return;
1152 : }
1153 : // Only for a document this device has open: an awareness state is about
1154 : // where its editor is, and there is no editor before that.
1155 1 : if (auto session = findSession(conversationId, documentId))
1156 1 : publishAwareness(session, state);
1157 : }
1158 :
1159 : void
1160 16 : CollaborativeEditing::publishAwareness(const std::shared_ptr<Session>& session, const std::string& state)
1161 : {
1162 16 : AwarenessWire entry;
1163 : {
1164 16 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1165 : // Withdrawing a state that was never shared would tell the members about
1166 : // an editor they were never told about in the first place.
1167 16 : if (state.empty() && session->localState.empty())
1168 14 : return;
1169 2 : entry.clientId = clientId_;
1170 2 : entry.clock = ++session->localClock;
1171 : // "null" is how a client that is no longer there is spelled.
1172 3 : entry.state = state.empty() ? "null" : state;
1173 2 : session->localState = state;
1174 2 : session->localAnnounced = std::chrono::steady_clock::now();
1175 16 : }
1176 6 : broadcastFrame(session, FRAME_AWARENESS, encodeAwareness({entry}));
1177 2 : scheduleAwarenessUpkeep(session);
1178 18 : }
1179 :
1180 : void
1181 4 : CollaborativeEditing::scheduleAwarenessUpkeep(const std::shared_ptr<Session>& session)
1182 : {
1183 4 : if (!session->awarenessTimer)
1184 0 : return;
1185 : {
1186 4 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1187 : // One timer at a time, and none at all while nobody is editing: a
1188 : // document nobody has open must not keep waking the process up.
1189 4 : if (session->upkeepRunning)
1190 2 : return;
1191 2 : if (session->awareness.empty() && session->localState.empty())
1192 0 : return;
1193 2 : session->upkeepRunning = true;
1194 2 : session->awarenessTimer->expires_after(AWARENESS_SWEEP);
1195 4 : }
1196 2 : std::weak_ptr<CollaborativeEditing> wthis = weak_from_this();
1197 2 : std::weak_ptr<Session> wsession = session;
1198 2 : session->awarenessTimer->async_wait([wthis, wsession](const asio::error_code& ec) {
1199 2 : auto sthis = wthis.lock();
1200 2 : auto session = wsession.lock();
1201 2 : if (!sthis || !session)
1202 2 : return;
1203 : {
1204 0 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1205 0 : session->upkeepRunning = false;
1206 0 : }
1207 0 : if (!ec)
1208 0 : sthis->awarenessUpkeep(session);
1209 4 : });
1210 2 : }
1211 :
1212 : void
1213 0 : CollaborativeEditing::awarenessUpkeep(const std::shared_ptr<Session>& session)
1214 : {
1215 0 : const auto now = std::chrono::steady_clock::now();
1216 0 : std::vector<std::pair<uint64_t, std::string>> expired;
1217 0 : std::string renew;
1218 : {
1219 0 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1220 0 : for (auto it = session->awareness.begin(); it != session->awareness.end();) {
1221 0 : if (now - it->second.lastSeen >= AWARENESS_TIMEOUT) {
1222 0 : expired.emplace_back(it->first, it->second.owner);
1223 0 : it = session->awareness.erase(it);
1224 : } else {
1225 0 : ++it;
1226 : }
1227 : }
1228 : // Re-announced well before it would expire elsewhere, so that a single
1229 : // lost message does not make this device blink out of the document.
1230 0 : if (!session->localState.empty() && now - session->localAnnounced >= AWARENESS_RENEW)
1231 0 : renew = session->localState;
1232 0 : }
1233 0 : for (const auto& [clientId, owner] : expired)
1234 0 : emitSignal<libjami::ConversationSignal::CollaborativeParticipantLeft>(accountId_,
1235 0 : session->conversationId,
1236 0 : session->documentId,
1237 : owner,
1238 : clientId);
1239 0 : if (!renew.empty())
1240 0 : publishAwareness(session, renew); // rearms the timer on its way out
1241 : else
1242 0 : scheduleAwarenessUpkeep(session);
1243 0 : }
1244 :
1245 : void
1246 13 : CollaborativeEditing::onLocalUpdate(const std::shared_ptr<Session>& session, const YrsDocument::Bytes& update)
1247 : {
1248 : // Real-time path: hand the incremental update to the devices editing along.
1249 13 : broadcastFrame(session, FRAME_UPDATE, update);
1250 : // Durable path: accumulate the update for the next checkpoint.
1251 13 : queueUpdate(session, update);
1252 13 : }
1253 :
1254 : void
1255 37 : CollaborativeEditing::replayStoredUpdates(const std::shared_ptr<Session>& session)
1256 : {
1257 37 : auto conversation = documentConversation(session->documentId);
1258 37 : if (!conversation)
1259 0 : return;
1260 : // The updates come from peers' commits, so their content is not ours to
1261 : // trust: a malformed one must cost that one update, not the whole replay.
1262 51 : for (const auto& encoded : conversation->documentUpdates()) {
1263 : try {
1264 14 : session->doc->applyUpdate(base64::decode(encoded));
1265 0 : } catch (const std::exception& e) {
1266 0 : JAMI_WARNING("[Account {}] [Document {}] Skipping unreadable stored update: {}",
1267 : accountId_,
1268 : session->documentId,
1269 : e.what());
1270 0 : }
1271 37 : }
1272 37 : }
1273 :
1274 : void
1275 24 : CollaborativeEditing::loadPersistedState(const std::shared_ptr<Session>& session)
1276 : {
1277 24 : if (!documentConversation(session->documentId))
1278 20 : return; // The pending clone will replay its state when it completes.
1279 : {
1280 9 : std::lock_guard<std::mutex> lk(mutex_);
1281 9 : if (session->persistedLoaded)
1282 5 : return;
1283 9 : }
1284 : // Replay the updates stored in the document's repository. Nothing is
1285 : // signalled: the caller encodes the converged state and hands it to the
1286 : // client that asked to open the document.
1287 4 : replayStoredUpdates(session);
1288 : // What is already stored is not news: the client resolves the attachments
1289 : // of the state it is being handed. Only what arrives afterwards is signalled.
1290 : // The asymmetry is deliberate. For a document being opened the client pulls,
1291 : // asking for each reference it meets while rendering; announcing every stored
1292 : // attachment here would push bytes nobody has asked for yet, on a document the
1293 : // user may never scroll through. The signal exists for the opposite case: an
1294 : // attachment landing in an already open document, which the client has no
1295 : // reason to look for.
1296 4 : std::vector<std::string> ids;
1297 4 : if (auto conversation = documentConversation(session->documentId))
1298 4 : ids = conversation->documentAttachmentIds();
1299 : // Only now: a session flagged as loaded is never replayed again, so flagging
1300 : // it before the replay would freeze a partially rebuilt document.
1301 4 : std::lock_guard<std::mutex> lk(mutex_);
1302 4 : session->persistedLoaded = true;
1303 4 : for (auto& id : ids)
1304 0 : session->knownAttachments.insert(std::move(id));
1305 4 : }
1306 :
1307 : void
1308 13 : CollaborativeEditing::queueUpdate(const std::shared_ptr<Session>& session, const YrsDocument::Bytes& update)
1309 : {
1310 13 : const bool held = documentConversation(session->documentId) != nullptr;
1311 13 : bool capReached = false;
1312 : {
1313 13 : std::lock_guard<std::mutex> lk(session->pendingMutex);
1314 : // Without a repository nothing will ever drain this: keep the last batch
1315 : // so a repository appearing later still saves recent work, and drop the
1316 : // rest rather than growing without bound.
1317 13 : if (!held && session->pending.size() >= PENDING_HARD_CAP)
1318 0 : session->pending.erase(session->pending.begin(), session->pending.begin() + CHECKPOINT_MAX_PENDING);
1319 13 : session->pending.emplace_back(base64::encode(update));
1320 13 : capReached = session->pending.size() >= CHECKPOINT_MAX_PENDING;
1321 13 : }
1322 : // Checkpointing reads the document back, so it must never run inline on the
1323 : // path that just wrote to it: scheduleCheckpoint() only arms a timer.
1324 13 : if (capReached)
1325 0 : session->checkpointDue = true;
1326 13 : scheduleCheckpoint(session, capReached ? std::chrono::seconds(0) : CHECKPOINT_IDLE);
1327 13 : }
1328 :
1329 : void
1330 13 : CollaborativeEditing::scheduleCheckpoint(const std::shared_ptr<Session>& session, std::chrono::seconds delay)
1331 : {
1332 13 : if (!session->checkpointTimer)
1333 0 : return;
1334 : // Once the batch has reached its cap, stop letting new edits push the
1335 : // deadline further away: sustained typing would otherwise never checkpoint.
1336 13 : if (delay.count() > 0 && session->checkpointDue)
1337 0 : return;
1338 13 : std::weak_ptr<CollaborativeEditing> wthis = weak_from_this();
1339 13 : std::weak_ptr<Session> wsession = session;
1340 13 : std::lock_guard<std::mutex> lk(session->timerMutex);
1341 13 : session->checkpointTimer->expires_after(delay);
1342 13 : session->checkpointTimer->async_wait([wthis, wsession](const asio::error_code& ec) {
1343 13 : if (ec) // cancelled by a newer edit (debounce) or by shutdown
1344 8 : return;
1345 5 : auto sthis = wthis.lock();
1346 5 : auto session = wsession.lock();
1347 5 : if (sthis && session)
1348 5 : sthis->checkpointNow(session);
1349 5 : });
1350 13 : }
1351 :
1352 : void
1353 58 : CollaborativeEditing::checkpointNow(const std::shared_ptr<Session>& session)
1354 : {
1355 58 : session->checkpointDue = false;
1356 58 : auto account = account_.lock();
1357 58 : if (!account)
1358 0 : return;
1359 58 : auto* cm = account->convModule();
1360 58 : if (!cm)
1361 0 : return;
1362 : // Before draining anything: a drained batch with nowhere to go is lost.
1363 : // Keeping the batch queued means a replica appearing later -- the document
1364 : // being opened, which is what clones it -- still saves recent work.
1365 58 : if (!documentConversation(session->documentId))
1366 4 : return;
1367 54 : std::vector<std::string> batch;
1368 : {
1369 54 : std::lock_guard<std::mutex> lk(session->pendingMutex);
1370 54 : batch.swap(session->pending);
1371 54 : }
1372 54 : if (batch.empty())
1373 41 : return;
1374 :
1375 : // The checkpoint is a commit in the document's own swarm: committing it is
1376 : // also what announces it, so the other holders fetch it through the same
1377 : // pipeline that moves conversation messages. Nothing else needs sending.
1378 39 : cm->createCommit(session->documentId,
1379 26 : CommitMessage::checkpoint(batch),
1380 : true,
1381 : {},
1382 26 : [w = weak_from_this(), wsession = std::weak_ptr<Session>(session), batch](bool ok,
1383 : const std::string&) {
1384 13 : if (ok)
1385 13 : return;
1386 : // Keep the updates queued so the next checkpoint retries
1387 : // them rather than silently losing the edits they carry,
1388 : // and make sure a retry is actually scheduled even if the
1389 : // user has stopped typing.
1390 0 : auto sthis = w.lock();
1391 0 : auto session = wsession.lock();
1392 0 : if (!sthis || !session)
1393 0 : return;
1394 : {
1395 0 : std::lock_guard<std::mutex> lk(session->pendingMutex);
1396 0 : session->pending.insert(session->pending.begin(), batch.begin(), batch.end());
1397 0 : }
1398 0 : sthis->scheduleCheckpoint(session, CHECKPOINT_IDLE);
1399 0 : });
1400 99 : }
1401 :
1402 : std::vector<std::map<std::string, std::string>>
1403 105 : CollaborativeEditing::history(const std::string& /*conversationId*/, const std::string& documentId, size_t max)
1404 : {
1405 105 : auto conversation = documentConversation(documentId);
1406 210 : return conversation ? conversation->documentHistory(max) : std::vector<std::map<std::string, std::string>> {};
1407 105 : }
1408 :
1409 : YrsDocument::Bytes
1410 3 : CollaborativeEditing::documentStateAt(const std::string& /*conversationId*/,
1411 : const std::string& documentId,
1412 : const std::string& commitId)
1413 : {
1414 3 : auto conversation = documentConversation(documentId);
1415 3 : if (!conversation)
1416 0 : return {};
1417 :
1418 : // Nothing at all when the checkpoint is unknown, which is what the public
1419 : // contract promises. It has to be told apart from a checkpoint that exists
1420 : // and holds nothing: the two would otherwise be the same answer, and a
1421 : // client restoring an early, legitimately empty version could not tell
1422 : // whether it was allowed to.
1423 3 : const auto stored = conversation->documentUpdatesAt(commitId);
1424 3 : if (!stored)
1425 0 : return {};
1426 :
1427 : // Replay into a throwaway replica: the live document must not be touched.
1428 : // What the client does with that state -- show it, restore it, diff it -- is
1429 : // its own business, and depends on a document type the daemon ignores.
1430 3 : YrsDocument snapshot {replicaId()};
1431 7 : for (const auto& encoded : *stored) {
1432 : try {
1433 4 : snapshot.applyUpdate(base64::decode(encoded));
1434 0 : } catch (const std::exception& e) {
1435 0 : JAMI_WARNING("[Account {}] [Document {}] Skipping unreadable stored update: {}",
1436 : accountId_,
1437 : documentId,
1438 : e.what());
1439 0 : }
1440 : }
1441 3 : return snapshot.encodeStateAsUpdate();
1442 3 : }
1443 :
1444 : void
1445 81 : CollaborativeEditing::onDocumentAnnounced(const std::string& conversationId, const std::string& documentId)
1446 : {
1447 : {
1448 81 : std::lock_guard<std::mutex> lk(announcedMtx_);
1449 81 : if (auto it = removed_.find(conversationId); it != removed_.end() && it->second.count(documentId) != 0)
1450 0 : return;
1451 81 : announced_[conversationId].emplace(documentId);
1452 81 : if (!replicating_.emplace(key(conversationId, documentId)).second)
1453 0 : return;
1454 81 : }
1455 : // The caller can hold the parent conversation's locks. Fetch only after
1456 : // leaving that path, without pretending a client has opened the document.
1457 81 : dht::ThreadPool::io().run([w = weak_from_this(), conversationId, documentId] {
1458 81 : if (auto sthis = w.lock()) {
1459 81 : sthis->ensureDocumentReplica(conversationId, documentId, false);
1460 : {
1461 81 : std::lock_guard<std::mutex> lk(sthis->announcedMtx_);
1462 81 : sthis->replicating_.erase(key(conversationId, documentId));
1463 81 : }
1464 : // A clone in progress reports when it lands; otherwise report now.
1465 81 : auto account = sthis->account_.lock();
1466 81 : auto* cm = account ? account->convModule() : nullptr;
1467 81 : if (cm && !cm->isCloningDocument(documentId)) {
1468 78 : auto stored = sthis->documentConversation(documentId) != nullptr;
1469 78 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentReplicated>(sthis->accountId_,
1470 78 : conversationId,
1471 78 : documentId,
1472 : stored);
1473 : }
1474 162 : }
1475 81 : });
1476 : }
1477 :
1478 : void
1479 1003 : CollaborativeEditing::syncDocuments(const std::string& conversationId)
1480 : {
1481 1003 : auto account = account_.lock();
1482 1003 : if (!account)
1483 0 : return;
1484 1003 : auto* cm = account->convModule();
1485 1003 : auto parent = cm ? cm->getConversation(conversationId) : nullptr;
1486 2005 : if (!parent || parent->mode() == ConversationMode::DOCUMENT || parent->isRemoving()
1487 2005 : || !parent->isMember(account->getUsername()))
1488 10 : return;
1489 999 : for (const auto& document : parent->collaborativeDocuments())
1490 1005 : onDocumentAnnounced(conversationId, document.at("id"));
1491 1013 : }
1492 :
1493 : void
1494 4 : CollaborativeEditing::onDocumentRemoved(const std::string& conversationId, const std::string& documentId)
1495 : {
1496 : {
1497 4 : std::lock_guard<std::mutex> lk(announcedMtx_);
1498 4 : removed_[conversationId].emplace(documentId);
1499 4 : if (auto it = announced_.find(conversationId); it != announced_.end())
1500 4 : it->second.erase(documentId);
1501 4 : }
1502 4 : dropLocalReplica(conversationId, documentId);
1503 : // The repository goes too: a document nobody can open again would otherwise
1504 : // outlive its own removal on every device that held it. From another
1505 : // thread: this runs while addToHistory() holds the parent conversation's
1506 : // lock, and tearing a conversation down takes locks of its own.
1507 4 : dht::ThreadPool::io().run([w = account_, documentId] {
1508 4 : if (auto account = w.lock())
1509 4 : if (auto* cm = account->convModule())
1510 4 : cm->removeDocumentReplica(documentId);
1511 4 : });
1512 4 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentRemoved>(accountId_, conversationId, documentId, true);
1513 4 : }
1514 :
1515 : void
1516 47 : CollaborativeEditing::onRepositoryUpdated(const std::string& conversationId,
1517 : const std::string& documentId,
1518 : bool contentChanged)
1519 : {
1520 : // A remote rename can land on a document nobody has open here, and both
1521 : // early returns below are reachable in that case: drop the cached name first
1522 : // or a client would keep showing the old one until the account restarts.
1523 : {
1524 47 : std::lock_guard<std::mutex> lk(mutex_);
1525 47 : nameCache_.erase(key(conversationId, documentId));
1526 47 : ++nameEpoch_;
1527 47 : }
1528 47 : auto session = findSession(conversationId, documentId);
1529 47 : auto conversation = documentConversation(documentId);
1530 47 : if (!conversation)
1531 2 : return;
1532 45 : if (!session) {
1533 12 : if (contentChanged)
1534 8 : emitUpdate(conversationId, documentId, {});
1535 12 : return;
1536 : }
1537 : // What the replica knows before the replay, so that what it learns from it
1538 : // can be told apart from what it already had.
1539 33 : const auto before = session->doc->encodeStateVector();
1540 33 : session->doc->takeChanged();
1541 : // Applying an update the replica already knows is a no-op for a CRDT, so
1542 : // replaying the whole set is correct, just more work than strictly needed.
1543 33 : replayStoredUpdates(session);
1544 : {
1545 33 : std::lock_guard<std::mutex> lk(mutex_);
1546 33 : session->persistedLoaded = true;
1547 33 : }
1548 : // Nothing at all when the replay taught us nothing -- a rename-only commit,
1549 : // or updates the real-time path had already delivered -- or every client
1550 : // would light an "unread" badge for a document nobody touched. The question
1551 : // is put to yrs rather than answered by measuring the diff: yrs appends the
1552 : // whole deletion set to a diff without diffing it, so a diff carrying no new
1553 : // content still measures a few bytes on any document where a character was
1554 : // ever erased.
1555 : //
1556 : // What is sent is only what the replay brought, not the whole document: a
1557 : // synchronization usually carries a handful of keystrokes, and re-encoding
1558 : // a 300 kB document for each of them would push megabytes a minute through
1559 : // the client API for nothing. A closed holder's client is not handed the
1560 : // content either -- it gets the converged state when it reopens -- but it
1561 : // is told that there is some: an update with an empty payload, which is
1562 : // what lets it badge a document nobody here is watching.
1563 : bool tellClient;
1564 : {
1565 33 : std::lock_guard<std::mutex> lk(mutex_);
1566 33 : tellClient = session->open;
1567 33 : }
1568 33 : if (session->doc->takeChanged())
1569 3 : emitUpdate(conversationId, documentId, tellClient ? session->doc->encodeDiff(before) : YrsDocument::Bytes {});
1570 : // Independent of the updates above: an attachment is not part of the CRDT,
1571 : // so a synchronization can bring the payload of a reference the real-time
1572 : // path delivered long before, with no update at all to show for it.
1573 33 : emitNewAttachments(session);
1574 : // The name travels with the repository now, so a remote rename lands here.
1575 : // It cannot be detected by reading the name around the replay: the caller
1576 : // merges before calling us, so both reads would return the name from after
1577 : // the merge. What we compare against is the last name we told the clients.
1578 33 : std::string name;
1579 : {
1580 33 : auto infos = conversation->infos();
1581 66 : if (auto it = infos.find("title"); it != infos.end())
1582 33 : name = it->second;
1583 33 : }
1584 33 : auto renamed = false;
1585 : {
1586 33 : std::lock_guard<std::mutex> lk(mutex_);
1587 33 : nameCache_[key(conversationId, documentId)] = name;
1588 33 : ++nameEpoch_;
1589 : // Only a name the clients have already been told can be seen to change.
1590 33 : if (session->announcedName && *session->announcedName != name)
1591 14 : renamed = true;
1592 33 : if (session->announcedName)
1593 33 : session->announcedName = name;
1594 33 : }
1595 33 : if (renamed)
1596 14 : emitRename(conversationId, documentId, name);
1597 : // A synchronization is also how new editors become reachable: the clone
1598 : // this session may have been waiting for just landed, or a joiner's
1599 : // membership commits were just merged -- the very merge that entitles the
1600 : // device this replica refused a moment ago to its channel.
1601 33 : connectRealtimeChannels(session);
1602 61 : }
1603 :
1604 : void
1605 492 : CollaborativeEditing::flush()
1606 : {
1607 492 : std::vector<std::shared_ptr<Session>> sessions;
1608 : {
1609 492 : std::lock_guard<std::mutex> lk(mutex_);
1610 492 : sessions.reserve(sessions_.size());
1611 530 : for (const auto& [_, session] : sessions_)
1612 38 : sessions.emplace_back(session);
1613 492 : }
1614 530 : for (const auto& session : sessions) {
1615 38 : if (session->checkpointTimer) {
1616 38 : std::lock_guard<std::mutex> lk(session->timerMutex);
1617 38 : session->checkpointTimer->cancel();
1618 38 : }
1619 : // No compaction here: the edits are already durable, packing is pure
1620 : // housekeeping, and doing it inline would stall the account
1621 : // unregistration for seconds per open document.
1622 38 : checkpointNow(session);
1623 : }
1624 492 : }
1625 :
1626 : void
1627 12 : CollaborativeEditing::emitUpdate(const std::string& conversationId,
1628 : const std::string& documentId,
1629 : const YrsDocument::Bytes& update)
1630 : {
1631 : // One channel for every document type: the payload is an opaque Y-CRDT
1632 : // update, and the client merges it into its own replica. That is what makes
1633 : // a plain-text editor and a rich-text editor listen to the same signal.
1634 12 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentUpdate>(accountId_, conversationId, documentId, update);
1635 12 : }
1636 :
1637 : void
1638 16 : CollaborativeEditing::emitRename(const std::string& conversationId,
1639 : const std::string& documentId,
1640 : const std::string& name)
1641 : {
1642 16 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentRenamed>(accountId_, conversationId, documentId, name);
1643 16 : }
1644 :
1645 : void
1646 33 : CollaborativeEditing::emitNewAttachments(const std::shared_ptr<Session>& session)
1647 : {
1648 33 : auto conversation = documentConversation(session->documentId);
1649 33 : if (!conversation)
1650 0 : return;
1651 33 : auto ids = conversation->documentAttachmentIds();
1652 33 : std::vector<std::string> fresh;
1653 : {
1654 33 : std::lock_guard<std::mutex> lk(mutex_);
1655 35 : for (auto& id : ids)
1656 2 : if (session->knownAttachments.insert(id).second)
1657 1 : fresh.push_back(std::move(id));
1658 33 : }
1659 34 : for (const auto& id : fresh)
1660 2 : emitSignal<libjami::ConversationSignal::CollaborativeAttachmentAdded>(accountId_,
1661 1 : session->conversationId,
1662 1 : session->documentId,
1663 : id);
1664 33 : }
1665 :
1666 : std::string
1667 1 : CollaborativeEditing::addAttachment(const std::string& conversationId,
1668 : const std::string& documentId,
1669 : const std::vector<uint8_t>& data)
1670 : {
1671 1 : if (data.empty() || data.size() > MAX_ATTACHMENT_SIZE) {
1672 0 : JAMI_WARNING("[Account {}] [Document {}] Attachment refused: {} byte(s), limit is {}",
1673 : accountId_,
1674 : documentId,
1675 : data.size(),
1676 : MAX_ATTACHMENT_SIZE);
1677 0 : return {};
1678 : }
1679 1 : auto account = account_.lock();
1680 1 : if (!account)
1681 0 : return {};
1682 1 : auto* cm = account->convModule();
1683 1 : if (!cm)
1684 0 : return {};
1685 : // The payload is a commit in the document's own swarm, and committing it is
1686 : // also what announces it: peers fetch it straight away rather than showing a
1687 : // placeholder until the next checkpoint.
1688 1 : auto id = cm->addDocumentAttachment(documentId, data);
1689 1 : if (id.empty())
1690 0 : return {};
1691 1 : if (auto session = findSession(conversationId, documentId)) {
1692 : // Ours already: the client that stored it holds the bytes, and the next
1693 : // synchronization must not announce them back to it.
1694 1 : std::lock_guard<std::mutex> lk(mutex_);
1695 1 : session->knownAttachments.insert(id);
1696 2 : }
1697 1 : return id;
1698 1 : }
1699 :
1700 : std::vector<uint8_t>
1701 3 : CollaborativeEditing::attachment(const std::string& /*conversationId*/,
1702 : const std::string& documentId,
1703 : const std::string& attachmentId)
1704 : {
1705 : // Readable without an editing session: a client browsing the history of a
1706 : // document it has not opened still has to resolve what it refers to.
1707 3 : auto conversation = documentConversation(documentId);
1708 6 : return conversation ? conversation->documentAttachment(attachmentId) : std::vector<uint8_t> {};
1709 3 : }
1710 :
1711 : } // namespace jami
|
