From 8ae45e9e395eedc2aa07958c9ba6c2bfb358d3cc Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 6 Oct 2026 12:31:16 +0200 Subject: [PATCH] Manage avatar replication, background downloads and shared cache --- tools/chatgui/src/avatars/avatarmanager.cpp | 335 ++++++++++++++++++++ tools/chatgui/src/avatars/avatarmanager.h | 64 ++++ 2 files changed, 399 insertions(+) create mode 100644 tools/chatgui/src/avatars/avatarmanager.cpp create mode 100644 tools/chatgui/src/avatars/avatarmanager.h diff --git a/tools/chatgui/src/avatars/avatarmanager.cpp b/tools/chatgui/src/avatars/avatarmanager.cpp new file mode 100644 index 00000000..28f13374 --- /dev/null +++ b/tools/chatgui/src/avatars/avatarmanager.cpp @@ -0,0 +1,335 @@ +#include "avatarmanager.h" +#include "../../db/db_manager.h" +#include "../../transport/gui_bridge.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +extern "C" { +#include "chat/chat_core.h" +#include "chat/chat_member.h" +#include "member_files/member_files.h" +#include "mem.h" +#include "debug_config.h" +} + +static AvatarManager* s_manager = nullptr; +static QString text(const char* s) { return QCoreApplication::translate("ChatGui", s); } +QString AvatarManager::key(quint64 group, quint64 owner) { return QString::number(group) + ':' + QString::number(owner); } +AvatarManager* AvatarManager::instance() { return s_manager; } + +AvatarManager::AvatarManager(DbManager* db, QObject* parent) : QObject(parent), m_db(db) { + s_manager = this; m_pool.setMaxThreadCount(1); + m_timer.setInterval(2000); + connect(&m_timer, &QTimer::timeout, this, [this] { + if (!m_ready) return; + for (auto group : m_servers.keys()) scheduleOwn(group); + for (const auto& context : m_wanted) if (m_refs.contains(context)) image(QString::number(m_refs[context].group), m_refs[context].owner); + pump(); + }); + m_timer.start(); +} + +AvatarManager::~AvatarManager() { + m_timer.stop(); s_manager = nullptr; m_pool.waitForDone(); + for (auto id : m_jobs.keys()) { + if (!gui_bridge_is_uasync_ready()) break; + auto* q = (mf_cancel_request*)u_calloc(1, sizeof(mf_cancel_request)); + if (q) { q->inst = gui_bridge_get_inst(); q->id = id; gui_bridge_post_uasync_fn(member_files_gui_cancel, q); } + } +} + +void AvatarManager::initialize() { + if (!m_db || !m_db->isOpen() || m_ready || !m_directory.isEmpty()) return; + const QString base = m_db->dbPath().isEmpty() ? QStandardPaths::writableLocation(QStandardPaths::AppLocalDataLocation) : m_db->dbPath(); + m_directory = base + QStringLiteral("/avatar_cache/") + QString::number(m_db->myNodeId(), 16); + m_defaultServer = m_db->getUiState("avatar_default_server").toULongLong(nullptr, 16); + const QString file = m_db->getUiState("avatar_default_file"); + if (mf_avatar_file_valid(file.toLatin1().constData())) m_defaultHash = file.left(64); + const QString directory = m_directory; + m_pool.start(QRunnable::create([this, directory] { + bool ready = AvatarCache::initialize(directory); + QMetaObject::invokeMethod(this, [this, ready] { + m_ready = ready; + if (!ready) { emit statusChanged(text("Cannot open avatar cache")); return; } + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "avatar: cache ready path=%s", qPrintable(m_directory)); + if (!m_defaultHash.isEmpty()) loadCache(m_defaultHash); + refreshMembers(); pump(); + }, Qt::QueuedConnection); + })); +} + +void AvatarManager::refreshMembers() { + if (!m_db || !m_db->isOpen() || !gui_bridge_is_uasync_ready()) return; + for (const auto& channel : m_db->getChannels()) { + auto* q = (chat_member_list_req*)u_calloc(1, sizeof(chat_member_list_req)); + if (!q) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "avatar: member request allocation failed"); return; } + q->inst = gui_bridge_get_inst(); const QByteArray id = channel.channelId.toLatin1(); + if (id.size() >= int(sizeof(q->ch_id))) { u_free(q); continue; } + memcpy(q->ch_id, id.constData(), id.size()); gui_bridge_post_uasync_fn(chat_core_request_member_list_trampoline, q); + } +} + +void AvatarManager::onMembers(const QString& channel, int count, const uint8_t* data, int length, bool full) { + const quint64 group = channel.toULongLong(); + if (!group || count < 0 || count > length / CHAT_MEMBER_DISPLAY_SIZE) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "avatar: malformed member snapshot channel=%s count=%d bytes=%d", qPrintable(channel), count, length); + return; + } + QSet seen; + if (full) m_servers[group].clear(); + for (int i = 0; i < count; i++) { + chat_member_display member{}; chat_core_deserialize_member(data + i * CHAT_MEMBER_DISPLAY_SIZE, &member); + const QString context = key(group, member.node_id); seen.insert(context); + if (member.flags & CHAT_MEMBER_FLAG_DELETED) { removeMember(channel, member.node_id); continue; } + if (member.storage) m_servers[group][member.node_id] = QString::fromUtf8(member.name); + else m_servers[group].remove(member.node_id); + Ref ref{group, member.node_id, member.avatar_server, QString::fromLatin1(member.avatar_file).left(64)}; + if (!ref.server || !mf_avatar_file_valid(member.avatar_file)) { ref.server = 0; ref.hash.clear(); } + const Ref old = m_refs.value(context); m_refs[context] = ref; + if (old.hash != ref.hash || old.server != ref.server) { + m_retry.remove(context); m_failures.remove(context); emit changed(channel, ref.owner); + } + } + if (full) for (const auto& context : m_refs.keys()) + if (m_refs[context].group == group && !seen.contains(context)) removeMember(channel, m_refs[context].owner); + emit serversChanged(); scheduleOwn(group); pump(); +} + +void AvatarManager::removeMember(const QString& channel, quint64 owner) { + const quint64 group = channel.toULongLong(); const QString context = key(group, owner); + m_refs.remove(context); m_wanted.remove(context); m_servers[group].remove(owner); + emit changed(channel, owner); +} + +QMap AvatarManager::servers() const { + QMap result; + for (const auto& group : m_servers) for (auto it = group.begin(); it != group.end(); ++it) result[it.key()] = it.value(); + return result; +} + +QImage AvatarManager::image(const QString& channel, quint64 owner) { + const QString context = key(channel.toULongLong(), owner); const Ref ref = m_refs.value(context); + if (!m_ready || !ref.server || ref.hash.isEmpty()) return {}; + m_wanted.insert(context); + if (auto* image = m_images.object(ref.hash)) return *image; + if (!m_missing.contains(ref.hash)) { loadCache(ref.hash); return {}; } + if (m_retry.value(context) > QDateTime::currentMSecsSinceEpoch()) return {}; + Job job; job.ref = ref; enqueue(job); pump(); return {}; +} + +QImage AvatarManager::defaultImage() const { + auto* image = m_images.object(m_defaultHash); return image ? *image : QImage(); +} + +void AvatarManager::notify(const QString& hash) { + for (const auto& ref : m_refs) if (ref.hash == hash) emit changed(QString::number(ref.group), ref.owner); + emit serversChanged(); +} + +void AvatarManager::loadCache(const QString& hash) { + if (m_loading.contains(hash)) return; + m_loading.insert(hash); const QString path = m_directory + '/' + hash + QStringLiteral(".png"); + m_pool.start(QRunnable::create([this, hash, path] { + AvatarCache::Result result = AvatarCache::load(path, hash); + if (result.image.isNull() && QFile::exists(path)) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "avatar: invalid cached image hash=%s reason=%s", qPrintable(hash), qPrintable(result.error)); + if (!QFile::remove(path)) DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "avatar: cannot remove invalid cache hash=%s", qPrintable(hash)); + } + QMetaObject::invokeMethod(this, [this, hash, result] { + m_loading.remove(hash); + if (result.image.isNull()) m_missing.insert(hash); + else { + DEBUG_DEBUG(DEBUG_CATEGORY_MEDIA, "avatar: disk cache hit hash=%s", qPrintable(hash)); + m_images.insert(hash, new QImage(result.image), qMax(1, int(result.image.sizeInBytes() / 1024))); m_missing.remove(hash); + } + notify(hash); + }, Qt::QueuedConnection); + })); +} + +void AvatarManager::saveDefault() { + const char* keys[] = {"avatar_default_server", "avatar_default_file"}; + const QString values[] = {QString::number(m_defaultServer, 16), m_defaultHash.isEmpty() ? QString() : m_defaultHash + ".png"}; + for (int i = 0; i < 2; i++) { + auto* q = (save_ui_state_arg*)u_calloc(1, sizeof(save_ui_state_arg)); + if (!q) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "avatar: preference allocation failed"); return; } + q->inst = gui_bridge_get_inst(); const QByteArray value = values[i].toLatin1(); + memcpy(q->data, keys[i], strlen(keys[i]) + 1); memcpy(q->data + strlen(keys[i]) + 1, value.constData(), value.size()); + gui_bridge_post_uasync_fn(chat_core_save_ui_state_trampoline, q); + } +} + +void AvatarManager::uploadDefault(const QString& source, quint64 server) { + if (!m_ready || m_preparing || !server || !gui_bridge_is_uasync_ready()) return; + m_preparing = true; emit statusChanged(text("Preparing avatar…")); + const QString directory = m_directory; const quint64 generation = ++m_generation; cancelUploads(); + m_pool.start(QRunnable::create([this, source, directory, server, generation] { + const auto result = AvatarCache::prepare(source, directory); + QMetaObject::invokeMethod(this, [this, result, server, generation] { + m_preparing = false; + if (generation != m_generation) return; + if (result.image.isNull()) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "avatar: preparation failed reason=%s", qPrintable(result.error)); + emit statusChanged(text("Cannot prepare avatar: %1").arg(result.error)); return; + } + m_defaultHash = result.hash; m_defaultServer = server; saveDefault(); + m_images.insert(result.hash, new QImage(result.image), qMax(1, int(result.image.sizeInBytes() / 1024))); + m_missing.remove(result.hash); emit serversChanged(); emit statusChanged(text("Uploading avatar…")); + for (auto group : m_servers.keys()) { m_retry.remove(key(group, m_db->myNodeId())); scheduleOwn(group); } + pump(); + }, Qt::QueuedConnection); + })); +} + +void AvatarManager::cancelUploads() { + for (const auto& job : m_jobs) if (job.upload && job.stage != 100 && gui_bridge_is_uasync_ready()) { + auto* q = (mf_cancel_request*)u_calloc(1, sizeof(mf_cancel_request)); + if (q) { q->inst = gui_bridge_get_inst(); q->id = job.id; gui_bridge_post_uasync_fn(member_files_gui_cancel, q); } + } + QQueue keep; while (!m_queue.isEmpty()) { auto job = m_queue.dequeue(); if (!job.upload) keep.enqueue(job); } m_queue = keep; +} + +void AvatarManager::clearDefault() { + if (!m_ready || !gui_bridge_is_uasync_ready()) return; + ++m_generation; cancelUploads(); m_defaultHash.clear(); m_defaultServer = 0; saveDefault(); + for (const auto& ref : m_refs) if (ref.owner == m_db->myNodeId() && ref.server) { + Job job; job.ref = ref; job.ref.server = 0; job.ref.hash.clear(); job.upload = true; job.generation = m_generation; + enqueue(job); + } + pump(); emit serversChanged(); +} + +void AvatarManager::scheduleOwn(quint64 group) { + if (!m_ready || m_preparing || !m_defaultServer || m_defaultHash.isEmpty() || !m_servers.value(group).contains(m_defaultServer) || + !m_images.contains(m_defaultHash)) return; + const QString context = key(group, m_db->myNodeId()); + if (!m_refs.contains(context) || m_retry.value(context) > QDateTime::currentMSecsSinceEpoch()) return; + const Ref old = m_refs.value(context); + if (old.server == m_defaultServer && old.hash == m_defaultHash) return; + Job job; job.ref = {group, m_db->myNodeId(), m_defaultServer, m_defaultHash}; job.upload = true; job.generation = m_generation; + enqueue(job); +} + +void AvatarManager::enqueue(const Job& job) { + for (const auto& old : m_jobs) if ((old.ref.group == job.ref.group && old.ref.owner == job.ref.owner) || + (!job.upload && !old.upload && old.ref.hash == job.ref.hash)) return; + for (const auto& old : m_queue) if ((old.ref.group == job.ref.group && old.ref.owner == job.ref.owner) || + (!job.upload && !old.upload && old.ref.hash == job.ref.hash)) return; + m_queue.enqueue(job); +} + +void AvatarManager::pump() { + if (!m_ready || !gui_bridge_is_uasync_ready()) return; + while (m_jobs.size() < 2 && !m_queue.isEmpty()) { + Job job = m_queue.dequeue(); const QString context = key(job.ref.group, job.ref.owner); + if (!m_refs.contains(context) || (job.upload && job.generation != m_generation) || + (!job.upload && (m_refs[context].hash != job.ref.hash || m_refs[context].server != job.ref.server))) continue; + do { job.id = QRandomGenerator::global()->generate64(); } while (!job.id || m_jobs.contains(job.id)); + job.local = m_directory + '/' + job.ref.hash + (job.upload ? QStringLiteral(".png") : QString(".download-%1").arg(job.id)); + m_jobs.insert(job.id, job); auto& stored = m_jobs[job.id]; + if (job.upload) { if (job.ref.server) post(stored, MF_MKDIR); else publish(stored); } + else transfer(stored); + } +} + +bool AvatarManager::post(Job& job, int command) { + auto* q = (mf_gui_request*)u_calloc(1, sizeof(mf_gui_request)); + if (!q) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "avatar: request allocation failed"); finish(job.id, MF_IO_ERROR); return false; } + q->inst = gui_bridge_get_inst(); q->group = job.ref.group; q->server = job.ref.server; + q->message.kind = MF_REQUEST; q->message.command = command; q->message.owner = job.ref.owner; + // Каждый новый шаг имеет новый ID: сервер хранит квитанции мутаций по request_id. + quint64 id; do { id = QRandomGenerator::global()->generate64(); } while (!id || m_jobs.contains(id)); + Job next = job; m_jobs.remove(job.id); next.id = id; next.stage = command; m_jobs.insert(id, next); + q->message.request_id = id; + const QByteArray path = command == MF_MKDIR ? QByteArray("avatars") : QByteArray("avatars/") + next.ref.hash.toLatin1() + ".png"; + memcpy(q->message.path, path.constData(), path.size()); + gui_bridge_post_uasync_fn(member_files_gui_request, q); return true; +} + +void AvatarManager::transfer(Job& job) { + auto* q = (mf_transfer_request*)u_calloc(1, sizeof(mf_transfer_request)); + if (!q) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "avatar: transfer allocation failed"); finish(job.id, MF_IO_ERROR); return; } + q->inst = gui_bridge_get_inst(); q->id = job.id; q->group = job.ref.group; q->server = job.ref.server; + q->owner = job.ref.owner; q->upload = job.upload; q->max_bytes = AvatarCache::MaxBytes; + const QByteArray local = job.local.toUtf8(), remote = "avatars/" + job.ref.hash.toLatin1() + ".png"; + if (local.size() > MF_PATH_MAX) { u_free(q); finish(job.id, MF_INVALID_PATH); return; } + memcpy(q->local, local.constData(), local.size()); memcpy(q->remote, remote.constData(), remote.size()); job.stage = 99; + DEBUG_INFO(DEBUG_CATEGORY_MEDIA, "avatar: %s group=%llu owner=%016llx server=%016llx hash=%s", + job.upload ? "upload" : "download", (unsigned long long)job.ref.group, (unsigned long long)job.ref.owner, + (unsigned long long)job.ref.server, qPrintable(job.ref.hash)); + gui_bridge_post_uasync_fn(member_files_gui_transfer, q); +} + +void AvatarManager::publish(Job& job) { + if (job.generation != m_generation) { finish(job.id, MF_CANCELLED); return; } + auto* q = (chat_avatar_req*)u_calloc(1, sizeof(chat_avatar_req)); + if (!q) { DEBUG_ERROR(DEBUG_CATEGORY_MEDIA, "avatar: publication allocation failed"); finish(job.id, MF_IO_ERROR); return; } + q->inst = gui_bridge_get_inst(); q->id = job.id; q->group = job.ref.group; q->server = job.ref.server; + if (job.ref.server) { const QByteArray file = job.ref.hash.toLatin1() + ".png"; memcpy(q->file, file.constData(), file.size()); } + job.stage = 100; gui_bridge_post_uasync_fn(chat_core_set_my_avatar_trampoline, q); +} + +void AvatarManager::onResponse(const uint8_t* data, int length) { + if (length < 16 + MF_WIRE_HEADER) return; + mf_message response{}; if (mf_decode(data + 16, length - 16, &response) < 0 || !m_jobs.contains(response.request_id)) return; + auto& job = m_jobs[response.request_id]; + if (job.ref.group != mf_get64(data) || job.ref.server != mf_get64(data + 8) || response.command != job.stage) return; + if (job.generation != m_generation) { finish(job.id, MF_CANCELLED); return; } + if (job.stage == MF_MKDIR && (response.status == MF_OK || response.status == MF_EXISTS)) { post(job, MF_STAT); return; } + if (job.stage == MF_STAT && response.status == MF_NOT_FOUND) { transfer(job); return; } + if (job.stage == MF_STAT && response.status == MF_OK && response.type == MF_FILE && response.size <= AvatarCache::MaxBytes) { + publish(job); return; + } + finish(job.id, response.status ? response.status : MF_INVALID_RANGE); +} + +void AvatarManager::onTransfer(const uint8_t* data, int length) { + if (length != 51 || data[50] != 2) return; + const quint64 id = mf_get64(data); if (!m_jobs.contains(id)) return; + auto& job = m_jobs[id]; const int status = data[48] | (int(data[49]) << 8); + if (status) { finish(id, status); return; } + if (job.upload) { publish(job); return; } + const Job snapshot = job; const QString directory = m_directory; job.stage = 101; + QStringList pinned; if (!m_defaultHash.isEmpty()) pinned << m_defaultHash; + for (const auto& ref : m_refs) if (m_wanted.contains(key(ref.group, ref.owner))) pinned << ref.hash; + m_pool.start(QRunnable::create([this, snapshot, directory, pinned] { + const auto result = AvatarCache::publish(snapshot.local, directory, snapshot.ref.hash); AvatarCache::prune(directory, pinned); + QMetaObject::invokeMethod(this, [this, snapshot, result] { + if (!m_jobs.contains(snapshot.id)) return; + if (result.image.isNull()) { + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "avatar: validation failed hash=%s reason=%s", qPrintable(snapshot.ref.hash), qPrintable(result.error)); + finish(snapshot.id, MF_HASH_MISMATCH); return; + } + m_images.insert(result.hash, new QImage(result.image), qMax(1, int(result.image.sizeInBytes() / 1024))); + m_missing.remove(result.hash); finish(snapshot.id, MF_OK); notify(result.hash); + }, Qt::QueuedConnection); + })); +} + +void AvatarManager::onPublished(const uint8_t* data, int length) { + if (length != 17 || !m_jobs.contains(mf_get64(data))) return; + auto& job = m_jobs[mf_get64(data)]; if (job.stage != 100 || job.ref.group != mf_get64(data + 8)) return; + finish(job.id, data[16] ? MF_IO_ERROR : MF_OK); +} + +void AvatarManager::finish(quint64 id, int status) { + const Job job = m_jobs.take(id); const QString context = key(job.ref.group, job.ref.owner); + if (status && status != MF_CANCELLED) { + unsigned attempt = qMin(6u, m_failures.value(context) + 1); m_failures[context] = attempt; + m_retry[context] = QDateTime::currentMSecsSinceEpoch() + (5000LL << (attempt - 1)); + DEBUG_WARN(DEBUG_CATEGORY_MEDIA, "avatar: operation failed group=%llu owner=%016llx hash=%s status=%s retry_ms=%lld", + (unsigned long long)job.ref.group, (unsigned long long)job.ref.owner, qPrintable(job.ref.hash), mf_status_text(status), + (long long)(5000LL << (attempt - 1))); + if (job.upload && job.generation == m_generation) emit statusChanged(text("Avatar upload failed: %1").arg(mf_status_text(status))); + } else if (!status) { + m_retry.remove(context); m_failures.remove(context); + if (job.upload && job.generation == m_generation) emit statusChanged(text("Avatar updated")); + } + QTimer::singleShot(0, this, &AvatarManager::pump); +} diff --git a/tools/chatgui/src/avatars/avatarmanager.h b/tools/chatgui/src/avatars/avatarmanager.h new file mode 100644 index 00000000..28c9a116 --- /dev/null +++ b/tools/chatgui/src/avatars/avatarmanager.h @@ -0,0 +1,64 @@ +/* Один менеджер на аккаунт: подписанные ссылки, общий кеш и ограниченная очередь передач. */ +#pragma once +#include "avatarcache.h" +#include +#include +#include +#include +#include +#include +#include + +class DbManager; +class AvatarManager : public QObject { + Q_OBJECT +public: + explicit AvatarManager(DbManager* db, QObject* parent = nullptr); + ~AvatarManager() override; + static AvatarManager* instance(); + void initialize(); + void refreshMembers(); + void onMembers(const QString& channel, int count, const uint8_t* data, int length, bool full = true); + void removeMember(const QString& channel, quint64 owner); + void onResponse(const uint8_t* data, int length); + void onTransfer(const uint8_t* data, int length); + void onPublished(const uint8_t* data, int length); + QImage image(const QString& channel, quint64 owner); + QMap servers() const; + void uploadDefault(const QString& source, quint64 server); + void clearDefault(); + QImage defaultImage() const; +signals: + void changed(const QString& channel, quint64 owner); + void serversChanged(); + void statusChanged(const QString& status); +private: + struct Ref { quint64 group = 0, owner = 0, server = 0; QString hash; }; + struct Job { Ref ref; quint64 id = 0, generation = 0; bool upload = false; int stage = 0; QString local; }; + static QString key(quint64 group, quint64 owner); + void loadCache(const QString& hash); + void notify(const QString& hash); + void scheduleOwn(quint64 group); + void enqueue(const Job& job); + void pump(); + bool post(Job& job, int command); + void transfer(Job& job); + void publish(Job& job); + void finish(quint64 id, int status); + void saveDefault(); + void cancelUploads(); + DbManager* m_db; + QString m_directory, m_defaultHash; + quint64 m_defaultServer = 0, m_generation = 1; + bool m_ready = false, m_preparing = false; + QHash m_refs; + QHash> m_servers; + QCache m_images{32 * 1024}; + QSet m_loading, m_missing, m_wanted; + QHash m_retry; + QHash m_failures; + QQueue m_queue; + QHash m_jobs; + QThreadPool m_pool; + QTimer m_timer; +};