#include "EventHandler.h" #include #include #include #include #include #include #include #include #include #include "dao/AccountDAO.h" #include "dao/ApplicationDAO.h" #include "dao/BankDAO.h" #include "dao/DeviceDAO.h" #include "dao/EventDAO.h" #include "dao/TransactionDAO.h" #include "db/ApplicationInfo.h" #include "adb/AdbUtils.h" #include "net/NetworkService.h" #include "black/GetLastTransactionsScript.h" #include "black/LoginAndCheckAccountsScript.h" #include "ozon/GetLastTransactionsScript.h" #include "ozon/LoginAndCheckAccountsScript.h" #include "black/PayByCardScript.h" #include "ozon/PayByCardScript.h" #include "ozon/PayByPhoneScript.h" static void sendTaskResult(const QString &type, const QString &materialId, const QString &bankName, const QJsonValue &data); static void sendTaskError(const QString &type, const QString &materialId, const QString &bankName, const QString &description); EventHandler::EventHandler(QObject *parent) : QObject(parent) {} EventHandler::~EventHandler() = default; void EventHandler::start() { qDebug() << "[EventHandler] start() called, thread:" << QThread::currentThread(); m_network = new NetworkService(this); connect(m_network, &NetworkService::tasksReceived, this, &EventHandler::onTasksReceived); m_pollTimer = new QTimer(this); m_pollTimer->setInterval(5000); connect(m_pollTimer, &QTimer::timeout, this, [this]() { // Пробуем запустить задачи из очереди для свободных устройств const QStringList onlineDevices = DeviceDAO::getOnlineDeviceCodes(); for (const QString &deviceId : onlineDevices) { processNextTask(deviceId); } // Запрашиваем новые задачи только для устройств без pending events const QSet busyDevices = EventDAO::getDevicesWithPendingEvents(); const QStringList materialIds = AccountDAO::getActiveMaterialIdsExcludingDevices(busyDevices); if (!materialIds.isEmpty()) { m_network->getTasks(materialIds); } }); m_pollTimer->start(); // Закрываем зависшие события с прошлого запуска (только не отправленные на сервер) for (const EventStatus st : {EventStatus::Wait, EventStatus::InProgress}) { const QList stale = EventDAO::getAllEvents(st); for (const EventInfo &e : stale) { if (e.sentToServer) continue; EventDAO::updateEvent(e.id, EventStatus::Error, "App restarted"); EventDAO::markSentToServer(e.id); sendTaskError(eventTypeToString(e.type), e.externalId, e.bankName, "App restarted, event was pending"); qDebug() << "[EventHandler] Closed stale event id:" << e.id; } } // Первый запрос сразу при старте qDebug() << "[EventHandler] starting first getTasks()"; m_network->getTasks(); } void EventHandler::onTasksReceived(const QJsonArray &tasks) { if (!m_running) return; if (!tasks.isEmpty()) qDebug() << "[EventHandler] onTasksReceived, count:" << tasks.size(); // Получаем список онлайн устройств один раз const QStringList onlineDevices = DeviceDAO::getOnlineDeviceCodes(); // 1. Сохраняем новые задачи в БД for (const auto &taskVal : tasks) { const QJsonObject task = taskVal.toObject(); const QString bankName = task["bank_name"].toString(); const QString materialId = task["material_id"].isString() ? task["material_id"].toString() : QString::number(task["material_id"].toInt()); const QString typeStr = task["type"].toString(); const EventType eventType = eventTypeFromString(typeStr); if (eventType == EventType::Unknown) { qWarning() << "[EventHandler] Unknown task type:" << typeStr; continue; } // Ищем аккаунт по material_id напрямую const AccountInfo account = AccountDAO::findByMaterialId(materialId); if (account.id == -1) { qWarning() << "[EventHandler] Account not found for bank:" << bankName << "material_id:" << materialId; sendTaskError(typeStr, materialId, bankName, "Account not found for material_id: " + materialId); continue; } EventInfo event; event.status = EventStatus::Wait; event.type = eventType; event.externalId = materialId; event.accountId = account.id; event.deviceId = account.deviceId; event.bankName = bankName; event.timestamp = QDateTime::currentDateTimeUtc(); event.json = QJsonDocument(task).toJson(QJsonDocument::Compact); const QJsonObject body = task["body"].toObject(); if (eventType == EventType::CreateTransaction) { event.amount = body["amount"].toDouble() / 100.0; event.phone = body["bank_participator_id"].toString(); // Резолвим название банка-получателя по алиасу const QString bankAlias = body["extra"].toObject()["bank_participator_name"].toString(); if (!bankAlias.isEmpty()) { const BankInfo bank = BankDAO::findByAlias(bankAlias); event.comment = !bank.nspkName.isEmpty() ? bank.nspkName : bank.name; } } const int eventId = EventDAO::insertEvent(event); if (eventId == -1) { qCritical() << "[EventHandler] Failed to insert event, external_id:" << materialId; continue; } // Если устройство офлайн — сразу отправляем ошибку на сервер if (!onlineDevices.contains(account.deviceId)) { qWarning() << "[EventHandler] Device offline for material_id:" << materialId << "device:" << account.deviceId; EventDAO::updateEvent(eventId, EventStatus::Error, "Device is offline"); sendTaskError(typeStr, materialId, bankName, "Device is offline"); } } // 2. Для каждого онлайн устройства без активного потока — запускаем следующую задачу for (const QString &deviceId : onlineDevices) { processNextTask(deviceId); } } static void sendTaskResult(const QString &type, const QString &materialId, const QString &bankName, const QJsonValue &data, int eventId = -1) { QJsonObject obj; obj["type"] = type; obj["material_id"] = materialId; // UUID string obj["bank_name"] = bankName; obj["data"] = data; auto *thread = new QThread; auto *networkService = new NetworkService; networkService->moveToThread(thread); networkService->setPayload(QJsonDocument(obj).toJson()); QObject::connect(thread, &QThread::started, networkService, [networkService, eventId]() { networkService->postTaskResult(); if (eventId != -1) { EventDAO::markSentToServer(eventId); } }); QObject::connect(networkService, &NetworkService::finished, thread, &QThread::quit); QObject::connect(networkService, &NetworkService::finished, networkService, &QObject::deleteLater); QObject::connect(thread, &QThread::finished, thread, &QThread::deleteLater); thread->start(); } static void sendTaskError(const QString &type, const QString &materialId, const QString &bankName, const QString &description) { const QString message = QStringLiteral("[%1] bank=%2 material=%3: %4") .arg(type, bankName, materialId, description); auto *thread = new QThread; auto *networkService = new NetworkService; networkService->moveToThread(thread); QObject::connect(thread, &QThread::started, networkService, [networkService, message]() { networkService->postMonitoringLog("error", "task_error", message, {}); emit networkService->finished(); }); QObject::connect(networkService, &NetworkService::finished, thread, &QThread::quit); QObject::connect(networkService, &NetworkService::finished, networkService, &QObject::deleteLater); QObject::connect(thread, &QThread::finished, thread, &QThread::deleteLater); thread->start(); } static void sendMonitoringLog(const QString &type, const QString &event, const QString &message, const QByteArray &image = {}) { auto *thread = new QThread; auto *networkService = new NetworkService; networkService->moveToThread(thread); QObject::connect(thread, &QThread::started, networkService, [networkService, type, event, message, image]() { networkService->postMonitoringLog(type, event, message, image); emit networkService->finished(); }); QObject::connect(networkService, &NetworkService::finished, thread, &QThread::quit); QObject::connect(networkService, &NetworkService::finished, networkService, &QObject::deleteLater); QObject::connect(thread, &QThread::finished, thread, &QThread::deleteLater); thread->start(); } static void sendEventStatus(const QString &deviceId, const QString &externalId, const int eventId, const QString &status, const QString &error) { QJsonObject obj; obj["id"] = externalId; obj["externalId"] = eventId; obj["status"] = status; obj["error"] = error; auto *paymentThread = new QThread; auto *networkService = new NetworkService; networkService->moveToThread(paymentThread); networkService->setDeviceId(deviceId); networkService->setPayload(QJsonDocument(obj).toJson()); QObject::connect(paymentThread, &QThread::started, networkService, &NetworkService::postEventStatus); QObject::connect(networkService, &NetworkService::finished, paymentThread, &QThread::quit); QObject::connect(networkService, &NetworkService::finished, networkService, &QObject::deleteLater); QObject::connect(paymentThread, &QThread::finished, paymentThread, &QThread::deleteLater); paymentThread->start(); } void EventHandler::updateEventThread(const QString &key, const EventInfo &event, const QString &result) { const EventStatus newStatus = result.isEmpty() ? EventStatus::Complete : EventStatus::Error; if (!EventDAO::updateEvent(event.id, newStatus, result)) { qCritical() << "[EventHandler] Failed to update event:" << event.id; } else { if (newStatus == EventStatus::Error) { sendTaskError(eventTypeToString(event.type), event.externalId, event.bankName, result); EventDAO::markSentToServer(event.id); } if (newStatus != EventStatus::Error) { const AccountInfo account = AccountDAO::getAccountById(event.accountId); const QMap currencyCodes = { {"RUB", 643}, {"USD", 840}, {"KRW", 410}, {"AZN", 944}, {"ARS", 32} }; const int currencyCode = currencyCodes.value(account.currency.toUpper(), 643); QJsonValue taskData; if (event.type == EventType::FetchProfile) { EventDAO::updateEventAmount(event.id, account.amount); QJsonObject obj; obj["bank_account_id"] = QString(); obj["amount"] = static_cast(account.amount * 100); obj["currency"] = currencyCode; taskData = obj; } else if (event.type == EventType::FetchTransactions) { QList txList = TransactionDAO::getTransactionsWithinHours(event.accountId, 24); QJsonArray arr; for (const auto &tx : txList) { QJsonObject txObj; txObj["bank_transaction_id"] = tx.bankTrExternalId.isEmpty() ? QJsonValue(QJsonValue::Null) : QJsonValue(tx.bankTrExternalId); txObj["amount"] = static_cast(tx.amount * 100); txObj["currency"] = currencyCode; txObj["transaction_type"] = (tx.type == TransactionType::Phone) ? "top-up" : "bank"; txObj["bank_participator_id"]= tx.phone; txObj["status"] = tx.status == TransactionStatus::Complete ? "completed" : (tx.status == TransactionStatus::Wait ? "progress" : "completed"); QJsonObject extraObj; extraObj["fee"] = static_cast(tx.fee * 100); txObj["extra"] = extraObj; if (tx.timestamp.isValid()) { txObj["created_at"] = tx.timestamp.toTimeZone(QTimeZone("Europe/Moscow")).toString(Qt::ISODateWithMs); } else if (!tx.bankTime.isEmpty()) { QDateTime dt = QDateTime::fromString(tx.bankTime, "dd.MM.yyyy, HH:mm"); if (!dt.isValid()) dt = QDateTime::fromString(tx.bankTime, "dd.MM.yyyy HH:mm"); if (dt.isValid()) { dt.setTimeZone(QTimeZone("Europe/Moscow")); txObj["created_at"] = dt.toString(Qt::ISODateWithMs); } else { txObj["created_at"] = QJsonValue(QJsonValue::Null); } } else { txObj["created_at"] = QJsonValue(QJsonValue::Null); } arr.append(txObj); } taskData = arr; } else if (event.type == EventType::CreateTransaction) { // Определяем тип транзакции из JSON задачи const QJsonObject taskJson = QJsonDocument::fromJson(event.json.toUtf8()).object(); const QString txType = taskJson["body"].toObject()["type"].toString(); QJsonObject obj; obj["amount"] = -static_cast(event.amount * 100); obj["currency"] = currencyCode; obj["transaction_type"] = txType.isEmpty() ? "top-up" : txType; obj["bank_participator_id"]= event.phone; obj["status"] = "completed"; obj["created_at"] = QDateTime::currentDateTime().toTimeZone(QTimeZone("Europe/Moscow")).toString(Qt::ISODateWithMs); // Берём bank_transaction_id и fee из последней сохранённой транзакции QList txList = TransactionDAO::getTransactionsWithinHours(event.accountId, 1); int feeInCents = 0; for (const auto &tx : txList) { if (!tx.bankTrExternalId.isEmpty() && qAbs(tx.amount) == event.amount && tx.phone == event.phone) { obj["bank_transaction_id"] = tx.bankTrExternalId; feeInCents = static_cast(tx.fee * 100); break; } } QJsonObject extraObj; extraObj["fee"] = feeInCents; obj["extra"] = extraObj; taskData = obj; } else { QJsonObject obj; obj["status"] = "completed"; taskData = obj; } sendTaskResult( eventTypeToString(event.type), event.externalId, event.bankName, taskData, event.id ); } if (newStatus == EventStatus::Complete && event.type == EventType::CreateTransaction) { const QString msg = QStringLiteral("[CREATE_TRANSACTION] bank=%1 material=%2 amount=%3 phone=%4") .arg(event.bankName, event.externalId, QString::number(event.amount, 'f', 2), event.phone); QByteArray screenshot; const QString tmpPath = QDir::tempPath() + "/arc_tx_" + QString::number(event.id); if (AdbUtils::takeScreenshot(event.deviceId, tmpPath)) { QFile f(tmpPath + ".png"); if (f.open(QIODevice::ReadOnly)) { screenshot = f.readAll(); f.close(); } f.remove(); } sendMonitoringLog("info", "task_success", msg, screenshot); } } if (const auto tm = m_timers.take(key)) { tm->stop(); tm->deleteLater(); } m_threads.remove(key); // Сразу пробуем запустить следующую задачу для этого устройства processNextTask(key); } void EventHandler::startThreadForTask(const AccountInfo &account, const EventInfo &event) { const auto thr = new QThread(this); const QString key = account.deviceId; // Generic helper: подключает сигналы и переносит worker в поток auto setup = [&](auto *worker) { using W = std::remove_pointer_t; worker->moveToThread(thr); connect(thr, &QThread::started, worker, &W::start); connect(worker, &W::finished, thr, &QThread::quit); connect(worker, &W::finished, worker, &QObject::deleteLater); connect(thr, &QThread::finished, thr, &QObject::deleteLater); connect(worker, &W::finishedWithResult, this, [this, key, event](const QString &result) { updateEventThread(key, event, result); } ); }; const QString &appCode = account.appCode; const QString &deviceId = account.deviceId; if (event.type == EventType::FetchProfile) { if (appCode == "ozon") setup(new Ozon::LoginAndCheckAccountsScript(deviceId, appCode)); else if (appCode == "black") setup(new Black::LoginAndCheckAccountsScript(deviceId, appCode)); else qWarning() << "[EventHandler] FETCH_PROFILE: unknown appCode:" << appCode; } if (event.type == EventType::FetchTransactions) { if (appCode == "ozon") setup(new Ozon::GetLastTransactionsScript(deviceId, appCode)); else if (appCode == "black") setup(new Black::GetLastTransactionsScript(deviceId, appCode)); else qWarning() << "[EventHandler] FETCH_TRANSACTIONS: unknown appCode:" << appCode; } if (event.type == EventType::CreateTransaction) { const QString recipientBank = event.comment; // резолвленное имя банка-получателя // Определяем тип транзакции из JSON задачи const QJsonObject taskJson = QJsonDocument::fromJson(event.json.toUtf8()).object(); const QString txType = taskJson["body"].toObject()["type"].toString(); if (txType == "bank") { // Перевод по номеру карты if (appCode == "ozon") setup(new Ozon::PayByCardScript(account, event.phone, event.amount)); else if (appCode == "black") setup(new Black::PayByCardScript(account, event.phone, event.amount)); else qWarning() << "[EventHandler] CREATE_TRANSACTION bank: unknown appCode:" << appCode; } else { // Перевод по номеру телефона (top-up) if (appCode == "ozon") setup(new Ozon::PayByPhoneScript(account, event.phone, recipientBank, event.amount)); else qWarning() << "[EventHandler] CREATE_TRANSACTION: unknown appCode:" << appCode; } } // // Watchdog-таймер: 5 минут — убиваем зависший поток // const auto timer = new QTimer(this); // timer->setSingleShot(true); // connect(timer, &QTimer::timeout, this, [this, key]() { // qWarning() << "[EventHandler] Timeout for key" << key << "- killing stuck thread."; // if (QThread *thread = m_threads.value(key); thread && thread->isRunning()) { // thread->requestInterruption(); // if (!thread->wait(500)) { // qWarning() << "[EventHandler] Hard terminate for key" << key; // thread->terminate(); // thread->wait(); // } // } // if (const auto tm = m_timers.take(key)) { // tm->deleteLater(); // } // m_threads.remove(key); // }); m_threads[key] = thr; // m_timers[key] = timer; // timer->start(5 * 60 * 1000); thr->start(); } void EventHandler::processNextTask(const QString &deviceId) { if (!m_running) return; if (const QThread *thr = m_threads.value(deviceId, nullptr); thr && thr->isRunning()) return; while (true) { const EventInfo event = EventDAO::getFirstWaitEventByDeviceId(deviceId); if (event.id == -1) break; const qint64 waitSecs = event.timestamp.secsTo(QDateTime::currentDateTimeUtc()); if (waitSecs > 5 * 60) { qWarning() << "[EventHandler] Task expired after" << waitSecs << "s, event id:" << event.id; EventDAO::updateEvent(event.id, EventStatus::Error, "Task expired: waited " + QString::number(waitSecs) + "s"); sendTaskError(eventTypeToString(event.type), event.externalId, event.bankName, "Task expired: not started within 5 minutes"); continue; } const AccountInfo account = AccountDAO::getAccountById(event.accountId); const ApplicationInfo app = ApplicationDAO::getApplication(deviceId, account.appCode); if (app.status != "active") break; if (!EventDAO::updateEvent(event.id, EventStatus::InProgress, "")) { qCritical() << "[EventHandler] Failed to set InProgress, event id:" << event.id; break; } startThreadForTask(account, event); break; } } void EventHandler::clearAll() { for (QThread *thread : m_threads) { if (thread->isRunning()) { thread->requestInterruption(); thread->terminate(); thread->wait(); } thread->deleteLater(); } m_threads.clear(); for (QTimer *timer : m_timers) { timer->stop(); timer->deleteLater(); } m_timers.clear(); } void EventHandler::stop() { m_running = false; if (m_pollTimer) m_pollTimer->stop(); clearAll(); }