feat: Store message on sequencer DB (#190)

* feat: Store message on sequencer DB * Add SEQ_DATABASE_URL * Update src/helpers/mysql.ts Co-authored-by: Wan <[email protected]> * Update src/helpers/highlight.ts Co-authored-by: Wan <[email protected]> * chore: fix tests * Update highlight.ts * Fix test cases --------- Co-authored-by: Wan <[email protected]>
snapshot-labs · Oct 9, 2023 · 45073f5 · 45073f5
1 parent 0d3a7f5
commit 45073f5
Show file tree

Hide file tree

Showing 6 changed files with 59 additions and 19 deletions.
diff --git a/.env.example b/.env.example
@@ -1,7 +1,8 @@
 PORT=3001
 NETWORK=testnet
 MAINTENANCE=
-DATABASE_URL=mysql://...
+HUB_DATABASE_URL=mysql://...
+SEQ_DATABASE_URL=mysql://...
 RELAYER_PK=0x123...
 DEFAULT_NETWORK=1
 SHUTTER_URL=https://...

diff --git a/src/helpers/highlight.ts b/src/helpers/highlight.ts
@@ -1,8 +1,9 @@
-import db from './mysql';
+import { default as hubDB, sequencerDB } from './mysql';
 
 export async function storeMsg(id, ipfs, address, version, timestamp, space, type, sig, receipt) {
   const query = 'INSERT INTO messages SET ?';
-  await db.queryAsync(query, [
+  // TODO: remove this once migration is done
+  const result = await hubDB.queryAsync(query, [
     {
       id,
       ipfs,
@@ -15,9 +16,26 @@ export async function storeMsg(id, ipfs, address, version, timestamp, space, typ
       receipt
     }
   ]);
+  if (result.insertId) {
+    // TODO: remove IGNORE once migration is done
+    await sequencerDB.queryAsync('INSERT IGNORE INTO messages SET ?', [
+      {
+        id,
+        mci: result.insertId,
+        ipfs,
+        address,
+        version,
+        timestamp,
+        space,
+        type,
+        sig,
+        receipt
+      }
+    ]);
+  }
 }
 
 export async function doesMessageExist(id: string): Promise<boolean> {
-  const result = await db.queryAsync('SELECT 1 FROM messages WHERE id = ? LIMIT 1', id);
+  const result = await sequencerDB.queryAsync('SELECT 1 FROM messages WHERE id = ? LIMIT 1', id);
   return result.length > 0;
 }
diff --git a/src/helpers/mysql.ts b/src/helpers/mysql.ts
@@ -9,17 +9,31 @@ const connectionLimit = parseInt(process.env.CONNECTION_LIMIT ?? '25');
 log.info(`[mysql] connection limit ${connectionLimit}`);
 
 // @ts-ignore
-const config = parse(process.env.DATABASE_URL);
-config.connectionLimit = connectionLimit;
-config.multipleStatements = true;
-config.database = config.path[0];
-config.host = config.hosts[0].name;
-config.port = config.hosts[0].port;
-config.connectTimeout = 60e3;
-config.acquireTimeout = 60e3;
-config.timeout = 60e3;
-config.charset = 'utf8mb4';
+const hubConfig = parse(process.env.HUB_DATABASE_URL);
+hubConfig.connectionLimit = connectionLimit;
+hubConfig.multipleStatements = true;
+hubConfig.database = hubConfig.path[0];
+hubConfig.host = hubConfig.hosts[0].name;
+hubConfig.port = hubConfig.hosts[0].port;
+hubConfig.connectTimeout = 60e3;
+hubConfig.acquireTimeout = 60e3;
+hubConfig.timeout = 60e3;
+hubConfig.charset = 'utf8mb4';
 bluebird.promisifyAll([Pool, Connection]);
-const db = mysql.createPool(config);
+const hubDB = mysql.createPool(hubConfig);
 
-export default db;
+// @ts-ignore
+const sequencerConfig = parse(process.env.SEQ_DATABASE_URL);
+sequencerConfig.connectionLimit = connectionLimit;
+sequencerConfig.multipleStatements = true;
+sequencerConfig.database = sequencerConfig.path[0];
+sequencerConfig.host = sequencerConfig.hosts[0].name;
+sequencerConfig.port = sequencerConfig.hosts[0].port;
+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 };
diff --git a/test/.env.test b/test/.env.test
@@ -1,3 +1,4 @@
-DATABASE_URL=mysql://root:[email protected]:3306/snapshot_sequencer_test
+HUB_DATABASE_URL=mysql://root:[email protected]:3306/snapshot_sequencer_test
+SEQ_DATABASE_URL=mysql://root:[email protected]:3306/snapshot_sequencer_test
 NETWORK=main
 RELAYER_PK=01686849e86499c1860ea0afc97f29c11018cbac049abf843df875c60054076e
diff --git a/test/integration/helpers/highlight.test.ts b/test/integration/helpers/highlight.test.ts
@@ -3,6 +3,12 @@ import db from '../../../src/helpers/mysql';
 
 describe('highlight', () => {
   describe('doesMessageExist()', () => {
+    beforeAll(async () => {
+      await db.queryAsync(
+        'DELETE from snapshot_sequencer_test.messages where id = ?',
+        'test-exists'
+      );
+    });
     afterAll(async () => {
       await db.queryAsync(
         'DELETE from snapshot_sequencer_test.messages where id = ?',

diff --git a/test/integration/ingestor.test.ts b/test/integration/ingestor.test.ts
@@ -3,7 +3,7 @@ import proposalInput from '../fixtures/ingestor-payload/proposal.json';
 import voteInput from '../fixtures/ingestor-payload/vote.json';
 import cloneDeep from 'lodash/cloneDeep';
 import omit from 'lodash/omit';
-import db from '../../src/helpers/mysql';
+import { default as db, sequencerDB } from '../../src/helpers/mysql';
 import relayer from '../../src/helpers/relayer';
 
 jest.mock('../../src/helpers/moderation', () => {
@@ -99,7 +99,7 @@ describe('ingestor', () => {
   afterAll(async () => {
     await db.queryAsync('DELETE FROM snapshot_sequencer_test.proposals;');
     await db.queryAsync('DELETE FROM snapshot_sequencer_test.messages;');
-    return db.endAsync();
+    return await db.endAsync();
   });
 
   it('rejects when the submitter IP is banned', async () => {