Migrated notifications to Firebase

.env.example

@@ -4,6 +4,7 @@ DATABASE_URL=
 SERVICE_EVENTS=
 SERVICE_EVENTS_SALT=
 SERVICE_PUSH_NOTIFICATIONS=
-SERVICE_PUSHER_BEAMS_INSTANCE_ID=
-SERVICE_PUSHER_BEAMS_SECRET_KEY=
 HUB_URL=
+FIREBASE_PROJECT_ID=
+FIREBASE_CLIENT_EMAIL=
+FIREBASE_PRIVATE_KEY=

.gitignore

@@ -4,7 +4,6 @@ dist
 build
 .env
 
-
 # Remove some common IDE working directories
 .idea
 .vscode

.nvmrc

@@ -0,0 +1 @@
+v16.17.0

package.json

@@ -19,6 +19,7 @@
     "dotenv": "^10.0.0",
     "eslint": "^6.7.2",
     "express": "^4.17.1",
+    "firebase-admin": "^11.0.1",
     "lodash.chunk": "^4.2.0",
     "mysql": "^2.18.1",
     "nodemon": "^2.0.15",

src/api.ts

@@ -1,6 +1,14 @@
 import express from 'express';
 import { sendEvent } from './events';
 import pkg from '../package.json';
+import {
+  isTokenValid,
+  sendPushNotification,
+  subscribeUser,
+  toggleSpaceNotification,
+  unsubscribeUser
+} from './helpers/firebase';
+import { sendError } from './helpers/utils';
 
 const router = express.Router();
 
@@ -14,18 +22,58 @@ router.get('/', async (req, res) => {
 router.get('/test', async (req, res) => {
   const url: any = req.query.url || '';
   const event = {
-    id: `proposal/0x38c654c0f81b63ea1839ec3b221fad6ecba474aa0c4e8b4e8bc957f70100e753`,
-    space: 'pistachiodao.eth',
+    id: `proposal/0xc59f05d899cd80178300b724c4bf43a037a908741369690243aa4a77bbafbf18`,
+    space: 'fabien.eth',
     event: 'proposal/created',
-    expire: 1647343155
+    expire: 1664272182
   };
   try {
-    new URL(url);
-    await sendEvent(event, url);
+    // new URL(url);
+    // await sendEvent(event, url);
+    await sendPushNotification(event);
     return res.json({ url, success: true });
   } catch (e) {
     return res.json({ url, error: e });
   }
 });
 
+router.post('/device', async (req, res) => {
+  const owner = req.body?.owner;
+  const token = req.body?.token;
+  try {
+    // Add token to DB and subscribe user to topics
+    await subscribeUser(token, owner);
+    return res.json({ success: true });
+  } catch (error) {
+    console.log('[notifications] Error adding device', error);
+    return sendError(res, 'Error enabling push notifications');
+  }
+});
+
+router.delete('/device', async (req, res) => {
+  const owner = req.body?.owner;
+  const token = req.body?.token;
+  try {
+    // Delete token from DB and unsubscribe user from topics
+    await unsubscribeUser(token, owner);
+    return res.json({ success: true });
+  } catch (error) {
+    console.log('[notifications] Error deleting device', error);
+    return sendError(res, 'Error disabling push notifications');
+  }
+});
+
+router.post('/subscribed', async (req, res) => {
+  const owner = req.body?.owner;
+  const token = req.body?.token;
+  try {
+    // Validate token from DB and subscribe user to topics
+    const subscribed = await isTokenValid(token, owner);
+    return res.json({ subscribed });
+  } catch (error) {
+    console.log('[notifications] Error validating device', error);
+    return sendError(res, 'Error validating device');
+  }
+});
+
 export default router;

src/events.ts

@@ -1,7 +1,7 @@
 import fetch from 'cross-fetch';
 import snapshot from '@snapshot-labs/snapshot.js';
-import { sendEventToDiscordSubscribers } from './discord';
-import { sendPushNotification } from './helpers/beams';
+// import { sendEventToDiscordSubscribers } from './discord';
+import { sendPushNotification } from './helpers/firebase';
 import db from './helpers/mysql';
 import { sha256 } from './helpers/utils';
 import { getProposal, getProposalScores } from './helpers/proposal';
@@ -110,7 +110,7 @@ async function processEvents(subscribers) {
     // Send event to discord subscribers and webhook subscribers and then delete event from db
     // TODO: handle errors and retry
     if (servicePushNotifications && event.event === 'proposal/start') sendPushNotification(event);
-    sendEventToDiscordSubscribers(event.event, proposalId);
+    // sendEventToDiscordSubscribers(event.event, proposalId);
     sendEventToWebhookSubscribers(event, subscribers);
 
     try {

src/helpers/firebase.ts

@@ -0,0 +1,142 @@
+import snapshot from '@snapshot-labs/snapshot.js';
+import { initializeApp, cert, ServiceAccount } from 'firebase-admin/app';
+import { getMessaging } from 'firebase-admin/messaging';
+
+import db from './mysql';
+import { getProposal } from './proposal';
+
+const { FIREBASE_PROJECT_ID, FIREBASE_CLIENT_EMAIL, FIREBASE_PRIVATE_KEY, HUB_URL } = process.env;
+
+const firebaseApp = initializeApp({
+  credential: cert({
+    projectId: FIREBASE_PROJECT_ID,
+    clientEmail: FIREBASE_CLIENT_EMAIL,
+    privateKey: (FIREBASE_PRIVATE_KEY || '').replace(/\\n/g, '\n')
+  } as ServiceAccount)
+});
+const messaging = getMessaging(firebaseApp);
+const hubURL = HUB_URL || 'https://hub.snapshot.org';
+
+async function getUserSpaces(user: string) {
+  let userSpaces: { [key: string]: any } = [];
+  const query = {
+    follows: {
+      __args: {
+        where: { follower: user }
+      },
+      space: { id: true }
+    }
+  };
+  try {
+    const result = await snapshot.utils.subgraphRequest(`${hubURL}/graphql`, query);
+    userSpaces = result.follows || [];
+  } catch (error) {
+    console.log('[notifications] Snapshot hub error:', error);
+  }
+  return userSpaces.map(follow => follow.space.id);
+}
+
+async function toggleTopicSubscription(
+  token: string,
+  owner: string,
+  space: string,
+  unsubscribe: boolean
+) {
+  try {
+    if (unsubscribe) {
+      await messaging.unsubscribeFromTopic(token, space);
+    } else {
+      await messaging.subscribeToTopic(token, space);
+    }
+
+    console.log(
+      `[notifications] Successfully ${
+        unsubscribe ? 'un' : ''
+      }subscribed user: ${owner} to space: ${space}`
+    );
+  } catch (e) {
+    console.error(
+      `[notifications] Error ${unsubscribe ? 'un' : ''}subscribing user ${owner} to space ${space}`,
+      e
+    );
+  }
+}
+
+export async function isTokenValid(token: string, owner: string) {
+  // TODO: Check token validity in Firebase
+  const device = await db.queryAsync(
+    'SELECT * FROM device_tokens WHERE token = ? AND owner = ? LIMIT 1',
+    [token, owner]
+  );
+  return Boolean(device.length);
+}
+
+export async function sendPushNotification(event) {
+  const proposal = await getProposal(event.id.replace('proposal/', ''));
+  if (!proposal) {
+    console.log('[notifications] Proposal not found', event.id);
+    return;
+  }
+
+  messaging
+    .sendToTopic(event.space, {
+      notification: {
+        title: event.space,
+        body: proposal.title,
+        clickAction: `${process.env.SNAPSHOT_URI}/#/${event.space}/${event.id}`
+      }
+    })
+    .then(response => {
+      console.log('Notification sent successfully!', response.messageId);
+    })
+    .catch(error => console.error(error.errorInfo.code));
+}
+
+export async function subscribeUser(token: string, owner: string) {
+  try {
+    const ts = parseInt((Date.now() / 1e3).toFixed());
+    await db.queryAsync(
+      `INSERT INTO device_tokens (token, owner, created, updated) VALUES (?, ?, ?, ?)`,
+      [token, owner, ts, ts]
+    );
+
+    const userSpaces = await getUserSpaces(owner);
+    userSpaces.map(async space => toggleTopicSubscription(token, owner, space, false));
+  } catch (e) {
+    console.error('[notifications] subscribeUser:', e);
+  }
+}
+
+export async function unsubscribeUser(token: string, owner: string) {
+  try {
+    await db.queryAsync('DELETE FROM device_tokens WHERE token = ? AND owner = ? LIMIT 1', [
+      token,
+      owner
+    ]);
+
+    const userSpaces = await getUserSpaces(owner);
+    userSpaces.map(async space => toggleTopicSubscription(token, owner, space, true));
+  } catch (e) {
+    console.error('[notifications] unsubscribeUser:', e);
+  }
+}
+
+export async function toggleSpaceNotification(
+  owner: string,
+  spaceId: string,
+  unsubscribe: boolean
+) {
+  try {
+    const deviceTokens = await db.queryAsync(`SELECT token FROM device_tokens WHERE owner = ?`, [
+      owner
+    ]);
+
+    if (deviceTokens.length !== 0) {
+      deviceTokens.map(async deviceToken =>
+        toggleTopicSubscription(deviceToken.token, owner, spaceId, unsubscribe)
+      );
+    }
+  } catch (e) {
+    console.error('[notifications] toggleSpaceNotification:', e);
+  }
+}

src/helpers/schema.sql

@@ -39,3 +39,14 @@ CREATE TABLE subscribers (
   INDEX active (active),
   INDEX created (created)
 );
+
+CREATE TABLE device_tokens (
+  token VARCHAR(256) NOT NULL,
+  owner VARCHAR(256) NOT NULL,
+  created VARCHAR(64) NOT NULL,
+  updated VARCHAR(64) NOT NULL,
+  PRIMARY KEY (token),
+  INDEX owner (owner),
+  INDEX created (created),
+  INDEX updated (updated)
+);

src/helpers/utils.ts

@@ -5,7 +5,12 @@ export function shortenAddress(str = '') {
 }
 
 export function sha256(str) {
-  return createHash('sha256')
-    .update(str)
-    .digest('hex');
+  return createHash('sha256').update(str).digest('hex');
+}
+
+export function sendError(res, description, status = 400) {
+  return res.status(status).json({
+    error: 'Bad request',
+    error_description: description
+  });
 }

src/index.ts

@@ -4,7 +4,7 @@ import bodyParser from 'body-parser';
 import cors from 'cors';
 import api from './api';
 import './replay';
-import './discord';
+// import './discord';
 
 const app = express();
 const PORT = process.env.PORT || 3000;