Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "data_transfer.h"
19 :
20 : #include "base64.h"
21 : #include "fileutils.h"
22 : #include "manager.h"
23 : #include "client/jami_signal.h"
24 :
25 : #include <algorithm>
26 : #include <mutex>
27 : #include <cstdlib> // mkstemp
28 : #include <filesystem>
29 : #include <limits>
30 :
31 : #include <opendht/rng.h>
32 : #include <opendht/thread_pool.h>
33 :
34 : namespace jami {
35 :
36 : libjami::DataTransferId
37 71 : generateUID(std::mt19937_64& engine)
38 : {
39 71 : return std::uniform_int_distribution<libjami::DataTransferId> {1, JAMI_ID_MAX_VAL}(engine);
40 : }
41 :
42 : std::string
43 56 : getFileId(const std::string& commitId, const std::string& tid, const std::string& displayName)
44 : {
45 56 : auto extension = fileutils::getFileExtension(displayName);
46 56 : if (extension.empty())
47 106 : return fmt::format("{}_{}", commitId, tid);
48 6 : return fmt::format("{}_{}.{}", commitId, tid, extension);
49 : }
50 :
51 139 : FileInfo::FileInfo(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
52 : const std::string& fileId,
53 : const std::string& interactionId,
54 139 : const libjami::DataTransferInfo& info)
55 139 : : fileId_(fileId)
56 139 : , interactionId_(interactionId)
57 139 : , info_(info)
58 278 : , channel_(channel)
59 139 : {}
60 :
61 : void
62 170 : FileInfo::emit(libjami::DataTransferEventCode code)
63 : {
64 170 : if (finishedCb_ && code >= libjami::DataTransferEventCode::finished)
65 101 : finishedCb_(uint32_t(code));
66 170 : if (interactionId_ != "") {
67 : // Else it's an internal transfer
68 35 : runOnMainThread([info = info_, iid = interactionId_, fid = fileId_, code]() {
69 35 : emitSignal<libjami::DataTransferSignal::DataTransferEvent>(info.accountId,
70 35 : info.conversationId,
71 35 : iid,
72 35 : fid,
73 : uint32_t(code));
74 35 : });
75 : }
76 170 : }
77 :
78 70 : OutgoingFile::OutgoingFile(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
79 : const std::string& fileId,
80 : const std::string& interactionId,
81 : const libjami::DataTransferInfo& info,
82 : size_t start,
83 70 : size_t end)
84 : : FileInfo(channel, fileId, interactionId, info)
85 70 : , start_(start)
86 70 : , end_(end)
87 : {
88 70 : std::filesystem::path fpath(info_.path);
89 70 : if (!std::filesystem::is_regular_file(fpath)) {
90 74 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
91 37 : return;
92 : }
93 33 : stream_.open(fpath, std::ios::binary | std::ios::in);
94 33 : if (!stream_ || !stream_.is_open()) {
95 0 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
96 0 : return;
97 : }
98 70 : }
99 :
100 70 : OutgoingFile::~OutgoingFile()
101 : {
102 70 : if (stream_ && stream_.is_open())
103 0 : stream_.close();
104 70 : if (channel_) {
105 66 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
106 : }
107 70 : }
108 :
109 : void
110 70 : OutgoingFile::process()
111 : {
112 70 : if (!channel_ or !stream_ or !stream_.is_open())
113 37 : return;
114 33 : auto correct = false;
115 33 : stream_.seekg(static_cast<long>(start_), std::ios::beg);
116 : try {
117 33 : std::vector<char> buffer(UINT16_MAX, 0);
118 33 : std::error_code ec;
119 33 : auto pos = start_;
120 275 : while (!stream_.eof()) {
121 246 : stream_.read(buffer.data(),
122 246 : end_ > start_ ? static_cast<long>(std::min(end_ - pos, buffer.size()))
123 246 : : static_cast<long>(buffer.size()));
124 246 : auto gcount = stream_.gcount();
125 246 : pos += gcount;
126 246 : channel_->write(reinterpret_cast<const uint8_t*>(buffer.data()), gcount, ec);
127 246 : if (ec)
128 4 : break;
129 : }
130 33 : if (!ec)
131 29 : correct = true;
132 33 : stream_.close();
133 33 : } catch (const std::exception& e) {
134 0 : JAMI_WARNING("Failed to read from stream: {}", e.what());
135 0 : }
136 33 : if (!isUserCancelled_) {
137 : // NOTE: emit(code) MUST be changed to improve handling of multiple destinations
138 : // But for now, we can just avoid to emit errors to the client, because for outgoing
139 : // transfer in a swarm, for outgoingFiles, we know that the file is ok. And the peer
140 : // will retry the transfer if they need, so we don't need to show errors.
141 33 : if (!interactionId_.empty() && !correct)
142 1 : return;
143 32 : auto code = correct ? libjami::DataTransferEventCode::finished : libjami::DataTransferEventCode::closed_by_peer;
144 32 : emit(code);
145 : }
146 : }
147 :
148 : void
149 0 : OutgoingFile::cancel()
150 : {
151 : // Remove link, not original file
152 0 : auto path = fileutils::get_data_dir() / "conversation_data" / info_.accountId / info_.conversationId / fileId_;
153 0 : if (std::filesystem::is_symlink(path))
154 0 : dhtnet::fileutils::remove(path);
155 0 : isUserCancelled_ = true;
156 0 : emit(libjami::DataTransferEventCode::closed_by_host);
157 0 : }
158 :
159 69 : IncomingFile::IncomingFile(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
160 : const libjami::DataTransferInfo& info,
161 : const std::string& fileId,
162 : const std::string& interactionId,
163 69 : const std::string& sha3Sum)
164 : : FileInfo(channel, fileId, interactionId, info)
165 69 : , sha3Sum_(sha3Sum)
166 138 : , path_(info.path + ".tmp")
167 : {
168 69 : stream_.open(path_, std::ios::binary | std::ios::out | std::ios::app);
169 69 : if (!stream_)
170 0 : return;
171 :
172 69 : emit(libjami::DataTransferEventCode::ongoing);
173 0 : }
174 :
175 69 : IncomingFile::~IncomingFile()
176 : {
177 : {
178 69 : std::lock_guard<std::mutex> lk(streamMtx_);
179 69 : if (stream_ && stream_.is_open())
180 1 : stream_.close();
181 69 : }
182 69 : if (channel_)
183 136 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
184 69 : }
185 :
186 : void
187 1 : IncomingFile::cancel()
188 : {
189 1 : isUserCancelled_ = true;
190 1 : emit(libjami::DataTransferEventCode::closed_by_peer);
191 1 : if (channel_)
192 2 : dht::ThreadPool::io().run([channel = std::move(channel_)] { channel->shutdown(); });
193 1 : }
194 :
195 : void
196 69 : IncomingFile::process()
197 : {
198 69 : channel_->setOnRecv([w = weak_from_this()](const uint8_t* buf, size_t len) {
199 221 : if (auto shared = w.lock()) {
200 221 : std::lock_guard<std::mutex> lk(shared->streamMtx_);
201 221 : if (shared->stream_.is_open())
202 221 : shared->stream_.write(reinterpret_cast<const char*>(buf), static_cast<long>(len));
203 221 : shared->info_.bytesProgress = shared->stream_.tellp();
204 221 : return static_cast<int>(len);
205 442 : }
206 : // Data received after destruction
207 0 : JAMI_ERROR("{} bytes received after IncomingFile destruction.", len);
208 0 : return -1;
209 : });
210 69 : channel_->onShutdown([w = weak_from_this()](const std::error_code& /*error_code*/) {
211 69 : auto shared = w.lock();
212 69 : if (!shared)
213 1 : return;
214 : {
215 68 : std::lock_guard<std::mutex> lk(shared->streamMtx_);
216 68 : if (shared->stream_ && shared->stream_.is_open())
217 68 : shared->stream_.close();
218 68 : }
219 68 : auto correct = shared->sha3Sum_.empty();
220 68 : std::error_code ec;
221 68 : if (!correct) {
222 16 : if (shared->isUserCancelled_) {
223 0 : std::filesystem::remove(shared->path_, ec);
224 16 : } else if (shared->info_.bytesProgress < shared->info_.totalSize) {
225 1 : JAMI_WARNING("Channel for {} shut down before transfer was complete (progress: {}/{})",
226 : shared->info_.path,
227 : shared->info_.bytesProgress,
228 : shared->info_.totalSize);
229 15 : } else if (shared->info_.totalSize != 0 && shared->info_.bytesProgress > shared->info_.totalSize) {
230 0 : JAMI_WARNING("Removing {} larger than announced: {}/{}",
231 : shared->path_,
232 : shared->info_.bytesProgress,
233 : shared->info_.totalSize);
234 0 : std::filesystem::remove(shared->path_, ec);
235 : } else {
236 15 : auto sha3Sum = fileutils::sha3File(shared->path_);
237 15 : if (shared->sha3Sum_ == sha3Sum) {
238 13 : JAMI_LOG("New file received: {}", shared->info_.path);
239 13 : correct = true;
240 : } else {
241 2 : JAMI_WARNING(
242 : "Removing {} with expected size ({} bytes) but invalid sha3sum (expected: {}, actual: {})",
243 : shared->path_,
244 : shared->info_.totalSize,
245 : shared->sha3Sum_,
246 : sha3Sum);
247 2 : std::filesystem::remove(shared->path_, ec);
248 : }
249 15 : }
250 16 : if (ec) {
251 0 : JAMI_ERROR("Failed to remove file {}: {}", shared->path_, ec.message());
252 : }
253 : }
254 68 : if (correct) {
255 65 : std::filesystem::rename(shared->path_, shared->info_.path, ec);
256 65 : if (ec) {
257 0 : JAMI_ERROR("Failed to rename file from {} to {}: {}", shared->path_, shared->info_.path, ec.message());
258 0 : correct = false;
259 : }
260 : }
261 68 : if (shared->isUserCancelled_)
262 0 : return;
263 68 : auto code = correct ? libjami::DataTransferEventCode::finished : libjami::DataTransferEventCode::closed_by_host;
264 68 : shared->emit(code);
265 68 : dht::ThreadPool::io().run([s = std::move(shared)] {});
266 69 : });
267 69 : }
268 :
269 : //==============================================================================
270 :
271 : class TransferManager::Impl
272 : {
273 : public:
274 996 : Impl(const std::string& accountId, const std::string& accountUri, const std::string& to, const std::mt19937_64& rand)
275 996 : : accountId_(accountId)
276 996 : , accountUri_(accountUri)
277 996 : , to_(to)
278 996 : , rand_(rand)
279 : {
280 996 : if (!to_.empty()) {
281 392 : conversationDataPath_ = fileutils::get_data_dir() / accountId_ / "conversation_data" / to_;
282 392 : dhtnet::fileutils::check_dir(conversationDataPath_);
283 392 : waitingPath_ = conversationDataPath_ / "waiting";
284 : }
285 996 : profilesPath_ = fileutils::get_data_dir() / accountId_ / "profiles";
286 996 : accountProfilePath_ = fileutils::get_data_dir() / accountId / "profile.vcf";
287 996 : loadWaiting();
288 996 : }
289 :
290 996 : ~Impl()
291 : {
292 996 : std::lock_guard lk {mapMutex_};
293 1034 : for (auto& [channel, _] : outgoings_) {
294 76 : dht::ThreadPool::io().run([c = std::move(channel)] { c->shutdown(); });
295 : }
296 996 : outgoings_.clear();
297 996 : incomings_.clear();
298 996 : vcards_.clear();
299 996 : }
300 :
301 996 : void loadWaiting()
302 : {
303 : try {
304 : // read file
305 1992 : auto file = fileutils::loadFile(waitingPath_);
306 : // load values
307 0 : msgpack::object_handle oh = msgpack::unpack((const char*) file.data(), file.size());
308 0 : std::lock_guard lk {mapMutex_};
309 0 : oh.get().convert(waitingIds_);
310 996 : } catch (const std::exception& e) {
311 996 : return;
312 996 : }
313 : }
314 21 : void saveWaiting()
315 : {
316 21 : std::ofstream file(waitingPath_, std::ios::trunc | std::ios::binary);
317 21 : msgpack::pack(file, waitingIds_);
318 21 : }
319 :
320 : std::string accountId_ {};
321 : std::string accountUri_ {};
322 : std::string to_ {};
323 : std::filesystem::path waitingPath_ {};
324 : std::filesystem::path profilesPath_ {};
325 : std::filesystem::path accountProfilePath_ {};
326 : std::filesystem::path conversationDataPath_ {};
327 :
328 : std::mutex mapMutex_ {};
329 : std::map<std::string, WaitingRequest> waitingIds_ {};
330 : std::map<std::shared_ptr<dhtnet::ChannelSocket>, std::shared_ptr<OutgoingFile>> outgoings_ {};
331 : std::map<std::string, std::shared_ptr<IncomingFile>> incomings_ {};
332 : std::map<std::pair<std::string, std::string>, std::shared_ptr<IncomingFile>> vcards_ {};
333 :
334 : std::mt19937_64 rand_;
335 : };
336 :
337 996 : TransferManager::TransferManager(const std::string& accountId,
338 : const std::string& accountUri,
339 : const std::string& to,
340 996 : const std::mt19937_64& rand)
341 996 : : pimpl_ {std::make_unique<Impl>(accountId, accountUri, to, rand)}
342 996 : {}
343 :
344 996 : TransferManager::~TransferManager() {}
345 :
346 : void
347 72 : TransferManager::transferFile(const std::shared_ptr<dhtnet::ChannelSocket>& channel,
348 : const std::string& fileId,
349 : const std::string& interactionId,
350 : const std::string& path,
351 : size_t start,
352 : size_t end,
353 : OnFinishedCb onFinished)
354 : {
355 72 : std::lock_guard lk {pimpl_->mapMutex_};
356 72 : if (pimpl_->outgoings_.find(channel) != pimpl_->outgoings_.end())
357 2 : return;
358 70 : libjami::DataTransferInfo info;
359 70 : info.accountId = pimpl_->accountId_;
360 70 : info.conversationId = pimpl_->to_;
361 70 : info.path = path;
362 70 : auto f = std::make_shared<OutgoingFile>(channel, fileId, interactionId, info, start, end);
363 70 : f->onFinished([w = weak(), channel, onFinished = std::move(onFinished)](uint32_t code) {
364 32 : if (code == uint32_t(libjami::DataTransferEventCode::finished) && onFinished) {
365 4 : onFinished();
366 : }
367 : // schedule destroy outgoing transfer as not needed
368 32 : dht::ThreadPool().computation().run([w, channel] {
369 32 : if (auto sthis_ = w.lock()) {
370 32 : auto& pimpl = sthis_->pimpl_;
371 32 : std::lock_guard lk {pimpl->mapMutex_};
372 32 : auto itO = pimpl->outgoings_.find(channel);
373 32 : if (itO != pimpl->outgoings_.end())
374 32 : pimpl->outgoings_.erase(itO);
375 64 : }
376 32 : });
377 32 : });
378 70 : auto [outFile, _] = pimpl_->outgoings_.emplace(channel, std::move(f));
379 70 : dht::ThreadPool::io().run([w = std::weak_ptr<OutgoingFile>(outFile->second)] {
380 70 : if (auto of = w.lock())
381 70 : of->process();
382 70 : });
383 72 : }
384 :
385 : bool
386 1 : TransferManager::cancel(const std::string& fileId)
387 : {
388 1 : std::lock_guard lk {pimpl_->mapMutex_};
389 : // Remove from waiting, this avoid auto-download
390 1 : auto itW = pimpl_->waitingIds_.find(fileId);
391 1 : if (itW != pimpl_->waitingIds_.end()) {
392 1 : pimpl_->waitingIds_.erase(itW);
393 1 : JAMI_LOG("Cancel {}", fileId);
394 1 : pimpl_->saveWaiting();
395 : }
396 1 : auto itC = pimpl_->incomings_.find(fileId);
397 1 : if (itC == pimpl_->incomings_.end())
398 0 : return false;
399 1 : itC->second->cancel();
400 1 : return true;
401 1 : }
402 :
403 : bool
404 3 : TransferManager::info(const std::string& fileId, std::string& path, int64_t& total, int64_t& progress) const noexcept
405 : {
406 3 : std::unique_lock lk {pimpl_->mapMutex_};
407 3 : if (pimpl_->to_.empty())
408 0 : return false;
409 :
410 3 : auto itI = pimpl_->incomings_.find(fileId);
411 3 : auto itW = pimpl_->waitingIds_.find(fileId);
412 3 : std::filesystem::path transferPath;
413 : try {
414 3 : transferPath = this->path(fileId);
415 3 : path = transferPath.string();
416 0 : } catch (const std::filesystem::filesystem_error& e) {
417 0 : JAMI_WARNING("Unable to resolve transfer path for {}: {}", fileId, e.what());
418 0 : progress = 0;
419 0 : return false;
420 0 : }
421 3 : if (itI != pimpl_->incomings_.end()) {
422 0 : total = itI->second->info().totalSize;
423 0 : progress = itI->second->info().bytesProgress;
424 0 : return true;
425 : }
426 :
427 3 : std::error_code ec;
428 3 : if (std::filesystem::is_regular_file(transferPath, ec)) {
429 1 : auto fileSize = std::filesystem::file_size(transferPath, ec);
430 1 : if (ec) {
431 0 : JAMI_WARNING("Unable to read transfer file size for {}: {}", path, ec.message());
432 0 : progress = 0;
433 0 : return false;
434 : }
435 1 : progress = static_cast<int64_t>(
436 1 : std::min<uintmax_t>(fileSize, static_cast<uintmax_t>(std::numeric_limits<int64_t>::max())));
437 1 : if (itW != pimpl_->waitingIds_.end()) {
438 0 : total = static_cast<int64_t>(itW->second.totalSize);
439 : } else {
440 : // If not waiting it's finished
441 1 : total = progress;
442 : }
443 1 : return true;
444 : }
445 2 : if (ec) {
446 2 : JAMI_WARNING("Unable to inspect transfer path {}: {}", path, ec.message());
447 : }
448 2 : if (itW != pimpl_->waitingIds_.end()) {
449 0 : total = static_cast<int64_t>(itW->second.totalSize);
450 0 : progress = 0;
451 0 : return true;
452 : }
453 : // Else we don't know infos there.
454 2 : progress = 0;
455 2 : return false;
456 3 : }
457 :
458 : void
459 13 : TransferManager::waitForTransfer(const std::string& fileId,
460 : const std::string& interactionId,
461 : const std::string& sha3sum,
462 : const std::string& path,
463 : std::size_t total)
464 : {
465 13 : std::unique_lock lk(pimpl_->mapMutex_);
466 13 : auto itW = pimpl_->waitingIds_.find(fileId);
467 13 : if (itW != pimpl_->waitingIds_.end())
468 1 : return;
469 12 : pimpl_->waitingIds_[fileId] = {fileId, interactionId, sha3sum, path, total};
470 12 : pimpl_->saveWaiting();
471 25 : }
472 :
473 : void
474 12 : TransferManager::onIncomingFileTransfer(const std::string& fileId,
475 : const std::shared_ptr<dhtnet::ChannelSocket>& channel,
476 : size_t start)
477 : {
478 12 : std::lock_guard lk(pimpl_->mapMutex_);
479 : // Check if not already an incoming file for this id and that we are waiting this file
480 12 : auto itC = pimpl_->incomings_.find(fileId);
481 12 : if (itC != pimpl_->incomings_.end()) {
482 0 : dht::ThreadPool().io().run([channel] { channel->shutdown(); });
483 0 : return;
484 : }
485 12 : auto itW = pimpl_->waitingIds_.find(fileId);
486 12 : if (itW == pimpl_->waitingIds_.end()) {
487 0 : dht::ThreadPool().io().run([channel] { channel->shutdown(); });
488 0 : return;
489 : }
490 :
491 12 : libjami::DataTransferInfo info;
492 12 : info.accountId = pimpl_->accountId_;
493 12 : info.conversationId = pimpl_->to_;
494 12 : info.path = itW->second.path;
495 12 : info.totalSize = static_cast<int64_t>(itW->second.totalSize);
496 12 : info.bytesProgress = static_cast<int64_t>(start);
497 :
498 : // Generate the file path within the conversation data directory
499 : // using the file id if no path has been specified, otherwise create
500 : // a symlink(Note: this will not work on Windows).
501 12 : auto filePath = path(fileId);
502 12 : if (info.path.empty()) {
503 0 : info.path = filePath.string();
504 : } else {
505 : // We don't need to check if this is an existing symlink here, as
506 : // the attempt to create one should report the error string correctly.
507 12 : fileutils::createFileLink(filePath, info.path);
508 : }
509 :
510 12 : auto ifile = std::make_shared<IncomingFile>(std::move(channel),
511 : info,
512 : fileId,
513 12 : itW->second.interactionId,
514 24 : itW->second.sha3sum);
515 12 : auto res = pimpl_->incomings_.emplace(fileId, std::move(ifile));
516 12 : if (res.second) {
517 12 : res.first->second->onFinished([w = weak(), fileId](uint32_t code) {
518 : // schedule destroy transfer as not needed
519 12 : dht::ThreadPool().computation().run([w, fileId, code] {
520 12 : if (auto sthis_ = w.lock()) {
521 12 : auto& pimpl = sthis_->pimpl_;
522 12 : std::lock_guard lk {pimpl->mapMutex_};
523 12 : auto itO = pimpl->incomings_.find(fileId);
524 12 : if (itO != pimpl->incomings_.end())
525 12 : pimpl->incomings_.erase(itO);
526 12 : if (code == uint32_t(libjami::DataTransferEventCode::finished)) {
527 8 : auto itW = pimpl->waitingIds_.find(fileId);
528 8 : if (itW != pimpl->waitingIds_.end()) {
529 8 : pimpl->waitingIds_.erase(itW);
530 8 : pimpl->saveWaiting();
531 : }
532 : }
533 24 : }
534 12 : });
535 12 : });
536 12 : res.first->second->process();
537 : }
538 12 : }
539 :
540 : std::filesystem::path
541 40 : TransferManager::path(const std::string& fileId) const
542 : {
543 40 : return pimpl_->conversationDataPath_ / fileId;
544 : }
545 :
546 : void
547 61 : TransferManager::onIncomingProfile(const std::shared_ptr<dhtnet::ChannelSocket>& channel, const std::string& sha3Sum)
548 : {
549 61 : if (!channel)
550 0 : return;
551 :
552 61 : auto chName = channel->name();
553 61 : std::string_view name = chName;
554 61 : auto sep = name.find_last_of('?');
555 61 : if (sep != std::string::npos)
556 9 : name = name.substr(0, sep);
557 :
558 61 : auto lastSep = name.find_last_of('/');
559 61 : auto fileId = name.substr(lastSep + 1);
560 :
561 61 : auto deviceId = channel->deviceId().toString();
562 61 : auto cert = channel->peerCertificate();
563 61 : if (!cert || !cert->issuer || fileId.find(".vcf") == std::string::npos)
564 0 : return;
565 :
566 70 : auto uri = fileId == "profile.vcf" ? cert->issuer->getId().toString()
567 122 : : std::string(fileId.substr(0, fileId.size() - 4 /*.vcf*/));
568 :
569 61 : std::lock_guard lk(pimpl_->mapMutex_);
570 61 : auto idx = std::make_pair(deviceId, uri);
571 : // Check if not already an incoming file for this id and that we are waiting this file
572 61 : auto itV = pimpl_->vcards_.find(idx);
573 61 : if (itV != pimpl_->vcards_.end()) {
574 8 : dht::ThreadPool().io().run([channel] { channel->shutdown(); });
575 4 : return;
576 : }
577 :
578 57 : auto tid = generateUID(pimpl_->rand_);
579 57 : libjami::DataTransferInfo info;
580 57 : info.accountId = pimpl_->accountId_;
581 57 : info.conversationId = pimpl_->to_;
582 :
583 57 : auto recvDir = fileutils::get_cache_dir() / pimpl_->accountId_ / "vcard";
584 57 : dhtnet::fileutils::recursive_mkdir(recvDir);
585 114 : info.path = (recvDir / fmt::format("{:s}_{:s}_{}", deviceId, uri, tid)).string();
586 :
587 57 : auto ifile = std::make_shared<IncomingFile>(std::move(channel), info, "profile.vcf", "", sha3Sum);
588 57 : auto res = pimpl_->vcards_.emplace(idx, std::move(ifile));
589 57 : if (res.second) {
590 228 : res.first->second->onFinished([w = weak(),
591 57 : uri = std::move(uri),
592 57 : deviceId = std::move(deviceId),
593 57 : accountId = pimpl_->accountId_,
594 57 : cert = std::move(cert),
595 : path = info.path](uint32_t code) {
596 285 : dht::ThreadPool().computation().run([w,
597 57 : uri = std::move(uri),
598 57 : deviceId = std::move(deviceId),
599 57 : accountId = std::move(accountId),
600 57 : path = std::move(path),
601 : code] {
602 57 : if (auto sthis_ = w.lock()) {
603 57 : auto& pimpl = sthis_->pimpl_;
604 :
605 57 : auto destPath = sthis_->profilePath(uri);
606 : try {
607 : // Move profile to destination path
608 57 : std::lock_guard lock(dhtnet::fileutils::getFileLock(destPath));
609 57 : dhtnet::fileutils::recursive_mkdir(destPath.parent_path());
610 57 : std::filesystem::rename(path, destPath);
611 57 : if (!pimpl->accountUri_.empty() && uri == pimpl->accountUri_) {
612 : // If this is the account profile, link or copy it to the account profile path
613 1 : if (!fileutils::createFileLink(pimpl->accountProfilePath_, destPath)) {
614 0 : std::error_code ec;
615 0 : std::filesystem::copy_file(destPath, pimpl->accountProfilePath_, ec);
616 : }
617 : }
618 57 : } catch (const std::exception& e) {
619 0 : JAMI_ERROR("{}", e.what());
620 0 : }
621 :
622 57 : std::lock_guard lk {pimpl->mapMutex_};
623 57 : auto itO = pimpl->vcards_.find({deviceId, uri});
624 57 : if (itO != pimpl->vcards_.end())
625 57 : pimpl->vcards_.erase(itO);
626 57 : if (code == uint32_t(libjami::DataTransferEventCode::finished)) {
627 57 : emitSignal<libjami::ConfigurationSignal::ProfileReceived>(accountId, uri, destPath.string());
628 : }
629 114 : }
630 57 : });
631 57 : });
632 57 : res.first->second->process();
633 : }
634 81 : }
635 :
636 : std::filesystem::path
637 69 : TransferManager::profilePath(const std::string& contactId) const
638 : {
639 138 : return pimpl_->profilesPath_ / fmt::format("{}.vcf", base64::encode(contactId));
640 : }
641 :
642 : std::vector<WaitingRequest>
643 1635 : TransferManager::waitingRequests() const
644 : {
645 1635 : std::vector<WaitingRequest> res;
646 1635 : std::lock_guard lk(pimpl_->mapMutex_);
647 1636 : for (const auto& [fileId, req] : pimpl_->waitingIds_) {
648 1 : auto itC = pimpl_->incomings_.find(fileId);
649 1 : if (itC == pimpl_->incomings_.end())
650 1 : res.emplace_back(req);
651 : }
652 3270 : return res;
653 1635 : }
654 :
655 : bool
656 4 : TransferManager::isWaiting(const std::string& fileId) const
657 : {
658 4 : std::lock_guard lk(pimpl_->mapMutex_);
659 8 : return pimpl_->waitingIds_.find(fileId) != pimpl_->waitingIds_.end();
660 4 : }
661 :
662 : } // namespace jami
|
