fix: use graphql directly

package.json

@@ -15,6 +15,7 @@
   },
   "prettier": "@snapshot-labs/prettier-config",
   "dependencies": {
+    "@apollo/client": "^3.8.3",
     "@ethersproject/wallet": "^5.7.0",
     "@pusher/push-notifications-server": "^1.2.5",
     "@snapshot-labs/snapshot-metrics": "^1.3.0",
@@ -28,13 +29,11 @@
     "discord.js": "^14.3.0",
     "dotenv": "^10.0.0",
     "express": "^4.17.1",
-    "express-prom-bundle": "^6.6.0",
-    "json-to-graphql-query": "^2.2.5",
+    "graphql": "^16.8.0",
     "lodash.chunk": "^4.2.0",
     "mysql": "^2.18.1",
     "node-fetch": "^1.7.0",
     "nodemon": "^3.0.1",
-    "prom-client": "^14.2.0",
     "remove-markdown": "^0.3.0",
     "ts-node": "^10.9.1",
     "typescript": "^4.8.3"

src/events.ts

@@ -1,7 +1,7 @@
 import snapshot from '@snapshot-labs/snapshot.js';
 import { capture } from '@snapshot-labs/snapshot-sentry';
 import db from './helpers/mysql';
-import { getProposal, getSubscribers } from './helpers/utils';
+import { getProposal, getSubscribers } from './helpers/snapshot';
 import providers from './providers';
 
 const DELAY = 5;

src/helpers/snapshot.ts

@@ -0,0 +1,190 @@
+import { gql, ApolloClient, InMemoryCache, HttpLink } from '@apollo/client/core';
+import { capture } from '@snapshot-labs/snapshot-sentry';
+import { fetchWithKeepAlive } from './utils';
+
+const HUB_URL = `${process.env.HUB_URL || 'https://hub.snapshot.org'}/graphql`;
+
+const client = new ApolloClient({
+  link: new HttpLink({ uri: HUB_URL, fetch: fetchWithKeepAlive as any }),
+  cache: new InMemoryCache({
+    addTypename: false
+  }),
+  defaultOptions: {
+    query: {
+      fetchPolicy: 'no-cache'
+    }
+  }
+});
+
+type Message = {
+  mci: number;
+  id: string;
+  ipfs: string;
+  type: string;
+  timestamp: number;
+  space: string;
+};
+
+type Space = {
+  id: string;
+  name: string;
+  avatar?: string;
+};
+
+type Proposal = {
+  space: Space;
+  id: string;
+  type: string;
+  author: string;
+  title: string;
+  body: string;
+  choices: string[];
+  created: number;
+  start: number;
+  end: number;
+  link: string;
+  snapshot: string;
+};
+
+type Subscriber = {
+  address: string;
+};
+
+const MESSAGES_QUERY = gql`
+  query Messages(
+    $type_in: [String]
+    $first: Int
+    $mci: Int
+    $orderDirection: OrderDirection
+    $orderBy: String
+  ) {
+    messages(
+      where: { mci_gt: $mci, type_in: $type_in }
+      first: $first
+      orderBy: $orderBy
+      orderDirection: $orderDirection
+    ) {
+      mci
+      id
+      ipfs
+      type
+      timestamp
+      space
+    }
+  }
+`;
+
+const SPACE_QUERY = gql`
+  query Space($id: String) {
+    space(id: $id) {
+      id
+      name
+    }
+  }
+`;
+
+const PROPOSAL_QUERY = gql`
+  query Proposal($id: String) {
+    proposal(id: $id) {
+      space {
+        id
+        name
+        avatar
+      }
+      id
+      type
+      author
+      title
+      body
+      choices
+      created
+      start
+      end
+      link
+      snapshot
+    }
+  }
+`;
+
+const SUBSCRIPTIONS_QUERY = gql`
+  query Subscriptions($space: String) {
+    subscriptions(where: { space: $space }) {
+      address
+    }
+  }
+`;
+
+export async function getNextMessages(mci: number) {
+  try {
+    const {
+      data: { messages }
+    }: { data: { messages: Message[] } } = await client.query({
+      query: MESSAGES_QUERY,
+      variables: {
+        mci,
+        type_in: ['proposal', 'delete-proposal'],
+        first: 10,
+        orderBy: 'mci',
+        orderDirection: 'asc'
+      }
+    });
+
+    return messages;
+  } catch (e: any) {
+    capture(e);
+    return [];
+  }
+}
+
+export async function getSpace(id: string) {
+  try {
+    const {
+      data: { space }
+    }: { data: { space: Space | null } } = await client.query({
+      query: SPACE_QUERY,
+      variables: {
+        id
+      }
+    });
+
+    return space;
+  } catch (e: any) {
+    capture(e);
+    return null;
+  }
+}
+
+export async function getProposal(id: string) {
+  try {
+    const {
+      data: { proposal }
+    }: { data: { proposal: Proposal | null } } = await client.query({
+      query: PROPOSAL_QUERY,
+      variables: {
+        id
+      }
+    });
+
+    return proposal;
+  } catch (e: any) {
+    capture(e);
+    return null;
+  }
+}
+
+export async function getSubscribers(space: string) {
+  try {
+    const {
+      data: { subscribers }
+    }: { data: { subscribers: Subscriber[] | null } } = await client.query({
+      query: SUBSCRIPTIONS_QUERY,
+      variables: {
+        space
+      }
+    });
+    return (subscribers || []).map(subscriber => subscriber.address);
+  } catch (e: any) {
+    capture(e);
+    return [];
+  }
+}

src/helpers/utils.ts

@@ -1,8 +1,6 @@
 import { createHash } from 'crypto';
-import snapshot from '@snapshot-labs/snapshot.js';
-import { capture } from '@snapshot-labs/snapshot-sentry';
-
-const HUB_URL = process.env.HUB_URL || 'https://hub.snapshot.org';
+import http from 'node:http';
+import https from 'node:https';
 
 export function shortenAddress(str = '') {
   return `${str.slice(0, 6)}...${str.slice(str.length - 4)}`;
@@ -12,83 +10,14 @@ export function sha256(str) {
   return createHash('sha256').update(str).digest('hex');
 }
 
-export async function getSubscribers(space) {
-  let subscriptions: { [key: string]: any } = [];
-  const query = {
-    subscriptions: {
-      __args: {
-        where: { space }
-      },
-      address: true
-    }
-  };
-  try {
-    const result = await snapshot.utils.subgraphRequest(`${HUB_URL}/graphql`, query);
-    subscriptions = result.subscriptions || [];
-  } catch (e: any) {
-    capture(e, { contexts: { input: { query, space } } });
-  }
-  return subscriptions.map(subscription => subscription.address);
-}
-
-export async function getProposal(id) {
-  let proposal: { [key: string]: any } | null = null;
-  const query = {
-    proposal: {
-      __args: {
-        id
-      },
-      space: {
-        id: true,
-        name: true,
-        avatar: true
-      },
-      id: true,
-      type: true,
-      author: true,
-      title: true,
-      body: true,
-      choices: true,
-      start: true,
-      end: true,
-      link: true,
-      snapshot: true
-    }
-  };
+const agentOptions = { keepAlive: true };
+const httpAgent = new http.Agent(agentOptions);
+const httpsAgent = new https.Agent(agentOptions);
 
-  try {
-    const result = await snapshot.utils.subgraphRequest(`${HUB_URL}/graphql`, query);
-    if (result.errors) {
-      console.error(`[events] Errors in subgraph request for proposal id: ${id}`);
-    }
-    proposal = result.proposal || null;
-    return proposal;
-  } catch (e: any) {
-    capture(e, { contexts: { input: { query, id } } });
-    return null;
-  }
+function agent(url: string) {
+  return new URL(url).protocol === 'http:' ? httpAgent : httpsAgent;
 }
 
-export async function getSpace(id) {
-  let space: { [key: string]: any } | null = null;
-  const query = {
-    space: {
-      __args: {
-        id
-      },
-      id: true,
-      name: true
-    }
-  };
-  try {
-    const result = await snapshot.utils.subgraphRequest(`${HUB_URL}/graphql`, query);
-    if (result.errors) {
-      console.error(`[events] Errors in subgraph request for proposal id: ${id}`);
-    }
-    space = result.space || null;
-    return space;
-  } catch (e: any) {
-    capture(e, { contexts: { input: { query, id } } });
-    return null;
-  }
-}
+export const fetchWithKeepAlive = (uri: any, options: any = {}) => {
+  return fetch(uri, { agent: agent(uri), ...options });
+};

src/providers/beams.ts

@@ -10,8 +10,10 @@ const beams = new PushNotifications({
   secretKey: process.env.SERVICE_PUSHER_BEAMS_SECRET_KEY ?? ''
 });
 
-export async function send(event, proposal, subscribers) {
-  if (!SERVICE_PUSH_NOTIFICATIONS || event.event !== 'proposal/start') return;
+export async function send(event, proposal, subscribers: string[]) {
+  if (!SERVICE_PUSH_NOTIFICATIONS || event.event !== 'proposal/start' || subscribers.length === 0) {
+    return;
+  }
 
   const walletsChunks = chunk(subscribers, 100);
   const end = timeOutgoingRequest.startTimer({ provider: 'beams' });

src/providers/discord.ts

@@ -16,7 +16,7 @@ import {
 import db from '../helpers/mysql';
 import removeMd from 'remove-markdown';
 import { shortenAddress } from '../helpers/utils';
-import { getSpace } from '../helpers/utils';
+import { getSpace } from '../helpers/snapshot';
 import { capture } from '@snapshot-labs/snapshot-sentry';
 import { timeOutgoingRequest } from '../helpers/metrics';
 

src/providers/xmtp.ts

@@ -1,9 +1,9 @@
 import { ApiUrls, Client } from '@xmtp/xmtp-js';
 import { Wallet } from '@ethersproject/wallet';
-import { getSpace } from '../helpers/utils';
+import { capture } from '@snapshot-labs/snapshot-sentry';
+import { getSpace } from '../helpers/snapshot';
 import db from '../helpers/mysql';
 import { timeOutgoingRequest } from '../helpers/metrics';
-import { capture } from '@snapshot-labs/snapshot-sentry';
 
 const XMTP_PK = process.env.XMTP_PK || Wallet.createRandom().privateKey;
 const XMTP_ENV = (process.env.XMTP_ENV || 'dev') as keyof typeof ApiUrls;
@@ -94,7 +94,7 @@ export async function send(event, proposal, subscribers) {
 }
 
 async function sendMessages(addresses: string[], msg) {
-  if (!client) return;
+  if (!client || addresses.length === 0) return;
 
   const canMessage = await client.canMessage(addresses);