Merge pull request #2 from antonkorotkov/dev

Dev

.env-sample

@@ -0,0 +1,2 @@
+RABBITMQ_DEFAULT_USER=
+RABBITMQ_DEFAULT_PASS=

app/.env-sample

@@ -0,0 +1,4 @@
+DB_CONNECTION_STRING=mongodb://mongodb:27017/coinMonitor
+BOT_TOKEN=
+WCI_KEY=
+AMQP_CONNECTION_STRING=amqp://<>:<>@rabbitmq

app/Dockerfile

@@ -1,19 +1,13 @@
-# Use the official Node.js image as the base image
-FROM node:18
+FROM node:20-alpine
 
-# Create and change to the app directory
 WORKDIR /usr/src/app
 
-# Copy application dependency manifests to the container image.
 COPY package*.json ./
 
-# Install dependencies
 RUN npm install
 
-# Copy the rest of the application code to the container image.
 COPY . .
 
 ENV DEBUG=*
 
-# Run the application
 CMD ["node", "index.js"]

app/amqp/index.js

@@ -0,0 +1,38 @@
+const amqp = require('amqplib');
+
+class AMQP {
+    #connection;
+
+    /** @type {amqp.ConfirmChannel | undefined} */
+    #channel;
+
+    /**
+     * @param {string} connectionString
+     */
+    constructor(connectionString) {
+        this.#connection = amqp.connect(connectionString);
+    }
+
+    /**
+     * @param {string} queueName
+     * @returns {Promise<amqp.ConfirmChannel>}
+     */
+    async getChannel(queueName) {
+        if (!this.#channel) {
+            this.#channel = await (await this.#connection).createConfirmChannel();
+
+            await this.#channel.assertQueue(queueName, {
+                durable: true
+            });
+        }
+
+        return this.#channel;
+    }
+
+    async closeChannel() {
+        await this.#channel.close();
+        this.#channel = undefined;
+    }
+}
+
+module.exports = AMQP;

app/app.js

@@ -2,14 +2,20 @@ class App {
     #db;
     #mongoose;
     #log;
+
+    /** @type {import('./market/MarketsService')} */
     #marketsService;
+
+    /** @type {import('./market/MarketChangesService')} */
+    #marketChangesService;
     #telegramBot;
 
     constructor(options) {
         this.#mongoose = options.mongoose;
         this.#db = options.db;
         this.#log = options.logger.debug(this.constructor.name);
         this.#marketsService = options.marketsService;
+        this.#marketChangesService = options.marketChangesService;
         this.#telegramBot = options.telegramBot;
 
         this.#log('Initializing...');
@@ -19,8 +25,12 @@ class App {
         this.#mongoose.connection.once('open', async () => {
             this.#log('Connected to database!');
 
-            await this.#marketsService.watch(changed => {
-                this.#log(changed.length);
+            await this.#marketsService.watch(async changedMarkets => {
+                for (const market of changedMarkets) {
+                    await this.#marketChangesService.push(market);
+                }
+
+                await this.#marketChangesService.closeChannel();
             });
 
             this.#telegramBot.start();

app/bot/TelegramBot.js

@@ -24,9 +24,10 @@ class TelegramBot {
         this.#bot.catch(args => this.errorHandler(args));
         this.#bot.use(options.logger.middleware(this.constructor.name));
         this.#bot.use(i18n);
-        this.#bot.use(session({ initial: () => ({}) }));
+        this.#bot.use(session({ initial: () => ({}), type: 'single' }));
         this.#bot.use(conversations());
         this.#bot.use(createConversation(options.conversationAddMonitor(this.#bot)));
+        this.#bot.use(createConversation(options.conversationDeleteMonitor(this.#bot)));
         this.#bot.use(createConversation(options.conversationSearch(this.#bot)));
 
         this.#bot.command('start', start);

app/bot/commands/search.js

@@ -1,5 +1,8 @@
 const search = async ctx => {
-    await ctx.conversation.enter('searchConversation');
+    const stats = await ctx.conversation.active();
+
+    if (!Object.keys(stats).length)
+        await ctx.conversation.enter('searchConversation');
 };
 
 module.exports = search;

app/bot/conversations/addMonitor.js

@@ -1,9 +1,67 @@
-module.exports = options => bot => {
+const { InlineKeyboard } = require("grammy");
+const Monitor = require("../../db/schemas/Monitor");
 
+const TYPE_FIXED = 'fixed';
+const TYPE_PERCENTAGE = 'percentage';
+const CONFIRM_CREATE = 'create';
+const CONFIRM_CANCEL = 'cancel';
+
+module.exports = ({ logger }) => _ => {
     return async function addMonitorConversation(conversation, ctx) {
-        await ctx.reply('Add monitor conversation started');
-        await ctx.reply(ctx.state.coin);
+        const typeKeyboard = new InlineKeyboard();
+        typeKeyboard.text(ctx.t('monitor_type_fixed'), TYPE_FIXED).text(ctx.t('monitor_type_percentage'), TYPE_PERCENTAGE);
+
+        await ctx.reply(ctx.t('monitor_type_prompt'), {
+            reply_markup: typeKeyboard
+        });
+
+        const monitorTypeCtx = await conversation.waitForCallbackQuery([TYPE_FIXED, TYPE_PERCENTAGE], {
+            otherwise: ctx => ctx.reply(ctx.t('use_buttons_alert'), { reply_markup: typeKeyboard }),
+        });
+
+        await monitorTypeCtx.editMessageText(ctx.t('monitor_threshold_prompt', { type: monitorTypeCtx.match === TYPE_PERCENTAGE ? '%' : '$' }), {
+            reply_markup: undefined
+        });
+
+        const value = await conversation.form.number(
+            ctx => ctx.reply(ctx.t('monitor_number_validation'), { parse_mode: 'HTML' })
+        );
+
+        if (value <= 0 || value > 1_000_000)
+            return await ctx.reply(ctx.t('monitor_bad_threshold'));
+
+        const confirmationKeyboard = new InlineKeyboard();
+        confirmationKeyboard.text(ctx.t('create'), CONFIRM_CREATE).text(ctx.t('cancel'), CONFIRM_CANCEL);
+        await ctx.reply(ctx.t('add_monitor_confirmation', { type: monitorTypeCtx.match, coin: ctx.state.coin, value: `${value}${monitorTypeCtx.match === TYPE_PERCENTAGE ? '%' : '$'}` }), {
+            parse_mode: 'HTML',
+            reply_markup: confirmationKeyboard
+        });
+
+        const confirmationCtx = await conversation.waitForCallbackQuery([CONFIRM_CREATE, CONFIRM_CANCEL], {
+            otherwise: ctx => ctx.reply(ctx.t('use_buttons_alert'), { reply_markup: confirmationKeyboard }),
+        });
+
+        if (confirmationCtx.match === CONFIRM_CANCEL)
+            return await ctx.reply('😒');
+
+        try {
+            await Monitor.createOrUpdate({
+                telegramId: ctx.chat.id,
+                coinId: ctx.state.id,
+                coin: ctx.state.coin,
+                coinName: ctx.state.name,
+                lastPrice: ctx.state.price,
+                threshold: {
+                    type: monitorTypeCtx.match,
+                    value: value
+                }
+            });
+
+            await ctx.reply(ctx.t('monitor_created'));
+        } catch (err) {
+            logger.debug('addMonitor')(err);
 
-        const { msg: { text } } = await conversation.waitFor("message:text");
+            await ctx.reply('Sorry, there was an error on our side. Please try again later or contact administrator.');
+        }
     };
 };

app/bot/conversations/deleteMonitor.js

@@ -0,0 +1,36 @@
+const { InlineKeyboard } = require("grammy");
+const Monitor = require("../../db/schemas/Monitor");
+
+const DELETE_CREATE = 'delete';
+const CONFIRM_CANCEL = 'cancel';
+
+module.exports = ({ logger }) => _ => {
+    return async function deleteMonitorConversation(conversation, ctx) {
+        try {
+            const confirmationKeyboard = new InlineKeyboard();
+            const { id } = ctx.state;
+
+            if (!id)
+                return await ctx.reply('Something went wrong. No monitor selected for deletion.');
+
+            confirmationKeyboard.text(ctx.t('delete'), DELETE_CREATE).text(ctx.t('cancel'), CONFIRM_CANCEL);
+            await ctx.reply(ctx.t('are_you_sure'), {
+                reply_markup: confirmationKeyboard
+            });
+
+            const confirmationCtx = await conversation.waitForCallbackQuery([DELETE_CREATE, CONFIRM_CANCEL], {
+                otherwise: ctx => ctx.reply(ctx.t('use_buttons_alert'), { reply_markup: confirmationKeyboard }),
+            });
+
+            if (confirmationCtx.match === CONFIRM_CANCEL)
+                return await ctx.reply('👍');
+
+            await Monitor.findByIdAndDelete(id);
+            await ctx.reply(ctx.t('monitor_deleted'));
+        } catch (err) {
+            logger.debug('deleteMonitor')(err);
+
+            await ctx.reply('Sorry, there was an error on our side. Please try again later or contact administrator.');
+        }
+    };
+};

app/bot/conversations/index.js

@@ -1,7 +1,9 @@
 const searchConversation = require('./search');
 const addMonitorConversation = require('./addMonitor');
+const deleteMonitorConversation = require('./deleteMonitor');
 
 module.exports = {
     searchConversation,
-    addMonitorConversation
+    addMonitorConversation,
+    deleteMonitorConversation
 }

app/bot/conversations/search.js

@@ -4,17 +4,18 @@ module.exports = options => bot => {
     const marketsService = options.marketsService;
 
     /**
-     * @param {string} coin
+     * @param {string} id
      * @param {import('grammy').CallbackQueryContext} ctx
      */
-    const onItemClick = async (coin, ctx) => {
-        const market = marketsService.getMarketByCoin(coin);
+    const onItemClick = async (id, ctx) => {
+        const market = marketsService.getMarketById(id);
 
         if (!market)
             return await ctx.reply(ctx.t('coin_not_found'));
 
         const userId = ctx.from.id;
-        const message = marketDetails.setMarket(userId, market).getMessage(userId);
+        const details = await marketDetails.setMarket(userId, market);
+        const message = details.getMessage(userId);
         const reply_markup = await marketDetails.getMarkup(userId, ctx);
 
         if (!message || !reply_markup)
@@ -26,20 +27,27 @@ module.exports = options => bot => {
         });
     }
 
-    const onAddMonitor = async (coin, ctx) => {
-        ctx.state = { coin };
-        await ctx.conversation.enter('addMonitorConversation');
+    const onAddMonitor = async (id, coin, price, name, ctx) => {
+        ctx.state = { id, coin, price, name };
+        const stats = await ctx.conversation.active();
+
+        if (!Object.keys(stats).length)
+            await ctx.conversation.enter('addMonitorConversation');
     };
 
-    const onShowMonitor = (coin, ctx) => {
-        ctx.reply(`Show Monitor for ${coin}`);
+    const onDeleteMonitor = async (monitor, ctx) => {
+        ctx.state = { id: monitor?.id };
+        const stats = await ctx.conversation.active();
+
+        if (!Object.keys(stats).length)
+            await ctx.conversation.enter('deleteMonitorConversation');
     };
 
-    marketDetails.init(bot, { onAddMonitor, onShowMonitor });
+    marketDetails.init(bot, { onAddMonitor, onDeleteMonitor });
     pagination.init(bot, {
         onItemClick,
         renderItem: i => `${i.getName()} (${i.getCoin()})`,
-        resolveItemId: i => i.getCoin()
+        resolveItemId: i => i.getId()
     });
 
     return async function searchConversation(conversation, ctx) {

app/bot/keyboards/AbstractPersonalizedCache.js

@@ -47,13 +47,13 @@ class AbstractPersonalizedCache {
         if (!this.#cache[userId])
             return this;
 
-        if (entityName && this.#cache[userId][entityName]) {
-            this.#cache[userId][entityName] = undefined;
-
+        if (!entityName) {
+            this.#cache[userId] = undefined;
             return this;
         }
 
-        this.#cache[userId] = undefined;
+        if (this.#cache[userId][entityName])
+            this.#cache[userId][entityName] = undefined;
 
         return this;
     }