jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Manage avatar replication, background downloads and shared cache

master
evgeny 6 days ago
parent
f91251be14
commit
8ae45e9e39
2 changed files with 399 additions and 0 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 335
      tools/chatgui/src/avatars/avatarmanager.cpp
  2. 64
      tools/chatgui/src/avatars/avatarmanager.h

335
tools/chatgui/src/avatars/avatarmanager.cpp
Unescape View File

@ -0,0 +1,335 @@
#include "avatarmanager.h"
#include "../../db/db_manager.h"
#include "../../transport/gui_bridge.h"
#include <QCoreApplication>
#include <QDateTime>
#include <QDir>
#include <QFile>
#include <QMetaObject>
#include <QRandomGenerator>
#include <QRunnable>
#include <QStandardPaths>
#include <cstring>
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<QString> 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<quint64, QString> AvatarManager::servers() const {
QMap<quint64, QString> 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<Job> 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);
}

64
tools/chatgui/src/avatars/avatarmanager.h
Unescape View File

@ -0,0 +1,64 @@
/* Один менеджер на аккаунт: подписанные ссылки, общий кеш и ограниченная очередь передач. */
#pragma once
#include "avatarcache.h"
#include <QObject>
#include <QCache>
#include <QHash>
#include <QQueue>
#include <QSet>
#include <QThreadPool>
#include <QTimer>
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<quint64, QString> 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<QString, Ref> m_refs;
QHash<quint64, QMap<quint64, QString>> m_servers;
QCache<QString, QImage> m_images{32 * 1024};
QSet<QString> m_loading, m_missing, m_wanted;
QHash<QString, qint64> m_retry;
QHash<QString, unsigned> m_failures;
QQueue<Job> m_queue;
QHash<quint64, Job> m_jobs;
QThreadPool m_pool;
QTimer m_timer;
};
Write Preview
Loading…
Cancel
Save