feat: add email subscriptions

.env.example

@@ -3,6 +3,7 @@ NETWORK=testnet
 MAINTENANCE=
 HUB_DATABASE_URL=mysql://...
 SEQ_DATABASE_URL=mysql://...
+ENVELOP_DATABASE_URL=mysql://
 RELAYER_PK=0x123...
 DEFAULT_NETWORK=1
 SHUTTER_URL=https://...

src/helpers/mysql.ts

@@ -33,7 +33,21 @@ sequencerConfig.connectTimeout = 60e3;
 sequencerConfig.acquireTimeout = 60e3;
 sequencerConfig.timeout = 60e3;
 sequencerConfig.charset = 'utf8mb4';
-bluebird.promisifyAll([Pool, Connection]);
 const sequencerDB = mysql.createPool(sequencerConfig);
 
-export { hubDB as default, sequencerDB };
+// @ts-ignore
+const envelopConfig = parse(process.env.ENVELOP_DATABASE_URL);
+envelopConfig.connectionLimit = connectionLimit;
+envelopConfig.multipleStatements = true;
+envelopConfig.database = envelopConfig.path[0];
+envelopConfig.host = envelopConfig.hosts[0].name;
+envelopConfig.port = envelopConfig.hosts[0].port;
+envelopConfig.connectTimeout = 60e3;
+envelopConfig.acquireTimeout = 60e3;
+envelopConfig.timeout = 60e3;
+envelopConfig.charset = 'utf8mb4';
+const envelopDB = mysql.createPool(envelopConfig);
+
+bluebird.promisifyAll([Pool, Connection]);
+
+export { hubDB as default, sequencerDB, envelopDB };

src/ingestor.ts

@@ -31,6 +31,10 @@ const NETWORK_METADATA = {
   }
 };
 
+function shouldPinIpfs(type: string, message: any) {
+  return !(type === 'email-subscription' && message.email);
+}
+
 export default async function ingestor(req) {
   if (flaggedIps.includes(sha256(getIp(req)))) {
     return Promise.reject('unauthorized');
@@ -92,7 +96,11 @@ export default async function ingestor(req) {
     }
 
     let aliased = false;
-    if (!['settings', 'alias', 'profile'].includes(type)) {
+    if (
+      !['settings', 'alias', 'profile', 'email-subscription', 'delete-email-subscription'].includes(
+        type
+      )
+    ) {
       if (!message.space) return Promise.reject('unknown space');
 
       try {
@@ -106,7 +114,16 @@ export default async function ingestor(req) {
     }
 
     // Check if signing address is an alias
-    const aliasTypes = ['follow', 'unfollow', 'subscribe', 'unsubscribe', 'profile', 'statement'];
+    const aliasTypes = [
+      'follow',
+      'unfollow',
+      'subscribe',
+      'unsubscribe',
+      'profile',
+      'statement',
+      'email-subscription',
+      'delete-email-subscription'
+    ];
     const aliasOptionTypes = ['vote', 'vote-array', 'vote-string', 'proposal', 'delete-proposal'];
     if (body.address !== message.from) {
       if (!aliasTypes.includes(type) && !aliasOptionTypes.includes(type))
@@ -206,6 +223,13 @@ export default async function ingestor(req) {
       type = 'vote';
     }
 
+    if (type === 'email-subscription') {
+      payload = {
+        email: message.email,
+        subscriptions: message.subscriptions
+      };
+    }
+
     let legacyBody: any = {
       address: message.from,
       msg: JSON.stringify({
@@ -245,7 +269,7 @@ export default async function ingestor(req) {
         ...restBody
       };
       [pinned, receipt] = await Promise.all([
-        pin(ipfsBody, process.env.PINEAPPLE_URL),
+        shouldPinIpfs(type, message) ? pin(ipfsBody, process.env.PINEAPPLE_URL) : { cid: '' },
         issueReceipt(formattedSignature)
       ]);
     } catch (e) {

src/writer/delete-email-subscription.ts

@@ -0,0 +1,22 @@
+import { envelopDB } from '../helpers/mysql';
+
+type Message = { address: string };
+
+export async function verify(message: Message): Promise<boolean> {
+  const result = await envelopDB.queryAsync(
+    'SELECT * FROM subscribers WHERE address = ? AND verified > 0 LIMIT 1',
+    [message.address]
+  );
+
+  if (!result[0]) {
+    return Promise.reject('user not subscribed');
+  }
+
+  return true;
+}
+
+export async function action(message: Message): Promise<void> {
+  await envelopDB.queryAsync('DELETE FROM subscribers WHERE address = ? AND verified > 0 LIMIT 1', [
+    message.address
+  ]);
+}

src/writer/email-subscription.ts

@@ -0,0 +1,85 @@
+import snapshot from '@snapshot-labs/snapshot.js';
+import log from '../helpers/log';
+import { envelopDB } from '../helpers/mysql';
+import { jsonParse } from '../helpers/utils';
+
+type Message = { msg: string; address: string };
+type Payload = {
+  email?: string;
+  subscriptions?: string[];
+};
+
+function extractPayload(message: Message): Payload {
+  return jsonParse(message.msg).payload;
+}
+
+export async function verify(message: Message): Promise<boolean> {
+  const payload = extractPayload(message);
+
+  const schemaIsValid = snapshot.utils.validateSchema(snapshot.schemas.emailSubscription, payload);
+  if (schemaIsValid !== true) {
+    log.warn(`[writer] Wrong email subscription format ${JSON.stringify(schemaIsValid)}`);
+    return Promise.reject('wrong email subscription format');
+  }
+
+  if (payload.email?.length) {
+    return verifySubscriptionCreation(message, payload);
+  } else {
+    return verifySubscriptionUpdate(message, payload);
+  }
+}
+
+export async function action(message: Message): Promise<void> {
+  const payload = extractPayload(message);
+
+  if (payload.email?.length) {
+    await createAction(message, payload);
+  } else {
+    await updateAction(message, payload);
+  }
+}
+
+async function verifySubscriptionCreation(message: Message, payload: Payload): Promise<boolean> {
+  const result = await envelopDB.queryAsync(
+    `SELECT * FROM subscribers WHERE address = ? AND email = ? LIMIT 1`,
+    [message.address, payload.email]
+  );
+
+  if (result[0]) {
+    return Promise.reject('email already subscribed');
+  }
+
+  return true;
+}
+
+async function verifySubscriptionUpdate(message: Message, payload: Payload): Promise<boolean> {
+  const result = await envelopDB.queryAsync(
+    `SELECT * FROM subscribers WHERE address = ? ORDER BY verified DESC LIMIT 1`,
+    [message.address, payload.email]
+  );
+
+  if (!result[0]) {
+    return Promise.reject('email not subscribed');
+  }
+
+  if (!result[0].verified) {
+    return Promise.reject('email not verified');
+  }
+
+  return true;
+}
+
+async function createAction(message: Message, payload: Payload) {
+  await envelopDB.queryAsync(`INSERT INTO subscribers (email, address, created) VALUES(?, ?, ?)`, [
+    payload.email,
+    message.address,
+    (Date.now() / 1e3).toFixed()
+  ]);
+}
+
+async function updateAction(message: Message, payload: Payload) {
+  await envelopDB.queryAsync(
+    `UPDATE subscribers SET subscriptions = ? WHERE address = ? AND verified > 0 LIMIT 1`,
+    [JSON.stringify(payload.subscriptions), message.address]
+  );
+}

src/writer/index.ts

@@ -1,6 +1,8 @@
 import * as alias from './alias';
+import * as deleteEmailSubscription from './delete-email-subscription';
 import * as deleteProposal from './delete-proposal';
 import * as deleteSpace from './delete-space';
+import * as emailSubscription from './email-subscription';
 import * as flagProposal from './flag-proposal';
 import * as follow from './follow';
 import * as profile from './profile';
@@ -25,6 +27,8 @@ export default {
   unfollow,
   subscribe,
   unsubscribe,
+  'email-subscription': emailSubscription,
+  'delete-email-subscription': deleteEmailSubscription,
   alias,
   profile,
   statement

src/writer/statement.ts

@@ -5,6 +5,7 @@ import { DEFAULT_NETWORK_ID, jsonParse, NETWORK_IDS } from '../helpers/utils';
 
 export async function verify(body): Promise<any> {
   const msg = jsonParse(body.msg, {});
+
   const schemaIsValid = snapshot.utils.validateSchema(snapshot.schemas.statement, msg.payload);
   if (schemaIsValid !== true) {
     log.warn(`[writer] Wrong statement format ${JSON.stringify(schemaIsValid)}`);

test/.env.test

@@ -1,5 +1,6 @@
 HUB_DATABASE_URL=mysql://root:[email protected]:3306/snapshot_sequencer_test
 SEQ_DATABASE_URL=mysql://root:[email protected]:3306/snapshot_sequencer_test
+ENVELOP_DATABASE_URL=mysql://root:[email protected]:3306/snapshot_sequencer_test
 NETWORK=mainnet
 RELAYER_PK=01686849e86499c1860ea0afc97f29c11018cbac049abf843df875c60054076e
 NODE_ENV=test

test/integration/writer/delete-email-subscription.test.ts

@@ -0,0 +1,71 @@
+import db, { envelopDB, sequencerDB } from '../../../src/helpers/mysql';
+import { action, verify } from '../../../src/writer/delete-email-subscription';
+
+describe('writer/delete-subscription', () => {
+  const TEST_PREFIX = 'test-delete-subscription';
+
+  afterAll(async () => {
+    await envelopDB.queryAsync('DELETE FROM subscribers');
+    await envelopDB.endAsync();
+    await db.endAsync();
+    await sequencerDB.endAsync();
+  });
+
+  describe('verify()', () => {
+    beforeAll(async () => {
+      await Promise.all([
+        envelopDB.queryAsync(
+          'INSERT INTO subscribers SET address = ?, email = ?, subscriptions = ?, created = ?, verified = ?',
+          [`${TEST_PREFIX}-0x0`, '[email protected]', '[]', 0, 0]
+        ),
+        envelopDB.queryAsync(
+          'INSERT INTO subscribers SET address = ?, email = ?, subscriptions = ?, created = ?, verified = ?',
+          [`${TEST_PREFIX}-0x1`, '[email protected]', '[]', 0, 1]
+        )
+      ]);
+    });
+
+    it('rejects when the address is not subscribed', () => {
+      return expect(verify({ address: '0x0' })).rejects.toEqual(`user not subscribed`);
+    });
+
+    it('rejects when the address is not verified', () => {
+      return expect(verify({ address: `${TEST_PREFIX}-0x0` })).rejects.toEqual(
+        `user not subscribed`
+      );
+    });
+
+    it('resolves when the address is verified', () => {
+      expect(verify({ address: `${TEST_PREFIX}-0x1` })).resolves;
+    });
+  });
+
+  describe('action()', () => {
+    const address = `${TEST_PREFIX}-0x3`;
+
+    beforeAll(async () => {
+      await Promise.all([
+        envelopDB.queryAsync(
+          'INSERT INTO subscribers SET address = ?, email = ?, subscriptions = ?, created = ?, verified = ?',
+          [address, '[email protected]', '[]', 0, 0]
+        ),
+        envelopDB.queryAsync(
+          'INSERT INTO subscribers SET address = ?, email = ?, subscriptions = ?, created = ?, verified = ?',
+          [address, '[email protected]', '[]', 0, 1]
+        )
+      ]);
+    });
+
+    it('deletes the subscription', async () => {
+      await action({ address: address });
+
+      const results = await envelopDB.queryAsync('SELECT * FROM subscribers WHERE address = ?', [
+        address
+      ]);
+
+      // Only delete the verified subscription
+      expect(results.length).toBe(1);
+      expect(results[0].email).toEqual('[email protected]');
+    });
+  });
+});

test/integration/writer/email-subscription.test.ts

@@ -0,0 +1,61 @@
+import db, { envelopDB, sequencerDB } from '../../../src/helpers/mysql';
+import { action, verify } from '../../../src/writer/email-subscription';
+
+describe('writer/subscription', () => {
+  const TEST_PREFIX = 'test-subscription';
+  const msg = JSON.stringify({ payload: { email: '[email protected]', subscriptions: [] } });
+
+  afterAll(async () => {
+    await envelopDB.queryAsync('DELETE FROM subscribers');
+    await envelopDB.endAsync();
+    await db.endAsync();
+    await sequencerDB.endAsync();
+  });
+
+  describe('verify()', () => {
+    const address = `${TEST_PREFIX}-0x0`;
+    const invalidMsg = JSON.stringify({
+      payload: { email: 'not an email' }
+    });
+
+    beforeAll(async () => {
+      await envelopDB.queryAsync(
+        'INSERT INTO subscribers SET address = ?, email = ?, subscriptions = ?, created = ?, verified = ?',
+        [address, '[email protected]', '[]', 0, 0]
+      );
+    });
+
+    it('rejects when the address is already subscribed', () => {
+      return expect(verify({ address: address, msg })).rejects.toEqual('email already subscribed');
+    });
+
+    it('rejects when the subscription type is not valid', () => {
+      return expect(verify({ address: address, msg: invalidMsg })).rejects.toEqual(
+        'wrong email subscription format'
+      );
+    });
+
+    it('resolves when all args are valid', () => {
+      return expect(verify({ address: `${TEST_PREFIX}-0x1`, msg })).resolves.toBe(true);
+    });
+  });
+
+  describe('action()', () => {
+    const address = `${TEST_PREFIX}-0x1`;
+
+    it('creates a subscription', async () => {
+      await action({
+        address: address,
+        msg
+      });
+
+      const result = await envelopDB.queryAsync(
+        `SELECT * FROM subscribers WHERE address = ? LIMIT 1`,
+        [address]
+      );
+
+      expect(result[0].email).toEqual('[email protected]');
+      expect(result[0].verified).toEqual(0);
+    });
+  });
+});