Merge pull request #1780 from Giveth/feat/stellar_integration

fix: change from WS to SSE

package.json

@@ -69,8 +69,7 @@
     "twitter-api-sdk": "^1.0.9",
     "type-graphql": "2.0.0-beta.1",
     "typedi": "0.8.0",
-    "typeorm": "0.3.20",
-    "ws": "^8.18.0"
+    "typeorm": "0.3.20"
   },
   "lint-staged": {
     "*.ts": [

src/resolvers/draftDonationResolver.ts

@@ -27,6 +27,7 @@ import {
 import { RecurringDonation } from '../entities/recurringDonation';
 import { checkTransactions } from '../services/cronJobs/checkQRTransactionJob';
 import { findProjectById } from '../repositories/projectRepository';
+import { notifyDonationFailed } from '../services/sse/sse';
 
 const draftDonationEnabled = process.env.ENABLE_DRAFT_DONATION === 'true';
 const draftRecurringDonationEnabled =
@@ -419,9 +420,12 @@ export class DraftDonationResolver {
       );
 
       // Notify clients of new donation
-      (global as any).notifyDraftDonationFailed({
-        draftDonationId: id,
-        expiresAt: draftDonation.expiresAt,
+      notifyDonationFailed({
+        type: 'draft-donation-failed',
+        data: {
+          draftDonationId: id,
+          expiresAt: draftDonation.expiresAt,
+        },
       });
 
       return true;

src/server/bootstrap.ts

@@ -7,7 +7,7 @@ import { ApolloServer } from '@apollo/server';
 import { expressMiddleware } from '@apollo/server/express4';
 import { ApolloServerPluginSchemaReporting } from '@apollo/server/plugin/schemaReporting';
 import { ApolloServerPluginLandingPageGraphQLPlayground } from '@apollo/server-plugin-landing-page-graphql-playground';
-import express, { json, Request } from 'express';
+import express, { json, Request, Response } from 'express';
 import { Container } from 'typedi';
 import { Resource } from '@adminjs/typeorm';
 import { validate } from 'class-validator';
@@ -68,7 +68,7 @@ import { runDraftDonationMatchWorkerJob } from '../services/cronJobs/draftDonati
 import { runCheckUserSuperTokenBalancesJob } from '../services/cronJobs/checkUserSuperTokenBalancesJob';
 import { runCheckPendingRecurringDonationsCronJob } from '../services/cronJobs/syncRecurringDonationsWithNetwork';
 import { runCheckQRTransactionJob } from '../services/cronJobs/checkQRTransactionJob';
-import { startWebSocketServer } from '../services/ws/webSocketServer';
+import { addClient } from '../services/sse/sse';
 
 Resource.validate = validate;
 
@@ -290,13 +290,12 @@ export async function bootstrap() {
     app.post('/fiat_webhook', onramperWebhookHandler);
     app.post('/transak_webhook', webhookHandler);
 
-    const httpServer = http.createServer(app);
+    // Route to handle SSE connections
+    app.get('/events', (_req: Request, res: Response) => {
+      addClient(res);
+    });
 
-    // Start WebSocket server
-    const { notifyDonationAdded, notifyDraftDonationFailed } =
-      startWebSocketServer(httpServer);
-    (global as any).notifyDonationAdded = notifyDonationAdded;
-    (global as any).notifyDraftDonationFailed = notifyDraftDonationFailed;
+    const httpServer = http.createServer(app);
 
     await new Promise<void>((resolve, reject) => {
       httpServer

src/services/cronJobs/checkQRTransactionJob.ts

@@ -15,6 +15,7 @@ import { findUserById } from '../../repositories/userRepository';
 import { relatedActiveQfRoundForProject } from '../qfRoundService';
 import { QfRound } from '../../entities/qfRound';
 import { syncDonationStatusWithBlockchainNetwork } from '../donationService';
+import { notifyClients } from '../sse/sse';
 
 const STELLAR_HORIZON_API =
   (config.get('STELLAR_HORIZON_API_URL') as string) ||
@@ -182,9 +183,12 @@ export async function checkTransactions(
         });
 
         // Notify clients of new donation
-        (global as any).notifyDonationAdded({
-          donationId: returnedDonation.id,
-          draftDonationId: donation.id,
+        notifyClients({
+          type: 'new-donation',
+          data: {
+            donationId: returnedDonation.id,
+            draftDonationId: donation.id,
+          },
         });
 
         return;

src/services/sse/sse.ts

@@ -0,0 +1,45 @@
+import { Response } from 'express';
+
+let clients: Response[] = [];
+type TNewDonation = {
+  type: 'new-donation';
+  data: {
+    donationId: number;
+    draftDonationId: number;
+  };
+};
+
+type TDraftDonationFailed = {
+  type: 'draft-donation-failed';
+  data: {
+    draftDonationId: number;
+    expiresAt?: Date;
+  };
+};
+
+// Add a new client to the SSE stream
+export function addClient(res: Response) {
+  res.setHeader('Access-Control-Allow-Origin', '*');
+  res.setHeader('Content-Type', 'text/event-stream');
+  res.setHeader('Cache-Control', 'no-cache');
+  res.setHeader('Connection', 'keep-alive');
+
+  res.flushHeaders();
+
+  clients.push(res);
+
+  // Remove the client on disconnect
+  res.on('close', () => {
+    clients = clients.filter(client => client !== res);
+  });
+}
+
+// Notify all connected clients about a new donation
+export function notifyClients(data: TNewDonation) {
+  clients.forEach(client => client.write(`data: ${JSON.stringify(data)}\n\n`));
+}
+
+// Notify all connected clients about a failed donation
+export function notifyDonationFailed(data: TDraftDonationFailed) {
+  clients.forEach(client => client.write(`data: ${JSON.stringify(data)}\n\n`));
+}