Active subscribers

src/events.ts

@@ -5,6 +5,11 @@ import { sendPushNotification } from './helpers/beams';
 import db from './helpers/mysql';
 import { sha256 } from './helpers/utils';
 import { getProposal, getProposalScores } from './helpers/proposal';
+import {
+  getActiveSubscribers,
+  updateActiveSubscribers,
+  updateSubscribers
+} from './helpers/subscribers';
 
 const delay = 5;
 const interval = 15;
@@ -62,32 +67,42 @@ export async function sendEvent(event, to) {
   event.secret = sha256(`${to}${serviceEventsSalt}`);
   const headerSecret = sha256(`${to}${process.env.SERVICE_EVENTS_SALT}`);
   try {
+    const controller = new AbortController();
+    setTimeout(() => controller.abort(), 10000); // Abort request after 10 seconds if there is no response
     const res = await fetch(to, {
+      signal: controller.signal,
       method: 'POST',
       headers: {
         'Content-Type': 'application/json',
         Authentication: headerSecret
       },
       body: JSON.stringify(event)
     });
-    return res.text();
+    if (res.status !== 200) {
+      throw new Error(`${res.status} ${res.statusText}`);
+    }
+    return 'success';
   } catch (error) {
     console.log('[events] Error sending event data to webhook', to, JSON.stringify(error));
-    return;
+    return 'failed';
   }
 }
 
-const sendEventToWebhookSubscribers = (event, subscribers) => {
-  Promise.allSettled(
-    subscribers
-      .filter(subscriber => [event.space, '*'].includes(subscriber.space))
-      .map(subscriber => sendEvent(event, subscriber.url))
-  )
-    .then(() => console.log('[events] Process event done'))
-    .catch(e => console.log('[events] Process event failed', e));
+const sendEventToWebhookSubscribers = async event => {
+  const subscribers = await getActiveSubscribers(event.space);
+  console.log('[events] Subscribers for this event: ', subscribers.length);
+  if (!subscribers.length) return;
+  try {
+    const allSubscribersStatus = await Promise.all(
+      subscribers.map(subscriber => sendEvent(event, subscriber.url))
+    );
+    updateSubscribers(subscribers, allSubscribersStatus);
+  } catch (e) {
+    console.log('[events] sendEventToWebhookSubscribers failed', e);
+  }
 };
 
-async function processEvents(subscribers) {
+async function processEvents() {
   const ts = parseInt((Date.now() / 1e3).toFixed()) - delay;
   const events = await db.queryAsync('SELECT * FROM events WHERE expire <= ?', [ts]);
 
@@ -108,7 +123,7 @@ async function processEvents(subscribers) {
     // TODO: handle errors and retry
     if (servicePushNotifications && event.event === 'proposal/start') sendPushNotification(event);
     sendEventToDiscordSubscribers(event.event, proposalId);
-    sendEventToWebhookSubscribers(event, subscribers);
+    sendEventToWebhookSubscribers(event);
 
     try {
       await db.queryAsync('DELETE FROM events WHERE id = ? AND event = ? LIMIT 1', [
@@ -123,15 +138,14 @@ async function processEvents(subscribers) {
 }
 
 async function run() {
+  await snapshot.utils.sleep(interval * 1e3);
   try {
-    const subscribers = await db.queryAsync('SELECT * FROM subscribers');
-    console.log('[events] Subscribers', subscribers.length);
-    await processEvents(subscribers);
+    await processEvents();
   } catch (e) {
     console.log('[events] Failed to process', e);
   }
-  await snapshot.utils.sleep(interval * 1e3);
-  await run();
+  updateActiveSubscribers();
+  run();
 }
 
-if (serviceEvents) setTimeout(() => run(), interval * 1e3);
+if (serviceEvents) run();

src/helpers/schema.sql

@@ -31,8 +31,10 @@ CREATE TABLE subscribers (
   owner VARCHAR(256) NOT NULL,
   url TEXT NOT NULL,
   space VARCHAR(256) NOT NULL,
-  active INT(11) NOT NULL,
-  created INT(11) NOT NULL,
+  last_active INT(11) DEFAULT 0 NOT NULL,
+  last_attempt INT(11) DEFAULT 0 NOT NULL,
+  active int DEFAULT 1 NOT NULL,
+  created TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
   PRIMARY KEY (id),
   INDEX owner (owner),
   INDEX space (space),

src/helpers/subscribers.ts

@@ -0,0 +1,44 @@
+import db from './mysql';
+
+const subcribersActiveTime = 30; // days
+
+export const getActiveSubscribers = async space => {
+  const subscribersQuery = `
+    SELECT * FROM subscribers
+    WHERE active = 1 AND (space LIKE CONCAT('%', ?, '%') OR space = "*")
+  `;
+  const subscribersParams = [space];
+  return db.queryAsync(subscribersQuery, subscribersParams);
+};
+
+export const updateSubscribers = async (subscribers, allSubscribersStatus) => {
+  const subscriberSuccessIds = subscribers
+    .filter((_subscriber, index) => allSubscribersStatus[index] === 'success')
+    .map(subscriber => subscriber.id);
+
+  const ts = parseInt((Date.now() / 1e3).toFixed());
+  let query = ` 
+          UPDATE subscribers SET last_attempt = ? WHERE id IN (?); 
+        `;
+  const params = [ts, subscribers.map(subscriber => subscriber.id)];
+  if (subscriberSuccessIds.length) {
+    query += `UPDATE subscribers SET last_active = ? WHERE id IN (?);`;
+    params.push(ts, subscriberSuccessIds);
+  }
+  return db.queryAsync(query, params).catch(e => {
+    console.log('[events] updateSubscribers failed', e);
+  });
+};
+
+export const updateActiveSubscribers = async () => {
+  const ts = subcribersActiveTime * 24 * 60 * 60;
+  const query = `
+    UPDATE subscribers SET active = 0 
+    WHERE active = 1 AND
+    DATEDIFF(CURRENT_TIMESTAMP, created) > ? AND
+    last_active + ? < last_attempt;`;
+  const params = [subcribersActiveTime, ts];
+  return db.queryAsync(query, params).catch(e => {
+    console.log('[events] updateActiveSubscribers failed', e);
+  });;
+};

src/subscriptions.ts

@@ -9,5 +9,5 @@ export async function loadSubscriptions() {
     if (!subs[sub.space]) subs[sub.space] = [];
     subs[sub.space].push(sub);
   });
-  console.log('Subscriptions', Object.keys(subs).length);
+  console.log('Discord Subscriptions', Object.keys(subs).length);
 }