vnext: Telegram bot (WIP)

1 files changed, 28 insertions, 5 deletions

diff --git a/vnext/server/middleware/event.js b/vnext/server/middleware/event.js
index 54611280..1267d1c4 100644
--- a/vnext/server/middleware/event.js
+++ b/vnext/server/middleware/event.js
@@ -1,8 +1,27 @@
 import { simpleParser } from 'mailparser';
 import { isPM, isReply, isService } from '../common/MessageUtils';
+import { sendTelegramNotification } from '../durov';
 import { subscribers } from '../http';
 import { sendNotification, buildNotification } from '../sender';
-var debug = require('debug')('event');
+import debug from 'debug';
+var log = debug('event');
+import config from 'config';
+import EventSource from 'eventsource';
+
+const es = new EventSource(config.get('service.baseURL')+ '/api/events');
+es.addEventListener('msg', (msg) => {
+    log(msg.data);
+    processMessageEvent(JSON.parse(msg.data));
+});
+es.addEventListener('read', (msg) => {
+    log(msg);
+});
+es.addEventListener('open', () => {
+    log('online');
+});
+es.onerror = (evt) => {
+    log(`err: ${JSON.stringify(evt)}`);
+};
 
 /** @type {number[]} */
 const allSandboxIds = [];
@@ -23,7 +42,7 @@ function processMessageEvent(msg) {
     }
     subscribers(new URLSearchParams(JSON.parse(JSON.stringify(params)))).then(users => {
         users.forEach(user => {
-            debug(`${user.uname}: ${user.unreadCount}`);
+            log(`${user.uname}: ${user.unreadCount}`);
             let [sandboxTokens, productionTokens] = (user.tokens || [])
                 .filter(t => ['mpns', 'apns', 'gcm'].includes(t.type))
                 .map(t => t.token)
@@ -34,8 +53,12 @@ function processMessageEvent(msg) {
                     return result;
                 }, [[], []]);
             sendNotification(productionTokens, sandboxTokens, buildNotification(user, msg));
+            let durovIds = (user.tokens || [])
+                .filter(t => ['durov'].includes(t.type))
+                .map(t => t.token);
+            sendTelegramNotification(msg, durovIds);
         });
-    }).catch(console.error);
+    }).catch(log);
 }
 
 /**
@@ -47,7 +70,7 @@ export default function handleMessage(req, res) {
     return simpleParser(req.body, {})
         .then(parsed => {
             const new_version = parsed.headers.get('x-event-version') == '1.0';
-            debug(`New event: ${parsed.text}, new version: ${new_version}`);
+            log(`New event: ${parsed.text}, new version: ${new_version}`);
             if (new_version) {
                 /** @type {import('../../client').SystemEvent} */
                 const event = JSON.parse(parsed.text || '');
@@ -71,5 +94,5 @@ export default function handleMessage(req, res) {
             }
             res.end();
         })
-        .catch(err => { console.error(err); res.status(400).send('Invalid request'); });
+        .catch(err => { log(err); res.status(400).send('Invalid request'); });
 }