Давайте немного о коде и решениях 😄

Имперский Стражник — большая система с кучей разных источников сообщений. Сообщение может уйти из одного из 200 классов ядра или ещё большего зоопарка внешних модулей, но есть одна важная проблема: отправку сообщений надо контролировать.

Надо следить за лимитами, чтобы не передалбывать Bot API зазря. Надо следить за ответами, чтобы, если мы вышли за лимит, подождать и повторно отправить запрос. Надо вообще убеждаться в доставке, синхронизировать состояние и ответы.

Короче, следить надо за всем. Так что давайте я расскажу, что мы сделали, чтобы всё это работало и жило с учётом нашего стека.

Начнём с учёта отправки, лимитов и структуризации. Я решил, что за отправку сообщений будет отвечать синглтон-сервис, на который все остальные сервисы будут скидывать задачи. Но как это всё соединить, учитывая необходимость быстрого масштабирования и подключения новых модулей?

Тут на помощь пришёл Bull (не реклама, если что, просто делюсь либой, на которой делал решение). Давайте посмотрим, как выглядит воркер отправки:

new Worker(process.env.DEBUG ? "guard_dev_send" : "guard_send", async (job) => { // тут я переключаю очередь в зависимости от режима, в котором запущен сервис, чтобы не задевать прод во время тестирования
    let telegram = job.data.target === "media" ? mediaBot.telegram : bot.telegram; // в данный момент бот, через который будет идти отправка, выбирается немного костыльно, но мы в работе :D
    try {
        return await telegram.sendMessage(job.data.id, job.data.text, job.data.extra); // отправляем сообщение
    } catch (e: any) {
        if (e.code === 429) {
            await job.moveToDelayed((Date.now() / 1000) + 30000); // словили рейтлимит, ждём 30 секунд и пробуем опять
        } else {
            e._sourceStack = job.data.stack; // подпихиваем стек до очереди в ошибку
            e._stack = e.stack; // стек внутри очереди
            e._message = e.message; // и причину
            return e;
        }
    }
}, {
    connection: {
        host: process.env.REDIS_HOST,
        port: Number.parseInt(process.env.REDIS_PORT ?? "6379")
    },
    limiter: { // режем скорость отправки
        max: 2,
        duration: 1000
    },
    concurrency: 1, // и ограничиваем параллельность
}).on("failed", (job, err) => {
    Queues.logger.error({
        job,
        err
    }, "Failed to send message"); // если что-то пошло не так, логируем
});

Небольшой такой код, но он позволяет нам чётко контролировать очередь и лимиты. А дальше посмотрим на отправку и приём)

В комментариях разрешается открыть портал в холивар))

Давайте опять поговорим про очереди в Стражнике. Мы уже увидели сниппет кода, отвечающий за обработку, но как мы реализовали отправку? Вот тут всё начинает становиться сильно интереснее.

Основная сложность, с которой мы столкнулись, — необходимость рефакторить весь код. Хотелось избежать редактирования вообще всех мест, где идёт отправка сообщений. Поэтому мы вспомнили очень важную вещь, которая есть в JS: почти всё можно перезаписать по своему желанию. Этим мы и воспользовались, перезаписав метод отправки сообщения и подменив логику отправки в Telegram логикой отправки в очередь.

Получился вот такой прикольчик:

ctx.reply = async function (text: string, options?: ExtraSendMessage): Promise<Message> {
    return Queues.getInstance().sendMessage("replyQueued", this.chat.id, text, options);
}

Тут мы перезаписали оригинальный метод ответа на сообщение своим собственным. Собственно, вуаля — половина работы отменена)

Дальше задача номер два, которая стала сильно интереснее: отправка сообщений разделяется на два типа — там, где нам нужен ответ, и там, где нам наплевать на ответ.

Нашей первой реализацией была идея внутри каждой операции отправки сообщения создавать листенер, который будет ждать ответа о том, что сообщение отправилось, и резолвить промис с ответом. А делать await промиса или нет — уже решение на уровне бизнес-логики.

Что могло пойти не так? Да, как оказалось, много чего. У нас утекли листенеры. Их стало так много, что Node.js перестал справляться с их контролем.

Встал вопрос: как сделать решение, которое не перегрузит среду, но сможет дожидаться ответа по отправке сотен сообщений и делать это асинхронно?

И тут я придумал прикольное решение: создаём Map, который будет хранить ID задачи на отправку и коллбэк.

private _promisedMessages = new Map<string, { resolve: (message: Message) => void, reject: (error: Error) => void }>();

В отправке мы создаём задачу и промис, записывая в мапу ID задачи и resolve этого промиса в качестве коллбэка.

async sendMessage(target: TargetBot, name: string, id: number, text: string, extra: ExtraSendMessage = {}, priority = 10) {
    let error = new Error("Stack trace"); // получаем стектрейс до этого момента
    let job = await this._sendQueue.add(name, {id, text, extra, target, stack: error.stack}, {priority});
    return new Promise<Message>((resolve, reject) => {
        this._promisedMessages.set(job.id!, {resolve, reject});
    });
}

И вместо сотен листенеров делаем один-единственный глобальный, который будет работать до самого завершения процесса:

new QueueEvents(process.env.DEBUG ? "guard_dev_send" : "guard_send", {
    connection: {
        host: process.env.REDIS_HOST,
        port: Number.parseInt(process.env.REDIS_PORT ?? "6379"),
    },
})
    .on("completed", async ({jobId, returnvalue}) => {
        if (this._promisedMessages.has(jobId)) { // проверяем, что это сообщение ожидается на этом воркере
            let {resolve, reject} = this._promisedMessages.get(jobId)!; // получаем коллбэки этого сообщения
            this._promisedMessages.delete(jobId); // удаляем его из мапы
            resolve(returnvalue); // отдаём результат промису
        }
});

И тем самым получается, что в команде код выглядит как самый обычный вызов:

if (!peer.isUserKnown()) {
    await ctx.reply(await ctx.lang.t("commands.general.userNotFound"));
    return;
}

А на самом деле за пару миллисекунд запрос обходит две очереди и несколько сервисов, выполняя работу чётенько и безопасненько)

С первого взгляда выглядит как быстрое и логичное решение, но мы к этому шли месяц-полтора.

Вот как-то так.

Я живой (вроде), так что попробуем оживится ещё и в ваших глазах.
Встречаемся в 22:30 (МСК) в уже знакомом нам месте.

https://www.twitch.tv/sleepless_code

Потихоньку начинаем, ждём всех и изучаем новый плейлист музыки)

https://www.twitch.tv/sleepless_code

Игры всегда надо рассматривать с нескольких сторон. Всегда можно уйти в две крайности:
Первая — игры это зло, которое надо запретить.
Вторая — игры идеальны и ни в чём не виноваты.

Как говорили великие, нет ничего абсолютно чёрного и абсолютно белого. Это было бы слишком примитивно. Проблема в том, что реальность намного сложнее. (Если, конечно, не вспоминать The Horses.)

Игры могут делать совершенно разные вещи. Они могут:
- Помогать
- Развивать
- Объединять
- Давать убежище

Но при этом они могут:
- Вызывать зависимость (в частности, азартные)
- Усиливать агрессию
- Портить здоровье (особенно режим сна)

Из-за этого любая дискуссия легко превращается в банальный обмен простыми характеристиками. Что, кстати, очень напоминает комментарии в интернете.

Мне кажется, игры (хотя, возможно, не все — тут это уже моя субъективная точка зрения) не могут быть просто хорошими или плохими. Всё зависит от того, какие они, для кого они и что именно делают.
И это даже не вопрос качества уровня Concord. Это вопрос морально-этического осмысления. Потому что одна и та же игра для одного человека может стать способом социализации и поиска друзей, а для другого — способом убежать от проблем.

Именно поэтому мне кажется, что вопрос должен звучать не «хорошие игры или плохие», а «какое влияние оказывают конкретные игры и почему».

Давайте продолжим рассуждать вместе.
Как вы думаете, какие игры для кого и для чего существуют?