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 18 : truncatedName(std::string name)
159 : {
160 18 : if (name.size() <= MAX_DOCUMENT_NAME_SIZE)
161 18 : 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 168 : CollaborativeEditing::CollaborativeEditing(const std::shared_ptr<JamiAccount>& account)
247 168 : : account_(account)
248 168 : , accountId_(account->getAccountID())
249 336 : , ioContext_(Manager::instance().ioContext())
250 168 : {}
251 :
252 168 : CollaborativeEditing::~CollaborativeEditing() = default;
253 :
254 : uint64_t
255 33 : 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 33 : if (clientId_ != 0)
269 2 : return clientId_;
270 31 : 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 31 : std::uniform_int_distribution<uint64_t> dist(1, (uint64_t(1) << 53) - 1);
274 31 : std::mt19937_64 gen(rd());
275 31 : clientId_ = dist(gen);
276 31 : return clientId_;
277 31 : }
278 :
279 : std::string
280 193 : CollaborativeEditing::key(const std::string& conversationId, const std::string& documentId)
281 : {
282 193 : return conversationId + '/' + documentId;
283 : }
284 :
285 : std::shared_ptr<Conversation>
286 314 : CollaborativeEditing::documentConversation(const std::string& documentId)
287 : {
288 314 : auto account = account_.lock();
289 314 : if (!account)
290 0 : return nullptr;
291 314 : auto* cm = account->convModule(true);
292 314 : if (!cm)
293 0 : return nullptr;
294 314 : auto conversation = cm->getConversation(documentId);
295 314 : if (!conversation || conversation->mode() != ConversationMode::DOCUMENT)
296 63 : return nullptr;
297 251 : return conversation;
298 314 : }
299 :
300 : std::shared_ptr<CollaborativeEditing::Session>
301 57 : CollaborativeEditing::findSession(const std::string& conversationId, const std::string& documentId)
302 : {
303 57 : std::lock_guard<std::mutex> lk(mutex_);
304 57 : auto it = sessions_.find(key(conversationId, documentId));
305 114 : return it != sessions_.end() ? it->second : nullptr;
306 57 : }
307 :
308 : std::shared_ptr<CollaborativeEditing::Session>
309 23 : CollaborativeEditing::findSessionByDocument(const std::string& documentId)
310 : {
311 23 : std::lock_guard<std::mutex> lk(mutex_);
312 23 : for (const auto& [_, session] : sessions_)
313 23 : if (session && session->documentId == documentId)
314 23 : return session;
315 0 : return nullptr;
316 23 : }
317 :
318 : std::shared_ptr<CollaborativeEditing::Session>
319 37 : CollaborativeEditing::ensureSession(const std::string& conversationId, const std::string& documentId)
320 : {
321 37 : std::lock_guard<std::mutex> lk(mutex_);
322 37 : auto k = key(conversationId, documentId);
323 37 : if (auto it = sessions_.find(k); it != sessions_.end())
324 4 : return it->second;
325 :
326 33 : auto session = std::make_shared<Session>();
327 33 : session->conversationId = conversationId;
328 33 : session->documentId = documentId;
329 33 : session->doc = std::make_unique<YrsDocument>(replicaId());
330 33 : session->checkpointTimer = std::make_unique<asio::steady_timer>(*ioContext_);
331 33 : session->awarenessTimer = std::make_unique<asio::steady_timer>(*ioContext_);
332 33 : sessions_.emplace(k, session);
333 33 : return session;
334 37 : }
335 :
336 : std::string
337 16 : 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 16 : const std::string type = mimeType.empty() ? DEFAULT_DOC_MIME_TYPE : mimeType;
346 :
347 16 : auto account = account_.lock();
348 16 : if (!account)
349 0 : return {};
350 16 : auto* cm = account->convModule();
351 16 : 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 16 : auto documentId = cm->startDocument(conversationId, type);
356 16 : 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 16 : const auto stored = truncatedName(name);
365 16 : if (!stored.empty())
366 48 : cm->updateConversationInfos(documentId, {{"title", stored}}, false);
367 :
368 16 : auto session = ensureSession(conversationId, documentId);
369 : {
370 16 : std::lock_guard<std::mutex> lk(mutex_);
371 16 : session->persistedLoaded = true; // the repository is newborn: nothing to replay
372 16 : session->announcedName = stored;
373 16 : nameCache_[key(conversationId, documentId)] = stored;
374 16 : ++nameEpoch_;
375 16 : }
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 16 : std::lock_guard<std::mutex> lk(announcedMtx_);
381 16 : announced_[conversationId].emplace(documentId);
382 16 : }
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 16 : auto announceResult = std::make_shared<std::promise<bool>>();
391 16 : auto announcedDone = announceResult->get_future();
392 48 : cm->createCommit(conversationId,
393 32 : CommitMessage::collabDocCreated(documentId, stored, type),
394 : true,
395 : {},
396 48 : [w = weak_from_this(),
397 : conversationId,
398 : documentId,
399 16 : accountId = accountId_,
400 : announceResult](bool ok, const std::string&) {
401 16 : announceResult->set_value(ok);
402 16 : if (ok)
403 16 : 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 16 : 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 0 : JAMI_ERROR("[Account {}] Document {} announcement never completed in conversation {}",
434 : accountId_,
435 : documentId,
436 : conversationId);
437 0 : return {};
438 : }
439 16 : if (!announcedDone.get())
440 0 : return {};
441 16 : return documentId;
442 32 : }
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 2 : CollaborativeEditing::removeDocumentLocally(const std::string& conversationId, const std::string& documentId)
484 : {
485 2 : 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 2 : if (auto account = account_.lock())
495 2 : if (auto* cm = account->convModule())
496 2 : cm->removeDocumentReplica(documentId);
497 2 : dropLocalReplica(conversationId, documentId);
498 2 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentRemoved>(accountId_, conversationId, documentId, false);
499 2 : 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 2 : 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 2 : const auto k = key(conversationId, documentId);
547 2 : uint64_t epoch = 0;
548 : {
549 2 : std::lock_guard<std::mutex> lk(mutex_);
550 2 : if (auto it = nameCache_.find(k); it != nameCache_.end())
551 2 : return it->second;
552 0 : epoch = nameEpoch_;
553 2 : }
554 0 : auto conversation = documentConversation(documentId);
555 0 : if (!conversation)
556 0 : 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 75 : CollaborativeEditing::documents(const std::string& conversationId)
572 : {
573 75 : auto account = account_.lock();
574 75 : if (!account)
575 0 : return {};
576 75 : auto* cm = account->convModule();
577 75 : if (!cm)
578 0 : return {};
579 75 : auto conversation = cm->getConversation(conversationId);
580 75 : if (!conversation)
581 17 : return {};
582 58 : 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 101 : for (auto& doc : docs) {
588 43 : auto it = doc.find("id");
589 129 : doc[DOCUMENT_STORED_LOCALLY] = (it != doc.end() && documentConversation(it->second)) ? TRUE_STR : FALSE_STR;
590 : }
591 58 : return docs;
592 75 : }
593 :
594 : bool
595 23 : 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 23 : std::lock_guard<std::mutex> lk(announcedMtx_);
602 23 : if (auto rmIt = removed_.find(conversationId); rmIt != removed_.end() && rmIt->second.count(documentId) != 0)
603 0 : return false;
604 23 : if (auto it = announced_.find(conversationId); it != announced_.end())
605 23 : return it->second.count(documentId) != 0;
606 23 : }
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 149 : CollaborativeEditing::knowsDocument(const std::string& documentId)
626 : {
627 149 : std::lock_guard<std::mutex> lk(announcedMtx_);
628 149 : for (const auto& [_, ids] : announced_)
629 0 : if (ids.count(documentId) != 0)
630 0 : return true;
631 149 : return false;
632 149 : }
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 5 : CollaborativeEditing::dropLocalReplica(const std::string& conversationId, const std::string& documentId)
644 : {
645 5 : std::shared_ptr<Session> session;
646 : {
647 5 : std::lock_guard<std::mutex> lk(mutex_);
648 5 : auto k = key(conversationId, documentId);
649 5 : if (auto it = sessions_.find(k); it != sessions_.end()) {
650 4 : session = it->second;
651 4 : sessions_.erase(it);
652 : }
653 5 : nameCache_.erase(k);
654 5 : ++nameEpoch_;
655 5 : }
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 5 : if (session && session->checkpointTimer) {
661 4 : std::lock_guard<std::mutex> lk(session->timerMutex);
662 4 : session->checkpointTimer->cancel();
663 4 : }
664 5 : if (session)
665 4 : closeRealtimeChannels(session);
666 5 : }
667 :
668 : YrsDocument::Bytes
669 21 : CollaborativeEditing::openDocument(const std::string& conversationId, const std::string& documentId)
670 : {
671 : // A document only exists once the conversation announced it. Opening one
672 : // that was never announced would clone from any id a caller cares to name,
673 : // and bypass the very gate that decides which documents exist here.
674 21 : if (!isAnnouncedDocument(conversationId, documentId)) {
675 0 : JAMI_WARNING("[Account {}] Refusing to open document {}: it was not announced in conversation {}",
676 : accountId_,
677 : documentId,
678 : conversationId);
679 0 : return {};
680 : }
681 21 : auto account = account_.lock();
682 21 : if (!account)
683 0 : return {};
684 21 : auto* cm = account->convModule();
685 21 : if (!cm)
686 0 : return {};
687 21 : auto conversation = documentConversation(documentId);
688 21 : if (!conversation) {
689 : // Opening is what opts this device into holding a replica: clone the
690 : // document's swarm from its announcer. The clone lands asynchronously
691 : // -- through the very pipeline a conversation invite uses -- and
692 : // reports through onRepositoryUpdated(), which replays it into this
693 : // session and hands the client the difference. Until then the document
694 : // is open and empty, exactly like a conversation still syncing.
695 18 : std::string announcer;
696 18 : std::string announcedName;
697 18 : for (const auto& doc : documents(conversationId)) {
698 18 : auto idIt = doc.find("id");
699 18 : if (idIt == doc.end() || idIt->second != documentId)
700 0 : continue;
701 36 : if (auto authorIt = doc.find("author"); authorIt != doc.end())
702 18 : announcer = authorIt->second;
703 36 : if (auto nameIt = doc.find("displayName"); nameIt != doc.end())
704 18 : announcedName = nameIt->second;
705 18 : break;
706 18 : }
707 18 : if (announcer.empty()) {
708 0 : JAMI_WARNING("[Account {}] Unable to open document {}: its announcer is unknown", accountId_, documentId);
709 0 : return {};
710 : }
711 18 : auto session = ensureSession(conversationId, documentId);
712 : {
713 : // Nothing on disk yet, so nothing to replay; flagging it now is
714 : // what lets the clone's completion replay into this session.
715 18 : std::lock_guard<std::mutex> lk(mutex_);
716 18 : session->persistedLoaded = true;
717 18 : session->open = true;
718 : // The client saw the announcement's name; recording it is what lets
719 : // a rename that lands with (or after) the clone be seen as one.
720 18 : if (!session->announcedName)
721 17 : session->announcedName = announcedName;
722 18 : }
723 : // The announcer is the likeliest holder, but no peer is a reliable
724 : // one: it may be offline, or be this very account -- its creator
725 : // reopening after a leave took the replica away, and only the
726 : // account's other devices can then serve it. Which members hold a
727 : // copy cannot be known without the repository, so every joined
728 : // member is a candidate; the module fetches from a couple and lets
729 : // its fallback rounds walk the rest.
730 54 : std::vector<std::string> candidates {announcer};
731 54 : for (const auto& member : cm->getConversationMembers(conversationId)) {
732 36 : auto uriIt = member.find("uri");
733 36 : if (uriIt == member.end() || uriIt->second == announcer)
734 18 : continue;
735 : // Invited, banned and left members cannot hold a replica:
736 : // holding one starts with an open, which they are refused.
737 18 : auto roleIt = member.find("role");
738 18 : if (roleIt == member.end() || (roleIt->second != "admin" && roleIt->second != "member"))
739 0 : continue;
740 18 : candidates.emplace_back(uriIt->second);
741 18 : }
742 : // The parent swarm already knows who is reachable right now: members
743 : // with a connected device come first, so the initial fetches go to
744 : // peers that can actually answer. Ties keep the announcer in front
745 : // as the likeliest holder.
746 18 : if (auto parent = cm->getConversation(conversationId)) {
747 18 : std::set<std::string> online;
748 36 : for (const auto& device : parent->peersToSyncWith()) {
749 18 : auto uri = parent->uriFromDevice(device.toString());
750 18 : if (!uri.empty())
751 18 : online.emplace(std::move(uri));
752 36 : }
753 18 : std::stable_partition(candidates.begin(), candidates.end(), [&](const auto& uri) {
754 36 : return online.count(uri) != 0;
755 : });
756 36 : }
757 18 : cm->cloneDocumentFrom(documentId, candidates);
758 : // No channels yet: they need the members recorded in the repository, so
759 : // the clone's completion is what opens them. Until then the document is
760 : // open and empty, exactly like a conversation still syncing.
761 18 : return session->doc->encodeStateAsUpdate();
762 18 : }
763 3 : auto session = ensureSession(conversationId, documentId);
764 : // Rebuild the CRDT state from persisted commits if this session was just created,
765 : // so a document opens with its full content even when the daemon restarted or the
766 : // commits were never replayed yet.
767 3 : loadPersistedState(session);
768 : // Remember the name the client is about to see, so a later rename can be
769 : // told from it.
770 : {
771 3 : auto infos = conversation->infos();
772 3 : auto it = infos.find("title");
773 3 : const auto name = it != infos.end() ? it->second : std::string {};
774 3 : std::lock_guard<std::mutex> lk(mutex_);
775 3 : session->open = true;
776 3 : if (!session->announcedName)
777 0 : session->announcedName = name;
778 3 : }
779 3 : auto state = session->doc->encodeStateAsUpdate();
780 : // Reach out to the other devices editing the document. What they produced
781 : // while nothing was open here is not asked for -- there is no handshake --
782 : // and does not need to be: it arrives with its producer's next checkpoint,
783 : // while everything from here on arrives live.
784 3 : connectRealtimeChannels(session);
785 3 : return state;
786 39 : }
787 :
788 : void
789 8 : CollaborativeEditing::closeDocument(const std::string& conversationId, const std::string& documentId)
790 : {
791 8 : auto session = findSession(conversationId, documentId);
792 8 : if (!session)
793 0 : return;
794 : {
795 8 : std::lock_guard<std::mutex> lk(mutex_);
796 8 : session->open = false;
797 8 : }
798 : // Flush pending edits, but keep the in-memory CRDT replica so that reopening
799 : // the document shows its current content. The session stays consistent via
800 : // persisted commits (replayed on load) and live updates from other members.
801 8 : if (session->checkpointTimer) {
802 8 : std::lock_guard<std::mutex> lk(session->timerMutex);
803 8 : session->checkpointTimer->cancel();
804 8 : }
805 8 : checkpointNow(session);
806 : // Withdraw this device's awareness state, which is what clears its cursor
807 : // for the other members. Done explicitly rather than left to expire so that
808 : // closing an editor is seen at once instead of a timeout later.
809 8 : publishAwareness(session, {});
810 : // The channels only exist between devices that are editing, and this one no
811 : // longer is. A reopen starts them over.
812 8 : closeRealtimeChannels(session);
813 8 : }
814 :
815 : void
816 6 : CollaborativeEditing::applyUpdate(const std::string& conversationId,
817 : const std::string& documentId,
818 : const YrsDocument::Bytes& update)
819 : {
820 6 : auto session = findSession(conversationId, documentId);
821 6 : if (!session)
822 0 : return;
823 6 : if (update.size() > MAX_UPDATE_SIZE) {
824 0 : JAMI_WARNING("[Account {}] [Document {}] Discarding a {} byte update from the client: "
825 : "over the {} byte limit",
826 : accountId_,
827 : documentId,
828 : update.size(),
829 : MAX_UPDATE_SIZE);
830 0 : return;
831 : }
832 : // Merge before forwarding: an update the engine rejects must not be sent to
833 : // the members nor written to the repository.
834 6 : if (!session->doc->applyUpdate(update))
835 0 : return;
836 6 : onLocalUpdate(session, update);
837 6 : }
838 :
839 : YrsDocument::Bytes
840 5 : CollaborativeEditing::documentState(const std::string& conversationId, const std::string& documentId)
841 : {
842 5 : auto session = findSession(conversationId, documentId);
843 10 : return session ? session->doc->encodeStateAsUpdate() : YrsDocument::Bytes {};
844 5 : }
845 :
846 : namespace {
847 :
848 : /// Frame a payload and write it to one channel.
849 : void
850 3 : writeFrame(const std::shared_ptr<dhtnet::ChannelSocket>& socket, uint8_t tag, const std::vector<uint8_t>& payload)
851 : {
852 3 : std::vector<uint8_t> frame;
853 3 : frame.reserve(payload.size() + 11);
854 3 : appendVarUint(frame, payload.size());
855 3 : frame.push_back(tag);
856 3 : frame.insert(frame.end(), payload.begin(), payload.end());
857 3 : std::error_code ec;
858 3 : socket->write(frame.data(), frame.size(), ec);
859 3 : if (ec)
860 0 : JAMI_WARNING("Unable to send a {} byte collaborative frame: {}", frame.size(), ec.message());
861 3 : }
862 :
863 : } // namespace
864 :
865 : bool
866 17 : CollaborativeEditing::acceptsRealtimeChannel(const std::string& documentId,
867 : const std::string& peer,
868 : const std::string& deviceId)
869 : {
870 : // Only for a document being edited here. A closed holder replicates through
871 : // the swarm and has no use for live frames; a device that never held the
872 : // document has nothing to accept them into; and answering for ids nobody
873 : // opened would let a peer probe what this device knows.
874 17 : auto session = findSessionByDocument(documentId);
875 17 : if (!session)
876 0 : return false;
877 : {
878 17 : std::lock_guard<std::mutex> lk(mutex_);
879 17 : if (!session->open)
880 14 : return false;
881 17 : }
882 : // Membership per the document's own replica, never the parent
883 : // conversation's. Invited counts as in: a document member is only ever
884 : // invited by a holder having just vouched for it while serving its clone,
885 : // and its join follows by itself -- there is no pending-invitation state to
886 : // wrongly admit. A joiner whose commits have not reached us at all is still
887 : // refused, and reached out to the moment they are merged.
888 3 : auto conversation = documentConversation(documentId);
889 3 : if (!conversation)
890 0 : return false; // the clone is still in flight: nothing to check against
891 3 : return conversation->isPeerAuthorized(peer, deviceId, true);
892 17 : }
893 :
894 : void
895 6 : CollaborativeEditing::onRealtimeChannel(const std::string& documentId,
896 : const std::string& peer,
897 : const std::string& deviceId,
898 : std::shared_ptr<dhtnet::ChannelSocket> socket)
899 : {
900 6 : auto session = findSessionByDocument(documentId);
901 6 : bool open = false;
902 6 : if (session) {
903 6 : std::lock_guard<std::mutex> lk(mutex_);
904 6 : open = session->open;
905 6 : }
906 6 : if (!open) {
907 : // The document was closed while the channel was being set up.
908 0 : socket->shutdown();
909 0 : return;
910 : }
911 : {
912 6 : std::lock_guard<std::mutex> lk(session->protocolMutex);
913 6 : session->channels[deviceId].push_back(socket);
914 6 : }
915 :
916 : // The parsing state of this one channel. Only its own receive callback
917 : // touches it, and those are delivered in order, so it needs no lock.
918 6 : auto buffer = std::make_shared<std::vector<uint8_t>>();
919 6 : std::weak_ptr<CollaborativeEditing> wthis = weak_from_this();
920 6 : std::weak_ptr<Session> wsession = session;
921 6 : std::weak_ptr<dhtnet::ChannelSocket> wsocket = socket;
922 6 : socket->setOnRecv([wthis, wsession, wsocket, peer, buffer](const uint8_t* data, size_t size) {
923 3 : auto sthis = wthis.lock();
924 3 : auto session = wsession.lock();
925 3 : if (!sthis || !session)
926 0 : return size;
927 3 : buffer->insert(buffer->end(), data, data + size);
928 3 : size_t pos = 0;
929 6 : while (pos < buffer->size()) {
930 3 : uint64_t length = 0;
931 3 : size_t consumed = 0;
932 3 : const auto res = readVarUint(buffer->data() + pos, buffer->size() - pos, length, consumed);
933 3 : if (res == VarUint::INCOMPLETE)
934 0 : break;
935 : // The length is judged before anything is held against it: a peer
936 : // must not have us buffer megabytes of a frame that could only be
937 : // refused once complete. Nothing recovers a framing violation --
938 : // there is no way back into sync with a stream that lies about its
939 : // lengths -- so the channel goes down with it.
940 3 : if (res == VarUint::MALFORMED || length > MAX_UPDATE_SIZE) {
941 0 : buffer->clear();
942 0 : if (auto socket = wsocket.lock())
943 0 : socket->shutdown();
944 0 : return size;
945 : }
946 3 : if (buffer->size() - pos - consumed < 1 + length)
947 0 : break; // the rest of the frame is still in flight
948 3 : const uint8_t tag = (*buffer)[pos + consumed];
949 3 : sthis->onFrame(session, peer, tag, buffer->data() + pos + consumed + 1, length);
950 3 : pos += consumed + 1 + length;
951 : }
952 3 : buffer->erase(buffer->begin(), buffer->begin() + pos);
953 3 : return size;
954 3 : });
955 6 : socket->onShutdown([wsession, wsocket, deviceId](const std::error_code& /*ec*/) {
956 6 : auto session = wsession.lock();
957 6 : if (!session)
958 0 : return;
959 6 : auto socket = wsocket.lock();
960 6 : std::lock_guard<std::mutex> lk(session->protocolMutex);
961 6 : auto it = session->channels.find(deviceId);
962 6 : if (it == session->channels.end())
963 1 : return;
964 5 : auto& list = it->second;
965 10 : list.erase(std::remove_if(list.begin(), list.end(), [&](const auto& s) { return !socket || s == socket; }),
966 5 : list.end());
967 5 : if (list.empty())
968 5 : session->channels.erase(it);
969 8 : });
970 :
971 : // Tell the newcomer at once who this device is in the document: our
972 : // awareness state only re-announces itself at the renewal cadence, and an
973 : // editor joining a session should not stare at an empty document for
974 : // fifteen seconds before the cursors appear.
975 6 : std::vector<uint8_t> hello;
976 : {
977 6 : std::lock_guard<std::mutex> lk(session->protocolMutex);
978 6 : if (!session->localState.empty())
979 0 : hello = encodeAwareness({{clientId_, session->localClock, session->localState}});
980 6 : }
981 6 : if (!hello.empty())
982 0 : writeFrame(socket, FRAME_AWARENESS, hello);
983 6 : }
984 :
985 : void
986 39 : CollaborativeEditing::connectRealtimeChannels(const std::shared_ptr<Session>& session)
987 : {
988 39 : auto account = account_.lock();
989 39 : if (!account)
990 0 : return;
991 : {
992 39 : std::lock_guard<std::mutex> lk(mutex_);
993 39 : if (!session->open)
994 14 : return;
995 39 : }
996 25 : auto conversation = documentConversation(session->documentId);
997 25 : if (!conversation)
998 0 : return; // the clone is still in flight: its completion comes back here
999 : // Every device the repository lists is asked, not just the ones the
1000 : // document's swarm is connected to right now: the channel dials through the
1001 : // connection manager, which reaches a device the DRT has not settled on
1002 : // yet. Most are declined -- the other end only accepts while it has the
1003 : // document open -- and the ones that matter are the editors.
1004 25 : const auto ownDevice = std::string(account->currentDeviceId());
1005 72 : for (const auto& [member, devices] : conversation->memberDevices()) {
1006 94 : for (const auto& device : devices) {
1007 47 : const auto deviceId = device.toString();
1008 47 : if (deviceId == ownDevice)
1009 24 : continue;
1010 : {
1011 23 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1012 23 : if (session->channels.count(deviceId) != 0)
1013 3 : continue; // already talking to it
1014 23 : }
1015 : // The socket is handled where the accepting side's is, in
1016 : // YdocChannelHandler::onReady; a refusal needs nothing done.
1017 20 : account->connectYdocDevice(device, session->documentId);
1018 47 : }
1019 25 : }
1020 39 : }
1021 :
1022 : void
1023 12 : CollaborativeEditing::closeRealtimeChannels(const std::shared_ptr<Session>& session)
1024 : {
1025 12 : decltype(session->channels) channels;
1026 : {
1027 12 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1028 12 : channels = std::move(session->channels);
1029 12 : session->channels.clear();
1030 12 : }
1031 : // Outside the lock: shutting a socket down runs its shutdown handler, which
1032 : // takes the same lock to unregister -- and finds nothing left to.
1033 13 : for (auto& [_, list] : channels)
1034 2 : for (auto& socket : list)
1035 1 : socket->shutdown();
1036 12 : }
1037 :
1038 : void
1039 8 : CollaborativeEditing::broadcastFrame(const std::shared_ptr<Session>& session,
1040 : uint8_t tag,
1041 : const std::vector<uint8_t>& payload)
1042 : {
1043 8 : if (payload.empty())
1044 0 : return;
1045 : // Written outside the lock: a send can stall on a congested peer, and the
1046 : // receive path needs the lock to route what the others are saying.
1047 8 : std::vector<std::shared_ptr<dhtnet::ChannelSocket>> sockets;
1048 : {
1049 8 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1050 11 : for (const auto& [_, list] : session->channels)
1051 3 : sockets.insert(sockets.end(), list.begin(), list.end());
1052 8 : }
1053 11 : for (const auto& socket : sockets)
1054 3 : writeFrame(socket, tag, payload);
1055 8 : }
1056 :
1057 : void
1058 3 : CollaborativeEditing::onFrame(
1059 : const std::shared_ptr<Session>& session, const std::string& from, uint8_t tag, const uint8_t* payload, size_t size)
1060 : {
1061 3 : switch (tag) {
1062 1 : case FRAME_UPDATE:
1063 1 : onRemoteUpdate(session, YrsDocument::Bytes(payload, payload + size));
1064 1 : break;
1065 2 : case FRAME_AWARENESS:
1066 2 : onAwarenessPayload(session, from, payload, size);
1067 2 : break;
1068 0 : default:
1069 : // A frame type from a newer daemon. Skipping it -- its length is known
1070 : // -- is what lets one be added without every peer upgrading first.
1071 0 : break;
1072 : }
1073 3 : }
1074 :
1075 : void
1076 1 : CollaborativeEditing::onRemoteUpdate(const std::shared_ptr<Session>& session, const YrsDocument::Bytes& update)
1077 : {
1078 1 : if (update.empty())
1079 0 : return;
1080 : // Cleared before the update rather than read after it alone, so that what
1081 : // this update brought is not confused with what an earlier one did.
1082 1 : session->doc->takeChanged();
1083 1 : if (!session->doc->applyUpdate(update))
1084 0 : return; // malformed: don't hand it to the clients
1085 : // Nothing when the update taught the replica nothing: a frame that raced a
1086 : // checkpoint fetch carries what the replay already merged, and forwarding
1087 : // it would light an "unread" badge on a document nobody touched.
1088 1 : if (!session->doc->takeChanged())
1089 0 : return;
1090 : // Channels only exist while the document is open here, but a frame can slip
1091 : // in between the close and the sockets going down; reopening hands the
1092 : // merged state over instead.
1093 : {
1094 1 : std::lock_guard<std::mutex> lk(mutex_);
1095 1 : if (!session->open)
1096 0 : return;
1097 1 : }
1098 : // Not persisted here: the device that produced it checkpoints it into its own
1099 : // repository and it reaches ours through synchronization. Storing it again
1100 : // would keep one copy per member of every single edit.
1101 1 : emitUpdate(session->conversationId, session->documentId, update);
1102 : }
1103 :
1104 : void
1105 2 : CollaborativeEditing::onAwarenessPayload(const std::shared_ptr<Session>& session,
1106 : const std::string& from,
1107 : const uint8_t* data,
1108 : size_t size)
1109 : {
1110 2 : if (size > MAX_AWARENESS_MESSAGE_SIZE)
1111 0 : return;
1112 2 : std::vector<AwarenessWire> entries;
1113 : try {
1114 2 : msgpack::object_handle oh = msgpack::unpack(reinterpret_cast<const char*>(data), size);
1115 2 : oh.get().convert(entries);
1116 2 : } catch (const std::exception&) {
1117 0 : return; // a peer sent something that is not an awareness message
1118 0 : }
1119 :
1120 : // What the local clients have to be told, gathered while the table is locked
1121 : // and emitted once it is not: a signal handler is entitled to call back into
1122 : // this manager.
1123 2 : std::vector<std::pair<uint64_t, std::string>> changed;
1124 2 : std::vector<uint64_t> left;
1125 2 : bool contested = false;
1126 2 : const auto now = std::chrono::steady_clock::now();
1127 : {
1128 2 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1129 4 : for (const auto& entry : entries) {
1130 2 : if (entry.state.size() > MAX_AWARENESS_SIZE)
1131 0 : continue; // a presence state, not a payload
1132 2 : if (entry.clientId == clientId_) {
1133 : // Someone is speaking for this device. Nothing legitimate does
1134 : // that, and the protocol's own answer -- outrun it so that peers
1135 : // keep our real state -- is also the right one here.
1136 0 : contested = true;
1137 0 : continue;
1138 : }
1139 2 : auto it = session->awareness.find(entry.clientId);
1140 2 : if (it != session->awareness.end()) {
1141 1 : if (it->second.owner != from)
1142 0 : continue; // a member may not speak for another's client id
1143 : // Strictly greater: two states with the same clock are the same
1144 : // state having taken two routes, and re-emitting one of them
1145 : // would make a cursor jump back to where it already was.
1146 1 : if (entry.clock <= it->second.clock)
1147 0 : continue;
1148 : } else {
1149 1 : if (session->awareness.size() >= MAX_AWARENESS_PEERS)
1150 0 : continue;
1151 1 : if (entry.state.empty() || entry.state == "null")
1152 0 : continue; // a client that is gone and was never here
1153 : }
1154 2 : const bool gone = entry.state.empty() || entry.state == "null";
1155 2 : if (gone) {
1156 1 : session->awareness.erase(entry.clientId);
1157 1 : left.push_back(entry.clientId);
1158 : } else {
1159 1 : auto& peer = session->awareness[entry.clientId];
1160 1 : peer.clock = entry.clock;
1161 1 : peer.state = entry.state;
1162 1 : peer.lastSeen = now;
1163 1 : peer.owner = from;
1164 1 : changed.emplace_back(entry.clientId, entry.state);
1165 : }
1166 : }
1167 2 : }
1168 :
1169 3 : for (const auto& [clientId, state] : changed)
1170 2 : emitSignal<libjami::ConversationSignal::CollaborativeAwarenessChanged>(accountId_,
1171 1 : session->conversationId,
1172 1 : session->documentId,
1173 : from,
1174 : clientId,
1175 : state);
1176 3 : for (auto clientId : left)
1177 2 : emitSignal<libjami::ConversationSignal::CollaborativeParticipantLeft>(accountId_,
1178 1 : session->conversationId,
1179 1 : session->documentId,
1180 : from,
1181 : clientId);
1182 2 : if (contested) {
1183 0 : JAMI_WARNING("[Account {}] [Document {}] {} announced this device's client id; re-announcing",
1184 : accountId_,
1185 : session->documentId,
1186 : from);
1187 0 : std::string state;
1188 : {
1189 0 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1190 0 : state = session->localState;
1191 0 : }
1192 0 : publishAwareness(session, state);
1193 0 : }
1194 2 : scheduleAwarenessUpkeep(session);
1195 2 : }
1196 :
1197 : void
1198 1 : CollaborativeEditing::setAwareness(const std::string& conversationId,
1199 : const std::string& documentId,
1200 : const std::string& state)
1201 : {
1202 1 : if (state.size() > MAX_AWARENESS_SIZE) {
1203 0 : JAMI_WARNING("[Account {}] [Document {}] Refusing to broadcast an oversized awareness state",
1204 : accountId_,
1205 : documentId);
1206 0 : return;
1207 : }
1208 : // Only for a document this device has open: an awareness state is about
1209 : // where its editor is, and there is no editor before that.
1210 1 : if (auto session = findSession(conversationId, documentId))
1211 1 : publishAwareness(session, state);
1212 : }
1213 :
1214 : void
1215 9 : CollaborativeEditing::publishAwareness(const std::shared_ptr<Session>& session, const std::string& state)
1216 : {
1217 9 : AwarenessWire entry;
1218 : {
1219 9 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1220 : // Withdrawing a state that was never shared would tell the members about
1221 : // an editor they were never told about in the first place.
1222 9 : if (state.empty() && session->localState.empty())
1223 7 : return;
1224 2 : entry.clientId = clientId_;
1225 2 : entry.clock = ++session->localClock;
1226 : // "null" is how a client that is no longer there is spelled.
1227 3 : entry.state = state.empty() ? "null" : state;
1228 2 : session->localState = state;
1229 2 : session->localAnnounced = std::chrono::steady_clock::now();
1230 9 : }
1231 6 : broadcastFrame(session, FRAME_AWARENESS, encodeAwareness({entry}));
1232 2 : scheduleAwarenessUpkeep(session);
1233 11 : }
1234 :
1235 : void
1236 4 : CollaborativeEditing::scheduleAwarenessUpkeep(const std::shared_ptr<Session>& session)
1237 : {
1238 4 : if (!session->awarenessTimer)
1239 0 : return;
1240 : {
1241 4 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1242 : // One timer at a time, and none at all while nobody is editing: a
1243 : // document nobody has open must not keep waking the process up.
1244 4 : if (session->upkeepRunning)
1245 2 : return;
1246 2 : if (session->awareness.empty() && session->localState.empty())
1247 0 : return;
1248 2 : session->upkeepRunning = true;
1249 2 : session->awarenessTimer->expires_after(AWARENESS_SWEEP);
1250 4 : }
1251 2 : std::weak_ptr<CollaborativeEditing> wthis = weak_from_this();
1252 2 : std::weak_ptr<Session> wsession = session;
1253 2 : session->awarenessTimer->async_wait([wthis, wsession](const asio::error_code& ec) {
1254 2 : auto sthis = wthis.lock();
1255 2 : auto session = wsession.lock();
1256 2 : if (!sthis || !session)
1257 2 : return;
1258 : {
1259 0 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1260 0 : session->upkeepRunning = false;
1261 0 : }
1262 0 : if (!ec)
1263 0 : sthis->awarenessUpkeep(session);
1264 4 : });
1265 2 : }
1266 :
1267 : void
1268 0 : CollaborativeEditing::awarenessUpkeep(const std::shared_ptr<Session>& session)
1269 : {
1270 0 : const auto now = std::chrono::steady_clock::now();
1271 0 : std::vector<std::pair<uint64_t, std::string>> expired;
1272 0 : std::string renew;
1273 : {
1274 0 : std::lock_guard<std::mutex> lk(session->protocolMutex);
1275 0 : for (auto it = session->awareness.begin(); it != session->awareness.end();) {
1276 0 : if (now - it->second.lastSeen >= AWARENESS_TIMEOUT) {
1277 0 : expired.emplace_back(it->first, it->second.owner);
1278 0 : it = session->awareness.erase(it);
1279 : } else {
1280 0 : ++it;
1281 : }
1282 : }
1283 : // Re-announced well before it would expire elsewhere, so that a single
1284 : // lost message does not make this device blink out of the document.
1285 0 : if (!session->localState.empty() && now - session->localAnnounced >= AWARENESS_RENEW)
1286 0 : renew = session->localState;
1287 0 : }
1288 0 : for (const auto& [clientId, owner] : expired)
1289 0 : emitSignal<libjami::ConversationSignal::CollaborativeParticipantLeft>(accountId_,
1290 0 : session->conversationId,
1291 0 : session->documentId,
1292 : owner,
1293 : clientId);
1294 0 : if (!renew.empty())
1295 0 : publishAwareness(session, renew); // rearms the timer on its way out
1296 : else
1297 0 : scheduleAwarenessUpkeep(session);
1298 0 : }
1299 :
1300 : void
1301 6 : CollaborativeEditing::onLocalUpdate(const std::shared_ptr<Session>& session, const YrsDocument::Bytes& update)
1302 : {
1303 : // Real-time path: hand the incremental update to the devices editing along.
1304 6 : broadcastFrame(session, FRAME_UPDATE, update);
1305 : // Durable path: accumulate the update for the next checkpoint.
1306 6 : queueUpdate(session, update);
1307 6 : }
1308 :
1309 : void
1310 36 : CollaborativeEditing::replayStoredUpdates(const std::shared_ptr<Session>& session)
1311 : {
1312 36 : auto conversation = documentConversation(session->documentId);
1313 36 : if (!conversation)
1314 0 : return;
1315 : // The updates come from peers' commits, so their content is not ours to
1316 : // trust: a malformed one must cost that one update, not the whole replay.
1317 47 : for (const auto& encoded : conversation->documentUpdates()) {
1318 : try {
1319 11 : session->doc->applyUpdate(base64::decode(encoded));
1320 0 : } catch (const std::exception& e) {
1321 0 : JAMI_WARNING("[Account {}] [Document {}] Skipping unreadable stored update: {}",
1322 : accountId_,
1323 : session->documentId,
1324 : e.what());
1325 0 : }
1326 36 : }
1327 36 : }
1328 :
1329 : void
1330 3 : CollaborativeEditing::loadPersistedState(const std::shared_ptr<Session>& session)
1331 : {
1332 : {
1333 3 : std::lock_guard<std::mutex> lk(mutex_);
1334 3 : if (session->persistedLoaded)
1335 3 : return;
1336 3 : }
1337 : // Replay the updates stored in the document's repository. Nothing is
1338 : // signalled: the caller encodes the converged state and hands it to the
1339 : // client that asked to open the document.
1340 0 : replayStoredUpdates(session);
1341 : // What is already stored is not news: the client resolves the attachments
1342 : // of the state it is being handed. Only what arrives afterwards is signalled.
1343 : // The asymmetry is deliberate. For a document being opened the client pulls,
1344 : // asking for each reference it meets while rendering; announcing every stored
1345 : // attachment here would push bytes nobody has asked for yet, on a document the
1346 : // user may never scroll through. The signal exists for the opposite case: an
1347 : // attachment landing in an already open document, which the client has no
1348 : // reason to look for.
1349 0 : std::vector<std::string> ids;
1350 0 : if (auto conversation = documentConversation(session->documentId))
1351 0 : ids = conversation->documentAttachmentIds();
1352 : // Only now: a session flagged as loaded is never replayed again, so flagging
1353 : // it before the replay would freeze a partially rebuilt document.
1354 0 : std::lock_guard<std::mutex> lk(mutex_);
1355 0 : session->persistedLoaded = true;
1356 0 : for (auto& id : ids)
1357 0 : session->knownAttachments.insert(std::move(id));
1358 0 : }
1359 :
1360 : void
1361 6 : CollaborativeEditing::queueUpdate(const std::shared_ptr<Session>& session, const YrsDocument::Bytes& update)
1362 : {
1363 6 : const bool held = documentConversation(session->documentId) != nullptr;
1364 6 : bool capReached = false;
1365 : {
1366 6 : std::lock_guard<std::mutex> lk(session->pendingMutex);
1367 : // Without a repository nothing will ever drain this: keep the last batch
1368 : // so a repository appearing later still saves recent work, and drop the
1369 : // rest rather than growing without bound.
1370 6 : if (!held && session->pending.size() >= PENDING_HARD_CAP)
1371 0 : session->pending.erase(session->pending.begin(), session->pending.begin() + CHECKPOINT_MAX_PENDING);
1372 6 : session->pending.emplace_back(base64::encode(update));
1373 6 : capReached = session->pending.size() >= CHECKPOINT_MAX_PENDING;
1374 6 : }
1375 : // Checkpointing reads the document back, so it must never run inline on the
1376 : // path that just wrote to it: scheduleCheckpoint() only arms a timer.
1377 6 : if (capReached)
1378 0 : session->checkpointDue = true;
1379 6 : scheduleCheckpoint(session, capReached ? std::chrono::seconds(0) : CHECKPOINT_IDLE);
1380 6 : }
1381 :
1382 : void
1383 6 : CollaborativeEditing::scheduleCheckpoint(const std::shared_ptr<Session>& session, std::chrono::seconds delay)
1384 : {
1385 6 : if (!session->checkpointTimer)
1386 0 : return;
1387 : // Once the batch has reached its cap, stop letting new edits push the
1388 : // deadline further away: sustained typing would otherwise never checkpoint.
1389 6 : if (delay.count() > 0 && session->checkpointDue)
1390 0 : return;
1391 6 : std::weak_ptr<CollaborativeEditing> wthis = weak_from_this();
1392 6 : std::weak_ptr<Session> wsession = session;
1393 6 : std::lock_guard<std::mutex> lk(session->timerMutex);
1394 6 : session->checkpointTimer->expires_after(delay);
1395 6 : session->checkpointTimer->async_wait([wthis, wsession](const asio::error_code& ec) {
1396 6 : if (ec) // cancelled by a newer edit (debounce) or by shutdown
1397 1 : return;
1398 5 : auto sthis = wthis.lock();
1399 5 : auto session = wsession.lock();
1400 5 : if (sthis && session)
1401 5 : sthis->checkpointNow(session);
1402 5 : });
1403 6 : }
1404 :
1405 : void
1406 42 : CollaborativeEditing::checkpointNow(const std::shared_ptr<Session>& session)
1407 : {
1408 42 : session->checkpointDue = false;
1409 42 : auto account = account_.lock();
1410 42 : if (!account)
1411 0 : return;
1412 42 : auto* cm = account->convModule();
1413 42 : if (!cm)
1414 0 : return;
1415 : // Before draining anything: a drained batch with nowhere to go is lost.
1416 : // Keeping the batch queued means a replica appearing later -- the document
1417 : // being opened, which is what clones it -- still saves recent work.
1418 42 : if (!documentConversation(session->documentId))
1419 5 : return;
1420 37 : std::vector<std::string> batch;
1421 : {
1422 37 : std::lock_guard<std::mutex> lk(session->pendingMutex);
1423 37 : batch.swap(session->pending);
1424 37 : }
1425 37 : if (batch.empty())
1426 31 : return;
1427 :
1428 : // The checkpoint is a commit in the document's own swarm: committing it is
1429 : // also what announces it, so the other holders fetch it through the same
1430 : // pipeline that moves conversation messages. Nothing else needs sending.
1431 18 : cm->createCommit(session->documentId,
1432 12 : CommitMessage::checkpoint(batch),
1433 : true,
1434 : {},
1435 12 : [w = weak_from_this(), wsession = std::weak_ptr<Session>(session), batch](bool ok,
1436 : const std::string&) {
1437 6 : if (ok)
1438 6 : return;
1439 : // Keep the updates queued so the next checkpoint retries
1440 : // them rather than silently losing the edits they carry,
1441 : // and make sure a retry is actually scheduled even if the
1442 : // user has stopped typing.
1443 0 : auto sthis = w.lock();
1444 0 : auto session = wsession.lock();
1445 0 : if (!sthis || !session)
1446 0 : return;
1447 : {
1448 0 : std::lock_guard<std::mutex> lk(session->pendingMutex);
1449 0 : session->pending.insert(session->pending.begin(), batch.begin(), batch.end());
1450 0 : }
1451 0 : sthis->scheduleCheckpoint(session, CHECKPOINT_IDLE);
1452 0 : });
1453 73 : }
1454 :
1455 : std::vector<std::map<std::string, std::string>>
1456 61 : CollaborativeEditing::history(const std::string& /*conversationId*/, const std::string& documentId, size_t max)
1457 : {
1458 61 : auto conversation = documentConversation(documentId);
1459 122 : return conversation ? conversation->documentHistory(max) : std::vector<std::map<std::string, std::string>> {};
1460 61 : }
1461 :
1462 : YrsDocument::Bytes
1463 0 : CollaborativeEditing::documentStateAt(const std::string& /*conversationId*/,
1464 : const std::string& documentId,
1465 : const std::string& commitId)
1466 : {
1467 0 : auto conversation = documentConversation(documentId);
1468 0 : if (!conversation)
1469 0 : return {};
1470 :
1471 : // Nothing at all when the checkpoint is unknown, which is what the public
1472 : // contract promises. It has to be told apart from a checkpoint that exists
1473 : // and holds nothing: the two would otherwise be the same answer, and a
1474 : // client restoring an early, legitimately empty version could not tell
1475 : // whether it was allowed to.
1476 0 : const auto stored = conversation->documentUpdatesAt(commitId);
1477 0 : if (!stored)
1478 0 : return {};
1479 :
1480 : // Replay into a throwaway replica: the live document must not be touched.
1481 : // What the client does with that state -- show it, restore it, diff it -- is
1482 : // its own business, and depends on a document type the daemon ignores.
1483 0 : YrsDocument snapshot {replicaId()};
1484 0 : for (const auto& encoded : *stored) {
1485 : try {
1486 0 : snapshot.applyUpdate(base64::decode(encoded));
1487 0 : } catch (const std::exception& e) {
1488 0 : JAMI_WARNING("[Account {}] [Document {}] Skipping unreadable stored update: {}",
1489 : accountId_,
1490 : documentId,
1491 : e.what());
1492 0 : }
1493 : }
1494 0 : return snapshot.encodeStateAsUpdate();
1495 0 : }
1496 :
1497 : void
1498 54 : CollaborativeEditing::onDocumentAnnounced(const std::string& conversationId, const std::string& documentId)
1499 : {
1500 : // The author may have retired this announcement. Answered from the cache
1501 : // alone, never by walking the conversation again: this runs while
1502 : // addToHistory() holds the conversation lock, and asking the conversation
1503 : // anything from here is what deadlocks the caller. addToHistory() applies the
1504 : // removals of a batch before its announcements, and a removal is always newer
1505 : // than the announcement it retires, so the cache is already right by now.
1506 : //
1507 : // Nothing is replicated here: holding a replica is a per-device choice, made
1508 : // by opening the document. The announcement only records that it exists.
1509 54 : std::lock_guard<std::mutex> lk(announcedMtx_);
1510 54 : if (auto it = removed_.find(conversationId); it != removed_.end() && it->second.count(documentId) != 0)
1511 0 : return;
1512 54 : announced_[conversationId].emplace(documentId);
1513 54 : }
1514 :
1515 : void
1516 3 : CollaborativeEditing::onDocumentRemoved(const std::string& conversationId, const std::string& documentId)
1517 : {
1518 : {
1519 3 : std::lock_guard<std::mutex> lk(announcedMtx_);
1520 3 : removed_[conversationId].emplace(documentId);
1521 3 : if (auto it = announced_.find(conversationId); it != announced_.end())
1522 3 : it->second.erase(documentId);
1523 3 : }
1524 3 : dropLocalReplica(conversationId, documentId);
1525 : // The repository goes too: a document nobody can open again would otherwise
1526 : // outlive its own removal on every device that held it. From another
1527 : // thread: this runs while addToHistory() holds the parent conversation's
1528 : // lock, and tearing a conversation down takes locks of its own.
1529 3 : dht::ThreadPool::io().run([w = account_, documentId] {
1530 3 : if (auto account = w.lock())
1531 3 : if (auto* cm = account->convModule())
1532 3 : cm->removeDocumentReplica(documentId);
1533 3 : });
1534 3 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentRemoved>(accountId_, conversationId, documentId, true);
1535 3 : }
1536 :
1537 : void
1538 36 : CollaborativeEditing::onRepositoryUpdated(const std::string& conversationId, const std::string& documentId)
1539 : {
1540 : // A remote rename can land on a document nobody has open here, and both
1541 : // early returns below are reachable in that case: drop the cached name first
1542 : // or a client would keep showing the old one until the account restarts.
1543 : {
1544 36 : std::lock_guard<std::mutex> lk(mutex_);
1545 36 : nameCache_.erase(key(conversationId, documentId));
1546 36 : ++nameEpoch_;
1547 36 : }
1548 36 : auto session = findSession(conversationId, documentId);
1549 36 : auto conversation = documentConversation(documentId);
1550 36 : if (!session || !conversation)
1551 0 : return; // not being edited here; the repository is up to date on disk
1552 : {
1553 36 : std::lock_guard<std::mutex> lk(mutex_);
1554 36 : if (!session->persistedLoaded)
1555 0 : return; // never opened: it will be replayed on open
1556 36 : }
1557 : // What the replica knows before the replay, so that what it learns from it
1558 : // can be told apart from what it already had.
1559 36 : const auto before = session->doc->encodeStateVector();
1560 36 : session->doc->takeChanged();
1561 : // Applying an update the replica already knows is a no-op for a CRDT, so
1562 : // replaying the whole set is correct, just more work than strictly needed.
1563 36 : replayStoredUpdates(session);
1564 : // Nothing at all when the replay taught us nothing -- a rename-only commit,
1565 : // or updates the real-time path had already delivered -- or every client
1566 : // would light an "unread" badge for a document nobody touched. The question
1567 : // is put to yrs rather than answered by measuring the diff: yrs appends the
1568 : // whole deletion set to a diff without diffing it, so a diff carrying no new
1569 : // content still measures a few bytes on any document where a character was
1570 : // ever erased.
1571 : //
1572 : // What is sent is only what the replay brought, not the whole document: a
1573 : // synchronization usually carries a handful of keystrokes, and re-encoding
1574 : // a 300 kB document for each of them would push megabytes a minute through
1575 : // the client API for nothing. A closed holder's client is not handed the
1576 : // content either -- it gets the converged state when it reopens -- but it
1577 : // is told that there is some: an update with an empty payload, which is
1578 : // what lets it badge a document nobody here is watching.
1579 : bool tellClient;
1580 : {
1581 36 : std::lock_guard<std::mutex> lk(mutex_);
1582 36 : tellClient = session->open;
1583 36 : }
1584 36 : if (session->doc->takeChanged())
1585 5 : emitUpdate(conversationId, documentId, tellClient ? session->doc->encodeDiff(before) : YrsDocument::Bytes {});
1586 : // Independent of the updates above: an attachment is not part of the CRDT,
1587 : // so a synchronization can bring the payload of a reference the real-time
1588 : // path delivered long before, with no update at all to show for it.
1589 36 : emitNewAttachments(session);
1590 : // The name travels with the repository now, so a remote rename lands here.
1591 : // It cannot be detected by reading the name around the replay: the caller
1592 : // merges before calling us, so both reads would return the name from after
1593 : // the merge. What we compare against is the last name we told the clients.
1594 36 : std::string name;
1595 : {
1596 36 : auto infos = conversation->infos();
1597 72 : if (auto it = infos.find("title"); it != infos.end())
1598 36 : name = it->second;
1599 36 : }
1600 36 : auto renamed = false;
1601 : {
1602 36 : std::lock_guard<std::mutex> lk(mutex_);
1603 36 : nameCache_[key(conversationId, documentId)] = name;
1604 36 : ++nameEpoch_;
1605 : // Only a name the clients have already been told can be seen to change.
1606 36 : if (session->announcedName && *session->announcedName != name)
1607 1 : renamed = true;
1608 36 : if (session->announcedName)
1609 36 : session->announcedName = name;
1610 36 : }
1611 36 : if (renamed)
1612 1 : emitRename(conversationId, documentId, name);
1613 : // A synchronization is also how new editors become reachable: the clone
1614 : // this session may have been waiting for just landed, or a joiner's
1615 : // membership commits were just merged -- the very merge that entitles the
1616 : // device this replica refused a moment ago to its channel.
1617 36 : connectRealtimeChannels(session);
1618 36 : }
1619 :
1620 : void
1621 174 : CollaborativeEditing::flush()
1622 : {
1623 174 : std::vector<std::shared_ptr<Session>> sessions;
1624 : {
1625 174 : std::lock_guard<std::mutex> lk(mutex_);
1626 174 : sessions.reserve(sessions_.size());
1627 203 : for (const auto& [_, session] : sessions_)
1628 29 : sessions.emplace_back(session);
1629 174 : }
1630 203 : for (const auto& session : sessions) {
1631 29 : if (session->checkpointTimer) {
1632 29 : std::lock_guard<std::mutex> lk(session->timerMutex);
1633 29 : session->checkpointTimer->cancel();
1634 29 : }
1635 : // No compaction here: the edits are already durable, packing is pure
1636 : // housekeeping, and doing it inline would stall the account
1637 : // unregistration for seconds per open document.
1638 29 : checkpointNow(session);
1639 : }
1640 174 : }
1641 :
1642 : void
1643 6 : CollaborativeEditing::emitUpdate(const std::string& conversationId,
1644 : const std::string& documentId,
1645 : const YrsDocument::Bytes& update)
1646 : {
1647 : // One channel for every document type: the payload is an opaque Y-CRDT
1648 : // update, and the client merges it into its own replica. That is what makes
1649 : // a plain-text editor and a rich-text editor listen to the same signal.
1650 6 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentUpdate>(accountId_, conversationId, documentId, update);
1651 6 : }
1652 :
1653 : void
1654 3 : CollaborativeEditing::emitRename(const std::string& conversationId,
1655 : const std::string& documentId,
1656 : const std::string& name)
1657 : {
1658 3 : emitSignal<libjami::ConversationSignal::CollaborativeDocumentRenamed>(accountId_, conversationId, documentId, name);
1659 3 : }
1660 :
1661 : void
1662 36 : CollaborativeEditing::emitNewAttachments(const std::shared_ptr<Session>& session)
1663 : {
1664 36 : auto conversation = documentConversation(session->documentId);
1665 36 : if (!conversation)
1666 0 : return;
1667 36 : auto ids = conversation->documentAttachmentIds();
1668 36 : std::vector<std::string> fresh;
1669 : {
1670 36 : std::lock_guard<std::mutex> lk(mutex_);
1671 37 : for (auto& id : ids)
1672 1 : if (session->knownAttachments.insert(id).second)
1673 1 : fresh.push_back(std::move(id));
1674 36 : }
1675 37 : for (const auto& id : fresh)
1676 2 : emitSignal<libjami::ConversationSignal::CollaborativeAttachmentAdded>(accountId_,
1677 1 : session->conversationId,
1678 1 : session->documentId,
1679 : id);
1680 36 : }
1681 :
1682 : std::string
1683 1 : CollaborativeEditing::addAttachment(const std::string& conversationId,
1684 : const std::string& documentId,
1685 : const std::vector<uint8_t>& data)
1686 : {
1687 1 : if (data.empty() || data.size() > MAX_ATTACHMENT_SIZE) {
1688 0 : JAMI_WARNING("[Account {}] [Document {}] Attachment refused: {} byte(s), limit is {}",
1689 : accountId_,
1690 : documentId,
1691 : data.size(),
1692 : MAX_ATTACHMENT_SIZE);
1693 0 : return {};
1694 : }
1695 1 : auto account = account_.lock();
1696 1 : if (!account)
1697 0 : return {};
1698 1 : auto* cm = account->convModule();
1699 1 : if (!cm)
1700 0 : return {};
1701 : // The payload is a commit in the document's own swarm, and committing it is
1702 : // also what announces it: peers fetch it straight away rather than showing a
1703 : // placeholder until the next checkpoint.
1704 1 : auto id = cm->addDocumentAttachment(documentId, data);
1705 1 : if (id.empty())
1706 0 : return {};
1707 1 : if (auto session = findSession(conversationId, documentId)) {
1708 : // Ours already: the client that stored it holds the bytes, and the next
1709 : // synchronization must not announce them back to it.
1710 1 : std::lock_guard<std::mutex> lk(mutex_);
1711 1 : session->knownAttachments.insert(id);
1712 2 : }
1713 1 : return id;
1714 1 : }
1715 :
1716 : std::vector<uint8_t>
1717 3 : CollaborativeEditing::attachment(const std::string& /*conversationId*/,
1718 : const std::string& documentId,
1719 : const std::string& attachmentId)
1720 : {
1721 : // Readable without an editing session: a client browsing the history of a
1722 : // document it has not opened still has to resolve what it refers to.
1723 3 : auto conversation = documentConversation(documentId);
1724 6 : return conversation ? conversation->documentAttachment(attachmentId) : std::vector<uint8_t> {};
1725 3 : }
1726 :
1727 : } // namespace jami
|
